diff --git a/docs/guides/workflow/retry_config/index.md b/docs/guides/workflow/retry_config/index.md
index 5bab66ca9b0..17095dee39f 100644
--- a/docs/guides/workflow/retry_config/index.md
+++ b/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
diff --git a/src/google/adk/events/event.py b/src/google/adk/events/event.py
index 18d089afb87..1e75b1e94ac 100644
--- a/src/google/adk/events/event.py
+++ b/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."""
diff --git a/src/google/adk/workflow/_node_runner.py b/src/google/adk/workflow/_node_runner.py
index e0e89023aa0..49b999cdedb 100644
--- a/src/google/adk/workflow/_node_runner.py
+++ b/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
diff --git a/src/google/adk/workflow/utils/_rehydration_utils.py b/src/google/adk/workflow/utils/_rehydration_utils.py
index e12d4314a06..21912b8621d 100644
--- a/src/google/adk/workflow/utils/_rehydration_utils.py
+++ b/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 (
diff --git a/src/google/adk/workflow/utils/_replay_interceptor.py b/src/google/adk/workflow/utils/_replay_interceptor.py
index 1d45fa90897..d4ed9cc5f62 100644
--- a/src/google/adk/workflow/utils/_replay_interceptor.py
+++ b/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,
diff --git a/tests/unittests/workflow/test_workflow_dynamic_nodes.py b/tests/unittests/workflow/test_workflow_dynamic_nodes.py
index 805e7dc0ae0..4508d3fa07a 100644
--- a/tests/unittests/workflow/test_workflow_dynamic_nodes.py
+++ b/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]
 
diff --git a/tests/unittests/workflow/test_workflow_failures.py b/tests/unittests/workflow/test_workflow_failures.py
index b361194a44a..44d9e5360c4 100644
--- a/tests/unittests/workflow/test_workflow_failures.py
+++ b/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."""
diff --git a/tests/unittests/workflow/test_workflow_hitl.py b/tests/unittests/workflow/test_workflow_hitl.py
index fa73af16597..3be870c8680 100644
--- a/tests/unittests/workflow/test_workflow_hitl.py
+++ b/tests/unittests/workflow/test_workflow_hitl.py
@@ -3081,3 +3081,212 @@ def collect(node_input: dict[str, Any]) -> str:
   assert b_runs == 1
   outputs3 = [e.output for e in events3 if e.output is not None]
   assert outputs3[-1] == 'approved:True|from_b'
