fix(workflow): record runs that finish without a result so resume replays them

docs/guides/workflow/retry_config/index.md

@@ -103,7 +103,7 @@ inner = Workflow(
 )
 ```
 
-A retried sub-workflow does not redo work it already finished. Children that produced an output or a state change are replayed from the session's events, so only the failed node and whatever follows it run again. A child that produced neither leaves nothing to replay and runs again on every attempt, so give a side-effecting node an output or a state change if repeating it would be harmful.
+A retried sub-workflow does not redo work it already finished. Finished children are replayed from the session's events, so only the failed node and whatever follows it run again. This includes a child that finished without an output or a state change: ADK records that it finished, so its side effects are not repeated.
 
 Setting `RetryConfig` on the node that can fail is still the narrower option, and it retries just that node rather than the whole sub-workflow.
 
@@ -124,7 +124,6 @@ test_retry_config = RetryConfig(
 
 - **The attempt counter does not survive an interrupt.** It is **not** persisted. If the workflow is interrupted, whether by a human-in-the-loop input downstream or by the application restarting, and is later resumed, a node that has to run again starts counting from 1 with its full budget of attempts back.
 - **Retries are local to the node.** They happen inside that node's execution and nowhere else. Once a node has used up its attempts it enters the `FAILED` state, and the workflow execution fails with it, or follows whatever error handling you have configured.
-- **A retried `Workflow` runs some finished children again.** A child that produced no output and no state change runs again, because the replay has no record that it completed. See [Retrying a nested workflow](#retrying-a-nested-workflow).
 - **A cancelled child can be replayed as finished.** A child that was cancelled right after it emitted an event without an output, such as a progress message, looks finished to the replay. The retry then hands back no output for it instead of running it again. A parallel worker item that is cancelled because another item failed is the usual case.
 
 ## Related samples

src/google/adk/events/event.py

@@ -75,6 +75,16 @@ class NodeInfo(BaseModel):
   carries the output value.
   """
 
+  completed: bool | None = None
+  """When True, the node run at ``path`` finished without emitting any other
+  event and without an output, route or transfer.
+
+  Such a run leaves no other trace in the session, so this event records that
+  it finished. On resume the run is replayed from history instead of running
+  again. For a ``wait_for_output`` node this means the run ended ``WAITING``;
+  agents and ``rerun_on_resume`` nodes, which rerun on resume, get no marker.
+  """
+
   @property
   def run_id(self) -> str:
     """The run ID of the node that generated the event."""

src/google/adk/workflow/_node_runner.py

@@ -142,6 +142,11 @@ def __init__(
     self._prior_output = prior_output
     self._prior_interrupt_ids = prior_interrupt_ids
 
+    # Whether the current attempt has put an event of its own in the session.
+    self._emitted_in_attempt = False
+    # Whether the current attempt stopped on a child that is still waiting.
+    self._interrupted_in_attempt = False
+
   @property
   def run_id(self) -> str:
     """The run ID assigned to this node run."""
@@ -163,6 +168,8 @@ async def run(
       ctx = self._create_child_context(
           resume_inputs, attempt_count=attempt_count
       )
+      self._emitted_in_attempt = False
+      self._interrupted_in_attempt = False
       # A retry after a dynamic child failure rewinds to this point, so it
       # replays only the children this attempt completed.
       counters_at_start = self._snapshot_dynamic_children(ctx)
@@ -182,6 +189,7 @@ async def run(
               ctx.error, attempt_count
           ):
             await self._flush_output_and_deltas(ctx)
+            await self._record_completion_without_result(ctx)
             logger.debug("node %s end.", ctx.node_path)
             return ctx
       except asyncio.CancelledError:
@@ -425,6 +433,7 @@ async def _execute_node(
       # A dynamic child interrupted via ctx.run_node().
       # Tool calls have their own Context, so copy their interrupt IDs too.
       ctx._interrupt_ids.update(exc.interrupt_ids)
+      self._interrupted_in_attempt = True
 
   async def _run_node_loop(self, ctx: Context, node_input: Any) -> None:
     """Iterate node.run(), track events in context, and enqueue them."""
@@ -503,6 +512,7 @@ async def _enqueue_event(self, event: Event, ctx: Context) -> None:
     if not event.partial:
       self._flush_deltas(event, ctx)
     await ctx._invocation_context._enqueue_event(event)
+    self._emitted_in_attempt = True
 
     if event.output is not None:
       ctx._output_emitted = True
@@ -537,11 +547,53 @@ async def _flush_output_and_deltas(self, ctx: Context) -> None:
 
     self._enrich_event(event, ctx)
     await ctx._invocation_context._enqueue_event(event)
+    self._emitted_in_attempt = True
     if has_deferred_output:
       ctx._output_emitted = True
     if has_unflushed_route:
       ctx._route_emitted = True
 
+  async def _record_completion_without_result(self, ctx: Context) -> None:
+    """Emit a completion marker for a run that finished leaving no trace.
+
+    A run that emitted no event and has no output, route or transfer leaves
+    nothing in the session that shows it ran, so resume would run it (and its
+    side effects) again. The marker lets the scan replay it instead.
+    """
+    from ..agents.base_agent import BaseAgent
+    from ..events.event import Event
+    from ._workflow import Workflow
+
+    if (
+        self._emitted_in_attempt
+        or self._interrupted_in_attempt
+        # A nested workflow always reruns and replays its own children.
+        or isinstance(self._node, Workflow)
+        or ctx.error is not None
+        or ctx.interrupt_ids
+        or ctx.output is not None
+        or ctx._output_delegated
+        or ctx.route is not None
+        or ctx.actions.transfer_to_agent is not None
+        # A waiting agent (task or chat mode) or rerun_on_resume node reruns on
+        # resume to take new input, so its run must not be recorded as ended.
+        or (
+            self._node.wait_for_output
+            and (
+                self._node.rerun_on_resume or isinstance(self._node, BaseAgent)
+            )
+        )
+        # Tests wrap Context around invocation-context stubs that lack
+        # is_aborted or return a truthy Mock; neither means aborted.
+        or getattr(ctx._invocation_context, "is_aborted", False) is True
+    ):
+      return
+
+    event = Event()
+    event.node_info.completed = True
+    self._enrich_event(event, ctx)
+    await ctx._invocation_context._enqueue_event(event)
+
   def _flush_deltas(self, event: Event, ctx: Context) -> None:
     """Move pending state/artifact deltas from ctx onto the event."""
     from ..events.event_actions import EventActions

src/google/adk/workflow/utils/_rehydration_utils.py

@@ -94,6 +94,14 @@ class _ChildScanState:
   finished_after_resume: bool = False
   """Whether the child emitted a direct completion event since its last answer (or start)."""
 
+  completed: bool = False
+  """Whether the child's run recorded that it finished without a result.
+
+  Set from a completion marker (see ``_is_completion_marker``). Unlike
+  ``finished_after_resume``, which any direct non-interrupt event sets, this
+  is only set once the run has actually ended.
+  """
+
 
 def _wrap_response(value: Any) -> dict[str, Any]:
   """Wraps a value into a dict suitable for FunctionResponse.response.
@@ -551,6 +559,11 @@ def get_owner_key(event_path_builder: _NodePathBuilder) -> str | None:
         if schema_json:
           schemas_by_id[interrupt_id] = schema_json
 
+    if interrupt_ids_to_process:
+      child.completed = False
+    elif is_direct and _is_completion_marker(event):
+      child.completed = True
+
     if is_direct:
       child.finished_after_resume = (
           not interrupt_ids_to_process
@@ -563,6 +576,11 @@ def get_owner_key(event_path_builder: _NodePathBuilder) -> str | None:
   return scan_states
 
 
+def _is_completion_marker(event: Event) -> bool:
+  """Returns whether ``event`` records a run that finished without a result."""
+  return bool(event.node_info and getattr(event.node_info, 'completed', None))
+
+
 def _event_has_result(event: Event) -> bool:
   """Returns whether ``event`` carries a node output, route, or transfer."""
   return (

src/google/adk/workflow/utils/_replay_interceptor.py

@@ -191,9 +191,14 @@ def check_interception(
         should_run = True
         resume_inputs = recovered.resolved_responses
     else:
-      # Allow fresh execution for crashed/timeout dynamic nodes;
-      # static nodes with no outcome (e.g. return None) should be fast-forwarded.
-      should_run = current_run is not None and not current_run.is_static
+      # Allow fresh execution for crashed/timeout dynamic nodes, but not for
+      # dynamic nodes that recorded finishing without a result; static nodes
+      # with no outcome (e.g. return None) should be fast-forwarded.
+      should_run = (
+          current_run is not None
+          and not current_run.is_static
+          and not recovered.completed
+      )
 
   return InterceptionResult(
       should_run=should_run,

tests/unittests/workflow/test_workflow_dynamic_nodes.py

@@ -773,9 +773,6 @@ async def parent(*, ctx, node_input):
 
 
 @pytest.mark.asyncio
-@pytest.mark.xfail(
-    reason='No completion marker event for None output (event design flaw)'
-)
 async def test_dynamic_node_none_output_not_rerun():
   """Dynamic child that completed with None output is not re-run.
 
@@ -786,9 +783,6 @@ async def test_dynamic_node_none_output_not_rerun():
   Assert:
     - On Run 3, A should NOT re-run (it already completed).
     - A should return None (cached), B resumes.
-    - Currently fails because A's None completion leaves no
-      trace in session events — the lazy scan thinks A still
-      needs to re-run.
   """
   run_count_a = [0]
 

tests/unittests/workflow/test_workflow_failures.py

@@ -1294,6 +1294,38 @@ def flaky_node(ctx: Context, node_input: Any):
   assert completed_attempts == 1
 
 
+@pytest.mark.asyncio
+async def test_retry_config_on_nested_workflow_replays_child_without_output():
+  """A retried sub-workflow does not redo a child that finished with no output."""
+  side_effect_attempts = 0
+  flaky_attempts = 0
+
+  @node()
+  def side_effect_node(ctx: Context):
+    nonlocal side_effect_attempts
+    side_effect_attempts += 1
+
+  @node()
+  def flaky_node(ctx: Context, node_input: Any):
+    nonlocal flaky_attempts
+    flaky_attempts += 1
+    if flaky_attempts < 3:
+      raise CustomError('Node failed')
+    yield 'recovered'
+
+  inner_wf = Workflow(
+      name='inner_wf',
+      edges=[(START, side_effect_node, flaky_node)],
+      retry_config=RetryConfig(max_attempts=3, initial_delay=0.0, jitter=0.0),
+  )
+  outer_wf = Workflow(name='outer_wf', edges=[(START, inner_wf)])
+
+  await _run_workflow(outer_wf)
+
+  assert flaky_attempts == 3
+  assert side_effect_attempts == 1
+
+
 @pytest.mark.asyncio
 async def test_retry_config_on_outer_workflow_retries_nested_failure():
   """A failure inside a sub-workflow is a failure of the outer workflow too."""