@@ -112,6 +112,7 @@ validation_join = JoinNode(
 
 - **The output is always a dictionary.** A `JoinNode` emits predecessor names mapped to their outputs and nothing else. When you need a different shape, transform it in the node downstream of the join.
 - **A predecessor on an untaken conditional path stops the join firing.** If a `JoinNode` declares a predecessor that sits on a conditional path, and that path is not taken, the join never triggers. The run does not hang waiting for it. The workflow finishes as soon as the branches that did run are done, and the join and everything after it is skipped silently, with no error anywhere. Every static predecessor a join declares in the graph has to execute before the join fires.
+- **In a loop, a `JoinNode` waits for every predecessor to arrive again in the current iteration.** If the loop restarts before the join fires, the arrivals already recorded carry over and the next firing can combine outputs from different iterations.
 
 ## Related samples
 
 
@@ -99,6 +99,14 @@ class _LoopState(DynamicNodeState):
   node_branches: dict[str, str] = field(default_factory=dict)
   """Cached static node branches."""
 
+  join_arrivals: dict[str, set[str]] = field(default_factory=dict)
+  """Predecessors that arrived since each wait-for-all node last fired.
+
+  Keyed by the wait-for-all node's name. Cleared when that node fires, so in
+  a loop it waits for every predecessor's next run instead of counting runs
+  left over from an earlier iteration.
+  """
+
   pending_tasks: dict[str, asyncio.Task[Context]] = field(default_factory=dict)
   """Running static node tasks."""
 
@@ -798,6 +806,8 @@ async def _handle_completion(
       node_state.resume_inputs.clear()
     if child_ctx.output is not None:
       loop_state.node_outputs[node_name] = child_ctx.output
+    else:
+      loop_state.node_outputs.pop(node_name, None)
     loop_state.node_branches[node_name] = (
         child_ctx._invocation_context.branch or ""
     )
@@ -824,22 +834,26 @@ def _buffer_barrier_trigger(
   ) -> None:
     """Buffer a trigger for target_name once all predecessors have completed.
 
-    No-op while any predecessor is still outstanding.
+    No-op while any predecessor is still outstanding. Only runs completed
+    since target_name last fired count, so each firing consumes them.
     """
     graph = self._require_graph()
     predecessors = {
         e.from_node.name for e in graph.edges if e.to_node.name == target_name
     }
+    arrivals = loop_state.join_arrivals.get(target_name, set())
     # START never executes, so it is satisfied as soon as the workflow begins.
     if not all(
         p == START.name
         or (
-            loop_state.nodes.get(p)
+            p in arrivals
+            and loop_state.nodes.get(p)
             and loop_state.nodes[p].status == NodeStatus.COMPLETED
         )
         for p in predecessors
     ):
       return
+    loop_state.join_arrivals.pop(target_name, None)
 
     outputs = {p: loop_state.node_outputs.get(p) for p in predecessors}
     branches = [loop_state.node_branches.get(p, "") for p in predecessors]
@@ -870,6 +884,7 @@ def _buffer_downstream_triggers(
       target_node = self._get_static_node_by_name(target_name)
 
       if target_node._requires_all_predecessors:
+        loop_state.join_arrivals.setdefault(target_name, set()).add(node_name)
         self._buffer_barrier_trigger(loop_state, target_name)
       else:
         # Normal node logic
 
@@ -14,13 +14,20 @@
 
 """Testings for the JoinNode."""
 
+import asyncio
+
 from google.adk import workflow
 from google.adk.apps import app
+from google.adk.apps.app import ResumabilityConfig
+from google.adk.events._interrupts import extract_event_interrupt_ids
+from google.adk.events.event import Event
+from google.adk.events.request_input import RequestInput
 from google.adk.workflow import _base_node as base_node
 from google.adk.workflow import _graph as workflow_graph
 from google.adk.workflow import _join_node as join_node
 from google.adk.workflow import START
 from google.adk.workflow._workflow import Workflow
+from google.adk.workflow.utils import _workflow_hitl_utils as hitl_utils
 from pydantic import BaseModel
 import pytest
 
@@ -348,3 +355,213 @@ async def test_join_node_computes_common_branch_prefix(
   assert len(capture_events) > 0
   for e in capture_events:
     assert e.branch is None
+
+
+@pytest.mark.asyncio
+async def test_join_node_in_loop_waits_for_each_iteration(
+    request: pytest.FixtureRequest,
+) -> None:
+  """Tests JoinNode in a loop waits for every predecessor's new run."""
+  join_inputs: list[dict[str, str]] = []
+
+  def node_a() -> str:
+    return 'A2' if join_inputs else 'A1'
+
+  def node_b() -> str:
+    return 'B2' if join_inputs else 'B1'
+
+  def node_check(node_input: dict[str, str]) -> Event | None:
+    join_inputs.append(node_input)
+    return Event(route='again') if len(join_inputs) == 1 else None
+
+  join = join_node.JoinNode(name='join')
+  agent = Workflow(
+      name='wf',
+      edges=[
+          (START, node_a, node_b),
+          ((node_a, node_b), join, node_check),
+          (node_check, {'again': node_a}),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=app.App(name=request.function.__name__, root_agent=agent)
+  )
+
+  await runner.run_async('start')
+
+  assert join_inputs == [
+      {'node_a': 'A1', 'node_b': 'B1'},
+      {'node_a': 'A2', 'node_b': 'B2'},
+  ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('is_resumable', [False, True])
+async def test_join_node_in_loop_waits_for_predecessor_paused_for_input(
+    request: pytest.FixtureRequest, is_resumable: bool
+) -> None:
+  """Tests JoinNode in a loop waits for a predecessor that asks the user."""
+  join_inputs: list[dict[str, str]] = []
+
+  def node_a() -> str:
+    return 'A2' if join_inputs else 'A1'
+
+  def node_b() -> RequestInput | str:
+    return RequestInput(message='Continue?') if join_inputs else 'B1'
+
+  def node_check(node_input: dict[str, str]) -> Event | None:
+    join_inputs.append(node_input)
+    return Event(route='again') if len(join_inputs) == 1 else None
+
+  join = join_node.JoinNode(name='join')
+  agent = Workflow(
+      name='wf',
+      edges=[
+          (START, node_a, node_b),
+          ((node_a, node_b), join, node_check),
+          (node_check, {'again': node_a}),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=app.App(
+          name=request.function.__name__,
+          root_agent=agent,
+          resumability_config=ResumabilityConfig(is_resumable=is_resumable),
+      )
+  )
+
+  events = await runner.run_async('start')
+
+  assert join_inputs == [{'node_a': 'A1', 'node_b': 'B1'}]
+
+  request_events = workflow_testing_utils.get_request_input_events(events)
+  [interrupt_id] = extract_event_interrupt_ids(request_events[0])
+  await runner.run_async(
+      new_message=hitl_utils.create_request_input_response(
+          interrupt_id, {'result': 'B2'}
+      ),
+      invocation_id=events[0].invocation_id,
+  )
+
+  assert join_inputs == [
+      {'node_a': 'A1', 'node_b': 'B1'},
+      {'node_a': 'A2', 'node_b': 'B2'},
+  ]
+
+
+@pytest.mark.asyncio
+async def test_join_node_in_loop_clears_prior_output_when_node_returns_none(
+    request: pytest.FixtureRequest,
+) -> None:
+  """Tests JoinNode receives None when a rerun predecessor returns None."""
+  join_inputs: list[dict[str, str | None]] = []
+
+  def node_a() -> str | None:
+    return None if join_inputs else 'A1'
+
+  def node_b() -> str:
+    return 'B2' if join_inputs else 'B1'
+
+  def node_check(node_input: dict[str, str | None]) -> Event | None:
+    join_inputs.append(node_input)
+    return Event(route='again') if len(join_inputs) == 1 else None
+
+  join = join_node.JoinNode(name='join')
+  agent = Workflow(
+      name='wf',
+      edges=[
+          (START, node_a, node_b),
+          ((node_a, node_b), join, node_check),
+          (node_check, {'again': node_a}),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=app.App(name=request.function.__name__, root_agent=agent)
+  )
+
+  await runner.run_async('start')
+
+  assert join_inputs == [
+      {'node_a': 'A1', 'node_b': 'B1'},
+      {'node_a': None, 'node_b': 'B2'},
+  ]
+
+
+@pytest.mark.asyncio
+async def test_join_node_does_not_fire_when_predecessor_routes_away(
+    request: pytest.FixtureRequest,
+) -> None:
+  """Tests JoinNode does not fire when a predecessor routes away from it."""
+
+  def node_a() -> Event:
+    return Event(output='A1', route='away')
+
+  def node_b() -> str:
+    return 'B1'
+
+  join = join_node.JoinNode(name='join')
+  other_node = workflow_testing_utils.InputCapturingNode(name='other_node')
+  capture = workflow_testing_utils.InputCapturingNode(name='capture')
+  agent = Workflow(
+      name='wf',
+      edges=[
+          (START, (node_a, node_b)),
+          (node_a, {'to_join': join, 'away': other_node}),
+          (node_b, join),
+          (join, capture),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=app.App(name=request.function.__name__, root_agent=agent)
+  )
+
+  await runner.run_async('start')
+
+  assert other_node.received_inputs == ['A1']
+  assert capture.received_inputs == []
+
+
+@pytest.mark.asyncio
+async def test_join_node_waits_for_retriggered_in_flight_predecessor(
+    request: pytest.FixtureRequest,
+) -> None:
+  """Tests JoinNode waits when an arrived predecessor has a run in flight."""
+  a_runs = 0
+  b_runs = 0
+
+  async def node_a() -> str:
+    nonlocal a_runs
+    a_runs += 1
+    if a_runs == 2:
+      for _ in range(5):
+        await asyncio.sleep(0)
+    return f'A{a_runs}'
+
+  def node_b() -> Event:
+    nonlocal b_runs
+    b_runs += 1
+    route = 'retry' if b_runs == 1 else 'to_join'
+    return Event(output=f'B{b_runs}', route=route)
+
+  def restart() -> None:
+    return None
+
+  join = join_node.JoinNode(name='join')
+  capture = workflow_testing_utils.InputCapturingNode(name='capture')
+  agent = Workflow(
+      name='wf',
+      edges=[
+          (START, (node_a, node_b)),
+          (node_a, join),
+          (node_b, {'to_join': join, 'retry': restart}),
+          (restart, (node_a, node_b)),
+          (join, capture),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=app.App(name=request.function.__name__, root_agent=agent)
+  )
+
+  await runner.run_async('start')
+
+  assert capture.received_inputs == [{'node_a': 'A2', 'node_b': 'B2'}]