From 04cd7f053797816277ff6b7451140ff6ef40d378 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 16:26:44 +0000
Subject: [PATCH 1/6] Prepare registry SDK readbacks for the producer physical
 seed

---
 scripts/conformance/command_keys/README.md    |  41 ++++
 .../command_keys/physical_readback.py         | 186 ++++++++++++++++++
 tests/test_physical_command_key_readback.py   |  85 ++++++++
 3 files changed, 312 insertions(+)
 create mode 100644 scripts/conformance/command_keys/physical_readback.py
 create mode 100644 tests/test_physical_command_key_readback.py

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 803ff64..e33e128 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -69,3 +69,44 @@ accepted common stopped-copy helper is separately maintained by
 owns the real published historical seed/upgrade. Namespace/run fences also remain
 distinct installed cells. No full shared216
 PASS follows from this adapter alone.
+
+## Physical upgrade consumer
+
+`physical_readback.py` consumes the producer-owned JSON receipt contract from
+[Server1210](https://github.com/durable-workflow/server/pull/1210). Source/execution
+reuse is held until that PR's final source review and checks qualify. This is
+separate from the accepted synthetic-copy helper and its older manifest hash.
+
+The consumer invokes the registry-only Python Client against an already upgraded
+disposable runtime. The producer receipt must include all original signal/update
+open/completed/continued cases, complete captured histories and original update
+identities. The manifest includes the historical unkeyed-signal clarification.
+It refuses a changed receipt hash, incomplete seed, missing identities or package
+overlay before issuing a command.
+
+An outer runner must bind the producer source, stopped backup and receipt hashes,
+upgrade a private copy with the qualified image, preserve the original, disable
+unrelated maintenance writes and compare complete logical snapshots around this
+consumer. It owns the image/source check, exact runtime/database/credential
+cleanup and the final qualification receipt. Never run against the retained
+original database or reuse a production runtime credential.
+
+```sh
+/path/to/registry-venv/bin/python -I \
+  scripts/conformance/command_keys/physical_readback.py \
+  --upstream http://127.0.0.1:12345 --server-version 3.9.2 \
+  --credential-file /path/to/private/fixture.env \
+  --seed /path/to/producer-receipt.json --seed-sha256 FULL_PRODUCER_RECEIPT_SHA256 \
+  --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \
+  --output /path/to/private/new-consumer-receipt.json
+```
+
+For each original keyed update it issues four concurrent retries with changed
+input, requires original command/run/update/acceptance identities and explicit
+`intent_verified: false`, checks a changed-name conflict, then uses GET-only
+outcome reads to distinguish pending from completed-null. Python's update helper
+uses its unscoped route; this does not claim a new explicit-run update API.
+Historical NULL-key signals receive no fabricated retry key. Their compatibility
+control reads the original history, including completed/continued boundaries.
+Neither this consumer's completion flag nor a producer-only upgrade receipt alone
+qualifies the installed SDK physical cell without the outer snapshot/cleanup proof.
diff --git a/scripts/conformance/command_keys/physical_readback.py b/scripts/conformance/command_keys/physical_readback.py
new file mode 100644
index 0000000..adac25b
--- /dev/null
+++ b/scripts/conformance/command_keys/physical_readback.py
@@ -0,0 +1,186 @@
+"""Installed SDK readbacks for the producer-owned physical seed, after upgrade.
+
+The outer runner owns the verified seed/database, exact image, upgrade, complete
+zero-write snapshots and cleanup. This consumer neither seeds nor upgrades data.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import hashlib
+import importlib.util
+import json
+import stat
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import urlsplit
+
+from durable_workflow import Client
+from durable_workflow.errors import UpdateFailed
+
+# Physical seed authority includes the producer's unkeyed-signal clarification.
+# It is distinct from the older manifest bound to the synthetic-copy helper.
+MANIFEST_SHA256 = "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f"
+
+
+def bound_source():
+    path = Path(__file__).with_name("response_loss.py")
+    spec = importlib.util.spec_from_file_location("installed_keys_provenance", path)
+    if spec is None or spec.loader is None:
+        raise ValueError("installed package verifier is unavailable")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module, hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def seed_cases(document):
+    if (
+        document.get("schema") != "durable-workflow.command-key-physical-seed.v1"
+        or document.get("manifest_sha256") != MANIFEST_SHA256
+    ):
+        raise ValueError("require the producer-owned physical seed contract")
+    cases = document["cases"]
+    expected = {(kind, state) for kind in ("signal", "update") for state in ("open", "completed", "continued")}
+    if len(cases) != len(expected) or {(case["kind"], case["state"]) for case in cases} != expected:
+        raise ValueError("physical seed must contain each original state/type exactly once")
+    for case in cases:
+        receipt = case["receipt"]
+        fields = ("command_id", "run_id") + (("update_id", "accepted_at") if case["kind"] == "update" else ())
+        if any(not isinstance(receipt.get(field), str) or not receipt[field] for field in fields):
+            raise ValueError("original receipt has incomplete durable identities")
+        if receipt["workflow_id"] != case["workflow_id"]:
+            raise ValueError("original receipt names a different workflow")
+        if case["kind"] == "update" and not case["original_body"].get("request_id"):
+            raise ValueError("original keyed update identity is missing")
+        if case["history"]["next_page_token"] is not None:
+            raise ValueError("physical seed history was not fully captured")
+    return cases
+
+
+def legacy_receipt(original, received):
+    if received.get("retained") is not True or received.get("intent_verified") is not False:
+        raise ValueError("legacy readback must be retained and explicitly unverified")
+    for field in ("command_id", "run_id", "update_id", "accepted_at"):
+        if received.get(field) != original[field]:
+            raise ValueError("legacy readback changed original " + field)
+
+
+async def run(args):
+    if not __debug__:
+        raise ValueError("acceptance assertions require Python without optimization")
+    if args.output.exists():
+        raise ValueError("refusing to overwrite existing evidence")
+    upstream = urlsplit(args.upstream)
+    if (
+        upstream.scheme != "http"
+        or upstream.hostname != "127.0.0.1"
+        or not upstream.port
+        or args.upstream != f"http://127.0.0.1:{upstream.port}"
+    ):
+        raise ValueError("synthetic runtime must be canonical numeric loopback")
+    raw_seed = args.seed.read_bytes()
+    if hashlib.sha256(raw_seed).hexdigest() != args.seed_sha256:
+        raise ValueError("producer seed receipt hash mismatch")
+    cases = seed_cases(json.loads(raw_seed))
+    verifier, verifier_hash = bound_source()
+    package = verifier.installed_identity(args.wheel)
+    mode = args.credential_file.lstat().st_mode
+    if not stat.S_ISREG(mode) or mode & 0o077:
+        raise ValueError("fixture credential file must be owner-only")
+    token = dict(line.split("=", 1) for line in args.credential_file.read_text().splitlines())["DW_AUTH_TOKEN"]
+    report = {
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "sdk": package,
+        "seed_sha256": args.seed_sha256,
+        "verifier_sha256": verifier_hash,
+        "consumer_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "scope": "installed SDK physical-upgrade readbacks only",
+        "cells": [],
+        "complete": False,
+    }
+    try:
+        async with Client(args.upstream, token=token) as client:
+            info = await client.get_cluster_info()
+            if info["version"] != args.server_version:
+                raise ValueError("runtime differs from the outer verified upgrade tuple")
+            report["server_version"] = info["version"]
+            for case in cases:
+                original = case["receipt"]
+                workflow_id = case["workflow_id"]
+                observed = [event async for event in client.iter_history(workflow_id, original["run_id"])]
+                if observed != case["history"]["events"]:
+                    raise ValueError("upgrade history differs from the original producer capture")
+                cell = {
+                    "kind": case["kind"],
+                    "state": case["state"],
+                    "command_id": original["command_id"],
+                    "run_id": original["run_id"],
+                    "original_history_preserved": True,
+                }
+                if case["kind"] == "signal":
+                    # Published historical signals have NULL stored keys. Never
+                    # manufacture keyed recovery from an unrelated current key.
+                    cell["scope"] = "unkeyed historical signal compatibility only"
+                else:
+                    key = case["original_body"]["request_id"]
+                    name = original["update_name"]
+                    receipts = await asyncio.gather(
+                        *(
+                            client.update_workflow(workflow_id, name, args=[999], request_id=key, wait_for="accepted")
+                            for _ in range(4)
+                        )
+                    )
+                    for receipt in receipts:
+                        legacy_receipt(original, receipt)
+                        assert receipt["request_id"] == key
+                        if case["state"] != "open":
+                            assert receipt["update_status"] == "completed" and receipt["result"] is None
+                    try:
+                        await client.update_workflow(workflow_id, "other", args=[999], request_id=key)
+                    except UpdateFailed as error:
+                        assert error.reason == "request_id_conflict"
+                        assert error.details["retained_command_id"] == original["command_id"]
+                    else:
+                        raise AssertionError("changed legacy command name was admitted")
+                    outcome = await client.update_outcome_receipt(workflow_id, original["update_id"])
+                    assert outcome["update_id"] == original["update_id"]
+                    if case["state"] == "open":
+                        assert outcome["update_status"] == "accepted"
+                    else:
+                        assert outcome["update_status"] == "completed" and outcome["result"] is None
+                        assert await client.update_outcome(workflow_id, original["update_id"]) is None
+                    cell.update(
+                        {
+                            "scope": "original keyed update; unscoped SDK route returns original run",
+                            "update_id": original["update_id"],
+                            "concurrent_retries": len(receipts),
+                            "retained": True,
+                            "intent_verified": False,
+                            "changed_name": "request_id_conflict",
+                        }
+                    )
+                report["cells"].append(cell)
+        report["complete"] = True
+        report["finished_at"] = datetime.now(timezone.utc).isoformat()
+    except Exception as error:
+        report["failure_type"] = type(error).__name__
+        raise
+    finally:
+        args.output.write_text(json.dumps(report, indent=2) + "\n")
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--upstream", required=True)
+    parser.add_argument("--server-version", required=True)
+    parser.add_argument("--credential-file", type=Path, required=True)
+    parser.add_argument("--seed", type=Path, required=True)
+    parser.add_argument("--seed-sha256", required=True)
+    parser.add_argument("--wheel", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    asyncio.run(run(parse_args()))
diff --git a/tests/test_physical_command_key_readback.py b/tests/test_physical_command_key_readback.py
new file mode 100644
index 0000000..c392b1b
--- /dev/null
+++ b/tests/test_physical_command_key_readback.py
@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import copy
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+PATH = Path(__file__).parents[1] / "scripts/conformance/command_keys/physical_readback.py"
+SPEC = importlib.util.spec_from_file_location("physical_readback", PATH)
+assert SPEC is not None and SPEC.loader is not None
+consumer = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(consumer)
+
+
+def seed():
+    cases = []
+    for kind in ("signal", "update"):
+        for state in ("open", "completed", "continued"):
+            identity = kind + "-" + state
+            receipt = {"workflow_id": identity, "command_id": "command-" + identity, "run_id": "run-" + identity}
+            body = {}
+            if kind == "update":
+                receipt.update({"update_id": "update-" + identity, "accepted_at": "2026-01-01T00:00:00Z"})
+                body["request_id"] = "original-update-key"
+            cases.append(
+                {
+                    "kind": kind,
+                    "state": state,
+                    "workflow_id": identity,
+                    "receipt": receipt,
+                    "original_body": body,
+                    "history": {"events": [], "next_page_token": None},
+                }
+            )
+    return {
+        "schema": "durable-workflow.command-key-physical-seed.v1",
+        "manifest_sha256": consumer.MANIFEST_SHA256,
+        "cases": cases,
+    }
+
+
+def test_accepts_real_unkeyed_signal_compatibility_without_inventing_keys():
+    cases = consumer.seed_cases(seed())
+    assert all("request_id" not in case["original_body"] for case in cases if case["kind"] == "signal")
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda value: value.update(manifest_sha256="unreviewed"),
+        lambda value: value["cases"].pop(),
+        lambda value: value["cases"].append(copy.deepcopy(value["cases"][0])),
+        lambda value: value["cases"][0].update(state="unsupported"),
+        lambda value: value["cases"][0]["receipt"].pop("command_id"),
+        lambda value: value["cases"][0]["receipt"].update(workflow_id="other"),
+        lambda value: value["cases"][3]["original_body"].pop("request_id"),
+        lambda value: value["cases"][0]["history"].update(next_page_token="unfinished"),
+    ],
+)
+def test_incomplete_or_wrong_seed_refuses_before_client_creation(mutation):
+    document = seed()
+    mutation(document)
+    with pytest.raises(ValueError):
+        consumer.seed_cases(document)
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        {"retained": False},
+        {"intent_verified": True},
+        {"intent_verified": None},
+        {"command_id": "other"},
+        {"run_id": "successor"},
+        {"update_id": "other"},
+        {"accepted_at": "changed"},
+    ],
+)
+def test_legacy_receipt_cannot_claim_verification_or_change_original_identities(mutation):
+    original = seed()["cases"][3]["receipt"]
+    received = {**original, "retained": True, "intent_verified": False}
+    consumer.legacy_receipt(original, received)
+    with pytest.raises(ValueError):
+        consumer.legacy_receipt(original, {**received, **mutation})

From 632ec55b221dc5a598833eceb36af0c15d08ebb7 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 16:33:57 +0000
Subject: [PATCH 2/6] Bind physical readbacks to the frozen backup and full
 request-window snapshots

---
 scripts/conformance/command_keys/README.md    |  17 ++
 .../command_keys/run_physical_readback.py     | 242 ++++++++++++++++++
 tests/test_physical_command_key_readback.py   |  35 +++
 3 files changed, 294 insertions(+)
 create mode 100644 scripts/conformance/command_keys/run_physical_readback.py

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index e33e128..50880dc 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -91,6 +91,23 @@ consumer. It owns the image/source check, exact runtime/database/credential
 cleanup and the final qualification receipt. Never run against the retained
 original database or reuse a production runtime credential.
 
+`run_physical_readback.py` supplies that outer integration for the frozen producer
+handoff. It binds the exact stopped-backup/JSON/helper hashes, copies into a new
+private directory, checks native schema markers, runs the qualified image's
+`schema-upgrade`, and verifies the empty intent table. It then brackets this
+consumer with complete typed logical snapshots from the accepted helper.
+The request window ends before shutdown; later repair work cannot be relabeled
+as part of that zero-write window. The retained original backup is hash-checked
+again after cleanup. Two sequential containers have the same 1 CPU/1 GiB/256-PID
+limits, read-only roots and bounded tmpfs paths; no compile or historical reseed
+occurs. Original seed/upgrade authority remains Server-owned.
+
+Once the producer's final source gate permits execution reuse, invoke the outer
+controller with `--python`, `--wheel`, `--baseline`, `--seed`, `--snapshot-helper`
+and a new private `--output` directory. The original file inputs come from the
+producer handoff and accepted common snapshot helper, not an invented fixture.
+This draft has no installed physical result yet.
+
 ```sh
 /path/to/registry-venv/bin/python -I \
   scripts/conformance/command_keys/physical_readback.py \
diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py
new file mode 100644
index 0000000..96a52c8
--- /dev/null
+++ b/scripts/conformance/command_keys/run_physical_readback.py
@@ -0,0 +1,242 @@
+"""Qualify one installed SDK consumer on a private copy of the producer seed."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+import secrets
+import shutil
+import sqlite3
+import stat
+import subprocess
+import time
+import urllib.request
+from contextlib import closing
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+HELPER_SHA256 = "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad"
+BASELINE_SHA256 = "7d1826f7681342bb3a5974509648e2e8d75e7a21d262f65f3ec1f58727ccbc0c"
+SEED_SHA256 = "8c728ce827df3a11621dfb0e82cdd404bbab48784182a5e92c8d632d49047fc5"
+
+
+def load(path, name):
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise ValueError("required source module is unavailable")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def verify_file(path, expected):
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or hashlib.sha256(path.read_bytes()).hexdigest() != expected:
+        raise ValueError("required producer/helper input identity differs")
+
+
+def native_marker(connection, expected):
+    if connection.execute("SELECT engine,version FROM dw_server_schema").fetchall() != [("rust-development", expected)]:
+        raise ValueError("native schema ownership/version marker differs")
+
+
+def run(args):
+    # No Docker action or backup copy precedes all input identity checks.
+    verify_file(args.baseline, BASELINE_SHA256)
+    verify_file(args.seed, SEED_SHA256)
+    verify_file(args.snapshot_helper, HELPER_SHA256)
+    legacy = load(args.snapshot_helper, "accepted_sqlite_snapshot")
+    controls = load(Path(__file__).with_name("run_response_loss.py"), "native_runtime_identity")
+    root = args.output.absolute()
+    root.mkdir(mode=0o700, parents=False, exist_ok=False)
+    database = root / "database"
+    database.mkdir(mode=0o700)
+    copied = database / "upgraded.sqlite"
+    config = root / "docker-config"
+    config.mkdir(mode=0o700)
+    (config / "config.json").write_text('{"auths":{}}\n')
+    environment = {"PATH": os.defpath, "TMPDIR": str(root), "PYTHONUNBUFFERED": "1"}
+    docker = ["docker", "--config", str(config)]
+
+    def call(*arguments, check=True, timeout=45):
+        return subprocess.run(
+            docker + list(arguments), env=environment, check=check, capture_output=True, text=True, timeout=timeout
+        )
+
+    def snapshot():
+        with closing(sqlite3.connect(copied.as_uri() + "?mode=ro", uri=True, timeout=2)) as connection:
+            connection.execute("BEGIN")
+            return legacy.snapshot(connection)
+
+    image = "ghcr.io/durable-workflow/server@" + controls.EXPECTED_SERVER["index"]
+    observed = controls.image_identity(json.loads(call("image", "inspect", image).stdout))
+    suffix = uuid4().hex[:12]
+    names = ["sdk-python-216-physical-upgrade-" + suffix, "sdk-python-216-physical-runtime-" + suffix]
+    credential = root / "fixture.env"
+    descriptor = os.open(credential, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+    with os.fdopen(descriptor, "w") as stream:
+        stream.write(
+            "DW_AUTH_TOKEN="
+            + secrets.token_hex(24)
+            + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDW_RUST_MAINTENANCE_LOOP=0"
+            + "\nDB_CONNECTION=sqlite\nDB_DATABASE=/cohort/upgraded.sqlite\n"
+        )
+    consumer = Path(__file__).with_name("physical_readback.py")
+    report = {
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "baseline_sha256": BASELINE_SHA256,
+        "seed_sha256": SEED_SHA256,
+        "snapshot_helper_sha256": HELPER_SHA256,
+        "expected_server": controls.EXPECTED_SERVER,
+        "observed_image": observed,
+        "controller_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "consumer_sha256": hashlib.sha256(consumer.read_bytes()).hexdigest(),
+        "complete": False,
+    }
+    common = [
+        "--init",
+        "--read-only",
+        "--cpus",
+        "1",
+        "--memory",
+        "1g",
+        "--pids-limit",
+        "256",
+        "--tmpfs",
+        "/tmp:mode=1777,size=128m",
+        "--tmpfs",
+        "/app/storage:uid=1000,gid=1000,mode=0700,size=64m",
+        "--tmpfs",
+        "/app/database:uid=1000,gid=1000,mode=0700,size=16m",
+        "-v",
+        str(database) + ":/cohort:rw",
+        "--env-file",
+        str(credential),
+    ]
+    clean = []
+    try:
+        shutil.copyfile(args.baseline, copied)
+        copied.chmod(0o600)
+        verify_file(copied, BASELINE_SHA256)
+        with closing(sqlite3.connect(copied)) as connection:
+            native_marker(connection, 12)
+        call("create", "--name", names[0], "--network", "none", *common, image, "schema-upgrade")
+        call("start", names[0])
+        assert call("wait", names[0]).stdout.strip() == "0"
+        upgrade = json.loads(call("logs", names[0]).stdout)
+        assert upgrade == {"backend": "sqlite", "from_schema_version": 12, "schema_version": 13, "status": "upgraded"}
+        report["upgrade"] = upgrade
+        upgrade_state = json.loads(call("inspect", names[0]).stdout)[0]
+        assert upgrade_state["State"]["OOMKilled"] is False
+        call("rm", "-v", names[0])
+        with closing(sqlite3.connect(copied)) as connection:
+            native_marker(connection, 13)
+            assert connection.execute("SELECT count(*) FROM dw_command_intents").fetchone()[0] == 0
+        call("create", "--name", names[1], "-p", "127.0.0.1::8080", *common, image)
+        call("start", names[1])
+        address = call("port", names[1], "8080/tcp").stdout.strip()
+        host, port = address.split(":")
+        assert host == "127.0.0.1" and port.isdigit() and 1 <= int(port) <= 65535
+        upstream = "http://" + address
+        deadline = time.monotonic() + 45
+        while time.monotonic() < deadline:
+            try:
+                with urllib.request.urlopen(upstream + "/api/health", timeout=2) as response:
+                    if response.status == 200:
+                        break
+            except OSError:
+                time.sleep(0.2)
+        else:
+            raise TimeoutError("upgraded synthetic runtime readiness")
+        before = snapshot()
+        with (root / "consumer.log").open("x") as log:
+            result = subprocess.run(
+                [
+                    str(args.python.absolute()),
+                    "-I",
+                    str(consumer),
+                    "--upstream",
+                    upstream,
+                    "--server-version",
+                    controls.EXPECTED_SERVER["version"],
+                    "--credential-file",
+                    str(credential),
+                    "--seed",
+                    str(args.seed.absolute()),
+                    "--seed-sha256",
+                    SEED_SHA256,
+                    "--wheel",
+                    str(args.wheel.absolute()),
+                    "--output",
+                    str(root / "consumer.json"),
+                ],
+                cwd=root,
+                env=environment,
+                stdout=log,
+                stderr=subprocess.STDOUT,
+                timeout=120,
+            )
+        after = snapshot()
+        report["consumer_exit_code"] = result.returncode
+        report["request_window_before_sha256"] = legacy.fingerprint(before)
+        report["request_window_after_sha256"] = legacy.fingerprint(after)
+        assert before == after, "consumer request window changed complete logical rows/schema"
+        assert result.returncode == 0, "installed physical consumer failed"
+        proof = json.loads((root / "consumer.json").read_text())
+        assert proof["complete"] and proof["consumer_sha256"] == report["consumer_sha256"]
+        report["consumer_complete"] = True
+    except Exception as error:
+        report["failure_type"] = type(error).__name__
+        raise
+    finally:
+        for name in names:
+            inspection = call("inspect", name, check=False)
+            volumes = []
+            if inspection.returncode == 0:
+                state = json.loads(inspection.stdout)[0]
+                volumes = [mount["Name"] for mount in state["Mounts"] if mount["Type"] == "volume"]
+                report.setdefault("oom_by_container", {})[name] = state["State"]["OOMKilled"]
+                call("rm", "-f", "-v", name)
+            absent = controls.missing_object(call("inspect", name, check=False), name)
+            volumes_absent = all(
+                controls.missing_object(call("volume", "inspect", volume, check=False), volume, volume=True)
+                for volume in volumes
+            )
+            clean.append(absent and volumes_absent)
+        credential.unlink(missing_ok=True)
+        if all(clean):
+            for path in database.iterdir():
+                path.unlink()
+            database.rmdir()
+        verify_file(args.baseline, BASELINE_SHA256)
+        report["cleanup"] = {
+            "containers_and_volumes_absent": all(clean),
+            "private_database_removed": not database.exists(),
+            "credential_removed": not credential.exists(),
+            "original_backup_unchanged": True,
+        }
+        report["finished_at"] = datetime.now(timezone.utc).isoformat()
+        report["complete"] = (
+            all(clean)
+            and report.get("consumer_complete") is True
+            and not any(report.get("oom_by_container", {}).values())
+            and "failure_type" not in report
+        )
+        (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
+        if not all(clean):
+            raise RuntimeError("owned physical runtime cleanup incomplete")
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description=__doc__)
+    for name in ("python", "wheel", "baseline", "seed", "snapshot-helper", "output"):
+        parser.add_argument("--" + name, type=Path, required=True)
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    run(parse_args())
diff --git a/tests/test_physical_command_key_readback.py b/tests/test_physical_command_key_readback.py
index c392b1b..298b74a 100644
--- a/tests/test_physical_command_key_readback.py
+++ b/tests/test_physical_command_key_readback.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import copy
+import hashlib
 import importlib.util
+import sqlite3
 from pathlib import Path
 
 import pytest
@@ -11,6 +13,12 @@
 assert SPEC is not None and SPEC.loader is not None
 consumer = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(consumer)
+CONTROLLER_SPEC = importlib.util.spec_from_file_location(
+    "physical_controller", PATH.with_name("run_physical_readback.py")
+)
+assert CONTROLLER_SPEC is not None and CONTROLLER_SPEC.loader is not None
+controller = importlib.util.module_from_spec(CONTROLLER_SPEC)
+CONTROLLER_SPEC.loader.exec_module(controller)
 
 
 def seed():
@@ -83,3 +91,30 @@ def test_legacy_receipt_cannot_claim_verification_or_change_original_identities(
     consumer.legacy_receipt(original, received)
     with pytest.raises(ValueError):
         consumer.legacy_receipt(original, {**received, **mutation})
+
+
+def test_input_hash_and_regular_file_refusal(tmp_path):
+    path = tmp_path / "synthetic"
+    path.write_bytes(b"original")
+    digest = hashlib.sha256(path.read_bytes()).hexdigest()
+    controller.verify_file(path, digest)
+    path.write_bytes(b"changed")
+    with pytest.raises(ValueError, match="identity differs"):
+        controller.verify_file(path, digest)
+    link = tmp_path / "linked"
+    link.symlink_to(path)
+    with pytest.raises(ValueError, match="identity differs"):
+        controller.verify_file(link, hashlib.sha256(path.read_bytes()).hexdigest())
+
+
+def test_schema_authority_uses_native_marker_not_sqlite_user_version():
+    with sqlite3.connect(":memory:") as connection:
+        connection.execute("PRAGMA user_version=12")
+        connection.execute("CREATE TABLE dw_server_schema(engine,version)")
+        connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)")
+        controller.native_marker(connection, 13)
+        with pytest.raises(ValueError, match="marker differs"):
+            controller.native_marker(connection, 12)
+        connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)")
+        with pytest.raises(ValueError, match="marker differs"):
+            controller.native_marker(connection, 13)

From b2383603fd13d788246313cbe0523732f489afac Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 16:41:27 +0000
Subject: [PATCH 3/6] Validate exact upgrade journal metadata and preserve all
 business rows

---
 scripts/conformance/command_keys/README.md    |  9 +-
 .../command_keys/run_physical_readback.py     | 92 +++++++++++++++++--
 tests/test_physical_command_key_readback.py   | 65 +++++++++++++
 3 files changed, 156 insertions(+), 10 deletions(-)

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 50880dc..53aa71a 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -96,6 +96,12 @@ handoff. It binds the exact stopped-backup/JSON/helper hashes, copies into a new
 private directory, checks native schema markers, runs the qualified image's
 `schema-upgrade`, and verifies the empty intent table. It then brackets this
 consumer with complete typed logical snapshots from the accepted helper.
+The upgrade itself preserves every original business row and schema object.
+Its only metadata exceptions are the native marker and a single SQLx journal
+row, validated field by field against tagged description/checksum/version,
+success, zero execution time and the actual timestamp window. The added empty
+table/catalog comes from the hash-bound tagged SQL step in an isolated SQLite
+catalog. No business row or migration-journal field is exempted from comparison.
 The request window ends before shutdown; later repair work cannot be relabeled
 as part of that zero-write window. The retained original backup is hash-checked
 again after cleanup. Two sequential containers have the same 1 CPU/1 GiB/256-PID
@@ -103,7 +109,8 @@ limits, read-only roots and bounded tmpfs paths; no compile or historical reseed
 occurs. Original seed/upgrade authority remains Server-owned.
 
 Once the producer's final source gate permits execution reuse, invoke the outer
-controller with `--python`, `--wheel`, `--baseline`, `--seed`, `--snapshot-helper`
+controller with `--python`, `--wheel`, `--baseline`, `--seed`, `--snapshot-helper`,
+the tagged `--step-sql`,
 and a new private `--output` directory. The original file inputs come from the
 producer handoff and accepted common snapshot helper, not an invented fixture.
 This draft has no installed physical result yet.
diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py
index 96a52c8..b914e7c 100644
--- a/scripts/conformance/command_keys/run_physical_readback.py
+++ b/scripts/conformance/command_keys/run_physical_readback.py
@@ -4,7 +4,6 @@
 
 import argparse
 import hashlib
-import importlib.util
 import json
 import os
 import secrets
@@ -15,21 +14,36 @@
 import time
 import urllib.request
 from contextlib import closing
+from copy import deepcopy
 from datetime import datetime, timezone
 from pathlib import Path
+from types import ModuleType
 from uuid import uuid4
 
 HELPER_SHA256 = "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad"
 BASELINE_SHA256 = "7d1826f7681342bb3a5974509648e2e8d75e7a21d262f65f3ec1f58727ccbc0c"
 SEED_SHA256 = "8c728ce827df3a11621dfb0e82cdd404bbab48784182a5e92c8d632d49047fc5"
+STEP_SHA256 = "fc3ad1730682c09704027dcbabc17cd6ded9c642a3ff20d7988d5f4b4a5726fd"
+JOURNAL_SHA384 = "53b752529a1f7addaefe62ab61160dd42b3990f805cfdc4ecd1b48927a6c2f7ce42526fc04522ac887a37208f343fc73"
+JOURNAL_DESCRIPTION = (
+    "full frozen PHP schema, native worker incarnations, session holders, stream reclaim deadlines, "
+    "service call deadlines, service admission counters and committed external payload deletions and "
+    "task ready origins and poll expiry indexes and bound run command contracts and task queue activity "
+    "dispatch pauses and keyed command intents"
+)
 
 
-def load(path, name):
-    spec = importlib.util.spec_from_file_location(name, path)
-    if spec is None or spec.loader is None:
-        raise ValueError("required source module is unavailable")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
+def load(path, name, expected=None):
+    data = path.read_bytes()
+    digest = hashlib.sha256(data).hexdigest()
+    if expected is not None and digest != expected:
+        raise ValueError("required executable source identity differs")
+    module = ModuleType(name)
+    module.__file__ = str(path)
+    # Execute precisely the bytes that were hashed, rather than reopening the
+    # external helper through an import loader after its identity check.
+    exec(compile(data, str(path), "exec"), module.__dict__)
+    module.source_sha256 = digest
     return module
 
 
@@ -44,12 +58,63 @@ def native_marker(connection, expected):
         raise ValueError("native schema ownership/version marker differs")
 
 
+def validate_journal(table, began, finished):
+    columns = table["columns"]
+    if (
+        set(columns) != {"version", "description", "installed_on", "success", "checksum", "execution_time"}
+        or len(table["rows"]) != 1
+    ):
+        raise ValueError("upgraded migration journal shape differs")
+    row = dict(zip(columns, table["rows"][0], strict=True))
+    expected = {
+        "version": ["integer", 13],
+        "description": ["text", JOURNAL_DESCRIPTION],
+        "success": ["integer", 1],
+        "checksum": ["blob", JOURNAL_SHA384],
+        "execution_time": ["integer", 0],
+    }
+    if any(row[key] != value for key, value in expected.items()):
+        raise ValueError("upgraded migration journal field differs from tagged source")
+    installed = row["installed_on"]
+    if installed[0] != "text":
+        raise ValueError("migration timestamp type differs")
+    timestamp = datetime.fromisoformat(installed[1]).replace(tzinfo=timezone.utc).timestamp()
+    if not int(began) <= timestamp <= finished:
+        raise ValueError("migration timestamp is outside the actual upgrade window")
+
+
+def validate_upgrade(before, after, added_schema, began, finished):
+    # Preserve every business row and catalog entry. Only the exact source-defined
+    # marker/journal changes and empty new intent table may differ.
+    if sorted(before["schema"] + added_schema) != after["schema"]:
+        raise ValueError("upgrade changed unexpected schema objects")
+    validate_journal(after["tables"]["_sqlx_migrations"], began, finished)
+    expected = deepcopy(before)
+    expected["schema"] = after["schema"]
+    marker = expected["tables"]["dw_server_schema"]
+    version = marker["columns"].index("version")
+    if len(marker["rows"]) != 1 or marker["rows"][0][version] != ["integer", 12]:
+        raise ValueError("baseline marker differs")
+    marker["rows"][0][version] = ["integer", 13]
+    expected["tables"]["_sqlx_migrations"] = after["tables"]["_sqlx_migrations"]
+    expected["tables"]["dw_command_intents"] = {"columns": ["command_id", "namespace", "intent_digest"], "rows": []}
+    if expected != after:
+        raise ValueError("upgrade changed business rows or unexpected metadata")
+
+
 def run(args):
+    if not __debug__:
+        raise ValueError("acceptance assertions require Python without optimization")
     # No Docker action or backup copy precedes all input identity checks.
     verify_file(args.baseline, BASELINE_SHA256)
     verify_file(args.seed, SEED_SHA256)
     verify_file(args.snapshot_helper, HELPER_SHA256)
-    legacy = load(args.snapshot_helper, "accepted_sqlite_snapshot")
+    verify_file(args.step_sql, STEP_SHA256)
+    legacy = load(args.snapshot_helper, "accepted_sqlite_snapshot", HELPER_SHA256)
+    with closing(sqlite3.connect(":memory:")) as catalog:
+        catalog.execute("CREATE TABLE dw_server_schema(engine TEXT, version INTEGER)")
+        catalog.executescript(args.step_sql.read_text())
+        added_schema = [row for row in legacy.snapshot(catalog)["schema"] if row[2] != "dw_server_schema"]
     controls = load(Path(__file__).with_name("run_response_loss.py"), "native_runtime_identity")
     root = args.output.absolute()
     root.mkdir(mode=0o700, parents=False, exist_ok=False)
@@ -91,6 +156,8 @@ def snapshot():
         "baseline_sha256": BASELINE_SHA256,
         "seed_sha256": SEED_SHA256,
         "snapshot_helper_sha256": HELPER_SHA256,
+        "runtime_verifier_sha256": controls.source_sha256,
+        "tagged_step_sha256": STEP_SHA256,
         "expected_server": controls.EXPECTED_SERVER,
         "observed_image": observed,
         "controller_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
@@ -124,18 +191,25 @@ def snapshot():
         verify_file(copied, BASELINE_SHA256)
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 12)
+        baseline_snapshot = snapshot()
+        upgrade_began = time.time()
         call("create", "--name", names[0], "--network", "none", *common, image, "schema-upgrade")
         call("start", names[0])
         assert call("wait", names[0]).stdout.strip() == "0"
         upgrade = json.loads(call("logs", names[0]).stdout)
         assert upgrade == {"backend": "sqlite", "from_schema_version": 12, "schema_version": 13, "status": "upgraded"}
         report["upgrade"] = upgrade
