JoinNode in a loop fires early with the previous iteration's predecessor outputs, then fires again #7498

🔴 Required Information

Describe the Bug:
A JoinNode inside a loop fires before all of its predecessors have re-run, using outputs left over from the previous iteration, and then fires again when the rest arrive.

Workflow._buffer_barrier_trigger (src/google/adk/workflow/_workflow.py) treats the join as ready when every predecessor's NodeState.status == COMPLETED. That status is never consumed when the join fires, so on the next iteration a predecessor that has not re-run yet still counts as done. As soon as the first predecessor of iteration N completes, the join fires with the other predecessors' iteration N-1 outputs. Everything downstream of the join then runs an extra time per iteration.

This contradicts the documented contract (.agents/skills/adk-agent-builder/references/parallel-and-fanout.md): "A JoinNode waits for every predecessor, then fires once with all their outputs".

It is also visible with human-in-the-loop: if the slower branch pauses for RequestInput on iteration 2, the join fires with its stale iteration-1 output while it is still waiting for the user.

Steps to Reproduce:

1. pip install google-adk==2.11.0
2. Run the snippet below. One branch (X -> B) is a hop longer than the other (A), so on the second iteration A completes first.

Expected Behavior:
One join firing per iteration, each with that iteration's outputs:

{'a': 'A1', 'b': 'B1'}
{'a': 'A2', 'b': 'B2'}

Observed Behavior:
The join fires twice in iteration 2, the first time with B's stale iteration-1 output:

{'a': 'A1', 'b': 'B1'}
{'a': 'A2', 'b': 'B1'}
{'a': 'A2', 'b': 'B2'}

Environment Details:

• ADK Library Version: 2.11.0 (also reproduced on main at bcac35a)
• Desktop OS: macOS
• Python Version: 3.12.13

Model Information:

• Are you using LiteLLM: No
• Which model is being used: N/A (function nodes only)

🟡 Optional Information

Minimal Reproduction Code:

import asyncio

from google.adk.events.event import Event
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import Edge, JoinNode, START, Workflow, node
from google.genai import types

iteration = 0
join_inputs = []


def a(node_input):
  global iteration
  iteration += 1
  return f"A{iteration}"


def x(node_input):
  return "x"


def b(node_input):  # one hop further from START than `a`
  return f"B{iteration}"


def check(node_input):
  join_inputs.append(node_input)
  if len(join_inputs) < 2:
    yield Event(route="again")
  else:
    yield Event(route="done", output="finished")


def end(node_input):
  return node_input


A, X, B, C, E = node(a), node(x), node(b), node(check), node(end)
J = JoinNode(name="J")
wf = Workflow(
    name="wf",
    edges=[
        Edge(from_node=START, to_node=A),
        Edge(from_node=START, to_node=X),
        Edge(from_node=X, to_node=B),
        Edge(from_node=A, to_node=J),
        Edge(from_node=B, to_node=J),
        Edge(from_node=J, to_node=C),
        Edge(from_node=C, to_node=A, route="again"),
        Edge(from_node=C, to_node=X, route="again"),
        Edge(from_node=C, to_node=E, route="done"),
    ],
)


async def main():
  runner = Runner(app_name="app", node=wf, session_service=InMemorySessionService())
  session = await runner.session_service.create_session(app_name="app", user_id="u")
  async for _ in runner.run_async(
      user_id="u",
      session_id=session.id,
      new_message=types.Content(role="user", parts=[types.Part(text="go")]),
  ):
    pass
  for i in join_inputs:
    print(dict(sorted(i.items())))


asyncio.run(main())

How often has this issue occurred?:

• Always (100%)

Proposed fix: Track, per wait-for-all node, which predecessors have completed since it last fired, and clear that set when it fires; check that set instead of NodeState.status. Resume is unaffected because resume replays recorded completions through the same completion path, which rebuilds the set. I have a PR ready with unit tests (plain loop, and a loop where the slower branch pauses for input, resumable and non-resumable).