"""Phase 4AG shared-APL-component closed-loop Brian oracle.

Brian has no Torch dependency.  A separate, pinned science-Python process
executes the unchanged legacy float32 APL operations from Brian's own spikes.
This is independent LIF/scheduling validation with a shared APL numerical
component, NOT an all-components-independent CPU oracle.
"""
from __future__ import annotations

import csv
import hashlib
import importlib.metadata
import json
import math
import os
import platform
import queue
import struct
import subprocess
import sys
import threading
import time
import types
from contextlib import ExitStack
from pathlib import Path

import numpy as np


BLOCK_BYTES = 8 << 20
PROTOCOL = "phase4ag-apl-v1"
AUTHOR_PATH = Path("/data/digital-drosophila/experiments/phase4ad-author-semantics/external-resources/model.py")
AUTHOR_SHA256 = "fc45837d7122c6ce2a7f3f2f23c515992e4b232aadb919efabb72337fac88e4e"
AE_PATH = Path("/data/digital-drosophila/models/male-cns-lif/author-runtime-lab-v1/oracle_stress.py")
AE_SHA256 = "32eb1c42383068ce52701d2139a376ce155ad7072cde019949f8ca2d55492409"
AF_PATH = Path("/data/digital-drosophila/models/male-cns-lif/author-fullgraph-lab-v1/oracle_full.py")
AF_SHA256 = "d72003d98f0f1624b12126eff50bd361c43d888f4a64d90ba9247a58976f1961"
EVENT_DTYPE = np.dtype([("step", "<i4"), ("neuron_idx", "<i4")])
BUNDLE_FILES = ("input_matrix.npy", "output_matrix.npy", "kc_globals.npy", "apl_globals.npy")
_PROCESS_USED = False


def _check(condition, message):
    if not condition:
        raise ValueError(message)


def _save(path, value):
    with Path(path).open("x", encoding="utf-8") as handle:
        json.dump(value, handle, indent=2, sort_keys=True, allow_nan=False)
        handle.write("\n")


def _digest(path):
    result = hashlib.sha256()
    with Path(path).open("rb") as handle:
        for block in iter(lambda: handle.read(BLOCK_BYTES), b""):
            result.update(block)
    return result.hexdigest()


def _array_hash(array, dtype):
    return hashlib.sha256(np.ascontiguousarray(array, dtype=dtype).tobytes()).hexdigest()


def _load_checked(path, expected, name):
    data = Path(path).read_bytes()
    _check(hashlib.sha256(data).hexdigest() == expected, f"Pinned source hash mismatch: {path}")
    module = types.ModuleType(name)
    module.__file__ = str(path)
    exec(compile(data, str(path), "exec"), module.__dict__)
    return module


def _new_memmap(path, dtype, shape):
    _check(not Path(path).exists(), f"Refusing to overwrite output: {path}")
    return np.lib.format.open_memmap(path, mode="w+", dtype=dtype, shape=shape)


def _budget(root, limit, reserve=0):
    total = 0
    for directory, _, filenames in os.walk(root, followlinks=False):
        for filename in filenames:
            total += (Path(directory) / filename).stat(follow_symlinks=False).st_size
    _check(total + int(reserve) <= limit,
           f"Output budget exceeded: {total}+{reserve}>{limit}; inputs and duration were not adjusted")
    return total


class _PipeReader:
    """One bounded reader thread, with timeout/EOF checks for every frame."""

    def __init__(self, pipe, timeout):
        self.pipe, self.timeout = pipe, float(timeout)
        self.chunks = queue.Queue(maxsize=8)
        self.buffer = bytearray()
        self.thread = threading.Thread(target=self._pump, daemon=True)
        self.thread.start()

    def _pump(self):
        try:
            while True:
                chunk = os.read(self.pipe.fileno(), 65536)
                self.chunks.put(chunk)
                if not chunk:
                    return
        except BaseException as error:
            self.chunks.put(error)

    def _fill(self, deadline):
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("APL sidecar response timed out")
        try:
            chunk = self.chunks.get(timeout=remaining)
        except queue.Empty as error:
            raise TimeoutError("APL sidecar response timed out") from error
        if isinstance(chunk, BaseException):
            raise RuntimeError("APL sidecar stdout reader failed") from chunk
        if not chunk:
            raise EOFError("APL sidecar ended before a complete response")
        self.buffer.extend(chunk)

    def read(self, size):
        deadline = time.monotonic() + self.timeout
        while len(self.buffer) < size:
            self._fill(deadline)
        result = bytes(self.buffer[:size])
        del self.buffer[:size]
        return result

    def line(self):
        deadline = time.monotonic() + self.timeout
        while b"\n" not in self.buffer:
            _check(len(self.buffer) < 1 << 20, "APL sidecar handshake exceeds 1 MiB")
            self._fill(deadline)
        end = self.buffer.index(b"\n") + 1
        result = bytes(self.buffer[:end])
        del self.buffer[:end]
        return result


