From 0e301503b29d802327238c32f5ae3377797a564c Mon Sep 17 00:00:00 2001
From: Sagar Thakrar <sagar@cybai.in>
Date: Tue, 13 Oct 2026 22:53:47 +0530
Subject: [PATCH] Tooling: opt-in parallel U8 layer-end controls with
 byte-identical receipt, 3.0x on 10 cores (no new kappa)

---
 Makefile                                      |  10 ++
 .../BENCHMARK-macos-arm64.json                |  12 ++
 .../layer-end-controls-parallel/CLAIM.json    |  22 +++
 .../layer-end-controls-parallel/PINS.json     |  27 ++++
 .../layer-end-controls-parallel/README.md     |  88 ++++++++++
 .../audit_worker.py                           |  50 ++++++
 .../controls_parallel.py                      | 153 ++++++++++++++++++
 .../test_package.py                           |  51 ++++++
 .../layer-end-controls-parallel/verify.py     | 124 ++++++++++++++
 9 files changed, 537 insertions(+)
 create mode 100644 research/layer-end-controls-parallel/BENCHMARK-macos-arm64.json
 create mode 100644 research/layer-end-controls-parallel/CLAIM.json
 create mode 100644 research/layer-end-controls-parallel/PINS.json
 create mode 100644 research/layer-end-controls-parallel/README.md
 create mode 100644 research/layer-end-controls-parallel/audit_worker.py
 create mode 100644 research/layer-end-controls-parallel/controls_parallel.py
 create mode 100644 research/layer-end-controls-parallel/test_package.py
 create mode 100644 research/layer-end-controls-parallel/verify.py

