Issue in Tool Confirmation

[Opty0602]

🔴 Required Information

Please ensure all items in this section are completed to allow for efficient
triaging. Requests without complete information may be rejected / deprioritized.
If an item is not applicable to you - please mark it as N/A

Describe the Bug:
In adk 1.0 version the tool confirmation worked as intended, but in 2.0 when agent invokes a tool which has Tool Confirmation, resuming the flow makes the whole invocation to happen again and again.
So i have this architecture sequential agent (planner -> executor -> validator) with first two agent having request_human_input tool containing ToolConfirmation

Steps to Reproduce:
provided the agent.py file and test_runner.py file.

1. install google adk 1.*
2. python test_runner.py

adk web as is not working as intended for both (adk 1.* and 2.*)

Expected Behavior:
Since PlannerAgent, ExecutorAgent and ValidatorAgent are in sequential manner, with first two agent having request_human_input tool with tool_confirmation implemented. when user confirms the tool and provide no feedback, the control should be passed to the next agent. if feedback given and confirm tool is false then the current agent should refine its answer and recall this human tool to interact with user via tool call.
Observed Behavior:

In google adk 2.0, instead of passing control over to next agent, the planner agent keeps on rerunning and re invoking this tool. (i tried passing invocation_id in the runner for adk 2.0 as well as resumability config is set to True, still it was not working)
Note:
i didn't pass invocation_id in runner.run_async in adk 1.*, yet it was working fine. I have a FastAPI version of this backend which is perfectly working, but to reproduce i created test_runner.py.

Environment Details:

• ADK Library Version (pip show google-adk): 2.2.0
• Desktop OS: linux
• Python Version (python -V): 3.12

Model Information:

• Are you using LiteLLM: No
• Which model is being used: gemini-2.5-flash

---

🟡 Optional Information

Regression:
Yes it worked in google adk 1.*

Minimal Reproduction Code:
agent.py

"""
================================================================================
  SequentialConfirmationAgent — Agent Definitions
  Standard Template v1.0
================================================================================
  Workflow:
    User Message
       │
       ▼
    PlannerAgent  ──► request_human_approval(task_name="plan")
       │                └─ pauses until user approves/rejects
       ▼
    ExecutorAgent ──► request_human_approval(task_name="execution_result")
       │                └─ reads state["plan"], pauses until user approves/rejects
       ▼
    ValidatorAgent──► request_human_approval(task_name="validation_result")
                       └─ reads state["execution_result"], pauses until user approves/rejects

  State keys written by `request_human_approval` on approval:
    state["plan"]              → set by PlannerAgent
    state["execution_result"]  → set by ExecutorAgent
    state["validation_result"] → set by ValidatorAgent
    NOTE: Since we aren't using `output_key` parameter of llmAgent() or Agent() we set the agent responses through the tool itself.

  These keys are referenced in the agent instructions using ADK's
  {key?} template syntax (the `?` makes the substitution optional,
  so the agent doesn't error if the key isn't populated yet).

  Human-in-the-Loop (HITL) Mechanism:
    - `request_human_approval` is a standard Python function.
    - ADK injects `tool_context: ToolContext` automatically — the agent
      does NOT see or control this parameter.
    - On first call: tool_context.tool_confirmation is None →
        tool_context.request_confirmation(...) is called → ADK pauses
        the entire invocation and emits an "adk_request_confirmation"
        function call event upstream (caught by main.py's event_generator).
    - On resume: tool_context.tool_confirmation is populated with the
        user's response → the function proceeds to approved/rejected branch.

  skip_summarization Note:
    - `tool_context.actions.skip_summarization = True` prevents ADK from
      making an extra LLM call to summarise the tool's return value.
    - It is only set on the APPROVED path — rejection allows the agent to
      re-reason about the feedback naturally.
    - Known ADK behaviour: skip_summarization is global per-step, so setting
      it in one tool affects all tools in that same step. This is low-risk
      here since each agent calls request_human_approval exactly once per turn.
================================================================================
"""

from google.adk.agents import LlmAgent
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.tools import FunctionTool, AgentTool
from google.adk.tools.tool_context import ToolContext


