Skip to content

Commit 34c5427

Browse files
cursoragentZsanz3
andcommitted
fix(flows): replay unanswered sibling on parallel tool resume
decide_step_resume treated one answered sibling as coverage for the whole event, so a parallel call that never ran was skipped. Require full id and name coverage (issubset) before continuing. Fixes #7108 Co-authored-by: Zsanz3 <Zsanz3@users.noreply.github.com>
1 parent 7ae1c9b commit 34c5427

2 files changed

Lines changed: 82 additions & 37 deletions

File tree

‎src/google/adk/flows/llm_flows/core/_resume.py‎

Lines changed: 30 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -110,9 +110,8 @@ def _pause_left_calls_unanswered(
110110
fr.id for ev in events for fr in ev.get_function_responses() if fr.id
111111
}
112112
# `issubset`, not `&`: this asks whether *any* awaited id is still open, so a
113-
# partially answered pause keeps waiting. `decide_resume` asks the opposite
114-
# question of its own ids -- whether *none* are answered -- and drops
115-
# `issubset` for that reason. The two are not interchangeable.
113+
# partially answered pause keeps waiting. Intersection would treat one
114+
# sibling answer as coverage for the rest.
116115
return bool(awaited) and not awaited.issubset(answered)
117116

118117

@@ -191,20 +190,26 @@ def _needs_call_replay(
191190
call_names: set[str | None],
192191
answers: list[types.FunctionResponse],
193192
from_sub_branch: bool,
193+
call_ids: set[str | None] | None = None,
194+
answered_ids: set[str] | None = None,
194195
) -> bool:
195196
"""Whether the calls named by `call_names` still have to be run.
196197
197-
`call_names` holds every name on the call event, not just the first: one
198-
event can carry parallel calls, and an answer to the second is not evidence
199-
the first never ran.
198+
Coverage is all-or-nothing: an answer to one parallel call is not
199+
evidence the others ran. Names and ids both have to be covered --
200+
two calls can share a name, so names alone cannot see a missing twin.
200201
"""
201202
if not call_names:
202203
return False
203-
return (
204-
not answers
205-
or any(fr.name not in call_names for fr in answers)
206-
or from_sub_branch
207-
)
204+
answered_names = {fr.name for fr in answers}
205+
names_uncovered = not call_names.issubset(answered_names)
206+
ids_uncovered = False
207+
if call_ids is not None and answered_ids is not None:
208+
concrete_ids = {i for i in call_ids if i is not None}
209+
ids_uncovered = bool(concrete_ids) and not concrete_ids.issubset(
210+
answered_ids
211+
)
212+
return names_uncovered or ids_uncovered or from_sub_branch
208213

209214

