From 8236de8acb772860883bd4239a00d0fd76316641 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:06:56 +0000
Subject: [PATCH 1/4] test: qualify installed Rust readbacks after physical
 SQLite upgrade

---
 .../command-keys/physical-sqlite-profile.json |  70 +++++++
 .../command-keys/physical_upgrade.py          | 189 ++++++++++++++++++
 scripts/acceptance/command-keys/run_open.py   |  70 ++++++-
 .../command-keys/synthetic-sqlite.md          |  50 +++++
 .../command-keys/test_physical_upgrade.py     |  82 ++++++++
 5 files changed, 460 insertions(+), 1 deletion(-)
 create mode 100644 scripts/acceptance/command-keys/physical-sqlite-profile.json
 create mode 100644 scripts/acceptance/command-keys/physical_upgrade.py
 create mode 100644 scripts/acceptance/command-keys/test_physical_upgrade.py

diff --git a/scripts/acceptance/command-keys/physical-sqlite-profile.json b/scripts/acceptance/command-keys/physical-sqlite-profile.json
new file mode 100644
index 0000000..37f3699
--- /dev/null
+++ b/scripts/acceptance/command-keys/physical-sqlite-profile.json
@@ -0,0 +1,70 @@
+{
+  "baseline_index": "sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e",
+  "baseline_source": "28f084513235f6647e740c4b3d9e3aace7a323b7",
+  "baseline_version": "3.8.0",
+  "journal_checksum_sha384": "53b752529a1f7addaefe62ab61160dd42b3990f805cfdc4ecd1b48927a6c2f7ce42526fc04522ac887a37208f343fc73",
+  "journal_description": "full frozen PHP schema, native worker incarnations, session holders, stream reclaim deadlines, service call deadlines, service admission counters and committed external payload deletions and task ready origins and poll expiry indexes and bound run command contracts and task queue activity dispatch pauses and keyed command intents",
+  "new_schema_objects": [
+    "dw_command_intents",
+    "dw_command_intents_namespace",
+    "sqlite_autoindex_dw_command_intents_1"
+  ],
+  "producer_manifest_commit": "c542ae7928a73215e5dc9d1d67085e1a4191ba57",
+  "producer_manifest_sha256": "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f",
+  "schema": "durable-workflow.physical-sqlite-upgrade-profile.v1",
+  "schema_after": 13,
+  "schema_before": 12,
+  "server_index": "sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9",
+  "server_source": "5614274d2629d03f92932a07dfee0fb10349ddd9",
+  "server_version": "3.9.2",
+  "source_inputs": {
+    "rust/migrations/sqlite/0003_full_schema.sql": {
+      "git_blob": "3e58585408652a8e0622db183915f04e62349551",
+      "sha256": "661e8dc4ca329b5798bd661ebf1678fb43d813cdf61dd2d0c4e29978d908ccee"
+    },
+    "rust/migrations/sqlite/0004_worker_incarnations.sql": {
+      "git_blob": "375d6507b12fb469714180d4d7f57bdda0efc520",
+      "sha256": "51c59ea1bbd59129e5696660114139cadde1f0afdd984bba34f51c485b4798fe"
+    },
+    "rust/migrations/sqlite/0005_worker_session_holders.sql": {
+      "git_blob": "d7e099452e707fe6c93cb0b01187d2d92b8e3826",
+      "sha256": "9c6bc35eafb7f62535388b2f890509f63b2c877676e3e8fe63eb938e589e3fee"
+    },
+    "rust/migrations/sqlite/0006_stream_reclaims.sql": {
+      "git_blob": "481e73c9f0df21950e797dcd2caf784bde6aed20",
+      "sha256": "b7e99137ee8690612535639682c94e71a632d1ec45bb2b071b098425986a8a51"
+    },
+    "rust/migrations/sqlite/0007_service_calls.sql": {
+      "git_blob": "fa2de782c9ab06e195430bf47760c6b847f00451",
+      "sha256": "752a1a40556d940c0a99d22fd5e4b4419e9a97078cde29802d80337fbcf17ed9"
+    },
+    "rust/migrations/sqlite/0008_external_payload_deletions.sql": {
+      "git_blob": "3faec1972524eaa783fd8039c26d92be591107cf",
+      "sha256": "d66f652cb2bdd96ffb57c402d94b9ac3831e6425040c90cc6c2653c50dad961b"
+    },
+    "rust/migrations/sqlite/0009_task_ready_origin.sql": {
+      "git_blob": "559740942e5e7c15afbd75241adcea616bc564b0",
+      "sha256": "e5d47c5062ca0953151e315086323e173f5b2bd222c0806cef035aaa9417fcd8"
+    },
+    "rust/migrations/sqlite/0010_poll_expiry_indexes.sql": {
+      "git_blob": "23d7c3e6797e8c0a211bab6706fba2359d4e4e39",
+      "sha256": "4d4fb5e03ad1780c191dda35af0721de022009af2efedb9408bdefe1ace3a5b4"
+    },
+    "rust/migrations/sqlite/0011_run_command_contracts.sql": {
+      "git_blob": "eaacac26c1ee44289da98804cc669f1cddd8b416",
+      "sha256": "385f4134f409493178a11044348643ee15022ce050919ea88836a3e1d00c19f9"
+    },
+    "rust/migrations/sqlite/0012_task_queue_pauses.sql": {
+      "git_blob": "fa43906db894d91d7142c0a76d6213b311796ba2",
+      "sha256": "484ab3a84c09fbc27b6bb39965b4075a1ce6fac4751a209670fe148812beed1f"
+    },
+    "rust/migrations/sqlite/0013_command_intents.sql": {
+      "git_blob": "295c7f89e281fd9a8275a4100d525d566687145b",
+      "sha256": "fc3ad1730682c09704027dcbabc17cd6ded9c642a3ff20d7988d5f4b4a5726fd"
+    },
+    "rust/src/runtime/schema.rs": {
+      "git_blob": "78f4fc9f804f9c8b5130846db9cc9164c85179aa",
+      "sha256": "1575bb959f0abf3092da1c369bcd5ac8f13f621706f1f640da1d8e7cf2ddadd2"
+    }
+  }
+}
diff --git a/scripts/acceptance/command-keys/physical_upgrade.py b/scripts/acceptance/command-keys/physical_upgrade.py
new file mode 100644
index 0000000..1c1e06d
--- /dev/null
+++ b/scripts/acceptance/command-keys/physical_upgrade.py
@@ -0,0 +1,189 @@
+"""Read a producer-owned physical seed without synthesizing legacy identities."""
+from concurrent.futures import ThreadPoolExecutor
+from copy import deepcopy
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+
+import execution_guards as guards
+import sqlite_legacy
+
+
+PROFILE = Path(__file__).with_name('physical-sqlite-profile.json')
+PROFILE_SHA256 = '9ab24b2c8fe8fd2ae6976fe70770093ad0cf677b5a0c7855e5741b25918af3e9'
+
+
+def profile():
+    raw = PROFILE.read_bytes()
+    if hashlib.sha256(raw).hexdigest() != PROFILE_SHA256:
+        raise ValueError('physical profile differs from the reviewed tagged source')
+    return json.loads(raw)
+
+
+def read_owned(path, digest, limit):
+    path = Path(path)
+    if (not path.is_absolute() or path.resolve() != path
+            or not re.fullmatch('[0-9a-f]{64}', digest)):
+        raise ValueError('physical input needs a canonical path and exact SHA-256')
+    descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
+    try:
+        if not 0 < os.fstat(descriptor).st_size <= limit:
+            raise ValueError('physical input exceeds its byte bound')
+        record = guards._record(descriptor)
+        raw = os.pread(descriptor, limit + 1, 0)
+        if (record['sha256'] != digest or len(raw) != record['size']
+                or hashlib.sha256(raw).hexdigest() != digest
+                or guards._record(descriptor) != record
+                or guards._identity(path.lstat()) != tuple(record['identity'])):
+            raise ValueError('physical input identity/content differs')
+        return raw, record
+    finally:
+        os.close(descriptor)
+
+
+def prepare(backup, backup_sha, receipts, receipts_sha):
+    selected = profile()
+    raw, original = read_owned(backup, backup_sha, sqlite_legacy.MAX_DATABASE_BYTES)
+    for suffix in ('-wal', '-journal'):
+        sidecar = Path(str(backup) + suffix)
+        if sidecar.is_symlink() or (sidecar.exists() and sidecar.stat().st_size):
+            raise ValueError('physical backup must be stopped and checkpointed')
+    packet_raw, _ = read_owned(receipts, receipts_sha, 4 * 1024 * 1024)
+    packet = json.loads(packet_raw)
+    assert packet['schema'] == 'durable-workflow.command-key-physical-seed.v1'
+    assert packet['manifest_sha256'] == selected['producer_manifest_sha256']
+    assert packet['expected_external_baseline'] == {
+        'tag':selected['baseline_version'],'source_commit':selected['baseline_source'],
+        'index_digest':selected['baseline_index']}
+    assert len(packet['cases']) == 6
+    assert len({row['workflow_id'] for row in packet['cases']}) == 6
+    assert len({row['receipt']['command_id'] for row in packet['cases']}) == 6
+    assert {(row['kind'],row['state']) for row in packet['cases']} == {
+        (kind,state) for kind in ('signal','update') for state in ('open','completed','continued')}
+    for row in packet['cases']:
+        receipt = row['receipt']
+        assert receipt['workflow_id'] == row['workflow_id']
+        assert all(isinstance(receipt.get(key),str) and receipt[key] for key in ('command_id','run_id'))
+        if row['kind'] == 'update':
+            assert all(isinstance(receipt.get(key),str) and receipt[key] for key in ('update_id','accepted_at','update_name'))
+            assert isinstance(row['original_body']['request_id'],str) and row['original_body']['request_id']
+    return raw, original, packet, selected
+
+
+def copy_backup(raw, destination):
+    parent = destination.parent.stat()
+    if (not destination.is_absolute() or destination.parent.resolve() != destination.parent
+            or not stat.S_ISDIR(parent.st_mode) or parent.st_uid != os.getuid()
+            or stat.S_IMODE(parent.st_mode) != 0o700):
+        raise ValueError('physical copy needs an owned private parent')
+    with open(os.open(destination,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600),'wb') as output:
+        output.write(raw)
+
+
+def validate_upgrade(before, after, observed, started, finished, selected):
+    assert observed == {'backend':'sqlite','from_schema_version':selected['schema_before'],
+                        'schema_version':selected['schema_after'],'status':'upgraded'}, observed
+    assert set(after['tables']) == set(before['tables']) | {'dw_command_intents'}
+    assert not after['tables']['dw_command_intents']['rows']
+    expected = deepcopy(before)
+    version = expected['tables']['dw_server_schema']['columns'].index('version')
+    assert len(expected['tables']['dw_server_schema']['rows']) == 1
+    assert expected['tables']['dw_server_schema']['rows'][0][version] == ['integer',selected['schema_before']]
+    expected['tables']['dw_server_schema']['rows'][0][version] = ['integer',selected['schema_after']]
+    journal = after['tables']['_sqlx_migrations']
+    assert len(journal['rows']) == 1
+    entry = dict(zip(journal['columns'],journal['rows'][0]))
+    assert set(entry) == {'version','description','installed_on','success','checksum','execution_time'}
+    assert entry['version'] == ['integer',selected['schema_after']] and entry['success'] == ['integer',1]
+    assert entry['execution_time'] == ['integer',0]
+    assert entry['description'] == ['text',selected['journal_description']]
+    assert entry['checksum'] == ['blob',selected['journal_checksum_sha384']]
+    assert entry['installed_on'][0] == 'text'
+    installed = datetime.strptime(entry['installed_on'][1],'%Y-%m-%d %H:%M:%S').replace(tzinfo=timezone.utc).timestamp()
+    assert started - 1 <= installed <= finished + 1
+    expected['tables']['_sqlx_migrations'] = journal
+    projected = deepcopy(after)
+    del projected['tables']['dw_command_intents']
+    added = [row for row in projected['schema'] if row not in before['schema']]
+    assert {row[1] for row in added} == set(selected['new_schema_objects'])
+    assert all(row[2] == 'dw_command_intents' for row in added)
+    projected['schema'] = [row for row in projected['schema'] if row in before['schema']]
+    assert projected == expected, 'physical upgrade altered undeclared business rows/catalog'
+
+
+def validate_original_commands(value, packet):
+    commands = value['tables']['workflow_commands']
+    columns = commands['columns']
+    for case in packet['cases']:
+        original = case['receipt']
+        matches = [dict(zip(columns,row)) for row in commands['rows']
+                   if row[columns.index('id')] == ['text',original['command_id']]]
+        assert len(matches) == 1
+        row = matches[0]
+        assert row['command_type'] == ['text',case['kind']]
+        assert row['workflow_instance_id'] == ['text',case['workflow_id']]
+        assert row['workflow_run_id'] == ['text',original['run_id']]
+        expected_key = ['null'] if case['kind'] == 'signal' else ['text',case['original_body']['request_id']]
+        assert row['request_id'] == expected_key
+
+
+def readbacks(packet, invoke, snapshot, work):
+    before = snapshot()
+    validate_original_commands(before,packet)
+    assert not before['tables']['dw_command_intents']['rows']
+    outcomes = []
+    for case in packet['cases']:
+        original = case['receipt']
+        history_args = {'op':'history','workflow_id':case['workflow_id'],'run_id':original['run_id']}
+        history = invoke(history_args)
+        assert history['kind'] == 'ok'
+        events = history['value']['history_events']
+        event_type = 'SignalReceived' if case['kind'] == 'signal' else 'UpdateAccepted'
+        expected_count = sum(event['event_type'] == event_type for event in case['history']['events'])
+        assert expected_count > 0 and sum(event['type'] == event_type for event in events) == expected_count
+        if case['kind'] == 'signal':
+            outcomes.append({'kind':'signal','state':case['state'],'scope':'NULL-key compatibility only',
+                             'original_command_id':original['command_id'],'original_run_id':original['run_id'],
+                             'history_sha256':sqlite_legacy.fingerprint(events)})
+            continue
+        request = {'op':'update_raw','workflow_id':case['workflow_id'],'name':original['update_name'],
+                   'request_id':case['original_body']['request_id'],'input':[999]}
+        fields = ('command_id','run_id','update_id','accepted_at')
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            replies = list(pool.map(lambda _:invoke(request),range(4)))
+        replies.append(invoke(dict(request,run_id=original['run_id'])))
+        for reply in replies:
+            assert reply['kind'] == 'ok' and reply['value']['status'] in (200,202), reply
+            body = reply['value']['body']
+            assert body['retained'] is True and body['intent_verified'] is False
+            assert body['request_id'] == request['request_id']
+            assert all(body[field] == original[field] for field in fields)
+            if case['state'] != 'open':assert body['update_status'] == 'completed'
+        high = invoke(dict(request,op='update'))
+        if case['state'] == 'open':
+            assert high['kind'] == 'pending' and high['receipt']['retained'] is True
+            assert high['receipt']['intent_verified'] is False
+            assert all(high['receipt'][field] == original[field] for field in fields)
+        else:
+            assert high['kind'] == 'ok' and high['value'] is None
+        for changed in (dict(request,name='other'),dict(request,run_id='00000000000000000000000000')):
+            refused = invoke(changed)
+            assert refused['kind'] == 'ok' and refused['value']['status'] == 409
+            assert refused['value']['body']['reason'] == 'request_id_conflict'
+            assert refused['value']['body']['retained_command_id'] == original['command_id']
+        assert invoke(history_args)['value']['history_events'] == events
+        outcomes.append({'kind':'update','state':case['state'],'outcome':'pass',
+                         'original':{field:original[field] for field in fields},
+                         'returned':{field:replies[0]['value']['body'][field] for field in fields},
+                         'retained':True,'intent_verified':False,'concurrency':4})
+    after = snapshot()
+    assert after == before, 'physical SDK request window altered full logical rows/schema'
+    for label,value in (('before',before),('after',after)):
+        with open(os.open(work/('physical-'+label+'-snapshot.json'),os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600),'w') as output:
+            json.dump(value,output,sort_keys=True)
+    return {'cell':'physical-sqlite-upgrade-sdk-readback-only','outcome':'pass',
+            'cases':outcomes,'request_window_sha256':sqlite_legacy.fingerprint(before)}
diff --git a/scripts/acceptance/command-keys/run_open.py b/scripts/acceptance/command-keys/run_open.py
index c75d99c..28337a2 100644
--- a/scripts/acceptance/command-keys/run_open.py
+++ b/scripts/acceptance/command-keys/run_open.py
@@ -22,6 +22,7 @@
 import sqlite_legacy as legacy
 import execution_guards as guards
 import response_loss