class _APLWorker:
    def __init__(self, python, script, bundle_dir, stderr, timeout=180):
        self.command = [str(python), "-B", str(script), "--bundle-dir", str(bundle_dir), "--dt-ms", "0.1"]
        self.process = subprocess.Popen(self.command, stdin=subprocess.PIPE,
            stdout=subprocess.PIPE, stderr=stderr, bufsize=0, shell=False)
        self.reader = _PipeReader(self.process.stdout, timeout)
        self.tick = 0
        try:
            self.hello = json.loads(self.reader.line().decode("utf-8"))
            _check(self.hello.get("ok") is True, "APL sidecar did not acknowledge initialization")
            _check(self.hello.get("protocol") == PROTOCOL, "Unexpected APL sidecar protocol")
            for key, value in (("n_neurons", 166700), ("n_apl", 2), ("n_kc", 4064)):
                _check(self.hello.get(key) == value, f"APL sidecar changed {key}")
            _check(self.hello.get("dt_ms") == .1, "APL sidecar changed dt")
            expected_decay = f"{int(np.float32(math.exp(-.1 / 20.)).view(np.uint32)):08x}"
            expected_gain = f"{int(np.float32(.01).view(np.uint32)):08x}"
            _check(self.hello.get("decay_f32_bits") == expected_decay and self.hello.get("gain_f32_bits") == expected_gain,
                   "APL sidecar decay/gain differs from the frozen float32 scalars")
            _check(self.hello.get("files_sha256") == {name: _digest(bundle_dir / name) for name in BUNDLE_FILES},
                   "APL sidecar bundle hashes differ from the independently read bundle")
        except BaseException:
            self.abort()
            raise

    def step(self, tick, global_spikes):
        _check(tick == self.tick, "APL sidecar request ticks must be consecutive")
        nodes = np.asarray(global_spikes)
        _check(nodes.ndim == 1 and nodes.dtype == np.dtype("int32"), "APL input spikes must be int32 globals")
        _check(not len(nodes) or (nodes[0] >= 0 and nodes[-1] < 166700 and np.all(np.diff(nodes) > 0)),
               "APL input spikes must be sorted, unique, and in the frozen global range")
        packet = struct.pack("<qQ", int(tick), len(nodes)) + np.ascontiguousarray(nodes, dtype="<i4").tobytes()
        view = memoryview(packet)
        while view:
            written = self.process.stdin.write(view)
            if not written:
                raise BrokenPipeError("APL sidecar stdin failed")
            view = view[written:]
        self.process.stdin.flush()
        response_tick, byte_count = struct.unpack("<qQ", self.reader.read(16))
        _check(response_tick == tick and byte_count == (2 + 2 + 4064) * 4,
               "APL sidecar frame tick/length mismatch")
        values = np.frombuffer(self.reader.read(byte_count), dtype="<f4")
        _check(np.isfinite(values).all() and np.all(values >= 0), "APL sidecar returned invalid nonnegative f32 states")
        self.tick += 1
        return values[:2], values[2:4], values[4:]

    def finish(self, steps):
        _check(self.tick == steps, "APL sidecar did not process every tick")
        self.process.stdin.close()
        status = self.process.wait(timeout=30)
        _check(status == 0, f"APL sidecar exited unsuccessfully: {status}")
        self.reader.thread.join(timeout=5)
        _check(not self.reader.thread.is_alive(), "APL stdout did not reach EOF")
        extra = bytes(self.reader.buffer)
        while not self.reader.chunks.empty():
            chunk = self.reader.chunks.get_nowait()
            _check(not isinstance(chunk, BaseException), "APL stdout reader failed after the last frame")
            extra += chunk
        _check(not extra, "APL sidecar emitted unrequested trailing stdout bytes")
        self.process.stdout.close()
        return status

    def abort(self):
        if self.process.poll() is None:
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait(timeout=5)
        for handle in (self.process.stdin, self.process.stdout):
            if handle is not None and not handle.closed:
                handle.close()


def _graph_inputs(directory):
    arrays = {}
    definitions = {"nodes": "int32", "edge_id": "int64", "src": "int32",
                   "dst": "int32", "signed_raw": "int64", "ptr": "int64", "weights": "float64"}
    for name, dtype in definitions.items():
        value = np.load(directory / (name + ".npy"), mmap_mode="r", allow_pickle=False)
        _check(value.ndim == 1 and value.dtype == np.dtype(dtype), f"Invalid graph array {name}")
        arrays[name] = value
    nodes, edge_ids = arrays["nodes"], arrays["edge_id"]
    n, edges = len(nodes), len(edge_ids)
    _check(0 < n <= 166700 and edges > 0, "A case needs real global nodes and explicit nonempty connectivity")
    _check(nodes[0] >= 0 and nodes[-1] < 166700 and np.all(np.diff(nodes) > 0), "Invalid global node order")
    for name in ("src", "dst", "signed_raw", "weights"):
        _check(arrays[name].shape == (edges,), f"Graph array {name} has the wrong shape")
    pointer = arrays["ptr"]
    _check(pointer.shape == (n + 1,) and pointer[0] == 0 and pointer[-1] == edges and
           np.all(np.diff(pointer) >= 0), "Invalid graph CSR pointer")
    source_counts = np.zeros(n, np.int64)
    identity_edges = True
    previous_source, previous_edge = -1, -1
    width = BLOCK_BYTES // 8
    for begin in range(0, edges, width):
        stop = min(edges, begin + width)
        source, target, ids = arrays["src"][begin:stop], arrays["dst"][begin:stop], edge_ids[begin:stop]
        _check(np.all((source >= 0) & (source < n)) and np.all((target >= 0) & (target < n)),
               "Graph edge endpoint is outside the local node array")
        _check(np.all(ids >= 0), "Graph edge IDs must be nonnegative")
        _check(source[0] > previous_source or (source[0] == previous_source and ids[0] > previous_edge),
               "Graph edge blocks are not in canonical source/edge order")
        _check(np.all((source[1:] > source[:-1]) |
                      ((source[1:] == source[:-1]) & (ids[1:] > ids[:-1]))),
               "Graph edges are not in canonical source/edge order")
        previous_source, previous_edge = int(source[-1]), int(ids[-1])
        identity_edges &= np.array_equal(ids, np.arange(begin, stop, dtype=np.int64))
        source_counts += np.bincount(source, minlength=n)
        _check(np.isfinite(arrays["weights"][begin:stop]).all(), "Nonfinite graph weights")
    _check(np.array_equal(source_counts, np.diff(pointer)), "CSR pointer and source rows disagree")
    if not identity_edges:
        _check(len(np.unique(edge_ids)) == edges, "Original edge IDs must be unique")
    for name in ("completeness.csv", "connectivity.parquet"):
        _check((directory / name).is_file(), f"Missing unchanged author input: {name}")
    if n == 166700:
        _check(edges == 25582938 and identity_edges and np.array_equal(nodes, np.arange(n, dtype=np.int32)),
               "A full probe must use the unchanged complete canonical graph")
    return arrays, identity_edges


