digital-drosophila/registry/phase4an-engineering-validation/source-snapshot/runtime_light.py

"""Engineering-only, fixed-weight trial execution derived from frozen AJ AST.

Public APIs run_light(case_dir,out_dir,build_root,overlay_file,overlay_manifest)
and run_reference(...) have identical arguments. Neither trains nor carries a
previous trial's neural state. The reference retains AJ's eager diagnostics;
the light function removes only an explicit diagnostic whitelist. Both accept
an explicit spec.max_steps instead of AJ's 1000-step admission bound.

The parent protocol owns conditioning, cue/tail schedules, frozen input/mask
authority and success criteria. A completed trial is not a learning PASS.
Before any scientific execution, the coordinator must verify externally fixed
hashes for the entire graph, APL bundle, coefficients/model, and preregistered
case spec (including max_steps). Inherited AJ consistency checks and this
adapter's input hashes are not a replacement for those external authorities.
"""
from __future__ import annotations

import ast
import copy
import hashlib
import importlib.util
import json
from pathlib import Path
import re
import sys

import numpy as np


AJ_SHA = 'af1975ba5e5e700208c0a598d3cbe648a3cf10885c6231239b6e4b28f76d173b'
AM_OVERLAY_SHA = '7f891d0ba79b393c96e75fd2dee2eddc51e57d8213eaac2779703bc2c18712b4'
MASK_SHA = '8b0fb6eb2e6c31da4d17dc20d890aa3f367e514f05642de5bad6b42a1611c05b'
V2_SHA = 'abb4574da85f00bd9ef61024c65993ef81d42651ff8d8d63c15dd69c6222471e'
PINS = {
    'apl-timing-lab-v1/gpu_timing.py': AJ_SHA,
    'apl-timing-lab-v1/timing_queue.py': '54df6d4818e2a0c26e96fb0aaf5d9b71cd964981706da937562d48af79aca14e',
    'author-apl-lab-v2/gpu_apl.py': '69f5d893f47dafff4ef7c3186c3df0b9014666b32c8f4e943b57016bec92b18a',
    'author-apl-lab-v2/legacy_apl_worker.py': '21ceec65fe42581dc55872010e0025d660e80358905f1cf4df1dadf341a3b291',
    'author-runtime-lab-v1/gpu_backend.py': '03758d07572f4b44d272059d9308fac534268ad201b57bdd77d69fee95e67278',
    'author-runtime-lab-v1/author_ext.cpp': '208a937d17bbb041b9bf6b25b89264ee566830b60609e58491b44583d8541f15',
    'author-runtime-lab-v1/author_ext.cu': '669f166e18ce9a3f63fc723db8b0a8b74f7fa4757194b1bd65b4cf6008a7787e',
    'author-fullgraph-lab-v1/oracle_full.py': 'd72003d98f0f1624b12126eff50bd361c43d888f4a64d90ba9247a58976f1961',
}
_LOADED = {}
_INSTALL = '_runtime.install(weights, graph, apl_out_mask, upload, out_dir, source_provenance, device)'
_FINISH = '_runtime.finish(ring, apl_ring, apl, steps, out_dir)'
_MATH_TARGETS = frozenset(('indptr', 'dst', 'global_nodes_gpu', 'base_weights', 'weights',
    'local_apl', 'apl_out_mask', 'common_kc', 'local_kc', 'full_kc', 'kc_local_gpu', 'kc_full_gpu',
    'v', 'g', 'ref', 'counts', 'requests', 'ring', 'apl_delay', 'apl_ring', 'requested_local_gpu',
    'spikes', 'due', 'sources', 'degrees', 'offsets', 'edge_count', 'expanded', 'destination',
    'order', 'ordered_edges', 'ordered_targets', 'boundary', 'starts', 'targets', 'full_spikes',
    'drive', 'apl_state', 'h', 'slot', 'applied_h', 'delta_si'))


def require(ok, message):
    if not ok:
        raise ValueError(message)


def digest(path):
    h = hashlib.sha256()
    with Path(path).open('rb') as stream:
        for block in iter(lambda: stream.read(8 << 20), b''):
            h.update(block)
    return h.hexdigest()