+import physical_upgrade
 
 
 
@@ -48,6 +49,11 @@ def main():
     parser.add_argument('--scenario', choices=['readback', 'two-rows', 'legacy-token', 'modern-raw', 'case-only'], default='readback')
     parser.add_argument('--response-loss-fixture', type=Path,
                         help='accepted shared proxy file; selects the separate postcommit-loss cell')
+    parser.add_argument('--physical-backup',type=Path)
+    parser.add_argument('--physical-backup-sha256')
+    parser.add_argument('--physical-receipts',type=Path)
+    parser.add_argument('--physical-receipts-sha256')
+    parser.add_argument('--physical-source-commit',help='accepted producer seed-source commit from its owner handoff')
     parser.add_argument('--work-dir',type=Path,required=True)
     parser.add_argument('--consumer-binary',type=Path,required=True)
     parser.add_argument('--consumer-lock',type=Path,required=True)
@@ -59,6 +65,13 @@ def main():
     parser.add_argument('--server-version',required=True)
     parser.add_argument('--server-source',required=True)
     options=parser.parse_args()
+    physical_args = (options.physical_backup, options.physical_backup_sha256, options.physical_receipts,
+                     options.physical_receipts_sha256, options.physical_source_commit)
+    physical_mode = any(value is not None for value in physical_args)
+    if physical_mode and (not all(physical_args) or options.state != 'open' or options.scenario != 'readback'
+                          or options.response_loss_fixture is not None
+                          or not re.fullmatch('[0-9a-f]{40}',options.physical_source_commit)):
+        parser.error('physical upgrade needs all exact input identities and its own cell')
     if options.state != 'open' and options.scenario != 'readback':
         parser.error('ambiguity/case scenarios require their own open-state cell')
     if options.response_loss_fixture is not None and (options.state != 'open' or options.scenario != 'readback'):