# ==============================================================================
# SECTION 1 — Shared Tool: request_human_approval
#
# This is the single HITL gate used by all three agents.
# It is wrapped once as a FunctionTool and reused across all agent definitions
# for consistency (avoids the mixed raw-function vs FunctionTool pattern).
# ==============================================================================

def request_human_approval(tool_context: ToolContext, task_name: str, response: str):
    """
    Pauses agent execution and requests human approval for a completed task.

    ADK intercepts `tool_context` and injects it automatically — the LLM
    never sees this parameter in the tool's schema.

    Args:
        tool_context: Injected by ADK. Provides access to confirmation state,
                      session state, and action flags.
        task_name:    Logical name of the task being approved.
                      Also used as the state key on approval
                      (e.g., "plan", "execution_result", "validation_result").
        response:     The full output to present to the human for review.

    Returns:
        dict: A status message indicating the outcome of the approval request.
    """
    # --- First call: no prior confirmation exists → request one ---
    if not tool_context.tool_confirmation:
        tool_context.request_confirmation(
            hint=f"Please review and approve the {task_name}.",
            payload={
                "response": response,
                "feedback": None,
            },
        )
        # ADK pauses the invocation here. This return value is never seen
        # by the LLM because the invocation is suspended before summarization.
        return {"status": f"Awaiting {task_name} approval."}

    # --- Resume call: tool_confirmation is populated with the user's response ---
    if tool_context.tool_confirmation.confirmed:
        # APPROVED: persist the response in session state for downstream agents.
        # The state key matches the {task_name?} template used in agent instructions.
        tool_context.state[task_name] = response

        # skip_summarization = True means ADK will NOT make an extra LLM call
        # to narrate this tool result. The agent moves on directly to the next step.
        # Only set on the approved path — rejection lets the agent re-reason with feedback.
        tool_context.actions.skip_summarization = True
        return {"status": f"{task_name} approved."}

    else:
        # REJECTED: return feedback (if any) so the agent can revise and resubmit.
        feedback = tool_context.tool_confirmation.payload.get("feedback")
        if feedback:
            return {"status": f"{task_name} rejected. Feedback: {feedback}"}
        return {"status": f"{task_name} rejected. Please revise and resubmit."}


# Wrap once as FunctionTool — reused by all three agents below.
# Using FunctionTool explicitly across all agents is the consistent, recommended
# pattern. Passing a raw function also works (ADK auto-wraps it), but explicit
# is clearer in a shared codebase / team standard template.
approval_tool = FunctionTool(func=request_human_approval)


# ==============================================================================
# SECTION 2 — Agent Definitions
# ==============================================================================

# ------------------------------------------------------------------------------
# Agent 1: PlannerAgent
#
# Responsibility: Produce a detailed plan for the user's problem.
# State written: state["plan"] (on approval)
# ------------------------------------------------------------------------------
planner_agent = LlmAgent(
    name="PlannerAgent",
    model="gemini-2.5-flash",
    instruction="""
You are first agent in the Sequential agent execution flow (PlannerAgent -> ExecutorAgent -> ValidatorAgent)
Your task is to create a detailed plan to solve the user's problem.

Once your plan is complete, you MUST call the `request_human_approval` tool with:
  - task_name = "plan"
  - response  = the full plan text

Do not engage in conversation or explain your reasoning outside the tool call.
If your plan is rejected with feedback, revise it to incorporate the feedback
and call `request_human_approval` again with the updated plan.

Your response will be seed value for next agent to work with.
""",
    tools=[approval_tool],
    # include_contents='none'
)

# ------------------------------------------------------------------------------
# Agent 2: ExecutorAgent
#
# Responsibility: Execute the approved plan from PlannerAgent.
# State read:    state["plan"]       ← injected via {plan?} template
# State written: state["execution_result"] (on approval)
#
# NOTE: The template variable {plan?} MUST match the state key written by
#       PlannerAgent ("plan"). The `?` suffix makes it optional so ADK
#       does not error if the key is absent (e.g., on a fresh session).
# ------------------------------------------------------------------------------
executor_agent = LlmAgent(
    name="ExecutorAgent",
    model="gemini-2.5-flash",
    instruction="""
You are second agent in the Sequential agent execution flow (PlannerAgent -> ExecutorAgent -> ValidatorAgent)    
Your task is to execute the following approved plan:
---
{plan?}
---

Think through each step carefully, then call the `request_human_approval` tool with:
  - task_name = "execution_result"
  - response  = the full execution result

Do not engage in conversation or explain your reasoning outside the tool call.
If your result is rejected with feedback, revise it based on the feedback
and call `request_human_approval` again with the updated result.

Your response will be seed value for next agent to work with.
""",
    tools=[approval_tool],
    include_contents="none"
)

