test: qualify published workflow identity through cold workers

tests/fixtures/workflow_identity_consumer/README.md

@@ -0,0 +1,22 @@
+# Published workflow identity consumer
+
+This fixture normally installs the selected published Python SDK and compares
+its installed source with the exact immutable release. Set the required Compose
+environment values, use a uniquely named project and a disposable data/state
+directory, then start the stack.
+
+Run `/opt/venv/bin/python /fixture/consumer.py execute` in the client container,
+followed by the same command with `cold` in a separate process. Every original
+has a thirty-second whole observer and thirty-second run/execution deadlines.
+The first workflow node, activity node and cold workflow node use separate
+Worker instances and registrations. The final cold client verifies original
+typed results and unchanged retained histories.
+
+The eight originals cover default/named namespaces, literal numeric workflow
+IDs, activity suspension/cold replay, and continue-as-new successor identity.
+Continue-as-new result inspection explicitly pins the successor run; it does
+not treat an original run handle as a promise to follow a chain.
+
+Stop and audit the API/database, then remove the uniquely named project and all
+task-owned data, dependencies and raw output. The fixture does not qualify any
+native runtime or a broader SDK contract.

tests/fixtures/workflow_identity_consumer/compose.yml

@@ -0,0 +1,68 @@
+services:
+  bootstrap:
+    image: ${IDENTITY_SERVER_IMAGE:?Set the immutable published image}
+    user: "1000:1000"
+    cpus: 1
+    mem_limit: 512m
+    memswap_limit: 512m
+    command: [server-bootstrap]
+    environment: &server_environment
+      DB_CONNECTION: sqlite
+      DB_DATABASE: /data/database.sqlite
+      DB_SQLITE_JOURNAL_MODE: DELETE
+      CACHE_STORE: file
+      QUEUE_CONNECTION: database
+      DW_MODE: service
+      DW_POLLING_CACHE_PATH: /data/poll-cache
+      DW_AUTH_DRIVER: token
+      DW_AUTH_TOKEN: identity-client
+      DW_ADMIN_TOKEN: identity-admin
+      DW_WORKER_TOKEN: identity-worker
+      DW_BOOTSTRAP_RETRIES: 1
+    volumes: &server_volumes
+      - ${IDENTITY_PHASE_ROOT:?Set the disposable phase directory}/data:/data
+
+  api:
+    image: ${IDENTITY_SERVER_IMAGE:?}
+    user: "1000:1000"
+    cpus: 1
+    mem_limit: 512m
+    memswap_limit: 512m
+    environment: *server_environment
+    volumes: *server_volumes
+    depends_on:
+      bootstrap:
+        condition: service_completed_successfully
+
+  client:
+    image: ${IDENTITY_CLIENT_IMAGE:?Set the existing pinned Python image}
+    user: "1000:1000"
+    init: true
+    cpus: 1
+    mem_limit: 512m
+    memswap_limit: 512m
+    environment:
+      HOME: /tmp
+      SERVER_URL: http://api:8080
+      IDENTITY_COHORT: ${IDENTITY_COHORT:?Set a new cohort}
+      IDENTITY_SDK_VERSION: ${IDENTITY_SDK_VERSION:?Set the published version}
+    entrypoint: [sh, -ec]
+    command:
+      - |
+        python -m venv /opt/venv
+        /opt/venv/bin/pip install --index-url https://pypi.org/simple --no-cache-dir --quiet durable-workflow=="$$IDENTITY_SDK_VERSION"
+        exec sleep infinity
+    tmpfs:
+      - /opt/venv:rw,exec,uid=1000,gid=1000,size=268435456
+    healthcheck:
+      test: [CMD, /opt/venv/bin/python, -c, "import durable_workflow"]
+      interval: 2s
+      timeout: 5s
+      retries: 30
+    volumes:
+      - ${IDENTITY_PHASE_ROOT:?}/state:/state
+      - ${IDENTITY_EXPECTED_SOURCE:?Set the exact tagged package source}:/expected-source:ro
+      - ./consumer.py:/fixture/consumer.py:ro
+    depends_on:
+      api:
+        condition: service_healthy

tests/fixtures/workflow_identity_consumer/consumer.py

