From 93d0bb380c2347edf6a2bf84e84f2dd851ad37bd Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 15:32:32 -0500 Subject: [PATCH] fix: surface a deterministic exception from ParallelWorker on concurrent failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit asyncio.wait returns completed tasks as an unordered set. _ParallelWorker iterated that set directly to pick the exception to raise, so when two or more branches failed within the same wake-up, which branch's exception surfaced depended on set iteration order (object-hash dependent). This made failure output differ between runs — and between record and replay for frameworks that resume or replay agent workflows and filter retries by exception type (RetryConfig.exceptions). Iterate completed tasks in input-index order instead, so the lowest-index failed branch's exception is surfaced consistently. Behavior is otherwise unchanged. The regression test fails ~50% of the time without the fix (two branches failing in the same wake-up) and always passes with it. --- src/google/adk/workflow/_parallel_worker.py | 5 ++- .../workflow/test_workflow_parallel_worker.py | 34 +++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/src/google/adk/workflow/_parallel_worker.py b/src/google/adk/workflow/_parallel_worker.py index 4e2a9420e7e..c329161dbad 100644 --- a/src/google/adk/workflow/_parallel_worker.py +++ b/src/google/adk/workflow/_parallel_worker.py @@ -116,7 +116,10 @@ async def _run_impl( done, pending = await asyncio.wait( pending_tasks, return_when=asyncio.FIRST_COMPLETED ) - for task in done: + # asyncio.wait returns completed tasks as an unordered set; iterate in + # input order so that, when several tasks fail in the same wake-up, + # the surfaced exception is deterministic across runs and replays. + for task in sorted(done, key=lambda t: getattr(t, '_worker_index')): exc = task.exception() if exc is not None: # If a task failed, cancel all other pending tasks. diff --git a/tests/unittests/workflow/test_workflow_parallel_worker.py b/tests/unittests/workflow/test_workflow_parallel_worker.py index 632ecc873a3..e4d616e5c74 100644 --- a/tests/unittests/workflow/test_workflow_parallel_worker.py +++ b/tests/unittests/workflow/test_workflow_parallel_worker.py @@ -1090,3 +1090,37 @@ async def hitl_concurrency_worker( }, ), ] + + +@pytest.mark.asyncio +async def test_parallel_worker_simultaneous_failures_raise_lowest_index( + request: pytest.FixtureRequest, +): + """The exception surfaced from concurrent failures is deterministic. + + Setup: 2 items whose workers both fail immediately, so both tasks can + complete within the same asyncio.wait wake-up. + Assert: the propagated exception is always the lowest-index item's. + Previously the failed task was picked by iterating the unordered set + returned by asyncio.wait, so the surfaced exception could differ + between runs (and between record and replay). + """ + + async def _worker_always_fails(node_input: str) -> str: + raise ValueError(f'{node_input} failed') + + for _ in range(10): + node_a = _ProducerNode(items=['item-0', 'item-1'], name='NodeA') + worker = ParallelWorker(node=_worker_always_fails) + agent = Workflow( + name='test_agent_simultaneous_fail', + edges=[ + (START, node_a), + (node_a, worker), + ], + ) + app = App(name=request.function.__name__, root_agent=agent) + runner = testing_utils.InMemoryRunner(app=app) + + with pytest.raises(ValueError, match='item-0 failed'): + await runner.run_async(testing_utils.get_user_content('start'))