# ------------------------------------------------------------------------------
# Agent 3: ValidatorAgent
#
# Responsibility: Validate the execution result from ExecutorAgent.
# State read:    state["execution_result"] ← injected via {execution_result?} template
# State written: state["validation_result"] (on approval)
#
# NOTE: The template variable {execution_result?} MUST match the state key
#       written by ExecutorAgent ("execution_result").
#       Previous version used {TASK_EXECUTION?} which did NOT match the key
#       and would always inject an empty string. This is now corrected.
# ------------------------------------------------------------------------------
validator_agent = LlmAgent(
    name="ValidatorAgent",
    model="gemini-2.5-flash",
    instruction="""
You are last agent in the Sequential agent execution flow (PlannerAgent -> ExecutorAgent -> ValidatorAgent)     
Your task is to validate the following execution result as per the plan:

Plan:    
---
{plan?}    

Execution Result
---
{execution_result?}
---

Assess correctness, completeness, and quality. 

---

- If provided:
Then call the `request_human_approval` tool with:
  - task_name = "validation_result"
  - response  = your full validation report

Do not engage in conversation or explain your reasoning outside the tool call.
If your report is rejected, re-evaluate and resubmit an updated report
using `request_human_approval`.
---
""",
    # tools=[approval_tool],
    output_key="validation_result"
)


# ==============================================================================
# SECTION 3 — Sequential Workflow Assembly
#
# SequentialAgent runs sub_agents in order: Planner → Executor → Validator.
# Each agent is gated by a human approval step before the next one starts.
# ADK's ResumabilityConfig (set in main.py on the App object) ensures the
# workflow can pause and resume across HTTP requests without losing state.
# ==============================================================================

sequential_workflow = SequentialAgent(
    name="SequentialConfirmationAgent",
    description=(
        "A three-stage sequential workflow (Plan → Execute → Validate) "
        "with human approval gates between each stage."
        "requires stringified version for the provided manner: {'user_query':<user query>}"
    ),
    sub_agents=[planner_agent, executor_agent, validator_agent],
)

manager_agent = LlmAgent(
    name = "Manager",
    instruction=" you are manager agent, your task is to help user to get their work done, always acknowledge user before going ahead with any tool call/activity.",
    model = 'gemini-2.5-flash',
    sub_agents=[sequential_workflow]
    
                         )

# root_agent is the required ADK entry point.
# The Runner in app.py is initialized with the App that wraps this agent.
root_agent = manager_agent

test_runner.py
( i orignally implemented streaming with react UI, bellow is the close version i could get, the intent is clear that after planner agent, executor agent is getting the control and is doing its share of work.)

import asyncio
from google.adk.runners import Runner
from SequentialConfirmationAgent.agent import root_agent
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
from google.genai import types
from dotenv import load_dotenv
from pprint import pp
import logging
from google.adk.apps import ResumabilityConfig

# --- Configuration ---
APP_NAME = "SequentialConfirmationApp"
USER_ID = "test_user"
SESSION_ID = "test_session1"


