Digital Drosophila Archive
196 lines
11 KiB
Python
196 lines
11 KiB
Python
"""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()
|