diff --git a/README.md b/README.md
index 7de27ee..c373287 100644
--- a/README.md
+++ b/README.md
@@ -72,6 +72,20 @@ mismatches remain valid only when all evidence is non-passing. The stable wrappe
 inactive resolver, Control closure, and scanner select it together. The prior
 generation remains immutable and restorable.
 
+## Inactive offline delivery replay
+
+`delivery/v1/replay.py` replays one already-supplied local materialization input
+through the existing local Git materializer. It then checks one repo-relative
+candidate blob against a supplied SHA-256 and records a private, resumable state.
+It never executes candidate code or a user command string.
+
+Review and publisher records are supplied offline test observations. Each names the
+exact request digest, candidate tree, and candidate commit, and all three must match
+the recorded materialization: two candidate commits can carry one tree, so the commit
+is what binds an observation to this candidate. They still do not authenticate an
+actor or authorize a real publication. Missing review stays waiting; a completed
+receipt is explicitly an offline simulation with no authority or qualification.
+
 ## Inactive fake adapter contract matrix
 
 `adapter-tests/v1/` runs a fixed 2×2 producer/forge matrix against one unrelated
diff --git a/RESTORE.md b/RESTORE.md
index 22091d2..782ced6 100644
--- a/RESTORE.md
+++ b/RESTORE.md
@@ -55,6 +55,22 @@ with the restored `scripts/core-contract.sh`. No compiled helper is installed or
 restored. A future activation must separately qualify and bind a production trusted
 parent; restoring these files does not select a live profile.
 
+### Restore the inactive offline delivery replay
+
+Restore the three paths listed under “Inactive offline delivery replay” in
+[`ci/required-files.txt`](ci/required-files.txt), then run:
+
+```sh
+bash scripts/test/delivery-replay.test.sh
+```
+
+The replay is a local offline simulation. It materializes only a caller-owned
+candidate, reads one fixed candidate blob, and records test observations. Each
+observation must name the request digest, candidate tree, and candidate commit of
+the recorded materialization, and completion holds the candidate ref itself under a
+git transaction it owns. It does not execute candidate code, select a profile,
+authenticate review, publish, merge, deploy, or contact a provider or target.
+
 ### Restore the inactive default profile assembly
 
 Restore the eight paths listed under “Inactive default profile assembly” in
diff --git a/ci/required-files.txt b/ci/required-files.txt
index 3b916a0..0160bca 100644
--- a/ci/required-files.txt
+++ b/ci/required-files.txt
@@ -270,6 +270,11 @@ adapters/local-git-materializer/v1/materialize.sh
 adapters/local-git-materializer/v1/object-closure.c
 scripts/test/local-git-materializer-adapter.test.sh
 
+# Inactive offline delivery replay
+delivery/v1/replay.py
+scripts/test/delivery-replay.test.sh
+work/delivery-loop-first/plan.md
+
 # Inactive Claude Code producer normalizer payload
 adapters/claude-code-producer/v1/normalize.jq
 scripts/test/default-claude-code-producer-adapter.test.sh
