| import json |
| import unittest |
|
|
| from terrarium.engine import ( |
| ACTION_FAMILIES, StateError, apply_proposal, cognition_packet, create_random_resident, |
| create_resident, export_state, import_state, schedule_action, validate_state, |
| ) |
|
|
|
|
| class EngineTests(unittest.TestCase): |
| def setUp(self): |
| self.state = create_resident("Moss", "Glass Observatory", "Moss Machine", "brass seed") |
|
|
| def proposal(self, action, target): |
| return { |
| "action": action, "target": target, |
| "intention": "Moss follows the scheduled action carefully.", |
| "narration": "Moss pauses beneath the amber light and acts with deliberate curiosity.", |
| } |
|
|
| def quiet_proposal(self, action, target): |
| return { |
| "action": action, "target": target, |
| "intention": "The scheduled life action proceeds.", |
| "narration": f"The resident chooses {action.replace('_', ' ')}.", |
| } |
|
|
| def test_genesis_is_valid(self): |
| validate_state(self.state) |
| self.assertEqual(self.state["cycle"], 0) |
|
|
| def test_scheduler_penalizes_immediate_repetition(self): |
| first, target = schedule_action(self.state) |
| state = apply_proposal(self.state, self.proposal(first, target), first, target) |
| second, _ = schedule_action(state) |
| self.assertNotEqual(first, second) |
|
|
| def test_model_cannot_override_scheduled_action(self): |
| action, target = schedule_action(self.state) |
| bad = self.proposal("rest", "self") |
| with self.assertRaises(StateError): |
| apply_proposal(self.state, bad, action, target) |
| self.assertEqual(self.state["cycle"], 0) |
|
|
| def test_export_import_round_trip(self): |
| restored = import_state(export_state(self.state)) |
| self.assertEqual(restored, self.state) |
|
|
| def test_tampered_save_is_rejected(self): |
| envelope = json.loads(export_state(self.state)) |
| envelope["resident"]["cycle"] = 9 |
| with self.assertRaises(StateError): |
| import_state(json.dumps(envelope)) |
|
|
| def test_invalid_brain_proposal_leaves_state_unchanged(self): |
| action, target = schedule_action(self.state) |
| bad = self.proposal(action, target) |
| bad["narration"] = "" |
| before = json.dumps(self.state, sort_keys=True) |
| with self.assertRaises(StateError): |
| apply_proposal(self.state, bad, action, target) |
| self.assertEqual(json.dumps(self.state, sort_keys=True), before) |
|
|
| def test_rechecksummed_malformed_save_is_rejected(self): |
| envelope = json.loads(export_state(self.state)) |
| envelope["resident"]["needs"] = {"curiosity": "very"} |
| canonical = json.dumps(envelope["resident"], sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| import hashlib |
| envelope["sha256"] = hashlib.sha256(canonical.encode()).hexdigest() |
| with self.assertRaises(StateError): |
| import_state(json.dumps(envelope)) |
|
|
| def test_seeded_genesis_is_reproducible(self): |
| first = create_random_resident(seed=8675309) |
| second = create_random_resident(seed=8675309) |
| first.pop("created_at") |
| second.pop("created_at") |
| self.assertEqual(first, second) |
|
|
| def test_genesis_varies_meaningfully_across_seeds(self): |
| residents = [create_random_resident(seed=seed) for seed in range(24)] |
| fingerprints = { |
| ( |
| item["resident"]["name"], item["resident"]["form"], item["habitat"]["name"], |
| next(iter(item["artifacts"].values()))["name"], |
| tuple(item["profile"]["temperament"].values()), item["profile"]["dominant_drive"], |
| ) |
| for item in residents |
| } |
| self.assertGreaterEqual(len(fingerprints), 22) |
|
|
| def test_separate_arrivals_do_not_share_mutable_state(self): |
| first = create_random_resident(seed=101) |
| second = create_random_resident(seed=202) |
| first["memories"].append({"cycle": 0, "text": "Only the first resident remembers this.", "salience": 0.5}) |
| self.assertNotEqual(first["genesis"]["seed"], second["genesis"]["seed"]) |
| self.assertEqual(len(second["memories"]), 1) |
|
|
| def test_profile_reaches_cognition_packet(self): |
| resident = create_random_resident(seed=4242) |
| action, target = schedule_action(resident) |
| packet = cognition_packet(resident, action, target) |
| self.assertEqual(packet["genesis"], resident["genesis"]) |
| self.assertEqual(packet["profile"], resident["profile"]) |
|
|
| def test_legacy_v1_save_remains_compatible(self): |
| envelope = json.loads(export_state(self.state)) |
| envelope["resident"].pop("genesis") |
| envelope["resident"].pop("profile") |
| envelope["resident"].pop("mood") |
| envelope["resident"]["needs"].pop("comfort") |
| envelope["resident"]["needs"].pop("play") |
| canonical = json.dumps(envelope["resident"], sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| import hashlib |
| envelope["sha256"] = hashlib.sha256(canonical.encode()).hexdigest() |
| restored = import_state(json.dumps(envelope)) |
| validate_state(restored) |
| action, target = schedule_action(restored) |
| packet = cognition_packet(restored, action, target) |
| self.assertEqual(packet["genesis"]["version"], 0) |
| self.assertEqual(packet["profile"]["dominant_drive"], "curiosity") |
|
|
| def test_weighted_ecology_is_reproducible_and_balanced(self): |
| family_counts = {} |
| action_sequences = [] |
| ritual_count = 0 |
| for seed in range(12): |
| resident = create_random_resident(seed=seed) |
| actions = [] |
| for _ in range(100): |
| action, target = schedule_action(resident) |
| actions.append(action) |
| family = ACTION_FAMILIES[action] |
| family_counts[family] = family_counts.get(family, 0) + 1 |
| ritual_count += action == "perform_ritual" |
| resident = apply_proposal(resident, self.quiet_proposal(action, target), action, target) |
| self.assertGreaterEqual(len(set(actions)), 12) |
| self.assertFalse(any(left == right for left, right in zip(actions, actions[1:]))) |
| action_sequences.append(actions) |
| total = sum(family_counts.values()) |
| discovery_share = family_counts["discover"] / total |
| self.assertGreater(discovery_share, 0.12) |
| self.assertLess(discovery_share, 0.28) |
| self.assertGreater(sum(value for key, value in family_counts.items() if key != "discover") / total, 0.70) |
| self.assertTrue(all(family_counts.get(family, 0) / total > 0.04 for family in { |
| "discover", "enjoy", "live", "create", "play", "reflect", "restore", |
| })) |
| self.assertGreater(ritual_count, 10) |
|
|
| replay = create_random_resident(seed=0) |
| replay_actions = [] |
| for _ in range(100): |
| action, target = schedule_action(replay) |
| replay_actions.append(action) |
| replay = apply_proposal(replay, self.quiet_proposal(action, target), action, target) |
| self.assertEqual(replay_actions, action_sequences[0]) |
|
|
| def test_moods_persist_then_change_within_bounds(self): |
| resident = create_random_resident(seed=31337) |
| observed = [] |
| for _ in range(40): |
| observed.append(resident["mood"]["name"]) |
| action, target = schedule_action(resident) |
| resident = apply_proposal(resident, self.quiet_proposal(action, target), action, target) |
| self.assertGreaterEqual(resident["mood"]["remaining"], 1) |
| self.assertLessEqual(resident["mood"]["remaining"], 5) |
| runs = [] |
| for mood in observed: |
| if not runs or runs[-1][0] != mood: |
| runs.append([mood, 1]) |
| else: |
| runs[-1][1] += 1 |
| self.assertGreater(len(runs), 5) |
| self.assertTrue(any(length >= 2 for _, length in runs)) |
|
|
| def test_export_import_preserves_future_ecology_sequence(self): |
| original = create_random_resident(seed=404) |
| for _ in range(25): |
| action, target = schedule_action(original) |
| original = apply_proposal(original, self.quiet_proposal(action, target), action, target) |
| restored = import_state(export_state(original)) |
| original_future = [] |
| restored_future = [] |
| for _ in range(30): |
| action, target = schedule_action(original) |
| original_future.append((action, target, original["mood"].copy())) |
| original = apply_proposal(original, self.quiet_proposal(action, target), action, target) |
| action, target = schedule_action(restored) |
| restored_future.append((action, target, restored["mood"].copy())) |
| restored = apply_proposal(restored, self.quiet_proposal(action, target), action, target) |
| self.assertEqual(original_future, restored_future) |
|
|
| def test_invalid_mood_is_rejected(self): |
| resident = create_random_resident(seed=78) |
| resident["mood"] = {"name": "obsessed", "remaining": 99} |
| with self.assertRaises(StateError): |
| validate_state(resident) |
|
|
| def test_invalid_profile_is_rejected(self): |
| resident = create_random_resident(seed=77) |
| resident["profile"]["dominant_drive"] = "escape" |
| with self.assertRaises(StateError): |
| validate_state(resident) |
|
|
| def test_scheduler_never_addresses_observer(self): |
| resident = create_random_resident(seed=991) |
| resident["observer_message"] = "Legacy save text that must be ignored." |
| for _ in range(20): |
| action, target = schedule_action(resident) |
| self.assertNotEqual(action, "address_observer") |
| self.assertNotEqual(target, "observer") |
| resident = apply_proposal(resident, self.proposal(action, target), action, target) |
|
|
| def test_engine_memories_follow_actions_without_duplicates_or_new_threads(self): |
| resident = create_random_resident(seed=4444) |
| initial_threads = json.loads(json.dumps(resident["threads"])) |
| for _ in range(200): |
| action, target = schedule_action(resident) |
| resident = apply_proposal(resident, self.quiet_proposal(action, target), action, target) |
| memory_texts = [item["text"].casefold() for item in resident["memories"]] |
| self.assertEqual(len(memory_texts), len(set(memory_texts))) |
| self.assertEqual(resident["threads"], initial_threads) |
| self.assertTrue(any("enjoy" in item or "comfort" in item for item in memory_texts)) |
|
|
| def test_legacy_observer_action_history_remains_importable(self): |
| resident = create_random_resident(seed=992) |
| resident["recent_actions"].append({"cycle": 1, "action": "address_observer", "target": "observer"}) |
| resident["history"].append({ |
| "cycle": 1, "action": "address_observer", "target": "observer", |
| "intention": "A legacy intention.", "narration": "A legacy narration.", "before_hash": "legacy", |
| }) |
| validate_state(resident) |
| restored = import_state(export_state(resident)) |
| self.assertEqual(restored["history"][-1]["action"], "address_observer") |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
