Add real published Waterline entry-refusal and selected-run repair acceptance

tests/fixtures/entry_recovery_http/README.md

@@ -0,0 +1,89 @@
+# Published entry-recovery Waterline fixture
+
+This revision uses normally installed Python SDK 2.8.0, published Server 2.10.1
+and published Waterline 2.12.1. It extends the earlier HTTP fixture below with
+real browser interaction; no route or response is mocked. Run only
+`required-before` and `required-after` for this two-original UI cohort. The first
+uses desktop/dark and the second mobile/light. The real selected-run Repair
+action replaces the fixture's direct repair request. The original deadlines and
+same-run assertions are unchanged. The stopped-state audit expects two completed
+originals, two deliberate repairs and no registered workers.
+
+Use `mcr.microsoft.com/playwright/python:v1.55.0-noble` for the client/browser
+and install `requirements.txt` normally into `/opt/venv`. The published SDK wheel
+SHA-256 is `908267666c35890fce13ab9c8e62a5c93abac78bdffbb0e56423eb3b3e57a77b`.
+Mount these fixture files individually under `/phase`, with only the disposable
+`data` and `consumer` directories writable. Keep SDK source off `PYTHONPATH`.
+The packaged viewer resolves as `http://viewer:8080/waterline`, uses the same
+synthetic API credentials and must be configured with
+`WATERLINE_ACCESS_MODE=operator`. Require both services and the browser to be
+ready before admitting either original. Preserve concise failed observations;
+retire images pulled solely for the phase, browsers, dependencies and all raw
+state immediately after review.
+
+## Historical HTTP cohort
+
+This is the five-original installed-package acceptance fixture for
+[shared issue 221](https://github.com/durable-workflow/.github/issues/221).
+It uses a normal PyPI installation, not this checkout's runtime code.
+
+The recorded cohort uses Python SDK 2.6.2 (`1b03ed2a`) and Server 2.9.9,
+image index `sha256:d45d197e9556292ec096531030717e43d69ea4ea9ab197170971492f14cc0869`.
+That Server image is historical: use synthetic data on an isolated internal
+Docker network without published host ports when reproducing this exact tuple.
+Production deployments should use a current patched Server release.
+
+## Preparation
+
+Copy these files into a disposable directory, mounted as `/phase` in the client,
+and create its `data` and `consumer` subdirectories. The API resolves as
+`http://api:8080` on their private network. Mount only `data` at `/data` in the
+published API image, preserving its packaged bootstrap/seeder files.
+
+Bootstrap a fresh SQLite database with `DB_CONNECTION=sqlite`,
+`DB_DATABASE=/data/database.sqlite`, `CACHE_STORE=file`,
+`QUEUE_CONNECTION=database`, `DW_POLLING_CACHE_PATH=/data/poll-cache`,
+`DW_AUTH_TOKEN=general221-fixture`, `DW_ADMIN_TOKEN=general221-fixture`, and
+`DW_WORKER_TOKEN=general221-worker`. These are synthetic fixture credentials.
+Run the image's `server-bootstrap` and its normal Apache entry point.
+Require `/api/ready` to return 200 before business admission.
+
+Install `requirements.txt` into `/opt/venv` in the client. Save the PyPI wheel
+as `/phase/sdk.whl`; its SHA-256 is
+`28379c16bcd598c9aa545071d074bee297e449158a550e580d1df0ef9fc27ab8`.
+Run `verify_install.py` before and after the cohort. It checks the installed
+production files against both their RECORD hashes and the published wheel.
+Do not put an SDK checkout on `PYTHONPATH`. An executable tmpfs is needed if
+the venv, including extension modules, lives in tmpfs.
+
+## Execute and inspect
+
+For each of `required-before`, `required-after`, `optional`, `body-before`, and
+`body-after`, use a fresh cohort identifier and run:
+
+```sh
+EXPECTED_BINDING_REFUSAL=1 /opt/venv/bin/python /phase/consumer.py CASE COHORT
+```
+
+Every original retains a 30-second run and 60-second execution deadline, and
+each observer has its original 30-second total bound. The two changed required
+signatures must refuse without entering the body. Supported run-scoped repair
+and compatible code must finish the same original with unchanged history
+prefix and deadlines. The optional argument completes normally; genuine body
+TypeErrors remain terminal business failures. The fixture records every
+business Worker exit and actual task submission, including failure reason.
+
+For each completed original, start a separate process with
+`/opt/venv/bin/python /phase/cold.py WORKFLOW_ID RUN_ID`. Its result must retain
+the exact signed-int64 maximum, null and Unicode values.
+
+Stop the API before SQL inspection. Run `audit.py` with the entire phase mounted
+kernel read-only at `/phase`, a disposable writable `/audit`, no network and no
+SDK dependency. It copies the stopped database/WAL, opens that copy read-only,
+checks five originals and two deliberate repairs, and requires source and
+copied database/WAL bytes to stay exact. Only a transient SQLite index may be
+created in the disposable copy. Its report is written to stdout.
+
+Retain concise results and source identities in the owning issue. Remove all
+owned containers, network, dependencies, database, wheel and raw traces after
+inspection; preserve borrowed images and other owners' resources.

tests/fixtures/entry_recovery_http/audit.py

@@ -0,0 +1,89 @@
+import datetime
+import hashlib
+import json
+import os
+import shutil
+import sqlite3
+from pathlib import Path
+
+root = Path('/phase')
+path = root / 'data/database.sqlite'
+files = [p for p in (path, Path(str(path) + '-wal')) if p.exists()]
+hashes = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in files}
+copy = Path('/audit')
+for source in files:
+    shutil.copyfile(source, copy / source.name)
+db = sqlite3.connect('file:' + str(copy / path.name) + '?mode=ro', uri=True)
+db.row_factory = sqlite3.Row
+db.execute('PRAGMA query_only=ON')
+results = []
+for evidence_path in sorted((root / 'consumer').glob('*/evidence.json')):
+    evidence = json.loads(evidence_path.read_text())
+    run_id = evidence['after']['run_id']
+    run = dict(db.execute('SELECT * FROM workflow_runs WHERE id=?', (run_id,)).fetchone())
+    tasks = [dict(row) for row in db.execute(
+        'SELECT id,status,attempt_count,repair_count FROM workflow_tasks WHERE workflow_run_id=? ORDER BY id',
+        (run_id,),
+    )]
+    history = [dict(row) for row in db.execute(
+        'SELECT sequence,event_type FROM workflow_history_events WHERE workflow_run_id=? ORDER BY sequence', (run_id,),
+    )]
+    assert run['status'] == evidence['after']['status']
+    assert [(r['sequence'], r['event_type']) for r in history] == [
+        (r['sequence'], r['event_type']) for r in evidence['history']['events']
+    ]
+    assert evidence['prefix']['events'] == evidence['history']['events'][:len(evidence['prefix']['events'])]
+    started = datetime.datetime.fromisoformat(run['started_at'])
+    assert (datetime.datetime.fromisoformat(run['run_deadline_at']) - started).total_seconds() == 30
+    assert (datetime.datetime.fromisoformat(run['execution_deadline_at']) - started).total_seconds() == 60
+    assert started.microsecond != 0
+    if run['status'] == 'completed':
+        assert all(task['status'] == 'completed' for task in tasks), tasks
+    else:
+        assert all(task['status'] in ('completed', 'failed') for task in tasks), tasks
+        assert sum(row['event_type'] == 'WorkflowFailed' for row in history) == 1
+    repaired = evidence.get('refusal') is not None
+    assert sum(task['repair_count'] for task in tasks) == int(repaired), tasks
+    assert sum(task['attempt_count'] - 1 for task in tasks) == int(repaired), tasks
+    assert sum(row['event_type'] == 'WorkflowCompleted' for row in history) == int(run['status'] == 'completed')
+    if repaired:
+        assert not evidence['refusal']['is_terminal']
+        assert not any(row['event_type'] == 'WorkflowFailed' for row in history)
+    assert all(code == 0 for code in evidence['worker_exits'])
+    results.append({'case': evidence['case'], 'cohort': evidence['cohort'], 'run_id': run_id,
+                    'status': run['status'], 'tasks': len(tasks), 'repairs': int(repaired),
+                    'events': len(history), 'worker_exits': evidence['worker_exits'],
+                    'run_deadline_at': run['run_deadline_at'], 'execution_deadline_at': run['execution_deadline_at']})
+assert len(results) == 2, results
+failed_originals = []
+for original_path in sorted((root / 'consumer').glob('*/original.json')):
+    if (original_path.parent / 'evidence.json').exists():
+        continue
+    original = json.loads(original_path.read_text())
+    run_id = original['before']['run_id']
+    run = db.execute('SELECT status FROM workflow_runs WHERE id=?', (run_id,)).fetchone()
+    tasks = [dict(row) for row in db.execute(
+        'SELECT status,attempt_count,repair_count FROM workflow_tasks WHERE workflow_run_id=?', (run_id,),
+    )]
+    history = [row[0] for row in db.execute(
+        'SELECT event_type FROM workflow_history_events WHERE workflow_run_id=? ORDER BY sequence', (run_id,),
+    )]
+    assert run['status'] == 'pending', (run_id, dict(run))
+    assert tasks == [{'status': 'failed', 'attempt_count': 1, 'repair_count': 0}], tasks
+    assert history == ['StartAccepted', 'WorkflowStarted'], history
+    assert (original_path.parent / 'waterline-failure.txt').exists()
+    failed_originals.append({'run_id': run_id, 'status': run['status'], 'repairs': 0, 'events': len(history)})
+assert db.execute('SELECT COUNT(*) FROM workflow_runs').fetchone()[0] == len(results) + len(failed_originals)
+assert sum(row['status'] == 'completed' for row in results) == 2
+assert sum(row['status'] == 'failed' for row in results) == 0
+assert db.execute('SELECT COUNT(*) FROM workflow_worker_registrations').fetchone()[0] == 0
+db.close()
+assert hashes == {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in files}
+assert hashes == {p.name: hashlib.sha256((copy / p.name).read_bytes()).hexdigest() for p in files}
+print(json.dumps({'sdk': '2.8.0', 'server': '2.10.1', 'waterline': '2.12.1', 'cases': results,
+                  'waterline_asset_source': os.environ.get('WATERLINE_ASSET_SOURCE', 'published'),
+                  'failed_originals': failed_originals,
+                  'events': sum(r['events'] for r in results),
+                  'tasks': sum(r['tasks'] for r in results), 'repairs': sum(r['repairs'] for r in results),
+                  'workers': sum(len(r['worker_exits']) for r in results), 'database_and_wal_unchanged': True,
+                  'source_hashes': hashes}, indent=2))

