Skip to content

Commit cf84e6b

Browse files
DABHclaude
andcommitted
fix: surface a deterministic exception from ParallelWorker on concurrent failures
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. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 40cf97b commit cf84e6b

2 files changed

Lines changed: 38 additions & 1 deletion

File tree

src/google/adk/workflow/_parallel_worker.py

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,10 @@ async def _run_impl(
116116
done, pending = await asyncio.wait(
117117
pending_tasks, return_when=asyncio.FIRST_COMPLETED
118118
)
119-
for task in done:
119+
# asyncio.wait returns completed tasks as an unordered set; iterate in
120+
# input order so that, when several tasks fail in the same wake-up,
121+
# the surfaced exception is deterministic across runs and replays.
122+
for task in sorted(done, key=lambda t: getattr(t, '_worker_index')):
120123
exc = task.exception()
121124
if exc is not None:
122125
# If a task failed, cancel all other pending tasks.

tests/unittests/workflow/test_workflow_parallel_worker.py

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1090,3 +1090,37 @@ async def hitl_concurrency_worker(
10901090
},
10911091
),
10921092
]
1093+
1094+
1095+
@pytest.mark.asyncio
1096+
async def test_parallel_worker_simultaneous_failures_raise_lowest_index(
1097+
request: pytest.FixtureRequest,
1098+
):
1099+
"""The exception surfaced from concurrent failures is deterministic.
1100+
1101+
Setup: 2 items whose workers both fail immediately, so both tasks can
1102+
complete within the same asyncio.wait wake-up.
1103+
Assert: the propagated exception is always the lowest-index item's.
1104+
Previously the failed task was picked by iterating the unordered set
1105+
returned by asyncio.wait, so the surfaced exception could differ
1106+
between runs (and between record and replay).
1107+
"""
1108+
1109+
async def _worker_always_fails(node_input: str) -> str:
1110+
raise ValueError(f'{node_input} failed')
1111+
1112+
for _ in range(10):
1113+
node_a = _ProducerNode(items=['item-0', 'item-1'], name='NodeA')
1114+
worker = ParallelWorker(node=_worker_always_fails)
1115+
agent = Workflow(
1116+
name='test_agent_simultaneous_fail',
1117+
edges=[
1118+
(START, node_a),
1119+
(node_a, worker),
1120+
],
1121+
)
1122+
app = App(name=request.function.__name__, root_agent=agent)
1123+
runner = testing_utils.InMemoryRunner(app=app)
1124+
1125+
with pytest.raises(ValueError, match='item-0 failed'):
1126+
await runner.run_async(testing_utils.get_user_content('start'))

0 commit comments

Comments
 (0)