Workflow: node returning None in a later loop iteration leaves its previous output for JoinNode #7515

🔴 Required Information

Describe the Bug:
In a looping workflow, when a node returns None on a later iteration, a downstream JoinNode still receives that node's output from the previous iteration.

Workflow._handle_completion (src/google/adk/workflow/_workflow.py) only writes to loop_state.node_outputs when the output is not None:

if child_ctx.output is not None:
  loop_state.node_outputs[node_name] = child_ctx.output

So the old value stays in place, and the join reads it with loop_state.node_outputs.get(p). On the first iteration the same node returning None gives the join None. On later iterations it gives stale data, so the result depends on what happened in an earlier pass.

This is the follow-up I mentioned in #7498. It is a separate bug: it reproduces on main with or without #7499.

Steps to Reproduce:

1. pip install google-adk==2.11.0
2. Run the snippet below. a returns "A1" on the first iteration and None on the second.

Expected Behavior:

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

Observed Behavior:

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

Environment Details:

• ADK Library Version: 2.11.0 (also reproduced on main at b3cbb47)
• 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 "A1" if iteration == 1 else None  # no output on iteration 2


def b(node_input):
  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, B, C, E = node(a), 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=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=B, 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 (looking for guidance before I open a PR)

I'd like to fix this, but there are a few ways to do it and they behave differently, so I'd like to check which one fits the design first.

1. Remove the entry when a node completes with None (loop_state.node_outputs.pop(node_name, None)). A key in node_outputs would still mean "this node produced output", so _finalize and _has_terminal_output keep working as they do now. The join gets None through .get(), which matches what it gets on the first iteration. This is my preferred option: it changes one line and only affects nodes that run more than once.
2. Always store the output, including None. This is simpler, but it changes how terminal outputs work. A terminal node that returns None would count as having produced output, which affects _has_terminal_output and the "multiple terminal nodes produced output" check.
3. Clear the join's predecessor outputs when the join fires. This only affects joins, but node_outputs could still go stale for other readers, such as a terminal node that runs more than once.

Questions:

• Is None meant to mean "no output", so the join should see 'a': None? Or should a predecessor that returned nothing be left out of the join's input dict entirely?
• Does option 1 fit, or would you prefer another approach?
• Should this be a separate small PR, or folded into fix(workflow): make JoinNode in a loop wait for each iteration's predecessors #7499?

Once a direction is confirmed, I'll open a PR with a unit test in tests/unittests/workflow/test_join_node.py.