unseen-eval: calibrated replay deadline + native agreement check, docker-lane diagnostics, proof 4.4 record; engine runner cleans up fetched sources

gVisor replays exhausted the hook's 30 s deadline in the snapshot step on a
1,725-file repository; cluster replays now use 30 s x a measured slowdown
factor, verified on a native sample. The engine runner deletes each pin's fetched
source (1.8 GB had accumulated), caps the cache at 500 MB and refuses below
1.5 GB free. Proof 4.4: two engine pins via the generic dispatcher match the VM
baseline exactly. Implemented with Codex.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

.kiro/specs/unseen-repo-evaluation/design.md

@@ -219,6 +219,22 @@ what the user sees (30 s) and what the engine could catch given time (300 s), so
 told apart from a miss by the engine. It is a measurement arm, not a proposed default: adopting a longer hook
 deadline would be a separate decision under 7.2.
 
+**Decision 4.0a: calibrated cluster deadline and native check** (maintainer, 2026-10-05).
+The gVisor proof exhausted 30 seconds in snapshot preparation on a 1,725-file repository.
+Cluster replays therefore scale the 30/300-second arms by the median paired cluster/native
+time through the quick tier, rounded to integer seconds. Native runc replays retain 30/300 seconds.
+A deterministic sample (default 20, or all of a smaller proof manifest) runs both lanes at
+600 seconds for calibration. Report per-change stage times and median slowdown with bootstrap
+95% intervals for time through quick, engine build and engine queries separately. Incomplete
+timings or degraded scans invalidate calibration. Records contain counts, times and opaque digests.
+
+A deterministic native sample accompanies the cluster slice. Compare whether each push flags
+anything, with a Wilson 95% interval. If its lower bound is below 95%, or any sample pair is
+missing/failed, flag the run and exclude cluster results from the baseline. Disagreements name
+change IDs only. The scorer reports applied deadlines and factors. Docker pulls explicitly after
+checking its data-root free space against an unpacked image size upper bound plus 1 GiB; failures
+retain sanitized stderr tails and exit codes. See the unseen README for coordinator commands.
+
 **Decision 4.1: catch, per family.** A finding counts if its family is the change's family. Engine streams carry a
 family; a quick-rule match maps its CWE through the family CWE table. Its head location must also be in the
 vulnerable function or within 10 lines of the change's head-side hunks (protocol v1's window).

benchmarks/ax/batch.py

@@ -387,7 +387,15 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o
             for artifact in attempt_output.iterdir():
                 if artifact.is_file():
                     shutil.copyfile(artifact, output / artifact.name)
-            attempts.append({"lane": lane, "status": record.get("status"), "attempt": attempt, "transport": record.get("ax_attempts", [])})
+            attempts.append(
+                {
+                    "lane": lane,
+                    "status": record.get("status"),
+                    "attempt": attempt,
+                    "transport": record.get("ax_attempts", []),
+                    "docker_errors": record.get("docker_errors", []),
+                }
+            )
             if record.get("status") not in {"sandbox_failure", "instrument_failure"}:
                 break
         return {**record, "attempts": (previous or {}).get("attempts", []) + attempts, "input_digest": item.input_digest, "done": True}
@@ -402,9 +410,76 @@ def dispatch(items, workload, out, executors, *, lanes=("ax", "ax", "docker"), o
     return schedule(pending, lanes, out, execute, complete, fallbacks=fallbacks, overflow=overflow)
 
 
+def docker_diagnostic(text):
+    """Bound diagnostics and remove bearer URLs, environment secrets and assignments."""
+    text = str(text or "")
+    for key, value in os.environ.items():
+        if len(value) >= 4 and any(word in key.upper() for word in ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "KEY")):
+            text = text.replace(value, "[redacted]")
+    text = re.sub(r"https?://[^\s\"']+", "[redacted-url]", text)
+    text = re.sub(
+        r"(?i)(authorization|bearer|password|token|secret|signature|credential|[\w-]*key)\s*[:= ]\s*[^\s,;]+", r"\1=[redacted]", text
+    )
+    return "\n".join(text.splitlines()[-10:])[-4000:]
+
+
 class DockerLane(AXLane):
     """VM lane uses the same presigned-object protocol and pinned replay image."""
 
