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..879e367 --- /dev/null +++ b/scripts/acceptance/command-keys/physical-sqlite-profile.json @@ -0,0 +1,78 @@ +{ + "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", + "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_contract_commit": "a09f156399416216d2f097c257a2164f83c7eda7", + "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..f09da2d --- /dev/null +++ b/scripts/acceptance/command-keys/physical_upgrade.py @@ -0,0 +1,207 @@ +"""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 = '8f9c4695b814a6c3f923885e51d1900506906138b7e0d27dc87b8df592a36e9e' + + +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 require_handoff(commit): + selected = profile() + if commit != selected['producer_contract_commit']: + raise ValueError('producer handoff commit differs from the reviewed physical profile') + 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 + 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() + 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) + 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..db84e64 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 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) @@ -59,6 +65,17 @@ 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 physical_mode: + 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'): @@ -84,6 +101,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 +125,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 +183,52 @@ 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(**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, + 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..4fa596a 100644 --- a/scripts/acceptance/command-keys/synthetic-sqlite.md +++ b/scripts/acceptance/command-keys/synthetic-sqlite.md @@ -194,3 +194,60 @@ 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. +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: + +```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..2377dc0 --- /dev/null +++ b/scripts/acceptance/command-keys/test_physical_upgrade.py @@ -0,0 +1,162 @@ +"""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 +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_another_well_formed_handoff_sha_refuses_before_input_or_runtime_effects(self): + self.assertEqual(self.profile,physical.require_handoff(self.profile['producer_contract_commit'])) + 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: + 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_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_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) + 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()