Qualify installed Python physical-upgrade readbacks

scripts/conformance/command_keys/README.md

@@ -81,3 +81,69 @@ 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.
+
+## Physical upgrade consumer
+
+The producer-owned [frozen handoff](https://github.com/durable-workflow/server/pull/1210#issuecomment-6860138558)
+is accepted at contract source `a09f156399416216d2f097c257a2164f83c7eda7`.
+The actual original seed execution remains `09a4ccda3da11853606af031a8970bbde4b2d6ec`,
+driver SHA-256 `01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b`.
+These are different authorities and must remain distinct in every receipt.
+
+`run_physical_readback.py` consumes that stopped backup/JSON plus the accepted
+common snapshot helper and hash-bound tagged SQL step. It captures every input
+with the same Git/owner/no-follow/held-identity boundary as the response-loss
+runner. Only a new private copy is upgraded. Start/inspect/wait/log/remove/absence
+checks use captured full creation IDs and ownership labels. Upgrade and runtime
+are sequential; the runtime and registry consumer are the only concurrent pair.
+The fixed image/bootstrap/hash-locked registry environment is shared with the
+reviewed response-loss runner. No host interpreter or executable input pathname
+is accepted.
+
+```sh
+python3 -I scripts/conformance/command_keys/run_physical_readback.py \
+  --runner-commit FULL_REVIEWED_RUNNER_COMMIT \
+  --wheel /path/to/downloaded/durable_workflow-2.17.2-py3-none-any.whl \
+  --baseline /path/to/producer/stopped-backup.sqlite \
+  --seed /path/to/producer/seed-receipts.json \
+  --snapshot-helper /path/to/accepted/sqlite_legacy.py \
+  --step-sql /path/to/tagged/0013_command_intents.sql \
+  --output /path/to/private-disk-scratch/new-physical-run
+```
+
+The controller binds the frozen backup/packet/helper/step hashes. It checks the
+native marker transition and validates every business row/schema object, allowing
+only the exact new empty intent catalog and source-defined marker/journal changes.
+Every SQLx journal field is checked: version, description, tagged cumulative SQL
+checksum, success, zero execution time and the actual upgrade timestamp window.
+The installed Client's readbacks are bracketed by complete typed logical snapshots.
+That interval ends before shutdown; subsequent repair work is not relabeled as
+part of the zero-write claim. Original inputs are rechecked and preserved.
+
+For each original keyed update in open/completed/continued state, four concurrent
+changed-input retries must return the original command/run/update/acceptance
+identities with explicit `intent_verified: false`. Changed names must conflict;
+GET-only outcome reads distinguish pending from completed-null. The Python update
+helper uses its unscoped public route, so no new explicit-run update API is claimed.
+Historical NULL-key signals receive no fabricated retry key. Their compatibility
+control preserves their original history and business identities only.
+
+A producer upgrade receipt alone does not qualify this SDK. The exact consumer
+source, installed proof, complete snapshots, immutable input/toolchain rechecks,
+all three captured-ID cleanup records and independent review remain required.
+
+The host workspace also retains no-follow descriptors for every ancestor,
+private directory and created file. Config/credential bytes are immutable;
+the copied SQLite file may change contents but must keep its owned single-link
+inode. Docker configuration and credential reads use held
+`/proc/<controller-pid>/fd` paths. The canonical database bind has a preflight
+inside the immutable published image: it verifies the mounted directory/file
+device, inode, owner, mode and single-link identities before executing the
+native entrypoint. Host snapshots and lifecycle calls check workspace
+identities before and after use. Cleanup uses an explicit empty kernel FD
+configuration directory, so workspace substitution cannot block known-ID cleanup. Evidence is created exclusively relative to the
+held directory, with the creation descriptor retained through verification.
+A substituted parent, copied database, configuration, credential or report refuses
+qualification. Known-container cleanup stays bound to captured IDs and preserves
+replacement paths. Workspace substitution controls must pass before installed
+response-loss or physical results qualify.

scripts/conformance/command_keys/consumer_bootstrap.py

@@ -22,6 +22,7 @@
 
 RESULT_PREFIX = "SDK_PYTHON_CONSUMER_RESULT="
 ROLES = {"adapter", "bootstrap", "proxy", "wheel", "requirements"}
+PHYSICAL_ROLES = {"adapter", "bootstrap", "verifier", "seed", "wheel", "requirements"}
 MAX_BUNDLE_BYTES = 8 * 1024 * 1024
 IMAGE_INTERPRETER = Path("/usr/local/bin/python3.12")
 
@@ -35,7 +36,9 @@ def approved_environment_link(relative, target):
 
 
 def decode_inputs(bundle):
-    if set(bundle["sources"]) != ROLES:
+    scenario = bundle.get("scenario", "response_loss")
+    roles = ROLES if scenario == "response_loss" else PHYSICAL_ROLES if scenario == "physical_readback" else None
+    if roles is None or set(bundle["sources"]) != roles:
         raise ValueError("captured source roles differ")
     result = {}
     for role, entry in bundle["sources"].items():
@@ -185,14 +188,23 @@ def main():
     descriptor = os.open(credential, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
     with os.fdopen(descriptor, "w") as stream:
         stream.write("DW_AUTH_TOKEN=" + bundle["token"] + "\n")
-    adapter = execute_source(sources["adapter"], "installed_command_keys")
+    scenario = bundle.get("scenario", "response_loss")
+    verifier = execute_source(sources.get("verifier", sources["adapter"]), "installed_package_verifier")
+    package = verifier.installed_identity(sources["wheel"])
+    adapter = (
+        verifier if scenario == "response_loss" else execute_source(sources["adapter"], "installed_physical_readback")
+    )
     try:
         result = asyncio.run(
             adapter.main(
                 SimpleNamespace(
                     upstream=bundle["upstream"],
                     credential_file=credential,
-                    proxy_data=sources["proxy"],
+                    proxy_data=sources.get("proxy"),
+                    seed_data=sources.get("seed"),
+                    seed_sha256=bundle["sources"].get("seed", {}).get("sha256"),
+                    package_identity=package,
+                    verifier_sha256=bundle["sources"].get("verifier", bundle["sources"]["adapter"])["sha256"],
                     wheel_data=sources["wheel"],
                     adapter_sha256=bundle["sources"]["adapter"]["sha256"],
                     server_version=bundle["server_version"],

scripts/conformance/command_keys/physical_readback.py

@@ -0,0 +1,161 @@
+"""Installed SDK readbacks for the producer-owned physical seed, after upgrade.
+
+The outer runner owns the verified seed/database, exact image, upgrade, complete
+zero-write snapshots and cleanup. This consumer neither seeds nor upgrades data.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import stat
+from datetime import datetime, timezone
+from urllib.parse import urlsplit
+
+from durable_workflow import Client
+from durable_workflow.errors import UpdateFailed
+
+# Physical seed authority includes the producer's unkeyed-signal clarification.
+# It is distinct from the older manifest bound to the synthetic-copy helper.
+MANIFEST_SHA256 = "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f"
+
+
+def seed_cases(document):
+    if (
+        document.get("schema") != "durable-workflow.command-key-physical-seed.v1"
+        or document.get("manifest_sha256") != MANIFEST_SHA256
+    ):
+        raise ValueError("require the producer-owned physical seed contract")
+    cases = document["cases"]
+    expected = {(kind, state) for kind in ("signal", "update") for state in ("open", "completed", "continued")}
+    if len(cases) != len(expected) or {(case["kind"], case["state"]) for case in cases} != expected:
+        raise ValueError("physical seed must contain each original state/type exactly once")
+    for case in cases:
+        receipt = case["receipt"]
+        fields = ("command_id", "run_id") + (("update_id", "accepted_at") if case["kind"] == "update" else ())
+        if any(not isinstance(receipt.get(field), str) or not receipt[field] for field in fields):
+            raise ValueError("original receipt has incomplete durable identities")
+        if receipt["workflow_id"] != case["workflow_id"]:
+            raise ValueError("original receipt names a different workflow")
+        if case["kind"] == "update" and not case["original_body"].get("request_id"):
+            raise ValueError("original keyed update identity is missing")
+        if case["history"]["next_page_token"] is not None:
+            raise ValueError("physical seed history was not fully captured")
+    return cases
+
+
+def legacy_receipt(original, received):
+    if received.get("retained") is not True or received.get("intent_verified") is not False:
+        raise ValueError("legacy readback must be retained and explicitly unverified")
+    for field in ("command_id", "run_id", "update_id", "accepted_at"):
+        if received.get(field) != original[field]:
+            raise ValueError("legacy readback changed original " + field)
+
+
+async def main(args):
+    if not __debug__:
+        raise ValueError("acceptance assertions require Python without optimization")
+    if args.output.exists():
+        raise ValueError("refusing to overwrite existing evidence")
+    upstream = urlsplit(args.upstream)
+    if (
+        upstream.scheme != "http"
+        or upstream.hostname != "127.0.0.1"
+        or not upstream.port
+        or args.upstream != f"http://127.0.0.1:{upstream.port}"
+    ):
+        raise ValueError("synthetic runtime must be canonical numeric loopback")
+    raw_seed = args.seed_data
+    if hashlib.sha256(raw_seed).hexdigest() != args.seed_sha256:
+        raise ValueError("producer seed receipt hash mismatch")
+    cases = seed_cases(json.loads(raw_seed))
+    package = args.package_identity
+    mode = args.credential_file.lstat().st_mode
+    if not stat.S_ISREG(mode) or mode & 0o077:
+        raise ValueError("fixture credential file must be owner-only")
+    token = dict(line.split("=", 1) for line in args.credential_file.read_text().splitlines())["DW_AUTH_TOKEN"]
+    report = {
+        "started_at": datetime.now(timezone.utc).isoformat(),
+        "sdk": package,
+        "seed_sha256": args.seed_sha256,
+        "verifier_sha256": args.verifier_sha256,
+        "adapter_sha256": args.adapter_sha256,
+        "scope": "installed SDK physical-upgrade readbacks only",
+        "cells": [],
+        "complete": False,
+    }
+    try:
+        async with Client(args.upstream, token=token) as client:
+            info = await client.get_cluster_info()
+            if info["version"] != args.server_version:
+                raise ValueError("runtime differs from the outer verified upgrade tuple")
+            report["server_version"] = info["version"]
+            for case in cases:
+                original = case["receipt"]
+                workflow_id = case["workflow_id"]
+                observed = [event async for event in client.iter_history(workflow_id, original["run_id"])]
+                if observed != case["history"]["events"]:
+                    raise ValueError("upgrade history differs from the original producer capture")
+                cell = {
+                    "kind": case["kind"],
+                    "state": case["state"],
+                    "command_id": original["command_id"],
+                    "run_id": original["run_id"],
+                    "original_history_preserved": True,
+                }
+                if case["kind"] == "signal":
+                    # Published historical signals have NULL stored keys. Never
+                    # manufacture keyed recovery from an unrelated current key.
+                    cell["scope"] = "unkeyed historical signal compatibility only"
+                else:
+                    key = case["original_body"]["request_id"]
+                    name = original["update_name"]
+                    receipts = await asyncio.gather(
+                        *(
+                            client.update_workflow(workflow_id, name, args=[999], request_id=key, wait_for="accepted")
+                            for _ in range(4)
+                        )
+                    )
+                    for receipt in receipts:
+                        legacy_receipt(original, receipt)
+                        assert receipt["request_id"] == key
+                        if case["state"] != "open":
+                            assert receipt["update_status"] == "completed" and receipt["result"] is None
+                    try:
+                        await client.update_workflow(workflow_id, "other", args=[999], request_id=key)
+                    except UpdateFailed as error:
+                        assert error.reason == "request_id_conflict"
+                        assert error.details["retained_command_id"] == original["command_id"]
+                    else:
+                        raise AssertionError("changed legacy command name was admitted")
+                    outcome = await client.update_outcome_receipt(workflow_id, original["update_id"])
+                    assert outcome["update_id"] == original["update_id"]
+                    if case["state"] == "open":
+                        assert outcome["update_status"] == "accepted"
+                    else:
+                        assert outcome["update_status"] == "completed" and outcome["result"] is None
+                        assert await client.update_outcome(workflow_id, original["update_id"]) is None
+                    cell.update(
+                        {
+                            "scope": "original keyed update; unscoped SDK route returns original run",
+                            "update_id": original["update_id"],
+                            "concurrent_retries": len(receipts),
+                            "retained": True,
+                            "intent_verified": False,
+                            "changed_name": "request_id_conflict",
+                        }
+                    )
+                report["cells"].append(cell)
+        report["complete"] = True
+        report["finished_at"] = datetime.now(timezone.utc).isoformat()
+    except Exception as error:
+        report["failure_type"] = type(error).__name__
+        raise
+    finally:
+        args.output.write_text(json.dumps(report, indent=2) + "\n")
+    return report
+
+
+if __name__ == "__main__":
+    raise SystemExit("Use the captured-input run_physical_readback.py controller")