def _requests(case_dir, nodes, steps):
    with np.load(case_dir / "requests.npz", allow_pickle=False) as archive:
        _check(set(archive.files) == {"tick", "neuron_idx"}, "Unexpected sparse request schema")
        ticks, requested = archive["tick"].copy(), archive["neuron_idx"].copy()
    _check(ticks.dtype == requested.dtype == np.dtype("int32") and ticks.ndim == requested.ndim == 1 and
           ticks.shape == requested.shape, "Requests must be equal-length int32 vectors")
    _check(np.all((ticks >= 0) & (ticks < steps)), "Request time outside the case")
    _check(not len(ticks) or np.all((ticks[1:] > ticks[:-1]) |
           ((ticks[1:] == ticks[:-1]) & (requested[1:] > requested[:-1]))),
           "Requests must be sorted and unique by tick/global node")
    local = np.searchsorted(nodes, requested)
    _check(np.all(local < len(nodes)) and np.array_equal(nodes[local], requested),
           "Request references a global neuron absent from this case")
    counts = np.bincount(ticks, minlength=steps).astype(np.int32)
    pointer = np.concatenate((np.zeros(1, np.int64), np.cumsum(counts, dtype=np.int64)))
    return requested, local.astype(np.int32), pointer, counts


class _QueueProxy:
    def __init__(self, original, get_tick, available, arrays, identity_edges,
                 due_counts, spike_ring, delay_steps, handle, records, summarize):
        self._original_queue, self._get_tick = original, get_tick
        self._available, self._arrays, self._identity_edges = available, arrays, identity_edges
        self._due_counts, self._spike_ring, self._delay_steps = due_counts, spike_ring, delay_steps
        self._handle, self._records, self._summarize = handle, records, summarize

    def peek(self, *args, **kwargs):
        original = self._original_queue.peek(*args, **kwargs)
        local_edges = np.asarray(original)
        tick = self._get_tick()
        _check(tick == len(self._records["rows"]), "Queue peek ticks must be consecutive")
        _check(local_edges.ndim == 1 and np.all((local_edges >= 0) & (local_edges < len(self._arrays["edge_id"]))),
               "Original queue contains an invalid local edge index")
        _check(len(local_edges) == self._due_counts[tick % len(self._due_counts)],
               "Queue arrivals differ from the fixed-delay outgoing degree of actual spikes")
        fired_then = np.zeros(len(self._arrays["nodes"]), dtype=bool)
        if tick >= self._delay_steps:
            fired_then[self._spike_ring[(tick - self._delay_steps) % len(self._spike_ring)]] = True

        def accepted(begin, stop):
            part = local_edges[begin:stop]
            _check(np.all(fired_then[self._arrays["src"][part]]), "Arrival lacks a presynaptic spike at author delay")
            return self._available[self._arrays["dst"][part]]

        # Only the observed ID view changes; the queue is returned verbatim.
        edge_ids = local_edges if self._identity_edges else self._arrays["edge_id"][local_edges]
        summary = self._summarize(tick, edge_ids, accepted)
        self._handle.write(json.dumps(summary, sort_keys=True) + "\n")
        self._records["rows"].append(summary)
        self._records["peek_calls"] += 1
        return original

    def advance(self, *args, **kwargs):
        self._records["advance_calls"] += 1
        return self._original_queue.advance(*args, **kwargs)

    def __getattr__(self, name):
        return getattr(self._original_queue, name)