diff --git a/delivery/v1/replay.py b/delivery/v1/replay.py
new file mode 100755
index 0000000..1c741c6
--- /dev/null
+++ b/delivery/v1/replay.py
@@ -0,0 +1,875 @@
+#!/usr/bin/env python3
+"""Run one inactive, offline delivery replay without executing candidate code."""
+
+import argparse
+from contextlib import contextmanager
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import select
+import signal
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+
+
+LOADED_DRIVER_BYTES = globals().get("_REPLAY_DRIVER_BYTES")
+if LOADED_DRIVER_BYTES is None and __name__ == "__main__":
+    try:
+        driver_descriptor = os.open(__file__, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+        with os.fdopen(driver_descriptor, "rb") as driver_handle:
+            driver_source = driver_handle.read(8 * 1024 * 1024 + 1)
+        if len(driver_source) > 8 * 1024 * 1024:
+            raise OSError("replay driver exceeds its size limit")
+        driver_code = compile(driver_source, __file__, "exec")
+    except (OSError, SyntaxError, TypeError, ValueError) as error:
+        print(f"delivery replay: loaded replay driver identity is unavailable: {error}", file=sys.stderr)
+        raise SystemExit(1) from error
+    globals()["_REPLAY_DRIVER_BYTES"] = driver_source
+    exec(driver_code, globals())
+    raise SystemExit(1)
+
+MAX_INPUT_BYTES = 8 * 1024 * 1024
+MAX_OBSERVATION_BYTES = 64 * 1024
+MAX_VERIFIED_BLOB_BYTES = 1024 * 1024
+GUARD_ACKNOWLEDGEMENT_SECONDS = 5
+MAX_GUARD_LINE_BYTES = 4096
+OID = re.compile(r"[0-9a-f]{40}|[0-9a-f]{64}\Z")
+ACTOR = re.compile(r"[a-z0-9][a-z0-9._:-]{0,127}\Z")
+# The core contract's id rule; the source repository id is journaled, so it is
+# bounded before anything is written.
+REPOSITORY_ID = re.compile(r"[a-z0-9][a-z0-9._:-]{0,127}\Z")
+GIT_ENVIRONMENT = {
+    "PATH": "/usr/bin:/bin", "LC_ALL": "C", "GIT_CONFIG_NOSYSTEM": "1",
+    "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_NO_REPLACE_OBJECTS": "1",
+    "GIT_NO_LAZY_FETCH": "1", "GIT_TERMINAL_PROMPT": "0",
+}
+NATIVE_EXECUTABLE_MAGICS = (
+    b"\x7fELF", b"\xcf\xfa\xed\xfe", b"\xfe\xed\xfa\xcf",
+    b"\xca\xfe\xba\xbe", b"\xbe\xba\xfe\xca",
+)
+PACKAGE_FILES = (
+    "adapters/local-git-materializer/v1/materialize.sh",
+    "adapters/local-git-materializer/v1/protocol.jq",
+    "scripts/core-contract.sh",
+    "core/v2/generation-registry.json",
+)
+GENERATION_FILES = (
+    "core-ingress.sh",
+    "contracts.jq",
+    "modules/schema.jq",
+    "modules/profile_graph.jq",
+    "modules/stage_request.jq",
+    "modules/result_facts.jq",
+    "modules/result_truth.jq",
+)
+
+
+class ReplayError(Exception):
+    pass
+
+
+def digest_bytes(value):
+    return hashlib.sha256(value).hexdigest()
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def read_bytes(path, limit):
+    # Open without blocking and refuse anything but a regular file before the
+    # first read, so a FIFO or device cannot stall the replay.
+    flags = os.O_RDONLY | os.O_NONBLOCK | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError as error:
+        raise ReplayError("input is not readable: %s" % path) from error
+    try:
+        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
+            raise ReplayError("input is not a regular file: %s" % path)
+        chunks = []
+        remaining = limit + 1
+        while remaining:
+            chunk = os.read(descriptor, remaining)
+            if not chunk:
+                break
+            chunks.append(chunk)
+            remaining -= len(chunk)
+        data = b"".join(chunks)
+    finally:
+        os.close(descriptor)
+    if len(data) > limit:
+        raise ReplayError("input exceeds its size limit")
+    return data
+
+
+def parse_json(data):
+    try:
+        return json.loads(data)
+    except (ValueError, UnicodeDecodeError, RecursionError) as error:
+        # Deeply nested input within the byte limit raises RecursionError; it is
+        # still just input this program cannot accept, never a crash.
+        raise ReplayError("input is not JSON") from error
+
+
+def private_directory(path):
+    value = Path(path)
+    stat = value.stat()
+    if value.is_symlink() or not value.is_dir() or stat.st_uid != os.getuid():
+        raise ReplayError("state directory is not a caller-owned directory")
+    if stat.st_mode & 0o077:
+        raise ReplayError("state directory is not private")
+    return value.resolve()
+
+
+def trusted_file(path):
+    value = Path(path)
+    stat = value.stat()
+    if value.is_symlink() or not value.is_file() or stat.st_size > MAX_INPUT_BYTES:
+        raise ReplayError("trusted tool is unavailable")
+    return value.resolve()
+
+
+def disjoint(*paths):
+    resolved = [Path(path).resolve() for path in paths]
+    for index, left in enumerate(resolved):
+        for right in resolved[index + 1:]:
+            if left == right or left in right.parents or right in left.parents:
+                raise ReplayError("caller-owned boundaries overlap")
+
+
+def atomic_json(path, value):
+    atomic_bytes(path, canonical(value) + b"\n")
+
+
+def atomic_bytes(path, encoded):
+    descriptor, temporary = tempfile.mkstemp(prefix=".replay-", dir=path.parent)
+    try:
+        with os.fdopen(descriptor, "wb") as handle:
+            handle.write(encoded)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temporary, path)
+        directory = os.open(path.parent, os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def safe_path(value):
+    if not isinstance(value, str) or not value or len(value) > 4096:
+        raise ReplayError("verification path is invalid")
+    parts = value.split("/")
+    if any(part in {"", ".", "..", ".git"} or part.endswith((".", " ")) for part in parts):
+        raise ReplayError("verification path is invalid")
+    if any("\\" in part or any(ord(char) < 32 for char in part) for part in parts):
+        raise ReplayError("verification path is invalid")
+    return value
+
+
+def package_paths(generation):
+    root = f"core/v2/generations/{generation}"
+    return PACKAGE_FILES + tuple(f"{root}/{name}" for name in GENERATION_FILES)
+
+
+def execution_source_bytes(repository, arguments):
+    core_relative = "scripts/core-contract.sh"
+    core = read_bytes(trusted_file(repository / core_relative), MAX_INPUT_BYTES)
+    match = re.search(
+        rb"^PORTABLE_CORE_GENERATION='(g-[0-9a-f]{64})'$", core, re.MULTILINE
+    )
+    if match is None:
+        raise ReplayError("materializer package generation is unavailable")
+    generation = match.group(1).decode()
+    package = {
+        relative: core if relative == core_relative else
+        read_bytes(trusted_file(repository / relative), MAX_INPUT_BYTES)
+        for relative in package_paths(generation)
+    }
+    dependency_sources = {
+        ".dependencies/object-closure": trusted_file(arguments.closure_helper),
+        ".dependencies/jq": trusted_file(arguments.jq_bin),
+    }
+    for source in dependency_sources.values():
+        # The snapshot copies dependencies with execute permission, so the caller
+        # must already hold an executable file; bytes alone never confer that.
+        if not os.access(source, os.X_OK) or not (os.stat(source).st_mode & 0o111):
+            raise ReplayError("dependency is not executable")
+    dependencies = {
+        relative: read_bytes(source, MAX_INPUT_BYTES)
+        for relative, source in dependency_sources.items()
+    }
+    if any(not data.startswith(NATIVE_EXECUTABLE_MAGICS) for data in dependencies.values()):
+        raise ReplayError("dependency is not a native executable")
+    return package | dependencies
+
+
+def owned_staging_token(owner_path):
+    if not owner_path.exists() or owner_path.is_symlink() or not owner_path.is_file():
+        return None
+    value = read_bytes(owner_path, 128)
+    match = re.fullmatch(rb"ystack-delivery-execution-v1:([0-9a-f]{64})\n", value)
+    return match.group(1) if match is not None else None
+
+
+def create_execution_snapshot(repository, arguments, state_dir):
+    root = state_dir / "execution"
+    staging = state_dir / ".execution-building"
+    owner_path = state_dir / ".execution-building.owner"
+    if root.is_symlink() or root.exists():
+        if root.is_symlink() or not root.is_dir():
+            raise ReplayError("execution bundle is unavailable")
+        # An existing bundle is judged by execution_sources_match, so changed or
+        # invalid dependencies read as a mismatch there, not as a build failure here.
+        return root
+    source_bytes = execution_source_bytes(repository, arguments)
+    token = owned_staging_token(owner_path)
+    if staging.is_symlink() or staging.exists():
+        if staging.is_symlink() or not staging.is_dir() or token is None:
+            raise ReplayError("execution bundle staging is not program-owned")
+        marker = staging / ".owner"
+        entries = list(staging.iterdir())
+        if entries and (
+            marker not in entries or marker.is_symlink() or not marker.is_file() or
+            read_bytes(marker, 128) != token + b"\n"
+        ):
+            raise ReplayError("execution bundle staging is not program-owned")
+        shutil.rmtree(staging)
+    if token is None:
+        if owner_path.exists():
+            raise ReplayError("execution bundle staging owner is invalid")
+        token = os.urandom(32).hex().encode()
+        descriptor = os.open(owner_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL |
+                             getattr(os, "O_NOFOLLOW", 0), 0o600)
+        with os.fdopen(descriptor, "wb") as owner:
+            owner.write(b"ystack-delivery-execution-v1:" + token + b"\n")
+            owner.flush()
+            os.fsync(owner.fileno())
+    os.mkdir(staging, 0o700)
+    atomic_bytes(staging / ".owner", token + b"\n")
+    os.chmod(staging / ".owner", 0o400)
+    for relative, data in source_bytes.items():
+        destination = staging / relative
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        atomic_bytes(destination, data)
+        mode = 0o500 if relative.endswith(".sh") or relative.startswith(".dependencies/") else 0o400
+        os.chmod(destination, mode)
+    os.replace(staging, root)
+    directory = os.open(state_dir, os.O_DIRECTORY)
+    try:
+        os.fsync(directory)
+    finally:
+        os.close(directory)
+    os.unlink(owner_path)
+    return root
+
+
+def execution_sources_match(repository, arguments, execution):
+    try:
+        return all(
+            digest_bytes(data) == digest_bytes(read_bytes(trusted_file(execution / relative), MAX_INPUT_BYTES))
+            for relative, data in execution_source_bytes(repository, arguments).items()
+        )
+    except (OSError, ReplayError):
+        return False
+
+
+def driver_identity():
+    if not isinstance(LOADED_DRIVER_BYTES, bytes) or len(LOADED_DRIVER_BYTES) > MAX_INPUT_BYTES:
+        raise ReplayError("loaded replay driver identity is unavailable")
+    return digest_bytes(LOADED_DRIVER_BYTES)
+
+
+def materializer_package_identity(repository):
+    core_path = trusted_file(repository / "scripts/core-contract.sh")
+    core_bytes = core_path.read_bytes()
+    match = re.search(
+        rb"^PORTABLE_CORE_GENERATION='(g-[0-9a-f]{64})'$", core_bytes, re.MULTILINE
+    )
+    if match is None:
+        raise ReplayError("materializer package generation is unavailable")
+    generation = match.group(1).decode()
+    files = {
+        relative: digest_bytes(trusted_file(repository / relative).read_bytes())
+        for relative in package_paths(generation)
+    }
+    package = {"generation_id": generation, "files": files}
+    package["sha256"] = digest_bytes(canonical(package))
+    return package
+
+
+def input_identity(input_value, input_sha, arguments, execution):
+    try:
+        request = input_value["stage_request"]
+        request_sha = request["sha256"]
+        body = request["content"]["body"]
+        source = body["target_revision"]["value"]
+        source_tree_id = next(
+            item["value"]["value"]["value"]["object_id"]
+            for item in body["inputs"]
+            if item["input_id"] == body["operation"]["arguments"]["source_tree_input_id"]
+        )
+    except (KeyError, StopIteration, TypeError) as error:
+        raise ReplayError("materialization input lacks an exact source identity") from error
+    if not isinstance(request_sha, str) or not re.fullmatch(r"[0-9a-f]{64}", request_sha):
+        raise ReplayError("materialization input request identity is invalid")
+    if not isinstance(source_tree_id, str) or not OID.fullmatch(source_tree_id):
+        raise ReplayError("materialization input tree identity is invalid")
+    if not isinstance(source, dict) or not isinstance(source.get("commit_id"), str) or \
+       not OID.fullmatch(source["commit_id"]):
+        raise ReplayError("materialization input commit identity is invalid")
+    if not isinstance(source.get("hash_algorithm"), str) or \
+       source["hash_algorithm"] not in {"sha1", "sha256"}:
+        raise ReplayError("materialization input hash algorithm is invalid")
+    expected = arguments.expected_sha256
+    if not re.fullmatch(r"[0-9a-f]{64}", expected):
+        raise ReplayError("expected verifier digest is invalid")
+    package = materializer_package_identity(execution)
+    identity = {
+        "input_sha256": input_sha,
+        "request_sha256": request_sha,
+        "source_commit_id": source["commit_id"],
+        "source_tree_id": source_tree_id,
+        "source_hash_algorithm": source.get("hash_algorithm"),
+        "verifier": {
+            "id": "delivery.fixed-content-sha256.v1",
+            "path": safe_path(arguments.verify_path),
+            "expected_sha256": expected,
+        },
+        "driver_sha256": driver_identity(),
+        "materializer_sha256": package["files"][PACKAGE_FILES[0]],
+        "materializer_package": package,
+        "closure_helper_sha256": digest_bytes(read_bytes(
+            trusted_file(execution / ".dependencies/object-closure"), MAX_INPUT_BYTES
+        )),
+        "jq_sha256": digest_bytes(read_bytes(
+            trusted_file(execution / ".dependencies/jq"), MAX_INPUT_BYTES
+        )),
+        "source_repository_id": arguments.source_repository_id,
+    }
+    identity["run_key"] = digest_bytes(canonical(identity))
+    return identity
+
+
+def run_materializer(arguments, execution, input_path, identity, candidate_root=None, scratch_root=None):
+    candidate_root = Path(arguments.candidate_root).resolve() if candidate_root is None else candidate_root
+    scratch_root = Path(arguments.scratch_root).resolve() if scratch_root is None else scratch_root
+    command = [
+        str(execution / PACKAGE_FILES[0]), "materialize", str(input_path),
+        arguments.source_repository_id, str(Path(arguments.source_git_dir).resolve()),
+        str(candidate_root), str(scratch_root),
+        str(execution / ".dependencies/object-closure"), str(execution / ".dependencies/jq"),
+    ]
+    environment = {"PATH": "/usr/bin:/bin", "LC_ALL": "C"}
+    result = subprocess.run(command, env=environment, stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE, check=False)
+    if result.returncode != 0 or len(result.stdout) > MAX_INPUT_BYTES:
+        raise ReplayError("materialization did not complete")
+    try:
+        response = json.loads(result.stdout)
+        receipt_text = response["payloads"][0]["data"]
+        receipt = json.loads(receipt_text)
+        candidate = receipt["candidate"]
+        receipt_sha = response["payloads"][0]["sha256"]
+        if (
+            receipt_sha != digest_bytes(receipt_text.encode()) or
+            receipt["request_ref"]["sha256"] != identity["request_sha256"] or
+            response["stage_result"]["body"]["request_ref"]["sha256"] != identity["request_sha256"] or
+            receipt["source"] != {
+                "repository_id": identity["source_repository_id"],
+                "hash_algorithm": identity["source_hash_algorithm"],
+                "commit_id": identity["source_commit_id"],
+                "tree_id": identity["source_tree_id"],
+            }
+        ):
+            raise ReplayError("materializer response does not match the input snapshot")
+        return {
+            "response_sha256": digest_bytes(result.stdout),
+            "receipt_sha256": receipt_sha,
+            "candidate_commit_id": candidate["commit_id"],
+            "candidate_tree_id": candidate["tree_id"],
+            "candidate_parent_commit_id": candidate["parent_commit_id"],
+        }
+    except (KeyError, IndexError, TypeError, json.JSONDecodeError) as error:
+        raise ReplayError("materializer response is malformed") from error
+
+
+def candidate_identity(candidate_root, source_commit):
+    repository = Path(candidate_root).resolve() / "repository.git"
+    if not repository.is_dir() or repository.is_symlink():
+        return None
+    values = []
+    for revision in ("refs/heads/candidate", "refs/heads/candidate^{tree}"):
+        result = subprocess.run(["/usr/bin/git", f"--git-dir={repository}", "rev-parse", revision],
+                                env=GIT_ENVIRONMENT, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False)
+        value = result.stdout.decode().strip()
+        if result.returncode != 0 or not OID.fullmatch(value):
+            return None
+        values.append(value)
+    if values[0] == source_commit:
+        return {"candidate_commit_id": values[0], "candidate_tree_id": values[1],
+                "candidate_parent_commit_id": source_commit}
+    result = subprocess.run(["/usr/bin/git", f"--git-dir={repository}", "rev-parse", "refs/heads/candidate^"],
+                            env=GIT_ENVIRONMENT, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False)
+    parent = result.stdout.decode().strip()
+    if result.returncode != 0 or not OID.fullmatch(parent):
+        return None
+    return {"candidate_commit_id": values[0], "candidate_tree_id": values[1],
+            "candidate_parent_commit_id": parent}
+
+
+def await_guard_prepared(process):
+    # Ownership of the candidate ref comes from git's own transaction
+    # acknowledgement, never from the lock file existing: a lock another process
+    # holds fails our prepare, and git then closes stdout without "prepare: ok".
+    deadline = time.monotonic() + GUARD_ACKNOWLEDGEMENT_SECONDS
+    descriptor = process.stdout.fileno()
+    pending = b""
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            process.kill()
+            raise ReplayError("candidate repository identity guard timed out")
+        readable, _, _ = select.select([descriptor], [], [], remaining)
+        if not readable:
+            process.kill()
+            raise ReplayError("candidate repository identity guard timed out")
+        chunk = os.read(descriptor, MAX_GUARD_LINE_BYTES)
+        if not chunk:
+            raise ReplayError("candidate repository identity guard failed")
+        lines = (pending + chunk).split(b"\n")
+        pending = lines.pop()
+        if len(pending) > MAX_GUARD_LINE_BYTES:
+            raise ReplayError("candidate repository identity guard failed")
+        if b"prepare: ok" in lines:
+            return
+
+
+@contextmanager
+def hold_candidate_ref(candidate_root, expected_commit):
+    repository = Path(candidate_root).resolve() / "repository.git"
+    lock_path = repository / "refs/heads/candidate.lock"
+    command = ["/usr/bin/git", f"--git-dir={repository}", "-c", "core.hooksPath=/dev/null",
+               "update-ref", "--stdin"]
+    process = subprocess.Popen(command, env=GIT_ENVIRONMENT, stdin=subprocess.PIPE,
+                               stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    try:
+        process.stdin.write(
+            f"option no-deref\nstart\nverify refs/heads/candidate {expected_commit}\nprepare\n".encode()
+        )
+        process.stdin.flush()
+        await_guard_prepared(process)
+        if lock_path.is_symlink() or not lock_path.is_file():
+            raise ReplayError("candidate repository identity guard failed")
+        symbolic = subprocess.run(
+            ["/usr/bin/git", f"--git-dir={repository}", "symbolic-ref", "-q", "refs/heads/candidate"],
+            env=GIT_ENVIRONMENT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False
+        )
+        if symbolic.returncode != 1:
+            raise ReplayError("candidate repository identity guard failed")
+        yield
+    finally:
+        if process.poll() is None:
+            try:
+                process.stdin.write(b"abort\n")
+                process.stdin.flush()
+            except (BrokenPipeError, OSError):
+                pass
+        if process.stdin is not None:
+            process.stdin.close()
+        try:
+            process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            process.terminate()
+            process.wait(timeout=5)
+        if process.stdout is not None:
+            process.stdout.close()
+
+
+def reconcile_materialization(arguments, execution, input_path, identity, state_dir):
+    existing = candidate_identity(arguments.candidate_root, identity["source_commit_id"])
+    if existing is None:
+        return None
+    # One fixed pair of staging directories, cleared on entry and on exit, so a
+    # crash mid-reconcile can never accumulate materialized repositories.
+    recovery_candidate = state_dir / "reconcile-candidate"
+    recovery_scratch = state_dir / "reconcile-scratch"
+    for stale in (recovery_candidate, recovery_scratch):
+        if stale.is_symlink() or stale.exists():
+            if stale.is_symlink() or not stale.is_dir():
+                raise ReplayError("reconcile staging is unavailable")
+            shutil.rmtree(stale)
+    os.mkdir(recovery_candidate, 0o700)
+    os.mkdir(recovery_scratch, 0o700)
+    try:
+        recomputed = run_materializer(arguments, execution, input_path, identity,
+                                      recovery_candidate, recovery_scratch)
+    finally:
+        shutil.rmtree(recovery_candidate, ignore_errors=True)
+        shutil.rmtree(recovery_scratch, ignore_errors=True)
+    if existing != {
+        "candidate_commit_id": recomputed["candidate_commit_id"],
+        "candidate_tree_id": recomputed["candidate_tree_id"],
+        "candidate_parent_commit_id": recomputed["candidate_parent_commit_id"],
+    }:
+        raise ReplayError("existing candidate does not match frozen materialization input")
+    return recomputed
+
+
+def verify_candidate(candidate_root, candidate_tree, path, expected):
+    repository = Path(candidate_root).resolve() / "repository.git"
+    if not repository.is_dir() or repository.is_symlink() or not OID.fullmatch(candidate_tree):
+        raise ReplayError("candidate repository identity is unavailable")
+    object_name = f"{candidate_tree}:{path}"
+    size = subprocess.run(["/usr/bin/git", f"--git-dir={repository}", "cat-file", "-s", object_name],
+                          env=GIT_ENVIRONMENT, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False)
+    if size.returncode != 0 or not size.stdout.strip().isdigit() or int(size.stdout) > MAX_VERIFIED_BLOB_BYTES:
+        raise ReplayError("fixed verifier cannot read the candidate blob")
+    blob = subprocess.run(["/usr/bin/git", f"--git-dir={repository}", "cat-file", "blob", object_name],
+                          env=GIT_ENVIRONMENT, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False)
+    if blob.returncode != 0 or len(blob.stdout) != int(size.stdout):
+        raise ReplayError("fixed verifier could not read the candidate blob")
+    actual = digest_bytes(blob.stdout)
+    if actual != expected:
+        raise ReplayError("fixed verifier digest mismatch")
+    return actual
+
+
+def revalidate_candidate(arguments, state):
+    expected = {
+        name: state["materialization"][name]
+        for name in ("candidate_commit_id", "candidate_tree_id", "candidate_parent_commit_id")
+    }
+    if candidate_identity(arguments.candidate_root, state["identity"]["source_commit_id"]) != expected:
+        raise ReplayError("candidate repository no longer matches saved materialization")
+    verify_candidate(arguments.candidate_root, expected["candidate_tree_id"],
+                     state["identity"]["verifier"]["path"],
+                     state["identity"]["verifier"]["expected_sha256"])
+
+
+def observation(path, kind, identity, candidate_commit_id, field):
+    if path is None:
+        return None
+    source = read_bytes(path, MAX_OBSERVATION_BYTES)
+    value = parse_json(source)
+    source_sha = digest_bytes(source)
+    if not isinstance(value, dict) or value.get("schema_version") != 1 or value.get("kind") != kind:
+        raise ReplayError("offline observation is malformed")
+    if not isinstance(value.get("actor_id"), str) or not ACTOR.fullmatch(value["actor_id"]):
+        raise ReplayError("offline observation actor is invalid")
+    # Two candidate commits can carry one tree, so the commit binds the
+    # observation to this exact candidate and the tree alone never does.
+    if value.get("request_sha256") != identity["request_sha256"] or \
+       value.get("candidate_tree_id") != identity["candidate_tree_id"] or \
+       value.get("candidate_commit_id") != candidate_commit_id:
+        raise ReplayError("offline observation does not match this candidate")
+    return {"actor_id": value["actor_id"], field: value.get(field), "sha256": source_sha}
+
+
+def validate_state(state, identity):
+    if not isinstance(state, dict) or state.get("schema_version") != 1 or \
+       state.get("kind") != "delivery_replay_state" or state.get("authority") != "none" or \
+       state.get("qualification") != "unavailable":
+        raise ReplayError("state journal is malformed")
+    saved = state.get("identity")
+    if not isinstance(saved, dict) or any(
+        not isinstance(saved.get(name), str) or not re.fullmatch(r"[0-9a-f]{64}", saved[name])
+        for name in ("input_sha256", "request_sha256", "driver_sha256", "materializer_sha256",
+                     "closure_helper_sha256", "jq_sha256", "run_key")
+    ) or not isinstance(saved.get("source_repository_id"), str) or \
+       not REPOSITORY_ID.fullmatch(saved["source_repository_id"]) or \
+       saved.get("source_hash_algorithm") not in {"sha1", "sha256"} or \
+       any(not isinstance(saved.get(name), str) or not OID.fullmatch(saved[name])
+           for name in ("source_commit_id", "source_tree_id")) or \
+       not isinstance(saved.get("verifier"), dict) or \
+       not isinstance(saved["verifier"].get("id"), str) or \
+       not isinstance(saved["verifier"].get("path"), str) or \
+       not re.fullmatch(r"[0-9a-f]{64}", str(saved["verifier"].get("expected_sha256", ""))):
+        raise ReplayError("state journal identity is malformed")
+    package = saved.get("materializer_package")
+    if not isinstance(package, dict) or not isinstance(package.get("generation_id"), str) or \
+       not re.fullmatch(r"g-[0-9a-f]{64}", package["generation_id"]) or \
+       not isinstance(package.get("files"), dict) or \
+       set(package["files"]) != set(package_paths(package["generation_id"])) or any(
+           not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value)
+           for value in package["files"].values()
+       ) or not isinstance(package.get("sha256"), str) or \
+       package["sha256"] != digest_bytes(canonical({
+           "generation_id": package["generation_id"], "files": package["files"]
+       })) or saved["materializer_sha256"] != package["files"][PACKAGE_FILES[0]]:
+        raise ReplayError("state journal materializer package is malformed")
+    phase = state.get("phase")
+    if phase not in {"materializing", "verifying", "review-wait", "publish-wait", "completed-offline", "failed"}:
+        raise ReplayError("state journal phase is malformed")
+    needs_materialization = phase in {"verifying", "review-wait", "publish-wait", "completed-offline"}
+    for name in ("candidate_commit_id", "candidate_tree_id"):
+        if (needs_materialization and name not in saved) or (
+            name in saved and (not isinstance(saved[name], str) or not OID.fullmatch(saved[name]))
+        ):
+            raise ReplayError("state journal candidate identity is malformed")
+    materialization = state.get("materialization")
+    if needs_materialization and (not isinstance(materialization, dict) or any(
+        not isinstance(materialization.get(name), str) or not OID.fullmatch(materialization[name])
+        for name in ("candidate_commit_id", "candidate_tree_id", "candidate_parent_commit_id")
+    ) or any(
+        not isinstance(materialization.get(name), str) or not re.fullmatch(r"[0-9a-f]{64}", materialization[name])
+        for name in ("response_sha256", "receipt_sha256")
+    )):
+        raise ReplayError("state journal materialization is malformed")
+    if needs_materialization and any(
+        saved[name] != materialization[name]
+        for name in ("candidate_commit_id", "candidate_tree_id")
+    ):
+        raise ReplayError("state journal candidate identity does not match materialization")
+    if phase in {"review-wait", "publish-wait", "completed-offline"}:
+        verification = state.get("verification")
+        if verification != {
+            "id": saved["verifier"]["id"],
+            "path": saved["verifier"]["path"],
+            "sha256": saved["verifier"]["expected_sha256"],
+        }:
+            raise ReplayError("state journal verification is malformed")
+    if phase in {"publish-wait", "completed-offline"}:
+        review = state.get("review")
+        if not isinstance(review, dict) or not isinstance(review.get("actor_id"), str) or \
+           not ACTOR.fullmatch(review["actor_id"]) or \
+           review.get("verdict") != "clean" or not re.fullmatch(r"[0-9a-f]{64}", str(review.get("sha256", ""))):
+            raise ReplayError("state journal review is malformed")
+    if phase == "completed-offline":
+        publisher = state.get("publisher")
+        if not isinstance(publisher, dict) or not isinstance(publisher.get("actor_id"), str) or \
+           not ACTOR.fullmatch(publisher["actor_id"]) or \
+           publisher.get("disposition") != "offline-simulated" or \
+           not re.fullmatch(r"[0-9a-f]{64}", str(publisher.get("sha256", ""))):
+            raise ReplayError("state journal publisher is malformed")
+    if phase == "failed" and not isinstance(state.get("reason"), str):
+        raise ReplayError("state journal failure is malformed")
+
+
+def result(state):
+    print(json.dumps({"kind": "delivery_replay_receipt", "authority": "none",
+                      "qualification": "unavailable", "offline_simulation": True,
+                      "state": state}, sort_keys=True, separators=(",", ":")), flush=True)
+
+
+def stop_if_interrupted(state, interrupted):
+    if not interrupted["value"]:
+        return False
+    if state is not None:
+        result(state)
+    return True
+
+
+def replay_locked(arguments, state_dir):
+    repository = Path(__file__).resolve().parents[2]
+    state_path = state_dir / "run.json"
+    input_snapshot_path = state_dir / "materialization-input.json"
+    lock_path = state_dir / "replay.lock"
+    interrupted = {"value": False}
+    previous_term = signal.getsignal(signal.SIGTERM)
+    previous_int = signal.getsignal(signal.SIGINT)
+    signal.signal(signal.SIGTERM, lambda *_: interrupted.__setitem__("value", True))
+    signal.signal(signal.SIGINT, lambda *_: interrupted.__setitem__("value", True))
+    try:
+        lock_descriptor = os.open(lock_path, os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600)
+        with os.fdopen(lock_descriptor, "a+b") as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX)
+            execution = create_execution_snapshot(repository, arguments, state_dir)
+            sources_match = execution_sources_match(repository, arguments, execution)
+            if not REPOSITORY_ID.fullmatch(arguments.source_repository_id):
+                raise ReplayError("source repository id is invalid")
+            input_bytes = read_bytes(arguments.input, MAX_INPUT_BYTES)
+            input_value = parse_json(input_bytes)
+            input_sha = digest_bytes(input_bytes)
+            identity = input_identity(input_value, input_sha, arguments, execution)
+            state = None
+            if state_path.exists():
+                state = parse_json(read_bytes(state_path, MAX_OBSERVATION_BYTES))
+                validate_state(state, identity)
+            if state is not None and any(state["identity"].get(name) != value for name, value in identity.items()):
+                result({"phase": "stale", "reason": "run identity changed"})
+                return 2
+            if not sources_match:
+                if state is not None:
+                    result({"phase": "stale", "reason": "execution dependencies changed"})
+                    return 2
+                raise ReplayError("execution bundle does not match current dependencies")
+            if stop_if_interrupted(state, interrupted):
+                return 75
+            fresh_run = state is None
+            if fresh_run:
+                state = {"schema_version": 1, "kind": "delivery_replay_state", "identity": identity,
+                         "phase": "materializing", "authority": "none", "qualification": "unavailable"}
+                atomic_bytes(input_snapshot_path, input_bytes)
+                atomic_json(state_path, state)
+            elif not input_snapshot_path.is_file() or input_snapshot_path.is_symlink() or (
+                digest_bytes(read_bytes(input_snapshot_path, MAX_INPUT_BYTES)) != identity["input_sha256"]
+            ):
+                raise ReplayError("saved materialization input snapshot is unavailable")
+            if state["phase"] == "failed":
+                if state.get("recoverable"):
+                    state["recovery"] = "start a new replay with fresh empty candidate, scratch, and state directories"
+                    atomic_json(state_path, state)
+                result(state)
+                return 1
+            if state["phase"] == "completed-offline":
+                revalidate_candidate(arguments, state)
+                if stop_if_interrupted(state, interrupted):
+                    return 75
+                for supplied, kind, field, recorded in (
+                    (arguments.review_observation, "delivery_replay_review_observation", "verdict", state.get("review")),
+                    (arguments.publisher_observation, "delivery_replay_publisher_observation", "disposition", state.get("publisher")),
+                ):
+                    if supplied is not None:
+                        supplied_observation = observation(
+                            supplied, kind, state["identity"],
+                            state["materialization"]["candidate_commit_id"], field
+                        )
+                        if stop_if_interrupted(state, interrupted):
+                            return 75
+                        if supplied_observation != recorded:
+                            raise ReplayError("supplied offline observation changed after completion")
+                result(state)
+                return 0
+            if state["phase"] == "materializing":
+                try:
+                    # Only a resumed run may adopt a candidate that is already in
+                    # the candidate root; a fresh run always goes through the
+                    # materializer, whose root check refuses a pre-populated root.
+                    reconciled = None if fresh_run else reconcile_materialization(
+                        arguments, execution, input_snapshot_path, identity, state_dir
+                    )
+                    state["materialization"] = reconciled or run_materializer(
+                        arguments, execution, input_snapshot_path, identity
+                    )
+                except ReplayError as error:
+                    if stop_if_interrupted(state, interrupted):
+                        return 75
+                    state.update({"phase": "failed", "recoverable": True, "reason": str(error)})
+                    atomic_json(state_path, state)
+                    result(state)
+                    return 1
+                state["identity"].update({
+                    "candidate_commit_id": state["materialization"]["candidate_commit_id"],
+                    "candidate_tree_id": state["materialization"]["candidate_tree_id"],
+                })
+                state["phase"] = "verifying"
+                atomic_json(state_path, state)
+                if stop_if_interrupted(state, interrupted):
+                    return 75
+            if state["phase"] == "verifying":
+                try:
+                    state["verification"] = {"id": identity["verifier"]["id"], "path": identity["verifier"]["path"],
+                                             "sha256": verify_candidate(arguments.candidate_root, state["identity"]["candidate_tree_id"],
+                                                                        identity["verifier"]["path"], identity["verifier"]["expected_sha256"])}
+                except ReplayError as error:
+                    if stop_if_interrupted(state, interrupted):
+                        return 75
+                    state.update({"phase": "failed", "recoverable": False, "reason": str(error)})
+                    atomic_json(state_path, state)
+                    result(state)
+                    return 1
+                state["phase"] = "review-wait"
+                atomic_json(state_path, state)
+                if stop_if_interrupted(state, interrupted):
+                    return 75
+            if state["phase"] == "review-wait":
+                revalidate_candidate(arguments, state)
+                if stop_if_interrupted(state, interrupted):
+                    return 75
+                review = observation(arguments.review_observation, "delivery_replay_review_observation",
+                                     state["identity"], state["materialization"]["candidate_commit_id"], "verdict")
+                if stop_if_interrupted(state, interrupted):
+                    return 75
+                if review is None:
+                    result(state)
+                    return 0
+                if review["verdict"] != "clean":
+                    state.update({"phase": "failed", "recoverable": False, "reason": "offline review did not report clean"})
+                    atomic_json(state_path, state)
+                    result(state)
+                    return 1
+                state["review"] = review
+                state["phase"] = "publish-wait"
+                atomic_json(state_path, state)
+            if state["phase"] == "publish-wait":
+                with hold_candidate_ref(arguments.candidate_root,
+                                        state["materialization"]["candidate_commit_id"]):
+                    revalidate_candidate(arguments, state)
+                    if stop_if_interrupted(state, interrupted):
+                        return 75
+                    if arguments.review_observation is not None:
+                        supplied_review = observation(arguments.review_observation,
+                                                      "delivery_replay_review_observation",
+                                                      state["identity"],
+                                                      state["materialization"]["candidate_commit_id"], "verdict")
+                        if stop_if_interrupted(state, interrupted):
+                            return 75
+                        if supplied_review != state.get("review"):
+                            raise ReplayError("supplied offline review changed after review wait")
+                    publisher = observation(arguments.publisher_observation,
+                                            "delivery_replay_publisher_observation",
+                                            state["identity"],
+                                            state["materialization"]["candidate_commit_id"], "disposition")
+                    if stop_if_interrupted(state, interrupted):
+                        return 75
+                    if publisher is None:
+                        result(state)
+                        return 0
+                    if publisher["disposition"] != "offline-simulated":
+                        state.update({"phase": "failed", "recoverable": False, "reason": "offline publisher disposition is invalid"})
+                        atomic_json(state_path, state)
+                        result(state)
+                        return 1
+                    state["publisher"] = publisher
+                    state["phase"] = "completed-offline"
+                    atomic_json(state_path, state)
+                    result(state)
+                    return 0
+            result(state)
+            return 1
+    finally:
+        signal.signal(signal.SIGTERM, previous_term)
+        signal.signal(signal.SIGINT, previous_int)
+
+
+def replay(arguments):
+    state_dir = private_directory(arguments.state_dir)
+    disjoint(state_dir, arguments.source_git_dir, arguments.candidate_root, arguments.scratch_root)
+    return replay_locked(arguments, state_dir)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", required=True)
+    parser.add_argument("--source-repository-id", required=True)
+    parser.add_argument("--source-git-dir", required=True)
+    parser.add_argument("--candidate-root", required=True)
+    parser.add_argument("--scratch-root", required=True)
+    parser.add_argument("--state-dir", required=True)
+    parser.add_argument("--closure-helper", required=True)
+    parser.add_argument("--jq-bin", required=True)
+    parser.add_argument("--verify-path", required=True)
+    parser.add_argument("--expected-sha256", required=True)
+    parser.add_argument("--review-observation")
+    parser.add_argument("--publisher-observation")
+    try:
+        return replay(parser.parse_args())
+    except (OSError, ReplayError) as error:
+        print(f"delivery replay: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/test/delivery-replay.test.sh b/scripts/test/delivery-replay.test.sh
new file mode 100755
index 0000000..8fab3e4
--- /dev/null
+++ b/scripts/test/delivery-replay.test.sh
@@ -0,0 +1,1152 @@
+#!/usr/bin/env bash
+# shellcheck disable=SC2016
+set -euo pipefail
+export LC_ALL=C
+export PYTHONDONTWRITEBYTECODE=1
+umask 077
+
+root=$(CDPATH='' cd -P -- "${BASH_SOURCE[0]%/*}/../.." && pwd -P)
+replay="$root/delivery/v1/replay.py"
+fixture_builder="$root/scripts/test/local-git-materializer-fixtures.sh"
+closure_source="$root/adapters/local-git-materializer/v1/object-closure.c"
+python_with_int_limit=$(command -v python3)
+tmp=$(/usr/bin/mktemp -d "${TMPDIR:-/tmp}/ystack-delivery-replay.XXXXXX")
+cleanup() { /bin/rm -rf -- "$tmp"; }
+trap cleanup EXIT
+
+sha_file() { /usr/bin/shasum -a 256 "$1" | /usr/bin/awk '{print $1}'; }
+fail() { printf 'FAIL: %s\n' "$1" >&2; exit 1; }
+passed=0
+pass() { passed=$((passed + 1)); printf 'ok %s - %s\n' "$passed" "$1"; }
+
+platform=$(/usr/bin/uname -s):$(/usr/bin/uname -m)
+case "$platform" in
+  Linux:x86_64) asset=jq-linux64; asset_sha=af986793a515d500ab2d35f8d2aecd656e764504b789b66d7e1a0b727a124c44 ;;
+  Darwin:x86_64|Darwin:arm64) asset=jq-osx-amd64; asset_sha=5c0a0a3ea600f302ee458b30317425dd9632d1ad8882259fcaf4e9b868b2b1ef ;;
+  *) fail "unsupported host $platform" ;;
+esac
+# Run alone on a fresh restore this suite cannot rely on another suite having
+# filled the shared jq 1.6 cache, so it fetches the pinned release itself.
+jq_cache_dir="${TMPDIR:-/tmp}/ystack-portable-core-jq16"
+/bin/mkdir -p "$jq_cache_dir"
+jq_bin="$jq_cache_dir/$asset"
+if [ ! -f "$jq_bin" ] || [ -L "$jq_bin" ] || [ "$(sha_file "$jq_bin")" != "$asset_sha" ]; then
+  download=$(/usr/bin/mktemp "$jq_cache_dir/.jq-1.6.XXXXXX")
+  /usr/bin/curl --proto '=https' --tlsv1.2 -fsSL \
+    "https://github.com/jqlang/jq/releases/download/jq-1.6/$asset" -o "$download"
+  [ "$(sha_file "$download")" = "$asset_sha" ] || fail 'jq release digest'
+  /bin/chmod 0555 "$download"
+  /bin/mv "$download" "$jq_bin"
+fi
+[ -f "$jq_bin" ] && [ ! -L "$jq_bin" ] && [ "$(sha_file "$jq_bin")" = "$asset_sha" ] ||
+  fail 'pinned jq 1.6 is required'
+
+runtime="$tmp/runtime"
+/bin/mkdir -m 700 "$runtime" "$tmp/home"
+/bin/cp "$jq_bin" "$runtime/jq"
+/bin/chmod 0555 "$runtime/jq"
+jq_bin="$runtime/jq"
+PATH="$runtime:/usr/bin:/bin"
+export PATH
+[ "$(command -v jq)" = "$runtime/jq" ] || fail 'private jq is not first on PATH'
+/usr/bin/cc -std=c11 -Wall -Wextra -Werror -O2 "$closure_source" -o "$runtime/object-closure"
+/bin/chmod 0555 "$runtime/object-closure"
+
+git_clean() {
+  /usr/bin/env -i HOME="$tmp/home" TMPDIR="$tmp" PATH=/usr/bin:/bin LC_ALL=C \
+    GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null GIT_NO_REPLACE_OBJECTS=1 \
+    GIT_NO_LAZY_FETCH=1 GIT_TERMINAL_PROMPT=0 /usr/bin/git --no-replace-objects "$@"
+}
+make_source() {
+  local destination=$1 blob tree commit
+  /bin/mkdir -m 700 "$destination"
+  git_clean init -q --bare "$destination"
+  blob=$(printf '%s\n' alpha beta | git_clean --git-dir="$destination" hash-object -w --stdin)
+  tree=$(printf '100644 blob %s\tsource.txt\n' "$blob" | git_clean --git-dir="$destination" mktree)
+  commit=$(printf '%s\n' source | /usr/bin/env -i HOME="$tmp/home" PATH=/usr/bin:/bin LC_ALL=C \
+    GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null GIT_AUTHOR_NAME=fixture \
+    GIT_AUTHOR_EMAIL=fixture@example.invalid GIT_COMMITTER_NAME=fixture \
+    GIT_COMMITTER_EMAIL=fixture@example.invalid GIT_AUTHOR_DATE=2000-01-01T00:00:00Z \
+    GIT_COMMITTER_DATE=2000-01-01T00:00:00Z /usr/bin/git --git-dir="$destination" commit-tree "$tree")
+  git_clean --git-dir="$destination" update-ref refs/heads/main "$commit"
+  printf '%s %s\n' "$commit" "$tree"
+}
+
+make_source_with_ancestor() {
+  local destination=$1 blob tree base commit
+  /bin/mkdir -m 700 "$destination"
+  git_clean init -q --bare "$destination"
+  blob=$(printf '%s\n' alpha beta | git_clean --git-dir="$destination" hash-object -w --stdin)
+  tree=$(printf '100644 blob %s\tsource.txt\n' "$blob" | git_clean --git-dir="$destination" mktree)
+  base=$(printf '%s\n' base | /usr/bin/env -i HOME="$tmp/home" PATH=/usr/bin:/bin LC_ALL=C \
+    GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null GIT_AUTHOR_NAME=fixture \
+    GIT_AUTHOR_EMAIL=fixture@example.invalid GIT_COMMITTER_NAME=fixture \
+    GIT_COMMITTER_EMAIL=fixture@example.invalid /usr/bin/git --git-dir="$destination" commit-tree "$tree")
+  commit=$(printf '%s\n' source | /usr/bin/env -i HOME="$tmp/home" PATH=/usr/bin:/bin LC_ALL=C \
+    GIT_CONFIG_NOSYSTEM=1 GIT_CONFIG_GLOBAL=/dev/null GIT_AUTHOR_NAME=fixture \
+    GIT_AUTHOR_EMAIL=fixture@example.invalid GIT_COMMITTER_NAME=fixture \
+    GIT_COMMITTER_EMAIL=fixture@example.invalid /usr/bin/git --git-dir="$destination" commit-tree "$tree" -p "$base")
+  git_clean --git-dir="$destination" update-ref refs/heads/main "$commit"
+  printf '%s %s\n' "$commit" "$tree"
+}
+
+make_empty_input() {
+  local input=$1 output=$2
+  local intermediate="$output.intermediate" request="$output.request"
+  "$jq_bin" -S -c '(.stage_request.content.body.inputs[] | select(.input_id=="input.producer-patch") | .value.value.value.sha256) = $sha |
+    (.payloads[] | select(.input_id=="input.producer-patch") | .data) = "" |
+    (.trust_context.verified_payloads[] | select(.input_id=="input.producer-patch") | .content.data) = "" |
+    (.trust_context.verified_payloads[] | select(.input_id=="input.producer-patch") | .sha256) = $sha' \
+    --arg sha "$(printf '' | /usr/bin/shasum -a 256 | /usr/bin/awk '{print $1}')" "$input" >"$intermediate"
+  "$jq_bin" -S -c '.stage_request.content' "$intermediate" >"$request"
+  "$jq_bin" -S -c --arg sha "$(sha_file "$request")" '.stage_request.sha256=$sha' "$intermediate" >"$output"
+}
+
+read -r source_commit source_tree < <(make_source "$tmp/source.git")
+"$fixture_builder" build "$tmp/fixture" "$jq_bin" sha1 "$source_commit" "$source_tree"
+base_input="$tmp/fixture/input.json"
+expected_changed=$(printf '%s\n' alpha beta gamma | /usr/bin/shasum -a 256 | /usr/bin/awk '{print $1}')
+
+run_replay() {
+  local name=$1 input=$2 expected=$3
+  local state="$tmp/$name-state" candidate="$tmp/$name-candidate" scratch="$tmp/$name-scratch"
+  /bin/mkdir -m 700 "$state" "$candidate" "$scratch"
+  python3 "$replay" --input "$input" --source-repository-id fixture.target \
+    --source-git-dir "$tmp/source.git" --candidate-root "$candidate" --scratch-root "$scratch" \
+    --state-dir "$state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+    --verify-path source.txt --expected-sha256 "$expected"
+}
+
+printf '%s\n' '#!/bin/sh' "exec '$jq_bin' \"\$@\"" >"$tmp/jq-launcher"
+/bin/chmod 0555 "$tmp/jq-launcher"
+/bin/mkdir -m 700 "$tmp/launcher-state" "$tmp/launcher-candidate" "$tmp/launcher-scratch"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target \
+  --source-git-dir "$tmp/source.git" --candidate-root "$tmp/launcher-candidate" --scratch-root "$tmp/launcher-scratch" \
+  --state-dir "$tmp/launcher-state" --closure-helper "$runtime/object-closure" --jq-bin "$tmp/jq-launcher" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/launcher.out" 2>&1; then
+  fail dependency-launcher
+fi
+grep -Fq 'delivery replay: dependency is not a native executable' "$tmp/launcher.out" ||
+  fail dependency-launcher-error
+[ ! -e "$tmp/launcher-state/run.json" ] || fail dependency-launcher-journal
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target \
+  --source-git-dir "$tmp/source.git" --candidate-root "$tmp/launcher-candidate" --scratch-root "$tmp/launcher-scratch" \
+  --state-dir "$tmp/launcher-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/launcher-retry.out"
+jq -e '.state.phase=="review-wait"' "$tmp/launcher-retry.out" >/dev/null || fail dependency-launcher-retry
+pass 'a corrected native dependency can reuse state after launcher rejection'
+
+/bin/cp "$jq_bin" "$tmp/jq-noexec"
+/bin/chmod 0444 "$tmp/jq-noexec"
+/bin/mkdir -m 700 "$tmp/noexec-state" "$tmp/noexec-candidate" "$tmp/noexec-scratch"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target \
+  --source-git-dir "$tmp/source.git" --candidate-root "$tmp/noexec-candidate" --scratch-root "$tmp/noexec-scratch" \
+  --state-dir "$tmp/noexec-state" --closure-helper "$runtime/object-closure" --jq-bin "$tmp/jq-noexec" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/noexec.out" 2>&1; then
+  fail dependency-noexec
+fi
+grep -Fq 'delivery replay: dependency is not executable' "$tmp/noexec.out" || fail dependency-noexec-error
+[ ! -e "$tmp/noexec-state/run.json" ] && [ ! -e "$tmp/noexec-state/execution" ] || fail dependency-noexec-state
+pass 'a non-executable native dependency is rejected before any snapshot grants it execute permission'
+
+python3 -c 'import sys; sys.stdout.write("[" * 100000 + "]" * 100000)' >"$tmp/deep.json"
+/bin/mkdir -m 700 "$tmp/deep-state" "$tmp/deep-candidate" "$tmp/deep-scratch"
+if python3 "$replay" --input "$tmp/deep.json" --source-repository-id fixture.target \
+  --source-git-dir "$tmp/source.git" --candidate-root "$tmp/deep-candidate" --scratch-root "$tmp/deep-scratch" \
+  --state-dir "$tmp/deep-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/deep.out" 2>&1; then
+  fail deep-json-accepted
+fi
+grep -Fq 'delivery replay: input is not JSON' "$tmp/deep.out" || fail deep-json-error
+! grep -Fq 'Traceback' "$tmp/deep.out" || fail deep-json-traceback
+pass 'deeply nested JSON is rejected as input, never as a crash'
+
+
+
+snapshot_interrupt_wrapper="$tmp/snapshot-interrupt.py"
+printf '%s\n' \
+  'import importlib.util, pathlib, sys' \
+  'path, point, arguments = sys.argv[1], sys.argv[2], sys.argv[3:]' \
+  'spec = importlib.util.spec_from_file_location("replay", path)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(path).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, path, "exec"), module.__dict__)' \
+  'if point == "directory":' \
+  '    original_mkdir = module.os.mkdir' \
+  '    def interrupted_mkdir(target, *args, **kwargs):' \
+  '        result = original_mkdir(target, *args, **kwargs)' \
+  '        if pathlib.Path(target).name == ".execution-building": raise module.ReplayError("snapshot interrupted after directory creation")' \
+  '        return result' \
+  '    module.os.mkdir = interrupted_mkdir' \
+  'else:' \
+  '    original_atomic = module.atomic_bytes' \
+  '    writes = {"count": 0}' \
+  '    def interrupted_atomic(target, data):' \
+  '        original_atomic(target, data)' \
+  '        if ".execution-building" in pathlib.Path(target).parts:' \
+  '            writes["count"] += 1' \
+  '            if writes["count"] == 2: raise module.ReplayError("snapshot interrupted during copy")' \
+  '    module.atomic_bytes = interrupted_atomic' \
+  'sys.argv = [path] + arguments' \
+  'raise SystemExit(module.main())' >"$snapshot_interrupt_wrapper"
+for snapshot_point in directory copy; do
+  mkdir -m 700 "$tmp/snapshot-$snapshot_point-state" "$tmp/snapshot-$snapshot_point-candidate" \
+    "$tmp/snapshot-$snapshot_point-scratch"
+  if python3 "$snapshot_interrupt_wrapper" "$replay" "$snapshot_point" \
+    --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/snapshot-$snapshot_point-candidate" --scratch-root "$tmp/snapshot-$snapshot_point-scratch" \
+    --state-dir "$tmp/snapshot-$snapshot_point-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+    --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/snapshot-$snapshot_point.out" 2>&1; then
+    fail "snapshot-$snapshot_point-interrupt"
+  fi
+  [ ! -e "$tmp/snapshot-$snapshot_point-state/run.json" ] || fail "snapshot-$snapshot_point-journal"
+  [ -d "$tmp/snapshot-$snapshot_point-state/.execution-building" ] || fail "snapshot-$snapshot_point-staging"
+  python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/snapshot-$snapshot_point-candidate" --scratch-root "$tmp/snapshot-$snapshot_point-scratch" \
+    --state-dir "$tmp/snapshot-$snapshot_point-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+    --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/snapshot-$snapshot_point-retry.out"
+  jq -e '.state.phase=="review-wait"' "$tmp/snapshot-$snapshot_point-retry.out" >/dev/null ||
+    fail "snapshot-$snapshot_point-retry"
+done
+
+mkdir -m 700 "$tmp/foreign-staging-state" "$tmp/foreign-staging-state/.execution-building" \
+  "$tmp/foreign-staging-candidate" "$tmp/foreign-staging-scratch"
+printf '%s\n' preserve >"$tmp/foreign-staging-state/.execution-building/sentinel"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/foreign-staging-candidate" --scratch-root "$tmp/foreign-staging-scratch" \
+  --state-dir "$tmp/foreign-staging-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/foreign-staging.out" 2>&1; then
+  fail foreign-staging
+fi
+[ "$(cat "$tmp/foreign-staging-state/.execution-building/sentinel")" = preserve ] || fail foreign-staging-preserved
+mkdir -m 700 "$tmp/foreign-link-state" "$tmp/foreign-link-target" "$tmp/foreign-link-candidate" "$tmp/foreign-link-scratch"
+printf '%s\n' preserve >"$tmp/foreign-link-target/sentinel"
+ln -s "$tmp/foreign-link-target" "$tmp/foreign-link-state/.execution-building"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/foreign-link-candidate" --scratch-root "$tmp/foreign-link-scratch" \
+  --state-dir "$tmp/foreign-link-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/foreign-link.out" 2>&1; then
+  fail foreign-staging-link
+fi
+[ -L "$tmp/foreign-link-state/.execution-building" ] && \
+  [ "$(cat "$tmp/foreign-link-target/sentinel")" = preserve ] || fail foreign-staging-link-preserved
+pass 'transactional bundle recovery preserves foreign staging data and links'
+
+run_replay changed "$base_input" "$expected_changed" >"$tmp/changed.out"
+jq -e '.state.phase=="review-wait" and .authority=="none" and .offline_simulation==true' "$tmp/changed.out" >/dev/null ||
+  fail missing-review-waits
+request_sha=$(jq -r '.identity.request_sha256' "$tmp/changed-state/run.json")
+candidate_tree=$(jq -r '.identity.candidate_tree_id' "$tmp/changed-state/run.json")
+candidate_commit=$(jq -r '.materialization.candidate_commit_id' "$tmp/changed-state/run.json")
+moved_candidate=$(printf '%s\n' moved | /usr/bin/env -i HOME="$tmp/home" PATH=/usr/bin:/bin LC_ALL=C \
+  GIT_AUTHOR_NAME=fixture GIT_AUTHOR_EMAIL=fixture@example.invalid GIT_COMMITTER_NAME=fixture \
+  GIT_COMMITTER_EMAIL=fixture@example.invalid /usr/bin/git --git-dir="$tmp/changed-candidate/repository.git" \
+  commit-tree "$candidate_tree" -p "$candidate_commit")
+expect_candidate_move_rejected() {
+  local phase=$1
+  shift
+  /usr/bin/git --git-dir="$tmp/changed-candidate/repository.git" update-ref refs/heads/candidate "$moved_candidate"
+  if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+    --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+    --expected-sha256 "$expected_changed" "$@" >"$tmp/candidate-moved-$phase.out" 2>&1; then
+    /usr/bin/git --git-dir="$tmp/changed-candidate/repository.git" update-ref refs/heads/candidate "$candidate_commit"
+    fail "candidate-moved-$phase"
+  fi
+  /usr/bin/git --git-dir="$tmp/changed-candidate/repository.git" update-ref refs/heads/candidate "$candidate_commit"
+  grep -Eq 'candidate repository no longer matches saved materialization|candidate repository identity guard failed' \
+    "$tmp/candidate-moved-$phase.out" ||
+    fail "candidate-moved-$phase-error"
+}
+pass 'changed materialization and fixed read-only verification wait for review'
+
+cp "$base_input" "$tmp/mutable-input.json"
+mkdir -m 700 "$tmp/mutation-state" "$tmp/mutation-candidate" "$tmp/mutation-scratch"
+python3 "$replay" --input "$tmp/mutable-input.json" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/mutation-candidate" --scratch-root "$tmp/mutation-scratch" --state-dir "$tmp/mutation-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/mutation.out" &
+mutation_pid=$!
+mutation_wait=0
+while [ ! -f "$tmp/mutation-state/materialization-input.json" ]; do
+  if ! kill -0 "$mutation_pid" 2>/dev/null; then
+    wait "$mutation_pid" || :
+    sed -n '1,12p' "$tmp/mutation.out" >&2
+    fail input-snapshot-start
+  fi
+  mutation_wait=$((mutation_wait + 1))
+  if [ "$mutation_wait" -gt 100 ]; then
+    kill -TERM "$mutation_pid" 2>/dev/null || :
+    wait "$mutation_pid" || :
+    fail input-snapshot-timeout
+  fi
+  sleep 0.1
+done
+printf '%s\n' '{"replaced":"after snapshot"}' >"$tmp/mutable-input.json"
+wait "$mutation_pid" || fail input-snapshot-run
+[ "$(sha_file "$tmp/mutation-state/materialization-input.json")" = "$(sha_file "$base_input")" ] || fail input-snapshot-bytes
+jq -e '.state.phase=="review-wait" and .state.identity.input_sha256==$sha' --arg sha "$(sha_file "$base_input")" \
+  "$tmp/mutation.out" >/dev/null || fail input-snapshot-output
+pass 'replacement of the original input after snapshot cannot change materialization'
+
+kill_wrapper="$tmp/kill-after-materialize.py"
+printf '%s\n' \
+  'import importlib.util, os, pathlib, signal, sys' \
+  'path, arguments = sys.argv[1], sys.argv[2:]' \
+  'spec = importlib.util.spec_from_file_location("replay", path)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(path).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, path, "exec"), module.__dict__)' \
+  'original = module.run_materializer' \
+  'def stop_after_materialization(*args, **kwargs):' \
+  '    result = original(*args, **kwargs)' \
+  '    os.kill(os.getpid(), signal.SIGKILL)' \
+  '    return result' \
+  'module.run_materializer = stop_after_materialization' \
+  'sys.argv = [path] + arguments' \
+  'raise SystemExit(module.main())' >"$kill_wrapper"
+mkdir -m 700 "$tmp/reconcile-state" "$tmp/reconcile-candidate" "$tmp/reconcile-scratch"
+if python3 "$kill_wrapper" "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-candidate" --scratch-root "$tmp/reconcile-scratch" --state-dir "$tmp/reconcile-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/reconcile-killed.out" 2>&1; then fail reconcile-kill; fi
+[ "$(jq -r '.phase' "$tmp/reconcile-state/run.json")" = materializing ] && [ -d "$tmp/reconcile-candidate/repository.git" ] ||
+  fail reconcile-window
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-candidate" --scratch-root "$tmp/reconcile-scratch" --state-dir "$tmp/reconcile-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/reconcile-retry.out"
+jq -e '.state.phase=="review-wait"' "$tmp/reconcile-retry.out" >/dev/null || fail reconcile-retry
+pass 'SIGKILL after materializer output reconciles the existing candidate once'
+
+# A fresh run (no journal yet) must not adopt a candidate already sitting in
+# the candidate root, even one that matches the input: only the materializer's
+# empty-root check may admit a root on a first run.
+mkdir -m 700 "$tmp/fresh-reuse-state" "$tmp/fresh-reuse-scratch"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-candidate" --scratch-root "$tmp/fresh-reuse-scratch" --state-dir "$tmp/fresh-reuse-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/fresh-reuse.out" 2>&1; then fail fresh-reuse-accepted; fi
+grep -Fq Traceback "$tmp/fresh-reuse.out" && fail fresh-reuse-traceback
+jq -e '.state.phase=="failed" and .state.reason=="materialization did not complete"' "$tmp/fresh-reuse.out" >/dev/null ||
+  fail fresh-reuse-outcome
+pass 'a fresh run never adopts a pre-populated candidate root'
+
+mkdir -m 700 "$tmp/repeated-kill-state" "$tmp/repeated-kill-candidate" "$tmp/repeated-kill-scratch"
+for kill_round in 1 2 3; do
+  if python3 "$kill_wrapper" "$replay" --input "$base_input" --source-repository-id fixture.target \
+    --source-git-dir "$tmp/source.git" --candidate-root "$tmp/repeated-kill-candidate" \
+    --scratch-root "$tmp/repeated-kill-scratch" --state-dir "$tmp/repeated-kill-state" \
+    --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+    --expected-sha256 "$expected_changed" >"$tmp/repeated-kill-$kill_round.out" 2>&1; then
+    fail "repeated-kill-$kill_round"
+  fi
+  [ "$(find "$tmp/repeated-kill-state" -maxdepth 1 -type d -name 'execution*' | wc -l | tr -d ' ')" = 1 ] ||
+    fail "repeated-kill-snapshot-count-$kill_round"
+done
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/repeated-kill-candidate" --scratch-root "$tmp/repeated-kill-scratch" \
+  --state-dir "$tmp/repeated-kill-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/repeated-kill-resume.out"
+jq -e '.state.phase=="review-wait"' "$tmp/repeated-kill-resume.out" >/dev/null || fail repeated-kill-resume
+pass 'repeated SIGKILL recovery reuses one bounded execution bundle'
+
+[ "$(find "$tmp/repeated-kill-state" -maxdepth 1 -name 'reconcile-*' | wc -l | tr -d ' ')" = 0 ] ||
+  fail repeated-kill-reconcile-leftovers
+pass 'crash recovery leaves no reconcile staging directories behind'
+
+
+mkdir -m 700 "$tmp/reconcile-bad-state" "$tmp/reconcile-bad-candidate" "$tmp/reconcile-bad-scratch"
+if python3 "$kill_wrapper" "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-bad-candidate" --scratch-root "$tmp/reconcile-bad-scratch" --state-dir "$tmp/reconcile-bad-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/reconcile-bad-killed.out" 2>&1; then fail reconcile-bad-kill; fi
+bad_repo="$tmp/reconcile-bad-candidate/repository.git"
+bad_commit=$(printf '%s\n' mismatch | /usr/bin/env -i HOME="$tmp/home" PATH=/usr/bin:/bin LC_ALL=C \
+  GIT_AUTHOR_NAME=fixture GIT_AUTHOR_EMAIL=fixture@example.invalid GIT_COMMITTER_NAME=fixture \
+  GIT_COMMITTER_EMAIL=fixture@example.invalid /usr/bin/git --git-dir="$bad_repo" commit-tree "$source_tree" -p "$source_commit")
+/usr/bin/git --git-dir="$bad_repo" update-ref refs/heads/candidate "$bad_commit"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-bad-candidate" --scratch-root "$tmp/reconcile-bad-scratch" --state-dir "$tmp/reconcile-bad-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/reconcile-bad.out" 2>&1; then fail reconcile-mismatch; fi
+jq -e '.state.phase=="failed" and (.state.reason|contains("does not match frozen"))' "$tmp/reconcile-bad.out" >/dev/null ||
+  fail reconcile-mismatch-state
+pass 'a mismatched interrupted candidate is rejected without cleanup'
+
+group_interrupt_wrapper="$tmp/materialization-group-interrupt.py"
+printf '%s\n' \
+  'import importlib.util, os, pathlib, signal, sys' \
+  'path, point, signal_name, arguments = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4:]' \
+  'os.setsid()' \
+  'spec = importlib.util.spec_from_file_location("replay", path)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(path).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, path, "exec"), module.__dict__)' \
+  'original = module.run_materializer' \
+  'def interrupt_materialization(*args, **kwargs):' \
+  '    if point == "before":' \
+  '        os.killpg(os.getpgrp(), getattr(signal, signal_name))' \
+  '        raise module.ReplayError("materialization did not complete")' \
+  '    original(*args, **kwargs)' \
+  '    os.killpg(os.getpgrp(), getattr(signal, signal_name))' \
+  '    raise module.ReplayError("materialization did not complete")' \
+  'def interrupt_verification(*_args, **_kwargs):' \
+  '    os.killpg(os.getpgrp(), getattr(signal, signal_name))' \
+  '    raise module.ReplayError("fixed verifier could not read the candidate blob")' \
+  'if point == "verify":' \
+  '    module.verify_candidate = interrupt_verification' \
+  'else:' \
+  '    module.run_materializer = interrupt_materialization' \
+  'sys.argv = [path] + arguments' \
+  'raise SystemExit(module.main())' >"$group_interrupt_wrapper"
+for interrupt_case in before after verify; do
+  if [ "$interrupt_case" = after ]; then interrupt_signal=SIGTERM; else interrupt_signal=SIGINT; fi
+  mkdir -m 700 "$tmp/group-$interrupt_case-state" "$tmp/group-$interrupt_case-candidate" \
+    "$tmp/group-$interrupt_case-scratch"
+  if python3 "$group_interrupt_wrapper" "$replay" "$interrupt_case" "$interrupt_signal" \
+    --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/group-$interrupt_case-candidate" --scratch-root "$tmp/group-$interrupt_case-scratch" \
+    --state-dir "$tmp/group-$interrupt_case-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+    --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/group-$interrupt_case.out"; then
+    fail "group-$interrupt_case-status"
+  else
+    interrupt_status=$?
+  fi
+  [ "$interrupt_status" -eq 75 ] || fail "group-$interrupt_case-code"
+  if [ "$interrupt_case" = before ]; then
+    expected_interrupt_phase=materializing
+    [ ! -e "$tmp/group-$interrupt_case-candidate/repository.git" ] || fail group-before-effect
+  else
+    if [ "$interrupt_case" = after ]; then expected_interrupt_phase=materializing; else expected_interrupt_phase=verifying; fi
+    [ -d "$tmp/group-$interrupt_case-candidate/repository.git" ] || fail group-after-candidate
+  fi
+  jq -e --arg phase "$expected_interrupt_phase" '.phase==$phase' \
+    "$tmp/group-$interrupt_case-state/run.json" >/dev/null || fail "group-$interrupt_case-state"
+  python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/group-$interrupt_case-candidate" --scratch-root "$tmp/group-$interrupt_case-scratch" \
+    --state-dir "$tmp/group-$interrupt_case-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+    --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/group-$interrupt_case-resume.out"
+  jq -e '.state.phase=="review-wait"' "$tmp/group-$interrupt_case-resume.out" >/dev/null ||
+    fail "group-$interrupt_case-resume"
+done
+pass 'process-group cancellation during materialization or verification resumes the same attempt'
+
+for tree_case in numeric list null; do
+  case "$tree_case" in
+    numeric) tree_value=123 ;;
+    list) tree_value='[]' ;;
+    null) tree_value=null ;;
+  esac
+  "$jq_bin" -S -c "(.stage_request.content.body.operation.arguments.source_tree_input_id) as \$id |
+    (.stage_request.content.body.inputs[] | select(.input_id == \$id) |
+    .value.value.value.object_id) = $tree_value" "$base_input" >"$tmp/$tree_case-tree.json"
+  mkdir -m 700 "$tmp/$tree_case-tree-state" "$tmp/$tree_case-tree-candidate" "$tmp/$tree_case-tree-scratch"
+  if python3 "$replay" --input "$tmp/$tree_case-tree.json" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/$tree_case-tree-candidate" --scratch-root "$tmp/$tree_case-tree-scratch" \
+    --state-dir "$tmp/$tree_case-tree-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+    --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/$tree_case-tree.out" 2>&1; then
+    fail "$tree_case-source-tree"
+  fi
+  if ! grep -Fq 'delivery replay: materialization input tree identity is invalid' "$tmp/$tree_case-tree.out" ||
+     grep -Fq Traceback "$tmp/$tree_case-tree.out"; then
+    fail "$tree_case-source-tree-error"
+  fi
+done
+pass 'non-string source tree identities fail without a traceback'
+
+for identity_field in commit_id hash_algorithm; do
+  case "$identity_field" in
+    commit_id) identity_error='materialization input commit identity is invalid' ;;
+    hash_algorithm) identity_error='materialization input hash algorithm is invalid' ;;
+  esac
+  for identity_case in numeric list null; do
+    case "$identity_case" in
+      numeric) identity_value=1111111111111111111111111111111111111111 ;;
+      list) identity_value='[]' ;;
+      null) identity_value=null ;;
+    esac
+    "$jq_bin" -S -c ".stage_request.content.body.target_revision.value.$identity_field = $identity_value" \
+      "$base_input" >"$tmp/$identity_field-$identity_case.json"
+    mkdir -m 700 "$tmp/$identity_field-$identity_case-state" "$tmp/$identity_field-$identity_case-candidate" \
+      "$tmp/$identity_field-$identity_case-scratch"
+    if python3 "$replay" --input "$tmp/$identity_field-$identity_case.json" --source-repository-id fixture.target \
+      --source-git-dir "$tmp/source.git" --candidate-root "$tmp/$identity_field-$identity_case-candidate" \
+      --scratch-root "$tmp/$identity_field-$identity_case-scratch" --state-dir "$tmp/$identity_field-$identity_case-state" \
+      --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+      --expected-sha256 "$expected_changed" >"$tmp/$identity_field-$identity_case.out" 2>&1; then
+      fail "$identity_field-$identity_case"
+    fi
+    [ ! -e "$tmp/$identity_field-$identity_case-state/run.json" ] || fail "$identity_field-$identity_case-journal"
+    grep -Fq "delivery replay: $identity_error" \
+      "$tmp/$identity_field-$identity_case.out" || fail "$identity_field-$identity_case-error"
+  done
+done
+pass 'non-string commit and hash-algorithm identities are rejected before journaling'
+
+mkdir -m 700 "$tmp/caller-execution-root" "$tmp/caller-execution-state" \
+  "$tmp/caller-execution-candidate" "$tmp/caller-execution-scratch"
+printf '%s\n' keep >"$tmp/caller-execution-root/sentinel"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/caller-execution-candidate" --scratch-root "$tmp/caller-execution-scratch" \
+  --state-dir "$tmp/caller-execution-state" --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --execution-root "$tmp/caller-execution-root" >"$tmp/caller-execution.out" 2>&1; then
+  fail caller-execution-root
+fi
+[ "$(cat "$tmp/caller-execution-root/sentinel")" = keep ] || fail caller-execution-root-deleted
+grep -Fq 'unrecognized arguments: --execution-root' "$tmp/caller-execution.out" ||
+  fail caller-execution-root-error
+pass 'the replay CLI has no caller-selected execution-root path'
+
+huge_integer=$(printf '1%.0s' {1..5000})
+printf '{"huge":%s}\n' "$huge_integer" >"$tmp/huge-input.json"
+mkdir -m 700 "$tmp/huge-input-state" "$tmp/huge-input-candidate" "$tmp/huge-input-scratch"
+if PYTHONINTMAXSTRDIGITS=4300 "$python_with_int_limit" "$replay" --input "$tmp/huge-input.json" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/huge-input-candidate" --scratch-root "$tmp/huge-input-scratch" --state-dir "$tmp/huge-input-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/huge-input.out" 2>&1; then fail huge-integer-input; fi
+if ! grep -Fq 'delivery replay: input is not JSON' "$tmp/huge-input.out" ||
+   grep -Fq Traceback "$tmp/huge-input.out"; then
+  fail huge-integer-input-error
+fi
+printf '{"huge":%s}\n' "$huge_integer" >"$tmp/huge-observation.json"
+if PYTHONINTMAXSTRDIGITS=4300 "$python_with_int_limit" "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-candidate" --scratch-root "$tmp/reconcile-scratch" --state-dir "$tmp/reconcile-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/huge-observation.json" >"$tmp/huge-observation.out" 2>&1; then fail huge-integer-observation; fi
+if ! grep -Fq 'delivery replay: input is not JSON' "$tmp/huge-observation.out" ||
+   grep -Fq Traceback "$tmp/huge-observation.out"; then
+  fail huge-integer-observation-error
+fi
+mkdir -m 700 "$tmp/huge-journal-state" "$tmp/huge-journal-candidate" "$tmp/huge-journal-scratch"
+printf '{"huge":%s}\n' "$huge_integer" >"$tmp/huge-journal-state/run.json"
+if PYTHONINTMAXSTRDIGITS=4300 "$python_with_int_limit" "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/huge-journal-candidate" --scratch-root "$tmp/huge-journal-scratch" --state-dir "$tmp/huge-journal-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/huge-journal.out" 2>&1; then fail huge-integer-journal; fi
+if ! grep -Fq 'delivery replay: input is not JSON' "$tmp/huge-journal.out" ||
+   grep -Fq Traceback "$tmp/huge-journal.out"; then
+  fail huge-integer-journal-error
+fi
+pass 'huge JSON integers in input, observation, and journal fail without a traceback'
+
+long_repository_id=$(printf 'a%.0s' {1..200})
+mkdir -m 700 "$tmp/long-id-state" "$tmp/long-id-candidate" "$tmp/long-id-scratch"
+if python3 "$replay" --input "$base_input" --source-repository-id "$long_repository_id" --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/long-id-candidate" --scratch-root "$tmp/long-id-scratch" --state-dir "$tmp/long-id-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/long-id.out" 2>&1; then fail long-repository-id; fi
+if ! grep -Fq 'delivery replay: source repository id is invalid' "$tmp/long-id.out" ||
+   grep -Fq Traceback "$tmp/long-id.out"; then
+  fail long-repository-id-error
+fi
+[ ! -e "$tmp/long-id-state/run.json" ] || fail long-repository-id-journaled
+pass 'an out-of-contract source repository id is refused before anything is journaled'
+
+printf '\377' >"$tmp/invalid-input.json"
+mkdir -m 700 "$tmp/invalid-input-state" "$tmp/invalid-input-candidate" "$tmp/invalid-input-scratch"
+if python3 "$replay" --input "$tmp/invalid-input.json" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/invalid-input-candidate" --scratch-root "$tmp/invalid-input-scratch" --state-dir "$tmp/invalid-input-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/invalid-input.out" 2>&1; then fail invalid-utf8-input; fi
+if ! grep -Fq 'delivery replay: input is not JSON' "$tmp/invalid-input.out" ||
+   grep -Fq Traceback "$tmp/invalid-input.out"; then
+  fail invalid-utf8-input-error
+fi
+mkfifo "$tmp/fifo-input.json"
+mkdir -m 700 "$tmp/fifo-input-state" "$tmp/fifo-input-candidate" "$tmp/fifo-input-scratch"
+# A FIFO with no writer would block a plain open-then-read forever; the replay
+# must refuse it before reading. The background watchdog only fires on a hang.
+python3 "$replay" --input "$tmp/fifo-input.json" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/fifo-input-candidate" --scratch-root "$tmp/fifo-input-scratch" --state-dir "$tmp/fifo-input-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/fifo-input.out" 2>&1 &
+fifo_pid=$!
+( sleep 20; kill -9 "$fifo_pid" 2>/dev/null ) &
+fifo_watchdog=$!
+if wait "$fifo_pid"; then fail fifo-input; fi
+kill "$fifo_watchdog" 2>/dev/null || true
+if ! grep -Fq 'delivery replay: input is not a regular file' "$tmp/fifo-input.out" ||
+   grep -Fq Traceback "$tmp/fifo-input.out"; then
+  fail fifo-input-error
+fi
+printf '\377' >"$tmp/reconcile-state/invalid-review.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/reconcile-candidate" --scratch-root "$tmp/reconcile-scratch" --state-dir "$tmp/reconcile-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/reconcile-state/invalid-review.json" >"$tmp/invalid-review.out" 2>&1; then fail invalid-utf8-review; fi
+if ! grep -Fq 'delivery replay: input is not JSON' "$tmp/invalid-review.out" ||
+   grep -Fq Traceback "$tmp/invalid-review.out"; then
+  fail invalid-utf8-review-error
+fi
+mkdir -m 700 "$tmp/invalid-journal-state" "$tmp/invalid-journal-candidate" "$tmp/invalid-journal-scratch"
+printf '\377' >"$tmp/invalid-journal-state/run.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/invalid-journal-candidate" --scratch-root "$tmp/invalid-journal-scratch" --state-dir "$tmp/invalid-journal-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/invalid-journal.out" 2>&1; then fail invalid-utf8-journal; fi
+if ! grep -Fq 'delivery replay: input is not JSON' "$tmp/invalid-journal.out" ||
+   grep -Fq Traceback "$tmp/invalid-journal.out"; then
+  fail invalid-utf8-journal-error
+fi
+pass 'invalid UTF-8 input, review, and journal records fail without a traceback'
+
+printf '%s\n' '{"schema_version":1,"kind":"delivery_replay_review_observation","actor_id":"test.reviewer","request_sha256":"'"$request_sha"'","candidate_tree_id":"'"$candidate_tree"'","candidate_commit_id":"'"$candidate_commit"'","verdict":"clean"}' >"$tmp/review.json"
+printf '%s\n' '{"schema_version":1,"kind":"delivery_replay_publisher_observation","actor_id":"test.publisher","request_sha256":"'"$request_sha"'","candidate_tree_id":"'"$candidate_tree"'","candidate_commit_id":"'"$candidate_commit"'","disposition":"offline-simulated"}' >"$tmp/publisher.json"
+expect_candidate_move_rejected review-wait --review-observation "$tmp/review.json"
+lock_holder="$tmp/lock-holder.py"
+printf '%s\n' \
+  'import fcntl, pathlib, sys, time' \
+  'lock_path, ready, release = map(pathlib.Path, sys.argv[1:])' \
+  'with lock_path.open("a+b") as lock:' \
+  '    fcntl.flock(lock, fcntl.LOCK_EX)' \
+  '    ready.write_text("ready")' \
+  '    while not release.exists(): time.sleep(0.01)' >"$lock_holder"
+lock_wrapper="$tmp/lock-replay.py"
+printf '%s\n' \
+  'import importlib.util, pathlib, sys' \
+  'path, marker, arguments = sys.argv[1], pathlib.Path(sys.argv[2]), sys.argv[3:]' \
+  'spec = importlib.util.spec_from_file_location("replay", path)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(path).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, path, "exec"), module.__dict__)' \
+  'original = module.fcntl.flock' \
+  'def marked_flock(*args, **kwargs):' \
+  '    marker.write_text("waiting")' \
+  '    return original(*args, **kwargs)' \
+  'module.fcntl.flock = marked_flock' \
+  'sys.argv = [path] + arguments' \
+  'raise SystemExit(module.main())' >"$lock_wrapper"
+mkdir -m 700 "$tmp/lock-state"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/lock-state/"
+python3 "$lock_holder" "$tmp/lock-state/replay.lock" "$tmp/lock-ready" "$tmp/lock-release" &
+lock_holder_pid=$!
+while [ ! -f "$tmp/lock-ready" ]; do sleep 0.01; done
+python3 "$lock_wrapper" "$replay" "$tmp/lock-waiting" \
+  --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/lock-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --review-observation "$tmp/review.json" >"$tmp/lock-cancel.out" &
+lock_replay_pid=$!
+while [ ! -f "$tmp/lock-waiting" ]; do sleep 0.01; done
+kill -TERM "$lock_replay_pid"
+touch "$tmp/lock-release"
+wait "$lock_holder_pid"
+if wait "$lock_replay_pid"; then fail lock-cancel-status; else lock_status=$?; fi
+[ "$lock_status" -eq 75 ] || fail lock-cancel-code
+jq -e '(.phase=="review-wait") and (has("review")|not)' "$tmp/lock-state/run.json" >/dev/null ||
+  fail lock-cancel-state
+
+observation_wrapper="$tmp/observation-interrupt.py"
+printf '%s\n' \
+  'import importlib.util, os, pathlib, signal, sys' \
+  'path, signal_name, arguments = sys.argv[1], sys.argv[2], sys.argv[3:]' \
+  'spec = importlib.util.spec_from_file_location("replay", path)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(path).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, path, "exec"), module.__dict__)' \
+  'original = module.observation' \
+  'def interrupt_after_observation(*args, **kwargs):' \
+  '    result = original(*args, **kwargs)' \
+  '    os.kill(os.getpid(), getattr(signal, signal_name))' \
+  '    return result' \
+  'module.observation = interrupt_after_observation' \
+  'sys.argv = [path] + arguments' \
+  'raise SystemExit(module.main())' >"$observation_wrapper"
+mkdir -m 700 "$tmp/review-cancel-state"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/review-cancel-state/"
+if python3 "$observation_wrapper" "$replay" SIGINT \
+  --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/review-cancel-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --review-observation "$tmp/review.json" >"$tmp/review-cancel.out"; then
+  fail review-cancel-status
+else
+  review_cancel_status=$?
+fi
+[ "$review_cancel_status" -eq 75 ] || fail review-cancel-code
+jq -e '(.phase=="review-wait") and (has("review")|not)' "$tmp/review-cancel-state/run.json" >/dev/null ||
+  fail review-cancel-state
+printf '%s\n' '{"schema_version":1,"kind":"delivery_replay_review_observation","actor_id":123,"request_sha256":"'"$request_sha"'","candidate_tree_id":"'"$candidate_tree"'","candidate_commit_id":"'"$candidate_commit"'","verdict":"clean"}' >"$tmp/numeric-review.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/numeric-review.json" >"$tmp/numeric-review.out" 2>&1; then fail numeric-review-actor; fi
+grep -Fq 'offline observation actor is invalid' "$tmp/numeric-review.out" || fail numeric-review-actor-error
+pass 'offline observations require a string actor identity'
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/review.json" >"$tmp/publish-wait.out"
+jq -e '.state.phase=="publish-wait"' "$tmp/publish-wait.out" >/dev/null || fail missing-publisher-waits
+expect_candidate_move_rejected publish-wait --publisher-observation "$tmp/publisher.json"
+git_clean --git-dir="$tmp/changed-candidate/repository.git" update-ref refs/heads/alternate "$candidate_commit"
+git_clean --git-dir="$tmp/changed-candidate/repository.git" symbolic-ref refs/heads/candidate refs/heads/alternate
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/candidate-symref.out" 2>&1; then
+  fail candidate-symref
+fi
+git_clean --git-dir="$tmp/changed-candidate/repository.git" symbolic-ref --delete refs/heads/candidate
+git_clean --git-dir="$tmp/changed-candidate/repository.git" update-ref refs/heads/candidate "$candidate_commit"
+git_clean --git-dir="$tmp/changed-candidate/repository.git" update-ref -d refs/heads/alternate
+grep -Fq 'candidate repository identity guard failed' "$tmp/candidate-symref.out" || fail candidate-symref-error
+pass 'candidate completion guard rejects a symbolic candidate ref'
+mkdir -m 700 "$tmp/publish-cancel-state"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/publish-cancel-state/"
+if python3 "$observation_wrapper" "$replay" SIGTERM \
+  --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/publish-cancel-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/publish-cancel.out"; then
+  fail publish-cancel-status
+else
+  publish_cancel_status=$?
+fi
+[ "$publish_cancel_status" -eq 75 ] || fail publish-cancel-code
+jq -e '(.phase=="publish-wait") and (has("publisher")|not)' "$tmp/publish-cancel-state/run.json" >/dev/null ||
+  fail publish-cancel-state
+pass 'SIGTERM at the lock and SIGINT or SIGTERM after wait observations do not advance state'
+
+atomic_wrapper="$tmp/atomic-candidate-observation.py"
+printf '%s\n' \
+  'import importlib.util, os, pathlib, signal, subprocess, sys' \
+  'path, mode, marker, moved, arguments = sys.argv[1], sys.argv[2], pathlib.Path(sys.argv[3]), sys.argv[4], sys.argv[5:]' \
+  'spec = importlib.util.spec_from_file_location("replay", path)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(path).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, path, "exec"), module.__dict__)' \
+  'original = module.observation' \
+  'def act_during_publisher(*args, **kwargs):' \
+  '    result = original(*args, **kwargs)' \
+  '    if args[1] == "delivery_replay_publisher_observation":' \
+  '        if mode == "move":' \
+  '            candidate = pathlib.Path(arguments[arguments.index("--candidate-root") + 1]) / "repository.git"' \
+  '            attempt = subprocess.run(["/usr/bin/git", f"--git-dir={candidate}", "-c", "core.hooksPath=/dev/null", "update-ref", "refs/heads/candidate", moved], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)' \
+  '            marker.write_text(str(attempt.returncode))' \
+  '        else:' \
+  '            os.kill(os.getpid(), signal.SIGKILL)' \
+  '    return result' \
+  'module.observation = act_during_publisher' \
+  'sys.argv = [path] + arguments' \
+  'raise SystemExit(module.main())' >"$atomic_wrapper"
+mkdir -m 700 "$tmp/atomic-move-state"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/atomic-move-state/"
+python3 "$atomic_wrapper" "$replay" move "$tmp/atomic-move-status" "$moved_candidate" \
+  --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/atomic-move-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/atomic-move.out"
+jq -e '.state.phase=="completed-offline"' "$tmp/atomic-move.out" >/dev/null || fail atomic-move-completion
+[ "$(cat "$tmp/atomic-move-status")" != 0 ] || fail atomic-move-lock
+[ "$(git_clean --git-dir="$tmp/changed-candidate/repository.git" rev-parse refs/heads/candidate)" = "$candidate_commit" ] ||
+  fail atomic-move-ref
+
+mkdir -m 700 "$tmp/atomic-kill-state"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/atomic-kill-state/"
+if python3 "$atomic_wrapper" "$replay" kill "$tmp/unused-kill-status" "$moved_candidate" \
+  --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/atomic-kill-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/atomic-kill.out" 2>&1; then
+  fail atomic-kill-status
+fi
+atomic_lock="$tmp/changed-candidate/repository.git/refs/heads/candidate.lock"
+atomic_wait=0
+while [ -e "$atomic_lock" ]; do
+  atomic_wait=$((atomic_wait + 1))
+  [ "$atomic_wait" -le 100 ] || fail atomic-kill-lock-release
+  sleep 0.01
+done
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/atomic-kill-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/atomic-kill-resume.out"
+jq -e '.state.phase=="completed-offline"' "$tmp/atomic-kill-resume.out" >/dev/null || fail atomic-kill-resume
+pass 'candidate ref guard blocks publisher-time moves and releases after SIGKILL'
+
+mkdir -m 700 "$tmp/git-env-state" "$tmp/git-env-hooks" "$tmp/git-env-work-tree"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/git-env-state/"
+printf '%s\n' '#!/bin/sh' "printf hook >'$tmp/git-env-hook-ran'" >"$tmp/git-env-hooks/reference-transaction"
+/bin/chmod 0555 "$tmp/git-env-hooks/reference-transaction"
+printf '%s\n' '[core]' "hooksPath = $tmp/git-env-hooks" >"$tmp/git-env-global"
+git_clean init -q --bare "$tmp/git-env-other.git"
+/usr/bin/env GIT_NAMESPACE=poison GIT_COMMON_DIR="$tmp/git-env-other.git" GIT_DIR="$tmp/git-env-other.git" \
+  GIT_WORK_TREE="$tmp/git-env-work-tree" GIT_CONFIG_NOSYSTEM=0 GIT_CONFIG_GLOBAL="$tmp/git-env-global" \
+  GIT_CONFIG_COUNT=1 GIT_CONFIG_KEY_0=core.hooksPath GIT_CONFIG_VALUE_0="$tmp/git-env-hooks" \
+  python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/git-env-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/git-env.out"
+jq -e '.state.phase=="completed-offline"' "$tmp/git-env.out" >/dev/null || fail git-env-completion
+[ ! -e "$tmp/git-env-hook-ran" ] || fail git-env-hook
+[ "$(git_clean --git-dir="$tmp/changed-candidate/repository.git" rev-parse refs/heads/candidate)" = "$candidate_commit" ] ||
+  fail git-env-candidate
+pass 'all candidate Git operations ignore ambient repository, namespace, config, work-tree, and hook settings'
+printf '%s\n' '{"schema_version":1,"kind":"delivery_replay_publisher_observation","actor_id":123,"request_sha256":"'"$request_sha"'","candidate_tree_id":"'"$candidate_tree"'","candidate_commit_id":"'"$candidate_commit"'","disposition":"offline-simulated"}' >"$tmp/numeric-publisher.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --publisher-observation "$tmp/numeric-publisher.json" >"$tmp/numeric-publisher.out" 2>&1; then fail numeric-publisher-actor; fi
+grep -Fq 'offline observation actor is invalid' "$tmp/numeric-publisher.out" || fail numeric-publisher-actor-error
+expect_malformed_state() {
+  local name=$1 filter=$2
+  local state_root="$tmp/malformed-$name-state"
+  /bin/mkdir -m 700 "$state_root"
+  cp "$tmp/changed-state/materialization-input.json" "$state_root/materialization-input.json"
+  jq -S -c "$filter" "$tmp/changed-state/run.json" >"$state_root/run.json"
+  if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+    --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$state_root" \
+    --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+    >"$tmp/malformed-$name.out" 2>&1; then fail "malformed-$name"; fi
+  if ! grep -Fq 'delivery replay: state journal' "$tmp/malformed-$name.out" ||
+     grep -Fq Traceback "$tmp/malformed-$name.out"; then
+    fail "malformed-$name-error"
+  fi
+}
+expect_malformed_state identity-type '.identity=[]'
+expect_malformed_state missing-phase 'del(.phase)'
+expect_malformed_state invalid-phase '.phase="unknown"'
+expect_malformed_state missing-materialization '(.phase="verifying") | del(.materialization)'
+expect_malformed_state missing-verification '(.phase="review-wait") | del(.verification)'
+expect_malformed_state missing-review '(.phase="publish-wait") | del(.review)'
+expect_malformed_state missing-publisher '(.phase="completed-offline") | del(.publisher)'
+for candidate_phase in verifying review-wait publish-wait completed-offline; do
+  expect_malformed_state "missing-candidate-commit-$candidate_phase" \
+    "(.phase=\"$candidate_phase\") | del(.identity.candidate_commit_id)"
+  expect_malformed_state "missing-candidate-tree-$candidate_phase" \
+    "(.phase=\"$candidate_phase\") | del(.identity.candidate_tree_id)"
+done
+expect_malformed_state mismatched-candidate-commit \
+  '(.phase="verifying") | .identity.candidate_commit_id="0000000000000000000000000000000000000000"'
+expect_malformed_state mismatched-candidate-tree \
+  '(.phase="completed-offline") | .identity.candidate_tree_id="0000000000000000000000000000000000000000"'
+expect_malformed_state numeric-review-actor '.review.actor_id=123'
+expect_malformed_state numeric-publisher-actor \
+  '(.phase="completed-offline") | .publisher={"actor_id":123,"disposition":"offline-simulated","sha256":"0000000000000000000000000000000000000000000000000000000000000000"}'
+for verification_phase in review-wait publish-wait completed-offline; do
+  expect_malformed_state "verification-id-$verification_phase" \
+    "(.phase=\"$verification_phase\") | .verification.id=\"delivery.other.v1\""
+  expect_malformed_state "verification-path-$verification_phase" \
+    "(.phase=\"$verification_phase\") | .verification.path=\"other.txt\""
+  expect_malformed_state "verification-sha-$verification_phase" \
+    "(.phase=\"$verification_phase\") | .verification.sha256=\"0000000000000000000000000000000000000000000000000000000000000000\""
+done
+pass 'malformed state phases and nested records fail without a traceback'
+jq -S -c '.note="changed after review wait"' "$tmp/review.json" >"$tmp/changed-review.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/changed-review.json" --publisher-observation "$tmp/publisher.json" >"$tmp/changed-review.out" 2>&1; then fail changed-review-after-wait; fi
+grep -Fq 'review changed after review wait' "$tmp/changed-review.out" || fail changed-review-after-wait-error
+pass 'a changed supplied review cannot advance publish wait'
+ref_holder="$tmp/candidate-ref-holder.py"
+printf '%s\n' \
+  'import pathlib, subprocess, sys, time' \
+  'repository, commit, ready, release = sys.argv[1], sys.argv[2], pathlib.Path(sys.argv[3]), pathlib.Path(sys.argv[4])' \
+  'environment = {"PATH": "/usr/bin:/bin", "LC_ALL": "C", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null"}' \
+  'command = ["/usr/bin/git", "--git-dir=" + repository, "update-ref", "--stdin"]' \
+  'holder = subprocess.Popen(command, env=environment, stdin=subprocess.PIPE, stdout=subprocess.PIPE)' \
+  'holder.stdin.write(("option no-deref\nstart\nverify refs/heads/candidate " + commit + "\nprepare\n").encode())' \
+  'holder.stdin.flush()' \
+  'while True:' \
+  '    line = holder.stdout.readline()' \
+  '    if not line: raise SystemExit("holder could not lock the candidate ref")' \
+  '    if line.strip() == b"prepare: ok": break' \
+  'ready.write_text("held")' \
+  'while not release.exists(): time.sleep(0.01)' \
+  'holder.stdin.write(b"abort\n")' \
+  'holder.stdin.flush()' \
+  'holder.stdin.close()' \
+  'raise SystemExit(holder.wait())' >"$ref_holder"
+mkdir -m 700 "$tmp/held-lock-state"
+cp "$tmp/changed-state/materialization-input.json" "$tmp/changed-state/run.json" "$tmp/held-lock-state/"
+python3 "$ref_holder" "$tmp/changed-candidate/repository.git" "$candidate_commit" \
+  "$tmp/held-lock-ready" "$tmp/held-lock-release" &
+ref_holder_pid=$!
+held_lock_wait=0
+while [ ! -f "$tmp/held-lock-ready" ]; do
+  kill -0 "$ref_holder_pid" 2>/dev/null || fail held-lock-holder
+  held_lock_wait=$((held_lock_wait + 1))
+  [ "$held_lock_wait" -le 1000 ] || fail held-lock-holder-timeout
+  sleep 0.01
+done
+# Another process already holds the candidate ref lock, so this guard never owns
+# it. The background watchdog only fires if the guard hangs instead of failing.
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/held-lock-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/held-lock.out" 2>&1 &
+held_lock_pid=$!
+( sleep 60; kill -9 "$held_lock_pid" 2>/dev/null ) &
+held_lock_watchdog=$!
+if wait "$held_lock_pid"; then fail held-lock-status; fi
+kill "$held_lock_watchdog" 2>/dev/null || true
+if ! grep -Fq 'delivery replay: candidate repository identity guard failed' "$tmp/held-lock.out" ||
+   grep -Fq Traceback "$tmp/held-lock.out"; then
+  fail held-lock-error
+fi
+jq -e '.phase=="publish-wait" and (has("publisher")|not)' "$tmp/held-lock-state/run.json" >/dev/null ||
+  fail held-lock-state
+touch "$tmp/held-lock-release"
+wait "$ref_holder_pid" || fail held-lock-release
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/held-lock-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --publisher-observation "$tmp/publisher.json" >"$tmp/held-lock-resume.out"
+jq -e '.state.phase=="completed-offline"' "$tmp/held-lock-resume.out" >/dev/null || fail held-lock-resume
+pass 'a candidate ref lock held elsewhere blocks completion until it is released'
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/review.json" --publisher-observation "$tmp/publisher.json" >"$tmp/completed.out"
+jq -e '.state.phase=="completed-offline" and .state.publisher.disposition=="offline-simulated"' "$tmp/completed.out" >/dev/null ||
+  fail completed-offline
+expect_candidate_move_rejected completed-offline
+pass 'review, publish, and completed waits reject a moved same-tree candidate ref'
+cp "$tmp/completed.out" "$tmp/completed-first.out"
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/changed-candidate" --scratch-root "$tmp/changed-scratch" --state-dir "$tmp/changed-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/completed-repeat.out"
+cmp "$tmp/completed-first.out" "$tmp/completed-repeat.out" || fail duplicate-completion-output
+pass 'offline review and publisher observations complete once and replay deterministically'
+
+if run_replay verifier-failure "$base_input" "$(printf '0%.0s' {1..64})" >"$tmp/verifier-failure.out" 2>&1; then
+  fail fixed-verifier-failure
+fi
+jq -e '.state.phase=="failed" and (.state.reason|contains("digest mismatch"))' "$tmp/verifier-failure.out" >/dev/null ||
+  fail fixed-verifier-failure-state
+pass 'fixed verifier failure is terminal and explicit'
+
+mkdir -m 700 "$tmp/mismatch-state" "$tmp/mismatch-candidate" "$tmp/mismatch-scratch"
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/mismatch-candidate" --scratch-root "$tmp/mismatch-scratch" --state-dir "$tmp/mismatch-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" > /dev/null
+mismatch_commit=$(jq -r '.materialization.candidate_commit_id' "$tmp/mismatch-state/run.json")
+mismatch_tree=$(jq -r '.materialization.candidate_tree_id' "$tmp/mismatch-state/run.json")
+zero_oid=$(printf '0%.0s' {1..40})
+printf '%s\n' '{"schema_version":1,"kind":"delivery_replay_review_observation","actor_id":"test.reviewer","request_sha256":"'"$request_sha"'","candidate_tree_id":"'"$zero_oid"'","candidate_commit_id":"'"$mismatch_commit"'","verdict":"clean"}' >"$tmp/mismatch-review.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/mismatch-candidate" --scratch-root "$tmp/mismatch-scratch" --state-dir "$tmp/mismatch-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/mismatch-review.json" >"$tmp/mismatch.out" 2>&1; then fail mismatched-review; fi
+grep -Fq 'does not match this candidate' "$tmp/mismatch.out" || fail mismatched-review-error
+pass 'mismatched supplied review cannot complete the replay'
+
+# The same tree can belong to two candidate commits, so a review naming this
+# request and tree but another commit must not advance this candidate.
+printf '%s\n' '{"schema_version":1,"kind":"delivery_replay_review_observation","actor_id":"test.reviewer","request_sha256":"'"$request_sha"'","candidate_tree_id":"'"$mismatch_tree"'","candidate_commit_id":"'"$zero_oid"'","verdict":"clean"}' >"$tmp/other-commit-review.json"
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/mismatch-candidate" --scratch-root "$tmp/mismatch-scratch" --state-dir "$tmp/mismatch-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  --review-observation "$tmp/other-commit-review.json" >"$tmp/other-commit.out" 2>&1; then fail other-commit-review; fi
+if ! grep -Fq 'does not match this candidate' "$tmp/other-commit.out" ||
+   grep -Fq Traceback "$tmp/other-commit.out"; then
+  fail other-commit-review-error
+fi
+jq -e '.phase=="review-wait" and (has("review")|not)' "$tmp/mismatch-state/run.json" >/dev/null ||
+  fail other-commit-review-state
+pass 'a review bound to another candidate commit with this tree is refused'
+
+make_empty_input "$base_input" "$tmp/empty-final.json"
+source_digest=$(printf '%s\n' alpha beta | /usr/bin/shasum -a 256 | /usr/bin/awk '{print $1}')
+run_replay no-change "$tmp/empty-final.json" "$source_digest" >"$tmp/no-change.out"
+jq -e '.state.phase=="review-wait" and .state.materialization.candidate_tree_id==.state.identity.source_tree_id' "$tmp/no-change.out" >/dev/null ||
+  fail no-change
+pass 'empty producer patch records a no-change candidate before review'
+
+recover_no_change() {
+  local name=$1 input=$2 source=$3
+  local state="$tmp/$name-state" candidate="$tmp/$name-candidate" scratch="$tmp/$name-scratch"
+  /bin/mkdir -m 700 "$state" "$candidate" "$scratch"
+  if python3 "$kill_wrapper" "$replay" --input "$input" --source-repository-id fixture.target --source-git-dir "$source" \
+    --candidate-root "$candidate" --scratch-root "$scratch" --state-dir "$state" \
+    --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$source_digest" \
+    >"$tmp/$name-killed.out" 2>&1; then fail "$name-kill"; fi
+  [ "$(jq -r '.phase' "$state/run.json")" = materializing ] && [ -d "$candidate/repository.git" ] || fail "$name-window"
+  python3 "$replay" --input "$input" --source-repository-id fixture.target --source-git-dir "$source" \
+    --candidate-root "$candidate" --scratch-root "$scratch" --state-dir "$state" \
+    --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$source_digest" \
+    >"$tmp/$name-retry.out"
+  jq -e '.state.phase=="review-wait" and .state.materialization.candidate_commit_id==.state.identity.source_commit_id' \
+    "$tmp/$name-retry.out" >/dev/null || fail "$name-retry"
+}
+recover_no_change no-change-root "$tmp/empty-final.json" "$tmp/source.git"
+read -r ancestor_commit ancestor_tree < <(make_source_with_ancestor "$tmp/ancestor-source.git")
+"$fixture_builder" build "$tmp/ancestor-fixture" "$jq_bin" sha1 "$ancestor_commit" "$ancestor_tree"
+make_empty_input "$tmp/ancestor-fixture/input.json" "$tmp/ancestor-empty.json"
+recover_no_change no-change-ancestor "$tmp/ancestor-empty.json" "$tmp/ancestor-source.git"
+pass 'SIGKILL no-change recovery accepts both root and ancestor source commits'
+
+mkdir -m 700 "$tmp/interrupted-state" "$tmp/interrupted-candidate" "$tmp/interrupted-scratch"
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/interrupted-candidate" --scratch-root "$tmp/interrupted-scratch" --state-dir "$tmp/interrupted-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/interrupted.out" &
+interrupted_pid=$!
+interrupted_wait=0
+while [ ! -f "$tmp/interrupted-state/run.json" ]; do
+  if ! kill -0 "$interrupted_pid" 2>/dev/null; then
+    wait "$interrupted_pid" || :
+    sed -n '1,12p' "$tmp/interrupted.out" >&2
+    fail interrupted-start
+  fi
+  interrupted_wait=$((interrupted_wait + 1))
+  if [ "$interrupted_wait" -gt 100 ]; then
+    kill -TERM "$interrupted_pid" 2>/dev/null || :
+    wait "$interrupted_pid" || :
+    sed -n '1,12p' "$tmp/interrupted.out" >&2
+    fail interrupted-start-timeout
+  fi
+  sleep 0.1
+done
+kill -TERM "$interrupted_pid"
+if wait "$interrupted_pid"; then fail interrupted-run; fi
+[ "$(jq -r '.phase' "$tmp/interrupted-state/run.json")" = verifying ] || fail interrupted-state
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/interrupted-candidate" --scratch-root "$tmp/interrupted-scratch" --state-dir "$tmp/interrupted-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/interrupted-retry.out"
+jq -e '.state.phase=="review-wait"' "$tmp/interrupted-retry.out" >/dev/null || fail interrupted-retry
+pass 'interruption after materialization resumes without a duplicate candidate output'
+
+mkdir -m 700 "$tmp/stale-state" "$tmp/stale-candidate" "$tmp/stale-scratch"
+python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/stale-candidate" --scratch-root "$tmp/stale-scratch" --state-dir "$tmp/stale-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" > /dev/null
+if python3 "$replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/stale-candidate" --scratch-root "$tmp/stale-scratch" --state-dir "$tmp/stale-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$source_digest" >"$tmp/stale.out"; then fail changed-input-stale; fi
+jq -e '.state.phase=="stale"' "$tmp/stale.out" >/dev/null || fail changed-input-stale-state
+pass 'changed verifier input cannot reuse the prior run'
+changed_input="$tmp/changed-input.json"
+jq -S -c '.attempt.attempt_id="attempt.changed"' "$base_input" >"$changed_input"
+if python3 "$replay" --input "$changed_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/stale-candidate" --scratch-root "$tmp/stale-scratch" --state-dir "$tmp/stale-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/changed-input-stale.out"; then fail changed-materialization-input-stale; fi
+jq -e '.state.phase=="stale"' "$tmp/changed-input-stale.out" >/dev/null || fail changed-materialization-input-stale-state
+pass 'changed materialization input cannot reuse the prior run'
+
+package_root="$tmp/replay-package"
+generation=$(/usr/bin/sed -n \
+  "s/^PORTABLE_CORE_GENERATION='\(g-[0-9a-f]\{64\}\)'$/\1/p" "$root/scripts/core-contract.sh")
+/bin/mkdir -p "$package_root/delivery/v1" "$package_root/adapters/local-git-materializer/v1" \
+  "$package_root/scripts" "$package_root/core/v2/generations"
+/bin/cp "$replay" "$package_root/delivery/v1/replay.py"
+/bin/cp "$root/adapters/local-git-materializer/v1/materialize.sh" \
+  "$root/adapters/local-git-materializer/v1/protocol.jq" \
+  "$package_root/adapters/local-git-materializer/v1/"
+/bin/cp "$root/scripts/core-contract.sh" "$package_root/scripts/core-contract.sh"
+/bin/cp "$root/core/v2/generation-registry.json" "$package_root/core/v2/generation-registry.json"
+/bin/cp -R "$root/core/v2/generations/$generation" "$package_root/core/v2/generations/"
+package_replay="$package_root/delivery/v1/replay.py"
+/bin/mkdir -m 700 "$tmp/package-state" "$tmp/package-candidate" "$tmp/package-scratch"
+python3 "$package_replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/package-candidate" --scratch-root "$tmp/package-scratch" --state-dir "$tmp/package-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/package-first.out"
+printf '\n' >>"$package_root/adapters/local-git-materializer/v1/protocol.jq"
+if python3 "$package_replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/package-candidate" --scratch-root "$tmp/package-scratch" --state-dir "$tmp/package-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/package-dependency-stale.out"; then fail changed-package-dependency; fi
+jq -e '.state.phase=="stale"' "$tmp/package-dependency-stale.out" >/dev/null || fail changed-package-dependency-state
+/bin/cp "$root/adapters/local-git-materializer/v1/protocol.jq" \
+  "$package_root/adapters/local-git-materializer/v1/protocol.jq"
+printf '\n' >>"$package_replay"
+if python3 "$package_replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/package-candidate" --scratch-root "$tmp/package-scratch" --state-dir "$tmp/package-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt --expected-sha256 "$expected_changed" \
+  >"$tmp/package-driver-stale.out"; then fail changed-replay-driver; fi
+jq -e '.state.phase=="stale"' "$tmp/package-driver-stale.out" >/dev/null || fail changed-replay-driver-state
+pass 'changed executable package or replay driver cannot reuse a prior run'
+
+/bin/cp "$replay" "$package_replay"
+/bin/cp "$root/adapters/local-git-materializer/v1/protocol.jq" \
+  "$package_root/adapters/local-git-materializer/v1/protocol.jq"
+printf '%s\n' keep >"$package_root/sentinel"
+if python3 "$package_replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/package-candidate" --scratch-root "$tmp/package-scratch" --state-dir "$tmp/package-state" \
+  --closure-helper "$runtime/object-closure" --jq-bin "$jq_bin" --verify-path source.txt \
+  --expected-sha256 "$expected_changed" --execution-root "$package_root" >"$tmp/copied-root-bypass.out" 2>&1; then
+  fail copied-root-bypass
+fi
+grep -Fq 'unrecognized arguments: --execution-root' "$tmp/copied-root-bypass.out" || fail copied-root-bypass-error
+[ "$(cat "$package_root/sentinel")" = keep ] || fail copied-root-bypass-deleted
+pass 'a copied expected layout cannot select or delete an execution root'
+
+driver_wrapper="$tmp/driver-identity.py"
+printf '%s\n' \
+  'import hashlib, importlib.util, pathlib, sys' \
+  'driver = pathlib.Path(sys.argv[1])' \
+  'saved = driver.read_bytes()' \
+  'spec = importlib.util.spec_from_file_location("replay", driver)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = saved' \
+  'exec(compile(saved, str(driver), "exec"), module.__dict__)' \
+  'assert module.driver_identity() == hashlib.sha256(saved).hexdigest()' \
+  'for changed in (saved + b"\n# comment after load\n", saved + b"\n\n", saved + b"\nCHANGED_EXECUTABLE_STATEMENT = True\n", saved + b"\n\\xff\n"):' \
+  '    driver.write_bytes(changed)' \
+  '    assert module.driver_identity() == hashlib.sha256(saved).hexdigest()' \
+  'driver.write_bytes(saved)' >"$driver_wrapper"
+python3 "$driver_wrapper" "$package_replay" || fail driver-loaded-identity
+pass 'driver identity remains bound to the exact source buffer loaded once'
+
+/bin/cp "$runtime/object-closure" "$tmp/race-object-closure"
+/bin/cp "$jq_bin" "$tmp/race-jq"
+/bin/chmod 0555 "$tmp/race-object-closure" "$tmp/race-jq"
+race_wrapper="$tmp/snapshot-race.py"
+printf '%s\n' \
+  'import argparse, importlib.util, pathlib, stat, sys' \
+  'driver, repository, state_dir, helper, jq_bin, *arguments = sys.argv[1:]' \
+  'spec = importlib.util.spec_from_file_location("replay", driver)' \
+  'module = importlib.util.module_from_spec(spec)' \
+  'module._REPLAY_DRIVER_BYTES = pathlib.Path(driver).read_bytes()' \
+  'exec(compile(module._REPLAY_DRIVER_BYTES, driver, "exec"), module.__dict__)' \
+  'value = lambda name: arguments[arguments.index(name) + 1]' \
+  'values = argparse.Namespace(input=value("--input"), source_repository_id=value("--source-repository-id"), source_git_dir=value("--source-git-dir"), candidate_root=value("--candidate-root"), scratch_root=value("--scratch-root"), state_dir=value("--state-dir"), closure_helper=helper, jq_bin=jq_bin, verify_path=value("--verify-path"), expected_sha256=value("--expected-sha256"), review_observation=None, publisher_observation=None)' \
+  'module.create_execution_snapshot(pathlib.Path(repository), values, pathlib.Path(state_dir))' \
+  'targets = [pathlib.Path(repository) / "adapters/local-git-materializer/v1/protocol.jq", pathlib.Path(helper), pathlib.Path(jq_bin)]' \
+  'saved = [target.read_bytes() for target in targets]' \
+  'modes = [stat.S_IMODE(target.stat().st_mode) for target in targets]' \
+  'try:' \
+  '    for target in targets:' \
+  '        target.chmod(0o700)' \
+  '        target.write_bytes(b"replaced after execution snapshot\n")' \
+  '    try:' \
+  '        module.replay_locked(values, pathlib.Path(state_dir))' \
+  '    except module.ReplayError as error:' \
+  '        assert "execution bundle does not match current dependencies" in str(error)' \
+  '    else:' \
+  '        raise AssertionError("changed execution sources were accepted")' \
+  'finally:' \
+  '    for target, data, mode in zip(targets, saved, modes):' \
+  '        target.write_bytes(data)' \
+  '        target.chmod(mode)' >"$race_wrapper"
+/bin/mkdir -m 700 "$tmp/race-state" "$tmp/race-candidate" "$tmp/race-scratch"
+driver_sha=$(sha_file "$package_replay")
+package_sha=$(sha_file "$package_root/adapters/local-git-materializer/v1/protocol.jq")
+helper_sha=$(sha_file "$tmp/race-object-closure")
+jq_sha=$(sha_file "$tmp/race-jq")
+python3 "$race_wrapper" "$package_replay" "$package_root" "$tmp/race-state" \
+  "$tmp/race-object-closure" "$tmp/race-jq" \
+  --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/race-candidate" --scratch-root "$tmp/race-scratch" --state-dir "$tmp/race-state" \
+  --closure-helper "$tmp/race-object-closure" --jq-bin "$tmp/race-jq" \
+  --verify-path source.txt --expected-sha256 "$expected_changed"
+python3 "$package_replay" --input "$base_input" --source-repository-id fixture.target --source-git-dir "$tmp/source.git" \
+  --candidate-root "$tmp/race-candidate" --scratch-root "$tmp/race-scratch" --state-dir "$tmp/race-state" \
+  --closure-helper "$tmp/race-object-closure" --jq-bin "$tmp/race-jq" \
+  --verify-path source.txt --expected-sha256 "$expected_changed" >"$tmp/race.out"
+jq -e '.state.phase=="review-wait" and
+  .state.identity.driver_sha256==$driver and
+  .state.identity.materializer_package.files["adapters/local-git-materializer/v1/protocol.jq"]==$package and
+  .state.identity.closure_helper_sha256==$helper and .state.identity.jq_sha256==$jq' \
+  --arg driver "$driver_sha" --arg package "$package_sha" --arg helper "$helper_sha" --arg jq "$jq_sha" \
+  "$tmp/race.out" >/dev/null || fail immutable-execution-snapshot
+pass 'the one state-owned execution bundle rejects drift and records its exact bytes'
+
+printf 'delivery replay: %s focused checks passed\n' "$passed"
diff --git a/work/delivery-loop-first/plan.md b/work/delivery-loop-first/plan.md
new file mode 100644
index 0000000..877e5a3
--- /dev/null
+++ b/work/delivery-loop-first/plan.md
@@ -0,0 +1,25 @@
+# Delivery loop first plan
+
+This PR adds one inactive offline replay slice. It owns only these paths:
+
+- `delivery/v1/replay.py`
+- `scripts/test/delivery-replay.test.sh` and its test-owned fixtures
+- `work/delivery-loop-first/plan.md`
+- `README.md`, `RESTORE.md`, and `ci/required-files.txt`
+
+The replay accepts an existing canonical local-materialization input and
+caller-owned source, candidate, scratch, and private state directories. It calls
+the existing local materializer, verifies one repo-relative candidate blob against
+one supplied SHA-256, and journals identities and phase atomically. It never runs
+candidate code or a user command string.
+
+States are `materializing`, `verifying`, `review-wait`, `publish-wait`, `failed`,
+and `completed-offline`. Review and publisher records are supplied offline test
+observations. They name the exact request digest, candidate tree, and candidate
+commit plus an actor, but do not authenticate anyone or authorize publication. The
+commit is required because one tree can belong to two candidate commits. Missing
+review remains waiting.
+
+The slice is not profile selection, qualification, model execution, target access,
+deployment, release, install, merge, or production publication. It is not the
+whole delivery loop.