@@ -84,6 +97,14 @@ def main():
         sdk_version=options.sdk_version, sdk_sha256=options.sdk_sha256, sdk_source=options.sdk_source)
     if options.response_loss_fixture is not None:
         response_loss.fixture(options.response_loss_fixture)  # Refuse changed source before effects.
+    if physical_mode:
+        physical_raw, physical_original, physical_packet, physical_profile = physical_upgrade.prepare(
+            options.physical_backup,options.physical_backup_sha256,
+            options.physical_receipts,options.physical_receipts_sha256)
+        if (options.server_source != physical_profile['server_source']
+                or options.server_index != physical_profile['server_index']
+                or options.server_version != physical_profile['server_version']):
+            parser.error('physical profile does not bind this exact published producer tuple')
     image='ghcr.io/durable-workflow/server@'+options.server_index
     info=json.loads(capture('docker','image','inspect',image))[0]
     assert info['Config']['Labels']['org.opencontainers.image.revision']==options.server_source
@@ -100,7 +121,7 @@ def main():
     key = 'base64:' + base64.b64encode(secrets.token_bytes(32)).decode()
     with open(os.open(envfile, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600), 'w') as file:
         file.write('DW_AUTH_TOKEN='+token+'\nDW_SERVER_KEY='+key+'\n')
-    name = 'sdk-rust-216-legacy-' + options.state + '-' + secrets.token_hex(5)
+    name = ('sdk-rust-216-physical-' if physical_mode else 'sdk-rust-216-legacy-'+options.state+'-') + secrets.token_hex(5)
     owned_id = None
     binary = None
     worker_pool = None
@@ -158,6 +179,53 @@ def snapshot(path):
 
     try:
         binary = guards.BoundExecutable(options.consumer_binary, build['binary'])
