bug(graph): OR-join node double-executes when 2+ of its predecessors land in the same dispatch batch #3628

Found while reviewing PR #3448 (the max_concurrency feature) — confirmed present on main / with no max_concurrency set, so it's unrelated to that PR. Filing separately per review convention.

Repro topology: x -> a0..a4, x -> bb, a0..a4 -> bb (a fan-out from x into 5 siblings that all also feed bb, plus bb also depends directly on x).

Batch 1 = [x]. Once x completes, _find_newly_ready_nodes([x]) (graph.py:981) correctly adds a0..a4 and bb to batch 2, since bb has a direct x -> bb edge — this is intended OR-join behavior.

Batch 2 = [a0, a1, a2, a3, a4, bb] all dispatch together. After batch 2 finishes, _execute_graph (graph.py:829-830) calls _find_newly_ready_nodes(current_batch) with current_batch = all 6 nodes just dispatched, bb included. _is_node_ready_with_conditions (graph.py:996) only checks "is there a satisfied incoming edge from something in this batch" — it does not exclude the target node itself from being re-evaluated, and does not check whether the target already ran. Since a0->bb (and a1->bb, etc.) are real edges and a0..a4 are members of the batch that just completed, bb is deemed "newly ready" again and gets dispatched a second time in batch 3 — i.e. bb.stream_async/invoke_async runs twice for one logical graph execution.

This reproduces with max_concurrency unset (None), using only currently-shipped, unmodified _find_newly_ready_nodes/_is_node_ready_with_conditions/_execute_graph (verified byte-identical to main while reviewing #3448).

from .test_graph import create_mock_agent
from strands.hooks import BeforeNodeCallEvent
from strands.multiagent.graph import GraphBuilder
import pytest

@pytest.mark.asyncio
async def test_or_join_double_fires_within_batch():
    n = 5
    builder = GraphBuilder()
    builder.add_node(create_mock_agent("x", "x done"), "x")
    builder.set_entry_point("x")
    for i in range(n):
        builder.add_node(create_mock_agent(f"a{i}", f"a{i} done"), f"a{i}")
        builder.add_edge("x", f"a{i}")
    agent_bb = create_mock_agent("bb", "bb done")
    builder.add_node(agent_bb, "bb")
    builder.add_edge("x", "bb")
    for i in range(n):
        builder.add_edge(f"a{i}", "bb")
    graph = builder.build()  # no max_concurrency set

    calls = []
    graph.add_hook(lambda e: calls.append(e.node_id) if e.node_id == "bb" else None, BeforeNodeCallEvent)

    await graph.invoke_async("task")
    assert len(calls) == 1, f"bb's BeforeNodeCallEvent fired {len(calls)} times, expected 1"

Observed: bb's BeforeNodeCallEvent fires twice, and agent_bb.stream_async is called twice, for a single invoke_async. The final GraphResult still reports Status.COMPLETED, so this is silent (extra cost/side-effects for a non-idempotent node, doubled token usage, no visible error).

Suspected root cause: _find_newly_ready_nodes/_is_node_ready_with_conditions (graph.py:981-1013) treat every member of the just-completed current_batch as a fresh "source" for edge-satisfaction checks, without excluding edges whose target is also a member of that same batch (i.e. a node that already ran this batch via an earlier-satisfied OR-edge shouldn't be re-armed by a later-satisfied edge from a batch-mate).

Not urgent/security-sensitive, but worth a look — needs an idempotency/already-completed guard in _find_newly_ready_nodes or _execute_node, and a regression test for a node with ≥2 real incoming edges that land in the same batch.