+    def __init__(self, *args, runner=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.runner = runner or subprocess.run
+        self._prepared = False
+
+    def __call__(self, item, output, attempt):
+        self._diagnostics = []
+        try:
+            if not self._prepared:
+                self._prepare_image()
+                self._prepared = True
+        except (ValueError, OSError, subprocess.SubprocessError) as exc:
+            if not self._diagnostics:
+                self._diagnostics.append({"step": "preflight", "exit_code": None, "stderr": docker_diagnostic(str(exc))})
+            record = {"status": "instrument_failure", "docker_errors": self._diagnostics}
+            write_json(output / "docker-error.json", record)
+            return record
+        try:
+            record = super().__call__(item, output, attempt)
+        finally:
+            if self._diagnostics:
+                write_json(output / "docker-error.json", {"docker_errors": self._diagnostics})
+        if self._diagnostics:
+            record["docker_errors"] = self._diagnostics
+        return record
+
+    def _run_docker(self, command, timeout):
+        try:
+            done = self.runner(command, capture_output=True, text=True, timeout=timeout)
+        except (OSError, subprocess.SubprocessError) as exc:
+            self._diagnostics.append(
+                {"step": command[1], "exit_code": None, "stderr": docker_diagnostic(getattr(exc, "stderr", "") or type(exc).__name__)}
+            )
+            raise
+        if done.returncode:
+            self._diagnostics.append({"step": command[1], "exit_code": done.returncode, "stderr": docker_diagnostic(done.stderr)})
+            raise ValueError(f"docker {command[1]} failed (exit {done.returncode})")
+        return done
+
+    def _prepare_image(self):
+        # Docker's data root, not the coordinator checkout, is where layers consume space.
+        root = self._run_docker(["docker", "info", "--format", "{{.DockerRootDir}}"], 30).stdout.strip()
+        size = getattr(self.args, "docker_image_bytes", None)
+        if not isinstance(size, int) or size <= 0:
+            raise ValueError("set --docker-image-bytes to the unpacked image size upper bound before pulling")
+        free = self.runner(["df", "-B1", "--output=avail", root], capture_output=True, text=True, timeout=30)
+        if free.returncode:
+            raise ValueError("cannot check free space on Docker data root")
+        available = int(free.stdout.splitlines()[-1].strip())
+        required = size + 1024**3
+        if available < required:
+            raise ValueError(f"refusing Docker pull: free bytes {available} below image + 1 GiB ({required})")
+        self._run_docker(["docker", "pull", self.workload.image], 900)
+
     def _execute_attempt(self, item, output, attempt, attempts):
         # Reuse all object/validation/cleanup logic, replacing only the task runner.
         import copy
@@ -417,18 +492,44 @@ class DockerLane(AXLane):
     def _docker_command(self, *args, manifest=None, timeout=30):
         if args[0] == "apply":
             self._docker_name = manifest["metadata"]["name"]
-            command = ["docker", "run", "-d", "--name", self._docker_name, "--entrypoint", manifest["spec"]["command"][0]]
+            self._docker_created = True  # Even a failed run can leave a created container.
+            command = [
+                "docker",
+                "run",
+                "-d",
+                "--runtime",
+                "runc",
+                "--pull",
+                "never",
+                "--name",
+                self._docker_name,
+                "--entrypoint",
+                manifest["spec"]["command"][0],
+            ]
             for env in manifest["spec"]["env"]:
                 command += ["-e", env["name"] + "=" + env["value"]]
             command += [manifest["spec"]["image"], *manifest["spec"]["command"][1:], "--heap-profile", "vm"]
         elif args[0] == "get":
             command = ["docker", "inspect", "--format", "{{.State.Running}} {{.State.ExitCode}}", self._docker_name]
         else:
+            if not getattr(self, "_docker_created", False):
+                return subprocess.CompletedProcess([], 0, "", "")
             command = ["docker", "rm", "-f", self._docker_name or args[2]]
-        done = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
+        try:
-        if done.returncode:
+            done = self._run_docker(command, timeout)
-            raise ValueError("docker command failed")
+        except ValueError:
+            if args[0] == "delete" and "No such container" in self._diagnostics[-1]["stderr"]:
+                self._diagnostics.pop()
+                return subprocess.CompletedProcess(command, 0, "", "")
+            raise
+        if args[0] == "apply":
+            self._docker_created = True
         if args[0] == "get":
             phase = "Running" if done.stdout.startswith("true") else "Failed"
+            if phase == "Failed":
+                logs = self._run_docker(["docker", "logs", "--tail", "10", self._docker_name], timeout)
+                self._diagnostics.append(
+                    {"step": "container", "exit_code": int(done.stdout.split()[-1]), "stderr": docker_diagnostic(logs.stderr)}
+                )
             done.stdout = json.dumps({"metadata": {"name": self._docker_name}, "status": {"phase": phase}})
         return done

benchmarks/learn/engine_trace.py

@@ -179,9 +179,11 @@ def make_plan(
 class Inputs:
     """Resolve spent corpora; fetch pair source commits into the host cache."""
 
-    def __init__(self, root, cache, allow_excerpt_fallback=False):
+    def __init__(self, root, cache, allow_excerpt_fallback=False, *, keep_sources=False, source_cache_mb=500):
         self.cache = cache
         self.allow_excerpt_fallback = allow_excerpt_fallback
+        self.keep_sources = keep_sources
+        self.source_cache_bytes = source_cache_mb * 1024 * 1024
         self.catalog_rows = defaultdict(list)
         for catalog in (root / "benchmarks/pairs").rglob("catalog.toml"):
             for item in tomllib.loads(catalog.read_text()).get("pair", []):
@@ -259,8 +261,30 @@ class Inputs:
             raise ValueError(f"missing catalog repository for {case.name}")
         return repo, parent if side == "vuln" else commit
 
+    def source_path(self, repo, commit, roots):
+        return self.cache / "trace-sources" / digest([repo, commit, roots])
+
+    def prepare_source_cache(self):
+        cache = self.cache / "trace-sources"
+        cache.mkdir(parents=True, exist_ok=True)
+        entries = []
+        for entry in cache.iterdir():
+            if entry.is_dir() and not entry.is_symlink():
+                size = sum(p.stat().st_size for p in entry.rglob("*") if not p.is_symlink() and p.is_file())
+                entries.append((entry.stat().st_mtime_ns, entry, size))
+        total = sum(size for _, _, size in entries)
+        for _, entry, size in sorted(entries):
+            if total <= self.source_cache_bytes:
+                break
+            shutil.rmtree(entry)
+            total -= size
+        free = shutil.disk_usage(cache).free
+        if free < 1.5 * 1024**3:
+            raise OSError(f"refusing source fetch: {free / 1024**3:.2f} GB free at {cache}; at least 1.5 GB required")
+
     def fetch_source(self, repo, commit, roots):
-        clone = self.cache / "trace-sources" / digest([repo, commit, roots])
+        self.prepare_source_cache()
+        clone = self.source_path(repo, commit, roots)
         clone.mkdir(parents=True, exist_ok=True)
         url = repo if repo.startswith("https://") else f"https://github.com/{repo.removesuffix('.git')}.git"
         env = {**os.environ, "GIT_TERMINAL_PROMPT": "0"}
@@ -348,6 +372,13 @@ class Inputs:
                 for case, side in resolved:
                     _materialize_side(target, case, side=side)
                 return {"mode": "excerpt-fallback", "reason": str(exc), "repo": repo, "commit": commit}
+            finally:
+                # Only the exported copy is used by executors. Drop the clone even
+                # on a partial fetch, timeout, failed export or excerpt fallback.
+                if not self.keep_sources:
+                    clone = self.source_path(repo, commit, roots)
+                    if clone.exists():
+                        shutil.rmtree(clone)
             verify_functions(target, rows)
             return {**proof, "mode": "source", "repo": repo, "commit": commit}
         if all(r["source"] == "dev-php" for r in rows):
@@ -845,6 +876,8 @@ def main():
     parser.add_argument("--examples", type=Path, help="frozen example JSONL instead of the local memory store")
     parser.add_argument("--memory", type=Path, default=Path.home() / "ousast-results/plane/memory")
     parser.add_argument("--cache", type=Path, default=Path.home() / ".cache/openultrasast")
+    parser.add_argument("--keep-sources", action="store_true", help="retain fetched sources after export (still subject to cache eviction)")
+    parser.add_argument("--source-cache-mb", type=int, default=500, help="maximum retained source MB before each fetch (default: 500)")
     parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
     parser.add_argument("--image", help="default: openultrasast:dev for Docker; public 2.0.1 image for k8s/mixed")
     parser.add_argument("--executor", choices=("docker", "k8s", "k8s-jobs", "mixed", "queue", "ax", "mixed-ax"), default="docker")
@@ -872,6 +905,8 @@ def main():
     parser.add_argument("--dry-run", action="store_true")
     parser.add_argument("--summary", action="store_true")
     args = parser.parse_args()
+    if args.source_cache_mb < 0:
+        parser.error("--source-cache-mb must be nonnegative")
     if args.parallel is None:
         args.parallel = 2 if args.executor in {"ax", "mixed-ax"} else 1
     if args.executor in {"ax", "mixed-ax"}:
@@ -912,7 +947,7 @@ def main():
         units = join_units(read_jsonl(args.units), examples, args.include_typescript)
     except ValueError as exc:
         parser.error(str(exc))
-    inputs = Inputs(ROOT, args.cache, args.allow_excerpt_fallback)
+    inputs = Inputs(ROOT, args.cache, args.allow_excerpt_fallback, keep_sources=args.keep_sources, source_cache_mb=args.source_cache_mb)
     # Resolve pair source names for leak-audit coverage before calculating the resume key.
     for unit in units:
         if unit.get("excerpt") or unit["source"].startswith("advisory-fixes"):

benchmarks/measurements/2026-10-04-unseen-04-dispatcher/record.json

@@ -0,0 +1,110 @@
+{
+  "cleanup": "completed-before-checkpoint",
+  "commit": "5dc86c2f3f54b4dfb59060c88fb3ab5b307ef448",
+  "differences": 0,
+  "image": "ghcr.io/norandom/ousast-engine-task@sha256:b73c9bbd48fbf39c14ac013e5b0ff0a9d536b3d6388b50b1bd4cddfa22264b91",
+  "pins": [
+    {
+      "baseline_digest": "c12c675f816a24ab33a63336a21efdc93db10a37dcf83437d91bec6f728cef1f",
+      "equal": true,
+      "instrument": [
+        {
+          "bytes": 1761,
+          "cpg_bytes": 138723,
+          "jvm": [
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 11.88592994,
+              "success": null
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 35.821929360999995,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 7.4493522819999995,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 2.9803447330000026,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 0.5151375319999971,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 9.691500790999996,
+              "success": true
+            }
+          ]
+        }
+      ],
+      "pin_digest": "63c992b500590a9d4b643c9458df1d6cf71d770b06cc70594acbdba99b06c00b",
+      "record_digest": "c12c675f816a24ab33a63336a21efdc93db10a37dcf83437d91bec6f728cef1f",
+      "seconds": 158.85258329089265
+    },
+    {
+      "baseline_digest": "5a60ed069d99f9325a1d3f3a7612fcf0d853d4fabfd53add81151b051132a5ee",
+      "equal": true,
+      "instrument": [
+        {
+          "bytes": 1791,
+          "cpg_bytes": 138886,
+          "jvm": [
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 7.656527941999999,
+              "success": null
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 27.532885242,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 9.524187851,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 5.0155780199999995,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 0.6809096819999993,
+              "success": true
+            },
+            {
+              "exit_code": null,
+              "heap_mb": 1700,
+              "seconds": 13.824248304000001,
+              "success": true
+            }
+          ]
+        }
+      ],
+      "pin_digest": "8bd2f48dfb2d00cfc5b60a5dee3a8c87e6970671121caa17221f2449c1254fb2",
+      "record_digest": "5a60ed069d99f9325a1d3f3a7612fcf0d853d4fabfd53add81151b051132a5ee",
+      "seconds": 142.5083134449087
+    }
+  ]
+}