def _read_json(path):
    def pairs(items):
        out = {}
        for key, value in items:
            require(key not in out, 'Duplicate JSON key: ' + key)
            out[key] = value
        return out
    return json.loads(Path(path).read_text(encoding='utf-8'), object_pairs_hook=pairs,
                      parse_constant=lambda value: (_ for _ in ()).throw(ValueError('Nonfinite JSON: ' + value)))


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


def _raw(value):
    return hashlib.sha256(np.ascontiguousarray(value, dtype=value.dtype.newbyteorder('<')).tobytes()).hexdigest()


def _sha(value):
    return isinstance(value, str) and re.fullmatch('[0-9a-f]{64}', value) is not None


def _load(name, path, expected):
    path = Path(path).absolute()
    raw = path.read_bytes()
    require(hashlib.sha256(raw).hexdigest() == expected, 'Unpinned source: ' + str(path))
    if name in sys.modules:
        module = sys.modules[name]
        require(_LOADED.get(name) == (path, expected, module), 'Existing unverified module binding: ' + name)
        return module
    spec = importlib.util.spec_from_file_location(name, path)
    require(spec is not None and spec.loader is not None, 'Missing module loader')
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    try:
        exec(compile(raw, str(path), 'exec'), module.__dict__)
    except BaseException:
        sys.modules.pop(name, None)
        raise
    _LOADED[name] = (path, expected, module)
    return module


def validate_spec(spec, case_dir):
    """Admission policy only. No implicit early stopping or checkpoint support."""
    require(type(spec.get('max_steps')) is int and 0 < spec['max_steps'] <= np.iinfo(np.int32).max,
            'An explicit positive int32-range max_steps is mandatory')
    require(type(spec.get('steps')) is int and 0 < spec['steps'] <= spec['max_steps'], 'steps exceeds its declared bound')
    require(spec.get('dt_ms') == .1 and spec.get('duration_ms') == spec['steps'] * .1 and
            type(spec.get('apl_delay_steps')) is int and spec['apl_delay_steps'] == 18,
            'Frozen clock/duration/D18 contract changed')
    require(spec.get('record_history') is False, 'Only wholebrain fresh-state trials are supported')
    require(not (Path(case_dir) / 'arrays.npz').exists(), 'Initial arrays/checkpoint continuation are not supported in this version')
    require(not any('checkpoint' in str(key).lower() or str(key) in ('initial_state', 'resume', 'continuation') for key in spec),
            'Checkpoint/continuation configuration is unsupported')
    require(spec.get('within_trial_weight_updates', False) is False, 'No within-trial plasticity is supported')
    return spec


