fix(confirmation): run a confirmed tool once per approval

src/google/adk/flows/llm_flows/tools/_confirmation.py

@@ -306,22 +306,42 @@ async def run_async(
     # the session and the toolset have moved on since the approval, so the
     # strict checks in `_resolve_confirmation_targets` can now legitimately fail
     # and abort the invocation.
+    #
+    # The same confirmation response can also arrive again in a later user
+    # event (a client retry, a resubmitted form). The tool result from the
+    # first resume then sits before the last user event, so look for results
+    # after the first user event that answered each confirmation instead.
+    # The "requires confirmation" placeholder result is emitted before that
+    # answer, so it does not count.
     confirmation_to_original_fc_id = _map_confirmation_to_original_fc_ids(
         events, set(confirmations_by_fc_id.keys())
     )
-    responded_fc_ids: set[str] = set()
-    for event in reversed(events):
-      if event.author == "user":
-        break
+    first_answer_index: dict[str, int] = {}
+    last_response_index: dict[str, int] = {}
+    for i, event in enumerate(events):
       for function_response in event.get_function_responses():
-        if function_response.id:
-          responded_fc_ids.add(function_response.id)
+        if not function_response.id:
+          continue
+        if event.author == "user":
+          if function_response.id in confirmations_by_fc_id:
+            first_answer_index.setdefault(function_response.id, i)
+        else:
+          last_response_index[function_response.id] = i
+
+    consumed_confirmation_fc_ids: set[str] = set()
+    for confirmation_fc_id in confirmations_by_fc_id:
+      original_fc_id = confirmation_to_original_fc_id.get(confirmation_fc_id)
+      if (
+          original_fc_id
+          and last_response_index.get(original_fc_id, -1)
+          > first_answer_index[confirmation_fc_id]
+      ):
+        consumed_confirmation_fc_ids.add(confirmation_fc_id)
 
     confirmations_by_fc_id = {
         confirmation_fc_id: confirmation
         for confirmation_fc_id, confirmation in confirmations_by_fc_id.items()
-        if confirmation_to_original_fc_id.get(confirmation_fc_id)
-        not in responded_fc_ids
+        if confirmation_fc_id not in consumed_confirmation_fc_ids
     }
 
     if not confirmations_by_fc_id:

tests/unittests/runners/test_run_tool_confirmation.py

@@ -24,12 +24,18 @@
 from google.adk.agents.sequential_agent import SequentialAgent
 from google.adk.agents.sequential_agent import SequentialAgentState
 from google.adk.apps.app import App
+from google.adk.apps.app import EventsCompactionConfig
 from google.adk.apps.app import ResumabilityConfig
+from google.adk.apps.base_events_summarizer import BaseEventsSummarizer
+from google.adk.events.event import Event
+from google.adk.events.event_actions import EventActions
+from google.adk.events.event_actions import EventCompaction
 from google.adk.events.ui_widget import UiWidget
 from google.adk.flows.llm_flows.functions import REQUEST_CONFIRMATION_FUNCTION_CALL_NAME
 from google.adk.tools.function_tool import FunctionTool
 from google.adk.tools.tool_context import ToolContext
 from google.adk.utils.context_utils import Aclosing
+from google.genai.types import Content
 from google.genai.types import FunctionCall
 from google.genai.types import FunctionResponse
 from google.genai.types import GenerateContentResponse
@@ -1120,3 +1126,145 @@ def _sibling_tool(tool_context: ToolContext) -> dict[str, str]:
         gated_tool.name: expected_gated_response,
         sibling_tool.name: {"result": "sibling ran"},
     }]
