"""Synthetic plans only; no neural input generation, files, remote or execution.""" import copy import unittest from unittest import mock import evaluation_plan import evaluation_policy as policy import trial_plan class EvaluationPolicyTests(unittest.TestCase): @classmethod def setUpClass(cls): # Explicit synthetic fixture seed, never a proposed production master. cls.trial = trial_plan.build_trial_plan(461) cls.result = policy.build_evaluation_policy(cls.trial) cls.registry = policy._registry_index(cls.trial) cls.rows = cls.result["audit"]["bindings"] def suite(self, assay="normal"): block = self.trial["plan"]["blocks"][0] suite = next(s for s in block["suites"] if s["assay"] == assay) return block, copy.deepcopy(suite) def check_suite(self, block, suite, registry=None): return policy._suite_policy(block, suite, self.trial["plan"]["master"], self.registry if registry is None else registry) def test_fixed_config_and_complete_consumer_inventory(self): config = self.result["config"] self.assertEqual(set(config), evaluation_plan.CONFIG_FIELDS) self.assertEqual(config["probe_steps"], 1200) self.assertEqual(config["cs_start_tick"], 0) self.assertEqual(config["cs_stop_tick_exclusive"], 1000) self.assertEqual(config["reversal_checkpoints"], [0, 1, 2, 5, 10, 20, 50]) self.assertEqual(config["baseline_execution"], "execute_each_snapshot") self.assertEqual(len(config["snapshot_bindings"]), 4480) # The original public consumer performs its own full authority check. inventory = evaluation_plan.build_evaluation_plan(self.trial, config) self.assertEqual(inventory["budget"]["evaluation_probes"], 9440) self.assertEqual(inventory["budget"]["snapshot_count"], 4720) self.assertEqual(inventory["budget"]["evaluation_steps"], 11328000) self.assertEqual(inventory["budget"]["planned_scientific_steps_including_evaluation"], 118928000) self.assertEqual(inventory["budget"]["probes_by_assay"], {"normal": 3920, "delayed": 3920, "reversal": 1120, "retention": 480}) self.assertFalse(inventory["execution_authorized"]) self.assertFalse(inventory["budget"]["baseline_output_deduplication"]) def test_each_binding_audited_in_exact_order(self): bindings = self.result["config"]["snapshot_bindings"] self.assertEqual([r["binding"] for r in self.rows], bindings) identities = [(r["block_id"], r["assay"], r["arm"], r["checkpoint"]) for r in bindings] self.assertEqual(len(identities), len(set(identities))) self.assertEqual({r["checkpoint"] for r in bindings if r["assay"] == "reversal"}, set(policy.POINTS)) self.assertEqual(self.result["audit"]["coverage"]["snapshots_by_assay"], {"normal": 1960, "delayed": 1960, "reversal": 560}) def test_five_independent_replicates_not_relabelled_from_probes(self): coverage = self.result["audit"]["coverage"] self.assertEqual(coverage["training_blocks"], 40) self.assertEqual(coverage["training_replicates_per_cell"], 5) self.assertEqual(coverage["new_random_streams"], 0) self.assertFalse(coverage["new_master_seed_selected"]) self.assertEqual(self.result["seed_manifest_sha256"], self.trial["seed_manifest"]["payload_sha256"]) def test_normal_and_delayed_controls_match_actual_ordered_source_prefixes(self): selected = [r for r in self.rows if r["binding"]["assay"] in ("normal", "delayed") and r["binding"]["arm"] != "dan_only"] self.assertEqual(len(selected), 3360) for row in selected: self.assertIs(row["ordered_CS_prefix_exact"], True) self.assertEqual(row["actual"]["ordered_CS_source_identity_sha256"], row["reference"]["ordered_CS_source_identity_sha256"]) self.assertEqual(row["actual"]["CS_counts_by_cue"], row["reference"]["CS_counts_by_cue"]) if row["binding"]["arm"] != "unpaired": self.assertEqual(row["binding"]["completed_epochs"], row["reference_completed_epochs"]) self.assertTrue(row["elapsed_ticks_equal_to_reference"]) def test_unpaired_earliest_matching_prefix_and_full_terminal_including_tail(self): any_earlier_US = False any_trailing_US = False for block in self.trial["plan"]["blocks"]: for suite in block["suites"][:2]: arms = {a["arm"]: a for a in suite["arms"]} paired, unpaired = arms["paired"], arms["unpaired"] rows = [r for r in self.rows if r["binding"]["block_id"] == block["block_id"] and r["binding"]["assay"] == suite["assay"] and r["binding"]["arm"] == "unpaired"] for row in rows: point = row["binding"]["checkpoint"] count = row["binding"]["completed_epochs"] needed = row["reference_completed_epochs"] if point == 0: self.assertEqual(count, 0) elif point == 50: self.assertEqual(count, 150) self.assertEqual(row["actual"]["CS_epochs"], 100) self.assertEqual(row["actual"]["requested_US_epochs"], 50) self.assertEqual(row["elapsed_ticks_difference"], 50 * paired["epochs"][0]["steps"]) any_trailing_US |= unpaired["epochs"][-1]["cs"] is None else: actual_positions = [i + 1 for i, e in enumerate(unpaired["epochs"]) if e["cs"] is not None] self.assertEqual(count, actual_positions[needed - 1]) self.assertIsNotNone(unpaired["epochs"][count - 1]["cs"]) self.assertEqual(row["actual"]["requested_US_epochs"], count - needed) any_earlier_US |= count > needed self.assertTrue(any_earlier_US) self.assertTrue(any_trailing_US) def test_DAN_only_is_yoked_zero_CS_not_equal_exposure(self): rows = [r for r in self.rows if r["binding"]["arm"] == "dan_only"] self.assertEqual(len(rows), 560) for row in rows: self.assertEqual(row["actual"]["CS_epochs"], 0) self.assertIsNone(row["ordered_CS_prefix_exact"]) self.assertFalse(row["CS_exposure_equal"]) self.assertTrue(row["DAN_only_yoke_verified"]) self.assertTrue(row["requested_US_count_equal_to_reference"]) self.assertTrue(row["elapsed_ticks_equal_to_reference"]) self.assertEqual(row["binding"]["completed_epochs"], row["reference_completed_epochs"]) def test_reversal_each_own_US_boundary_and_original_terminal(self): saw_different_middle_CS_prefix = False for block in self.trial["plan"]["blocks"]: suite = block["suites"][2] arms = {a["arm"]: a for a in suite["arms"]} for row in [r for r in self.rows if r["binding"]["block_id"] == block["block_id"] and r["binding"]["assay"] == "reversal"]: binding = row["binding"] arm = arms[binding["arm"]] point, count = binding["checkpoint"], binding["completed_epochs"] if point in (0, 50): self.assertEqual(count, 0 if point == 0 else 100) self.assertTrue(row["ordered_CS_prefix_exact"]) else: ordinal_positions = [i + 1 for i, e in enumerate(arm["epochs"]) if e["us"]["requested"]] self.assertEqual(count, ordinal_positions[point - 1]) self.assertTrue(arm["epochs"][count - 1]["us"]["requested"]) saw_different_middle_CS_prefix |= not row["ordered_CS_prefix_exact"] self.assertEqual(row["actual"]["requested_US_epochs"], point) self.assertEqual(row["actual"]["enabled_nonzero_D_update_opportunities"], point) self.assertTrue(saw_different_middle_CS_prefix) def test_intermediate_US_time_and_update_mismatches_are_explicit_not_failed(self): middle = [r for r in self.rows if r["binding"]["checkpoint"] not in (0, 50)] self.assertTrue(any(r["binding"]["arm"] == "random_reward" and r["requested_US_count_difference"] != 0 for r in middle)) self.assertTrue(any(r["binding"]["arm"] == "unpaired" and r["elapsed_ticks_difference"] > 0 for r in middle)) self.assertTrue(any(r["binding"]["arm"] == "frozen" and r["nonzero_D_update_opportunity_difference"] < 0 for r in middle)) self.assertTrue(all(r["all_US_timing_or_total_exposure_matched_claimed"] is False for r in self.rows)) self.assertTrue(all(r["actual"]["actual_accepted_US_or_DAN_spikes_known"] is False for r in self.rows)) def test_all_boundaries_follow_complete_epoch_update_not_US_tick(self): for row in self.rows: self.assertEqual(row["boundary"], "after complete epoch and scheduled delta update; not at the US tick") steps = 1800 if row["binding"]["assay"] == "delayed" else 1200 self.assertEqual(row["actual"]["elapsed_training_ticks"], row["binding"]["completed_epochs"] * steps) self.assertEqual(row["terminal_includes_all_epochs"], row["binding"]["checkpoint"] == 50) def test_control_seed_or_source_alias_mutations_fail_without_relying_on_counts(self): for field, change in (("seed", lambda v: v + 1), ("source_stream_id", lambda v: v + "x"), ("stream_id", lambda v: v + "x"), ("start_tick", lambda v: False)): with self.subTest(field=field): block, suite = self.suite() cs = suite["arms"][1]["epochs"][0]["cs"] cs[field] = change(cs[field]) with self.assertRaises(ValueError): self.check_suite(block, suite) def test_reordered_cues_with_same_counts_and_valid_individual_aliases_fail(self): block, suite = self.suite() epochs = suite["arms"][1]["epochs"] index = next(i for i in range(1, len(epochs)) if epochs[i]["cs"]["cue"] != epochs[0]["cs"]["cue"]) epochs[0]["cs"], epochs[index]["cs"] = epochs[index]["cs"], epochs[0]["cs"] with self.assertRaisesRegex(ValueError, "CS occurrence order|ordered CS|CS sequence"): self.check_suite(block, suite) def test_corrupt_alias_registry_is_not_excused_by_equal_CS_seed(self): block, suite = self.suite() independent, aliases = self.registry changed = copy.deepcopy(aliases) cs = suite["arms"][1]["epochs"][0]["cs"] key = (block["block_id"], "normal", "CS", cs["cue"], f'{cs["occurrence"]:02d}', "d_zero") changed[("training", key)]["source_key"][-1] = "frozen" with self.assertRaisesRegex(ValueError, "alias is not registered"): self.check_suite(block, suite, (independent, changed)) def test_DAN_yoked_cue_US_or_added_CS_mutations_fail(self): for mutation in ("cue", "US", "CS"): with self.subTest(mutation=mutation): block, suite = self.suite() paired = suite["arms"][0] dan = suite["arms"][-1] epoch = dan["epochs"][0] if mutation == "cue": epoch["yoked_cue_removed"] = "B" if epoch["yoked_cue_removed"] == "A" else "A" elif mutation == "US": epoch["us"]["nominal_slot_tick"] += 1 else: epoch["cs"] = copy.deepcopy(paired["epochs"][0]["cs"]) with self.assertRaises(ValueError): self.check_suite(block, suite) def test_missing_reordered_or_duplicate_control_arms_fail(self): for mutation in ("missing", "reorder", "duplicate"): with self.subTest(mutation=mutation): block, suite = self.suite() if mutation == "missing": suite["arms"].pop() elif mutation == "reorder": suite["arms"][0], suite["arms"][1] = suite["arms"][1], suite["arms"][0] else: suite["arms"].append(suite["arms"][0]) with self.assertRaisesRegex(ValueError, "control arms"): self.check_suite(block, suite) def test_result_not_aliased_no_execution_or_new_scientific_gate(self): fields = ("protocol_approval_claimed", "execution_authorized", "training_executed", "inputs_generated", "input_generator_implementation_verified", "source_and_snapshot_authority_verified", "candidate_generated", "learning_pass_claimed", "scientific_gate_added") self.assertTrue(all(self.result[field] is False for field in fields)) self.assertEqual(self.result["status"], "DRAFT-NOT-REGISTERED") self.assertTrue(self.result["choices_are_new_draft_proposals_not_inherited_gates"]) self.assertTrue(self.result["audit"]["unchanged_retention_scope"]["same_continuous_ten_second_lineage"]) self.assertFalse(self.result["audit"]["unchanged_retention_scope"]["retained_snapshot_execution_or_lineage_verified"]) # Config and audit may describe the same row, but neither is a caller alias. block = self.trial["plan"]["blocks"][0] self.assertIsNot(self.result["config"]["reversal_checkpoints"], block["evaluation_checkpoints"]) self.assertEqual(self.result["trial_payload_sha256"], self.trial["payload_sha256"]) def test_validation_rejects_rehashed_config_type_and_audit_mutations(self): # Separate reconstruction is supplied by the already independently built # fixture to keep mutation tests small; public full reconstruction is # exercised once below, not hidden behind this mocked trust boundary. with mock.patch.object(policy, "build_evaluation_policy", return_value=self.result): for mutation in ("point", "type", "audit", "authorization"): with self.subTest(mutation=mutation): changed = copy.deepcopy(self.result) if mutation == "point": changed["config"]["snapshot_bindings"][1]["completed_epochs"] += 1 elif mutation == "type": changed["config"]["cs_start_tick"] = False elif mutation == "audit": changed["audit"]["bindings"][0]["actual"]["CS_epochs"] += 1 else: changed["execution_authorized"] = True changed["payload_sha256"] = policy._sha({k: v for k, v in changed.items() if k != "payload_sha256"}) with self.assertRaisesRegex(ValueError, "exact reconstruction"): policy.validate_evaluation_policy(changed, self.trial) def test_public_validation_full_reconstruction(self): self.assertTrue(policy.validate_evaluation_policy(self.result, self.trial)) def test_trial_authority_hash_rejected_before_planning(self): changed = dict(self.trial) changed["payload_sha256"] = "0" * 64 with self.assertRaisesRegex(ValueError, "payload SHA"): policy.build_evaluation_policy(changed) if __name__ == "__main__": unittest.main()