@@ -76,7 +76,10 @@ class _ChildScanState:
   """The agent the child asked to transfer to, or None if it asked for none."""
 
   interrupt_ids: set[str] = field(default_factory=set)
-  """Every interrupt the child raised."""
+  """Every interrupt the child (or its descendants) raised."""
+
+  direct_interrupt_ids: set[str] = field(default_factory=set)
+  """Interrupts raised directly by the child node itself."""
 
   resolved_ids: set[str] = field(default_factory=set)
   """The subset of ``interrupt_ids`` a user response has come back for."""
@@ -425,6 +428,8 @@ def get_owner_key(event_path_builder: _NodePathBuilder) -> str | None:
     if interrupt_ids_to_process:
       for interrupt_id in interrupt_ids_to_process:
         child.interrupt_ids.add(interrupt_id)
+        if is_direct:
+          child.direct_interrupt_ids.add(interrupt_id)
         interrupt_owner[interrupt_id] = owner_key
 
         schema_json = _extract_schema_from_event(event, interrupt_id)
 
@@ -90,6 +90,7 @@ def check_interception(
     )
 
   unresolved = recovered.interrupt_ids - recovered.resolved_ids
+  direct_unresolved = recovered.direct_interrupt_ids - recovered.resolved_ids
 
   should_run = False
   output = None
@@ -99,9 +100,15 @@ def check_interception(
 
   if unresolved:
     # Case 2: Cross-turn unresolved interrupts remain.
-    # Rerun natively with resolved inputs if the node supports rerun and some
-    # progress was made; otherwise remain waiting and bubble unresolved interrupts.
-    if node.rerun_on_resume and recovered.resolved_ids:
+    # A node with unresolved direct interrupts of its own must wait until all of
+    # them are answered before rerunning. A parent node whose unresolved
+    # interrupts belong only to descendant dynamic nodes reruns when progress was
+    # made so the unblocked child can advance.
+    if (
+        not direct_unresolved
+        and node.rerun_on_resume
+        and recovered.resolved_ids
+    ):
       should_run = True
       resume_inputs = recovered.resolved_responses
     else:
 
@@ -1053,23 +1053,7 @@ async def collect_events(agen):
   assert node_b.received_inputs == ['final_output']
 
 
-@pytest.mark.parametrize(
-    'resumable',
-    [
-        False,
-        pytest.param(
-            True,
-            marks=pytest.mark.xfail(
-                reason=(
-                    'A rerun_on_resume node is re-run as soon as one interrupt'
-                    ' resolves instead of waiting for all pending interrupts;'
-                    ' fixing this needs a change to the shared replay'
-                    ' interception logic.'
-                )
-            ),
-        ),
-    ],
-)
+@pytest.mark.parametrize('resumable', [False, True])
 @pytest.mark.asyncio
 async def test_rerun_on_resume_waits_for_all_interrupts(
     request: pytest.FixtureRequest,
@@ -1182,14 +1166,10 @@ async def _run_impl(
     )
     assert final_state2['nodes']['NodeA']['interrupts'] == ['req2']
 
-  # The node should NOT have produced any RequestInput or data output in resumable mode.
-  # In non-resumable mode, it re-yields the pending interrupt 'req2'.
+  # The node should NOT have produced any RequestInput or data output since it
+  # did not rerun while req2 was still pending.
   req_events2 = workflow_testing_utils.get_request_input_events(events2)
-  if resumable:
-    assert len(req_events2) == 0
-  else:
-    assert len(req_events2) == 1
-    assert get_request_input_interrupt_ids(req_events2[0]) == ['req2']
+  assert len(req_events2) == 0
 
   # Run 3: provide req2 — now all interrupts resolved, node should rerun.
   events3 = await runner.run_async(
 
@@ -96,12 +96,17 @@ def test_cross_turn_unresolved_interrupts_no_rerun():
   assert result.interrupts == {'fc-2'}
 
 
-def test_cross_turn_unresolved_interrupts_rerun():
-  """Cross-turn unresolved interrupts with rerun resolves progress and reruns."""
-  # Given unresolved interrupts and node with rerun_on_resume
+def test_cross_turn_unresolved_descendant_interrupts_rerun():
+  """A rerun_on_resume node reruns once its own interrupts are all answered.
+
+  The remaining unresolved interrupt belongs to a descendant dynamic node, so
+  the parent reruns to let that child advance.
+  """
+  # Given the node's only direct interrupt resolved and a descendant's pending
   recovered = _ChildScanState(
       run_id='1',
       interrupt_ids={'fc-1', 'fc-2'},
+      direct_interrupt_ids={'fc-1'},
       resolved_ids={'fc-1'},
       resolved_responses={'fc-1': 'ans'},
   )
@@ -118,6 +123,30 @@ def test_cross_turn_unresolved_interrupts_rerun():
   assert result.resume_inputs == {'fc-1': 'ans'}
 
 
+def test_cross_turn_unresolved_direct_interrupts_wait():
+  """A rerun_on_resume node waits while any of its own interrupts is pending."""
+  # Given two direct interrupts of which only one has been answered
+  recovered = _ChildScanState(
+      run_id='1',
+      interrupt_ids={'fc-1', 'fc-2'},
+      direct_interrupt_ids={'fc-1', 'fc-2'},
+      resolved_ids={'fc-1'},
+      resolved_responses={'fc-1': 'ans'},
+  )
+  node = BaseNode(name='node', rerun_on_resume=True)
+
+  # When checked
+  result = check_interception(
+      node=node,
+      recovered=recovered,
+  )
+
+  # Then it keeps waiting on the unanswered one instead of rerunning
+  assert not result.should_run
+  assert result.interrupts == {'fc-2'}
+  assert result.resume_inputs is None
+
+
 def test_cross_turn_completed():
   """Cross-turn completed run fast-forwards output and route."""
   # Given a completed run from history