"""Synthetic sparse-event updates only; no recorded trajectories or training.""" import copy import hashlib import importlib.util import json import os from pathlib import Path import tempfile import unittest import numpy as np import conditioning_update as m def source_path(): # Same PHASE4J_SOURCE and ordered local candidates as frozen AM tests. explicit = os.environ.get("PHASE4J_SOURCE") if explicit: return Path(explicit) parent = Path(__file__).resolve().parents[1] paths = (parent / "models/male-cns-lif/phase4j.py", parent / "mb-physiology-v1/phase4j.py") for path in paths: if path.is_file(): return path raise FileNotFoundError("Set PHASE4J_SOURCE to the SHA-pinned original phase4j.py") SOURCE = source_path() BRIDGE = Path(os.environ.get("AM_ELIGIBILITY_BRIDGE", str(Path(__file__).resolve().parents[1] / "wholebrain-plastic-interface-v1/eligibility_bridge.py"))) def args(rows=((0, 1), (3, 3)), us_tick=3, steps=8): events = np.asarray(rows, np.int32).reshape(-1, 2) return dict(step=events[:, 0], neuron_idx=events[:, 1], kc=np.array([1, 3, 7], np.int32), candidate_edge_id=np.array([1, 4, 5, 10, 12], np.int64), candidate_sources=np.array([1, 1, 3, 7, 3], np.int32), candidate_targets=np.array([2, 2, 5, 5, 9], np.int32), candidate_compartments=np.array(["y5B'2a", "a1", "y5B'2a", "y5B'2a", "y2a'1"]), delta_before=np.zeros(5, np.float32), base_candidate=np.array([.275, .55, .0001, .275, 2.75], np.float32), steps=steps, n_neurons=10, compartment="y5B'2a", us_tick=us_tick, d_command=1.0 if us_tick is not None else 0.0, frozen=False, bridge_source=BRIDGE, phase4j_source=SOURCE, trace_backend="numpy") class ConditioningUpdate(unittest.TestCase): def test_pinned_original_calls_and_recomputable_receipt(self): inputs = args() result = m.update_epoch(**inputs) bridge = m._load_bridge(BRIDGE) traced = bridge.trace_from_events(inputs["step"], inputs["neuron_idx"], inputs["kc"], 8, [3], phase4j_source=SOURCE, backend="numpy", n_neurons=10) expected, _ = bridge.apply_eligibility(traced.last_command_trace, inputs["kc"], inputs["candidate_sources"], inputs["candidate_compartments"], inputs["compartment"], inputs["delta_before"], inputs["base_candidate"], d=1., phase4j_source=SOURCE, n_neurons=10) self.assertTrue(bridge.bit_equal(result.delta_after, expected)) self.assertTrue(m.verify_receipt(result.receipt, **inputs)) self.assertEqual(result.receipt["source_pins"]["eligibility_bridge_sha256"], hashlib.sha256(BRIDGE.read_bytes()).hexdigest()) self.assertEqual(result.receipt["trace_audit"]["sampling"], "after_current_tick_spike") json.dumps(result.receipt, allow_nan=False) def test_zero_d_and_frozen_preserve_existing_delta(self): for d, frozen in ((0., False), (1., True), (0., True)): inputs = args() inputs["delta_before"][:] = -.25 * inputs["base_candidate"] inputs.update(d_command=d, frozen=frozen) result = m.update_epoch(**inputs) self.assertEqual(result.delta_after.tobytes(), inputs["delta_before"].tobytes()) self.assertEqual(result.receipt["counts"]["original_updated_edges_attempted"], 0) self.assertEqual(result.receipt["counts"]["actual_bitwise_changed_edges"], 0) def test_us_before_cs_does_not_use_future_trace(self): result = m.update_epoch(**args(rows=((2, 1),), us_tick=1)) self.assertFalse(np.any(result.sampled_trace)) self.assertTrue(np.any(result.final_trace)) self.assertFalse(np.any(result.delta_after)) def test_current_tick_and_reentrant_kc_saturate_before_sample(self): result = m.update_epoch(**args(rows=((0, 1), (1, 1), (3, 1), (3, 3)), us_tick=3)) self.assertEqual(result.sampled_trace.tolist(), [1., 1., 0.]) self.assertEqual(result.delta_after[0].view(np.uint32), np.float32(-.00275).view(np.uint32)) self.assertEqual(result.receipt["counts"]["actual_KC_events"], 4) def test_non_kc_only_changes_event_evidence_not_update(self): a = m.update_epoch(**args(rows=((0, 1), (3, 3)))) b = m.update_epoch(**args(rows=((0, 0), (0, 1), (2, 9), (3, 3)))) self.assertEqual(a.delta_after.tobytes(), b.delta_after.tobytes()) self.assertEqual(a.sampled_trace.tobytes(), b.sampled_trace.tobytes()) self.assertNotEqual(a.receipt["input_arrays"], b.receipt["input_arrays"]) def test_no_us_uses_zero_not_final_trace(self): result = m.update_epoch(**args(us_tick=None)) self.assertFalse(np.any(result.sampled_trace)) self.assertFalse(np.any(np.signbit(result.sampled_trace))) self.assertTrue(np.any(result.final_trace)) self.assertEqual(result.receipt["counts"]["actual_value_changed_edges"], 0) def test_attempts_and_fully_saturated_changes_are_distinct(self): inputs = args(rows=((3, 1), (3, 3), (3, 7))) inputs["delta_before"][:] = -.25 * inputs["base_candidate"] result = m.update_epoch(**inputs) counts = result.receipt["counts"] self.assertEqual(counts["original_updated_edges_attempted"], 3) self.assertEqual(counts["actual_value_changed_edges"], 0) self.assertEqual(counts["selected_at_floor_before"], 3) self.assertEqual(counts["selected_at_floor_after"], 3) self.assertEqual(counts["selected_newly_at_floor"], 0) def test_new_floor_and_off_compartment_identity(self): inputs = args() inputs["delta_before"][1] = -.01 result = m.update_epoch(**inputs) self.assertEqual(result.receipt["counts"]["selected_newly_at_floor"], 1) self.assertEqual(result.receipt["counts"]["actual_bitwise_changed_edges"], 2) self.assertEqual(result.delta_after[[1, 3, 4]].tobytes(), inputs["delta_before"][[1, 3, 4]].tobytes()) def test_owned_readonly_results_and_all_input_bytes_unchanged(self): inputs = args() before = {k: v.tobytes() for k, v in inputs.items() if isinstance(v, np.ndarray)} result = m.update_epoch(**inputs) for key, raw in before.items(): self.assertEqual(inputs[key].tobytes(), raw) for output in (result.delta_after, result.sampled_trace, result.final_trace): self.assertTrue(output.flags.owndata) self.assertFalse(output.flags.writeable) self.assertTrue(all(not np.shares_memory(output, inputs[key]) for key in before)) inputs["delta_before"][:] = -.000001 self.assertNotEqual(result.delta_after.tobytes(), inputs["delta_before"].tobytes()) def test_scope_claims_do_not_prove_graph_or_dan_delivery(self): receipt = m.update_epoch(**args()).receipt for field in ("complete_KC_identity_verified", "complete_mask_identity_verified", "full_graph_authority_verified", "brain_dynamics_executed", "weights_installed", "base_modified", "training_executed_by_this_function", "scientific_learning_pass_claimed", "candidate_generated", "numpy_torch_bitwise_comparison_performed", "frozen_Torch_expression_executed"): self.assertIs(receipt[field], False) self.assertFalse(receipt["semantics"]["actual_DAN_delivery_observed"]) self.assertFalse(receipt["semantics"]["actual_DAN_delivery_used_to_gate"]) def test_invalid_events_mapping_commands_and_values_fail_closed(self): mutations = ( ("step", np.array([0, 3], np.int64)), ("steps", True), ("us_tick", False), ("d_command", True), ("d_command", float("nan")), ("frozen", 1), ("trace_backend", "auto"), ("candidate_edge_id", np.array([1, 1, 5, 10, 12], np.int64)), ("candidate_targets", np.array([2, 2, 5, 5, 10], np.int32)), ("candidate_sources", np.array([0, 1, 3, 7, 3], np.int32)), ("base_candidate", np.array([.275, .55, float("inf"), .275, 2.75], np.float32)), ("delta_before", np.full(5, -0., np.float32)), ("kc", np.array([1, 1, 7], np.int32)), ) for key, value in mutations: inputs = args() inputs[key] = value with self.subTest(key=key), self.assertRaises(ValueError): m.update_epoch(**inputs) for rows in (((0, 1), (0, 1)), ((3, 3), (0, 1)), ((8, 1),)): with self.assertRaises(ValueError): m.update_epoch(**args(rows=rows)) inputs = args(us_tick=None) inputs["d_command"] = 1 with self.assertRaises(ValueError): m.update_epoch(**inputs) def test_receipt_forgery_fails_after_new_self_hash(self): inputs = args() bad = copy.deepcopy(m.update_epoch(**inputs).receipt) bad["counts"]["actual_value_changed_edges"] = 99 bad["payload_sha256"] = m._sha({k: v for k, v in bad.items() if k != "payload_sha256"}) with self.assertRaisesRegex(ValueError, "exact recomputation"): m.verify_receipt(bad, **inputs) def test_source_drift_rejected_before_loading(self): with tempfile.TemporaryDirectory() as directory: for field, source in (("bridge_source", BRIDGE), ("phase4j_source", SOURCE)): path = Path(directory) / (field + ".py") path.write_bytes(source.read_bytes() + b"\nraise RuntimeError('never execute')\n") inputs = args() inputs[field] = path with self.subTest(field=field), self.assertRaisesRegex(ValueError, "SHA256 mismatch"): m.update_epoch(**inputs) @unittest.skipUnless(importlib.util.find_spec("torch") is not None, "Torch absent: per-call NumPy/Torch exact compatibility NOT established locally") def test_frozen_torch_cpu_matches_numpy_on_synthetic_events(self): inputs = args(rows=((0, 1), (1, 1), (3, 3), (7, 7)), us_tick=7) numpy_result = m.update_epoch(**inputs) inputs["trace_backend"] = "torch" torch_result = m.update_epoch(**inputs) for field in ("delta_after", "sampled_trace", "final_trace"): self.assertEqual(getattr(numpy_result, field).tobytes(), getattr(torch_result, field).tobytes()) self.assertTrue(torch_result.receipt["frozen_Torch_expression_executed"]) self.assertFalse(torch_result.receipt["numpy_torch_bitwise_comparison_performed"]) if __name__ == "__main__": unittest.main()