diff --git a/scripts/check_ladder_apr_release_pin.sh b/scripts/check_ladder_apr_release_pin.sh
new file mode 100755
index 0000000000..ee643d7397
--- /dev/null
+++ b/scripts/check_ladder_apr_release_pin.sh
@@ -0,0 +1,101 @@
+#!/usr/bin/env bash
+# check_ladder_apr_release_pin.sh — the release cells run the apr UNDER TEST, never a PATH lookup (#3715 A3).
+#
+# WHY. Operator 2026-09-28 17:28Z: "cells and every gate step invoke the binary UNDER TEST by absolute path
+# (the rc.1 artifact) with a sha256 check; each receipt records `apr --version` SHA == rc.1 tag SHA.
+# Falsifier: plant a different-sha apr first on PATH -> cell refuses. PATH lookup of apr on the gate path = 0."
+# Four `apr` binaries once coexisted on one box and a bare `apr` resolved to a 26-day-old one. A cell measured
+# on the wrong binary is a confident verdict about code nobody is shipping.
+#
+# WHAT. model_ladder.sh step 0, release mode (APR_RELEASE_SHA256 / APR_RELEASE_COMMIT set), driven through
+# `--lock-probe` -- the same apr_locked every cell goes through -- in a temp root whose PATH has a PLANTED apr
+# first. Every fake apr logs each invocation, so "the planted apr ran zero times" is observed, not assumed.
+#
+# Exit: 0 every case as expected and every mutant killed · 1 a case or mutant landed wrong · 2 could not check.
+set -uo pipefail
+cd "$(dirname "$0")/.." || exit 2
+LADDER_SH=scripts/model_ladder.sh
+for t in sha256sum flock choom; do command -v "$t" > /dev/null || { echo "check_ladder_apr_release_pin: $t absent" >&2; exit 2; }; done
+[ -f "$LADDER_SH" ] || { echo "check_ladder_apr_release_pin: $LADDER_SH absent" >&2; exit 2; }
+
+W=$(mktemp -d "${TMPDIR:-/tmp}/apr-pin.XXXXXX") || exit 2
+trap 'rm -rf -- "${W:?}"' EXIT
+GOOD=abcdef1234567890abcdef1234567890abcdef12   # the "release commit"
+OTHER=0123456789abcdef0123456789abcdef01234567
+mkdir -p "$W/root" "$W/rc" "$W/path" "$W/imp"
+for d in contracts scripts evidence Cargo.toml; do ln -s "$PWD/$d" "$W/root/$d"; done
+LOG="$W/calls"
+fake() { # <file> <label> <commit-it-claims>
+  printf '#!/bin/sh\necho "%s $*" >> "%s"\n[ "$1" = --version ] && { echo "apr 0.70.0 (%s)"; exit 0; }\nexit 0\n' "$2" "$LOG" "${3:0:9}" > "$1"
+  chmod +x "$1"
+}
+fake "$W/rc/apr" rc "$GOOD"          # the artifact under test
+fake "$W/path/apr" planted "$OTHER"  # a different-sha apr, first on PATH
+fake "$W/imp/apr" impostor "$GOOD"   # different bytes, but CLAIMS the release commit
+# A bare name is checked in one place and executed from another: the ladder's cwd holds the real bytes (so
+# -f and sha256 pass on them), while exec resolves the same name through PATH -- to an impostor claiming the
+# release commit. Only the ABSOLUTE-path rule stops this one: bytes checked != bytes run.
+cp "$W/rc/apr" "$W/root/rc-apr"; chmod +x "$W/root/rc-apr"
+fake "$W/path/rc-apr" impostor "$GOOD"
+sum() { local s; s=$(sha256sum -- "$1") && printf '%s' "${s%% *}"; }
+RC_SUM=$(sum "$W/rc/apr"); PL_SUM=$(sum "$W/path/apr")
+
+# probe <script> <APR> <sha|-> <commit|-> -> sets RC, CALLS (non---version invocations by label)
+probe() {
+  : > "$LOG"
+  local env=(PATH="$W/path:$PATH" MODEL_LADDER_ROOT="$W/root" MODEL_LADDER_GPU_LOCK="$W/lock" MODEL_LADDER_LOCK_WAIT=5 APR="$2")
+  [ "$3" != - ] && env+=(APR_RELEASE_SHA256="$3")
+  [ "$4" != - ] && env+=(APR_RELEASE_COMMIT="$4")
+  env "${env[@]}" timeout 60 bash "$1" --lock-probe qa cell > "$W/out" 2>&1; RC=$?
+  CALLS=$(grep -v -- ' --version' "$LOG" | cut -d' ' -f1 | sort | uniq -c | tr -s ' ' | tr '\n' ';')
+}
+
+# name | APR | sha | commit | want rc | want the ONLY cell invocation (label or "none")
+CASES="green-planted-on-path|$W/rc/apr|$RC_SUM|$GOOD|0|rc
+bare-apr-is-a-path-lookup|apr|$RC_SUM|$GOOD|2|none
+bare-name-checked-in-cwd-run-from-path|rc-apr|$RC_SUM|$GOOD|2|none
+absolute-planted|$W/path/apr|$RC_SUM|$GOOD|2|none
+impostor-claims-the-commit|$W/imp/apr|$RC_SUM|$GOOD|2|none
+right-bytes-wrong-commit|$W/path/apr|$PL_SUM|$GOOD|2|none
+half-pin-sha-only|$W/rc/apr|$RC_SUM|-|2|none
+half-pin-commit-only|$W/rc/apr|-|$GOOD|2|none
+sha-not-hex|$W/rc/apr|nothex|$GOOD|2|none"
+
+run_cases() { # <script> [quiet] -> number of cases that landed wrong
+  local s="$1" q="${2:-}" bad=0 name apr sha commit wrc wcall want
+  while IFS='|' read -r name apr sha commit wrc wcall; do
+    probe "$s" "$apr" "$sha" "$commit"
+    want=$([ "$wcall" = none ] && echo "" || echo " 1 $wcall;")
+    if [ "$RC" = "$wrc" ] && [ "$CALLS" = "$want" ]; then [ -z "$q" ] && echo "ok    $name (rc $RC, cell ran: ${wcall})"
+    else bad=$((bad + 1)); [ -z "$q" ] && { echo "FAIL  $name: rc $RC (want $wrc), cell invocations '${CALLS}' (want '${want}')"; sed 's/^/        /' "$W/out" | tail -n 3; }
+    fi
+  done <<< "$CASES"
+  return "$bad"
+}
+
+echo "== cases (planted apr first on PATH in every case)"
+run_cases "$LADDER_SH"; bad=$?
+rows=$(printf '%s\n' "$CASES" | wc -l)
+
+echo "== mutants (each must turn a case RED)"
+mutant() { # <name> <python-literal old> <new>
+  local m="$W/mut-$1.sh"
+  python3 - "$LADDER_SH" "$m" "$2" "$3" <<'PY' || { echo "FAIL  mutant $1: anchor not found once -- the check no longer tracks $LADDER_SH"; bad=$((bad + 1)); return; }
+import sys
+src, dst, old, new = sys.argv[1:]
+s = open(src).read()
+sys.exit(1) if s.count(old) != 1 else open(dst, "w").write(s.replace(old, new, 1))
+PY
+  if run_cases "$m" quiet; then echo "FAIL  mutant $1 SURVIVED"; bad=$((bad + 1)); else echo "ok    mutant $1 killed"; fi
+}
+mutant no-release-pin '  release_pin || exit 2' '  true || exit 2'
+mutant no-absolute-check '    /*) ;;
+    *) echo "decline: release mode needs' '    *) ;;
+    /*) echo "decline: release mode needs'
+mutant no-sha256-check '[ "$got" = "$want" ] ||' 'true ||'
+mutant no-version-check '*"(${commit:0:9})"*) ;;' '*) ;;'
+mutant path-lookup-in-cell 'choom -n 1000 -- "$APR" "$@"; }' 'choom -n 1000 -- apr "$@"; }'
+
+[ "$rows" -ge 9 ] || { echo "VACUOUS $rows case(s), fewer than the 9 declared" >&2; exit 1; }
+[ "$bad" = 0 ] && { echo "PASS $rows cases + 5 mutants: release cells run only the pinned artifact; PATH lookups of apr = 0"; exit 0; }
+echo "RED $bad"; exit 1
diff --git a/scripts/check_ladder_cell_timeout_takt.sh b/scripts/check_ladder_cell_timeout_takt.sh
new file mode 100755
index 0000000000..a77ca93aa1
--- /dev/null
+++ b/scripts/check_ladder_cell_timeout_takt.sh
@@ -0,0 +1,119 @@
+#!/usr/bin/env bash
+# check_ladder_cell_timeout_takt.sh — H1 (hard per-cell timeout = FAIL) and H5 (takt tripwires) in the cells producer.
+#
+# WHY. Operator 2026-09-28 17:50Z: "H1 hard per-cell timeout. Timeout = FAIL, blocks FINAL, never a skip. Default
+# 10 min; after sampling, per class = 4 x measured p95 (floor 5 min). The timeout value is recorded in every
+# receipt." and "H5 Takt tripwires after rc.1: at +2 h, +4 h, +6 h, completed cells >= 90% of plan, or ANDON
+# immediately." A hung 4B cell held lambda's GPU for hours, and the old subprocess timeout killed only `flock`:
+# the apr grandchild kept the card after the cell was written down. So the case that matters is not "the timer
+# fired" but "the process GROUP is gone and the row says FAIL with the timeout it ran under".
+#
+# Runs the SHIPPED scripts/lib/model_ladder_cells_produce.py (imported), then each mutant must turn a case RED.
+# Exit: 0 all cases + mutants as expected · 1 a case or mutant landed wrong · 2 could not check.
+set -uo pipefail
+cd "$(dirname "$0")/.." || exit 2
+LIB=scripts/lib
+[ -f "$LIB/model_ladder_cells_produce.py" ] || { echo "check_ladder_cell_timeout_takt: producer absent" >&2; exit 2; }
+W=$(mktemp -d "${TMPDIR:-/tmp}/cell-tmo.XXXXXX") || exit 2
+trap 'rm -rf -- "${W:?}"' EXIT
+
+cases() { # <libdir> [quiet] -> 0 when every case lands as expected
+  python3 - "$1" "$W" "${2:-}" <<'PY'
+import importlib, json, os, sys, time
+lib, w, quiet = sys.argv[1], sys.argv[2], sys.argv[3]
+sys.path.insert(0, lib)
+P = importlib.import_module("model_ladder_cells_produce")
+bad = 0
+def case(name, ok, why):
+    global bad
+    if not ok:
+        bad += 1
+    if not quiet or not ok:
+        print(("ok    " if ok else "FAIL  ") + name + ("" if ok else ": " + why))
+
+# H1-a: a hung apr whose grandchild holds on -- the whole group must be gone, rc 124, timed_out set.
+apr = os.path.join(w, "hang-apr")
+pidf = os.path.join(w, "grandchild.pid")
+open(apr, "w").write(f"#!/bin/sh\nsleep 20 &\necho $! > {pidf}\nwait\n")
+os.chmod(apr, 0o755)
+R = P.Runner(apr, "", 5, 2)
+t0 = time.monotonic(); rc, out, err = R.call(["run"]); el = time.monotonic() - t0
+time.sleep(0.3)
+gpid = int(open(pidf).read())
+alive = os.path.exists(f"/proc/{gpid}") and "Z" not in open(f"/proc/{gpid}/stat").read().split()[2]
+case("timeout-kills-the-group", rc == 124 and R.timed_out and not alive and el < 10,
+     f"rc={rc} timed_out={R.timed_out} grandchild_alive={alive} elapsed={el:.1f}s")
+if alive:
+    os.kill(gpid, 9)
+
+# H1-a2: the timeout bounds the CELL, not each apr call: two 1.5 s calls under one 2 s cell deadline -> the 2nd times out.
+slow = os.path.join(w, "slow-apr")
+open(slow, "w").write("#!/bin/sh\nsleep 1.5\n")
+os.chmod(slow, 0o755)
+R = P.Runner(slow, "", 5, 2)
+R.deadline = time.monotonic() + R.timeout
+rc1 = R.call(["run"])[0]; rc2 = R.call(["run"])[0]
+case("timeout-bounds-the-cell-not-the-call", rc1 == 0 and rc2 == 124 and R.timed_out, f"rc1={rc1} rc2={rc2}")
+
+# H1-b: the row a timeout produces is a FAIL naming the timeout, and carries timeout_s.
+row = P.stamp({"verdict": "pass", "reason": "answered"}, time.monotonic(), 600, True)
+case("timeout-row-is-fail", row["verdict"] == "fail" and row["reason"].startswith("TIMEOUT") and row["timeout_s"] == 600
+     and row["timed_out"] is True, json.dumps(row))
+row = P.stamp({"verdict": "pass", "reason": "answered"}, time.monotonic(), 600, False)
+case("no-timeout-row-keeps-verdict", row["verdict"] == "pass" and row["timeout_s"] == 600 and "wall_s" in row, json.dumps(row))
+
+# H1-c: class timeouts floor at 5 min; no class -> the 10-min default.
+class A: pass
+a = A(); a.timeout = P.DEFAULT_CELL_TIMEOUT_S; a.class_timeouts = {"m.gguf|run": 100, "m.gguf|chat": 1200}
+case("class-timeout-floor", P.cell_timeout(a, {"file": "m.gguf"}, "run") == 300, str(P.cell_timeout(a, {"file": "m.gguf"}, "run")))
+case("class-timeout-used", P.cell_timeout(a, {"file": "m.gguf"}, "chat") == 1200, "")
+case("default-timeout-10min", P.cell_timeout(a, {"file": "m.gguf"}, "serve") == 600 and P.DEFAULT_CELL_TIMEOUT_S == 600, "")
+
+# H5: at +2 h 9/10 is met (90%), at +4 h 17/20 is an ANDON, +6 h not yet reached -> not judged.
+plan = os.path.join(w, "plan.json"); log = os.path.join(w, "takt.jsonl")
+json.dump({"start_epoch": 1000, "checkpoints": [{"at_s": 7200, "planned_cells": 10}, {"at_s": 14400, "planned_cells": 20},
+                                                {"at_s": 21600, "planned_cells": 30}]}, open(plan, "w"))
+now = [1000 + 7200]
+T = P.Takt(plan, log, clock=lambda: now[0])
+P.COMPLETED[0] = 9
+r1 = T.tick()
+case("takt-90pct-met", len(r1) == 1 and not r1[0]["andon"] and not os.path.exists(log + ".andon"), json.dumps(r1))
+now[0] = 1000 + 14400 + 5; P.COMPLETED[0] = 17
+r2 = T.tick()
+andon = open(log + ".andon").read() if os.path.exists(log + ".andon") else ""
+case("takt-below-90pct-andon", len(r2) == 1 and r2[0]["andon"] and "ANDON H5" in andon, json.dumps(r2) + " andon=" + andon)
+case("takt-future-not-judged", T.fired == 2 and len(open(log).read().splitlines()) == 2, f"fired={T.fired}")
+try:
+    json.dump({"start_epoch": 1, "checkpoints": []}, open(plan, "w")); P.Takt(plan, log); empty_ok = False
+except ValueError:
+    empty_ok = True
+case("takt-empty-plan-refused", empty_ok, "an empty plan was accepted")
+sys.exit(1 if bad else 0)
+PY
+}
+
+echo "== cases"
+timeout 120 bash -c "$(declare -f cases); W=$W; cases $LIB"; bad=$?
+echo "== mutants (each must turn a case RED)"
+mutant() { # <name> <old> <new>
+  local md="$W/mut-$1"; mkdir -p "$md"; cp "$LIB"/*.py "$md/"
+  python3 - "$md/model_ladder_cells_produce.py" "$2" "$3" <<'PY' || { echo "FAIL  mutant $1 did not apply -- it proves nothing"; bad=1; return; }
+import sys
+p, old, new = sys.argv[1:]
+s = open(p).read()
+sys.exit(1) if s.count(old) != 1 else open(p, "w").write(s.replace(old, new, 1))
+PY
+  if timeout 120 bash -c "$(declare -f cases); W=$W; cases $md quiet" > /dev/null 2>&1; then echo "FAIL  mutant $1 SURVIVED"; bad=1; else echo "ok    mutant $1 killed"; fi
+}
+mutant kill-only-leader 'os.killpg(p.pid, signal.SIGKILL)' 'p.kill()'
+mutant no-new-session 'text=True, start_new_session=True)' 'text=True)'
+mutant timeout-not-fail '        row["verdict"] = "fail"
+' '        pass
+'
+mutant per-call-not-per-cell 'left = self.timeout if self.deadline is None else' 'left = self.timeout if True else'
+mutant no-floor 'return max(int(t), TIMEOUT_FLOOR_S) if' 'return int(t) if'
+mutant takt-always-ok 'ok = done >= self.RATIO * plan' 'ok = True'
+mutant takt-no-andon-file '                with open(self.log + ".andon", "a") as f:
+                    f.write(line + "\n")' '                pass'
+[ "$bad" = 0 ] && { echo "PASS H1 timeout (group kill, FAIL, timeout_s) + H5 takt: all cases and 7 mutants"; exit 0; }
+echo "RED"; exit 1
diff --git a/scripts/check_release_models_t1.sh b/scripts/check_release_models_t1.sh
index 6b19770fad..4694f4f007 100755
--- a/scripts/check_release_models_t1.sh
+++ b/scripts/check_release_models_t1.sh
@@ -82,6 +82,9 @@ A_CHOOM_R='choom -n 1000 -- bash scripts/model_ladder.sh --host $REMOTE_HOST'
 A_CHOOM_L='    choom -n 1000 -- bash scripts/model_ladder.sh --host "$LOCAL_HOST"'
 A_CELLS_R='bash scripts/model_ladder.sh --host $REMOTE_HOST --cells --out'
 A_CELLS_L='bash scripts/model_ladder.sh --host "$LOCAL_HOST" --cells --out'
+# A3 (operator 2026-09-28 17:28Z): both legs hand the ladder the artifact under test by ABSOLUTE path + sha256 + commit.
+A_APR_L='    APR="$apr" APR_RELEASE_SHA256="${sum%% *}" APR_RELEASE_COMMIT="$sha" \'
+A_APR_R='APR="\$apr" APR_RELEASE_SHA256="\${sum%% *}" APR_RELEASE_COMMIT="$sha" \\'
 R_VER='    out="$(cd "$root" && bash "$judge" --version "$version" 2>&1)"; rc=$?'
 R_RED='        *) printf '"'"'FAIL  R7 model matrix NOT green for %s (rc %s):\n%s\n'"'"' "$version" "$rc" \'
 # The FAIL-lines line also appears on the emergency-scope path (#4046 merge-back), so the
@@ -94,7 +97,7 @@ R_NOJUDGE='    if [ ! -f "$judge" ]; then'
 env_die() { printf 'ENV   %s -- the table judged nothing, not a pass\n' "$*" >&2; exit 2; }
 for f in "$AUTOPILOT" "$MODELS" "$PARAMS" "$PREFLIGHT"; do [ -r "$f" ] || env_die "no $f"; done
 for t in git python3; do command -v "$t" > /dev/null 2>&1 || env_die "no $t"; done
-for a in "$A_SHARED" "$A_255" "$A_BUILD" "$A_NORCPT" "$A_RED" "$A_DECLINE" "$A_PROOF" "$A_DISK" "$A_CHOOM_R" "$A_CHOOM_L" "$A_CELLS_R" "$A_CELLS_L"; do
+for a in "$A_SHARED" "$A_255" "$A_BUILD" "$A_NORCPT" "$A_RED" "$A_DECLINE" "$A_PROOF" "$A_DISK" "$A_CHOOM_R" "$A_CHOOM_L" "$A_CELLS_R" "$A_CELLS_L" "$A_APR_L" "$A_APR_R"; do
     grep -qF -- "$a" "$MODELS" || env_die "models_t1.sh has no '$a' line -- the subject moved"
 done
 for a in "$A_DIE" "$A_RDIE" "$A_RNODR"; do
@@ -207,9 +210,14 @@ fixture() {
 #!/usr/bin/env bash
 while [ $# -gt 0 ]; do case "$1" in --host) h=$2; shift 2 ;; --out) o=$2; shift 2 ;; --cells) c=1; shift ;; *) shift ;; esac; done
 [ "${c:-0}" = 1 ] || { echo "fixture: no --cells on $h, so the receipt would carry no cells[] (#3715 B1)"; exit 2; }
+# A3: the real ladder's step 0 refuses these; the stub refuses the same way so the wiring is what is tested.
+case "${APR:-}" in /*) ;; *) echo "fixture: \$APR is not an absolute path on $h ('${APR:-}') -- A3"; exit 2 ;; esac
+s=$(sha256sum -- "$APR") && [ "${s%% *}" = "${APR_RELEASE_SHA256:-}" ] \
+    || { echo "fixture: \$APR sha256 is not APR_RELEASE_SHA256 on $h -- A3"; exit 2; }
+c=${APR_RELEASE_COMMIT:-unset}; case "$("$APR" --version)" in *"(${c:0:9})"*) ;; *) echo "fixture: \$APR is not built from APR_RELEASE_COMMIT on $h -- A3"; exit 2 ;; esac
 printf 'ladder host=%s oom=%s flock=%s\n' "$h" "${FX_OOM:-none}" "${FX_FLOCK:-0}" >> "$FX_LOG"
 [ "${FX_NO_RECEIPT_HOST:-}" = "$h" ] && { echo "fixture: measured nothing on $h"; exit 2; }
-apr="${CARGO_TARGET_DIR:-$PWD/target}/release/apr"
+apr="$APR"
 v=$(sed -n 's/^version = "\(.*\)"$/\1/p' Cargo.toml | head -n 1)
 red=0; [ "${FX_RED_HOST:-}" = "$h" ] && red=1
 mkdir -p "$o"
@@ -458,6 +466,8 @@ mutant no-choom         "$MODELS" "$A_CHOOM_R" 'bash scripts/model_ladder.sh --h
 mutant wrapper-flock    "$MODELS" "$A_CHOOM_L" '    flock /tmp/apr-gpu.lock choom -n 1000 -- bash scripts/model_ladder.sh --host "$LOCAL_HOST"' oom_victim
 mutant no-cells-remote   "$MODELS" "$A_CELLS_R" 'bash scripts/model_ladder.sh --host $REMOTE_HOST --out' green_pair
 mutant no-cells-local    "$MODELS" "$A_CELLS_L" 'bash scripts/model_ladder.sh --host "$LOCAL_HOST" --out' green_pair
+mutant apr-unpinned-local "$MODELS" "$A_APR_L" '    APR="$apr" \' green_pair
+mutant apr-bare-remote  "$MODELS" "$A_APR_R" 'APR=apr APR_RELEASE_SHA256="\${sum%% *}" APR_RELEASE_COMMIT="$sha" \\' green_pair
 mutant r7-no-version    "$PREFLIGHT" "$R_VER" '    out="$(cd "$root" && bash "$judge" 2>&1)"; rc=$?' r7_green preflight
 mutant r7-red-is-go     "$PREFLIGHT" "$R_RED" '        *) return 0; printf '"'"'FAIL  R7 model matrix NOT green for %s (rc %s):\n%s\n'"'"' "$version" "$rc" \' r7_red preflight
 mutant r7-no-fail-lines "$PREFLIGHT" "$R_LINES" "$R_LINES_HEAD"$'\n''               "" ;;' r7_missing preflight
@@ -465,6 +475,6 @@ mutant r7-decline-is-go "$PREFLIGHT" "$R_DECLINE" '        2) return 0; echo "FA
 mutant r7-no-judge-ok   "$PREFLIGHT" "$R_NOJUDGE" '    if false; then' r7_nojudge preflight
 
 # VACUITY FLOOR: a table that ran fewer rows than it declares is not a pass.
-[ "$rows" -ge 41 ] || { printf 'VACUOUS %s row(s) ran, fewer than the 41 declared\n' "$rows" >&2; exit 1; }
+[ "$rows" -ge 43 ] || { printf 'VACUOUS %s row(s) ran, fewer than the 43 declared\n' "$rows" >&2; exit 1; }
 [ "$fails" -eq 0 ] || { printf 'RED   %s of %s row(s) failed\n' "$fails" "$rows" >&2; exit 1; }
 printf 'PASS  %s row(s): the model matrix runs at T-1 on both hosts, every failure to prove the release STOPs before the tag, and R7 refuses the same failures at T-4 (#3717)\n' "$rows"
diff --git a/scripts/lib/model_ladder_cells_produce.py b/scripts/lib/model_ladder_cells_produce.py
index 134fe849e7..11648c667e 100644
--- a/scripts/lib/model_ladder_cells_produce.py
+++ b/scripts/lib/model_ladder_cells_produce.py
@@ -39,6 +39,7 @@
 import socket
 import subprocess
 import sys
+import threading
 import time
 import urllib.error
 import urllib.request
@@ -167,22 +168,83 @@ def refusal(stderr):
     return tuple(int(float(x) * MIB) for x in m.groups())
 
 
+# H1 (operator 2026-09-28 17:50Z): default 10 min per cell; after sampling, per class 4 x measured p95, floor 5 min.
+DEFAULT_CELL_TIMEOUT_S = 600
+TIMEOUT_FLOOR_S = 300
+COMPLETED = [0]  # cells finished (any verdict), read by the H5 takt watch
+
+
+class Takt:
+    """H5 (operator 2026-09-28 17:50Z): takt tripwires after rc.1 -- at +2 h, +4 h, +6 h the completed cells must be
+    >= 90% of plan, or ANDON immediately. A clock-driven thread, not a per-model check: a model that runs long is
+    exactly when the checkpoint passes unseen. Plan: {"start_epoch": s, "checkpoints": [{"at_s": 7200,
+    "planned_cells": n}, ...]}. Every checkpoint (met or not) is one JSON line in the takt log; a miss also prints
+    `ANDON H5 ...` to stderr and to the andon file next to the log. It signals; it never skips or stops a cell."""
+    RATIO = 0.9
+
+    def __init__(self, plan_path, log_path, clock=time.time):
+        plan = json.load(open(plan_path))
+        self.start, self.clock, self.log = float(plan["start_epoch"]), clock, log_path
+        self.checks = sorted((int(c["at_s"]), int(c["planned_cells"])) for c in plan["checkpoints"])
+        if not self.checks:
+            raise ValueError("takt plan has no checkpoints -- a tripwire with nothing to trip is not H5")
+        self.fired = 0
+
+    def tick(self):
+        """Judge every checkpoint whose time has passed; returns the lines written this tick."""
+        out, now = [], self.clock()
+        while self.fired < len(self.checks) and now >= self.start + self.checks[self.fired][0]:
+            at, plan = self.checks[self.fired]
+            done = COMPLETED[0]
+            ok = done >= self.RATIO * plan
+            rec = {"at_s": at, "planned_cells": plan, "completed_cells": done, "ratio": round(done / plan, 3) if plan else None,
+                   "threshold": self.RATIO, "andon": not ok, "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now))}
+            with open(self.log, "a") as f:
+                f.write(json.dumps(rec) + "\n")
+            if not ok:
+                line = f"ANDON H5 takt +{at / 3600:g}h: {done} cells completed < {self.RATIO:.0%} of plan {plan}"
+                print(line, file=sys.stderr, flush=True)
+                with open(self.log + ".andon", "a") as f:
+                    f.write(line + "\n")
+            out.append(rec)
+            self.fired += 1
+        return out
+
+    def watch(self, poll_s=30):
+        def loop():
+            while self.fired < len(self.checks):
+                self.tick()
+                time.sleep(poll_s)
+        threading.Thread(target=loop, daemon=True).start()
+
+
 class Runner:
     """Runs apr under the fleet GPU lock: `flock -E 75 -w WAIT LOCK choom -n 1000 -- apr ...`."""
 
     def __init__(self, apr, lock, wait, timeout):
         self.apr, self.lock, self.wait, self.timeout = apr, lock, wait, timeout
+        self.timed_out = False
+        self.deadline = None  # H1: one deadline per CELL, shared by every apr call in it (density retries)
 
     def argv(self, args):
         pre = ["flock", "-E", "75", "-w", str(self.wait), self.lock, "choom", "-n", "1000", "--"] if self.lock else []
         return pre + [self.apr] + args
 
     def call(self, args, stdin=None):
+        # H1 (operator 2026-09-28 17:50Z): a hard per-cell timeout, and a timeout is a FAIL -- never a skip. The whole
+        # process GROUP is killed: subprocess.run's timeout killed only `flock`, and the apr grandchild it had
+        # forked kept running and kept the GPU (and the fleet lock's fd) after the cell was already recorded.
+        p = subprocess.Popen(self.argv(args), stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
+                             stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True)
         try:
-            p = subprocess.run(self.argv(args), input=stdin, capture_output=True, text=True, timeout=self.timeout)
-            return p.returncode, p.stdout, p.stderr
+            left = self.timeout if self.deadline is None else max(1.0, self.deadline - time.monotonic())
+            out, err = p.communicate(stdin, timeout=left)
+            return p.returncode, out, err
         except subprocess.TimeoutExpired:
-            return 124, "", f"timed out after {self.timeout} s"
+            os.killpg(p.pid, signal.SIGKILL)
+            p.communicate()
+            self.timed_out = True
+            return 124, "", f"TIMEOUT after {self.timeout} s -- a FAIL, never a skip (H1)"
 
 
 def base_row(item, verb, mode, rid, max_tokens):
@@ -269,6 +331,7 @@ def __init__(self, R, path, ceiling_s):
         self.proc = subprocess.Popen(R.argv(["serve", "run", path, "--port", str(self.port), "--gpu"]),
                                      stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True)
         self.up = self._wait(ceiling_s)
+        self.timed_out = False
 
     def _wait(self, ceiling_s):
         end = time.time() + ceiling_s
@@ -290,6 +353,9 @@ def ask(self, prompt, mode, max_tokens, timeout):
         except urllib.error.HTTPError as e:
             return e.code, {}, e.read().decode(errors="replace")
         except (urllib.error.URLError, OSError, json.JSONDecodeError) as e:
+            if isinstance(e, TimeoutError) or "timed out" in str(e):
+                self.timed_out = True
+                return 124, {}, f"TIMEOUT after {timeout} s -- a FAIL, never a skip (H1)"
             return 1, {}, str(e)
 
     def close(self):
@@ -337,9 +403,35 @@ def cells_for(item, L, rungs_doc):
     return out
 
 
+def cell_timeout(a, item, verb):
+    """H1: the class timeout (file|verb) when sampled, else the default; never under the 5-minute floor."""
+    t = (a.class_timeouts or {}).get(f"{item['file']}|{verb}", a.timeout)
+    return max(int(t), TIMEOUT_FLOOR_S) if f"{item['file']}|{verb}" in (a.class_timeouts or {}) else int(t)
+
+
+def stamp(row, t0, timeout, timed_out):
+    """Every row records its wall time and the timeout it ran under (H1: the value is in every receipt)."""
+    row["wall_s"], row["timeout_s"], row["timed_out"] = round(time.monotonic() - t0, 1), timeout, bool(timed_out)
+    COMPLETED[0] += 1
+    if timed_out:
+        row["verdict"] = "fail"
+        if not row["reason"].startswith("TIMEOUT"):
+            row["reason"] = f"TIMEOUT after {timeout} s -- a FAIL, never a skip (H1); " + row["reason"]
+    return row
+
+
 def measure_item(R, item, path, L, rungs_doc, a):
     rows, density = [], a.chars_per_token
     owed = cells_for(item, L, rungs_doc)
+    for verb in sorted({o[3] for o in owed}):
+        print(f"cells: owed {item['file']} {verb} {sum(o[3] == verb for o in owed)}", file=sys.stderr)
+    if a.select:
+        # H4 sampling: one (rung, thinking) per model, all verbs -- a class sample, never a release verdict.
+        # `<rung>:max` takes the heavier mode the model owes at that rung (on, when it thinks at all).
+        rid, _, mode = a.select.partition(":")
+        if mode == "max":
+            mode = "on" if any(o[0] == rid and o[2] == "on" for o in owed) else "off"
+        owed = [o for o in owed if o[0] == rid and o[2] == mode]
     by_rung = {}
     for rid, tok, mode, verb in owed:
         by_rung.setdefault((rid, tok), []).append((mode, verb))
@@ -370,6 +462,8 @@ def measure_item(R, item, path, L, rungs_doc, a):
                 verbs = sorted((v for m, v in jobs if m == mode), key=lambda v: v != "run")  # run first: it counts
                 for verb in verbs:
                     row = base_row(item, verb, mode, rid, budget)
+                    R.timeout, R.timed_out, t0 = cell_timeout(a, item, verb), False, time.monotonic()
+                    R.deadline = t0 + R.timeout  # the timeout bounds the CELL: 3 density retries share it
                     if verb == "run":
                         for _attempt in range(3):  # the first rung's density is a guess; apr's count corrects it
                             row = measure_run(R, path, pf, mode, budget, base_row(item, verb, mode, rid, budget))
@@ -388,13 +482,17 @@ def measure_item(R, item, path, L, rungs_doc, a):
                         continue
                     else:
                         row = (measure_chat if verb == "chat" else measure_code)(R, path, prompt, mode, budget, row)
+                    stamp(row, t0, R.timeout, R.timed_out)
                     if row["prompt_tokens"] is None and measured:
                         row["prompt_tokens"] = measured
                         row["prompt_tokens_source"] = "apr run, identical prompt text (this verb prints no count)"
                     rows.append(row)
+        R.deadline = None
         for i, prompt, mode, budget, measured in deferred:
+            t0, tmo = time.monotonic(), cell_timeout(a, item, "serve")
             serve = serve or Serve(R, path, a.serve_ceiling)
-            row = measure_serve(serve, prompt, mode, budget, rows[i], R.timeout)
+            serve.timed_out = False
+            row = stamp(measure_serve(serve, prompt, mode, budget, rows[i], tmo), t0, tmo, serve.timed_out)
             if row["prompt_tokens"] is None and measured:
                 row["prompt_tokens"] = measured
                 row["prompt_tokens_source"] = "apr run, identical prompt text (this verb prints no count)"
@@ -429,7 +527,11 @@ def cmd_measure(a):
     rungs_doc = json.load(open(a.rungs))
     paths = dict(l.rstrip("\n").split("|", 1) for l in open(a.models) if "|" in l)
     R = Runner(a.apr, a.lock, a.lock_wait, a.timeout)
+    a.class_timeouts = json.load(open(a.timeout_classes)) if a.timeout_classes else {}
     rows, matched = [], 0
+    takt = Takt(a.takt_plan, a.takt_log or a.out + ".takt.jsonl") if a.takt_plan else None
+    if takt:
+        takt.watch()
     dec, drc = J.declaimed(L.get("cells") or {}, {h.get("id") for h in L.get("hosts") or []}, lambda w: print("cells: refused -- " + w, file=sys.stderr))
     if drc:
         return 1
@@ -449,6 +551,8 @@ def cmd_measure(a):
         # selects none. Zero rows at rc 0 would read as "this host owes no cells" (quorum 2, #3715).
         print(f"cells: refused -- --only {a.only!r} names no inventory model; cells are measured per model", file=sys.stderr)
         return 2
+    if takt:
+        takt.tick()
     json.dump(rows, open(a.out, "w"), indent=1)
     print(f"cells: {len(rows)} row(s), {sum(r['verdict'] == 'pass' for r in rows)} pass, "
           f"{sum(r['verdict'] == 'refused' for r in rows)} refused")
@@ -472,7 +576,12 @@ def main(argv=None):
     m.add_argument("--lock", default="")
     m.add_argument("--host", default="", help="this host's ladder id: a cells.declaimed arch on it is not measured")
     m.add_argument("--lock-wait", type=int, default=1800)
-    m.add_argument("--timeout", type=int, default=3600, help="per apr call, seconds")
+    m.add_argument("--timeout", type=int, default=DEFAULT_CELL_TIMEOUT_S,
+                   help="H1 hard per-cell timeout, seconds; a timeout is a FAIL (default 600)")
+    m.add_argument("--timeout-classes", default="", help='H1 JSON {"<file>|<verb>": seconds} (4 x measured p95); floor 300')
+    m.add_argument("--takt-plan", default="", help='H5 JSON {"start_epoch": s, "checkpoints": [{"at_s", "planned_cells"}]}')
+    m.add_argument("--takt-log", default="", help="H5 checkpoint JSON lines (default <out>.takt.jsonl; misses also in <log>.andon)")
+    m.add_argument("--select", default="", help="H4 sampling: only the cells of this <rung>:<thinking>, e.g. 20k:off")
     m.add_argument("--serve-ceiling", type=int, default=600)
     m.add_argument("--max-tokens", type=int, default=64)
     m.add_argument("--max-tokens-thinking", type=int, default=1024, help="a think block must have room to CLOSE")
diff --git a/scripts/model_ladder.sh b/scripts/model_ladder.sh
index 3ae4abc12a..2c89620c93 100755
--- a/scripts/model_ladder.sh
+++ b/scripts/model_ladder.sh
@@ -76,7 +76,34 @@ cd "${MODEL_LADDER_ROOT:-$(dirname "$0")/..}" || exit 2
 [ -f "$LADDER" ] || { echo "decline: $LADDER not found" >&2; exit 2; }
 
 # Step 0 — pin the binary. A diagnostic against the wrong apr is worse than none.
-if [ "${DOGFOOD_ALLOW_UNPINNED:-0}" = "1" ] && [ -n "${APR:-}" ]; then
+# A3 (operator 2026-09-28 17:28Z): the RELEASE cells run the artifact UNDER TEST — the rc build —
+# named by ABSOLUTE path and proven twice: its bytes by sha256, and the commit its own `--version`
+# prints against the release commit. PATH is never consulted: a bare or relative $APR is refused
+# before it could resolve, so a different-sha apr planted first on PATH cannot run a single cell.
+# Release mode is on when either variable is set; half a pin is a decline, never a fallback.
+release_pin() {
+  local want="${APR_RELEASE_SHA256:-}" commit="${APR_RELEASE_COMMIT:-}" got ver
+  [[ "$want" =~ ^[0-9a-f]{64}$ ]] || { echo "decline: APR_RELEASE_SHA256 is not a 64-hex sha256: '$want'" >&2; return 2; }
+  [[ "$commit" =~ ^[0-9a-f]{40}$ ]] || { echo "decline: APR_RELEASE_COMMIT is not a 40-hex commit: '$commit'" >&2; return 2; }
+  case "${APR:-}" in
+    /*) ;;
+    *) echo "decline: release mode needs \$APR as an ABSOLUTE path to the rc artifact, got '${APR:-}' (a PATH lookup could run another apr)" >&2; return 2 ;;
+  esac
+  [ -f "$APR" ] && [ -x "$APR" ] || { echo "decline: \$APR is not an executable file: '$APR'" >&2; return 2; }
+  got=$(sha256sum -- "$APR") || { echo "decline: sha256sum of '$APR' failed" >&2; return 2; }
+  got=${got%% *}
+  [ "$got" = "$want" ] || { echo "decline: \$APR sha256 $got != APR_RELEASE_SHA256 $want -- not the artifact under test" >&2; return 2; }
+  ver=$("$APR" --version 2>/dev/null) || { echo "decline: '$APR --version' failed" >&2; return 2; }
+  case "$ver" in
+    *"(${commit:0:9})"*) ;;
+    *) echo "decline: '$APR --version' says '${ver%%$'\n'*}', not built from release commit ${commit:0:9}" >&2; return 2 ;;
+  esac
+}
+APR_PIN=head
+if [ -n "${APR_RELEASE_SHA256:-}${APR_RELEASE_COMMIT:-}" ]; then
+  release_pin || exit 2
+  APR_PIN=release
+elif [ "${DOGFOOD_ALLOW_UNPINNED:-0}" = "1" ] && [ -n "${APR:-}" ]; then
   : # published-crate mode (apr-dogfood G13): caller pinned $APR deliberately
 else
   # shellcheck disable=SC1091
@@ -481,6 +508,8 @@ SHA=$(git rev-parse --short HEAD 2>/dev/null || echo unknown)
 # apr_sha: the full 40-hex HEAD, which scripts/apr_bin.sh proved the binary was built from. The
 # release-readiness shape (#3715) compares it to the release commit by exact equality.
 APR_SHA=$(git rev-parse HEAD 2>/dev/null || echo unknown)
+# A3: in release mode the binary was proven against the release commit, not against this tree's HEAD.
+[ "$APR_PIN" = release ] && APR_SHA="$APR_RELEASE_COMMIT"
 VERSION=$(cargo metadata --no-deps --offline --format-version 1 2>/dev/null | python3 -c '
 import json, os, sys
 m = json.load(sys.stdin)
@@ -1428,8 +1457,14 @@ fi
 CELLS_JSON=""
 if [ "$CELLS" = 1 ]; then
   CELLS_JSON="$WORK/cells.json"
+  # H1/H5 (operator 2026-09-28 17:50Z): every cell runs under a hard timeout (default 10 min; per class from
+  # MODEL_LADDER_TIMEOUT_CLASSES once sampled) and a timeout is a FAIL. MODEL_LADDER_TAKT_PLAN arms the takt
+  # tripwires (+2/+4/+6 h, >= 90% of plan); a miss lands in $OUT_DIR/takt-$HOST.jsonl.andon the moment it happens.
+  pace=()
+  [ -n "${MODEL_LADDER_TIMEOUT_CLASSES:-}" ] && pace+=(--timeout-classes "$MODEL_LADDER_TIMEOUT_CLASSES")
+  [ -n "${MODEL_LADDER_TAKT_PLAN:-}" ] && pace+=(--takt-plan "$MODEL_LADDER_TAKT_PLAN" --takt-log "$OUT_DIR/takt-$HOST.jsonl")
   python3 "$PRODUCE" measure --apr "$APR" --inventory "$INV_RECEIPT" --models "$WORK/models.txt" \
-      --ladder "$LADDER" --rungs evidence/release/context-rungs.json --work "$WORK" \
+      --ladder "$LADDER" --rungs evidence/release/context-rungs.json --work "$WORK" "${pace[@]}" \
       --lock "$GPU_LOCK" --lock-wait "$LOCK_WAIT" --only "$ONLY" --host "$HOST" --out "$CELLS_JSON" > "$WORK/cells.log" 2>&1
   cells_rc=$?
   tail -1 "$WORK/cells.log"
@@ -1439,7 +1474,7 @@ if [ "$CELLS" = 1 ]; then
 fi
 why=$(ladder_disk_probe "$OUT_DIR") || ladder_write_decline "before the receipt: $why"
 python3 - "$ROWS" "$RECEIPT_TMP" "$HOST" "$VERSION" "$SHA" "${GPU_NAME:-}" "${GPU_CC:-}" "$EXECUTED" "$RED" "$APR_VERSION" "$INV_RECEIPT" "$INV_DIRS" "$INV_PATTERNS" "$APR_SHA" "$ONLY" "$GPU_MEM" "$CELLS_JSON" "$LADDER" <<'PY'
-import json, sys, datetime, platform
+import json, os, sys, datetime, platform
 rows = [json.loads(l) for l in open(sys.argv[1]) if l.strip()]
 inv = [json.loads(l) for l in open(sys.argv[11]) if l.strip()]
 # #4590: a (host, arch) the release DE-CLAIMS leaves inventory[] -- the universe released pv owes cells on --
@@ -1454,6 +1489,8 @@ out = {"schema": "apr-model-ladder-receipt/v2", "host": sys.argv[3], "version":
        "date": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%MZ"),
        "apr_version": sys.argv[10], "executed": int(sys.argv[8]), "red": int(sys.argv[9]),
        "only": (sys.argv[15] or None),
+       # A3: release mode records the sha256 step 0 verified; absent (not null-faked) otherwise.
+       **({"apr_sha256": os.environ["APR_RELEASE_SHA256"]} if os.environ.get("APR_RELEASE_SHA256") else {}),
        "inventory": inv, "inventory_dirs": sys.argv[12].split(":"), "inventory_patterns": sys.argv[13].split(","),
        "rungs": rows}
 if held:
diff --git a/scripts/release/models_t1.sh b/scripts/release/models_t1.sh
index 3c8e32bcc1..3eb663e98c 100644
--- a/scripts/release/models_t1.sh
+++ b/scripts/release/models_t1.sh
@@ -96,13 +96,17 @@ mkdir -p "$out" || exit 2
 rm -f -- "$out/$LOCAL_HOST.json" "$out/$REMOTE_HOST.json" "$out/$REMOTE_HOST.json.part"
 
 local_leg() {
-    local tdir got
+    local tdir got apr sum
     cargo build --release -p apr-cli --bin apr --features cuda --locked \
         || { echo "MODELS-LEG $LOCAL_HOST BUILD-FAILED"; return 3; }
     tdir=${CARGO_TARGET_DIR:-$(cargo metadata --no-deps --format-version 1 | jq -r .target_directory)}
-    got=$("$tdir/release/apr" --version 2>/dev/null | head -n 1)
+    apr="$tdir/release/apr"; case "$apr" in /*) ;; *) apr="$PWD/$apr" ;; esac
+    got=$("$apr" --version 2>/dev/null | head -n 1)
     [ "$got" = "$want" ] || { echo "MODELS-LEG $LOCAL_HOST NOT-THE-RELEASE: '$got' (want '$want')"; return 3; }
-    choom -n 1000 -- bash scripts/model_ladder.sh --host "$LOCAL_HOST" --cells --out "$out"
+    # A3: the ladder runs THIS file, by absolute path, re-proven by sha256 + --version commit (its step 0).
+    sum=$(sha256sum -- "$apr") || { echo "MODELS-LEG $LOCAL_HOST NOT-THE-RELEASE: sha256sum failed"; return 3; }
+    APR="$apr" APR_RELEASE_SHA256="${sum%% *}" APR_RELEASE_COMMIT="$sha" \
+        choom -n 1000 -- bash scripts/model_ladder.sh --host "$LOCAL_HOST" --cells --out "$out"
 }
 
 remote_leg() {
@@ -130,10 +134,13 @@ cd "\$dir/wt" || exit 3
 export CARGO_TARGET_DIR="\$dir/target"
 cargo build --release -p apr-cli --bin apr --features cuda --locked > "\$dir/build.log" 2>&1 \
   || { tail -n 20 "\$dir/build.log"; echo "MODELS-LEG $REMOTE_HOST BUILD-FAILED"; exit 3; }
-got=\$("\$CARGO_TARGET_DIR/release/apr" --version 2>/dev/null | head -n 1)
+apr="\$CARGO_TARGET_DIR/release/apr"
+got=\$("\$apr" --version 2>/dev/null | head -n 1)
 [ "\$got" = "$want" ] || { echo "MODELS-LEG $REMOTE_HOST NOT-THE-RELEASE: '\$got' (want '$want')"; exit 3; }
+sum=\$(sha256sum -- "\$apr") || { echo "MODELS-LEG $REMOTE_HOST NOT-THE-RELEASE: sha256sum failed"; exit 3; }
 rm -rf -- "\$dir/out"
-choom -n 1000 -- bash scripts/model_ladder.sh --host $REMOTE_HOST --cells --out "\$dir/out"; lrc=\$?
+APR="\$apr" APR_RELEASE_SHA256="\${sum%% *}" APR_RELEASE_COMMIT="$sha" \\
+  choom -n 1000 -- bash scripts/model_ladder.sh --host $REMOTE_HOST --cells --out "\$dir/out"; lrc=\$?
 if [ -f "\$dir/out/$REMOTE_HOST.json" ]; then
   echo "---RECEIPT $REMOTE_HOST---"; cat "\$dir/out/$REMOTE_HOST.json"; echo "---END RECEIPT---"
 fi