def load_overlay(overlay_file, manifest_file):
    """New manifest policy, frozen AM array validators; no old-global mutation.

    phase4an-overlay-v1 requires overlay_name zero/static25/learned,
    overlay_sha256, original mask_path/mask_sha256, external positive f32-mV
    base_candidate_path/base_candidate_sha256, aj_model_dir,
    am_overlay_path/am_overlay_sha256. Learned snapshots additionally require
    parent_snapshot_sha256 and eligibility_v2_source_sha256=the frozen 4J pin.
    These hashes bind provenance; the coordinator must verify the parent chain.
    NPZ contains ONLY edge_id:int64[14232], plastic_delta_mv:float32[14232].
    """
    overlay_file, manifest_file = Path(overlay_file).absolute(), Path(manifest_file).absolute()
    manifest_sha = digest(manifest_file)
    meta = _read_json(manifest_file)
    keys = {'schema', 'overlay_name', 'overlay_sha256', 'mask_path', 'mask_sha256',
            'base_candidate_path', 'base_candidate_sha256', 'aj_model_dir', 'am_overlay_path', 'am_overlay_sha256'}
    require(isinstance(meta, dict) and keys <= meta.keys() and meta['schema'] == 'phase4an-overlay-v1', 'Invalid overlay manifest schema')
    require(meta['overlay_name'] in ('zero', 'static25', 'learned'), 'Unknown overlay snapshot type')
    learned_keys = {'parent_snapshot_sha256', 'eligibility_v2_source_sha256'}
    allowed = keys | (learned_keys if meta['overlay_name'] == 'learned' else set())
    require(not (meta.keys() - allowed), 'Unknown or inapplicable overlay manifest fields')
    require(meta['mask_sha256'] == MASK_SHA and meta['am_overlay_sha256'] == AM_OVERLAY_SHA,
            'Original mask or frozen AM helper authority changed')
    require(_sha(meta['overlay_sha256']) and digest(overlay_file) == meta['overlay_sha256'], 'Overlay SHA mismatch')
    inputs = {str(overlay_file): meta['overlay_sha256'], str(manifest_file): manifest_sha}
    for label in ('mask', 'base_candidate', 'am_overlay'):
        path = Path(meta[label + '_path'])
        require(path.is_absolute() and _sha(meta[label + '_sha256']) and digest(path) == meta[label + '_sha256'],
                'Missing/mismatched anchored input: ' + label)
        inputs[str(path)] = meta[label + '_sha256']
    helper = _load('phase4an_sealed_am_overlay', meta['am_overlay_path'], AM_OVERLAY_SHA)
    mask = np.load(meta['mask_path'], allow_pickle=False)
    base = np.load(meta['base_candidate_path'], allow_pickle=False)
    require(mask.dtype == np.int64 and mask.shape == (helper.CANDIDATE_COUNT,), 'Original complete 14,232 mask is required')
    with np.load(overlay_file, allow_pickle=False) as archive:
        require(len(archive.files) == 2 and set(archive.files) == {'edge_id', 'plastic_delta_mv'}, 'No embedded base or additional overlay arrays allowed')
        plan = helper.validate_arrays(archive['edge_id'], archive['plastic_delta_mv'], mask, base)
    delta = plan['plastic_delta_mv']
    if meta['overlay_name'] == 'zero':
        require(not np.any(delta != 0), 'Zero snapshot contains a nonzero delta')
    elif meta['overlay_name'] == 'static25':
        require(helper.bit_equal(delta, np.multiply(np.float32(-.25), base)), 'static25 must equal the original f32 floor')
    else:
        require(_sha(meta.get('parent_snapshot_sha256')) and meta.get('eligibility_v2_source_sha256') == V2_SHA,
                'Learned snapshot needs parent hash and explicit frozen eligibility-v2 authority')
    aj_dir = Path(meta['aj_model_dir'])
    require(aj_dir.is_absolute() and aj_dir.name == 'apl-timing-lab-v1', 'AJ must use its real original directory')
    for relative, expected in PINS.items():
        path = aj_dir.parent / relative
        require(digest(path) == expected, 'Sealed dependency changed: ' + relative)
        inputs[str(path)] = expected
    require(all(digest(path) == expected for path, expected in inputs.items()), 'Input changed while loading')
    return helper, plan, meta, inputs


def _dump(node):
    return ast.dump(node, annotate_fields=True, include_attributes=False)


def _target_roots(node):
    if isinstance(node, ast.Name):
        return {node.id}
    if isinstance(node, (ast.Tuple, ast.List)):
        return set().union(*(_target_roots(value) for value in node.elts))
    if isinstance(node, (ast.Subscript, ast.Attribute)):
        return _target_roots(node.value)
    return set()


def math_projection(function):
    """Ordered/control-structured frozen math statements, excluding diagnostics.

    Whitelist explicitly covers state/weight initialization, request mutation,
    expansion/stable grouping, original CUDA calls, LegacyAPL and ring/g write.
    Tests also require the entire reference loop AST to equal the original.
    """
    def block(statements):
        result = []
        for node in statements:
            if isinstance(node, ast.FunctionDef) and node.name == 'upload':
                result.append(copy.deepcopy(node))
            elif isinstance(node, ast.Assign) and set().union(*(_target_roots(t) for t in node.targets)) & _MATH_TARGETS:
                result.append(copy.deepcopy(node))
            elif isinstance(node, ast.Expr) and isinstance(node.value, ast.Call):
                text = ast.unparse(node.value.func)
                if text.startswith('ext.') or text in ('requests.zero_', 'apl.spikes.zero_', 'apl_ring[slot].copy_'):
                    result.append(copy.deepcopy(node))
            elif isinstance(node, (ast.With, ast.Try)):
                result.extend(block(node.body))
            elif isinstance(node, (ast.If, ast.For)):
                body, other = block(node.body), block(node.orelse)
                if body or other:
                    keep = copy.deepcopy(node)
                    keep.body, keep.orelse = body, other
                    result.append(keep)
        return result
    return _dump(ast.Module(body=block(function.body), type_ignores=[]))