async def main():
    """
    Runs the sequential confirmation agent, handling multiple turns of user approval.
    """
    # --- 1. Initialization ---
    load_dotenv(verbose=True)
    session_service = InMemorySessionService()
    await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    print("Session created successfully.\\n")
    runner = Runner(agent=root_agent, session_service=session_service, app_name=APP_NAME)
    runner.resumability_config = ResumabilityConfig(is_resumable=True)

    # --- 2. Turn-Based Conversation Loop ---
    # The conversation starts with the user's initial text prompt.
    # Each iteration of this loop represents a single "turn".
    # The `next_message` will be either the user's text or a FunctionResponse for a confirmation.
    # The loop continues as long as there is a message to send to the agent.
    
    initial_prompt = "create me a simple hello world website with a colorful ui."
    next_message = Content(parts=[Part(text=initial_prompt)], role="user")
    
    turn_counter = 1
    final_events = []
    

    while next_message:
        print(f"--- Turn {turn_counter}: Sending message to agent ---")
        if next_message.parts[0].function_response:
            print("Message type: FunctionResponse (User Approval)")
        else:
            print(f"Message type: TextPrompt ('{next_message.parts[0].text}')")
        print("-" * 50)

        events_this_turn = []
        invocation_id = None
        async for event in runner.run_async(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=next_message,
            # invocation_id = invocation_id
            ):
            events_this_turn.append(event)
            
        
        final_events.extend(events_this_turn) # Store the latest events for final processing
        # pp(events_this_turn)
        print(f"events generated turn {turn_counter} : ", len(events_this_turn))
        # After each turn, assume there is no next message unless we find a confirmation request.
        next_message = None

        # Inspect the results of the completed turn to see if a confirmation is needed.
        for event in events_this_turn:
            function_calls = event.get_function_calls()
            if function_calls and function_calls[0].name == "adk_request_confirmation":
                invocation_id = event.invocation_id
                agent_name = event.author
                task_hint = function_calls[0].args["toolConfirmation"]["hint"]
                print(f"\\n>>> Agent '{agent_name}' requested confirmation.")
                print(f">>> Hint: {task_hint}\\n")
                print(f"""Agent's Response: {function_calls[0].args['originalFunctionCall']["args"].get("response","")}""")

                # For this automated test, we always approve.
                confirmation_response = {
                    "confirmed": True,
                    "payload": function_calls[0].args["toolConfirmation"]["payload"]
                }

                # Construct the message for the *next* turn. This message contains the
                # user's (our script's) approval, which will resume the agent flow.
                next_message = Content(
                    parts=[
                        Part(function_response=types.FunctionResponse(
                            id=function_calls[0].id,
                            name="adk_request_confirmation",
                            response=confirmation_response
                        ))
                    ],
                    role="user"
                )
                break # A turn can only have one confirmation, so we can stop looking.
        
        if not next_message:
            # If we looped through all events and did not find a new confirmation request,
            # it means the agent has finished its work and the conversation is over.
            print("\\n--- Agent sequence complete. No more confirmations requested. ---")

        turn_counter += 1

    # --- 3. Final Output ---
    # The final agent response is in the last batch of events that were processed.
    print("\\n--- Final Agent Response ---")
    final_response_found = False
    for event in final_events:
        if event.content and event.content.role == "model" and event.content.parts[0].text:
            print(event.content.parts[0].text)
            final_response_found = True
            # break # Print the first text part from the model
    
    if not final_response_found:
        print("No final text response from agent.")

    # --- 4. Final State ---
    print("\\n--- Final Session State ---")
    session = await session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    pp(session.state)
    await runner.close()


if __name__ == "__main__":
    asyncio.run(main())

How often has this issue occurred?:

• Always (100%)

[rpelevin]

This looks like a resume-boundary bug in confirmation consumption, not just a sequential-agent routing problem.

For a tool confirmation, the approval response should be consumed against the parked original tool invocation exactly once. After an approved confirmation, the runtime should record that confirmation result before any model re-entry, then advance the original invocation cursor to the next agent or step. It should not ask the model to recreate the same confirmation tool call just because the prior call was pending when the run paused.

The invariant I would want in the core runner is:

1. The confirmation response is correlated to the original function-call id, session id, invocation id, tool name, and canonical arguments.
2. Once consumed, that confirmation has a terminal state for this resume attempt: approved, rejected, cancelled, or expired.
3. An approved no-feedback confirmation becomes the completed result for that parked tool call and cannot trigger the same confirmation request again.
4. A rejected confirmation with feedback may re-enter the same agent, but as a new revision path, not as a replay of the already-consumed confirmation.
5. A stale or mismatched confirmation response fails closed instead of being applied to a different pending call.

The regression I would add is a three-agent sequence with confirmation on the first two agents: approve the planner result, assert the executor starts next, and assert the planner confirmation tool is not invoked a second time. A second test should reject with feedback and prove the planner revises once with a fresh confirmation identity. A third should resume with a stale confirmation id and prove no downstream agent runs.

Boundary: architecture and test feedback only; no claim about using this project or running its code.