File size: 11,315 Bytes
a1fe1e8 5fde0d1 91a684f a1fe1e8 5fde0d1 a1fe1e8 91a684f 5fde0d1 91a684f 5fde0d1 91a684f b4603ca ff6eb9d b4603ca a1fe1e8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 | 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()
|