+
+
+class TestHITLConfirmationReplay:
+  """Tests delivering the same confirmation response more than once."""
+
+  @pytest.mark.parametrize("confirmed", [True, False])
+  @pytest.mark.asyncio
+  async def test_repeated_confirmation_does_not_rerun_tool(
+      self, confirmed: bool
+  ):
+    """A confirmation already acted on does not run the tool again.
+
+    The same response delivered again (a client retry, a resubmitted form)
+    becomes the latest user event, so the tool result from the first resume
+    sits before it. The call is not executed or rejected a second time; it
+    asks for a fresh confirmation instead.
+    """
+    tool_calls = []
+
+    def _transfer(tool_context: ToolContext) -> dict[str, str]:
+      tool_calls.append(tool_context.function_call_id)
+      return {"result": "transferred"}
+
+    tool = FunctionTool(func=_transfer, require_confirmation=True)
+    mock_model = testing_utils.MockModel(
+        responses=[
+            _create_llm_response_from_tools([tool]),
+            _create_llm_response_from_text("response after first resume"),
+            _create_llm_response_from_text("response after replay"),
+        ]
+    )
+    agent = LlmAgent(name="root_agent", model=mock_model, tools=[tool])
+    # The repeated confirmation response is itself a second response to the
+    # same function call, which the invariant checker rejects by design.
+    runner = testing_utils.InMemoryRunner(
+        root_agent=agent, check_invariants=False
+    )
+
+    events = await runner.run_async(
+        testing_utils.UserContent("test user query")
+    )
+    user_confirmation = testing_utils.UserContent(
+        Part(
+            function_response=FunctionResponse(
+                id=events[1].content.parts[0].function_call.id,
+                name=REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+                response={"confirmed": confirmed},
+            )
+        )
+    )
+    await runner.run_async(user_confirmation)
+    assert len(tool_calls) == (1 if confirmed else 0)
+
+    events = await runner.run_async(user_confirmation)
+
+    assert len(tool_calls) == (1 if confirmed else 0)
+    tool_results = [
+        fr.response
+        for event in events
+        for fr in event.get_function_responses()
+        if fr.name == tool.name
+    ]
+    assert tool_results == [TOOL_CALL_ERROR_RESPONSE]
+    assert [
+        fc.name for event in events for fc in event.get_function_calls()
+    ] == [REQUEST_CONFIRMATION_FUNCTION_CALL_NAME]
+
+  @pytest.mark.asyncio
+  async def test_repeated_confirmation_after_compaction_does_not_rerun_tool(
+      self,
+  ):
+    """A confirmation replayed after compaction does not run the tool again.
+
+    Compaction only appends summary events; the original events stay in the
+    session, so the result of the first resume is still found.
+    """
+
+    class _SummarizeAll(BaseEventsSummarizer):
+
+      async def maybe_summarize_events(self, *, events: list[Event]) -> Event:
+        return Event(
+            author="user",
+            actions=EventActions(
+                compaction=EventCompaction(
+                    start_timestamp=events[0].timestamp,
+                    end_timestamp=events[-1].timestamp,
+                    compacted_content=Content(
+                        role="model", parts=[Part(text="summary")]
+                    ),
+                )
+            ),
+        )
+
+    tool_calls = []
+
+    def _transfer(tool_context: ToolContext) -> dict[str, str]:
+      tool_calls.append(tool_context.function_call_id)
+      return {"result": "transferred"}
+
+    tool = FunctionTool(func=_transfer, require_confirmation=True)
+    mock_model = testing_utils.MockModel(
+        responses=[
+            _create_llm_response_from_tools([tool]),
+            _create_llm_response_from_text("response after first resume"),
+            _create_llm_response_from_text("response to follow-up"),
+            _create_llm_response_from_text("response after replay"),
+        ]
+    )
+    app = App(
+        name="test_app",
+        root_agent=LlmAgent(name="root_agent", model=mock_model, tools=[tool]),
+        events_compaction_config=EventsCompactionConfig(
+            summarizer=_SummarizeAll(),
+            compaction_interval=1,
+            overlap_size=0,
+        ),
+    )
+    runner = testing_utils.InMemoryRunner(app=app, check_invariants=False)
+
+    events = await runner.run_async(
+        testing_utils.UserContent("test user query")
+    )
+    user_confirmation = testing_utils.UserContent(
+        Part(
+            function_response=FunctionResponse(
+                id=events[1].content.parts[0].function_call.id,
+                name=REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+                response={"confirmed": True},
+            )
+        )
+    )
+    await runner.run_async(user_confirmation)
+    await runner.run_async(testing_utils.UserContent("follow-up"))
+    assert any(event.actions.compaction for event in runner.session.events)
+    assert len(tool_calls) == 1
+
+    events = await runner.run_async(user_confirmation)
+
+    assert len(tool_calls) == 1
+    assert [
+        fc.name for event in events for fc in event.get_function_calls()
+    ] == [REQUEST_CONFIRMATION_FUNCTION_CALL_NAME]