test: exercise legacy ambiguity and committed response loss

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, '<accepted-committed-response-loss>', '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')

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,16 +21,33 @@
 sys.path.insert(0, str(SOURCE))
 import sqlite_legacy as legacy
 import execution_guards as guards
+import response_loss
 
 
 
 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,