def derive_ast(source_bytes, mode):
    """Derive only from exact AJ bytes; return ASTs plus an explicit edit receipt."""
    require(mode in ('light', 'reference'), 'Unknown diagnostic mode')
    require(hashlib.sha256(source_bytes).hexdigest() == AJ_SHA, 'AJ source pin mismatch')
    source = ast.parse(source_bytes)
    original = next(node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == 'run_gpu')
    loader = copy.deepcopy(next(node for node in source.body if isinstance(node, ast.FunctionDef) and node.name == 'load_inputs'))
    bounds = [node for node in ast.walk(loader) if isinstance(node, ast.Compare) and
              ast.unparse(node) == '0 < steps <= 1000']
    require(len(bounds) == 1, 'Expected one explicit step-admission bound')
    bounds[0].comparators[-1] = ast.parse("spec['max_steps']", mode='eval').body
    function = copy.deepcopy(original)
    body = next(node for node in function.body if isinstance(node, ast.Try)).body
    anchor = next(i for i, node in enumerate(body) if ast.unparse(node) == 'effective_hash = hashlib.sha256()')
    require(ast.unparse(body[anchor - 1]) == 'if bool(np.any(apl_out_mask)):\n    weights[upload(apl_out_mask, torch.bool)] = 0.0',
            'Installation no longer follows APL fast-edge zeroing')
    body.insert(anchor, ast.parse(_INSTALL).body[0])
    sync = [i for i, node in enumerate(body) if ast.unparse(node) == 'torch.cuda.synchronize(device)']
    require(len(sync) == 1, 'Expected one post-loop synchronization')
    body.insert(sync[0] + 1, ast.parse(_FINISH).body[0])
    removed, replacements = [], []

    class ReportingOnly(ast.NodeTransformer):
        def note(self, node):
            removed.append(_dump(node))
            return None

        def visit_Assign(self, node):
            names = set().union(*(_target_roots(t) for t in node.targets))
            if names & {'expanded_global', 'accepted_cpu', 'queue', 'apl_cpu', 'actual', 'finite', 'row'}:
                return self.note(node)
            if names == {'maps'} and isinstance(node.targets[0], ast.Subscript):
                return self.note(node)
            if names == {'timing'}:
                require(ast.unparse(node.value.func) == 'audit_sequence', 'Unexpected timing assignment')
                replacements.append(_dump(node))
                return ast.parse('timing = _runtime.timing(steps, pending)').body[0]
            if names == {'destination'}:
                return [node, ast.parse('_runtime.arrivals(expanded, destination, ref)').body[0]]
            if ast.unparse(node) == 'g[kc_local_gpu] = g[kc_local_gpu] - delta_si':
                return [node, ast.parse('_runtime.after_tick(tick, v, g, ref, drive, apl_state, h, applied_h)').body[0]]
            return node

        def visit_AugAssign(self, node):
            if _target_roots(node.target) & {'due_total', 'accepted_due_total'}:
                return self.note(node)
            return node

        def visit_For(self, node):
            text = ast.unparse(node.iter)
            if text in ("(('drive', 2), ('state', 2), ('inhibition', 4064))", 'apl_cpu.items()', 'actual.items()'):
                return self.note(node)
            return self.generic_visit(node)

        def visit_If(self, node):
            if ast.unparse(node.test) in ("spec['record_history']", 'not finite'):
                return self.note(node)
            return self.generic_visit(node)

        def visit_Expr(self, node):
            if isinstance(node.value, ast.Call) and ast.unparse(node.value.func) in ('queue_output.write', 'hash_output.write'):
                return self.note(node)
            return node

        def visit_With(self, node):
            targets = [getattr(item.optional_vars, 'id', None) for item in node.items]
            if targets == ['queue_output', 'hash_output', 'request_output']:
                replacements.append(_dump(node.items[0])); replacements.append(_dump(node.items[1]))
                node.items = [node.items[-1]]
            return self.generic_visit(node)

    if mode == 'light':
        function = ReportingOnly().visit(function)
        require(len(removed) == 20 and len(replacements) == 3, 'Reporting whitelist changed; stop instead of broad rewriting')
    loops = lambda f: [node for node in ast.walk(f) if isinstance(node, ast.For) and
                      isinstance(node.target, ast.Name) and node.target.id == 'tick']
    require(len(loops(original)) == len(loops(function)) == 1, 'Missing/duplicated trial loop')
    if mode == 'reference':
        require(_dump(loops(original)[0]) == _dump(loops(function)[0]), 'Reference loop must stay entirely unchanged')
    original_math, derived_math = math_projection(original), math_projection(function)
    require(original_math == derived_math, 'Ordered mathematical AST changed')
    extra_calls = [ast.unparse(node) for node in ast.walk(function) if isinstance(node, ast.Call) and
                   ast.unparse(node.func).startswith('_runtime.')]
    expected_calls = [_INSTALL, _FINISH]
    if mode == 'light':
        expected_calls += ['_runtime.arrivals(expanded, destination, ref)',
            '_runtime.after_tick(tick, v, g, ref, drive, apl_state, h, applied_h)',
            '_runtime.timing(steps, pending)']
    require(sorted(extra_calls) == sorted(expected_calls), 'Unlisted runtime callback added')
    tree = ast.fix_missing_locations(ast.Module(body=[loader, function], type_ignores=[]))
    sha = lambda value: hashlib.sha256(value.encode('utf-8')).hexdigest()
    receipt = {'mode': mode, 'aj_source_sha256': AJ_SHA, 'math_ast_sha256': sha(original_math),
        'derived_math_ast_sha256': sha(derived_math), 'mathematical_ast_identical': True,
        'reference_entire_tick_loop_identical': mode == 'reference',
        'admission_edit': '0 < steps <= 1000 -> 0 < steps <= spec.max_steps',
        'pre_loop_installation': _INSTALL, 'post_loop_terminal_recording': _FINISH,
        'added_runtime_callbacks': extra_calls,
        'removed_diagnostic_statements': removed, 'replaced_diagnostic_nodes': replacements,
        'derived_module_ast_sha256': sha(_dump(tree)), 'old_globals_patched': False}
    return tree, receipt


