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(cli): remove legacy shell tool aliases and harden HITL widget cleanup (#3340)

Browse Source 
Closes #1091

---

When the agent calls a shell tool that requires approval, the command
was rendered twice on screen at the same time — once in the streamed
`ToolCallMessage` header, and again in the approval dialog. For
non-shell tools (`write_file`, `edit_file`) the approval dialog shows
diffs/file contents so the duplication is hidden, but for shell-family
tools (`shell`, `bash`, `execute`) both surfaces render the same command
string.

Adds `set_awaiting_approval` / `clear_awaiting_approval` on
`ToolCallMessage` that toggle the widget's `display` attribute, and
wraps the approval await in `execute_task_textual` with a
`try`/`finally` that suppresses any in-flight shell tool messages while
the prompt is visible. The widget is restored before the post-decision
`set_running` / `set_rejected` calls run, so the existing status-update
flow is unchanged.

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
Co-authored-by: Mason Daugherty <61371264+mdrxy@users.noreply.github.com>
Co-authored-by: Mason Daugherty <github@mdrxy.com>
This commit is contained in:
open-swe[bot]
2026-05-11 12:35:31 -07:00
committed by GitHub
parent 7ab6e14565
commit 398f7f4512
12 changed files with 372 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/cli/deepagents_cli/agent.py
+8 -9
View File
@@ -88,11 +88,11 @@ REQUIRE_COMPACT_TOOL_APPROVAL: bool = True
class ShellAllowListMiddleware(AgentMiddleware):
"""Validate shell commands against an allow-list without HITL interrupts.
When the agent invokes a shell tool (any tool in `SHELL_TOOL_NAMES`),
this middleware checks the command against the configured allow-list
**before execution**. Rejected commands are returned as error `ToolMessage`
objects — the graph never pauses, so LangSmith traces stay as a single
continuous run.
When the agent invokes the `execute` shell tool, this middleware checks
the command against the configured allow-list **before execution**.
Rejected commands are returned as error `ToolMessage` objects — the
graph never pauses, so LangSmith traces stay as a single continuous
run.
Use this middleware in non-interactive mode to avoid the
interrupt/resume cycle that fragments traces.
@@ -135,10 +135,9 @@ class ShellAllowListMiddleware(AgentMiddleware):
"""
from langchain_core.messages import ToolMessage as LCToolMessage
from deepagents_cli.config import SHELL_TOOL_NAMES, is_shell_command_allowed
from deepagents_cli.config import is_shell_command_allowed
tool_name = request.tool_call["name"]
if tool_name not in SHELL_TOOL_NAMES:
if request.tool_call["name"] != "execute":
return None
args = request.tool_call.get("args") or {}
@@ -155,7 +154,7 @@ class ShellAllowListMiddleware(AgentMiddleware):
f"Allowed commands: {allowed_str}. "
f"Please use an allowed command or try another approach."
),
name=tool_name,
name="execute",
tool_call_id=request.tool_call["id"],
status="error",
)
libs/cli/deepagents_cli/app.py
+1 -2
View File
@@ -3378,7 +3378,6 @@ class DeepAgentsApp(App):
A Future that resolves to the user's decision.
"""
from deepagents_cli.config import (
SHELL_TOOL_NAMES,
is_shell_command_allowed,
settings,
)
@@ -3392,7 +3391,7 @@ class DeepAgentsApp(App):
approved_commands = []
for req in action_requests:
if req.get("name") in SHELL_TOOL_NAMES:
if req.get("name") == "execute":
command = req.get("args", {}).get("command", "")
if is_shell_command_allowed(command, settings.shell_allow_list):
approved_commands.append(command)
libs/cli/deepagents_cli/config.py
-8
View File
@@ -1464,14 +1464,6 @@ class SessionState:
return self.auto_approve
SHELL_TOOL_NAMES: frozenset[str] = frozenset({"bash", "shell", "execute"})
"""Tool names recognized as shell/command-execution tools.
Only `'execute'` is registered by the SDK and CLI backends in practice.
`'bash'` and `'shell'` are legacy names carried over and kept as
backwards-compatible aliases.
"""
DANGEROUS_SHELL_PATTERNS = (
"$(", # Command substitution
"`", # Backtick command substitution
libs/cli/deepagents_cli/non_interactive.py
+1 -2
View File
@@ -41,7 +41,6 @@ from deepagents_cli._version import __version__
from deepagents_cli.agent import DEFAULT_AGENT_NAME
from deepagents_cli.config import (
SHELL_ALLOW_ALL,
SHELL_TOOL_NAMES,
build_langsmith_thread_url,
create_model,
is_shell_command_allowed,
@@ -470,7 +469,7 @@ def _make_hitl_decision(
action_name = action_request.get("name", "")
if action_name in SHELL_TOOL_NAMES:
if action_name == "execute":
if not settings.shell_allow_list:
command = action_request.get("args", {}).get("command", "")
console.print(
libs/cli/deepagents_cli/textual_adapter.py
+26 -4
View File
@@ -1150,10 +1150,32 @@ async def execute_task_textual(
]
},
)
future = await adapter._request_approval(
action_requests, assistant_id
)
decision = await future
# Hide shell tool widgets while the approval renders the
# same command; restore before processing the decision
# so subsequent status updates render on the visible
# widget.
suppressed_tool_msgs = [
tool_msg
for tool_msg in adapter._current_tool_messages.values()
if tool_msg.tool_name == "execute"
]
for tool_msg in suppressed_tool_msgs:
tool_msg.set_awaiting_approval()
try:
future = await adapter._request_approval(
action_requests, assistant_id
)
decision = await future
finally:
for tool_msg in suppressed_tool_msgs:
try:
tool_msg.clear_awaiting_approval()
except Exception:
logger.exception(
"Failed to clear awaiting-approval "
"state on tool widget %s",
tool_msg.tool_name,
)
if isinstance(decision, dict):
decision_type = decision.get("type")
libs/cli/deepagents_cli/widgets/approval.py
+3 -4
View File
@@ -18,7 +18,6 @@ if TYPE_CHECKING:
from deepagents_cli import theme
from deepagents_cli.config import (
SHELL_TOOL_NAMES,
get_glyphs,
is_ascii_mode,
)
@@ -128,7 +127,7 @@ class ApprovalMenu(Container):
self.decision = decision
# Tools that don't need detailed info display (already shown in tool call)
_MINIMAL_TOOLS: ClassVar[frozenset[str]] = SHELL_TOOL_NAMES
_MINIMAL_TOOLS: ClassVar[frozenset[str]] = frozenset({"execute"})
def __init__(
self,
@@ -161,7 +160,7 @@ class ApprovalMenu(Container):
self._future: asyncio.Future[dict[str, str]] | None = None
self._option_widgets: list[Static] = []
self._tool_info_container: Vertical | None = None
# Minimal display if ALL tools are bash/shell
# Minimal display if ALL tools are shell-execution tools
self._is_minimal = all(name in self._MINIMAL_TOOLS for name in self._tool_names)
# For expandable shell commands
self._command_expanded = False
@@ -182,7 +181,7 @@ class ApprovalMenu(Container):
if len(self._action_requests) != 1:
return False
req = self._action_requests[0]
if req.get("name", "") not in SHELL_TOOL_NAMES:
if req.get("name", "") != "execute":
return False
command = str(req.get("args", {}).get("command", ""))
return _is_command_too_long(command)
libs/cli/deepagents_cli/widgets/messages.py
+25 -9
View File
@@ -139,8 +139,6 @@ _TOOLS_WITH_HEADER_INFO: set[str] = {
"glob",
"grep",
"execute", # sandbox shell
# Shell tools
"shell", # local shell
# Web tools
"web_search",
"fetch_url",
@@ -823,6 +821,9 @@ class ToolCallMessage(Vertical):
self._deferred_status: str | None = None
self._deferred_output: str | None = None
self._deferred_expanded: bool = False
# Whether the widget is currently hidden because an approval prompt
# is rendering the same content (see `set_awaiting_approval`).
self._awaiting_approval: bool = False
def compose(self) -> ComposeResult:
"""Compose the tool call message layout.
@@ -1018,11 +1019,7 @@ class ToolCallMessage(Vertical):
self._stop_animation()
self._status = "error"
# For shell commands, prepend the full command so users can see what failed
command = (
self._args.get("command")
if self._tool_name in {"shell", "bash", "execute"}
else None
)
command = self._args.get("command") if self._tool_name == "execute" else None
if command and isinstance(command, str) and command.strip():
self._output = f"$ {command}\n\n{error}"
else:
@@ -1063,6 +1060,27 @@ class ToolCallMessage(Vertical):
self._status_widget.update(Content.styled("- Skipped", "dim"))
self._status_widget.display = True
def set_awaiting_approval(self) -> None:
"""Hide the tool call while an approval prompt mirrors its content.
Used to avoid showing the same shell command in both the streamed tool
call header and the HITL approval dialog at the same time. The widget
is restored via `clear_awaiting_approval` once the user decides.
"""
self._awaiting_approval = True
self.display = False
def clear_awaiting_approval(self) -> None:
"""Restore the tool call after `set_awaiting_approval`.
No-op if `set_awaiting_approval` was not previously called, so the
method is safe to call unconditionally from a `finally` block.
"""
if not self._awaiting_approval:
return
self._awaiting_approval = False
self.display = True
def toggle_output(self) -> None:
"""Toggle expansion of the tool's preview/full output."""
if not self._output:
@@ -1114,8 +1132,6 @@ class ToolCallMessage(Vertical):
"edit_file": self._format_file_output,
"grep": self._format_search_output,
"glob": self._format_search_output,
"shell": self._format_shell_output,
"bash": self._format_shell_output,
"execute": self._format_shell_output,
"web_search": self._format_web_output,
"fetch_url": self._format_web_output,
libs/cli/tests/unit_tests/test_agent.py
+4 -17
View File
@@ -1919,17 +1919,8 @@ class TestShellAllowListMiddleware:
handler.assert_awaited_once_with(request)
assert result == "ok"
@pytest.mark.parametrize(
("tool_name", "command"),
[
pytest.param("execute", "ls -la", id="execute-tool"),
pytest.param("bash", "cat README.md", id="bash-tool"),
],
)
async def test_allows_approved_shell_command(
self, tool_name: str, command: str
) -> None:
"""Shell commands in the allow-list pass through for all SHELL_TOOL_NAMES."""
async def test_allows_approved_shell_command(self) -> None:
"""Shell commands in the allow-list pass through to the handler."""
from unittest.mock import AsyncMock
from deepagents_cli.agent import ShellAllowListMiddleware
@@ -1937,8 +1928,8 @@ class TestShellAllowListMiddleware:
middleware = ShellAllowListMiddleware(allow_list=["ls", "cat"])
request = Mock()
request.tool_call = {
"name": tool_name,
"args": {"command": command},
"name": "execute",
"args": {"command": "ls -la"},
"id": "tc2",
}
handler = AsyncMock(return_value="output")
@@ -2049,8 +2040,6 @@ class TestShellAllowListMiddleware:
def test_rejects_empty_allow_list(self) -> None:
"""Constructor rejects empty allow-list."""
import pytest
from deepagents_cli.agent import ShellAllowListMiddleware
with pytest.raises(ValueError, match="must not be empty"):
@@ -2058,8 +2047,6 @@ class TestShellAllowListMiddleware:
def test_rejects_shell_allow_all(self) -> None:
"""Constructor rejects SHELL_ALLOW_ALL sentinel."""
import pytest
from deepagents_cli.agent import ShellAllowListMiddleware
from deepagents_cli.config import SHELL_ALLOW_ALL
libs/cli/tests/unit_tests/test_approval.py
+38 -49
View File
@@ -18,18 +18,18 @@ class TestCheckExpandableCommand:
def test_shell_command_over_threshold_is_expandable(self) -> None:
"""Test that shell commands longer than threshold are expandable."""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 10)
menu = ApprovalMenu({"name": "shell", "args": {"command": long_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": long_command}})
assert menu._has_expandable_command is True
def test_shell_command_at_threshold_not_expandable(self) -> None:
"""Test that shell commands at exactly the threshold are not expandable."""
exact_command = "x" * _SHELL_COMMAND_TRUNCATE_LENGTH
menu = ApprovalMenu({"name": "shell", "args": {"command": exact_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": exact_command}})
assert menu._has_expandable_command is False
def test_shell_command_under_threshold_not_expandable(self) -> None:
"""Test that short shell commands are not expandable."""
menu = ApprovalMenu({"name": "shell", "args": {"command": "echo hello"}})
menu = ApprovalMenu({"name": "execute", "args": {"command": "echo hello"}})
assert menu._has_expandable_command is False
def test_execute_tool_is_expandable(self) -> None:
@@ -49,15 +49,15 @@ class TestCheckExpandableCommand:
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 10)
menu = ApprovalMenu(
[
{"name": "shell", "args": {"command": long_command}},
{"name": "shell", "args": {"command": "echo hello"}},
{"name": "execute", "args": {"command": long_command}},
{"name": "execute", "args": {"command": "echo hello"}},
]
)
assert menu._has_expandable_command is False
def test_missing_command_arg_not_expandable(self) -> None:
"""Test that shell requests without command arg are not expandable."""
menu = ApprovalMenu({"name": "shell", "args": {}})
menu = ApprovalMenu({"name": "execute", "args": {}})
assert menu._has_expandable_command is False
def test_multiline_command_over_line_threshold_is_expandable(self) -> None:
@@ -67,13 +67,13 @@ class TestCheckExpandableCommand:
only if the line-count check is missing.
"""
command = "\n".join(["echo line"] * (_SHELL_COMMAND_TRUNCATE_LINES + 1))
menu = ApprovalMenu({"name": "shell", "args": {"command": command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": command}})
assert menu._has_expandable_command is True
def test_multiline_command_at_line_threshold_not_expandable(self) -> None:
"""Commands at exactly the line threshold are not expandable."""
command = "\n".join(["echo line"] * _SHELL_COMMAND_TRUNCATE_LINES)
menu = ApprovalMenu({"name": "shell", "args": {"command": command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": command}})
assert menu._has_expandable_command is False
@@ -82,7 +82,7 @@ class TestGetCommandDisplay:
def test_short_command_shows_full(self) -> None:
"""Test that short commands display in full regardless of expanded state."""
menu = ApprovalMenu({"name": "shell", "args": {"command": "echo hello"}})
menu = ApprovalMenu({"name": "execute", "args": {"command": "echo hello"}})
display = menu._get_command_display(expanded=False)
assert "echo hello" in display.plain
assert "press 'e' to expand" not in display.plain
@@ -90,7 +90,7 @@ class TestGetCommandDisplay:
def test_long_command_truncated_when_not_expanded(self) -> None:
"""Test that long commands are truncated with expand hint."""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 50)
menu = ApprovalMenu({"name": "shell", "args": {"command": long_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": long_command}})
display = menu._get_command_display(expanded=False)
assert get_glyphs().ellipsis in display.plain
assert "press 'e' to expand" in display.plain
@@ -100,7 +100,7 @@ class TestGetCommandDisplay:
def test_long_command_shows_full_when_expanded(self) -> None:
"""Test that long commands display in full when expanded."""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 50)
menu = ApprovalMenu({"name": "shell", "args": {"command": long_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": long_command}})
display = menu._get_command_display(expanded=True)
assert long_command in display.plain
assert "press 'e' to expand" not in display.plain
@@ -108,7 +108,7 @@ class TestGetCommandDisplay:
def test_short_command_shows_full_even_when_expanded_true(self) -> None:
"""Test that short commands show in full even when expanded=True."""
menu = ApprovalMenu({"name": "shell", "args": {"command": "echo hello"}})
menu = ApprovalMenu({"name": "execute", "args": {"command": "echo hello"}})
display = menu._get_command_display(expanded=True)
assert "echo hello" in display.plain
assert "press 'e' to expand" not in display.plain
@@ -117,7 +117,7 @@ class TestGetCommandDisplay:
def test_command_at_boundary_plus_one_is_expandable(self) -> None:
"""Test off-by-one: command at exactly threshold + 1 is expandable."""
boundary_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 1)
menu = ApprovalMenu({"name": "shell", "args": {"command": boundary_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": boundary_command}})
assert menu._has_expandable_command is True
display = menu._get_command_display(expanded=False)
assert get_glyphs().ellipsis in display.plain
@@ -125,14 +125,14 @@ class TestGetCommandDisplay:
def test_none_command_value_handled(self) -> None:
"""Test that None command value is handled gracefully."""
menu = ApprovalMenu({"name": "shell", "args": {"command": None}})
menu = ApprovalMenu({"name": "execute", "args": {"command": None}})
assert menu._has_expandable_command is False
display = menu._get_command_display(expanded=False)
assert "None" in display.plain
def test_integer_command_value_handled(self) -> None:
"""Test that integer command value is converted to string."""
menu = ApprovalMenu({"name": "shell", "args": {"command": 12345}})
menu = ApprovalMenu({"name": "execute", "args": {"command": 12345}})
assert menu._has_expandable_command is False
display = menu._get_command_display(expanded=False)
assert "12345" in display.plain
@@ -140,14 +140,14 @@ class TestGetCommandDisplay:
def test_command_display_escapes_markup_tags(self) -> None:
"""Shell command display should safely render literal bracket sequences."""
command = "echo [/dim] [literal]"
menu = ApprovalMenu({"name": "shell", "args": {"command": command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": command}})
display = menu._get_command_display(expanded=True)
assert command in display.plain
def test_command_display_with_hidden_unicode_shows_warning(self) -> None:
"""Hidden Unicode should be surfaced with explicit warning details."""
command = "echo a\u202eb"
menu = ApprovalMenu({"name": "shell", "args": {"command": command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": command}})
display = menu._get_command_display(expanded=True)
assert "echo ab" in display.plain
assert "hidden chars detected" in display.plain
@@ -158,7 +158,7 @@ class TestGetCommandDisplay:
"""Multi-line commands collapse to the line cap with an expand hint."""
lines = [f"echo {i}" for i in range(_SHELL_COMMAND_TRUNCATE_LINES + 3)]
command = "\n".join(lines)
menu = ApprovalMenu({"name": "shell", "args": {"command": command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": command}})
display = menu._get_command_display(expanded=False)
plain = display.plain
assert get_glyphs().ellipsis in plain
@@ -171,7 +171,7 @@ class TestGetCommandDisplay:
"""Expanded multi-line commands show every line."""
lines = [f"echo {i}" for i in range(_SHELL_COMMAND_TRUNCATE_LINES + 3)]
command = "\n".join(lines)
menu = ApprovalMenu({"name": "shell", "args": {"command": command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": command}})
display = menu._get_command_display(expanded=True)
for line in lines:
assert line in display.plain
@@ -184,7 +184,7 @@ class TestToggleExpand:
def test_toggle_changes_expanded_state(self) -> None:
"""Test that toggling changes the expanded state."""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 10)
menu = ApprovalMenu({"name": "shell", "args": {"command": long_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": long_command}})
# Need to set up command widget for toggle to work
menu._command_widget = MagicMock()
@@ -197,7 +197,7 @@ class TestToggleExpand:
def test_toggle_updates_widget_with_correct_content(self) -> None:
"""Test that toggling calls widget.update() with correct display content."""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 10)
menu = ApprovalMenu({"name": "shell", "args": {"command": long_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": long_command}})
menu._command_widget = MagicMock()
# First toggle: expand
@@ -217,7 +217,7 @@ class TestToggleExpand:
def test_toggle_does_nothing_for_non_expandable(self) -> None:
"""Test that toggling does nothing for non-expandable commands."""
menu = ApprovalMenu({"name": "shell", "args": {"command": "echo hello"}})
menu = ApprovalMenu({"name": "execute", "args": {"command": "echo hello"}})
menu._command_widget = MagicMock()
assert menu._command_expanded is False
@@ -227,7 +227,7 @@ class TestToggleExpand:
def test_toggle_does_nothing_without_widget(self) -> None:
"""Test that toggling does nothing if command widget is not set."""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 10)
menu = ApprovalMenu({"name": "shell", "args": {"command": long_command}})
menu = ApprovalMenu({"name": "execute", "args": {"command": long_command}})
# Explicitly ensure no widget
menu._command_widget = None
@@ -236,41 +236,28 @@ class TestToggleExpand:
assert menu._command_expanded is False
class TestToolSetConsistency:
"""Tests for tool set consistency between _MINIMAL_TOOLS and SHELL_TOOL_NAMES."""
def test_bash_tool_is_expandable(self) -> None:
"""Test that bash tool commands can be expandable like shell commands.
The 'bash' tool is in _MINIMAL_TOOLS, so it should also support
expandable command display when the command is long.
"""
long_command = "x" * (_SHELL_COMMAND_TRUNCATE_LENGTH + 10)
menu = ApprovalMenu({"name": "bash", "args": {"command": long_command}})
# bash should be expandable just like shell
assert menu._has_expandable_command is True
def test_bash_short_command_not_expandable(self) -> None:
"""Test that short bash commands are not expandable."""
menu = ApprovalMenu({"name": "bash", "args": {"command": "ls -la"}})
assert menu._has_expandable_command is False
class TestExecuteToolMinimalDisplay:
"""Tests confirming `execute` is treated as the shell-execution tool."""
def test_execute_tool_is_minimal(self) -> None:
"""Test that execute tool uses minimal display like shell.
The 'execute' tool is in SHELL_TOOL_NAMES, so it should use minimal display.
"""
"""The `execute` tool should use minimal display."""
menu = ApprovalMenu({"name": "execute", "args": {"command": "echo hello"}})
# execute should use minimal display like shell/bash
assert menu._is_minimal is True
def test_non_shell_tool_is_not_minimal(self) -> None:
"""Tools other than `execute` should not use minimal display."""
menu = ApprovalMenu({"name": "write_file", "args": {"path": "f.py"}})
assert menu._is_minimal is False
class TestSecurityWarnings:
"""Tests for approval-level Unicode/URL warning collection."""
def test_collects_hidden_unicode_warning(self) -> None:
"""Hidden Unicode in args should populate security warnings."""
menu = ApprovalMenu({"name": "shell", "args": {"command": "echo he\u200bllo"}})
menu = ApprovalMenu(
{"name": "execute", "args": {"command": "echo he\u200bllo"}}
)
assert menu._security_warnings
assert any("hidden Unicode" in warning for warning in menu._security_warnings)
@@ -289,7 +276,7 @@ class TestGetCommandDisplayGuard:
def test_raises_on_empty_action_requests(self) -> None:
"""Test that _get_command_display raises RuntimeError with empty requests."""
menu = ApprovalMenu({"name": "shell", "args": {"command": "echo hello"}})
menu = ApprovalMenu({"name": "execute", "args": {"command": "echo hello"}})
# Artificially empty the action_requests to test the guard
menu._action_requests = []
with pytest.raises(RuntimeError, match="empty action_requests"):
@@ -371,7 +358,9 @@ class TestOptionOrdering:
class ApprovalTestApp(App[None]):
def compose(self) -> ComposeResult:
yield ApprovalMenu({"name": "shell", "args": {"command": "echo hello"}})
yield ApprovalMenu(
{"name": "execute", "args": {"command": "echo hello"}}
)
def on_approval_menu_decided(self, event: ApprovalMenu.Decided) -> None:
nonlocal decision_received
libs/cli/tests/unit_tests/test_messages.py
+57 -25
View File
@@ -637,7 +637,7 @@ class TestToolCallMessageShellCommand:
long_cmd = "pip install " + " ".join(f"package{i}" for i in range(50))
assert len(long_cmd) > 120 # Exceeds truncation limit
msg = ToolCallMessage("shell", {"command": long_cmd})
msg = ToolCallMessage("execute", {"command": long_cmd})
msg.set_error("Command not found: pip")
# The error output should include the full command
@@ -646,31 +646,13 @@ class TestToolCallMessageShellCommand:
def test_shell_error_command_prefix(self) -> None:
"""Error output should have shell prompt prefix."""
cmd = "echo hello"
msg = ToolCallMessage("shell", {"command": cmd})
msg = ToolCallMessage("execute", {"command": cmd})
msg.set_error("Permission denied")
# Output should have shell prompt prefix
assert msg._output.startswith("$ ")
assert cmd in msg._output
def test_bash_error_includes_full_command(self) -> None:
"""Error output should include full command for bash tool too."""
cmd = "make build"
msg = ToolCallMessage("bash", {"command": cmd})
msg.set_error("make: *** No rule to make target")
assert msg._output.startswith("$ ")
assert cmd in msg._output
def test_execute_error_includes_full_command(self) -> None:
"""Error output should include full command for execute tool too."""
cmd = "docker build ."
msg = ToolCallMessage("execute", {"command": cmd})
msg.set_error("Cannot connect to Docker daemon")
assert msg._output.startswith("$ ")
assert cmd in msg._output
def test_non_shell_error_unchanged(self) -> None:
"""Non-shell tools should not have command prepended."""
msg = ToolCallMessage("read_file", {"path": "/etc/passwd"})
@@ -682,7 +664,7 @@ class TestToolCallMessageShellCommand:
def test_shell_error_with_none_command(self) -> None:
"""Shell tool with None command should fall back to error-only output."""
msg = ToolCallMessage("shell", {"command": None})
msg = ToolCallMessage("execute", {"command": None})
error = "Some error"
msg.set_error(error)
@@ -691,7 +673,7 @@ class TestToolCallMessageShellCommand:
def test_shell_error_with_empty_command(self) -> None:
"""Shell tool with empty command should fall back to error-only output."""
msg = ToolCallMessage("shell", {"command": ""})
msg = ToolCallMessage("execute", {"command": ""})
error = "Some error"
msg.set_error(error)
@@ -700,7 +682,7 @@ class TestToolCallMessageShellCommand:
def test_shell_error_with_whitespace_command(self) -> None:
"""Shell tool with whitespace command should fall back to error-only output."""
msg = ToolCallMessage("shell", {"command": " "})
msg = ToolCallMessage("execute", {"command": " "})
error = "Some error"
msg.set_error(error)
@@ -708,7 +690,7 @@ class TestToolCallMessageShellCommand:
def test_shell_error_with_no_command_key(self) -> None:
"""Shell tool with no command key should fall back to error-only output."""
msg = ToolCallMessage("shell", {"other_arg": "value"})
msg = ToolCallMessage("execute", {"other_arg": "value"})
error = "Some error"
msg.set_error(error)
@@ -717,7 +699,7 @@ class TestToolCallMessageShellCommand:
def test_format_shell_output_styles_only_first_line_dim(self) -> None:
"""Shell output formatting should only style the first command line in dim."""
msg = ToolCallMessage("shell", {"command": "echo test"})
msg = ToolCallMessage("execute", {"command": "echo test"})
output = "$ echo test\ntest output\n$ not a command"
result = msg._format_shell_output(output, is_preview=False)
@@ -731,6 +713,56 @@ class TestToolCallMessageShellCommand:
assert "dim" not in lines[2].markup
class TestToolCallMessageAwaitingApproval:
"""Tests for `set_awaiting_approval` / `clear_awaiting_approval`."""
def test_set_awaiting_approval_hides_widget(self) -> None:
"""`set_awaiting_approval` should mark the widget as hidden."""
msg = ToolCallMessage("execute", {"command": "echo hi"})
assert msg._awaiting_approval is False
msg.set_awaiting_approval()
assert msg._awaiting_approval is True
assert msg.display is False
def test_clear_awaiting_approval_restores_widget(self) -> None:
"""`clear_awaiting_approval` should restore visibility."""
msg = ToolCallMessage("execute", {"command": "echo hi"})
msg.set_awaiting_approval()
msg.clear_awaiting_approval()
assert msg._awaiting_approval is False
assert msg.display is True
def test_clear_awaiting_approval_no_op_when_not_set(self) -> None:
"""Clearing before setting should not touch widget visibility."""
msg = ToolCallMessage("execute", {"command": "echo hi"})
msg.clear_awaiting_approval()
assert msg._awaiting_approval is False
async def test_awaiting_approval_round_trip_in_mounted_widget(self) -> None:
"""Mounted widget should hide on set, reappear on clear."""
from textual.app import App, ComposeResult
class _Harness(App[None]):
def __init__(self) -> None:
super().__init__()
self.msg = ToolCallMessage("execute", {"command": "echo hi"})
def compose(self) -> ComposeResult:
yield self.msg
app = _Harness()
async with app.run_test() as pilot:
await pilot.pause()
msg = app.msg
assert msg.display is True
msg.set_awaiting_approval()
await pilot.pause()
assert msg.display is False
msg.clear_awaiting_approval()
await pilot.pause()
assert msg.display is True
class TestUserMessageHighlighting:
"""Test UserMessage highlighting of `@mentions` and `/commands`."""
libs/cli/tests/unit_tests/test_non_interactive.py
+3 -6
View File
@@ -132,15 +132,12 @@ class TestMakeHitlDecision:
)
assert result == {"type": "approve"}
@pytest.mark.parametrize("tool_name", ["bash", "shell", "execute"])
def test_all_shell_tool_names_recognised(
self, tool_name: str, console: Console
) -> None:
"""All SHELL_TOOL_NAMES variants should be gated by the allow-list."""
def test_execute_tool_gated_by_allow_list(self, console: Console) -> None:
"""The `execute` shell tool is gated by the allow-list."""
with patch("deepagents_cli.non_interactive.settings") as mock_settings:
mock_settings.shell_allow_list = ["ls"]
result = _make_hitl_decision(
{"name": tool_name, "args": {"command": "rm -rf /"}}, console
{"name": "execute", "args": {"command": "rm -rf /"}}, console
)
assert result["type"] == "reject"
libs/cli/tests/unit_tests/test_textual_adapter.py
+206
View File
@@ -1139,6 +1139,212 @@ class TestExecuteTaskTextualTextThenToolSpinner:
)
class TestExecuteTaskTextualHITLShellSuppression:
"""Tests for shell-tool widget suppression during HITL approval."""
async def _run_with_decision(
self,
*,
tool_call_name: str,
tool_call_id: str,
approval_decision: dict[str, Any],
extra_tool_calls: list[tuple[str, dict[str, Any], str]] | None = None,
) -> tuple[
TextualUIAdapter,
list[object],
dict[str, tuple[bool, bool]],
]:
"""Drive a HITL flow and snapshot widget visibility during the await.
Returns the adapter, the mounted widgets, and a mapping of
`tool_call_id -> (display, _awaiting_approval)` captured while the
approval future is pending.
"""
mounted: list[object] = []
snapshots: dict[str, tuple[bool, bool]] = {}
async def mount_message(widget: object) -> None:
await asyncio.sleep(0)
mounted.append(widget)
future: asyncio.Future[object] = asyncio.Future()
async def request_approval(
_action_requests: list[dict[str, Any]],
_assistant_id: str | None,
) -> asyncio.Future[object]:
await asyncio.sleep(0)
for tid, tool_msg in adapter._current_tool_messages.items():
snapshots[tid] = (
bool(tool_msg.display),
tool_msg._awaiting_approval,
)
future.set_result(approval_decision)
return future
message_chunks: list[tuple[Any, ...]] = [
(
(),
"messages",
(
_tool_call_message(
tool_call_name, {"command": "echo hi"}, tool_call_id
),
{},
),
)
]
for name, args, tid in extra_tool_calls or []:
message_chunks.append(
((), "messages", (_tool_call_message(name, args, tid), {}))
)
action_requests = [{"name": tool_call_name, "args": {"command": "echo hi"}}]
for name, args, _tid in extra_tool_calls or []:
action_requests.append({"name": name, "args": args})
agent = _SequencedAgent(
streams_by_call=[
[
*message_chunks,
_hitl_interrupt_chunk(
{
"action_requests": action_requests,
"review_configs": [
{
"action_name": req["name"],
"allowed_decisions": ["approve", "reject"],
}
for req in action_requests
],
}
),
],
[],
]
)
adapter = TextualUIAdapter(
mount_message=mount_message,
update_status=_noop_status,
request_approval=request_approval,
)
await execute_task_textual(
user_input="hello",
agent=agent,
assistant_id="assistant",
session_state=SimpleNamespace(thread_id="thread-1", auto_approve=False),
adapter=adapter,
)
return adapter, mounted, snapshots
async def test_shell_tool_widget_suppressed_during_approval(self) -> None:
"""`execute` widget should be hidden during the await and restored after."""
_adapter, mounted, snapshots = await self._run_with_decision(
tool_call_name="execute",
tool_call_id="tool-shell",
approval_decision={"type": "approve"},
)
tool_rows = [w for w in mounted if isinstance(w, ToolCallMessage)]
assert len(tool_rows) == 1
# While the future was pending, the widget was hidden.
assert snapshots["tool-shell"] == (False, True)
# After the finally block, it was restored.
assert tool_rows[0].display is True
assert tool_rows[0]._awaiting_approval is False
async def test_non_shell_tool_widget_not_suppressed(self) -> None:
"""`read_file` widget should stay visible — only shell tools are hidden."""
_adapter, mounted, snapshots = await self._run_with_decision(
tool_call_name="read_file",
tool_call_id="tool-read",
approval_decision={"type": "approve"},
)
tool_rows = [w for w in mounted if isinstance(w, ToolCallMessage)]
assert len(tool_rows) == 1
# Visible the whole time, never marked as awaiting approval.
assert snapshots["tool-read"] == (True, False)
assert tool_rows[0].display is True
assert tool_rows[0]._awaiting_approval is False
async def test_mixed_batch_only_shell_suppressed(self) -> None:
"""Parallel shell + non-shell tools: only the shell row is hidden."""
_adapter, _mounted, snapshots = await self._run_with_decision(
tool_call_name="execute",
tool_call_id="tool-shell",
approval_decision={"type": "approve"},
extra_tool_calls=[("read_file", {"path": "notes.txt"}, "tool-read")],
)
assert snapshots["tool-shell"] == (False, True)
assert snapshots["tool-read"] == (True, False)
async def test_shell_widget_restored_when_approval_raises(self) -> None:
"""`finally` must restore the widget even if approval raises."""
mounted: list[object] = []
async def mount_message(widget: object) -> None:
await asyncio.sleep(0)
mounted.append(widget)
async def request_approval(
_action_requests: list[dict[str, Any]],
_assistant_id: str | None,
) -> asyncio.Future[object]:
await asyncio.sleep(0)
msg = "boom"
raise RuntimeError(msg)
agent = _SequencedAgent(
streams_by_call=[
[
(
(),
"messages",
(
_tool_call_message(
"execute", {"command": "echo hi"}, "tool-shell"
),
{},
),
),
_hitl_interrupt_chunk(
{
"action_requests": [
{"name": "execute", "args": {"command": "echo hi"}}
],
"review_configs": [
{
"action_name": "execute",
"allowed_decisions": ["approve", "reject"],
}
],
}
),
],
[],
]
)
adapter = TextualUIAdapter(
mount_message=mount_message,
update_status=_noop_status,
request_approval=request_approval,
)
with pytest.raises(RuntimeError, match="boom"):
await execute_task_textual(
user_input="hello",
agent=agent,
assistant_id="assistant",
session_state=SimpleNamespace(thread_id="thread-1", auto_approve=False),
adapter=adapter,
)
tool_rows = [w for w in mounted if isinstance(w, ToolCallMessage)]
assert len(tool_rows) == 1
assert tool_rows[0].display is True
assert tool_rows[0]._awaiting_approval is False
class TestExecuteTaskTextualAskUser:
"""Tests for ask_user interrupt handling in the Textual adapter."""