tests/fixtures/entry_recovery_http/cold.py

@@ -0,0 +1,17 @@
+import asyncio
+import json
+import sys
+
+from durable_workflow import Client
+
+
+async def main():
+    workflow_id, run_id = sys.argv[1:]
+    async with Client('http://api:8080', token='general221-fixture') as client:
+        handle = client.get_workflow_handle(workflow_id, run_id=run_id, workflow_type='general221.entry')
+        result = await handle.result(timeout=3)
+        assert result == {'amount': 7, 'typed': [9223372036854775807, None, 'Київ']}, result
+        print(json.dumps({'run_id': run_id, 'typed_result': 'exact'}))
+
+
+asyncio.run(main())

tests/fixtures/entry_recovery_http/compose.yml

@@ -0,0 +1,92 @@
+services:
+  bootstrap:
+    image: ${ENTRY_SERVER_IMAGE:?Set the immutable published Server 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_POLLING_CACHE_PATH: /data/poll-cache
+      DW_AUTH_DRIVER: token
+      DW_AUTH_TOKEN: general221-fixture
+      DW_ADMIN_TOKEN: general221-fixture
+      DW_WORKER_TOKEN: general221-worker
+    volumes:
+      - ${ENTRY_PHASE_ROOT:?Set a disposable phase directory}/data:/data
+
+  api:
+    image: ${ENTRY_SERVER_IMAGE:?}
+    user: "1000:1000"
+    cpus: 1
+    mem_limit: 512m
+    memswap_limit: 512m
+    environment: *server_environment
+    volumes:
+      - ${ENTRY_PHASE_ROOT:?}/data:/data
+    depends_on:
+      bootstrap:
+        condition: service_completed_successfully
+
+  viewer:
+    image: ${ENTRY_WATERLINE_IMAGE:?Set the immutable published Waterline image}
+    cpus: 0.5
+    mem_limit: 256m
+    memswap_limit: 256m
+    init: true
+    environment:
+      WATERLINE_SERVER_ENDPOINT: http://api:8080
+      WATERLINE_SERVER_TOKEN: general221-fixture
+      WATERLINE_NAMESPACE: default
+      WATERLINE_ACCESS_MODE: operator
+      WATERLINE_ALLOW_UNAUTHENTICATED: "true"
+    volumes:
+      - viewer-data:/data
+    depends_on:
+      api:
+        condition: service_healthy
+
+  client:
+    image: ${ENTRY_BROWSER_IMAGE:?Set the pinned Python Playwright image}
+    user: "1000:1000"
+    init: true
+    cpus: 2
+    mem_limit: 2g
+    memswap_limit: 2g
+    shm_size: 256m
+    environment:
+      HOME: /tmp
+      EXPECTED_BINDING_REFUSAL: "1"
+      WATERLINE_URL: http://viewer:8080/waterline
+    entrypoint: [sh, -ec]
+    command:
+      - |
+        python -m venv --without-pip /opt/venv
+        python -m pip --python /opt/venv/bin/python install --index-url https://pypi.org/simple --no-cache-dir --quiet -r /phase/requirements.txt
+        /opt/venv/bin/python /phase/setup.py
+        exec sleep infinity
+    tmpfs:
+      - /opt/venv:rw,exec,uid=1000,gid=1000,size=536870912
+    volumes:
+      - ${ENTRY_PHASE_ROOT:?}:/phase
+      - ./requirements.txt:/phase/requirements.txt:ro
+      - ./setup.py:/phase/setup.py:ro
+      - ./consumer.py:/phase/consumer.py:ro
+      - ./worker.py:/phase/worker.py:ro
+      - ./ui.py:/phase/ui.py:ro
+      - ./cold.py:/phase/cold.py:ro
+      - ./audit.py:/phase/audit.py:ro
+      - ./verify_install.py:/phase/verify_install.py:ro
+    depends_on:
+      api:
+        condition: service_healthy
+      viewer:
+        condition: service_healthy
+
+volumes:
+  viewer-data: