runner: prevent duplicate user event append on invocation retry with same invocation_id

[davidahmann]

Problem

Retrying/resuming with the same invocation_id and same user message can append duplicate user events to the session timeline. That weakens invocation idempotency and can trigger duplicate downstream effects.

Expected behavior

If the same user event for the same invocation is already persisted, retry should be a no-op for message append.

Scope (Phase 1)

• Add duplicate-event guard before appending user message event.
• Add runner-level regression test for repeated resume/retry with same invocation ID + message.

Acceptance criteria

• Duplicate retry does not append a second identical user event.
• Existing non-duplicate message append behavior remains unchanged.
• Regression test covers repeated retry path.

[llalitkumarrr]

Hello @davidahmann ,

We would like to analyze this issue in detail. Could you please let us know which version of google-adk you are using? It would also be very helpful if you could provide reproduction steps and a minimal reproducible code snippet to assist in our analysis.

[davidahmann]

@llalitkumarrr Thanks for the follow-up. I reproduced this on a released build.

Environment:

• google-adk: 1.25.0
• Python: 3.13.7
• OS: macOS arm64

Minimal repro:

import asyncio
from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.apps.app import App, ResumabilityConfig
from google.adk.events.event import Event
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.genai import types

class StubAgent(BaseAgent):
    async def _run_async_impl(self, invocation_context: InvocationContext):
        yield Event(
            invocation_id=invocation_context.invocation_id,
            author=self.name,
            content=types.Content(role="model", parts=[types.Part(text="ok")]),
        )

async def main():
    ss = InMemorySessionService()
    app = App(
        name="app",
        root_agent=StubAgent(name="stub", sub_agents=[]),
        resumability_config=ResumabilityConfig(is_resumable=True),
    )
    runner = Runner(app=app, session_service=ss, artifact_service=InMemoryArtifactService())
    session = await ss.create_session(app_name="app", user_id="u")
    msg = types.Content(role="user", parts=[types.Part(text="hello")])

    # Simulate first attempt already persisted for invocation_id inv-1
    await ss.append_event(session=session, event=Event(invocation_id="inv-1", author="user", content=msg))

    # Retry same invocation_id with same user message
    async for _ in runner.run_async(
        user_id="u",
        session_id=session.id,
        invocation_id="inv-1",
        new_message=types.Content(role="user", parts=[types.Part(text="hello")]),
    ):
        pass

    fresh = await ss.get_session(app_name="app", user_id="u", session_id=session.id)
    print(len([e for e in fresh.events if e.author == "user" and e.invocation_id == "inv-1"]))  # -> 2 on 1.25.0

asyncio.run(main())
Observed:

user events for inv-1: 2 (duplicate append)
Expected:

1 user event for the retried invocation (idempotent append)
I also validated the fix branch: same repro returns 1 user event, and targeted tests pass.