class FiniteObserver:
    """GPU scalar diagnostics only; never writes a neuronal/APL input tensor."""
    def __init__(self, torch, device):
        self.torch = torch
        self.ok = torch.ones((), device=device, dtype=torch.bool)
        self.first = torch.full((), -1, device=device, dtype=torch.int64)
        self.accepted = torch.zeros((), device=device, dtype=torch.int64)
        self.due = 0
        self.ticks = 0

    def arrivals(self, expanded, destination, ref):
        self.due += expanded.numel()
        self.accepted.add_((ref[destination.long()] == 0).to(self.torch.int64).sum())

    def observe(self, tick, *values):
        require(tick == self.ticks, 'Finite observer tick coverage changed')
        finite = self.torch.ones_like(self.ok)
        for value in values:
            finite.logical_and_(self.torch.isfinite(value).all())
        self.first = self.torch.where((self.first < 0) & (~finite), self.torch.full_like(self.first, tick), self.first)
        self.ok.logical_and_(finite)
        self.ticks += 1

    def result(self):
        return {'all_finite': bool(self.ok.item()), 'first_failure_tick': int(self.first.item()),
                'finite_observed_ticks': self.ticks, 'arriving_edges': self.due,
                'accepted_arriving_edges': int(self.accepted.item())}


def _copy_input(source, target, expected):
    with Path(source).open('rb') as src, Path(target).open('xb') as dst:
        for block in iter(lambda: src.read(8 << 20), b''):
            dst.write(block)
    require(digest(target) == expected, 'Copied input differs')


