Make condition-wait fingerprints stable across Python versions and processes

CHANGELOG.md

@@ -10,6 +10,12 @@ 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. 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).

src/durable_workflow/_condition_fingerprint.py

@@ -0,0 +1,229 @@
+"""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] = {}
+_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"]
+
+
+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
+    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"{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__ == "<lambda>":
+        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)
+        # 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)]
+            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:
+            return None
+        # 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)
+        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 _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
+    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), "<condition-fingerprint>", "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) == "<lambda>"),
+        None,
+    )
+    return candidate is not None and (
+        (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
+        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 "<locals>" in const)
+    ]
+
+
+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__