Advance replay time on persisted scalar child outcomes

src/durable_workflow/replay_clock.py

@@ -22,6 +22,10 @@
     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})
+_CHILD_FAMILY = "child workflow"
 
 
 def _task() -> object | None:
@@ -125,6 +129,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 +144,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 +210,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_FAMILY}:
             return
-        if not self._scalar_boundary(index, family, sequence):
+        boundary = (self._child_boundary(index, sequence) if family == _CHILD_FAMILY
+                    else self._scalar_boundary(index, family, sequence))
+        if not boundary:
             return
         event = self.events[index]
         payload = event["payload"]
@@ -209,6 +224,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_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))
+
     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 +300,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:

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])

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")

tests/fixtures/replay_regressions/business-replay-clock-child-results.json

@@ -0,0 +1,133 @@
+{
+  "$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": "2026-11-01T00:00:01+00:00"
+      },
+      "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": "2026-11-01T00:00:01+00:00"
+      },
+      "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"
+          }
+        }
+      }
+    ]
+  }
+}