+        upgrade_finished = time.time()
         upgrade_state = json.loads(call("inspect", names[0]).stdout)[0]
         assert upgrade_state["State"]["OOMKilled"] is False
         call("rm", "-v", names[0])
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 13)
             assert connection.execute("SELECT count(*) FROM dw_command_intents").fetchone()[0] == 0
+        upgraded_snapshot = snapshot()
+        validate_upgrade(baseline_snapshot, upgraded_snapshot, added_schema, upgrade_began, upgrade_finished)
+        report["upgrade_row_preservation"] = True
+        report["journal_sha384"] = JOURNAL_SHA384
         call("create", "--name", names[1], "-p", "127.0.0.1::8080", *common, image)
         call("start", names[1])
         address = call("port", names[1], "8080/tcp").stdout.strip()
@@ -233,7 +307,7 @@ def snapshot():
 
 def parse_args():
     parser = argparse.ArgumentParser(description=__doc__)
-    for name in ("python", "wheel", "baseline", "seed", "snapshot-helper", "output"):
+    for name in ("python", "wheel", "baseline", "seed", "snapshot-helper", "step-sql", "output"):
         parser.add_argument("--" + name, type=Path, required=True)
     return parser.parse_args()
 
diff --git a/tests/test_physical_command_key_readback.py b/tests/test_physical_command_key_readback.py
index 298b74a..872dcf5 100644
--- a/tests/test_physical_command_key_readback.py
+++ b/tests/test_physical_command_key_readback.py
@@ -118,3 +118,68 @@ def test_schema_authority_uses_native_marker_not_sqlite_user_version():
         connection.execute("INSERT INTO dw_server_schema VALUES ('rust-development',13)")
         with pytest.raises(ValueError, match="marker differs"):
             controller.native_marker(connection, 13)