benchmarks/unseen/README.md

@@ -289,3 +289,84 @@ proof is missing, or quick-tier findings differ. It does not convert a quick-onl
 without a JVM into an engine proof. Pool replays use `--lane mixed` (default) for the two
 AX slots and single VM overflow/fallback lane; `--arm long_deadline` changes only the
 hook deadline to 300 seconds and is a separate input digest.
+
+### Deadline calibration and native agreement
+
+The 2026-10-05 maintainer decision supersedes the unscaled cluster deadline above.
+gVisor snapshot preparation can consume 30 seconds before any analysis. Calibrate
+the pinned replay image against the VM's native Docker **runc** runtime first.
+The default deterministic sample is the first 20 changes ordered by SHA-256 of
+change ID; a smaller manifest (including the five-change proof) uses all changes.
+Sampling never depends on manifest order.
+
+`calibrate` runs both lanes with a 600-second hook deadline (override `--deadline`
+if stages still time out). It reads resolution, provenance, preparation, snapshot,
+and quick timings from the push artifact; snapshot is nested in preparation.
+`quick_total` sums resolution + provenance + preparation + quick, without counting
+snapshot twice. Engine build and query times sum across the hook's scans. Hook
+wall time comes from the instrument block. Missing/nonpositive timings, deadline
+exhaustion, degraded engine scans or failed instruments prevent a successful
+calibration; failed changes are never silently dropped. Each factor is the median
+of paired cluster/native ratios with a deterministic 2,000-resample percentile
+bootstrap 95% interval. The measurement record contains opaque change digests,
+counts and timings, with no repository names. Raw artifacts remain outside the repo.
+
+Set `REPLAY_IMAGE` to the newly built replay image's pinned digest and
+`REPLAY_IMAGE_BYTES` to an **unpacked image size upper bound in bytes** from the
+image build. Do not use compressed registry layer size. Docker refuses to pull
+without this bound or when `df` at its data root reports less than that bound plus
+1 GiB free. It then explicitly pulls the image before issuing presigned task URLs.
+Supply `--docker-image-bytes "$REPLAY_IMAGE_BYTES"` to VM and mixed replay commands,
+including the earlier VM proof command.
+Launch/pull failures retain exit codes and the last ten sanitized stderr lines in
+`docker-error.json`, the result record and attempt history. URLs, known environment
+secrets and credential assignments are redacted. The original VM failure's cause
+remains unconfirmed until this diagnostic is captured; the old transport discarded it.
+
+Coordinator commands (these run networked workloads; offline tests use fakes):
+
+```sh
+export PATH=/home/mc/Source/OpenUltraSAST/.venv/bin:$PATH
+export PYTHONPATH=$PWD/src
+export S3_ENDPOINT=https://files.because-security.com
+CALIBRATION_RECORD="benchmarks/measurements/$(date +%F)-unseen-calibration/record.json"
+python -m benchmarks.unseen.calibrate \
+  --manifest "$HOME/ousast-results/unseen/proof-05-inputs.json" \
+  --sample 20 --deadline 600 --image "$REPLAY_IMAGE" \
+  --docker-image-bytes "$REPLAY_IMAGE_BYTES" --kubeconfig "$AX_KUBECONFIG" \
+  --out "$HOME/ousast-results/unseen/calibration" --record "$CALIBRATION_RECORD"
+
+python -m benchmarks.unseen.replay --manifest benchmarks/unseen/pool-p1.toml \
+  --pool p1 --slice 1 --arm default --lane ax --image "$REPLAY_IMAGE" \
+  --calibration "$CALIBRATION_RECORD" --native-sample 100 \
+  --docker-image-bytes "$REPLAY_IMAGE_BYTES" --kubeconfig "$AX_KUBECONFIG" \
+  --out "$HOME/ousast-results/unseen/p1/1/calibrated-default"
+python -m benchmarks.unseen.score "$HOME/ousast-results/unseen/p1/1/calibrated-default" \
+  --out "$HOME/ousast-results/unseen/p1/1/calibrated-default-score.json"
+```
+
+For a 20-change slice calibration, replace the proof manifest with
+`benchmarks/unseen/pool-p1.toml --slice 1`; both commands accept `--private-manifest`.
+For the long arm use `--arm long_deadline` and a separate output root. Calibration
+requires the same image digest as replay. An explicit `--deadline-scale FACTOR`
+is an alternative to `--calibration`, not an additional multiplier.
+
+Cluster deadlines are `round(30 * factor)` or `round(300 * factor)` seconds;
+native deadlines remain 30 or 300, including VM fallback in mixed runs.
+Records carry configured and applied scale, base/applied deadline and actual lane;
+the scorer reports their counts. Task, subprocess and presigned-object budgets
+grow with the applied deadline.
+
+Use `--lane ax` (or `kind`) with `--native-sample K` for an agreement run, so every
+sampled change has a cluster counterpart. The native sample lives in `OUT/native`;
+the scorer detects it automatically, or accepts `--native PATH`. Missing, failed,
+wrong-lane or mismatched-image/arm pairs are unknown, never quiet agreements.
+The decision is whether any of the three finding streams flags the push, not
+whether the finding matches the vulnerability label.
+
+**Decision rule:** if the Wilson 95% agreement lower bound is below 95%, do not use
+the cluster results for the baseline. The score sets `flagged=true`,
+`baseline_eligible=false` and the adoption gate to `not_run`. Unknown pairs also
+flag the run. Disagreements are listed by change ID only, alongside per-change
+flagged/not-flagged decisions. Even perfect agreement needs at least 73 pairs to
+clear this bound; the five-change proof or a 20-change sample cannot qualify it.

