fix(code): skip tool interrupts once auto-approve is set (#4092)

Deep Agents Code no longer splits a turn into many runs when "approve
always" is enabled; gated tool interrupts are now suppressed at the
source.

---

When a user enables "approve always" in Deep Agents Code,
`HumanInTheLoopMiddleware` kept interrupting on every gated tool call
and the client auto-resolved each one. Each interrupt/resume cycle
starts a fresh agent run, so a single turn fragmented into many runs and
produced noisy, hard-to-parse traces.

The fix carries the auto-approve decision in run-scoped CLI context
(`CLIContext` / `CLIContextSchema.auto_approve`) rather than graph
state, and each `interrupt_on` config gains a `when` predicate
(`_should_interrupt_tool_call`) that reads `request.runtime.context`.
Once approve-always is in effect the predicate returns `False`, so the
middleware skips the interrupt entirely instead of pausing and being
patched on the client. The initial turn seeds the flag from session
settings (`app.py`), and `textual_adapter.py` refreshes it into context
on every stream iteration — so choosing "auto-approve all" mid-turn
propagates to the resuming stream and the remaining tool calls in that
same run also stop interrupting.

Auto-approve is sourced from run context rather than graph state for two
reasons: seeding state would require a first-turn `Command(update=...)`,
which the LangGraph API server rebuilds with `goto=None` and crashes
`_control_branch` on a fresh thread; and context is safer because the
model cannot self-approve by writing state. The predicate is fail-closed
— missing or malformed context still interrupts — and consumers accept
both a coerced `CLIContextSchema` (in-process) and a plain dict (over
the LangGraph API / RemoteGraph).

Made by [Open
SWE](https://openswe.vercel.app/agents/4b76ff9f-a307-5834-5471-a0d441f2a147)

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>

libs/code/deepagents_code/_cli_context.py

@@ -1,6 +1,7 @@
-"""Lightweight runtime context type for app model overrides.
+"""Lightweight runtime context types for the CLI agent graph.
 
-Extracted from `configurable_model` so hot-path modules (`app`,
+Carries per-run overrides (model swap/params, approval mode) passed via
+`context=`. Extracted from `configurable_model` so hot-path modules (`app`,
 `textual_adapter`) can import `CLIContext` without pulling in the langchain
 middleware stack.
 """
@@ -13,18 +14,40 @@ from typing import Any, TypedDict
 
 @dataclass
 class CLIContextSchema:
-    """Server-side schema for run-scoped CLI context."""
+    """Declared `context_schema` for the agent graph.
+
+    Registered via `context_schema=` when the graph is built, so LangGraph
+    coerces each run's `context=` payload into this dataclass — in-process,
+    `runtime.context` is a `CLIContextSchema` instance.
+
+    It exists alongside `CLIContext` (below) because the payload is shaped
+    differently on each side of the API boundary: in-process it is coerced to
+    this dataclass, but over the LangGraph API server (RemoteGraph) it is
+    serialized to JSON and arrives as a plain dict. Consumers
+    (`configurable_model._get_context`, `_should_interrupt_tool_call`)
+    therefore accept both shapes. `CLIContext` is the client-facing builder for
+    constructing that payload.
+
+    Fields mirror `CLIContext`; see its per-field docstrings for semantics.
+    """
 
     model: str | None = None
+
     model_params: dict[str, Any] = field(default_factory=dict)
+
     effective_model: str | None = None
 
+    auto_approve: bool = False
+
 
 class CLIContext(TypedDict, total=False):
-    """Runtime context passed via `context=` to the LangGraph graph.
+    """Client-facing builder for the per-run graph context payload.
 
-    Carries per-invocation overrides that `ConfigurableModelMiddleware`
-    reads from `request.runtime.context`.
+    Callers populate this and pass it via `context=` to `astream`/`ainvoke`.
+    `ConfigurableModelMiddleware` and the `interrupt_on` `when` predicate read
+    it from `request.runtime.context`. In-process LangGraph coerces it into
+    `CLIContextSchema` (the registered `context_schema`); over the API it stays
+    a plain dict — which is why consumers handle both shapes.
     """
 
     model: str | None
@@ -44,3 +67,12 @@ class CLIContext(TypedDict, total=False):
     resolved yet (e.g. credentials not configured), in which case nothing is
     recorded rather than a malformed spec.
     """
+
+    auto_approve: bool
+    """Whether gated tool calls should skip the human-approval interrupt.
+
+    Sourced from the client session (not graph state) so the model cannot
+    self-approve by writing state. The `interrupt_on` `when` predicate reads
+    this to suppress interrupts at the source when "approve always" is on,
+    avoiding the interrupt-then-auto-resolve round-trip.
+    """

libs/code/deepagents_code/agent.py

@@ -1008,6 +1008,55 @@ def _format_execute_description(
     return "\n".join(lines)
 
 
+def _is_auto_approve_enabled(value: object) -> bool:
+    """Return whether a context value explicitly enables auto-approve."""
+    return isinstance(value, bool) and value
+
+
+def _should_interrupt_tool_call(request: ToolCallRequest) -> bool:
+    """Decide whether a gated tool call should pause for human approval.
+
+    Returns `False` once the run context carries `auto_approve=True` so
+    `HumanInTheLoopMiddleware` skips the interrupt entirely. This avoids the
+    interrupt-then-auto-resolve pattern that previously split each turn into a
+    separate run after every tool call, producing noisy traces.
+
+    Auto-approve is read from the run-scoped `CLIContext` (set by the client)
+    rather than graph state. Sourcing it from state required seeding it with a
+    first-turn `Command(update=...)`, which the LangGraph API server rebuilds
+    with `goto=None` — crashing `_control_branch` on a fresh thread. Context
+    is also safer: the model cannot self-approve by writing state.
+
+    Args:
+        request: The pending tool call under review.
+
+    Returns:
+        `True` to interrupt for approval, `False` to auto-approve.
+    """
+    runtime = getattr(request, "runtime", None)
+    ctx = getattr(runtime, "context", None)
+    if isinstance(ctx, CLIContextSchema):
+        return not _is_auto_approve_enabled(ctx.auto_approve)
+    if isinstance(ctx, dict):
+        # Type-checked (not truthiness) check: over the JSON/RemoteGraph boundary a
+        # malformed payload (e.g. "yes", 1) must fail closed and interrupt, not
+        # silently auto-approve. Only a genuine boolean `True` suppresses.
+        return not _is_auto_approve_enabled(ctx.get("auto_approve"))
+    if ctx is not None:
+        # Context is present but neither expected shape. The registered
+        # `context_schema=CLIContextSchema` guarantees in-process coercion to
+        # that dataclass, and RemoteGraph delivers a dict — so this means the
+        # context-plumbing contract broke (likely an SDK change). Fail closed
+        # (interrupt), but surface it: otherwise auto-approve silently stops
+        # working with no error, looking like a feature that just "broke".
+        logger.warning(
+            "auto-approve predicate received unexpected context type %s; "
+            "interrupting for safety",
+            type(ctx).__name__,
+        )
+    return True
+
+
 def _add_interrupt_on() -> dict[str, InterruptOnConfig]:
     """Configure human-in-the-loop interrupt settings for all gated tools.
 
@@ -1016,42 +1065,53 @@ def _add_interrupt_on() -> dict[str, InterruptOnConfig]:
     delegation) is gated behind an approval prompt unless auto-approve
     is enabled.
 
+    Each config carries a `when` predicate so that enabling "approve always"
+    mid-session (carried in run-scoped context, not graph state) suppresses
+    the interrupt itself instead of relying on the client to auto-resolve it.
+
     Returns:
         Dictionary mapping tool names to their interrupt configuration.
     """
     execute_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": _format_execute_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
+        "when": _should_interrupt_tool_call,
     }
 
     write_file_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": _format_write_file_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
+        "when": _should_interrupt_tool_call,
     }
 
     edit_file_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": _format_edit_file_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
+        "when": _should_interrupt_tool_call,
     }
 
     web_search_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": _format_web_search_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
+        "when": _should_interrupt_tool_call,
     }
 
     fetch_url_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": _format_fetch_url_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
+        "when": _should_interrupt_tool_call,
     }
 
     task_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": _format_task_description,  # ty: ignore[invalid-argument-type]  # Callable description narrower than TypedDict expects
+        "when": _should_interrupt_tool_call,
     }
 
     async_subagent_interrupt_config: InterruptOnConfig = {
         "allowed_decisions": ["approve", "reject"],
         "description": "Launch, update, or cancel a remote async subagent.",
+        "when": _should_interrupt_tool_call,
     }
 
     interrupt_map: dict[str, InterruptOnConfig] = {
@@ -1075,6 +1135,7 @@ def _add_interrupt_on() -> dict[str, InterruptOnConfig]:
                 "window space. Recent messages are kept as-is. "
                 "Full history remains available for retrieval."
             ),
+            "when": _should_interrupt_tool_call,
         }
 
     return interrupt_map

libs/code/deepagents_code/app.py

@@ -7767,6 +7767,9 @@ class DeepAgentsApp(App):
                 image_tracker=self._image_tracker,
                 sandbox_type=self._sandbox_type,
                 message_kwargs=message_kwargs,
+                # `auto_approve` is intentionally omitted here: execute_task_textual
+                # writes it into this context from `session_state.auto_approve` at
+                # the top of every stream iteration, so seeding it would be dead.
                 context=CLIContext(
                     model=self._model_override,
                     model_params=self._model_params_override or {},

libs/code/deepagents_code/textual_adapter.py

@@ -47,7 +47,7 @@ if TYPE_CHECKING:
 
 
 from deepagents_code._ask_user_types import AskUserRequest
-from deepagents_code._cli_context import CLIContext  # noqa: TC001
+from deepagents_code._cli_context import CLIContext
 from deepagents_code._constants import SYSTEM_MESSAGE_PREFIX
 from deepagents_code._session_stats import (
     ModelStats as ModelStats,
@@ -401,8 +401,11 @@ async def execute_task_textual(
         adapter: The TextualUIAdapter for UI operations
         backend: Optional backend for file operations
         image_tracker: Optional tracker for images
-        context: Optional `CLIContext` with model override and params, passed
-            to the graph via `context=`.
+        context: Optional `CLIContext` with model override and params. The
+            current approval mode (`session_state.auto_approve`) is written
+            into `context["auto_approve"]` on every stream iteration before it
+            is passed to the graph via `context=`, so the `interrupt_on` `when`
+            predicate can suppress interrupts at the source.
         sandbox_type: Sandbox provider name for trace metadata, or `None`
             if no sandbox is active.
         message_kwargs: Extra fields merged into the stream input message
@@ -519,6 +522,10 @@ async def execute_task_textual(
     user_msg: dict[str, Any] = {"role": "user", "content": message_content}
     if message_kwargs:
         user_msg.update(message_kwargs)
+    # Auto-approve is carried via run context (set per stream iteration below),
+    # not graph state — so the initial input is a plain dict. A first-turn
+    # `Command(update=...)` would be rebuilt with `goto=None` by the LangGraph
+    # API server and crash `_control_branch` on a fresh thread.
     stream_input: dict | Command = {"messages": [user_msg]}
 
     # Track summarization lifecycle so spinner status and notification stay in sync.
@@ -531,6 +538,14 @@ async def execute_task_textual(
             pending_interrupts: dict[str, HITLRequest] = {}
             pending_ask_user: dict[str, AskUserRequest] = {}
 
+            # Carry the current approval mode into run context so the
+            # `interrupt_on` `when` predicate can suppress interrupts at the
+            # source. Refreshed each iteration so enabling "approve always"
+            # mid-turn propagates to the resuming stream.
+            if context is None:
+                context = CLIContext()
+            context["auto_approve"] = bool(session_state.auto_approve)
+
             # Show the Thinking spinner before each astream iteration so
             # both the first turn and HITL/ask_user resumes surface feedback
             # while the model processes input. Skip when
@@ -1253,6 +1268,12 @@ async def execute_task_textual(
 
                             if decision_type == "auto_approve_all":
                                 session_state.auto_approve = True
+                                # The resuming stream re-reads
+                                # `session_state.auto_approve` into run context
+                                # at the top of the loop, so the `interrupt_on`
+                                # `when` predicate suppresses interrupts on the
+                                # remaining tool calls in this turn — keeping it
+                                # a single run instead of resuming after each.
                                 if adapter._on_auto_approve_enabled:
                                     adapter._on_auto_approve_enabled()
                                 decisions = [

libs/code/tests/unit_tests/test_agent.py

@@ -1,6 +1,8 @@
 """Unit tests for agent formatting functions."""
 
+from dataclasses import fields
 from pathlib import Path
+from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import Mock, patch
 
@@ -11,9 +13,10 @@ from langchain_core.messages import AIMessage
 if TYPE_CHECKING:
     from langchain.agents.middleware.types import AgentState
     from langchain.messages import ToolCall
+    from langgraph.prebuilt.tool_node import ToolCallRequest
     from langgraph.runtime import Runtime
 
-from deepagents_code._cli_context import CLIContextSchema
+from deepagents_code._cli_context import CLIContext, CLIContextSchema
 from deepagents_code.agent import (
     DEFAULT_AGENT_NAME,
     _add_interrupt_on,
@@ -25,6 +28,7 @@ from deepagents_code.agent import (
     _format_web_search_description,
     _format_write_file_description,
     _sanitize_agent_message_name,
+    _should_interrupt_tool_call,
     build_model_identity_section,
     create_cli_agent,
     get_available_agent_names,
@@ -51,6 +55,100 @@ def test_add_interrupt_on_gates_async_task_tools() -> None:
         assert tool_name in interrupt_on
 
 
+def test_add_interrupt_on_attaches_auto_approve_predicate() -> None:
+    """Every gated tool carries the `when` predicate that honors auto-approve."""
+    interrupt_on = _add_interrupt_on()
+
+    assert interrupt_on
+    for config in interrupt_on.values():
+        assert config.get("when") is _should_interrupt_tool_call
+
+
+def _request_with_context(context: object) -> "ToolCallRequest":
+    return cast(
+        "ToolCallRequest",
+        SimpleNamespace(runtime=SimpleNamespace(context=context)),
+    )
+
+
+def test_should_interrupt_tool_call_respects_auto_approve_context() -> None:
+    """The predicate suppresses interrupts once auto-approve is in run context."""
+    # Dataclass-shaped context (in-process path).
+    assert _should_interrupt_tool_call(
+        _request_with_context(CLIContextSchema(auto_approve=False))
+    )
+    assert not _should_interrupt_tool_call(
+        _request_with_context(CLIContextSchema(auto_approve=True))
+    )
+    # Dict-shaped context (LangGraph API / RemoteGraph path).
+    assert _should_interrupt_tool_call(_request_with_context({"auto_approve": False}))
+    assert not _should_interrupt_tool_call(
+        _request_with_context({"auto_approve": True})
+    )
+
+
+def test_should_interrupt_tool_call_defaults_to_interrupting() -> None:
+    """Missing or malformed context must not auto-approve."""
+    assert _should_interrupt_tool_call(_request_with_context({}))
+    assert _should_interrupt_tool_call(_request_with_context(None))
+    assert _should_interrupt_tool_call(
+        cast("ToolCallRequest", SimpleNamespace(runtime=None))
+    )
+    assert _should_interrupt_tool_call(cast("ToolCallRequest", SimpleNamespace()))
+
+
+def test_should_interrupt_tool_call_truthy_non_bool_fails_closed() -> None:
+    """A truthy non-bool context value must interrupt, not auto-approve.
+
+    Context can arrive as a dataclass after in-process `context_schema`
+    coercion, or as a dict from the JSON/RemoteGraph boundary. A malformed
+    value in either shape must not slip past the gate on mere truthiness.
+    """
+    assert _should_interrupt_tool_call(
+        _request_with_context(CLIContextSchema(auto_approve=cast("Any", 1)))
+    )
+    assert _should_interrupt_tool_call(
+        _request_with_context(CLIContextSchema(auto_approve=cast("Any", "yes")))
+    )
+    assert _should_interrupt_tool_call(_request_with_context({"auto_approve": 1}))
+    assert _should_interrupt_tool_call(_request_with_context({"auto_approve": "yes"}))
+
+
+def test_should_interrupt_tool_call_warns_on_unexpected_context_shape(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A non-None context that is neither shape interrupts and logs a warning.
+
+    Reaching this branch means the `context_schema` coercion contract broke
+    (likely an SDK change); fail closed but surface it so a silently-degraded
+    auto-approve is observable instead of looking like a feature that "broke".
+    """
+    with caplog.at_level("WARNING", logger="deepagents_code.agent"):
+        assert _should_interrupt_tool_call(_request_with_context("garbage"))
+        assert _should_interrupt_tool_call(
+            _request_with_context(SimpleNamespace(auto_approve=True))
+        )
+
+    assert "unexpected context type" in caplog.text
+    # A legitimate absent context must stay silent — not every default is an anomaly.
+    caplog.clear()
+    with caplog.at_level("WARNING", logger="deepagents_code.agent"):
+        assert _should_interrupt_tool_call(_request_with_context(None))
+    assert "unexpected context type" not in caplog.text
+
+
+def test_cli_context_field_parity() -> None:
+    """`CLIContext` and `CLIContextSchema` must declare the same field set.
+
+    The two types model the same run-context payload on opposite sides of the
+    API boundary; the docstrings note "fields mirror" but nothing structural
+    enforces it. This locks in parity so a field added to one is added to both.
+    """
+    typed_dict_keys = set(CLIContext.__annotations__)
+    dataclass_keys = {f.name for f in fields(CLIContextSchema)}
+    assert typed_dict_keys == dataclass_keys
+
+
 def test_sanitize_agent_message_name_replaces_provider_unsafe_chars() -> None:
     """Agent display names with spaces must become valid message names."""
     assert _sanitize_agent_message_name("my agent") == "my_agent"

libs/code/tests/unit_tests/test_textual_adapter.py

@@ -771,20 +771,133 @@ class _SequencedAgent:
     def __init__(self, streams_by_call: list[list[tuple[Any, ...]]]) -> None:
         self._streams_by_call = streams_by_call
         self.stream_inputs: list[dict | Command] = []
+        self.contexts: list[Any] = []
 
     async def astream(
         self,
         stream_input: dict | Command,
         *_: Any,
+        context: object = None,
         **__: Any,
     ) -> AsyncIterator[tuple[Any, ...]]:
-        """Yield chunks for this invocation and record stream inputs."""
+        """Yield chunks for this invocation and record stream inputs/context.
+
+        `execute_task_textual` mutates a single `context` dict in place across
+        stream iterations (production reads the value at each call), so snapshot
+        a copy here to capture the per-iteration state rather than aliasing the
+        final mutation.
+        """
         self.stream_inputs.append(stream_input)
+        self.contexts.append(dict(context) if isinstance(context, dict) else context)
         chunks = self._streams_by_call.pop(0) if self._streams_by_call else []
         for chunk in chunks:
             yield chunk
 
 
+class TestExecuteTaskTextualAutoApproveInput:
+    """Auto-approve must ride on run context, never a first-turn `Command`."""
+
+    async def test_pre_enabled_auto_approve_uses_plain_dict_and_context(self) -> None:
+        """A fresh turn sends a plain dict input; auto-approve rides on context.
+
+        A first-turn `Command(update=...)` is rebuilt with `goto=None` by the
+        LangGraph API server's `map_cmd`, crashing `_control_branch` on a fresh
+        thread. The flag must travel via run context instead.
+        """
+        agent = _SequencedAgent([[]])
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=_mock_approval,
+        )
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=agent,
+            assistant_id="assistant",
+            session_state=SimpleNamespace(thread_id="thread-1", auto_approve=True),
+            adapter=adapter,
+        )
+
+        stream_input = agent.stream_inputs[0]
+        assert not isinstance(stream_input, Command)
+        assert stream_input == {"messages": [{"role": "user", "content": "hi"}]}
+        assert agent.contexts[0]["auto_approve"] is True
+
+    async def test_mid_turn_auto_approve_all_propagates_to_resume_context(
+        self,
+    ) -> None:
+        """Choosing "auto-approve all" mid-turn flips the resuming stream's context.
+
+        The PR's headline behavior: iteration 1 interrupts for approval, the
+        user picks `auto_approve_all`, and the per-iteration context refresh
+        re-reads `session_state.auto_approve` so iteration 2 (the resume)
+        carries `auto_approve=True`. Guards against hoisting the refresh out of
+        the stream loop (which would leave the first-iteration value frozen and
+        keep interrupting the rest of the turn).
+        """
+        action_requests = [{"name": "execute", "args": {"command": "echo hi"}}]
+        agent = _SequencedAgent(
+            streams_by_call=[
+                [
+                    (
+                        (),
+                        "messages",
+                        (
+                            _tool_call_message(
+                                "execute", {"command": "echo hi"}, "tool-1"
+                            ),
+                            {},
+                        ),
+                    ),
+                    _hitl_interrupt_chunk(
+                        {
+                            "action_requests": action_requests,
+                            "review_configs": [
+                                {
+                                    "action_name": "execute",
+                                    "allowed_decisions": ["approve", "reject"],
+                                }
+                            ],
+                        }
+                    ),
+                ],
+                [],
+            ]
+        )
+
+        async def request_approval(
+            _action_requests: list[dict[str, Any]],
+            _assistant_id: str | None,
+        ) -> asyncio.Future[object]:
+            await asyncio.sleep(0)
+            future: asyncio.Future[object] = asyncio.Future()
+            future.set_result({"type": "auto_approve_all"})
+            return future
+
+        adapter = TextualUIAdapter(
+            mount_message=_mock_mount,
+            update_status=_noop_status,
+            request_approval=request_approval,
+        )
+        session_state = SimpleNamespace(thread_id="thread-1", auto_approve=False)
+
+        await execute_task_textual(
+            user_input="hi",
+            agent=agent,
+            assistant_id="assistant",
+            session_state=session_state,
+            adapter=adapter,
+        )
+
+        # Two stream iterations: the initial turn and the resume after the
+        # decision. The flag must flip between them, not stay frozen.
+        assert len(agent.contexts) == 2
+        assert agent.contexts[0]["auto_approve"] is False
+        assert agent.contexts[1]["auto_approve"] is True
+        assert session_state.auto_approve is True
+
+
 def _ask_user_interrupt_chunk(payload: dict[str, Any]) -> tuple[Any, ...]:
     """Build an updates-stream chunk containing one ask_user interrupt."""
     interrupt = SimpleNamespace(id="interrupt-1", value=payload)