+
+
+def upgrade_snapshots():
+    columns = ["version", "description", "installed_on", "success", "checksum", "execution_time"]
+    before = {
+        "schema": [("table", "business", "business", "business schema")],
+        "user_version": 0,
+        "application_id": 0,
+        "tables": {
+            "business": {"columns": ["id"], "rows": [[["text", "preserve"]]]},
+            "dw_server_schema": {
+                "columns": ["engine", "version"],
+                "rows": [[["text", "rust-development"], ["integer", 12]]],
+            },
+            "_sqlx_migrations": {"columns": columns, "rows": []},
+        },
+    }
+    added = [("table", "dw_command_intents", "dw_command_intents", "tagged schema")]
+    after = copy.deepcopy(before)
+    after["schema"] = sorted(before["schema"] + added)
+    after["tables"]["dw_server_schema"]["rows"][0][1] = ["integer", 13]
+    after["tables"]["dw_command_intents"] = {"columns": ["command_id", "namespace", "intent_digest"], "rows": []}
+    after["tables"]["_sqlx_migrations"]["rows"] = [
+        [
+            ["integer", 13],
+            ["text", controller.JOURNAL_DESCRIPTION],
+            ["text", "2026-01-01 00:00:01"],
+            ["integer", 1],
+            ["blob", controller.JOURNAL_SHA384],
+            ["integer", 0],
+        ]
+    ]
+    return before, after, added
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda value: value["tables"]["business"]["rows"].clear(),
+        lambda value: value.update(user_version=13),
+        lambda value: value["tables"]["dw_command_intents"]["rows"].append([["text", "unexpected"]]),
+        lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(0, ["integer", 12]),
+        lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(1, ["text", "other description"]),
+        lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(2, ["text", "2025-01-01 00:00:00"]),
+        lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(3, ["integer", 0]),
+        lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(4, ["blob", "wrong checksum"]),
+        lambda value: value["tables"]["_sqlx_migrations"]["rows"][0].__setitem__(5, ["integer", 1]),
+    ],
+)
+def test_upgrade_preserves_all_business_rows_and_closes_every_journal_field(mutation):
+    before, after, added = upgrade_snapshots()
+    began, finished = 1767225600.5, 1767225602.5
+    controller.validate_upgrade(before, after, added, began, finished)
+    mutation(after)
+    with pytest.raises(ValueError):
+        controller.validate_upgrade(before, after, added, began, finished)
+
+
+def test_unapproved_helper_bytes_cannot_execute(tmp_path):
+    script = tmp_path / "helper.py"
+    marker = tmp_path / "executed"
+    script.write_text(f"from pathlib import Path\nPath({str(marker)!r}).touch()\n")
+    with pytest.raises(ValueError, match="source identity differs"):
+        controller.load(script, "unapproved_fixture", "0" * 64)
+    assert not marker.exists()

From db71738b2355feebd6cce42c396bafb1085dc1c5 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 18:20:24 +0000
Subject: [PATCH 4/6] Use captured inputs and the reviewed fixed interpreter
 for physical readbacks

---
 scripts/conformance/command_keys/README.md    |  97 +++---
 .../command_keys/consumer_bootstrap.py        |  18 +-
 .../command_keys/physical_readback.py         |  39 +--
 .../command_keys/run_physical_readback.py     | 298 ++++++++++--------
 .../command_keys/run_response_loss.py         | 111 ++++---
 tests/test_physical_command_key_readback.py   |  36 ++-
 6 files changed, 330 insertions(+), 269 deletions(-)

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 6dd5332..b6a7e76 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -84,65 +84,50 @@ This adapter alone does not complete shared216.
 
 ## Physical upgrade consumer
 
