Tooling: opt-in parallel U8 layer-end controls, byte-identical receipt, 3.0x on 10 cores (no new kappa)

Makefile

@@ -604,3 +604,13 @@ frontier890-verify: selected-record-check
 frontier942-verify: selected-record-check
 	@work=$$(mktemp -d "$${TMPDIR:-/tmp}/integer-mult-frontier942.XXXXXX"); \
 	python3 -B scripts/verify_frontier942.py --work "$$work/replay"
+
+# Tooling: opt-in parallel U8 layer-end controls with a byte-identical receipt (no new kappa).
+.PHONY: layer-end-controls-parallel-verify layer-end-controls-parallel-benchmark
+layer-end-controls-parallel-verify:
+	@work=$$(mktemp -d "$${TMPDIR:-/tmp}/integer-mult-layer-end.XXXXXX"); \
+	python3 -B research/layer-end-controls-parallel/verify.py --work "$$work/replay" && \
+	python3 -B -m unittest discover -s research/layer-end-controls-parallel -p test_package.py
+layer-end-controls-parallel-benchmark:
+	@work=$$(mktemp -d "$${TMPDIR:-/tmp}/integer-mult-layer-end.XXXXXX"); \
+	python3 -B research/layer-end-controls-parallel/verify.py --work "$$work/replay" --serial

research/layer-end-controls-parallel/BENCHMARK-macos-arm64.json

@@ -0,0 +1,12 @@
+{
+  "platform": "macOS-26.5.2-arm64-arm-64bit-Mach-O",
+  "python": "3.14.8",
+  "cpu_count": 10,
+  "jobs": 10,
+  "parallel_seconds": 19.9,
+  "serial_seconds": 52.5,
+  "speedup": 2.64,
+  "controls": [
+    "C1_corrupted_unmutated_word_fails_in_replay"
+  ]
+}

research/layer-end-controls-parallel/CLAIM.json

@@ -0,0 +1,22 @@
+{
+ "ai_assistance": "substantial: Claude (Anthropic) via Claude Code",
+ "assignment_seed": 85,
+ "base_commit": "2921874911956718fcaaaf9396fe8267b3cfa9f0",
+ "claims": [
+  "controls_parallel.py produces COLD-END-CONTROLS.json byte-identical to the pinned layer_end_controls.py for the same arguments (sha256 99a17435...), with identical printed control lines and the archived 95 rejected controls.",
+  "Measured on one machine (macOS arm64, 10 cores): serial 49.2 s, parallel 10 jobs 16.4 s (3.0x); harness run 52.5 s vs 19.9 s.",
+  "The pinned script is neither edited nor re-implemented: real outcomes from the pinned audit on byte-identical words are replayed through its own assertions; a corrupted unmutated word fails at the pinned assertion."
+ ],
+ "contributor": "Sagar Thakrar (thakrarsagar)",
+ "expected_last_line": "PASS_LAYER_END_CONTROLS_PARALLEL",
+ "license": "Apache-2.0",
+ "not_claimed": [
+  "any change to the selected kappa or to any mathematical obligation",
+  "wiring into reproduce_barch.py or scripts/verify_frontier942.py",
+  "speedups on other machines beyond the recorded measurement"
+ ],
+ "reproduce": "make layer-end-controls-parallel-verify",
+ "seed_source": "Python secrets (OS CSPRNG)",
+ "status": "tooling improvement (verification engineering); no new kappa",
+ "track": "85-94 verification engineering"
+}

research/layer-end-controls-parallel/PINS.json

@@ -0,0 +1,27 @@
+{
+  "base_commit": "2921874911956718fcaaaf9396fe8267b3cfa9f0",
+  "t37_archive": {
+    "path": "research/community-frontier942-audit/t37-replay.tar.gz",
+    "sha256": "48285025105210b56a32d3d4d57a21b7591760acc709d38073e4dca339eb06fe",
+    "word_member": "native/barch/FULL-WORD.json.gz",
+    "receipt_member": "native/controls/COLD-END-CONTROLS.json"
+  },
+  "recipe": {
+    "path": "research/complex-t21-owners/vendor/pr870/construction/barch/RECIPE-U8.json",
+    "sha256": "292f0e9f82824c4111728cf4de929bf6e43320fbbdc77658aae9681906b0d652",
+    "binding": "expect_sha256 is the payload digest of the U8 barch word"
+  },
+  "routing": {
+    "path": "research/complex-k492-owner-repair/vendor/multicold/vendor/pr558/vendor/pr552/complex-local/inputs/phase-baseline/CANONICAL-WORD.json.gz",
+    "sha256": "e070ae2c78b4a1b644b418774af62b40d8d16e6ba89602f0474afb557d507c37"
+  },
+  "script": {
+    "path": "research/complex-t21-owners/vendor/pr870/construction/layers_p6n/layer_end_controls.py",
+    "sha256": "fc58c5699ba6191c1163a3041241f6f0b619626c4462cd99bcf88210bac9e81e"
+  },
+  "pr635_code": {
+    "path": "research/complex-sw4-frame8/vendor/pr635/research/complex-571-cold14-st2/code",
+    "audit_sha256": "c9a05ff291016b870b4fc925b317abd9c3a604c73ffc8920dfdbbeb475e20aa3"
+  },
+  "layer_controls": 95
+}