class _Runtime:
    def __init__(self, mode, spec, helper, plan, meta, inputs, torch, overlay_file, manifest_file):
        self.mode, self.spec, self.helper, self.plan, self.meta, self.inputs = mode, spec, helper, plan, meta, inputs
        self.torch, self.overlay_file, self.manifest_file = torch, Path(overlay_file).absolute(), Path(manifest_file).absolute()
        self.observer = None
        self.installations = 0
        self.finished = False
        self.result = None

    def install(self, weights, graph, apl_mask, upload, out_dir, inherited, device):
        require(self.installations == 0, 'Repeated installation is forbidden')
        require(len(graph['nodes']) == 166700 and len(graph['edge_id']) == self.helper.EDGE_COUNT,
                'Only the intact full graph is permitted')
        require(int(apl_mask.sum()) == 5845, 'Original fast APL mask changed')
        mapping = self.helper.map_installation(graph['weights'], graph['edge_id'], apl_mask, self.plan)
        require(mapping['missing_candidate_count'] == 0, 'All original candidates must be present')
        rows = mapping['rows']
        if len(rows):
            index = upload(rows, self.torch.int64)
            nonzero = self.plan['plastic_delta_mv'] != np.float32(0)
            change = upload(self.plan['plastic_delta_mv'][nonzero], self.torch.float64) * 0.001
            weights[index] = weights[index] + change
            require(self.helper.bit_equal(weights[index].cpu().numpy(), mapping['expected_si']), 'Independent SI installation mismatch')
        raw_base, expected = self.helper.expected_hashes(graph['weights'], apl_mask, mapping)
        actual = hashlib.sha256()
        for start in range(0, weights.numel(), 1 << 20):
            actual.update(weights[start:start + (1 << 20)].cpu().numpy().astype('<f8', copy=False).tobytes())
        require(actual.hexdigest() == expected, 'Effective weight bytes differ from declared overlay')
        self.effective_sha = expected
        self.observer = FiniteObserver(self.torch, device)
        self.installations += 1
        _copy_input(self.overlay_file, out_dir / 'overlay-input.npz', self.meta['overlay_sha256'])
        _copy_input(self.manifest_file, out_dir / 'overlay-manifest.json', self.inputs[str(self.manifest_file)])
        _save(out_dir / 'overlay-audit.json', {'schema': 'phase4an-overlay-install-v1', 'complete': True,
            'overlay_name': self.meta['overlay_name'], 'candidate_count': len(self.plan['edge_id']),
            'nonzero_delta_count': len(rows), 'am_helper_sha256': AM_OVERLAY_SHA, 'mask_sha256': MASK_SHA,
            'base_candidate_sha256': self.meta['base_candidate_sha256'], 'delta_raw_sha256': _raw(self.plan['plastic_delta_mv']),
            'base_weights_raw_sha256': raw_base, 'effective_weights_raw_sha256': expected,
            'delta_dtype': 'float32', 'delta_units': 'mV', 'state_dtype': 'float64', 'bridge': self.helper.BRIDGE,
            'zero_rows_untouched': True, 'weights_updated_during_run': False, 'apl_fast_edges_zeroed': int(apl_mask.sum()),
            'parent_snapshot_sha256': self.meta.get('parent_snapshot_sha256'),
            'eligibility_v2_source_sha256': self.meta.get('eligibility_v2_source_sha256'),
            'parent_chain_verified_by_runtime': False, 'inherited_AJ_provenance_only': inherited})

    def arrivals(self, expanded, destination, ref):
        self.observer.arrivals(expanded, destination, ref)

    def after_tick(self, tick, v, g, ref, drive, state, h, applied):
        self.observer.observe(tick, v, g, ref, drive, state, h, applied)

    def finish(self, ring, apl_ring, apl, steps, out_dir):
        require(not self.finished and self.installations == 1, 'Invalid terminal recording sequence')
        if self.mode == 'light':
            self.result = self.observer.result()
            require(self.result['finite_observed_ticks'] == steps, 'Missing finite checks')
        else:
            # Reaching this point means AJ's original per-tick finite check passed.
            self.result = {'all_finite': True, 'first_failure_tick': -1, 'finite_observed_ticks': steps}
        _save(out_dir / 'finite-audit.json', dict(self.result, mode=self.mode,
            scope='every completed tick; no neuronal or APL tensor mutated by observer'))
        raw_ring, raw_apl = ring.cpu().numpy().copy(), apl_ring.cpu().numpy().copy()
        head = steps % len(raw_apl)
        pending = np.concatenate((raw_apl[head:], raw_apl[:head]), axis=0)
        apl_state = apl.state.cpu().numpy().copy()
        require(raw_ring.dtype == np.uint8 and raw_ring.shape == (19, 166700) and
                apl_state.dtype == pending.dtype == np.float32 and apl_state.shape == (2,) and pending.shape == (18, 4064),
                'Terminal queue dtype/shape changed')
        with (out_dir / 'terminal-queues.npz').open('xb') as target:
            np.savez_compressed(target, spike_ring=raw_ring, apl_state=apl_state, apl_pending=pending,
                                next_tick=np.asarray(steps, dtype=np.int64))
        self.finished = True
        require(self.result['all_finite'], 'Nonfinite value observed; trial is invalid and must not produce a learning update')

    def timing(self, steps, pending):
        return {'apl_delay_steps': 18, 'steps': steps, 'pending_generated_count': min(steps, 18),
            'pending_initial_zero_prefix': max(18 - steps, 0), 'brain_tail_advanced': False,
            'full_sequence_bitwise_audit_performed': False,
            'scope': 'Original D18 ring AST retained; generated/applied histories not recorded'}

    def summary(self, value):
        result = dict(value)
        result.pop('learning', None)
        result.update(diagnostic_mode=self.mode, within_trial_weight_updates=False,
            outer_protocol=self.spec.get('outer_protocol'), outer_learning_status='not evaluated by this single-trial executor',
            learning_pass_claimed=False, checkpoint_continuation_used=False,
            max_steps=self.spec['max_steps'], explicit_steps_executed=self.spec['steps'],
            entire_neural_trial_executed=True, hidden_early_stop=False,
            full_APL_history_recorded=self.mode == 'reference', full_tick_state_hashes_recorded=self.mode == 'reference',
            full_queue_hashes_recorded=self.mode == 'reference',
            full_sequence_bitwise_audit_performed=self.mode == 'reference',
            first_failure_tick=self.result['first_failure_tick'],
            pass_all_scope='single fixed-weight trial finite/recording integrity only; no learning or equivalence claim')
        if self.mode == 'light':
            result.update(arriving_edges=self.result['arriving_edges'], accepted_arriving_edges=self.result['accepted_arriving_edges'])
            if 'apl_timing' in result:
                result['apl_timing'] = dict(result['apl_timing'], generated_trace=None, applied_trace=None,
                                            full_sequence_bitwise_audit_performed=False)
        return result

    def write_json(self, path, value):
        path = Path(path)
        if path.name == 'summary.json':
            value = self.summary(value)
        elif path.name == 'apl-timing-audit.json' and self.mode == 'light':
            value = dict(value)
            value.update(generated_trace=None, applied_trace=None, full_sequence_bitwise_audit_performed=False)
        _save(path, value)