-`physical_readback.py` consumes the producer-owned JSON receipt contract from
-[Server1210](https://github.com/durable-workflow/server/pull/1210). Source/execution
-reuse is held until that PR's final source review and checks qualify. This is
-separate from the accepted synthetic-copy helper and its older manifest hash.
-
-The consumer invokes the registry-only Python Client against an already upgraded
-disposable runtime. The producer receipt must include all original signal/update
-open/completed/continued cases, complete captured histories and original update
-identities. The manifest includes the historical unkeyed-signal clarification.
-It refuses a changed receipt hash, incomplete seed, missing identities or package
-overlay before issuing a command.
-
-An outer runner must bind the producer source, stopped backup and receipt hashes,
-upgrade a private copy with the qualified image, preserve the original, disable
-unrelated maintenance writes and compare complete logical snapshots around this
-consumer. It owns the image/source check, exact runtime/database/credential
-cleanup and the final qualification receipt. Never run against the retained
-original database or reuse a production runtime credential.
-
-`run_physical_readback.py` supplies that outer integration for the frozen producer
-handoff. It binds the exact stopped-backup/JSON/helper hashes, copies into a new
-private directory, checks native schema markers, runs the qualified image's
-`schema-upgrade`, and verifies the empty intent table. It then brackets this
-consumer with complete typed logical snapshots from the accepted helper.
-The upgrade itself preserves every original business row and schema object.
-Its only metadata exceptions are the native marker and a single SQLx journal
-row, validated field by field against tagged description/checksum/version,
-success, zero execution time and the actual timestamp window. The added empty
-table/catalog comes from the hash-bound tagged SQL step in an isolated SQLite
-catalog. No business row or migration-journal field is exempted from comparison.
-The request window ends before shutdown; later repair work cannot be relabeled
-as part of that zero-write window. The retained original backup is hash-checked
-again after cleanup. Two sequential containers have the same 1 CPU/1 GiB/256-PID
-limits, read-only roots and bounded tmpfs paths; no compile or historical reseed
-occurs. Original seed/upgrade authority remains Server-owned.
-
-Once the producer's final source gate permits execution reuse, invoke the outer
-controller with `--python`, `--wheel`, `--baseline`, `--seed`, `--snapshot-helper`,
-the tagged `--step-sql`,
-and a new private `--output` directory. The original file inputs come from the
-producer handoff and accepted common snapshot helper, not an invented fixture.
-This draft has no installed physical result yet.
+The producer-owned [frozen handoff](https://github.com/durable-workflow/server/pull/1210#issuecomment-6860138558)
+is accepted at contract source `a09f156399416216d2f097c257a2164f83c7eda7`.
+The actual original seed execution remains `09a4ccda3da11853606af031a8970bbde4b2d6ec`,
+driver SHA-256 `01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b`.
+These are different authorities and must remain distinct in every receipt.
+
+`run_physical_readback.py` consumes that stopped backup/JSON plus the accepted
+common snapshot helper and hash-bound tagged SQL step. It captures every input
+with the same Git/owner/no-follow/held-identity boundary as the response-loss
+runner. Only a new private copy is upgraded. Start/inspect/wait/log/remove/absence
+checks use captured full creation IDs and ownership labels. Upgrade and runtime
+are sequential; the runtime and registry consumer are the only concurrent pair.
+The fixed image/bootstrap/hash-locked registry environment is shared with the
+reviewed response-loss runner. No host interpreter or executable input pathname
+is accepted.
 
 ```sh
-/path/to/registry-venv/bin/python -I \
-  scripts/conformance/command_keys/physical_readback.py \
-  --upstream http://127.0.0.1:12345 --server-version 3.9.2 \
-  --credential-file /path/to/private/fixture.env \
-  --seed /path/to/producer-receipt.json --seed-sha256 FULL_PRODUCER_RECEIPT_SHA256 \
+python3 -I scripts/conformance/command_keys/run_physical_readback.py \
+  --runner-commit FULL_REVIEWED_RUNNER_COMMIT \
   --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \
-  --output /path/to/private/new-consumer-receipt.json
+  --baseline /path/to/producer/stopped-backup.sqlite \
+  --seed /path/to/producer/seed-receipts.json \
+  --snapshot-helper /path/to/accepted/sqlite_legacy.py \
+  --step-sql /path/to/tagged/0013_command_intents.sql \
+  --output /path/to/private-disk-scratch/new-physical-run
 ```
 
-For each original keyed update it issues four concurrent retries with changed
-input, requires original command/run/update/acceptance identities and explicit
-`intent_verified: false`, checks a changed-name conflict, then uses GET-only
-outcome reads to distinguish pending from completed-null. Python's update helper
-uses its unscoped route; this does not claim a new explicit-run update API.
+The controller binds the frozen backup/packet/helper/step hashes. It checks the
+native marker transition and validates every business row/schema object, allowing
+only the exact new empty intent catalog and source-defined marker/journal changes.
+Every SQLx journal field is checked: version, description, tagged cumulative SQL
+checksum, success, zero execution time and the actual upgrade timestamp window.
+The installed Client's readbacks are bracketed by complete typed logical snapshots.
+That interval ends before shutdown; subsequent repair work is not relabeled as
+part of the zero-write claim. Original inputs are rechecked and preserved.
+
+For each original keyed update in open/completed/continued state, four concurrent
+changed-input retries must return the original command/run/update/acceptance
+identities with explicit `intent_verified: false`. Changed names must conflict;
+GET-only outcome reads distinguish pending from completed-null. The Python update
+helper uses its unscoped public route, so no new explicit-run update API is claimed.
 Historical NULL-key signals receive no fabricated retry key. Their compatibility
-control reads the original history, including completed/continued boundaries.
-Neither this consumer's completion flag nor a producer-only upgrade receipt alone
-qualifies the installed SDK physical cell without the outer snapshot/cleanup proof.
+control preserves their original history and business identities only.
+
+A producer upgrade receipt alone does not qualify this SDK. The exact consumer
+source, installed proof, complete snapshots, immutable input/toolchain rechecks,
+all three captured-ID cleanup records and independent review remain required.
diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py
index 21067db..60a5d13 100644
--- a/scripts/conformance/command_keys/consumer_bootstrap.py
+++ b/scripts/conformance/command_keys/consumer_bootstrap.py
@@ -22,6 +22,7 @@
 
 RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT="
 ROLES = {"adapter", "bootstrap", "proxy", "wheel", "requirements"}
+PHYSICAL_ROLES = {"adapter", "bootstrap", "verifier", "seed", "wheel", "requirements"}
 MAX_BUNDLE_BYTES = 8 * 1024 * 1024
 IMAGE_INTERPRETER = Path("/usr/local/bin/python3.12")
 
@@ -35,7 +36,9 @@ def approved_environment_link(relative, target):
 
 
 def decode_inputs(bundle):
-    if set(bundle["sources"]) != ROLES:
+    scenario = bundle.get("scenario", "response_loss")
+    roles = ROLES if scenario == "response_loss" else PHYSICAL_ROLES if scenario == "physical_readback" else None
+    if roles is None or set(bundle["sources"]) != roles:
         raise ValueError("captured source roles differ")
     result = {}
     for role, entry in bundle["sources"].items():
@@ -185,14 +188,23 @@ def main():
     descriptor = os.open(credential, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
     with os.fdopen(descriptor, "w") as stream:
         stream.write("DW_AUTH_TOKEN=" + bundle["token"] + "\n")
-    adapter = execute_source(sources["adapter"], "installed_command_keys")
+    scenario = bundle.get("scenario", "response_loss")
+    verifier = execute_source(sources.get("verifier", sources["adapter"]), "installed_package_verifier")
+    package = verifier.installed_identity(sources["wheel"])
+    adapter = (
+        verifier if scenario == "response_loss" else execute_source(sources["adapter"], "installed_physical_readback")
+    )
     try:
         result = asyncio.run(
             adapter.main(
                 SimpleNamespace(
                     upstream=bundle["upstream"],
                     credential_file=credential,
-                    proxy_data=sources["proxy"],
+                    proxy_data=sources.get("proxy"),
+                    seed_data=sources.get("seed"),
+                    seed_sha256=bundle["sources"].get("seed", {}).get("sha256"),
+                    package_identity=package,
+                    verifier_sha256=bundle["sources"].get("verifier", bundle["sources"]["adapter"])["sha256"],
                     wheel_data=sources["wheel"],
                     adapter_sha256=bundle["sources"]["adapter"]["sha256"],
                     server_version=bundle["server_version"],
diff --git a/scripts/conformance/command_keys/physical_readback.py b/scripts/conformance/command_keys/physical_readback.py
index adac25b..095a131 100644
--- a/scripts/conformance/command_keys/physical_readback.py
+++ b/scripts/conformance/command_keys/physical_readback.py
@@ -6,14 +6,11 @@
 
 from __future__ import annotations
 
-import argparse
 import asyncio
 import hashlib
-import importlib.util
 import json
 import stat
 from datetime import datetime, timezone
-from pathlib import Path
 from urllib.parse import urlsplit
 
 from durable_workflow import Client
@@ -24,16 +21,6 @@
 MANIFEST_SHA256 = "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f"
 
 
-def bound_source():
-    path = Path(__file__).with_name("response_loss.py")
-    spec = importlib.util.spec_from_file_location("installed_keys_provenance", path)
-    if spec is None or spec.loader is None:
-        raise ValueError("installed package verifier is unavailable")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module, hashlib.sha256(path.read_bytes()).hexdigest()
-
-
 def seed_cases(document):
     if (
         document.get("schema") != "durable-workflow.command-key-physical-seed.v1"
@@ -66,7 +53,7 @@ def legacy_receipt(original, received):
             raise ValueError("legacy readback changed original " + field)
 
 
-async def run(args):
+async def main(args):
     if not __debug__:
         raise ValueError("acceptance assertions require Python without optimization")
     if args.output.exists():
@@ -79,12 +66,11 @@ async def run(args):
         or args.upstream != f"http://127.0.0.1:{upstream.port}"
     ):
         raise ValueError("synthetic runtime must be canonical numeric loopback")
-    raw_seed = args.seed.read_bytes()
+    raw_seed = args.seed_data
     if hashlib.sha256(raw_seed).hexdigest() != args.seed_sha256:
         raise ValueError("producer seed receipt hash mismatch")
     cases = seed_cases(json.loads(raw_seed))
-    verifier, verifier_hash = bound_source()
-    package = verifier.installed_identity(args.wheel)
+    package = args.package_identity
     mode = args.credential_file.lstat().st_mode
     if not stat.S_ISREG(mode) or mode & 0o077:
         raise ValueError("fixture credential file must be owner-only")
@@ -93,8 +79,8 @@ async def run(args):
         "started_at": datetime.now(timezone.utc).isoformat(),
         "sdk": package,
         "seed_sha256": args.seed_sha256,
-        "verifier_sha256": verifier_hash,
-        "consumer_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "verifier_sha256": args.verifier_sha256,
+        "adapter_sha256": args.adapter_sha256,
         "scope": "installed SDK physical-upgrade readbacks only",
         "cells": [],
         "complete": False,
@@ -168,19 +154,8 @@ async def run(args):
         raise
     finally:
         args.output.write_text(json.dumps(report, indent=2) + "\n")
-
-
-def parse_args():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--upstream", required=True)
-    parser.add_argument("--server-version", required=True)
-    parser.add_argument("--credential-file", type=Path, required=True)
-    parser.add_argument("--seed", type=Path, required=True)
-    parser.add_argument("--seed-sha256", required=True)
-    parser.add_argument("--wheel", type=Path, required=True)
-    parser.add_argument("--output", type=Path, required=True)
-    return parser.parse_args()
+    return report
 
 
 if __name__ == "__main__":
-    asyncio.run(run(parse_args()))
+    raise SystemExit("Use the captured-input run_physical_readback.py controller")
diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py
index b914e7c..0f213e2 100644
--- a/scripts/conformance/command_keys/run_physical_readback.py
+++ b/scripts/conformance/command_keys/run_physical_readback.py
@@ -3,17 +3,17 @@
 from __future__ import annotations
 
 import argparse
+import base64
 import hashlib
 import json
 import os
+import re
 import secrets
-import shutil
 import sqlite3
-import stat
 import subprocess
 import time
 import urllib.request
-from contextlib import closing
+from contextlib import ExitStack, closing
 from copy import deepcopy
 from datetime import datetime, timezone
 from pathlib import Path
@@ -33,26 +33,16 @@
 )
 
 
-def load(path, name, expected=None):
-    data = path.read_bytes()
-    digest = hashlib.sha256(data).hexdigest()
-    if expected is not None and digest != expected:
+def load_bytes(data, name, expected, *, filename=None):
+    if hashlib.sha256(data).hexdigest() != expected:
         raise ValueError("required executable source identity differs")
     module = ModuleType(name)
-    module.__file__ = str(path)
-    # Execute precisely the bytes that were hashed, rather than reopening the
-    # external helper through an import loader after its identity check.
-    exec(compile(data, str(path), "exec"), module.__dict__)
-    module.source_sha256 = digest
+    module.__file__ = filename or "captured-" + name + ".py"
+    exec(compile(data, module.__file__, "exec"), module.__dict__)
+    module.source_sha256 = expected
     return module
 
 
-def verify_file(path, expected):
-    info = path.lstat()
-    if not stat.S_ISREG(info.st_mode) or hashlib.sha256(path.read_bytes()).hexdigest() != expected:
-        raise ValueError("required producer/helper input identity differs")
-
-
 def native_marker(connection, expected):
     if connection.execute("SELECT engine,version FROM dw_server_schema").fetchall() != [("rust-development", expected)]:
         raise ValueError("native schema ownership/version marker differs")
@@ -102,20 +92,25 @@ def validate_upgrade(before, after, added_schema, began, finished):
         raise ValueError("upgrade changed business rows or unexpected metadata")
 
 
-def run(args):
-    if not __debug__:
-        raise ValueError("acceptance assertions require Python without optimization")
-    # No Docker action or backup copy precedes all input identity checks.
-    verify_file(args.baseline, BASELINE_SHA256)
-    verify_file(args.seed, SEED_SHA256)
-    verify_file(args.snapshot_helper, HELPER_SHA256)
-    verify_file(args.step_sql, STEP_SHA256)
-    legacy = load(args.snapshot_helper, "accepted_sqlite_snapshot", HELPER_SHA256)
+def controls_from_git(args):
+    if re.fullmatch(r"[0-9a-f]{40}", args.runner_commit) is None:
+        raise ValueError("require one exact runner commit")
+    root = Path(__file__).absolute().parents[3]
+    relative = "scripts/conformance/command_keys/run_response_loss.py"
+    data = subprocess.check_output(
+        ["git", "-C", str(root), "show", args.runner_commit + ":" + relative], env={"PATH": os.defpath}, timeout=10
+    )
+    return load_bytes(
+        data, "reviewed_runtime_controls", hashlib.sha256(data).hexdigest(), filename=str(root / relative)
+    )
+
+
+def run_bound(args, inputs, controls):
+    legacy = load_bytes(inputs["snapshot_helper"].data, "accepted_snapshot", HELPER_SHA256)
     with closing(sqlite3.connect(":memory:")) as catalog:
         catalog.execute("CREATE TABLE dw_server_schema(engine TEXT, version INTEGER)")
-        catalog.executescript(args.step_sql.read_text())
+        catalog.executescript(inputs["step_sql"].data.decode())
         added_schema = [row for row in legacy.snapshot(catalog)["schema"] if row[2] != "dw_server_schema"]
-    controls = load(Path(__file__).with_name("run_response_loss.py"), "native_runtime_identity")
     root = args.output.absolute()
     root.mkdir(mode=0o700, parents=False, exist_ok=False)
     database = root / "database"
@@ -124,7 +119,7 @@ def run(args):
     config = root / "docker-config"
     config.mkdir(mode=0o700)
     (config / "config.json").write_text('{"auths":{}}\n')
-    environment = {"PATH": os.defpath, "TMPDIR": str(root), "PYTHONUNBUFFERED": "1"}
+    environment = {"PATH": os.defpath, "TMPDIR": str(root)}
     docker = ["docker", "--config", str(config)]
 
     def call(*arguments, check=True, timeout=45):
@@ -138,30 +133,31 @@ def snapshot():
             return legacy.snapshot(connection)
 
     image = "ghcr.io/durable-workflow/server@" + controls.EXPECTED_SERVER["index"]
+    python_image = "docker.io/library/python@" + controls.PYTHON_INDEX
     observed = controls.image_identity(json.loads(call("image", "inspect", image).stdout))
-    suffix = uuid4().hex[:12]
-    names = ["sdk-python-216-physical-upgrade-" + suffix, "sdk-python-216-physical-runtime-" + suffix]
+    interpreter = controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout))
+    nonce = uuid4().hex
+    token = secrets.token_hex(24)
     credential = root / "fixture.env"
     descriptor = os.open(credential, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
     with os.fdopen(descriptor, "w") as stream:
         stream.write(
-            "DW_AUTH_TOKEN="
-            + secrets.token_hex(24)
-            + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDW_RUST_MAINTENANCE_LOOP=0"
-            + "\nDB_CONNECTION=sqlite\nDB_DATABASE=/cohort/upgraded.sqlite\n"
+            "DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\n"
+            "DW_RUST_MAINTENANCE_LOOP=0\nDB_CONNECTION=sqlite\nDB_DATABASE=/cohort/upgraded.sqlite\n"
         )
-    consumer = Path(__file__).with_name("physical_readback.py")
     report = {
         "started_at": datetime.now(timezone.utc).isoformat(),
-        "baseline_sha256": BASELINE_SHA256,
-        "seed_sha256": SEED_SHA256,
-        "snapshot_helper_sha256": HELPER_SHA256,
-        "runtime_verifier_sha256": controls.source_sha256,
-        "tagged_step_sha256": STEP_SHA256,
+        "runner_commit": args.runner_commit,
+        "accepted_contract_source": "a09f156399416216d2f097c257a2164f83c7eda7",
+        "actual_seed_source": "09a4ccda3da11853606af031a8970bbde4b2d6ec",
+        "actual_seed_driver_sha256": "01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b",
         "expected_server": controls.EXPECTED_SERVER,
         "observed_image": observed,
-        "controller_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
-        "consumer_sha256": hashlib.sha256(consumer.read_bytes()).hexdigest(),
+        "interpreter_image": interpreter,
+        "ownership_nonce": nonce,
+        "owned_containers": {},
+        "inputs": {role: {"sha256": value.sha256, "bytes": len(value.data)} for role, value in inputs.items()},
+        "cleanup_by_id": {},
         "complete": False,
     }
     common = [
@@ -173,6 +169,10 @@ def snapshot():
         "1g",
         "--pids-limit",
         "256",
+        "--label",
+        controls.OWNER_LABEL + "=sdk-python",
+        "--label",
+        controls.RUN_LABEL + "=" + nonce,
         "--tmpfs",
         "/tmp:mode=1777,size=128m",
         "--tmpfs",
@@ -184,130 +184,182 @@ def snapshot():
         "--env-file",
         str(credential),
     ]
-    clean = []
+
+    def create(role, *arguments):
+        name = "sdk-python-216-physical-" + role + "-" + nonce[:12]
+        identity = controls.created_id(call("create", "--name", name, *common, *arguments).stdout)
+        if identity in [entry["id"] for entry in report["owned_containers"].values()]:
+            raise ValueError("duplicate creation ID")
+        report["owned_containers"][role] = {"id": identity, "requested_name": name}
+        if controls.inspect_owned(call, identity, nonce) is None:
+            raise ValueError("created object vanished")
+        return identity
+
     try:
-        shutil.copyfile(args.baseline, copied)
-        copied.chmod(0o600)
-        verify_file(copied, BASELINE_SHA256)
+        for value in inputs.values():
+            value.recheck()
+        descriptor = os.open(copied, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        with os.fdopen(descriptor, "wb") as stream:
+            stream.write(inputs["baseline"].data)
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 12)
-        baseline_snapshot = snapshot()
-        upgrade_began = time.time()
-        call("create", "--name", names[0], "--network", "none", *common, image, "schema-upgrade")
-        call("start", names[0])
-        assert call("wait", names[0]).stdout.strip() == "0"
-        upgrade = json.loads(call("logs", names[0]).stdout)
-        assert upgrade == {"backend": "sqlite", "from_schema_version": 12, "schema_version": 13, "status": "upgraded"}
+        baseline = snapshot()
+        began = time.time()
+        upgrade_id = create("upgrade", "--network", "none", image, "schema-upgrade")
+        call("start", upgrade_id)
+        if call("wait", upgrade_id).stdout.strip() != "0":
+            raise RuntimeError("schema upgrade failed")
+        upgrade = json.loads(call("logs", upgrade_id).stdout)
+        if upgrade != {"backend": "sqlite", "from_schema_version": 12, "schema_version": 13, "status": "upgraded"}:
+            raise ValueError("unexpected upgrade result")
+        finished = time.time()
         report["upgrade"] = upgrade
-        upgrade_finished = time.time()
-        upgrade_state = json.loads(call("inspect", names[0]).stdout)[0]
-        assert upgrade_state["State"]["OOMKilled"] is False
-        call("rm", "-v", names[0])
+        report["cleanup_by_id"]["upgrade"] = controls.cleanup_owned(call, upgrade_id, nonce)
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 13)
-            assert connection.execute("SELECT count(*) FROM dw_command_intents").fetchone()[0] == 0
-        upgraded_snapshot = snapshot()
-        validate_upgrade(baseline_snapshot, upgraded_snapshot, added_schema, upgrade_began, upgrade_finished)
+        validate_upgrade(baseline, snapshot(), added_schema, began, finished)
         report["upgrade_row_preservation"] = True
         report["journal_sha384"] = JOURNAL_SHA384
-        call("create", "--name", names[1], "-p", "127.0.0.1::8080", *common, image)
-        call("start", names[1])
-        address = call("port", names[1], "8080/tcp").stdout.strip()
+        runtime_id = create("runtime", "-p", "127.0.0.1::8080", image)
+        call("start", runtime_id)
+        address = call("port", runtime_id, "8080/tcp").stdout.strip()
         host, port = address.split(":")
-        assert host == "127.0.0.1" and port.isdigit() and 1 <= int(port) <= 65535
+        if host != "127.0.0.1" or not port.isdigit() or not 1 <= int(port) <= 65535:
+            raise ValueError("runtime did not bind numeric loopback")
         upstream = "http://" + address
+        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
         deadline = time.monotonic() + 45
         while time.monotonic() < deadline:
             try:
-                with urllib.request.urlopen(upstream + "/api/health", timeout=2) as response:
+                with opener.open(upstream + "/api/health", timeout=2) as response:
                     if response.status == 200:
                         break
             except OSError:
                 time.sleep(0.2)
         else:
-            raise TimeoutError("upgraded synthetic runtime readiness")
+            raise TimeoutError("upgraded runtime readiness")
+        bundle = {
+            "scenario": "physical_readback",
+            "sources": {
+                role: {"sha256": value.sha256, "base64": base64.b64encode(value.data).decode()}
+                for role, value in inputs.items()
+                if role in {"adapter", "bootstrap", "requirements", "verifier", "seed", "wheel"}
+            },
+            "upstream": upstream,
+            "token": token,
+            "server_version": controls.EXPECTED_SERVER["version"],
+        }
+        for value in inputs.values():
+            value.recheck()
+        consumer_id = controls.create_consumer(call, "sdk-python-216-physical-consumer-" + nonce[:12], nonce, inputs)
+        if consumer_id in [entry["id"] for entry in report["owned_containers"].values()]:
+            raise ValueError("duplicate consumer ID")
+        report["owned_containers"]["consumer"] = {"id": consumer_id}
+        if controls.inspect_owned(call, consumer_id, nonce) is None:
+            raise ValueError("consumer vanished")
         before = snapshot()
-        with (root / "consumer.log").open("x") as log:
-            result = subprocess.run(
-                [
-                    str(args.python.absolute()),
-                    "-I",
-                    str(consumer),
-                    "--upstream",
-                    upstream,
-                    "--server-version",
-                    controls.EXPECTED_SERVER["version"],
-                    "--credential-file",
-                    str(credential),
-                    "--seed",
-                    str(args.seed.absolute()),
-                    "--seed-sha256",
-                    SEED_SHA256,
-                    "--wheel",
-                    str(args.wheel.absolute()),
-                    "--output",
-                    str(root / "consumer.json"),
-                ],
-                cwd=root,
-                env=environment,
-                stdout=log,
-                stderr=subprocess.STDOUT,
-                timeout=120,
-            )
+        result = subprocess.run(
+            docker + ["start", "-a", "-i", consumer_id],
+            input=json.dumps(bundle),
+            env=environment,
+            capture_output=True,
+            text=True,
+            timeout=600,
+        )
         after = snapshot()
-        report["consumer_exit_code"] = result.returncode
+        (root / "consumer.log").write_text(result.stdout + result.stderr)
         report["request_window_before_sha256"] = legacy.fingerprint(before)
         report["request_window_after_sha256"] = legacy.fingerprint(after)
-        assert before == after, "consumer request window changed complete logical rows/schema"
-        assert result.returncode == 0, "installed physical consumer failed"
-        proof = json.loads((root / "consumer.json").read_text())
-        assert proof["complete"] and proof["consumer_sha256"] == report["consumer_sha256"]
+        if before != after:
+            raise ValueError("consumer window changed complete logical rows/schema")
+        state = controls.inspect_owned(call, consumer_id, nonce)
+        if state is None or state["State"]["Running"]:
+            raise ValueError("consumer final state unverified")
+        report["consumer_exit_code"] = state["State"]["ExitCode"]
+        if result.returncode != 0 or report["consumer_exit_code"] != 0:
+            raise RuntimeError("installed consumer failed")
+        for value in inputs.values():
+            value.recheck()
+        report["input_identities_rechecked"] = True
+        if controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) != interpreter:
+            raise ValueError("interpreter image changed")
+        report["interpreter_identity_rechecked"] = True
+        proof = controls.consumer_result(result.stdout, inputs, scenario="physical_readback")
+        (root / "consumer.json").write_text(json.dumps(proof, indent=2) + "\n")
         report["consumer_complete"] = True
     except Exception as error:
         report["failure_type"] = type(error).__name__
         raise
     finally:
-        for name in names:
-            inspection = call("inspect", name, check=False)
-            volumes = []
-            if inspection.returncode == 0:
-                state = json.loads(inspection.stdout)[0]
-                volumes = [mount["Name"] for mount in state["Mounts"] if mount["Type"] == "volume"]
-                report.setdefault("oom_by_container", {})[name] = state["State"]["OOMKilled"]
-                call("rm", "-f", "-v", name)
-            absent = controls.missing_object(call("inspect", name, check=False), name)
-            volumes_absent = all(
-                controls.missing_object(call("volume", "inspect", volume, check=False), volume, volume=True)
-                for volume in volumes
-            )
-            clean.append(absent and volumes_absent)
+        for role, owned in report["owned_containers"].items():
+            if role in report["cleanup_by_id"]:
+                continue
+            try:
+                report["cleanup_by_id"][role] = controls.cleanup_owned(call, owned["id"], nonce)
+            except Exception as error:
+                report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__}
         credential.unlink(missing_ok=True)
-        if all(clean):
+        clean = all(value.get("absent") and value.get("volumes_absent") for value in report["cleanup_by_id"].values())
+        if clean:
             for path in database.iterdir():
                 path.unlink()
             database.rmdir()
-        verify_file(args.baseline, BASELINE_SHA256)
+        try:
+            for value in inputs.values():
+                value.recheck()
+            report["original_inputs_unchanged"] = True
+        except Exception as error:
+            report["post_input_failure_type"] = type(error).__name__
         report["cleanup"] = {
-            "containers_and_volumes_absent": all(clean),
+            "containers_and_volumes_absent": clean,
             "private_database_removed": not database.exists(),
             "credential_removed": not credential.exists(),
-            "original_backup_unchanged": True,
         }
         report["finished_at"] = datetime.now(timezone.utc).isoformat()
         report["complete"] = (
-            all(clean)
+            len(report["cleanup_by_id"]) == 3
+            and clean
             and report.get("consumer_complete") is True
-            and not any(report.get("oom_by_container", {}).values())
+            and all(value.get("oom_killed") is False for value in report["cleanup_by_id"].values())
             and "failure_type" not in report
+            and "post_input_failure_type" not in report
         )
         (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
-        if not all(clean):
-            raise RuntimeError("owned physical runtime cleanup incomplete")
+        if not clean:
+            raise RuntimeError("owned physical cleanup incomplete")
+        if not report["complete"] and "failure_type" not in report:
+            raise RuntimeError("physical evidence incomplete")
+
+
+def run(args):
+    if not __debug__:
+        raise ValueError("acceptance requires assertions enabled")
+    controls = controls_from_git(args)
+    with ExitStack() as stack:
+        inputs = controls.captured_inputs(
+            args,
+            stack,
+            source_files={
+                "controller": "run_physical_readback.py",
+                "support": "run_response_loss.py",
+                "adapter": "physical_readback.py",
+                "verifier": "response_loss.py",
+            },
+            data_files={
+                "wheel": (args.wheel, 4 * 1024 * 1024, controls.WHEEL_SHA256),
+                "baseline": (args.baseline, 64 * 1024 * 1024, BASELINE_SHA256),
+                "seed": (args.seed, 256 * 1024, SEED_SHA256),
+                "snapshot_helper": (args.snapshot_helper, 128 * 1024, HELPER_SHA256),
+                "step_sql": (args.step_sql, 16 * 1024, STEP_SHA256),
+            },
+        )
+        run_bound(args, inputs, controls)
 
 
 def parse_args():
     parser = argparse.ArgumentParser(description=__doc__)
-    for name in ("python", "wheel", "baseline", "seed", "snapshot-helper", "step-sql", "output"):
+    parser.add_argument("--runner-commit", required=True)
+    for name in ("wheel", "baseline", "seed", "snapshot-helper", "step-sql", "output"):
         parser.add_argument("--" + name, type=Path, required=True)
     return parser.parse_args()
 
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index da57ce6..16e6bd0 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -124,7 +124,7 @@ def git_bytes(root, commit, relative):
     )
 
 
-def captured_inputs(args, stack):
+def captured_inputs(args, stack, *, source_files=None, data_files=None):
     if re.fullmatch(r"[0-9a-f]{40}", args.runner_commit) is None:
         raise ValueError("runner commit must be one exact full Git SHA")
     root = Path(__file__).absolute().parents[3]
@@ -139,27 +139,33 @@ def captured_inputs(args, stack):
     # Load the immutable Git object, never a second lookup of a mutable file.
     exec(compile(guard_bytes, "reviewed-input-guards.py", "exec"), guards.__dict__)
     inputs = {}
-    for role, filename in {
+    sources = {
         "controller": "run_response_loss.py",
         "guard": "input_guards.py",
         "adapter": "response_loss.py",
         "bootstrap": "consumer_bootstrap.py",
         "requirements": "consumer-requirements.txt",
-    }.items():
+    }
+    if source_files:
+        sources.update(source_files)
+    for role, filename in sources.items():
         expected = hashlib.sha256(git_bytes(root, head, prefix + filename)).hexdigest()
         inputs[role] = stack.enter_context(
             guards.CapturedInput(root / prefix / filename, role, maximum=256 * 1024, expected=expected)
         )
-    inputs["proxy"] = stack.enter_context(
-        guards.CapturedInput(args.proxy, "proxy", maximum=64 * 1024, expected=PROXY_SHA256)
-    )
-    inputs["wheel"] = stack.enter_context(
-        guards.CapturedInput(args.wheel, "wheel", maximum=4 * 1024 * 1024, expected=WHEEL_SHA256)
+    data = (
+        data_files
+        if data_files is not None
+        else {"proxy": (args.proxy, 64 * 1024, PROXY_SHA256), "wheel": (args.wheel, 4 * 1024 * 1024, WHEEL_SHA256)}
     )
+    for role, (path, maximum, expected) in data.items():
+        inputs[role] = stack.enter_context(guards.CapturedInput(path, role, maximum=maximum, expected=expected))
     return inputs
 
 
-def consumer_result(stdout, inputs):
+def consumer_result(stdout, inputs, *, scenario="response_loss"):
+    if scenario not in {"response_loss", "physical_readback"}:
+        raise ValueError("unknown installed consumer scenario")
     lines = [line[len(RESULT_PREFIX) :] for line in stdout.splitlines() if line.startswith(RESULT_PREFIX)]
     if len(lines) != 1:
         raise ValueError("consumer did not emit exactly one bound result")
@@ -167,16 +173,61 @@ def consumer_result(stdout, inputs):
     if (
         result.get("complete") is not True
         or result.get("adapter_sha256") != inputs["adapter"].sha256
-        or result.get("proxy_sha256") != PROXY_SHA256
         or result.get("sdk", {}).get("wheel_sha256") != WHEEL_SHA256
         or result.get("sdk", {}).get("python") != PYTHON_VERSION
         or result.get("toolchain", {}).get("post_inventory_equal") is not True
         or result["toolchain"].get("requirements_sha256") != inputs["requirements"].sha256
     ):
         raise ValueError("consumer result does not bind the captured inputs/toolchain")
+    if scenario == "response_loss" and result.get("proxy_sha256") != PROXY_SHA256:
+        raise ValueError("consumer proxy identity differs")
+    if scenario == "physical_readback" and result.get("seed_sha256") != inputs["seed"].sha256:
+        raise ValueError("consumer seed identity differs")
     return result
 
 
+def create_consumer(call, name, nonce, inputs):
+    return created_id(
+        call(
+            "create",
+            "--init",
+            "--read-only",
+            "--name",
+            name,
+            "--label",
+            OWNER_LABEL + "=sdk-python",
+            "--label",
+            RUN_LABEL + "=" + nonce,
+            "--network",
+            "host",
+            "--cpus",
+            "1",
+            "--memory",
+            "1g",
+            "--pids-limit",
+            "256",
+            "--cap-drop",
+            "ALL",
+            "--security-opt",
+            "no-new-privileges",
+            "--user",
+            f"{os.getuid()}:{os.getgid()}",
+            "--tmpfs",
+            "/tmp:mode=1777,size=256m",
+            "--tmpfs",
+            f"/opt/sdk-python:mode=0700,uid={os.getuid()},gid={os.getgid()},size=256m,exec",
+            "-i",
+            "--entrypoint",
+            "/usr/local/bin/python3.12",
+            "docker.io/library/python@" + PYTHON_INDEX,
+            "-I",
+            "-B",
+            "-c",
+            inputs["bootstrap"].data.decode(),
+        ).stdout
+    )
+
+
 def run_bound(args, inputs):
     root = args.output.absolute()
     root.mkdir(mode=0o700, parents=False, exist_ok=False)
@@ -284,45 +335,7 @@ def call(*arguments, check=True, timeout=45):
         # Host networking is needed only for the fixture's numeric loopback.
         # It has no host mounts or inherited host environment. Only the synthetic
         # runtime token is delivered through the captured stdin bundle.
-        consumer_id = created_id(
-            call(
-                "create",
-                "--init",
-                "--read-only",
-                "--name",
-                names[1],
-                "--label",
-                OWNER_LABEL + "=sdk-python",
-                "--label",
-                RUN_LABEL + "=" + nonce,
-                "--network",
-                "host",
-                "--cpus",
-                "1",
-                "--memory",
-                "1g",
-                "--pids-limit",
-                "256",
-                "--cap-drop",
-                "ALL",
-                "--security-opt",
-                "no-new-privileges",
-                "--user",
-                f"{os.getuid()}:{os.getgid()}",
-                "--tmpfs",
-                "/tmp:mode=1777,size=256m",
-                "--tmpfs",
-                f"/opt/sdk-python:mode=0700,uid={os.getuid()},gid={os.getgid()},size=256m,exec",
-                "-i",
-                "--entrypoint",
-                "/usr/local/bin/python3.12",
-                python_image,
-                "-I",
-                "-B",
-                "-c",
-                inputs["bootstrap"].data.decode(),
-            ).stdout
-        )
+        consumer_id = create_consumer(call, names[1], nonce, inputs)
         if consumer_id == runtime_id:
             raise ValueError("Docker create reused an existing container ID")
         report["owned_containers"]["consumer"] = {"id": consumer_id, "requested_name": names[1]}
diff --git a/tests/test_physical_command_key_readback.py b/tests/test_physical_command_key_readback.py
index 872dcf5..d8d08fa 100644
--- a/tests/test_physical_command_key_readback.py
+++ b/tests/test_physical_command_key_readback.py
@@ -19,6 +19,14 @@
 assert CONTROLLER_SPEC is not None and CONTROLLER_SPEC.loader is not None
 controller = importlib.util.module_from_spec(CONTROLLER_SPEC)
 CONTROLLER_SPEC.loader.exec_module(controller)
+GUARD_SPEC = importlib.util.spec_from_file_location("physical_input_guards", PATH.with_name("input_guards.py"))
+assert GUARD_SPEC is not None and GUARD_SPEC.loader is not None
+guards = importlib.util.module_from_spec(GUARD_SPEC)
+GUARD_SPEC.loader.exec_module(guards)
+BOOTSTRAP_SPEC = importlib.util.spec_from_file_location("physical_bootstrap", PATH.with_name("consumer_bootstrap.py"))
+assert BOOTSTRAP_SPEC is not None and BOOTSTRAP_SPEC.loader is not None
+bootstrap = importlib.util.module_from_spec(BOOTSTRAP_SPEC)
+BOOTSTRAP_SPEC.loader.exec_module(bootstrap)
 
 
 def seed():
@@ -97,14 +105,30 @@ def test_input_hash_and_regular_file_refusal(tmp_path):
     path = tmp_path / "synthetic"
     path.write_bytes(b"original")
     digest = hashlib.sha256(path.read_bytes()).hexdigest()
-    controller.verify_file(path, digest)
+    with guards.CapturedInput(path, "baseline", maximum=100, expected=digest) as captured:
+        assert captured.data == b"original"
     path.write_bytes(b"changed")
-    with pytest.raises(ValueError, match="identity differs"):
-        controller.verify_file(path, digest)
+    with pytest.raises(ValueError, match="input refused"):
+        guards.CapturedInput(path, "baseline", maximum=100, expected=digest)
     link = tmp_path / "linked"
     link.symlink_to(path)
-    with pytest.raises(ValueError, match="identity differs"):
-        controller.verify_file(link, hashlib.sha256(path.read_bytes()).hexdigest())
+    with pytest.raises(ValueError, match="input refused"):
+        guards.CapturedInput(link, "baseline", maximum=100, expected=hashlib.sha256(path.read_bytes()).hexdigest())
+
+
+def test_physical_bundle_closes_seed_and_verifier_roles():
+    import base64
+
+    sources = {
+        role: {"base64": base64.b64encode(b"original").decode(), "sha256": hashlib.sha256(b"original").hexdigest()}
+        for role in bootstrap.PHYSICAL_ROLES
+    }
+    assert bootstrap.decode_inputs({"scenario": "physical_readback", "sources": sources})["seed"] == b"original"
+    sources["proxy"] = sources["seed"]
+    with pytest.raises(ValueError, match="roles differ"):
+        bootstrap.decode_inputs({"scenario": "physical_readback", "sources": sources})
+    with pytest.raises(ValueError, match="roles differ"):
+        bootstrap.decode_inputs({"scenario": "unknown", "sources": sources})
 
 
 def test_schema_authority_uses_native_marker_not_sqlite_user_version():
@@ -181,5 +205,5 @@ def test_unapproved_helper_bytes_cannot_execute(tmp_path):
     marker = tmp_path / "executed"
     script.write_text(f"from pathlib import Path\nPath({str(marker)!r}).touch()\n")
     with pytest.raises(ValueError, match="source identity differs"):
-        controller.load(script, "unapproved_fixture", "0" * 64)
+        controller.load_bytes(script.read_bytes(), "unapproved_fixture", "0" * 64)
     assert not marker.exists()

From ae35b2e7928497e461c85a75c0552350a36b4ea8 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 18:52:13 +0000
Subject: [PATCH 5/6] Hold workspace identities across mounts snapshots and
 evidence writes

---
 scripts/conformance/command_keys/README.md    |  12 ++
 .../command_keys/run_physical_readback.py     |  79 ++++----
 .../command_keys/run_response_loss.py         |  58 ++++--
 .../command_keys/workspace_guards.py          | 171 ++++++++++++++++++
 tests/test_command_keys_workspace.py          | 133 ++++++++++++++
 5 files changed, 397 insertions(+), 56 deletions(-)
 create mode 100644 scripts/conformance/command_keys/workspace_guards.py
 create mode 100644 tests/test_command_keys_workspace.py

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index b6a7e76..f4ef388 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -131,3 +131,15 @@ control preserves their original history and business identities only.
 A producer upgrade receipt alone does not qualify this SDK. The exact consumer
 source, installed proof, complete snapshots, immutable input/toolchain rechecks,
 all three captured-ID cleanup records and independent review remain required.
+
+The host workspace also retains no-follow descriptors for every ancestor,
+private directory and created file. Config/credential bytes are immutable;
+the copied SQLite file may change contents but must keep its owned single-link
+inode. Docker configuration, credential reads and database binds use held
+`/proc/<controller-pid>/fd` paths. Snapshots and lifecycle calls check workspace
+identities before and after use. Evidence is created exclusively relative to the
+held directory, with the creation descriptor retained through verification.
+A substituted parent, copied database, configuration, credential or report refuses
+qualification. Known-container cleanup stays bound to captured IDs and preserves
+replacement paths. Workspace substitution controls must pass before installed
+response-loss or physical results qualify.
diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py
index 0f213e2..96df527 100644
--- a/scripts/conformance/command_keys/run_physical_readback.py
+++ b/scripts/conformance/command_keys/run_physical_readback.py
@@ -105,32 +105,41 @@ def controls_from_git(args):
     )
 
 
-def run_bound(args, inputs, controls):
+def run_bound(args, inputs, controls, workspace):
     legacy = load_bytes(inputs["snapshot_helper"].data, "accepted_snapshot", HELPER_SHA256)
     with closing(sqlite3.connect(":memory:")) as catalog:
         catalog.execute("CREATE TABLE dw_server_schema(engine TEXT, version INTEGER)")
         catalog.executescript(inputs["step_sql"].data.decode())
         added_schema = [row for row in legacy.snapshot(catalog)["schema"] if row[2] != "dw_server_schema"]
-    root = args.output.absolute()
-    root.mkdir(mode=0o700, parents=False, exist_ok=False)
-    database = root / "database"
-    database.mkdir(mode=0o700)
-    copied = database / "upgraded.sqlite"
-    config = root / "docker-config"
-    config.mkdir(mode=0o700)
-    (config / "config.json").write_text('{"auths":{}}\n')
-    environment = {"PATH": os.defpath, "TMPDIR": str(root)}
-    docker = ["docker", "--config", str(config)]
+    database = workspace.directory("database")
+    copied = Path(workspace.proc(database, "upgraded.sqlite"))
+    config = workspace.directory("docker-config")
+    configuration = workspace.write("config.json", '{"auths":{}}\n', parent=config)
+    environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)}
+    docker = ["docker", "--config", workspace.proc(config)]
 
     def call(*arguments, check=True, timeout=45):