210215
def decide_resume(
@@ -257,18 +262,25 @@ def decide_resume(
257262
# short-circuits both unanswered tests rather than being repeated in each.
258263
from_sub_branch = _is_sub_branch_answer(answer_event, call_event)
259264
answers = answer_event.get_function_responses()
260-
# `ids & answered` alone decides these: a set that is a subset of the
261-
# answered ids necessarily intersects it, so testing `issubset` as well
262-
# never changes the outcome.
263-
lro_unanswered = bool(lro_ids) and not lro_ids & answered_ids
265+
concrete_call_ids = {i for i in call_ids if i is not None}
266+
# `issubset`, not `&`: one answered id does not cover a sibling that
267+
# never ran. Pause only when nothing matched (no id and no name);
268+
# leftover ids after a sibling answer are a replay, not a pause.
269+
lro_unanswered = bool(lro_ids) and not lro_ids.issubset(answered_ids)
264270
call_unanswered = (
265-
bool(call_ids)
266-
and not call_ids & answered_ids
271+
bool(concrete_call_ids)
272+
and not concrete_call_ids.issubset(answered_ids)
267273
and not any(fr.name in call_names for fr in answers)
268274
)
269275
if not from_sub_branch and (lro_unanswered or call_unanswered):
270276
pause = True
271-
elif _needs_call_replay(call_names, answers, from_sub_branch):
277+
elif _needs_call_replay(
278+
call_names,
279+
answers,
280+
from_sub_branch,
281+
call_ids=concrete_call_ids,
282+
answered_ids=answered_ids,
283+
):
272284
return ResumeDecision(ResumeAction.REPLAY_CALLS, call_event)
273285

274286
return ResumeDecision(ResumeAction.PAUSE if pause else ResumeAction.CONTINUE)

‎tests/unittests/flows/llm_flows/core/test_resume.py‎

Lines changed: 52 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,22 @@ def _call_event(name: str, call_id: str, *, lro: bool = False) -> Event:
5252
)
5353

5454

55+
def _parallel_call_event(pairs: list[tuple[str, str]]) -> Event:
56+
return Event(
57+
author='agent',
58+
invocation_id='inv-1',
59+
content=types.Content(
60+
role='model',
61+
parts=[
62+
types.Part(
63+
function_call=types.FunctionCall(id=i, name=n, args={})
64+
)
65+
for i, n in pairs
66+
],
67+
),
68+
)
69+
70+
5571
def _response_event(
5672
name: str,
5773
response_id: str | None,
@@ -292,25 +308,7 @@ def test_parallel_calls_all_answered_continue(self):
292308
# One event can carry parallel calls. Matching answers against only the
293309
# first call's name reads the second answer as a foreign name, so a fully
294310
# answered event is replayed and both tools run a second time.
295-
call = Event(
296-
author='agent',
297-
invocation_id='inv-1',
298-
content=types.Content(
299-
role='model',
300-
parts=[
301-
types.Part(
302-
function_call=types.FunctionCall(
303-
id='c1', name='ask', args={}
304-
)
305-
),
306-
types.Part(
307-
function_call=types.FunctionCall(
308-
id='c2', name='fetch', args={}
309-
)
310-
),
311-
],
312-
),
313-
)
311+
call = _parallel_call_event([('c1', 'ask'), ('c2', 'fetch')])
314312
events = [
315313
call,
316314
_response_event('ask', 'c1'),
@@ -321,6 +319,29 @@ def test_parallel_calls_all_answered_continue(self):
321319
)
322320
assert decision.action is ResumeAction.CONTINUE
323321

322+
def test_parallel_calls_partially_answered_replay(self):
323+
# A sibling answer is not coverage for a call that never ran. Any-answered
324+
# (name match or id intersection) would CONTINUE here and drop fetch.
325+
call = _parallel_call_event([('c1', 'ask'), ('c2', 'fetch')])
326+
events = [call, _response_event('ask', 'c1')]
327+
328+
decision = decide_resume(
329+
self._ctx(), events, {'ask': object(), 'fetch': object()}
330+
)
331+
332+
assert decision.action is ResumeAction.REPLAY_CALLS
333+
assert decision.event is call
334+
335+
def test_parallel_same_name_partially_answered_replay(self):
336+
# Two calls can share a name, so names alone cannot see the missing twin.
337+
call = _parallel_call_event([('c1', 'ask'), ('c2', 'ask')])
338+
events = [call, _response_event('ask', 'c1')]
339+
340+
decision = decide_resume(self._ctx(), events, {'ask': object()})
341+
342+
assert decision.action is ResumeAction.REPLAY_CALLS
343+
assert decision.event is call
344+
324345
def test_sub_branch_answer_replays_instead_of_pausing(self):
325346
# A HITL answer returned against the branch the call opened resolves it,
326347
# even though it carries none of the call's ids.
@@ -434,3 +455,15 @@ def test_the_agents_own_trailing_call_is_still_replayed(self):
434455
)
435456
assert decision.action is ResumeAction.REPLAY_CALLS
436457
assert decision.replay_event() is call
458+
459+
def test_a_partially_answered_parallel_step_is_replayed(self):
460+
# The entry point must not CONTINUE just because a sibling was answered.
461+
call = _parallel_call_event([('c1', 'ask'), ('c2', 'fetch')])
462+
events = [call, _response_event('ask', 'c1')]
463+
464+
decision = decide_step_resume(
465+
self._ctx(events), {'ask': object(), 'fetch': object()}
466+
)
467+
468+
assert decision.action is ResumeAction.REPLAY_CALLS
469+
assert decision.replay_event() is call

0 commit comments

Comments
 (0)