def compile_runtime(aj_module, runtime, mode):
    """Fresh namespace; __file__ is this real adapter, not a disguised old file."""
    tree, audit = derive_ast(Path(aj_module.__file__).read_bytes(), mode)
    namespace = dict(aj_module.__dict__)
    namespace.update(__name__=__name__ + '.derived_' + mode, __file__=str(Path(__file__).absolute()),
                     _runtime=runtime, write_json=runtime.write_json)
    exec(compile(tree, str(Path(__file__).absolute()) + '::AJ-AST-' + mode, 'exec'), namespace)
    audit.update(derived_namespace_file=namespace['__file__'], original_module_file=str(Path(aj_module.__file__).absolute()),
        output_metadata_binding='new write_json only labels recording/within-trial scope; no old global mutated')
    return namespace['run_gpu'], audit


def execution_paths(case_dir, out_dir, build_root, protected=()):
    """Resolve directory aliases before containment checks, without any writes.

    Only data/source/output directories are canonicalized here. This helper
    must not be used for a Python executable: resolving a venv entry changes
    interpreter semantics. The coordinator must keep input paths stable during
    execution; this is not a defense against concurrent filesystem mutation.
    """
    case_dir = Path(case_dir).resolve(strict=True)
    require(case_dir.is_dir(), 'Input case must be an existing directory')
    out_dir, build_root = (Path(path).resolve(strict=False) for path in (out_dir, build_root))
    require(not out_dir.exists(), 'Output must be a new directory')
    require(not build_root.exists() or build_root.is_dir(), 'Build cache must be a directory')
    roots = [case_dir]
    for path in protected:
        root = Path(path).resolve(strict=True)
        require(root.is_dir(), 'Protected input/source root must be a directory')
        roots.append(root)
    require(not any(out_dir.is_relative_to(root) or build_root.is_relative_to(root) for root in roots),
            'Output/build paths cannot write into frozen source or input directories')
    return case_dir, out_dir, build_root