+        workspace.check()
+        result = subprocess.run(
+            docker + list(arguments), env=environment, check=check, capture_output=True, text=True, timeout=timeout
+        )
+        if arguments[0] != "create":
+            workspace.check()
+        return result
+
+    def cleanup_call(*arguments, check=True, timeout=45):
+        workspace.check_file(configuration)
         return subprocess.run(
             docker + list(arguments), env=environment, check=check, capture_output=True, text=True, timeout=timeout
         )
 
     def snapshot():
+        workspace.check()
         with closing(sqlite3.connect(copied.as_uri() + "?mode=ro", uri=True, timeout=2)) as connection:
             connection.execute("BEGIN")
-            return legacy.snapshot(connection)
+            result = legacy.snapshot(connection)
+        workspace.check()
+        return result
 
     image = "ghcr.io/durable-workflow/server@" + controls.EXPECTED_SERVER["index"]
     python_image = "docker.io/library/python@" + controls.PYTHON_INDEX
@@ -138,13 +147,11 @@ def snapshot():
     interpreter = controls.interpreter_identity(json.loads(call("image", "inspect", python_image).stdout))
     nonce = uuid4().hex
     token = secrets.token_hex(24)
-    credential = root / "fixture.env"
-    descriptor = os.open(credential, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
-    with os.fdopen(descriptor, "w") as stream:
-        stream.write(
-            "DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\n"
-            "DW_RUST_MAINTENANCE_LOOP=0\nDB_CONNECTION=sqlite\nDB_DATABASE=/cohort/upgraded.sqlite\n"
-        )
+    credential = workspace.write(
+        "fixture.env",
+        "DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\n"
+        "DW_RUST_MAINTENANCE_LOOP=0\nDB_CONNECTION=sqlite\nDB_DATABASE=/cohort/upgraded.sqlite\n",
+    )
     report = {
         "started_at": datetime.now(timezone.utc).isoformat(),
         "runner_commit": args.runner_commit,
@@ -180,9 +187,9 @@ def snapshot():
         "--tmpfs",
         "/app/database:uid=1000,gid=1000,mode=0700,size=16m",
         "-v",
-        str(database) + ":/cohort:rw",
+        workspace.proc(database) + ":/cohort:rw",
         "--env-file",
-        str(credential),
+        workspace.proc(credential["fd"]),
     ]
 
     def create(role, *arguments):
@@ -198,9 +205,7 @@ def create(role, *arguments):
     try:
         for value in inputs.values():
             value.recheck()
-        descriptor = os.open(copied, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
-        with os.fdopen(descriptor, "wb") as stream:
-            stream.write(inputs["baseline"].data)
+        workspace.write("upgraded.sqlite", inputs["baseline"].data, parent=database, immutable=False)
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 12)
         baseline = snapshot()
@@ -214,7 +219,7 @@ def create(role, *arguments):
             raise ValueError("unexpected upgrade result")
         finished = time.time()
         report["upgrade"] = upgrade
-        report["cleanup_by_id"]["upgrade"] = controls.cleanup_owned(call, upgrade_id, nonce)
+        report["cleanup_by_id"]["upgrade"] = controls.cleanup_owned(cleanup_call, upgrade_id, nonce)
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 13)
         validate_upgrade(baseline, snapshot(), added_schema, began, finished)
