From a630ef4c2a9e796259623609dfebfd9bec0c2535 Mon Sep 17 00:00:00 2001 From: Ray Neto Date: Tue, 11 Aug 2026 16:34:55 -0300 Subject: [PATCH 1/2] test(runners): cover duplicate user event append on invocation retry Four tests driving the public run_async API: retrying an invocation with the same invocation_id and the same new_message must leave exactly one user event in the session, on both the resumable and the non-resumable path. Two non-regression tests pin that a different message or a different state_delta still appends. Both dedup tests fail on main today, on the bug reported in https://github.com/google/adk-python/issues/4506. --- tests/unittests/test_runners.py | 169 ++++++++++++++++++++++++++++++++ 1 file changed, 169 insertions(+) diff --git a/tests/unittests/test_runners.py b/tests/unittests/test_runners.py index 05460f50f2..0b0a731515 100644 --- a/tests/unittests/test_runners.py +++ b/tests/unittests/test_runners.py @@ -2983,5 +2983,174 @@ async def test_base_agent_run_live_does_not_leak_context(): otel_context.detach(token) +def _user_events_for(session: Session, invocation_id: str) -> list[Event]: + return [ + event + for event in session.events + if event.author == "user" and event.invocation_id == invocation_id + ] + + +async def _drain_events(agen) -> None: + async with aclosing(agen) as events: + async for _ in events: + pass + + +def _user_message(text: str) -> types.Content: + """A fresh Content per call: run_async mutates `role` on the object it gets.""" + return types.Content(role="user", parts=[types.Part(text=text)]) + + +@pytest.mark.asyncio +async def test_resumable_retry_with_same_message_appends_one_user_event(): + """Resuming an invocation with the same new_message must not duplicate it. + + Regression test for https://github.com/google/adk-python/issues/4506. + + Setup: a resumable app runs one invocation to completion. + Act: run_async again with that invocation_id and an identical new_message. + Assert: the invocation still holds exactly one user event. + """ + session_service = InMemorySessionService() + runner = Runner( + app=App( + name=TEST_APP_ID, + root_agent=MockAgent("root_agent"), + resumability_config=ResumabilityConfig(is_resumable=True), + ), + session_service=session_service, + artifact_service=InMemoryArtifactService(), + ) + session = await session_service.create_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID + ) + await _drain_events( + runner.run_async( + user_id=TEST_USER_ID, + session_id=session.id, + new_message=_user_message("hello"), + ) + ) + started = await session_service.get_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID, session_id=session.id + ) + invocation_id = next( + event for event in started.events if event.author == "user" + ).invocation_id + + await _drain_events( + runner.run_async( + user_id=TEST_USER_ID, + session_id=session.id, + invocation_id=invocation_id, + new_message=_user_message("hello"), + ) + ) + + stored = await session_service.get_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID, session_id=session.id + ) + assert len(_user_events_for(stored, invocation_id)) == 1 + + +@pytest.mark.asyncio +async def test_non_resumable_retry_with_same_message_appends_one_user_event(): + """The same guarantee on a non-resumable app given an explicit invocation_id. + + A non-resumable app never enters the resume path: run_async routes an explicit + invocation_id straight to _setup_context_for_new_invocation. An at-least-once + task queue retrying the same request duplicates the user event there too. + """ + session_service = InMemorySessionService() + runner = Runner( + app_name=TEST_APP_ID, + agent=MockAgent("root_agent"), + session_service=session_service, + artifact_service=InMemoryArtifactService(), + ) + session = await session_service.create_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID + ) + + for _ in range(2): + await _drain_events( + runner.run_async( + user_id=TEST_USER_ID, + session_id=session.id, + invocation_id="inv-task-retry", + new_message=_user_message("hello"), + ) + ) + + stored = await session_service.get_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID, session_id=session.id + ) + assert len(_user_events_for(stored, "inv-task-retry")) == 1 + + +@pytest.mark.asyncio +async def test_retry_with_a_different_message_still_appends(): + """The guard must be scoped to identical content, not to the invocation id.""" + session_service = InMemorySessionService() + runner = Runner( + app_name=TEST_APP_ID, + agent=MockAgent("root_agent"), + session_service=session_service, + artifact_service=InMemoryArtifactService(), + ) + session = await session_service.create_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID + ) + + for text in ("first", "second"): + await _drain_events( + runner.run_async( + user_id=TEST_USER_ID, + session_id=session.id, + invocation_id="inv-distinct", + new_message=_user_message(text), + ) + ) + + stored = await session_service.get_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID, session_id=session.id + ) + assert len(_user_events_for(stored, "inv-distinct")) == 2 + + +@pytest.mark.asyncio +async def test_retry_with_a_different_state_delta_still_appends(): + """A retry carrying a different state delta still has an effect to persist.""" + session_service = InMemorySessionService() + runner = Runner( + app_name=TEST_APP_ID, + agent=MockAgent("root_agent"), + session_service=session_service, + artifact_service=InMemoryArtifactService(), + ) + session = await session_service.create_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID + ) + + for attempt in (1, 2): + await _drain_events( + runner.run_async( + user_id=TEST_USER_ID, + session_id=session.id, + invocation_id="inv-state-delta", + new_message=_user_message("same text"), + state_delta={"attempt": attempt}, + ) + ) + + stored = await session_service.get_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID, session_id=session.id + ) + events = _user_events_for(stored, "inv-state-delta") + assert len(events) == 2 + assert [event.actions.state_delta["attempt"] for event in events] == [1, 2] + + if __name__ == "__main__": pytest.main([__file__]) From 50fa7496d3cac7b911cd77677c62cea4d10c099a Mon Sep 17 00:00:00 2001 From: Ray Neto Date: Tue, 11 Aug 2026 16:48:59 -0300 Subject: [PATCH 2/2] fix(runners): skip duplicate user event append on invocation retry _append_new_message_to_session appended the user event unconditionally, so an invocation that re-sends the same new_message recorded the user turn twice. Both call paths reach it: resuming a resumable invocation, and a non-resumable app given an explicit invocation_id by a caller such as an at-least-once task queue. The guard sits at that convergence point, after the plugin callback has run, and matches on both content and state_delta so a retry carrying a different delta still persists. Related: https://github.com/google/adk-python/issues/4506 --- src/google/adk/runners.py | 50 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/src/google/adk/runners.py b/src/google/adk/runners.py index 626b6ee5d1..b618cc8366 100644 --- a/src/google/adk/runners.py +++ b/src/google/adk/runners.py @@ -1802,6 +1802,25 @@ async def _append_new_message_to_session( new_message.parts[i] = types.Part( text=f'Uploaded file: {file_name}. It is saved into artifacts' ) + + # A resumed or retried invocation may re-send the same user message — an + # at-least-once task queue replaying the request with the same + # invocation_id, for example. Appending it again duplicates the user turn in + # the session timeline, so skip the append when this invocation already + # recorded an identical user event. + # See https://github.com/google/adk-python/issues/4506. + if self._invocation_has_user_event( + session=session, + invocation_id=invocation_context.invocation_id, + new_message=new_message, + state_delta=state_delta, + ): + logger.info( + 'Skipping duplicate user event append for invocation_id %s.', + invocation_context.invocation_id, + ) + return + # Appends only. We do not yield the event because it's not from the model. if state_delta: event = Event( @@ -1823,6 +1842,37 @@ async def _append_new_message_to_session( session=invocation_context.session, event=event ) + def _invocation_has_user_event( + self, + *, + session: Session, + invocation_id: str, + new_message: types.Content, + state_delta: Optional[dict[str, Any]], + ) -> bool: + """Whether this invocation already recorded this exact user event. + + Both the content and the state delta must match: a retry that carries a + different state delta still has an effect left to persist. + + Args: + session: The session to inspect. + invocation_id: The invocation the user event would belong to. + new_message: The user message about to be appended. + state_delta: The state changes the append would carry. + + Returns: + True if an identical user event is already recorded for the invocation. + """ + expected_state_delta = state_delta or {} + return any( + event.author == 'user' + and event.invocation_id == invocation_id + and event.content == new_message + and event.actions.state_delta == expected_state_delta + for event in session.events + ) + async def run_live( self, *,