def run_case(case_dir: Path, out_dir: Path) -> dict:
    """Run one continuous case, with exclusive outputs and no candidate trace input.

    Graph inputs are the AF array/author-table schema plus original weights.npy.
    requests.npz contains sparse tick/neuron_idx int32 vectors.  Optional
    arrays.npz contains initial_v_mv/initial_g_mv float64 vectors only.
    Full probes write state hashes, sparse events and APL traces, not dense
    whole-brain histories.  Each case requires a fresh process.
    """
    global _PROCESS_USED
    _check(not _PROCESS_USED, "Every oracle case requires a fresh process")
    _PROCESS_USED = True
    case_dir, out_dir = Path(case_dir).resolve(strict=True), Path(out_dir).resolve()
    spec = json.loads((case_dir / "spec.json").read_text(encoding="utf-8"))
    graph_dir = Path(spec["graph_dir"]).resolve(strict=True)
    bundle_dir = Path(spec["apl_bundle_dir"]).resolve(strict=True)
    worker_path = Path(spec["legacy_apl_worker"]).resolve(strict=True)
    science_python = Path(spec["science_python"]).resolve(strict=True)
    author_path = Path(spec.get("author_path", AUTHOR_PATH)).resolve(strict=True)
    ae_path = Path(spec.get("ae_oracle_path", AE_PATH)).resolve(strict=True)
    af_path = Path(spec.get("af_oracle_path", AF_PATH)).resolve(strict=True)
    coefficients_path = Path(spec["coefficients_path"]).resolve(strict=True)
    steps = int(spec["steps"])
    _check(steps == spec["steps"] and 0 < steps <= np.iinfo(np.int32).max, "Invalid number of steps")
    _check(float(spec["dt_ms"]) == .1 and np.isclose(float(spec["duration_ms"]), steps * .1, rtol=0, atol=1e-12),
           "The frozen APL protocol requires dt_ms=0.1 and a matching duration")
    _check(spec.get("precision", "float64") == "float64", "Author state must be float64")
    _check(spec.get("training", False) is False and spec.get("poi_used", False) is False,
           "Training and original poi are outside this gate")
    _check(spec["apl"] == {"tau_ms": 20, "gain": .01, "normalization": 1}, "Frozen APL parameters changed")
    record_history = spec["record_history"]
    _check(isinstance(record_history, bool), "record_history must be a boolean")
    budget_root = Path(spec.get("budget_root", out_dir.parent)).resolve()
    budget_limit = int(spec.get("output_budget_bytes", 15 * 1024 ** 3))
    _check(budget_limit > 0 and out_dir.is_relative_to(budget_root), "Invalid output budget/root")
    out_dir.mkdir(parents=True, exist_ok=False)
    mapped_files = []
    saved_preferences, old_dt, worker = None, None, None
    last_complete_tick = -1
    try:
        arrays, identity_edges = _graph_inputs(graph_dir)
        nodes, n, edges = arrays["nodes"], len(arrays["nodes"]), len(arrays["edge_id"])
        _check(n < 166700 or not record_history, "Full probes must not save dense V/g/ref histories")
        bundle = {name: np.load(bundle_dir / (name + ".npy"), mmap_mode="r", allow_pickle=False)
                  for name in ("input_matrix", "output_matrix", "kc_globals", "apl_globals")}
        for name, shape in (("input_matrix", (2, 166700)), ("output_matrix", (2, 4064))):
            value = bundle[name]
            _check(value.dtype == np.dtype("float32") and value.shape == shape and value.flags.c_contiguous,
                   f"Frozen APL {name} shape/dtype/layout changed")
            _check(np.isfinite(value).all() and np.all(value >= 0), f"Invalid positive raw-normalized APL {name}")
        for name, count in (("kc_globals", 4064), ("apl_globals", 2)):
            value = bundle[name]
            _check(value.dtype == np.dtype("int32") and value.shape == (count,) and value[0] >= 0 and
                   value[-1] < 166700 and np.all(np.diff(value) > 0), f"Frozen APL identity {name} changed")
        _check(not np.intersect1d(bundle["kc_globals"], bundle["apl_globals"]).size, "APL and KC identities overlap")
        kc_positions = np.searchsorted(nodes, bundle["kc_globals"])
        present = kc_positions < n
        present[present] &= nodes[kc_positions[present]] == bundle["kc_globals"][present]
        kc_local, kc_in_bundle = kc_positions[present], np.flatnonzero(present)
        requested_globals, request_local, request_ptr, request_count = _requests(case_dir, nodes, steps)
        initial = None
        if (case_dir / "arrays.npz").exists():
            with np.load(case_dir / "arrays.npz", allow_pickle=False) as archive:
                _check(set(archive.files) == {"initial_v_mv", "initial_g_mv"}, "Unexpected initial-state schema")
                initial = {name: archive[name].copy() for name in archive.files}
            for name, value in initial.items():
                _check(value.dtype == np.dtype("float64") and value.shape == (n,) and np.isfinite(value).all(),
                       f"Invalid diagnostic initial state: {name}")
        history_bytes = steps * n * 40 if record_history else 0
        _budget(budget_root, budget_limit, history_bytes + steps * (2 + 2 + 4064) * 4 + edges * 8 + (64 << 20))
        versions = {name: importlib.metadata.version(name) for name in
                    ("brian2", "numpy", "pandas", "pyarrow", "sympy", "scipy", "cython")}
        _save(out_dir / "environment.json", {"python": sys.version, "platform": platform.platform(),
              "packages": versions, "codegen": "numpy", "author_precision": "float64",
              "APL_precision": "float32 in separate science-Python process", "Torch_imported_into_Brian": False})
        for name, required in (("brian2", "2.5.1"), ("numpy", "1.22.3"), ("sympy", "1.12")):
            _check(versions[name] == required, f"{name} must remain pinned to {required}")
        import brian2 as b2
        import pandas as pd
        from brian2.core.base import BrianObject

        ae = _load_checked(ae_path, AE_SHA256, "phase4ag_sealed_ae_helpers")
        af = _load_checked(af_path, AF_SHA256, "phase4ag_sealed_af_helpers")
        author = _load_checked(author_path, AUTHOR_SHA256, "phase4ag_unchanged_author")
        _check("torch" not in sys.modules, "Torch must not be imported into the pinned Brian process")
        comp = pd.read_csv(graph_dir / "completeness.csv", index_col=0)
        _check(np.array_equal(comp.index.to_numpy(), nodes) and
               np.array_equal(comp["local_index"].to_numpy(), np.arange(n)), "Author CSV neuron identity differs")
        del comp
        graph_hashes = {name: _digest(graph_dir / name) for name in
            tuple(key + ".npy" for key in arrays) + ("completeness.csv", "connectivity.parquet")}
        bundle_hashes = {name: _digest(bundle_dir / name) for name in BUNDLE_FILES}
        case_hashes = {name: _digest(case_dir / name) for name in ("spec.json", "requests.npz")}
        if initial is not None:
            case_hashes["arrays.npz"] = _digest(case_dir / "arrays.npz")
        worker_hash = _digest(worker_path)
        coefficient_hash = _digest(coefficients_path)
        baseline_coefficients = json.loads(coefficients_path.read_text(encoding="utf-8"))
        if "legacy_apl_worker_sha256" in spec:
            _check(worker_hash == spec["legacy_apl_worker_sha256"], "APL worker differs from preregistered source")
        _save(out_dir / "input-audit.json", {"spec": spec, "case_hashes": case_hashes,
              "graph_hashes": graph_hashes, "bundle_hashes": bundle_hashes,
              "author_sha256": AUTHOR_SHA256, "ae_sha256": AE_SHA256, "af_sha256": AF_SHA256,
              "legacy_apl_worker_sha256": worker_hash, "coefficients_sha256": coefficient_hash,
              "candidate_trace_used_as_input": False,
              "scope": "shared-APL-component closed-loop oracle, not all-components-independent CPU equivalence"})
        saved_preferences = {key: b2.prefs[key] for key in
            ("codegen.target", "core.default_float_dtype", "legacy.refractory_timing")}
        old_dt = b2.defaultclock.dt
        b2.start_scope()
        b2.prefs["codegen.target"] = "numpy"
        b2.prefs["core.default_float_dtype"] = np.float64
        b2.prefs["legacy.refractory_timing"] = False
        b2.defaultclock.dt = .1 * b2.ms
        parameters = dict(author.default_params)
        _save(out_dir / "parameter-checks.json", ae._parameters_match(parameters, spec["model"], b2))
        neu, syn, spikes = author.create_model(graph_dir / "completeness.csv", graph_dir / "connectivity.parquet", parameters)
        _check(len(neu) == n and len(syn) == edges, "Author loaded an unexpected graph size")
        _check(all(np.dtype(neu.variables[key].dtype) == np.float64 for key in ("v", "g")) and
               np.dtype(syn.variables["w"].dtype) == np.float64, "Author states/weights are not float64")
        effective = _new_memmap(out_dir / "weights.npy", "<f8", (edges,))
        mapped_files.append(effective)
        width, w_unit, mask_count = BLOCK_BYTES // 8, float(parameters["w_syn"] / b2.volt), 0
        mask_hasher = hashlib.sha256()
        base_hasher = hashlib.sha256()
        effective_hasher = hashlib.sha256()
        for begin in range(0, edges, width):
            stop = min(edges, begin + width)
            np.testing.assert_array_equal(np.asarray(syn.i[begin:stop]), arrays["src"][begin:stop])
            np.testing.assert_array_equal(np.asarray(syn.j[begin:stop]), arrays["dst"][begin:stop])
            original = np.asarray(syn.w_[begin:stop], dtype=np.float64).copy()
            raw_expected = arrays["signed_raw"][begin:stop] * w_unit
            _check(np.array_equal(original.view(np.uint64), raw_expected.view(np.uint64)) and
                   np.array_equal(original.view(np.uint64), arrays["weights"][begin:stop].view(np.uint64)),
                   "Actual original Brian weights differ from unchanged signed raw or AF base weights")
            base_hasher.update(np.ascontiguousarray(original, dtype="<f8").tobytes())
            mask = np.isin(nodes[arrays["src"][begin:stop]], bundle["apl_globals"])
            mask_count += int(mask.sum())
            mask_hasher.update(np.ascontiguousarray(arrays["edge_id"][begin:stop][mask], dtype="<i8").tobytes())
            original[mask] = np.float64(0.0)
            effective_hasher.update(np.ascontiguousarray(original, dtype="<f8").tobytes())
            effective[begin:stop] = original
            syn.w_[begin:stop] = original
        if n == 166700:
            _check(mask_count == 5845, "Full legacy APL fast-edge mask must contain exactly 5845 existing edges")
        effective.flush()
        _save(out_dir / "apl-mask-audit.json", {"count": mask_count, "ordered_original_edge_id_sha256": mask_hasher.hexdigest(),
              "base_weight_raw_bytes_sha256": base_hasher.hexdigest(), "effective_weights_file_sha256": _digest(out_dir / "weights.npy"),
              "effective_weights_raw_sha256": effective_hasher.hexdigest(),
              "rule": "zero +0 all existing fast edges whose source GLOBAL neuron is in apl_globals; no graph edge removal",
              "graded_targets": "only KC identities present in this LIF case", "KC_targets_in_case": len(kc_local),
              "base_graph_files_modified": False, "fullgraph_required_mask_count": 5845})
        if initial is not None:
            neu.v = initial["initial_v_mv"] * b2.mV
            neu.g = initial["initial_g_mv"] * b2.mV
        v_raw = neu.variables["v"].get_value()
        g_raw = neu.variables["g"].get_value()
        last_raw = neu.variables["lastspike"].get_value()
        available_raw = neu.variables["not_refractory"].get_value()
        histories = {}
        pre_apl_histories = {}
        if record_history:
            for name, dtype in (("v", "float64"), ("g", "float64"), ("ref", "int32")):
                histories[name] = _new_memmap(out_dir / f"{name}-history.{dtype}.npy", np.dtype(dtype), (steps, n))
                mapped_files.append(histories[name])
                pre_apl_histories[name] = _new_memmap(out_dir / f"preAPL-{name}.{dtype}.npy", np.dtype(dtype), (steps, n))
                mapped_files.append(pre_apl_histories[name])
        apl_history = {}
        for name, count in (("drive", 2), ("state", 2), ("inhibition", 4064)):
            apl_history[name] = _new_memmap(out_dir / f"apl-{name}.float32.npy", "<f4", (steps, count))
            mapped_files.append(apl_history[name])
        _save(out_dir / "output-schema.json", {"history_recorded": record_history, "history_shape": [steps, n],
              "state_hash_file": "tick-state-hashes.jsonl", "state_hash_encoding": "SHA256 of contiguous little-endian raw f64 V/g and i32 ref",
              "APL_dtype": "float32", "APL_h_units": "mV; gain already applied in float32",
              "APL_SI_conversion": "h.astype(float64) * float64(0.001), then separate float64 g subtraction",
              "APL_write_schedule": "end, order=0, after original reset, unconditional raw g write; recorder end/order=1",
              "APL_input": "Brian's own actual global spikes, never candidate spikes or candidate inhibition",
              "APL_shapes": {"drive": [steps, 2], "state": [steps, 2], "inhibition": [steps, 4064]},
              "request_columns": ["tick", "neuron_idx", "attempted", "accepted", "rejected"],
              "queue_protocol": "sealed Phase4AF queue_summary; original global edge IDs, including zero-weight/ref-rejected arrivals",
              "state_npz": ["step", "neuron_idx", "counts", "v", "g", "ref"],
              "complete_marker": "summary.json complete=true only after worker exit and all checks"})
        delay_steps = int(round(float(parameters["t_dly"] / b2.ms) / .1))
        ref_steps = int(round(float(parameters["t_rfc"] / b2.ms) / .1))
        _check(delay_steps == 18 and ref_steps == 22, "Author delay/refractory changed")
        queue_len = delay_steps + 1
        due_counts = np.zeros(queue_len, np.int64)
        spike_ring = [np.empty(0, np.int32) for _ in range(queue_len)]
        outdegree = np.diff(arrays["ptr"])
        request_seen = np.zeros(steps, np.uint8)
        apl_seen = np.zeros(steps, np.uint8)
        applicable_count = np.zeros(steps, np.int32)
        rejected_count = np.zeros(steps, np.int32)
        spike_count = np.zeros(steps, np.int32)
        total_counts = np.zeros(n, np.int32)
        records = {"rows": [], "peek_calls": 0, "advance_calls": 0, "installations": []}
        last_ticks = np.zeros(n, np.int64)
        final_ref = np.zeros(n, np.int32)
        permitted = np.empty(0, np.int32)
        total_events = 0

        def tick_now():
            return int(round(float(neu.clock.t / b2.second) / .0001))

        def current_ref(tick):
            seen = np.isfinite(last_raw) & (last_raw >= 0)
            last_ticks.fill(0)
            last_ticks[seen] = np.rint(last_raw[seen] / .0001).astype(np.int64)
            result = np.where(seen, np.maximum(ref_steps - (tick - last_ticks), 0), 0).astype(np.int32)
            _check(np.array_equal(result == 0, available_raw), "Ref history disagrees with author not_refractory")
            return result

        with ExitStack() as stack:
            stderr = stack.enter_context((out_dir / "apl-worker.stderr.log").open("xb"))
            worker = _APLWorker(science_python, worker_path, bundle_dir, stderr, float(spec.get("worker_timeout_seconds", 180)))
            _save(out_dir / "apl-worker-handshake.json", {"command": worker.command, "handshake": worker.hello})
            queue_file = stack.enter_context((out_dir / "queue.jsonl").open("x", encoding="utf-8"))
            event_file = stack.enter_context((out_dir / "events.bin").open("xb"))
            hash_file = stack.enter_context((out_dir / "tick-state-hashes.jsonl").open("x", encoding="utf-8"))
            tick_file = stack.enter_context((out_dir / "ticks.jsonl").open("x", encoding="utf-8"))
            requests_file = stack.enter_context((out_dir / "requests.csv").open("x", encoding="utf-8", newline=""))
            request_writer = csv.DictWriter(requests_file, fieldnames=["tick", "neuron_idx", "attempted", "accepted", "rejected"])
            request_writer.writeheader()

            def request_operation():
                nonlocal permitted
                tick = tick_now()
                _check(0 <= tick < steps and not request_seen[tick], "Invalid/duplicate common request tick")
                request_seen[tick] = 1
                begin, stop = request_ptr[tick:tick + 2]
                local = request_local[begin:stop]
                available = available_raw[local].copy()
                permitted = local[available]
                applicable_count[tick] = len(permitted)
                rejected_count[tick] = len(local) - len(permitted)
                for node, accepted in zip(requested_globals[begin:stop], available):
                    request_writer.writerow({"tick": tick, "neuron_idx": int(node), "attempted": 1,
                                             "accepted": int(accepted), "rejected": int(not accepted)})
                if len(permitted):
                    neu.v[permitted] = parameters["v_th"] + 1 * b2.mV

            def apl_operation():
                tick = tick_now()
                _check(0 <= tick < steps and not apl_seen[tick] and request_seen[tick], "Invalid/duplicate APL tick")
                active = np.asarray(neu.spikes, dtype=np.int32)
                _check(not len(active) or np.all(np.diff(active) > 0), "Brian actual spikes are not canonical/unique")
                _check(np.all(g_raw[active] == 0.0), "APL ran before the original author g reset")
                if record_history:
                    pre_apl_histories["v"][tick] = v_raw
                    pre_apl_histories["g"][tick] = g_raw
                    pre_apl_histories["ref"][tick] = current_ref(tick)
                drive, state, inhibition = worker.step(tick, np.asarray(nodes[active], dtype=np.int32))
                apl_history["drive"][tick] = drive
                apl_history["state"][tick] = state
                apl_history["inhibition"][tick] = inhibition
                # These are separate array operations.  Do not move the unit
                # conversion into a fused subtract; do not guard by refractory.
                delta_si = inhibition[kc_in_bundle].astype(np.float64) * np.float64(.001)
                g_raw[kc_local] = g_raw[kc_local] - delta_si
                apl_seen[tick] = 1

            def record_end():
                nonlocal last_complete_tick, total_events, final_ref
                tick = tick_now()
                _check(tick == last_complete_tick + 1 and tick < steps and apl_seen[tick], "End tick is incomplete/out of order")
                _check(len(records["rows"]) == tick + 1, "End state lacks queue observation")
                _check(np.isfinite(v_raw).all() and np.isfinite(g_raw).all(), "Nonfinite author/APL state")
                final_ref = current_ref(tick)
                if record_history:
                    histories["v"][tick], histories["g"][tick], histories["ref"][tick] = v_raw, g_raw, final_ref
                active = np.asarray(neu.spikes, dtype=np.int32)
                _check(np.isin(permitted, active).all(), "An applicable request did not produce a real spike")
                spike_count[tick] = len(active)
                total_counts[active] += 1
                if len(active):
                    events = np.empty(len(active), EVENT_DTYPE)
                    events["step"], events["neuron_idx"] = tick, nodes[active]
                    event_file.write(memoryview(events).cast("B"))
                total_events += len(active)
                spike_ring[tick % queue_len] = active.copy()
                due_counts[(tick + delay_steps) % queue_len] = int(outdegree[active].sum(dtype=np.int64))
                hash_file.write(json.dumps({"tick": tick, "v_sha256": _array_hash(v_raw, "<f8"),
                    "g_sha256": _array_hash(g_raw, "<f8"), "ref_sha256": _array_hash(final_ref, "<i4")}, sort_keys=True) + "\n")
                tick_file.write(json.dumps({"tick": tick, "spikes": len(active), "requests": int(request_count[tick]),
                    "applicable_requests": int(applicable_count[tick]), "ref_rejected_requests": int(rejected_count[tick]),
                    "event_sha256": _array_hash(nodes[active], "<i4"),
                    "due_arrivals": records["rows"][tick]["count"], "accepted_arrivals": records["rows"][tick]["accepted_count"],
                    "APL_drive_sha256": _array_hash(apl_history["drive"][tick], "<f4"),
                    "APL_state_sha256": _array_hash(apl_history["state"][tick], "<f4"),
                    "APL_inhibition_sha256": _array_hash(apl_history["inhibition"][tick], "<f4")}, sort_keys=True) + "\n")
                _budget(budget_root, budget_limit, reserve=1 << 20)
                last_complete_tick = tick
                if (tick + 1) % 100 == 0 or tick + 1 == steps:
                    for mapped in mapped_files:
                        mapped.flush()
                    for handle in (queue_file, event_file, hash_file, tick_file, requests_file):
                        handle.flush()
                    print(json.dumps({"oracle": spec["name"], "completed_ticks": tick + 1,
                                      "spikes": total_events, "APL_worker_ticks": worker.tick}), flush=True)

            class QueueInstaller(BrianObject):
                def __init__(self):
                    super().__init__(clock=neu.clock, when="end", order=1_000_000, name="zz_apl_queue_installer")

                def before_run(self, run_namespace):
                    installed = []
                    for code in syn.pre.code_objects:
                        if "_queue" not in code.namespace or "_queue.peek()" not in str(code.code):
                            continue
                        original = code.namespace["_queue"]
                        if isinstance(original, _QueueProxy):
                            original = original._original_queue
                        code.namespace["_queue"] = _QueueProxy(original, tick_now, available_raw, arrays, identity_edges,
                            due_counts, spike_ring, delay_steps, queue_file, records, af.queue_summary)
                        installed.append(code.name)
                    _check(len(installed) == 1, "Expected one original anatomical queue consumer")
                    records["installations"].append(installed[0])

            objects = [neu, syn, spikes,
                b2.NetworkOperation(request_operation, clock=neu.clock, when="before_thresholds", order=0, name="oracle_common_requests"),
                b2.NetworkOperation(apl_operation, clock=neu.clock, when="end", order=0, name="oracle_legacy_apl_unconditional"),
                b2.NetworkOperation(record_end, clock=neu.clock, when="end", order=1, name="oracle_after_apl_record"), QueueInstaller()]
            network = b2.Network(*objects)
            _save(out_dir / "schedule.json", {"schedule": list(network.schedule), "summary": str(network.scheduling_summary())})
            network.run(0 * b2.ms, namespace={})
            _save(out_dir / "compiled-code.json", ae._code_records(network))
            compiled_coefficients = ae._coefficients(neu, parameters, b2, .0001)
            coefficient_keys = ("gd", "c0", "cg", "cv", "threshold", "reset", "seed_volt")
            _check(all(float(compiled_coefficients[key]).hex() == float(baseline_coefficients[key]).hex()
                       for key in coefficient_keys), "Compiled Brian coefficients differ from sealed AF")
            _check(last_complete_tick == -1 and worker.tick == 0 and not records["rows"] and not request_seen.any(),
                   "Zero-duration compilation unexpectedly advanced state")
            network.run(steps * .1 * b2.ms, namespace={})
            _save(out_dir / "generated-code.json", ae._code_records(network))
            worker_status = worker.finish(steps)
            worker = None

        _check(last_complete_tick == steps - 1 and np.all(request_seen == 1) and np.all(apl_seen == 1), "Incomplete trial")
        _check(records["peek_calls"] == records["advance_calls"] == steps, "Incomplete queue audit")
        _check(total_events == int(spikes.num_spikes) and np.array_equal(total_counts, np.asarray(spikes.count[:])),
               "Sparse events disagree with original SpikeMonitor")
        _check(np.array_equal(request_count, applicable_count + rejected_count), "Incomplete request accounting")
        _check(np.all(np.asarray(neu.rfc[:] / b2.ms) == float(parameters["t_rfc"] / b2.ms)), "Author refractory duration changed")
        for begin in range(0, edges, width):
            stop = min(edges, begin + width)
            _check(np.array_equal(np.asarray(syn.w_[begin:stop]).view(np.uint64), effective[begin:stop].view(np.uint64)),
                   "Effective synaptic weights changed during the trial")
        event_values = np.memmap(out_dir / "events.bin", dtype=EVENT_DTYPE, mode="r", shape=(total_events,)) if total_events else np.empty(0, EVENT_DTYPE)
        _check((out_dir / "events.bin").stat().st_size == total_events * 8, "Sparse event byte count mismatch")
        for begin in range(0, total_events, width):
            stop = min(total_events, begin + width)
            spike_local = np.asarray(spikes.i[begin:stop], dtype=np.int32)
            np.testing.assert_array_equal(event_values["neuron_idx"][begin:stop], nodes[spike_local])
            times = np.rint(np.asarray(spikes.t[begin:stop] / b2.second) / .0001).astype(np.int32)
            np.testing.assert_array_equal(event_values["step"][begin:stop], times)
        _budget(budget_root, budget_limit, total_events * 8 + n * 24 + (8 << 20))
        with (out_dir / "state.npz").open("xb") as handle:
            np.savez_compressed(handle, step=event_values["step"], neuron_idx=event_values["neuron_idx"],
                                counts=total_counts, v=v_raw, g=g_raw, ref=final_ref)
        with (out_dir / "tick-summary.npz").open("xb") as handle:
            np.savez_compressed(handle, request_count=request_count, applicable_count=applicable_count,
                rejected_count=rejected_count, spike_count=spike_count,
                arrival_count=np.asarray([row["count"] for row in records["rows"]], np.int64),
                accepted_arrival_count=np.asarray([row["accepted_count"] for row in records["rows"]], np.int64))
        coefficients = ae._coefficients(neu, parameters, b2, .0001)
        keys = ("gd", "c0", "cg", "cv", "threshold", "reset", "seed_volt")
        _check(all(float(coefficients[key]).hex() == float(baseline_coefficients[key]).hex() for key in keys),
               "Actual Brian coefficients differ from sealed AF coefficients")
        _save(out_dir / "coefficients.json", coefficients)
        _save(out_dir / "queue-audit.json", {"peek_calls": records["peek_calls"], "advance_calls": records["advance_calls"],
              "installations": records["installations"], "same_peek_object_returned": True,
              "original_edge_ids_not_sorted_or_filtered": True, "zero_weight_and_refractory_rejected_edges_retained": True,
              "post_threshold_pre_synaptic_guard_observed": True, "forwarding_contract": "sealed AE/AF design",
              "new_unwrapped_repeat_performed": False, "limit": "Current wrapper was not independently rerun without the queue proxy"})
        for path, expected in ((author_path, AUTHOR_SHA256), (ae_path, AE_SHA256), (af_path, AF_SHA256), (worker_path, worker_hash)):
            _check(_digest(path) == expected, f"Source changed during the run: {path}")
        _check(_digest(coefficients_path) == coefficient_hash, "Sealed coefficient reference changed during the run")
        for directory, hashes in ((graph_dir, graph_hashes), (bundle_dir, bundle_hashes), (case_dir, case_hashes)):
            for name, expected in hashes.items():
                _check(_digest(directory / name) == expected, f"Read-only input changed during the run: {name}")
        for mapped in mapped_files:
            mapped.flush()
        _check("torch" not in sys.modules, "Pinned Brian process unexpectedly imported Torch")
        output_bytes = _budget(budget_root, budget_limit, 1 << 20)
        summary = {"name": spec["name"], "complete": True, "precision": "float64", "APL_precision": "float32",
            "backend": "shared-APL-component closed-loop Brian2 oracle", "steps": steps, "completed_ticks": last_complete_tick + 1,
            "nodes": n, "edges": edges, "spikes": total_events, "requests": int(request_count.sum()),
            "applicable_requests": int(applicable_count.sum()), "ref_rejected_requests": int(rejected_count.sum()),
            "APL_fast_edges_zeroed": mask_count, "APL_worker_exit_status": worker_status, "APL_worker_ticks": steps,
            "apl_fast_edges_zeroed": mask_count, "effective_weights_raw_sha256": effective_hasher.hexdigest(),
            "kc_targets_present": len(kc_local),
            "candidate_trace_used_as_input": False, "independent_CPU_all_components_claim": False,
            "history_recorded": record_history, "single_continuous_run": True, "no_training": True, "no_poi": True,
            "source_and_inputs_unchanged": True, "pass_all": True,
            "pass_all_scope": "Oracle construction, protocol and internal self-checks only; candidate comparison is a separate gate",
            "scope": "Only this fixed trajectory; legacy Torch APL is a shared numerical component, not an independent CPU APL implementation",
            "output_budget_bytes": budget_limit, "output_bytes_before_summary": output_bytes,
            "state_hashes_sha256": _digest(out_dir / "tick-state-hashes.jsonl"), "queue_sha256": _digest(out_dir / "queue.jsonl"),
            "events_sha256": _digest(out_dir / "events.bin"), "weights_sha256": _digest(out_dir / "weights.npy")}
        _save(out_dir / "summary.json", summary)
        return summary
    except Exception as error:
        for mapped in mapped_files:
            mapped.flush()
        _save(out_dir / "error.json", {"complete": False, "last_complete_tick": last_complete_tick,
              "type": type(error).__name__, "message": str(error), "partial_files_preserved": True,
              "inputs_or_duration_adjusted": False})
        raise
    finally:
        if worker is not None:
            worker.abort()
        if saved_preferences is not None:
            for key, value in saved_preferences.items():
                b2.prefs[key] = value
            b2.defaultclock.dt = old_dt