@@ -267,7 +272,7 @@ def create(role, *arguments):
             timeout=600,
         )
         after = snapshot()
-        (root / "consumer.log").write_text(result.stdout + result.stderr)
+        workspace.write("consumer.log", result.stdout + result.stderr)
         report["request_window_before_sha256"] = legacy.fingerprint(before)
         report["request_window_after_sha256"] = legacy.fingerprint(after)
         if before != after:
@@ -285,7 +290,9 @@ def create(role, *arguments):
             raise ValueError("interpreter image changed")
         report["interpreter_identity_rechecked"] = True
         proof = controls.consumer_result(result.stdout, inputs, scenario="physical_readback")
-        (root / "consumer.json").write_text(json.dumps(proof, indent=2) + "\n")
+        workspace.write("consumer.json", json.dumps(proof, indent=2) + "\n")
+        workspace.check()
+        report["workspace_identities_rechecked"] = True
         report["consumer_complete"] = True
     except Exception as error:
         report["failure_type"] = type(error).__name__
@@ -295,15 +302,13 @@ def create(role, *arguments):
             if role in report["cleanup_by_id"]:
                 continue
             try:
-                report["cleanup_by_id"][role] = controls.cleanup_owned(call, owned["id"], nonce)
+                report["cleanup_by_id"][role] = controls.cleanup_owned(cleanup_call, owned["id"], nonce)
             except Exception as error:
                 report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__}
-        credential.unlink(missing_ok=True)
+        workspace.remove(credential)
         clean = all(value.get("absent") and value.get("volumes_absent") for value in report["cleanup_by_id"].values())
         if clean:
