"""Synthetic CPU array/receipt mutation tests; no files or neural execution.""" import copy import hashlib import json import unittest from unittest import mock import numpy as np import conditioning_preflight_compare as P from test_engineering_compare import synthetic_parts def rehash(receipt): receipt["payload_sha256"] = P._sha({k: v for k, v in receipt.items() if k != "payload_sha256"}) def pair(*, active=True): reference = synthetic_parts([64], active=active)[0] light = synthetic_parts([64], active=active, traces=False, queues=False)[0] for record, mode in ((reference, "reference"), (light, "light")): summary = record["summary"] summary.update(edges=6, explicit_steps_executed=64, within_trial_weight_updates=False, hidden_early_stop=False, entire_neural_trial_executed=True, record_history=False, state_units="SI volts", first_failure_tick=-1, counts_match_sparse_events=True, total_spikes=len(record["state"]["step"]), requested=len(record["requests"]), accepted_requests=int(record["requests"][:, 3].sum()), rejected_requests=int(record["requests"][:, 4].sum()), full_APL_history_recorded=mode == "reference", full_tick_state_hashes_recorded=mode == "reference", full_queue_hashes_recorded=mode == "reference", full_sequence_bitwise_audit_performed=mode == "reference", arriving_edges=sum(row["count"] for row in reference["queues"]), accepted_arriving_edges=sum(row["accepted_count"] for row in reference["queues"]), input_sha256={"base_weights.npy": "a" * 64, "coefficients.json": "b" * 64, "requests.npz": "c" * 64, "spec.json": "d" * 64}) record["terminal_origin"] = "recorded_terminal_queues" return reference, light, reference["requests"][:, :2].astype(np.int32) def delta_fixture(): before = np.zeros(P.CANDIDATE_COUNT, np.float32) after = before.copy() after[1] = np.float32(-.001) receipt = dict(schema="phase4an-conditioning-update-draft-v1", status="DRAFT-COMPUTED-EPOCH-UPDATE-NOT-INSTALLED", source_pins={"eligibility_bridge_sha256": P.ELIGIBILITY_BRIDGE_SHA256, "phase4j_sha256": P.PHASE4J_SHA256}, input_arrays={"delta_before": P._evidence(before)}, outputs={"delta_after": P._evidence(after)}, command={"D_scheduled": 1.0, "frozen": False}, semantics={"application": "epoch_boundary_after_all_recorded_ticks; affects subsequent epoch only", "delta_dtype": "float32", "delta_units": "mV", "D_definition": "artificial scheduled scalar", "actual_DAN_delivery_observed": False, "actual_DAN_delivery_used_to_gate": False}, counts={"candidates": P.CANDIDATE_COUNT, "actual_bitwise_changed_edges": 1, "actual_value_changed_edges": 1}, weights_installed=False, base_modified=False, brain_dynamics_executed=False, scientific_learning_pass_claimed=False, candidate_generated=False) rehash(receipt) manifest = dict(schema="phase4an-overlay-v1", overlay_name="learned", overlay_sha256="1" * 64, mask_path="/synthetic/mask.npy", mask_sha256=P.MASK_SHA256, base_candidate_path="/synthetic/base.npy", base_candidate_sha256="2" * 64, aj_model_dir="/synthetic/apl-timing-lab-v1", am_overlay_path="/synthetic/overlay.py", am_overlay_sha256=P.AM_OVERLAY_SHA256, parent_snapshot_sha256="3" * 64, eligibility_v2_source_sha256=P.PHASE4J_SHA256) return before, after, after.copy(), receipt, manifest class EpochPairTests(unittest.TestCase): def setUp(self): # Pure three-neuron fixtures; production constants remain full MaleCNS. self.patches = [mock.patch.object(P, "N", 3), mock.patch.object(P, "E", 6), mock.patch.object(P, "N_KC", 2)] for patch in self.patches: patch.start() self.addCleanup(patch.stop) def compare(self, reference, light, requests): return P.compare_epoch_pair(reference, light, expected_requests=requests, expected_steps=64, expected_effective_weight_sha256="e" * 64) def test_exact_actual_arrays_with_explicit_light_limits(self): reference, light, requests = pair() report = self.compare(reference, light, requests) self.assertTrue(report["pass_all"], report["failed_checks"]) self.assertTrue(report["numerical_exact"]) self.assertTrue(report["nonvacuous_activity"]) self.assertFalse(report["light_intermediate_states_observed"]) self.assertFalse(report["light_per_tick_V_g_ref_identity_claimed"]) self.assertEqual(report["recorded_pair_availability"], {"APL_traces": False, "state_hashes": False, "queue_hashes": False}) self.assertFalse(report["external_file_seals_process_and_input_authority_verified_here"]) self.assertFalse(report["learning_pass_claimed"]) self.assertFalse(report["candidate_generated"]) json.dumps(report, allow_nan=False) def test_equal_quiet_records_are_not_new_activity_or_learning_pass(self): reference, light, requests = pair(active=False) report = self.compare(reference, light, requests) self.assertTrue(report["numerical_exact"]) self.assertFalse(report["nonvacuous_activity"]) self.assertTrue(report["activity_is_descriptive_not_a_new_gate"]) self.assertFalse(report["learning_pass_claimed"]) def test_signed_zero_final_voltage_difference_fails(self): reference, light, requests = pair() light["state"]["g"][0] = np.float64(0.) report = self.compare(reference, light, requests) self.assertFalse(report["pass_all"]) self.assertIn("pair.state.g", report["failed_checks"]) def test_equal_but_wrong_original_requests_fail(self): reference, light, requests = pair() wrong = requests.copy() wrong[0, 1] = 1 report = self.compare(reference, light, wrong) self.assertFalse(report["pass_all"]) self.assertIn("reference.input_request_array", report["failed_checks"]) self.assertIn("light.input_request_array", report["failed_checks"]) def test_reference_missing_diagnostics_cannot_be_excused_by_light(self): for key in ("traces", "state_hashes", "queues"): with self.subTest(key=key): reference, light, requests = pair() reference[key] = {} if key == "traces" else None report = self.compare(reference, light, requests) self.assertFalse(report["pass_all"]) self.assertIn("reference.full_diagnostics", report["failed_checks"]) def test_partial_or_corrupt_array_schema_fails_closed(self): reference, light, requests = pair() del reference["traces"]["drive"] with self.assertRaises(ValueError): self.compare(reference, light, requests) reference, light, requests = pair() light["state"]["v"][0] = np.nan with self.assertRaises(ValueError): self.compare(reference, light, requests) def test_reported_modes_population_delay_and_scope_fail_closed(self): mutations = (("nodes", 2), ("edges", 7), ("apl_delay_steps", 17), ("diagnostic_mode", "reference"), ("first_failure_tick", 0), ("within_trial_weight_updates", True), ("explicit_steps_executed", 63), ("all_finite", 1), ("parent_checkpoint_manifest_sha256", "f" * 64)) for key, value in mutations: with self.subTest(key=key): reference, light, requests = pair() light["summary"][key] = value self.assertFalse(self.compare(reference, light, requests)["pass_all"]) def test_effective_weight_and_base_coefficient_hash_binding(self): for mutation in ("effective", "base_weights.npy", "coefficients.json"): with self.subTest(mutation=mutation): reference, light, requests = pair() if mutation == "effective": for record in (reference, light): record["summary"]["effective_weights_raw_sha256"] = "f" * 64 else: light["summary"]["input_sha256"][mutation] = "f" * 64 self.assertFalse(self.compare(reference, light, requests)["pass_all"]) def test_performance_is_not_compared_and_path_only_spec_identity_is_external(self): reference, light, requests = pair() reference["summary"].update(wall_seconds=1., peak_allocated_vram_bytes=1) light["summary"].update(wall_seconds=99., peak_allocated_vram_bytes=999) light["summary"]["input_sha256"]["spec.json"] = "f" * 64 report = self.compare(reference, light, requests) self.assertTrue(report["pass_all"]) self.assertFalse(report["external_file_seals_process_and_input_authority_verified_here"]) def test_declared_light_full_diagnostics_or_derived_terminal_is_rejected(self): for mutation in ("recorded", "origin"): reference, light, requests = pair() if mutation == "recorded": light["summary"]["full_tick_state_hashes_recorded"] = True else: light["terminal_origin"] = "AM_spike_ring_derived_from_fresh_events_APL_bits_from_timing" self.assertFalse(self.compare(reference, light, requests)["pass_all"]) def test_actual_summary_accounting_cannot_be_self_reported_wrong(self): for key in ("requested", "accepted_requests", "rejected_requests", "total_spikes", "arriving_edges", "accepted_arriving_edges"): with self.subTest(key=key): reference, light, requests = pair() light["summary"][key] += 1 self.assertFalse(self.compare(reference, light, requests)["pass_all"]) def test_invalid_planned_request_seed_arrays_are_not_silently_coerced(self): reference, light, requests = pair() for bad in (requests.astype(np.int64), requests[::-1], np.concatenate((requests[:1], requests))): with self.assertRaises(ValueError): self.compare(reference, light, bad) with self.assertRaises(ValueError): P.compare_epoch_pair(reference, light, expected_requests=requests, expected_steps=True, expected_effective_weight_sha256="e" * 64) class DeltaBindingTests(unittest.TestCase): def audit(self, values): before, after, installed, receipt, manifest = values return P.audit_delta_binding(before, after, installed, update_receipt=receipt, parent_snapshot_sha256="3" * 64, child_manifest=manifest) def test_exact_array_binding_preserves_explicit_external_authority_limits(self): values = delta_fixture() for value in values[:3]: value.setflags(write=False) original = [value.tobytes() for value in values[:3]] result = self.audit(values) self.assertTrue(result["pass_all"], result["failed_checks"]) self.assertEqual(result["actual_changed_edges"], 1) for name in ("eligibility_v2_independently_recomputed_here", "full_mask_identity_verified_here", "file_bytes_and_parent_snapshot_authority_verified_here", "installation_in_runtime_verified_here", "actual_DAN_delivery_verified_here", "learning_pass_claimed"): self.assertFalse(result[name]) self.assertEqual([value.tobytes() for value in values[:3]], original) json.dumps(result, allow_nan=False) def test_parent_overlay_file_sha_is_not_raw_delta_or_checkpoint_sha(self): values = delta_fixture() self.assertNotEqual(values[4]["overlay_sha256"], P._evidence(values[1])["raw_c_order_sha256"]) self.assertTrue(self.audit(values)["pass_all"]) for false_parent in (P._evidence(values[0])["raw_c_order_sha256"], "4" * 64): values[4]["parent_snapshot_sha256"] = false_parent result = self.audit(values) self.assertFalse(result["pass_all"]) self.assertIn("manifest.parent_overlay_file_sha", result["failed_checks"]) def test_installed_delta_signed_zero_and_row_mutation_rejected(self): for mutation in ("zero", "row"): values = delta_fixture() if mutation == "zero": values[2][0] = np.float32(-0.) else: values[2][1], values[2][2] = values[2][2], values[2][1] result = self.audit(values) self.assertFalse(result["pass_all"]) self.assertIn("computed_installed_delta_bitwise", result["failed_checks"]) def test_rehashed_receipt_must_still_bind_supplied_arrays(self): for location, key in (("input_arrays", "delta_before"), ("outputs", "delta_after")): values = delta_fixture() values[3][location][key]["raw_c_order_sha256"] = "f" * 64 rehash(values[3]) result = self.audit(values) self.assertFalse(result["pass_all"]) self.assertIn("receipt." + key, result["failed_checks"]) def test_rehashed_false_count_and_boolean_integer_ambiguity_rejected(self): for field, value in (("actual_bitwise_changed_edges", 2), ("actual_value_changed_edges", True), ("candidates", float(P.CANDIDATE_COUNT))): values = delta_fixture() values[3]["counts"][field] = value rehash(values[3]) self.assertFalse(self.audit(values)["pass_all"]) def test_stale_receipt_or_changed_frozen_rule_manifest_rejected(self): for mutation in ("hash", "v2", "mask", "AMhelper", "extra", "wrong_name"): values = delta_fixture() if mutation == "hash": values[3]["payload_sha256"] = "0" * 64 elif mutation == "extra": values[4]["unregistered_override"] = True elif mutation == "wrong_name": values[4]["overlay_name"] = "static25" else: key = {"v2": "eligibility_v2_source_sha256", "mask": "mask_sha256", "AMhelper": "am_overlay_sha256"}[mutation] values[4][key] = "0" * 64 self.assertFalse(self.audit(values)["pass_all"]) def test_zero_D_and_frozen_require_exact_no_update_including_signed_zero(self): for mode in ("D0", "frozen"): values = delta_fixture() values[3]["command"]["D_scheduled" if mode == "D0" else "frozen"] = 0.0 if mode == "D0" else True rehash(values[3]) result = self.audit(values) self.assertFalse(result["pass_all"]) self.assertIn("zero_D_or_frozen_identity", result["failed_checks"]) values[1][:] = values[0] values[2][:] = values[0] values[3]["outputs"]["delta_after"] = P._evidence(values[1]) values[3]["counts"].update(actual_value_changed_edges=0, actual_bitwise_changed_edges=0) rehash(values[3]) self.assertTrue(self.audit(values)["pass_all"]) def test_actual_DAN_or_immediate_US_update_claim_does_not_match_interface(self): for field, value in (("actual_DAN_delivery_used_to_gate", True), ("actual_DAN_delivery_observed", True), ("application", "at US tick"), ("delta_units", "V")): values = delta_fixture() values[3]["semantics"][field] = value rehash(values[3]) self.assertFalse(self.audit(values)["pass_all"]) def test_wrong_dtype_shape_nonfinite_or_positive_delta_rejected(self): values = delta_fixture() with self.assertRaises(ValueError): P.audit_delta_binding(values[0].astype(np.float64), values[1], values[2], update_receipt=values[3], parent_snapshot_sha256="3" * 64, child_manifest=values[4]) values = list(delta_fixture()) values[0] = values[0][:-1] with self.assertRaises(ValueError): self.audit(values) values = delta_fixture() values[0][0] = np.nan with self.assertRaises(ValueError): self.audit(values) values = delta_fixture() values[1][1] = values[2][1] = np.float32(.001) values[3]["outputs"]["delta_after"] = P._evidence(values[1]) rehash(values[3]) result = self.audit(values) self.assertFalse(result["pass_all"]) self.assertIn("delta.nonpositive", result["failed_checks"]) if __name__ == "__main__": unittest.main()