fix(code): avoid redundant Auto approval prompts

libs/code/deepagents_code/_ask_user_types.py

@@ -69,6 +69,20 @@ class AskUserRequest(TypedDict):
     """ID of the originating tool call, used to route the response back."""
 
 
+ASK_USER_AUTHORIZATION_METADATA_KEY = "deepagents_code_ask_user_authorization"
+MAX_ASK_USER_AUTHORIZATION_ANSWER_CHARS = 4000
+
+
+class AskUserAuthorizationReceipt(TypedDict):
+    """Trusted same-turn authorization recorded after an ask_user response."""
+
+    version: Literal[1]
+    thread_id: str
+    turn_id: str
+    tool_call_id: str
+    answers: list[str]
+
+
 class AskUserAnswered(TypedDict):
     """Widget result when the user submits answers."""
 

libs/code/deepagents_code/_cli_context.py

@@ -47,6 +47,8 @@ class CLIContextSchema:
 
     thread_id: str | None = None
 
+    turn_id: str | None = None
+
     offload_tool_call_id: str | None = None
 
 
@@ -96,6 +98,9 @@ class CLIContext(TypedDict, total=False):
     session-affinity headers.
     """
 
+    turn_id: str | None
+    """Current user-turn ID for binding trusted interactive responses."""
+
     offload_tool_call_id: str | None
     """The sole tool-call ID authorized during a server-driven `/offload` run.
 

libs/code/deepagents_code/agent.py