@@ -0,0 +1,168 @@
+"""Normally installed identity acceptance against a real published Server."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import importlib.metadata
+import json
+import os
+import sys
+import time
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import durable_workflow
+from durable_workflow import Client, Worker, activity, workflow
+
+STATE = Path('/state')
+URL = os.environ['SERVER_URL']
+COHORT = os.environ['IDENTITY_COHORT']
+
+
+def verify_install():
+    dist = importlib.metadata.distribution('durable-workflow')
+    assert dist.version == os.environ['IDENTITY_SDK_VERSION']
+    assert '/opt/venv/' in str(Path(durable_workflow.__file__).resolve())
+    hashes = []
+    for entry in dist.files or []:
+        if entry.hash is None:
+            continue
+        assert entry.hash.mode == 'sha256'
+        path = Path(dist.locate_file(entry))
+        digest = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest()).decode().rstrip('=')
+        assert digest == entry.hash.value, str(entry)
+        hashes.append(str(entry))
+    sources = list(Path('/expected-source').rglob('*.py'))
+    installed = Path(durable_workflow.__file__).parent
+    for source in sources:
+        assert source.read_bytes() == (installed / source.relative_to('/expected-source')).read_bytes(), str(source)
+    return {'version': dist.version, 'hashed_record_entries': len(hashes), 'matched_source_files': len(sources)}
+
+
+@activity.defn(name='qualification.identity-echo')
+def echo(value):
+    return value
+
+
+@workflow.defn(name='qualification.workflow-identity')
+class IdentityWorkflow:
+    def run(self, ctx, again=False, prior=None):
+        if again:
+            yield ctx.continue_as_new(False, [ctx.workflow_id, ctx.run_id])
+            return None
+        before = [ctx.workflow_id, ctx.run_id]
+        value = yield ctx.schedule_activity('qualification.identity-echo', [before + [9007199254740993, None, 'λ']])
+        return {'before': before, 'after': [ctx.workflow_id, ctx.run_id], 'prior': prior, 'echo': value}
+
+
+class ObservedClient(Client):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.registered = asyncio.Event()
+
+    async def register_worker(self, **kwargs):
+        reply = await super().register_worker(**kwargs)
+        self.registered.set()
+        return reply
+
+
+@asynccontextmanager
+async def node(namespace, queue, role):
+    async with ObservedClient(URL, token='identity-worker', namespace=namespace) as client:
+        worker = Worker(
+            client, task_queue=queue,
+            workflows=[IdentityWorkflow] if role != 'activity' else [],
+            activities=[echo] if role == 'activity' else [],
+            worker_id=f'{COHORT}-{namespace}-{queue}-{role}',
+            poll_timeout=1, heartbeat_interval=1,
+        )
+        task = asyncio.create_task(worker.run())
+        try:
+            await asyncio.wait_for(client.registered.wait(), 8)
+            yield
+        finally:
+            await worker.stop()
+            await asyncio.wait_for(task, 5)
+
+
+async def wait_event(client, workflow_id, event_type):
+    while True:
+        desc = await client.describe_workflow(workflow_id)
+        history = await client.get_history(workflow_id, desc.run_id)
+        if any(event['event_type'] == event_type for event in history['events']):
+            return desc.run_id, history['events']
+        await asyncio.sleep(0.1)
+
+
+async def original(namespace, workflow_id, again):
+    started = time.monotonic()
+    queue = 'identity-' + workflow_id
+    async with Client(URL, token='identity-client', namespace=namespace) as client:
+        handle = await client.start_workflow(
+            workflow_type='qualification.workflow-identity', task_queue=queue,
+            workflow_id=workflow_id, input=[again],
+            run_timeout_seconds=30, execution_timeout_seconds=30,
+        )
+        original_run_id = handle.run_id
+        async with node(namespace, queue, 'first'):
+            final_run_id, _ = await wait_event(client, workflow_id, 'ActivityScheduled')
+        async with node(namespace, queue, 'activity'):
+            await wait_event(client, workflow_id, 'ActivityCompleted')
+        async with node(namespace, queue, 'cold'):
+            result = await client.get_workflow_handle(workflow_id, run_id=final_run_id).result(timeout=20)
+        expected = [workflow_id, final_run_id]
+        assert result == {
+            'before': expected, 'after': expected,
+            'prior': [workflow_id, original_run_id] if again else None,
+            'echo': expected + [9007199254740993, None, 'λ'],
+        }, result
+        assert (final_run_id != original_run_id) is again
+        original_history = (await client.get_history(workflow_id, original_run_id))['events']
+        assert any(e['event_type'] == 'WorkflowContinuedAsNew' for e in original_history) is again
+        history = (await client.get_history(workflow_id, final_run_id))['events']
+        assert sum(e['event_type'] == 'ActivityStarted' for e in history) == 1
+        assert sum(e['event_type'] == 'WorkflowCompleted' for e in history) == 1
+        elapsed = time.monotonic() - started
+        assert elapsed < 30
+        return {'namespace': namespace, 'workflow_id': workflow_id, 'original_run_id': original_run_id,
+                'run_id': final_run_id, 'continued': again, 'result': result, 'elapsed_seconds': elapsed,
+                'history': history, 'original_history': original_history}
+
+
+async def main():
+    installation = verify_install()
+    stage = sys.argv[1]
+    path = STATE / 'originals.json'
+    if stage == 'execute':
+        async with Client(URL, token='identity-admin') as admin:
+            await admin.create_namespace(name='identity-named')
+        originals = []
+        path.write_text('[]')
+        for namespace in ['default', 'identity-named']:
+            prefix = '' if namespace == 'default' else 'named-'
+            for suffix, again in [('0', False), ('17', False), ('017', False), ('continue', True)]:
+                result = await asyncio.wait_for(original(namespace, prefix + suffix, again), 30)
+                originals.append(result)
+                path.write_text(json.dumps(originals, ensure_ascii=False, indent=2))
+                print(json.dumps({'workflow_id': result['workflow_id'], 'namespace': namespace,
+                                  'elapsed_seconds': result['elapsed_seconds'], 'passed': True}), flush=True)
+    elif stage == 'cold':
+        originals = json.loads(path.read_text())
+        assert len(originals) == 8
+        for record in originals:
+            async with Client(URL, token='identity-client', namespace=record['namespace']) as client:
+                handle = client.get_workflow_handle(record['workflow_id'], run_id=record['run_id'])
+                assert await handle.result(timeout=5) == record['result']
+                assert (await handle.get_history())['events'] == record['history']
+                assert (await client.get_history(
+                    record['workflow_id'], record['original_run_id']
+                ))['events'] == record['original_history']
+    else:
+        raise ValueError('Choose execute or cold.')
+    assert verify_install() == installation
+    print(json.dumps({'stage': stage, 'installation': installation, 'originals': len(originals)}), flush=True)
+
+
+asyncio.run(main())