-            for path in database.iterdir():
-                path.unlink()
-            database.rmdir()
+            workspace.remove_database(database)
         try:
             for value in inputs.values():
                 value.recheck()
@@ -312,8 +317,8 @@ def create(role, *arguments):
             report["post_input_failure_type"] = type(error).__name__
         report["cleanup"] = {
             "containers_and_volumes_absent": clean,
-            "private_database_removed": not database.exists(),
-            "credential_removed": not credential.exists(),
+            "private_database_removed": clean,
+            "credential_removed": True,
         }
         report["finished_at"] = datetime.now(timezone.utc).isoformat()
         report["complete"] = (
@@ -324,7 +329,8 @@ def create(role, *arguments):
             and "failure_type" not in report
             and "post_input_failure_type" not in report
         )
-        (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
+        workspace.write("run.json", json.dumps(report, indent=2) + "\n")
+        workspace.check()
         if not clean:
             raise RuntimeError("owned physical cleanup incomplete")
         if not report["complete"] and "failure_type" not in report:
@@ -353,7 +359,8 @@ def run(args):
                 "step_sql": (args.step_sql, 16 * 1024, STEP_SHA256),
             },
         )
-        run_bound(args, inputs, controls)
+        workspace = stack.enter_context(controls.workspace_for(inputs, args.output))
+        run_bound(args, inputs, controls, workspace)
 
 
 def parse_args():
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index 16e6bd0..6887fe6 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -145,6 +145,7 @@ def captured_inputs(args, stack, *, source_files=None, data_files=None):
         "adapter": "response_loss.py",
         "bootstrap": "consumer_bootstrap.py",
         "requirements": "consumer-requirements.txt",
+        "workspace": "workspace_guards.py",
     }
     if source_files:
         sources.update(source_files)
@@ -228,16 +229,30 @@ def create_consumer(call, name, nonce, inputs):
     )
 
 
-def run_bound(args, inputs):
-    root = args.output.absolute()
-    root.mkdir(mode=0o700, parents=False, exist_ok=False)
-    config = root / "docker-config"
-    config.mkdir(mode=0o700)
-    (config / "config.json").write_text('{"auths":{}}\n')
-    environment = {"PATH": os.defpath, "TMPDIR": str(root)}
-    docker = ["docker", "--config", str(config)]
+def workspace_for(inputs, path):
+    module = ModuleType("reviewed-workspace-guards")
+    exec(compile(inputs["workspace"].data, "captured-workspace-guards.py", "exec"), module.__dict__)
+    return module.Workspace(path)
+
+
+def run_bound(args, inputs, workspace):
+    config = workspace.directory("docker-config")
+    configuration = workspace.write("config.json", '{"auths":{}}\n', parent=config)
+    environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)}
+    docker = ["docker", "--config", workspace.proc(config)]
 
     def call(*arguments, check=True, timeout=45):
+        workspace.check()
+        result = subprocess.run(
+            docker + list(arguments), check=check, capture_output=True, text=True, timeout=timeout, env=environment
+        )
+        # Capture create's returned ownership ID before the caller's next guard.
+        if arguments[0] != "create":
+            workspace.check()
+        return result
+
+    def cleanup_call(*arguments, check=True, timeout=45):
+        workspace.check_file(configuration)
         return subprocess.run(
             docker + list(arguments), check=check, capture_output=True, text=True, timeout=timeout, env=environment
         )
@@ -250,10 +265,9 @@ def call(*arguments, check=True, timeout=45):
     suffix = nonce[:12]
     names = ["sdk-python-216-response-loss-" + suffix, "sdk-python-216-consumer-" + suffix]
     token = secrets.token_hex(24)
-    credential = root / "fixture.env"
-    descriptor = os.open(credential, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
-    with os.fdopen(descriptor, "w") as stream:
-        stream.write("DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDB_CONNECTION=sqlite\n")
+    credential = workspace.write(
+        "fixture.env", "DW_AUTH_TOKEN=" + token + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDB_CONNECTION=sqlite\n"
+    )
     report = {
         "started_at": datetime.now(timezone.utc).isoformat(),
         "runner_commit": args.runner_commit,
@@ -294,7 +308,7 @@ def call(*arguments, check=True, timeout=45):
                 "-p",
                 "127.0.0.1::8080",
                 "--env-file",
-                str(credential),
+                workspace.proc(credential["fd"]),
                 "-e",
                 "DB_DATABASE=/app/database/response-loss.sqlite",
                 image,
@@ -348,7 +362,7 @@ def call(*arguments, check=True, timeout=45):
             env=environment,
             timeout=600,
         )
-        (root / "consumer.log").write_text(result.stdout + result.stderr)
+        workspace.write("consumer.log", result.stdout + result.stderr)
         consumer_state = inspect_owned(call, consumer_id, nonce)
         if consumer_state is None or consumer_state["State"]["Running"]:
             raise ValueError("owned consumer final state is unverified")
@@ -363,7 +377,9 @@ def call(*arguments, check=True, timeout=45):
         if result.returncode != 0 or report["consumer_exit_code"] != 0:
             raise RuntimeError("installed consumer failed; retain its private incomplete evidence")
         consumer = consumer_result(result.stdout, inputs)
-        (root / "consumer.json").write_text(json.dumps(consumer, indent=2) + "\n")
+        workspace.write("consumer.json", json.dumps(consumer, indent=2) + "\n")
+        workspace.check()
+        report["workspace_identities_rechecked"] = True
         report["consumer_complete"] = True
     except Exception as error:
         report["failure_type"] = type(error).__name__
@@ -375,7 +391,7 @@ def call(*arguments, check=True, timeout=45):
             report["cleanup_by_id"] = {}
             for role, owned in report["owned_containers"].items():
                 try:
-                    outcome = cleanup_owned(call, owned["id"], nonce)
+                    outcome = cleanup_owned(cleanup_call, owned["id"], nonce)
                 except Exception as error:
                     report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__}
                     clean.append(False)
@@ -387,10 +403,10 @@ def call(*arguments, check=True, timeout=45):
         except Exception as error:
             report["cleanup_failure_type"] = type(error).__name__
         finally:
