From 7947da918963c77909bfd8904f0b7f24cfb7eb89 Mon Sep 17 00:00:00 2001 From: nandanadileep <110280757+nandanadileep@users.noreply.github.com> Date: Thu, 5 Nov 2026 03:04:59 +0000 Subject: [PATCH 1/2] fix: preserve replay state in virtual children --- .../context.py | 11 +- .../tests/context_test.py | 176 ++++++++++++++++++ 2 files changed, 185 insertions(+), 2 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py index e1957932e..678e2911c 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py @@ -606,10 +606,17 @@ def _replay_aware( was_replaying and next_exists and not next_terminal and not next_is_step ) # Before-the-op flips: - # - brand-new next op (no checkpoint): always flip to NEW. + # - brand-new next op (no checkpoint): flip to NEW, unless the operation + # is virtual. A virtual container never checkpoints itself, so the + # absence of a checkpoint at its id is by design and is not evidence + # of new work; its inner operations may still have history to replay. + # Keep replay status and let the child refine to NEW via its own + # look-ahead against its inner step ids. # - non-terminal STEP-type op (brand-new or retrying): the user function # is about to run real work, so flip to NEW before it. - if was_replaying and (not next_exists or (next_is_step and not next_terminal)): + if was_replaying and ( + (not next_exists and not is_virtual) or (next_is_step and not next_terminal) + ): self._set_replay_status_new() try: yield diff --git a/packages/aws-durable-execution-sdk-python/tests/context_test.py b/packages/aws-durable-execution-sdk-python/tests/context_test.py index beb9df4ad..9274c8dce 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -3303,6 +3303,182 @@ def on_operation_end(self, info): assert child_body_calls == [] +def _named_wait_op( + operation_id: str, + name: str, + parent_id: str | None, + status: OperationStatus = OperationStatus.SUCCEEDED, +) -> Operation: + """Build a WAIT checkpoint with the identity fields replay validation checks.""" + return Operation( + operation_id=operation_id, + operation_type=OperationType.WAIT, + status=status, + sub_type=OperationSubType.WAIT, + name=name, + parent_id=parent_id, + ) + + +def _install_update_recorder(state: ExecutionState) -> list[str]: + """Record operation ids passed to emit_operation_update_hook.""" + updates: list[str] = [] + state.emit_operation_update_hook = lambda op: updates.append(op.operation_id) # type: ignore[method-assign] + return updates + + +def test_virtual_child_preserves_replay_and_emits_update_for_inner_wait(): + """A virtual child with no container checkpoint keeps replay status. + + A virtual container intentionally has no checkpoint, so the absence of one + at its id must not flip the creator to NEW before the child is created. + Otherwise the inner wait's stored history is bypassed and its update + notification is dropped. Regression test for issue #889. + """ + state = _replay_state({}) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + context_id = parent._peek_next_operation_id() # noqa: SLF001 + child_probe = parent.create_child_context(context_id, is_virtual=True) + wait_id = child_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[wait_id] = _named_wait_op( # noqa: SLF001 + wait_id, "virtual-wait", parent_id=None + ) + state._updated_operation_ids.add(wait_id) # noqa: SLF001 + + child_entered_replay: list[bool] = [] + emitted_updates: list[str] = _install_update_recorder(state) + + def body(child: DurableContext) -> str: + child_entered_replay.append(child.is_replaying()) + child.wait(Duration.from_seconds(5), name="virtual-wait") + return "ok" + + result = parent.run_in_child_context( + body, name="virtual-context", config=ChildConfig(is_virtual=True) + ) + + # The child inherits REPLAY and the updated wait's notification is delivered. + assert child_entered_replay == [True] + assert emitted_updates == [wait_id] + assert result == "ok" + + +def test_replay_aware_brand_new_non_virtual_op_flips_to_new(): + """A brand-new non-virtual operation is new work and flips to NEW. + + Control for the virtual case: without `is_virtual`, a missing checkpoint at + the operation id means the operation is brand-new, so the context flips to + NEW before the body runs. + """ + ctx = DurableContext( + state=_replay_state({}), + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + inside_status: list[bool] = [] + with ctx._replay_aware(is_virtual=False): # noqa: SLF001 + inside_status.append(ctx.is_replaying()) + + assert inside_status == [False] + + +def test_replay_aware_brand_new_virtual_op_keeps_replay(): + """A brand-new virtual operation keeps replay status. + + A virtual container never checkpoints itself, so the missing checkpoint is + by design and must not flip the context to NEW before its child is created. + """ + ctx = DurableContext( + state=_replay_state({}), + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + inside_status: list[bool] = [] + with ctx._replay_aware(is_virtual=True): # noqa: SLF001 + inside_status.append(ctx.is_replaying()) + + assert inside_status == [True] + + +def test_virtual_child_refines_to_new_at_genuinely_new_inner_operation(): + """A virtual child keeps replay until its own look-ahead finds new work. + + The virtual container's missing checkpoint preserves the child in REPLAY, + but the first inner operation with no checkpoint is genuine new work and + must flip the child to NEW before it runs. + """ + state = _replay_state({}) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + child_entry_replay: list[bool] = [] + child_after_new_op: list[bool] = [] + + def body(child: DurableContext) -> str: + child_entry_replay.append(child.is_replaying()) + with child._replay_aware(): # noqa: SLF001 + child._create_step_id() # noqa: SLF001 - consume the brand-new op id + child_after_new_op.append(child.is_replaying()) + return "ok" + + parent.run_in_child_context( + body, name="virtual-context", config=ChildConfig(is_virtual=True) + ) + + assert child_entry_replay == [True] + assert child_after_new_op == [False] + + +def test_nested_virtual_child_preserves_replay_and_emits_update_for_inner_wait(): + """A virtual child nested in a virtual child still preserves replay status. + + Each virtual layer has no checkpoint by design. The inner wait's history + must survive both layers and its update notification delivered once. + """ + state = _replay_state({}) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + outer_id = parent._peek_next_operation_id() # noqa: SLF001 + outer_probe = parent.create_child_context(outer_id, is_virtual=True) + inner_id = outer_probe._peek_next_operation_id() # noqa: SLF001 + inner_probe = outer_probe.create_child_context(inner_id, is_virtual=True) + wait_id = inner_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[wait_id] = _named_wait_op( # noqa: SLF001 + wait_id, "nested-wait", parent_id=None + ) + state._updated_operation_ids.add(wait_id) # noqa: SLF001 + + inner_entered_replay: list[bool] = [] + emitted_updates: list[str] = _install_update_recorder(state) + + def inner_body(inner: DurableContext) -> str: + inner_entered_replay.append(inner.is_replaying()) + inner.wait(Duration.from_seconds(5), name="nested-wait") + return "ok" + + def outer_body(outer: DurableContext) -> str: + return outer.run_in_child_context( + inner_body, name="inner-context", config=ChildConfig(is_virtual=True) + ) + + result = parent.run_in_child_context( + outer_body, name="outer-context", config=ChildConfig(is_virtual=True) + ) + + assert inner_entered_replay == [True] + assert emitted_updates == [wait_id] + assert result == "ok" + + def test_replay_aware_emits_update_hook_for_operation_updated_since_last_invocation(): """Updated terminal operations emit operation_end, not replay start+end.""" captured: list[tuple[str, str, bool, OperationStatus]] = [] From 84b7b61fb9feefe312740348d573dba508d73cfb Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 6 Nov 2026 07:11:32 +0000 Subject: [PATCH 2/2] fix: keep replay after virtual children via first-inner look-ahead _next_operation_exists now treats a virtual child's first inner checkpoint as existence, matching the JS SDK. Tests cover a replayed step before a virtual child and two virtual children in a row, asserting plugin on_operation_end instead of patching emit_operation_update_hook. Co-authored-by: Nandana Dileep --- .../context.py | 23 +- .../tests/context_test.py | 302 +++++++++++++----- 2 files changed, 248 insertions(+), 77 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py index 678e2911c..8e6580e93 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py @@ -520,8 +520,17 @@ def _peek_next_checkpoint(self) -> CheckpointedResult: ) def _next_operation_exists(self) -> bool: - """True if a checkpoint exists for this context's next operation.""" - return self._peek_next_checkpoint().is_existent() + """True if the next operation or its first inner operation has a checkpoint.""" + if self._peek_next_checkpoint().is_existent(): + return True + # A virtual child never checkpoints itself, so its history starts at its + # first inner operation. + first_child_id: str = OperationIdNamespace( + self._peek_next_operation_id() + ).create_id_for_step(1) + return self.state.get_checkpoint_result( + first_child_id, claim_end_hook=False + ).is_existent() @contextmanager def _replay_aware( @@ -542,14 +551,20 @@ def _replay_aware( - Existence flip (before the op): if we are replaying and the next operation has no checkpoint at all, it is brand-new code, so flip to - NEW immediately so the operation and its logs count as new. + NEW immediately so the operation and its logs count as new. A virtual + child never checkpoints itself, so a missing checkpoint at its id is + not this flip: with `is_virtual=True` the context stays in replay and + the child refines against its inner operation ids. - Deferred status flip (after the op): a non-terminal next operation is a pure resume point with no user body. We keep replay status through the operation and flip to NEW afterwards, so the resuming operation's own logs stay de-duplicated but subsequent code counts as new. - Post-op existence flip (after the op): if we are still replaying once the operation completes and the *following* operation does not exist - yet, we have reached the replay boundary. + yet, we have reached the replay boundary. Existence includes a + checkpoint at the following operation's own id or, when that is + missing, at its first inner operation id. Virtual children never + checkpoint themselves, so their history starts at that inner id. """ was_replaying: bool = self.is_replaying() # Only look up a checkpoint when replaying. Operation call sites pass diff --git a/packages/aws-durable-execution-sdk-python/tests/context_test.py b/packages/aws-durable-execution-sdk-python/tests/context_test.py index 9274c8dce..67fecbb01 100644 --- a/packages/aws-durable-execution-sdk-python/tests/context_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/context_test.py @@ -2812,14 +2812,20 @@ def named(ctx: DurableContext) -> str: # region per-context replay status -def _replay_state(operations: dict[str, Operation]) -> ExecutionState: +def _replay_state( + operations: dict[str, Operation], + *, + plugin_executor: PluginExecutor | None = None, + updated_operation_ids: list[str] | None = None, +) -> ExecutionState: """Build a real ExecutionState seeded with the given operations.""" return ExecutionState( durable_execution_arn="arn:aws:durable:us-east-1:123456789012:execution/test", initial_checkpoint_token="token", # noqa: S106 operations=operations, service_client=Mock(), - plugin_executor=PluginExecutor(plugins=None), + plugin_executor=plugin_executor or PluginExecutor(plugins=None), + updated_operation_ids=updated_operation_ids, ) @@ -3320,11 +3326,43 @@ def _named_wait_op( ) -def _install_update_recorder(state: ExecutionState) -> list[str]: - """Record operation ids passed to emit_operation_update_hook.""" - updates: list[str] = [] - state.emit_operation_update_hook = lambda op: updates.append(op.operation_id) # type: ignore[method-assign] - return updates +def _named_step_op( + operation_id: str, + name: str, + result: str, + parent_id: str | None = None, +) -> Operation: + """Build a succeeded STEP checkpoint with the identity fields replay validation checks.""" + return Operation( + operation_id=operation_id, + operation_type=OperationType.STEP, + status=OperationStatus.SUCCEEDED, + sub_type=OperationSubType.STEP, + name=name, + parent_id=parent_id, + step_details=StepDetails(result=json.dumps(result)), + ) + + +def _capturing_end_plugin() -> tuple[ + list[tuple[str, str, bool, OperationStatus]], PluginExecutor +]: + """Return a recorder and a PluginExecutor that captures on_operation_end.""" + captured: list[tuple[str, str, bool, OperationStatus]] = [] + + class _CapturingPlugin(DurableInstrumentationPlugin): + def on_operation_end(self, info): + captured.append(("end", info.operation_id, info.is_replayed, info.status)) + + return captured, PluginExecutor(plugins=[plugin_factory(_CapturingPlugin())]) + + +def _wait_update_ends( + captured: list[tuple[str, str, bool, OperationStatus]], *wait_ids: str +) -> list[tuple[str, str, bool, OperationStatus]]: + """Return on_operation_end records for the given wait ids, in capture order.""" + wanted = set(wait_ids) + return [event for event in captured if event[1] in wanted] def test_virtual_child_preserves_replay_and_emits_update_for_inner_wait(): @@ -3335,57 +3373,41 @@ def test_virtual_child_preserves_replay_and_emits_update_for_inner_wait(): Otherwise the inner wait's stored history is bypassed and its update notification is dropped. Regression test for issue #889. """ - state = _replay_state({}) - parent = DurableContext( - state=state, - execution_context=ExecutionContext(durable_execution_arn="arn"), - replay_status=ReplayStatus.REPLAY, - ) - context_id = parent._peek_next_operation_id() # noqa: SLF001 - child_probe = parent.create_child_context(context_id, is_virtual=True) - wait_id = child_probe._peek_next_operation_id() # noqa: SLF001 - state._operations[wait_id] = _named_wait_op( # noqa: SLF001 - wait_id, "virtual-wait", parent_id=None - ) - state._updated_operation_ids.add(wait_id) # noqa: SLF001 + captured, plugin_executor = _capturing_end_plugin() + with plugin_invocation(plugin_executor): + state = _replay_state({}, plugin_executor=plugin_executor) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + context_id = parent._peek_next_operation_id() # noqa: SLF001 + child_probe = parent.create_child_context(context_id, is_virtual=True) + wait_id = child_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[wait_id] = _named_wait_op( # noqa: SLF001 + wait_id, "virtual-wait", parent_id=None + ) + state._updated_operation_ids.add(wait_id) # noqa: SLF001 - child_entered_replay: list[bool] = [] - emitted_updates: list[str] = _install_update_recorder(state) + child_entered_replay: list[bool] = [] - def body(child: DurableContext) -> str: - child_entered_replay.append(child.is_replaying()) - child.wait(Duration.from_seconds(5), name="virtual-wait") - return "ok" + def body(child: DurableContext) -> str: + child_entered_replay.append(child.is_replaying()) + child.wait(Duration.from_seconds(5), name="virtual-wait") + return "ok" - result = parent.run_in_child_context( - body, name="virtual-context", config=ChildConfig(is_virtual=True) - ) + result = parent.run_in_child_context( + body, name="virtual-context", config=ChildConfig(is_virtual=True) + ) - # The child inherits REPLAY and the updated wait's notification is delivered. + # The child inherits REPLAY and the plugin gets on_operation_end for the wait. assert child_entered_replay == [True] - assert emitted_updates == [wait_id] + assert _wait_update_ends(captured, wait_id) == [ + ("end", wait_id, False, OperationStatus.SUCCEEDED) + ] assert result == "ok" -def test_replay_aware_brand_new_non_virtual_op_flips_to_new(): - """A brand-new non-virtual operation is new work and flips to NEW. - - Control for the virtual case: without `is_virtual`, a missing checkpoint at - the operation id means the operation is brand-new, so the context flips to - NEW before the body runs. - """ - ctx = DurableContext( - state=_replay_state({}), - execution_context=ExecutionContext(durable_execution_arn="arn"), - replay_status=ReplayStatus.REPLAY, - ) - inside_status: list[bool] = [] - with ctx._replay_aware(is_virtual=False): # noqa: SLF001 - inside_status.append(ctx.is_replaying()) - - assert inside_status == [False] - - def test_replay_aware_brand_new_virtual_op_keeps_replay(): """A brand-new virtual operation keeps replay status. @@ -3441,44 +3463,178 @@ def test_nested_virtual_child_preserves_replay_and_emits_update_for_inner_wait() Each virtual layer has no checkpoint by design. The inner wait's history must survive both layers and its update notification delivered once. """ + captured, plugin_executor = _capturing_end_plugin() + with plugin_invocation(plugin_executor): + state = _replay_state({}, plugin_executor=plugin_executor) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + outer_id = parent._peek_next_operation_id() # noqa: SLF001 + outer_probe = parent.create_child_context(outer_id, is_virtual=True) + inner_id = outer_probe._peek_next_operation_id() # noqa: SLF001 + inner_probe = outer_probe.create_child_context(inner_id, is_virtual=True) + wait_id = inner_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[wait_id] = _named_wait_op( # noqa: SLF001 + wait_id, "nested-wait", parent_id=None + ) + state._updated_operation_ids.add(wait_id) # noqa: SLF001 + + inner_entered_replay: list[bool] = [] + + def inner_body(inner: DurableContext) -> str: + inner_entered_replay.append(inner.is_replaying()) + inner.wait(Duration.from_seconds(5), name="nested-wait") + return "ok" + + def outer_body(outer: DurableContext) -> str: + return outer.run_in_child_context( + inner_body, name="inner-context", config=ChildConfig(is_virtual=True) + ) + + result = parent.run_in_child_context( + outer_body, name="outer-context", config=ChildConfig(is_virtual=True) + ) + + assert inner_entered_replay == [True] + assert _wait_update_ends(captured, wait_id) == [ + ("end", wait_id, False, OperationStatus.SUCCEEDED) + ] + assert result == "ok" + + +def test_next_operation_exists_when_only_first_inner_checkpoint_is_present(): + """Look-ahead treats a virtual child's first inner checkpoint as existence. + + After a completed operation, `_next_operation_exists` must not require a + checkpoint at the next operation's own id. Virtual children never have one. + """ state = _replay_state({}) - parent = DurableContext( + ctx = DurableContext( state=state, execution_context=ExecutionContext(durable_execution_arn="arn"), replay_status=ReplayStatus.REPLAY, ) - outer_id = parent._peek_next_operation_id() # noqa: SLF001 - outer_probe = parent.create_child_context(outer_id, is_virtual=True) - inner_id = outer_probe._peek_next_operation_id() # noqa: SLF001 - inner_probe = outer_probe.create_child_context(inner_id, is_virtual=True) - wait_id = inner_probe._peek_next_operation_id() # noqa: SLF001 - state._operations[wait_id] = _named_wait_op( # noqa: SLF001 - wait_id, "nested-wait", parent_id=None + ctx._create_step_id() # noqa: SLF001 - previous completed op advances the counter + next_id = ctx._peek_next_operation_id() # noqa: SLF001 + child_probe = ctx.create_child_context(next_id, is_virtual=True) + first_inner_id = child_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[first_inner_id] = _named_wait_op( # noqa: SLF001 + first_inner_id, "inner-wait", parent_id=None ) - state._updated_operation_ids.add(wait_id) # noqa: SLF001 - inner_entered_replay: list[bool] = [] - emitted_updates: list[str] = _install_update_recorder(state) + assert ctx._next_operation_exists() is True # noqa: SLF001 - def inner_body(inner: DurableContext) -> str: - inner_entered_replay.append(inner.is_replaying()) - inner.wait(Duration.from_seconds(5), name="nested-wait") - return "ok" - def outer_body(outer: DurableContext) -> str: - return outer.run_in_child_context( - inner_body, name="inner-context", config=ChildConfig(is_virtual=True) +def test_virtual_child_after_replayed_step_preserves_replay_and_emits_update(): + """A replayed step before a virtual child must not flip the context to NEW. + + After the step, look-ahead finds no checkpoint at the virtual child's own + id. Existence of the child's first inner wait must keep replay active so + the wait's on_operation_end is delivered. + """ + captured, plugin_executor = _capturing_end_plugin() + with plugin_invocation(plugin_executor): + state = _replay_state({}, plugin_executor=plugin_executor) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + step_id = parent._create_step_id_for_logical_step(1) + context_id = parent._create_step_id_for_logical_step(2) + child_probe = parent.create_child_context(context_id, is_virtual=True) + wait_id = child_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[step_id] = _named_step_op( # noqa: SLF001 + step_id, "before-child", "done" ) + state._operations[wait_id] = _named_wait_op( # noqa: SLF001 + wait_id, "virtual-wait", parent_id=None + ) + state._updated_operation_ids.add(wait_id) # noqa: SLF001 - result = parent.run_in_child_context( - outer_body, name="outer-context", config=ChildConfig(is_virtual=True) - ) + child_entered_replay: list[bool] = [] - assert inner_entered_replay == [True] - assert emitted_updates == [wait_id] + def body(child: DurableContext) -> str: + child_entered_replay.append(child.is_replaying()) + child.wait(Duration.from_seconds(5), name="virtual-wait") + return "ok" + + step_result = parent.step( + lambda _step_context: "should-not-run", name="before-child" + ) + assert step_result == "done" + result = parent.run_in_child_context( + body, name="virtual-context", config=ChildConfig(is_virtual=True) + ) + + assert child_entered_replay == [True] + assert _wait_update_ends(captured, wait_id) == [ + ("end", wait_id, False, OperationStatus.SUCCEEDED) + ] assert result == "ok" +def test_two_virtual_children_in_a_row_preserve_replay_and_emit_update(): + """The second of two consecutive virtual children stays in replay. + + After the first virtual child, look-ahead finds no checkpoint at the second + child's own id. The second child's first inner wait must keep replay active. + """ + captured, plugin_executor = _capturing_end_plugin() + with plugin_invocation(plugin_executor): + state = _replay_state({}, plugin_executor=plugin_executor) + parent = DurableContext( + state=state, + execution_context=ExecutionContext(durable_execution_arn="arn"), + replay_status=ReplayStatus.REPLAY, + ) + first_id = parent._create_step_id_for_logical_step(1) + second_id = parent._create_step_id_for_logical_step(2) + first_probe = parent.create_child_context(first_id, is_virtual=True) + second_probe = parent.create_child_context(second_id, is_virtual=True) + first_wait_id = first_probe._peek_next_operation_id() # noqa: SLF001 + second_wait_id = second_probe._peek_next_operation_id() # noqa: SLF001 + state._operations[first_wait_id] = _named_wait_op( # noqa: SLF001 + first_wait_id, "first-wait", parent_id=None + ) + state._operations[second_wait_id] = _named_wait_op( # noqa: SLF001 + second_wait_id, "second-wait", parent_id=None + ) + state._updated_operation_ids.add(first_wait_id) # noqa: SLF001 + state._updated_operation_ids.add(second_wait_id) # noqa: SLF001 + + first_entered_replay: list[bool] = [] + second_entered_replay: list[bool] = [] + + def first_body(child: DurableContext) -> str: + first_entered_replay.append(child.is_replaying()) + child.wait(Duration.from_seconds(5), name="first-wait") + return "first" + + def second_body(child: DurableContext) -> str: + second_entered_replay.append(child.is_replaying()) + child.wait(Duration.from_seconds(5), name="second-wait") + return "second" + + first_result = parent.run_in_child_context( + first_body, name="first-virtual", config=ChildConfig(is_virtual=True) + ) + second_result = parent.run_in_child_context( + second_body, name="second-virtual", config=ChildConfig(is_virtual=True) + ) + + assert first_entered_replay == [True] + assert second_entered_replay == [True] + assert _wait_update_ends(captured, first_wait_id, second_wait_id) == [ + ("end", first_wait_id, False, OperationStatus.SUCCEEDED), + ("end", second_wait_id, False, OperationStatus.SUCCEEDED), + ] + assert first_result == "first" + assert second_result == "second" + + def test_replay_aware_emits_update_hook_for_operation_updated_since_last_invocation(): """Updated terminal operations emit operation_end, not replay start+end.""" captured: list[tuple[str, str, bool, OperationStatus]] = []