langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-22 01:35:28 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

fix(code): sync approval toggles during active runs (#4239)

Browse Source 
Approval mode now has a live control path for remote `dcode` sessions
instead of relying only on the run-context snapshot captured at stream
start. That lets mid-run toggles take effect before the next gated tool
call, while failing closed to manual approval when the live state cannot
be synced or read.

## Changes
- **Live approval state:** Added `approval_mode_key`,
`approval_mode_payload`, `read_approval_mode_from_store`, and
`awrite_approval_mode` to store per-thread auto-approve state in the
LangGraph Store under a deterministic hashed thread key.
- **Interrupt decisions:** Updated `_should_interrupt_tool_call` to
prefer the live store value over stale context, so toggling from
auto-approve back to manual can interrupt subsequent gated tool calls in
the same active run.
- **TUI synchronization:** Updated
`DeepAgentsApp.action_toggle_auto_approve` and
`_on_auto_approve_enabled` to write the live approval mode whenever the
user toggles approval behavior.
- **Fail-closed behavior:** If manual approval cannot sync while an
agent is running, the app warns the user and cancels the active run
rather than allowing stale auto-approval to continue.
- **Remote support:** Added `RemoteAgent.aput_store_item` so remote
sessions can write unindexed LangGraph Store records used by the
server-side approval predicate.
- **Context reconstruction:** Preserved `auto_approve` and
`approval_mode_key` when dict-based runtime context is reconstructed
into `CLIContextSchema`.
- **Stream propagation:** Updated `execute_task_textual` to write live
approval state before each stream iteration and await async auto-approve
callbacks triggered from approval prompts.
This commit is contained in:
Mason Daugherty
2026-06-24 23:12:40 -04:00
committed by GitHub
parent 649443a426
commit 4600365ea0
14 changed files with 934 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/code/deepagents_code/_cli_context.py
+11
View File
@@ -39,6 +39,8 @@ class CLIContextSchema:
auto_approve: bool = False
approval_mode_key: str | None = None
class CLIContext(TypedDict, total=False):
"""Client-facing builder for the per-run graph context payload.
@@ -76,3 +78,12 @@ class CLIContext(TypedDict, total=False):
this to suppress interrupts at the source when "approve always" is on,
avoiding the interrupt-then-auto-resolve round-trip.
"""
approval_mode_key: str | None
"""Store key for the live approval-mode control record.
The TUI updates this record when the user toggles approval mode mid-run.
The server-side interrupt predicate reads it from the LangGraph Store on
each gated tool call so auto-to-manual changes can take effect before the
current stream returns.
"""
libs/code/deepagents_code/agent.py
+39
View File
@@ -1030,6 +1030,38 @@ def _is_auto_approve_enabled(value: object) -> bool:
return isinstance(value, bool) and value
def _read_live_auto_approve(store: object, key: str | None) -> bool | None:
"""Return live approval mode from the LangGraph Store when configured.
Args:
store: `request.runtime.store` from the graph server.
key: Live approval-mode store key, or `None` when this run has no live
control record.
Returns:
`None` when no live key is configured for this run — the caller should
fall back to the static `auto_approve` context snapshot.
`True` or `False` when a live key is configured: these reflect
the stored mode, and `False` is also returned when the key
is configured but the store is unreadable (missing item,
malformed value, read error), so an unreadable live mode fails
closed and interrupts.
`None` therefore means "feature not in play," the opposite of the store
reader's `None` ("unreadable, be careful").
"""
if not key:
return None
from deepagents_code.approval_mode import read_approval_mode_from_store
value = read_approval_mode_from_store(store, key)
if value is None:
logger.warning(
"Approval-mode store item is unavailable; interrupting for safety"
)
return False
return value
def _should_interrupt_tool_call(request: ToolCallRequest) -> bool:
"""Decide whether a gated tool call should pause for human approval.
@@ -1052,9 +1084,16 @@ def _should_interrupt_tool_call(request: ToolCallRequest) -> bool:
"""
runtime = getattr(request, "runtime", None)
ctx = getattr(runtime, "context", None)
store = getattr(runtime, "store", None)
if isinstance(ctx, CLIContextSchema):
if (live := _read_live_auto_approve(store, ctx.approval_mode_key)) is not None:
return not live
return not _is_auto_approve_enabled(ctx.auto_approve)
if isinstance(ctx, dict):
raw_key = ctx.get("approval_mode_key")
key = raw_key if isinstance(raw_key, str) else None
if (live := _read_live_auto_approve(store, key)) is not None:
return not live
# 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.
libs/code/deepagents_code/app.py
+57 -2
View File
@@ -1351,6 +1351,7 @@ class TextualSessionState:
"""
self.auto_approve = auto_approve
self.thread_id = thread_id or _new_thread_id()
self.approval_mode_key: str | None = None
def reset_thread(self) -> str:
"""Reset to a new thread.
@@ -1359,6 +1360,7 @@ class TextualSessionState:
The new thread_id.
"""
self.thread_id = _new_thread_id()
self.approval_mode_key = None
return self.thread_id
@@ -5553,7 +5555,35 @@ class DeepAgentsApp(App):
)
await self._mount_approval_widget(menu, result_future)
def _on_auto_approve_enabled(self) -> None:
async def _write_live_approval_mode(self) -> bool:
"""Persist the current approval mode for the active thread.
Returns:
`True` when no write was needed or the write succeeded, otherwise
`False`.
"""
if self._session_state is None or self._agent is None:
return True
from deepagents_code.approval_mode import awrite_approval_mode
try:
live_key = await awrite_approval_mode(
self._agent,
self._session_state.thread_id,
auto_approve=bool(self._session_state.auto_approve),
)
except Exception:
self._session_state.approval_mode_key = None
logger.warning("Failed to write live approval-mode state", exc_info=True)
return False
self._session_state.approval_mode_key = live_key
return True
def _warn_live_approval_mode_unavailable(self, message: str) -> None:
"""Surface live approval-mode degradation to the user."""
self.notify(message, severity="warning", timeout=8, markup=False)
async def _on_auto_approve_enabled(self) -> None:
"""Handle auto-approve being enabled via the HITL approval menu.
Called when the user selects "Auto-approve all" from an approval
@@ -5566,6 +5596,11 @@ class DeepAgentsApp(App):
self._status_bar.set_auto_approve(enabled=True)
if self._session_state:
self._session_state.auto_approve = True
if not await self._write_live_approval_mode():
self._warn_live_approval_mode_unavailable(
"Auto-approve could not sync to the running agent; "
"approval prompts may continue."
)
async def _remove_ask_user_widget( # noqa: PLR6301 # Shared helper used by ask_user event handlers
self,
@@ -9315,7 +9350,7 @@ class DeepAgentsApp(App):
restore_iterm_cursor_guide()
super().exit(result=result, return_code=return_code, message=message)
def action_toggle_auto_approve(self) -> None:
async def action_toggle_auto_approve(self) -> None:
"""Toggle auto-approve mode for the current session.
When enabled, all tool calls (shell execution, file writes/edits,
@@ -9373,6 +9408,26 @@ class DeepAgentsApp(App):
self._status_bar.set_auto_approve(enabled=self._auto_approve)
if self._session_state:
self._session_state.auto_approve = self._auto_approve
if not await self._write_live_approval_mode():
if self._auto_approve:
self._warn_live_approval_mode_unavailable(
"Auto-approve could not sync to the running agent; "
"approval prompts may continue."
)
elif self._agent_running:
# Switching to manual mid-run, but the agent never saw it:
# cancel the active run rather than let it keep auto-approving.
self._session_state.approval_mode_key = None
self._warn_live_approval_mode_unavailable(
"Manual approval could not sync to the running agent; "
"the active run was cancelled for safety."
)
self._force_interrupt_active_work()
else:
self._warn_live_approval_mode_unavailable(
"Manual approval could not sync to the running agent; "
"start a new run before continuing."
)
def action_toggle_tool_output(self) -> None:
"""Toggle expand/collapse of the most recent tool output or skill body."""
libs/code/deepagents_code/approval_mode.py
+131
View File
@@ -0,0 +1,131 @@
"""Live approval-mode state shared through the LangGraph Store."""
from __future__ import annotations
import logging
from collections.abc import Mapping
from hashlib import sha256
from typing import TypedDict
logger = logging.getLogger(__name__)
APPROVAL_MODE_NAMESPACE: tuple[str, str] = ("deepagents_code", "approval_mode")
"""Store namespace for per-thread approval-mode control records."""
class ApprovalModePayload(TypedDict):
"""Stored approval-mode control payload."""
auto_approve: bool
def approval_mode_key(thread_id: str) -> str:
"""Return the store key for a thread's live approval mode.
Args:
thread_id: LangGraph thread id for the active session.
Returns:
Deterministic store key that does not expose the raw thread id.
"""
return sha256(thread_id.encode("utf-8")).hexdigest()
def approval_mode_payload(*, auto_approve: bool) -> ApprovalModePayload:
"""Return the stored approval-mode payload.
Args:
auto_approve: Whether gated tool calls should skip HITL approval.
Returns:
JSON-serializable store value.
"""
return {"auto_approve": auto_approve}
def _item_value(item: object) -> object:
"""Extract a store item's value from SDK and runtime item shapes.
Returns:
The item's stored value, or `None` when the shape is unrecognized.
"""
if isinstance(item, Mapping):
return item.get("value")
return getattr(item, "value", None)
def read_approval_mode_from_store(store: object, key: str | None) -> bool | None:
"""Read a live approval mode from the server-side LangGraph Store.
Args:
store: `request.runtime.store` from the graph server.
key: Store key produced by `approval_mode_key`. The `isinstance` guard
below still rejects non-string keys as defense-in-depth, since the
value crosses the JSON/RemoteGraph boundary before reaching here.
Returns:
`True` or `False` when the store contains a valid mode, otherwise
`None`. Callers should treat `None` as fail-closed.
"""
if store is None:
logger.debug("Approval-mode store is unavailable")
return None
if not isinstance(key, str) or not key:
logger.debug("Approval-mode store key is missing or invalid")
return None
get = getattr(store, "get", None)
if get is None:
logger.debug("Approval-mode store does not expose get()")
return None
try:
item = get(APPROVAL_MODE_NAMESPACE, key)
except Exception:
logger.warning("Could not read approval-mode store item", exc_info=True)
return None
if item is None:
logger.debug("Approval-mode store item is missing")
return None
value = _item_value(item)
auto_approve = value.get("auto_approve") if isinstance(value, Mapping) else None
if isinstance(auto_approve, bool):
return auto_approve
logger.debug("Approval-mode store item has invalid contents")
return None
async def awrite_approval_mode(
agent: object,
thread_id: str,
*,
auto_approve: bool,
) -> str | None:
"""Persist approval mode through an agent's remote store client.
Args:
agent: Agent object. Remote agents expose `aput_store_item`; agents
without a writer use run context only.
thread_id: LangGraph thread id for the active session.
auto_approve: Whether gated tool calls should skip HITL approval.
Returns:
Store key written, or `None` when the agent has no store writer.
Notes:
Remote agents rely on the server-side store being visible to the
running graph before the next gated tool predicate executes.
"""
put = getattr(agent, "aput_store_item", None)
if put is None:
return None
key = approval_mode_key(thread_id)
await put(
APPROVAL_MODE_NAMESPACE,
key,
approval_mode_payload(auto_approve=auto_approve),
)
return key
libs/code/deepagents_code/configurable_model.py
+3
View File
@@ -71,10 +71,13 @@ def _get_context(request: ModelRequest) -> CLIContextSchema | None:
if isinstance(ctx, CLIContextSchema):
return ctx
if isinstance(ctx, dict):
raw_key = ctx.get("approval_mode_key")
return CLIContextSchema(
model=ctx.get("model"),
model_params=ctx.get("model_params") or {},
effective_model=ctx.get("effective_model"),
auto_approve=bool(ctx.get("auto_approve", False)),
approval_mode_key=raw_key if isinstance(raw_key, str) else None,
)
return None
libs/code/deepagents_code/remote_client.py
+36
View File
@@ -367,6 +367,42 @@ class RemoteAgent:
)
raise
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
) -> None:
"""Write an item to the server-side LangGraph Store.
Args:
namespace: Store namespace.
key: Item key within `namespace`.
value: JSON-serializable item value.
Notes:
A failed write is logged at debug and re-raised. The re-raise is
load-bearing: callers (`awrite_approval_mode` and its callers)
depend on the failure propagating so they can fail closed — drop
the live approval-mode key and interrupt rather than keep
auto-approving. Removing the `raise` would turn the debug log into
a silent-failure hole, so the higher-severity logging is left to
those callers, which re-log at warning with `exc_info`.
"""
graph = self._get_graph()
try:
client = graph._validate_client()
await client.store.put_item(namespace, key, value, index=False)
except Exception:
logger.debug(
"Failed to write store item %s/%s",
".".join(namespace),
key,
exc_info=True,
)
# Load-bearing: see Notes. Callers fail closed on this propagation.
raise
async def aensure_thread(self, config: dict[str, Any]) -> None:
"""Ensure the remote thread record exists before mutating state.
libs/code/deepagents_code/textual_adapter.py
+33 -5
View File
@@ -227,7 +227,7 @@ class TextualUIAdapter:
mount_message: Callable[..., Awaitable[None]],
update_status: Callable[[str], None],
request_approval: Callable[..., Awaitable[Any]],
on_auto_approve_enabled: Callable[[], None] | None = None,
on_auto_approve_enabled: Callable[[], Awaitable[None] | None] | None = None,
set_spinner: Callable[[SpinnerStatus], Awaitable[None]] | None = None,
set_active_message: Callable[[str | None], None] | None = None,
sync_message_content: Callable[[str, str], None] | None = None,
@@ -434,6 +434,8 @@ async def execute_task_textual(
from langgraph.types import Command
from pydantic import ValidationError
from deepagents_code.approval_mode import awrite_approval_mode
hitl_request_adapter = _get_hitl_request_adapter(HITLRequest)
ask_user_adapter = _get_ask_user_adapter()
@@ -540,11 +542,35 @@ async def execute_task_textual(
# 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.
# source. Also write the live store item that the server-side
# predicate re-reads on each tool call, so toggling approval mode
# mid-stream (either direction) takes effect before the current
# stream returns. Turning auto-approve off is the safety-critical
# direction, but the same store write also propagates turning it on.
if context is None:
context = CLIContext()
context["auto_approve"] = bool(session_state.auto_approve)
auto_approve = bool(session_state.auto_approve)
context["auto_approve"] = auto_approve
try:
live_key = await awrite_approval_mode(
agent,
thread_id,
auto_approve=auto_approve,
)
except Exception:
logger.warning(
"Failed to write live approval mode; interrupting for safety",
exc_info=True,
)
context["auto_approve"] = False
context.pop("approval_mode_key", None)
session_state.approval_mode_key = None
else:
if live_key is None:
context.pop("approval_mode_key", None)
else:
context["approval_mode_key"] = live_key
session_state.approval_mode_key = live_key
# Show the Thinking spinner before each astream iteration so
# both the first turn and HITL/ask_user resumes surface feedback
@@ -1275,7 +1301,9 @@ async def execute_task_textual(
# 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()
callback_result = adapter._on_auto_approve_enabled()
if callback_result is not None:
await callback_result
decisions = [
ApproveDecision(type="approve")
for _ in action_requests
libs/code/tests/unit_tests/test_agent.py
+153 -4
View File
@@ -1,9 +1,9 @@
"""Unit tests for agent formatting functions."""
import warnings
from collections.abc import Iterator
from collections.abc import Iterator, Mapping
from contextlib import contextmanager
from dataclasses import fields
from dataclasses import dataclass, fields
from pathlib import Path
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, cast
@@ -45,6 +45,27 @@ from deepagents_code.managed_tools import BIN_DIR
from deepagents_code.project_utils import ProjectContext
@dataclass
class _StoreItem:
value: dict[str, Any]
class _FakeStore:
def __init__(self) -> None:
self.items: dict[tuple[tuple[str, ...], str], _StoreItem] = {}
def put(
self,
namespace: tuple[str, ...],
key: str,
value: Mapping[str, Any],
) -> None:
self.items[namespace, key] = _StoreItem(dict(value))
def get(self, namespace: tuple[str, ...], key: str) -> _StoreItem | None:
return self.items.get((namespace, key))
def _make_fake_chat_model() -> GenericFakeChatModel:
"""Create a fake chat model compatible with summarization middleware."""
model = GenericFakeChatModel(messages=iter([AIMessage(content="ok")]))
@@ -81,10 +102,14 @@ def test_add_interrupt_on_attaches_auto_approve_predicate() -> None:
assert config.get("when") is _should_interrupt_tool_call
def _request_with_context(context: object) -> "ToolCallRequest":
def _request_with_context(
context: object,
*,
store: object | None = None,
) -> "ToolCallRequest":
return cast(
"ToolCallRequest",
SimpleNamespace(runtime=SimpleNamespace(context=context)),
SimpleNamespace(runtime=SimpleNamespace(context=context, store=store)),
)
@@ -104,6 +129,130 @@ def test_should_interrupt_tool_call_respects_auto_approve_context() -> None:
)
def test_should_interrupt_tool_call_prefers_live_approval_mode() -> None:
"""A live manual toggle overrides an auto-approve run-context snapshot."""
from deepagents_code.approval_mode import (
APPROVAL_MODE_NAMESPACE,
approval_mode_key,
approval_mode_payload,
)
store = _FakeStore()
key = approval_mode_key("thread-1")
store.put(
APPROVAL_MODE_NAMESPACE,
key,
approval_mode_payload(auto_approve=False),
)
assert _should_interrupt_tool_call(
_request_with_context(
{"auto_approve": True, "approval_mode_key": key},
store=store,
)
)
assert _should_interrupt_tool_call(
_request_with_context(
CLIContextSchema(auto_approve=True, approval_mode_key=key),
store=store,
)
)
store.put(
APPROVAL_MODE_NAMESPACE,
key,
approval_mode_payload(auto_approve=True),
)
assert not _should_interrupt_tool_call(
_request_with_context(
{"auto_approve": False, "approval_mode_key": key},
store=store,
)
)
async def test_live_approval_round_trip_flips_interrupt_decision() -> None:
"""A mode written via `awrite_approval_mode` is read back by the predicate.
Exercises the full writer -> store -> reader contract across the shared
`approval_mode_key` seam. The isolated write- and read-side tests would both
stay green even if the two ever derived the key differently; only crossing
the seam catches that — a key mismatch would surface here as an unexpected
fail-closed interrupt.
"""
from deepagents_code.approval_mode import approval_mode_key, awrite_approval_mode
store = _FakeStore()
class _StoreWriter:
"""Agent double whose store writer feeds the same store the reader uses."""
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: Mapping[str, Any],
) -> None:
store.put(namespace, key, value)
agent = _StoreWriter()
key = approval_mode_key("thread-1")
written = await awrite_approval_mode(agent, "thread-1", auto_approve=True)
assert written == key
# Live auto-approve suppresses the interrupt even though the context
# snapshot still says manual.
assert not _should_interrupt_tool_call(
_request_with_context(
{"auto_approve": False, "approval_mode_key": key},
store=store,
)
)
await awrite_approval_mode(agent, "thread-1", auto_approve=False)
# Flipping the stored mode to manual interrupts despite an auto context.
assert _should_interrupt_tool_call(
_request_with_context(
{"auto_approve": True, "approval_mode_key": key},
store=store,
)
)
def test_cli_context_schema_fields_mirror_typed_dict() -> None:
"""`CLIContextSchema` and `CLIContext` must stay structurally identical.
The two shapes carry the same payload across the API boundary (dataclass
in-process, dict over RemoteGraph). A field added to one but not the other
would silently drop across that boundary; this pins the documented mirror.
"""
from deepagents_code._cli_context import CLIContext
assert {f.name for f in fields(CLIContextSchema)} == set(CLIContext.__annotations__)
def test_should_interrupt_tool_call_fails_closed_when_live_mode_missing() -> None:
"""A configured but missing live mode should interrupt for safety."""
from deepagents_code.approval_mode import approval_mode_key
assert _should_interrupt_tool_call(
_request_with_context(
{"auto_approve": True, "approval_mode_key": approval_mode_key("thread-1")},
store=_FakeStore(),
)
)
def test_should_interrupt_tool_call_fails_closed_without_live_mode_store() -> None:
"""A configured live-mode key with no runtime store should interrupt."""
from deepagents_code.approval_mode import approval_mode_key
assert _should_interrupt_tool_call(
_request_with_context(
{"auto_approve": True, "approval_mode_key": approval_mode_key("thread-1")}
)
)
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({}))
libs/code/tests/unit_tests/test_app.py
+164
View File
@@ -13187,6 +13187,170 @@ class TestForceInterruptActiveWork:
app._force_interrupt_active_work()
class _ApprovalModeWriter:
def __init__(self) -> None:
self.item: tuple[tuple[str, ...], str, dict[str, Any]] | None = None
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
) -> None:
self.item = (namespace, key, value)
class _FailingApprovalModeWriter:
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
) -> None:
_ = (namespace, key, value)
msg = "store unavailable"
raise RuntimeError(msg)
class TestLiveApprovalModeWrites:
"""Verify live approval-mode write and toggle failure behavior."""
async def test_write_live_approval_mode_records_key(self) -> None:
from deepagents_code.approval_mode import (
APPROVAL_MODE_NAMESPACE,
approval_mode_key,
)
app = DeepAgentsApp()
writer = _ApprovalModeWriter()
app._agent = cast("Any", writer)
app._session_state = TextualSessionState(
thread_id="thread-1",
auto_approve=True,
)
assert await app._write_live_approval_mode()
assert app._session_state.approval_mode_key == approval_mode_key("thread-1")
assert writer.item == (
APPROVAL_MODE_NAMESPACE,
approval_mode_key("thread-1"),
{"auto_approve": True},
)
async def test_write_live_approval_mode_clears_key_on_failure(self) -> None:
app = DeepAgentsApp()
app._agent = cast("Any", _FailingApprovalModeWriter())
app._session_state = TextualSessionState(
thread_id="thread-1",
auto_approve=False,
)
app._session_state.approval_mode_key = "stale"
assert not await app._write_live_approval_mode()
assert app._session_state.approval_mode_key is None
async def test_toggle_off_failed_write_cancels_running_agent(self) -> None:
app = DeepAgentsApp(auto_approve=True)
async with app.run_test() as pilot:
await pilot.pause()
app._session_state = TextualSessionState(
thread_id="thread-1",
auto_approve=True,
)
app._session_state.approval_mode_key = "stale"
app._agent_running = True
with (
patch.object(
app,
"_write_live_approval_mode",
new=AsyncMock(return_value=False),
),
patch.object(app, "_force_interrupt_active_work") as force,
patch.object(app, "notify") as notify,
):
await app.action_toggle_auto_approve()
assert app._auto_approve is False
assert app._session_state.auto_approve is False
assert app._session_state.approval_mode_key is None
force.assert_called_once()
notify.assert_called_once()
assert notify.call_args.kwargs["severity"] == "warning"
async def test_toggle_off_failed_write_does_not_cancel_when_idle(self) -> None:
app = DeepAgentsApp(auto_approve=True)
async with app.run_test() as pilot:
await pilot.pause()
app._session_state = TextualSessionState(
thread_id="thread-1",
auto_approve=True,
)
app._agent_running = False
with (
patch.object(
app,
"_write_live_approval_mode",
new=AsyncMock(return_value=False),
),
patch.object(app, "_force_interrupt_active_work") as force,
patch.object(app, "notify") as notify,
):
await app.action_toggle_auto_approve()
force.assert_not_called()
notify.assert_called_once()
# The idle branch emits a distinct message from the cancel branch.
assert "start a new run" in notify.call_args.args[0]
async def test_toggle_on_failed_write_does_not_cancel_running_agent(self) -> None:
app = DeepAgentsApp(auto_approve=False)
async with app.run_test() as pilot:
await pilot.pause()
app._session_state = TextualSessionState(
thread_id="thread-1",
auto_approve=False,
)
app._agent_running = True
with (
patch.object(
app,
"_write_live_approval_mode",
new=AsyncMock(return_value=False),
),
patch.object(app, "_force_interrupt_active_work") as force,
patch.object(app, "notify") as notify,
):
await app.action_toggle_auto_approve()
assert app._auto_approve is True
assert app._session_state.auto_approve is True
force.assert_not_called()
notify.assert_called_once()
# Toggling on emits the auto-approve warning, not the manual one.
assert "Auto-approve could not sync" in notify.call_args.args[0]
async def test_auto_approve_all_failed_write_warns(self) -> None:
app = DeepAgentsApp(auto_approve=False)
app._session_state = TextualSessionState(
thread_id="thread-1",
auto_approve=False,
)
with (
patch.object(
app,
"_write_live_approval_mode",
new=AsyncMock(return_value=False),
),
patch.object(app, "notify") as notify,
):
await app._on_auto_approve_enabled()
assert app._auto_approve is True
assert app._session_state.auto_approve is True
notify.assert_called_once()
assert notify.call_args.kwargs["severity"] == "warning"
class TestExternalBypassFieldHonored:
"""`event.bypass` overrides queue when set on a prompt event."""
libs/code/tests/unit_tests/test_approval_mode.py
+129
View File
@@ -0,0 +1,129 @@
"""Tests for live approval-mode store helpers."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, cast
import pytest
from deepagents_code.approval_mode import (
APPROVAL_MODE_NAMESPACE,
approval_mode_key,
approval_mode_payload,
awrite_approval_mode,
read_approval_mode_from_store,
)
@dataclass
class _StoreItem:
value: object
class _Store:
def __init__(self, item: object = None) -> None:
self.item = item
def get(self, namespace: tuple[str, ...], key: str) -> object:
assert namespace == APPROVAL_MODE_NAMESPACE
assert key
return self.item
class _FailingStore:
def get(self, namespace: tuple[str, ...], key: str) -> object:
_ = (namespace, key)
msg = "store unavailable"
raise RuntimeError(msg)
class _Writer:
def __init__(self) -> None:
self.items: list[tuple[tuple[str, ...], str, dict[str, Any]]] = []
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
) -> None:
self.items.append((namespace, key, value))
def test_approval_mode_payload_shape() -> None:
assert approval_mode_payload(auto_approve=True) == {"auto_approve": True}
def test_read_approval_mode_from_store_accepts_mapping_item() -> None:
key = approval_mode_key("thread-1")
item = {"value": {"auto_approve": True}}
assert read_approval_mode_from_store(_Store(item), key) is True
def test_read_approval_mode_from_store_accepts_attribute_item() -> None:
key = approval_mode_key("thread-1")
item = _StoreItem({"auto_approve": False})
assert read_approval_mode_from_store(_Store(item), key) is False
@pytest.mark.parametrize(
("store", "key"),
[
(None, approval_mode_key("thread-1")),
(object(), approval_mode_key("thread-1")), # store has no get()
(_Store(None), approval_mode_key("thread-1")),
(_Store(_StoreItem(["not", "a", "mapping"])), approval_mode_key("thread-1")),
(_Store(_StoreItem({"auto_approve": "yes"})), approval_mode_key("thread-1")),
(_Store(_StoreItem({"auto_approve": 1})), approval_mode_key("thread-1")),
(_Store(_StoreItem({"auto_approve": True})), ""),
(_Store(_StoreItem({"auto_approve": True})), None),
],
)
def test_read_approval_mode_from_store_fails_closed(
store: object,
key: str | None,
) -> None:
assert read_approval_mode_from_store(store, key) is None
def test_read_approval_mode_from_store_non_string_key_fails_closed() -> None:
"""A non-string key still fails closed via the runtime guard.
The declared `key` type is `str | None`, but the value crosses the
JSON/RemoteGraph boundary, so the `isinstance` guard remains as
defense-in-depth against a malformed payload.
"""
item = _StoreItem({"auto_approve": True})
assert read_approval_mode_from_store(_Store(item), cast("str", object())) is None
def test_read_approval_mode_from_store_exception_fails_closed(
caplog: pytest.LogCaptureFixture,
) -> None:
with caplog.at_level("WARNING", logger="deepagents_code.approval_mode"):
assert (
read_approval_mode_from_store(
_FailingStore(),
approval_mode_key("thread-1"),
)
is None
)
assert "Could not read approval-mode store item" in caplog.text
async def test_awrite_approval_mode_writes_payload() -> None:
writer = _Writer()
key = await awrite_approval_mode(writer, "thread-1", auto_approve=True)
assert key == approval_mode_key("thread-1")
assert writer.items == [
(APPROVAL_MODE_NAMESPACE, approval_mode_key("thread-1"), {"auto_approve": True})
]
async def test_awrite_approval_mode_returns_none_without_writer() -> None:
assert (await awrite_approval_mode(object(), "thread-1", auto_approve=True)) is None
libs/code/tests/unit_tests/test_configurable_model.py
+34 -1
View File
@@ -6,6 +6,7 @@ from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import MagicMock, patch
import pytest
from langchain.agents.middleware.types import ModelRequest, ModelResponse
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage
@@ -14,6 +15,7 @@ from deepagents_code._cli_context import CLIContext, CLIContextSchema
from deepagents_code.agent import build_model_identity_section
from deepagents_code.configurable_model import (
ConfigurableModelMiddleware,
_get_context,
_is_anthropic_model,
)
@@ -29,7 +31,7 @@ def _make_model(name: str) -> MagicMock:
def _make_request(
model: BaseChatModel,
context: CLIContext | CLIContextSchema | None = None,
context: object = None,
model_settings: dict[str, Any] | None = None,
system_prompt: str | None = None,
) -> ModelRequest:
@@ -94,6 +96,37 @@ class TestNoOverride:
)
assert captured[0] is request
def test_dict_context_reconstructs_approval_fields(self) -> None:
request = _make_request(
_make_model("claude-sonnet-4-6"),
context={
"auto_approve": True,
"approval_mode_key": "approval-key",
},
)
ctx = _get_context(request)
assert ctx is not None
assert ctx.auto_approve is True
assert ctx.approval_mode_key == "approval-key"
@pytest.mark.parametrize("key", [None, 1, object()])
def test_dict_context_coerces_non_string_approval_key(self, key: object) -> None:
request = _make_request(
_make_model("claude-sonnet-4-6"),
context={
"auto_approve": True,
"approval_mode_key": key,
},
)
ctx = _get_context(request)
assert ctx is not None
assert ctx.auto_approve is True
assert ctx.approval_mode_key is None
def test_same_model_spec(self) -> None:
request = _make_request(
_make_model("claude-sonnet-4-6"),
libs/code/tests/unit_tests/test_remote_client.py
+39
View File
@@ -2,6 +2,7 @@
import uuid
from collections.abc import Sequence
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
@@ -791,6 +792,44 @@ class TestRemoteAgentUpdateStateConflictRecovery:
runs_cancel.assert_not_called()
class TestRemoteAgentStore:
async def test_aput_store_item_uses_unindexed_put(self) -> None:
agent = RemoteAgent(url="http://localhost:8123", graph_name="agent")
store = SimpleNamespace(put_item=AsyncMock())
client = SimpleNamespace(store=store)
graph = MagicMock()
graph._validate_client.return_value = client
agent._graph = graph
await agent.aput_store_item(("ns",), "key", {"auto_approve": True})
store.put_item.assert_awaited_once_with(
("ns",),
"key",
{"auto_approve": True},
index=False,
)
async def test_aput_store_item_logs_and_reraises(
self,
caplog: pytest.LogCaptureFixture,
) -> None:
agent = RemoteAgent(url="http://localhost:8123", graph_name="agent")
store = SimpleNamespace(put_item=AsyncMock(side_effect=RuntimeError("boom")))
client = SimpleNamespace(store=store)
graph = MagicMock()
graph._validate_client.return_value = client
agent._graph = graph
with (
caplog.at_level("DEBUG", logger="deepagents_code.remote_client"),
pytest.raises(RuntimeError, match="boom"),
):
await agent.aput_store_item(("ns",), "key", {"auto_approve": True})
assert "Failed to write store item ns/key" in caplog.text
class TestRemoteAgentEnsureThread:
"""Verify remote thread registration before state writes."""
libs/code/tests/unit_tests/test_sessions.py
+11
View File
@@ -507,6 +507,17 @@ class TestTextualSessionState:
assert uuid.UUID(new_id).version == 7
assert state.thread_id == new_id
def test_reset_thread_clears_approval_mode_key(self):
"""A new thread must not inherit the prior thread's live approval key.
The key is hashed per thread; leaving a stale key would point the
interrupt predicate at the previous thread's mode.
"""
state = TextualSessionState(thread_id="original")
state.approval_mode_key = "stale"
state.reset_thread()
assert state.approval_mode_key is None
class TestFindSimilarThreads:
"""Tests for find_similar_threads function."""
libs/code/tests/unit_tests/test_textual_adapter.py
+94 -1
View File
@@ -3,7 +3,7 @@
import asyncio
import sys
from asyncio import Future
from collections.abc import AsyncIterator, Generator
from collections.abc import AsyncIterator, Awaitable, Callable, Generator
from datetime import datetime
from importlib.metadata import PackageNotFoundError
from io import StringIO
@@ -20,6 +20,7 @@ from rich.console import Console
from deepagents_code import config as config_module
from deepagents_code._ask_user_types import AskUserWidgetResult, Question
from deepagents_code.approval_mode import APPROVAL_MODE_NAMESPACE, approval_mode_key
from deepagents_code.config import build_stream_config
from deepagents_code.textual_adapter import (
ModelStats,
@@ -772,6 +773,16 @@ class _SequencedAgent:
self._streams_by_call = streams_by_call
self.stream_inputs: list[dict | Command] = []
self.contexts: list[Any] = []
self.store_items: list[tuple[tuple[str, ...], str, dict[str, Any]]] = []
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
) -> None:
"""Record store writes requested by `execute_task_textual`."""
self.store_items.append((namespace, key, value))
async def astream(
self,
@@ -794,6 +805,21 @@ class _SequencedAgent:
yield chunk
class _FailingApprovalStoreAgent(_SequencedAgent):
"""Agent test double whose approval-mode store writes fail."""
async def aput_store_item(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
) -> None:
"""Raise while preserving the production store-writer signature."""
_ = (namespace, key, value)
msg = "approval-mode store unavailable"
raise RuntimeError(msg)
class TestExecuteTaskTextualAutoApproveInput:
"""Auto-approve must ride on run context, never a first-turn `Command`."""
@@ -823,9 +849,46 @@ class TestExecuteTaskTextualAutoApproveInput:
assert not isinstance(stream_input, Command)
assert stream_input == {"messages": [{"role": "user", "content": "hi"}]}
assert agent.contexts[0]["auto_approve"] is True
key = approval_mode_key("thread-1")
assert agent.contexts[0]["approval_mode_key"] == key
assert agent.store_items == [
(APPROVAL_MODE_NAMESPACE, key, {"auto_approve": True})
]
async def test_live_approval_write_failure_fails_closed_context(self) -> None:
"""A failed live-mode write must not reuse a stale approval key."""
agent = _FailingApprovalStoreAgent([[]])
adapter = TextualUIAdapter(
mount_message=_mock_mount,
update_status=_noop_status,
request_approval=_mock_approval,
)
session_state = SimpleNamespace(
thread_id="thread-1",
auto_approve=True,
approval_mode_key="stale",
)
await execute_task_textual(
user_input="hi",
agent=agent,
assistant_id="assistant",
session_state=session_state,
adapter=adapter,
)
stream_input = agent.stream_inputs[0]
assert not isinstance(stream_input, Command)
assert agent.contexts[0]["auto_approve"] is False
assert "approval_mode_key" not in agent.contexts[0]
assert agent.store_items == []
# The stale key must be cleared so later turns don't reuse it.
assert session_state.approval_mode_key is None
@pytest.mark.parametrize("use_async_callback", [True, False])
async def test_mid_turn_auto_approve_all_propagates_to_resume_context(
self,
use_async_callback: bool,
) -> None:
"""Choosing "auto-approve all" mid-turn flips the resuming stream's context.
@@ -835,6 +898,10 @@ class TestExecuteTaskTextualAutoApproveInput:
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).
Parametrized over an async and a sync `on_auto_approve_enabled` callback
to cover the `Awaitable[None] | None` union the adapter awaits only when
the result is non-`None`.
"""
action_requests = [{"name": "execute", "args": {"command": "echo hi"}}]
agent = _SequencedAgent(
@@ -875,10 +942,28 @@ class TestExecuteTaskTextualAutoApproveInput:
future.set_result({"type": "auto_approve_all"})
return future
callback_seen: list[bool] = []
on_auto_approve_enabled: Callable[[], Awaitable[None] | None]
if use_async_callback:
async def _async_callback() -> None:
await asyncio.sleep(0)
callback_seen.append(True)
on_auto_approve_enabled = _async_callback
else:
def _sync_callback() -> None:
callback_seen.append(True)
on_auto_approve_enabled = _sync_callback
adapter = TextualUIAdapter(
mount_message=_mock_mount,
update_status=_noop_status,
request_approval=request_approval,
on_auto_approve_enabled=on_auto_approve_enabled,
)
session_state = SimpleNamespace(thread_id="thread-1", auto_approve=False)
@@ -895,6 +980,14 @@ class TestExecuteTaskTextualAutoApproveInput:
assert len(agent.contexts) == 2
assert agent.contexts[0]["auto_approve"] is False
assert agent.contexts[1]["auto_approve"] is True
key = approval_mode_key("thread-1")
assert agent.contexts[0]["approval_mode_key"] == key
assert agent.contexts[1]["approval_mode_key"] == key
assert agent.store_items == [
(APPROVAL_MODE_NAMESPACE, key, {"auto_approve": False}),
(APPROVAL_MODE_NAMESPACE, key, {"auto_approve": True}),
]
assert callback_seen == [True]
assert session_state.auto_approve is True