From 238e55799b97dc567251b0ced746e192d260016c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 04:30:17 +0000 Subject: [PATCH 1/2] Advance business replay time on correlated child outcomes --- src/durable_workflow/replay_clock.py | 53 +++- src/durable_workflow/workflow.py | 6 +- tests/business_clock_workflows.py | 30 ++- .../business-replay-clock-child-results.json | 135 ++++++++++ tests/test_business_replay_clock_child.py | 238 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 2 + 6 files changed, 454 insertions(+), 10 deletions(-) create mode 100644 tests/fixtures/replay_regressions/business-replay-clock-child-results.json create mode 100644 tests/test_business_replay_clock_child.py diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 017b08a..8573830 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -22,6 +22,9 @@ r"(?:\.[0-9]{1,6})?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])", ) _ACTIVITY_TERMINALS = frozenset({"ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled"}) +_CHILD_TERMINALS = frozenset({"ChildRunCompleted", "ChildRunFailed", "ChildRunCancelled", "ChildRunTerminated"}) +_CHILD_IDENTITIES = ("child_call_id", "child_workflow_instance_id", "child_workflow_run_id", "child_workflow_type") +_CHILD_EVENTS = frozenset({"ChildWorkflowScheduled", "ChildRunStarted", *_CHILD_TERMINALS}) def _task() -> object | None: @@ -125,6 +128,7 @@ class BusinessReplayClock: def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id: str) -> None: self.events = events self._by_sequence: dict[int, list[int]] = {} + self._child_calls: dict[str, list[int]] = {} self._source_run: str | None = None self._resume_step: int | None = None self._redrive_inserted: datetime | None = None @@ -139,9 +143,17 @@ def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id self._pending = self.initial self._current = self.initial for index, event in enumerate(events): - payload = event.get("payload") - if isinstance(payload, Mapping) and (sequence := _sequence(payload)) is not None: - self._by_sequence.setdefault(sequence, []).append(index) + self._index_event(index, event) + + def _index_event(self, index: int, event: Mapping[str, Any]) -> None: + payload = event.get("payload") + if not isinstance(payload, Mapping): + return + if (sequence := _sequence(payload)) is not None: + self._by_sequence.setdefault(sequence, []).append(index) + call = payload.get("child_call_id") + if _kind(event) in _CHILD_EVENTS and _identity(call): + self._child_calls.setdefault(call, []).append(index) def _start_prefix(self, index: int, payload: Mapping[str, Any]) -> bool: if index == 0: @@ -197,9 +209,11 @@ def unavailable(self) -> None: def observe(self, index: int | None, family: str, sequence: int) -> None: self._pending = (None, None) - if index is None or not 0 <= index < len(self.events) or family not in {"activity", "timer"}: + if index is None or not 0 <= index < len(self.events) or family not in {"activity", "timer", "child workflow"}: return - if not self._scalar_boundary(index, family, sequence): + boundary = (self._child_boundary(index, sequence) if family == "child workflow" + else self._scalar_boundary(index, family, sequence)) + if not boundary: return event = self.events[index] payload = event["payload"] @@ -209,6 +223,31 @@ def observe(self, index: int | None, family: str, sequence: int) -> None: time = _time(_recorded_value(event)) self._pending = (index, time) + def _child_boundary(self, index: int, sequence: int) -> bool: + indices = self._by_sequence.get(sequence, []) + if len(indices) not in (2, 3) or indices[-1] != index: + return False + kinds = [_kind(self.events[position]) for position in indices] + if kinds[0] != "ChildWorkflowScheduled" or kinds[-1] not in _CHILD_TERMINALS: + return False + if len(kinds) == 3 and kinds[1] != "ChildRunStarted": + return False + payload = self.events[index].get("payload") + if not isinstance(payload, Mapping) or _sequence(payload) != sequence: + return False + if not all(_identity(payload.get(key)) for key in _CHILD_IDENTITIES): + return False + if self._child_calls.get(payload["child_call_id"]) != indices: + return False + return all(self._child_row(position, payload) for position in indices) + + def _child_row(self, index: int, terminal: Mapping[str, Any]) -> bool: + payload = self.events[index].get("payload") + return (self._scalar_row(index, "child workflow", "child_call_id", terminal["child_call_id"], _CHILD_EVENTS) + and isinstance(payload, Mapping) + and payload.get("workflow_type", terminal["child_workflow_type"]) == terminal["child_workflow_type"] + and all(payload.get(key) == terminal[key] for key in _CHILD_IDENTITIES)) + def _reused_time(self, payload: Mapping[str, Any], family: str, sequence: int) -> datetime | None: if (self._source_run is None or payload.get("reused_from_run_id") != self._source_run or self._resume_step is None or sequence >= self._resume_step): @@ -260,7 +299,9 @@ def _scalar_lifecycle( return False return not openings or openings[0] < index - def _scalar_row(self, index: int, family: str, resource: str, identity: str, allowed: set[str]) -> bool: + def _scalar_row( + self, index: int, family: str, resource: str, identity: str, allowed: set[str] | frozenset[str], + ) -> bool: row = self.events[index] body = row.get("payload") if not isinstance(body, Mapping) or _kind(row) not in allowed or body.get(resource) != identity: diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 2dc7a87..2f3f0f9 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2859,8 +2859,8 @@ def now(self) -> datetime: def replay_now(self) -> datetime: """Return the proved UTC instant of the supported consumed boundary. - This additive clock supports scalar persisted activities/timers and an - explicit redrive start. Ordinary entry, inline/local/grouped operations + This additive clock supports scalar persisted activities, timers and + children, plus an explicit redrive start. Ordinary entry, inline/local/grouped operations and unproved callbacks remain unavailable. Repeated reads are stable; equal/backward recorded times retain their values and precision. Adopt business-time decisions through normal workflow versioning. ``now()`` @@ -7949,7 +7949,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None _assert_next_step_matches(cmd) val = resolved_results[result_cursor] step = recorded_steps[result_cursor] - if isinstance(cmd, ScheduleActivity | StartTimer): + if isinstance(cmd, ScheduleActivity | StartTimer | StartChildWorkflow): ctx._business_clock.observe(step.history_index, step.shape, step.workflow_sequence) if not isinstance(cmd, RecordLocalActivity) or prepare_local_activities: _advance_cancellation_clock([recorded_steps[result_cursor].history_index]) diff --git a/tests/business_clock_workflows.py b/tests/business_clock_workflows.py index cf620d5..cbc05fb 100644 --- a/tests/business_clock_workflows.py +++ b/tests/business_clock_workflows.py @@ -6,7 +6,7 @@ from typing import Any from durable_workflow import workflow -from durable_workflow.errors import DurableWorkflowError +from durable_workflow.errors import ChildWorkflowFailed, DurableWorkflowError from durable_workflow.workflow import WorkflowContext @@ -77,3 +77,31 @@ def run(self, ctx: WorkflowContext, name: str, seconds: int): # type: ignore[no yield ctx.start_timer(seconds) observations.append(clock_text(ctx)) return {"clock": observations, "legacy": ctx.now().isoformat(timespec="microseconds")} + + +@workflow.defn(name="tests.replay.business-clock-child") +class ChildBusinessClockWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + self.ctx = ctx + observations = [clock_text(ctx)] + legacy = [ctx.now().isoformat(timespec="microseconds")] + value = yield ctx.start_child_workflow("clock.child", []) + observations.append(clock_text(ctx)) + repeated = clock_text(ctx) == observations[-1] + legacy.append(ctx.now().isoformat(timespec="microseconds")) + failure = None + try: + yield ctx.start_child_workflow("clock.child", []) + except ChildWorkflowFailed as error: + failure = {"kind": error.failure_kind, "run": error.child_workflow_run_id, "message": str(error)} + observations.append(clock_text(ctx)) + legacy.append(ctx.now().isoformat(timespec="microseconds")) + return {"clock": observations, "legacy": legacy, "repeated": repeated, "result": value, "failure": failure} + + @workflow.query("clock") + def clock(self) -> str | None: + return clock_text(self.ctx) + + @workflow.query("explode") + def explode(self) -> None: + raise ValueError("child-clock-query-error") diff --git a/tests/fixtures/replay_regressions/business-replay-clock-child-results.json b/tests/fixtures/replay_regressions/business-replay-clock-child-results.json new file mode 100644 index 0000000..c29dc66 --- /dev/null +++ b/tests/fixtures/replay_regressions/business-replay-clock-child-results.json @@ -0,0 +1,135 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "business-replay-clock-child-results", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.business-clock-child", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-01T00:00:00.000000+00:00", + "payload": {}, + "sequence": 1 + }, + { + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-01T00:00:00.000000+00:00", + "payload": { + "sequence": 1, + "child_call_id": "call-1", + "child_workflow_instance_id": "child-1", + "child_workflow_run_id": "run-1", + "child_workflow_type": "clock.child" + }, + "sequence": 2 + }, + { + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-01T00:00:00.000000+00:00", + "payload": { + "sequence": 1, + "child_call_id": "call-1", + "child_workflow_instance_id": "child-1", + "child_workflow_run_id": "run-1", + "child_workflow_type": "clock.child" + }, + "sequence": 3 + }, + { + "event_type": "ChildRunCompleted", + "recorded_at": "2026-11-01T00:00:05.654321+00:00", + "payload": { + "sequence": 1, + "child_call_id": "call-1", + "child_workflow_instance_id": "child-1", + "child_workflow_run_id": "run-1", + "child_workflow_type": "clock.child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoQcmVjb3JkZWQ=" + }, + "closed_at": "2098-01-01T00:00:00Z", + "reused_recorded_at": "2099-01-01T00:00:00Z" + }, + "sequence": 4 + }, + { + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-01T00:00:00.000000+00:00", + "payload": { + "sequence": 2, + "child_call_id": "call-2", + "child_workflow_instance_id": "child-2", + "child_workflow_run_id": "run-2", + "child_workflow_type": "clock.child" + }, + "sequence": 5 + }, + { + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-01T00:00:00.000000+00:00", + "payload": { + "sequence": 2, + "child_call_id": "call-2", + "child_workflow_instance_id": "child-2", + "child_workflow_run_id": "run-2", + "child_workflow_type": "clock.child" + }, + "sequence": 6 + }, + { + "event_type": "ChildRunFailed", + "recorded_at": "2026-11-01T00:00:03.123456+00:00", + "payload": { + "sequence": 2, + "child_call_id": "call-2", + "child_workflow_instance_id": "child-2", + "child_workflow_run_id": "run-2", + "child_workflow_type": "clock.child", + "message": "recorded child failure", + "closed_at": "2098-01-01T00:00:00Z", + "reused_recorded_at": "2099-01-01T00:00:00Z" + }, + "sequence": 7 + }, + { + "event_type": "WorkflowTaskCompleted", + "recorded_at": "2099-01-01T00:00:00Z", + "payload": {}, + "sequence": 8 + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "clock": [ + null, + "2026-11-01T00:00:05.654321+00:00", + "2026-11-01T00:00:03.123456+00:00" + ], + "legacy": [ + "2026-11-01T00:00:00.000000+00:00", + "2026-11-01T00:00:00.000000+00:00", + "2026-11-01T00:00:00.000000+00:00" + ], + "repeated": true, + "result": "recorded", + "failure": { + "kind": "child_workflow", + "run": "run-2", + "message": "recorded child failure" + } + } + } + ] + } +} diff --git a/tests/test_business_replay_clock_child.py b/tests/test_business_replay_clock_child.py new file mode 100644 index 0000000..bc8d26e --- /dev/null +++ b/tests/test_business_replay_clock_child.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +import pytest + +from durable_workflow import ReplayClockUnavailable, serializer +from durable_workflow.errors import ChildWorkflowFailed, NonDeterministicReplayError, QueryFailed +from durable_workflow.workflow import CompleteWorkflow, FailWorkflow, WorkflowContext, query_state, replay +from tests.business_clock_workflows import ChildBusinessClockWorkflow, clock_text +from tests.test_business_replay_clock import result + +START = "2026-11-01T00:00:00.000000+00:00" +COMPLETED = "2026-11-01T00:00:05.654321+00:00" +FAILED = "2026-11-01T00:00:03.123456+00:00" + + +def child_rows(sequence: int, kind: str, time: Any, *, started: bool = True) -> list[dict[str, Any]]: + payload = {"sequence": sequence, "child_call_id": f"call-{sequence}", + "child_workflow_instance_id": f"child-{sequence}", "child_workflow_run_id": f"run-{sequence}", + "child_workflow_type": "clock.child"} + rows = [{"event_type": "ChildWorkflowScheduled", "recorded_at": START, "payload": dict(payload)}] + if started: + rows.append({"event_type": "ChildRunStarted", "recorded_at": START, "payload": dict(payload)}) + if kind == "ChildRunCompleted": + payload["result"] = serializer.envelope("recorded") + else: + payload["message"] = "recorded child failure" + payload.update(closed_at="2098-01-01T00:00:00Z", reused_recorded_at="2099-01-01T00:00:00Z") + rows.append({"event_type": kind, "recorded_at": time, "payload": payload}) + return rows + + +def history() -> list[dict[str, Any]]: + return [{"event_type": "WorkflowStarted", "recorded_at": START, "payload": {}}, + *child_rows(1, "ChildRunCompleted", COMPLETED), *child_rows(2, "ChildRunFailed", FAILED)] + + +def execute(events: list[dict[str, Any]]) -> dict[str, Any]: + return dict(result(events, ChildBusinessClockWorkflow)) + + +def test_actual_completion_caught_failure_and_cold_replay_use_parent_resolution_time() -> None: + events = history() + events.append({"event_type": "WorkflowTaskCompleted", "recorded_at": "2099-01-01T00:00:00Z", "payload": {}}) + original = deepcopy(events) + first = execute(events) + assert first == execute(events) + assert first == {"clock": [None, COMPLETED, FAILED], "legacy": [START] * 3, "repeated": True, + "result": "recorded", "failure": {"kind": "child_workflow", "run": "run-2", + "message": "recorded child failure"}} + assert events == original + + +@pytest.mark.parametrize("time,expected", [(COMPLETED, COMPLETED), (FAILED, FAILED), + ("2026-11-01T02:00:03.123456+02:00", FAILED)]) +def test_equal_backward_and_offset_times_preserve_the_exact_consumed_instant(time: str, expected: str) -> None: + events = history() + events[-1]["recorded_at"] = time + assert execute(events)["clock"] == [None, COMPLETED, expected] + + +@pytest.mark.parametrize("kind,failure_kind", [("ChildRunCompleted", None), ("ChildRunFailed", "child_workflow"), + ("ChildRunCancelled", "cancelled"), ("ChildRunTerminated", "terminated")]) +def test_each_child_terminal_preserves_its_original_outcome_and_own_time(kind: str, failure_kind: str | None) -> None: + class Single: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + value = yield ctx.start_child_workflow("clock.child", []) + outcome = value + except ChildWorkflowFailed as error: + outcome = (error.failure_kind, error.child_workflow_run_id) + return outcome, clock_text(ctx), ctx.replay_now() == ctx.replay_now() + + rows = child_rows(1, kind, COMPLETED) + expected = "recorded" if failure_kind is None else (failure_kind, "run-1") + assert result(rows, Single) == (expected, COMPLETED, True) + + +@pytest.mark.parametrize("field", ["child_call_id", "child_workflow_instance_id", "child_workflow_run_id"]) +@pytest.mark.parametrize("row", [1, 2, 3]) +@pytest.mark.parametrize("replacement", [None, "", "foreign", 17]) +def test_missing_empty_conflicting_or_wrong_type_identity_keeps_legacy_result_without_time( + field: str, row: int, replacement: Any, +) -> None: + events = history() + if replacement is None: + events[row]["payload"].pop(field) + else: + events[row]["payload"][field] = replacement + value = execute(events) + assert value["result"] == "recorded" + assert value["clock"] == [None, None, FAILED] + + +def test_type_is_required_and_a_conflicting_alias_cannot_supply_authority() -> None: + missing = history() + for row in missing[1:4]: + row["payload"].pop("child_workflow_type") + assert execute(missing)["clock"] == [None, None, FAILED] + conflict = history() + for row in conflict[1:4]: + row["payload"].update(workflow_type="clock.child", child_workflow_type="foreign.child") + assert execute(conflict)["clock"] == [None, None, FAILED] + + +def test_existing_child_type_mismatch_still_raises_the_original_validation_error() -> None: + events = history() + for row in events[1:4]: + row["payload"]["child_workflow_type"] = "foreign.child" + with pytest.raises(NonDeterministicReplayError): + execute(events) + + +@pytest.mark.parametrize("bad", [None, False, 17, "", "not-a-time", "2026-11-01T00:00:03", + "2026-02-30T00:00:00Z", "2026-11-01T24:00:00Z"]) +def test_invalid_resolution_time_does_not_borrow_other_time_then_next_child_can_recover(bad: Any) -> None: + events = history() + events[3]["recorded_at"] = bad + assert execute(events)["clock"] == [None, None, FAILED] + + +def test_public_timestamp_is_supported_but_invalid_present_value_cannot_fall_back() -> None: + events = history() + # Keep the established legacy start-clock carrier; this changes only the + # public child rows consumed by the additive clock. + for row in events[1:]: + row["timestamp"] = row.pop("recorded_at") + assert execute(events)["clock"] == [None, COMPLETED, FAILED] + events[3]["recorded_at"] = COMPLETED + events[3]["timestamp"] = "invalid" + assert execute(events)["clock"] == [None, None, FAILED] + + +def test_optional_start_is_supported_but_duplicate_and_changing_run_lineage_are_unavailable() -> None: + events = history() + del events[2] + assert execute(events)["clock"] == [None, COMPLETED, FAILED] + duplicate = history() + other = deepcopy(duplicate[3]) + other["recorded_at"] = "2099-01-01T00:00:00Z" + duplicate.insert(4, other) + assert execute(duplicate)["clock"] == [None, None, FAILED] + changed_run = history() + for row in changed_run[2:4]: + row["payload"]["child_workflow_run_id"] = "continued-child-run" + assert execute(changed_run)["clock"] == [None, None, FAILED] + + +def test_cross_call_or_unindexed_start_cannot_complete_another_childs_proof() -> None: + events = history() + events[2]["payload"]["sequence"] = "1" + assert execute(events)["clock"] == [None, None, FAILED] + events = history() + for row in events[4:]: + row["payload"]["child_call_id"] = "call-1" + assert execute(events)["clock"] == [None, None, None] + + +@pytest.mark.parametrize("row", [1, 2]) +def test_duplicate_opening_or_start_keeps_legacy_outcome_but_not_time(row: int) -> None: + events = history() + events.insert(row, deepcopy(events[row])) + value = execute(events) + assert value["result"] == "recorded" + assert value["clock"] == [None, None, FAILED] + + +def test_contradictory_terminals_cannot_supply_time_or_change_legacy_failure() -> None: + observed: list[str | None] = [] + + class Parent: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.start_child_workflow("clock.child", []) + observed.append(clock_text(ctx)) + yield ctx.start_child_workflow("clock.child", []) + + events = history() + conflicting = deepcopy(events[3]) + conflicting["event_type"] = "ChildRunFailed" + conflicting["payload"]["message"] = "contradictory failure" + events.insert(4, conflicting) + outcome = replay(Parent, events, [], payload_codec="avro") + try: + assert len(outcome.commands) == 1 + failure = outcome.commands[0] + assert isinstance(failure, FailWorkflow) + assert failure.exception_type == "ChildWorkflowFailed" + assert failure.message == "contradictory failure" + finally: + outcome.close() + assert observed == [None] + + +def test_query_prefix_nested_replay_error_and_scope_exit_preserve_parent_time() -> None: + class Parent: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.start_child_workflow("clock.child", []) + before = ctx.replay_now() + assert execute(history())["clock"] == [None, COMPLETED, FAILED] + assert ctx.replay_now() == before + with pytest.raises(QueryFailed, match="child-clock-query-error"): + query_state(ChildBusinessClockWorkflow, history(), [], "explode", payload_codec="avro") + assert ctx.replay_now() == before + return clock_text(ctx) + + assert result(history()[:4], Parent) == COMPLETED + assert query_state(ChildBusinessClockWorkflow, history(), [], "clock", payload_codec="avro") == FAILED + prefix = history()[:4] + assert query_state(ChildBusinessClockWorkflow, prefix, [], "clock", payload_codec="avro") == COMPLETED + prefix.append({"event_type": "WorkflowTaskCompleted", "recorded_at": "2099-01-01T00:00:00Z", "payload": {}}) + assert query_state(ChildBusinessClockWorkflow, prefix, [], "clock", payload_codec="avro") is None + + +def test_child_time_does_not_survive_an_inline_boundary_on_first_or_cold_replay() -> None: + retained: list[WorkflowContext] = [] + + class Inline: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + retained.append(ctx) + yield ctx.start_child_workflow("clock.child", []) + before = clock_text(ctx) + yield ctx.side_effect(lambda: "recorded") + return before, clock_text(ctx) + + events = history()[:4] + outcome = replay(Inline, events, [], payload_codec="avro") + try: + assert isinstance(outcome.commands[-1], CompleteWorkflow) + assert outcome.commands[-1].result == (COMPLETED, None) + finally: + outcome.close() + events.append({"event_type": "SideEffectRecorded", "payload": { + "sequence": 2, "result": serializer.envelope("recorded")}, "recorded_at": FAILED}) + assert result(events, Inline) == (COMPLETED, None) + with pytest.raises(ReplayClockUnavailable): + retained[0].replay_now() diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 7ed4872..e025720 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -26,6 +26,7 @@ query_state, ) from tests.business_clock_workflows import ( + ChildBusinessClockWorkflow, NativeRedriveBusinessClockWorkflow, PrefixedCoreTimerBusinessClockWorkflow, ScalarBusinessClockWorkflow, @@ -376,6 +377,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: WORKFLOWS = [ + ChildBusinessClockWorkflow, NativeRedriveBusinessClockWorkflow, PrefixedCoreTimerBusinessClockWorkflow, ScalarBusinessClockWorkflow, From 8199fe8054f50d8a050429fdd8cdcd163f349f46 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 04:33:44 +0000 Subject: [PATCH 2/2] Keep child clock evidence outside redrive profiles --- src/durable_workflow/replay_clock.py | 7 ++++--- .../business-replay-clock-child-results.json | 6 ++---- tests/test_business_replay_clock_child.py | 3 ++- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 8573830..e827081 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -25,6 +25,7 @@ _CHILD_TERMINALS = frozenset({"ChildRunCompleted", "ChildRunFailed", "ChildRunCancelled", "ChildRunTerminated"}) _CHILD_IDENTITIES = ("child_call_id", "child_workflow_instance_id", "child_workflow_run_id", "child_workflow_type") _CHILD_EVENTS = frozenset({"ChildWorkflowScheduled", "ChildRunStarted", *_CHILD_TERMINALS}) +_CHILD_FAMILY = "child workflow" def _task() -> object | None: @@ -209,9 +210,9 @@ def unavailable(self) -> None: def observe(self, index: int | None, family: str, sequence: int) -> None: self._pending = (None, None) - if index is None or not 0 <= index < len(self.events) or family not in {"activity", "timer", "child workflow"}: + if index is None or not 0 <= index < len(self.events) or family not in {"activity", "timer", _CHILD_FAMILY}: return - boundary = (self._child_boundary(index, sequence) if family == "child workflow" + boundary = (self._child_boundary(index, sequence) if family == _CHILD_FAMILY else self._scalar_boundary(index, family, sequence)) if not boundary: return @@ -243,7 +244,7 @@ def _child_boundary(self, index: int, sequence: int) -> bool: def _child_row(self, index: int, terminal: Mapping[str, Any]) -> bool: payload = self.events[index].get("payload") - return (self._scalar_row(index, "child workflow", "child_call_id", terminal["child_call_id"], _CHILD_EVENTS) + return (self._scalar_row(index, _CHILD_FAMILY, "child_call_id", terminal["child_call_id"], _CHILD_EVENTS) and isinstance(payload, Mapping) and payload.get("workflow_type", terminal["child_workflow_type"]) == terminal["child_workflow_type"] and all(payload.get(key) == terminal[key] for key in _CHILD_IDENTITIES)) diff --git a/tests/fixtures/replay_regressions/business-replay-clock-child-results.json b/tests/fixtures/replay_regressions/business-replay-clock-child-results.json index c29dc66..8532b6f 100644 --- a/tests/fixtures/replay_regressions/business-replay-clock-child-results.json +++ b/tests/fixtures/replay_regressions/business-replay-clock-child-results.json @@ -55,8 +55,7 @@ "codec": "avro", "blob": "wwHioz3/VYAiNwoQcmVjb3JkZWQ=" }, - "closed_at": "2098-01-01T00:00:00Z", - "reused_recorded_at": "2099-01-01T00:00:00Z" + "closed_at": "2026-11-01T00:00:01+00:00" }, "sequence": 4 }, @@ -94,8 +93,7 @@ "child_workflow_run_id": "run-2", "child_workflow_type": "clock.child", "message": "recorded child failure", - "closed_at": "2098-01-01T00:00:00Z", - "reused_recorded_at": "2099-01-01T00:00:00Z" + "closed_at": "2026-11-01T00:00:01+00:00" }, "sequence": 7 }, diff --git a/tests/test_business_replay_clock_child.py b/tests/test_business_replay_clock_child.py index bc8d26e..68a8fd6 100644 --- a/tests/test_business_replay_clock_child.py +++ b/tests/test_business_replay_clock_child.py @@ -200,8 +200,9 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] before = ctx.replay_now() assert execute(history())["clock"] == [None, COMPLETED, FAILED] assert ctx.replay_now() == before + query_events = history() with pytest.raises(QueryFailed, match="child-clock-query-error"): - query_state(ChildBusinessClockWorkflow, history(), [], "explode", payload_codec="avro") + query_state(ChildBusinessClockWorkflow, query_events, [], "explode", payload_codec="avro") assert ctx.replay_now() == before return clock_text(ctx)