"""Synthetic planning tests only; no input generation, files or brain runs.""" import copy from collections import Counter import unittest from unittest.mock import patch import evaluation_plan as m import trial_plan def synthetic_config(trial, reversal_points=(0, 50)): """An explicit TEST choice, not a proposed production default. Controls here use a simple fixed epoch fraction; the planner never supplies this policy. Only paired/reversal checkpoints track actual scheduled USs. """ config = dict(schema=m.CONFIG_SCHEMA, probe_steps=1200, cs_start_tick=0, cs_stop_tick_exclusive=1000, reversal_checkpoints=list(reversal_points), baseline_execution="execute_each_snapshot", snapshot_binding_policy="explicit_completed_epochs", snapshot_bindings=[]) for block in trial["plan"]["blocks"]: for suite in block["suites"]: points = list(reversal_points) if suite["assay"] == "reversal" else block["evaluation_checkpoints"] for arm in suite["arms"]: for point in points: if point in (0, 50): completed = 0 if point == 0 else len(arm["epochs"]) elif arm["arm"] in ("paired", "reversal", "no_reversal"): rewarded = [i + 1 for i, e in enumerate(arm["epochs"]) if e["us"]["requested"]] completed = rewarded[point - 1] else: completed = point * len(arm["epochs"]) // 50 config["snapshot_bindings"].append(dict(block_id=block["block_id"], assay=suite["assay"], arm=arm["arm"], checkpoint=point, completed_epochs=completed)) return config class EvaluationPlan(unittest.TestCase): @classmethod def setUpClass(cls): # Existing synthetic master from the trial-plan tests, not a real choice. cls.trial = trial_plan.build_trial_plan(461) cls.config = synthetic_config(cls.trial) cls.plan = m.build_evaluation_plan(cls.trial, cls.config) def test_public_full_authority_reconstruction(self): self.assertTrue(m.validate_evaluation_plan(self.plan, self.trial, self.config)) def test_exact_inventory_budget_and_no_execution_claim(self): b = self.plan["budget"] self.assertEqual(b["probes_by_assay"], dict(normal=3920, delayed=3920, reversal=320, retention=480)) self.assertEqual(b["evaluation_probes"], 8640) self.assertEqual(b["snapshot_count"], 4320) self.assertEqual(b["evaluation_steps"], 10368000) self.assertEqual(b["planned_scientific_steps_including_evaluation"], 117968000) self.assertEqual(b["independent_evaluation_inputs"], 80) for name in ("execution_authorized", "training_executed", "inputs_generated", "candidate_generated", "learning_pass_claimed", "scientific_gate_added", "external_source_and_snapshot_authority_verified"): self.assertIs(self.plan[name], False) def test_full_cells_and_all_control_coverage(self): cells = {} for p in self.plan["probes"]: cells.setdefault((p["mode"], p["dan_type"], p["original_reinforced_cue"]), set()).add(p["training_replicate"]) self.assertEqual(p["odor"], trial_plan.ODORS[p["cue"]]) self.assertFalse(p["weight_updates_enabled"]) self.assertFalse(p["US_requested"]) self.assertFalse(p["feedback_to_training_or_idle"]) self.assertFalse(p["adds_training_replicate"]) self.assertEqual(len(cells), 8) self.assertTrue(all(v == {1, 2, 3, 4, 5} for v in cells.values())) first = self.trial["plan"]["blocks"][0]["block_id"] normal = [p for p in self.plan["probes"] if p["block_id"] == first and p["assay"] == "normal"] self.assertEqual(Counter(p["arm"] for p in normal), Counter({arm: 14 for arm in trial_plan.ARMS})) def test_stream_keys_disjoint_and_reused_as_paired_probes(self): non_eval = {r["seed"] for r in self.trial["seed_manifest"]["independent_streams"] if r["split"] != "evaluation"} grouped = {} for p in self.plan["probes"]: grouped.setdefault((p["block_id"], p["cue"]), set()).add((p["input"]["seed"], p["input"]["stream_id"])) self.assertEqual(p["input"]["semantic_key"], [p["block_id"], "evaluation", "CS", p["cue"], "00", "base-probe"]) self.assertEqual(p["input"]["seed_bits"], 63) seeds = {next(iter(values))[0] for values in grouped.values()} self.assertEqual(len(grouped), 80) self.assertTrue(all(len(v) == 1 for v in grouped.values())) self.assertEqual(len(seeds), 80) self.assertFalse(seeds & non_eval) self.assertTrue(any(seed > 2**32 for seed in seeds)) def test_reversal_original_baseline_and_orientation(self): for s in self.plan["snapshots"]: if s["assay"] == "reversal": self.assertEqual(s["baseline_snapshot_id"], s["block_id"] + "/normal/paired/checkpoint-00") if s["checkpoint"] == 0: self.assertEqual(s["delta_source"], {"kind": "fork", "source": s["block_id"] + "/normal/paired/terminal-delta"}) self.assertEqual(s["neural_state_source"]["kind"], "common_frozen_reset") def test_retention_same_idle_lineage_independent_forks(self): grouped = {} for s in self.plan["snapshots"]: if s["assay"] != "retention": continue src = s["neural_state_source"] grouped.setdefault((s["block_id"], s["arm"]), []).append(src) self.assertFalse(src["reset_or_analytic_time_advance"]) self.assertFalse(s["snapshot_mutated_by_planning"]) self.assertEqual(s["baseline_snapshot_id"], s["block_id"] + "/normal/" + s["arm"] + "/checkpoint-00") self.assertEqual(len(grouped), 80) for rows in grouped.values(): self.assertEqual([r["elapsed_ticks"] for r in rows], [0, 10000, 100000]) self.assertEqual(len({r["idle_lineage_id"] for r in rows}), 1) self.assertEqual(rows[0]["kind"], "actual_terminal_checkpoint") self.assertEqual(rows[1]["kind"], "continuous_idle_checkpoint") def test_missing_choices_no_hidden_default(self): for key in m.CONFIG_FIELDS: bad = copy.deepcopy(self.config) del bad[key] with self.subTest(key=key), self.assertRaisesRegex(ValueError, "missing="): m.build_evaluation_plan(self.trial, bad) self.assertIn("snapshot_bindings", m.required_choices()) def test_bad_configuration_types_and_unimplemented_reuse(self): for key, value in (("probe_steps", 1200.0), ("cs_start_tick", False), ("cs_stop_tick_exclusive", True), ("cs_stop_tick_exclusive", 1201), ("reversal_checkpoints", [False, 50]), ("reversal_checkpoints", [50]), ("baseline_execution", "reuse_identical"), ("snapshot_binding_policy", "guess_from_paired")): bad = copy.deepcopy(self.config) bad[key] = value with self.subTest(key=key, value=value), self.assertRaises(ValueError): m._assemble(self.trial, bad) def test_binding_coverage_order_types_and_pairing_boundary(self): changes = ( lambda rows: rows.pop(), lambda rows: rows.reverse(), lambda rows: rows[0].update(checkpoint=False), lambda rows: rows[0].update(completed_epochs=False), lambda rows: rows[1].update(completed_epochs=0), lambda rows: rows[6].update(completed_epochs=99), lambda rows: rows[1].update(completed_epochs=100), lambda rows: rows[0].update(extra="unknown"), ) for index, change in enumerate(changes): bad = copy.deepcopy(self.config) change(bad["snapshot_bindings"]) with self.subTest(index=index), self.assertRaises(ValueError): m._assemble(self.trial, bad) def test_explicit_alternative_curve_changes_budget_not_seed_inventory(self): config = synthetic_config(self.trial, trial_plan.PROBE_CHECKPOINTS) result = m._assemble(self.trial, config) self.assertEqual(result["budget"]["evaluation_probes"], 9440) self.assertEqual(result["budget"]["probes_by_assay"]["reversal"], 1120) self.assertEqual(result["budget"]["independent_evaluation_inputs"], 80) config["cs_stop_tick_exclusive"] = 1200 alternate = m._assemble(self.trial, config) self.assertNotEqual(result["config_sha256"], alternate["config_sha256"]) self.assertTrue(all(p["cs_stop_tick_exclusive"] == 1200 for p in alternate["probes"])) def test_payload_mutations_rejected_even_when_self_hash_updated(self): changes = ( lambda p: p["probes"].pop(), lambda p: p["probes"].reverse(), lambda p: p["probes"][0]["input"].update(seed=1), lambda p: p["probes"][0].update(training_replicate=True), lambda p: p["probes"][0].update(adds_training_replicate=True), lambda p: p.update(execution_authorized=True), ) # Full trial authority was validated in setUpClass and again above. # Isolate evaluation reconstruction here, avoiding six repeated rebuilds # of the large, unchanged 68k-epoch synthetic training schedule. with patch.object(m.trial_plan, "validate_trial_plan", return_value=True): for index, change in enumerate(changes): bad = copy.deepcopy(self.plan) change(bad) bad["payload_sha256"] = m._sha({k: v for k, v in bad.items() if k != "payload_sha256"}) with self.subTest(index=index), self.assertRaises(ValueError): m.validate_evaluation_plan(bad, self.trial, self.config) def test_parent_authority_failure_propagates(self): with patch.object(m.trial_plan, "validate_trial_plan", side_effect=ValueError("invalid full parent")): with self.assertRaisesRegex(ValueError, "invalid full parent"): m.build_evaluation_plan(self.trial, self.config) def test_no_mutable_alias_to_caller(self): result = m._assemble(self.trial, self.config) result["config"]["snapshot_bindings"][0]["completed_epochs"] = 900 result["snapshots"][0]["delta_source"]["kind"] = "corruption" result["probes"][0]["input"]["semantic_key"][0] = "corruption" self.assertEqual(self.config["snapshot_bindings"][0]["completed_epochs"], 0) self.assertEqual(self.trial["plan"]["blocks"][0]["suites"][0]["arms"][0]["delta_start"]["kind"], "zero") self.assertEqual(m._assemble(self.trial, self.config), self.plan) if __name__ == "__main__": unittest.main()