@@ -2595,10 +2595,13 @@ def _subagent_cli_middleware(
     agent_middleware.extend([ResumeStateMiddleware(), GoalToolsMiddleware()])
 
     # Add ask_user middleware (must be early so its tool is available)
+    trusted_ask_user_tool: BaseTool | None = None
     if enable_ask_user:
         from deepagents_code.ask_user import AskUserMiddleware
 
-        agent_middleware.append(AskUserMiddleware())
+        ask_user_middleware = AskUserMiddleware()
+        agent_middleware.append(ask_user_middleware)
+        trusted_ask_user_tool = ask_user_middleware.tools[0]
 
     # Add memory middleware
     if enable_memory:
@@ -2796,13 +2799,12 @@ def _subagent_cli_middleware(
         )
 
     interrupt_on: dict[str, bool | InterruptOnConfig] | None
+    auto_mode_config: tuple[Path, list[str]] | None = None
     if resolved_interrupt_on is None:
         interrupt_on = {}
     else:
         interrupt_on = resolved_interrupt_on  # ty: ignore[invalid-assignment]  # InterruptOnConfig is compatible at runtime
         if auto_mode_enabled:
-            from deepagents_code.auto_mode import AutoModeHITLMiddleware
-
             configured_allow_list = shell_allow_list or settings.shell_allow_list
             narrow_allow_list = (
                 configured_allow_list if isinstance(configured_allow_list, list) else []
@@ -2813,13 +2815,7 @@ def _subagent_cli_middleware(
                 and project_context.project_root is not None
                 else effective_cwd or Path.cwd()
             )
-            agent_middleware.append(
-                AutoModeHITLMiddleware(
-                    resolved_interrupt_on,
-                    worktree_root=trusted_root,
-                    shell_allow_list=narrow_allow_list,
-                )
-            )
+            auto_mode_config = (Path(trusted_root), narrow_allow_list)
 
     # Set up composite backend with routing.
     if sandbox is None:
@@ -2861,6 +2857,21 @@ def _subagent_cli_middleware(
             routes={},
         )
 
+    compaction_middleware = _create_cli_compaction_middleware(model, composite_backend)
+    if auto_mode_config is not None and resolved_interrupt_on is not None:
+        from deepagents_code.auto_mode import AutoModeHITLMiddleware
+
+        trusted_root, narrow_allow_list = auto_mode_config
+        agent_middleware.append(
+            AutoModeHITLMiddleware(
+                resolved_interrupt_on,
+                worktree_root=trusted_root,
+                shell_allow_list=narrow_allow_list,
+                trusted_ask_user_tool=trusted_ask_user_tool,
+                trusted_compaction_tool=compaction_middleware.tools[0],
+            )
+        )
+
     if fs_tools is not None:
         # `fs_tools` is an explicit allowlist here (`--allow-fs-tools all` and an
         # omitted flag both arrive as `None`, leaving the SDK default in place).
@@ -2931,7 +2942,7 @@ def _subagent_cli_middleware(
             GoalCriteriaMiddleware(criteria_agent, criteria_fallback_agent)
         )
 
-    agent_middleware.append(_create_cli_compaction_middleware(model, composite_backend))
+    agent_middleware.append(compaction_middleware)
 
     grader_context_tools = _normalize_rubric_grader_context_tools(
         rubric_grader_tools or ()

libs/code/deepagents_code/ask_user.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Annotated, Any, cast
 
 if TYPE_CHECKING:
@@ -16,12 +17,18 @@
     ModelResponse,
     ResponseT,
 )
-from langchain.tools import InjectedToolCallId
-from langchain_core.messages import AIMessage, SystemMessage, ToolMessage
+from langchain.tools import InjectedToolCallId, ToolRuntime
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
 from langchain_core.tools import tool
 from langgraph.types import Command, interrupt
 
-from deepagents_code._ask_user_types import AskUserRequest, Question
+from deepagents_code._ask_user_types import (
+    ASK_USER_AUTHORIZATION_METADATA_KEY,
+    MAX_ASK_USER_AUTHORIZATION_ANSWER_CHARS,
+    AskUserAuthorizationReceipt,
+    AskUserRequest,
+    Question,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,10 +104,46 @@ def _validate_questions(questions: list[Question]) -> None:
             raise ValueError(msg)
 
 
+def _context_string(context: object, name: str) -> str | None:
+    value = (
+        context.get(name)
+        if isinstance(context, Mapping)
+        else getattr(context, name, None)
+    )
+    return value if isinstance(value, str) and value else None
+
+
+def _execution_thread_id(runtime: object) -> str | None:
+    execution_info = getattr(runtime, "execution_info", None)
+    thread_id = getattr(execution_info, "thread_id", None)
+    return thread_id if isinstance(thread_id, str) and thread_id else None
+
+
+def _active_turn_id(runtime: object) -> str | None:
+    from deepagents_code.auto_mode import USER_PROMPT_METADATA_KEY
+
+    state = getattr(runtime, "state", None)
+    messages = state.get("messages") if isinstance(state, Mapping) else None
+    if not isinstance(messages, list):
+        return None
+    for message in reversed(messages):
+        if not isinstance(message, HumanMessage):
+            continue
+        metadata = message.additional_kwargs.get(USER_PROMPT_METADATA_KEY)
+        if not isinstance(metadata, Mapping):
+            return None
+        turn_id = metadata.get("turn_id")
+        return turn_id if isinstance(turn_id, str) and turn_id else None
+    return None
+
+
 def _parse_answers(
     response: object,
     questions: list[Question],
     tool_call_id: str,
+    *,
+    thread_id: str | None = None,
+    turn_id: str | None = None,
 ) -> Command[Any]:
     """Parse an interrupt response into a `Command` with a `ToolMessage`.
 
@@ -117,12 +160,15 @@ def _parse_answers(
         response: Raw value returned by `interrupt()`.
         questions: The questions that were asked.
         tool_call_id: Originating tool call ID for the `ToolMessage`.
+        thread_id: Trusted runtime thread identity.
+        turn_id: Trusted runtime user-turn identity.
 
     Returns:
         `Command` containing a formatted `ToolMessage` with Q&A pairs.
     """
     status: str = "answered"
     error_text: str | None = None
+    answers_are_strings = False
     answers: list[str]
     if not isinstance(response, dict):
         logger.error(
@@ -153,6 +199,9 @@ def _parse_answers(
         else:
             raw_answers = response_dict["answers"]
             if isinstance(raw_answers, list):
+                answers_are_strings = all(
+                    isinstance(answer, str) for answer in raw_answers
+                )
                 answers = [str(answer) for answer in raw_answers]
             else:
                 logger.error(
@@ -190,14 +239,41 @@ def _parse_answers(
         detail = error_text or "ask_user interaction failed"
         answers = [f"(error: {detail})" for _ in questions]
 
+    additional_kwargs: dict[str, object] = {}
+    if (
+        status == "answered"
+        and answers_are_strings
+        and len(answers) == len(questions)
+        and all(
+            len(answer) <= MAX_ASK_USER_AUTHORIZATION_ANSWER_CHARS for answer in answers
+        )
+        and thread_id is not None
+        and turn_id is not None
+    ):
+        receipt = AskUserAuthorizationReceipt(
+            version=1,
+            thread_id=thread_id,
+            turn_id=turn_id,
+            tool_call_id=tool_call_id,
+            answers=list(answers),
+        )
+        additional_kwargs[ASK_USER_AUTHORIZATION_METADATA_KEY] = receipt
+
     formatted_answers = []
-    for i, q in enumerate(questions):
+    for i, question in enumerate(questions):
         answer = answers[i] if i < len(answers) else "(no answer)"
-        formatted_answers.append(f"Q: {q['question']}\nA: {answer}")
+        formatted_answers.append(f"Q: {question['question']}\nA: {answer}")
     result_text = "\n\n".join(formatted_answers)
     return Command(
         update={
-            "messages": [ToolMessage(result_text, tool_call_id=tool_call_id)],
+            "messages": [
+                ToolMessage(
+                    result_text,
+                    name="ask_user",
+                    tool_call_id=tool_call_id,
+                    additional_kwargs=additional_kwargs,
+                )
+            ],
         }
     )
 
@@ -232,12 +308,14 @@ def __init__(
         def _ask_user(
             questions: list[Question],
             tool_call_id: Annotated[str, InjectedToolCallId],
+            runtime: ToolRuntime[Any, Any],
         ) -> Command[Any]:
             """Ask the user one or more questions.
 
             Args:
                 questions: Questions to present to the user.
                 tool_call_id: Tool call identifier injected by LangChain.
+                runtime: Trusted graph runtime for thread and turn identity.
 
             Returns:
                 `Command` containing the parsed user answers as a `ToolMessage`.
@@ -254,7 +332,25 @@ def _ask_user(
             # GraphBubbleUp — a broad `except Exception` (e.g. ToolRetryMiddleware)
             # would swallow this interrupt and silently break ask_user.
             response = interrupt(ask_request)
-            return _parse_answers(response, questions, tool_call_id)
+            execution_thread_id = _execution_thread_id(runtime)
+            context_thread_id = _context_string(runtime.context, "thread_id")
+            context_turn_id = _context_string(runtime.context, "turn_id")
+            active_turn_id = _active_turn_id(runtime)
+            runtime_tool_call_id = runtime.tool_call_id
+            return _parse_answers(
+                response,
+                questions,
+                tool_call_id,
+                thread_id=(
+                    execution_thread_id
+                    if execution_thread_id == context_thread_id
+                    and runtime_tool_call_id == tool_call_id
+                    else None
+                ),
+                turn_id=(
+                    context_turn_id if context_turn_id == active_turn_id else None
+                ),
+            )
 
         _ask_user.name = "ask_user"
         self.tools = [_ask_user]