diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md
index 826fd5b74..a7f00920c 100644
--- a/.claude/CLAUDE.md
+++ b/.claude/CLAUDE.md
@@ -834,14 +834,17 @@ then on "Custom firecracker not found", because a hand-rolled runner called
 | `bench-chromium-hostcdp` | `bench-chromium-request-build` | host-container CDP baseline, no VM; `COMPARISON_LABEL=` (default `standalone`), `CPU_BUDGET=` (default `unlimited`), `CPUS=` (requires `CPU_BUDGET=vm-matched`), `BENCH_RESOLVE_ALL_TO=` |
 | `bench-chromium-fault` | `build` + `setup-default` | `FAULT_OUT=` (required), `FAULT_ARGS=` |
 
-- **verify/diag/run must never gain a `build` dependency.** reqbench.sh seals
-  fcvm, fc-agent and its six sources into a hash-bound runtime bundle; the
-  run refuses a golden recorded under a different bundle hash. A rebuild, an
+- **verify/diag/run and bench-chromium-scale must never gain a `build` dependency.**
+  reqbench.sh seals fcvm, fc-agent, the request harness's six sources and the
+  scale harness's five into a hash-bound runtime bundle; the run, and the scale
+  run through `reqbench.sh scale`, refuse a golden recorded under a different
+  bundle hash. A rebuild, an
   edit to a sealed file, or a new commit between golden and run invalidates
   the chain: the run also refuses a golden whose recorded source_revision is
   not the current git HEAD. Regolden instead of working around the seal.
   Sealed set: reqbench.{sh,py}, reqanalyze.py, cdpdrive.py, render.py,
-  wddrive.py, fcvm, fc-agent. The Makefile and test files are not sealed:
+  wddrive.py, reqscale.py, reqscale_analyze.py, faulttrace.bt, guardexec.py,
+  guardsupervise.py, fcvm, fc-agent. The Makefile and test files are not sealed:
   uncommitted edits to them do not affect a golden or a run in progress, but
   committing them before the run does.
 - Hugepage goldens are part of the snapshot identity: distinct tag, e.g.
diff --git a/Makefile b/Makefile
index 463bbe244..030e9dccc 100644
--- a/Makefile
+++ b/Makefile
@@ -1184,8 +1184,15 @@ test-chromium-fault:
 #
 # Every cell and every publication gate is required: an accidental benchmark is
 # worse than no benchmark, so there are no defaults to fall back on.
-bench-chromium-scale: private SHELL := $(TARGET_LEASE_SHELL)
-bench-chromium-scale: build
+# The memory server's mode for the UFFD backend, and the host control's page.
+# SCALE_URL may be a comma-separated list (the corpus); then SCALE_CONTROL_URL
+# names the one page the control renders, and SCALE_CONTROL_RESOLVE_ALL_TO maps
+# the control's names to a replay server.
+SCALE_UFFD_MODE ?= copy
+SCALE_UFFD_PREFETCH ?= on
+SCALE_CONTROL_URL ?=
+SCALE_CONTROL_RESOLVE_ALL_TO ?=
+bench-chromium-scale:
 	@test -n "$(SCALE_RATES)" || (echo "ERROR: SCALE_RATES required (for example 2,4,8)"; exit 1)
 	@test -n "$(SCALE_BURSTS)" || (echo "ERROR: SCALE_BURSTS required (must be at least 5)"; exit 1)
 	@test -n "$(SCALE_SEED)" || (echo "ERROR: SCALE_SEED required"; exit 1)
@@ -1199,8 +1206,7 @@ bench-chromium-scale: build
 	@test -n "$(SCALE_MAX_LAUNCH_LAG_MS)" || (echo "ERROR: SCALE_MAX_LAUNCH_LAG_MS required"; exit 1)
 	@test -n "$(SCALE_MAX_CONTROL_DRIFT_PCT)" || (echo "ERROR: SCALE_MAX_CONTROL_DRIFT_PCT required"; exit 1)
 	@echo "==> Running open-loop Chromium request scalability benchmark..."
-	sudo -E env RUST_LOG=fcvm=debug python3 bench/chromium/reqscale.py \
-		--fcvm ./target/release/fcvm --snapshot-tag "$(SCALE_TAG)" \
+	TAG="$(SCALE_TAG)" RESULTS="$(RESULTS)" bash bench/chromium/reqbench.sh scale \
 		--url "$(SCALE_URL)" --rates "$(SCALE_RATES)" \
 		--bursts "$(SCALE_BURSTS)" --control-chromium "$(SCALE_CONTROL_CHROMIUM)" \
 		--max-offered-rps-error-pct "$(SCALE_MAX_OFFERED_ERROR_PCT)" \
@@ -1209,6 +1215,9 @@ bench-chromium-scale: build
 		--max-p95-launch-lag-ms "$(SCALE_MAX_LAUNCH_LAG_MS)" \
 		--max-control-median-drift-pct "$(SCALE_MAX_CONTROL_DRIFT_PCT)" \
 		--seed "$(SCALE_SEED)" \
+		--uffd-mode "$(SCALE_UFFD_MODE)" --uffd-prefetch "$(SCALE_UFFD_PREFETCH)" \
+		$(if $(SCALE_CONTROL_URL),--control-url "$(SCALE_CONTROL_URL)") \
+		$(if $(SCALE_CONTROL_RESOLVE_ALL_TO),--control-resolve-all-to "$(SCALE_CONTROL_RESOLVE_ALL_TO)") \
 		--out-dir "$(SCALE_OUT)" $(SCALE_TRACE_ARGS)
 
 analyze-chromium-scale:
diff --git a/bench/chromium/AGENTS.md b/bench/chromium/AGENTS.md
index 19aa188e7..21fe81d14 100644
--- a/bench/chromium/AGENTS.md
+++ b/bench/chromium/AGENTS.md
@@ -199,8 +199,9 @@ hostname URLs and records no resolver (`guest_dns` null, no
 2026-08-16 corpus record, whose guests resolved through ambient DNS. A marker writer must first open the results directory and
 take an exclusive `flock` on that directory descriptor, then remove completion
 records and atomically rename the marker while holding the lock.
-`campaign_summary.py` and `compare.py` hold shared locks on every run directory
-whose authorization they consume, across validation and publication. This
+`campaign_summary.py`, `compare.py` and `reqscale_analyze.py` hold shared locks
+on every run directory whose authorization they consume, across validation and
+publication. This
 makes a withdrawal order before the publication (which refuses it) or after it
 (which invalidates it), with no unchecked interval between the last marker read
 and publication. The marker is tracked (`!results/**/WITHDRAWN` in
diff --git a/bench/chromium/corpus_campaign.sh b/bench/chromium/corpus_campaign.sh
index e9bba6f7b..d8c7a78f8 100755
--- a/bench/chromium/corpus_campaign.sh
+++ b/bench/chromium/corpus_campaign.sh
@@ -77,9 +77,41 @@ done
 # DIAG_ONLY=1 ends the campaign after golden, verify and diag: no measured
 # run, no analysis. For the throwaway golden round.
 DIAG_ONLY="${DIAG_ONLY:-0}"
+# MEASURE=scale replaces the serial reqbench run with the open-loop throughput
+# benchmark (make bench-chromium-scale) over the same corpus, inside the same
+# golden, verify and DNS-evidence bracket. Its rates, bursts, seed, gates and
+# control Chromium come from the caller's SCALE_* variables; the corpus URLs,
+# tag, memory-server mode and output directory come from this campaign. The
+# host control renders SCALE_CONTROL_URL (default: the corpus's first URL)
+# with every name mapped to this host's replay server.
+MEASURE="${MEASURE:-reqbench}"
+case "$MEASURE" in
+    reqbench) ;;
+    scale)
+        [ "${ENGINE:-chromium}" = chromium ] \
+            || { echo "BLOCKED: MEASURE=scale drives Chromium only (ENGINE=$ENGINE)" >&2; exit 2; }
+        [ "$BACKEND" = uffd ] \
+            || { echo "BLOCKED: MEASURE=scale pairs FILE and UFFD restores itself; BACKEND=$BACKEND has no meaning there" >&2; exit 2; }
+        ;;
+    *) echo "BLOCKED: MEASURE must be reqbench or scale (got '$MEASURE')" >&2; exit 2 ;;
+esac
 STAMP="$(date +%Y%m%d-%H%M%S)"
 RESULTS="${RESULTS:-$REPO/bench/chromium/results/reqbench-$STAMP-corpus}"
 LOGDIR="${LOGDIR:-/tmp/corpus-campaign-$STAMP}"
+# The analysis the DNS evidence names its run from: the serial run's own, or
+# the scale run's first analysis (the one that may publish comes after the
+# evidence exists).
+if [ "$MEASURE" = scale ]; then
+    # Each attempt measures into its own directory beside the campaign's
+    # evidence, which the analyzer reads from the run directory's parent.
+    # Nothing removes one: a retry never meets reqscale.py's refusal to reuse
+    # a directory, a withdrawn attempt keeps its WITHDRAWN marker, and a second
+    # campaign on the same RESULTS cannot touch this one's run.
+    SCALE_DIR="$RESULTS/scale-$STAMP-$$"
+    ANALYSIS_JSON="$SCALE_DIR/analysis-pre-evidence.json"
+else
+    ANALYSIS_JSON="$RESULTS/analysis.json"
+fi
 mkdir -p "$LOGDIR"
 # Created here, not left to reqbench: the replay server's logs and the resolver
 # evidence below are written into it before any reqbench phase runs.
