diff --git a/scripts/acceptance/command-keys/response_loss.py b/scripts/acceptance/command-keys/response_loss.py new file mode 100644 index 0000000..e548196 --- /dev/null +++ b/scripts/acceptance/command-keys/response_loss.py @@ -0,0 +1,127 @@ +"""Installed Rust caller of the accepted shared committed-response-loss fixture.""" +import concurrent.futures +from datetime import datetime, timezone +import hashlib +import json +import os +import stat +import threading +import types + +import sqlite_legacy + + +FIXTURE_COMMIT = '117118c83093822469b3824ba9587ff4068ed29f' +FIXTURE_SHA256 = 'bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725' + + +def fixture(path): + """Execute only the exact accepted source bytes, without a second path read.""" + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + info = os.fstat(descriptor) + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() + or info.st_nlink != 1 or info.st_mode & 0o022 or not 0 < info.st_size <= 64 * 1024): + raise ValueError('loss fixture must be an owned bounded single-link regular file') + source = os.pread(descriptor, 64 * 1024 + 1, 0) + if hashlib.sha256(source).hexdigest() != FIXTURE_SHA256: + raise ValueError('loss fixture differs from accepted public source') + finally: + os.close(descriptor) + module = types.ModuleType('accepted_committed_response_loss') + exec(compile(source, '', 'exec'), module.__dict__) + return module + + +def rows(snapshot, table): + data = snapshot['tables'][table] + return [dict(zip(data['columns'], (value[1] if len(value) == 2 else None for value in row))) + for row in data['rows']] + + +def committed_identity(before, kind, workflow_id, run_id, request_id): + token = kind + '-v1:' + hashlib.sha256(request_id.encode()).hexdigest() + commands = [row for row in rows(before, 'workflow_commands') + if row['workflow_instance_id'] == workflow_id and row['workflow_run_id'] == run_id + and row['command_type'] == kind and row['request_id'] == token] + assert len(commands) == 1 and commands[0]['id'] + original = {'command_id':commands[0]['id'], 'run_id':run_id} + if kind == 'update': + updates = [row for row in rows(before, 'workflow_updates') + if row['workflow_command_id'] == original['command_id']] + assert len(updates) == 1 and updates[0]['id'] and updates[0]['accepted_at'] + assert updates[0]['workflow_instance_id'] == workflow_id and updates[0]['workflow_run_id'] == run_id + original.update(update_id=updates[0]['id'], accepted_at=updates[0]['accepted_at']) + return original + + +def timestamp(value, *, sqlite_utc=False): + result = datetime.fromisoformat(value.replace('Z', '+00:00')) + if result.tzinfo is None: + # Native SQLite stores UTC without a zone; wire receipts must carry one. + assert sqlite_utc + result = result.replace(tzinfo=timezone.utc) + return result.astimezone(timezone.utc) + + +def run(path, origin, workflow_id, run_id, invoke, snapshot, work): + """No worker, transform or upgrade: prove committed loss and explicit retries.""" + module = fixture(path) + proxy = module.CommittedResponseLoss(origin) + thread = threading.Thread(target=proxy.serve_forever, kwargs={'poll_interval':0.01}, daemon=True) + thread.start() + cases = [] + try: + for kind, operation in [('signal','touch'), ('update','mutate')]: + request_id = 'committed-loss-' + kind + args = {'op':kind, 'workflow_id':workflow_id, 'name':operation, + 'input':[1], 'request_id':request_id} + route = '/api/workflows/' + workflow_id + '/' + kind + '/' + operation + proxy.arm('POST', route) + lost = invoke(args, proxy.url) + assert lost['kind'] == 'transport', lost + dropped = [row for row in proxy.records if row['path'] == route and row['dropped']] + assert len(dropped) == 1 and dropped[0]['method'] == 'POST' + assert dropped[0]['upstream_status'] == 202 and dropped[0]['upstream_complete'] is True + assert 0 < dropped[0]['upstream_body_bytes'] <= module.MAX_RESPONSE_BYTES + # Capture the committed identity before any same-key retry. The + # accepted proxy retains hashes/metadata only, never response bodies. + before = snapshot() + original = committed_identity(before, kind, workflow_id, run_id, request_id) + exported = invoke({'op':'history','workflow_id':workflow_id,'run_id':run_id}, origin) + assert exported['kind'] == 'ok' + events = exported['value']['history_events'] + event_type = 'SignalReceived' if kind == 'signal' else 'UpdateAccepted' + assert sum(event['type'] == event_type for event in events) == 1 + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + replies = list(pool.map(lambda _:invoke(args, proxy.url), range(4))) + for reply in replies: + assert reply['kind'] == ('ok' if kind == 'signal' else 'pending'), reply + receipt = reply['value'] if kind == 'signal' else reply['receipt'] + assert receipt['retained'] is True and receipt['intent_verified'] is True + assert receipt['request_id'] == request_id + for field,value in original.items(): + if field == 'accepted_at': + assert timestamp(receipt[field]) == timestamp(value, sqlite_utc=True) + else: + assert value and receipt[field] == value + conflict = invoke(dict(args,input=[True]), origin) + assert conflict['kind'] == 'http' and conflict['status'] == 409 and conflict['conflict'] is True + assert conflict['request_id'] == request_id and conflict['body']['reason'] == 'request_id_conflict' + assert invoke({'op':'history','workflow_id':workflow_id,'run_id':run_id}, origin)['value']['history_events'] == events + after = snapshot() + assert after == before, 'postcommit retries/conflict changed logical rows/schema' + for suffix,value in [('before',before), ('after',after)]: + with open(os.open(work / (kind+'-loss-'+suffix+'.json'), + os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600),'w') as output: + json.dump(value,output,sort_keys=True) + cases.append({'command_type':kind,'outcome':'pass','original':original, + 'returned':{field:receipt[field] for field in original},'concurrency':4, + 'retained':True,'intent_verified':True, + 'upstream':dropped[0],'zero_write_sha256':sqlite_legacy.fingerprint(before)}) + return {'cell':'postcommit-response-loss-only','cases':cases, + 'fixture_commit':FIXTURE_COMMIT,'fixture_sha256':FIXTURE_SHA256} + finally: + proxy.shutdown(); proxy.server_close(); thread.join(timeout=5) + if thread.is_alive(): + raise ValueError('loss fixture failed to stop') diff --git a/scripts/acceptance/command-keys/run_open.py b/scripts/acceptance/command-keys/run_open.py index 5b49814..c75d99c 100644 --- a/scripts/acceptance/command-keys/run_open.py +++ b/scripts/acceptance/command-keys/run_open.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""One synthetic legacy state cell; no physical upgrade or response-loss authority.""" +"""One named command-key acceptance cell, with separate evidence scopes.""" import argparse import re import tomllib @@ -21,6 +21,7 @@ sys.path.insert(0, str(SOURCE)) import sqlite_legacy as legacy import execution_guards as guards +import response_loss @@ -28,9 +29,25 @@ def capture(*args): return subprocess.check_output(args, text=True, timeout=30).strip() +def ambiguity_receipt(result, request_id, expected): + assert result['kind'] == 'ok' and result['value']['status'] == expected['status'], result + body = result['value']['body'] + assert body['reason'] == expected['reason'], body + if 'request_id' in body: + assert body['request_id'] == request_id, body + for field in ('retained', 'intent_verified', 'retryable'): + assert body[field] is expected[field], body + for field in ('command_id', 'run_id', 'update_id', 'accepted_at', 'retained_command_id', 'receipt'): + assert body.get(field) is None, body + return body + + def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--state', choices=['open', 'completed', 'continued'], default='open') + 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('--work-dir',type=Path,required=True) parser.add_argument('--consumer-binary',type=Path,required=True) parser.add_argument('--consumer-lock',type=Path,required=True) @@ -42,6 +59,10 @@ def main(): parser.add_argument('--server-version',required=True) parser.add_argument('--server-source',required=True) options=parser.parse_args() + 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'): + parser.error('postcommit loss uses its own open-state admissions without a legacy transform') if not __debug__: parser.error('run conformance assertions without Python optimization') if (not re.fullmatch(r'sha256:[0-9a-f]{64}',options.server_index) @@ -61,6 +82,8 @@ def main(): build, build_receipt_sha = guards.verify_build_receipt( options.build_receipt, SOURCE, options.consumer_binary, options.consumer_lock, 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. 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 @@ -117,9 +140,9 @@ def stop(): capture('docker','rm','-v',owned_id) owned_id=None - def invoke(args): + def invoke(args, base=None): env={'PATH':os.environ['PATH'],'TMPDIR':str(work), - 'DURABLE_WORKFLOW_LOCAL_ISOLATED':'1','DURABLE_WORKFLOW_SERVER_URL':origin, + 'DURABLE_WORKFLOW_LOCAL_ISOLATED':'1','DURABLE_WORKFLOW_SERVER_URL':base or origin, 'DURABLE_WORKFLOW_AUTH_TOKEN':token} result=binary.run(input=json.dumps(args), env=env, timeout=20) return json.loads(result.stdout) @@ -140,15 +163,41 @@ def snapshot(path): ok({'op':'register','queue':queue,'null_update':True}) workflow_type = 'keys.continue' if options.state == 'continued' else 'keys.wait' created=ok({'op':'start','workflow_type':workflow_type,'workflow_id':queue,'queue':queue,'input':[0]}) + if options.response_loss_fixture is not None: + cell = response_loss.run(options.response_loss_fixture, origin, queue, created['run_id'], + invoke, lambda:snapshot(database/'source.sqlite'), work) + stop() + binary.check() + cell.update(outcome='pass', started_at=started_at, + finished_at=datetime.now(timezone.utc).isoformat(), + 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(), + loss_adapter_sha256=hashlib.sha256((SOURCE/'response_loss.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 seed={};selections=[] + caller_key = document['keys']['case_exact'] if options.scenario == 'case-only' else document['keys']['original'] for kind,operation in [('signal','touch'),('update','mutate')]: - admission={'op':kind+'_raw','workflow_id':queue,'name':operation,'input':[document['legacy_readback']['original_input']],'request_id':document['keys']['original']} + admission={'op':kind+'_raw','workflow_id':queue,'name':operation,'input':[document['legacy_readback']['original_input']],'request_id':caller_key} receipt=ok(admission);assert receipt['status']==202,receipt body=receipt['body'];assert body['command_id'] and body['run_id']==created['run_id'] seed[kind]=(admission,body) - selections.append({'command_id':body['command_id'],'command_type':kind, + selected = {'command_id':body['command_id'],'command_type':kind, 'workflow_instance_id':queue,'workflow_run_id':body['run_id'], - 'caller_key':document['keys']['original'],'legacy_request_id':document['keys']['original']}) + 'caller_key':caller_key,'legacy_request_id':caller_key} + if options.scenario == 'two-rows': + second = ok(dict(admission, input=[2], request_id=document['keys']['second'])) + assert second['status'] == 202 and second['body']['run_id'] == created['run_id'] + assert second['body']['command_id'] and second['body']['command_id'] != body['command_id'] + selected.update(command_id=second['body']['command_id'], caller_key=document['keys']['second']) + elif options.scenario == 'legacy-token': + selected['legacy_request_id'] = document['keys']['tokens'][kind] + if options.scenario != 'modern-raw': + selections.append(selected) successor_run_id = None if options.state != 'open': # Use the same held descriptor and pre/post checks for the worker. @@ -185,14 +234,40 @@ def snapshot(path): assert 'command_type' in columns catalog=connection.execute("SELECT sql FROM sqlite_schema WHERE name='workflow_commands'").fetchone()[0] (work/'producer-command-catalog.sql').write_text(catalog+';\n') - transform=legacy.transform_copy(source,copied,selections,stopped=True) - print(json.dumps({'stage':'real-api-copy-transformed','catalog_column':'command_type', + if options.scenario == 'modern-raw': + # The accepted vector explicitly preserves the modern admission. + # No synthetic mutation or physical-upgrade authority is claimed. + copied = source + digest = legacy.fingerprint(snapshot(source)) + transform = {'cell':'modern-raw-unchanged', 'transformed':0, + 'manifest_sha256':legacy.MANIFEST_SHA256, + 'before_sha256':digest, 'after_sha256':digest} + else: + transform=legacy.transform_copy(source,copied,selections,stopped=True) + print(json.dumps({'stage':('real-api-modern-admission-retained' if options.scenario == 'modern-raw' else 'real-api-copy-transformed'),'catalog_column':'command_type', 'transform':transform}),flush=True) - start('legacy.sqlite') + start(copied.name) before=snapshot(copied) guards.verify_restart_baseline(before, transform['after_sha256']) for kind,(admission,original) in seed.items(): retry=dict(admission,input=[document['legacy_readback']['retry_input']]) + if options.scenario in ('two-rows', 'legacy-token', 'modern-raw'): + if options.scenario == 'modern-raw': + retry['request_id'] = document['keys']['tokens'][kind] + with concurrent.futures.ThreadPoolExecutor(max_workers=document['legacy_readback']['concurrency']) as pool: + replies=list(pool.map(lambda _:invoke(retry),range(document['legacy_readback']['concurrency']))) + for reply in replies: + ambiguity_receipt(reply, retry['request_id'], document['ambiguity_expected']) + high = invoke(dict(retry,op=kind)) + assert high['kind'] == 'http' and high['status'] == 409, high + assert high['request_id'] == high['body'].get('request_id'), high + ambiguity_receipt({'kind':'ok','value':{'status':high['status'],'body':high['body']}}, + retry['request_id'], document['ambiguity_expected']) + print(json.dumps({'stage':'ambiguity-readback','shape':options.scenario, + 'command_type':kind,'outcome':'pass','selected_receipt':False, + 'retained':True,'intent_verified':False,'retryable':False, + 'concurrency':document['legacy_readback']['concurrency']}),flush=True) + continue fields = ['command_id', 'run_id'] + (['update_id', 'accepted_at'] if kind == 'update' else []) assert all(original[field] for field in fields) with concurrent.futures.ThreadPoolExecutor(max_workers=document['legacy_readback']['concurrency']) as pool: @@ -231,13 +306,34 @@ def snapshot(path): 'concurrency':document['legacy_readback']['concurrency']}),flush=True) after=snapshot(copied) assert after==before,'readbacks or conflicts changed complete logical database rows/schema' + case_admissions = [] + if options.scenario == 'case-only': + # SQLite's binary comparison admits the different byte string. + # Zero-write proof above applies only to exact-byte legacy retries. + for kind,(admission,original) in seed.items(): + fresh = ok(dict(admission, request_id=document['keys']['case_other'])) + assert fresh['status'] == document['case_only']['sqlite_postgresql_binary']['status'], fresh + body = fresh['body'] + assert body['command_id'] and body['command_id'] != original['command_id'], body + assert body['run_id'] == original['run_id'] and body['request_id'] == document['keys']['case_other'] + case_admissions.append({'command_type':kind,'original_command_id':original['command_id'], + 'distinct_command_id':body['command_id'],'run_id':body['run_id']}) + admitted=snapshot(copied) + assert admitted != after + commands = after['tables']['workflow_commands'] + assert all(row in admitted['tables']['workflow_commands']['rows'] for row in commands['rows']) + with open(os.open(work/'case-admitted-snapshot.json',os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600),'w') as file: + json.dump(admitted,file,sort_keys=True) for filename, value in [('before-snapshot.json', before), ('after-snapshot.json', after)]: with open(os.open(work / filename, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600), 'w') as file: json.dump(value, file, sort_keys=True) stop() binary.check() - print(json.dumps({'cell':'synthetic-legacy-'+options.state+'-only','outcome':'pass', + cell = options.state if options.scenario == 'readback' else options.scenario + print(json.dumps({'cell':'synthetic-legacy-'+cell+'-only','outcome':'pass', 'state':options.state,'successor_run_id':successor_run_id, + 'scenario':options.scenario,'case_distinct_admissions':case_admissions, + 'zero_write_scope':'before/after exact legacy readbacks or ambiguity only; excludes case-distinct admissions', 'started_at':started_at,'finished_at':datetime.now(timezone.utc).isoformat(), 'zero_write_sha256':legacy.fingerprint(before),'server_source':options.server_source, 'server_index':options.server_index,'server_version':options.server_version, diff --git a/scripts/acceptance/command-keys/synthetic-sqlite.md b/scripts/acceptance/command-keys/synthetic-sqlite.md index a41de72..cc1a2f0 100644 --- a/scripts/acceptance/command-keys/synthetic-sqlite.md +++ b/scripts/acceptance/command-keys/synthetic-sqlite.md @@ -148,6 +148,49 @@ while retries and explicit original-run reads retain the original command/run. The complete database comparison includes the successor, so these readbacks cannot add executions or durable rows there. -Ambiguity and case controls, actual physical upgrade and post-commit loss still -require their own named results. Python +For a separate open-state ambiguity or case cell, add `--scenario` with one +of these accepted manifest shapes. Other states refuse these scenarios before +any runtime effect. + +| Scenario | Setup and required result | +| --- | --- | +| `two-rows` | Admit both original and second keys through the API. Transform only the second command to the original raw key. Both command types refuse four concurrent retries and the high-level call with409 `request_id_ambiguous`, retained/unverified/nonretryable markers and no selected receipt. | +| `legacy-token` | Transform the original command to the manifest's literal stored-token string and remove its intent. The original caller key has the same ambiguity refusal. | +| `modern-raw` | Keep the modern original admission unchanged, stop/restart its owned database without a transform, and retry the literal stored-token string. The receipt explicitly reports zero transforms. | +| `case-only` | Transform the exact case-sensitive key, prove exact-byte readbacks and conflicts are zero-write, then submit the differently cased key. SQLite admits a distinct new command for each type. Its admissions are separately recorded and explicitly excluded from the zero-write scope. Other backends are not inferred. | + +Ambiguity cells compare all logical rows/schema across the raw and high-level retries. +The case-only cell retains a third private snapshot after distinct admissions +and requires every preexisting command row to remain present unchanged. +Actual physical upgrade and post-commit loss still require their own named +results. Python optimization is refused because it would disable conformance assertions. + +## Run the separate committed-response-loss cell + +Obtain the accepted shared +[`committed_response_loss.py`](https://github.com/durable-workflow/.github/blob/117118c83093822469b3824ba9587ff4068ed29f/conformance/command-keys/committed_response_loss.py) +at commit `117118c83093822469b3824ba9587ff4068ed29f`, SHA-256 +`bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725`. +Use the same build and runtime arguments above with +`--response-loss-fixture "$ACCEPTED_LOSS_FIXTURE"`, keeping state `open`. +The runner refuses changed source before runtime effects and executes only +the verified source buffer. No proxy is reimplemented or modified here. + +This creates a separate fresh runtime and real signal/update admissions through +the registry SDK. The accepted proxy must consume a complete successful upstream +response before closing the client socket. The client must observe a transport +error. Before the first retry, the runner captures the actual committed command +and related update identity from a complete read-only logical snapshot and +checks one corresponding durable history event. Four explicit same-key retries +must retain those identities with verified intent. Integer-to-boolean changed +intent must fail with the typed409 conflict and exact caller key. All rows/schema +and history must remain unchanged across retries/conflict. The fixture retains +only bounded metadata and body hashes. + +The terminal `postcommit-response-loss-only` receipt binds the accepted fixture, +public adapter and recorded executable/build identities, original/returned +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. diff --git a/scripts/acceptance/command-keys/test_ambiguity_receipt.py b/scripts/acceptance/command-keys/test_ambiguity_receipt.py new file mode 100644 index 0000000..18dd35b --- /dev/null +++ b/scripts/acceptance/command-keys/test_ambiguity_receipt.py @@ -0,0 +1,48 @@ +"""An ambiguity refusal must not silently select a retained command receipt.""" +from copy import deepcopy +import unittest + +import run_open +import sqlite_legacy + + +class AmbiguityReceiptTest(unittest.TestCase): + def setUp(self): + self.expected = sqlite_legacy.manifest()['ambiguity_expected'] + self.result = {'kind':'ok','value':{'status':409,'body':{ + 'request_id':'synthetic-key','reason':'request_id_ambiguous', + 'retained':True,'intent_verified':False,'retryable':False}}} + + def test_exact_unverified_nonretryable_refusal_accepts(self): + self.assertEqual(self.result['value']['body'], run_open.ambiguity_receipt( + self.result, 'synthetic-key', self.expected)) + # The frozen ambiguity contract does not require an echoed caller key. + # Error::request_id() likewise reads only a key actually echoed by HTTP. + absent = deepcopy(self.result) + del absent['value']['body']['request_id'] + self.assertEqual(absent['value']['body'], run_open.ambiguity_receipt( + absent, 'synthetic-key', self.expected)) + + def test_selected_original_or_alternative_receipt_refuses(self): + for field in ('command_id','run_id','update_id','accepted_at','retained_command_id','receipt'): + result = deepcopy(self.result) + result['value']['body'][field] = 'selected-identity' + with self.subTest(field=field), self.assertRaises(AssertionError): + run_open.ambiguity_receipt(result, 'synthetic-key', self.expected) + + def test_success_conflict_wrong_key_or_verified_intent_cannot_qualify_as_ambiguous(self): + for field,value in (('reason','request_id_conflict'), ('request_id','other-key'), + ('retained',False), ('retained',1), ('intent_verified',True), + ('intent_verified',0), ('retryable',True)): + result = deepcopy(self.result) + result['value']['body'][field] = value + with self.subTest(field=field,value=value), self.assertRaises(AssertionError): + run_open.ambiguity_receipt(result, 'synthetic-key', self.expected) + result = deepcopy(self.result) + result['value']['status'] = 202 + with self.assertRaises(AssertionError): + run_open.ambiguity_receipt(result, 'synthetic-key', self.expected) + + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/acceptance/command-keys/test_response_loss.py b/scripts/acceptance/command-keys/test_response_loss.py new file mode 100644 index 0000000..dec11f3 --- /dev/null +++ b/scripts/acceptance/command-keys/test_response_loss.py @@ -0,0 +1,72 @@ +"""Inert evidence/refusal controls for the registry Rust loss adapter.""" +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +import tempfile +import unittest + +import response_loss + + +def table(columns, values): + return {'columns':columns, 'rows':[[['text',value] for value in row] for row in values]} + + +class LossEvidenceTest(unittest.TestCase): + def setUp(self): + self.snapshot = {'tables':{ + 'workflow_commands':table( + ['id','workflow_instance_id','workflow_run_id','command_type','request_id'], + [['command','workflow','run','update','update-v1:' + response_loss.hashlib.sha256(b'key').hexdigest()]]), + 'workflow_updates':table( + ['id','workflow_command_id','workflow_instance_id','workflow_run_id','accepted_at'], + [['update','command','workflow','run','2026-01-01 00:00:00.123000']])}} + + def identity(self, value): + return response_loss.committed_identity(value, 'update','workflow','run','key') + + def test_before_retry_identity_joins_exact_command_run_token_and_update(self): + self.assertEqual({'command_id':'command','run_id':'run','update_id':'update', + 'accepted_at':'2026-01-01 00:00:00.123000'}, self.identity(self.snapshot)) + + def test_absent_duplicate_or_foreign_rows_refuse_an_original_identity(self): + for target,column in [('workflow_commands','workflow_run_id'),('workflow_commands','request_id'), + ('workflow_updates','workflow_command_id'),('workflow_updates','workflow_run_id')]: + value=deepcopy(self.snapshot) + columns=value['tables'][target]['columns'] + value['tables'][target]['rows'][0][columns.index(column)]=['text','foreign'] + with self.subTest(target=target,column=column), self.assertRaises(AssertionError): + self.identity(value) + for target in ('workflow_commands','workflow_updates'): + value=deepcopy(self.snapshot) + value['tables'][target]['rows'] *= 2 + with self.subTest(target=target), self.assertRaises(AssertionError): + self.identity(value) + + def test_wire_timestamp_requires_zone_and_sqlite_utc_precision_is_preserved(self): + expected=datetime(2026,1,1,0,0,0,123000,tzinfo=timezone.utc) + self.assertEqual(expected,response_loss.timestamp('2026-01-01T00:00:00.123Z')) + self.assertEqual(expected,response_loss.timestamp('2026-01-01 00:00:00.123000',sqlite_utc=True)) + with self.assertRaises(AssertionError): + response_loss.timestamp('2026-01-01 00:00:00.123000') + + def test_changed_fixture_refuses_before_code_execution(self): + with tempfile.TemporaryDirectory() as directory: + root=Path(directory) + marker=root/'executed' + fixture=root/'fixture.py' + fixture.write_text('open('+repr(str(marker))+',"w").write("executed")\n') + fixture.chmod(0o600) + with self.assertRaisesRegex(ValueError,'differs'): + response_loss.fixture(fixture) + self.assertFalse(marker.exists()) + link=root/'link.py';link.symlink_to(fixture) + with self.assertRaises(OSError): + response_loss.fixture(link) + link.unlink();link.hardlink_to(fixture) + with self.assertRaisesRegex(ValueError,'single-link'): + response_loss.fixture(fixture) + + +if __name__ == '__main__': + unittest.main()