def _run(mode, case_dir, out_dir, build_root, overlay_file, overlay_manifest):
    require(sys.dont_write_bytecode, 'Use python -B; frozen dependencies must not receive bytecode writes')
    case_dir, out_dir, build_root = execution_paths(case_dir, out_dir, build_root)
    spec = validate_spec(_read_json(case_dir / 'spec.json'), case_dir)
    helper, plan, meta, inputs = load_overlay(overlay_file, overlay_manifest)
    own_sha = digest(__file__)
    inputs[str(case_dir / 'spec.json')] = digest(case_dir / 'spec.json')
    inputs[str(case_dir / 'requests.npz')] = digest(case_dir / 'requests.npz')
    inputs[str(Path(spec['graph_dir']) / 'weights.npy')] = digest(Path(spec['graph_dir']) / 'weights.npy')
    aj_dir = Path(meta['aj_model_dir']).resolve(strict=True)
    protected = (aj_dir.parent, Path(meta['am_overlay_path']).resolve(strict=True).parent,
                 Path(spec['graph_dir']), Path(spec['apl_bundle_dir']),
                 *(Path(path).resolve(strict=True).parent for path in
                   (overlay_file, overlay_manifest, meta['mask_path'], meta['base_candidate_path'], spec['coefficients_path'])))
    case_dir, out_dir, build_root = execution_paths(case_dir, out_dir, build_root, protected)
    _load('timing_queue', aj_dir / 'timing_queue.py', PINS['apl-timing-lab-v1/timing_queue.py'])
    aj = _load('phase4an_sealed_aj_gpu', aj_dir / 'gpu_timing.py', AJ_SHA)
    runtime = _Runtime(mode, spec, helper, plan, meta, inputs, aj.torch, overlay_file, overlay_manifest)
    function, ast_audit = compile_runtime(aj, runtime, mode)
    try:
        inherited = function(case_dir, out_dir, build_root)
        require(runtime.finished and runtime.installations == 1 and inherited['complete'], 'Incomplete single-trial execution')
        require(inherited['effective_weights_raw_sha256'] == runtime.effective_sha, 'Effective weight identity changed during trial')
        require(digest(__file__) == own_sha and all(digest(path) == expected for path, expected in inputs.items()),
                'Source or input changed during execution')
        summary = _read_json(out_dir / 'summary.json')
        require(summary == runtime.summary(inherited), 'Stored summary does not bind actual trial')
        _save(out_dir / 'runtime-provenance.json', {'schema': 'phase4an-reporting-adapter-v1', 'complete': True,
            'source_file': str(Path(__file__).absolute()), 'source_sha256': own_sha, 'mode': mode, 'ast': ast_audit,
            'input_sha256': inputs, 'old_globals_patched': False, 'math_changed': False,
            'admission_bound_changed_explicitly': True, 'weights_updated_during_run': False,
            'learning_pass_claimed': False, 'parent_chain_verification_owned_by_coordinator': True,
            'summary_sha256': digest(out_dir / 'summary.json'), 'finite_audit_sha256': digest(out_dir / 'finite-audit.json'),
            'terminal_queues_sha256': digest(out_dir / 'terminal-queues.npz'),
            'overlay_audit_sha256': digest(out_dir / 'overlay-audit.json')})
        return summary
    except Exception as error:
        if out_dir.is_dir() and not (out_dir / 'runtime-error.json').exists():
            _save(out_dir / 'runtime-error.json', {'complete': False, 'type': type(error).__name__, 'message': str(error),
                                                'learning_update_forbidden': True})
        raise


def run_light(case_dir, out_dir, build_root, overlay_file, overlay_manifest):
    return _run('light', case_dir, out_dir, build_root, overlay_file, overlay_manifest)


def run_reference(case_dir, out_dir, build_root, overlay_file, overlay_manifest):
    return _run('reference', case_dir, out_dir, build_root, overlay_file, overlay_manifest)