From 830ae6b38d318c42e5051abaa0d22fc465cf981a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 02:17:51 +0000 Subject: [PATCH 1/5] Prepare published cold Worker timer redrive acceptance --- .github/workflows/ci.yml | 29 +++ scripts/ci/activity_acceptance_consumer.py | 21 +- scripts/ci/published-activity-acceptance.py | 38 ++- scripts/ci/published-timer-redrive.md | 46 ++++ tests/fixtures/published-timer-redrive.json | 11 + .../integration/test_native_timer_redrive.py | 227 ++++++++++++++++++ tests/test_activity_acceptance_consumer.py | 19 ++ tests/test_native_timer_redrive_cell.py | 133 ++++++++++ tests/test_published_activity_acceptance.py | 11 + 9 files changed, 522 insertions(+), 13 deletions(-) create mode 100644 scripts/ci/published-timer-redrive.md create mode 100644 tests/fixtures/published-timer-redrive.json create mode 100644 tests/integration/test_native_timer_redrive.py create mode 100644 tests/test_native_timer_redrive_cell.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 389730d..ceb5948 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,6 +23,10 @@ on: description: Run the reviewed published SDK/Server activity identity tuple type: boolean default: false + published_timer_redrive: + description: Run the reviewed published SDK/native timer-redrive tuple + type: boolean + default: false permissions: contents: read @@ -32,6 +36,31 @@ concurrency: cancel-in-progress: ${{ github.event_name == 'pull_request' }} jobs: + published-timer-redrive: + if: github.event_name == 'workflow_dispatch' && inputs.published_timer_redrive + needs: [action-policy, lint] + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 + with: + persist-credentials: false + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 + with: + python-version: "3.12.16" + - name: Bootstrap patched pip trust + uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7 # clean-trust + with: + clients: pip,python + - name: Run the published cold Worker timer-redrive case + run: python scripts/ci/published-activity-acceptance.py --scenario timer-redrive + - name: Retain published tuple and scenario evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: published-timer-redrive-${{ github.sha }} + path: published-activity-results/ + retention-days: 7 published-activity-acceptance: if: github.event_name == 'workflow_dispatch' && inputs.published_activity_acceptance needs: [action-policy, lint] diff --git a/scripts/ci/activity_acceptance_consumer.py b/scripts/ci/activity_acceptance_consumer.py index c0dcb11..dd9136b 100644 --- a/scripts/ci/activity_acceptance_consumer.py +++ b/scripts/ci/activity_acceptance_consumer.py @@ -1,4 +1,4 @@ -"""Install a closed wheel set, then await admission to the ordinary retry case.""" +"""Install a closed wheel set, then await admission to one fixed reviewed case.""" from __future__ import annotations @@ -23,6 +23,10 @@ LIMIT = 64 * 1024 * 1024 JUNIT = DATA / "identity.xml" CAPTURED_VERIFIER = "" +CASE_FILES = { + "activity-identity": {"test_activity_identity.py"}, + "timer-redrive": {"test_native_timer_redrive.py", "redrive_timer_clock_workflow.py"}, +} def receive(maximum): @@ -53,9 +57,11 @@ def run(arguments, environment, input_data=None): def validate_setup(packet): - if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version"}: + if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version", "scenario"}: raise ValueError("consumer setup shape differs") - expected_files = {"requirements.txt", "verify_activity_install.py", "test_activity_identity.py", "conftest.py"} + if packet["scenario"] not in CASE_FILES: + raise ValueError("consumer scenario differs") + expected_files = {"requirements.txt", "verify_activity_install.py", "conftest.py"} | CASE_FILES[packet["scenario"]] if set(packet["files"]) != expected_files or not 1 <= len(packet["wheels"]) <= 30: raise ValueError("consumer setup membership differs") if packet["wheel_name"] not in packet["wheels"]: @@ -218,6 +224,8 @@ def prepare_environment(): def execute_case(verifier, python, packet, before, environment, admission): + redrive = packet["scenario"] == "timer-redrive" + test_name = "test_native_timer_redrive.py" if redrive else "test_activity_identity.py" environment.update( { "DURABLE_WORKFLOW_SERVER_URL": admission["server_url"], @@ -226,6 +234,9 @@ def execute_case(verifier, python, packet, before, environment, admission): "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", } ) + if redrive: + environment["DURABLE_WORKFLOW_REDRIVE_CELL"] = "1" + environment["DURABLE_WORKFLOW_REDRIVE_CASE_SHA256"] = packet["files"][test_name]["sha256"] result = subprocess.run( [ str(python), @@ -243,13 +254,13 @@ def execute_case(verifier, python, packet, before, environment, admission): "--import-mode=importlib", "-v", "--capture=tee-sys", - str(DATA / "test_activity_identity.py"), + str(DATA / test_name), "--junitxml=" + str(JUNIT), ], env=environment, capture_output=True, text=True, - timeout=90, + timeout=180 if redrive else 90, ) after = inventory(verifier, python, packet) if after != before: diff --git a/scripts/ci/published-activity-acceptance.py b/scripts/ci/published-activity-acceptance.py index 0931ac5..00fd2c2 100644 --- a/scripts/ci/published-activity-acceptance.py +++ b/scripts/ci/published-activity-acceptance.py @@ -1,7 +1,8 @@ -"""Run one exact published SDK retry-identity case inside an owned internal network.""" +"""Run one fixed published SDK acceptance case inside an owned internal network.""" from __future__ import annotations +import argparse import base64 import hashlib import ipaddress @@ -38,6 +39,15 @@ LABEL_FILTER = "label=" ID_FORMAT = "{{.ID}}" DIAGNOSTIC_LIMIT = 64 * 1024 +SCENARIO = "activity-identity" +SCENARIOS = { + "activity-identity": ( + "test_activity_identity.py", "test_retry_preserves_received_execution_id_and_changes_attempt_id", + ), + "timer-redrive": ( + "test_native_timer_redrive.py", "test_cold_workers_preserve_fractional_timer_clock_across_repeated_redrive", + ), +} # No inherited context, registry credentials, proxy, SSH or remote Docker endpoint. # This kernel directory is private to each CLI process and contains no config.json. DOCKER_ENVIRONMENT = { @@ -111,7 +121,7 @@ def verify_case_result(result): raise ValueError("published retry case reported failure") root = ElementTree.fromstring(result["junit"]) cases = list(root.iter("testcase")) - expected = "test_retry_preserves_received_execution_id_and_changes_attempt_id" + expected = SCENARIOS[SCENARIO][1] if len(cases) != 1 or cases[0].get("name") != expected: raise ValueError("published retry case inventory differs") if any(cases[0].find(tag) is not None for tag in ("failure", "error", "skipped")): @@ -144,11 +154,15 @@ def capture_source(report): paths = { REQUIREMENTS: "scripts/ci/published-activity-requirements.txt", VERIFIER_FILENAME: "scripts/ci/verify_activity_install.py", - "test_activity_identity.py": "tests/integration/test_activity_identity.py", + SCENARIOS[SCENARIO][0]: "tests/integration/" + SCENARIOS[SCENARIO][0], "conftest.py": "tests/integration/conftest.py", "bootstrap.py": "scripts/ci/activity_acceptance_consumer.py", - "tuple.json": "tests/fixtures/published-activity-acceptance.json", + "tuple.json": "tests/fixtures/" + ( + "published-timer-redrive.json" if SCENARIO == "timer-redrive" else "published-activity-acceptance.json" + ), } + if SCENARIO == "timer-redrive": + paths["redrive_timer_clock_workflow.py"] = "tests/integration/redrive_timer_clock_workflow.py" captured = { name: call("git", "show", commit + ":" + path, cwd=ROOT).stdout.encode() for name, path in paths.items() } @@ -196,10 +210,14 @@ def download_wheels(selected, captured, files): wheels[path.name] = packet_file(path.read_bytes()) if wheels[wheel_name]["sha256"] != approval["files"][wheel_name]: raise ValueError("registry wheel differs from the independently reviewed release") + inputs = [REQUIREMENTS, VERIFIER_FILENAME, SCENARIOS[SCENARIO][0], "conftest.py"] + if SCENARIO == "timer-redrive": + inputs.append("redrive_timer_clock_workflow.py") return { + "scenario": SCENARIO, "files": { name: packet_file(captured[name]) - for name in (REQUIREMENTS, VERIFIER_FILENAME, "test_activity_identity.py", "conftest.py") + for name in inputs }, "wheels": wheels, "wheel_name": wheel_name, @@ -468,8 +486,9 @@ def run_case(owned, selected, images, packet, bootstrap, report): if ready["sdk_inventory"] != result["sdk_inventory"]: raise ValueError("installed SDK bytes changed during the case") owned.files.write("sdk-inventory.json", json.dumps(result["sdk_inventory"], indent=2)) - owned.files.write("identity.log", result["case_output"]) - owned.files.write("identity.xml", result["junit"]) + name = "timer-redrive" if SCENARIO == "timer-redrive" else "identity" + owned.files.write(name + ".log", result["case_output"]) + owned.files.write(name + ".xml", result["junit"]) report["consumer_ready"] = {key: value for key, value in ready.items() if key != "sdk_inventory"} session.send({"release": True}) code = session.finish() @@ -482,7 +501,7 @@ def run_case(owned, selected, images, packet, bootstrap, report): def main(): files = output_files.Output(OUTPUT) - report = {"complete": False, "started_at": datetime.now(timezone.utc).isoformat()} + report = {"complete": False, "scenario": SCENARIO, "started_at": datetime.now(timezone.utc).isoformat()} owned = OwnedDocker(uuid4().hex, files, report) signal.signal(signal.SIGALRM, deadline) signal.alarm(360) @@ -513,4 +532,7 @@ def main(): if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--scenario", choices=tuple(SCENARIOS), default=SCENARIO) + SCENARIO = parser.parse_args().scenario main() diff --git a/scripts/ci/published-timer-redrive.md b/scripts/ci/published-timer-redrive.md new file mode 100644 index 0000000..9703b8b --- /dev/null +++ b/scripts/ci/published-timer-redrive.md @@ -0,0 +1,46 @@ +The timer-redrive cell uses the exact published tuple in +`tests/fixtures/published-timer-redrive.json` and the same closed install, +internal network, container authority, SDK inventory and independent cleanup +profile as [activity identity](published-activity-acceptance.md). The identity +tuple and its default command remain unchanged. + +After the changed case, tuple and runner have source review, terminal normal +checks and normal integration, run the ordinary GitHub Actions cell: + +```bash +gh workflow run ci.yml --ref main -f published_timer_redrive=true +``` + +One unchanged workflow definition first completes an activity, waits for a +timer, and fails its second activity. The source TimerFired must have a real +nonzero fractional event timestamp. The test never edits published history. +It obtains original start authority from the persisted run, then performs first +and repeated redrive through the public client API. Each of the three Workers +runs in a fresh Python process with the same definition and fingerprint. +The first successor fails the second activity again; the next succeeds. + +The retained observations prove callback identity and clock, emitted commands, +new shared timer-pair identities, immediate-predecessor lineage, original start +anchor, completed-result reuse and unchanged source history. Successors must +already have the reused timer prefix before a Worker starts, issue no timer +command, expose no timer wait, and execute only the failed activity. Parsed UTC +instants retain all supported microseconds; unrepresentable precision refuses +rather than being silently rounded. The original definition does not read an +ordinary entry clock, so this does not advertise that separate SDK capability. +Database timer/task invariants remain bound to the qualified native producer; +the SDK case observes public history, Worker commands and callbacks. + +The trusted consumer binds case source to its Git-captured digest before each +cold child. A literal child entry verifies those captured bytes before compile +and execution; it opens no host dispatcher. Each Worker has a 45-second SDK +budget and a 55-second child wall limit. The selected pytest case has a +180-second bound within the existing 240-second consumer exchange and +six-minute controller deadline; cleanup keeps its independent per-call bounds. +The older identity case retains its 90-second pytest bound. No local SDK or +native runtime run is part of source qualification. + +The seven-day artifact retains the complete fixed wheel set, selected platform +and source, unchanged SDK inventory, before/after runtime authority, all three +cold-process fingerprints and command/callback observations, actual histories, +JUnit and full-ID/label absence. An installed cell is qualified only after its +retained evidence receives independent review. diff --git a/tests/fixtures/published-timer-redrive.json b/tests/fixtures/published-timer-redrive.json new file mode 100644 index 0000000..421f036 --- /dev/null +++ b/tests/fixtures/published-timer-redrive.json @@ -0,0 +1,11 @@ +{ + "schema": "sdk-python.published-timer-redrive/v1", + "sdk_version": "2.17.2", + "server_image": "ghcr.io/durable-workflow/server@sha256:39d218fc5f251a3e7c4ff1dcda06d9cc53f66c034f0c6d0f67e8371fd0cc97d6", + "server_source": "da38c0ede9b15046fcab2bb0f25412c85a3e3404", + "server_amd64_manifest": "sha256:8bc8400c4ff203c1e4d5c7f82028cfa1d429ab00a47a6e698453d8345ae4bcfc", + "server_qualification": "https://github.com/durable-workflow/server/issues/980#issuecomment-6867470082", + "python_image": "docker.io/library/python@sha256:935c3c5b12357a7e10caa9c1e6ca5692344d97cf229832ef584be2cec9fc03df", + "python_amd64_manifest": "sha256:0f086103b026048cc50cb5da25ec435c023c5241a83fc44daea593377abc61c4", + "python_version": "3.12.16" +} diff --git a/tests/integration/test_native_timer_redrive.py b/tests/integration/test_native_timer_redrive.py new file mode 100644 index 0000000..305e21e --- /dev/null +++ b/tests/integration/test_native_timer_redrive.py @@ -0,0 +1,227 @@ +"""Published-package cold Worker proof; run only through its reviewed manual cell.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import linecache +import os +import re +import subprocess +import sys +import types +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import pytest + +from durable_workflow import Client, Worker, activity + +DEFINITION_SHA = "af24e69d5d4ed57e3cb56915a5e2febe535438e8bb2b33ecfd998098c1557b2b" +WORKFLOW_TYPE = "tests.python-redrive-timer-clock" +CHILD_ENTRY = """import asyncio, hashlib, json, sys +p = json.loads(sys.stdin.read()) +source = p.pop('source') +if hashlib.sha256(source.encode()).hexdigest() != p.pop('source_sha256'): + raise ValueError('cold Worker source differs') +scope = {'__name__': 'cold_redrive_case', '__file__': ''} +exec(compile(source, '', 'exec'), scope) +asyncio.run(scope['cold_worker'](p)) +""" + + +def instant(value: Any) -> datetime: + assert isinstance(value, str) and value + assert not re.search(r"\.\d{7}", value), "clock precision exceeds Python's exact microsecond representation" + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + assert result.tzinfo is not None + return result.astimezone(timezone.utc) + + +def one(events: list[dict[str, Any]], kind: str, *, step: int | None = None) -> dict[str, Any]: + selected = [ + event for event in events + if event["event_type"] == kind and (step is None or event["payload"].get("sequence") == step) + ] + assert len(selected) == 1, (kind, step, len(selected)) + return selected[0] + + +def verify_reuse( + events: list[dict[str, Any]], predecessor: list[dict[str, Any]], *, + predecessor_id: str, source_clock: datetime, source_start: datetime, used_timer_ids: set[str], +) -> str: + started = one(events, "WorkflowStarted")["payload"] + assert started["continued_from_run_id"] == predecessor_id + assert started["recovery_kind"] == "redrive" + assert started["resume_step_sequence"] == 3 + assert instant(started["replayed_started_at"]) == source_start + scheduled = one(events, "TimerScheduled", step=2)["payload"] + fired = one(events, "TimerFired", step=2)["payload"] + timer_id = scheduled["timer_id"] + assert isinstance(timer_id, str) and timer_id and timer_id not in used_timer_ids + assert fired["timer_id"] == timer_id + assert fired["reused_from_run_id"] == predecessor_id + assert instant(fired["reused_recorded_at"]) == source_clock + assert instant(fired["fired_at"]) == instant(one(predecessor, "TimerFired", step=2)["payload"]["fired_at"]) + for payload in (scheduled, fired): + assert "task" not in payload and "command" not in payload + assert scheduled["fire_at"] == one(predecessor, "TimerScheduled", step=2)["payload"]["fire_at"] + original = one(predecessor, "ActivityCompleted", step=1)["payload"] + reused = one(events, "ActivityCompleted", step=1)["payload"] + assert reused["result"] == original["result"] + assert reused["reused_from_run_id"] == predecessor_id + assert reused["reused_activity_execution_id"] == original["activity_execution_id"] + assert not [event for event in events if event["event_type"] == "ActivityScheduled"] + return timer_id + + +def definition(source: str) -> type: + assert hashlib.sha256(source.encode()).hexdigest() == DEFINITION_SHA + name, filename = "accepted_redrive_timer", "" + module = types.ModuleType(name) + module.__file__ = filename + sys.modules[name] = module + linecache.cache[filename] = (len(source), None, source.splitlines(keepends=True), filename) + exec(compile(source, filename, "exec"), module.__dict__) + return module.RedriveTimerClockWorkflow + + +async def cold_worker(packet: dict[str, Any]) -> None: + callbacks: list[dict[str, Any]] = [] + completions: list[dict[str, Any]] = [] + + class ObservedClient(Client): + async def complete_workflow_task(self, **kwargs: Any) -> Any: + completions.append({"task_id": kwargs["task_id"], "commands": deepcopy(kwargs["commands"])}) + return await super().complete_workflow_task(**kwargs) + + @activity.defn(name="tests.redrive.first") + async def first(name: str) -> str: + callbacks.append({"activity": "first", "pid": os.getpid()}) + return f"Hello, {name}!" + + @activity.defn(name="tests.redrive.second") + async def second(greeting: str, clock: str) -> str: + info = activity.context().info + callbacks.append({ + "activity": "second", "clock": clock, "pid": os.getpid(), "run_id": info.run_id, + "task_id": info.task_id, "attempt_id": info.activity_attempt_id, + "execution_id": info.activity_execution_id, + }) + if packet["fail_second"]: + raise RuntimeError("fixed failed-activity redrive boundary") + return "resumed" + + async with ObservedClient(packet["server_url"], token=packet["token"], namespace="default") as client: + worker = Worker( + client, task_queue=packet["queue"], worker_id="python-clock-" + uuid4().hex, + workflows=[definition(packet["definition"])], activities=[first, second], sticky_cache_capacity=0, + ) + fingerprints = dict(worker.workflow_definition_fingerprints) + try: + outcome = await worker.run_until(workflow_id=packet["workflow_id"], timeout=45) + finally: + await worker.stop() + print("COLD_WORKER_RESULT=" + json.dumps({ + "pid": os.getpid(), "worker_id": worker.worker_id, "fingerprints": fingerprints, + "definition_sha256": DEFINITION_SHA, "status": outcome.status, + "callbacks": callbacks, "completions": completions, + }, sort_keys=True), flush=True) + + +def run_cold_worker(packet: dict[str, Any]) -> dict[str, Any]: + source = Path(__file__).read_text() + digest = hashlib.sha256(source.encode()).hexdigest() + assert digest == os.environ["DURABLE_WORKFLOW_REDRIVE_CASE_SHA256"] + packet = {**packet, "source": source, "source_sha256": digest} + environment = {"PATH": os.defpath, "PYTHONDONTWRITEBYTECODE": "1"} + result = subprocess.run( + [sys.executable, "-I", "-B", "-c", CHILD_ENTRY], input=json.dumps(packet), + env=environment, capture_output=True, text=True, timeout=55, check=False, + ) + assert result.returncode == 0, (result.returncode, result.stdout[-16384:], result.stderr[-16384:]) + lines = [line.removeprefix("COLD_WORKER_RESULT=") for line in result.stdout.splitlines() + if line.startswith("COLD_WORKER_RESULT=")] + assert len(lines) == 1 + return json.loads(lines[0]) + + +def verify_worker(observed: dict[str, Any], *, redrive: bool, clock: datetime) -> None: + assert observed["definition_sha256"] == DEFINITION_SHA + assert set(observed["fingerprints"]) == {WORKFLOW_TYPE} + assert observed["worker_id"] and isinstance(observed["pid"], int) and observed["pid"] > 0 + assert observed["completions"] + commands = [command for completion in observed["completions"] for command in completion["commands"]] + assert sum(command["type"] == "start_timer" for command in commands) == (0 if redrive else 1) + expected = ["tests.redrive.second"] if redrive else ["tests.redrive.first", "tests.redrive.second"] + assert [command["activity_type"] for command in commands if command["type"] == "schedule_activity"] == expected + assert [item["activity"] for item in observed["callbacks"]] == (["second"] if redrive else ["first", "second"]) + assert instant(observed["callbacks"][-1]["clock"]) == clock + + +@pytest.mark.skipif( + os.environ.get("DURABLE_WORKFLOW_REDRIVE_CELL") != "1", reason="reviewed manual published cell only", +) +async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redrive( + server_url: str, server_token: str, +) -> None: + source = Path(__file__).with_name("redrive_timer_clock_workflow.py").read_text() + assert hashlib.sha256(source.encode()).hexdigest() == DEFINITION_SHA + queue = "python-timer-clock-" + uuid4().hex + common = {"server_url": server_url, "token": server_token, "queue": queue, + "workflow_id": queue, "definition": source} + observations = [] + histories = [] + run_ids = [] + async with Client(server_url, token=server_token, namespace="default") as client: + handle = await client.start_workflow( + workflow_type=WORKFLOW_TYPE, workflow_id=queue, task_queue=queue, input=["Taylor", 1], + ) + assert handle.run_id + run_ids.append(handle.run_id) + observations.append(await asyncio.to_thread(run_cold_worker, {**common, "fail_second": True})) + assert observations[0]["status"] == "failed" + histories.append([event async for event in client.iter_history(queue, handle.run_id)]) + frozen_source = deepcopy(histories[0]) + original = await client.describe_workflow_run(queue, handle.run_id) + source_start = instant(original.started_at) + source_clock = instant(one(histories[0], "TimerFired", step=2)["recorded_at"]) + assert source_clock.microsecond != 0, "real TimerFired must expose nonzero source precision" + used_timer_ids = {one(histories[0], "TimerFired", step=2)["payload"]["timer_id"]} + for repeat in (1, 2): + receipt = await client.redrive_workflow(queue, run_ids[-1]) + assert receipt.outcome == "redriven" and receipt.run_id and receipt.run_id not in run_ids + run_id = receipt.run_id + prefix = [event async for event in client.iter_history(queue, run_id)] + timer_id = verify_reuse( + prefix, histories[-1], predecessor_id=run_ids[-1], source_clock=source_clock, + source_start=source_start, used_timer_ids=used_timer_ids, + ) + used_timer_ids.add(timer_id) + before = await client.describe_workflow_run(queue, run_id) + assert before.wait_kind != "timer" + observed = await asyncio.to_thread(run_cold_worker, {**common, "fail_second": repeat == 1}) + assert observed["status"] == ("failed" if repeat == 1 else "completed") + verify_worker(observed, redrive=True, clock=source_clock) + assert observed["callbacks"][0]["run_id"] == run_id + observations.append(observed) + run_ids.append(run_id) + histories.append([event async for event in client.iter_history(queue, run_id)]) + assert histories[-1][:len(prefix)] == prefix + assert [event async for event in client.iter_history(queue, handle.run_id)] == frozen_source + result = await client.get_workflow_handle(queue, run_id=run_ids[-1]).result(timeout=10) + verify_worker(observations[0], redrive=False, clock=source_clock) + assert len({item["pid"] for item in observations}) == len({item["worker_id"] for item in observations}) == 3 + assert len({json.dumps(item["fingerprints"], sort_keys=True) for item in observations}) == 1 + assert result == {"greeting": "Hello, Taylor!", "timer_clock": source_clock.isoformat(timespec="microseconds"), + "resumed": "resumed"} + print(json.dumps({ + "scope": "published cold-Worker timer redrive", "runs": run_ids, "source_started_at": original.started_at, + "source_timer_clock": source_clock.isoformat(), "definition_sha256": DEFINITION_SHA, + "workers": observations, "histories": histories, + }, sort_keys=True)) diff --git a/tests/test_activity_acceptance_consumer.py b/tests/test_activity_acceptance_consumer.py index e9fcf66..0bcb27d 100644 --- a/tests/test_activity_acceptance_consumer.py +++ b/tests/test_activity_acceptance_consumer.py @@ -135,6 +135,7 @@ def test_closed_consumer_refuses_sdk_wheel_omission_or_substitution(): "wheel_name": "sdk.whl", "wheel_sha256": "b" * 64, "sdk_version": "test", + "scenario": "activity-identity", } with pytest.raises(ValueError, match="differs from approval"): consumer.validate_setup(packet) @@ -143,6 +144,24 @@ def test_closed_consumer_refuses_sdk_wheel_omission_or_substitution(): consumer.validate_setup(packet) +@pytest.mark.parametrize("scenario", ["activity-identity", "timer-redrive", "foreign"]) +def test_consumer_scenario_has_closed_captured_file_membership(scenario): + inputs = {"requirements.txt", "verify_activity_install.py", "conftest.py"} + inputs |= consumer.CASE_FILES.get(scenario, set()) + packet = { + "scenario": scenario, "files": dict.fromkeys(inputs, {}), "wheels": {"sdk.whl": {"sha256": "a" * 64}}, + "wheel_name": "sdk.whl", "wheel_sha256": "a" * 64, "sdk_version": "test", + } + if scenario == "foreign": + with pytest.raises(ValueError, match="scenario differs"): + consumer.validate_setup(packet) + return + consumer.validate_setup(packet) + packet["files"]["unreviewed.py"] = {} + with pytest.raises(ValueError, match="membership differs"): + consumer.validate_setup(packet) + + def test_sdk_writable_verifier_replacement_cannot_change_the_retained_checker(monkeypatch, tmp_path): original = "def verify_sdk(*args, **kwargs): return {'checker': 'Git-captured'}\n" monkeypatch.setattr(consumer, "CAPTURED_VERIFIER", original) diff --git a/tests/test_native_timer_redrive_cell.py b/tests/test_native_timer_redrive_cell.py new file mode 100644 index 0000000..004e0dd --- /dev/null +++ b/tests/test_native_timer_redrive_cell.py @@ -0,0 +1,133 @@ +"""Offline contract controls for the separate installed cold-Worker cell.""" + +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +import sys +from copy import deepcopy +from pathlib import Path + +import pytest + +from tests.integration import test_native_timer_redrive as cell +from tests.test_native_timer_redrive_clock import successor + + +def prefix(): + previous, current = successor(), successor() + source_time = cell.instant("2026-11-03T00:00:06.654321+00:00") + source_start = cell.instant("2026-11-03T00:00:00+00:00") + for event in previous["history"]: + if event["event_type"] in {"TimerScheduled", "TimerFired"}: + event["payload"]["timer_id"] = "old-timer" + events = current["history"] + cell.one(events, "WorkflowStarted")["payload"]["continued_from_run_id"] = "previous" + fired = cell.one(events, "TimerFired")["payload"] + fired["reused_from_run_id"] = "previous" + fired["reused_recorded_at"] = source_time.isoformat() + completed = cell.one(events, "ActivityCompleted")["payload"] + completed["reused_from_run_id"] = "previous" + completed["reused_activity_execution_id"] = cell.one(previous["history"], "ActivityCompleted")["payload"][ + "activity_execution_id" + ] + return events, previous["history"], { + "predecessor_id": "previous", "source_clock": source_time, "source_start": source_start, + "used_timer_ids": {"old-timer"}, + } + + +def test_exact_fraction_and_immediate_predecessor_preserve_unchanged_inputs(): + events, previous, arguments = prefix() + original = deepcopy((events, previous, arguments)) + assert cell.verify_reuse(events, previous, **arguments) + assert (events, previous, arguments) == original + + +@pytest.mark.parametrize("fault", [ + "floor", "root_lineage", "new_start", "pair_id", "old_id", "timer_task", "scheduled_activity", "changed_result", +]) +def test_false_reuse_evidence_refuses(fault): + events, previous, arguments = prefix() + fired = cell.one(events, "TimerFired")["payload"] + if fault == "floor": + fired["reused_recorded_at"] = "2026-11-03T00:00:06+00:00" + elif fault == "root_lineage": + fired["reused_from_run_id"] = "root-instead-of-immediate-predecessor" + elif fault == "new_start": + cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"] = "2026-11-03T00:05:00Z" + elif fault == "pair_id": + fired["timer_id"] = "different-pair" + elif fault == "old_id": + arguments["used_timer_ids"].add(fired["timer_id"]) + elif fault == "timer_task": + fired["task"] = {"id": "new-timer-work"} + elif fault == "scheduled_activity": + events.append({"event_type": "ActivityScheduled", "payload": {"sequence": 1}}) + else: + cell.one(events, "ActivityCompleted")["payload"]["result"] = "changed" + with pytest.raises(AssertionError): + cell.verify_reuse(events, previous, **arguments) + + +@pytest.mark.parametrize("clock", ["2026-11-03T00:00:06.654321001Z", "2026-11-03T00:00:06", ""]) +def test_unrepresentable_or_unzoned_clock_cannot_hide_precision_loss(clock): + with pytest.raises(AssertionError): + cell.instant(clock) + + +def test_changed_definition_or_case_source_refuses_before_child(monkeypatch): + with pytest.raises(AssertionError): + cell.definition("changed definition") + monkeypatch.setenv("DURABLE_WORKFLOW_REDRIVE_CASE_SHA256", "0" * 64) + monkeypatch.setattr(cell.subprocess, "run", lambda *a, **k: pytest.fail("must refuse before any child")) + with pytest.raises(AssertionError): + cell.run_cold_worker({}) + + +@pytest.mark.parametrize("substitute", [False, True]) +def test_literal_child_bootstrap_binds_captured_bytes_without_opening_a_dispatcher(substitute): + source = "async def cold_worker(p): print('fixed inert result', p['marker'])\n" + packet = {"source": source, "source_sha256": hashlib.sha256(source.encode()).hexdigest(), "marker": "owned"} + if substitute: + packet["source"] += "\nraise RuntimeError('substitute executed')\n" + result = subprocess.run( + [sys.executable, "-I", "-B", "-c", cell.CHILD_ENTRY], input=json.dumps(packet), + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=5, + ) + if substitute: + assert result.returncode != 0 and "source differs" in result.stderr + assert "substitute executed" not in result.stderr and not result.stdout + else: + assert result.returncode == 0 and result.stdout == "fixed inert result owned\n" + + +def test_accepted_workflow_definition_bytes_remain_frozen(): + path = Path(__file__).parent / "integration/redrive_timer_clock_workflow.py" + assert hashlib.sha256(path.read_bytes()).hexdigest() == cell.DEFINITION_SHA + + +@pytest.mark.parametrize("fault", [None, "timer_command", "first_callback", "no_task", "floor"]) +def test_redrive_worker_evidence_requires_only_the_failed_activity_and_exact_clock(fault): + clock = cell.instant("2026-11-03T00:00:06.654321Z") + observed = { + "definition_sha256": cell.DEFINITION_SHA, "fingerprints": {cell.WORKFLOW_TYPE: "sha256:" + "a" * 64}, + "worker_id": "fixed-worker", "pid": 123, + "completions": [{"commands": [{"type": "schedule_activity", "activity_type": "tests.redrive.second"}]}], + "callbacks": [{"activity": "second", "clock": clock.isoformat()}], + } + if fault == "timer_command": + observed["completions"][0]["commands"].append({"type": "start_timer"}) + elif fault == "first_callback": + observed["callbacks"].insert(0, {"activity": "first"}) + elif fault == "no_task": + observed["completions"] = [] + elif fault == "floor": + observed["callbacks"][0]["clock"] = "2026-11-03T00:00:06Z" + if fault: + with pytest.raises(AssertionError): + cell.verify_worker(observed, redrive=True, clock=clock) + else: + cell.verify_worker(observed, redrive=True, clock=clock) diff --git a/tests/test_published_activity_acceptance.py b/tests/test_published_activity_acceptance.py index 31078ab..1dbe4a8 100644 --- a/tests/test_published_activity_acceptance.py +++ b/tests/test_published_activity_acceptance.py @@ -374,6 +374,17 @@ def test_complete_case_requires_the_actual_single_retry_test_to_pass(): runner.verify_case_result(case_result()) +def test_timer_scenario_cannot_borrow_a_passing_identity_receipt(monkeypatch): + monkeypatch.setattr(runner, "SCENARIO", "timer-redrive") + result = case_result() + with pytest.raises(ValueError, match="inventory differs"): + runner.verify_case_result(result) + result["junit"] = result["junit"].replace( + "test_retry_preserves_received_execution_id_and_changes_attempt_id", runner.SCENARIOS["timer-redrive"][1], + ) + runner.verify_case_result(result) + + @pytest.mark.parametrize("change", ["skip", "failure", "error", "wrong_case", "empty", "reported_failure"]) def test_exit_zero_cannot_qualify_a_skipped_missing_or_different_case(change): result = case_result() From a5e4e5fc4abd5fe0200a7cf0025f3f9413c4f736 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 03:27:01 +0000 Subject: [PATCH 2/5] Bind cold Worker startup and complete timer admission evidence --- scripts/ci/activity_acceptance_consumer.py | 71 ++++--- scripts/ci/activity_acceptance_entry.py | 160 ++++++++++++++ scripts/ci/published-activity-acceptance.md | 5 + scripts/ci/published-activity-acceptance.py | 11 +- scripts/ci/published-timer-redrive.md | 35 +++- .../integration/test_native_timer_redrive.py | 198 +++++++++++++++--- tests/test_activity_acceptance_entry.py | 113 ++++++++++ tests/test_native_timer_redrive_cell.py | 188 ++++++++++++++--- 8 files changed, 682 insertions(+), 99 deletions(-) create mode 100644 scripts/ci/activity_acceptance_entry.py create mode 100644 tests/test_activity_acceptance_entry.py diff --git a/scripts/ci/activity_acceptance_consumer.py b/scripts/ci/activity_acceptance_consumer.py index dd9136b..f51b756 100644 --- a/scripts/ci/activity_acceptance_consumer.py +++ b/scripts/ci/activity_acceptance_consumer.py @@ -23,6 +23,7 @@ LIMIT = 64 * 1024 * 1024 JUNIT = DATA / "identity.xml" CAPTURED_VERIFIER = "" +CAPTURED_ENTRY = "" CASE_FILES = { "activity-identity": {"test_activity_identity.py"}, "timer-redrive": {"test_native_timer_redrive.py", "redrive_timer_clock_workflow.py"}, @@ -138,28 +139,27 @@ def main(): verifier = retained_verifier() environment, python = prepare_environment() before = inventory(verifier, python, packet) + entry = captured_entry() + installed = entry["snapshot"](ENVIRONMENT) + site = ENVIRONMENT / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + distributions = sorted( + (item.metadata["Name"], item.version) + for item in importlib.metadata.distributions(path=[str(site / "site-packages")]) + ) ready = { "sdk_inventory": before, "python": sys.version, "environment": str(ENVIRONMENT), - "distributions": run( - [ - str(python), - "-I", - "-B", - "-c", - "import importlib.metadata,json; print(json.dumps(sorted((d.metadata['Name']," - "d.version) for d in importlib.metadata.distributions())))", - ], - environment, - ), + "distributions": json.dumps(distributions), + "environment_inventory": installed, + "environment_inventory_sha256": entry["digest"](installed), } print("ACTIVITY_CONSUMER_READY=" + json.dumps(ready, sort_keys=True), flush=True) admission = receive(8192) if set(admission) != {"runtime_ipv4", "token"}: raise ValueError("consumer admission shape differs") admission["server_url"] = await_runtime(admission["runtime_ipv4"]) - output = execute_case(verifier, python, packet, before, environment, admission) + output = execute_case(verifier, python, packet, before, environment, admission, installed) print("ACTIVITY_CONSUMER_RESULT=" + json.dumps(output, sort_keys=True), flush=True) # Remain attached while the controller rechecks the same two IDs and IP. if receive(8192) != {"release": True}: @@ -223,7 +223,29 @@ def prepare_environment(): return environment, python -def execute_case(verifier, python, packet, before, environment, admission): +def captured_entry(): + if not CAPTURED_ENTRY: + raise ValueError("controller did not bind the Git-captured entry") + namespace = {"__name__": "captured_entry"} + exec(compile(CAPTURED_ENTRY, "", "exec"), namespace) + return namespace + + +def entry_source(mode, case_name, case_sha, installed, worker_entry=""): + source = CAPTURED_ENTRY + values = { + "MODE": mode, "INVENTORY_SHA256": captured_entry()["digest"](installed), + "CASE_SHA256": case_sha, "CASE_NAME": case_name, "WORKER_ENTRY": worker_entry, + } + for name, value in values.items(): + placeholder = name + ' = ""' + if source.count(placeholder) != 1: + raise ValueError("captured entry binding placeholder differs") + source = source.replace(placeholder, name + " = " + repr(value), 1) + return source + + +def execute_case(verifier, python, packet, before, environment, admission, installed): redrive = packet["scenario"] == "timer-redrive" test_name = "test_native_timer_redrive.py" if redrive else "test_activity_identity.py" environment.update( @@ -236,28 +258,19 @@ def execute_case(verifier, python, packet, before, environment, admission): ) if redrive: environment["DURABLE_WORKFLOW_REDRIVE_CELL"] = "1" - environment["DURABLE_WORKFLOW_REDRIVE_CASE_SHA256"] = packet["files"][test_name]["sha256"] + worker_entry = entry_source("worker", test_name, packet["files"][test_name]["sha256"], installed) + parent_entry = entry_source("pytest", test_name, packet["files"][test_name]["sha256"], installed, worker_entry) result = subprocess.run( [ - str(python), + sys.executable, "-I", + "-S", "-B", - "-m", - "pytest", "-c", - "/dev/null", - "-p", - "pytest_asyncio.plugin", - "-p", - "no:cacheprovider", - "--asyncio-mode=auto", - "--import-mode=importlib", - "-v", - "--capture=tee-sys", - str(DATA / test_name), - "--junitxml=" + str(JUNIT), + parent_entry, ], env=environment, + input=json.dumps({"inventory": installed}), capture_output=True, text=True, timeout=180 if redrive else 90, @@ -265,6 +278,8 @@ def execute_case(verifier, python, packet, before, environment, admission): after = inventory(verifier, python, packet) if after != before: raise ValueError("installed SDK inventory changed") + if captured_entry()["snapshot"](ENVIRONMENT) != installed: + raise ValueError("private install changed during the case") return { "complete": result.returncode == 0, "sdk_inventory": after, diff --git a/scripts/ci/activity_acceptance_entry.py b/scripts/ci/activity_acceptance_entry.py new file mode 100644 index 0000000..35d5bbb --- /dev/null +++ b/scripts/ci/activity_acceptance_entry.py @@ -0,0 +1,160 @@ +"""Captured entry: verify source and the private install before enabling imports.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import os +import stat +import sys +import types +from pathlib import Path + +MODE = "" +INVENTORY_SHA256 = "" +CASE_SHA256 = "" +CASE_NAME = "" +WORKER_ENTRY = "" +ENVIRONMENT = Path("/run/acceptance-env") +DATA = Path("/run/acceptance-data") +MAXIMUM_FILE = 64 * 1024 * 1024 + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +def metadata(path): + value = path.lstat() + return [value.st_dev, value.st_ino, value.st_uid, value.st_gid, value.st_mode, value.st_size, value.st_mtime_ns] + + +def captured_file(path, maximum=MAXIMUM_FILE): + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + try: + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode) or before.st_size > maximum: + raise ValueError("captured file shape or size differs") + chunks, size = [], 0 + while block := os.read(descriptor, min(1024 * 1024, maximum + 1 - size)): + chunks.append(block) + size += len(block) + if size > maximum: + raise ValueError("captured file exceeds its bound") + after = os.fstat(descriptor) + if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != ( + after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, + ): + raise ValueError("captured file changed during its read") + return b"".join(chunks) + finally: + os.close(descriptor) + + +def snapshot(root): + before = metadata(root) + if not stat.S_ISDIR(before[4]): + raise ValueError("private install root is not a directory") + entries = {} + for path in sorted(root.rglob("*")): + info = metadata(path) + item = {"metadata": info} + if stat.S_ISREG(info[4]): + item["sha256"] = hashlib.sha256(captured_file(path)).hexdigest() + elif stat.S_ISLNK(info[4]): + item["target"] = os.readlink(path) + elif not stat.S_ISDIR(info[4]): + raise ValueError("private install contains an unsupported entry") + entries[path.relative_to(root).as_posix()] = item + if before != metadata(root): + raise ValueError("private install root changed during capture") + return {"root": before, "entries": entries} + + +def guarded_source(path, expected): + source = captured_file(path, 256 * 1024) + if hashlib.sha256(source).hexdigest() != expected: + raise ValueError("captured case source differs") + return source + + +def source_module(source): + module = types.ModuleType("captured_published_case") + module.__file__ = str(DATA / CASE_NAME) + sys.modules[module.__name__] = module + exec(compile(source, module.__file__, "exec", optimize=0), module.__dict__) + return module + + +def run_pytest(source, inventory): + # pytest never opens the writable test or conftest as executable code. + import pytest + + class CapturedModule(pytest.Module): + def _getobj(self): + module = source_module(source) + module._COLD_BOOTSTRAP = WORKER_ENTRY + module._COLD_INVENTORY = inventory + module._COLD_PYTHON = sys.executable + return module + + class CapturedPlugin: + @pytest.hookimpl(tryfirst=True) + def pytest_pycollect_makemodule(self, module_path, parent): + if module_path != DATA / CASE_NAME: + raise ValueError("pytest requested an uncaptured module") + return CapturedModule.from_parent(parent, path=module_path) + + @pytest.fixture(scope="session") + def server_url(self): + return os.environ["DURABLE_WORKFLOW_SERVER_URL"] + + @pytest.fixture(scope="session") + def server_token(self): + return os.environ["DURABLE_WORKFLOW_AUTH_TOKEN"] + + return pytest.main([ + "-c", "/dev/null", "--noconftest", "-p", "pytest_asyncio.plugin", "-p", "no:cacheprovider", + "-o", "asyncio_default_fixture_loop_scope=function", + "--asyncio-mode=auto", "--import-mode=importlib", "-v", "--capture=tee-sys", str(DATA / CASE_NAME), + "--junitxml=" + str(DATA / "identity.xml"), + ], plugins=[CapturedPlugin()]) + + +def main(): + if not (sys.flags.isolated and sys.flags.no_site and sys.dont_write_bytecode) or sys.flags.optimize: + raise ValueError("captured entry requires isolated no-site Python with active assertions") + payload = sys.stdin.buffer.read(4 * 1024 * 1024 + 1) + if len(payload) > 4 * 1024 * 1024: + raise ValueError("captured entry input exceeds its bound") + packet = json.loads(payload) + if set(packet) != ({"inventory", "worker"} if MODE == "worker" else {"inventory"}): + raise ValueError("captured entry input shape differs") + inventory = packet["inventory"] + if digest(inventory) != INVENTORY_SHA256: + raise ValueError("private install commitment differs") + source = guarded_source(DATA / CASE_NAME, CASE_SHA256) + if snapshot(ENVIRONMENT) != inventory: + raise ValueError("private install inventory changed") + # No site.addsitedir or site.main: .pth and sitecustomize are never run. + site = ENVIRONMENT / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + site /= "site-packages" + if not site.is_dir() or site.is_symlink(): + raise ValueError("verified site directory differs") + sys.path.insert(0, str(site)) + sys.prefix = sys.exec_prefix = str(ENVIRONMENT) + if MODE == "worker": + asyncio.run(source_module(source).cold_worker(packet["worker"])) + result = 0 + elif MODE == "pytest": + result = run_pytest(source, inventory) + else: + raise ValueError("captured entry mode differs") + if snapshot(ENVIRONMENT) != inventory: + raise ValueError("private install changed during execution") + raise SystemExit(result) + + +if __name__ == "__main__": + main() diff --git a/scripts/ci/published-activity-acceptance.md b/scripts/ci/published-activity-acceptance.md index fb275ac..645114d 100644 --- a/scripts/ci/published-activity-acceptance.md +++ b/scripts/ci/published-activity-acceptance.md @@ -19,6 +19,11 @@ SDK wheel member, complete distribution metadata and installed RECORD, and uses isolated Python with explicit pytest plugins. Installer-created metadata and console scripts have exact expected bytes. The checker stays in the trusted bootstrap's memory through the case. The checkout is never installed as the SDK. +The complete private installed-file inventory is captured before SDK admission. +Parent pytest uses the read-only base interpreter with `-I -S -B`; its literal +entry verifies captured test bytes and the installed inventory before enabling +the exact private site directory. A captured collector and fixed fixtures avoid +executing mutable test/conftest paths, and no `.pth` startup is processed. The published native runtime and official Python consumer share one owned internal network. Neither publishes a port or receives a host mount. The diff --git a/scripts/ci/published-activity-acceptance.py b/scripts/ci/published-activity-acceptance.py index 00fd2c2..01b9fe6 100644 --- a/scripts/ci/published-activity-acceptance.py +++ b/scripts/ci/published-activity-acceptance.py @@ -141,10 +141,12 @@ def packet_file(data): def bind_bootstrap(captured): source = captured["bootstrap.py"].decode() - placeholder = 'CAPTURED_VERIFIER = ""' - if source.count(placeholder) != 1: - raise ValueError("consumer verifier binding placeholder differs") - return source.replace(placeholder, "CAPTURED_VERIFIER = " + repr(captured[VERIFIER_FILENAME].decode()), 1) + for name, filename in (("CAPTURED_VERIFIER", VERIFIER_FILENAME), ("CAPTURED_ENTRY", "entry.py")): + placeholder = name + ' = ""' + if source.count(placeholder) != 1: + raise ValueError("consumer source binding placeholder differs") + source = source.replace(placeholder, name + " = " + repr(captured[filename].decode()), 1) + return source def capture_source(report): @@ -157,6 +159,7 @@ def capture_source(report): SCENARIOS[SCENARIO][0]: "tests/integration/" + SCENARIOS[SCENARIO][0], "conftest.py": "tests/integration/conftest.py", "bootstrap.py": "scripts/ci/activity_acceptance_consumer.py", + "entry.py": "scripts/ci/activity_acceptance_entry.py", "tuple.json": "tests/fixtures/" + ( "published-timer-redrive.json" if SCENARIO == "timer-redrive" else "published-activity-acceptance.json" ), diff --git a/scripts/ci/published-timer-redrive.md b/scripts/ci/published-timer-redrive.md index 9703b8b..9659f2d 100644 --- a/scripts/ci/published-timer-redrive.md +++ b/scripts/ci/published-timer-redrive.md @@ -19,20 +19,39 @@ and repeated redrive through the public client API. Each of the three Workers runs in a fresh Python process with the same definition and fingerprint. The first successor fails the second activity again; the next succeeds. -The retained observations prove callback identity and clock, emitted commands, -new shared timer-pair identities, immediate-predecessor lineage, original start -anchor, completed-result reuse and unchanged source history. Successors must +The retained observations prove callback identity and clock, closed emitted +command/task inventories, new shared timer-pair identities, immediate-predecessor +lineage, original start anchor, completed-result reuse and unchanged source +history. Recorded nonempty Worker fingerprints must match every original and +successor `WorkflowStarted` fingerprint with its `worker` source. Successors must already have the reused timer prefix before a Worker starts, issue no timer command, expose no timer wait, and execute only the failed activity. Parsed UTC instants retain all supported microseconds; unrepresentable precision refuses rather than being silently rounded. The original definition does not read an ordinary entry clock, so this does not advertise that separate SDK capability. -Database timer/task invariants remain bound to the qualified native producer; -the SDK case observes public history, Worker commands and callbacks. +Unpruned public history exports bind the selected workflow/run and include every +task/wait status. Original stored timer tasks/waits must each count one; +successors must each count zero before and after their cold Worker. All emitted +completion task IDs and activity callback task IDs must match the stored export +inventory. Full timer payloads must match their predecessor modulo the explicitly +authorized new pair ID, immediate predecessor and preserved clock fields. -The trusted consumer binds case source to its Git-captured digest before each -cold child. A literal child entry verifies those captured bytes before compile -and execution; it opens no host dispatcher. Each Worker has a 45-second SDK +An absent matching registration must refuse redrive without changing the chain +or source history. Admission then holds a matching source-cohort registration +that never polls or executes work; its metadata is copied from the actual source +Worker and its cleanup uses the returned registration generation token. Only the +fresh child may execute the accepted successor. Physical `workflow_run_timers` +counts remain a separate actual published-producer observation. Reconstructed +export timers and an immutable qualification link are not per-fixture physical +counts; this API case makes no such claim. + +The trusted consumer captures the complete private installed-file inventory +before admitting SDK execution. Parent pytest and every cold child use the +read-only base interpreter with `-I -S -B`. Their literal entry verifies the +independently fixed case digest and installed-inventory commitment before adding +the exact private site directory, without processing `.pth` or sitecustomize. +Pytest uses captured source bytes and fixed fixtures instead of reopening the +mutable test or conftest as executable code. Each Worker has a 45-second SDK budget and a 55-second child wall limit. The selected pytest case has a 180-second bound within the existing 240-second consumer exchange and six-minute controller deadline; cleanup keeps its independent per-call bounds. diff --git a/tests/integration/test_native_timer_redrive.py b/tests/integration/test_native_timer_redrive.py index 305e21e..a823d9f 100644 --- a/tests/integration/test_native_timer_redrive.py +++ b/tests/integration/test_native_timer_redrive.py @@ -11,6 +11,7 @@ import subprocess import sys import types +from collections import Counter from copy import deepcopy from datetime import datetime, timezone from pathlib import Path @@ -19,19 +20,15 @@ import pytest -from durable_workflow import Client, Worker, activity +from durable_workflow import ActivityRetryPolicy, Client, Worker, activity +from durable_workflow.errors import ServerError +from durable_workflow.workflow import CompleteWorkflow, ScheduleActivity, StartTimer DEFINITION_SHA = "af24e69d5d4ed57e3cb56915a5e2febe535438e8bb2b33ecfd998098c1557b2b" WORKFLOW_TYPE = "tests.python-redrive-timer-clock" -CHILD_ENTRY = """import asyncio, hashlib, json, sys -p = json.loads(sys.stdin.read()) -source = p.pop('source') -if hashlib.sha256(source.encode()).hexdigest() != p.pop('source_sha256'): - raise ValueError('cold Worker source differs') -scope = {'__name__': 'cold_redrive_case', '__file__': ''} -exec(compile(source, '', 'exec'), scope) -asyncio.run(scope['cold_worker'](p)) -""" +_COLD_BOOTSTRAP = "" +_COLD_INVENTORY: dict[str, Any] = {} +_COLD_PYTHON = "" def instant(value: Any) -> datetime: @@ -67,10 +64,15 @@ def verify_reuse( assert fired["timer_id"] == timer_id assert fired["reused_from_run_id"] == predecessor_id assert instant(fired["reused_recorded_at"]) == source_clock - assert instant(fired["fired_at"]) == instant(one(predecessor, "TimerFired", step=2)["payload"]["fired_at"]) - for payload in (scheduled, fired): - assert "task" not in payload and "command" not in payload - assert scheduled["fire_at"] == one(predecessor, "TimerScheduled", step=2)["payload"]["fire_at"] + expected_scheduled = deepcopy(one(predecessor, "TimerScheduled", step=2)["payload"]) + expected_fired = deepcopy(one(predecessor, "TimerFired", step=2)["payload"]) + for payload in (expected_scheduled, expected_fired): + payload.pop("task", None) + payload.pop("command", None) + payload["timer_id"] = timer_id + expected_fired.update(reused_from_run_id=predecessor_id, reused_recorded_at=fired["reused_recorded_at"]) + assert scheduled == expected_scheduled + assert fired == expected_fired original = one(predecessor, "ActivityCompleted", step=1)["payload"] reused = one(events, "ActivityCompleted", step=1)["payload"] assert reused["result"] == original["result"] @@ -94,15 +96,25 @@ def definition(source: str) -> type: async def cold_worker(packet: dict[str, Any]) -> None: callbacks: list[dict[str, Any]] = [] completions: list[dict[str, Any]] = [] + registration: dict[str, Any] = {} class ObservedClient(Client): + async def register_worker(self, **kwargs: Any) -> Any: + acknowledgement = await super().register_worker(**kwargs) + registration.update(deepcopy({key: value for key, value in kwargs.items() if key != "process_metrics"})) + return acknowledgement + async def complete_workflow_task(self, **kwargs: Any) -> Any: completions.append({"task_id": kwargs["task_id"], "commands": deepcopy(kwargs["commands"])}) return await super().complete_workflow_task(**kwargs) @activity.defn(name="tests.redrive.first") async def first(name: str) -> str: - callbacks.append({"activity": "first", "pid": os.getpid()}) + info = activity.context().info + callbacks.append({ + "activity": "first", "pid": os.getpid(), "run_id": info.run_id, + "task_id": info.task_id, "attempt_id": info.activity_attempt_id, "execution_id": info.activity_execution_id, + }) return f"Hello, {name}!" @activity.defn(name="tests.redrive.second") @@ -130,18 +142,17 @@ async def second(greeting: str, clock: str) -> str: print("COLD_WORKER_RESULT=" + json.dumps({ "pid": os.getpid(), "worker_id": worker.worker_id, "fingerprints": fingerprints, "definition_sha256": DEFINITION_SHA, "status": outcome.status, - "callbacks": callbacks, "completions": completions, + "callbacks": callbacks, "completions": completions, "registration": registration, + "registration_namespace": client.namespace, }, sort_keys=True), flush=True) def run_cold_worker(packet: dict[str, Any]) -> dict[str, Any]: - source = Path(__file__).read_text() - digest = hashlib.sha256(source.encode()).hexdigest() - assert digest == os.environ["DURABLE_WORKFLOW_REDRIVE_CASE_SHA256"] - packet = {**packet, "source": source, "source_sha256": digest} + assert _COLD_BOOTSTRAP and _COLD_INVENTORY and _COLD_PYTHON environment = {"PATH": os.defpath, "PYTHONDONTWRITEBYTECODE": "1"} result = subprocess.run( - [sys.executable, "-I", "-B", "-c", CHILD_ENTRY], input=json.dumps(packet), + [_COLD_PYTHON, "-I", "-S", "-B", "-c", _COLD_BOOTSTRAP], + input=json.dumps({"inventory": _COLD_INVENTORY, "worker": packet}), env=environment, capture_output=True, text=True, timeout=55, check=False, ) assert result.returncode == 0, (result.returncode, result.stdout[-16384:], result.stderr[-16384:]) @@ -151,19 +162,132 @@ def run_cold_worker(packet: dict[str, Any]) -> dict[str, Any]: return json.loads(lines[0]) -def verify_worker(observed: dict[str, Any], *, redrive: bool, clock: datetime) -> None: +def verify_export(document: dict[str, Any], workflow_id: str, run_id: str, *, terminal: bool, source: bool) -> None: + assert document["schema"] == "durable-workflow.v2.history-export" and document["schema_version"] == 2 + selected = document["workflow"] + assert selected["instance_id"] == workflow_id and selected["run_id"] == run_id + assert selected["workflow_type"] == WORKFLOW_TYPE + assert selected["details_state"] == "retained" and selected["details_pruned_at"] is None + assert document["history_complete"] is terminal + assert document["selected_run"]["waits_projection_source"] == "workflow_run_waits" + assert document["redaction"]["applied"] is False + for section, kind in (("tasks", "type"), ("waits", "kind")): + rows = document[section] + assert isinstance(rows, list) + assert all(isinstance(row, dict) and isinstance(row[kind], str) and row[kind] for row in rows) + assert all(isinstance(row["status"], str) and row["status"] for row in rows) + assert sum(row[kind] == "timer" for row in rows) == (1 if source else 0) + + +def verify_recorded_fingerprints(observations: list[dict[str, Any]], histories: list[list[dict[str, Any]]]) -> None: + assert len(observations) == len(histories) == 3 + expected = observations[0]["fingerprints"][WORKFLOW_TYPE] + assert isinstance(expected, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", expected) + for observed, events in zip(observations, histories, strict=True): + assert observed["fingerprints"] == {WORKFLOW_TYPE: expected} + started = one(events, "WorkflowStarted")["payload"] + assert started["workflow_definition_fingerprint"] == expected + assert started["workflow_definition_fingerprint_source"] == "worker" + + +def verify_worker( + observed: dict[str, Any], *, redrive: bool, clock: datetime, queue: str, exported: dict[str, Any], +) -> None: assert observed["definition_sha256"] == DEFINITION_SHA assert set(observed["fingerprints"]) == {WORKFLOW_TYPE} + assert re.fullmatch(r"sha256:[0-9a-f]{64}", observed["fingerprints"][WORKFLOW_TYPE]) assert observed["worker_id"] and isinstance(observed["pid"], int) and observed["pid"] > 0 - assert observed["completions"] - commands = [command for completion in observed["completions"] for command in completion["commands"]] - assert sum(command["type"] == "start_timer" for command in commands) == (0 if redrive else 1) - expected = ["tests.redrive.second"] if redrive else ["tests.redrive.first", "tests.redrive.second"] - assert [command["activity_type"] for command in commands if command["type"] == "schedule_activity"] == expected + assert all(set(completion) == {"task_id", "commands"} for completion in observed["completions"]) + task_ids = [completion["task_id"] for completion in observed["completions"]] + assert all(isinstance(identity, str) and identity for identity in task_ids) + assert len(set(task_ids)) == len(task_ids) == (2 if redrive else 4) + assert set(task_ids) == {task["id"] for task in exported["tasks"] if task["type"] == "workflow"} + counts = {"workflow": 2, "activity": 1} if redrive else {"workflow": 4, "activity": 2, "timer": 1} + assert Counter(task["type"] for task in exported["tasks"]) == counts assert [item["activity"] for item in observed["callbacks"]] == (["second"] if redrive else ["first", "second"]) + for callback in observed["callbacks"]: + assert all(isinstance(callback[name], str) and callback[name] for name in ( + "task_id", "attempt_id", "execution_id", "run_id", + )) + assert callback["run_id"] == exported["workflow"]["run_id"] + assert {callback["task_id"] for callback in observed["callbacks"]} == { + task["id"] for task in exported["tasks"] if task["type"] == "activity" + } + assert all(len(completion["commands"]) == 1 for completion in observed["completions"]) + commands = [command for completion in observed["completions"] for command in completion["commands"]] + clock_text = clock.isoformat(timespec="microseconds") + scheduled = ScheduleActivity( + "tests.redrive.second", ["Hello, Taylor!", clock_text], retry_policy=ActivityRetryPolicy(max_attempts=1), + ).to_server_command(queue) + prefix = [scheduled] if redrive else [ + ScheduleActivity("tests.redrive.first", ["Taylor"]).to_server_command(queue), + StartTimer(1).to_server_command(queue), scheduled, + ] + assert commands[:-1] == prefix + terminal = commands[-1] + if observed["status"] == "completed": + assert terminal == CompleteWorkflow({ + "greeting": "Hello, Taylor!", "timer_clock": clock_text, "resumed": "resumed", + }).to_server_command(queue) + else: + required = {"type", "message", "failed_step_sequence", "failed_activity_execution_id"} + allowed = required | {"exception_type", "exception_class", "exception", "non_retryable"} + assert required <= set(terminal) <= allowed + assert terminal["type"] == "fail_workflow" and terminal["failed_step_sequence"] == 3 + assert isinstance(terminal["message"], str) and terminal["message"] + assert terminal["failed_activity_execution_id"] == observed["callbacks"][-1]["execution_id"] assert instant(observed["callbacks"][-1]["clock"]) == clock +async def admit_redrive( + client: Client, queue: str, run_id: str, observed: dict[str, Any], events: list[dict[str, Any]], +): + started = one(events, "WorkflowStarted")["payload"] + fingerprint = observed["fingerprints"][WORKFLOW_TYPE] + assert re.fullmatch(r"sha256:[0-9a-f]{64}", fingerprint) + assert started["workflow_definition_fingerprint"] == fingerprint + assert started["workflow_definition_fingerprint_source"] == "worker" + assert client.namespace == observed["registration_namespace"] == "default" + before = [run.run_id for run in (await client.list_workflow_runs(queue)).runs] + try: + await client.redrive_workflow(queue, run_id) + except ServerError as error: + assert error.status == 409 and error.reason() == "workflow_definition_unavailable_or_changed" + assert isinstance(error.body, dict) and error.body["outcome"] == "rejected" + assert error.body["workflow_id"] == queue and error.body["run_id"] == run_id + assert error.body["accepted"] is False and error.body["workflow_run_id"] is None + assert error.body["task_id"] is None + refusal = {"status": error.status, "body": error.body} + else: + raise AssertionError("redrive unexpectedly admitted without a matching registration") + assert [run.run_id for run in (await client.list_workflow_runs(queue)).runs] == before + assert [event async for event in client.iter_history(queue, run_id)] == events + request = deepcopy(observed["registration"]) + assert request["task_queue"] == queue and request["supported_workflow_types"] == [WORKFLOW_TYPE] + assert request["workflow_definition_fingerprints"] == {WORKFLOW_TYPE: fingerprint} + holder_id = "python-clock-admission-" + uuid4().hex + request["worker_id"] = holder_id + acknowledgement = await client.register_worker(**request) + token = acknowledgement.get("registration_token") + assert isinstance(token, str) and re.fullmatch(r"[0-9a-f]{32}", token) + try: + assert acknowledgement["worker_id"] == holder_id and acknowledgement["registered"] is True + assert acknowledgement["namespace"] == "default" and acknowledgement["task_queue"] == queue + # The holder never creates a Worker or polls: only the later cold child + # may claim and execute the successor's task. + receipt = await client.redrive_workflow(queue, run_id) + finally: + removed = await client.deregister_worker_registration( + holder_id, registration_token=token, expected_namespace="default", timeout=10, + ) + assert removed["outcome"] == "deregistered" and removed["recovered_workflow_task_count"] == 0 + return receipt, { + "absent_registration": refusal, "matching_registration": request, + "registration_token_sha256": hashlib.sha256(token.encode()).hexdigest(), + "cleanup": {key: value for key, value in removed.items() if key != "registration_token"}, + } + + @pytest.mark.skipif( os.environ.get("DURABLE_WORKFLOW_REDRIVE_CELL") != "1", reason="reviewed manual published cell only", ) @@ -177,6 +301,8 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr "workflow_id": queue, "definition": source} observations = [] histories = [] + exports = [] + admissions = [] run_ids = [] async with Client(server_url, token=server_token, namespace="default") as client: handle = await client.start_workflow( @@ -192,9 +318,12 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr source_start = instant(original.started_at) source_clock = instant(one(histories[0], "TimerFired", step=2)["recorded_at"]) assert source_clock.microsecond != 0, "real TimerFired must expose nonzero source precision" + exports.append(await client.export_history(queue, handle.run_id)) + verify_export(exports[0], queue, handle.run_id, terminal=True, source=True) used_timer_ids = {one(histories[0], "TimerFired", step=2)["payload"]["timer_id"]} for repeat in (1, 2): - receipt = await client.redrive_workflow(queue, run_ids[-1]) + receipt, admission = await admit_redrive(client, queue, run_ids[-1], observations[-1], histories[-1]) + admissions.append(admission) assert receipt.outcome == "redriven" and receipt.run_id and receipt.run_id not in run_ids run_id = receipt.run_id prefix = [event async for event in client.iter_history(queue, run_id)] @@ -205,23 +334,28 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr used_timer_ids.add(timer_id) before = await client.describe_workflow_run(queue, run_id) assert before.wait_kind != "timer" + admitted_export = await client.export_history(queue, run_id) + verify_export(admitted_export, queue, run_id, terminal=False, source=False) observed = await asyncio.to_thread(run_cold_worker, {**common, "fail_second": repeat == 1}) assert observed["status"] == ("failed" if repeat == 1 else "completed") - verify_worker(observed, redrive=True, clock=source_clock) + completed_export = await client.export_history(queue, run_id) + verify_export(completed_export, queue, run_id, terminal=True, source=False) + verify_worker(observed, redrive=True, clock=source_clock, queue=queue, exported=completed_export) assert observed["callbacks"][0]["run_id"] == run_id observations.append(observed) run_ids.append(run_id) histories.append([event async for event in client.iter_history(queue, run_id)]) + exports.append({"before": admitted_export, "after": completed_export}) assert histories[-1][:len(prefix)] == prefix assert [event async for event in client.iter_history(queue, handle.run_id)] == frozen_source result = await client.get_workflow_handle(queue, run_id=run_ids[-1]).result(timeout=10) - verify_worker(observations[0], redrive=False, clock=source_clock) + verify_worker(observations[0], redrive=False, clock=source_clock, queue=queue, exported=exports[0]) assert len({item["pid"] for item in observations}) == len({item["worker_id"] for item in observations}) == 3 - assert len({json.dumps(item["fingerprints"], sort_keys=True) for item in observations}) == 1 + verify_recorded_fingerprints(observations, histories) assert result == {"greeting": "Hello, Taylor!", "timer_clock": source_clock.isoformat(timespec="microseconds"), "resumed": "resumed"} print(json.dumps({ "scope": "published cold-Worker timer redrive", "runs": run_ids, "source_started_at": original.started_at, "source_timer_clock": source_clock.isoformat(), "definition_sha256": DEFINITION_SHA, - "workers": observations, "histories": histories, + "workers": observations, "histories": histories, "exports": exports, "admissions": admissions, }, sort_keys=True)) diff --git a/tests/test_activity_acceptance_entry.py b/tests/test_activity_acceptance_entry.py new file mode 100644 index 0000000..15de3bc --- /dev/null +++ b/tests/test_activity_acceptance_entry.py @@ -0,0 +1,113 @@ +"""Inert first-open controls; no SDK, API, Worker or Docker invocation.""" + +from __future__ import annotations + +import hashlib +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest +from scripts.ci import activity_acceptance_consumer as consumer +from scripts.ci import activity_acceptance_entry as entry + + +@pytest.fixture +def sandbox(tmp_path, monkeypatch): + environment, data = tmp_path / "environment", tmp_path / "data" + site = environment / "lib" / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor)) + site = site / "site-packages" + site.mkdir(parents=True) + data.mkdir() + case = data / "test_captured.py" + case.write_text("async def cold_worker(p): print('fixed inert result', p['marker'])\n") + source = Path(entry.__file__).read_text() + source = source.replace('Path("/run/acceptance-env")', "Path(" + repr(str(environment)) + ")") + source = source.replace('Path("/run/acceptance-data")', "Path(" + repr(str(data)) + ")") + monkeypatch.setattr(consumer, "CAPTURED_ENTRY", source) + monkeypatch.setattr(entry, "ENVIRONMENT", environment) + monkeypatch.setattr(entry, "DATA", data) + monkeypatch.setattr(entry, "CASE_NAME", case.name) + return environment, data, site, case + + +def launch(case, installed, *, no_site=True): + source = consumer.entry_source( + "worker", case.name, hashlib.sha256(case.read_bytes()).hexdigest(), installed, + ) + arguments = [sys.executable, "-I", *(["-S"] if no_site else []), "-B", "-c", source] + return subprocess.run( + arguments, input=json.dumps({"inventory": installed, "worker": {"marker": "owned"}}), + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, + ) + + +def test_no_site_entry_can_execute_only_the_verified_inert_source(sandbox): + environment, _data, _site, case = sandbox + installed = entry.snapshot(environment) + result = launch(case, installed) + assert result.returncode == 0 and result.stdout == "fixed inert result owned\n" + + +def test_entry_refuses_normal_site_even_when_other_inputs_match(sandbox): + environment, _data, _site, case = sandbox + result = launch(case, entry.snapshot(environment), no_site=False) + assert result.returncode != 0 and "no-site Python" in result.stderr + assert not result.stdout + + +def test_added_startup_hook_refuses_before_any_case_code(sandbox): + environment, _data, site, case = sandbox + installed = entry.snapshot(environment) + (site / "sitecustomize.py").write_text("raise RuntimeError('startup replacement executed')\n") + result = launch(case, installed) + assert result.returncode != 0 and "install inventory changed" in result.stderr + assert "startup replacement executed" not in result.stderr and not result.stdout + + +def test_predecessor_runs_pth_before_guard_successor_refuses_source_before_effect(sandbox, tmp_path): + environment, _data, site, case = sandbox + subprocess.run( + [sys.executable, "-I", "-B", "-m", "venv", "--without-pip", "--copies", str(environment)], + check=True, capture_output=True, text=True, timeout=30, + ) + marker = tmp_path / "pre-open-marker" + (site / "preopen.pth").write_text( + "import pathlib; pathlib.Path(" + repr(str(marker)) + ").write_text('pre-open')\n", + ) + installed = entry.snapshot(environment) + source = consumer.entry_source("worker", case.name, hashlib.sha256(case.read_bytes()).hexdigest(), installed) + old = subprocess.run( + [str(environment / "bin/python"), "-I", "-B", "-c", "raise RuntimeError('late guard')"], + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, + ) + assert old.returncode != 0 and marker.read_text() == "pre-open" + marker.unlink() + case.write_text("raise RuntimeError('foreign case executed')\n") + corrected = subprocess.run( + [sys.executable, "-I", "-S", "-B", "-c", source], + input=json.dumps({"inventory": installed, "worker": {"marker": "owned"}}), + env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, + ) + assert corrected.returncode != 0 and "captured case source differs" in corrected.stderr + assert "foreign case executed" not in corrected.stderr and not marker.exists() and not corrected.stdout + + +def test_post_capture_path_replacement_keeps_the_verified_bytes(sandbox): + _environment, _data, _site, case = sandbox + case.write_text("chosen = 'captured'\n") + source = entry.guarded_source(case, hashlib.sha256(case.read_bytes()).hexdigest()) + case.write_text("raise RuntimeError('replacement must not execute')\n") + assert entry.source_module(source).chosen == "captured" + + +def test_pytest_collector_uses_captured_bytes_and_no_mutable_conftest(sandbox, monkeypatch): + _environment, data, _site, case = sandbox + source = b"def test_captured(): assert 2 + 2 == 4\n" + case.write_text("raise RuntimeError('mutable test must not execute')\n") + (data / "conftest.py").write_text("raise RuntimeError('mutable conftest must not execute')\n") + monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + assert entry.run_pytest(source, {}) == 0 + assert (data / "identity.xml").is_file() diff --git a/tests/test_native_timer_redrive_cell.py b/tests/test_native_timer_redrive_cell.py index 004e0dd..ac8383d 100644 --- a/tests/test_native_timer_redrive_cell.py +++ b/tests/test_native_timer_redrive_cell.py @@ -4,11 +4,9 @@ import hashlib import json -import os -import subprocess -import sys from copy import deepcopy from pathlib import Path +from types import SimpleNamespace import pytest @@ -48,6 +46,7 @@ def test_exact_fraction_and_immediate_predecessor_preserve_unchanged_inputs(): @pytest.mark.parametrize("fault", [ "floor", "root_lineage", "new_start", "pair_id", "old_id", "timer_task", "scheduled_activity", "changed_result", + "extra_fired", "extra_scheduled", "changed_delay", ]) def test_false_reuse_evidence_refuses(fault): events, previous, arguments = prefix() @@ -66,6 +65,12 @@ def test_false_reuse_evidence_refuses(fault): fired["task"] = {"id": "new-timer-work"} elif fault == "scheduled_activity": events.append({"event_type": "ActivityScheduled", "payload": {"sequence": 1}}) + elif fault == "extra_fired": + fired["unreviewed"] = True + elif fault == "extra_scheduled": + cell.one(events, "TimerScheduled")["payload"]["unreviewed"] = True + elif fault == "changed_delay": + cell.one(events, "TimerScheduled")["payload"]["delay_seconds"] = 999 else: cell.one(events, "ActivityCompleted")["payload"]["result"] = "changed" with pytest.raises(AssertionError): @@ -78,45 +83,44 @@ def test_unrepresentable_or_unzoned_clock_cannot_hide_precision_loss(clock): cell.instant(clock) -def test_changed_definition_or_case_source_refuses_before_child(monkeypatch): +def test_changed_definition_or_absent_trusted_entry_refuses_before_child(monkeypatch): with pytest.raises(AssertionError): cell.definition("changed definition") - monkeypatch.setenv("DURABLE_WORKFLOW_REDRIVE_CASE_SHA256", "0" * 64) monkeypatch.setattr(cell.subprocess, "run", lambda *a, **k: pytest.fail("must refuse before any child")) with pytest.raises(AssertionError): cell.run_cold_worker({}) -@pytest.mark.parametrize("substitute", [False, True]) -def test_literal_child_bootstrap_binds_captured_bytes_without_opening_a_dispatcher(substitute): - source = "async def cold_worker(p): print('fixed inert result', p['marker'])\n" - packet = {"source": source, "source_sha256": hashlib.sha256(source.encode()).hexdigest(), "marker": "owned"} - if substitute: - packet["source"] += "\nraise RuntimeError('substitute executed')\n" - result = subprocess.run( - [sys.executable, "-I", "-B", "-c", cell.CHILD_ENTRY], input=json.dumps(packet), - env={"PATH": os.defpath}, capture_output=True, text=True, timeout=5, - ) - if substitute: - assert result.returncode != 0 and "source differs" in result.stderr - assert "substitute executed" not in result.stderr and not result.stdout - else: - assert result.returncode == 0 and result.stdout == "fixed inert result owned\n" - - def test_accepted_workflow_definition_bytes_remain_frozen(): path = Path(__file__).parent / "integration/redrive_timer_clock_workflow.py" assert hashlib.sha256(path.read_bytes()).hexdigest() == cell.DEFINITION_SHA -@pytest.mark.parametrize("fault", [None, "timer_command", "first_callback", "no_task", "floor"]) +@pytest.mark.parametrize("fault", [ + None, "timer_command", "extra_command", "extra_completion", "missing_task", "foreign_task", + "first_callback", "no_task", "floor", "empty_fingerprint", +]) def test_redrive_worker_evidence_requires_only_the_failed_activity_and_exact_clock(fault): clock = cell.instant("2026-11-03T00:00:06.654321Z") observed = { "definition_sha256": cell.DEFINITION_SHA, "fingerprints": {cell.WORKFLOW_TYPE: "sha256:" + "a" * 64}, "worker_id": "fixed-worker", "pid": 123, - "completions": [{"commands": [{"type": "schedule_activity", "activity_type": "tests.redrive.second"}]}], - "callbacks": [{"activity": "second", "clock": clock.isoformat()}], + "status": "failed", + "completions": [ + {"task_id": "one", "commands": [cell.ScheduleActivity( + "tests.redrive.second", ["Hello, Taylor!", clock.isoformat(timespec="microseconds")], + retry_policy=cell.ActivityRetryPolicy(max_attempts=1), + ).to_server_command("queue")]}, + {"task_id": "two", "commands": [{"type": "fail_workflow", "message": "fixed failure", + "failed_step_sequence": 3, "failed_activity_execution_id": "execution"}]}, + ], + "callbacks": [{"activity": "second", "clock": clock.isoformat(), "execution_id": "execution", + "task_id": "activity", "attempt_id": "attempt", "run_id": "run"}], + } + exported = { + "workflow": {"run_id": "run"}, + "tasks": [{"id": "one", "type": "workflow"}, {"id": "two", "type": "workflow"}, + {"id": "activity", "type": "activity"}], } if fault == "timer_command": observed["completions"][0]["commands"].append({"type": "start_timer"}) @@ -124,10 +128,140 @@ def test_redrive_worker_evidence_requires_only_the_failed_activity_and_exact_clo observed["callbacks"].insert(0, {"activity": "first"}) elif fault == "no_task": observed["completions"] = [] + elif fault == "extra_command": + observed["completions"][0]["commands"].append({"type": "upsert_search_attributes"}) + elif fault == "extra_completion": + observed["completions"].append(deepcopy(observed["completions"][0])) + elif fault == "missing_task": + observed["completions"][0]["task_id"] = "" + elif fault == "foreign_task": + observed["completions"][0]["task_id"] = "foreign" + elif fault == "empty_fingerprint": + observed["fingerprints"][cell.WORKFLOW_TYPE] = "" elif fault == "floor": observed["callbacks"][0]["clock"] = "2026-11-03T00:00:06Z" if fault: with pytest.raises(AssertionError): - cell.verify_worker(observed, redrive=True, clock=clock) + cell.verify_worker(observed, redrive=True, clock=clock, queue="queue", exported=exported) + else: + cell.verify_worker(observed, redrive=True, clock=clock, queue="queue", exported=exported) + + +@pytest.mark.parametrize("fault", [None, "empty", "changed", "source"]) +def test_recorded_worker_fingerprint_is_exact_across_original_and_successors(fault): + fingerprint = "sha256:" + "a" * 64 + observations = [{"fingerprints": {cell.WORKFLOW_TYPE: fingerprint}} for _ in range(3)] + histories = [[{"event_type": "WorkflowStarted", "payload": { + "workflow_definition_fingerprint": fingerprint, "workflow_definition_fingerprint_source": "worker", + }}] for _ in range(3)] + if fault == "empty": + observations[0]["fingerprints"][cell.WORKFLOW_TYPE] = "" + elif fault == "changed": + histories[1][0]["payload"]["workflow_definition_fingerprint"] = "sha256:" + "b" * 64 + elif fault == "source": + histories[2][0]["payload"]["workflow_definition_fingerprint_source"] = "unrelated" + if fault: + with pytest.raises(AssertionError): + cell.verify_recorded_fingerprints(observations, histories) + else: + cell.verify_recorded_fingerprints(observations, histories) + + +@pytest.mark.parametrize("fault", [None, "done_timer", "cancelled_wait", "pruned", "run", "redacted", "incomplete"]) +def test_export_requires_bound_retained_all_status_inventory(fault): + document = { + "schema": "durable-workflow.v2.history-export", "schema_version": 2, "history_complete": True, + "workflow": {"instance_id": "workflow", "run_id": "run", "workflow_type": cell.WORKFLOW_TYPE, + "details_state": "retained", "details_pruned_at": None}, + "selected_run": {"waits_projection_source": "workflow_run_waits"}, "redaction": {"applied": False}, + "tasks": [{"id": "workflow-task", "type": "workflow", "status": "completed"}], "waits": [], + "timers": [{"id": "history-reconstructed-pair"}], + } + if fault == "done_timer": + document["tasks"].append({"id": "timer", "type": "timer", "status": "completed"}) + elif fault == "cancelled_wait": + document["waits"].append({"kind": "timer", "status": "cancelled"}) + elif fault == "pruned": + document["workflow"]["details_state"] = "pruned" + elif fault == "run": + document["workflow"]["run_id"] = "foreign" + elif fault == "redacted": + document["redaction"]["applied"] = True + elif fault == "incomplete": + document["history_complete"] = False + if fault: + with pytest.raises(AssertionError): + cell.verify_export(document, "workflow", "run", terminal=True, source=False) + else: + cell.verify_export(document, "workflow", "run", terminal=True, source=False) + + +@pytest.mark.parametrize("fault", [None, "absent_admitted", "wrong_refusal", "positive_failure"]) +async def test_matching_nonconsuming_registration_is_required_and_fenced_cleanup_runs(fault): + fingerprint = "sha256:" + "a" * 64 + events = [{"event_type": "WorkflowStarted", "payload": { + "workflow_definition_fingerprint": fingerprint, "workflow_definition_fingerprint_source": "worker", + }}] + observed = { + "fingerprints": {cell.WORKFLOW_TYPE: fingerprint}, "registration_namespace": "default", + "registration": {"worker_id": "original", "task_queue": "queue", + "supported_workflow_types": [cell.WORKFLOW_TYPE], + "workflow_definition_fingerprints": {cell.WORKFLOW_TYPE: fingerprint}, + "capability_manifest": {}}, + } + + class FakeClient: + namespace = "default" + + def __init__(self): + self.calls = [] + self.registered = False + + async def list_workflow_runs(self, queue): + return SimpleNamespace(runs=[SimpleNamespace(run_id="source")]) + + async def iter_history(self, queue, run_id): + for event in events: + yield event + + async def register_worker(self, **request): + self.calls.append(("register", request)) + self.registered = True + self.worker_id = request["worker_id"] + return {"worker_id": self.worker_id, "registered": True, "registration_token": "b" * 32, + "namespace": "default", "task_queue": "queue"} + + async def redrive_workflow(self, queue, run_id): + self.calls.append(("redrive", self.registered)) + if not self.registered and fault != "absent_admitted": + raise cell.ServerError(409, { + "reason": "other" if fault == "wrong_refusal" else "workflow_definition_unavailable_or_changed", + "outcome": "rejected", "workflow_id": queue, "run_id": run_id, + "workflow_run_id": None, "accepted": False, "task_id": None, + }) + if self.registered and fault == "positive_failure": + raise cell.ServerError(503, {"reason": "fixed admission failure"}) + return SimpleNamespace(outcome="redriven", run_id="next") + + async def deregister_worker_registration(self, worker_id, **arguments): + assert worker_id == self.worker_id and arguments == { + "registration_token": "b" * 32, "expected_namespace": "default", "timeout": 10, + } + self.calls.append(("deregister", worker_id)) + self.registered = False + return {"worker_id": worker_id, "registration_token": "b" * 32, + "outcome": "deregistered", "recovered_workflow_task_count": 0} + + client = FakeClient() + if fault: + with pytest.raises(cell.ServerError if fault == "positive_failure" else AssertionError): + await cell.admit_redrive(client, "queue", "source", observed, events) + assert not client.registered else: - cell.verify_worker(observed, redrive=True, clock=clock) + receipt, evidence = await cell.admit_redrive(client, "queue", "source", observed, events) + assert receipt.run_id == "next" and not client.registered + assert [name for name, _value in client.calls] == ["redrive", "register", "redrive", "deregister"] + assert "b" * 32 not in json.dumps(evidence) + assert evidence["matching_registration"]["workflow_definition_fingerprints"] == { + cell.WORKFLOW_TYPE: fingerprint, + } From 88dcd098362419f22ee6e2c99a32abba38e43781 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 03:45:50 +0000 Subject: [PATCH 3/5] Preserve literal source and inherited redrive clock strings --- scripts/ci/published-timer-redrive.md | 3 ++ .../integration/test_native_timer_redrive.py | 16 ++++++- tests/test_native_timer_redrive_cell.py | 42 +++++++++++++++++++ 3 files changed, 59 insertions(+), 2 deletions(-) diff --git a/scripts/ci/published-timer-redrive.md b/scripts/ci/published-timer-redrive.md index 9659f2d..d04c908 100644 --- a/scripts/ci/published-timer-redrive.md +++ b/scripts/ci/published-timer-redrive.md @@ -35,6 +35,9 @@ successors must each count zero before and after their cold Worker. All emitted completion task IDs and activity callback task IDs must match the stored export inventory. Full timer payloads must match their predecessor modulo the explicitly authorized new pair ID, immediate predecessor and preserved clock fields. +First reuse preserves the authoritative source start and TimerFired timestamp +strings exactly. Later reuse preserves the immediate predecessor's inherited +strings exactly as well as comparing their parsed UTC instants. An absent matching registration must refuse redrive without changing the chain or source history. Admission then holds a matching source-cohort registration diff --git a/tests/integration/test_native_timer_redrive.py b/tests/integration/test_native_timer_redrive.py index a823d9f..e50481a 100644 --- a/tests/integration/test_native_timer_redrive.py +++ b/tests/integration/test_native_timer_redrive.py @@ -51,12 +51,17 @@ def one(events: list[dict[str, Any]], kind: str, *, step: int | None = None) -> def verify_reuse( events: list[dict[str, Any]], predecessor: list[dict[str, Any]], *, predecessor_id: str, source_clock: datetime, source_start: datetime, used_timer_ids: set[str], + source_clock_text: str, source_start_text: str, ) -> str: started = one(events, "WorkflowStarted")["payload"] assert started["continued_from_run_id"] == predecessor_id assert started["recovery_kind"] == "redrive" assert started["resume_step_sequence"] == 3 assert instant(started["replayed_started_at"]) == source_start + inherited_start = one(predecessor, "WorkflowStarted")["payload"].get("replayed_started_at") + expected_start = source_start_text if inherited_start is None else inherited_start + assert isinstance(expected_start, str) and started["replayed_started_at"] == expected_start + assert instant(source_start_text) == source_start scheduled = one(events, "TimerScheduled", step=2)["payload"] fired = one(events, "TimerFired", step=2)["payload"] timer_id = scheduled["timer_id"] @@ -66,6 +71,10 @@ def verify_reuse( assert instant(fired["reused_recorded_at"]) == source_clock expected_scheduled = deepcopy(one(predecessor, "TimerScheduled", step=2)["payload"]) expected_fired = deepcopy(one(predecessor, "TimerFired", step=2)["payload"]) + inherited_clock = expected_fired.get("reused_recorded_at") + expected_clock = source_clock_text if inherited_clock is None else inherited_clock + assert isinstance(expected_clock, str) and fired["reused_recorded_at"] == expected_clock + assert instant(source_clock_text) == source_clock for payload in (expected_scheduled, expected_fired): payload.pop("task", None) payload.pop("command", None) @@ -316,7 +325,8 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr frozen_source = deepcopy(histories[0]) original = await client.describe_workflow_run(queue, handle.run_id) source_start = instant(original.started_at) - source_clock = instant(one(histories[0], "TimerFired", step=2)["recorded_at"]) + source_clock_text = one(histories[0], "TimerFired", step=2)["recorded_at"] + source_clock = instant(source_clock_text) assert source_clock.microsecond != 0, "real TimerFired must expose nonzero source precision" exports.append(await client.export_history(queue, handle.run_id)) verify_export(exports[0], queue, handle.run_id, terminal=True, source=True) @@ -330,6 +340,7 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr timer_id = verify_reuse( prefix, histories[-1], predecessor_id=run_ids[-1], source_clock=source_clock, source_start=source_start, used_timer_ids=used_timer_ids, + source_start_text=original.started_at, source_clock_text=source_clock_text, ) used_timer_ids.add(timer_id) before = await client.describe_workflow_run(queue, run_id) @@ -356,6 +367,7 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr "resumed": "resumed"} print(json.dumps({ "scope": "published cold-Worker timer redrive", "runs": run_ids, "source_started_at": original.started_at, - "source_timer_clock": source_clock.isoformat(), "definition_sha256": DEFINITION_SHA, + "source_timer_clock": source_clock_text, "source_timer_clock_utc": source_clock.isoformat(), + "definition_sha256": DEFINITION_SHA, "workers": observations, "histories": histories, "exports": exports, "admissions": admissions, }, sort_keys=True)) diff --git a/tests/test_native_timer_redrive_cell.py b/tests/test_native_timer_redrive_cell.py index ac8383d..ce445eb 100644 --- a/tests/test_native_timer_redrive_cell.py +++ b/tests/test_native_timer_redrive_cell.py @@ -21,6 +21,9 @@ def prefix(): for event in previous["history"]: if event["event_type"] in {"TimerScheduled", "TimerFired"}: event["payload"]["timer_id"] = "old-timer" + event["payload"].pop("reused_recorded_at", None) + event["payload"].pop("reused_from_run_id", None) + cell.one(previous["history"], "WorkflowStarted")["payload"].pop("replayed_started_at", None) events = current["history"] cell.one(events, "WorkflowStarted")["payload"]["continued_from_run_id"] = "previous" fired = cell.one(events, "TimerFired")["payload"] @@ -34,6 +37,7 @@ def prefix(): return events, previous["history"], { "predecessor_id": "previous", "source_clock": source_time, "source_start": source_start, "used_timer_ids": {"old-timer"}, + "source_clock_text": source_time.isoformat(), "source_start_text": source_start.isoformat(), } @@ -44,6 +48,44 @@ def test_exact_fraction_and_immediate_predecessor_preserve_unchanged_inputs(): assert (events, previous, arguments) == original +@pytest.mark.parametrize("reformat", [None, "clock", "start"]) +def test_repeated_reuse_preserves_inherited_strings_as_well_as_instants(reformat): + events, previous, arguments = prefix() + original_clock = arguments["source_clock"].isoformat() + original_start = arguments["source_start"].isoformat() + cell.one(previous, "TimerFired")["payload"]["reused_recorded_at"] = original_clock + cell.one(previous, "WorkflowStarted")["payload"]["replayed_started_at"] = original_start + if reformat == "clock": + cell.one(events, "TimerFired")["payload"]["reused_recorded_at"] = original_clock.replace("+00:00", "Z") + elif reformat == "start": + cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"] = original_start.replace("+00:00", "Z") + if reformat: + assert cell.instant(cell.one(events, "TimerFired")["payload"]["reused_recorded_at"]) == arguments[ + "source_clock" + ] + assert cell.instant(cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"]) == arguments[ + "source_start" + ] + with pytest.raises(AssertionError): + cell.verify_reuse(events, previous, **arguments) + else: + cell.verify_reuse(events, previous, **arguments) + + +@pytest.mark.parametrize("field", ["clock", "start"]) +def test_first_reuse_cannot_reformat_authoritative_source_strings(field): + events, previous, arguments = prefix() + kind, name, source_name = ( + ("TimerFired", "reused_recorded_at", "source_clock") if field == "clock" + else ("WorkflowStarted", "replayed_started_at", "source_start") + ) + payload = cell.one(events, kind)["payload"] + payload[name] = payload[name].replace("+00:00", "Z") + assert cell.instant(payload[name]) == arguments[source_name] + with pytest.raises(AssertionError): + cell.verify_reuse(events, previous, **arguments) + + @pytest.mark.parametrize("fault", [ "floor", "root_lineage", "new_start", "pair_id", "old_id", "timer_task", "scheduled_activity", "changed_result", "extra_fired", "extra_scheduled", "changed_delay", From a14000f177993c8ad9a7c5faccb1d43577a00178 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 03:59:29 +0000 Subject: [PATCH 4/5] fix: bind timer cell to public history timestamp semantics --- scripts/ci/published-timer-redrive.md | 9 ++-- .../integration/test_native_timer_redrive.py | 19 +++++---- tests/test_native_timer_redrive_cell.py | 42 +++++++++++++++++-- 3 files changed, 56 insertions(+), 14 deletions(-) diff --git a/scripts/ci/published-timer-redrive.md b/scripts/ci/published-timer-redrive.md index d04c908..62d2860 100644 --- a/scripts/ci/published-timer-redrive.md +++ b/scripts/ci/published-timer-redrive.md @@ -35,9 +35,12 @@ successors must each count zero before and after their cold Worker. All emitted completion task IDs and activity callback task IDs must match the stored export inventory. Full timer payloads must match their predecessor modulo the explicitly authorized new pair ID, immediate predecessor and preserved clock fields. -First reuse preserves the authoritative source start and TimerFired timestamp -strings exactly. Later reuse preserves the immediate predecessor's inherited -strings exactly as well as comparing their parsed UTC instants. +The paged public history supplies the original TimerFired `timestamp`. +First reuse compares the authoritative start and event clocks as full-precision +UTC instants. Native fallback serialization may change the public API's `Z` to +`+00:00` and omit insignificant fractional digits. Later reuse preserves the +immediate predecessor's inherited strings exactly as well as comparing their +parsed UTC instants. An absent matching registration must refuse redrive without changing the chain or source history. Admission then holds a matching source-cohort registration diff --git a/tests/integration/test_native_timer_redrive.py b/tests/integration/test_native_timer_redrive.py index e50481a..211df95 100644 --- a/tests/integration/test_native_timer_redrive.py +++ b/tests/integration/test_native_timer_redrive.py @@ -48,6 +48,13 @@ def one(events: list[dict[str, Any]], kind: str, *, step: int | None = None) -> return selected[0] +def source_timer_boundary(events: list[dict[str, Any]]) -> tuple[str, datetime]: + text = one(events, "TimerFired", step=2)["timestamp"] + clock = instant(text) + assert clock.microsecond != 0, "real TimerFired must expose nonzero source precision" + return text, clock + + def verify_reuse( events: list[dict[str, Any]], predecessor: list[dict[str, Any]], *, predecessor_id: str, source_clock: datetime, source_start: datetime, used_timer_ids: set[str], @@ -59,8 +66,8 @@ def verify_reuse( assert started["resume_step_sequence"] == 3 assert instant(started["replayed_started_at"]) == source_start inherited_start = one(predecessor, "WorkflowStarted")["payload"].get("replayed_started_at") - expected_start = source_start_text if inherited_start is None else inherited_start - assert isinstance(expected_start, str) and started["replayed_started_at"] == expected_start + if inherited_start is not None: + assert isinstance(inherited_start, str) and started["replayed_started_at"] == inherited_start assert instant(source_start_text) == source_start scheduled = one(events, "TimerScheduled", step=2)["payload"] fired = one(events, "TimerFired", step=2)["payload"] @@ -72,8 +79,8 @@ def verify_reuse( expected_scheduled = deepcopy(one(predecessor, "TimerScheduled", step=2)["payload"]) expected_fired = deepcopy(one(predecessor, "TimerFired", step=2)["payload"]) inherited_clock = expected_fired.get("reused_recorded_at") - expected_clock = source_clock_text if inherited_clock is None else inherited_clock - assert isinstance(expected_clock, str) and fired["reused_recorded_at"] == expected_clock + if inherited_clock is not None: + assert isinstance(inherited_clock, str) and fired["reused_recorded_at"] == inherited_clock assert instant(source_clock_text) == source_clock for payload in (expected_scheduled, expected_fired): payload.pop("task", None) @@ -325,9 +332,7 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr frozen_source = deepcopy(histories[0]) original = await client.describe_workflow_run(queue, handle.run_id) source_start = instant(original.started_at) - source_clock_text = one(histories[0], "TimerFired", step=2)["recorded_at"] - source_clock = instant(source_clock_text) - assert source_clock.microsecond != 0, "real TimerFired must expose nonzero source precision" + source_clock_text, source_clock = source_timer_boundary(histories[0]) exports.append(await client.export_history(queue, handle.run_id)) verify_export(exports[0], queue, handle.run_id, terminal=True, source=True) used_timer_ids = {one(histories[0], "TimerFired", step=2)["payload"]["timer_id"]} diff --git a/tests/test_native_timer_redrive_cell.py b/tests/test_native_timer_redrive_cell.py index ce445eb..79bbfff 100644 --- a/tests/test_native_timer_redrive_cell.py +++ b/tests/test_native_timer_redrive_cell.py @@ -18,12 +18,16 @@ def prefix(): previous, current = successor(), successor() source_time = cell.instant("2026-11-03T00:00:06.654321+00:00") source_start = cell.instant("2026-11-03T00:00:00+00:00") + for document in (previous, current): + for event in document["history"]: + event["timestamp"] = event.pop("recorded_at") for event in previous["history"]: if event["event_type"] in {"TimerScheduled", "TimerFired"}: event["payload"]["timer_id"] = "old-timer" event["payload"].pop("reused_recorded_at", None) event["payload"].pop("reused_from_run_id", None) cell.one(previous["history"], "WorkflowStarted")["payload"].pop("replayed_started_at", None) + cell.one(previous["history"], "TimerFired")["timestamp"] = "2026-11-03T00:00:06.654321Z" events = current["history"] cell.one(events, "WorkflowStarted")["payload"]["continued_from_run_id"] = "previous" fired = cell.one(events, "TimerFired")["payload"] @@ -37,7 +41,7 @@ def prefix(): return events, previous["history"], { "predecessor_id": "previous", "source_clock": source_time, "source_start": source_start, "used_timer_ids": {"old-timer"}, - "source_clock_text": source_time.isoformat(), "source_start_text": source_start.isoformat(), + "source_clock_text": "2026-11-03T00:00:06.654321Z", "source_start_text": "2026-11-03T00:00:00.000000Z", } @@ -72,20 +76,50 @@ def test_repeated_reuse_preserves_inherited_strings_as_well_as_instants(reformat cell.verify_reuse(events, previous, **arguments) +@pytest.mark.parametrize("fraction,rendered", [ + ("000000", ""), ("654000", ".654"), ("654321", ".654321"), +]) +def test_first_reuse_accepts_native_offset_and_precision_rendering(fraction, rendered): + events, previous, arguments = prefix() + arguments["source_clock_text"] = f"2026-11-03T00:00:06.{fraction}Z" + arguments["source_clock"] = cell.instant(arguments["source_clock_text"]) + arguments["source_start_text"] = f"2026-11-03T00:00:00.{fraction}Z" + arguments["source_start"] = cell.instant(arguments["source_start_text"]) + cell.one(events, "TimerFired")["payload"]["reused_recorded_at"] = f"2026-11-03T00:00:06{rendered}+00:00" + cell.one(events, "WorkflowStarted")["payload"]["replayed_started_at"] = f"2026-11-03T00:00:00{rendered}+00:00" + cell.verify_reuse(events, previous, **arguments) + + @pytest.mark.parametrize("field", ["clock", "start"]) -def test_first_reuse_cannot_reformat_authoritative_source_strings(field): +def test_first_reuse_refuses_changed_instant(field): events, previous, arguments = prefix() kind, name, source_name = ( ("TimerFired", "reused_recorded_at", "source_clock") if field == "clock" else ("WorkflowStarted", "replayed_started_at", "source_start") ) payload = cell.one(events, kind)["payload"] - payload[name] = payload[name].replace("+00:00", "Z") - assert cell.instant(payload[name]) == arguments[source_name] + payload[name] = arguments[source_name].replace(microsecond=1).isoformat() with pytest.raises(AssertionError): cell.verify_reuse(events, previous, **arguments) +def test_source_boundary_uses_timestamp_only_public_history_shape(): + events, previous, arguments = prefix() + assert all("timestamp" in event and "recorded_at" not in event for event in previous) + text, clock = cell.source_timer_boundary(previous) + assert (text, clock) == (arguments["source_clock_text"], arguments["source_clock"]) + arguments.update(source_clock_text=text, source_clock=clock) + cell.verify_reuse(events, previous, **arguments) + + +def test_export_recorded_at_cannot_replace_public_timestamp(): + _, previous, _ = prefix() + fired = cell.one(previous, "TimerFired") + fired["recorded_at"] = fired.pop("timestamp") + with pytest.raises(KeyError, match="timestamp"): + cell.source_timer_boundary(previous) + + @pytest.mark.parametrize("fault", [ "floor", "root_lineage", "new_start", "pair_id", "old_id", "timer_task", "scheduled_activity", "changed_result", "extra_fired", "extra_scheduled", "changed_delay", From 0a7cf41f0697ebc591f98b8e5e7e65c45760c591 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 04:04:21 +0000 Subject: [PATCH 5/5] test: report timer qualification assertions individually --- .../integration/test_native_timer_redrive.py | 70 +++++++++++++------ tests/test_activity_acceptance_entry.py | 22 ++++-- tests/test_native_timer_redrive_cell.py | 8 ++- 3 files changed, 68 insertions(+), 32 deletions(-) diff --git a/tests/integration/test_native_timer_redrive.py b/tests/integration/test_native_timer_redrive.py index 211df95..9dc0695 100644 --- a/tests/integration/test_native_timer_redrive.py +++ b/tests/integration/test_native_timer_redrive.py @@ -32,7 +32,8 @@ def instant(value: Any) -> datetime: - assert isinstance(value, str) and value + assert isinstance(value, str) + assert value assert not re.search(r"\.\d{7}", value), "clock precision exceeds Python's exact microsecond representation" result = datetime.fromisoformat(value.replace("Z", "+00:00")) assert result.tzinfo is not None @@ -67,12 +68,15 @@ def verify_reuse( assert instant(started["replayed_started_at"]) == source_start inherited_start = one(predecessor, "WorkflowStarted")["payload"].get("replayed_started_at") if inherited_start is not None: - assert isinstance(inherited_start, str) and started["replayed_started_at"] == inherited_start + assert isinstance(inherited_start, str) + assert started["replayed_started_at"] == inherited_start assert instant(source_start_text) == source_start scheduled = one(events, "TimerScheduled", step=2)["payload"] fired = one(events, "TimerFired", step=2)["payload"] timer_id = scheduled["timer_id"] - assert isinstance(timer_id, str) and timer_id and timer_id not in used_timer_ids + assert isinstance(timer_id, str) + assert timer_id + assert timer_id not in used_timer_ids assert fired["timer_id"] == timer_id assert fired["reused_from_run_id"] == predecessor_id assert instant(fired["reused_recorded_at"]) == source_clock @@ -80,7 +84,8 @@ def verify_reuse( expected_fired = deepcopy(one(predecessor, "TimerFired", step=2)["payload"]) inherited_clock = expected_fired.get("reused_recorded_at") if inherited_clock is not None: - assert isinstance(inherited_clock, str) and fired["reused_recorded_at"] == inherited_clock + assert isinstance(inherited_clock, str) + assert fired["reused_recorded_at"] == inherited_clock assert instant(source_clock_text) == source_clock for payload in (expected_scheduled, expected_fired): payload.pop("task", None) @@ -164,7 +169,9 @@ async def second(greeting: str, clock: str) -> str: def run_cold_worker(packet: dict[str, Any]) -> dict[str, Any]: - assert _COLD_BOOTSTRAP and _COLD_INVENTORY and _COLD_PYTHON + assert _COLD_BOOTSTRAP + assert _COLD_INVENTORY + assert _COLD_PYTHON environment = {"PATH": os.defpath, "PYTHONDONTWRITEBYTECODE": "1"} result = subprocess.run( [_COLD_PYTHON, "-I", "-S", "-B", "-c", _COLD_BOOTSTRAP], @@ -179,11 +186,14 @@ def run_cold_worker(packet: dict[str, Any]) -> dict[str, Any]: def verify_export(document: dict[str, Any], workflow_id: str, run_id: str, *, terminal: bool, source: bool) -> None: - assert document["schema"] == "durable-workflow.v2.history-export" and document["schema_version"] == 2 + assert document["schema"] == "durable-workflow.v2.history-export" + assert document["schema_version"] == 2 selected = document["workflow"] - assert selected["instance_id"] == workflow_id and selected["run_id"] == run_id + assert selected["instance_id"] == workflow_id + assert selected["run_id"] == run_id assert selected["workflow_type"] == WORKFLOW_TYPE - assert selected["details_state"] == "retained" and selected["details_pruned_at"] is None + assert selected["details_state"] == "retained" + assert selected["details_pruned_at"] is None assert document["history_complete"] is terminal assert document["selected_run"]["waits_projection_source"] == "workflow_run_waits" assert document["redaction"]["applied"] is False @@ -198,7 +208,8 @@ def verify_export(document: dict[str, Any], workflow_id: str, run_id: str, *, te def verify_recorded_fingerprints(observations: list[dict[str, Any]], histories: list[list[dict[str, Any]]]) -> None: assert len(observations) == len(histories) == 3 expected = observations[0]["fingerprints"][WORKFLOW_TYPE] - assert isinstance(expected, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", expected) + assert isinstance(expected, str) + assert re.fullmatch(r"sha256:[0-9a-f]{64}", expected) for observed, events in zip(observations, histories, strict=True): assert observed["fingerprints"] == {WORKFLOW_TYPE: expected} started = one(events, "WorkflowStarted")["payload"] @@ -212,7 +223,9 @@ def verify_worker( assert observed["definition_sha256"] == DEFINITION_SHA assert set(observed["fingerprints"]) == {WORKFLOW_TYPE} assert re.fullmatch(r"sha256:[0-9a-f]{64}", observed["fingerprints"][WORKFLOW_TYPE]) - assert observed["worker_id"] and isinstance(observed["pid"], int) and observed["pid"] > 0 + assert observed["worker_id"] + assert isinstance(observed["pid"], int) + assert observed["pid"] > 0 assert all(set(completion) == {"task_id", "commands"} for completion in observed["completions"]) task_ids = [completion["task_id"] for completion in observed["completions"]] assert all(isinstance(identity, str) and identity for identity in task_ids) @@ -249,8 +262,10 @@ def verify_worker( required = {"type", "message", "failed_step_sequence", "failed_activity_execution_id"} allowed = required | {"exception_type", "exception_class", "exception", "non_retryable"} assert required <= set(terminal) <= allowed - assert terminal["type"] == "fail_workflow" and terminal["failed_step_sequence"] == 3 - assert isinstance(terminal["message"], str) and terminal["message"] + assert terminal["type"] == "fail_workflow" + assert terminal["failed_step_sequence"] == 3 + assert isinstance(terminal["message"], str) + assert terminal["message"] assert terminal["failed_activity_execution_id"] == observed["callbacks"][-1]["execution_id"] assert instant(observed["callbacks"][-1]["clock"]) == clock @@ -268,10 +283,14 @@ async def admit_redrive( try: await client.redrive_workflow(queue, run_id) except ServerError as error: - assert error.status == 409 and error.reason() == "workflow_definition_unavailable_or_changed" - assert isinstance(error.body, dict) and error.body["outcome"] == "rejected" - assert error.body["workflow_id"] == queue and error.body["run_id"] == run_id - assert error.body["accepted"] is False and error.body["workflow_run_id"] is None + assert error.status == 409 + assert error.reason() == "workflow_definition_unavailable_or_changed" + assert isinstance(error.body, dict) + assert error.body["outcome"] == "rejected" + assert error.body["workflow_id"] == queue + assert error.body["run_id"] == run_id + assert error.body["accepted"] is False + assert error.body["workflow_run_id"] is None assert error.body["task_id"] is None refusal = {"status": error.status, "body": error.body} else: @@ -279,16 +298,20 @@ async def admit_redrive( assert [run.run_id for run in (await client.list_workflow_runs(queue)).runs] == before assert [event async for event in client.iter_history(queue, run_id)] == events request = deepcopy(observed["registration"]) - assert request["task_queue"] == queue and request["supported_workflow_types"] == [WORKFLOW_TYPE] + assert request["task_queue"] == queue + assert request["supported_workflow_types"] == [WORKFLOW_TYPE] assert request["workflow_definition_fingerprints"] == {WORKFLOW_TYPE: fingerprint} holder_id = "python-clock-admission-" + uuid4().hex request["worker_id"] = holder_id acknowledgement = await client.register_worker(**request) token = acknowledgement.get("registration_token") - assert isinstance(token, str) and re.fullmatch(r"[0-9a-f]{32}", token) + assert isinstance(token, str) + assert re.fullmatch(r"[0-9a-f]{32}", token) try: - assert acknowledgement["worker_id"] == holder_id and acknowledgement["registered"] is True - assert acknowledgement["namespace"] == "default" and acknowledgement["task_queue"] == queue + assert acknowledgement["worker_id"] == holder_id + assert acknowledgement["registered"] is True + assert acknowledgement["namespace"] == "default" + assert acknowledgement["task_queue"] == queue # The holder never creates a Worker or polls: only the later cold child # may claim and execute the successor's task. receipt = await client.redrive_workflow(queue, run_id) @@ -296,7 +319,8 @@ async def admit_redrive( removed = await client.deregister_worker_registration( holder_id, registration_token=token, expected_namespace="default", timeout=10, ) - assert removed["outcome"] == "deregistered" and removed["recovered_workflow_task_count"] == 0 + assert removed["outcome"] == "deregistered" + assert removed["recovered_workflow_task_count"] == 0 return receipt, { "absent_registration": refusal, "matching_registration": request, "registration_token_sha256": hashlib.sha256(token.encode()).hexdigest(), @@ -339,7 +363,9 @@ async def test_cold_workers_preserve_fractional_timer_clock_across_repeated_redr for repeat in (1, 2): receipt, admission = await admit_redrive(client, queue, run_ids[-1], observations[-1], histories[-1]) admissions.append(admission) - assert receipt.outcome == "redriven" and receipt.run_id and receipt.run_id not in run_ids + assert receipt.outcome == "redriven" + assert receipt.run_id + assert receipt.run_id not in run_ids run_id = receipt.run_id prefix = [event async for event in client.iter_history(queue, run_id)] timer_id = verify_reuse( diff --git a/tests/test_activity_acceptance_entry.py b/tests/test_activity_acceptance_entry.py index 15de3bc..1ff9eeb 100644 --- a/tests/test_activity_acceptance_entry.py +++ b/tests/test_activity_acceptance_entry.py @@ -48,13 +48,15 @@ def test_no_site_entry_can_execute_only_the_verified_inert_source(sandbox): environment, _data, _site, case = sandbox installed = entry.snapshot(environment) result = launch(case, installed) - assert result.returncode == 0 and result.stdout == "fixed inert result owned\n" + assert result.returncode == 0 + assert result.stdout == "fixed inert result owned\n" def test_entry_refuses_normal_site_even_when_other_inputs_match(sandbox): environment, _data, _site, case = sandbox result = launch(case, entry.snapshot(environment), no_site=False) - assert result.returncode != 0 and "no-site Python" in result.stderr + assert result.returncode != 0 + assert "no-site Python" in result.stderr assert not result.stdout @@ -63,8 +65,10 @@ def test_added_startup_hook_refuses_before_any_case_code(sandbox): installed = entry.snapshot(environment) (site / "sitecustomize.py").write_text("raise RuntimeError('startup replacement executed')\n") result = launch(case, installed) - assert result.returncode != 0 and "install inventory changed" in result.stderr - assert "startup replacement executed" not in result.stderr and not result.stdout + assert result.returncode != 0 + assert "install inventory changed" in result.stderr + assert "startup replacement executed" not in result.stderr + assert not result.stdout def test_predecessor_runs_pth_before_guard_successor_refuses_source_before_effect(sandbox, tmp_path): @@ -83,7 +87,8 @@ def test_predecessor_runs_pth_before_guard_successor_refuses_source_before_effec [str(environment / "bin/python"), "-I", "-B", "-c", "raise RuntimeError('late guard')"], env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, ) - assert old.returncode != 0 and marker.read_text() == "pre-open" + assert old.returncode != 0 + assert marker.read_text() == "pre-open" marker.unlink() case.write_text("raise RuntimeError('foreign case executed')\n") corrected = subprocess.run( @@ -91,8 +96,11 @@ def test_predecessor_runs_pth_before_guard_successor_refuses_source_before_effec input=json.dumps({"inventory": installed, "worker": {"marker": "owned"}}), env={"PATH": os.defpath}, capture_output=True, text=True, timeout=10, ) - assert corrected.returncode != 0 and "captured case source differs" in corrected.stderr - assert "foreign case executed" not in corrected.stderr and not marker.exists() and not corrected.stdout + assert corrected.returncode != 0 + assert "captured case source differs" in corrected.stderr + assert "foreign case executed" not in corrected.stderr + assert not marker.exists() + assert not corrected.stdout def test_post_capture_path_replacement_keeps_the_verified_bytes(sandbox): diff --git a/tests/test_native_timer_redrive_cell.py b/tests/test_native_timer_redrive_cell.py index 79bbfff..5c524d3 100644 --- a/tests/test_native_timer_redrive_cell.py +++ b/tests/test_native_timer_redrive_cell.py @@ -49,7 +49,7 @@ def test_exact_fraction_and_immediate_predecessor_preserve_unchanged_inputs(): events, previous, arguments = prefix() original = deepcopy((events, previous, arguments)) assert cell.verify_reuse(events, previous, **arguments) - assert (events, previous, arguments) == original + assert original == (events, previous, arguments) @pytest.mark.parametrize("reformat", [None, "clock", "start"]) @@ -320,7 +320,8 @@ async def redrive_workflow(self, queue, run_id): return SimpleNamespace(outcome="redriven", run_id="next") async def deregister_worker_registration(self, worker_id, **arguments): - assert worker_id == self.worker_id and arguments == { + assert worker_id == self.worker_id + assert arguments == { "registration_token": "b" * 32, "expected_namespace": "default", "timeout": 10, } self.calls.append(("deregister", worker_id)) @@ -335,7 +336,8 @@ async def deregister_worker_registration(self, worker_id, **arguments): assert not client.registered else: receipt, evidence = await cell.admit_redrive(client, "queue", "source", observed, events) - assert receipt.run_id == "next" and not client.registered + assert receipt.run_id == "next" + assert not client.registered assert [name for name, _value in client.calls] == ["redrive", "register", "redrive", "deregister"] assert "b" * 32 not in json.dumps(evidence) assert evidence["matching_registration"]["workflow_definition_fingerprints"] == {