+        if physical_mode:
+            physical_path=database/'physical.sqlite'
+            physical_upgrade.copy_backup(physical_raw,physical_path)
+            del physical_raw
+            original_snapshot=snapshot(physical_path)
+            physical_upgrade.validate_original_commands(original_snapshot,physical_packet)
+            upgrade_started=time.time()
+            owned_id=capture('docker','create','--init','--read-only','--name',name+'-upgrade',
+                '--cpus','1','--memory','1g','--pids-limit','256','--network','none',
+                '--cap-drop','ALL','--security-opt','no-new-privileges',
+                '--tmpfs','/tmp:mode=1777,size=128m',
+                '--tmpfs','/app/storage:uid=1000,gid=1000,mode=0700,size=64m',
+                '--tmpfs','/app/database:uid=1000,gid=1000,mode=0700,size=16m',
+                '-v',str(database)+':/cohort:rw','--env-file',str(envfile),
+                '-e','DB_CONNECTION=sqlite','-e','DB_DATABASE=/cohort/physical.sqlite',image,'schema-upgrade')
+            capture('docker','start',owned_id)
+            exit_code=capture('docker','wait',owned_id)
+            upgrade_finished=time.time()
+            upgrade_log=capture('docker','logs',owned_id)
+            with open(os.open(work/'schema-upgrade.log',os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600),'w') as output:
+                output.write(upgrade_log+'\n')
+            assert exit_code == '0', 'physical schema-upgrade exited unsuccessfully'
+            upgraded=snapshot(physical_path)
+            physical_upgrade.validate_upgrade(original_snapshot,upgraded,json.loads(upgrade_log),
+                                               upgrade_started,upgrade_finished,physical_profile)
+            physical_upgrade.validate_original_commands(upgraded,physical_packet)
+            capture('docker','rm','-v',owned_id);owned_id=None
+            start(physical_path.name)
+            cell=physical_upgrade.readbacks(physical_packet,invoke,lambda:snapshot(physical_path),work)
+            stop()
+            _, rechecked = physical_upgrade.read_owned(options.physical_backup,options.physical_backup_sha256,
+                                                       legacy.MAX_DATABASE_BYTES)
+            assert rechecked == physical_original, 'original baseline changed during the owned-copy experiment'
+            binary.check()
+            cell.update(started_at=started_at,finished_at=datetime.now(timezone.utc).isoformat(),
+                baseline_sha256=options.physical_backup_sha256,seed_receipts_sha256=options.physical_receipts_sha256,
+                producer_seed_source=options.physical_source_commit,
+                profile_sha256=physical_upgrade.PROFILE_SHA256,
+                server_source=options.server_source,server_index=options.server_index,
+                server_version=options.server_version,server_platform='linux/amd64',
+                sdk_version=options.sdk_version,sdk_sha256=options.sdk_sha256,sdk_source=options.sdk_source,
+                runner_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+                physical_adapter_sha256=hashlib.sha256((SOURCE/'physical_upgrade.py').read_bytes()).hexdigest(),
+                binary_sha256=binary.record['sha256'],build_receipt_sha256=build_receipt_sha,
+                build_subject=build['subject'],build_command=build['build_command'])
+            print(json.dumps(cell),flush=True)
+            return
         start('source.sqlite')
         queue='legacy-'+options.state+'-'+secrets.token_hex(5)
         ok({'op':'register','queue':queue,'null_update':True})
diff --git a/scripts/acceptance/command-keys/synthetic-sqlite.md b/scripts/acceptance/command-keys/synthetic-sqlite.md
index cc1a2f0..c4725c6 100644
--- a/scripts/acceptance/command-keys/synthetic-sqlite.md
+++ b/scripts/acceptance/command-keys/synthetic-sqlite.md
@@ -194,3 +194,53 @@ identities, proxy success/drop metadata and per-type zero-write digests. Private
 before/after snapshots remain mode0600 with the same seven-day retention. The
 cell performs no SQL transform, physical upgrade or worker execution. It does
 not qualify the synthetic legacy cells or the complete #216 cohort.
+
+## Consume a producer-owned physical upgrade seed
+
+The physical mode uses the separately accepted producer seed/receipt handoff.
+Run it only after that source's required CI and final independent review clear
+and its owner confirms the accepted source identity. A producer diagnostic or
+published image alone does not clear the seed-source gate. Do not run a retained
+owner-specific control script or create another seed or synthetic transform.
+
+Supply all five physical arguments with the normal recorded consumer and exact
+Server arguments:
+
+```sh
+--physical-backup "$STOPPED_BASELINE" \
+--physical-backup-sha256 "$BASELINE_SHA256" \
+--physical-receipts "$PRODUCER_SEED_RECEIPTS" \
+--physical-receipts-sha256 "$SEED_RECEIPTS_SHA256" \
+--physical-source-commit "$ACCEPTED_SEED_SOURCE"
+```
+
+The runner verifies owned canonical single-link bounded inputs, complete hashes,
+the physical seed's baseline/manifest identity and all six distinct cases. It
+copies the stopped checkpointed baseline into a new owned private file and runs
+the pinned image's `schema-upgrade` once, offline, at1CPU/1GiB/256PIDs. The
+original is never a write target and its complete identity/hash is rechecked.
+No legacy key or intent row is manufactured.
+
+The machine-owned [SQLite profile](physical-sqlite-profile.json) binds the exact
+published producer source/index, version step, tagged schema inputs, journal
+description/checksum and new schema objects. Every original row/catalog object
+must survive except the declared schema-version and source-defined SQLx journal
+replacement. All journal fields and the invocation timestamp interval are
+validated. The new intent table must be empty. Then the same image starts on
+that upgraded copy through the existing exact-peer/runtime guards.
+
+The installed Rust consumer performs four concurrent keyed-update retries in
+open/completed/continued states, original-run reads, typed pending or completed
+null results, and name/run409 fences retaining original identities. Genuine old
+NULL-key signals are checked through preserved rows and history only, without
+claiming a caller key, retained receipt or verified intent. Complete snapshots
+immediately before/after the SDK request window must match, including all task,
+history, update and intent rows. A post-stop repair snapshot is not substituted
+for that bounded request-window proof.
+
+The terminal receipt distinguishes the physical cell from synthetic/loss cells
+and binds producer source, original backup/receipt hashes, consumer build/binary
+and complete request-window digest. Runtime/upgrade containers, owned copy and
+env are removed. Private mode0600 snapshots/logs retain the same seven-day limit.
+This profile covers SQLite/amd64 only. It does not qualify other backends or the
+whole #216 cohort by itself.
diff --git a/scripts/acceptance/command-keys/test_physical_upgrade.py b/scripts/acceptance/command-keys/test_physical_upgrade.py
new file mode 100644
index 0000000..0f313e1
--- /dev/null
+++ b/scripts/acceptance/command-keys/test_physical_upgrade.py
@@ -0,0 +1,82 @@
+"""Inert physical-upgrade guards; no seed, SDK, Docker or schema CLI executes."""
+from copy import deepcopy
+from datetime import datetime, timezone
+from pathlib import Path
+import tempfile
+import unittest
+
+import physical_upgrade as physical
+
+
+class PhysicalUpgradeTest(unittest.TestCase):
+    def setUp(self):
+        self.profile = physical.profile()
+        self.started = datetime(2026,1,1,tzinfo=timezone.utc).timestamp()
+        self.before = {'schema':[['table','business','business','CREATE TABLE business(id TEXT)']],
+                       'tables':{'business':{'columns':['id'],'rows':[[['text','original']]]},
+                                 'dw_server_schema':{'columns':['engine','version'],
+                                    'rows':[[['text','rust-development'],['integer',12]]]},
+                                 '_sqlx_migrations':{'columns':['version'],'rows':[[['integer',12]]]}}}
+        self.after = deepcopy(self.before)
+        self.after['tables']['dw_server_schema']['rows'][0][1] = ['integer',13]
+        self.after['tables']['dw_command_intents'] = {'columns':['command_id','namespace','intent_digest'],'rows':[]}
+        self.after['tables']['_sqlx_migrations'] = {
+            'columns':['version','description','installed_on','success','checksum','execution_time'],
+            'rows':[[['integer',13],['text',self.profile['journal_description']],
+                     ['text','2026-01-01 00:00:00'],['integer',1],
+                     ['blob',self.profile['journal_checksum_sha384']],['integer',0]]]}
+        self.after['schema'] += [[('table' if name=='dw_command_intents' else 'index'),
+                                 name,'dw_command_intents',None] for name in self.profile['new_schema_objects']]
+        self.observed = {'backend':'sqlite','from_schema_version':12,'schema_version':13,'status':'upgraded'}
+
+    def validate(self, after):
+        physical.validate_upgrade(self.before,after,self.observed,self.started,self.started+1,self.profile)
+
+    def test_only_declared_version_journal_and_empty_intent_catalog_accept(self):
+        self.validate(self.after)
+
+    def test_original_business_row_or_catalog_change_refuses(self):
+        changed=deepcopy(self.after);changed['tables']['business']['rows'][0][0]=['text','changed']
+        with self.assertRaises(AssertionError):self.validate(changed)
+        changed=deepcopy(self.after);changed['schema'].pop(0)
+        with self.assertRaises(AssertionError):self.validate(changed)
+        changed=deepcopy(self.after);changed['schema'].append(['index','extra','dw_command_intents',None])
+        with self.assertRaises(AssertionError):self.validate(changed)
+
+    def test_every_journal_field_and_invocation_time_are_bound(self):
+        for index,value in enumerate((['integer',12],['text','other'],['text','2025-01-01 00:00:00'],
+                                     ['integer',0],['blob','00'],['integer',1])):
+            changed=deepcopy(self.after);changed['tables']['_sqlx_migrations']['rows'][0][index]=value
+            with self.subTest(index=index), self.assertRaises(AssertionError):self.validate(changed)
+        changed=deepcopy(self.after);changed['tables']['dw_command_intents']['rows']=[[]]
+        with self.assertRaises(AssertionError):self.validate(changed)
+
+    def test_unowned_shape_changed_hash_and_existing_destination_refuse(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root=Path(directory);source=root/'input';source.write_bytes(b'owned')
+            source.chmod(0o600)
+            digest=physical.hashlib.sha256(source.read_bytes()).hexdigest()
+            raw,_=physical.read_owned(source,digest,100)
+            with self.assertRaises(ValueError):physical.read_owned(source,'0'*64,100)
+            with self.assertRaises(ValueError):physical.read_owned(source,digest,2)
+            link=root/'link';link.symlink_to(source)
+            with self.assertRaises(ValueError):physical.read_owned(link,digest,100)
+            link.unlink();link.hardlink_to(source)
+            with self.assertRaises(ValueError):physical.read_owned(source,digest,100)
+            link.unlink();destination=root/'copy'
+            physical.copy_backup(raw,destination)
+            with self.assertRaises(FileExistsError):physical.copy_backup(b'other',destination)
+            self.assertEqual(raw,source.read_bytes())
+            self.assertEqual(raw,destination.read_bytes())
+
+    def test_null_key_signal_is_compatibility_only_and_cannot_be_relabelled(self):
+        packet={'cases':[{'kind':'signal','workflow_id':'workflow','receipt':{'command_id':'signal','run_id':'run'}}]}
+        snapshot={'tables':{'workflow_commands':{
+            'columns':['id','command_type','workflow_instance_id','workflow_run_id','request_id'],
+            'rows':[[['text','signal'],['text','signal'],['text','workflow'],['text','run'],['null']]]}}}
+        physical.validate_original_commands(snapshot,packet)
+        snapshot['tables']['workflow_commands']['rows'][0][-1]=['text','manufactured']
+        with self.assertRaises(AssertionError):physical.validate_original_commands(snapshot,packet)
+
+
+if __name__=='__main__':unittest.main()

From 069e26c11c8d153ce73b28ce1688089132828867 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:16:26 +0000
Subject: [PATCH 2/4] test: bind physical handoff identity before any runner
 effect

---
 .../command-keys/physical_upgrade.py          |  7 ++++++
 scripts/acceptance/command-keys/run_open.py   |  2 ++
 .../command-keys/synthetic-sqlite.md          |  2 ++
 .../command-keys/test_physical_upgrade.py     | 24 +++++++++++++++++++
 4 files changed, 35 insertions(+)

diff --git a/scripts/acceptance/command-keys/physical_upgrade.py b/scripts/acceptance/command-keys/physical_upgrade.py
index 1c1e06d..9736bb3 100644
--- a/scripts/acceptance/command-keys/physical_upgrade.py
+++ b/scripts/acceptance/command-keys/physical_upgrade.py
@@ -24,6 +24,13 @@ def profile():
     return json.loads(raw)
 
 
+def require_handoff(commit):
+    selected = profile()
+    if commit != selected['producer_manifest_commit']:
+        raise ValueError('producer handoff commit differs from the reviewed physical profile')
+    return selected
+
+
 def read_owned(path, digest, limit):
     path = Path(path)
     if (not path.is_absolute() or path.resolve() != path
diff --git a/scripts/acceptance/command-keys/run_open.py b/scripts/acceptance/command-keys/run_open.py
index 28337a2..5049555 100644
--- a/scripts/acceptance/command-keys/run_open.py
+++ b/scripts/acceptance/command-keys/run_open.py
@@ -72,6 +72,8 @@ def main():
                           or options.response_loss_fixture is not None
                           or not re.fullmatch('[0-9a-f]{40}',options.physical_source_commit)):
         parser.error('physical upgrade needs all exact input identities and its own cell')
+    if physical_mode:
+        physical_upgrade.require_handoff(options.physical_source_commit)
     if options.state != 'open' and options.scenario != 'readback':
         parser.error('ambiguity/case scenarios require their own open-state cell')
     if options.response_loss_fixture is not None and (options.state != 'open' or options.scenario != 'readback'):
diff --git a/scripts/acceptance/command-keys/synthetic-sqlite.md b/scripts/acceptance/command-keys/synthetic-sqlite.md
index c4725c6..48ca971 100644
--- a/scripts/acceptance/command-keys/synthetic-sqlite.md
+++ b/scripts/acceptance/command-keys/synthetic-sqlite.md
@@ -202,6 +202,8 @@ Run it only after that source's required CI and final independent review clear
 and its owner confirms the accepted source identity. A producer diagnostic or
 published image alone does not clear the seed-source gate. Do not run a retained
 owner-specific control script or create another seed or synthetic transform.
+The supplied source commit must exactly equal the frozen profile's accepted
+producer commit. Another well-formed SHA refuses before input or runtime effects.
 
 Supply all five physical arguments with the normal recorded consumer and exact
 Server arguments:
diff --git a/scripts/acceptance/command-keys/test_physical_upgrade.py b/scripts/acceptance/command-keys/test_physical_upgrade.py
index 0f313e1..54a7ff6 100644
--- a/scripts/acceptance/command-keys/test_physical_upgrade.py
+++ b/scripts/acceptance/command-keys/test_physical_upgrade.py
@@ -2,6 +2,8 @@
 from copy import deepcopy
 from datetime import datetime, timezone
 from pathlib import Path
+import subprocess
+import sys
 import tempfile
 import unittest
 
@@ -35,6 +37,28 @@ def validate(self, after):
     def test_only_declared_version_journal_and_empty_intent_catalog_accept(self):
         self.validate(self.after)
 
+    def test_another_well_formed_handoff_sha_refuses_before_input_or_runtime_effects(self):
+        self.assertEqual(self.profile,physical.require_handoff(self.profile['producer_manifest_commit']))
+        with self.assertRaisesRegex(ValueError,'handoff commit'):
+            physical.require_handoff('0'*40)
+
+    def test_cli_wrong_handoff_refuses_before_any_file_or_container_input(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root=Path(directory)
+            command=[sys.executable,'-B',str(Path(__file__).with_name('run_open.py')),
+                '--physical-backup',str(root/'missing-backup'),'--physical-backup-sha256','a'*64,
+                '--physical-receipts',str(root/'missing-seed'),'--physical-receipts-sha256','b'*64,
+                '--physical-source-commit','0'*40,'--work-dir',str(root/'output'),
+                '--consumer-binary',str(root/'missing-binary'),'--consumer-lock',str(root/'missing-lock'),
+                '--build-receipt',str(root/'missing-receipt'),'--sdk-version','4.16.0',
+                '--sdk-sha256','c'*64,'--sdk-source','d'*40,
+                '--server-index',self.profile['server_index'],'--server-version',self.profile['server_version'],
+                '--server-source',self.profile['server_source']]
+            result=subprocess.run(command,text=True,capture_output=True,timeout=10)
+            self.assertNotEqual(0,result.returncode)
+            self.assertIn('producer handoff commit differs',result.stderr)
+            self.assertEqual([],list(root.iterdir()))
+
     def test_original_business_row_or_catalog_change_refuses(self):
         changed=deepcopy(self.after);changed['tables']['business']['rows'][0][0]=['text','changed']
         with self.assertRaises(AssertionError):self.validate(changed)

From cb9c1f1f1de24cb127ee973b64a711ac4b9bfcf2 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:24:35 +0000
Subject: [PATCH 3/4] test: distinguish actual seed execution from fixture
 handoff authority

---
 .../command-keys/physical-sqlite-profile.json        | 10 +++++++++-
 scripts/acceptance/command-keys/physical_upgrade.py  |  7 +++++--
 scripts/acceptance/command-keys/run_open.py          |  5 +++--
 scripts/acceptance/command-keys/synthetic-sqlite.md  |  9 +++++++--
 .../acceptance/command-keys/test_physical_upgrade.py | 12 +++++++++++-
 5 files changed, 35 insertions(+), 8 deletions(-)

diff --git a/scripts/acceptance/command-keys/physical-sqlite-profile.json b/scripts/acceptance/command-keys/physical-sqlite-profile.json
index 37f3699..a66d32f 100644
--- a/scripts/acceptance/command-keys/physical-sqlite-profile.json
+++ b/scripts/acceptance/command-keys/physical-sqlite-profile.json
@@ -1,5 +1,13 @@
 {
   "baseline_index": "sha256:7988a23525d9e50e997a9f6f02e801300ed4f626de3cf6977033657ae7e9f58e",
+  "baseline_seed_execution": {
+    "backup_sha256": "7d1826f7681342bb3a5974509648e2e8d75e7a21d262f65f3ec1f58727ccbc0c",
+    "driver_git_blob": "96f3950e5113ac8484d72e64677ea2fbfab2facc",
+    "driver_path": "scripts/ci/command_key_upgrade_seed.py",
+    "driver_sha256": "01c0fa1251ac3c809723e58d71817881fe2c5c9498579bc4c2e6097324a3618b",
+    "receipts_sha256": "8c728ce827df3a11621dfb0e82cdd404bbab48784182a5e92c8d632d49047fc5",
+    "source_commit": "09a4ccda3da11853606af031a8970bbde4b2d6ec"
+  },
   "baseline_source": "28f084513235f6647e740c4b3d9e3aace7a323b7",
   "baseline_version": "3.8.0",
   "journal_checksum_sha384": "53b752529a1f7addaefe62ab61160dd42b3990f805cfdc4ecd1b48927a6c2f7ce42526fc04522ac887a37208f343fc73",
@@ -9,7 +17,7 @@
     "dw_command_intents_namespace",
     "sqlite_autoindex_dw_command_intents_1"
   ],
-  "producer_manifest_commit": "c542ae7928a73215e5dc9d1d67085e1a4191ba57",
+  "producer_contract_commit": "c542ae7928a73215e5dc9d1d67085e1a4191ba57",
   "producer_manifest_sha256": "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f",
   "schema": "durable-workflow.physical-sqlite-upgrade-profile.v1",
   "schema_after": 13,
diff --git a/scripts/acceptance/command-keys/physical_upgrade.py b/scripts/acceptance/command-keys/physical_upgrade.py
index 9736bb3..19497e1 100644
--- a/scripts/acceptance/command-keys/physical_upgrade.py
+++ b/scripts/acceptance/command-keys/physical_upgrade.py
@@ -14,7 +14,7 @@
 
 
 PROFILE = Path(__file__).with_name('physical-sqlite-profile.json')
-PROFILE_SHA256 = '9ab24b2c8fe8fd2ae6976fe70770093ad0cf677b5a0c7855e5741b25918af3e9'
+PROFILE_SHA256 = '8c29c63a8fefa04c0d090924550202437a812925d4e3230be09e119f0f850c03'
 
 
 def profile():
@@ -26,7 +26,7 @@ def profile():
 
 def require_handoff(commit):
     selected = profile()
-    if commit != selected['producer_manifest_commit']:
+    if commit != selected['producer_contract_commit']:
         raise ValueError('producer handoff commit differs from the reviewed physical profile')
     return selected
 
@@ -54,6 +54,9 @@ def read_owned(path, digest, limit):
 
 def prepare(backup, backup_sha, receipts, receipts_sha):
     selected = profile()
+    seed = selected['baseline_seed_execution']
+    if backup_sha != seed['backup_sha256'] or receipts_sha != seed['receipts_sha256']:
+        raise ValueError('physical baseline provenance differs from the frozen seed execution')
     raw, original = read_owned(backup, backup_sha, sqlite_legacy.MAX_DATABASE_BYTES)
     for suffix in ('-wal', '-journal'):
         sidecar = Path(str(backup) + suffix)
diff --git a/scripts/acceptance/command-keys/run_open.py b/scripts/acceptance/command-keys/run_open.py
index 5049555..5b955a4 100644
--- a/scripts/acceptance/command-keys/run_open.py
+++ b/scripts/acceptance/command-keys/run_open.py
@@ -53,7 +53,7 @@ def main():
     parser.add_argument('--physical-backup-sha256')
     parser.add_argument('--physical-receipts',type=Path)
     parser.add_argument('--physical-receipts-sha256')
-    parser.add_argument('--physical-source-commit',help='accepted producer seed-source commit from its owner handoff')
+    parser.add_argument('--physical-source-commit',help='accepted fixture-contract commit from its owner handoff')
     parser.add_argument('--work-dir',type=Path,required=True)
     parser.add_argument('--consumer-binary',type=Path,required=True)
     parser.add_argument('--consumer-lock',type=Path,required=True)
@@ -217,7 +217,8 @@ def snapshot(path):
             binary.check()
             cell.update(started_at=started_at,finished_at=datetime.now(timezone.utc).isoformat(),
                 baseline_sha256=options.physical_backup_sha256,seed_receipts_sha256=options.physical_receipts_sha256,
-                producer_seed_source=options.physical_source_commit,
+                accepted_contract_source=options.physical_source_commit,
+                baseline_seed_execution=physical_profile['baseline_seed_execution'],
                 profile_sha256=physical_upgrade.PROFILE_SHA256,
                 server_source=options.server_source,server_index=options.server_index,
                 server_version=options.server_version,server_platform='linux/amd64',
diff --git a/scripts/acceptance/command-keys/synthetic-sqlite.md b/scripts/acceptance/command-keys/synthetic-sqlite.md
index 48ca971..4fa596a 100644
--- a/scripts/acceptance/command-keys/synthetic-sqlite.md
+++ b/scripts/acceptance/command-keys/synthetic-sqlite.md
@@ -202,8 +202,13 @@ Run it only after that source's required CI and final independent review clear
 and its owner confirms the accepted source identity. A producer diagnostic or
 published image alone does not clear the seed-source gate. Do not run a retained
 owner-specific control script or create another seed or synthetic transform.
-The supplied source commit must exactly equal the frozen profile's accepted
-producer commit. Another well-formed SHA refuses before input or runtime effects.
+The supplied source commit must exactly equal the frozen profile's fixture-contract
+commit. Another well-formed SHA refuses before input or runtime effects. The
+historical baseline seed-execution commit, driver blob/checksum and exact
+backup/packet hashes are bound separately in that profile. Another artifact
+cannot inherit those historical execution identities. The terminal receipt
+reports accepted contract source and actual baseline seed execution separately.
+Accepting a later fixture source never relabels the historical seed run.
 
 Supply all five physical arguments with the normal recorded consumer and exact
 Server arguments:
diff --git a/scripts/acceptance/command-keys/test_physical_upgrade.py b/scripts/acceptance/command-keys/test_physical_upgrade.py
index 54a7ff6..6ff14c9 100644
--- a/scripts/acceptance/command-keys/test_physical_upgrade.py
+++ b/scripts/acceptance/command-keys/test_physical_upgrade.py
@@ -38,7 +38,7 @@ def test_only_declared_version_journal_and_empty_intent_catalog_accept(self):
         self.validate(self.after)
 
     def test_another_well_formed_handoff_sha_refuses_before_input_or_runtime_effects(self):
-        self.assertEqual(self.profile,physical.require_handoff(self.profile['producer_manifest_commit']))
+        self.assertEqual(self.profile,physical.require_handoff(self.profile['producer_contract_commit']))
         with self.assertRaisesRegex(ValueError,'handoff commit'):
             physical.require_handoff('0'*40)
 
@@ -59,6 +59,16 @@ def test_cli_wrong_handoff_refuses_before_any_file_or_container_input(self):
             self.assertIn('producer handoff commit differs',result.stderr)
             self.assertEqual([],list(root.iterdir()))
 
+    def test_another_artifact_cannot_inherit_the_frozen_seed_execution_provenance(self):
+        seed=self.profile['baseline_seed_execution']
+        with tempfile.TemporaryDirectory() as directory:
+            root=Path(directory)
+            for backup_sha,packet_sha in (('0'*64,seed['receipts_sha256']),
+                                           (seed['backup_sha256'],'0'*64)):
+                with self.assertRaisesRegex(ValueError,'baseline provenance'):
+                    physical.prepare(root/'absent-backup',backup_sha,root/'absent-packet',packet_sha)
+            self.assertEqual([],list(root.iterdir()))
+
     def test_original_business_row_or_catalog_change_refuses(self):
         changed=deepcopy(self.after);changed['tables']['business']['rows'][0][0]=['text','changed']
         with self.assertRaises(AssertionError):self.validate(changed)

From ba3f5f06462445765ce2875cf10fdb1b66c9ec78 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 13 Nov 2026 17:31:40 +0000
Subject: [PATCH 4/4] test: freeze accepted physical handoff and enforce
 receipt provenance

---
 .../command-keys/physical-sqlite-profile.json |  2 +-
 .../command-keys/physical_upgrade.py          | 10 +++-
 scripts/acceptance/command-keys/run_open.py   |  8 +--
 .../command-keys/test_physical_upgrade.py     | 50 ++++++++++++++++++-
 4 files changed, 62 insertions(+), 8 deletions(-)

diff --git a/scripts/acceptance/command-keys/physical-sqlite-profile.json b/scripts/acceptance/command-keys/physical-sqlite-profile.json
index a66d32f..879e367 100644
--- a/scripts/acceptance/command-keys/physical-sqlite-profile.json
+++ b/scripts/acceptance/command-keys/physical-sqlite-profile.json
@@ -17,7 +17,7 @@
     "dw_command_intents_namespace",
     "sqlite_autoindex_dw_command_intents_1"
   ],
-  "producer_contract_commit": "c542ae7928a73215e5dc9d1d67085e1a4191ba57",
+  "producer_contract_commit": "a09f156399416216d2f097c257a2164f83c7eda7",
   "producer_manifest_sha256": "ceb4695643e48c7bfe3f81eea380c6ab9aa0b973da830cb552d070344047319f",
   "schema": "durable-workflow.physical-sqlite-upgrade-profile.v1",
   "schema_after": 13,
diff --git a/scripts/acceptance/command-keys/physical_upgrade.py b/scripts/acceptance/command-keys/physical_upgrade.py
index 19497e1..f09da2d 100644
--- a/scripts/acceptance/command-keys/physical_upgrade.py
+++ b/scripts/acceptance/command-keys/physical_upgrade.py
@@ -14,7 +14,7 @@
 
 
 PROFILE = Path(__file__).with_name('physical-sqlite-profile.json')
-PROFILE_SHA256 = '8c29c63a8fefa04c0d090924550202437a812925d4e3230be09e119f0f850c03'
+PROFILE_SHA256 = '8f9c4695b814a6c3f923885e51d1900506906138b7e0d27dc87b8df592a36e9e'
 
 
 def profile():
@@ -31,6 +31,14 @@ def require_handoff(commit):
     return selected
 
 
+def attribution(commit, declared_execution):
+    selected = require_handoff(commit)
+    if declared_execution != selected['baseline_seed_execution']:
+        raise ValueError('historical seed source/driver identity differs from the frozen execution')
+    return {'accepted_contract_source':commit,
+            'baseline_seed_execution':deepcopy(selected['baseline_seed_execution'])}
+
+
 def read_owned(path, digest, limit):
     path = Path(path)
     if (not path.is_absolute() or path.resolve() != path
diff --git a/scripts/acceptance/command-keys/run_open.py b/scripts/acceptance/command-keys/run_open.py
index 5b955a4..db84e64 100644
--- a/scripts/acceptance/command-keys/run_open.py
+++ b/scripts/acceptance/command-keys/run_open.py
@@ -73,7 +73,9 @@ def main():
                           or not re.fullmatch('[0-9a-f]{40}',options.physical_source_commit)):
         parser.error('physical upgrade needs all exact input identities and its own cell')
     if physical_mode:
-        physical_upgrade.require_handoff(options.physical_source_commit)
+        selected_contract=physical_upgrade.require_handoff(options.physical_source_commit)
+        physical_attribution=physical_upgrade.attribution(options.physical_source_commit,
+                                                         selected_contract['baseline_seed_execution'])
     if options.state != 'open' and options.scenario != 'readback':
         parser.error('ambiguity/case scenarios require their own open-state cell')
     if options.response_loss_fixture is not None and (options.state != 'open' or options.scenario != 'readback'):
@@ -215,10 +217,8 @@ def snapshot(path):
                                                        legacy.MAX_DATABASE_BYTES)
             assert rechecked == physical_original, 'original baseline changed during the owned-copy experiment'
             binary.check()
-            cell.update(started_at=started_at,finished_at=datetime.now(timezone.utc).isoformat(),
+            cell.update(**physical_attribution,started_at=started_at,finished_at=datetime.now(timezone.utc).isoformat(),
                 baseline_sha256=options.physical_backup_sha256,seed_receipts_sha256=options.physical_receipts_sha256,
-                accepted_contract_source=options.physical_source_commit,
-                baseline_seed_execution=physical_profile['baseline_seed_execution'],
                 profile_sha256=physical_upgrade.PROFILE_SHA256,
                 server_source=options.server_source,server_index=options.server_index,
                 server_version=options.server_version,server_platform='linux/amd64',
diff --git a/scripts/acceptance/command-keys/test_physical_upgrade.py b/scripts/acceptance/command-keys/test_physical_upgrade.py
index 6ff14c9..2377dc0 100644
--- a/scripts/acceptance/command-keys/test_physical_upgrade.py
+++ b/scripts/acceptance/command-keys/test_physical_upgrade.py
@@ -1,7 +1,11 @@
 """Inert physical-upgrade guards; no seed, SDK, Docker or schema CLI executes."""
 from copy import deepcopy
+from contextlib import closing
 from datetime import datetime, timezone
 from pathlib import Path
+import hashlib
+import json
+import sqlite3
 import subprocess
 import sys
 import tempfile
@@ -39,8 +43,10 @@ def test_only_declared_version_journal_and_empty_intent_catalog_accept(self):
 
     def test_another_well_formed_handoff_sha_refuses_before_input_or_runtime_effects(self):
         self.assertEqual(self.profile,physical.require_handoff(self.profile['producer_contract_commit']))
-        with self.assertRaisesRegex(ValueError,'handoff commit'):
-            physical.require_handoff('0'*40)
+        for wrong in ('0'*40,'c542ae7928a73215e5dc9d1d67085e1a4191ba57',
+                      self.profile['baseline_seed_execution']['source_commit']):
+            with self.subTest(wrong=wrong), self.assertRaisesRegex(ValueError,'handoff commit'):
+                physical.require_handoff(wrong)
 
     def test_cli_wrong_handoff_refuses_before_any_file_or_container_input(self):
         with tempfile.TemporaryDirectory() as directory:
@@ -69,6 +75,46 @@ def test_another_artifact_cannot_inherit_the_frozen_seed_execution_provenance(se
                     physical.prepare(root/'absent-backup',backup_sha,root/'absent-packet',packet_sha)
             self.assertEqual([],list(root.iterdir()))
 
+    def test_historical_source_and_driver_are_not_relabelled_as_contract_acceptance(self):
+        seed=deepcopy(self.profile['baseline_seed_execution'])
+        result=physical.attribution(self.profile['producer_contract_commit'],seed)
+        self.assertEqual(self.profile['producer_contract_commit'],result['accepted_contract_source'])
+        self.assertEqual(seed,result['baseline_seed_execution'])
+        self.assertNotIn('producer_seed_source',result)
+        for field,value in (('source_commit',self.profile['producer_contract_commit']),
+                            ('driver_git_blob','0'*40),('driver_sha256','0'*64)):
+            changed=dict(seed,**{field:value})
+            with self.subTest(field=field), self.assertRaisesRegex(ValueError,'historical seed'):
+                physical.attribution(self.profile['producer_contract_commit'],changed)
+
+    def test_a_different_coherent_self_hashed_artifact_pair_refuses_at_preflight(self):
+        # This inert data models the preflight shape only; it is not an API seed
+        # or physical-upgrade proof. Neither schema CLI nor SDK is invoked.
+        with tempfile.TemporaryDirectory() as directory:
+            root=Path(directory);backup=root/'alternate.sqlite';packet_path=root/'alternate.json'
+            packet={'schema':'durable-workflow.command-key-physical-seed.v1',
+                'manifest_sha256':self.profile['producer_manifest_sha256'],
+                'expected_external_baseline':{'tag':self.profile['baseline_version'],
+                    'source_commit':self.profile['baseline_source'],'index_digest':self.profile['baseline_index']},
+                'cases':[]}
+            with closing(sqlite3.connect(backup)) as connection:
+                connection.execute('CREATE TABLE workflow_commands (id TEXT,command_type TEXT,workflow_instance_id TEXT,workflow_run_id TEXT,request_id TEXT)')
+                for kind in ('signal','update'):
+                    for state in ('open','completed','continued'):
+                        identity=kind+'-'+state
+                        receipt={'workflow_id':identity,'command_id':'command-'+identity,'run_id':'run-'+identity}
+                        if kind=='update':receipt.update(update_id='update-'+identity,accepted_at='2026-01-01T00:00:00Z',update_name='set')
+                        packet['cases'].append({'kind':kind,'state':state,'workflow_id':identity,
+                                                'receipt':receipt,'original_body':{'request_id':'key-'+identity}})
+                        connection.execute('INSERT INTO workflow_commands VALUES (?,?,?,?,?)',
+                            (receipt['command_id'],kind,identity,receipt['run_id'],None if kind=='signal' else 'key-'+identity))
+                connection.commit()
+            backup.chmod(0o600);packet_path.write_text(json.dumps(packet));packet_path.chmod(0o600)
+            backup_sha=hashlib.sha256(backup.read_bytes()).hexdigest()
+            packet_sha=hashlib.sha256(packet_path.read_bytes()).hexdigest()
+            with self.assertRaisesRegex(ValueError,'baseline provenance'):
+                physical.prepare(backup,backup_sha,packet_path,packet_sha)
+
     def test_original_business_row_or_catalog_change_refuses(self):
         changed=deepcopy(self.after);changed['tables']['business']['rows'][0][0]=['text','changed']
         with self.assertRaises(AssertionError):self.validate(changed)