research/layer-end-controls-parallel/README.md

@@ -0,0 +1,88 @@
+# Parallel U8 layer-end controls (opt-in, byte-identical receipt)
+
+**Status: tooling improvement (verification engineering). No new κ, and no check is changed or
+weakened.**
+Track 85–94 (verification engineering), assignment seed 85 (OS CSPRNG via Python `secrets`).
+Base commit `2921874911956718fcaaaf9396fe8267b3cfa9f0`.
+
+## What it speeds up
+
+The selected T37 replay's `u8-native` stage spends 77.0 s in `layer-end-controls`, the second
+most expensive step after the phase lowering. That step runs PR870's
+`layers_p6n/layer_end_controls.py`, which audits the unmutated U8 word and 95 mutated words
+with PR635's `hpe_audit_and_transport.run`. Every mutation must be rejected for a pinned reason.
+The 96 audits are independent pure computations.
+
+`controls_parallel.py` takes the same arguments and parallelizes them **without editing or
+re-implementing the pinned script**:
+
+1. **Record.** The pinned script runs once with a stand-in `hpe_audit_and_transport` that only
+   records the exact gzip bytes of each trial word. As each word is recorded, the real audit of
+   it starts in a worker process.
+2. **Evaluate.** Every distinct word is audited by the real pinned `run` (digest-checked) in a
+   worker, giving exactly the outcome `trial()` would observe.
+3. **Replay.** The unchanged pinned script runs again. Its stand-in returns or raises the
+   precomputed real outcome for the identical bytes. The script's own assertions accept or reject
+   every control, check every pinned reason and write `COLD-END-CONTROLS.json`. The sequence of
+   trial words must equal the recorded one.
+
+In this design the record pass's placeholder answers never decide anything. Only real outcomes,
+evaluated by the pinned audit on the byte-identical words, reach the script's assertions.
+
+## Benchmark (same pinned inputs, same machine)
+
+| Runner | Wall time | Speedup |
+|---|---:|---:|
+| pinned `layer_end_controls.py` | 49.2 s (standalone), 52.5 s (harness) | 1× |
+| `controls_parallel.py --jobs 10` | 16.4 s (standalone), 19.9 s (harness) | **3.0×** (2.6× under harness load) |
+| `controls_parallel.py --jobs 4` | 23.3 s (standalone) | 2.1× |
+
+Machine: macOS 26.5.2 arm64, 10 cores, Python 3.14.8. The receipt is in
+[`BENCHMARK-macos-arm64.json`](BENCHMARK-macos-arm64.json). The remaining serial cost is the
+pinned script's own mutant construction (deep copy, JSON and gzip of each word), which runs once
+per pass by design. On the recorded replay machine the serial step took 77.0 s.
+
+## Evidence of equivalence
+
+- `COLD-END-CONTROLS.json` is **byte-identical** to the pinned script's receipt for the same
+  arguments, at digest `99a17435…` in every 4-job and 10-job run.
+- The printed control lines are identical.
+- The 95 rejected controls equal the archived receipt of the selected T37 replay, in order.
+- **Control C1:** a corrupted unmutated word, with one gate coefficient doubled, makes the parallel
+  runner fail. The failure comes from the pinned script's own assertion, `unmutated cold-end word
+  must pass`, in the replay pass, and no receipt is written.
+
+## Using it in the selected replay (opt-in, not wired)
+
+`reproduce_barch.py` runs the pinned script as one step. An opt-in replay can invoke
+`controls_parallel.py` with the same three arguments and the same `PR635_CODE` environment, then
+check the receipt as `reproduce_barch.py` does. No pinned file is modified, and wiring is left to
+the maintainers. This complements the opt-in phase runner in PR1041. Together they would cut the
+`u8-native` stage from about 254 s to about a minute on the recorded machine, assuming similar
+core counts.
+
+## Reproduce
+
+```sh
+make layer-end-controls-parallel-verify      # parallel run, archived-receipt match, control C1 (~40 s)
+make layer-end-controls-parallel-benchmark   # also runs the pinned serial reference (~80 s)
+```
+
+The expected last line is `PASS_LAYER_END_CONTROLS_PARALLEL`. It uses standard-library Python
+3.10+ and no network. The barch word is read from the pinned T37 archive, and its payload is
+checked against PR870's `RECIPE-U8.json`.
+
+## Scope
+
+This accelerates one step. It discharges no mathematical obligation, and the timings are local
+measurements.
+
+## Credit
+
+- DreamingOfClouds: `layer_end_controls.py`, adapted from PR635's cold-end controls.
+- PR635's authors: `hpe_audit_and_transport.py`.
+- Jeremy Kothe and NewBArch: PR869/PR870, the barch word and pins.
+- Jonathan Capalbo: the PR942 replay.
+- Prior concurrency tooling: PR853 and PR921 (navinor), and PR1041.
+- Prepared by Sagar Thakrar (thakrarsagar) with substantial AI assistance from Claude
+  (Anthropic) via Claude Code. New material is Apache-2.0.

research/layer-end-controls-parallel/audit_worker.py

@@ -0,0 +1,50 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Worker side of controls_parallel.py: the real, pinned PR635 audit applied to exact trial bytes.
+
+Kept in its own module so that worker functions stay importable (and picklable) while
+the pinned script temporarily occupies __main__ during the record and replay passes.
+"""
+import hashlib
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+
+AUDIT_SHA256 = 'c9a05ff291016b870b4fc925b317abd9c3a604c73ffc8920dfdbbeb475e20aa3'
+AUDIT = None
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def init_worker(code_dir):
+    global AUDIT
+    path = Path(code_dir) / 'hpe_audit_and_transport.py'
+    if sha(path.read_bytes()) != AUDIT_SHA256:
+        raise RuntimeError('pinned hpe_audit_and_transport.py digest mismatch')
+    if str(code_dir) not in sys.path:
+        sys.path.insert(0, str(code_dir))
+    import importlib.util
+    spec = importlib.util.spec_from_file_location('pinned_hpe_audit_and_transport', path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    AUDIT = mod
+
+
+def evaluate(task):
+    """The outcome trial() of the pinned script would observe for these exact bytes."""
+    data, routing = task
+    tmp = Path(tempfile.mkdtemp())
+    try:
+        p = tmp / 'w.json.gz'
+        p.write_bytes(data)
+        try:
+            AUDIT.run(p, Path(routing), tmp / 'out')
+            return ('pass', ())
+        except AssertionError as e:
+            return ('assert', e.args)
+        except Exception as e:      # the pinned trial() does not catch these; the replay re-raises
+            return ('error', (repr(e),))
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)

research/layer-end-controls-parallel/controls_parallel.py

@@ -0,0 +1,153 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Opt-in parallel runner for the pinned U8 layer-end controls, with identical receipts.
+
+Usage: controls_parallel.py LAYERED-WORD.json.gz ROUTING OUTDIR [ENTERING] [--jobs N]
+       (environment PR635_CODE exactly as for the pinned script)
+
+The pinned script research/complex-t21-owners/vendor/pr870/construction/layers_p6n/
+layer_end_controls.py builds 95 mutated words and audits each one, plus the unmutated
+word, with PR635's hpe_audit_and_transport.run. Every trial is independent. The pinned
+script is never edited and never re-implemented:
+
+1. record:   the pinned script runs with a stand-in hpe_audit_and_transport whose run()
+             records the exact gzip bytes of each trial word. It answers the unmutated
+             word with success and every other word with a placeholder rejection that
+             satisfies any reason check. No receipt is written in this pass.
+2. evaluate: every distinct recorded word is audited by the real, pinned
+             hpe_audit_and_transport.run in a worker process, giving exactly the outcome
+             trial() would see (the AssertionError arguments, or success).
+3. replay:   the pinned script runs again with a stand-in that returns or raises the
+             precomputed real outcome for the identical bytes. Its own assertions accept
+             or reject every control and write COLD-END-CONTROLS.json. The sequence of
+             trial words must equal the recorded one.
+"""
+import argparse
+import hashlib
+import multiprocessing as mp
+import os
+import runpy
+import shutil
+import sys
+import tempfile
+import time
+import types
+from pathlib import Path
+
+from audit_worker import evaluate, init_worker
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / 'research/complex-t21-owners/vendor/pr870/construction/layers_p6n/layer_end_controls.py'
+SCRIPT_SHA256 = 'fc58c5699ba6191c1163a3041241f6f0b619626c4462cd99bcf88210bac9e81e'
+
+
+class AnyReason(str):
+    """Placeholder message for the record pass: every reason substring is 'in' it."""
+
+    def __contains__(self, item):
+        return True
+
+
+class Placeholder:
+    def __repr__(self):
+        return AnyReason('placeholder rejection (record pass only)')
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def run_pinned(argv, stub_run):
+    stub = types.ModuleType('hpe_audit_and_transport')
+    stub.run = stub_run
+    saved_argv, saved_mod = sys.argv, sys.modules.get('hpe_audit_and_transport')
+    sys.modules['hpe_audit_and_transport'] = stub
+    sys.argv = [str(SCRIPT), *argv]
+    try:
+        runpy.run_path(str(SCRIPT), run_name='__main__')
+    finally:
+        sys.argv = saved_argv
+        if saved_mod is None:
+            sys.modules.pop('hpe_audit_and_transport', None)
+        else:
+            sys.modules['hpe_audit_and_transport'] = saved_mod
+
+
+def main():
+    if not __debug__:
+        raise RuntimeError('assertions must remain enabled')
+    ap = argparse.ArgumentParser()
+    ap.add_argument('word')
+    ap.add_argument('routing')
+    ap.add_argument('outdir')
+    ap.add_argument('entering', nargs='?')
+    ap.add_argument('--jobs', type=int, default=os.cpu_count() or 1)
+    a = ap.parse_args()
+    if sha(SCRIPT.read_bytes()) != SCRIPT_SHA256:
+        raise RuntimeError('pinned layer_end_controls.py digest mismatch')
+    code_dir = os.environ['PR635_CODE']
+    argv = [a.word, a.routing, a.outdir] + ([a.entering] if a.entering else [])
+    start = time.monotonic()
+
+    # 1. record, while 2. the real audits already run in the pool as each word is recorded
+    recorded = []
+    pending = {}
+    routing_abs = str(Path(a.routing).resolve())
+    pool = None if a.jobs == 1 else mp.get_context('spawn').Pool(a.jobs, initializer=init_worker, initargs=(code_dir,))
+    if pool is None:
+        init_worker(code_dir)
+
+    def record(p, routing, out):
+        data = Path(p).read_bytes()
+        key = sha(data)
+        recorded.append(key)
+        if key not in pending:
+            pending[key] = (pool.apply_async(evaluate, ((data, routing_abs),)) if pool is not None
+                            else evaluate((data, routing_abs)))
+        if len(recorded) == 1:
+            return None                      # the pinned script's first trial is the unmutated word
+        raise AssertionError(Placeholder())
+    sink = tempfile.mkdtemp()
+    try:
+        devnull = open(os.devnull, 'w')
+        saved = sys.stdout
+        sys.stdout = devnull
+        try:
+            run_pinned([a.word, a.routing, sink] + argv[3:], record)
+        finally:
+            sys.stdout = saved
+            devnull.close()
+        print('recorded', len(recorded), 'trial words', len(pending), 'distinct', 'seconds',
+              round(time.monotonic() - start, 1), flush=True)
+        table = {k: (v.get() if pool is not None else v) for k, v in pending.items()}
+    finally:
+        shutil.rmtree(sink, ignore_errors=True)
+        if pool is not None:
+            pool.close()
+            pool.join()
+    print('evaluated', len(table), 'words with', a.jobs, 'jobs', 'seconds', round(time.monotonic() - start, 1), flush=True)
+
+    # 3. replay
+    seen = []
+
+    def replay(p, routing, out):
+        data = Path(p).read_bytes()
+        key = sha(data)
+        seen.append(key)
+        kind, args = table[key] if key in table else evaluate_here(data, routing)
+        if kind == 'pass':
+            return None
+        if kind == 'assert':
+            raise AssertionError(*args)
+        raise RuntimeError('audit raised a non-assertion error: ' + args[0])
+
+    def evaluate_here(data, routing):
+        init_worker(code_dir)
+        return evaluate((data, str(routing)))
+    run_pinned(argv, replay)
+    if seen != recorded:
+        raise RuntimeError('the replay produced a different sequence of trial words')
+    print('replayed', len(seen), 'trials; total seconds', round(time.monotonic() - start, 1), flush=True)
+
+
+if __name__ == '__main__':
+    main()