-            credential.unlink(missing_ok=True)
+            workspace.remove(credential)
             report["cleanup"] = {
                 "containers_and_volumes_absent": len(clean) == 2 and all(clean),
-                "credential_removed": not credential.exists(),
+                "credential_removed": True,
                 "oom_killed": (
                     True
                     if any(value is True for value in oom)
@@ -409,7 +425,8 @@ def call(*arguments, check=True, timeout=45):
                 and "failure_type" not in report
                 and "cleanup_failure_type" not in report
             )
-            (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
+            workspace.write("run.json", json.dumps(report, indent=2) + "\n")
+            workspace.check()
         if not report["cleanup"]["containers_and_volumes_absent"]:
             raise RuntimeError("owned runtime cleanup incomplete")
         if not report["complete"] and "failure_type" not in report:
@@ -419,7 +436,8 @@ def call(*arguments, check=True, timeout=45):
 def run(args):
     with ExitStack() as stack:
         inputs = captured_inputs(args, stack)
-        run_bound(args, inputs)
+        workspace = stack.enter_context(workspace_for(inputs, args.output))
+        run_bound(args, inputs, workspace)
 
 
 def parse_args():
diff --git a/scripts/conformance/command_keys/workspace_guards.py b/scripts/conformance/command_keys/workspace_guards.py
new file mode 100644
index 0000000..465ea0b
--- /dev/null
+++ b/scripts/conformance/command_keys/workspace_guards.py
@@ -0,0 +1,171 @@
+"""Held, no-follow fixture workspace identities and descriptor-relative outputs."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import stat
+from pathlib import Path
+
+
+def shape(info):
+    return info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, info.st_nlink
+
+
+class Workspace:
+    def __init__(self, path):
+        self.path = Path(path).absolute()
+        self.handles = []
+        self.directories = []
+        self.files = {}
+        if ".." in self.path.parts:
+            raise ValueError("workspace path is not canonical")
+        try:
+            parent = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+            self.handles.append(parent)
+            for index, part in enumerate(self.path.parts[1:]):
+                if index == len(self.path.parts) - 2:
+                    os.mkdir(part, mode=0o700, dir_fd=parent)
+                descriptor = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
+                self.handles.append(descriptor)
+                info = os.fstat(descriptor)
+                sticky_root = info.st_uid == 0 and bool(info.st_mode & stat.S_ISVTX)
+                if info.st_uid not in {0, os.getuid()} or (info.st_mode & 0o022 and not sticky_root):
+                    raise ValueError("workspace parent is not owner-controlled")
+                self.directories.append((parent, part, descriptor, shape(info)[:5]))
+                parent = descriptor
+            self.root = parent
+            self.check()
+        except BaseException:
+            self.close()
+            raise
+
+    def check(self):
+        for parent, name, descriptor, expected in self.directories:
+            if (
+                shape(os.fstat(descriptor))[:5] != expected
+                or shape(os.stat(name, dir_fd=parent, follow_symlinks=False))[:5] != expected
+            ):
+                raise ValueError("held workspace directory was replaced")
+        for item in self.files.values():
+            self.check_file(item)
+
+    def directory(self, name):
+        self.check()
+        self.name(name)
+        os.mkdir(name, mode=0o700, dir_fd=self.root)
+        descriptor = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.root)
+        self.handles.append(descriptor)
+        self.directories.append((self.root, name, descriptor, shape(os.fstat(descriptor))[:5]))
+        self.check()
+        return descriptor
+
+    @staticmethod
+    def name(name):
+        if not name or name in {".", ".."} or "/" in name or "\x00" in name:
+            raise ValueError("workspace entry must be one literal basename")
+
+    @staticmethod
+    def proc(descriptor, name=None):
+        return "/proc/" + str(os.getpid()) + "/fd/" + str(descriptor) + ("/" + name if name else "")
+
+    def check_file(self, item):
+        current = os.stat(item["name"], dir_fd=item["parent"], follow_symlinks=False)
+        held = os.fstat(item["fd"])
+        if shape(current) != item["shape"] or shape(held) != item["shape"]:
+            raise ValueError("held workspace file was replaced")
+        if held.st_size > item["maximum"]:
+            raise ValueError("workspace file exceeded its bound")
+        if item["sha256"] is not None:
+            data = os.pread(item["fd"], item["maximum"] + 1, 0)
+            if hashlib.sha256(data).hexdigest() != item["sha256"]:
+                raise ValueError("held workspace bytes changed")
+
+    def capture(self, parent, name, *, immutable=False, maximum=64 * 1024 * 1024):
+        self.check()
+        self.name(name)
+        descriptor = os.open(name, os.O_RDWR | os.O_NOFOLLOW, dir_fd=parent)
+        return self.remember(parent, name, descriptor, immutable=immutable, maximum=maximum)
+
+    def remember(self, parent, name, descriptor, *, immutable, maximum, initial_sha256=None):
+        self.handles.append(descriptor)
+        info = os.fstat(descriptor)
+        if (
+            not stat.S_ISREG(info.st_mode)
+            or info.st_uid != os.getuid()
+            or info.st_nlink != 1
+            or info.st_mode & 0o077
+            or info.st_size > maximum
+        ):
+            raise ValueError("workspace file is not private, regular and single-link")
+        item = {
+            "parent": parent,
+            "name": name,
+            "fd": descriptor,
+            "shape": shape(info),
+            "maximum": maximum,
+            "sha256": hashlib.sha256(os.pread(descriptor, maximum + 1, 0)).hexdigest() if immutable else None,
+        }
+        if (
+            initial_sha256 is not None
+            and hashlib.sha256(os.pread(descriptor, maximum + 1, 0)).hexdigest() != initial_sha256
+        ):
+            raise ValueError("created workspace bytes differ")
+        self.files[(parent, name)] = item
+        self.check()
+        return item
+
+    def write(self, name, data, *, parent=None, immutable=True):
+        self.check()
+        self.name(name)
+        parent = self.root if parent is None else parent
+        data = data.encode() if isinstance(data, str) else data
+        descriptor = os.open(name, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=parent)
+        try:
+            with os.fdopen(descriptor, "wb", closefd=False) as stream:
+                stream.write(data)
+                stream.flush()
+                os.fsync(descriptor)
+        except BaseException:
+            os.close(descriptor)
+            raise
+        item = self.remember(
+            parent,
+            name,
+            descriptor,
+            immutable=immutable,
+            maximum=max(64 * 1024 * 1024, len(data)),
+            initial_sha256=hashlib.sha256(data).hexdigest(),
+        )
+        self.check()
+        return item
+
+    def remove(self, item):
+        # Cleanup authority comes from the held parent/inode even if an outer
+        # pathname was renamed. A replacement entry is preserved.
+        self.check_file(item)
+        os.unlink(item["name"], dir_fd=item["parent"])
+        del self.files[(item["parent"], item["name"])]
+
+    def remove_database(self, directory):
+        self.check()
+        for name in os.listdir(directory):
+            item = self.files.get((directory, name))
+            if item is None:
+                item = self.capture(directory, name)
+            self.remove(item)
+        entry = next(value for value in self.directories if value[2] == directory)
+        self.check()
+        os.rmdir(entry[1], dir_fd=entry[0])
+        self.directories.remove(entry)
+
+    def close(self):
+        for descriptor in reversed(self.handles):
+            os.close(descriptor)
+        self.handles.clear()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_error):
+        self.close()
diff --git a/tests/test_command_keys_workspace.py b/tests/test_command_keys_workspace.py
new file mode 100644
index 0000000..0be4155
--- /dev/null
+++ b/tests/test_command_keys_workspace.py
@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import importlib.util
+import os
+import sqlite3
+from pathlib import Path
+
+import pytest
+
+PATH = Path(__file__).parents[1] / "scripts/conformance/command_keys/workspace_guards.py"
+SPEC = importlib.util.spec_from_file_location("command_workspace", PATH)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+def test_root_replacement_is_seen_and_descriptor_bind_keeps_original_database(tmp_path):
+    root = tmp_path / "work"
+    with module.Workspace(root) as workspace:
+        directory = workspace.directory("database")
+        workspace.write("copied.sqlite", b"original", parent=directory, immutable=False)
+        binding = Path(workspace.proc(directory)) / "copied.sqlite"
+        root.rename(tmp_path / "retained-original")
+        (root / "database").mkdir(parents=True)
+        (root / "database/copied.sqlite").write_bytes(b"substituted")
+        # The predecessor mount/snapshot path follows the replacement.
+        assert (root / "database/copied.sqlite").read_bytes() == b"substituted"
+        assert binding.read_bytes() == b"original"
+        with pytest.raises(ValueError, match="directory was replaced"):
+            workspace.check()
+        with pytest.raises(ValueError):
+            workspace.write("run.json", '{"complete":true}')
+        assert not (root / "run.json").exists()
+
+
+@pytest.mark.parametrize("entry", ["database", "docker-config"])
+def test_subdirectory_rename_replacement_refuses_use(tmp_path, entry):
+    root = tmp_path / "work"
+    with module.Workspace(root) as workspace:
+        workspace.directory(entry)
+        (root / entry).rename(root / (entry + "-original"))
+        (root / entry).mkdir()
+        with pytest.raises(ValueError):
+            workspace.check()
+
+
+@pytest.mark.parametrize("name", ["fixture.env", "consumer.log", "consumer.json", "run.json"])
+@pytest.mark.parametrize("replacement", ["rename", "symlink", "hardlink", "rewrite"])
+def test_workspace_output_and_credential_substitution_refuses(tmp_path, name, replacement):
+    root = tmp_path / "work"
+    with module.Workspace(root) as workspace:
+        item = workspace.write(name, b"original")
+        path = root / name
+        if replacement == "rewrite":
+            path.write_bytes(b"changed!")
+        else:
+            path.rename(root / (name + "-original"))
+            outside = tmp_path / "outside"
+            outside.write_bytes(b"foreign!")
+            if replacement == "rename":
+                path.write_bytes(b"foreign!")
+                path.chmod(0o600)
+            elif replacement == "symlink":
+                path.symlink_to(outside)
+            else:
+                os.link(outside, path)
+        # A pathname-only reader accepts the changed bytes; the held identity refuses.
+        assert path.read_bytes() != b"original"
+        with pytest.raises(ValueError):
+            workspace.check()
+        with pytest.raises(ValueError):
+            workspace.remove(item)
+        assert path.exists()
+
+
+def test_evidence_writer_refuses_existing_symlink_without_touching_marker(tmp_path):
+    root = tmp_path / "work"
+    marker = tmp_path / "marker"
+    marker.write_bytes(b"preserve")
+    with module.Workspace(root) as workspace:
+        (root / "run.json").symlink_to(marker)
+        with pytest.raises(FileExistsError):
+            workspace.write("run.json", '{"complete":true}')
+        assert marker.read_bytes() == b"preserve"
+
+
+def test_created_file_descriptor_is_held_without_reopen(tmp_path, monkeypatch):
+    root = tmp_path / "work"
+    with module.Workspace(root) as workspace:
+        original = workspace.remember
+
+        def replace_before_remember(parent, name, descriptor, **kwargs):
+            path = root / name
+            path.rename(root / "original")
+            path.write_bytes(b"false-output")
+            path.chmod(0o600)
+            return original(parent, name, descriptor, **kwargs)
+
+        monkeypatch.setattr(workspace, "remember", replace_before_remember)
+        with pytest.raises(ValueError):
+            workspace.write("consumer.json", b"true-output")
+
+
+def test_mutable_database_keeps_inode_across_legitimate_sqlite_updates(tmp_path):
+    root = tmp_path / "work"
+    with module.Workspace(root) as workspace:
+        directory = workspace.directory("database")
+        workspace.write("copied.sqlite", b"", parent=directory, immutable=False)
+        path = Path(workspace.proc(directory, "copied.sqlite"))
+        with sqlite3.connect(path) as database:
+            database.execute("CREATE TABLE values_table(value INTEGER)")
+            database.execute("INSERT INTO values_table VALUES (1)")
+        workspace.check()
+        with sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) as database:
+            assert database.execute("SELECT value FROM values_table").fetchall() == [(1,)]
+        workspace.check()
+        original = root / "database/copied.sqlite"
+        original.rename(root / "database/old.sqlite")
+        original.write_bytes(b"different")
+        original.chmod(0o600)
+        with pytest.raises(ValueError):
+            workspace.check()
+
+
+def test_workspace_parent_replacement_refuses_before_create(tmp_path):
+    parent = tmp_path / "parent"
+    parent.mkdir(mode=0o700)
+    with module.Workspace(parent / "work") as workspace:
+        parent.rename(tmp_path / "retained-parent")
+        (parent / "work").mkdir(parents=True)
+        with pytest.raises(ValueError):
+            workspace.write("consumer.log", "must not write")
+        assert not (parent / "work/consumer.log").exists()

From 377a3eeeece605da675e4b628c308dd2e6be527c Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 18:57:51 +0000
Subject: [PATCH 6/6] Verify native mount identities before entrypoint and
 decouple owned cleanup

---
 scripts/conformance/command_keys/README.md    | 10 ++--
 .../command_keys/run_physical_readback.py     | 34 +++++++++++---
 .../command_keys/run_response_loss.py         | 10 ++--
 .../command_keys/workspace_guards.py          | 37 +++++++++++++++
 tests/test_command_keys_workspace.py          | 47 +++++++++++++++++++
 5 files changed, 125 insertions(+), 13 deletions(-)

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index f4ef388..105bd7f 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -135,9 +135,13 @@ all three captured-ID cleanup records and independent review remain required.
 The host workspace also retains no-follow descriptors for every ancestor,
 private directory and created file. Config/credential bytes are immutable;
 the copied SQLite file may change contents but must keep its owned single-link
-inode. Docker configuration, credential reads and database binds use held
-`/proc/<controller-pid>/fd` paths. Snapshots and lifecycle calls check workspace
-identities before and after use. Evidence is created exclusively relative to the
+inode. Docker configuration and credential reads use held
+`/proc/<controller-pid>/fd` paths. The canonical database bind has a preflight
+inside the immutable published image: it verifies the mounted directory/file
+device, inode, owner, mode and single-link identities before executing the
+native entrypoint. Host snapshots and lifecycle calls check workspace
+identities before and after use. Cleanup uses an explicit empty kernel FD
+configuration directory, so workspace substitution cannot block known-ID cleanup. Evidence is created exclusively relative to the
 held directory, with the creation descriptor retained through verification.
 A substituted parent, copied database, configuration, credential or report refuses
 qualification. Known-container cleanup stays bound to captured IDs and preserves
diff --git a/scripts/conformance/command_keys/run_physical_readback.py b/scripts/conformance/command_keys/run_physical_readback.py
index 96df527..ce91df2 100644
--- a/scripts/conformance/command_keys/run_physical_readback.py
+++ b/scripts/conformance/command_keys/run_physical_readback.py
@@ -114,7 +114,7 @@ def run_bound(args, inputs, controls, workspace):
     database = workspace.directory("database")
     copied = Path(workspace.proc(database, "upgraded.sqlite"))
     config = workspace.directory("docker-config")
-    configuration = workspace.write("config.json", '{"auths":{}}\n', parent=config)
+    workspace.write("config.json", '{"auths":{}}\n', parent=config)
     environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)}
     docker = ["docker", "--config", workspace.proc(config)]
 
@@ -128,9 +128,13 @@ def call(*arguments, check=True, timeout=45):
         return result
 
     def cleanup_call(*arguments, check=True, timeout=45):
-        workspace.check_file(configuration)
         return subprocess.run(
-            docker + list(arguments), env=environment, check=check, capture_output=True, text=True, timeout=timeout
+            ["docker", "--config", "/proc/self/fd", *arguments],
+            env=environment,
+            check=check,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
         )
 
     def snapshot():
@@ -187,7 +191,7 @@ def snapshot():
         "--tmpfs",
         "/app/database:uid=1000,gid=1000,mode=0700,size=16m",
         "-v",
-        workspace.proc(database) + ":/cohort:rw",
+        workspace.mount_path(database) + ":/cohort:rw",
         "--env-file",
         workspace.proc(credential["fd"]),
     ]
@@ -205,12 +209,20 @@ def create(role, *arguments):
     try:
         for value in inputs.values():
             value.recheck()
-        workspace.write("upgraded.sqlite", inputs["baseline"].data, parent=database, immutable=False)
+        held_database = workspace.write("upgraded.sqlite", inputs["baseline"].data, parent=database, immutable=False)
         with closing(sqlite3.connect(copied)) as connection:
             native_marker(connection, 12)
         baseline = snapshot()
         began = time.time()
-        upgrade_id = create("upgrade", "--network", "none", image, "schema-upgrade")
+        upgrade_id = create(
+            "upgrade",
+            "--network",
+            "none",
+            "--entrypoint",
+            "/bin/sh",
+            image,
+            *workspace.native_arguments(database, held_database, "schema-upgrade"),
+        )
         call("start", upgrade_id)
         if call("wait", upgrade_id).stdout.strip() != "0":
             raise RuntimeError("schema upgrade failed")
@@ -225,7 +237,15 @@ def create(role, *arguments):
         validate_upgrade(baseline, snapshot(), added_schema, began, finished)
         report["upgrade_row_preservation"] = True
         report["journal_sha384"] = JOURNAL_SHA384
-        runtime_id = create("runtime", "-p", "127.0.0.1::8080", image)
+        runtime_id = create(
+            "runtime",
+            "-p",
+            "127.0.0.1::8080",
+            "--entrypoint",
+            "/bin/sh",
+            image,
+            *workspace.native_arguments(database, held_database),
+        )
         call("start", runtime_id)
         address = call("port", runtime_id, "8080/tcp").stdout.strip()
         host, port = address.split(":")
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index 6887fe6..df620a5 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -237,7 +237,7 @@ def workspace_for(inputs, path):
 
 def run_bound(args, inputs, workspace):
     config = workspace.directory("docker-config")
-    configuration = workspace.write("config.json", '{"auths":{}}\n', parent=config)
+    workspace.write("config.json", '{"auths":{}}\n', parent=config)
     environment = {"PATH": os.defpath, "TMPDIR": workspace.proc(workspace.root)}
     docker = ["docker", "--config", workspace.proc(config)]
 
@@ -252,9 +252,13 @@ def call(*arguments, check=True, timeout=45):
         return result
 
     def cleanup_call(*arguments, check=True, timeout=45):
-        workspace.check_file(configuration)
         return subprocess.run(
-            docker + list(arguments), check=check, capture_output=True, text=True, timeout=timeout, env=environment
+            ["docker", "--config", "/proc/self/fd", *arguments],
+            check=check,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            env=environment,
         )
 
     image = "ghcr.io/durable-workflow/server@" + EXPECTED_SERVER["index"]
diff --git a/scripts/conformance/command_keys/workspace_guards.py b/scripts/conformance/command_keys/workspace_guards.py
index 465ea0b..0a27e6d 100644
--- a/scripts/conformance/command_keys/workspace_guards.py
+++ b/scripts/conformance/command_keys/workspace_guards.py
@@ -7,6 +7,15 @@
 import stat
 from pathlib import Path
 
+NATIVE_PREFLIGHT = """set -eu
+test -d "$1" && test ! -L "$1" || exit 78
+test -f "$1/$2" && test ! -L "$1/$2" || exit 78
+test "$(stat -c '%d:%i:%u:%g:%a' -- "$1")" = "$3" || exit 78
+test "$(stat -c '%d:%i:%u:%g:%a:%h' -- "$1/$2")" = "$4" || exit 78
+shift 4
+exec /usr/local/bin/durable-workflow-server "$@"
+"""
+
 
 def shape(info):
     return info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid, info.st_nlink
@@ -69,6 +78,34 @@ def name(name):
     def proc(descriptor, name=None):
         return "/proc/" + str(os.getpid()) + "/fd/" + str(descriptor) + ("/" + name if name else "")
 
+    def mount_path(self, directory):
+        self.check()
+        entry = next(value for value in self.directories if value[2] == directory)
+        if entry[0] != self.root:
+            raise ValueError("bind source is not one held workspace child")
+        return str(self.path / entry[1])
+
+    def native_arguments(self, directory, database, *arguments):
+        self.check()
+        folder = os.fstat(directory)
+        file = os.fstat(database["fd"])
+        directory_key = (
+            f"{folder.st_dev}:{folder.st_ino}:{folder.st_uid}:{folder.st_gid}:{stat.S_IMODE(folder.st_mode):o}"
+        )
+        file_key = (
+            f"{file.st_dev}:{file.st_ino}:{file.st_uid}:{file.st_gid}:{stat.S_IMODE(file.st_mode):o}:{file.st_nlink}"
+        )
+        return [
+            "-c",
+            NATIVE_PREFLIGHT,
+            "checked-native",
+            "/cohort",
+            database["name"],
+            directory_key,
+            file_key,
+            *arguments,
+        ]
+
     def check_file(self, item):
         current = os.stat(item["name"], dir_fd=item["parent"], follow_symlinks=False)
         held = os.fstat(item["fd"])
diff --git a/tests/test_command_keys_workspace.py b/tests/test_command_keys_workspace.py
index 0be4155..000196c 100644
--- a/tests/test_command_keys_workspace.py
+++ b/tests/test_command_keys_workspace.py
@@ -131,3 +131,50 @@ def test_workspace_parent_replacement_refuses_before_create(tmp_path):
         with pytest.raises(ValueError):
             workspace.write("consumer.log", "must not write")
         assert not (parent / "work/consumer.log").exists()
+
+
+@pytest.mark.parametrize("replacement", ["none", "file", "directory", "symlink", "hardlink"])
+def test_native_preflight_refuses_substitution_before_native_execution(tmp_path, replacement):
+    import subprocess
+
+    root = tmp_path / "work"
+    marker = tmp_path / "native-executed"
+    with module.Workspace(root) as workspace:
+        directory = workspace.directory("database")
+        database = workspace.write("copied.sqlite", b"original", parent=directory, immutable=False)
+        arguments = workspace.native_arguments(directory, database, str(marker))
+        # Keep the actual inode/type preflight; substitute a harmless stand-in
+        # only for the immutable image's native executable in this unit control.
+        arguments[1] = arguments[1].replace("exec /usr/local/bin/durable-workflow-server", "exec /usr/bin/touch")
+        arguments[3] = str(root / "database")
+        path = root / "database/copied.sqlite"
+        if replacement == "directory":
+            (root / "database").rename(root / "original-database")
+            (root / "database").mkdir(mode=0o700)
+            path.write_bytes(b"foreign")
+        elif replacement != "none":
+            path.rename(root / "database/original.sqlite")
+            if replacement == "file":
+                path.write_bytes(b"foreign")
+                path.chmod(0o600)
+            elif replacement == "symlink":
+                path.symlink_to(root / "database/original.sqlite")
+            else:
+                os.link(root / "database/original.sqlite", path)
+        result = subprocess.run(["/bin/sh", *arguments], capture_output=True, text=True)
+        if replacement == "none":
+            assert result.returncode == 0 and marker.exists()
+        else:
+            assert result.returncode == 78 and not marker.exists()
+            # The predecessor entrypoint has no independent mounted-inode gate.
+            subprocess.run(["/usr/bin/touch", str(marker)], check=True)
+            assert marker.exists()
+
+
+def test_cleanup_configuration_is_an_immutable_empty_kernel_namespace():
+    # proc's FD directory contains only numeric names and cannot be seeded with
+    # a config.json after the controller's workspace/config path is replaced.
+    path = Path("/proc/self/fd/config.json")
+    assert not path.exists()
+    with pytest.raises(OSError):
+        path.write_text("must not create")