diff --git a/Makefile b/Makefile
index 66f00cfb30..2741ae0af9 100644
--- a/Makefile
+++ b/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
diff --git a/research/layer-end-controls-parallel/BENCHMARK-macos-arm64.json b/research/layer-end-controls-parallel/BENCHMARK-macos-arm64.json
new file mode 100644
index 0000000000..1d5f63f420
--- /dev/null
+++ b/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"
+  ]
+}
diff --git a/research/layer-end-controls-parallel/CLAIM.json b/research/layer-end-controls-parallel/CLAIM.json
new file mode 100644
index 0000000000..2ec513bfe0
--- /dev/null
+++ b/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"
+}
diff --git a/research/layer-end-controls-parallel/PINS.json b/research/layer-end-controls-parallel/PINS.json
new file mode 100644
index 0000000000..512f8d4797
--- /dev/null
+++ b/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
+}
diff --git a/research/layer-end-controls-parallel/README.md b/research/layer-end-controls-parallel/README.md
new file mode 100644
index 0000000000..ddd2fccd14
--- /dev/null
+++ b/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.
diff --git a/research/layer-end-controls-parallel/audit_worker.py b/research/layer-end-controls-parallel/audit_worker.py
new file mode 100644
index 0000000000..5f5a3d4592
--- /dev/null
+++ b/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)
diff --git a/research/layer-end-controls-parallel/controls_parallel.py b/research/layer-end-controls-parallel/controls_parallel.py
new file mode 100644
index 0000000000..c089c05ad4
--- /dev/null
+++ b/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()
diff --git a/research/layer-end-controls-parallel/test_package.py b/research/layer-end-controls-parallel/test_package.py
new file mode 100644
index 0000000000..085480f2e8
--- /dev/null
+++ b/research/layer-end-controls-parallel/test_package.py
@@ -0,0 +1,51 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Fast checks of the record/evaluate/replay runner.
+
+Run: python3 -B -m unittest discover -s research/layer-end-controls-parallel -p test_package.py -v
+"""
+import gzip
+import json
+import sys
+import tarfile
+import unittest
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+sys.path.insert(0, str(HERE))
+
+import audit_worker  # noqa: E402
+import controls_parallel as cp  # noqa: E402
+
+PINS = json.loads((HERE / 'PINS.json').read_text())
+
+
+def barch_word():
+    with tarfile.open(ROOT / PINS['t37_archive']['path'], 'r:gz') as tar:
+        return tar.extractfile(tar.getmember(PINS['t37_archive']['word_member'])).read()
+
+
+class RunnerTests(unittest.TestCase):
+    def test_placeholder_satisfies_any_reason_check(self):
+        msg = repr(cp.Placeholder())
+        self.assertIsInstance(msg, str)
+        for why in ("'arbitrary-dirt endpoint'", 'hpe cone pre-check (row / column)', 'anything at all'):
+            self.assertIn(why, msg)
+
+    def test_pins_match_runner_constants(self):
+        self.assertEqual(cp.SCRIPT_SHA256, PINS['script']['sha256'])
+        self.assertEqual(audit_worker.AUDIT_SHA256, PINS['pr635_code']['audit_sha256'])
+
+    def test_real_audit_accepts_word_and_rejects_corruption(self):
+        audit_worker.init_worker(str(ROOT / PINS['pr635_code']['path']))
+        routing = str(ROOT / PINS['routing']['path'])
+        data = barch_word()
+        self.assertEqual(audit_worker.evaluate((data, routing)), ('pass', ()))
+        w = json.loads(gzip.decompress(data))
+        w['A'][0][3] = w['A'][0][3] * 2
+        kind, args = audit_worker.evaluate((gzip.compress(json.dumps(w).encode(), mtime=0), routing))
+        self.assertEqual(kind, 'assert')
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/research/layer-end-controls-parallel/verify.py b/research/layer-end-controls-parallel/verify.py
new file mode 100644
index 0000000000..7a61f7f977
--- /dev/null
+++ b/research/layer-end-controls-parallel/verify.py
@@ -0,0 +1,124 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Check the opt-in parallel layer-end controls against the pinned serial reference.
+
+Usage: python3 -B research/layer-end-controls-parallel/verify.py --work NEW_EMPTY_DIR [--jobs N] [--serial]
+
+1. Extracts the pinned U8 barch word from the selected T37 archive (payload = PR870 RECIPE-U8 expect_sha256).
+2. Runs controls_parallel.py: its COLD-END-CONTROLS.json must report PASS with the archived receipt's
+   95 rejected controls, in order.
+3. With --serial, also runs the pinned layer_end_controls.py with the same arguments and requires a
+   byte-identical receipt and identical printed control lines (the timing receipt).
+4. Control: a corrupted unmutated word must make the parallel runner fail, as the serial script does,
+   because the pinned script's own assertion 'unmutated cold-end word must pass' fires in the replay pass.
+Expected last line: PASS_LAYER_END_CONTROLS_PARALLEL
+"""
+import argparse
+import gzip
+import hashlib
+import json
+import os
+import platform
+import subprocess
+import sys
+import tarfile
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+PINS = json.loads((HERE / 'PINS.json').read_text())
+
+
+def require(ok, message):
+    if not ok:
+        raise SystemExit('FAIL ' + message)
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+def pinned(key):
+    path = ROOT / PINS[key]['path']
+    require(sha(path.read_bytes()) == PINS[key]['sha256'], key + ' digest')
+    return path
+
+
+def member(name):
+    with tarfile.open(pinned('t37_archive'), 'r:gz') as tar:
+        return tar.extractfile(tar.getmember(name)).read()
+
+
+def run(cmd, log):
+    code = ROOT / PINS['pr635_code']['path']
+    require(sha((code / 'hpe_audit_and_transport.py').read_bytes()) == PINS['pr635_code']['audit_sha256'], 'audit digest')
+    src = ROOT / 'research/complex-k492-owner-repair/vendor/multicold'
+    env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1', PYTHONUTF8='1', PYTHONINTMAXSTRDIGITS='0',
+               MULTICOLD_SOURCE=str(src), GAIN_PR564_SOURCE=str(src), PR635_CODE=str(code))
+    t = time.monotonic()
+    with open(log, 'w') as f:
+        rc = subprocess.call(cmd, cwd=ROOT, env=env, stdout=f, stderr=subprocess.STDOUT)
+    return rc, time.monotonic() - t
+
+
+def control_lines(log):
+    return [x for x in Path(log).read_text().splitlines() if x.startswith(('rejected', 'PASS ', 'entering end'))]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument('--work', required=True)
+    ap.add_argument('--jobs', type=int, default=os.cpu_count() or 1)
+    ap.add_argument('--serial', action='store_true')
+    a = ap.parse_args()
+    work = Path(a.work).resolve()
+    require(not work.exists(), 'work directory must not exist')
+    work.mkdir(parents=True)
+    pinned('script')
+    recipe = json.loads(pinned('recipe').read_text())
+    data = member(PINS['t37_archive']['word_member'])
+    require(sha(gzip.decompress(data)) == recipe['expect_sha256'], 'barch word payload')
+    word = work / 'barch-word.json.gz'
+    word.write_bytes(data)
+    routing = pinned('routing')
+    archived = json.loads(member(PINS['t37_archive']['receipt_member']))
+    receipt = dict(platform=platform.platform(), python=platform.python_version(), cpu_count=os.cpu_count(), jobs=a.jobs)
+
+    out = work / 'parallel'
+    rc, secs = run([sys.executable, '-B', str(HERE / 'controls_parallel.py'), str(word), str(routing), str(out),
+                    '--jobs', str(a.jobs)], work / 'parallel.log')
+    require(rc == 0, 'parallel controls failed')
+    got = json.loads((out / 'COLD-END-CONTROLS.json').read_text())
+    require(got['status'] == archived['status'] == 'PASS_COLD_END_RAW_WORD_CONTROLS', 'status')
+    require(got['rejected'] == archived['rejected'] and len(got['rejected']) == PINS['layer_controls'],
+            'rejected controls differ from the archived receipt')
+    receipt['parallel_seconds'] = round(secs, 1)
+    if a.serial:
+        ser = work / 'serial'
+        rc, secs = run([sys.executable, '-B', str(ROOT / PINS['script']['path']), str(word), str(routing), str(ser)],
+                       work / 'serial.log')
+        require(rc == 0, 'serial reference failed')
+        require((ser / 'COLD-END-CONTROLS.json').read_bytes() == (out / 'COLD-END-CONTROLS.json').read_bytes(),
+                'serial and parallel receipts are not byte-identical')
+        require(control_lines(work / 'serial.log') == control_lines(work / 'parallel.log'), 'printed control lines differ')
+        receipt['serial_seconds'] = round(secs, 1)
+        receipt['speedup'] = round(receipt['serial_seconds'] / receipt['parallel_seconds'], 2)
+
+    # Control: a corrupted unmutated word fails in both runners.
+    w = json.loads(gzip.decompress(data))
+    w['A'][0][3] = w['A'][0][3] * 2                       # one gate coefficient doubled
+    bad = work / 'corrupted-word.json.gz'
+    bad.write_bytes(gzip.compress(json.dumps(w).encode(), mtime=0))
+    rc, _ = run([sys.executable, '-B', str(HERE / 'controls_parallel.py'), str(bad), str(routing), str(work / 'C1'),
+                 '--jobs', str(a.jobs)], work / 'C1.log')
+    log = (work / 'C1.log').read_text()
+    require(rc != 0 and 'unmutated cold-end word must pass' in log and not (work / 'C1' / 'COLD-END-CONTROLS.json').exists(),
+            'corrupted word not rejected by the parallel runner')
+    receipt['controls'] = ['C1_corrupted_unmutated_word_fails_in_replay']
+    (work / 'RECEIPT.json').write_text(json.dumps(receipt, indent=2) + '\n')
+    print(json.dumps(receipt, indent=2))
+    print('PASS_LAYER_END_CONTROLS_PARALLEL')
+
+
+if __name__ == '__main__':
+    main()