+
+
+async def _run_and_answer_first_request(
+    runner: testing_utils.InMemoryRunner,
+) -> list[Event]:
+  """Runs the workflow once, then answers the request it paused on."""
+  events1 = await runner.run_async('start')
+  [interrupt_id] = [
+      interrupt_id
+      for event in workflow_testing_utils.get_request_input_events(events1)
+      for interrupt_id in extract_event_interrupt_ids(event)
+  ]
+  return await runner.run_async(
+      new_message=testing_utils.UserContent(
+          create_request_input_response(interrupt_id, {'ok': True})
+      ),
+      invocation_id=events1[0].invocation_id,
+  )
+
+
+def _app(name: str, root: Workflow, resumable: bool) -> App:
+  return App(
+      name=name,
+      root_agent=root,
+      resumability_config=(
+          ResumabilityConfig(is_resumable=True) if resumable else None
+      ),
+  )
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_node_returning_none_is_not_rerun_on_resume(resumable: bool):
+  """A node that finished without output is replayed, not run again."""
+  emails_sent = []
+
+  def send_email():
+    emails_sent.append('email')
+
+  def ask():
+    return RequestInput(message='Approve?')
+
+  def finish(node_input: Any):
+    return node_input
+
+  wf = Workflow(name='wf', edges=[(START, send_email, ask, finish)])
+  runner = testing_utils.InMemoryRunner(
+      app=_app('none_output_seq', wf, resumable)
+  )
+
+  events2 = await _run_and_answer_first_request(runner)
+
+  assert emails_sent == ['email']
+  assert [
+      e.output for e in events2 if e.node_info.name == 'finish' and e.output
+  ] == [{'ok': True}]
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_parallel_node_returning_none_is_not_rerun_on_resume(
+    resumable: bool,
+):
+  """A parallel branch that finished without output is not run again."""
+  emails_sent = []
+
+  def send_email():
+    emails_sent.append('email')
+
+  def ask():
+    return RequestInput(message='Approve?')
+
+  wf = Workflow(name='wf', edges=[(START, send_email), (START, ask)])
+  runner = testing_utils.InMemoryRunner(
+      app=_app('none_output_parallel', wf, resumable)
+  )
+
+  await _run_and_answer_first_request(runner)
+
+  assert emails_sent == ['email']
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_dynamic_child_returning_none_is_not_rerun_on_resume(
+    resumable: bool,
+):
+  """A ctx.run_node child that finished without output is not run again."""
+  emails_sent = []
+
+  @node
+  def send_email():
+    emails_sent.append('email')
+
+  @node
+  def ask():
+    return RequestInput(message='Approve?')
+
+  @node(rerun_on_resume=True)
+  async def parent(*, ctx: Context, node_input: Any):
+    await ctx.run_node(send_email)
+    answer = await ctx.run_node(ask)
+    yield answer
+
+  wf = Workflow(name='wf', edges=[(START, parent)])
+  runner = testing_utils.InMemoryRunner(
+      app=_app('none_output_dynamic', wf, resumable)
+  )
+
+  events2 = await _run_and_answer_first_request(runner)
+
+  assert emails_sent == ['email']
+  assert {'ok': True} in [e.output for e in events2 if e.output is not None]
+
+
+class _QuietWaitingNode(BaseNode):
+  """A wait_for_output node that never emits, so it always ends WAITING."""
+
+  wait_for_output: bool = True
+  calls: list[Any] = Field(default_factory=list)
+
+  @override
+  async def _run_impl(
+      self, *, ctx: Context, node_input: Any
+  ) -> AsyncGenerator[Any, None]:
+    self.calls.append(node_input)
+    return
+    yield  # pylint: disable=unreachable
+
+
+@pytest.mark.parametrize(
+    ('rerun_on_resume', 'expect_marker'), [(False, True), (True, False)]
+)
+@pytest.mark.asyncio
+async def test_waiting_node_run_is_recorded_unless_it_reruns(
+    rerun_on_resume: bool, expect_marker: bool
+):
+  """A waiting run gets the marker, except on a node that reruns on resume."""
+
+  def ask():
+    return RequestInput(message='Approve?')
+
+  waiting = _QuietWaitingNode(name='waiting', rerun_on_resume=rerun_on_resume)
+  wf = Workflow(name='wf', edges=[(START, waiting), (START, ask)])
+  runner = testing_utils.InMemoryRunner(app=_app('waiting_marker', wf, False))
+
+  await runner.run_async('start')
+
+  markers = [
+      e
+      for e in runner.session.events
+      if e.node_info.name == 'waiting' and e.node_info.completed
+  ]
+  assert bool(markers) == expect_marker
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_waiting_dynamic_child_keeps_parent_waiting_on_resume(
+    resumable: bool,
+):
+  """A ctx.run_node child that ended WAITING is not handed back as a result.
+
+  With raise_on_wait=True the parent stays WAITING in the first turn, and it
+  must still be WAITING after another branch's request is answered, rather
+  than continuing as if the child had finished with None.
+  """
+  child = _QuietWaitingNode(name='child')
+  after_child = []
+
+  @node(rerun_on_resume=True)
+  async def parent(*, ctx: Context, node_input: Any):
+    await ctx.run_node(child, raise_on_wait=True)
+    after_child.append('ran')
+    yield 'parent done'
+
+  def ask():
+    return RequestInput(message='Approve?')
+
+  wf = Workflow(name='wf', edges=[(START, parent), (START, ask)])
+  runner = testing_utils.InMemoryRunner(
+      app=_app('waiting_dynamic_child', wf, resumable)
+  )
+
+  events2 = await _run_and_answer_first_request(runner)
+
+  assert after_child == []
+  assert 'parent done' not in [e.output for e in events2]
+  # The child's waiting run is replayed, not run again.
+  assert child.calls == [None]
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_quiet_waiting_node_is_not_rerun_on_resume(resumable: bool):
+  """A waiting run that emitted no event is replayed, not run again."""
+  waiting = _QuietWaitingNode(name='waiting')
+
+  def ask():
+    return RequestInput(message='Approve?')
+
+  wf = Workflow(name='wf', edges=[(START, waiting), (START, ask)])
+  runner = testing_utils.InMemoryRunner(
+      app=_app('quiet_waiting_not_rerun', wf, resumable)
+  )
+
+  await _run_and_answer_first_request(runner)
+
+  assert len(waiting.calls) == 1
diff --git a/tests/unittests/workflow/utils/test_replay_interceptor.py b/tests/unittests/workflow/utils/test_replay_interceptor.py
index 3919c2aa410..d5f6f3ef2ed 100644
--- a/tests/unittests/workflow/utils/test_replay_interceptor.py
+++ b/tests/unittests/workflow/utils/test_replay_interceptor.py
@@ -396,6 +396,26 @@ def test_cross_turn_no_outcome_waiting_agent_or_rerun_node_reruns(
   assert result.should_run
 
 
+def test_cross_turn_dynamic_run_recorded_completed_fast_forwards():
+  """Case 6 replays a dynamic run whose completion marker was recorded."""
+  recovered = _ChildScanState(run_id='1', completed=True)
+  node = BaseNode(name='node', rerun_on_resume=False, wait_for_output=False)
+  dynamic_run = DynamicNodeRun(
+      state=NodeState(run_id='1'),
+      recovered_state=recovered,
+      is_static=False,
+  )
+
+  result = check_interception(
+      node=node,
+      recovered=recovered,
+      current_run=dynamic_run,
+  )
+
+  assert not result.should_run
+  assert result.output is None
+
+
 # --- create_mock_context ---