From eebb5c984d054987d4ee07c040933a65f3feb2b0 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 16:07:55 +0000
Subject: [PATCH 1/7] Add installed command-response loss acceptance adapter

---
 scripts/conformance/command_keys/README.md    |  68 +++
 .../conformance/command_keys/response_loss.py | 409 ++++++++++++++++++
 .../command_keys/run_response_loss.py         | 198 +++++++++
 tests/test_command_keys_acceptance.py         | 113 +++++
 4 files changed, 788 insertions(+)
 create mode 100644 scripts/conformance/command_keys/README.md
 create mode 100644 scripts/conformance/command_keys/response_loss.py
 create mode 100644 scripts/conformance/command_keys/run_response_loss.py
 create mode 100644 tests/test_command_keys_acceptance.py

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
new file mode 100644
index 0000000..b0387e9
--- /dev/null
+++ b/scripts/conformance/command_keys/README.md
@@ -0,0 +1,68 @@
+# Installed command-response recovery
+
+This adapter exercises the actual published Python Client and Worker for shared
+[command-key acceptance](https://github.com/durable-workflow/.github/issues/216).
+It creates synthetic workflows and a disposable SQLite runtime. It does not
+modify SDK product source or provide a new transport fault implementation.
+
+The source binds the published wheel and accepted shared proxy by SHA-256.
+The controller binds the qualified native image index, source and amd64 platform
+from [Server's independent publication review](https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177).
+Expected platform receipt fields remain separate from observed local image
+identity and runtime version. Stable aliases and final Release selection are
+separate producer gates.
+
+Prepare a registry-only virtualenv outside the checkout. Install
+`durable-workflow==2.17.2` from PyPI with wheel-only dependencies and download its
+wheel separately with `pip download --no-deps --only-binary=:all:`. Do not install
+this checkout, an editable package or a local archive. The adapter refuses a
+direct URL installation, unexpected wheel hash, mismatched runtime files or a
+source import overlay. The controller invokes the virtualenv interpreter with
+`-I`; assertions must remain enabled.
+
+Download the frozen shared proxy from
+[`117118c8`](https://github.com/durable-workflow/.github/blob/117118c83093822469b3824ba9587ff4068ed29f/conformance/command-keys/committed_response_loss.py).
+Its bound SHA-256 is
+`bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725`.
+The proxy accepts numeric loopback only and retains bounded request/response
+counts and hashes. It receives only ephemeral synthetic credentials.
+
+After checking host/Docker capacity, pull the controller's exact image with an
+empty Docker config. The controller itself never pulls an image. Run:
+
+```sh
+python3 -I scripts/conformance/command_keys/run_response_loss.py \
+  --python /path/to/registry-venv/bin/python \
+  --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \
+  --proxy /path/to/verified/committed_response_loss.py \
+  --output /path/to/private-disk-scratch/new-run
+```
+
+The output parent must exist. The new private output directory must not exist.
+One runtime is capped at 1 CPU, 1 GiB and 256 PIDs, with a read-only root,
+`--init` and one ephemeral loopback port. The consumer is sequential and bounded
+to ten minutes. A private empty Docker config and minimal subprocess environment
+exclude host credentials. The controller owns exact-name runtime/anonymous-volume
+cleanup and removes its generated credential even after consumer failure.
+
+Six workflow cells exercise eight genuinely lost upstream success responses:
+automatic and manual recovery with generated and explicit keys, both signal and
+update routes, and one continuation. The adapter independently compares sender
+and proxy request bytes, requires successful complete upstream consumption,
+checks four concurrent identical retries and a typed changed-intent conflict,
+then verifies one business execution/admission and retained identities after
+closure or continuation. Typed nested values include integers, booleans, bytes
+and Unicode. Query reads consume the receipt's `result` field.
+
+Keep a failed attempt as NONPASS. `consumer.json` and `run.json` must both report
+`complete: true`, with no OOM, exact package/runtime/source identities and verified
+cleanup, before reporting this cell as passing. Record the public runner commit,
+file hashes, tuple, timestamps and sanitized outcomes in the owning GitHub record.
+Keep diagnostic logs private and temporary; do not commit generated payloads.
+
+This does not qualify synthetic legacy transforms or physical upgrades. The
+accepted common stopped-copy helper is separately maintained by
+[SDK Rust260](https://github.com/durable-workflow/sdk-rust/pull/260), and Server1210
+owns the real published historical seed/upgrade. Namespace/run fences and maximum
+Unicode key acceptance also remain distinct installed cells. No full shared216
+PASS follows from this adapter alone.
diff --git a/scripts/conformance/command_keys/response_loss.py b/scripts/conformance/command_keys/response_loss.py
new file mode 100644
index 0000000..84de621
--- /dev/null
+++ b/scripts/conformance/command_keys/response_loss.py
@@ -0,0 +1,409 @@
+"""Installed SDK adapter for shared #216's separately reviewed loopback fixture.
+
+Run only with a reviewed proxy file/hash and synthetic runtime credentials.
+No transport fault implementation or product source overlay lives here.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import hashlib
+import importlib.metadata
+import importlib.util
+import json
+import platform
+import stat
+import threading
+import time
+from collections import Counter
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+from zipfile import ZipFile
+
+import httpx
+
+from durable_workflow import Client, Worker, workflow
+from durable_workflow.errors import SignalFailed, UpdateFailed
+from durable_workflow.retry_policy import TransportRetryPolicy
+
+WORKFLOW_TYPE = "sdk-python-command-response-loss"
+SDK_VERSION = "2.17.2"
+WHEEL_SHA256 = "7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc"
+PROXY_COMMIT = "117118c83093822469b3824ba9587ff4068ed29f"
+PROXY_SHA256 = "bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725"
+PAYLOAD = {
+    "number": 1,
+    "flag": True,
+    "large": 2**62 + 7,
+    "unicode": "Ünïcødé ☃ 日本",
+    "nested": [None, False, "text", 1.25, b"\x00\xff"],
+}
+IDENTITIES = ("command_id", "command_sequence", "run_id", "resolved_run_id", "update_id", "accepted_at")
+
+
+def installed_identity(wheel):
+    """Compare installed code to the exact published wheel, without importing project source."""
+    if hashlib.sha256(wheel.read_bytes()).hexdigest() != WHEEL_SHA256:
+        raise ValueError("published wheel identity mismatch")
+    distribution = importlib.metadata.distribution("durable-workflow")
+    if distribution.version != SDK_VERSION or distribution.read_text("direct_url.json") is not None:
+        raise ValueError("require a registry-only installation of the bound SDK")
+    with ZipFile(wheel) as archive:
+        names = sorted(
+            name for name in archive.namelist() if name.startswith("durable_workflow/") and not name.endswith("/")
+        )
+        for name in names:
+            if distribution.locate_file(name).read_bytes() != archive.read(name):
+                raise ValueError("installed SDK file differs from published wheel")
+    if len(names) != 46:
+        raise ValueError("incomplete published SDK inventory")
+    # The imported package must also come from that installation, even when a
+    # caller accidentally runs without Python's documented isolated flag.
+    import durable_workflow
+
+    if Path(durable_workflow.__file__).resolve() != distribution.locate_file("durable_workflow/__init__.py").resolve():
+        raise ValueError("SDK import is outside the verified installation")
+    return {
+        "version": SDK_VERSION,
+        "wheel_sha256": WHEEL_SHA256,
+        "runtime_files_verified": len(names),
+        "python": platform.python_version(),
+        "dependencies": {name: importlib.metadata.version(name) for name in ("httpx", "fastavro")},
+    }
+
+
+@workflow.defn(name=WORKFLOW_TYPE)
+class Commands:
+    def __init__(self):
+        self.signals = 0
+        self.updates = 0
+        self.done = False
+        self.rotate = False
+        self.generation = 0
+
+    @workflow.signal("add")
+    def add(self, value):
+        assert value == PAYLOAD and type(value["number"]) is int and type(value["flag"]) is bool
+        self.signals += 1
+
+    @workflow.signal("finish")
+    def finish(self):
+        self.done = True
+
+    @workflow.signal("continue")
+    def continue_run(self):
+        self.rotate = True
+
+    @workflow.update("value")
+    def value(self, value):
+        self.updates += 1
+        return {"count": self.updates, "payload": value}
+
+    @workflow.query("state")
+    def state(self):
+        return {"signals": self.signals, "updates": self.updates, "generation": self.generation}
+
+    def run(self, ctx, generation=0):
+        self.generation = generation
+        yield ctx.wait_condition(lambda: self.done or self.rotate, key="control", timeout=600)
+        if self.rotate:
+            yield ctx.continue_as_new(generation + 1)
+        return self.state()
+
+
+async def history(client, handle):
+    return [event async for event in client.iter_history(handle.workflow_id, handle.run_id)]
+
+
+async def until(client, handle, event_type):
+    deadline = time.monotonic() + 60
+    while time.monotonic() < deadline:
+        events = await history(client, handle)
+        if any(event["event_type"] == event_type for event in events):
+            return events
+        await asyncio.sleep(0.1)
+    raise TimeoutError(event_type)
+
+
+async def state(client, workflow_id, signals, updates, generation=0):
+    deadline = time.monotonic() + 60
+    expected = {"signals": signals, "updates": updates, "generation": generation}
+    while time.monotonic() < deadline:
+        receipt = await client.query_workflow(workflow_id, "state")
+        assert receipt["workflow_id"] == workflow_id and receipt["query_name"] == "state"
+        assert receipt["reason"] is None
+        if receipt["result"] == expected:
+            return
+        await asyncio.sleep(0.1)
+    raise AssertionError("business state mismatch")
+
+
+async def command(client, kind, handle, key=None, payload=PAYLOAD):
+    if kind == "signal":
+        return await client.signal_workflow(
+            handle.workflow_id, "add", args=[payload], request_id=key, run_id=handle.run_id
+        )
+    return await client.update_workflow(
+        handle.workflow_id, "value", args=[payload], request_id=key, wait_for="accepted"
+    )
+
+
+def route(kind, handle):
+    if kind == "signal":
+        return f"/api/workflows/{handle.workflow_id}/runs/{handle.run_id}/signal/add"
+    return f"/api/workflows/{handle.workflow_id}/update/value"
+
+
+def retained(original, retry):
+    assert retry["request_id"] == original["request_id"]
+    assert retry["retained"] is True and retry["intent_verified"] is True
+    for key in IDENTITIES:
+        if key in original:
+            assert retry.get(key) == original[key], key
+
+
+def sanitized_record(record):
+    return {
+        key: record[key]
+        for key in (
+            "method",
+            "path",
+            "request_body_bytes",
+            "request_body_sha256",
+            "upstream_status",
+            "upstream_complete",
+            "upstream_body_bytes",
+            "upstream_body_sha256",
+            "dropped",
+        )
+    }
+
+
+def loss_records(proxy, path, count):
+    records = [record for record in proxy.records if record["method"] == "POST" and record["path"] == path]
+    assert len(records) == count, (path, len(records), count)
+    assert records[0]["dropped"] is True and 200 <= records[0]["upstream_status"] < 300
+    assert records[0]["upstream_body_bytes"] > 0
+    assert all(record["upstream_complete"] is True for record in records)
+    assert len(records[0]["upstream_body_sha256"]) == 64
+    assert all(record["dropped"] is False and 200 <= record["upstream_status"] < 300 for record in records[1:])
+    return [sanitized_record(record) for record in records]
+
+
+def compare_requests(records, requests):
+    """Independent sender and proxy observations must agree on every request byte."""
+    assert len(records) == len(requests)
+    assert len({item["sha256"] for item in requests}) == 1
+    for record, request in zip(records, requests, strict=True):
+        assert record["request_body_sha256"] == request["sha256"]
+        assert record["request_body_bytes"] == request["bytes"]
+
+
+async def main(args):
+    if not __debug__:
+        raise ValueError("acceptance assertions require Python without optimization")
+    if args.output.exists():
+        raise ValueError("refusing to overwrite an existing evidence file")
+    proxy_bytes = args.proxy_source.read_bytes()
+    if hashlib.sha256(proxy_bytes).hexdigest() != PROXY_SHA256:
+        raise ValueError("accepted proxy source hash mismatch")
+    package = installed_identity(args.wheel)
+    spec = importlib.util.spec_from_file_location("shared_committed_response_loss", args.proxy_source)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    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"]
+    proxy = module.CommittedResponseLoss(args.upstream)
+    thread = threading.Thread(target=proxy.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True)
+    thread.start()
+    report = {
+        "sdk": package,
+        "proxy_commit": PROXY_COMMIT,
+        "proxy_sha256": PROXY_SHA256,
+        "adapter_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "cells": [],
+        "complete": False,
+    }
+    wire = []
+
+    async def observe(request):
+        path = request.url.path
+        if request.method == "POST" and (path.endswith("/signal/add") or path.endswith("/update/value")):
+            body = request.content
+            wire.append(
+                {
+                    "path": path,
+                    "sha256": hashlib.sha256(body).hexdigest(),
+                    "bytes": len(body),
+                    "request_id": json.loads(body)["request_id"],
+                }
+            )
+
+    def save():
+        args.output.write_text(json.dumps(report, indent=2) + "\n")
+
+    queue = "sdk-python-loss-" + uuid4().hex
+    try:
+        async with Client(args.upstream, token=token) as direct:
+            info = await direct.get_cluster_info()
+            if info["version"] != args.server_version:
+                raise ValueError("runtime version differs from the externally verified tuple")
+            report["server_version"] = info["version"]
+            worker = Worker(direct, task_queue=queue, workflows=[Commands], poll_timeout=0.2)
+            running = asyncio.create_task(worker.run())
+            try:
+                for automatic, explicit, continued, kinds in (
+                    (True, False, False, ("signal", "update")),
+                    (True, True, True, ("signal", "update")),
+                    (False, False, False, ("signal",)),
+                    (False, True, False, ("signal",)),
+                    (False, False, False, ("update",)),
+                    (False, True, False, ("update",)),
+                ):
+                    handle = await direct.start_workflow(
+                        workflow_type=WORKFLOW_TYPE, workflow_id=queue + "-" + uuid4().hex, task_queue=queue, input=[0]
+                    )
+                    await until(direct, handle, "ConditionWaitOpened")
+                    originals = {}
+                    cell = {
+                        "automatic_retry": automatic,
+                        "explicit_key": explicit,
+                        "continued": continued,
+                        "commands": {},
+                    }
+                    policy = None if automatic else TransportRetryPolicy(max_attempts=1)
+                    async with Client(proxy.url, token=token, retry_policy=policy) as lossy:
+                        lossy._http.event_hooks["request"].append(observe)
+                        for kind in kinds:
+                            path = route(kind, handle)
+                            key = kind + "-" + uuid4().hex if explicit else None
+                            proxy.arm("POST", path)
+                            diagnostic = None
+                            if automatic:
+                                receipt = await command(lossy, kind, handle, key)
+                                assert receipt["retained"] is True and receipt["intent_verified"] is True
+                            else:
+                                try:
+                                    await command(lossy, kind, handle, key)
+                                except httpx.TransportError as error:
+                                    diagnostic = type(error).__name__
+                                    sent_key = getattr(error, "request_id", None)
+                                    assert isinstance(sent_key, str) and sent_key
+                                    assert sent_key == key if explicit else sent_key.startswith(f"sdk-python-{kind}-")
+                                else:
+                                    raise AssertionError("selected committed response was not lost")
+                                loss_records(proxy, path, 1)
+                                receipt = await command(direct, kind, handle, sent_key)
+                                assert receipt["retained"] is True and receipt["intent_verified"] is True
+                            expected = 2 if automatic else 1
+                            records = loss_records(proxy, path, expected)
+                            requests = [item for item in wire if item["path"] == path]
+                            assert len(requests) == expected
+                            compare_requests(records, requests)
+                            assert all(item["request_id"] == receipt["request_id"] for item in requests)
+                            assert (
+                                receipt["request_id"] == key
+                                if explicit
+                                else receipt["request_id"].startswith(f"sdk-python-{kind}-")
+                            )
+                            assert receipt["command_id"] and type(receipt["command_sequence"]) is int
+                            if kind == "update":
+                                result = await direct.update_outcome(
+                                    handle.workflow_id, receipt["update_id"], wait_timeout_seconds=30
+                                )
+                                assert result["count"] == 1 and result["payload"] == PAYLOAD
+                                assert (
+                                    type(result["payload"]["number"]) is int and type(result["payload"]["flag"]) is bool
+                                )
+                            originals[kind] = receipt
+                            assert receipt.get("resolved_run_id", receipt.get("run_id")) == handle.run_id
+                            # Concurrent retries must all resolve to one retained
+                            # command before any close/continuation control.
+                            retries = await asyncio.gather(
+                                *(command(direct, kind, handle, receipt["request_id"]) for _ in range(4))
+                            )
+                            for retry in retries:
+                                retained(receipt, retry)
+                            try:
+                                await command(direct, kind, handle, receipt["request_id"], {"different": True})
+                            except (SignalFailed, UpdateFailed) as error:
+                                assert error.reason == "request_id_conflict"
+                                assert error.request_id == receipt["request_id"]
+                            else:
+                                raise AssertionError("changed command intent was admitted")
+                            cell["commands"][kind] = {
+                                "wire": requests,
+                                "loss": records,
+                                "transport_exception": diagnostic,
+                                "concurrent_identical_retries": len(retries),
+                                "changed_intent": "request_id_conflict",
+                                "identity": {key: receipt[key] for key in IDENTITIES if key in receipt},
+                            }
+                    await state(direct, handle.workflow_id, int("signal" in kinds), int("update" in kinds))
+                    if continued:
+                        await direct.signal_workflow(handle.workflow_id, "continue", run_id=handle.run_id)
+                        await until(direct, handle, "WorkflowContinuedAsNew")
+                        current = await direct.describe_workflow(handle.workflow_id)
+                        assert current.run_id != handle.run_id
+                        await state(direct, handle.workflow_id, 0, 0, 1)
+                        for kind, original in originals.items():
+                            retained(original, await command(direct, kind, handle, original["request_id"]))
+                        await state(direct, handle.workflow_id, 0, 0, 1)
+                        # The original selected-run identity remains authoritative for recovery.
+                        before = len(await history(direct, handle))
+                        await direct.signal_workflow(current.workflow_id, "finish", run_id=current.run_id)
+                        await until(direct, current, "WorkflowCompleted")
+                    else:
+                        await direct.signal_workflow(handle.workflow_id, "finish", run_id=handle.run_id)
+                        await until(direct, handle, "WorkflowCompleted")
+                        before = len(await history(direct, handle))
+                    for kind, original in originals.items():
+                        retained(original, await command(direct, kind, handle, original["request_id"]))
+                    events = await history(direct, handle)
+                    assert len(events) == before, "retained closed-run reads wrote history"
+                    counts = Counter(event["event_type"] for event in events)
+                    assert counts["SignalReceived"] == int("signal" in kinds) + 1
+                    for event_type in ("UpdateAccepted", "UpdateApplied", "UpdateCompleted"):
+                        assert counts[event_type] == int("update" in kinds), (event_type, counts)
+                    cell["event_counts"] = dict(counts)
+                    cell["business_executions"] = {"signal": int("signal" in kinds), "update": int("update" in kinds)}
+                    report["cells"].append(cell)
+                    save()
+                    print(f"completed response-loss cell {len(report['cells'])}", flush=True)
+            finally:
+                await worker.stop()
+                await running
+        report["business_complete"] = True
+    except Exception as error:
+        report["failure_type"] = type(error).__name__
+        raise
+    finally:
+        proxy.shutdown()
+        proxy.server_close()
+        thread.join(timeout=5)
+        report["proxy_thread_stopped"] = not thread.is_alive()
+        save()
+        assert not thread.is_alive()
+    report["complete"] = True
+    report["finished_at"] = datetime.now(timezone.utc).isoformat()
+    save()
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--upstream", required=True)
+    parser.add_argument("--credential-file", type=Path, required=True)
+    parser.add_argument("--proxy-source", type=Path, required=True)
+    parser.add_argument("--wheel", type=Path, required=True)
+    parser.add_argument("--server-version", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    asyncio.run(main(parse_args()))
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
new file mode 100644
index 0000000..212748d
--- /dev/null
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -0,0 +1,198 @@
+"""Run one disposable installed-Python response-loss cohort, with exact cleanup."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import secrets
+import subprocess
+import time
+import urllib.request
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import uuid4
+
+EXPECTED_SERVER = {
+    "version": "3.9.2",
+    "source": "5614274d2629d03f92932a07dfee0fb10349ddd9",
+    "index": "sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9",
+    "linux_amd64": "sha256:813c9337c44be7d497efd29e1b463546519e488aa796b5ecf9dfc6f4f10d2128",
+    "qualification": "https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177",
+}
+
+
+def image_identity(document):
+    image = document[0]
+    if (
+        image["Architecture"] != "amd64"
+        or image["Os"] != "linux"
+        or image["Config"]["Labels"]["org.opencontainers.image.revision"] != EXPECTED_SERVER["source"]
+        or not any(value.endswith("@" + EXPECTED_SERVER["index"]) for value in image["RepoDigests"])
+    ):
+        raise ValueError("local runtime image differs from the qualified source/index/platform")
+    return {
+        "id": image["Id"],
+        "architecture": image["Architecture"],
+        "os": image["Os"],
+        "revision": image["Config"]["Labels"]["org.opencontainers.image.revision"],
+        "repo_digests": image["RepoDigests"],
+    }
+
+
+def run(args):
+    root = args.output.resolve()
+    root.mkdir(mode=0o700, parents=False, exist_ok=False)
+    config = root / "docker-config"
+    config.mkdir(mode=0o700)
+    (config / "config.json").write_text('{"auths":{}}\n')
+    scratch = root / "tmp"
+    scratch.mkdir(mode=0o700)
+    environment = {"PATH": os.defpath, "TMPDIR": str(scratch), "PYTHONUNBUFFERED": "1"}
+    docker = ["docker", "--config", str(config)]
+
+    def call(*arguments, check=True, timeout=45):
+        return subprocess.run(
+            docker + list(arguments), check=check, capture_output=True, text=True, timeout=timeout, env=environment
+        )
+
+    image = "ghcr.io/durable-workflow/server@" + EXPECTED_SERVER["index"]
+    observed = image_identity(json.loads(call("image", "inspect", image).stdout))
+    name = "sdk-python-216-response-loss-" + uuid4().hex[:12]
+    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\nDB_CONNECTION=sqlite\n"
+        )
+    source = Path(__file__).with_name("response_loss.py")
+    report = {
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "expected_server": EXPECTED_SERVER,
+        "observed_image": observed,
+        "controller_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "adapter_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
+        "complete": False,
+    }
+    volumes = []
+    created = False
+    try:
+        call(
+            "run",
+            "-d",
+            "--rm",
+            "--init",
+            "--read-only",
+            "--name",
+            name,
+            "--cpus",
+            "1",
+            "--memory",
+            "1g",
+            "--pids-limit",
+            "256",
+            "-p",
+            "127.0.0.1::8080",
+            "--env-file",
+            str(credential),
+            image,
+        )
+        created = True
+        inspection = json.loads(call("inspect", name).stdout)[0]
+        volumes = [mount["Name"] for mount in inspection["Mounts"] if mount["Type"] == "volume"]
+        port = call("port", name, "8080/tcp").stdout.strip().removeprefix("127.0.0.1:")
+        if not port.isascii() or not port.isdigit() or not 1 <= int(port) <= 65535:
+            raise ValueError("runtime did not bind an explicit loopback port")
+        upstream = "http://127.0.0.1:" + port
+        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("synthetic runtime readiness")
+        with (root / "consumer.log").open("x") as log:
+            result = subprocess.run(
+                [
+                    str(args.python.absolute()),
+                    "-I",
+                    str(source),
+                    "--upstream",
+                    upstream,
+                    "--credential-file",
+                    str(credential),
+                    "--proxy-source",
+                    str(args.proxy.resolve()),
+                    "--wheel",
+                    str(args.wheel.resolve()),
+                    "--server-version",
+                    EXPECTED_SERVER["version"],
+                    "--output",
+                    str(root / "consumer.json"),
+                ],
+                cwd=root,
+                env=environment,
+                stdout=log,
+                stderr=subprocess.STDOUT,
+                timeout=600,
+            )
+        report["consumer_exit_code"] = result.returncode
+        if result.returncode != 0:
+            raise RuntimeError("installed consumer failed; preserve private log and incomplete evidence")
+        consumer = json.loads((root / "consumer.json").read_text())
+        if not consumer["complete"] or consumer["adapter_sha256"] != report["adapter_sha256"]:
+            raise ValueError("consumer proof is incomplete or source changed")
+    except Exception as error:
+        report["failure_type"] = type(error).__name__
+        raise
+    finally:
+        final = call("inspect", name, check=False)
+        if final.returncode == 0:
+            inspected = json.loads(final.stdout)[0]
+            volumes = [mount["Name"] for mount in inspected["Mounts"] if mount["Type"] == "volume"]
+            report["oom_killed"] = inspected["State"]["OOMKilled"]
+            call("rm", "-f", "-v", name)
+        elif created:
+            report["runtime_disappeared_before_cleanup"] = True
+        credential.unlink(missing_ok=True)
+        checked = call("inspect", name, check=False)
+        absent = checked.returncode == 1 and f"No such object: {name}" in checked.stderr
+        remaining = []
+        for volume in volumes:
+            checked = call("volume", "inspect", volume, check=False)
+            if not (checked.returncode == 1 and f"no such volume: {volume}" in checked.stderr):
+                remaining.append(volume)
+        report["cleanup"] = {
+            "container_absent": absent,
+            "volumes_removed": len(volumes) - len(remaining),
+            "volumes_remaining": len(remaining),
+            "credential_removed": not credential.exists(),
+        }
+        report["finished_at"] = datetime.now(timezone.utc).isoformat()
+        report["complete"] = (
+            report.get("consumer_exit_code") == 0
+            and absent
+            and not remaining
+            and report.get("oom_killed") is False
+            and "failure_type" not in report
+        )
+        (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
+        if not absent or remaining:
+            raise RuntimeError("owned runtime cleanup incomplete")
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--python", type=Path, required=True, help="registry-only virtualenv Python")
+    parser.add_argument("--wheel", type=Path, required=True, help="downloaded bound PyPI wheel")
+    parser.add_argument("--proxy", type=Path, required=True, help="exact accepted shared proxy")
+    parser.add_argument("--output", type=Path, required=True, help="new private disk-backed directory")
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    run(parse_args())
diff --git a/tests/test_command_keys_acceptance.py b/tests/test_command_keys_acceptance.py
new file mode 100644
index 0000000..00c5f79
--- /dev/null
+++ b/tests/test_command_keys_acceptance.py
@@ -0,0 +1,113 @@
+"""Keep incomplete transport evidence from becoming an installed acceptance PASS."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+from zipfile import ZipFile
+
+import pytest
+
+SOURCE = Path(__file__).parents[1] / "scripts/conformance/command_keys/response_loss.py"
+SPEC = importlib.util.spec_from_file_location("command_keys_acceptance", SOURCE)
+assert SPEC is not None and SPEC.loader is not None
+adapter = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(adapter)
+
+
+def record(**changes):
+    return {
+        "method": "POST",
+        "path": "/api/workflows/synthetic/update/value",
+        "request_body_bytes": 100,
+        "request_body_sha256": "a" * 64,
+        "upstream_status": 200,
+        "upstream_complete": True,
+        "upstream_body_bytes": 50,
+        "upstream_body_sha256": "b" * 64,
+        "dropped": True,
+        **changes,
+    }
+
+
+def test_complete_loss_and_retry_agree_with_independent_sender():
+    records = [record(), record(dropped=False)]
+    observed = adapter.loss_records(SimpleNamespace(records=records), records[0]["path"], 2)
+    adapter.compare_requests(observed, [{"sha256": "a" * 64, "bytes": 100}] * 2)
+
+
+@pytest.mark.parametrize(
+    "changes",
+    [
+        {"dropped": False},
+        {"upstream_complete": False},
+        {"upstream_status": 502},
+        {"upstream_body_bytes": 0},
+        {"upstream_body_sha256": ""},
+    ],
+)
+def test_incomplete_or_unplanned_loss_refuses(changes):
+    item = record(**changes)
+    with pytest.raises(AssertionError):
+        adapter.loss_records(SimpleNamespace(records=[item]), item["path"], 1)
+
+
+@pytest.mark.parametrize(
+    "requests",
+    [
+        [],
+        [{"sha256": "a" * 64, "bytes": 99}],
+        [{"sha256": "c" * 64, "bytes": 100}],
+        [{"sha256": "a" * 64, "bytes": 100}, {"sha256": "c" * 64, "bytes": 100}],
+    ],
+)
+def test_truncated_or_changed_request_comparison_refuses(requests):
+    with pytest.raises(AssertionError):
+        adapter.compare_requests([record()], requests)
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        {"request_id": "different"},
+        {"command_id": "different"},
+        {"retained": False},
+        {"intent_verified": False},
+        {"run_id": "different"},
+    ],
+)
+def test_retained_identity_and_intent_are_required(change):
+    original = {"request_id": "synthetic-key", "command_id": "command-1", "run_id": "run-1"}
+    retry = {**original, "retained": True, "intent_verified": True, **change}
+    with pytest.raises(AssertionError):
+        adapter.retained(original, retry)
+
+
+def test_installed_file_substitution_and_source_overlay_refuse(tmp_path, monkeypatch):
+    wheel = tmp_path / "synthetic.whl"
+    root = tmp_path / "installed"
+    names = ["durable_workflow/__init__.py"] + [f"durable_workflow/file{i}.py" for i in range(45)]
+    with ZipFile(wheel, "w") as archive:
+        for name in names:
+            archive.writestr(name, b"# synthetic fixture\n")
+            target = root / name
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_bytes(b"# synthetic fixture\n")
+    monkeypatch.setattr(adapter, "WHEEL_SHA256", hashlib.sha256(wheel.read_bytes()).hexdigest())
+    distribution = SimpleNamespace(
+        version=adapter.SDK_VERSION, read_text=lambda _: None, locate_file=lambda name: root / name
+    )
+    monkeypatch.setattr(adapter.importlib.metadata, "distribution", lambda _: distribution)
+    import durable_workflow
+
+    monkeypatch.setattr(durable_workflow, "__file__", str(root / names[0]))
+    assert adapter.installed_identity(wheel)["runtime_files_verified"] == 46
+    (root / names[-1]).write_bytes(b"# substituted\n")
+    with pytest.raises(ValueError, match="differs"):
+        adapter.installed_identity(wheel)
+    (root / names[-1]).write_bytes(b"# synthetic fixture\n")
+    monkeypatch.setattr(durable_workflow, "__file__", str(tmp_path / "overlay/__init__.py"))
+    with pytest.raises(ValueError, match="outside"):
+        adapter.installed_identity(wheel)

From ace90eb7f4ee52e4d25a20fcf398b4803719be1a Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 16:10:51 +0000
Subject: [PATCH 2/7] Give disposable SQLite a bounded writable path and verify
 cleanup replies

---
 scripts/conformance/command_keys/README.md    |  2 +
 .../command_keys/run_response_loss.py         | 20 +++++++-
 tests/test_command_keys_acceptance.py         | 46 +++++++++++++++++++
 3 files changed, 66 insertions(+), 2 deletions(-)

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index b0387e9..3fa944d 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -41,6 +41,8 @@ python3 -I scripts/conformance/command_keys/run_response_loss.py \
 The output parent must exist. The new private output directory must not exist.
 One runtime is capped at 1 CPU, 1 GiB and 256 PIDs, with a read-only root,
 `--init` and one ephemeral loopback port. The consumer is sequential and bounded
+to ten minutes. Bounded tmpfs mounts supply writable SQLite, storage and temporary
+paths inside the container's memory ceiling. The consumer is sequential and bounded
 to ten minutes. A private empty Docker config and minimal subprocess environment
 exclude host credentials. The controller owns exact-name runtime/anonymous-volume
 cleanup and removes its generated credential even after consumer failure.
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index 212748d..f266cb0 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -41,6 +41,12 @@ def image_identity(document):
     }
 
 
+def missing_object(result, name, *, volume=False):
+    """Distinguish Docker's actual missing-object reply from daemon/tool errors."""
+    expected = f"Error response from daemon: get {name}: no such volume" if volume else f"error: no such object: {name}"
+    return result.returncode == 1 and result.stderr.strip() == expected
+
+
 def run(args):
     root = args.output.resolve()
     root.mkdir(mode=0o700, parents=False, exist_ok=False)
@@ -73,6 +79,7 @@ def call(*arguments, check=True, timeout=45):
         "observed_image": observed,
         "controller_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
         "adapter_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
+        "owned_container": name,
         "complete": False,
     }
     volumes = []
@@ -84,6 +91,12 @@ def call(*arguments, check=True, timeout=45):
             "--rm",
             "--init",
             "--read-only",
+            "--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=64m",
             "--name",
             name,
             "--cpus",
@@ -96,11 +109,14 @@ def call(*arguments, check=True, timeout=45):
             "127.0.0.1::8080",
             "--env-file",
             str(credential),
+            "-e",
+            "DB_DATABASE=/app/database/response-loss.sqlite",
             image,
         )
         created = True
         inspection = json.loads(call("inspect", name).stdout)[0]
         volumes = [mount["Name"] for mount in inspection["Mounts"] if mount["Type"] == "volume"]
+        report["owned_volumes"] = volumes
         port = call("port", name, "8080/tcp").stdout.strip().removeprefix("127.0.0.1:")
         if not port.isascii() or not port.isdigit() or not 1 <= int(port) <= 65535:
             raise ValueError("runtime did not bind an explicit loopback port")
@@ -160,11 +176,11 @@ def call(*arguments, check=True, timeout=45):
             report["runtime_disappeared_before_cleanup"] = True
         credential.unlink(missing_ok=True)
         checked = call("inspect", name, check=False)
-        absent = checked.returncode == 1 and f"No such object: {name}" in checked.stderr
+        absent = missing_object(checked, name)
         remaining = []
         for volume in volumes:
             checked = call("volume", "inspect", volume, check=False)
-            if not (checked.returncode == 1 and f"no such volume: {volume}" in checked.stderr):
+            if not missing_object(checked, volume, volume=True):
                 remaining.append(volume)
         report["cleanup"] = {
             "container_absent": absent,
diff --git a/tests/test_command_keys_acceptance.py b/tests/test_command_keys_acceptance.py
index 00c5f79..aebefb0 100644
--- a/tests/test_command_keys_acceptance.py
+++ b/tests/test_command_keys_acceptance.py
@@ -15,6 +15,12 @@
 assert SPEC is not None and SPEC.loader is not None
 adapter = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(adapter)
+CONTROLLER_SPEC = importlib.util.spec_from_file_location(
+    "command_keys_controller", SOURCE.with_name("run_response_loss.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 record(**changes):
@@ -111,3 +117,43 @@ def test_installed_file_substitution_and_source_overlay_refuse(tmp_path, monkeyp
     monkeypatch.setattr(durable_workflow, "__file__", str(tmp_path / "overlay/__init__.py"))
     with pytest.raises(ValueError, match="outside"):
         adapter.installed_identity(wheel)
+
+
+@pytest.mark.parametrize(
+    "volume,message",
+    [
+        (False, "error: no such object: synthetic-owned"),
+        (True, "Error response from daemon: get synthetic-owned: no such volume"),
+    ],
+)
+def test_cleanup_recognizes_only_missing_owned_objects(volume, message):
+    result = SimpleNamespace(returncode=1, stderr=message + "\n")
+    assert controller.missing_object(result, "synthetic-owned", volume=volume)
+    assert not controller.missing_object(result, "someone-elses-object", volume=volume)
+    result.stderr = "Cannot connect to the Docker daemon"
+    assert not controller.missing_object(result, "synthetic-owned", volume=volume)
+    result.stderr = message
+    result.returncode = 0
+    assert not controller.missing_object(result, "synthetic-owned", volume=volume)
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("Architecture", "arm64"),
+        ("Os", "windows"),
+        ("RepoDigests", []),
+        ("Config", {"Labels": {"org.opencontainers.image.revision": "unqualified-source"}}),
+    ],
+)
+def test_unqualified_local_image_refuses(field, value):
+    image = {
+        "Architecture": "amd64",
+        "Os": "linux",
+        "Id": "synthetic-config-id",
+        "RepoDigests": ["ghcr.io/durable-workflow/server@" + controller.EXPECTED_SERVER["index"]],
+        "Config": {"Labels": {"org.opencontainers.image.revision": controller.EXPECTED_SERVER["source"]}},
+    }
+    assert controller.image_identity([image])["revision"] == controller.EXPECTED_SERVER["source"]
+    with pytest.raises(ValueError, match="differs"):
+        controller.image_identity([{**image, field: value}])

From d4dbdaf259f6041f10bc96fbb6672d52a9ef4d94 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 16:12:42 +0000
Subject: [PATCH 3/7] Exercise maximum Unicode keys through real response loss

---
 scripts/conformance/command_keys/README.md        | 13 +++++++------
 scripts/conformance/command_keys/response_loss.py | 11 ++++++++++-
 tests/test_command_keys_acceptance.py             |  8 ++++++++
 3 files changed, 25 insertions(+), 7 deletions(-)

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 3fa944d..803ff64 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -40,9 +40,9 @@ python3 -I scripts/conformance/command_keys/run_response_loss.py \
 
 The output parent must exist. The new private output directory must not exist.
 One runtime is capped at 1 CPU, 1 GiB and 256 PIDs, with a read-only root,
-`--init` and one ephemeral loopback port. The consumer is sequential and bounded
-to ten minutes. Bounded tmpfs mounts supply writable SQLite, storage and temporary
-paths inside the container's memory ceiling. The consumer is sequential and bounded
+`--init` and one ephemeral loopback port. Bounded tmpfs mounts supply writable
+SQLite, storage and temporary paths inside the container's memory ceiling.
+The consumer is sequential and bounded
 to ten minutes. A private empty Docker config and minimal subprocess environment
 exclude host credentials. The controller owns exact-name runtime/anonymous-volume
 cleanup and removes its generated credential even after consumer failure.
@@ -53,7 +53,8 @@ update routes, and one continuation. The adapter independently compares sender
 and proxy request bytes, requires successful complete upstream consumption,
 checks four concurrent identical retries and a typed changed-intent conflict,
 then verifies one business execution/admission and retained identities after
-closure or continuation. Typed nested values include integers, booleans, bytes
+closure or continuation. Explicit keys use the corrected 255-code-point Unicode
+boundary, exceeding 255 UTF-8 bytes. Typed nested values include integers, booleans, bytes
 and Unicode. Query reads consume the receipt's `result` field.
 
 Keep a failed attempt as NONPASS. `consumer.json` and `run.json` must both report
@@ -65,6 +66,6 @@ Keep diagnostic logs private and temporary; do not commit generated payloads.
 This does not qualify synthetic legacy transforms or physical upgrades. The
 accepted common stopped-copy helper is separately maintained by
 [SDK Rust260](https://github.com/durable-workflow/sdk-rust/pull/260), and Server1210
-owns the real published historical seed/upgrade. Namespace/run fences and maximum
-Unicode key acceptance also remain distinct installed cells. No full shared216
+owns the real published historical seed/upgrade. Namespace/run fences also remain
+distinct installed cells. No full shared216
 PASS follows from this adapter alone.
diff --git a/scripts/conformance/command_keys/response_loss.py b/scripts/conformance/command_keys/response_loss.py
index 84de621..4001c35 100644
--- a/scripts/conformance/command_keys/response_loss.py
+++ b/scripts/conformance/command_keys/response_loss.py
@@ -43,6 +43,13 @@
 IDENTITIES = ("command_id", "command_sequence", "run_id", "resolved_run_id", "update_id", "accepted_at")
 
 
+def explicit_key(kind):
+    prefix = kind + "-" + uuid4().hex
+    # This is the corrected producer's 255-code-point boundary, deliberately
+    # longer than 255 UTF-8 bytes. Caller keys are opaque Unicode strings.
+    return prefix + "日" * (255 - len(prefix))
+
+
 def installed_identity(wheel):
     """Compare installed code to the exact published wheel, without importing project source."""
     if hashlib.sha256(wheel.read_bytes()).hexdigest() != WHEEL_SHA256:
@@ -281,7 +288,7 @@ def save():
                         lossy._http.event_hooks["request"].append(observe)
                         for kind in kinds:
                             path = route(kind, handle)
-                            key = kind + "-" + uuid4().hex if explicit else None
+                            key = explicit_key(kind) if explicit else None
                             proxy.arm("POST", path)
                             diagnostic = None
                             if automatic:
@@ -337,6 +344,8 @@ def save():
                             else:
                                 raise AssertionError("changed command intent was admitted")
                             cell["commands"][kind] = {
+                                "key_characters": len(receipt["request_id"]),
+                                "key_utf8_bytes": len(receipt["request_id"].encode()),
                                 "wire": requests,
                                 "loss": records,
                                 "transport_exception": diagnostic,
diff --git a/tests/test_command_keys_acceptance.py b/tests/test_command_keys_acceptance.py
index aebefb0..3ccf1f8 100644
--- a/tests/test_command_keys_acceptance.py
+++ b/tests/test_command_keys_acceptance.py
@@ -23,6 +23,14 @@
 CONTROLLER_SPEC.loader.exec_module(controller)
 
 
+@pytest.mark.parametrize("kind", ["signal", "update"])
+def test_explicit_key_exercises_unicode_character_not_byte_limit(kind):
+    key = adapter.explicit_key(kind)
+    assert len(key) == 255
+    assert len(key.encode()) > 255
+    assert key != adapter.explicit_key(kind)
+
+
 def record(**changes):
     return {
         "method": "POST",

From 089fe4689f3803b27d2f6ce84a9f855e8d416ed1 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:13:14 +0000
Subject: [PATCH 4/7] Capture reviewed inputs and bind the registry consumer to
 an immutable interpreter

---
 scripts/conformance/command_keys/README.md    | 112 +++---
 .../command_keys/consumer-requirements.txt    |   9 +
 .../command_keys/consumer_bootstrap.py        | 193 ++++++++++
 .../conformance/command_keys/input_guards.py  |  95 +++++
 .../conformance/command_keys/response_loss.py |  33 +-
 .../command_keys/run_response_loss.py         | 332 ++++++++++++------
 tests/test_command_keys_captured_inputs.py    | 166 +++++++++
 7 files changed, 767 insertions(+), 173 deletions(-)
 create mode 100644 scripts/conformance/command_keys/consumer-requirements.txt
 create mode 100644 scripts/conformance/command_keys/consumer_bootstrap.py
 create mode 100644 scripts/conformance/command_keys/input_guards.py
 create mode 100644 tests/test_command_keys_captured_inputs.py

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 803ff64..3959391 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -1,71 +1,77 @@
 # Installed command-response recovery
 
-This adapter exercises the actual published Python Client and Worker for shared
-[command-key acceptance](https://github.com/durable-workflow/.github/issues/216).
-It creates synthetic workflows and a disposable SQLite runtime. It does not
-modify SDK product source or provide a new transport fault implementation.
+This adapter exercises the published Python Client and actual Worker for
+[shared command-key acceptance](https://github.com/durable-workflow/.github/issues/216).
+It creates synthetic workflows and a disposable SQLite runtime. Its transport
+fault comes from the independently reviewed shared loopback fixture.
 
-The source binds the published wheel and accepted shared proxy by SHA-256.
-The controller binds the qualified native image index, source and amd64 platform
-from [Server's independent publication review](https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177).
-Expected platform receipt fields remain separate from observed local image
-identity and runtime version. Stable aliases and final Release selection are
-separate producer gates.
+The controller binds the qualified native image/source/platform from
+[Server's publication receipt](https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177).
+Its interpreter is the published official Python image index
+`sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df`,
+amd64 child `sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4`.
+It starts a fresh private virtualenv in that image and installs all nine registry
+packages from `consumer-requirements.txt`, with exact wheel-only hashes. The SDK
+wheel is the reviewed published artifact. The interpreter/config and entire
+installed environment are fingerprinted before and after the consumer.
 
-Prepare a registry-only virtualenv outside the checkout. Install
-`durable-workflow==2.17.2` from PyPI with wheel-only dependencies and download its
-wheel separately with `pip download --no-deps --only-binary=:all:`. Do not install
-this checkout, an editable package or a local archive. The adapter refuses a
-direct URL installation, unexpected wheel hash, mismatched runtime files or a
-source import overlay. The controller invokes the virtualenv interpreter with
-`-I`; assertions must remain enabled.
-
-Download the frozen shared proxy from
+Download the published SDK wheel identified by `WHEEL_SHA256` in the controller.
+Download the shared proxy from
 [`117118c8`](https://github.com/durable-workflow/.github/blob/117118c83093822469b3824ba9587ff4068ed29f/conformance/command-keys/committed_response_loss.py).
-Its bound SHA-256 is
+Its SHA-256 is
 `bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725`.
-The proxy accepts numeric loopback only and retains bounded request/response
-counts and hashes. It receives only ephemeral synthetic credentials.
+Store both as regular, single-link, owned files in owner-controlled directories.
+Before the experiment, check host/Docker capacity and pull both exact images
+with an empty Docker config. The controller itself does not pull images.
 
-After checking host/Docker capacity, pull the controller's exact image with an
-empty Docker config. The controller itself never pulls an image. Run:
+Run from an exact committed checkout:
 
 ```sh
 python3 -I scripts/conformance/command_keys/run_response_loss.py \
-  --python /path/to/registry-venv/bin/python \
+  --runner-commit FULL_REVIEWED_RUNNER_COMMIT \
   --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \
   --proxy /path/to/verified/committed_response_loss.py \
   --output /path/to/private-disk-scratch/new-run
 ```
 
-The output parent must exist. The new private output directory must not exist.
-One runtime is capped at 1 CPU, 1 GiB and 256 PIDs, with a read-only root,
-`--init` and one ephemeral loopback port. Bounded tmpfs mounts supply writable
-SQLite, storage and temporary paths inside the container's memory ceiling.
-The consumer is sequential and bounded
-to ten minutes. A private empty Docker config and minimal subprocess environment
-exclude host credentials. The controller owns exact-name runtime/anonymous-volume
-cleanup and removes its generated credential even after consumer failure.
+The controller and executable source inputs must equal that commit's Git blobs.
+It opens inputs without following symlinks, checks owner/mode/single-link/file
+bounds and retains descriptors plus captured bytes. The adapter, bootstrap and
+proxy execute from those bytes. Wheel inspection uses the same captured bytes.
+Pathname, inode, content and parent-directory identities are checked again before
+launch and after return. A substituted program cannot certify its own output by
+reporting the original hash. The host does not supply an interpreter or virtualenv.
+
+The new private output directory must not exist; its parent must already exist.
+One runtime and one consumer are each capped at 1 CPU, 1 GiB and 256 PIDs, with
+read-only roots and `--init`. Bounded tmpfs paths remain within each memory cap.
+The runtime publishes one ephemeral loopback port. The consumer uses host
+networking for the numeric loopback-only proxy and has no host mounts. Registry
+installation precedes creation of its synthetic credential file. Its entire
+bootstrap/input bundle travels through stdin; executable files are never reopened
+from host paths. Controller readiness explicitly disables environment proxies.
+The consumer has a ten-minute bound, and cleanup owns both containers, anonymous
+volumes and the generated fixture credential.
 
-Six workflow cells exercise eight genuinely lost upstream success responses:
-automatic and manual recovery with generated and explicit keys, both signal and
-update routes, and one continuation. The adapter independently compares sender
-and proxy request bytes, requires successful complete upstream consumption,
-checks four concurrent identical retries and a typed changed-intent conflict,
-then verifies one business execution/admission and retained identities after
-closure or continuation. Explicit keys use the corrected 255-code-point Unicode
-boundary, exceeding 255 UTF-8 bytes. Typed nested values include integers, booleans, bytes
-and Unicode. Query reads consume the receipt's `result` field.
+Six workflows exercise eight intentionally lost complete upstream successes:
+automatic/manual recovery with generated/explicit keys on both routes, including
+continuation. Sender and proxy request bytes agree, four concurrent identical
+retries retain one admission, and changed intent gets a typed conflict. Business
+execution/admission counts and closed/continued original identities are checked.
+Explicit keys contain 255 Unicode code points and exceed 255 UTF-8 bytes. Nested
+typed values include integers, booleans, bytes and Unicode. Query receipts are
+read through their `result` field.
 
-Keep a failed attempt as NONPASS. `consumer.json` and `run.json` must both report
-`complete: true`, with no OOM, exact package/runtime/source identities and verified
-cleanup, before reporting this cell as passing. Record the public runner commit,
-file hashes, tuple, timestamps and sanitized outcomes in the owning GitHub record.
-Keep diagnostic logs private and temporary; do not commit generated payloads.
+Both `consumer.json` and `run.json` must report complete success, with captured
+input and interpreter rechecks, equal environment inventories, no OOM and exact
+cleanup, before qualification. Record the reviewed runner commit, input hashes,
+tuple, timestamps and sanitized outcomes in GitHub. Keep failed attempts as
+NONPASS and diagnostic captures private and temporary. The earlier host-venv
+observation is retained under its Security provenance HOLD, not reused as the
+corrected runner's qualification.
 
-This does not qualify synthetic legacy transforms or physical upgrades. The
-accepted common stopped-copy helper is separately maintained by
-[SDK Rust260](https://github.com/durable-workflow/sdk-rust/pull/260), and Server1210
-owns the real published historical seed/upgrade. Namespace/run fences also remain
-distinct installed cells. No full shared216
-PASS follows from this adapter alone.
+Synthetic legacy transformations and physical upgrades remain separate cells.
+The accepted common stopped-copy helper is owned by
+[SDK Rust260](https://github.com/durable-workflow/sdk-rust/pull/260); Server1210
+owns the real historical seed. Namespace/run fences remain independently scoped.
+This adapter alone does not complete shared216.
diff --git a/scripts/conformance/command_keys/consumer-requirements.txt b/scripts/conformance/command_keys/consumer-requirements.txt
new file mode 100644
index 0000000..dfd01a9
--- /dev/null
+++ b/scripts/conformance/command_keys/consumer-requirements.txt
@@ -0,0 +1,9 @@
+anyio==4.16.0 --hash=sha256:c66ae634e0b8228396b801741beb3294326d4b6102f7bba4041c0dfb58c29026
+certifi==2026.10.23 --hash=sha256:e9b54de76e530bb6cf0ccd036787103acb58fd42ba930ea8bf948ddbaa2f61d1
+durable-workflow==2.17.2 --hash=sha256:7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc
+fastavro==1.13.1 --hash=sha256:2f56a127d71e45083306d2650efff827cad0f4b0744dd42cb69c631d77943b1d
+h11==0.16.0 --hash=sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86
+httpcore==1.0.9 --hash=sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55
+httpx==0.28.1 --hash=sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad
+idna==3.20 --hash=sha256:ab7ae7122974553370f0bdb919e1a960b2cd1bc1ef0276416d896db81c14582c
+typing_extensions==4.16.0 --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8
diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py
new file mode 100644
index 0000000..0cb0938
--- /dev/null
+++ b/scripts/conformance/command_keys/consumer_bootstrap.py
@@ -0,0 +1,193 @@
+"""Trusted argv bootstrap inside the fixed published Python image.
+
+No host interpreter, virtualenv, source directory or credential directory is
+mounted. The controller sends only already captured inputs and synthetic data.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import json
+import linecache
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+import venv
+from pathlib import Path
+from types import ModuleType, SimpleNamespace
+
+RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT="
+ROLES = {"adapter", "bootstrap", "proxy", "wheel", "requirements"}
+MAX_BUNDLE_BYTES = 8 * 1024 * 1024
+
+
+def decode_inputs(bundle):
+    if set(bundle["sources"]) != ROLES:
+        raise ValueError("captured source roles differ")
+    result = {}
+    for role, entry in bundle["sources"].items():
+        data = base64.b64decode(entry["base64"], validate=True)
+        if hashlib.sha256(data).hexdigest() != entry["sha256"]:
+            raise ValueError("captured bytes differ: " + role)
+        result[role] = data
+    return result
+
+
+def execute_source(data, name):
+    module = ModuleType(name)
+    module.__file__ = "captured-" + name + ".py"
+    sys.modules[name] = module
+    linecache.cache[module.__file__] = (len(data), None, data.decode().splitlines(keepends=True), module.__file__)
+    # The bytes were captured before runtime start. No executable input path
+    # is reopened or imported through a loader.
+    exec(compile(data, module.__file__, "exec"), module.__dict__)
+    return module
+
+
+def inventory(root):
+    entries = []
+    total = 0
+    for path in sorted(root.rglob("*")):
+        relative = str(path.relative_to(root))
+        info = path.lstat()
+        if stat.S_ISDIR(info.st_mode):
+            continue
+        if path.is_symlink():
+            if relative != "lib64" or os.readlink(path) != "lib":
+                raise ValueError("unexpected installed-environment link")
+            entries.append([relative, "symlink", "lib"])
+            continue
+        if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid():
+            raise ValueError("unexpected installed-environment file")
+        total += info.st_size
+        if total > 128 * 1024 * 1024 or len(entries) > 10_000:
+            raise ValueError("installed-environment inventory exceeds bound")
+        entries.append([relative, stat.S_IMODE(info.st_mode), hashlib.sha256(path.read_bytes()).hexdigest()])
+    digest = hashlib.sha256(json.dumps(entries, separators=(",", ":")).encode()).hexdigest()
+    return {"sha256": digest, "files": len(entries), "bytes": total}
+
+
+def prepare(bundle, sources):
+    if "prepared" in bundle:
+        raise ValueError("caller may not supply a prepared interpreter")
+    root = Path(tempfile.mkdtemp(prefix="sdk-python-installed-"))
+    environment = root / "venv"
+    venv.EnvBuilder(with_pip=True, symlinks=False).create(environment)
+    lock = root / "requirements.txt"
+    lock.write_bytes(sources["requirements"])
+    lock.chmod(0o400)
+    python = environment / "bin/python"
+    executable = python.lstat()
+    if not stat.S_ISREG(executable.st_mode) or executable.st_nlink != 1 or executable.st_uid != os.getuid():
+        raise ValueError("fresh interpreter identity differs")
+    binary_sha256 = hashlib.sha256(python.read_bytes()).hexdigest()
+    if binary_sha256 != hashlib.sha256(Path(sys.executable).read_bytes()).hexdigest():
+        raise ValueError("virtualenv interpreter differs from fixed image interpreter")
+    clean_environment = {"PATH": os.defpath, "TMPDIR": str(root), "PIP_CONFIG_FILE": "/dev/null"}
+    subprocess.run(
+        [
+            str(python),
+            "-I",
+            "-B",
+            "-m",
+            "pip",
+            "--isolated",
+            "install",
+            "--no-compile",
+            "--no-cache-dir",
+            "--only-binary=:all:",
+            "--require-hashes",
+            "--disable-pip-version-check",
+            "--progress-bar",
+            "off",
+            "--index-url",
+            "https://pypi.org/simple",
+            "-r",
+            str(lock),
+        ],
+        check=True,
+        env=clean_environment,
+        timeout=180,
+    )
+    before = inventory(environment)
+    bundle["prepared"] = {
+        "root": str(environment),
+        "binary_sha256": binary_sha256,
+        "config_sha256": hashlib.sha256((environment / "pyvenv.cfg").read_bytes()).hexdigest(),
+        "inventory": before,
+    }
+    # Both the child program and its inputs come from the same captured bytes.
+    # The interpreter is created here from the immutable image, never supplied
+    # by a host pathname. Its complete environment is checked around the call.
+    result = subprocess.run(
+        [str(python), "-I", "-B", "-c", sources["bootstrap"].decode(), "--prepared"],
+        input=json.dumps(bundle).encode(),
+        env=clean_environment,
+        timeout=300,
+    )
+    if inventory(environment) != before:
+        raise ValueError("installed environment changed")
+    raise SystemExit(result.returncode)
+
+
+def main():
+    raw = sys.stdin.buffer.read(MAX_BUNDLE_BYTES + 1)
+    if len(raw) > MAX_BUNDLE_BYTES:
+        raise ValueError("captured bundle exceeds bound")
+    bundle = json.loads(raw)
+    sources = decode_inputs(bundle)
+    if sys.argv[1:] != ["--prepared"]:
+        prepare(bundle, sources)
+    prepared = bundle["prepared"]
+    environment = Path(prepared["root"])
+    if Path(sys.prefix) != environment or Path(sys.executable).parent != environment / "bin":
+        raise ValueError("consumer did not start in its private registry environment")
+    if (
+        hashlib.sha256(Path(sys.executable).read_bytes()).hexdigest() != prepared["binary_sha256"]
+        or hashlib.sha256((environment / "pyvenv.cfg").read_bytes()).hexdigest() != prepared["config_sha256"]
+    ):
+        raise ValueError("prepared interpreter/config identity changed")
+    before = inventory(environment)
+    if before != prepared["inventory"]:
+        raise ValueError("prepared environment changed before consumer execution")
+    root = environment.parent
+    credential = root / "fixture.env"
+    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")
+    try:
+        result = asyncio.run(
+            adapter.main(
+                SimpleNamespace(
+                    upstream=bundle["upstream"],
+                    credential_file=credential,
+                    proxy_data=sources["proxy"],
+                    wheel_data=sources["wheel"],
+                    adapter_sha256=bundle["sources"]["adapter"]["sha256"],
+                    server_version=bundle["server_version"],
+                    output=root / "consumer.json",
+                )
+            )
+        )
+    finally:
+        credential.unlink(missing_ok=True)
+    after = inventory(environment)
+    if before != after:
+        raise ValueError("installed environment changed during consumer execution")
+    result["toolchain"] = {
+        "interpreter_sha256": prepared["binary_sha256"],
+        "venv_config_sha256": prepared["config_sha256"],
+        "environment_inventory": before,
+        "post_inventory_equal": True,
+        "requirements_sha256": bundle["sources"]["requirements"]["sha256"],
+    }
+    print(RESULT_PREFIX + json.dumps(result), flush=True)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/conformance/command_keys/input_guards.py b/scripts/conformance/command_keys/input_guards.py
new file mode 100644
index 0000000..c60887d
--- /dev/null
+++ b/scripts/conformance/command_keys/input_guards.py
@@ -0,0 +1,95 @@
+"""Capture reviewed local inputs once and detect pathname/inode substitutions."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import stat
+from pathlib import Path
+
+
+def identity(info):
+    return (
+        info.st_dev,
+        info.st_ino,
+        info.st_mode,
+        info.st_uid,
+        info.st_gid,
+        info.st_nlink,
+        info.st_size,
+        info.st_mtime_ns,
+        info.st_ctime_ns,
+    )
+
+
+def directory_identity(info):
+    return (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid)
+
+
+class CapturedInput:
+    """Retain no-follow descriptors and immutable bytes for one bounded input."""
+
+    def __init__(self, path, role, *, maximum, expected=None):
+        self.path = Path(path).absolute()
+        self.role = role
+        self.handles = []
+        self.parents = []
+        if ".." in self.path.parts:
+            raise ValueError("input path must be canonical")
+        try:
+            parent = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+            self.handles.append(parent)
+            for part in self.path.parts[1:-1]:
+                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("input parent is not owner-controlled")
+                self.parents.append((parent, part, descriptor, directory_identity(info)))
+                parent = descriptor
+            self.parent = parent
+            self.descriptor = os.open(self.path.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent)
+            self.handles.append(self.descriptor)
+            info = os.fstat(self.descriptor)
+            if (
+                not stat.S_ISREG(info.st_mode)
+                or info.st_uid != os.getuid()
+                or info.st_nlink != 1
+                or info.st_mode & 0o022
+                or not 0 < info.st_size <= maximum
+            ):
+                raise ValueError("input must be an owned bounded single-link regular file")
+            self.original = identity(info)
+            self.data = os.pread(self.descriptor, maximum + 1, 0)
+            self.sha256 = hashlib.sha256(self.data).hexdigest()
+            if len(self.data) != info.st_size or (expected is not None and self.sha256 != expected):
+                raise ValueError("captured input identity differs")
+            self.recheck()
+        except (OSError, ValueError):
+            self.close()
+            raise ValueError("reviewed input refused: " + role) from None
+
+    def recheck(self):
+        for parent, name, descriptor, original in self.parents:
+            if (
+                directory_identity(os.fstat(descriptor)) != original
+                or directory_identity(os.stat(name, dir_fd=parent, follow_symlinks=False)) != original
+            ):
+                raise ValueError("input parent was replaced: " + self.role)
+        current = os.stat(self.path.name, dir_fd=self.parent, follow_symlinks=False)
+        if identity(current) != self.original or identity(os.fstat(self.descriptor)) != self.original:
+            raise ValueError("held input was replaced or changed: " + self.role)
+        if hashlib.sha256(os.pread(self.descriptor, self.original[6] + 1, 0)).hexdigest() != self.sha256:
+            raise ValueError("held input bytes changed: " + self.role)
+
+    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/scripts/conformance/command_keys/response_loss.py b/scripts/conformance/command_keys/response_loss.py
index 4001c35..d2447bd 100644
--- a/scripts/conformance/command_keys/response_loss.py
+++ b/scripts/conformance/command_keys/response_loss.py
@@ -6,11 +6,11 @@
 
 from __future__ import annotations
 
-import argparse
 import asyncio
 import hashlib
 import importlib.metadata
 import importlib.util
+import io
 import json
 import platform
 import stat
@@ -19,6 +19,7 @@
 from collections import Counter
 from datetime import datetime, timezone
 from pathlib import Path
+from types import ModuleType
 from uuid import uuid4
 from zipfile import ZipFile
 
@@ -52,12 +53,13 @@ def explicit_key(kind):
 
 def installed_identity(wheel):
     """Compare installed code to the exact published wheel, without importing project source."""
-    if hashlib.sha256(wheel.read_bytes()).hexdigest() != WHEEL_SHA256:
+    wheel_bytes = wheel if isinstance(wheel, bytes) else wheel.read_bytes()
+    if hashlib.sha256(wheel_bytes).hexdigest() != WHEEL_SHA256:
         raise ValueError("published wheel identity mismatch")
     distribution = importlib.metadata.distribution("durable-workflow")
     if distribution.version != SDK_VERSION or distribution.read_text("direct_url.json") is not None:
         raise ValueError("require a registry-only installation of the bound SDK")
-    with ZipFile(wheel) as archive:
+    with ZipFile(io.BytesIO(wheel_bytes)) as archive:
         names = sorted(
             name for name in archive.namelist() if name.startswith("durable_workflow/") and not name.endswith("/")
         )
@@ -213,13 +215,12 @@ async def main(args):
         raise ValueError("acceptance assertions require Python without optimization")
     if args.output.exists():
         raise ValueError("refusing to overwrite an existing evidence file")
-    proxy_bytes = args.proxy_source.read_bytes()
+    proxy_bytes = args.proxy_data
     if hashlib.sha256(proxy_bytes).hexdigest() != PROXY_SHA256:
         raise ValueError("accepted proxy source hash mismatch")
-    package = installed_identity(args.wheel)
-    spec = importlib.util.spec_from_file_location("shared_committed_response_loss", args.proxy_source)
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
+    package = installed_identity(args.wheel_data)
+    module = ModuleType("shared_committed_response_loss")
+    exec(compile(proxy_bytes, "captured-committed-response-loss.py", "exec"), module.__dict__)
     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")
@@ -231,7 +232,7 @@ async def main(args):
         "sdk": package,
         "proxy_commit": PROXY_COMMIT,
         "proxy_sha256": PROXY_SHA256,
-        "adapter_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "adapter_sha256": args.adapter_sha256,
         "started_at": datetime.now(timezone.utc).isoformat(),
         "cells": [],
         "complete": False,
@@ -401,18 +402,8 @@ def save():
     report["complete"] = True
     report["finished_at"] = datetime.now(timezone.utc).isoformat()
     save()
-
-
-def parse_args():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--upstream", required=True)
-    parser.add_argument("--credential-file", type=Path, required=True)
-    parser.add_argument("--proxy-source", type=Path, required=True)
-    parser.add_argument("--wheel", type=Path, required=True)
-    parser.add_argument("--server-version", required=True)
-    parser.add_argument("--output", type=Path, required=True)
-    return parser.parse_args()
+    return report
 
 
 if __name__ == "__main__":
-    asyncio.run(main(parse_args()))
+    raise SystemExit("Use the bound run_response_loss.py controller; standalone executable paths are refused")
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index f266cb0..0ba42c5 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -1,17 +1,21 @@
-"""Run one disposable installed-Python response-loss cohort, with exact cleanup."""
+"""Run the published SDK in a bound interpreter with captured executable inputs."""
 
 from __future__ import annotations
 
 import argparse
+import base64
 import hashlib
 import json
 import os
+import re
 import secrets
 import subprocess
 import time
 import urllib.request
+from contextlib import ExitStack
 from datetime import datetime, timezone
 from pathlib import Path
+from types import ModuleType
 from uuid import uuid4
 
 EXPECTED_SERVER = {
@@ -21,6 +25,12 @@
     "linux_amd64": "sha256:813c9337c44be7d497efd29e1b463546519e488aa796b5ecf9dfc6f4f10d2128",
     "qualification": "https://github.com/durable-workflow/server/pull/1209#issuecomment-6858140177",
 }
+PYTHON_INDEX = "sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df"
+PYTHON_AMD64 = "sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4"
+PYTHON_VERSION = "3.12.16"
+WHEEL_SHA256 = "7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc"
+PROXY_SHA256 = "bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725"
+RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT="
 
 
 def image_identity(document):
@@ -41,21 +51,96 @@ def image_identity(document):
     }
 
 
+def interpreter_identity(document):
+    image = document[0]
+    versions = [value for value in image["Config"]["Env"] if value.startswith("PYTHON_VERSION=")]
+    if (
+        image["Architecture"] != "amd64"
+        or image["Os"] != "linux"
+        or not any(value.endswith("@" + PYTHON_INDEX) for value in image["RepoDigests"])
+        or versions != ["PYTHON_VERSION=" + PYTHON_VERSION]
+    ):
+        raise ValueError("require the fixed published Python interpreter image")
+    return {
+        "id": image["Id"],
+        "index": PYTHON_INDEX,
+        "expected_amd64": PYTHON_AMD64,
+        "version": PYTHON_VERSION,
+        "architecture": "amd64",
+    }
+
+
 def missing_object(result, name, *, volume=False):
-    """Distinguish Docker's actual missing-object reply from daemon/tool errors."""
     expected = f"Error response from daemon: get {name}: no such volume" if volume else f"error: no such object: {name}"
     return result.returncode == 1 and result.stderr.strip() == expected
 
 
-def run(args):
-    root = args.output.resolve()
+def git_bytes(root, commit, relative):
+    return subprocess.check_output(
+        ["git", "-C", str(root), "show", commit + ":" + relative], env={"PATH": os.defpath}, timeout=10
+    )
+
+
+def captured_inputs(args, stack):
+    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]
+    head = subprocess.check_output(
+        ["git", "-C", str(root), "rev-parse", "HEAD"], text=True, env={"PATH": os.defpath}, timeout=10
+    ).strip()
+    if head != args.runner_commit:
+        raise ValueError("checkout differs from the declared reviewed runner commit")
+    prefix = "scripts/conformance/command_keys/"
+    guard_bytes = git_bytes(root, head, prefix + "input_guards.py")
+    guards = ModuleType("reviewed_input_guards")
+    # 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 {
+        "controller": "run_response_loss.py",
+        "guard": "input_guards.py",
+        "adapter": "response_loss.py",
+        "bootstrap": "consumer_bootstrap.py",
+        "requirements": "consumer-requirements.txt",
+    }.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)
+    )
+    return inputs
+
+
+def consumer_result(stdout, inputs):
+    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")
+    result = json.loads(lines[0])
+    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")
+    return result
+
+
+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')
-    scratch = root / "tmp"
-    scratch.mkdir(mode=0o700)
-    environment = {"PATH": os.defpath, "TMPDIR": str(scratch), "PYTHONUNBUFFERED": "1"}
+    environment = {"PATH": os.defpath, "TMPDIR": str(root)}
     docker = ["docker", "--config", str(config)]
 
     def call(*arguments, check=True, timeout=45):
@@ -64,47 +149,47 @@ def call(*arguments, check=True, timeout=45):
         )
 
     image = "ghcr.io/durable-workflow/server@" + EXPECTED_SERVER["index"]
+    python_image = "docker.io/library/python@" + PYTHON_INDEX
     observed = image_identity(json.loads(call("image", "inspect", image).stdout))
-    name = "sdk-python-216-response-loss-" + uuid4().hex[:12]
+    interpreter = interpreter_identity(json.loads(call("image", "inspect", python_image).stdout))
+    suffix = uuid4().hex[: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=" + secrets.token_hex(24) + "\nDW_RUNTIME_CREDENTIALS_ENABLED=false\nDB_CONNECTION=sqlite\n"
-        )
-    source = Path(__file__).with_name("response_loss.py")
+        stream.write("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,
         "expected_server": EXPECTED_SERVER,
         "observed_image": observed,
-        "controller_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
-        "adapter_sha256": hashlib.sha256(source.read_bytes()).hexdigest(),
-        "owned_container": name,
+        "interpreter_image": interpreter,
+        "inputs": {role: {"sha256": value.sha256, "bytes": len(value.data)} for role, value in inputs.items()},
         "complete": False,
     }
-    volumes = []
-    created = False
     try:
+        for value in inputs.values():
+            value.recheck()
         call(
             "run",
             "-d",
-            "--rm",
             "--init",
             "--read-only",
-            "--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=64m",
             "--name",
-            name,
+            names[0],
             "--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=64m",
             "-p",
             "127.0.0.1::8080",
             "--env-file",
@@ -113,100 +198,149 @@ def call(*arguments, check=True, timeout=45):
             "DB_DATABASE=/app/database/response-loss.sqlite",
             image,
         )
-        created = True
-        inspection = json.loads(call("inspect", name).stdout)[0]
-        volumes = [mount["Name"] for mount in inspection["Mounts"] if mount["Type"] == "volume"]
-        report["owned_volumes"] = volumes
-        port = call("port", name, "8080/tcp").stdout.strip().removeprefix("127.0.0.1:")
-        if not port.isascii() or not port.isdigit() or not 1 <= int(port) <= 65535:
-            raise ValueError("runtime did not bind an explicit loopback port")
-        upstream = "http://127.0.0.1:" + port
+        address = call("port", names[0], "8080/tcp").stdout.strip()
+        host, port = address.split(":")
+        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("synthetic runtime readiness")
-        with (root / "consumer.log").open("x") as log:
-            result = subprocess.run(
-                [
-                    str(args.python.absolute()),
-                    "-I",
-                    str(source),
-                    "--upstream",
-                    upstream,
-                    "--credential-file",
-                    str(credential),
-                    "--proxy-source",
-                    str(args.proxy.resolve()),
-                    "--wheel",
-                    str(args.wheel.resolve()),
-                    "--server-version",
-                    EXPECTED_SERVER["version"],
-                    "--output",
-                    str(root / "consumer.json"),
-                ],
-                cwd=root,
-                env=environment,
-                stdout=log,
-                stderr=subprocess.STDOUT,
-                timeout=600,
-            )
+        bundle = {
+            "sources": {
+                role: {"sha256": value.sha256, "base64": base64.b64encode(value.data).decode()}
+                for role, value in inputs.items()
+                if role in {"adapter", "bootstrap", "requirements", "proxy", "wheel"}
+            },
+            "upstream": upstream,
+            "token": token,
+            "server_version": EXPECTED_SERVER["version"],
+        }
+        for value in inputs.values():
+            value.recheck()
+        # 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.
+        result = subprocess.run(
+            docker
+            + [
+                "run",
+                "--init",
+                "--read-only",
+                "--name",
+                names[1],
+                "--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=512m",
+                "-i",
+                "--entrypoint",
+                "/usr/local/bin/python3.12",
+                python_image,
+                "-I",
+                "-B",
+                "-c",
+                inputs["bootstrap"].data.decode(),
+            ],
+            input=json.dumps(bundle),
+            capture_output=True,
+            text=True,
+            env=environment,
+            timeout=600,
+        )
+        (root / "consumer.log").write_text(result.stdout + result.stderr)
         report["consumer_exit_code"] = result.returncode
+        for value in inputs.values():
+            value.recheck()
+        report["input_identities_rechecked"] = True
+        if interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) != interpreter:
+            raise ValueError("interpreter image identity changed")
+        report["interpreter_identity_rechecked"] = True
         if result.returncode != 0:
-            raise RuntimeError("installed consumer failed; preserve private log and incomplete evidence")
-        consumer = json.loads((root / "consumer.json").read_text())
-        if not consumer["complete"] or consumer["adapter_sha256"] != report["adapter_sha256"]:
-            raise ValueError("consumer proof is incomplete or source changed")
+            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")
+        report["consumer_complete"] = True
     except Exception as error:
         report["failure_type"] = type(error).__name__
         raise
     finally:
-        final = call("inspect", name, check=False)
-        if final.returncode == 0:
-            inspected = json.loads(final.stdout)[0]
-            volumes = [mount["Name"] for mount in inspected["Mounts"] if mount["Type"] == "volume"]
-            report["oom_killed"] = inspected["State"]["OOMKilled"]
-            call("rm", "-f", "-v", name)
-        elif created:
-            report["runtime_disappeared_before_cleanup"] = True
-        credential.unlink(missing_ok=True)
-        checked = call("inspect", name, check=False)
-        absent = missing_object(checked, name)
-        remaining = []
-        for volume in volumes:
-            checked = call("volume", "inspect", volume, check=False)
-            if not missing_object(checked, volume, volume=True):
-                remaining.append(volume)
-        report["cleanup"] = {
-            "container_absent": absent,
-            "volumes_removed": len(volumes) - len(remaining),
-            "volumes_remaining": len(remaining),
-            "credential_removed": not credential.exists(),
-        }
-        report["finished_at"] = datetime.now(timezone.utc).isoformat()
-        report["complete"] = (
-            report.get("consumer_exit_code") == 0
-            and absent
-            and not remaining
-            and report.get("oom_killed") is False
-            and "failure_type" not in report
-        )
-        (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
-        if not absent or remaining:
+        clean = []
+        oom = []
+        try:
+            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"]
+                    oom.append(state["State"]["OOMKilled"])
+                    call("rm", "-f", "-v", name)
+                absent = missing_object(call("inspect", name, check=False), name)
+                absent_volumes = all(
+                    missing_object(call("volume", "inspect", volume, check=False), volume, volume=True)
+                    for volume in volumes
+                )
+                clean.append(absent and absent_volumes)
+        except Exception as error:
+            report["cleanup_failure_type"] = type(error).__name__
+        finally:
+            credential.unlink(missing_ok=True)
+            report["cleanup"] = {
+                "containers_and_volumes_absent": len(clean) == 2 and all(clean),
+                "credential_removed": not credential.exists(),
+                "oom_killed": any(oom),
+            }
+            report["finished_at"] = datetime.now(timezone.utc).isoformat()
+            report["complete"] = (
+                report.get("consumer_complete") is True
+                and len(clean) == 2
+                and all(clean)
+                and len(oom) == 2
+                and not any(oom)
+                and "failure_type" not in report
+                and "cleanup_failure_type" not in report
+            )
+            (root / "run.json").write_text(json.dumps(report, indent=2) + "\n")
+        if not report["cleanup"]["containers_and_volumes_absent"]:
             raise RuntimeError("owned runtime cleanup incomplete")
+        if not report["complete"] and "failure_type" not in report:
+            raise RuntimeError("installed cohort evidence is incomplete")
+
+
+def run(args):
+    with ExitStack() as stack:
+        inputs = captured_inputs(args, stack)
+        run_bound(args, inputs)
 
 
 def parse_args():
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--python", type=Path, required=True, help="registry-only virtualenv Python")
-    parser.add_argument("--wheel", type=Path, required=True, help="downloaded bound PyPI wheel")
-    parser.add_argument("--proxy", type=Path, required=True, help="exact accepted shared proxy")
-    parser.add_argument("--output", type=Path, required=True, help="new private disk-backed directory")
+    parser.add_argument("--runner-commit", required=True)
+    parser.add_argument("--wheel", type=Path, required=True)
+    parser.add_argument("--proxy", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
     return parser.parse_args()
 
 
diff --git a/tests/test_command_keys_captured_inputs.py b/tests/test_command_keys_captured_inputs.py
new file mode 100644
index 0000000..435352e
--- /dev/null
+++ b/tests/test_command_keys_captured_inputs.py
@@ -0,0 +1,166 @@
+"""Substituted files cannot become executable authority or certify a result."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import importlib.util
+import json
+import runpy
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[1] / "scripts/conformance/command_keys"
+
+
+def module(name, path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec is not None and spec.loader is not None
+    result = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(result)
+    return result
+
+
+guards = module("capture_controls", ROOT / "input_guards.py")
+bootstrap = module("bootstrap_controls", ROOT / "consumer_bootstrap.py")
+controller = module("controller_controls", ROOT / "run_response_loss.py")
+ROLES = ["adapter", "proxy", "wheel", "bootstrap", "requirements"]
+
+
+@pytest.mark.parametrize("role", ROLES)
+@pytest.mark.parametrize("substitution", ["symlink", "rename-identical", "rewrite", "hardlink"])
+def test_all_input_roles_reject_symlinks_hardlinks_and_path_or_inode_changes(tmp_path, role, substitution):
+    path = tmp_path / "input"
+    path.write_bytes(b"original input")
+    digest = hashlib.sha256(path.read_bytes()).hexdigest()
+    if substitution == "symlink":
+        alias = tmp_path / "alias"
+        alias.symlink_to(path)
+        with pytest.raises(ValueError):
+            guards.CapturedInput(alias, role, maximum=1024, expected=digest)
+    elif substitution == "hardlink":
+        import os
+
+        os.link(path, tmp_path / "second-link")
+        with pytest.raises(ValueError):
+            guards.CapturedInput(path, role, maximum=1024, expected=digest)
+    else:
+        with guards.CapturedInput(path, role, maximum=1024, expected=digest) as captured:
+            if substitution == "rename-identical":
+                path.rename(tmp_path / "previous")
+                path.write_bytes(b"original input")
+            else:
+                path.write_bytes(b"substituted input")
+            assert captured.data == b"original input"
+            with pytest.raises(ValueError):
+                captured.recheck()
+
+
+def test_parent_directory_rename_refuses_even_identical_replacement(tmp_path):
+    parent = tmp_path / "owned"
+    parent.mkdir(mode=0o700)
+    path = parent / "input"
+    path.write_bytes(b"original")
+    with guards.CapturedInput(path, "adapter", maximum=100) as captured:
+        parent.rename(tmp_path / "renamed")
+        parent.mkdir(mode=0o700)
+        path.write_bytes(b"original")
+        with pytest.raises(ValueError, match="parent was replaced"):
+            captured.recheck()
+
+
+def test_substituted_adapter_cannot_self_certify_its_report(tmp_path):
+    path = tmp_path / "adapter.py"
+    path.write_text("def attest(): return {'complete': False}\n")
+    with guards.CapturedInput(path, "adapter", maximum=4096) as captured:
+        forged = {"complete": True, "adapter_sha256": captured.sha256}
+        path.rename(tmp_path / "original.py")
+        path.write_text("def attest(): return " + repr(forged) + "\n")
+        # The predecessor's hash-then-path-load and self-reported comparison
+        # accepts this synthetic forgery. The corrected execution sees only
+        # the captured original and its post-call identity check refuses.
+        predecessor = runpy.run_path(str(path))["attest"]()
+        assert predecessor["complete"] and predecessor["adapter_sha256"] == captured.sha256
+        executed = bootstrap.execute_source(captured.data, "captured_adapter_control")
+        assert executed.attest()["complete"] is False
+        with pytest.raises(ValueError):
+            captured.recheck()
+
+
+@pytest.mark.parametrize("role", ["adapter", "proxy", "wheel", "bootstrap", "requirements"])
+def test_bundle_byte_substitution_refuses_before_execution(role):
+    sources = {
+        key: {"base64": base64.b64encode(b"original").decode(), "sha256": hashlib.sha256(b"original").hexdigest()}
+        for key in bootstrap.ROLES
+    }
+    sources[role]["base64"] = base64.b64encode(b"substituted").decode()
+    with pytest.raises(ValueError, match="bytes differ"):
+        bootstrap.decode_inputs({"sources": sources})
+
+
+def test_no_host_interpreter_path_can_be_supplied(monkeypatch, capsys):
+    monkeypatch.setattr(
+        "sys.argv",
+        [
+            "run_response_loss.py",
+            "--runner-commit",
+            "a" * 40,
+            "--wheel",
+            "wheel",
+            "--proxy",
+            "proxy",
+            "--output",
+            "new-output",
+            "--python",
+            "/arbitrary/program",
+        ],
+    )
+    with pytest.raises(SystemExit):
+        controller.parse_args()
+    assert "unrecognized arguments: --python /arbitrary/program" in capsys.readouterr().err
+
+
+@pytest.mark.parametrize(
+    "changes",
+    [
+        {"Architecture": "arm64"},
+        {"RepoDigests": ["python@sha256:" + "0" * 64]},
+        {"Config": {"Env": ["PYTHON_VERSION=unqualified"]}},
+    ],
+)
+def test_unqualified_interpreter_image_refuses(changes):
+    image = {
+        "Architecture": "amd64",
+        "Os": "linux",
+        "Id": "qualified-image",
+        "RepoDigests": ["python@" + controller.PYTHON_INDEX],
+        "Config": {"Env": ["PYTHON_VERSION=" + controller.PYTHON_VERSION]},
+    }
+    assert controller.interpreter_identity([image])["version"] == controller.PYTHON_VERSION
+    with pytest.raises(ValueError, match="fixed published Python"):
+        controller.interpreter_identity([{**image, **changes}])
+
+
+def test_inventory_catches_installed_dependency_or_interpreter_mutation(tmp_path):
+    (tmp_path / "bin").mkdir()
+    (tmp_path / "bin/python").write_bytes(b"trusted interpreter fixture")
+    before = bootstrap.inventory(tmp_path)
+    (tmp_path / "bin/python").write_bytes(b"substituted interpreter fixture")
+    assert bootstrap.inventory(tmp_path) != before
+
+
+def test_requirement_lock_covers_only_exact_wheel_hashes():
+    lines = (ROOT / "consumer-requirements.txt").read_text().splitlines()
+    assert len(lines) == 9
+    assert all("==" in line and " --hash=sha256:" in line for line in lines)
+    assert controller.WHEEL_SHA256 in "\n".join(lines)
+    # This JSON framing round trip retains bytes rather than an executable path.
+    original = b"compiled from captured bytes\n"
+    bundle = {
+        "sources": {
+            role: {"base64": base64.b64encode(original).decode(), "sha256": hashlib.sha256(original).hexdigest()}
+            for role in bootstrap.ROLES
+        }
+    }
+    assert bootstrap.decode_inputs(json.loads(json.dumps(bundle)))["adapter"] == original

From 674763e97af7ea8db4df9efa4a25ba64295761e3 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:18:45 +0000
Subject: [PATCH 5/7] Keep scratch noexec and bind virtualenv links to the
 immutable image

---
 .../command_keys/consumer_bootstrap.py        | 37 ++++++++++++++++---
 tests/test_command_keys_captured_inputs.py    | 15 ++++++++
 2 files changed, 46 insertions(+), 6 deletions(-)

diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py
index 0cb0938..0a7b09c 100644
--- a/scripts/conformance/command_keys/consumer_bootstrap.py
+++ b/scripts/conformance/command_keys/consumer_bootstrap.py
@@ -23,6 +23,15 @@
 RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT="
 ROLES = {"adapter", "bootstrap", "proxy", "wheel", "requirements"}
 MAX_BUNDLE_BYTES = 8 * 1024 * 1024
+IMAGE_INTERPRETER = Path("/usr/local/bin/python3.12")
+
+
+def approved_environment_link(relative, target):
+    return (
+        (relative == "lib64" and target == "lib")
+        or (relative in {"bin/python", "bin/python3"} and target == "python3.12")
+        or (relative == "bin/python3.12" and target == str(IMAGE_INTERPRETER))
+    )
 
 
 def decode_inputs(bundle):
@@ -57,9 +66,10 @@ def inventory(root):
         if stat.S_ISDIR(info.st_mode):
             continue
         if path.is_symlink():
-            if relative != "lib64" or os.readlink(path) != "lib":
+            target = os.readlink(path)
+            if not approved_environment_link(relative, target):
                 raise ValueError("unexpected installed-environment link")
-            entries.append([relative, "symlink", "lib"])
+            entries.append([relative, "symlink", target])
             continue
         if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid():
             raise ValueError("unexpected installed-environment file")
@@ -76,16 +86,31 @@ def prepare(bundle, sources):
         raise ValueError("caller may not supply a prepared interpreter")
     root = Path(tempfile.mkdtemp(prefix="sdk-python-installed-"))
     environment = root / "venv"
-    venv.EnvBuilder(with_pip=True, symlinks=False).create(environment)
+    # Scratch remains noexec. These fresh internal links may target only the
+    # immutable interpreter in the fixed image, never a host/caller program.
+    venv.EnvBuilder(with_pip=True, symlinks=True).create(environment)
     lock = root / "requirements.txt"
     lock.write_bytes(sources["requirements"])
     lock.chmod(0o400)
     python = environment / "bin/python"
     executable = python.lstat()
-    if not stat.S_ISREG(executable.st_mode) or executable.st_nlink != 1 or executable.st_uid != os.getuid():
+    if (
+        not stat.S_ISLNK(executable.st_mode)
+        or executable.st_nlink != 1
+        or executable.st_uid != os.getuid()
+        or python.resolve() != IMAGE_INTERPRETER
+    ):
         raise ValueError("fresh interpreter identity differs")
+    image_binary = IMAGE_INTERPRETER.lstat()
+    if (
+        not stat.S_ISREG(image_binary.st_mode)
+        or image_binary.st_uid != 0
+        or image_binary.st_nlink != 1
+        or image_binary.st_mode & 0o022
+    ):
+        raise ValueError("immutable image interpreter shape differs")
     binary_sha256 = hashlib.sha256(python.read_bytes()).hexdigest()
-    if binary_sha256 != hashlib.sha256(Path(sys.executable).read_bytes()).hexdigest():
+    if binary_sha256 != hashlib.sha256(IMAGE_INTERPRETER.read_bytes()).hexdigest():
         raise ValueError("virtualenv interpreter differs from fixed image interpreter")
     clean_environment = {"PATH": os.defpath, "TMPDIR": str(root), "PIP_CONFIG_FILE": "/dev/null"}
     subprocess.run(
@@ -121,7 +146,7 @@ def prepare(bundle, sources):
         "inventory": before,
     }
     # Both the child program and its inputs come from the same captured bytes.
-    # The interpreter is created here from the immutable image, never supplied
+    # The interpreter links here to the immutable image, never supplied
     # by a host pathname. Its complete environment is checked around the call.
     result = subprocess.run(
         [str(python), "-I", "-B", "-c", sources["bootstrap"].decode(), "--prepared"],
diff --git a/tests/test_command_keys_captured_inputs.py b/tests/test_command_keys_captured_inputs.py
index 435352e..56cb567 100644
--- a/tests/test_command_keys_captured_inputs.py
+++ b/tests/test_command_keys_captured_inputs.py
@@ -150,6 +150,21 @@ def test_inventory_catches_installed_dependency_or_interpreter_mutation(tmp_path
     assert bootstrap.inventory(tmp_path) != before
 
 
+@pytest.mark.parametrize(
+    "relative,target",
+    [
+        ("bin/python", "python3.12"),
+        ("bin/python3", "python3.12"),
+        ("bin/python3.12", "/usr/local/bin/python3.12"),
+        ("lib64", "lib"),
+    ],
+)
+def test_fresh_venv_links_are_closed_to_the_immutable_image(relative, target):
+    assert bootstrap.approved_environment_link(relative, target)
+    assert not bootstrap.approved_environment_link(relative, "/host/arbitrary-program")
+    assert not bootstrap.approved_environment_link("lib/python3.12/site-packages/replaced", target)
+
+
 def test_requirement_lock_covers_only_exact_wheel_hashes():
     lines = (ROOT / "consumer-requirements.txt").read_text().splitlines()
     assert len(lines) == 9

From c1e8a0973fa8af336ed95d11df3ca277529fd1b7 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:24:02 +0000
Subject: [PATCH 6/7] Separate scratch from the private hash-locked native
 wheel environment

---
 scripts/conformance/command_keys/README.md             | 3 +++
 scripts/conformance/command_keys/consumer_bootstrap.py | 5 +++--
 scripts/conformance/command_keys/run_response_loss.py  | 4 +++-
 3 files changed, 9 insertions(+), 3 deletions(-)

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 3959391..276f906 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -45,6 +45,9 @@ reporting the original hash. The host does not supply an interpreter or virtuale
 The new private output directory must not exist; its parent must already exist.
 One runtime and one consumer are each capped at 1 CPU, 1 GiB and 256 PIDs, with
 read-only roots and `--init`. Bounded tmpfs paths remain within each memory cap.
+The consumer's ordinary scratch is noexec. Its separate private wheel environment
+permits native-extension mapping for the hash-locked codec wheel; it is not a
+host mount and accepts no unverified executable input.
 The runtime publishes one ephemeral loopback port. The consumer uses host
 networking for the numeric loopback-only proxy and has no host mounts. Registry
 installation precedes creation of its synthetic credential file. Its entire
diff --git a/scripts/conformance/command_keys/consumer_bootstrap.py b/scripts/conformance/command_keys/consumer_bootstrap.py
index 0a7b09c..21067db 100644
--- a/scripts/conformance/command_keys/consumer_bootstrap.py
+++ b/scripts/conformance/command_keys/consumer_bootstrap.py
@@ -85,7 +85,7 @@ def prepare(bundle, sources):
     if "prepared" in bundle:
         raise ValueError("caller may not supply a prepared interpreter")
     root = Path(tempfile.mkdtemp(prefix="sdk-python-installed-"))
-    environment = root / "venv"
+    environment = Path(tempfile.mkdtemp(prefix="environment-", dir="/opt/sdk-python")) / "venv"
     # Scratch remains noexec. These fresh internal links may target only the
     # immutable interpreter in the fixed image, never a host/caller program.
     venv.EnvBuilder(with_pip=True, symlinks=True).create(environment)
@@ -141,6 +141,7 @@ def prepare(bundle, sources):
     before = inventory(environment)
     bundle["prepared"] = {
         "root": str(environment),
+        "scratch": str(root),
         "binary_sha256": binary_sha256,
         "config_sha256": hashlib.sha256((environment / "pyvenv.cfg").read_bytes()).hexdigest(),
         "inventory": before,
@@ -179,7 +180,7 @@ def main():
     before = inventory(environment)
     if before != prepared["inventory"]:
         raise ValueError("prepared environment changed before consumer execution")
-    root = environment.parent
+    root = Path(prepared["scratch"])
     credential = root / "fixture.env"
     descriptor = os.open(credential, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
     with os.fdopen(descriptor, "w") as stream:
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index 0ba42c5..f6683ee 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -253,7 +253,9 @@ def call(*arguments, check=True, timeout=45):
                 "--user",
                 f"{os.getuid()}:{os.getgid()}",
                 "--tmpfs",
-                "/tmp:mode=1777,size=512m",
+                "/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",

From 3b207a0618f2315d012161282367441fa3324f1f Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:50:24 +0000
Subject: [PATCH 7/7] Bind fixture lifecycle and cleanup to validated container
 IDs

---
 scripts/conformance/command_keys/README.md    |   3 +
 .../command_keys/run_response_loss.py         | 174 +++++++++++++-----
 .../test_command_keys_container_ownership.py  | 108 +++++++++++
 3 files changed, 234 insertions(+), 51 deletions(-)
 create mode 100644 tests/test_command_keys_container_ownership.py

diff --git a/scripts/conformance/command_keys/README.md b/scripts/conformance/command_keys/README.md
index 276f906..84655be 100644
--- a/scripts/conformance/command_keys/README.md
+++ b/scripts/conformance/command_keys/README.md
@@ -55,6 +55,9 @@ bootstrap/input bundle travels through stdin; executable files are never reopene
 from host paths. Controller readiness explicitly disables environment proxies.
 The consumer has a ten-minute bound, and cleanup owns both containers, anonymous
 volumes and the generated fixture credential.
+Creation returns full container IDs bound to per-run ownership labels. Start,
+inspection, port lookup, removal and absence verification use those IDs. A reused
+human name grants no cleanup authority over its replacement or volumes.
 
 Six workflows exercise eight intentionally lost complete upstream successes:
 automatic/manual recovery with generated/explicit keys on both routes, including
diff --git a/scripts/conformance/command_keys/run_response_loss.py b/scripts/conformance/command_keys/run_response_loss.py
index f6683ee..da57ce6 100644
--- a/scripts/conformance/command_keys/run_response_loss.py
+++ b/scripts/conformance/command_keys/run_response_loss.py
@@ -31,6 +31,8 @@
 WHEEL_SHA256 = "7dc77d3882fa5cbddbeaee6a4cc63b05285e280523c8ea40dc31e03cc12a40bc"
 PROXY_SHA256 = "bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725"
 RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT="
+OWNER_LABEL = "dev.durable-workflow.fixture-owner"
+RUN_LABEL = "dev.durable-workflow.fixture-run"
 
 
 def image_identity(document):
@@ -75,6 +77,47 @@ def missing_object(result, name, *, volume=False):
     return result.returncode == 1 and result.stderr.strip() == expected
 
 
+def created_id(stdout):
+    value = stdout.strip()
+    if re.fullmatch(r"[0-9a-f]{64}", value) is None:
+        raise ValueError("Docker create did not return one full immutable container ID")
+    return value
+
+
+def inspect_owned(call, container_id, nonce):
+    created_id(container_id)
+    response = call("inspect", container_id, check=False)
+    if missing_object(response, container_id):
+        return None
+    if response.returncode != 0:
+        raise RuntimeError("owned container state is unverified")
+    records = json.loads(response.stdout)
+    if len(records) != 1:
+        raise ValueError("owned container inspection is ambiguous")
+    record = records[0]
+    labels = record.get("Config", {}).get("Labels") or {}
+    if record["Id"] != container_id or labels.get(OWNER_LABEL) != "sdk-python" or labels.get(RUN_LABEL) != nonce:
+        raise ValueError("container ID/ownership identity differs; preserve it")
+    return record
+
+
+def cleanup_owned(call, container_id, nonce):
+    record = inspect_owned(call, container_id, nonce)
+    volumes = []
+    oom = None
+    if record is not None:
+        volumes = [mount["Name"] for mount in record["Mounts"] if mount["Type"] == "volume"]
+        oom = record["State"]["OOMKilled"]
+        call("rm", "-f", "-v", container_id)
+    # Never inspect, remove or recover by a reusable name. If the owned ID
+    # vanished, any same-name replacement and its volumes remain untouched.
+    absent = inspect_owned(call, container_id, nonce) is None
+    volumes_absent = all(
+        missing_object(call("volume", "inspect", volume, check=False), volume, volume=True) for volume in volumes
+    )
+    return {"id": container_id, "absent": absent, "volumes_absent": volumes_absent, "oom_killed": oom}
+
+
 def git_bytes(root, commit, relative):
     return subprocess.check_output(
         ["git", "-C", str(root), "show", commit + ":" + relative], env={"PATH": os.defpath}, timeout=10
@@ -152,7 +195,8 @@ def call(*arguments, check=True, timeout=45):
     python_image = "docker.io/library/python@" + PYTHON_INDEX
     observed = image_identity(json.loads(call("image", "inspect", image).stdout))
     interpreter = interpreter_identity(json.loads(call("image", "inspect", python_image).stdout))
-    suffix = uuid4().hex[:12]
+    nonce = uuid4().hex
+    suffix = nonce[:12]
     names = ["sdk-python-216-response-loss-" + suffix, "sdk-python-216-consumer-" + suffix]
     token = secrets.token_hex(24)
     credential = root / "fixture.env"
@@ -166,39 +210,49 @@ def call(*arguments, check=True, timeout=45):
         "observed_image": observed,
         "interpreter_image": interpreter,
         "inputs": {role: {"sha256": value.sha256, "bytes": len(value.data)} for role, value in inputs.items()},
+        "owned_containers": {},
+        "ownership_nonce": nonce,
         "complete": False,
     }
     try:
         for value in inputs.values():
             value.recheck()
-        call(
-            "run",
-            "-d",
-            "--init",
-            "--read-only",
-            "--name",
-            names[0],
-            "--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=64m",
-            "-p",
-            "127.0.0.1::8080",
-            "--env-file",
-            str(credential),
-            "-e",
-            "DB_DATABASE=/app/database/response-loss.sqlite",
-            image,
+        runtime_id = created_id(
+            call(
+                "create",
+                "--init",
+                "--read-only",
+                "--name",
+                names[0],
+                "--label",
+                OWNER_LABEL + "=sdk-python",
+                "--label",
+                RUN_LABEL + "=" + nonce,
+                "--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=64m",
+                "-p",
+                "127.0.0.1::8080",
+                "--env-file",
+                str(credential),
+                "-e",
+                "DB_DATABASE=/app/database/response-loss.sqlite",
+                image,
+            ).stdout
         )
-        address = call("port", names[0], "8080/tcp").stdout.strip()
+        report["owned_containers"]["runtime"] = {"id": runtime_id, "requested_name": names[0]}
+        inspect_owned(call, runtime_id, nonce)
+        call("start", runtime_id)
+        address = call("port", runtime_id, "8080/tcp").stdout.strip()
         host, port = address.split(":")
         if host != "127.0.0.1" or not port.isdigit() or not 1 <= int(port) <= 65535:
             raise ValueError("runtime did not bind numeric loopback")
@@ -230,14 +284,17 @@ 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.
-        result = subprocess.run(
-            docker
-            + [
-                "run",
+        consumer_id = created_id(
+            call(
+                "create",
                 "--init",
                 "--read-only",
                 "--name",
                 names[1],
+                "--label",
+                OWNER_LABEL + "=sdk-python",
+                "--label",
+                RUN_LABEL + "=" + nonce,
                 "--network",
                 "host",
                 "--cpus",
@@ -264,7 +321,14 @@ def call(*arguments, check=True, timeout=45):
                 "-B",
                 "-c",
                 inputs["bootstrap"].data.decode(),
-            ],
+            ).stdout
+        )
+        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]}
+        inspect_owned(call, consumer_id, nonce)
+        result = subprocess.run(
+            docker + ["start", "-a", "-i", consumer_id],
             input=json.dumps(bundle),
             capture_output=True,
             text=True,
@@ -272,14 +336,18 @@ def call(*arguments, check=True, timeout=45):
             timeout=600,
         )
         (root / "consumer.log").write_text(result.stdout + result.stderr)
-        report["consumer_exit_code"] = result.returncode
+        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")
+        report["consumer_exit_code"] = consumer_state["State"]["ExitCode"]
+        report["docker_attach_exit_code"] = result.returncode
         for value in inputs.values():
             value.recheck()
         report["input_identities_rechecked"] = True
         if interpreter_identity(json.loads(call("image", "inspect", python_image).stdout)) != interpreter:
             raise ValueError("interpreter image identity changed")
         report["interpreter_identity_rechecked"] = True
-        if result.returncode != 0:
+        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")
@@ -291,20 +359,18 @@ def call(*arguments, check=True, timeout=45):
         clean = []
         oom = []
         try:
-            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"]
-                    oom.append(state["State"]["OOMKilled"])
-                    call("rm", "-f", "-v", name)
-                absent = missing_object(call("inspect", name, check=False), name)
-                absent_volumes = all(
-                    missing_object(call("volume", "inspect", volume, check=False), volume, volume=True)
-                    for volume in volumes
-                )
-                clean.append(absent and absent_volumes)
+            report["cleanup_by_id"] = {}
+            for role, owned in report["owned_containers"].items():
+                try:
+                    outcome = cleanup_owned(call, owned["id"], nonce)
+                except Exception as error:
+                    report["cleanup_by_id"][role] = {"id": owned["id"], "failure_type": type(error).__name__}
+                    clean.append(False)
+                    oom.append(None)
+                    continue
+                report["cleanup_by_id"][role] = outcome
+                oom.append(outcome["oom_killed"])
+                clean.append(outcome["absent"] and outcome["volumes_absent"])
         except Exception as error:
             report["cleanup_failure_type"] = type(error).__name__
         finally:
@@ -312,7 +378,13 @@ def call(*arguments, check=True, timeout=45):
             report["cleanup"] = {
                 "containers_and_volumes_absent": len(clean) == 2 and all(clean),
                 "credential_removed": not credential.exists(),
-                "oom_killed": any(oom),
+                "oom_killed": (
+                    True
+                    if any(value is True for value in oom)
+                    else False
+                    if len(oom) == 2 and all(value is False for value in oom)
+                    else None
+                ),
             }
             report["finished_at"] = datetime.now(timezone.utc).isoformat()
             report["complete"] = (
@@ -320,7 +392,7 @@ def call(*arguments, check=True, timeout=45):
                 and len(clean) == 2
                 and all(clean)
                 and len(oom) == 2
-                and not any(oom)
+                and all(value is False for value in oom)
                 and "failure_type" not in report
                 and "cleanup_failure_type" not in report
             )
diff --git a/tests/test_command_keys_container_ownership.py b/tests/test_command_keys_container_ownership.py
new file mode 100644
index 0000000..4ab1cb7
--- /dev/null
+++ b/tests/test_command_keys_container_ownership.py
@@ -0,0 +1,108 @@
+"""A reused human name grants no authority over another fixture's resources."""
+
+from __future__ import annotations
+
+import copy
+import importlib.util
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+PATH = Path(__file__).parents[1] / "scripts/conformance/command_keys/run_response_loss.py"
+SPEC = importlib.util.spec_from_file_location("id_bound_runner", PATH)
+assert SPEC is not None and SPEC.loader is not None
+controller = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(controller)
+OWN_ID, FOREIGN_ID = "a" * 64, "b" * 64
+NONCE = "owned-run"
+
+
+def document(identity, name, owner="sdk-python", nonce=NONCE):
+    return {
+        "Id": identity,
+        "Name": "/" + name,
+        "Config": {"Labels": {controller.OWNER_LABEL: owner, controller.RUN_LABEL: nonce}},
+        "State": {"OOMKilled": False},
+        "Mounts": [{"Type": "volume", "Name": identity + "-data"}],
+    }
+
+
+class DockerModel:
+    def __init__(self, own_present=False):
+        self.objects = {FOREIGN_ID: document(FOREIGN_ID, "reused-name", owner="foreign", nonce="foreign-run")}
+        if own_present:
+            self.objects[OWN_ID] = document(OWN_ID, "original-renamed")
+        self.names = {value["Name"].removeprefix("/"): key for key, value in self.objects.items()}
+        self.volumes = {value["Mounts"][0]["Name"] for value in self.objects.values()}
+        self.calls = []
+
+    def __call__(self, *args, **_options):
+        self.calls.append(args)
+        if args[0] == "inspect":
+            identity = self.names.get(args[1], args[1])
+            if identity in self.objects:
+                return SimpleNamespace(returncode=0, stdout=json.dumps([self.objects[identity]]), stderr="")
+            return SimpleNamespace(returncode=1, stdout="[]", stderr="error: no such object: " + args[1])
+        if args[0] == "rm":
+            identity = self.names.get(args[-1], args[-1])
+            value = self.objects.pop(identity)
+            self.volumes.remove(value["Mounts"][0]["Name"])
+            self.names.pop(value["Name"].removeprefix("/"))
+            return SimpleNamespace(returncode=0, stdout=identity, stderr="")
+        if args[:2] == ("volume", "inspect"):
+            if args[2] in self.volumes:
+                return SimpleNamespace(returncode=0, stdout="[]", stderr="")
+            return SimpleNamespace(
+                returncode=1, stdout="[]", stderr="Error response from daemon: get " + args[2] + ": no such volume"
+            )
+        raise AssertionError("unexpected Docker operation")
+
+
+@pytest.mark.parametrize("own_present", [False, True])
+def test_foreign_same_name_and_its_volume_survive_id_bound_cleanup(own_present):
+    docker = DockerModel(own_present)
+    foreign = copy.deepcopy(docker.objects[FOREIGN_ID])
+    result = controller.cleanup_owned(docker, OWN_ID, NONCE)
+    assert result["absent"] and result["volumes_absent"]
+    assert docker.objects[FOREIGN_ID] == foreign
+    assert FOREIGN_ID + "-data" in docker.volumes
+    assert all(call[-1] != "reused-name" for call in docker.calls)
+    assert [call for call in docker.calls if call[0] == "rm"] == ([("rm", "-f", "-v", OWN_ID)] if own_present else [])
+
+
+def test_predecessor_name_cleanup_deletes_the_foreign_replacement():
+    predecessor = DockerModel()
+    # The former controller resolved and removed by reusable name. The fake
+    # engine demonstrates that exact boundary without touching a real resource.
+    state = json.loads(predecessor("inspect", "reused-name").stdout)[0]
+    assert state["Id"] == FOREIGN_ID
+    predecessor("rm", "-f", "-v", "reused-name")
+    assert FOREIGN_ID not in predecessor.objects and FOREIGN_ID + "-data" not in predecessor.volumes
+    corrected = DockerModel()
+    controller.cleanup_owned(corrected, OWN_ID, NONCE)
+    assert FOREIGN_ID in corrected.objects and FOREIGN_ID + "-data" in corrected.volumes
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda value: value.update(Id=FOREIGN_ID),
+        lambda value: value["Config"]["Labels"].update({controller.OWNER_LABEL: "foreign"}),
+        lambda value: value["Config"]["Labels"].update({controller.RUN_LABEL: "different-run"}),
+    ],
+)
+def test_mismatched_id_or_ownership_refuses_without_mutating_anything(mutation):
+    docker = DockerModel(own_present=True)
+    mutation(docker.objects[OWN_ID])
+    with pytest.raises(ValueError, match="ownership identity differs"):
+        controller.cleanup_owned(docker, OWN_ID, NONCE)
+    assert not any(call[0] == "rm" for call in docker.calls)
+
+
+@pytest.mark.parametrize("value", ["reused-name", "a" * 12, "A" * 64, "a" * 64 + "\nforeign", ""])
+def test_only_a_full_created_id_can_authorize_operations(value):
+    with pytest.raises(ValueError, match="full immutable container ID"):
+        controller.created_id(value)
+    assert controller.created_id(OWN_ID + "\n") == OWN_ID