"""Synthetic eligibility tests only; never load recorded brain trajectories.""" import ast import importlib.util import json import math import os from pathlib import Path from types import SimpleNamespace import tempfile import unittest from typing import Any import numpy as np import eligibility_bridge as bridge def source_path(): 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() KC = np.array([1, 3, 7], np.int32) def run_trace(rows=(), steps=8, commands=(0, 3, 7), backend="numpy"): events = np.asarray(rows, np.int32).reshape(-1, 2) return bridge.trace_from_events(events[:, 0], events[:, 1], KC, steps, commands, phase4j_source=SOURCE, backend=backend, n_neurons=10) def update_args(): return dict(trace=np.array([1, .25, 0], np.float32), kc=KC.copy(), candidate_sources=np.array([1, 1, 3, 7, 3], np.int32), candidate_compartments=np.array(["y5B'2a", "a1", "y5B'2a", "y5B'2a", "y2a'1"]), compartment="y5B'2a", delta=np.zeros(5, np.float32), base_candidate=np.array([.275, .55, .0001, .275, 2.75], np.float32), d=1., phase4j_source=SOURCE, n_neurons=10) class FrozenTraceTests(unittest.TestCase): def test_no_input_positive_zero_and_no_command(self): result = run_trace(commands=()) self.assertEqual(result.snapshots.shape, (0, 3)) self.assertTrue(bridge.bit_equal(result.final, np.zeros(3, np.float32))) self.assertTrue(bridge.bit_equal(result.last_command_trace, result.final)) self.assertFalse(result.audit["training_executed"]) self.assertFalse(result.audit["torch_numpy_bitwise_comparison_performed"]) json.dumps(result.audit, allow_nan=False) def test_simultaneous_spike_is_seen_and_cap_is_one(self): result = run_trace([(0, 1), (1, 1), (3, 3), (3, 7)]) self.assertEqual(result.snapshots[0, 0], np.float32(1)) self.assertEqual(result.snapshots[1, 1:].tolist(), [1., 1.]) self.assertLessEqual(float(result.final.max()), 1.) def test_interval_decay_is_each_tick_not_power_jump(self): result = run_trace([(0, 1)], steps=1001, commands=(0, 1, 1000)) expected = np.float32(1) decay = np.float32(math.exp(-.1 / 100.)) for _ in range(1000): expected = np.float32(expected * decay) self.assertEqual(result.snapshots[-1, 0].view(np.uint32), expected.view(np.uint32)) self.assertEqual(result.snapshots[1, 0].view(np.uint32), decay.view(np.uint32)) def test_us_before_future_cs_has_zero_delta(self): result = run_trace([(2, 1)], commands=(1,)) args = update_args() args["trace"] = result.last_command_trace delta, _ = bridge.apply_eligibility(**args) self.assertTrue(bridge.bit_equal(delta, args["delta"])) def test_plus_1000_ms_ratio_strict_below_point01(self): result = run_trace([(0, 1)], steps=10001, commands=(0, 10000)) args = update_args() args["base_candidate"][:] = 100 # no saturation in this temporal check args["trace"] = result.snapshots[0] zero, _ = bridge.apply_eligibility(**args) args["trace"] = result.snapshots[1] late, _ = bridge.apply_eligibility(**args) self.assertLess(float(np.abs(late).sum() / np.abs(zero).sum()), .01) self.assertGreater(float(result.snapshots[-1, 0]), 0.) def test_non_kc_events_do_not_enter_trace(self): a = run_trace([(0, 1), (3, 3)]) b = run_trace([(0, 0), (0, 1), (2, 9), (3, 3)]) self.assertTrue(bridge.bit_equal(a.snapshots, b.snapshots)) self.assertEqual(b.audit["actual_kc_events"], 2) def test_inputs_and_outputs_are_not_mutated(self): step, node, kc = np.array([0, 7], np.int32), np.array([1, 3], np.int32), KC.copy() before = [a.copy() for a in (step, node, kc)] result = bridge.trace_from_events(step, node, kc, 8, [7], phase4j_source=SOURCE, n_neurons=10) for a, b in zip((step, node, kc), before): self.assertTrue(bridge.bit_equal(a, b)) for a in (result.final, result.snapshots, result.command_steps, result.last_command_trace): self.assertFalse(a.flags.writeable) def test_bad_event_and_command_schema_is_rejected(self): for rows in ([(0, 1), (0, 1)], [(1, 1), (0, 1)], [(-1, 1)], [(8, 1)], [(0, 10)]): with self.subTest(rows=rows), self.assertRaises(ValueError): run_trace(rows) for commands in ((1, 1), (3, 1), (-1,), (8,), (.5,), (True,), np.empty((0, 2), np.int32)): with self.subTest(commands=commands), self.assertRaises(ValueError): run_trace(commands=commands) with self.assertRaises(ValueError): bridge.trace_from_events(np.array([], np.int64), np.array([], np.int32), KC, 8, [], phase4j_source=SOURCE) def test_bad_kc_and_backend_are_rejected(self): for kc in (np.array([3, 1], np.int32), np.array([1, 1], np.int32), np.array([], np.int32)): with self.assertRaises(ValueError): bridge.trace_from_events(np.array([], np.int32), np.array([], np.int32), kc, 1, [], phase4j_source=SOURCE) with self.assertRaises(ValueError): run_trace(backend="auto") @unittest.skipUnless(importlib.util.find_spec("torch") is not None, "Torch absent: CPU bitwise compatibility NOT established locally") def test_torch_original_expression_bitwise_against_numpy(self): rng = np.random.default_rng(719) mask = rng.random((1001, 10)) < .07 tick, node = np.nonzero(mask) rows = np.stack((tick, node), axis=1).tolist() numpy_result = run_trace(rows, steps=1001, commands=(0, 1, 7, 999, 1000)) torch_result = run_trace(rows, steps=1001, commands=(0, 1, 7, 999, 1000), backend="torch") self.assertTrue(bridge.bit_equal(numpy_result.snapshots, torch_result.snapshots)) self.assertTrue(bridge.bit_equal(numpy_result.final, torch_result.final)) empty_np = run_trace([(0, 1)], steps=10001, commands=(0, 10000)) empty_torch = run_trace([(0, 1)], steps=10001, commands=(0, 10000), backend="torch") self.assertTrue(bridge.bit_equal(empty_np.snapshots, empty_torch.snapshots)) class FrozenUpdateTests(unittest.TestCase): def test_actual_source_sha_and_ast_only_extraction(self): _, report = bridge.apply_eligibility(**update_args()) self.assertEqual(report["source_sha256"], bridge.PHASE4J_SHA256) self.assertFalse(report["actual_dan_activity_used_to_gate"]) self.assertFalse(report["learning_pass_claimed"]) json.dumps(report, allow_nan=False) def test_source_drift_is_rejected_before_execution(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "phase4j.py" path.write_bytes(SOURCE.read_bytes() + b"\nraise RuntimeError('must not execute')\n") args = update_args() args["phase4j_source"] = path with self.assertRaisesRegex(ValueError, "SHA256 mismatch"): bridge.apply_eligibility(**args) def test_bitwise_against_independently_extracted_original_apply_v2(self): tree = ast.parse(SOURCE.read_text(encoding="utf-8")) function = next(x for x in tree.body if isinstance(x, ast.FunctionDef) and x.name == "apply_v2") namespace = {"np": np, "Validator": object, "Any": Any} exec(compile(ast.Module(body=[function], type_ignores=[]), "independent-old-apply-v2", "exec"), namespace) class Column: def __init__(self, values): self.values = values def to_numpy(self, dtype=None): return np.array(self.values, dtype=dtype) def astype(self, dtype): return Column(np.asarray(self.values).astype(dtype)) args = update_args() validator = SimpleNamespace(n=10, kc=KC, base_candidate=args["base_candidate"], cand={"source_neuron_idx": Column(args["candidate_sources"]), "target_compartment_verbatim": Column(args["candidate_compartments"])}) cfg = {"eligibility_v2": {"eta_effective_mv": .00275, "max_fractional_depression": .25}} for _ in range(50): actual, report = bridge.apply_eligibility(**args) expected, old_report = namespace["apply_v2"](validator, args["trace"], args["compartment"], 1., args["delta"], cfg) self.assertTrue(bridge.bit_equal(actual, expected)) self.assertEqual(report["original_report"], old_report) args["delta"] = actual def test_zero_d_and_frozen_preserve_existing_overlay(self): args = update_args() args["delta"][:] = -.25 * args["base_candidate"] for d, frozen in ((0., False), (1., True), (0., True)): args.update(d=d, frozen=frozen) result, report = bridge.apply_eligibility(**args) self.assertTrue(bridge.bit_equal(result, args["delta"])) self.assertEqual(report["actual_changed_edges"], 0) def test_off_compartment_and_inactive_source_exact_unchanged(self): args = update_args() args["delta"][1] = -.01 result, _ = bridge.apply_eligibility(**args) self.assertTrue(bridge.bit_equal(result[[1, 3, 4]], args["delta"][[1, 3, 4]])) self.assertEqual(result[0], np.float32(-.00275)) def test_original_floor_clamps_and_does_not_reverse_sign(self): args = update_args() result, report = bridge.apply_eligibility(**args) self.assertEqual(result[2], (-.25 * args["base_candidate"])[2]) args["delta"] = result again, _ = bridge.apply_eligibility(**args) self.assertEqual(again[2].view(np.uint32), result[2].view(np.uint32)) self.assertTrue(np.all(again <= 0)) # Original updated_edges counts attempted updates, including saturation. self.assertEqual(report["original_report"]["updated_edges"], 2) def test_all_original_compartments_are_literal_not_inferred(self): args = update_args() args["candidate_sources"] = np.full(4, 1, np.int32) args["candidate_compartments"] = np.array(bridge.COMPARTMENTS) args["delta"] = np.zeros(4, np.float32) args["base_candidate"] = np.ones(4, np.float32) for comp in bridge.COMPARTMENTS: args["compartment"] = comp result, _ = bridge.apply_eligibility(**args) self.assertEqual(int(np.count_nonzero(result)), 1) self.assertNotEqual(result[list(bridge.COMPARTMENTS).index(comp)], 0) def test_update_does_not_mutate_inputs_and_returns_readonly(self): args = update_args() before = {k: v.copy() for k, v in args.items() if isinstance(v, np.ndarray)} result, _ = bridge.apply_eligibility(**args) for k, value in before.items(): self.assertTrue(bridge.bit_equal(args[k], value)) self.assertFalse(result.flags.writeable) def test_invalid_units_dtypes_values_and_compartment_fail_closed(self): bad = [("trace", np.array([1, .25, 0], np.float64)), ("trace", np.array([1, np.nan, 0], np.float32)), ("trace", np.array([1.1, 0, 0], np.float32)), ("candidate_sources", np.array([1, 1, 3, 8, 3], np.int32)), ("compartment", "PAM15"), ("d", -.1), ("d", .5), ("d", float("nan")), ("d", True), ("frozen", 1), ("base_candidate", np.zeros(5, np.float32)), ("delta", np.ones(5, np.float32)), ("delta", np.full(5, -.5, np.float32)), ("delta", np.full(5, -0., np.float32))] for key, value in bad: with self.subTest(key=key, value=value): args = update_args() args[key] = value with self.assertRaises(ValueError): bridge.apply_eligibility(**args) def test_empty_candidate_subset_is_safe_but_not_claimed_full_mask(self): args = update_args() for key in ("candidate_sources", "candidate_compartments", "delta", "base_candidate"): args[key] = args[key][:0] result, report = bridge.apply_eligibility(**args) self.assertEqual(result.shape, (0,)) self.assertFalse(report["complete_mask_identity_verified_by_this_function"]) if __name__ == "__main__": unittest.main()