From c5cae933e7c3dfe10b50d53920af4ab2fb49f9ea Mon Sep 17 00:00:00 2001 From: chelsealong Date: Tue, 22 Sep 2026 14:27:55 +0000 Subject: [PATCH] fix(apps): stop orphaned function_call from permanently blocking compaction _longest_self_contained_prefix walks a compaction window keeping a set of open obligations keyed by call id, and only records a cut point where that set is empty. Once the first compaction has happened, an orphaned function_call with no matching function_response anywhere in the session (e.g. the process handling the tool died) becomes the first candidate event of every following window, so the set can never empty again and compaction silently stops forever. When the strict pass finds no balanced point, retry once while ignoring call ids that are provably dead: no response exists anywhere in the session (not just the window, to avoid splitting a pair whose response lives past the window boundary), the call isn't long-running or a pending auth/tool-confirmation request, and it wasn't opened by the newest invocation in the session (which may still be in flight). Everything else keeps blocking exactly as before. Fixes #7236 --- src/google/adk/apps/compaction.py | 95 +++++++++++++++++++++---- tests/unittests/apps/test_compaction.py | 40 +++++++++++ 2 files changed, 120 insertions(+), 15 deletions(-) diff --git a/src/google/adk/apps/compaction.py b/src/google/adk/apps/compaction.py index c34a3d544f2..a8b81aa4a5b 100644 --- a/src/google/adk/apps/compaction.py +++ b/src/google/adk/apps/compaction.py @@ -295,7 +295,9 @@ def _events_to_compact_for_token_threshold( event_retention_size=event_retention_size, ) events_to_compact = candidate_events[:split_index] - events_to_compact = _longest_self_contained_prefix(events_to_compact) + events_to_compact = _longest_self_contained_prefix( + events_to_compact, all_events=events + ) if not events_to_compact: return [] @@ -336,7 +338,44 @@ def _event_function_response_ids(event: Event) -> set[str]: return function_response_ids -def _longest_self_contained_prefix(events: list[Event]) -> list[Event]: +def _provably_dead_call_ids( + events: list[Event], + *, + all_events: list[Event], + newest_invocation_id: str | None, +) -> set[str]: + """Returns function-call ids opened in `events` that can never be answered. + + A call id qualifies once: it was opened by a plain function call (not one + marked long-running, nor one with a pending tool-confirmation/auth + request -- those are expected to stay open); no function response with the + same id exists anywhere in the session, not just the compaction window, so + a response that lives past the window boundary still protects its call; + and the call was not opened by the newest invocation in the session, which + may still be in flight. + """ + protected_ids: set[str] = set() + answered_ids: set[str] = set() + for event in all_events: + protected_ids |= event.long_running_tool_ids or set() + if event.actions: + protected_ids |= set(event.actions.requested_tool_confirmations) + protected_ids |= set(event.actions.requested_auth_configs) + answered_ids |= _event_function_response_ids(event) + + dead_ids: set[str] = set() + for event in events: + if event.invocation_id == newest_invocation_id: + continue + for call_id in _event_function_call_ids(event): + if call_id not in protected_ids and call_id not in answered_ids: + dead_ids.add(call_id) + return dead_ids + + +def _longest_self_contained_prefix( + events: list[Event], *, all_events: list[Event] +) -> list[Event]: """Returns the longest prefix of `events` that is safe to compact. Performs a single left-to-right pass tracking "open" obligations keyed by call @@ -345,19 +384,43 @@ def _longest_self_contained_prefix(events: list[Event]) -> list[Event]: opens within each event so a response only closes an obligation opened by an earlier event. The prefix is safe to summarize only at points where no obligation is open, so the longest prefix ending at such a balanced point is - returned (empty if the window never reaches a balanced point). + returned. + + If that strict pass never reaches a balanced point (e.g. the window opens + with a function call whose response will never arrive, because the process + handling it died), a second pass retries while ignoring call ids that + `_provably_dead_call_ids` proves can never be fulfilled, so healthy events + after a dead call can still be compacted. Without this, a single orphaned + call permanently blocks every future compaction, since it becomes the + first candidate event of every subsequent window once it is not covered by + the previous one. """ - open_ids: set[str] = set() - safe_length = 0 - for index, event in enumerate(events): - open_ids -= _event_function_response_ids(event) - open_ids |= _event_function_call_ids(event) - if event.actions: - open_ids |= set(event.actions.requested_tool_confirmations) - open_ids |= set(event.actions.requested_auth_configs) - if not open_ids: - safe_length = index + 1 - return events[:safe_length] + + def _pass(dead_ids: set[str]) -> int: + open_ids: set[str] = set() + safe_length = 0 + for index, event in enumerate(events): + open_ids -= _event_function_response_ids(event) + open_ids |= _event_function_call_ids(event) - dead_ids + if event.actions: + open_ids |= set(event.actions.requested_tool_confirmations) + open_ids |= set(event.actions.requested_auth_configs) + if not open_ids: + safe_length = index + 1 + return safe_length + + safe_length = _pass(set()) + if safe_length: + return events[:safe_length] + + newest_invocation_id = all_events[-1].invocation_id if all_events else None + dead_ids = _provably_dead_call_ids( + events, all_events=all_events, newest_invocation_id=newest_invocation_id + ) + if not dead_ids: + return events[:safe_length] + + return events[: _pass(dead_ids)] def _safe_token_compaction_split_index( @@ -600,7 +663,9 @@ async def _run_compaction_for_sliding_window( events_to_compact = [ e for e in events_to_compact if not e.actions.compaction ] - events_to_compact = _longest_self_contained_prefix(events_to_compact) + events_to_compact = _longest_self_contained_prefix( + events_to_compact, all_events=events + ) if not events_to_compact: return diff --git a/tests/unittests/apps/test_compaction.py b/tests/unittests/apps/test_compaction.py index 4a848fa7982..1a36c25e04e 100644 --- a/tests/unittests/apps/test_compaction.py +++ b/tests/unittests/apps/test_compaction.py @@ -1222,6 +1222,46 @@ async def test_sliding_window_plain_orphaned_function_call_dropped_from_contents self.assertEqual(result_contents[0].parts[0].text, 'Summary safe prefix') self.assertEqual(result_contents[1].parts[0].text, 'e3') + async def test_sliding_window_orphaned_call_after_prior_compaction_does_not_block_forever( + self, + ): + """An orphaned call opening a post-compaction window must not stall future compactions.""" + app = App( + name='test', + root_agent=Mock(spec=BaseAgent), + events_compaction_config=EventsCompactionConfig( + summarizer=self.mock_compactor, + compaction_interval=2, + overlap_size=0, + ), + ) + events = [ + self._create_compacted_event(1.0, 2.0, 'Summary 1-2'), + # inv3's call is orphaned: no matching response anywhere in the + # session, e.g. the process handling it died mid-call. + self._create_function_call_event(3.0, 'inv3', 'orphan-call-1'), + self._create_event(4.0, 'inv4', 'e4'), + ] + session = Session(app_name='test', user_id='u1', id='s1', events=events) + self.mock_compactor.maybe_summarize_events.side_effect = ( + lambda *, events: self._create_compacted_event( + events[0].timestamp, + events[-1].timestamp, + 'Summary inv3-inv4', + ) + ) + + await self._run_sliding_window(app, session, self.mock_session_service) + + # Without the fix, the orphaned call opens the window and never closes, + # so no compaction event is appended at all. + self.mock_session_service.append_event.assert_called_once() + appended_event = self.mock_session_service.append_event.call_args[1][ + 'event' + ] + self.assertEqual(appended_event.actions.compaction.start_timestamp, 3.0) + self.assertEqual(appended_event.actions.compaction.end_timestamp, 4.0) + async def test_token_threshold_excludes_pending_function_call_events(self): """Token-threshold compaction stays contiguous before pending calls.""" app = App(