@@ -81,7 +81,8 @@ class NodeInfo(BaseModel):
 
   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.
+  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
 
@@ -560,6 +560,7 @@ async def _record_completion_without_result(self, ctx: Context) -> None:
     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
 
@@ -574,9 +575,14 @@ async def _record_completion_without_result(self, ctx: Context) -> None:
         or ctx._output_delegated
         or ctx.route is not None
         or ctx.actions.transfer_to_agent is not None
-        # Finishing without output means a wait_for_output node is still
-        # waiting for more input, not that it is done.
-        or self._node.wait_for_output
+        # 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
 
@@ -3194,3 +3194,99 @@ async def parent(*, ctx: Context, node_input: Any):
 
   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