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()