benchmarks/unseen/calibrate.py

@@ -0,0 +1,128 @@
+"""Paired native/cluster deadline calibration; publish only opaque IDs and numbers."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import random
+import re
+import statistics
+from datetime import date
+from pathlib import Path
+
+from benchmarks.ax.batch import write_json
+from benchmarks.unseen.replay import canonical, load_slice, run_replays, sample_changes
+from benchmarks.unseen.score import coverage, load_records
+
+
+def stage_times(record):
+    if coverage(record) == "instrument_failure":
+        raise ValueError("calibration instrument failed")
+    push = record["push"]
+    if "deadline_exhausted" in json.dumps(push) or any(t.get("status") == "failed" for t in push.get("quick_tier", [])):
+        raise ValueError("calibration deadline exhausted or quick tier failed; increase deadline")
+    raw = push.get("timings", {})
+    timings = {}
+    for key, value in raw.items():
+        key = re.sub(r"^comparison_\d+_", "", key)
+        if not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0:
+            raise ValueError("invalid stage timing")
+        timings[key] = timings.get(key, 0) + value
+    required = ("resolution_seconds", "provenance_seconds", "preparation_seconds", "quick_seconds", "snapshot_seconds")
+    if any(key not in timings for key in required):
+        raise ValueError("missing completed quick-stage timings")
+    scans = [r["scan"] for r in push.get("scans", [])]
+    if any(s.get("degradations") for s in scans):
+        raise ValueError("incomplete engine stages; calibration cannot use degraded scans")
+    if not scans or any(k not in s for s in scans for k in ("build_seconds", "query_seconds")):
+        raise ValueError("missing engine timings")
+    result = {
+        "snapshot": timings["snapshot_seconds"],  # Nested inside preparation; never counted twice.
+        "preparation": timings["preparation_seconds"],
+        "quick": timings["quick_seconds"],
+        "quick_total": sum(timings[k] for k in required if k != "snapshot_seconds"),
+        "engine_build": sum(s["build_seconds"] for s in scans),
+        "engine_queries": sum(s["query_seconds"] for s in scans),
+        "hook": record["instrument"].get("timings", {}).get("pre-push"),
+    }
+    if any(not isinstance(v, (int, float)) or not math.isfinite(v) or v <= 0 for v in result.values()):
+        raise ValueError("missing or nonpositive stage timing")
+    return result
+
+
+def slowdown(native, cluster, *, seed=0, draws=2000):
+    if not native or len(native) != len(cluster):
+        raise ValueError("calibration needs paired timings")
+    if any(not math.isfinite(v) or v <= 0 for v in [*native, *cluster]):
+        raise ValueError("timings must be finite and positive")
+    ratios = [b / a for a, b in zip(native, cluster, strict=True)]
+    rng = random.Random(seed)
+    bootstrap = sorted(statistics.median(rng.choices(ratios, k=len(ratios))) for _ in range(draws))
+    return {
+        "n": len(ratios),
+        "factor": statistics.median(ratios),
+        "bootstrap95": [bootstrap[int(0.025 * (draws - 1))], bootstrap[int(0.975 * (draws - 1))]],
+    }
+
+
+def calibration_record(native, cluster):
+    nn, cc = ({r["id"]: r for r in rows} for rows in (native, cluster))
+    if not nn or nn.keys() != cc.keys() or len(nn) != len(native) or len(cc) != len(cluster):
+        raise ValueError("calibration requires exactly paired unique changes")
+    pairs = []
+    for key in sorted(nn):
+        if nn[key].get("lane") != "docker" or cc[key].get("lane") != "ax":
+            raise ValueError("calibration requires native docker and cluster ax lanes")
+        pairs.append(
+            {"change_id": hashlib.sha256(str(key).encode()).hexdigest(), "native": stage_times(nn[key]), "cluster": stage_times(cc[key])}
+        )
+    return {
+        "status": "ok",
+        "changes": len(pairs),
+        "pairs": pairs,
+        **{
+            stage: slowdown([p["native"][stage] for p in pairs], [p["cluster"][stage] for p in pairs])
+            for stage in ("quick_total", "engine_build", "engine_queries")
+        },
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--private-manifest", type=Path)
+    parser.add_argument("--slice", default="1")
+    parser.add_argument("--sample", type=int, default=20)
+    parser.add_argument("--deadline", type=int, default=600)
+    parser.add_argument("--image", required=True)
+    parser.add_argument("--docker-image-bytes", type=int, required=True, help="unpacked image size upper bound")
+    parser.add_argument("--ax-bin", default="ax")
+    parser.add_argument("--kubeconfig", type=Path)
+    parser.add_argument("--atespace", default="default")
+    parser.add_argument("--out", type=Path, default=Path.home() / "ousast-results/unseen/calibration")
+    parser.add_argument("--record", type=Path, default=Path("benchmarks/measurements") / f"{date.today()}-unseen-calibration/record.json")
+    args = parser.parse_args()
+    if args.deadline < 1 or args.sample < 1:
+        parser.error("deadline and sample must be positive")
+    args.arm = "default"
+    changes = sample_changes(load_slice(args.manifest, args.slice, args.private_manifest), args.sample)
+    failed = False
+    for lane, directory in (("docker", "native"), ("ax", "cluster")):
+        failed |= run_replays(changes, args, args.out / directory, lane, deadline=args.deadline)
+    result = {"status": "failed", "changes": len(changes)}
+    if not failed:
+        try:
+            result = calibration_record(load_records(args.out / "native"), load_records(args.out / "cluster"))
+        except ValueError as exc:
+            result["reason"] = str(exc)  # Only locally controlled diagnostics.
+    result.update(
+        image=args.image, deadline_seconds=args.deadline, sample_digest=hashlib.sha256(canonical([c["id"] for c in changes])).hexdigest()
+    )
+    write_json(args.record, result)
+    return 0 if result["status"] == "ok" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/unseen/proof.py

@@ -57,7 +57,7 @@ def replay_proof(roots):
 def engine_proof(root, baseline):
     current = records(root)
     old = {digest([r["repo"], r["pin"]]): r for r in records(baseline) if "repo" in r and "pin" in r}
-    current = [r for r in current if "repo" in r and "pin" in r]
+    current = [r for r in current if "repo" in r and "pin" in r and not all(u.get("status") == "unsupported" for u in r["units"])]
     if len(current) != 2:
         raise ValueError("engine proof requires exactly two pins")
     evidence = []

benchmarks/unseen/replay.py

@@ -6,6 +6,7 @@ import argparse
 import hashlib
 import io
 import json
+import math
 import subprocess
 import tarfile
 import tomllib
@@ -15,11 +16,26 @@ from benchmarks.ax.batch import AXLane, DockerLane, Item, Workload, dispatch, wr
 from benchmarks.unseen.score import coverage
 
 
-def make_item(change, arm, image):
+def make_item(change, arm, image, *, deadline_scale=1.0, lane="ax", deadline=None):
     if arm not in {"default", "long_deadline"}:
         raise ValueError("unknown replay arm")
     identity = hashlib.sha256(str(change["id"]).encode()).hexdigest()[:14]
-    spec = {**change, "id": identity, "hook_flags": [] if arm == "default" else ["--deadline", "300"], "image": image}
+    if not math.isfinite(deadline_scale) or deadline_scale <= 0:
+        raise ValueError("deadline scale must be finite and positive")
+    base = deadline if deadline is not None else 30 if arm == "default" else 300
+    applied = round(base * (1 if lane == "docker" else deadline_scale))
+    if applied < 1:
+        raise ValueError("scaled deadline must be at least one second")
+    spec = {
+        **change,
+        "id": identity,
+        "hook_flags": ["--deadline", str(applied)],
+        "image": image,
+        "base_deadline_seconds": base,
+        "deadline_seconds": applied,
+        "deadline_scale": deadline_scale,
+        "applied_deadline_scale": 1.0 if lane == "docker" else deadline_scale,
+    }
     data = json.dumps(spec, sort_keys=True).encode()
     digest = hashlib.sha256(data).hexdigest()
     return Item(
@@ -146,6 +162,11 @@ def main():
     parser.add_argument("--slice", default="1")
     parser.add_argument("--out", type=Path)
     parser.add_argument("--lane", choices=("ax", "kind", "docker", "mixed"), default="mixed")
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument("--deadline-scale", type=float, default=1.0)
+    group.add_argument("--calibration", type=Path)
+    parser.add_argument("--native-sample", type=int, default=0)
+    parser.add_argument("--docker-image-bytes", type=int, help="unpacked image size upper bound, required for VM pull")
     parser.add_argument("--ax-bin", default="ax")
     parser.add_argument("--kubeconfig", type=Path)
     parser.add_argument("--atespace", default="default")
@@ -157,7 +178,15 @@ def main():
     if not args.image:
         parser.error("--image digest is required")
     changes = load_slice(args.manifest, args.slice, args.private_manifest)
-    items = [make_item(change, args.arm, args.image) for change in changes]
+    factor = args.deadline_scale
+    if args.calibration:
+        calibration = json.loads(args.calibration.read_text())
+        if calibration.get("status") != "ok" or calibration.get("image") != args.image:
+            parser.error("calibration must be successful and match the replay image")
+        factor = calibration["quick_total"]["factor"]
+    if args.native_sample < 0 or (args.native_sample and args.lane not in {"ax", "kind"}):
+        parser.error("--native-sample requires an ax or kind cluster run and a positive K")
+    items = [make_item(change, args.arm, args.image, deadline_scale=factor, lane=args.lane) for change in changes]
     if len({i.id for i in items}) != len(items):
         parser.error("duplicate change identity")
     args.out = args.out or Path.home() / "ousast-results/unseen" / args.pool / args.slice / args.arm
@@ -166,27 +195,52 @@ def main():
         return 0
     if args.out.resolve().is_relative_to(Path(__file__).resolve().parents[2]):
         parser.error("raw replay output must be outside the repository")
+    failed = run_replays(changes, args, args.out, args.lane, deadline_scale=factor)
+    if args.native_sample:
+        failed |= run_replays(sample_changes(changes, args.native_sample), args, args.out / "native", "docker", deadline_scale=factor)
+    return 2 if failed else 0
+
+
+def sample_changes(changes, count):
+    if count < 1:
+        raise ValueError("sample size must be positive")
+    if len({str(c["id"]) for c in changes}) != len(changes):
+        raise ValueError("duplicate change identity")
+    return sorted(changes, key=lambda c: hashlib.sha256(str(c["id"]).encode()).digest())[:count]
+
+
+def run_replays(changes, args, out, lane, *, deadline_scale=1.0, deadline=None):
+    if out.resolve().is_relative_to(Path(__file__).resolve().parents[2]):
+        raise ValueError("raw replay output must be outside the repository")
+    items = [make_item(c, args.arm, args.image, deadline_scale=deadline_scale, lane=lane, deadline=deadline) for c in changes]
     write_json(
-        args.out / "inputs.json",
+        out / "inputs.json",
         [{"item_id": item.id, "change": change, "input_digest": item.input_digest} for item, change in zip(items, changes, strict=True)],
     )
-    workload = Workload(args.image, frozenset({"SPEC_URL", "RESULT_URL"}), 600 if args.arm == "default" else 900, validate_result)
+    maximum = max(
-    lanes = ("ax", "ax", "docker") if args.lane == "mixed" else (args.lane,) * (2 if args.lane in {"ax", "kind"} else 1)
+        max(json.loads(i.inputs["SPEC_URL"])["deadline_seconds"], json.loads(i.inputs["SPEC_URL"])["base_deadline_seconds"]) for i in items
+    )
+    workload = Workload(args.image, frozenset({"SPEC_URL", "RESULT_URL"}), maximum + 600, validate_result)
+    lanes = ("ax", "ax", "docker") if lane == "mixed" else (lane,) * (2 if lane in {"ax", "kind"} else 1)
+    original = {item.id: change for item, change in zip(items, changes, strict=True)}
     executors = {}
-    for lane in set(lanes):
+    for name in set(lanes):
-        transport = (DockerLane if lane == "docker" else AXLane)(args, workload)
+        transport = (DockerLane if name == "docker" else AXLane)(args, workload)
 
-        def execute(item, output, attempt, transport=transport):
+        def execute(item, output, attempt, transport=transport, name=name):
-            record = {**json.loads(item.inputs["SPEC_URL"]), **transport(item, output, attempt)}
+            spec = original[item.id]
+            # Re-render on placement, including VM fallback: native is always unscaled.
+            placed = make_item(spec, args.arm, args.image, deadline_scale=deadline_scale, lane=name, deadline=deadline)
+            placed.id = item.id
+            record = {**json.loads(placed.inputs["SPEC_URL"]), "lane": name, **transport(placed, output, attempt)}
             if (output / "result.dat").exists():
-                # Durable archive lives separately from attempt staging objects, which are deleted.
+                key = f"unseen/{placed.input_digest}/{name}/result.dat"
-                key = f"unseen/{item.input_digest}/result.dat"
                 transport.store._put(key, (output / "result.dat").read_bytes())
                 record["result_object"] = key
             return record
 
-        executors[lane] = execute
+        executors[name] = execute
-    return 2 if dispatch(items, workload, args.out, executors, lanes=lanes) else 0
+    return dispatch(items, workload, out, executors, lanes=lanes)
 
 
 if __name__ == "__main__":

benchmarks/unseen/replay_entry.py

@@ -52,7 +52,8 @@ def validate_spec(spec):
         raise ReplayFailure("repository must use public DNS")
     if any(not re.fullmatch("[0-9a-f]{40,64}", spec[key]) for key in ("base", "head")):
         raise ReplayFailure("invalid revision")
-    if spec.get("hook_flags", []) not in ([], ["--deadline", "300"]):
+    flags = spec.get("hook_flags", [])
+    if flags and (len(flags) != 2 or flags[0] != "--deadline" or not re.fullmatch(r"[1-9][0-9]*", str(flags[1]))):
         raise ReplayFailure("unsupported hook flags")
 
 
@@ -92,6 +93,8 @@ def main(*, work_root=Path("/tmp"), runner=subprocess.run, heap_profile="cluster
         spec = json.loads(spec_path.read_text())
         instrument["image"] = spec.get("image")
         validate_spec(spec)
+        instrument["deadline_seconds"] = int(spec["hook_flags"][1]) if spec.get("hook_flags") else 30
+        instrument["deadline_scale"] = spec.get("deadline_scale", 1.0)
         case = work_root / "case"
         case.mkdir()
         git = ["git", "-c", "core.hooksPath=/dev/null", "-c", "protocol.file.allow=never", "-C", str(case)]
@@ -148,7 +151,9 @@ def main(*, work_root=Path("/tmp"), runner=subprocess.run, heap_profile="cluster
             *spec.get("hook_flags", []),
         ]
         before = clock()
-        completed = runner(command, capture_output=True, text=True, timeout=330 if spec.get("hook_flags") else 60, env=env)
+        completed = runner(
+            command, capture_output=True, text=True, timeout=int(spec["hook_flags"][1]) + 30 if spec.get("hook_flags") else 60, env=env
+        )
         instrument["timings"][step] = clock() - before
         instrument["hook_exit"] = completed.returncode
         (output / "push.txt").write_text(completed.stdout)

benchmarks/unseen/score.py

@@ -199,12 +199,59 @@ def score(records, *, seed=0):
     return {
         **metrics(usable),
         "total": len(records),
+        "deadlines": [
+            {"lane": lane, "deadline_seconds": deadline, "deadline_scale": factor, "applied_deadline_scale": applied, "count": count}
+            for (lane, deadline, factor, applied), count in sorted(
+                Counter(
+                    (r.get("lane", "unknown"), r.get("deadline_seconds"), r.get("deadline_scale"), r.get("applied_deadline_scale"))
+                    for r in records
+                ).items(),
+                key=lambda pair: str(pair[0]),
+            )
+        ],
         "coverage": {state: counts[state] for state in STATES},
         "breakdowns": breakdown,
         "confirmed": metrics([r for r in usable if r.get("label_check", r.get("confirmed")) in ("confirmed", True)]),
     }
 
 
+def agreement(cluster, native):
+    """Missing, failed or wrong-lane records are unknown, never matching quiet pushes."""
+    cc = {r["id"]: r for r in cluster}
+    if len(cc) != len(cluster) or len({r["id"] for r in native}) != len(native):
+        raise ValueError("duplicate agreement identities")
+    pairs, unknown, disagreements = [], [], []
+    for n in sorted(native, key=lambda r: r["id"]):
+        c = cc.get(n["id"])
+        if (
+            c is None
+            or c.get("lane") not in {"ax", "kind"}
+            or n.get("lane") != "docker"
+            or c.get("image") != n.get("image")
+            or c.get("base_deadline_seconds") != n.get("base_deadline_seconds")
+            or coverage(c) == "instrument_failure"
+            or coverage(n) == "instrument_failure"
+        ):
+            unknown.append(n["id"])
+            continue
+        cluster_flag, native_flag = false_alarm(c), false_alarm(n)
+        same = cluster_flag == native_flag
+        pairs.append({"change_id": n["id"], "cluster_flagged": cluster_flag, "native_flagged": native_flag, "same": same})
+        if not same:
+            disagreements.append(n["id"])
+    report = rate([p["same"] for p in pairs])
+    eligible = bool(pairs) and not unknown and report["wilson95"][0] >= 0.95
+    return {
+        **report,
+        "requested": len(native),
+        "pairs": pairs,
+        "unknown": unknown,
+        "disagreements": disagreements,
+        "baseline_eligible": eligible,
+        "flagged": not eligible,
+    }
+
+
 def budgets(rows, *, thresholds=None, binary=False):
     ordinary = [r for r in rows if r["kind"] == "ordinary"]
     vulnerable = [r for r in rows if r["kind"] != "ordinary"]
@@ -295,12 +342,20 @@ def main():
     parser.add_argument("records", type=Path, help="JSON array of joined change/instrument/push records")
     parser.add_argument("--out", type=Path, required=True)
     parser.add_argument("--against", type=Path, help="production-arm records for a paired adoption gate")
+    parser.add_argument("--native", type=Path, help="native sample root; auto-detected under replay root")
     parser.add_argument("--slice-id")
     parser.add_argument("--informed", action="store_true")
     args = parser.parse_args()
     current = load_records(args.records)
     result = score(current)
+    native = args.native or (args.records / "native" if args.records.is_dir() and (args.records / "native").exists() else None)
+    if native:
+        result["native_agreement"] = agreement(current, load_records(native))
+        result["baseline_eligible"] = result["native_agreement"]["baseline_eligible"]
+        result["flagged"] = result["native_agreement"]["flagged"]
     result["unseen"] = gate(load_records(args.against) if args.against else [], current, slice_id=args.slice_id, informed=args.informed)
+    if native and result["flagged"]:
+        result["unseen"] = {"gate": "not_run", "reason": "native_agreement"}
     args.out.write_text(json.dumps(result, sort_keys=True, indent=2) + "\n")
 
 

tests/test_engine_trace.py

@@ -1719,3 +1719,104 @@ def test_worker_session_question_deadline_keeps_markers_and_does_not_replay(tmp_
     assert backend.timed_out and backend.asked == ["a"]
     assert not backend.answers and not backend.session_fallback
     assert closed == [True]
+
+
+@pytest.mark.parametrize("keep_sources", [False, True])
+@pytest.mark.parametrize("outcome", ["success", "failure", "timeout", "fetch-failure", "fetch-timeout", "export-failure"])
+def test_finished_pin_source_lifetime(tmp_path, monkeypatch, keep_sources, outcome):
+    inputs, case = pair_inputs(tmp_path)
+    inputs.keep_sources = keep_sources
+    pin = trace.make_plan([unit(file=case.relpath)], tmp_path)[0]
+    clone = inputs.source_path(case.repo, "a" * 40, ["pkg"])
+    monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln"))
+
+    def git(command, **kwargs):
+        (clone / "pkg").mkdir(exist_ok=True)
+        (clone / case.relpath).write_text("def run(): pass\n")
+        if outcome == "fetch-failure":
+            raise subprocess.CalledProcessError(1, command)
+        if outcome == "fetch-timeout":
+            raise subprocess.TimeoutExpired(command, 300)
+        if outcome == "export-failure":
+            (clone / case.relpath).unlink()
+        return subprocess.CompletedProcess(command, 0, "a" * 40 + "\n", "")
+
+    class Dispatcher:
+        def execute(self, checkout, output, prepared, name):
+            assert (checkout / case.relpath).read_text() == "def run(): pass\n"
+            if outcome == "timeout":
+                return None, "worker"
+            if outcome == "failure":
+                raise OSError("executor failed")
+            trace.write_json(
+                output / "result.json",
+                {**prepared, "done": True, "units": [worker.unit_record(u, "asked-nothing") for u in prepared["units"]]},
+            )
+            return subprocess.CompletedProcess([], 0, "", ""), "worker"
+
+    monkeypatch.setattr(trace.subprocess, "run", git)
+    monkeypatch.setattr(trace, "prepare_questions", lambda *_: {"a": {}})
+    args = k8s_args(tmp_path, executor="ax", ax_dispatcher=Dispatcher())
+    assert trace.run_pin(pin, None, {}, inputs, tmp_path, args) is (outcome != "success")
+    assert clone.exists() is keep_sources
+
+
+def test_source_cache_evicts_oldest_before_fetch(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+
+    inputs, _ = pair_inputs(tmp_path)
+    inputs.source_cache_bytes = 8
+    cache = inputs.cache / "trace-sources"
+    for name, modified in (("oldest", 10), ("newest", 30), ("middle", 20)):
+        entry = cache / name
+        entry.mkdir(parents=True)
+        (entry / "data").write_bytes(b"123456")
+        trace.os.utime(entry, (modified, modified))
+    # Other caches and symlink targets must remain untouched.
+    other = inputs.cache / "repos"
+    other.mkdir()
+    (other / "data").write_bytes(b"keep")
+    (cache / "link").symlink_to(other, target_is_directory=True)
+    monkeypatch.setattr(trace.shutil, "disk_usage", lambda path: SimpleNamespace(free=2 * 1024**3))
+    commands = []
+
+    def git(command, **kwargs):
+        assert not (cache / "oldest").exists()
+        assert not (cache / "middle").exists()
+        assert (cache / "newest/data").read_bytes() == b"123456"
+        assert (other / "data").read_bytes() == b"keep"
+        commands.append(command)
+        return subprocess.CompletedProcess(command, 0, "a" * 40 + "\n", "")
+
+    monkeypatch.setattr(trace.subprocess, "run", git)
+    inputs.fetch_source("owner/repo", "a" * 40, ["pkg"])
+    assert any("fetch" in command for command in commands)
+
+
+def test_source_fetch_refuses_low_disk_before_git(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+
+    inputs, _ = pair_inputs(tmp_path)
+    monkeypatch.setattr(trace.shutil, "disk_usage", lambda path: SimpleNamespace(free=int(1.5 * 1024**3) - 1))
+    monkeypatch.setattr(trace.subprocess, "run", lambda *a, **kw: pytest.fail("low disk must refuse before git"))
+    with pytest.raises(OSError, match="refusing source fetch: .*at least 1.5 GB required"):
+        inputs.fetch_source("owner/repo", "a" * 40, ["pkg"])
+    assert not inputs.source_path("owner/repo", "a" * 40, ["pkg"]).exists()
+
+
+@pytest.mark.parametrize("flags,keep,cap", [([], False, 500), (["--keep-sources", "--source-cache-mb", "7"], True, 7)])
+def test_source_cache_cli_options(tmp_path, monkeypatch, flags, keep, cap):
+    original = trace.Inputs
+    created = []
+
+    def inputs(*args, **kwargs):
+        result = original(*args, **kwargs)
+        created.append(result)
+        return result
+
+    monkeypatch.setattr(trace, "Inputs", inputs)
+    monkeypatch.setattr(trace, "read_jsonl", lambda path: [])
+    monkeypatch.setattr(sys, "argv", [str(SCRIPT), "--dry-run", "--examples", "unused", "--cache", str(tmp_path), *flags])
+    assert trace.main() == 0
+    assert created[0].keep_sources is keep
+    assert created[0].source_cache_bytes == cap * 1024 * 1024

tests/test_unseen_calibrate.py

@@ -0,0 +1,235 @@
+"""Offline calibration, placement, agreement and Docker diagnostic controls."""
+
+import importlib
+import json
+import subprocess
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+c = importlib.import_module("benchmarks.unseen.calibrate")
+r = importlib.import_module("benchmarks.unseen.replay")
+s = importlib.import_module("benchmarks.unseen.score")
+b = importlib.import_module("benchmarks.ax.batch")
+
+
+def timing_record(identity="opaque", factor=1, lane="docker"):
+    return {
+        "id": identity,
+        "lane": lane,
+        "repo": "never-publish",
+        "kind": "ordinary",
+        "instrument": {
+            "head_verified": True,
+            "base_verified": True,
+            "changed_bytes": {"app.py": 10},
+            "hook_exit": 0,
+            "hook_bytes_read": 10,
+            "timings": {"pre-push": 60 * factor},
+        },
+        "push": {
+            "timings": {
+                f"comparison_0_{k}_seconds": v * factor
+                for k, v in (("resolution", 1), ("provenance", 2), ("snapshot", 4), ("preparation", 6), ("quick", 1))
+            },
+            "quick_tier": [{"bytes_read": 10, "findings": []}],
+            "scans": [{"scan": {"build_seconds": 20 * factor, "query_seconds": 30 * factor}}],
+        },
+    }
+
+
+def test_calibration_median_bootstrap_and_privacy():
+    result = c.slowdown([1, 10, 100], [2, 20, 200])
+    assert result == {"n": 3, "factor": 2, "bootstrap95": [2, 2]}
+    varied = c.slowdown([1] * 5, [1, 2, 3, 4, 5])
+    assert varied["factor"] == 3
+    assert varied["bootstrap95"] == [1, 5]
+    assert varied == c.slowdown([1] * 5, [1, 2, 3, 4, 5])
+    record = c.calibration_record([timing_record()], [timing_record(factor=3, lane="ax")])
+    assert record["quick_total"]["factor"] == 3
+    assert record["pairs"][0]["native"]["quick_total"] == 10  # Snapshot is nested, not added twice.
+    assert record["engine_build"]["factor"] == record["engine_queries"]["factor"] == 3
+    assert "never-publish" not in json.dumps(record)
+    with pytest.raises(ValueError):
+        c.slowdown([0], [2])
+    bad = timing_record()
+    bad["push"]["snapshots"] = [{"boundaries": [{"reason": "deadline_exhausted"}]}]
+    with pytest.raises(ValueError, match="deadline"):
+        c.stage_times(bad)
+    bad = timing_record()
+    del bad["push"]["timings"]["comparison_0_quick_seconds"]
+    with pytest.raises(ValueError, match="missing"):
+        c.stage_times(bad)
+
+
+@pytest.mark.parametrize(
+    "lane,arm,expected",
+    [
+        ("ax", "default", 75),
+        ("ax", "long_deadline", 750),
+        ("docker", "default", 30),
+        ("docker", "long_deadline", 300),
+    ],
+)
+def test_scaled_task_contract(lane, arm, expected):
+    item = r.make_item({"id": "a"}, arm, "image", deadline_scale=2.5, lane=lane)
+    spec = json.loads(item.inputs["SPEC_URL"])
+    assert spec["hook_flags"] == ["--deadline", str(expected)]
+    assert spec["deadline_seconds"] == expected and spec["deadline_scale"] == 2.5
+    entry = importlib.import_module("benchmarks.unseen.replay_entry")
+    entry.validate_spec({**spec, "repository_url": "https://public.example/test", "base": "a" * 40, "head": "b" * 40})
+
+
+def test_native_sample_and_agreement():
+    native = [timing_record(str(i)) for i in range(100)]
+    cluster = [timing_record(str(i), lane="ax") for i in range(100)]
+    assert r.sample_changes(native, 20) == r.sample_changes(list(reversed(native)), 20)
+    report = s.agreement(cluster, native)
+    assert report["rate"] == 1 and report["baseline_eligible"]
+    assert s.agreement(cluster[:20], native[:20])["flagged"]  # Perfect small sample is insufficient.
+    cluster[0]["push"]["quick_tier"][0]["findings"] = [{"cwe": "CWE-89"}]
+    report = s.agreement(cluster, native)
+    assert report["disagreements"] == ["0"] and report["rate"] == 0.99
+    assert report["flagged"]
+    cluster[0]["instrument"] = {}
+    report = s.agreement(cluster, native)
+    assert report["unknown"] == ["0"] and report["flagged"]
+    assert s.agreement([], [])["flagged"]
+
+
+def test_docker_error_capture_and_preflight(tmp_path, monkeypatch):
+    calls = []
+    monkeypatch.setenv("TEST_SECRET", "hidden-value")
+    lane = object.__new__(b.DockerLane)
+    lane.args = SimpleNamespace(docker_image_bytes=100)
+    lane.workload = SimpleNamespace(image="pinned-image")
+    lane._prepared = False
+
+    def runner(command, **kwargs):
+        calls.append(command)
+        if command[:2] == ["docker", "info"]:
+            return subprocess.CompletedProcess(command, 0, "/docker-data\n", "")
+        if command[0] == "df":
+            return subprocess.CompletedProcess(command, 0, "Avail\n2000000000\n", "")
+        return subprocess.CompletedProcess(command, 125, "", "pull denied hidden-value https://host.example/?signature=private\nlast error")
+
+    lane.runner = runner
+    record = lane(None, tmp_path, 0)
+    assert record["status"] == "instrument_failure"
+    assert record["docker_errors"][0]["exit_code"] == 125
+    assert record["docker_errors"][0]["stderr"].endswith("last error")
+    assert "hidden-value" not in json.dumps(record) and "private" not in json.dumps(record)
+    assert [cmd[0] for cmd in calls] == ["docker", "df", "docker"]
+    assert calls[-1] == ["docker", "pull", "pinned-image"]
+    assert (tmp_path / "docker-error.json").exists()
+    lane.args.docker_image_bytes = 2000000000
+    calls.clear()
+    record = lane(None, tmp_path, 1)
+    assert "refusing Docker pull" in record["docker_errors"][0]["stderr"]
+    assert len(calls) == 2
+
+
+def test_docker_launch_error_survives_transport(tmp_path, monkeypatch):
+    monkeypatch.setenv("S3_ENDPOINT", "https://store.example")
+
+    class Store:
+        def _put(self, *args):
+            pass
+
+        def _delete(self, *args):
+            pass
+
+        def presign_get(self, *args):
+            return "https://store.example/secret"
+
+        presign_put = presign_get
+
+    image = "registry/image@sha256:" + "a" * 64
+    lane = b.DockerLane(
+        SimpleNamespace(atespace="default"),
+        b.Workload(image, frozenset({"SPEC_URL", "RESULT_URL"}), 60, None),
+        store=Store(),
+        runner=lambda command, **kw: subprocess.CompletedProcess(command, 0 if command[1] == "rm" else 125, "", "runtime missing"),
+    )
+    lane._prepared = True
+    item = r.make_item({"id": "x"}, "default", image)
+    record = lane(item, tmp_path, 0)
+    assert record["docker_errors"] == [{"step": "run", "exit_code": 125, "stderr": "runtime missing"}]
+    assert record["status"] == "instrument_failure"
+
+
+def test_placement_scales_cluster_but_not_vm_fallback(tmp_path, monkeypatch):
+    observed = []
+
+    class Transport:
+        def __init__(self, args, workload):
+            assert workload.deadline >= 750 + 600
+
+        def __call__(self, item, output, attempt):
+            spec = json.loads(item.inputs["SPEC_URL"])
+            observed.append(spec)
+            return {"status": "quick_only"}
+
+    def dispatch(items, workload, out, executors, **kwargs):
+        for lane in ("ax", "docker"):
+            record = executors[lane](items[0], out, 0)
+            assert record["lane"] == lane
+            assert record["deadline_seconds"] == (750 if lane == "ax" else 300)
+        return False
+
+    monkeypatch.setattr(r, "AXLane", Transport)
+    monkeypatch.setattr(r, "DockerLane", Transport)
+    monkeypatch.setattr(r, "dispatch", dispatch)
+    args = SimpleNamespace(arm="long_deadline", image="test-image")
+    assert not r.run_replays([{"id": "opaque"}], args, tmp_path, "mixed", deadline_scale=2.5)
+    assert observed[0]["id"] == observed[1]["id"]
+    assert observed[0]["hook_flags"] == ["--deadline", "750"]
+    assert observed[1]["hook_flags"] == ["--deadline", "300"]
+
+
+def test_calibration_cli_writes_anonymous_record(tmp_path, monkeypatch):
+    manifest, record = tmp_path / "manifest.json", tmp_path / "record.json"
+    changes = [{"id": str(i), "repo": "do-not-publish"} for i in range(25)]
+    monkeypatch.setattr(c, "load_slice", lambda *args: changes)
+    calls = []
+
+    def replay(selected, args, out, lane, **kwargs):
+        calls.append((selected, lane, kwargs))
+        return False
+
+    monkeypatch.setattr(c, "run_replays", replay)
+    monkeypatch.setattr(
+        c,
+        "load_records",
+        lambda path: [
+            timing_record(str(i), factor=3 if path.name == "cluster" else 1, lane="ax" if path.name == "cluster" else "docker")
+            for i in range(20)
+        ],
+    )
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "calibrate",
+            "--manifest",
+            str(manifest),
+            "--image",
+            "digest",
+            "--docker-image-bytes",
+            "100",
+            "--out",
+            str(tmp_path / "raw"),
+            "--record",
+            str(record),
+        ],
+    )
+    assert c.main() == 0
+    assert len(calls[0][0]) == 20 and calls[0][0] == calls[1][0]
+    assert [call[1] for call in calls] == ["docker", "ax"]
+    assert all(call[2]["deadline"] == 600 for call in calls)
+    payload = record.read_text()
+    assert "do-not-publish" not in payload and "never-publish" not in payload
+    assert json.loads(payload)["quick_total"]["factor"] == 3

tests/test_unseen_proof.py

@@ -54,3 +54,33 @@ def test_engine_proof_counts_differences_without_names(tmp_path):
         (baseline / f"{i}.json").write_text(json.dumps(record))
     proof = p.engine_proof(current, baseline)
     assert proof["differences"] == 0 and "synthetic" not in json.dumps(proof)
+
+
+@pytest.mark.parametrize("executed", [2, 3])
+def test_engine_proof_counts_only_executed_records(tmp_path, executed):
+    current, baseline = tmp_path / "current", tmp_path / "baseline"
+    current.mkdir()
+    baseline.mkdir()
+    (current / "progress.json").write_text(json.dumps({"completed": executed}))
+    for i in range(executed + 4):
+        record = {
+            "repo": "synthetic",
+            "pin": str(i),
+            "done": True,
+            "seconds": 20 if i < executed else 0,
+            "units": [{"unit": "unsupported", "status": "unsupported"}],
+        }
+        if i < executed:
+            record["units"].append(
+                {"unit": "executed", "status": "path", "instrument": {"bytes": 20, "jvm": [{"seconds": 8, "success": True}]}}
+            )
+            (baseline / f"{i}.json").write_text(json.dumps(record))
+        (current / f"{i}.json").write_text(json.dumps(record))
+
+    if executed == 3:
+        with pytest.raises(ValueError, match="engine proof requires exactly two pins"):
+            p.engine_proof(current, baseline)
+    else:
+        proof = p.engine_proof(current, baseline)
+        assert len(proof["pins"]) == 2
+        assert proof["differences"] == 0