From c3c88a79cea9021c23792408aeee5b58d722dbde Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:51:50 -0600 Subject: [PATCH 1/8] Make condition-wait fingerprints stable across Python versions and processes v1 hashed bytecode and repr(co_consts), so a Python upgrade failed replay at every recorded wait, and nested-function predicates (genexpr, nested lambda, comprehensions on 3.10/3.11) changed in every process. New waits record v2 (dwpy2:sha256:...) from module, qualname and a canonical source AST; without source no fingerprint is sent. Recorded v1 values match when this process computes the same v1 and are otherwise accepted as unverifiable. Reopens echo the first open's recorded value, and a keyless unverifiable v1 links only within its own chain. Evidence: golden v2 values recorded on python:3.12-slim and checked on every CI version and in fresh processes; corpus fixtures for a v2 history, a v1 history recorded by 2.14.0 on python:3.12, and a v1 genexpr history whose two opens were recorded in different processes. Fixes #258. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 + .../_condition_fingerprint.py | 167 +++++++++++++++ src/durable_workflow/workflow.py | 101 +++++---- tests/condition_fingerprint_predicates.py | 78 +++++++ tests/condition_fingerprint_workflows.py | 43 ++++ tests/fixtures/condition_fingerprints_v2.json | 18 ++ ...-fingerprint-v1-genexpr-cross-process.json | 106 ++++++++++ ...ndition-fingerprint-v1-py312-approval.json | 66 ++++++ .../condition-fingerprint-v2-approval.json | 61 ++++++ tests/test_condition_fingerprint.py | 198 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 3 + tests/test_wait_condition.py | 3 +- 12 files changed, 805 insertions(+), 43 deletions(-) create mode 100644 src/durable_workflow/_condition_fingerprint.py create mode 100644 tests/condition_fingerprint_predicates.py create mode 100644 tests/condition_fingerprint_workflows.py create mode 100644 tests/fixtures/condition_fingerprints_v2.json create mode 100644 tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json create mode 100644 tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json create mode 100644 tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json create mode 100644 tests/test_condition_fingerprint.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a13a544..9e8c573 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Opt-in typed workflow values (#184). `@workflow.defn(..., typed=True)` decodes the arguments of `run` (after `ctx`) and of every signal, query, update and update-validator handler to their annotations. `ctx.start_child_workflow(..., result_type=T)` decodes a child's result, and `WorkflowHandle.result(result_type=T)` / `Client.get_result(..., result_type=T)` decode a workflow's result on the client. Dataclass and `Enum` values in client workflow, signal, update, query and schedule arguments, child and continue-as-new arguments, and workflow and update results are sent as their neutral values. A typed `TypedValueError` names the path but never the value. Untyped workflows receive plain values as before, so in-flight histories replay unchanged. +### Fixed + +- `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. (#258) + ## [2.14.0] - 2026-11-04 **Upgrade from 2.13.2 and earlier.** Every earlier release fails replay when a condition wait has an orphaned duplicate open whose timer fires later, which Server 3.0.x can record (#253 below). diff --git a/src/durable_workflow/_condition_fingerprint.py b/src/durable_workflow/_condition_fingerprint.py new file mode 100644 index 0000000..23241b4 --- /dev/null +++ b/src/durable_workflow/_condition_fingerprint.py @@ -0,0 +1,167 @@ +"""Stable identity for ``wait_condition`` predicates (#258). + +The Server records the fingerprint a worker sends with ``open_condition_wait`` +and replay compares it, so the value must be the same on every Python version +and in every process. Version 2 hashes the predicate's module, qualified name +and a canonical form of its source AST. It never hashes bytecode, constant +reprs, memory addresses, file paths or line numbers. + +Version 1 (every release through 2.14.0) hashed bytecode and ``repr`` of code +constants, which change across Python versions and, for nested functions, +across processes. Recorded v1 values cannot be recomputed portably, so replay +treats a v1 value that differs from this process's v1 as unverifiable rather +than as a changed predicate (see :func:`fingerprint_verdict`). +""" + +from __future__ import annotations + +import ast +import hashlib +import inspect +import linecache +import re +from collections.abc import Callable +from typing import Any, Literal + +V2_PREFIX = "dwpy2:sha256:" +_LEGACY = re.compile(r"sha256:[0-9a-f]{64}\Z") +# Fields that carry no predicate meaning or differ between Python versions. +_SKIPPED_FIELDS = frozenset({"ctx", "kind", "type_comment", "type_params", "decorator_list", "returns"}) +_parsed: dict[tuple[str, int], ast.Module | None] = {} + +Verdict = Literal["match", "mismatch", "unverifiable"] + + +def legacy_fingerprint(predicate: Callable[[], bool]) -> str: + """The v1 fingerprint, kept only to recognise histories recorded by 2.14.0 and earlier.""" + h = hashlib.sha256() + h.update(b"durable-workflow-python.wait-condition.v1\0") + h.update(f"{getattr(predicate, '__module__', '')}\0".encode()) + h.update(f"{getattr(predicate, '__qualname__', '')}\0".encode()) + original = getattr(predicate, "__typed_workflow_method__", None) + code = getattr(original if original is not None else predicate, "__code__", None) + if code is None: + h.update(repr(predicate).encode()) + else: + h.update(repr(( + code.co_argcount, code.co_posonlyargcount, code.co_kwonlyargcount, code.co_code, + code.co_consts, code.co_names, code.co_varnames, code.co_freevars, + )).encode()) + return f"sha256:{h.hexdigest()}" + + +def is_legacy(value: Any) -> bool: + return isinstance(value, str) and _LEGACY.match(value) is not None + + +def fingerprint(predicate: Callable[[], bool]) -> str | None: + """Return the v2 fingerprint, or None when the predicate's source is unavailable. + + Without a fingerprint the wait is identified by its key and call order, + as in sdk-php. + """ + function = _authored_function(predicate) + if function is None: + return None + node = _predicate_node(function) + if node is None: + return None + h = hashlib.sha256() + h.update(b"durable-workflow-python.wait-condition.v2\0") + h.update(f"{getattr(function, '__module__', '')}\0".encode()) + h.update(f"{getattr(function, '__qualname__', '')}\0".encode()) + h.update(node.encode()) + return f"{V2_PREFIX}{h.hexdigest()}" + + +def fingerprint_verdict(recorded: Any, authored: str | None, predicate: Callable[[], bool]) -> Verdict: + """Compare a recorded fingerprint with the authored predicate. + + - No recorded value: nothing to compare. + - An identity supplied explicitly in another format (not v2, not None) + is compared exactly, as before. + - A v2 value is compared exactly; without source it cannot be checked. + - A v1 value equal to this process's v1 matches. Any other v1 value is + unverifiable, because v1 differs across Python versions and processes. + - Any other string is compared exactly, as before. + """ + if not isinstance(recorded, str): + return "match" + if authored is not None and not authored.startswith(V2_PREFIX): + return "match" if recorded == authored else "mismatch" + if recorded.startswith(V2_PREFIX): + if authored is None: + return "unverifiable" + return "match" if recorded == authored else "mismatch" + if is_legacy(recorded): + return "match" if recorded == legacy_fingerprint(predicate) else "unverifiable" + return "match" if recorded == authored else "mismatch" + + +def _authored_function(predicate: Any) -> Any: + target = getattr(predicate, "__typed_workflow_method__", None) or predicate + if inspect.ismethod(target): + target = target.__func__ + target = getattr(target, "__typed_workflow_method__", None) or target + return target if inspect.isfunction(target) else None + + +def _module_tree(function: Any) -> ast.Module | None: + try: + filename = inspect.getsourcefile(function) or inspect.getfile(function) + except (OSError, TypeError): + return None + lines = linecache.getlines(filename, function.__globals__) + if not lines: + return None + source = "".join(lines) + key = (filename, hash(source)) + if key not in _parsed: + try: + _parsed[key] = ast.parse(source) + except (SyntaxError, ValueError): + _parsed[key] = None + return _parsed[key] + + +def _predicate_node(function: Any) -> str | None: + tree = _module_tree(function) + if tree is None: + return None + code = function.__code__ + first_line = code.co_firstlineno + if function.__name__ == "": + lambdas = [node for node in ast.walk(tree) if isinstance(node, ast.Lambda) and node.lineno == first_line] + lambdas.sort(key=lambda node: node.col_offset) + if not lambdas: + return None + # Lambdas that start on the same line share one fingerprint covering + # all of them. Telling them apart needs column data that Python 3.10 + # lacks, and the fingerprint must be the same on every version. + return "|".join(_canonical(node) for node in lambdas) + definitions = [ + node for node in ast.walk(tree) + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) and node.name == function.__name__ + and min([node.lineno, *(decorator.lineno for decorator in node.decorator_list)]) == first_line + ] + if len(definitions) != 1: + return None + return _canonical(definitions[0]) + + +def _canonical(value: Any) -> str: + if isinstance(value, ast.AST): + parts = [type(value).__name__] + for name in value._fields: + if name in _SKIPPED_FIELDS: + continue + item = getattr(value, name, None) + if item is None or item == []: + continue + parts.append(f"{name}={_canonical(item)}") + return "(" + " ".join(parts) + ")" + if isinstance(value, list): + return "[" + ",".join(_canonical(item) for item in value) + "]" + if value is None or value is Ellipsis or isinstance(value, str | bytes | int | float | complex | bool): + return f"{type(value).__name__}:{value!r}" + return type(value).__name__ diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 9e6e5f1..372ead3 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -42,6 +42,8 @@ CommittedCancellationScopeHistory, ScopeBoundary, ) +from ._condition_fingerprint import fingerprint as _stable_condition_fingerprint +from ._condition_fingerprint import fingerprint_verdict, is_legacy from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history from ._failure_reporting import exception_message from ._typed_values import decode_payload_value, to_payload_value, to_payload_values, validate_annotation @@ -1759,31 +1761,9 @@ def to_server_command( } -def _condition_predicate_fingerprint(predicate: Callable[[], bool]) -> str: - h = hashlib.sha256() - h.update(b"durable-workflow-python.wait-condition.v1\0") - h.update(f"{getattr(predicate, '__module__', '')}\0".encode()) - h.update(f"{getattr(predicate, '__qualname__', '')}\0".encode()) - - # A typed=True wrapper is fingerprinted as the method it wraps, so turning - # typed on does not change a recorded predicate's identity (#184). - original = getattr(predicate, "__typed_workflow_method__", None) - code = getattr(original if original is not None else predicate, "__code__", None) - if code is None: - h.update(repr(predicate).encode()) - else: - h.update(repr(( - code.co_argcount, - code.co_posonlyargcount, - code.co_kwonlyargcount, - code.co_code, - code.co_consts, - code.co_names, - code.co_varnames, - code.co_freevars, - )).encode()) - - return f"sha256:{h.hexdigest()}" +def _condition_predicate_fingerprint(predicate: Callable[[], bool]) -> str | None: + """The stable v2 predicate fingerprint, or None without source (#258).""" + return _stable_condition_fingerprint(predicate) Command = ( @@ -6070,10 +6050,7 @@ def _condition_wait_mismatch( ) opened_fingerprint = opened.get("condition_definition_fingerprint") - if ( - isinstance(opened_fingerprint, str) - and cmd.condition_definition_fingerprint != opened_fingerprint - ): + if _fingerprint_verdict(opened, cmd) == "mismatch": return NonDeterministicReplayError( step.workflow_sequence, "condition wait", step.event_types, detail=( @@ -6086,7 +6063,19 @@ def _condition_wait_mismatch( return None - def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) -> bool: + def _fingerprint_verdict(opened: Mapping[str, Any], cmd: WaitCondition) -> str: + return fingerprint_verdict( + opened.get("condition_definition_fingerprint"), cmd.condition_definition_fingerprint, cmd.predicate, + ) + + def _same_logical_condition_wait( + opened: Mapping[str, Any], + cmd: WaitCondition, + *, + chain_first: Mapping[str, Any] | None = None, + chain_current: Mapping[str, Any] | None = None, + next_index: int | None = None, + ) -> bool: if _condition_wait_mismatch(opened, cmd) is not None: return False @@ -6095,10 +6084,24 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) return True opened_fingerprint = opened.get("condition_definition_fingerprint") + if not isinstance(opened_fingerprint, str) or not opened_fingerprint: + return False + verdict = _fingerprint_verdict(opened, cmd) + if verdict == "match": + return True + if verdict != "unverifiable": + return False + # A keyless wait whose fingerprint cannot be checked (a v1 value from + # another Python version or process, #258) links only within its own + # chain: the same recorded v1 value as the chain's first open, or a + # duplicate opened while the current wait was still open (#253). It + # never binds a recorded wait to another call. + first_fingerprint = chain_first.get("condition_definition_fingerprint") if chain_first is not None else None + if is_legacy(opened_fingerprint) and opened_fingerprint == first_fingerprint: + return True return ( - isinstance(opened_fingerprint, str) - and bool(opened_fingerprint) - and opened_fingerprint == cmd.condition_definition_fingerprint + chain_current is not None and next_index is not None + and _opened_before_terminal(next_index, chain_current) ) result_cursor = 0 @@ -6981,9 +6984,14 @@ def _assert_cancellation_call_matches(command: Any, boundary: CancellationDelive expected = ({"delay_seconds": leaf.delay_seconds} if isinstance(leaf, StartTimer) else { "condition_key": leaf.condition_key, "timeout_seconds": leaf.timeout_seconds, - "condition_definition_fingerprint": leaf.condition_definition_fingerprint, }) - if any(field in payload and payload[field] != actual for field, actual in expected.items()): + fingerprint_changed = isinstance(leaf, WaitCondition) and fingerprint_verdict( + payload.get("condition_definition_fingerprint"), leaf.condition_definition_fingerprint, + leaf.predicate, + ) == "mismatch" + if fingerprint_changed or any( + field in payload and payload[field] != actual for field, actual in expected.items() + ): raise NonDeterministicReplayError( sequence, "original group member", [event_type], detail="scoped group changed its original timer or condition definition", @@ -7107,7 +7115,7 @@ def _opened_before_terminal(next_index: int, opened: Mapping[str, Any] | None) - return terminal is not None and wait_opened_event_indexes[next_index] < terminal def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -> WaitCondition: - """Reopen a logical wait with the time left before its first deadline. + """Reopen a logical wait with its first fingerprint and remaining time. The deadline is fixed when the wait first opens, as in embedded Workflow: the first ConditionWaitOpened time plus the authored @@ -7116,14 +7124,20 @@ def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) - history time, rounded up and clamped to 0..authored (the sdk-php#269 rule). Without recorded times the authored timeout is kept. """ - authored = command.timeout_seconds - if authored is None or first_index >= len(wait_opened): + if first_index >= len(wait_opened): return command + reopened = copy(command) + # A reopen sends the fingerprint its first open recorded, so a wait + # first opened under v1 (2.14.0 and earlier) keeps v1 across the + # upgrade and links v1 to v1, never v1 to v2 (#258). + recorded = wait_opened[first_index].get("condition_definition_fingerprint") + if isinstance(recorded, str) and recorded: + reopened.condition_definition_fingerprint = recorded + authored = command.timeout_seconds opened_at = wait_opened_times[first_index] - if opened_at is None or history_time is None: - return command + if authored is None or opened_at is None or history_time is None: + return reopened remaining = math.ceil(authored - (history_time - opened_at).total_seconds()) - reopened = copy(command) reopened.timeout_seconds = min(authored, max(0, remaining)) return reopened @@ -7783,7 +7797,10 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None has_reopened_same_wait = ( opened is not None and next_wait_index < len(wait_opened) - and _same_logical_condition_wait(wait_opened[next_wait_index], cmd) + and _same_logical_condition_wait( + wait_opened[next_wait_index], cmd, chain_first=wait_opened[first_open_index], + chain_current=opened, next_index=next_wait_index, + ) ) if resolution == "timed_out": if has_reopened_same_wait and _opened_before_terminal(next_wait_index, opened): diff --git a/tests/condition_fingerprint_predicates.py b/tests/condition_fingerprint_predicates.py new file mode 100644 index 0000000..53cee3d --- /dev/null +++ b/tests/condition_fingerprint_predicates.py @@ -0,0 +1,78 @@ +"""Predicates whose v2 fingerprints are pinned in condition_fingerprints_v2.json (#258). + +The golden values were computed on the official python:3.12-slim image. Every +CI Python version must compute the same values, in this process and in a fresh +one, so a wait recorded on one version or process replays on any other. +Keep this module's source unchanged: editing it changes the fingerprints. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from durable_workflow import workflow + + +def _identity(predicate: Callable[[], bool]) -> Callable[[], bool]: + return predicate + + +@workflow.defn(name="tests.fingerprint.typed-ready", typed=True) +class TypedReady: + def __init__(self) -> None: + self.approvals: list[str] = [] + + @workflow.query("ready") + def ready(self, minimum: int = 2) -> bool: + return len(self.approvals) >= minimum + + def run(self, ctx: workflow.WorkflowContext) -> None: + return None + + +class Approvals: + def __init__(self) -> None: + self.items: list[int] = [1, 20] + self.approved = False + self.name = "x" + + def is_done(self) -> bool: + return self.approved + + def has_large(self) -> bool: + return any(n > 10 for n in self.items) + + def _multiline(self) -> Callable[[], bool]: + return _identity( + lambda: ( + self.approved + and len(self.items) > 0 + ) + ) + + def _same_line(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + return (lambda: self.approved), (lambda: not self.approved) + + def _inner(self) -> Callable[[], bool]: + def ready() -> bool: + return self.approved and bool(self.items) + + return ready + + def predicates(self) -> dict[str, Any]: + first, second = self._same_line() + return { + "lambda": lambda: self.approved, + "genexpr": lambda: any(n > 10 for n in self.items), + "listcomp": lambda: len([n for n in self.items if n]) > 2, + "nested_lambda": lambda: (lambda value: value > 3)(len(self.items)), + "fstring": lambda: f"{self.name}-{len(self.items)}" == "x-1", + "multiline_in_call": self._multiline(), + "same_line_first": first, + "same_line_second": second, + "bound_method": self.is_done, + "bound_genexpr_method": self.has_large, + "inner_def": self._inner(), + "typed_wrapper": TypedReady().ready, + } diff --git a/tests/condition_fingerprint_workflows.py b/tests/condition_fingerprint_workflows.py new file mode 100644 index 0000000..a00aaca --- /dev/null +++ b/tests/condition_fingerprint_workflows.py @@ -0,0 +1,43 @@ +"""Workflows for the condition-fingerprint corpus fixtures (#258). + +The v1 fixtures were recorded by durable-workflow 2.14.0 from PyPI on the +official python:3.12-slim image, replaying these classes; the v2 fixture uses +this branch's stable fingerprint. Keep this module's source unchanged: the +recorded fingerprints depend on it. +""" + +from __future__ import annotations + +from durable_workflow import workflow + + +@workflow.defn(name="tests.replay.fingerprint-approval") +class FingerprintApproval: + """A keyless wait whose fingerprint alone identifies it.""" + + def __init__(self) -> None: + self.approved = False + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + def run(self, ctx: workflow.WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approved) + return "approved" + + +@workflow.defn(name="tests.replay.fingerprint-genexpr") +class FingerprintGenexpr: + """A keyed wait whose v1 fingerprint differs in every process (nested genexpr).""" + + def __init__(self) -> None: + self.items: list[int] = [] + + @workflow.signal("add") + def add(self, value: int) -> None: + self.items.append(value) + + def run(self, ctx: workflow.WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: any(n > 10 for n in self.items), key="large") + return list(self.items) diff --git a/tests/fixtures/condition_fingerprints_v2.json b/tests/fixtures/condition_fingerprints_v2.json new file mode 100644 index 0000000..c7794a1 --- /dev/null +++ b/tests/fixtures/condition_fingerprints_v2.json @@ -0,0 +1,18 @@ +{ + "recorded_on": "python:3.12-slim (CPython 3.12.16)", + "module": "tests.condition_fingerprint_predicates", + "fingerprints": { + "lambda": "dwpy2:sha256:838e5598f19815ba337ba3b3483ba1036f162e1a39c115c1cb1e3259cc4bd0ca", + "genexpr": "dwpy2:sha256:7d23e5fca8a875c14dd2aa01263f8be68b681686cfaf890ff4b494893d2d09c8", + "listcomp": "dwpy2:sha256:905370895eeccf09cc08ca8d5611bfad01b227eaf8bf90402098d093681971a3", + "nested_lambda": "dwpy2:sha256:54054b9ed7882bf65e8879e11c9f0565be8b1d47b90a9156f324969fb89722a3", + "fstring": "dwpy2:sha256:e271b27eb6e6c71e8c2ae40088025003907b231c006da424731d92b40c780635", + "multiline_in_call": "dwpy2:sha256:53a9527b0234805ebffe2dcdd79715fd7743c23d001500b99f4616458e630cb5", + "same_line_first": "dwpy2:sha256:b4897053d25adf21b11dbec6ca5d06031e137a9d20c68f340a310c42160ab63b", + "same_line_second": "dwpy2:sha256:b4897053d25adf21b11dbec6ca5d06031e137a9d20c68f340a310c42160ab63b", + "bound_method": "dwpy2:sha256:4b9aabe93ef8c48c2f05481ccb92f743da7c19b76104816c4a3f12ab32d741b9", + "bound_genexpr_method": "dwpy2:sha256:9bbb6600801559c28420aa32085bfdc76ceef33e03c2d9194d2edb507053c934", + "inner_def": "dwpy2:sha256:7be3326bb020ab50b9d1bad496b395551db90324bf6b0b0af9fe888e9dcfdd84", + "typed_wrapper": "dwpy2:sha256:f5f8904a70119a2f2999a4a256bc42549f48aca0c74c47ab42d256db852b919a" + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json b/tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json new file mode 100644 index 0000000..44a24d3 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v1-genexpr-cross-process.json @@ -0,0 +1,106 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v1-genexpr-cross-process", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "source": { + "runtime": "sdk", + "package": "durable-workflow", + "version": "2.14.0" + }, + "workflow": { + "type": "tests.replay.fingerprint-genexpr", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-genexpr" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "large-1", + "sequence": 1, + "condition_definition_fingerprint": "sha256:dd6cc254ae46267f90fde3ff42b381ae7f6022af4de6fa42a07f1abbe2434b7f", + "condition_key": "large" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "add", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCBAoA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "large-1", + "sequence": 1, + "condition_key": "large" + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "large-2", + "sequence": 2, + "condition_definition_fingerprint": "sha256:b415051d8936710c8a887c6fc30e3bcb8bd20f2b9c79f398dfffa84215050aa3", + "condition_key": "large" + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "signal_name": "add", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCBCgA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 6 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "condition_wait_id": "large-2", + "sequence": 2, + "condition_key": "large" + }, + "sequence": 7 + } + ], + "expected": { + "type": "complete_workflow", + "result": [ + 5, + 20 + ] + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json b/tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json new file mode 100644 index 0000000..1aa3f11 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v1-py312-approval.json @@ -0,0 +1,66 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v1-py312-approval", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "source": { + "runtime": "sdk", + "package": "durable-workflow", + "version": "2.14.0" + }, + "workflow": { + "type": "tests.replay.fingerprint-approval", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-approval" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1, + "condition_definition_fingerprint": "sha256:0f5fd86d297870b40434e114ff4b7b42ce2c0d2df725c1069e8931539091c9d8" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1 + }, + "sequence": 4 + } + ], + "expected": { + "type": "complete_workflow", + "result": "approved" + } +} diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json b/tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json new file mode 100644 index 0000000..cc85aba --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v2-approval.json @@ -0,0 +1,61 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v2-approval", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.fingerprint-approval", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-approval" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1, + "condition_definition_fingerprint": "dwpy2:sha256:21b0cde657ecccaf9e2dc2a362d2e32e0e4d5d822c112576e5e23f4485a03199" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "approval-1", + "sequence": 1 + }, + "sequence": 4 + } + ], + "expected": { + "type": "complete_workflow", + "result": "approved" + } +} diff --git a/tests/test_condition_fingerprint.py b/tests/test_condition_fingerprint.py new file mode 100644 index 0000000..cfc772c --- /dev/null +++ b/tests/test_condition_fingerprint.py @@ -0,0 +1,198 @@ +"""Stable v2 condition fingerprints and v1 compatibility (#258).""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow._condition_fingerprint import ( + V2_PREFIX, + fingerprint, + fingerprint_verdict, + is_legacy, + legacy_fingerprint, +) +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.workflow import CompleteWorkflow, WaitCondition, WorkflowContext, replay +from tests.condition_fingerprint_predicates import Approvals + +ROOT = Path(__file__).resolve().parent.parent +GOLDEN = json.loads((ROOT / "tests/fixtures/condition_fingerprints_v2.json").read_text(encoding="utf-8")) +FOREIGN_V1 = "sha256:" + "1" * 64 # a v1 value recorded by another Python version or process +OTHER_V1 = "sha256:" + "2" * 64 + + +def test_v2_fingerprints_match_the_values_recorded_on_python_3_12() -> None: + # Every CI Python version (3.10 to 3.15) must reproduce the 3.12 values. + computed = {name: fingerprint(predicate) for name, predicate in Approvals().predicates().items()} + assert computed == GOLDEN["fingerprints"] + assert all(value.startswith(V2_PREFIX) for value in computed.values()) + + +def test_v2_fingerprints_are_the_same_in_fresh_processes() -> None: + # v1 changed in every process for nested functions (genexpr, nested lambda). + script = ( + "import json, sys; sys.path.insert(0, sys.argv[1]);" + "from tests.condition_fingerprint_predicates import Approvals;" + "from durable_workflow._condition_fingerprint import fingerprint;" + "print(json.dumps({k: fingerprint(v) for k, v in Approvals().predicates().items()}))" + ) + for _ in range(2): + output = subprocess.run([sys.executable, "-c", script, str(ROOT)], check=True, capture_output=True, + text=True, cwd=ROOT).stdout + assert json.loads(output) == GOLDEN["fingerprints"] + + +def test_v2_ignores_formatting_but_not_logic() -> None: + reformatted = lambda: ( # noqa: E731 + Approvals().approved # a comment + ) + flipped = lambda: not Approvals().approved # noqa: E731 + original = lambda: Approvals().approved # noqa: E731 + # Same module and qualname: only the predicate's logic decides. + assert fingerprint(reformatted) == fingerprint(original) + assert fingerprint(original) != fingerprint(flipped) + + +def test_a_predicate_without_source_has_no_fingerprint() -> None: + predicate = eval("lambda: False") # noqa: S307 - a predicate with no source file + assert fingerprint(predicate) is None + assert WorkflowContext(run_id="x").wait_condition(predicate, key="k").condition_definition_fingerprint is None + + +@pytest.mark.parametrize(("recorded", "authored", "expected"), [ + (None, "dwpy2:sha256:" + "a" * 64, "match"), + ("dwpy2:sha256:" + "a" * 64, "dwpy2:sha256:" + "a" * 64, "match"), + ("dwpy2:sha256:" + "a" * 64, "dwpy2:sha256:" + "b" * 64, "mismatch"), + ("dwpy2:sha256:" + "a" * 64, None, "unverifiable"), + (FOREIGN_V1, "dwpy2:sha256:" + "a" * 64, "unverifiable"), + (FOREIGN_V1, None, "unverifiable"), + ("sha256:different", "dwpy2:sha256:" + "a" * 64, "mismatch"), + (FOREIGN_V1, "changed", "mismatch"), # an explicitly supplied non-v2 identity is compared exactly + (FOREIGN_V1, FOREIGN_V1, "match"), +]) +def test_fingerprint_verdicts(recorded: Any, authored: str | None, expected: str) -> None: + assert fingerprint_verdict(recorded, authored, lambda: False) == expected + + +def test_a_v1_value_from_this_process_still_matches() -> None: + def predicate() -> bool: + return False + + assert is_legacy(legacy_fingerprint(predicate)) + assert fingerprint_verdict(legacy_fingerprint(predicate), fingerprint(predicate), predicate) == "match" + + +# Mixed chains across the upgrade (General's correction 1). +@workflow.defn(name="tests.fingerprint.keyless-approval") +class KeylessApproval: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + @workflow.signal("note") + def note(self) -> None: + return None + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approved) + return "approved" + + +AT = "2026-11-04T00:00:0{}+00:00" + + +def _opened(wait_id: str, sequence: int, value: str | None, second: int) -> dict[str, Any]: + payload: dict[str, Any] = {"condition_wait_id": wait_id, "sequence": sequence} + if value is not None: + payload["condition_definition_fingerprint"] = value + return {"event_type": "ConditionWaitOpened", "recorded_at": AT.format(second), "payload": payload} + + +def _signal(name: str, sequence: int, second: int) -> dict[str, Any]: + return {"event_type": "SignalReceived", "recorded_at": AT.format(second), "payload": { + "signal_name": name, "value": serializer.envelope([]), "payload_codec": "avro", + "workflow_sequence": sequence}} + + +def _satisfied(wait_id: str, sequence: int, second: int) -> dict[str, Any]: + return {"event_type": "ConditionWaitSatisfied", "recorded_at": AT.format(second), + "payload": {"condition_wait_id": wait_id, "sequence": sequence}} + + +START = {"event_type": "WorkflowStarted", "recorded_at": AT.format(0), "payload": {}} + + +def test_a_reopen_of_a_v1_first_open_sends_the_recorded_v1_value() -> None: + # Opened under 2.14.0 (v1), then a false wake under 2.14.1. The Server + # records ConditionWaitSatisfied with this task's reopen, so the task sees + # only the open and the signal. + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2)] + (command,) = replay(KeylessApproval, history, []).commands + assert isinstance(command, WaitCondition) + assert command.condition_definition_fingerprint == FOREIGN_V1 + + +def test_a_v1_chain_links_v1_to_v1_after_the_upgrade() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_a_reopen_without_source_keeps_echoing_the_recorded_value() -> None: + @workflow.defn(name="tests.fingerprint.no-source") + class NoSource: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("note") + def note(self) -> None: + return None + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(eval("lambda self=self: self.approved")) # noqa: S307 + return "approved" + + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2)] + (command,) = replay(NoSource, history, []).commands + assert isinstance(command, WaitCondition) and command.condition_definition_fingerprint == FOREIGN_V1 + + +# Keyless unverifiable values never bind a wait to another call (correction 2). +@workflow.defn(name="tests.fingerprint.two-keyless-waits") +class TwoKeylessWaits: + def __init__(self) -> None: + self.approvals = 0 + + @workflow.signal("approve") + def approve(self) -> None: + self.approvals += 1 + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(lambda: self.approvals >= 1) + yield ctx.wait_condition(lambda: self.approvals >= 2) + return self.approvals + + +def test_a_later_keyless_v1_open_binds_to_the_next_call_not_as_a_reopen() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("approve", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, OTHER_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(TwoKeylessWaits, history, []).commands) == [CompleteWorkflow(2)] + + +def test_a_changed_v2_predicate_is_refused() -> None: + (command,) = replay(KeylessApproval, [START], []).commands + assert isinstance(command, WaitCondition) and command.condition_definition_fingerprint.startswith(V2_PREFIX) + changed = "dwpy2:sha256:" + "0" * 64 + history = [START, _opened("w1", 1, changed, 1), _signal("approve", 1, 2), _satisfied("w1", 1, 2)] + with pytest.raises(NonDeterministicReplayError, match="condition_fingerprint_changed"): + replay(KeylessApproval, history, []) diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index b799968..603e662 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -25,6 +25,7 @@ commands_to_server_commands, query_state, ) +from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow from tests.test_condition_replay_drift import ConditionIdentityWorkflow from tests.test_generator_lifecycle import CleanupSideEffect, PendingFinally @@ -424,6 +425,8 @@ def run(self, ctx: WorkflowContext, message: str) -> None: OrphanedDuplicateDecision, TypedActivityResultsWorkflow, TypedOnboarding, + FingerprintApproval, + FingerprintGenexpr, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 8aa75e5..00391aa 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -250,7 +250,8 @@ def test_wait_condition_returns_dataclass_with_predicate_and_key(self) -> None: assert cmd.condition_key == "k" assert cmd.timeout_seconds == 5 assert cmd.condition_definition_fingerprint is not None - assert cmd.condition_definition_fingerprint.startswith("sha256:") + # v2: stable across Python versions and processes (#258). + assert cmd.condition_definition_fingerprint.startswith("dwpy2:sha256:") assert callable(cmd.predicate) assert cmd.predicate() is False From 51d4eeb00d1f58c752d52c52134e91944df459f5 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:01:57 -0600 Subject: [PATCH 2/8] Tell same-line lambdas apart by compiling each candidate A conditional choice of predicates on one line shared a fingerprint, so the condition-recovery integration test's drift went undetected. Each candidate lambda on the line is now compiled on the running interpreter inside a function binding the live free variables and compared with the live code; only the chosen node's AST is hashed, so values stay portable. Golden values regenerated on python:3.12-slim (identical on 3.13). Refs #258. Co-Authored-By: Claude Opus 5.5 --- .../_condition_fingerprint.py | 41 +++++++++++++++++-- tests/fixtures/condition_fingerprints_v2.json | 6 +-- tests/test_condition_fingerprint.py | 11 +++++ 3 files changed, 52 insertions(+), 6 deletions(-) diff --git a/src/durable_workflow/_condition_fingerprint.py b/src/durable_workflow/_condition_fingerprint.py index 23241b4..63e909c 100644 --- a/src/durable_workflow/_condition_fingerprint.py +++ b/src/durable_workflow/_condition_fingerprint.py @@ -133,11 +133,14 @@ def _predicate_node(function: Any) -> str | None: if function.__name__ == "": lambdas = [node for node in ast.walk(tree) if isinstance(node, ast.Lambda) and node.lineno == first_line] lambdas.sort(key=lambda node: node.col_offset) + if len(lambdas) > 1: + matching = [node for node in lambdas if _compiles_to(node, code)] + if len(matching) == 1: + lambdas = matching if not lambdas: return None - # Lambdas that start on the same line share one fingerprint covering - # all of them. Telling them apart needs column data that Python 3.10 - # lacks, and the fingerprint must be the same on every version. + # Lambdas that start on the same line and compile identically share + # one fingerprint covering all of them. return "|".join(_canonical(node) for node in lambdas) definitions = [ node for node in ast.walk(tree) @@ -149,6 +152,38 @@ def _predicate_node(function: Any) -> str | None: return _canonical(definitions[0]) +def _compiles_to(node: ast.Lambda, code: Any) -> bool: + """Whether *node* compiles to *code* on this interpreter. + + The candidate is compiled inside a function that binds the live code's + free variables, so names resolve the same way. Only the choice of node + uses bytecode, which is valid within one interpreter; the fingerprint + hashes the node's AST, so it stays portable. + """ + free = list(code.co_freevars) + lines = ["def __dw_outer__():"] + if free: + lines.append(f" {' = '.join(free)} = None") + lines.append(f" return ({ast.unparse(node)})") + try: + outer = compile("\n".join(lines), "", "exec") + except SyntaxError: + return False + function = next((const for const in outer.co_consts if hasattr(const, "co_code")), None) + candidate = next( + (const for const in getattr(function, "co_consts", ()) if getattr(const, "co_name", None) == ""), + None, + ) + return candidate is not None and ( + candidate.co_code == code.co_code + and candidate.co_names == code.co_names + and candidate.co_varnames == code.co_varnames + and candidate.co_freevars == code.co_freevars + and [c for c in candidate.co_consts if not hasattr(c, "co_code")] + == [c for c in code.co_consts if not hasattr(c, "co_code")] + ) + + def _canonical(value: Any) -> str: if isinstance(value, ast.AST): parts = [type(value).__name__] diff --git a/tests/fixtures/condition_fingerprints_v2.json b/tests/fixtures/condition_fingerprints_v2.json index c7794a1..3302ed4 100644 --- a/tests/fixtures/condition_fingerprints_v2.json +++ b/tests/fixtures/condition_fingerprints_v2.json @@ -5,11 +5,11 @@ "lambda": "dwpy2:sha256:838e5598f19815ba337ba3b3483ba1036f162e1a39c115c1cb1e3259cc4bd0ca", "genexpr": "dwpy2:sha256:7d23e5fca8a875c14dd2aa01263f8be68b681686cfaf890ff4b494893d2d09c8", "listcomp": "dwpy2:sha256:905370895eeccf09cc08ca8d5611bfad01b227eaf8bf90402098d093681971a3", - "nested_lambda": "dwpy2:sha256:54054b9ed7882bf65e8879e11c9f0565be8b1d47b90a9156f324969fb89722a3", + "nested_lambda": "dwpy2:sha256:a75240a391df56bc1a043357649e88182153055bf5b7ac2bb413abbff478ad9c", "fstring": "dwpy2:sha256:e271b27eb6e6c71e8c2ae40088025003907b231c006da424731d92b40c780635", "multiline_in_call": "dwpy2:sha256:53a9527b0234805ebffe2dcdd79715fd7743c23d001500b99f4616458e630cb5", - "same_line_first": "dwpy2:sha256:b4897053d25adf21b11dbec6ca5d06031e137a9d20c68f340a310c42160ab63b", - "same_line_second": "dwpy2:sha256:b4897053d25adf21b11dbec6ca5d06031e137a9d20c68f340a310c42160ab63b", + "same_line_first": "dwpy2:sha256:bd14b20c03220f1cddd0c6e07ea364aef66051dbdf7d14a73b2e03724d06c471", + "same_line_second": "dwpy2:sha256:351a2607e1e06e4d0bfa6020f47d577661c259fe1f7f5d43ccbe1e00e59140ad", "bound_method": "dwpy2:sha256:4b9aabe93ef8c48c2f05481ccb92f743da7c19b76104816c4a3f12ab32d741b9", "bound_genexpr_method": "dwpy2:sha256:9bbb6600801559c28420aa32085bfdc76ceef33e03c2d9194d2edb507053c934", "inner_def": "dwpy2:sha256:7be3326bb020ab50b9d1bad496b395551db90324bf6b0b0af9fe888e9dcfdd84", diff --git a/tests/test_condition_fingerprint.py b/tests/test_condition_fingerprint.py index cfc772c..d3f4e5c 100644 --- a/tests/test_condition_fingerprint.py +++ b/tests/test_condition_fingerprint.py @@ -196,3 +196,14 @@ def test_a_changed_v2_predicate_is_refused() -> None: history = [START, _opened("w1", 1, changed, 1), _signal("approve", 1, 2), _satisfied("w1", 1, 2)] with pytest.raises(NonDeterministicReplayError, match="condition_fingerprint_changed"): replay(KeylessApproval, history, []) + + +def test_two_lambdas_on_one_line_get_their_own_fingerprints() -> None: + # A conditional choice between predicates on one line (as in the + # condition-recovery integration test) must still detect drift. Each + # candidate is matched by compiling it on this interpreter; only its AST + # is hashed. + predicates = Approvals().predicates() + assert GOLDEN["fingerprints"]["same_line_first"] != GOLDEN["fingerprints"]["same_line_second"] + assert fingerprint(predicates["same_line_first"]) == GOLDEN["fingerprints"]["same_line_first"] + assert fingerprint(predicates["same_line_second"]) == GOLDEN["fingerprints"]["same_line_second"] From 8a5af449c279f56ddc3a5b79b6b19d92369ca71d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:14:27 -0600 Subject: [PATCH 3/8] Ignore Python 3.10 qualname constants when matching same-line lambdas On 3.10 a lambda that makes another function stores the inner qualname as a string constant, which differs between the live code and the synthetic compile, so the nested_lambda predicate fell back to the combined fingerprint and missed the 3.12 golden value. Qualname-like constants are now excluded from the comparison. Verified on python:3.10-slim and python:3.11-slim containers (match the 3.12 golden) and locally on 3.13. Refs #258. Co-Authored-By: Claude Opus 5.5 --- src/durable_workflow/_condition_fingerprint.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/durable_workflow/_condition_fingerprint.py b/src/durable_workflow/_condition_fingerprint.py index 63e909c..b9962ab 100644 --- a/src/durable_workflow/_condition_fingerprint.py +++ b/src/durable_workflow/_condition_fingerprint.py @@ -179,11 +179,18 @@ def _compiles_to(node: ast.Lambda, code: Any) -> bool: and candidate.co_names == code.co_names and candidate.co_varnames == code.co_varnames and candidate.co_freevars == code.co_freevars - and [c for c in candidate.co_consts if not hasattr(c, "co_code")] - == [c for c in code.co_consts if not hasattr(c, "co_code")] + and _plain_constants(candidate) == _plain_constants(code) ) +def _plain_constants(code: Any) -> list[Any]: + """Constants other than nested code and the qualnames Python 3.10 stores for them.""" + return [ + const for const in code.co_consts + if not hasattr(const, "co_code") and not (isinstance(const, str) and "" in const) + ] + + def _canonical(value: Any) -> str: if isinstance(value, ast.AST): parts = [type(value).__name__] From bd2755c743f5750d2147c2e17334d3d004be7b49 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:26:24 -0600 Subject: [PATCH 4/8] Address the #260 review: comprehensions, no-source chains, terminal cleanup - Drop PEP 709 comprehension segments (, , ) from the hashed qualname, so a lambda inside a list, dict or set comprehension has one v2 value on 3.11 and 3.12+. - Match same-line lambdas by names, argcount and constants first; use bytecode only to break a tie. Golden cases for lambdas in each comprehension and two same-line lambdas in a list comprehension; regenerated on python:3.12-slim and identical on 3.10, 3.11 and 3.13. - A keyless wait without source links its false-wake reopen within its chain (2.14.0 did, via its process-local v1). - Terminal reopen cleanup passes the chain's first open, so a trailing v1 duplicate in a keyless chain links across versions. - Cache the predicate node per code object; CHANGELOG notes rollback and mixed-fleet limits. Refs #258. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- .../_condition_fingerprint.py | 23 +++++++-- src/durable_workflow/workflow.py | 19 +++++++- tests/condition_fingerprint_predicates.py | 20 ++++++++ tests/fixtures/condition_fingerprints_v2.json | 7 ++- tests/test_condition_fingerprint.py | 48 +++++++++++++++++++ 6 files changed, 110 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e8c573..5bbed22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. (#258) +- `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. A keyless wait without source links its false-wake reopens within its chain, as before. **Rollback and mixed fleets:** a run that opened a wait under this release records a v2 fingerprint, which 2.14.0 and earlier refuse on replay. Don't roll a task queue back below this release once such waits exist, and upgrade every worker on a queue together. (#258) ## [2.14.0] - 2026-11-04 diff --git a/src/durable_workflow/_condition_fingerprint.py b/src/durable_workflow/_condition_fingerprint.py index b9962ab..88eb1d2 100644 --- a/src/durable_workflow/_condition_fingerprint.py +++ b/src/durable_workflow/_condition_fingerprint.py @@ -28,6 +28,10 @@ # Fields that carry no predicate meaning or differ between Python versions. _SKIPPED_FIELDS = frozenset({"ctx", "kind", "type_comment", "type_params", "decorator_list", "returns"}) _parsed: dict[tuple[str, int], ast.Module | None] = {} +_nodes: dict[Any, str | None] = {} +# Python 3.12 inlines list, dict and set comprehensions (PEP 709), which drops +# their segment from a nested function's __qualname__. +_INLINED_COMPREHENSION = re.compile(r"\.<(?:listcomp|dictcomp|setcomp)>") Verdict = Literal["match", "mismatch", "unverifiable"] @@ -63,13 +67,17 @@ def fingerprint(predicate: Callable[[], bool]) -> str | None: function = _authored_function(predicate) if function is None: return None - node = _predicate_node(function) + code = function.__code__ + if code not in _nodes: + _nodes[code] = _predicate_node(function) + node = _nodes[code] if node is None: return None + qualname = _INLINED_COMPREHENSION.sub("", getattr(function, "__qualname__", "")) h = hashlib.sha256() h.update(b"durable-workflow-python.wait-condition.v2\0") h.update(f"{getattr(function, '__module__', '')}\0".encode()) - h.update(f"{getattr(function, '__qualname__', '')}\0".encode()) + h.update(f"{qualname}\0".encode()) h.update(node.encode()) return f"{V2_PREFIX}{h.hexdigest()}" @@ -134,7 +142,11 @@ def _predicate_node(function: Any) -> str | None: lambdas = [node for node in ast.walk(tree) if isinstance(node, ast.Lambda) and node.lineno == first_line] lambdas.sort(key=lambda node: node.col_offset) if len(lambdas) > 1: - matching = [node for node in lambdas if _compiles_to(node, code)] + # Names and constants are stable across versions; bytecode only + # breaks a tie, and is compared within this interpreter. + matching = [node for node in lambdas if _compiles_to(node, code, compare_bytecode=False)] + if len(matching) > 1: + matching = [node for node in matching if _compiles_to(node, code, compare_bytecode=True)] if len(matching) == 1: lambdas = matching if not lambdas: @@ -152,7 +164,7 @@ def _predicate_node(function: Any) -> str | None: return _canonical(definitions[0]) -def _compiles_to(node: ast.Lambda, code: Any) -> bool: +def _compiles_to(node: ast.Lambda, code: Any, *, compare_bytecode: bool) -> bool: """Whether *node* compiles to *code* on this interpreter. The candidate is compiled inside a function that binds the live code's @@ -175,7 +187,8 @@ def _compiles_to(node: ast.Lambda, code: Any) -> bool: None, ) return candidate is not None and ( - candidate.co_code == code.co_code + (not compare_bytecode or candidate.co_code == code.co_code) + and candidate.co_argcount == code.co_argcount and candidate.co_names == code.co_names and candidate.co_varnames == code.co_varnames and candidate.co_freevars == code.co_freevars diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 372ead3..6c3238a 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -6085,7 +6085,16 @@ def _same_logical_condition_wait( opened_fingerprint = opened.get("condition_definition_fingerprint") if not isinstance(opened_fingerprint, str) or not opened_fingerprint: - return False + # A keyless wait without source has no identity beyond call order. + # Within its own chain (both opens recorded without a fingerprint) + # it is the same wait, as 2.14.0 linked it by its process-local v1. + first_fingerprint = ( + chain_first.get("condition_definition_fingerprint") if chain_first is not None else "absent" + ) + return ( + cmd.condition_definition_fingerprint is None and not cmd.condition_key + and chain_first is not None and not first_fingerprint + ) verdict = _fingerprint_verdict(opened, cmd) if verdict == "match": return True @@ -6167,6 +6176,8 @@ def _refresh_replay_logging() -> None: pending: list[Command] = [] advanced_cmd: Any = None terminal_condition_reopen_cmd: WaitCondition | None = None + # The first open of the chain a terminal reopen cleanup belongs to (#258). + terminal_condition_chain_first: Mapping[str, Any] | None = None version_decisions: dict[str, int] = {} def _version_error(detail: str, step: _RecordedStep | None = None) -> NonDeterministicReplayError: @@ -6742,7 +6753,9 @@ def _consume_terminal_condition_reopens() -> None: return while wait_yield_count < len(wait_opened): opened = wait_opened[wait_yield_count] - if not _same_logical_condition_wait(opened, terminal_condition_reopen_cmd): + if not _same_logical_condition_wait( + opened, terminal_condition_reopen_cmd, chain_first=terminal_condition_chain_first, + ): break opened_id = opened.get("condition_wait_id") if not isinstance(opened_id, str): @@ -7839,6 +7852,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if has_reopened_same_wait and consumed_reopen_for_current_wait else None ) + terminal_condition_chain_first = wait_opened[first_open_index] next_value = True wait_yield_count += 1 break @@ -7848,6 +7862,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if has_reopened_same_wait and consumed_reopen_for_current_wait else None ) + terminal_condition_chain_first = wait_opened[first_open_index] next_value = True wait_yield_count += 1 break diff --git a/tests/condition_fingerprint_predicates.py b/tests/condition_fingerprint_predicates.py index 53cee3d..0ca5501 100644 --- a/tests/condition_fingerprint_predicates.py +++ b/tests/condition_fingerprint_predicates.py @@ -60,8 +60,23 @@ def ready() -> bool: return ready + # Python 3.12 inlines list, dict and set comprehensions (PEP 709), so the + # comprehension disappears from these lambdas' __qualname__. + def _in_listcomp(self) -> Callable[[], bool]: + return [lambda: self.approved for _ in range(1)][0] + + def _in_dictcomp(self) -> Callable[[], bool]: + return {key: lambda: bool(self.items) for key in ["only"]}["only"] + + def _in_setcomp(self) -> Callable[[], bool]: + return next(iter({lambda: len(self.items) > 1 for _ in range(1)})) + + def _same_line_in_listcomp(self) -> list[Callable[[], bool]]: + return [(lambda: value > 0) if value else (lambda: not value) for value in [1, 0]] # noqa: B023 + def predicates(self) -> dict[str, Any]: first, second = self._same_line() + positive, negative = self._same_line_in_listcomp() return { "lambda": lambda: self.approved, "genexpr": lambda: any(n > 10 for n in self.items), @@ -75,4 +90,9 @@ def predicates(self) -> dict[str, Any]: "bound_genexpr_method": self.has_large, "inner_def": self._inner(), "typed_wrapper": TypedReady().ready, + "lambda_in_listcomp": self._in_listcomp(), + "lambda_in_dictcomp": self._in_dictcomp(), + "lambda_in_setcomp": self._in_setcomp(), + "same_line_in_listcomp_first": positive, + "same_line_in_listcomp_second": negative, } diff --git a/tests/fixtures/condition_fingerprints_v2.json b/tests/fixtures/condition_fingerprints_v2.json index 3302ed4..fb83d96 100644 --- a/tests/fixtures/condition_fingerprints_v2.json +++ b/tests/fixtures/condition_fingerprints_v2.json @@ -13,6 +13,11 @@ "bound_method": "dwpy2:sha256:4b9aabe93ef8c48c2f05481ccb92f743da7c19b76104816c4a3f12ab32d741b9", "bound_genexpr_method": "dwpy2:sha256:9bbb6600801559c28420aa32085bfdc76ceef33e03c2d9194d2edb507053c934", "inner_def": "dwpy2:sha256:7be3326bb020ab50b9d1bad496b395551db90324bf6b0b0af9fe888e9dcfdd84", - "typed_wrapper": "dwpy2:sha256:f5f8904a70119a2f2999a4a256bc42549f48aca0c74c47ab42d256db852b919a" + "typed_wrapper": "dwpy2:sha256:f5f8904a70119a2f2999a4a256bc42549f48aca0c74c47ab42d256db852b919a", + "lambda_in_listcomp": "dwpy2:sha256:f8f58909db214449f4f0e7fbe94937b06bf98c83bb0c995563352f1d760fdb80", + "lambda_in_dictcomp": "dwpy2:sha256:a9ff285d478d5859d137782384a0df64f3f7236a8601ee25278b5e750def4e3d", + "lambda_in_setcomp": "dwpy2:sha256:4d998d02bda534ead4a89dacb82846aaf72c96624aad0c68516c11653f3fe55f", + "same_line_in_listcomp_first": "dwpy2:sha256:3b5d4ea892eb8458abd074e793d6f6fc35118bf1032530a8d16773d68bdcb96d", + "same_line_in_listcomp_second": "dwpy2:sha256:7bbdd2c46be01c6ce8ea970a2fa68f8cf17b8643a8d58b752e8cc6c21e27a641" } } diff --git a/tests/test_condition_fingerprint.py b/tests/test_condition_fingerprint.py index d3f4e5c..83a6088 100644 --- a/tests/test_condition_fingerprint.py +++ b/tests/test_condition_fingerprint.py @@ -207,3 +207,51 @@ def test_two_lambdas_on_one_line_get_their_own_fingerprints() -> None: assert GOLDEN["fingerprints"]["same_line_first"] != GOLDEN["fingerprints"]["same_line_second"] assert fingerprint(predicates["same_line_first"]) == GOLDEN["fingerprints"]["same_line_first"] assert fingerprint(predicates["same_line_second"]) == GOLDEN["fingerprints"]["same_line_second"] + + +def test_inlined_comprehension_segments_are_dropped_from_the_qualname() -> None: + # 3.11 names a lambda in a list comprehension f...; + # 3.12 inlines the comprehension (PEP 709) and names it f... + predicates = Approvals().predicates() + for name in ("lambda_in_listcomp", "lambda_in_dictcomp", "lambda_in_setcomp"): + assert fingerprint(predicates[name]) == GOLDEN["fingerprints"][name] + + +# A keyless wait without source (eval, .pyc-only) still links its false-wake +# reopen within its chain, as 2.14.0 did with its process-local v1. +@workflow.defn(name="tests.fingerprint.keyless-no-source") +class KeylessNoSource: + def __init__(self) -> None: + self.approved = False + + @workflow.signal("approve") + def approve(self) -> None: + self.approved = True + + @workflow.signal("note") + def note(self) -> None: + return None + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(eval("lambda self=self: self.approved")) # noqa: S307 + return "approved" + + +def test_a_keyless_wait_without_source_links_its_false_wake_reopen() -> None: + history = [START, _opened("w1", 1, None, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, None, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(KeylessNoSource, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_a_keyless_wait_without_source_reopens_without_a_fingerprint() -> None: + (command,) = replay(KeylessNoSource, [START, _opened("w1", 1, None, 1), _signal("note", 1, 2)], []).commands + assert isinstance(command, WaitCondition) and command.condition_definition_fingerprint is None + + +def test_terminal_cleanup_links_a_trailing_v1_duplicate_within_its_chain() -> None: + # A v1 keyless chain (opened before the upgrade, reopened with the same v1), + # then a trailing duplicate open recorded by a second queued task. + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4), + _opened("w3", 3, FOREIGN_V1, 5)] + assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] From 4a3691e1d3024a485032100a867a347a8309794b Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:44:43 -0600 Subject: [PATCH 5/8] Keep the stalled-response test independent of TCP setup latency --- tests/test_client_timeouts.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/test_client_timeouts.py b/tests/test_client_timeouts.py index 9d808ac..33a5b2b 100644 --- a/tests/test_client_timeouts.py +++ b/tests/test_client_timeouts.py @@ -60,10 +60,15 @@ async def respond(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> @pytest.mark.asyncio @pytest.mark.parametrize("operation", ["control", "discovery", "worker"]) async def test_configured_timeout_bounds_a_stalled_response(operation: str) -> None: + async def allow_connection_setup(request: httpx.Request) -> None: + # Exercise the configured read budget even when cold TCP setup is slow. + request.extensions["timeout"]["connect"] = 1.0 + async with ( response_server(delay=None) as endpoint, Client(endpoint, timeout=0.05, retry_policy=TransportRetryPolicy(max_attempts=1)) as client, ): + client._http.event_hooks["request"].append(allow_connection_setup) if operation == "control": request = client.health() elif operation == "discovery": From 9f867e1884b87ff4bac6d7cf202791047c2241fb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:53:57 -0600 Subject: [PATCH 6/8] Keep same-line comprehension predicates portable before and after PEP 709 --- CHANGELOG.md | 2 ++ .../_condition_fingerprint.py | 9 ++++++++- tests/condition_fingerprint_predicates.py | 19 +++++++++++++++++++ tests/fixtures/condition_fingerprints_v2.json | 8 +++++++- tests/test_condition_fingerprint.py | 8 ++++++++ 5 files changed, 44 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5bbed22..52a6f2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. A keyless wait without source links its false-wake reopens within its chain, as before. **Rollback and mixed fleets:** a run that opened a wait under this release records a v2 fingerprint, which 2.14.0 and earlier refuse on replay. Don't roll a task queue back below this release once such waits exist, and upgrade every worker on a queue together. (#258) +- Same-line `wait_condition` lambdas whose bodies contain list, dict or set comprehensions share one combined fingerprint on every Python version. Put those predicates on separate lines to detect a change from one candidate to another. (#258) + ## [2.14.0] - 2026-11-04 **Upgrade from 2.13.2 and earlier.** Every earlier release fails replay when a condition wait has an orphaned duplicate open whose timer fires later, which Server 3.0.x can record (#253 below). diff --git a/src/durable_workflow/_condition_fingerprint.py b/src/durable_workflow/_condition_fingerprint.py index 88eb1d2..690d920 100644 --- a/src/durable_workflow/_condition_fingerprint.py +++ b/src/durable_workflow/_condition_fingerprint.py @@ -141,7 +141,14 @@ def _predicate_node(function: Any) -> str | None: if function.__name__ == "": lambdas = [node for node in ast.walk(tree) if isinstance(node, ast.Lambda) and node.lineno == first_line] lambdas.sort(key=lambda node: node.col_offset) - if len(lambdas) > 1: + # PEP 709 moves comprehension instructions into the enclosing lambda. + # Candidates indistinguishable before 3.12 can become distinct after + # inlining, so keep the combined AST on every version in this case. + has_inlined_comprehension = any( + isinstance(child, ast.ListComp | ast.DictComp | ast.SetComp) + for node in lambdas for child in ast.walk(node) + ) + if len(lambdas) > 1 and not has_inlined_comprehension: # Names and constants are stable across versions; bytecode only # breaks a tie, and is compared within this interpreter. matching = [node for node in lambdas if _compiles_to(node, code, compare_bytecode=False)] diff --git a/tests/condition_fingerprint_predicates.py b/tests/condition_fingerprint_predicates.py index 0ca5501..8184a40 100644 --- a/tests/condition_fingerprint_predicates.py +++ b/tests/condition_fingerprint_predicates.py @@ -74,9 +74,22 @@ def _in_setcomp(self) -> Callable[[], bool]: def _same_line_in_listcomp(self) -> list[Callable[[], bool]]: return [(lambda: value > 0) if value else (lambda: not value) for value in [1, 0]] # noqa: B023 + def _same_line_listcomp_bodies(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + return (lambda: any([n > 0 for n in self.items])), (lambda: any([n < 0 for n in self.items])) + + def _same_line_dictcomp_bodies(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + items = self.items + return (lambda: any({n: n > 0 for n in items}.values())), (lambda: any({n: n < 0 for n in items}.values())) + + def _same_line_setcomp_bodies(self) -> tuple[Callable[[], bool], Callable[[], bool]]: + return (lambda: any({n > 0 for n in self.items})), (lambda: any({n < 0 for n in self.items})) + def predicates(self) -> dict[str, Any]: first, second = self._same_line() positive, negative = self._same_line_in_listcomp() + list_first, list_second = self._same_line_listcomp_bodies() + dict_first, dict_second = self._same_line_dictcomp_bodies() + set_first, set_second = self._same_line_setcomp_bodies() return { "lambda": lambda: self.approved, "genexpr": lambda: any(n > 10 for n in self.items), @@ -95,4 +108,10 @@ def predicates(self) -> dict[str, Any]: "lambda_in_setcomp": self._in_setcomp(), "same_line_in_listcomp_first": positive, "same_line_in_listcomp_second": negative, + "same_line_listcomp_body_first": list_first, + "same_line_listcomp_body_second": list_second, + "same_line_dictcomp_body_first": dict_first, + "same_line_dictcomp_body_second": dict_second, + "same_line_setcomp_body_first": set_first, + "same_line_setcomp_body_second": set_second, } diff --git a/tests/fixtures/condition_fingerprints_v2.json b/tests/fixtures/condition_fingerprints_v2.json index fb83d96..ea1c5b6 100644 --- a/tests/fixtures/condition_fingerprints_v2.json +++ b/tests/fixtures/condition_fingerprints_v2.json @@ -18,6 +18,12 @@ "lambda_in_dictcomp": "dwpy2:sha256:a9ff285d478d5859d137782384a0df64f3f7236a8601ee25278b5e750def4e3d", "lambda_in_setcomp": "dwpy2:sha256:4d998d02bda534ead4a89dacb82846aaf72c96624aad0c68516c11653f3fe55f", "same_line_in_listcomp_first": "dwpy2:sha256:3b5d4ea892eb8458abd074e793d6f6fc35118bf1032530a8d16773d68bdcb96d", - "same_line_in_listcomp_second": "dwpy2:sha256:7bbdd2c46be01c6ce8ea970a2fa68f8cf17b8643a8d58b752e8cc6c21e27a641" + "same_line_in_listcomp_second": "dwpy2:sha256:7bbdd2c46be01c6ce8ea970a2fa68f8cf17b8643a8d58b752e8cc6c21e27a641", + "same_line_listcomp_body_first": "dwpy2:sha256:00ce99ef06cf9b60f5831755f511382ea0fc968bf840a5ce8975363612f8e877", + "same_line_listcomp_body_second": "dwpy2:sha256:00ce99ef06cf9b60f5831755f511382ea0fc968bf840a5ce8975363612f8e877", + "same_line_dictcomp_body_first": "dwpy2:sha256:940867da138c61a2bfcf8c892a11cc286f6575519260e6e453d6e1bff57c1208", + "same_line_dictcomp_body_second": "dwpy2:sha256:940867da138c61a2bfcf8c892a11cc286f6575519260e6e453d6e1bff57c1208", + "same_line_setcomp_body_first": "dwpy2:sha256:f7440ac9c37954c7265ed231940426db90d507250819b5f8cb9fb3bd2ceb4e7f", + "same_line_setcomp_body_second": "dwpy2:sha256:f7440ac9c37954c7265ed231940426db90d507250819b5f8cb9fb3bd2ceb4e7f" } } diff --git a/tests/test_condition_fingerprint.py b/tests/test_condition_fingerprint.py index 83a6088..6ade543 100644 --- a/tests/test_condition_fingerprint.py +++ b/tests/test_condition_fingerprint.py @@ -217,6 +217,14 @@ def test_inlined_comprehension_segments_are_dropped_from_the_qualname() -> None: assert fingerprint(predicates[name]) == GOLDEN["fingerprints"][name] +@pytest.mark.parametrize("kind", ["listcomp", "dictcomp", "setcomp"]) +def test_same_line_comprehension_bodies_share_a_portable_combined_fingerprint(kind: str) -> None: + predicates = Approvals().predicates() + first = fingerprint(predicates[f"same_line_{kind}_body_first"]) + second = fingerprint(predicates[f"same_line_{kind}_body_second"]) + assert first == second == GOLDEN["fingerprints"][f"same_line_{kind}_body_first"] + + # A keyless wait without source (eval, .pyc-only) still links its false-wake # reopen within its chain, as 2.14.0 did with its process-local v1. @workflow.defn(name="tests.fingerprint.keyless-no-source") From d4cb876062d564819715f6f564a9a84c2585764f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 01:02:50 -0600 Subject: [PATCH 7/8] Use the terminal wait boundary to link cross-process legacy duplicates --- src/durable_workflow/workflow.py | 5 + tests/condition_fingerprint_workflows.py | 7 ++ ...t-v1-terminal-cross-process-duplicate.json | 104 ++++++++++++++++++ tests/test_condition_fingerprint.py | 17 +++ tests/test_replay_regression_corpus.py | 3 +- 5 files changed, 135 insertions(+), 1 deletion(-) create mode 100644 tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 6c3238a..772a2a2 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -6178,6 +6178,8 @@ def _refresh_replay_logging() -> None: terminal_condition_reopen_cmd: WaitCondition | None = None # The first open of the chain a terminal reopen cleanup belongs to (#258). terminal_condition_chain_first: Mapping[str, Any] | None = None + # Its terminal position proves whether a different-process duplicate belongs. + terminal_condition_chain_current: Mapping[str, Any] | None = None version_decisions: dict[str, int] = {} def _version_error(detail: str, step: _RecordedStep | None = None) -> NonDeterministicReplayError: @@ -6755,6 +6757,7 @@ def _consume_terminal_condition_reopens() -> None: opened = wait_opened[wait_yield_count] if not _same_logical_condition_wait( opened, terminal_condition_reopen_cmd, chain_first=terminal_condition_chain_first, + chain_current=terminal_condition_chain_current, next_index=wait_yield_count, ): break opened_id = opened.get("condition_wait_id") @@ -7853,6 +7856,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None else None ) terminal_condition_chain_first = wait_opened[first_open_index] + terminal_condition_chain_current = opened next_value = True wait_yield_count += 1 break @@ -7863,6 +7867,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None else None ) terminal_condition_chain_first = wait_opened[first_open_index] + terminal_condition_chain_current = opened next_value = True wait_yield_count += 1 break diff --git a/tests/condition_fingerprint_workflows.py b/tests/condition_fingerprint_workflows.py index a00aaca..4ce1a1e 100644 --- a/tests/condition_fingerprint_workflows.py +++ b/tests/condition_fingerprint_workflows.py @@ -41,3 +41,10 @@ def add(self, value: int) -> None: def run(self, ctx: workflow.WorkflowContext): # type: ignore[no-untyped-def] yield ctx.wait_condition(lambda: any(n > 10 for n in self.items), key="large") return list(self.items) + + +@workflow.defn(name="tests.replay.fingerprint-terminal-duplicate") +class FingerprintTerminalDuplicate(FingerprintApproval): + @workflow.signal("note") + def note(self) -> None: + return None diff --git a/tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json b/tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json new file mode 100644 index 0000000..2ef3458 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-fingerprint-v1-terminal-cross-process-duplicate.json @@ -0,0 +1,104 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-fingerprint-v1-terminal-cross-process-duplicate", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.fingerprint-terminal-duplicate", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.fingerprint-terminal-duplicate" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "w1", + "sequence": 1, + "condition_definition_fingerprint": "sha256:1111111111111111111111111111111111111111111111111111111111111111" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "note", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "w1", + "sequence": 1 + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "w2", + "sequence": 2, + "condition_definition_fingerprint": "sha256:1111111111111111111111111111111111111111111111111111111111111111" + }, + "sequence": 5 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "w3", + "sequence": 3, + "condition_definition_fingerprint": "sha256:2222222222222222222222222222222222222222222222222222222222222222" + }, + "sequence": 6 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 7 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "condition_wait_id": "w2", + "sequence": 2 + }, + "sequence": 8 + } + ], + "expected": { + "type": "complete_workflow", + "result": "approved" + } +} diff --git a/tests/test_condition_fingerprint.py b/tests/test_condition_fingerprint.py index 6ade543..06b2245 100644 --- a/tests/test_condition_fingerprint.py +++ b/tests/test_condition_fingerprint.py @@ -263,3 +263,20 @@ def test_terminal_cleanup_links_a_trailing_v1_duplicate_within_its_chain() -> No _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4), _opened("w3", 3, FOREIGN_V1, 5)] assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_terminal_cleanup_links_a_cross_process_duplicate_opened_before_terminal() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _opened("w3", 3, OTHER_V1, 3), + _signal("approve", 2, 4), _satisfied("w2", 2, 4)] + assert list(replay(KeylessApproval, history, []).commands) == [CompleteWorkflow("approved")] + + +def test_terminal_cleanup_refuses_an_unrelated_later_keyless_v1_open() -> None: + history = [START, _opened("w1", 1, FOREIGN_V1, 1), _signal("note", 1, 2), _satisfied("w1", 1, 2), + _opened("w2", 2, FOREIGN_V1, 3), _signal("approve", 2, 4), _satisfied("w2", 2, 4), + _opened("w3", 3, OTHER_V1, 5)] + with pytest.raises(NonDeterministicReplayError) as captured: + replay(KeylessApproval, history, []) + assert captured.value.workflow_sequence == 3 + assert captured.value.expected_shape == "complete workflow" diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 603e662..0cbcb86 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -25,7 +25,7 @@ commands_to_server_commands, query_state, ) -from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr +from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow from tests.test_condition_replay_drift import ConditionIdentityWorkflow from tests.test_generator_lifecycle import CleanupSideEffect, PendingFinally @@ -427,6 +427,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: TypedOnboarding, FingerprintApproval, FingerprintGenexpr, + FingerprintTerminalDuplicate, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow From 49eacbdd20baa0ddb9f597d5456bffe00b4d5a74 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 01:08:07 -0600 Subject: [PATCH 8/8] Wait for the worker heartbeat instead of assuming startup latency --- tests/test_worker.py | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/tests/test_worker.py b/tests/test_worker.py index 141df05..b795a3d 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -4911,6 +4911,13 @@ class TestWorkerHeartbeats: async def test_run_drives_periodic_heartbeats_with_slot_state( self, mock_client: AsyncMock ) -> None: + heartbeat_sent = asyncio.Event() + + async def heartbeat(**kwargs: object) -> dict[str, bool]: + heartbeat_sent.set() + return {"acknowledged": True} + + mock_client.heartbeat_worker = AsyncMock(side_effect=heartbeat) worker = Worker( mock_client, task_queue="q1", @@ -4922,11 +4929,13 @@ async def test_run_drives_periodic_heartbeats_with_slot_state( heartbeat_interval=0.05, ) run_task = asyncio.create_task(worker.run()) - await asyncio.sleep(0.2) - await worker.stop() - run_task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await run_task + try: + await asyncio.wait_for(heartbeat_sent.wait(), timeout=2) + finally: + await worker.stop() + run_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await run_task assert mock_client.heartbeat_worker.call_count >= 1 kwargs = mock_client.heartbeat_worker.call_args.kwargs