@@ -484,6 +516,7 @@ write_dns_evidence() {
     # of this box names nothing; campaign_summary resolves each name beside
     # the record, and verify_file_sha256 is keyed by the same names.
     local verdict="$1" reason="${2:-}" owner_log="dns-owner.log"
+    local ANALYSIS_JSON="${ANALYSIS_JSON:-$RESULTS/analysis.json}"
     local log="$RESULTS/$owner_log" out="$RESULTS/dns-evidence.json"
     local samples=0 first_mismatch="" before=false after=false after_state f
     local sampler_alive=false load_stats load_samples=0 load_max=null
@@ -496,13 +529,13 @@ write_dns_evidence() {
     # ago; reqbench.py stamps that id into every record, so it survives a
     # re-analysis, which a hash of analysis.json would not.
     local run_id=""
-    if [ -f "$RESULTS/analysis.json" ]; then
+    if [ -f "$ANALYSIS_JSON" ]; then
         run_id=$(jq -r 'select((.run_id | type) == "string" and (.run_id | length) > 0)
-                        | .run_id' "$RESULTS/analysis.json" 2>/dev/null) || run_id=""
+                        | .run_id' "$ANALYSIS_JSON" 2>/dev/null) || run_id=""
     fi
     if [ -z "$run_id" ]; then
         verdict=unclean
-        reason="${reason:-$RESULTS/analysis.json names no run_id, so this evidence is bound to no measured run}"
+        reason="${reason:-$ANALYSIS_JSON names no run_id, so this evidence is bound to no measured run}"
     fi
     [ "$DNSMASQ_WAS_ACTIVE" = yes ] && before=true
     [ "$SAMPLER_ALIVE_AT_STOP" = yes ] && sampler_alive=true
@@ -915,14 +948,33 @@ say "box quiet (1-min load $load1)"
 run_verify before-run || campaign_fail "verify (before-run) failed after the settle wait"
 
 # --- measured run ----------------------------------------------------------
-say "measured run: $REPS reps/arm, warmup $WARMUP, arms $ARMS, $BACKEND/$UFFD_MODE prefetch=$UFFD_PREFETCH"
-start_dns_sampler
 run_rc=0
-TAG="$TAG" URL="$URLS" BACKEND="$BACKEND" UFFD_MODE="$UFFD_MODE" \
-    UFFD_PREFETCH="$UFFD_PREFETCH" ARMS="$ARMS" REPS="$REPS" WARMUP="$WARMUP" \
-    STALL_MAX_MS="$STALL_MAX_MS" RESULTS="$RESULTS" ENGINE="$ENGINE" \
-    make -C "$REPO" "$(engine_target run)" 2>&1 | tee "$LOGDIR/run.log" || run_rc=$?
-stop_dns_sampler
+if [ "$MEASURE" = scale ]; then
+    say "measured run: open-loop scale over the corpus, rates ${SCALE_RATES:-unset}, $UFFD_MODE prefetch=$UFFD_PREFETCH"
+    start_dns_sampler
+    SCALE_URL="$URLS" SCALE_TAG="$TAG" SCALE_OUT="$SCALE_DIR" \
+        SCALE_UFFD_MODE="$UFFD_MODE" SCALE_UFFD_PREFETCH="$UFFD_PREFETCH" \
+        SCALE_CONTROL_URL="${SCALE_CONTROL_URL:-${URLS%%,*}}" \
+        SCALE_CONTROL_RESOLVE_ALL_TO="${SCALE_CONTROL_RESOLVE_ALL_TO:-127.0.0.1}" \
+        make -C "$REPO" bench-chromium-scale 2>&1 | tee "$LOGDIR/run.log" || run_rc=$?
+    stop_dns_sampler
+    # The analyzer writes the run id the DNS evidence below binds to, and
+    # withholds publication until that evidence exists beside the run, so it
+    # runs once here and once more after the evidence is written.
+    if [ "$run_rc" -eq 0 ]; then
+        make -C "$REPO" analyze-chromium-scale SCALE_RUN_DIR="$SCALE_DIR" \
+            SCALE_ANALYSIS_JSON="$ANALYSIS_JSON" 2>&1 | tee "$LOGDIR/analyze.log" \
+            || run_rc=$?
+    fi
+else
+    say "measured run: $REPS reps/arm, warmup $WARMUP, arms $ARMS, $BACKEND/$UFFD_MODE prefetch=$UFFD_PREFETCH"
+    start_dns_sampler
+    TAG="$TAG" URL="$URLS" BACKEND="$BACKEND" UFFD_MODE="$UFFD_MODE" \
+        UFFD_PREFETCH="$UFFD_PREFETCH" ARMS="$ARMS" REPS="$REPS" WARMUP="$WARMUP" \
+        STALL_MAX_MS="$STALL_MAX_MS" RESULTS="$RESULTS" ENGINE="$ENGINE" \
+        make -C "$REPO" "$(engine_target run)" 2>&1 | tee "$LOGDIR/run.log" || run_rc=$?
+    stop_dns_sampler
+fi
 
 # The run's own exit is not the verdict: a run that measured cleanly against
 # the wrong resolver is worse than one that failed, so the after-run bracket
@@ -937,6 +989,17 @@ if [ "$run_rc" -ne 0 ] || [ "$after_rc" -ne 0 ] || [ "$verdict" != clean ]; then
     echo "FAILED: measured run exit $run_rc, after-run verify exit $after_rc, dns verdict $verdict" >&2
     exit 1
 fi
+if [ "$MEASURE" = scale ]; then
+    make -C "$REPO" analyze-chromium-scale SCALE_RUN_DIR="$SCALE_DIR" \
+        SCALE_ANALYSIS_JSON="$SCALE_DIR/analysis.json" 2>&1 \
+        | tee -a "$LOGDIR/analyze.log" \
+        || { echo "FAILED: the scale analysis after the DNS evidence did not run" >&2; exit 1; }
+    jq -e '.publishable == true' "$SCALE_DIR/analysis.json" >/dev/null || {
+        echo "FAILED: scale analysis not publishable: $(jq -r '.publication_blocked_by' "$SCALE_DIR/analysis.json" 2>&1)" >&2
+        exit 1
+    }
+    say "scale analysis: $SCALE_DIR/analysis.json (publishable)"
+fi
 
 say "records: $RESULTS"
 say "logs:    $LOGDIR"
diff --git a/bench/chromium/reqbench.py b/bench/chromium/reqbench.py
index 33d08d0cd..84a3729ff 100755
--- a/bench/chromium/reqbench.py
+++ b/bench/chromium/reqbench.py
@@ -1419,6 +1419,38 @@ def sha256_file(path: str) -> str:
     return h.hexdigest()
 
 
+def require_golden_runtime(snapshot_name: str, snapshot: dict, fcvm: str) -> dict:
+    """This process's staged runtime, refused unless it created the golden.
+
+    The runtime is the reqbench.sh bundle this process executes from: the
+    fcvm it runs, the hash of the bundle's MANIFEST.sha256, and the revision
+    it was staged at. reqbench.sh golden records the same three values for
+    the bundle that created the snapshot (snapshot_generation returns them).
+    """
+    runtime_bundle = os.environ.get("REQBENCH_RUNTIME_BUNDLE", "")
+    manifest_path = os.path.join(runtime_bundle, "MANIFEST.sha256")
+    if not runtime_bundle or os.path.realpath(runtime_bundle) != os.path.realpath(HERE):
+        raise RuntimeError(
+            f"{HERE} is not reqbench.sh's staged runtime bundle "
+            f"(REQBENCH_RUNTIME_BUNDLE={runtime_bundle!r}); measured phases run from it"
+        )
+    if not os.path.isfile(manifest_path):
+        raise RuntimeError(f"staged runtime manifest is missing: {manifest_path}")
+    runtime = {
+        "creator_fcvm_sha256": sha256_file(fcvm),
+        "creator_runtime_bundle_sha256": sha256_file(manifest_path),
+        "source_revision": os.environ.get("REQBENCH_SOURCE_REVISION", ""),
+    }
+    for field, current in runtime.items():
+        if snapshot[field] != current:
+            raise RuntimeError(
+                f"snapshot {snapshot_name} was created with {field}="
+                f"{snapshot[field]!r}, current runtime is {current!r}; recreate "
+                "the golden with this staged runtime"
+            )
+    return runtime
+
+
 # Every script that defines one request sample, for either engine. Must stay
 # equal to reqbench.sh's staged runtime sources minus the binaries and the
 # analyzer (which reads samples but defines none); asserted by
@@ -4398,27 +4430,13 @@ def main_with_resources(resources: ExitStack) -> int:
         p.error(str(error))
 
     args.fcvm = os.path.abspath(args.fcvm)
-    runtime_bundle = os.environ.get("REQBENCH_RUNTIME_BUNDLE", "")
-    manifest_path = os.path.join(runtime_bundle, "MANIFEST.sha256")
-    if os.path.realpath(runtime_bundle) != os.path.realpath(HERE):
-        p.error("reqbench.py must execute from reqbench.sh's staged runtime bundle")
-    if not os.path.isfile(manifest_path):
-        p.error(f"staged runtime manifest is missing: {manifest_path}")
-    current_fcvm_sha256 = sha256_file(args.fcvm)
-    current_runtime_bundle_sha256 = sha256_file(manifest_path)
-    current_source_revision = os.environ.get("REQBENCH_SOURCE_REVISION", "")
-    creator_identity = {
-        "creator_fcvm_sha256": current_fcvm_sha256,
-        "creator_runtime_bundle_sha256": current_runtime_bundle_sha256,
-        "source_revision": current_source_revision,
-    }
-    for field, current in creator_identity.items():
-        if snapshot[field] != current:
-            p.error(
-                f"snapshot {snapshot_name} was created with {field}="
-                f"{snapshot[field]!r}, current runtime is {current!r}; recreate "
-                "the golden with this staged runtime"
-            )
+    try:
+        runtime = require_golden_runtime(snapshot_name, snapshot, args.fcvm)
+    except RuntimeError as error:
+        p.error(str(error))
+    current_fcvm_sha256 = runtime["creator_fcvm_sha256"]
+    current_runtime_bundle_sha256 = runtime["creator_runtime_bundle_sha256"]
+    current_source_revision = runtime["source_revision"]
     os.makedirs(args.out_dir, exist_ok=True)
     arms = [a.strip() for a in args.arms.split(",") if a.strip()]
     allowed_arms = allowed_arms_for_engine(args.engine)
diff --git a/bench/chromium/reqbench.sh b/bench/chromium/reqbench.sh
index 822f4181e..cc9f4ef05 100755
--- a/bench/chromium/reqbench.sh
+++ b/bench/chromium/reqbench.sh
@@ -124,18 +124,18 @@ if [ "${BASH_SOURCE[0]}" = "$0" ] && [ "${REQBENCH_STAGED:-0}" != 1 ]; then
     source_revision_before=$(git -C "$REPO" rev-parse HEAD)
     mkdir -p "$RESULTS/runtime"
     stage_dir=$(mktemp -d "$RESULTS/runtime/.stage.XXXXXX")
-    for source in reqbench.sh reqbench.py reqanalyze.py cdpdrive.py render.py wddrive.py; do
+    # The request harness, then the scale harness (reqscale.HARNESS_SOURCES).
+    runtime_sources=(reqbench.sh reqbench.py reqanalyze.py cdpdrive.py render.py wddrive.py
+        reqscale.py reqscale_analyze.py faulttrace.bt guardexec.py guardsupervise.py)
+    for source in "${runtime_sources[@]}"; do
         cp --reflink=auto "$HERE/$source" "$stage_dir/$source"
     done
     cp --reflink=auto "$FC_AGENT" "$stage_dir/fc-agent"
     cp --reflink=auto "$FCVM" "$stage_dir/fcvm"
-    chmod 0555 "$stage_dir/fcvm" "$stage_dir/fc-agent" "$stage_dir/reqbench.sh" \
-        "$stage_dir/reqbench.py" "$stage_dir/reqanalyze.py" \
-        "$stage_dir/cdpdrive.py" "$stage_dir/render.py" "$stage_dir/wddrive.py"
     (
         cd "$stage_dir"
-        sha256sum fcvm fc-agent reqbench.sh reqbench.py reqanalyze.py cdpdrive.py render.py wddrive.py \
-            > MANIFEST.sha256
+        chmod 0555 fcvm fc-agent "${runtime_sources[@]}"
+        sha256sum fcvm fc-agent "${runtime_sources[@]}" > MANIFEST.sha256
     )
     bundle_hash=$(sha256sum "$stage_dir/MANIFEST.sha256" | cut -d' ' -f1)
     bundle_dir="$RESULTS/runtime/$bundle_hash"
@@ -2044,6 +2044,22 @@ cmd_diag() {
     return $rc
 }
 
+# The open-loop scale benchmark, from this bundle like every measured phase:
+# reqscale.py refuses a golden created by another bundle, fcvm or revision.
+cmd_scale() {
+    local rc=0
+    # reqscale.py hands the run directory it created back to this user itself
+    # (hand_back_to_invoker); a directory that already existed is refused.
+    sudo -E env RUST_LOG=fcvm=debug \
+        REQBENCH_RUNTIME_BUNDLE="${REQBENCH_RUNTIME_BUNDLE:-}" \
+        REQBENCH_SOURCE_REPO="$REPO" \
+        REQBENCH_SOURCE_REVISION="${REQBENCH_SOURCE_REVISION:-}" \
+        python3 "$HERE/reqscale.py" --snapshot-tag "$TAG" \
+        --data-root "$DATA_ROOT" --state-dir "$STATE_DIR" "$@" || rc=$?
+    verify_runtime_bundle || rc=1
+    return $rc
+}
+
 # Only dispatch when EXECUTED. Sourcing the file makes its helpers unit-testable
 # (see ReqbenchShell in test_reqbench.py) instead of reachable only through a
 # whole phase.
@@ -2055,6 +2071,7 @@ if [ "${BASH_SOURCE[0]}" = "$0" ]; then
         verify) cmd_verify ;;
         run)    cmd_run ;;
         diag)   cmd_diag ;;
+        scale)  shift; cmd_scale "$@" ;;
         all)
             # The chain's own build/golden/verify phases are the load the run
             # gate reads a minute later; default the settle window so a cold
@@ -2062,6 +2079,6 @@ if [ "${BASH_SOURCE[0]}" = "$0" ]; then
             # SETTLE_WAIT_SECS wins.
             export SETTLE_WAIT_SECS="${SETTLE_WAIT_SECS:-120}"
             cmd_build; cmd_golden; cmd_verify; cmd_run ;;
-        *) echo "usage: $0 {build|golden|verify|run|diag|all}" >&2; exit 2 ;;
+        *) echo "usage: $0 {build|golden|verify|run|diag|scale|all}" >&2; exit 2 ;;
     esac
 fi
diff --git a/bench/chromium/reqscale.py b/bench/chromium/reqscale.py
index 89022f43b..67bf977f6 100644
--- a/bench/chromium/reqscale.py
+++ b/bench/chromium/reqscale.py
@@ -31,12 +31,14 @@
 from __future__ import annotations
 
 import argparse
+import ipaddress
 import dataclasses
 import datetime as dt
 import fcntl
 import hashlib
 import json
 import math
+import atexit
 import os
 import platform
 import queue
@@ -55,7 +57,6 @@
 from typing import Callable, Iterable, Optional
 
 HERE = os.path.dirname(os.path.abspath(__file__))
-REPO = os.path.abspath(os.path.join(HERE, "..", ".."))
 sys.path.insert(0, HERE)
 
 import reqbench  # noqa: E402
@@ -270,6 +271,8 @@ class ScheduleConfig:
     score_seconds: float = SCORE_SECONDS
     trace_rate: Optional[float] = None
     trace_pairs: int = 0
+    # The pages requests render, cycled by pair index (request_url()).
+    urls: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -371,6 +374,8 @@ def _validate_schedule_config(config: ScheduleConfig) -> None:
         raise ValueError("scored bursts must use exactly a 15s ramp and 60s score")
     if not config.rates:
         raise ValueError("at least one target rate is required")
+    if not config.urls or not all(isinstance(url, str) and url for url in config.urls):
+        raise ValueError("the schedule needs the page or pages requests render")
     if len(set(config.rates)) != len(config.rates):
         raise ValueError("target rates must be unique")
     for rate in config.rates:
@@ -383,6 +388,14 @@ def _validate_schedule_config(config: ScheduleConfig) -> None:
                 f"rate {rate:g} supplies fewer than 200 scored requests per backend "
                 f"across {config.scored_bursts} bursts"
             )
+        # A corpus is only the same workload at every rate when each scored
+        # window renders every page equally often: whole cycles of the list.
+        score_pairs = _planned_count(rate, config.score_seconds)
+        if len(config.urls) > 1 and score_pairs % len(config.urls):
+            raise ValueError(
+                f"rate {rate:g} gives {score_pairs} scored pairs per burst, not a whole "
+                f"number of cycles of the {len(config.urls)}-page corpus"
+            )
     criteria = dataclasses.asdict(config.criteria)
     for field in (
         "max_offered_rps_error_pct", "max_p95_launch_lag_ms",
@@ -570,6 +583,8 @@ def build_schedule(config: ScheduleConfig, run_id: str) -> dict:
         "run_id": run_id,
         "seed": config.seed,
         "rates": list(config.rates),
+        "urls": list(config.urls),
+        "url_selection": "urls[pair_index % len(urls)]; both halves of a pair render the same page",
         "cells": [
             {
                 "cell_id": f"{backend}:r{format(rate, '.12g')}",
@@ -2358,12 +2373,17 @@ def sha256_file(path: str) -> str:
     return digest.hexdigest()
 
 
+# Every file a scale run reads or executes from its bundle; reqbench.sh stages
+# each of them (harness_hash_covers_every_staged_request_script).
+HARNESS_SOURCES = (
+    "reqscale.py", "reqscale_analyze.py", "reqbench.py", "cdpdrive.py",
+    "render.py", "faulttrace.bt", "guardexec.py", "guardsupervise.py",
+)
+
+
 def harness_sha256() -> str:
     digest = hashlib.sha256(b"fcvm-reqscale-harness-v1\0")
-    for name in (
-        "reqscale.py", "reqscale_analyze.py", "reqbench.py", "cdpdrive.py",
-        "render.py", "faulttrace.bt", "guardexec.py", "guardsupervise.py",
-    ):
+    for name in HARNESS_SOURCES:
         encoded = name.encode()
         digest.update(len(encoded).to_bytes(4, "big"))
         digest.update(encoded)
@@ -2420,6 +2440,37 @@ def quiet_host_snapshot(
     }
 
 
+def require_serve_mode(effective: str, requested: str) -> None:
+    """The memory server must run the mode the run asked for; fcvm coerces
+    minor to copy for NV2 snapshots, which would mislabel every UFFD cell."""
+    if effective != requested:
+        raise MeasurementInvalid(
+            f"the memory server runs {effective} mode, not the requested {requested}")
+
+
+def file_restore_refusal(snapshot_config: dict, environ) -> Optional[str]:
+    """Why this run's FILE arm would not be a file-backed restore, or None.
+
+    `fcvm snapshot run` without a memory server serves a hugepage snapshot, an
+    NV2 snapshot, or any restore under FCVM_FORCE_UFFD through an implicit
+    in-process UFFD server (src/commands/snapshot.rs direct_restore_memory),
+    so the FILE-versus-UFFD comparison would be UFFD against UFFD. NV2 comes
+    from the snapshot's kernel profile; any recorded profile is refused rather
+    than resolving whether it enables NV2.
+    """
+    metadata = snapshot_config.get("metadata")
+    if not isinstance(metadata, dict):
+        return "the snapshot config has no metadata to check its restore path against"
+    if metadata.get("hugepages"):
+        return "the snapshot uses hugepages, which fcvm restores through an implicit UFFD server"
+    if metadata.get("kernel_profile"):
+        return (f"the snapshot records kernel profile {metadata['kernel_profile']!r}, "
+                "which can restore through an implicit UFFD server (NV2)")
+    if "FCVM_FORCE_UFFD" in environ:
+        return "FCVM_FORCE_UFFD is set, which restores through an implicit UFFD server"
+    return None
+
+
 def snapshot_identity(data_root: str, snapshot_tag: str) -> dict:
     """Durable identity of the exact snapshot generation and runtime shape."""
     snapshots_root = os.path.realpath(os.path.join(data_root, "snapshots"))
@@ -2536,14 +2587,33 @@ def __exit__(self, _type, _value, _traceback):
         self.close()
 
 
+def golden_runtime(args, identity: dict) -> tuple[dict, dict]:
+    """The leased golden's creator runtime, and this run's, which must be it.
+
+    The serial run's check (reqbench.require_golden_runtime) against the
+    golden's reqbench-provenance.json, read under the same shared lease.
+    """
+    try:
+        golden = reqbench.snapshot_generation(args.data_root, args.snapshot_tag)
+        if (golden["generation_id"], golden["config_sha256"]) != (
+                identity["generation_id"], identity["config_sha256"]):
+            raise RuntimeError(
+                "the golden's reqbench provenance names another generation than the leased one")
+        runtime = reqbench.require_golden_runtime(args.snapshot_tag, golden, args.fcvm)
+    except RuntimeError as error:
+        raise MeasurementInvalid(str(error)) from error
+    return {field: golden[field] for field in runtime}, runtime
+
+
 def collect_provenance(args, schedule: dict, snapshot: dict) -> dict:
-    fcvm = os.path.abspath(args.fcvm)
-    if not os.path.isfile(fcvm):
-        raise MeasurementInvalid(f"fcvm binary is missing: {fcvm}")
-    revision = _command(["git", "-C", REPO, "rev-parse", "HEAD"])
-    if not re.fullmatch(r"[0-9a-f]{40}", revision):
-        raise MeasurementInvalid(f"git returned invalid revision {revision!r}")
-    dirty = _command(["git", "-C", REPO, "status", "--porcelain=v1", "--untracked-files=all"])
+    fcvm = args.fcvm
+    revision = _command(["git", "-C", args.source_repo, "rev-parse", "HEAD"])
+    if revision != args.runtime["source_revision"]:
+        raise MeasurementInvalid(
+            f"the source repository is at {revision!r}, not the revision "
+            f"{args.runtime['source_revision']!r} this runtime was staged from")
+    dirty = _command(["git", "-C", args.source_repo, "status", "--porcelain=v1",
+                      "--untracked-files=all"])
     if dirty:
         raise MeasurementInvalid(
             "benchmark source tree is dirty; commit the harness before measuring"
@@ -2556,7 +2626,7 @@ def collect_provenance(args, schedule: dict, snapshot: dict) -> dict:
             quiet,
         )
     return {
-        "schema": "fcvm.chromium.reqscale.provenance.v1",
+        "schema": "fcvm.chromium.reqscale.provenance.v2",
         "run_id": schedule["run_id"],
         "created_at": dt.datetime.now(dt.timezone.utc).isoformat(),
         "argv": list(sys.argv),
@@ -2566,13 +2636,16 @@ def collect_provenance(args, schedule: dict, snapshot: dict) -> dict:
         "source_status_sha256": hashlib.sha256(dirty.encode()).hexdigest(),
         "harness_sha256": harness_sha256(),
         "fcvm_path": fcvm,
-        "fcvm_sha256": sha256_file(fcvm),
+        "fcvm_sha256": args.runtime["creator_fcvm_sha256"],
+        "runtime_bundle_sha256": args.runtime["creator_runtime_bundle_sha256"],
+        "golden_creator": args.golden_creator,
         "fcvm_version": _command([fcvm, "--version"]),
         "host_control": {
             "chromium_path": args.control_chromium,
             "chromium_sha256": sha256_file(args.control_chromium),
             "chromium_version": _command([args.control_chromium, "--version"]),
             "url": args.control_url,
+            "resolve_all_to": args.control_resolve_all_to or None,
             "interval_seconds": CONTROL_INTERVAL_SECONDS,
             "timeout_seconds": args.control_timeout,
         },
@@ -2805,6 +2878,14 @@ def __init__(self, args, audit: CgroupAudit, cgroup_path: str, log_dir: str):
         self.state = None
         self.record = None
 
+    def command(self) -> list[str]:
+        """The memory server's argv, in the requested fault mode."""
+        return [
+            self.args.fcvm, "snapshot", "serve", self.args.snapshot_tag,
+            "--uffd-mode", getattr(self.args, "uffd_mode", "copy"),
+            "--uffd-prefetch", getattr(self.args, "uffd_prefetch", "on"),
+        ]
+
     def start(self) -> int:
         watch = reqbench.DirWatch(self.args.state_dir)
         try:
@@ -2815,10 +2896,7 @@ def start(self) -> int:
             }
             self.log_stream = open(self.log_path, "xb")
             self.proc = subprocess.Popen(
-                guarded_command(
-                    self.cgroup_path,
-                    [self.args.fcvm, "snapshot", "serve", self.args.snapshot_tag],
-                ),
+                guarded_command(self.cgroup_path, self.command()),
                 stdout=self.log_stream,
                 stderr=self.log_stream,
                 stdin=subprocess.DEVNULL,
@@ -2862,6 +2940,8 @@ def start(self) -> int:
                                 )
                             if config.get("uffd_mode") not in ("copy", "minor"):
                                 raise MeasurementInvalid(f"serve state {path} has invalid UFFD mode")
+                            require_serve_mode(
+                                config["uffd_mode"], getattr(self.args, "uffd_mode", "copy"))
                             self.state_path, self.state = path, state
                             self.record = {
                                 "schema": RECORD_SCHEMA,
@@ -2871,6 +2951,7 @@ def start(self) -> int:
                                 "pid_start_time_ticks": identity.start_time_ticks,
                                 "state_path": os.path.relpath(path, self.args.data_root),
                                 "uffd_mode": config["uffd_mode"],
+                                "uffd_prefetch": getattr(self.args, "uffd_prefetch", "on"),
                                 "snapshot_tag": self.args.snapshot_tag,
                                 "snapshot_generation_id": self.args.snapshot_identity[
                                     "generation_id"
@@ -3006,6 +3087,34 @@ def drive(self) -> dict:
             raise MeasurementInvalid("host control Chromium is not running")
         return cdpdrive.drive(self._drive_args(self.args.control_timeout))
 
+    def command(self) -> list[str]:
+        """The control Chromium's argv; profile_dir must already exist."""
+        return [
+            self.args.control_chromium,
+            "--headless=new",
+            "--no-sandbox",
+            "--remote-debugging-address=127.0.0.1",
+            "--remote-debugging-port=0",
+            "--remote-allow-origins=*",
+            "--ignore-certificate-errors",
+            "--disable-gpu",
+            "--disable-dev-shm-usage",
+            "--window-size=1280,800",
+            "--hide-scrollbars",
+            "--mute-audio",
+            "--no-first-run",
+            "--no-default-browser-check",
+            "--disable-background-networking",
+            "--disable-breakpad",
+            "--disable-component-update",
+            *(
+                [f"--host-resolver-rules=MAP * {self.args.control_resolve_all_to}"]
+                if self.args.control_resolve_all_to else []
+            ),
+            f"--user-data-dir={self.profile_dir}",
+            "about:blank",
+        ]
+
     def start(self) -> dict:
         self.profile_dir = tempfile.mkdtemp(
             prefix=f"fcvm-reqscale-control-{self.args.run_id}-",
@@ -3015,29 +3124,8 @@ def start(self) -> dict:
         pidfd = None
         try:
             self.log_stream = open(self.log_path, "xb")
-            command = [
-                self.args.control_chromium,
-                "--headless=new",
-                "--no-sandbox",
-                "--remote-debugging-address=127.0.0.1",
-                "--remote-debugging-port=0",
-                "--remote-allow-origins=*",
-                "--ignore-certificate-errors",
-                "--disable-gpu",
-                "--disable-dev-shm-usage",
-                "--window-size=1280,800",
-                "--hide-scrollbars",
-                "--mute-audio",
-                "--no-first-run",
-                "--no-default-browser-check",
-                "--disable-background-networking",
-                "--disable-breakpad",
-                "--disable-component-update",
-                f"--user-data-dir={self.profile_dir}",
-                "about:blank",
-            ]
             self.proc = subprocess.Popen(
-                supervised_command(self.cgroup_path, command),
+                supervised_command(self.cgroup_path, self.command()),
                 stdout=self.log_stream,
                 stderr=self.log_stream,
                 stdin=subprocess.DEVNULL,
@@ -3318,13 +3406,20 @@ def stop(self) -> dict:
         }
 
 
+def request_url(args, context: RequestContext) -> str:
+    """The page a request renders: the URL list cycled by pair, so the FILE and
+    UFFD halves of a pair always render the same page."""
+    urls = getattr(args, "urls", None) or [args.url]
+    return urls[context.pair_index % len(urls)]
+
+
 def _request_args(
     args, context: RequestContext, serve_pid: int, log_dir: str, probe,
 ) -> argparse.Namespace:
     return argparse.Namespace(
         serve_pid=serve_pid if context.backend == "uffd" else 0,
         snapshot_tag=args.snapshot_tag if context.backend == "file" else "",
-        url=args.url,
+        url=request_url(args, context),
         format=args.format,
         quality=args.quality,
         cdp_port=args.cdp_port,
@@ -3353,6 +3448,7 @@ def request(context: RequestContext) -> dict:
         # cgroup accounting.
         record = reqbench.run_cdp_request(request_args, rep, fast=True)
         record.update(
+            url=request_url(args, context),
             schema=RECORD_SCHEMA,
             kind="request",
             run_id=args.run_id,
@@ -3445,8 +3541,29 @@ def _interburst_membership(
     return members
 
 
+def hand_back_to_invoker(path: str) -> None:
+    """Give the run directory this process created to the user who ran sudo.
+
+    reqscale.py runs as root, and the campaign's analysis and evidence steps,
+    which write into the run directory, run as the invoking user. execute()
+    registers this only after its own os.mkdir of the directory succeeded, so a
+    directory that already existed is refused and never handed over.
+    """
+    if os.geteuid() != 0:
+        return
+    try:
+        uid, gid = int(os.environ["SUDO_UID"]), int(os.environ["SUDO_GID"])
+    except (KeyError, ValueError):
+        return
+    for top, dirs, files in os.walk(path, topdown=False):
+        for name in dirs + files:
+            os.lchown(os.path.join(top, name), uid, gid)
+    os.lchown(path, uid, gid)
+
+
 def execute(args, schedule: dict, provenance: dict) -> int:
     os.mkdir(args.out_dir)
+    atexit.register(hand_back_to_invoker, args.out_dir)
     _fsync_directory(os.path.dirname(args.out_dir))
     log_dir = os.path.join(args.out_dir, "logs")
     trace_dir = os.path.join(args.out_dir, "fault-trace")
@@ -3715,12 +3832,20 @@ def main() -> int:
     parser.add_argument("--max-control-median-drift-pct", type=float, required=True)
     parser.add_argument("--run-id", default="")
     parser.add_argument("--out-dir", required=True)
-    parser.add_argument("--fcvm", default=os.path.join(REPO, "target", "release", "fcvm"))
     parser.add_argument("--data-root", default="/mnt/fcvm-btrfs")
     parser.add_argument("--state-dir", default="")
     parser.add_argument("--cgroup-root", default="/sys/fs/cgroup")
     parser.add_argument("--control-chromium", default="chromium")
-    parser.add_argument("--control-url", default="")
+    parser.add_argument("--uffd-mode", choices=("copy", "minor"), default="copy",
+                        help="the memory server's fault mode for the UFFD backend")
+    parser.add_argument("--uffd-prefetch", choices=("on", "off"), default="on",
+                        help="the memory server's working-set replay")
+    parser.add_argument("--control-url", default="",
+                        help="the one page the host control Chromium renders; "
+                             "required when --url is a list")
+    parser.add_argument("--control-resolve-all-to", default="",
+                        help="map every host name the control Chromium resolves to "
+                             "this IPv4 address, e.g. a corpus replay server")
     parser.add_argument("--control-timeout", type=float, default=8.0)
     parser.add_argument("--control-tmp-root", default="/tmp")
     parser.add_argument("--format", choices=("png", "jpeg"), default="jpeg")
@@ -3744,11 +3869,21 @@ def main() -> int:
     args = parser.parse_args()
 
     args.run_id = args.run_id or uuid.uuid4().hex
-    args.fcvm = os.path.abspath(args.fcvm)
     args.data_root = os.path.abspath(args.data_root)
     args.state_dir = args.state_dir or os.path.join(args.data_root, "state")
     args.out_dir = os.path.abspath(args.out_dir)
+    args.urls = reqbench.parse_urls(args.url)
+    if len(args.urls) > 1 and not args.control_url:
+        parser.error("--control-url is required with a URL list: the host control "
+                     "renders one fixed page")
+    if len(reqbench.parse_urls(args.control_url or args.url)) != 1:
+        parser.error("--control-url must name exactly one URL")
     args.control_url = args.control_url or args.url
+    if args.control_resolve_all_to:
+        try:
+            ipaddress.IPv4Address(args.control_resolve_all_to)
+        except ValueError:
+            parser.error("--control-resolve-all-to must be an IPv4 address")
     args.control_tmp_root = os.path.abspath(args.control_tmp_root)
     try:
         _validate_snapshot_tag(args.snapshot_tag)
@@ -3771,6 +3906,7 @@ def main() -> int:
         parser.error("trace options require --trace-faults")
 
     config = ScheduleConfig(
+        urls=tuple(args.urls),
         rates=args.rates,
         scored_bursts=args.bursts,
         seed=args.seed,
@@ -3795,6 +3931,9 @@ def main() -> int:
         return 0
     if os.geteuid() != 0:
         parser.error("measured runs must be root so owned accounting cgroups can be created")
+    # The bundle reqbench.sh staged this file into: its fcvm is the one measured.
+    args.fcvm = os.path.join(HERE, "fcvm")
+    args.source_repo = os.environ.get("REQBENCH_SOURCE_REPO", "")
     resolved_chromium = (
         args.control_chromium if os.path.isabs(args.control_chromium)
         else shutil.which(args.control_chromium)
@@ -3813,6 +3952,14 @@ def main() -> int:
         with SnapshotGenerationLease(args.data_root, args.snapshot_tag) as lease:
             args.snapshot_generation_lease = lease
             args.snapshot_identity = dict(lease.identity)
+            config_path = os.path.join(
+                args.data_root, "snapshots", args.snapshot_tag, "config.json")
+            with open(config_path, "rb") as stream:
+                refusal = file_restore_refusal(
+                    strict_json_loads(stream.read(), config_path), os.environ)
+            if refusal:
+                raise MeasurementInvalid(f"refusing a FILE arm that is not one: {refusal}")
+            args.golden_creator, args.runtime = golden_runtime(args, lease.identity)
             provenance = collect_provenance(args, schedule, args.snapshot_identity)
             with TerminationFence():
                 return execute(args, schedule, provenance)
diff --git a/bench/chromium/reqscale_analyze.py b/bench/chromium/reqscale_analyze.py
index f1f5a2b1c..c0f1f6453 100644
--- a/bench/chromium/reqscale_analyze.py
+++ b/bench/chromium/reqscale_analyze.py
@@ -5,6 +5,7 @@
 
 import argparse
 import hashlib
+import ipaddress
 import json
 import math
 import os
@@ -17,6 +18,8 @@
 HERE = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, HERE)
 
+import campaign_summary  # noqa: E402
+import reqanalyze  # noqa: E402
 import reqscale  # noqa: E402
 
 
@@ -132,6 +135,56 @@ def _canonical_generation(value) -> str:
     return canonical
 
 
+def _null_or_ipv4(value) -> bool:
+    if value is None:
+        return True
+    try:
+        return isinstance(value, str) and str(ipaddress.IPv4Address(value)) == value
+    except ValueError:
+        return False
+
+
+def corpus_dns_gate(run_dir: str, schedule: dict, provenance: dict):
+    """None when publication needs no resolver evidence or has it, else why not.
+
+    A run needs resolver evidence when a URL's host is a name only a resolver
+    can answer (reqanalyze.url_needs_resolver, which treats a URL it cannot read
+    the way the browser does as needing one) or when the host control maps
+    names; IP literals and localhost resolve nothing, however many there are.
+    Whether each clone resolved those names through the replay server is
+    recorded only by the corpus campaign's DNS evidence beside this run
+    directory. Without a clean bundle that names this
+    run, a run against the wrong resolver is indistinguishable from a good one.
+    The bundle is held to campaign_summary's check, the one the index applies:
+    the run id, the :53 owner samples, the replay server's exit status, every
+    verify bracket and replay log against the sha256 recorded at the verdict,
+    and brackets that covered exactly this run's pages. The scale schedule
+    records no guest resolver, so the brackets are held to the one the first
+    of them names. A WITHDRAWN marker in the campaign directory, the results
+    directory the withdrawal rule in AGENTS.md governs, withdraws the run too.
+    """
+    corpus = provenance["host_control"].get("resolve_all_to") or any(
+        reqanalyze.url_needs_resolver(url) is not False for url in schedule["urls"])
+    if not corpus:
+        return None
+    campaign_dir = os.path.dirname(os.path.abspath(run_dir))
+    withdrawn = campaign_summary.withdrawal_errors([campaign_dir])
+    if withdrawn:
+        return f"the campaign is withdrawn: {withdrawn[0]}"
+    path = os.path.join(campaign_dir, "dns-evidence.json")
+    if not os.path.isfile(path):
+        return f"corpus run without the campaign's DNS evidence ({path})"
+    sources = campaign_summary.Sources(campaign_dir)
+    try:
+        evidence = sources.read_json(path)
+        campaign_summary.check_evidence(
+            campaign_dir, evidence, sources, None, list(schedule["urls"]), schedule["run_id"]
+        )
+    except campaign_summary.RunError as error:
+        return f"the campaign's DNS evidence does not hold: {error}"
+    return None
+
+
 def _validate_schedule(schedule: dict) -> reqscale.ScheduleConfig:
     if not isinstance(schedule, dict):
         raise AnalysisInvalid("schedule is not an object")
@@ -162,6 +215,7 @@ def _validate_schedule(schedule: dict) -> reqscale.ScheduleConfig:
             score_seconds=schedule["score_seconds"],
             trace_rate=schedule["trace_rate"],
             trace_pairs=schedule["trace_pairs"],
+            urls=tuple(schedule["urls"]),
         )
         rebuilt = reqscale.build_schedule(config, schedule["run_id"])
     except (
@@ -516,6 +570,7 @@ def _validate_requests(
                 "backend": planned.backend,
                 "segment": planned.segment,
                 "pair_index": planned.pair_index,
+                "url": schedule["urls"][planned.pair_index % len(schedule["urls"])],
                 "request_seed": planned.seed,
                 "population": spec.population,
                 "target_rps": spec.target_rps,
@@ -1197,19 +1252,33 @@ def _validate_provenance(provenance: dict, schedule: dict) -> tuple[str, str]:
         "source_revision", "source_dirty", "source_status_sha256",
         "harness_sha256", "fcvm_path", "fcvm_sha256", "fcvm_version",
         "host_control", "snapshot", "snapshot_generation_lease", "host",
-        "fault_trace",
+        "fault_trace", "runtime_bundle_sha256", "golden_creator",
     }
     if not isinstance(provenance, dict) or set(provenance) != required:
         raise AnalysisInvalid("provenance fields are incomplete or unknown")
-    if provenance.get("schema") != "fcvm.chromium.reqscale.provenance.v1":
+    if provenance.get("schema") != "fcvm.chromium.reqscale.provenance.v2":
         raise AnalysisInvalid("unsupported provenance schema")
     if provenance.get("run_id") != schedule["run_id"]:
         raise AnalysisInvalid("run identity differs between schedule and provenance")
     if provenance.get("source_dirty") is not False:
         raise AnalysisInvalid("measurement source tree was dirty")
     _require_hex(provenance.get("source_revision"), 40, "source revision")
-    for name in ("source_status_sha256", "harness_sha256", "fcvm_sha256"):
+    for name in ("source_status_sha256", "harness_sha256", "fcvm_sha256",
+                 "runtime_bundle_sha256"):
         _require_hex(provenance.get(name), 64, name)
+    measured_runtime = {
+        "creator_fcvm_sha256": provenance["fcvm_sha256"],
+        "creator_runtime_bundle_sha256": provenance["runtime_bundle_sha256"],
+        "source_revision": provenance["source_revision"],
+    }
+    golden = provenance.get("golden_creator")
+    if not isinstance(golden, dict) or set(golden) != set(measured_runtime):
+        raise AnalysisInvalid("provenance does not name the runtime that created its golden")
+    for field, measured in measured_runtime.items():
+        if golden[field] != measured:
+            raise AnalysisInvalid(
+                f"the golden was created with {field}={golden[field]!r}, the run "
+                f"measured with {measured!r}; it is not the golden's runtime")
     if not isinstance(provenance.get("created_at"), str) or not provenance["created_at"]:
         raise AnalysisInvalid("provenance has no creation time")
     if (
@@ -1269,7 +1338,7 @@ def _validate_provenance(provenance: dict, schedule: dict) -> tuple[str, str]:
         not isinstance(host_control, dict)
         or set(host_control) != {
             "chromium_path", "chromium_sha256", "chromium_version", "url",
-            "interval_seconds", "timeout_seconds",
+            "resolve_all_to", "interval_seconds", "timeout_seconds",
         }
         or host_control.get("interval_seconds") != reqscale.CONTROL_INTERVAL_SECONDS
         or not isinstance(host_control.get("chromium_path"), str)
@@ -1278,6 +1347,7 @@ def _validate_provenance(provenance: dict, schedule: dict) -> tuple[str, str]:
         or not host_control["chromium_version"]
         or not isinstance(host_control.get("url"), str)
         or not host_control["url"]
+        or not _null_or_ipv4(host_control.get("resolve_all_to"))
         or not 0 < _finite_number(
             host_control.get("timeout_seconds"), "host-control timeout", minimum=0,
         ) < reqscale.CONTROL_INTERVAL_SECONDS
@@ -1560,7 +1630,7 @@ def _validate_uffd_serve(
 ) -> None:
     required_fields = {
         "schema", "kind", "run_id", "pid", "pid_start_time_ticks",
-        "state_path", "uffd_mode", "snapshot_tag", "snapshot_generation_id",
+        "state_path", "uffd_mode", "uffd_prefetch", "snapshot_tag", "snapshot_generation_id",
         "snapshot_config_sha256",
     }
     if not isinstance(serve, dict) or set(serve) != required_fields:
@@ -1582,6 +1652,8 @@ def _validate_uffd_serve(
         raise AnalysisInvalid(f"UFFD serve is not bound to this snapshot generation: {mismatch}")
     if serve.get("uffd_mode") not in ("copy", "minor"):
         raise AnalysisInvalid("UFFD serve has an invalid memory mode")
+    if serve.get("uffd_prefetch") not in ("on", "off"):
+        raise AnalysisInvalid("UFFD serve has an invalid working-set prefetch setting")
     state_path = serve.get("state_path")
     if (
         not isinstance(state_path, str)
@@ -1916,11 +1988,17 @@ def analyze(run_dir: str) -> dict:
     else:
         trace_gate = {"enabled": False}
 
+    # A WITHDRAWN marker withdraws a scale run wherever it was measured.
+    withdrawn = campaign_summary.withdrawal_errors([run_dir])
+    block = withdrawn[0] if withdrawn else corpus_dns_gate(run_dir, schedule, provenance)
     return {
         "schema": ANALYSIS_SCHEMA,
         "run_id": run_id,
         "snapshot_generation_id": generation_id,
-        "publishable": True,
+        "publishable": block is None,
+        "publication_blocked_by": block,
+        "corpus": list(schedule["urls"]),
+        "uffd": {"mode": uffd_serve["uffd_mode"], "prefetch": uffd_serve["uffd_prefetch"]},
         # The report is the publication document, so the numbers that describe HOW the
         # run was scheduled, and on WHAT, have to come from the validated artifacts
         # rather than from prose written when the defaults happened to be these.
@@ -1938,6 +2016,7 @@ def analyze(run_dir: str) -> dict:
             "cpu_count": provenance["host"]["cpu_count"],
             "fcvm_version": provenance["fcvm_version"],
             "fcvm_sha256": provenance["fcvm_sha256"],
+            "runtime_bundle_sha256": provenance["runtime_bundle_sha256"],
             "source_revision": provenance["source_revision"],
             "source_dirty": provenance["source_dirty"],
             "chromium_version": provenance["host_control"]["chromium_version"],
@@ -1965,6 +2044,13 @@ def _format_seconds(value):
 
 
 def markdown_report(analysis: dict) -> str:
+    # The report says the run passed its checks, so an analysis that withholds
+    # publication gets none.
+    if analysis.get("publishable") is not True:
+        raise AnalysisInvalid(
+            "no report for an unpublishable analysis: "
+            f"{analysis.get('publication_blocked_by')}"
+        )
     sched = analysis["schedule"]
     prov = analysis["provenance"]
     warmup = sched["warmup_bursts"]
@@ -1981,7 +2067,8 @@ def markdown_report(analysis: dict) -> str:
         f"`{prov['source_revision'][:12]}`"
         f"{' with a DIRTY tree' if prov['source_dirty'] else ''}, driving "
         f"Chromium `{prov['chromium_version']}`. Full host and binary provenance is in "
-        f"`{prov['hostinfo']}`.",
+        f"`{prov['hostinfo']}`. The UFFD backend served in `{analysis['uffd']['mode']}` "
+        f"mode with working-set prefetch `{analysis['uffd']['prefetch']}`.",
         "",
         "FILE and UFFD were offered the same per-backend rate in one mixed stream. "
         "Each rate interval contained one request for each backend, separated by "
@@ -2076,33 +2163,47 @@ def main() -> int:
     args = parser.parse_args()
     if not args.json_out and not args.markdown_out:
         parser.error("at least one of --json-out or --markdown-out is required")
-    try:
-        analysis = analyze(os.path.abspath(args.run_dir))
-        if args.json_out:
-            reqscale.write_json_exclusive(os.path.abspath(args.json_out), analysis)
-        if args.markdown_out:
-            report_path = os.path.abspath(args.markdown_out)
-            directory = os.path.dirname(report_path)
-            os.makedirs(directory, exist_ok=True)
-            temp = os.path.join(
-                directory, f".{os.path.basename(report_path)}.{uuid.uuid4().hex}.tmp"
-            )
-            try:
-                fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
-                with os.fdopen(fd, "w") as stream:
-                    stream.write(markdown_report(analysis))
-                    stream.flush()
-                    os.fsync(stream.fileno())
-                os.link(temp, report_path)
-                reqscale._fsync_directory(directory)
-            finally:
+    run_dir = os.path.abspath(args.run_dir)
+    # The reader side of the WITHDRAWN protocol in AGENTS.md: shared locks on
+    # the run directory and the directory holding it (for a corpus run, the
+    # campaign whose marker and DNS evidence authorize it) from the first read
+    # until the outputs are installed. A withdrawal writer's exclusive lock
+    # then lands wholly before this analysis, which reads its marker, or after.
+    with campaign_summary.shared_run_directory_locks(
+            [run_dir, os.path.dirname(run_dir)]) as lock_errors:
+        try:
+            if lock_errors:
+                raise AnalysisInvalid("; ".join(lock_errors))
+            analysis = analyze(run_dir)
+            report = markdown_report(analysis) if args.markdown_out else None
+            moved = campaign_summary.locked_run_directory_errors(lock_errors.run_dirs)
+            if moved:
+                raise AnalysisInvalid("; ".join(moved))
+            if args.json_out:
+                reqscale.write_json_exclusive(os.path.abspath(args.json_out), analysis)
+            if report is not None:
+                report_path = os.path.abspath(args.markdown_out)
+                directory = os.path.dirname(report_path)
+                os.makedirs(directory, exist_ok=True)
+                temp = os.path.join(
+                    directory, f".{os.path.basename(report_path)}.{uuid.uuid4().hex}.tmp"
+                )
                 try:
-                    os.unlink(temp)
-                except FileNotFoundError:
-                    pass
-    except (AnalysisInvalid, reqscale.MeasurementInvalid, OSError, ValueError) as error:
-        print(f"{type(error).__name__}: {error}", file=sys.stderr)
-        return 4
+                    fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
+                    with os.fdopen(fd, "w") as stream:
+                        stream.write(report)
+                        stream.flush()
+                        os.fsync(stream.fileno())
+                    os.link(temp, report_path)
+                    reqscale._fsync_directory(directory)
+                finally:
+                    try:
+                        os.unlink(temp)
+                    except FileNotFoundError:
+                        pass
+        except (AnalysisInvalid, reqscale.MeasurementInvalid, OSError, ValueError) as error:
+            print(f"{type(error).__name__}: {error}", file=sys.stderr)
+            return 4
     return 0
 
 
diff --git a/bench/chromium/test_campaign.py b/bench/chromium/test_campaign.py
index df62ef1a5..1aa5c392d 100644
--- a/bench/chromium/test_campaign.py
+++ b/bench/chromium/test_campaign.py
@@ -495,6 +495,9 @@ class DnsBrackets(unittest.TestCase):
     FAKE_MAKE = """#!/bin/bash
 env > "$MAKE_ENV_DUMP"
 echo "$*" > "$MAKE_ARGV"
+for a in "$@"; do case "$a" in ""|-*|*=*|*/*) ;; *) target=$a; break ;; esac; done
+env > "$MAKE_ENV_DUMP.${target:-none}"
+echo "$*" >> "$MAKE_ARGV.all"
 [ -z "${MAKE_VERIFY_JSON:-}" ] || printf '%s\\n' "$MAKE_VERIFY_JSON" > "$RESULTS/verify-dns.json"
 [ -z "${MAKE_DIAG_JSON:-}" ] || { mkdir -p "$RESULTS/diag"; printf '%s\\n' "$MAKE_DIAG_JSON" > "$RESULTS/diag/summary.json"; }
 for qname in ${MAKE_DNS_QNAMES:-}; do
@@ -1140,6 +1143,62 @@ def test_the_run_sub_make_receives_the_stall_limit(self):
             with open(env["MAKE_ARGV"]) as handle:
                 self.assertIn("bench-chromium-request-run", handle.read())
 
+    def test_measure_scale_runs_the_open_loop_benchmark_over_the_corpus(self):
+        """MEASURE=scale replaces the serial run with bench-chromium-scale over
+        the same corpus, in the campaign's memory-server mode, with the host
+        control mapped to the replay server.
+
+        RED BEFORE THE FIX: there was no MEASURE knob and no scale branch.
+        """
+        body = campaign()
+        block = re.search(r'(if \[ "\$MEASURE" = scale \]; then\n    say "measured run: open-loop.*?\nfi\n)',
+                          body, re.S)
+        self.assertIsNotNone(block, "the campaign has no MEASURE=scale branch")
+        self.assertRegex(body, r'(?m)^MEASURE="\$\{MEASURE:-reqbench\}"$')
+        with tempfile.TemporaryDirectory() as tmp:
+            env, results = self._fakes(tmp)
+            script = ('set -euo pipefail\nsay() { :; }\n'
+                      f'URLS="{self._urls()}"\nBACKEND=uffd\nUFFD_MODE=minor\n'
+                      'UFFD_PREFETCH=on\nARMS=noop,cdp\nREPS=1\nWARMUP=1\n'
+                      'STALL_MAX_MS=15000\nMEASURE=scale\nTAG=cb-req-corpus\n'
+                      'SCALE_DIR="$RESULTS/scale-attempt"\nANALYSIS_JSON="$SCALE_DIR/analysis-pre-evidence.json"\n'
+                      f'{self._helpers()}\nrun_rc=0\n'
+                      f'{block.group(1)}\necho "run_rc=$run_rc"\n')
+            result = self._run(script, env)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            seen = self._make_env(dict(env, MAKE_ENV_DUMP=env["MAKE_ENV_DUMP"]
+                                       + ".bench-chromium-scale"))
+            self.assertEqual(seen.get("SCALE_URL"), self._urls())
+            self.assertEqual(seen.get("SCALE_TAG"), "cb-req-corpus")
+            self.assertEqual(seen.get("SCALE_OUT"), os.path.join(results, "scale-attempt"))
+            self.assertEqual(seen.get("SCALE_UFFD_MODE"), "minor")
+            self.assertEqual(seen.get("SCALE_UFFD_PREFETCH"), "on")
+            self.assertEqual(seen.get("SCALE_CONTROL_URL"), self._urls().split(",")[0])
+            self.assertEqual(seen.get("SCALE_CONTROL_RESOLVE_ALL_TO"), "127.0.0.1")
+            with open(env["MAKE_ARGV"] + ".all") as handle:
+                calls = handle.read().splitlines()
+            self.assertIn("bench-chromium-scale", calls[0])
+            self.assertFalse(any("bench-chromium-request-run" in c for c in calls))
+            # The run is analyzed before the DNS evidence, into the file the
+            # evidence names its run from (red before: the evidence read
+            # $RESULTS/analysis.json, which a scale run never writes).
+            self.assertEqual(len(calls), 2, calls)
+            self.assertIn("analyze-chromium-scale", calls[1])
+            self.assertIn(f"SCALE_RUN_DIR={results}/scale-attempt", calls[1])
+            self.assertIn("SCALE_ANALYSIS_JSON=$ANALYSIS_JSON".replace(
+                "$ANALYSIS_JSON", os.path.join(results, "scale-attempt", "analysis-pre-evidence.json")),
+                calls[1])
+        self.assertIn('ANALYSIS_JSON="$SCALE_DIR/analysis-pre-evidence.json"', body)
+        evidence = body[body.index("write_dns_evidence() {"):]
+        evidence = evidence[:evidence.index("\n}\n")]
+        self.assertIn('"$ANALYSIS_JSON"', evidence)
+        self.assertNotIn("$RESULTS/analysis.json", evidence.replace(
+            '${ANALYSIS_JSON:-$RESULTS/analysis.json}', ""))
+        # After a clean verdict the run is analyzed again into the analysis
+        # that may publish, and the campaign fails unless it does.
+        self.assertRegex(body, r'SCALE_ANALYSIS_JSON="\$SCALE_DIR/analysis\.json"')
+        self.assertIn("jq -e '.publishable == true' \"$SCALE_DIR/analysis.json\"", body)
+
     # From `mkdir -p "$RESULTS"` to the end of the rm, which may continue over
     # backslash-newlines, plus the lock release that closes the block. The
     # whole thing runs, so the lock the removal takes runs with it.
@@ -2356,5 +2415,33 @@ def test_the_diag_follows_the_golden_verify_in_both_phases_and_precedes_the_sett
                       "a failed diag does not end the campaign")
 
 
+
+class ScaleAttemptDirectory(unittest.TestCase):
+    def test_each_attempt_measures_into_its_own_scale_directory(self):
+        """RED ON a5d0159f: every attempt measured into $RESULTS/scale, and a
+        retry's startup cleanup deleted the earlier attempt's tree, WITHDRAWN
+        marker included, and could delete a concurrent campaign's run while it
+        was still writing. Each attempt now measures into its own directory
+        beside the campaign's evidence, and nothing removes one."""
+        body = campaign()
+        block = DnsBrackets.START_CLEANUP.search(body)
+        self.assertIsNotNone(block, "startup cleanup not found")
+        attempts = ("scale", "scale-20261009-010203-77")
+        with tempfile.TemporaryDirectory() as d:
+            results = os.path.join(d, "results")
+            os.makedirs(os.path.join(results, "diag"))
+            for attempt in attempts:
+                os.makedirs(os.path.join(results, attempt))
+                with open(os.path.join(results, attempt, "WITHDRAWN"), "w") as f:
+                    f.write("measured against the wrong resolver\n")
+            script = f'set -euo pipefail\nRESULTS="{results}"\n{block.group(1)}'
+            result = subprocess.run(["bash", "-c", script], capture_output=True, text=True, timeout=30)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            for attempt in attempts:
+                self.assertTrue(os.path.isfile(os.path.join(results, attempt, "WITHDRAWN")),
+                                f"the startup cleanup removed {attempt}'s WITHDRAWN marker")
+        self.assertRegex(body, r'(?m)^    SCALE_DIR="\$RESULTS/scale-\$STAMP-\$\$"$')
+        self.assertNotRegex(body, r'rm -rf[^\n]*scale')
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/bench/chromium/test_reqbench.py b/bench/chromium/test_reqbench.py
index c6f002d5f..46682f61d 100644
--- a/bench/chromium/test_reqbench.py
+++ b/bench/chromium/test_reqbench.py
@@ -9262,12 +9262,15 @@ class HarnessIdentity(unittest.TestCase):
 
     def test_harness_hash_covers_every_staged_request_script(self):
         sh = open(os.path.join(HERE, "reqbench.sh")).read()
-        m = re.search(r"for source in ([^;]+); do", sh)
+        m = re.search(r"runtime_sources=\(([^)]+)\)", sh)
         self.assertIsNotNone(m, "staged source list not found in reqbench.sh")
         staged = set(m.group(1).split())
         # reqanalyze.py is staged (the analysis step runs from the bundle) but
-        # defines no request sample.
-        self.assertEqual(set(reqbench.HARNESS_SOURCES), staged - {"reqanalyze.py"})
+        # defines no request sample; the scale run executes from the same bundle.
+        import reqscale
+        self.assertEqual(
+            staged,
+            set(reqbench.HARNESS_SOURCES) | {"reqanalyze.py"} | set(reqscale.HARNESS_SOURCES))
 
 
 # `make` invoked as a command in a repository script, ignoring comment lines:
diff --git a/bench/chromium/test_reqscale.py b/bench/chromium/test_reqscale.py
index 5a5c0a273..7633b2ea1 100644
--- a/bench/chromium/test_reqscale.py
+++ b/bench/chromium/test_reqscale.py
@@ -45,6 +45,7 @@ def config(self, **overrides):
             ),
             trace_rate=None,
             trace_pairs=0,
+            urls=("http://127.0.0.1/fixture",),
         )
         values.update(overrides)
         return reqscale.ScheduleConfig(**values)
@@ -228,7 +229,8 @@ def test_launches_follow_absolute_deadlines_and_drain_only_after_last_launch(sel
         clock = FakeClock()
         launcher = DeferredLauncher()
         records, summary = reqscale.run_open_loop_burst(
-            RUN_ID, spec, lambda context: {"backend": context.backend}, clock, launcher
+            RUN_ID, spec, lambda context: {"backend": context.backend,
+                                       "url": "http://127.0.0.1/fixture"}, clock, launcher
         )
 
         self.assertTrue(launcher.drain_called)
@@ -1140,7 +1142,8 @@ def _machine(raw, captured_ns):
     def fixture(cls):
         spec = OpenLoopIsActuallyOpenLoop.one_request_spec()
         records, summary = reqscale.run_open_loop_burst(
-            RUN_ID, spec, lambda context: {"backend": context.backend},
+            RUN_ID, spec, lambda context: {"backend": context.backend,
+                                       "url": "http://127.0.0.1/fixture"},
             FakeClock(), DeferredLauncher(),
         )
         for index, row in enumerate(records):
@@ -1190,7 +1193,8 @@ def fixture(cls):
                 "file": [], "uffd": [12],
             },
         )
-        schedule = {"run_id": RUN_ID, "bursts": [spec.to_dict()]}
+        schedule = {"run_id": RUN_ID, "urls": ["http://127.0.0.1/fixture"],
+                    "bursts": [spec.to_dict()]}
         return schedule, [summary], records
 
     def test_green_summary_cannot_hide_a_failed_raw_request(self):
@@ -1548,7 +1552,8 @@ def _sample(cls, index, scheduled, terminal, phase, run_path):
         }
 
     @classmethod
-    def build_run(cls, directory):
+    def build_run(cls, directory, url="http://127.0.0.1/fixture",
+                  resolve_all_to=None):
         criteria = reqscale.CapacityCriteria(
             max_offered_rps_error_pct=1.0,
             min_departure_ratio=0.95,
@@ -1559,6 +1564,7 @@ def build_run(cls, directory):
         schedule = reqscale.build_schedule(
             reqscale.ScheduleConfig(
                 rates=(0.8,), scored_bursts=5, seed=776, criteria=criteria,
+                urls=(url,),
             ),
             RUN_ID,
         )
@@ -1571,7 +1577,8 @@ def build_run(cls, directory):
             clock = FakeClock()
             clock.now_ns = next_start
             burst_rows, summary = reqscale.run_open_loop_burst(
-                RUN_ID, spec, lambda context: {"backend": context.backend},
+                RUN_ID, spec, lambda context: {"backend": context.backend,
+                                       "url": url},
                 clock, DeferredLauncher(),
             )
             for row in burst_rows:
@@ -1640,7 +1647,7 @@ def build_run(cls, directory):
 
         snapshot = cls._snapshot()
         provenance = {
-            "schema": "fcvm.chromium.reqscale.provenance.v1",
+            "schema": "fcvm.chromium.reqscale.provenance.v2",
             "run_id": RUN_ID,
             "created_at": "2026-08-09T00:00:00+00:00",
             "argv": ["reqscale.py", "--fixture"],
@@ -1650,6 +1657,12 @@ def build_run(cls, directory):
             "harness_sha256": "2" * 64,
             "fcvm_path": "/usr/bin/fcvm",
             "fcvm_sha256": "3" * 64,
+            "runtime_bundle_sha256": "5" * 64,
+            "golden_creator": {
+                "creator_fcvm_sha256": "3" * 64,
+                "creator_runtime_bundle_sha256": "5" * 64,
+                "source_revision": "1" * 40,
+            },
             "fcvm_version": "fcvm fixture",
             "schedule_sha256": reqscale.schedule_sha256(schedule),
             "snapshot": snapshot,
@@ -1671,7 +1684,8 @@ def build_run(cls, directory):
                 "chromium_path": "/usr/bin/chromium",
                 "chromium_version": "Chromium fixture",
                 "chromium_sha256": "4" * 64,
-                "url": "http://127.0.0.1/fixture",
+                "url": url,
+                "resolve_all_to": resolve_all_to,
                 "timeout_seconds": 8.0,
             },
             "fault_trace": {
@@ -1818,7 +1832,7 @@ def build_run(cls, directory):
             "schema": reqscale.RECORD_SCHEMA, "kind": "uffd-serve",
             "run_id": RUN_ID, "pid": cls.SERVE_PID,
             "pid_start_time_ticks": 300, "state_path": "state/serve.json",
-            "uffd_mode": "copy", "snapshot_tag": snapshot["tag"],
+            "uffd_mode": "copy", "uffd_prefetch": "on", "snapshot_tag": snapshot["tag"],
             "snapshot_generation_id": cls.GENERATION,
             "snapshot_config_sha256": cls.CONFIG_SHA,
         }
@@ -2295,13 +2309,45 @@ def test_tracing_without_a_predeclared_perturbation_limit_is_rejected(self):
             self.assertIn("max-trace-perturbation-pct", result.stderr)
             self.assertFalse(os.path.exists(os.path.join(d, "out")))
 
+    def _plan(self, *extra):
+        with tempfile.TemporaryDirectory() as d:
+            return subprocess.run(
+                [
+                    sys.executable, self.SCRIPT,
+                    "--snapshot-tag", "not-opened-in-plan-mode",
+                    "--rates", "2", "--bursts", "5",
+                    "--seed", "776", "--run-id", RUN_ID,
+                    "--out-dir", os.path.join(d, "plan"), "--plan-only",
+                    *self.criteria_args(), *extra,
+                ],
+                capture_output=True, text=True, timeout=30,
+            )
+
+    def test_a_url_list_needs_a_single_control_url(self):
+        """Red before URL lists: the list became the control's one URL."""
+        corpus = "https://a.example/,https://b.example/"
+        result = self._plan("--url", corpus)
+        self.assertEqual(result.returncode, 2)
+        self.assertIn("--control-url is required", result.stderr)
+        result = self._plan("--url", corpus, "--control-url", corpus)
+        self.assertEqual(result.returncode, 2)
+        self.assertIn("exactly one URL", result.stderr)
+        result = self._plan("--url", corpus, "--control-url", "https://a.example/",
+                            "--control-resolve-all-to", "127.0.0.1")
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_the_resolver_rule_must_be_an_ipv4_address(self):
+        result = self._plan("--url", "http://x/", "--control-resolve-all-to", "replay")
+        self.assertEqual(result.returncode, 2)
+        self.assertIn("IPv4", result.stderr)
+
 
 
 class ConcurrentRequestRecords(unittest.TestCase):
     @staticmethod
-    def _inputs():
+    def _inputs(urls=None, pair_index=0):
         args = SimpleNamespace(
-            url="http://x/", format="jpeg", quality=80, cdp_port=9222, ws_url="",
+            url="http://x/", urls=urls, format="jpeg", quality=80, cdp_port=9222, ws_url="",
             fcvm="fcvm", data_root="/d", state_dir="/s", timeout=10.0,
             teardown_timeout=5.0, rust_log="off", run_id="r", snapshot_tag="t",
             cgroup_paths={"uffd": "/sys/fs/cgroup/x/uffd", "file": "/sys/fs/cgroup/x/file"},
@@ -2311,11 +2357,26 @@ def _inputs():
                                population="score", traced=False, trace_pair_id=None)
         context = reqscale.RequestContext(
             run_id="r", burst_id="b", population="score", segment="score",
-            backend="uffd", target_rps=2.0, request_index=0, pair_index=0,
+            backend="uffd", target_rps=2.0, request_index=0, pair_index=pair_index,
             request_id="r:b:0", scheduled_ns=0, actual_launch_ns=1, request_seed=7,
         )
         return args, spec, context
 
+    def _record(self, cdp_record, urls=None, pair_index=0):
+        args, spec, context = self._inputs(urls, pair_index)
+        request = reqscale._make_request_fn(
+            args, spec, 1234, "/logs", {"uffd": mock.Mock(), "file": mock.Mock()}, None, 0)
+        seen = []
+
+        def run(request_args, rep, fast, **_options):
+            seen.append(request_args.url)
+            return dict(cdp_record)
+
+        with mock.patch.object(reqscale.reqbench, "run_cdp_request", side_effect=run):
+            record = request(context)
+        self.assertEqual(seen, [record["url"]], "the record names the page it rendered")
+        return record
+
     def test_a_concurrent_request_never_reads_the_memory_server(self):
         """Red while reqscale only dropped the samples from the record:
         run_cdp_request had already read the shared server's /proc/<pid>/stat
@@ -2344,5 +2405,293 @@ def sample(pid):
                     request(context)
         self.assertEqual(reads, [], "a concurrent UFFD request read the memory server")
 
+    def test_a_corpus_request_renders_its_pairs_url_and_says_so(self):
+        """Red before URL lists: every request rendered args.url, the whole
+        comma-separated spec, and no record named its page."""
+        urls = [f"https://site{i}.example/" for i in range(14)]
+        record = self._record({"ok": True}, urls=urls, pair_index=15)
+        self.assertEqual(record["url"], urls[1])
+
+
+CORPUS = tuple(f"https://site{i}.example/" for i in range(14))
+
+
+class CorpusPairing(unittest.TestCase):
+    def _config(self, rate):
+        return reqscale.ScheduleConfig(
+            rates=(rate,), scored_bursts=5, seed=776, urls=CORPUS,
+            criteria=reqscale.CapacityCriteria(
+                max_offered_rps_error_pct=1.0, min_departure_ratio=0.95,
+                max_score_end_backlog=8, max_p95_launch_lag_ms=25.0,
+                max_control_median_drift_pct=10.0))
+
+    def test_every_scored_window_renders_each_page_equally_often(self):
+        """Red before the whole-cycle rule: at 2 rps the 120 scored pairs of a
+        burst gave eight pages 9 renders and six pages 8, the same bias in
+        every burst, so the rate curve was confounded with the page mix."""
+        schedule = reqscale.build_schedule(self._config(1.4), RUN_ID)
+        self.assertEqual(schedule["urls"], list(CORPUS))
+        args = SimpleNamespace(url=",".join(CORPUS), urls=list(CORPUS))
+        for raw in schedule["bursts"]:
+            spec = reqscale.BurstSpec.from_dict(raw)
+            for backend in ("file", "uffd"):
+                pages = [reqscale.request_url(args, SimpleNamespace(pair_index=r.pair_index))
+                         for r in spec.requests
+                         if r.segment == "score" and r.backend == backend]
+                counts = {url: pages.count(url) for url in CORPUS}
+                self.assertEqual(set(counts.values()), {len(pages) // len(CORPUS)},
+                                 (raw["burst_id"], backend, counts))
+            halves = {}
+            for r in spec.requests:
+                halves.setdefault(r.pair_index, set()).add(
+                    reqscale.request_url(args, SimpleNamespace(pair_index=r.pair_index)))
+            self.assertTrue(all(len(urls) == 1 for urls in halves.values()))
+
+    def test_a_rate_that_splits_a_corpus_cycle_is_refused(self):
+        with self.assertRaisesRegex(ValueError, "whole number of cycles"):
+            reqscale.build_schedule(self._config(2.0), RUN_ID)
+
+    def test_a_single_url_is_unchanged(self):
+        args = SimpleNamespace(url="http://x/", urls=None)
+        self.assertEqual(reqscale.request_url(args, SimpleNamespace(pair_index=9)), "http://x/")
+
+
+class AnalyzerHoldsTheCorpus(unittest.TestCase):
+    def test_a_request_that_rendered_another_page_is_rejected(self):
+        """Red before the corpus was in the schedule: a producer rendering one
+        page for every request passed as the corpus experiment."""
+        schedule, summaries, records = AnalyzerRejectsCorruptEvidence.fixture()
+        records[0]["url"] = "http://elsewhere/"
+        with self.assertRaisesRegex(reqscale_analyze.AnalysisInvalid, "url"):
+            reqscale_analyze._validate_requests(
+                schedule, summaries, records,
+                AnalyzerRejectsCorruptEvidence.GENERATION,
+                AnalyzerRejectsCorruptEvidence.CONFIG_SHA)
+
+    def _gate(self, evidence):
+        with tempfile.TemporaryDirectory() as d:
+            run_dir = os.path.join(d, "scale")
+            os.mkdir(run_dir)
+            if evidence is not None:
+                with open(os.path.join(d, "dns-evidence.json"), "w") as handle:
+                    json.dump(evidence, handle)
+            return reqscale_analyze.corpus_dns_gate(
+                run_dir, {"run_id": RUN_ID, "urls": list(CORPUS)},
+                {"host_control": {"resolve_all_to": "127.0.0.1"}})
+
+    def test_a_corpus_run_needs_the_campaigns_clean_dns_evidence(self):
+        """Red before the gate: a corpus run's analysis said publishable with
+        no evidence that its clones resolved through the replay server."""
+        self.assertIn("without the campaign's DNS evidence", self._gate(None))
+        self.assertIn("not 'clean'", self._gate({"verdict": "unclean", "run_id": RUN_ID}))
+        self.assertIn("records run_id='other'",
+                      self._gate({"verdict": "clean", "run_id": "other"}))
+
+    def test_the_dns_evidence_is_held_to_every_file_it_pins(self):
+        """Red while the gate read only verdict and run_id: an object holding
+        just those two fields, or a bundle whose replay log changed after the
+        verdict, left a corpus run publishable. The gate is campaign_summary's
+        check, so a bundle the index would refuse is refused here too."""
+        from test_campaign_summary import write_run
+
+        bare = self._gate({"verdict": "clean", "run_id": RUN_ID})
+        self.assertIsNotNone(bare, "two fields passed as a whole DNS evidence bundle")
+        self.assertIn("samples", bare)
+        schedule = {"run_id": RUN_ID, "urls": ["https://example.com/"]}
+        provenance = {"host_control": {"resolve_all_to": "127.0.0.1"}}
+        with tempfile.TemporaryDirectory() as d:
+            write_run(d, analysis_overrides={"run_id": RUN_ID})
+            run_dir = os.path.join(d, "scale")
+            os.mkdir(run_dir)
+            self.assertIsNone(reqscale_analyze.corpus_dns_gate(run_dir, schedule, provenance))
+            with open(os.path.join(d, "corpus-dns.log"), "a") as handle:
+                handle.write("{}\n")
+            refused = reqscale_analyze.corpus_dns_gate(run_dir, schedule, provenance)
+        self.assertIsNotNone(refused, "a replay log changed after the verdict passed")
+        self.assertIn("corpus-dns.log sha256", refused)
+
+    def test_a_single_page_run_needs_no_resolver_evidence(self):
+        self.assertIsNone(reqscale_analyze.corpus_dns_gate(
+            "/nonexistent/scale", {"run_id": RUN_ID, "urls": ["http://127.0.0.1/f"]},
+            {"host_control": {"resolve_all_to": None}}))
+
+    def test_the_control_resolver_must_be_null_or_an_ipv4_address(self):
+        for good in (None, "127.0.0.1"):
+            self.assertTrue(reqscale_analyze._null_or_ipv4(good), good)
+        for bad in ("", "replay", "127.0.0.01", 7):
+            self.assertFalse(reqscale_analyze._null_or_ipv4(bad), bad)
+
+
+class ScaleOutputHonoursThePublicationGate(unittest.TestCase):
+    """main() is what the campaign and `make report-chromium-scale` run."""
+
+    @staticmethod
+    def _corpus_run(d, evidence=True):
+        from test_campaign_summary import write_run
+
+        run_dir = os.path.join(d, "scale")
+        os.mkdir(run_dir)
+        CompleteAnalyzerFixture.build_run(
+            run_dir, url="https://example.com/", resolve_all_to="127.0.0.1")
+        if evidence:
+            write_run(d, analysis_overrides={"run_id": RUN_ID})
+        return run_dir
+
+    @staticmethod
+    def _main(*argv):
+        err = io.StringIO()
+        with mock.patch.object(sys, "argv", ["reqscale_analyze.py", *argv]), \
+                mock.patch.object(reqscale_analyze, "BOOTSTRAP_DRAWS", 1000), \
+                redirect_stderr(err):
+            return reqscale_analyze.main(), err.getvalue()
+
+    @staticmethod
+    def _load(path):
+        with open(path) as stream:
+            return json.load(stream)
+
+    def test_an_unpublishable_analysis_gets_json_but_no_report(self):
+        """RED BEFORE THE FIX: with the campaign's DNS evidence absent the
+        analysis said publishable false, and --markdown-out still wrote a
+        report saying the run passed its checks, at exit 0. The JSON-only
+        analysis still exits 0, because the campaign writes it before the
+        evidence exists to bind the run id the evidence names."""
+        with tempfile.TemporaryDirectory() as d, tempfile.TemporaryDirectory() as out:
+            run_dir = self._corpus_run(d, evidence=False)
+            report = os.path.join(out, "report.md")
+            rc, err = self._main("--run-dir", run_dir, "--markdown-out", report)
+            self.assertNotEqual(rc, 0, "a report was written for an unpublishable analysis")
+            self.assertFalse(os.path.lexists(report))
+            self.assertIn("DNS evidence", err)
+            pre = os.path.join(out, "analysis-pre-evidence.json")
+            rc, err = self._main("--run-dir", run_dir, "--json-out", pre)
+            self.assertEqual(rc, 0, err)
+            self.assertFalse(self._load(pre)["publishable"])
+
+    def test_a_withdrawn_run_or_campaign_is_not_publishable(self):
+        """RED BEFORE THE FIX: the scale analyzer read no WITHDRAWN marker, so
+        re-analysing a withdrawn campaign whose clean DNS evidence stayed in
+        place said publishable, and its report was written."""
+        for where in ("campaign", "run"):
+            with self.subTest(where=where), tempfile.TemporaryDirectory() as d, \
+                    tempfile.TemporaryDirectory() as out:
+                run_dir = self._corpus_run(d)
+                before = os.path.join(out, "before.json")
+                rc, err = self._main("--run-dir", run_dir, "--json-out", before)
+                self.assertEqual(rc, 0, err)
+                self.assertTrue(self._load(before)["publishable"])
+                marked = d if where == "campaign" else run_dir
+                with open(os.path.join(marked, "WITHDRAWN"), "w") as marker:
+                    marker.write("resolver was ambient\n")
+                after = os.path.join(out, "after.json")
+                rc, err = self._main("--run-dir", run_dir, "--json-out", after)
+                self.assertEqual(rc, 0, err)
+                analysis = self._load(after)
+                self.assertFalse(analysis["publishable"], f"a withdrawn {where} stayed publishable")
+                self.assertIn("withdrawn: resolver was ambient",
+                              analysis["publication_blocked_by"])
+                report = os.path.join(out, "report.md")
+                rc, err = self._main("--run-dir", run_dir, "--markdown-out", report)
+                self.assertNotEqual(rc, 0, err)
+                self.assertFalse(os.path.lexists(report))
+
+    def test_the_run_and_its_campaign_stay_locked_while_the_analysis_reads_them(self):
+        """RED BEFORE THE FIX: the analyzer held no lock, so a withdrawal
+        writer following AGENTS.md (exclusive flock on the directory, then the
+        marker) took either directory in the middle of validation, and the
+        analysis published from the state before its marker."""
+        with tempfile.TemporaryDirectory() as d, tempfile.TemporaryDirectory() as out:
+            run_dir = self._corpus_run(d)
+            report = os.path.join(out, "report.md")
+            real_analyze = reqscale_analyze.analyze
+            writers = []
+
+            def analyze_beside_a_writer(path):
+                for directory in (run_dir, d):
+                    writers.append(subprocess.run(
+                        ["flock", "-n", "-x", directory, "sh", "-c",
+                         'printf "late\\n" > "$1"', "sh",
+                         os.path.join(directory, "WITHDRAWN")],
+                        capture_output=True, text=True, timeout=10,
+                    ).returncode)
+                return real_analyze(path)
+
+            with mock.patch.object(reqscale_analyze, "analyze",
+                                   side_effect=analyze_beside_a_writer):
+                rc, err = self._main("--run-dir", run_dir, "--markdown-out", report)
+            self.assertEqual(writers, [1, 1],
+                             "a withdrawal writer locked a directory mid-analysis")
+            self.assertEqual(rc, 0, err)
+            self.assertTrue(os.path.isfile(report))
+            for directory in (run_dir, d):
+                self.assertFalse(os.path.lexists(os.path.join(directory, "WITHDRAWN")))
+                released = subprocess.run(
+                    ["flock", "-n", "-x", directory, "true"],
+                    capture_output=True, timeout=10,
+                )
+                self.assertEqual(released.returncode, 0,
+                                 f"{directory}: the analysis kept its lock")
+
+
+class RestorePathIsWhatItSays(unittest.TestCase):
+    def test_a_file_arm_that_would_restore_through_uffd_is_refused(self):
+        """Red before the check: hugepage, NV2 (kernel profile) and forced
+        restores made the FILE arm an implicit UFFD one."""
+        plain = {"metadata": {"hugepages": False, "kernel_profile": None}}
+        self.assertIsNone(reqscale.file_restore_refusal(plain, {}))
+        self.assertIn("hugepages", reqscale.file_restore_refusal(
+            {"metadata": {"hugepages": True}}, {}))
+        self.assertIn("nested", reqscale.file_restore_refusal(
+            {"metadata": {"kernel_profile": "nested"}}, {}))
+        self.assertIn("FCVM_FORCE_UFFD", reqscale.file_restore_refusal(
+            plain, {"FCVM_FORCE_UFFD": "1"}))
+
+    def test_the_server_must_run_the_requested_mode(self):
+        """Red before the check: NV2 coerced minor to copy and the UFFD cells
+        were labelled minor."""
+        reqscale.require_serve_mode("minor", "minor")
+        with self.assertRaisesRegex(reqscale.MeasurementInvalid, "runs copy mode"):
+            reqscale.require_serve_mode("copy", "minor")
+
+
+class SpawnedArgv(unittest.TestCase):
+    """What start() hands to Popen, captured by a Popen that refuses."""
+
+    def _capture(self, start):
+        seen = []
+
+        def refuse(argv, **_kw):
+            seen.append(list(argv))
+            raise RuntimeError("captured")
+
+        with mock.patch.object(reqscale.subprocess, "Popen", side_effect=refuse):
+            with self.assertRaises(Exception):
+                start()
+        self.assertEqual(len(seen), 1)
+        return seen[0]
+
+    def test_the_memory_server_starts_in_the_requested_mode(self):
+        """Red on the parent: the serve argv had no --uffd-mode, so every run
+        served copy mode."""
+        with tempfile.TemporaryDirectory() as d:
+            args = SimpleNamespace(fcvm="/f", snapshot_tag="cb-req-corpus", state_dir=d,
+                                   data_root=d, uffd_mode="minor", uffd_prefetch="off",
+                                   rust_log="fcvm=debug")
+            serve = reqscale.UffdServe(args, mock.Mock(), "/sys/fs/cgroup/x/uffd", d)
+            argv = self._capture(serve.start)
+        self.assertEqual(argv[-8:], ["/f", "snapshot", "serve", "cb-req-corpus",
+                                     "--uffd-mode", "minor", "--uffd-prefetch", "off"])
+
+    def test_the_control_chromium_maps_names_to_the_replay_server(self):
+        """Red on the parent: the control argv had no resolver rule, so a
+        corpus control resolved on the live internet."""
+        with tempfile.TemporaryDirectory() as d:
+            args = SimpleNamespace(control_chromium="/c", control_resolve_all_to="127.0.0.1",
+                                   control_tmp_root=d, run_id="r")
+            control = reqscale.NativeChromiumControl(args, mock.Mock(), "/sys/fs/cgroup/x/c", d)
+            argv = self._capture(control.start)
+        self.assertIn("--host-resolver-rules=MAP * 127.0.0.1", argv)
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)
diff --git a/bench/chromium/test_reqscale_seal.py b/bench/chromium/test_reqscale_seal.py
new file mode 100644
index 000000000..6c578f282
--- /dev/null
+++ b/bench/chromium/test_reqscale_seal.py
@@ -0,0 +1,332 @@
+"""The scale run measures with the runtime that created its golden, as the serial run does."""
+import contextlib
+import hashlib
+import io
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+import reqbench  # noqa: E402
+import reqscale  # noqa: E402
+import reqscale_analyze  # noqa: E402
+import test_reqbench  # noqa: E402
+from test_reqscale import CompleteAnalyzerFixture  # noqa: E402
+
+GEN = "11111111-1111-4111-8111-111111111111"
+KEY = "a" * 64
+
+
+def write_golden(data_root, tag, creator):
+    snap = os.path.join(data_root, "snapshots", tag)
+    os.makedirs(snap)
+    paths = {}
+    for field, name in (("memory_path", "memory.bin"), ("vmstate_path", "vmstate.bin"),
+                        ("disk_path", "disk.raw")):
+        paths[field] = os.path.join(snap, name)
+        with open(paths[field], "wb") as f:
+            f.write(field.encode())
+    config = {
+        "name": tag, "vm_id": "vm-source", "generation_id": GEN,
+        "created_at": "2026-10-08T00:00:00Z", **paths,
+        "metadata": {
+            "image": "localhost/chromium-bench-req",
+            "image_disk_path": f"/image-cache/{KEY}.storage-v2.img",
+            "vcpu": 2, "memory_mib": 1024, "network_mode": "rootless",
+            "port_mappings": [{"host_ip": None, "host_port": 9222,
+                               "guest_port": 9222, "proto": "tcp"}],
+        },
+    }
+    raw = (json.dumps(config, sort_keys=True) + "\n").encode()
+    with open(os.path.join(snap, "config.json"), "wb") as f:
+        f.write(raw)
+    provenance = {
+        "snapshot_generation_id": GEN,
+        "snapshot_config_sha256": hashlib.sha256(raw).hexdigest(),
+        "snapshot_created_at": config["created_at"], "snapshot_vm_id": "vm-source",
+        "image": "localhost/chromium-bench-req", "image_id": "sha256:" + "b" * 64,
+        "image_digest": "sha256:" + KEY, "image_cache_key": KEY,
+        "guest_dns": None, "guest_env": [], **creator,
+    }
+    with open(os.path.join(snap, "reqbench-provenance.json"), "w") as f:
+        json.dump(provenance, f)
+
+
+class Reached(Exception):
+    pass
+
+
+class ScaleRefusesAGoldenFromAnotherRuntime(unittest.TestCase):
+    """reqscale.main() must refuse, before measuring, a golden whose recorded
+    creator runtime is not the staged runtime it executes from."""
+
+    def _run(self, d, creator_overrides):
+        bundle = os.path.join(d, "bundle")
+        os.makedirs(bundle)
+        with open(os.path.join(bundle, "fcvm"), "wb") as f:
+            f.write(b"#!/bin/sh\nexit 0\n")
+        with open(os.path.join(bundle, "MANIFEST.sha256"), "w") as f:
+            f.write("fixture manifest\n")
+        revision = "e" * 40
+        creator = {
+            "creator_fcvm_sha256": reqbench.sha256_file(os.path.join(bundle, "fcvm")),
+            "creator_runtime_bundle_sha256": reqbench.sha256_file(
+                os.path.join(bundle, "MANIFEST.sha256")),
+            "source_revision": revision,
+        }
+        creator.update(creator_overrides)
+        data_root = os.path.join(d, "data")
+        write_golden(data_root, "golden", creator)
+        argv = [
+            "reqscale.py", "--snapshot-tag", "golden", "--url", "http://127.0.0.1/x",
+            "--rates", "0.8", "--bursts", "5", "--seed", "776",
+            "--max-offered-rps-error-pct", "1", "--min-departure-ratio", "0.95",
+            "--max-score-end-backlog", "2", "--max-p95-launch-lag-ms", "20",
+            "--max-control-median-drift-pct", "10", "--out-dir", os.path.join(d, "out"),
+            "--data-root", data_root, "--control-chromium", sys.executable,
+            "--control-tmp-root", d, "--cgroup-root", os.path.join(d, "cgroup"),
+        ]
+        env = {"REQBENCH_RUNTIME_BUNDLE": bundle, "REQBENCH_SOURCE_REVISION": revision,
+               "REQBENCH_SOURCE_REPO": d}
+        err = io.StringIO()
+        reached = False
+        with mock.patch.object(sys, "argv", argv), \
+                mock.patch.dict(os.environ, env), \
+                mock.patch.object(reqscale.os, "geteuid", return_value=0), \
+                mock.patch.object(reqscale, "HERE", bundle), \
+                mock.patch.object(reqbench, "HERE", bundle), \
+                mock.patch.object(reqscale, "collect_provenance", side_effect=Reached), \
+                mock.patch.object(reqscale, "execute", side_effect=Reached), \
+                contextlib.redirect_stderr(err):
+            os.environ.pop("FCVM_FORCE_UFFD", None)
+            try:
+                rc = reqscale.main()
+            except Reached:
+                reached, rc = True, None
+        return reached, rc, err.getvalue()
+
+    def test_a_golden_created_by_another_runtime_is_refused_before_measuring(self):
+        with tempfile.TemporaryDirectory() as d:
+            reached, _rc, err = self._run(d, {})
+            self.assertTrue(reached, f"the golden's own runtime was refused: {err}")
+        for field, value in (("creator_fcvm_sha256", "1" * 64),
+                             ("creator_runtime_bundle_sha256", "2" * 64),
+                             ("source_revision", "3" * 40)):
+            with self.subTest(field=field), tempfile.TemporaryDirectory() as d:
+                reached, rc, err = self._run(d, {field: value})
+                self.assertFalse(reached, f"measured although the golden records {field}={value}")
+                self.assertEqual(rc, 4, err)
+                self.assertIn(f"was created with {field}='{value}'", err)
+
+
+class AnalyzerBindsTheRunToItsGoldensRuntime(unittest.TestCase):
+    @staticmethod
+    def _bind(provenance):
+        # what build_run writes once the fix lands
+        provenance["runtime_bundle_sha256"] = "5" * 64
+        provenance["golden_creator"] = {
+            "creator_fcvm_sha256": provenance["fcvm_sha256"],
+            "creator_runtime_bundle_sha256": "5" * 64,
+            "source_revision": provenance["source_revision"],
+        }
+
+    def test_a_run_not_bound_to_its_goldens_runtime_is_refused(self):
+        cases = {
+            "unbound": "provenance fields are incomplete or unknown",
+            "creator_fcvm_sha256": "creator_fcvm_sha256",
+            "creator_runtime_bundle_sha256": "creator_runtime_bundle_sha256",
+            "source_revision": "source_revision",
+        }
+        for case, expected in cases.items():
+            with self.subTest(case=case), tempfile.TemporaryDirectory() as d:
+                CompleteAnalyzerFixture.build_run(d)
+                path = os.path.join(d, "provenance.json")
+                with open(path) as f:
+                    provenance = json.load(f)
+                if "golden_creator" not in provenance:
+                    self._bind(provenance)
+                if case == "unbound":
+                    del provenance["golden_creator"], provenance["runtime_bundle_sha256"]
+                else:
+                    width = 40 if case == "source_revision" else 64
+                    provenance["golden_creator"][case] = "a" * width
+                with open(path, "w") as f:
+                    json.dump(provenance, f, sort_keys=True)
+                with self.assertRaisesRegex(reqscale_analyze.AnalysisInvalid, expected):
+                    reqscale_analyze.analyze(d)
+
+
+class ScaleShell(unittest.TestCase):
+    def test_scale_runs_reqscale_from_the_staged_bundle(self):
+        with tempfile.TemporaryDirectory() as d:
+            binx = os.path.join(d, "bin")
+            os.makedirs(binx)
+            for name, body in (("fcvm", "#!/bin/bash\nexit 0\n"),
+                               ("fc-agent", "#!/bin/bash\nexit 0\n"),
+                               ("bin/sudo", '#!/bin/bash\nprintf "%s\\n" "$@" > "$SUDO_ARGV"\n')):
+                with open(os.path.join(d, name), "w") as f:
+                    f.write(body)
+                os.chmod(os.path.join(d, name), 0o755)
+            os.makedirs(os.path.join(d, "state"))
+            env = dict(os.environ, PATH=binx + os.pathsep + os.environ["PATH"],
+                       RESULTS=os.path.join(d, "results"), STATE_DIR=os.path.join(d, "state"),
+                       RUNID="0" * 32, FCVM=os.path.join(d, "fcvm"),
+                       FC_AGENT=os.path.join(d, "fc-agent"), TAG="cb-scale",
+                       SUDO_ARGV=os.path.join(d, "argv"))
+            result = subprocess.run(
+                [os.path.join(HERE, "reqbench.sh"), "scale", "--url", "http://127.0.0.1/x"],
+                env=env, capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            with open(env["SUDO_ARGV"]) as f:
+                argv = f.read().splitlines()
+            script = next(a for a in argv if a.endswith("/reqscale.py"))
+            bundle = os.path.dirname(script)
+            self.assertEqual(os.path.dirname(bundle), os.path.join(d, "results", "runtime"))
+            with open(os.path.join(bundle, "MANIFEST.sha256")) as f:
+                sealed = {line.split()[1] for line in f}
+            self.assertLessEqual({"fcvm", "reqscale.py", "reqscale_analyze.py", "guardexec.py",
+                                  "guardsupervise.py", "faulttrace.bt"}, sealed)
+            self.assertIn(f"REQBENCH_RUNTIME_BUNDLE={bundle}", argv)
+            self.assertEqual(argv[argv.index("--snapshot-tag") + 1], "cb-scale")
+            self.assertEqual(argv[argv.index("--url") + 1], "http://127.0.0.1/x")
+
+
+
+class ScaleOutputOwnership(unittest.TestCase):
+    def test_a_pre_existing_output_directory_is_not_handed_back(self):
+        """RED ON f9f2fd96: reqbench.sh scale changed ownership of --out-dir
+        after reqscale.py returned, even when reqscale.py had refused it because
+        it already existed, so a mistyped SCALE_OUT could hand an existing
+        root-owned tree to the invoking user. Only reqscale.py, which knows
+        whether it created the directory, hands it back."""
+        with tempfile.TemporaryDirectory() as d:
+            binx = os.path.join(d, "bin")
+            os.makedirs(binx)
+            for name, body in (("fcvm", "#!/bin/bash\nexit 0\n"),
+                               ("fc-agent", "#!/bin/bash\nexit 0\n"),
+                               ("bin/sudo", '#!/bin/bash\nprintf "%s\\n" "$*" >> "$SUDO_LOG"\n')):
+                with open(os.path.join(d, name), "w") as f:
+                    f.write(body)
+                os.chmod(os.path.join(d, name), 0o755)
+            os.makedirs(os.path.join(d, "state"))
+            out = os.path.join(d, "results", "scale")
+            os.makedirs(out)  # already there before this run: reqscale.py refuses it
+            env = dict(os.environ, PATH=binx + os.pathsep + os.environ["PATH"],
+                       RESULTS=os.path.join(d, "results"), STATE_DIR=os.path.join(d, "state"),
+                       RUNID="0" * 32, FCVM=os.path.join(d, "fcvm"),
+                       FC_AGENT=os.path.join(d, "fc-agent"), TAG="cb-scale",
+                       SUDO_LOG=os.path.join(d, "sudo.log"))
+            result = subprocess.run(
+                [os.path.join(HERE, "reqbench.sh"), "scale", "--url", "http://127.0.0.1/x",
+                 "--out-dir", out],
+                env=env, capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            with open(env["SUDO_LOG"]) as f:
+                calls = f.read().splitlines()
+        self.assertTrue(any("/reqscale.py" in c for c in calls), calls)
+        self.assertEqual([c for c in calls if c.startswith("chown")], [],
+                         f"reqbench.sh changed ownership of an existing directory: {calls}")
+
+
+class HandBackToInvoker(unittest.TestCase):
+    def test_reqscale_hands_back_only_the_directory_it_created(self):
+        with tempfile.TemporaryDirectory() as d:
+            root = os.path.join(d, "out")
+            os.makedirs(os.path.join(root, "logs"))
+            open(os.path.join(root, "logs", "a.json"), "w").close()
+            seen = []
+            with mock.patch.object(reqscale.os, "geteuid", return_value=0), \
+                    mock.patch.dict(os.environ, {"SUDO_UID": "1234", "SUDO_GID": "567"}), \
+                    mock.patch.object(reqscale.os, "lchown",
+                                      side_effect=lambda p, u, g: seen.append((p, u, g))):
+                reqscale.hand_back_to_invoker(root)
+            self.assertEqual(sorted(seen), sorted([
+                (root, 1234, 567), (os.path.join(root, "logs"), 1234, 567),
+                (os.path.join(root, "logs", "a.json"), 1234, 567)]))
+            seen.clear()
+            with mock.patch.object(reqscale.os, "geteuid", return_value=1000), \
+                    mock.patch.object(reqscale.os, "lchown", side_effect=lambda p, u, g: seen.append(p)):
+                reqscale.hand_back_to_invoker(root)
+            self.assertEqual(seen, [], "a run that is not root changed ownership")
+        with open(os.path.join(HERE, "reqscale.py")) as f:
+            source = f.read()
+        body = source[source.index("def execute(args, schedule: dict, provenance: dict) -> int:"):]
+        self.assertEqual([line.strip() for line in body.splitlines()[1:3]],
+                         ["os.mkdir(args.out_dir)", "atexit.register(hand_back_to_invoker, args.out_dir)"],
+                         "the handback is not registered right after reqscale.py creates its run directory")
+
+
+class UffdServeRecordsPrefetch(unittest.TestCase):
+    def test_the_prefetch_setting_is_required_and_published(self):
+        """RED ON a5d0159f: uffd-serve.json recorded only the memory mode, and
+        the analyzer neither required nor published the working-set prefetch
+        setting, so an on run and an off run read as the same UFFD experiment."""
+        with tempfile.TemporaryDirectory() as d:
+            CompleteAnalyzerFixture.build_run(d)
+            analysis = reqscale_analyze.analyze(d)
+        self.assertEqual(analysis.get("uffd"), {"mode": "copy", "prefetch": "on"})
+        for case, value in (("missing", None), ("invalid", "maybe")):
+            with self.subTest(case=case), tempfile.TemporaryDirectory() as d:
+                CompleteAnalyzerFixture.build_run(d)
+                path = os.path.join(d, "uffd-serve.json")
+                with open(path) as f:
+                    serve = json.load(f)
+                if value is None:
+                    serve.pop("uffd_prefetch", None)
+                else:
+                    serve["uffd_prefetch"] = value
+                with open(path, "w") as f:
+                    json.dump(serve, f, sort_keys=True)
+                with self.assertRaisesRegex(reqscale_analyze.AnalysisInvalid, "UFFD serve"):
+                    reqscale_analyze.analyze(d)
+
+    def test_the_serve_record_carries_the_configured_prefetch(self):
+        with open(os.path.join(HERE, "reqscale.py")) as f:
+            source = f.read()
+        record = source[source.index('"kind": "uffd-serve",'):]
+        record = record[:record.index("}")]
+        self.assertIn('"uffd_prefetch": getattr(self.args, "uffd_prefetch", "on"),', record)
+
+
+class ResolverEvidenceGate(unittest.TestCase):
+    RUN_ID = "0" * 32
+
+    def test_several_ip_literal_urls_need_no_resolver_evidence(self):
+        """RED ON 25906ca7: the gate called any run with more than one URL a
+        corpus run, so a standalone run over IP-literal or local URLs, which
+        resolve nothing, was refused for lacking the campaign's DNS evidence."""
+        with tempfile.TemporaryDirectory() as d:
+            run_dir = os.path.join(d, "scale")
+            os.mkdir(run_dir)
+            gate = reqscale_analyze.corpus_dns_gate(
+                run_dir, {"run_id": self.RUN_ID,
+                          "urls": ["http://127.0.0.1/a", "http://localhost/b"]},
+                {"host_control": {"resolve_all_to": None}})
+        self.assertIsNone(gate, gate)
+
+    def test_a_hostname_url_needs_resolver_evidence_even_alone(self):
+        """RED ON 25906ca7: one hostname URL was not a corpus run, so it
+        published with no record of which resolver answered it."""
+        with tempfile.TemporaryDirectory() as d:
+            run_dir = os.path.join(d, "scale")
+            os.mkdir(run_dir)
+            gate = reqscale_analyze.corpus_dns_gate(
+                run_dir, {"run_id": self.RUN_ID, "urls": ["https://example.com/"]},
+                {"host_control": {"resolve_all_to": None}})
+        self.assertIn("without the campaign's DNS evidence", gate or "the gate passed it")
+
+class ScaleGraph(test_reqbench.MakefileBenchGraph):
+    def test_scale_never_rebuilds(self):
+        c = self.closure("bench-chromium-scale")
+        for forbidden in ("build", "setup-default", "cargo-target-link"):
+            self.assertNotIn(forbidden, c, f"bench-chromium-scale transitively reaches {forbidden}")
+
+
+if __name__ == "__main__":
+    unittest.main()