mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-21 17:25:26 -04:00
fix(code): gate delete file operations (#4299)
`delete` now follows the same approval, display, tracking, and memory-protection paths as other mutating file tools. This closes the gap where destructive deletes could bypass file-operation previews and managed memory protections. ## Changes - Register `delete` as a write-capable tool for PTC auditing and add it to the human approval interrupt map. - Add delete-specific approval previews that show removed file content as a unified diff when possible, or flag directories and unreadable paths explicitly. - Track completed delete operations in `FileOpTracker`, modeling successful file deletion as an empty after-state so removed lines appear in operation diffs. - Extend `ManagedMemoryGuardMiddleware` to reject deletes of guarded memory files, including parent-directory deletes that would remove a managed memory block. - Display `delete` calls consistently with other file tools in message headers and tool labels.
This commit is contained in:
@@ -217,7 +217,7 @@ class ShellAllowListMiddleware(AgentMiddleware):
|
||||
|
||||
|
||||
_INTERPRETER_WRITE_TOOLS: frozenset[str] = frozenset(
|
||||
{"execute", "write_file", "edit_file"}
|
||||
{"execute", "write_file", "edit_file", "delete"}
|
||||
)
|
||||
"""Tools considered write/shell capable for PTC auditing.
|
||||
|
||||
@@ -910,6 +910,17 @@ def _format_edit_file_description(
|
||||
return f"Action: Replace text ({scope})"
|
||||
|
||||
|
||||
def _format_delete_description(
|
||||
_tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
|
||||
) -> str:
|
||||
"""Format delete tool call for approval prompt.
|
||||
|
||||
Returns:
|
||||
Formatted description string for the delete tool call.
|
||||
"""
|
||||
return "Action: Delete file or directory"
|
||||
|
||||
|
||||
def _format_web_search_description(
|
||||
tool_call: ToolCall, _state: AgentState[Any], _runtime: Runtime[Any]
|
||||
) -> str:
|
||||
@@ -1146,6 +1157,12 @@ def _add_interrupt_on() -> dict[str, InterruptOnConfig]:
|
||||
"when": _should_interrupt_tool_call,
|
||||
}
|
||||
|
||||
delete_interrupt_config: InterruptOnConfig = {
|
||||
"allowed_decisions": ["approve", "reject"],
|
||||
"description": _format_delete_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
|
||||
@@ -1174,6 +1191,7 @@ def _add_interrupt_on() -> dict[str, InterruptOnConfig]:
|
||||
"execute": execute_interrupt_config,
|
||||
"write_file": write_file_interrupt_config,
|
||||
"edit_file": edit_file_interrupt_config,
|
||||
"delete": delete_interrupt_config,
|
||||
"web_search": web_search_interrupt_config,
|
||||
"fetch_url": fetch_url_interrupt_config,
|
||||
"task": task_interrupt_config,
|
||||
|
||||
@@ -210,6 +210,28 @@ def build_approval_preview(
|
||||
diff_title=f"Diff {display_path}",
|
||||
)
|
||||
|
||||
if tool_name == "delete":
|
||||
details = [f"File: {path_str}", "Action: Delete file or directory"]
|
||||
if physical_path is None:
|
||||
return ApprovalPreview(
|
||||
title=f"Delete {display_path}",
|
||||
details=details,
|
||||
error="Unable to resolve file path.",
|
||||
)
|
||||
before = _safe_read(physical_path)
|
||||
diff = None
|
||||
if before is not None:
|
||||
diff = compute_unified_diff(before, "", display_path, max_lines=100)
|
||||
details.append(f"Lines to delete: {_count_lines(before)}")
|
||||
elif physical_path.exists():
|
||||
details.append("Contents: directory or unreadable file")
|
||||
return ApprovalPreview(
|
||||
title=f"Delete {display_path}",
|
||||
details=details,
|
||||
diff=diff,
|
||||
diff_title=f"Diff {display_path}",
|
||||
)
|
||||
|
||||
if tool_name == "edit_file":
|
||||
if physical_path is None:
|
||||
return ApprovalPreview(
|
||||
@@ -287,10 +309,10 @@ class FileOpTracker:
|
||||
) -> None:
|
||||
"""Begin tracking a file operation.
|
||||
|
||||
Creates a record for the operation and, for write/edit operations,
|
||||
captures the file's content before modification.
|
||||
Creates a record for the operation and, for write/edit/delete
|
||||
operations, captures the file's content before the operation.
|
||||
"""
|
||||
if tool_name not in {"read_file", "write_file", "edit_file"}:
|
||||
if tool_name not in {"read_file", "write_file", "edit_file", "delete"}:
|
||||
return
|
||||
path_str = str(args.get("file_path") or args.get("path") or "")
|
||||
display_path = format_display_path(path_str)
|
||||
@@ -301,7 +323,7 @@ class FileOpTracker:
|
||||
tool_call_id=tool_call_id,
|
||||
args=args,
|
||||
)
|
||||
if tool_name in {"write_file", "edit_file"}:
|
||||
if tool_name in {"write_file", "edit_file", "delete"}:
|
||||
if self.backend and path_str:
|
||||
try:
|
||||
responses = self.backend.download_files([path_str])
|
||||
@@ -374,13 +396,22 @@ class FileOpTracker:
|
||||
if isinstance(limit, int) and lines > limit:
|
||||
record.metrics.end_line = (record.metrics.start_line or 1) + limit - 1
|
||||
else:
|
||||
# For write/edit operations, read back from backend (or local filesystem)
|
||||
self._populate_after_content(record)
|
||||
if record.after_content is None:
|
||||
record.status = "error"
|
||||
record.error = "Could not read updated file content."
|
||||
self._finalize(record)
|
||||
return record
|
||||
if record.tool_name == "delete":
|
||||
# Reached only after the success-status check above, so the
|
||||
# tool reported the path removed. Model an empty "after" to
|
||||
# diff the removed content against; there is nothing to read
|
||||
# back from disk. This trusts the tool's success status and
|
||||
# is sound for backends where a successful delete means the
|
||||
# path is gone.
|
||||
record.after_content = ""
|
||||
else:
|
||||
# Write/edit: read the updated content back from backend/disk.
|
||||
self._populate_after_content(record)
|
||||
if record.after_content is None:
|
||||
record.status = "error"
|
||||
record.error = "Could not read updated file content."
|
||||
self._finalize(record)
|
||||
return record
|
||||
record.metrics.lines_written = _count_lines(record.after_content)
|
||||
before_lines = _count_lines(record.before_content or "")
|
||||
diff = compute_unified_diff(
|
||||
|
||||
@@ -9,12 +9,16 @@ that file to persist learnings, nothing stops it from rewriting the managed
|
||||
block.
|
||||
|
||||
This middleware intercepts `write_file`/`edit_file` calls targeting the guarded
|
||||
file(s). When a call would change or remove the managed block, the model's other
|
||||
edits are kept (though surrounding whitespace may be normalized, and a fully
|
||||
removed block is re-appended rather than restored in place) while the managed
|
||||
block is restored, and an error is returned so the model learns the region is
|
||||
machine-managed. When the block was altered but the restore could not be
|
||||
completed, an error is still returned so the failure is never silent.
|
||||
file(s), and `delete` calls that would remove them. When a write or edit would
|
||||
change or remove the managed block, the model's other edits are kept (though
|
||||
surrounding whitespace may be normalized, and a fully removed block is
|
||||
re-appended rather than restored in place) while the managed block is restored,
|
||||
and an error is returned so the model learns the region is machine-managed. A
|
||||
`delete` call that would remove an existing managed block is rejected before the
|
||||
tool runs; a `delete` of a guarded file that exists but cannot be read is also
|
||||
rejected, failing closed rather than removing a file we cannot inspect. When the
|
||||
block was altered but the restore could not be completed, an error is still
|
||||
returned so the failure is never silent.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -44,7 +48,7 @@ if TYPE_CHECKING:
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_GUARDED_TOOLS: frozenset[str] = frozenset({"write_file", "edit_file"})
|
||||
_GUARDED_TOOLS: frozenset[str] = frozenset({"write_file", "edit_file", "delete"})
|
||||
"""Tool names whose calls can mutate a guarded file and must be inspected."""
|
||||
|
||||
_REJECTION_MESSAGE = (
|
||||
@@ -66,6 +70,14 @@ _RESTORE_FAILED_MESSAGE = (
|
||||
)
|
||||
"""Error returned when a managed-block edit could not be reverted."""
|
||||
|
||||
_DELETE_REJECTION_MESSAGE = (
|
||||
"The guarded memory file {path} contains a machine-managed region between "
|
||||
"the `deepagents:onboarding-name:start` and `deepagents:onboarding-name:end` "
|
||||
"markers and must not be deleted. Do not delete this file or a parent "
|
||||
"directory that contains it."
|
||||
)
|
||||
"""Error returned when a delete would remove a managed memory block."""
|
||||
|
||||
|
||||
class _RestoreOutcome(Enum):
|
||||
"""Result of attempting to restore a managed block after a tool call."""
|
||||
@@ -86,8 +98,10 @@ class ManagedMemoryGuardMiddleware(AgentMiddleware):
|
||||
Guards the managed onboarding-name block in a fixed set of memory files. A
|
||||
`write_file`/`edit_file` that leaves the managed block untouched passes
|
||||
through; one that alters or drops it has the block restored (other edits
|
||||
kept) and returns an error. If the restore itself fails, an error is still
|
||||
returned so the failure is never silent.
|
||||
kept) and returns an error. A `delete` targeting a guarded file (or a parent
|
||||
directory that contains one) is rejected outright before the tool runs when
|
||||
the file holds a managed block or exists but cannot be read. If the restore
|
||||
itself fails, an error is still returned so the failure is never silent.
|
||||
"""
|
||||
|
||||
def __init__(self, guarded_paths: Iterable[str | Path]) -> None:
|
||||
@@ -125,7 +139,8 @@ class ManagedMemoryGuardMiddleware(AgentMiddleware):
|
||||
Returns:
|
||||
The matching guarded `Path`, or `None` when the call is unrelated.
|
||||
"""
|
||||
if request.tool_call["name"] not in _GUARDED_TOOLS:
|
||||
tool_name = request.tool_call["name"]
|
||||
if tool_name not in _GUARDED_TOOLS:
|
||||
return None
|
||||
args = request.tool_call.get("args") or {}
|
||||
file_path = args.get("file_path")
|
||||
@@ -139,10 +154,17 @@ class ManagedMemoryGuardMiddleware(AgentMiddleware):
|
||||
logger.warning(
|
||||
"Could not resolve target path %r for %s",
|
||||
file_path,
|
||||
request.tool_call["name"],
|
||||
tool_name,
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
if tool_name == "delete":
|
||||
# `is_relative_to` is True when the guarded file is the delete
|
||||
# target itself or lives under a directory being deleted.
|
||||
for guarded in self._guarded:
|
||||
if guarded.is_relative_to(resolved):
|
||||
return guarded
|
||||
return None
|
||||
return resolved if resolved in self._guarded else None
|
||||
|
||||
@staticmethod
|
||||
@@ -336,6 +358,44 @@ class ManagedMemoryGuardMiddleware(AgentMiddleware):
|
||||
status="error",
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _reject_delete(path: Path, before: str | None) -> bool:
|
||||
"""Return whether a delete targeting a guarded path must be rejected.
|
||||
|
||||
The caller has already matched `path` as a guarded file (or a file
|
||||
inside a directory being deleted). The delete is rejected when:
|
||||
|
||||
- the guarded file currently holds a managed block, or
|
||||
- the guarded file exists but its content could not be read.
|
||||
|
||||
The second case fails closed on purpose: `_read` returns `None` for
|
||||
both a missing file and an existing-but-unreadable one (a permission
|
||||
error, or a symlink swap caught by `O_NOFOLLOW`). A missing guarded
|
||||
file has nothing to protect and is safe to remove, but an existing
|
||||
file we cannot inspect must not be deleted irreversibly on the
|
||||
assumption that it lacks a managed block.
|
||||
|
||||
Returns:
|
||||
`True` when the delete must be blocked, `False` when it may proceed.
|
||||
"""
|
||||
if before is not None:
|
||||
return extract_onboarding_name_block(before) is not None
|
||||
return path.exists()
|
||||
|
||||
@staticmethod
|
||||
def _delete_error(request: ToolCallRequest, path: Path) -> ToolMessage:
|
||||
"""Build the error result returned when a delete would remove memory.
|
||||
|
||||
Returns:
|
||||
An error-status `ToolMessage` explaining the protected file.
|
||||
"""
|
||||
return ToolMessage(
|
||||
content=_DELETE_REJECTION_MESSAGE.format(path=path),
|
||||
name=request.tool_call["name"],
|
||||
tool_call_id=request.tool_call["id"],
|
||||
status="error",
|
||||
)
|
||||
|
||||
def _result_after_restore(
|
||||
self,
|
||||
request: ToolCallRequest,
|
||||
@@ -372,12 +432,16 @@ class ManagedMemoryGuardMiddleware(AgentMiddleware):
|
||||
if path is None:
|
||||
return handler(request)
|
||||
before = self._read(path)
|
||||
if request.tool_call["name"] == "delete":
|
||||
if self._reject_delete(path, before):
|
||||
return self._delete_error(request, path)
|
||||
return handler(request)
|
||||
before_block = (
|
||||
extract_onboarding_name_block(before) if before is not None else None
|
||||
)
|
||||
result = handler(request)
|
||||
if before is None or before_block is None:
|
||||
return result
|
||||
return handler(request)
|
||||
result = handler(request)
|
||||
return self._result_after_restore(request, path, before, before_block, result)
|
||||
|
||||
async def awrap_tool_call(
|
||||
@@ -395,12 +459,16 @@ class ManagedMemoryGuardMiddleware(AgentMiddleware):
|
||||
if path is None:
|
||||
return await handler(request)
|
||||
before = await asyncio.to_thread(self._read, path)
|
||||
if request.tool_call["name"] == "delete":
|
||||
if await asyncio.to_thread(self._reject_delete, path, before):
|
||||
return self._delete_error(request, path)
|
||||
return await handler(request)
|
||||
before_block = (
|
||||
extract_onboarding_name_block(before) if before is not None else None
|
||||
)
|
||||
result = await handler(request)
|
||||
if before is None or before_block is None:
|
||||
return result
|
||||
return await handler(request)
|
||||
result = await handler(request)
|
||||
return await asyncio.to_thread(
|
||||
self._result_after_restore, request, path, before, before_block, result
|
||||
)
|
||||
|
||||
@@ -1363,6 +1363,7 @@ async def execute_task_textual(
|
||||
if tool_name in {
|
||||
"write_file",
|
||||
"edit_file",
|
||||
"delete",
|
||||
}:
|
||||
args = action_request.get("args", {})
|
||||
if isinstance(args, dict):
|
||||
@@ -1385,6 +1386,7 @@ async def execute_task_textual(
|
||||
if tool_name in {
|
||||
"write_file",
|
||||
"edit_file",
|
||||
"delete",
|
||||
}:
|
||||
args = action_request.get("args", {})
|
||||
if isinstance(args, dict):
|
||||
|
||||
@@ -189,7 +189,7 @@ def format_tool_display(tool_name: str, tool_args: dict) -> str:
|
||||
return path.name
|
||||
|
||||
# Tool-specific formatting - show the most important argument(s)
|
||||
if tool_name in {"read_file", "write_file", "edit_file"}:
|
||||
if tool_name in {"read_file", "write_file", "edit_file", "delete"}:
|
||||
# File operations: show the primary file path argument (file_path or path)
|
||||
path_value = tool_args.get("file_path")
|
||||
if path_value is None:
|
||||
|
||||
@@ -138,7 +138,7 @@ class ApprovalMenu(Container):
|
||||
def __init__(
|
||||
self,
|
||||
action_requests: list[dict[str, Any]] | dict[str, Any],
|
||||
_assistant_id: str | None = None,
|
||||
assistant_id: str | None = None,
|
||||
id: str | None = None, # noqa: A002 # Textual widget constructor uses `id` parameter
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
@@ -148,8 +148,8 @@ class ApprovalMenu(Container):
|
||||
action_requests: A single action request dictionary or a list of action
|
||||
request dictionaries requiring approval. Each dictionary should
|
||||
contain 'name' (tool name) and 'args' (tool arguments).
|
||||
_assistant_id: Optional assistant ID (currently unused, reserved for
|
||||
future use).
|
||||
assistant_id: Optional assistant ID for resolving virtual paths in
|
||||
file-operation previews.
|
||||
id: Optional widget ID. Defaults to 'approval-menu'.
|
||||
**kwargs: Additional keyword arguments passed to the Container base class.
|
||||
"""
|
||||
@@ -160,6 +160,7 @@ class ApprovalMenu(Container):
|
||||
else:
|
||||
self._action_requests = action_requests
|
||||
|
||||
self._assistant_id = assistant_id
|
||||
# For display purposes, get tool names
|
||||
self._tool_names = [r.get("name", "unknown") for r in self._action_requests]
|
||||
self._selected = 0
|
||||
@@ -397,7 +398,9 @@ class ApprovalMenu(Container):
|
||||
|
||||
# Get the appropriate renderer for this tool
|
||||
renderer = get_renderer(tool_name)
|
||||
widget_class, data = renderer.get_approval_widget(tool_args)
|
||||
widget_class, data = renderer.get_approval_widget(
|
||||
tool_args, assistant_id=self._assistant_id
|
||||
)
|
||||
approval_widget = widget_class(data)
|
||||
await self._tool_info_container.mount(approval_widget)
|
||||
|
||||
|
||||
@@ -112,6 +112,7 @@ _TOOLS_WITH_HEADER_INFO: set[str] = {
|
||||
"read_file",
|
||||
"write_file",
|
||||
"edit_file",
|
||||
"delete",
|
||||
"glob",
|
||||
"grep",
|
||||
"execute", # sandbox shell
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import difflib
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from deepagents_code.file_ops import build_approval_preview, format_display_path
|
||||
from deepagents_code.widgets.tool_widgets import (
|
||||
EditFileApprovalWidget,
|
||||
GenericApprovalWidget,
|
||||
@@ -27,11 +28,13 @@ class ToolRenderer:
|
||||
@staticmethod
|
||||
def get_approval_widget(
|
||||
tool_args: dict[str, Any],
|
||||
assistant_id: str | None = None, # noqa: ARG004
|
||||
) -> tuple[type[ToolApprovalWidget], dict[str, Any]]:
|
||||
"""Get the approval widget class and data for this tool.
|
||||
|
||||
Args:
|
||||
tool_args: The tool arguments from action_request
|
||||
tool_args: The tool arguments from action_request.
|
||||
assistant_id: Optional assistant identifier for resolving virtual paths.
|
||||
|
||||
Returns:
|
||||
Tuple of (widget_class, data_dict)
|
||||
@@ -45,6 +48,7 @@ class WriteFileRenderer(ToolRenderer):
|
||||
@staticmethod
|
||||
def get_approval_widget( # noqa: D102 # Protocol method — docstring on base class
|
||||
tool_args: dict[str, Any],
|
||||
assistant_id: str | None = None, # noqa: ARG004
|
||||
) -> tuple[type[ToolApprovalWidget], dict[str, Any]]:
|
||||
# Extract file extension for syntax highlighting
|
||||
file_path = tool_args.get("file_path", "")
|
||||
@@ -69,16 +73,52 @@ class TaskRenderer(ToolRenderer):
|
||||
@staticmethod
|
||||
def get_approval_widget( # noqa: D102 # Protocol method — docstring on base class
|
||||
tool_args: dict[str, Any], # noqa: ARG004 # Unused; interrupt description already formats task args
|
||||
assistant_id: str | None = None, # noqa: ARG004
|
||||
) -> tuple[type[ToolApprovalWidget], dict[str, Any]]:
|
||||
return GenericApprovalWidget, {}
|
||||
|
||||
|
||||
class DeleteFileRenderer(ToolRenderer):
|
||||
"""Renderer for delete tool - shows removed file content when available."""
|
||||
|
||||
@staticmethod
|
||||
def get_approval_widget( # noqa: D102 # Protocol method — docstring on base class
|
||||
tool_args: dict[str, Any],
|
||||
assistant_id: str | None = None,
|
||||
) -> tuple[type[ToolApprovalWidget], dict[str, Any]]:
|
||||
path = str(tool_args.get("file_path") or tool_args.get("path") or "")
|
||||
preview = build_approval_preview(
|
||||
"delete", {"file_path": path}, assistant_id=assistant_id
|
||||
)
|
||||
if preview is None:
|
||||
# `build_approval_preview` always returns a preview for "delete";
|
||||
# this guards its `ApprovalPreview | None` contract defensively.
|
||||
return GenericApprovalWidget, tool_args
|
||||
if preview.diff:
|
||||
return EditFileApprovalWidget, {
|
||||
"file_path": format_display_path(path),
|
||||
"diff_lines": preview.diff.splitlines(),
|
||||
"old_string": "",
|
||||
"new_string": "",
|
||||
}
|
||||
data: dict[str, Any] = {"file_path": format_display_path(path)}
|
||||
details = [
|
||||
detail for detail in preview.details if not detail.startswith("File:")
|
||||
]
|
||||
if details:
|
||||
data["details"] = "\n".join(details)
|
||||
if preview.error:
|
||||
data["error"] = preview.error
|
||||
return GenericApprovalWidget, data
|
||||
|
||||
|
||||
class EditFileRenderer(ToolRenderer):
|
||||
"""Renderer for edit_file tool - shows unified diff."""
|
||||
|
||||
@staticmethod
|
||||
def get_approval_widget( # noqa: D102 # Protocol method — docstring on base class
|
||||
tool_args: dict[str, Any],
|
||||
assistant_id: str | None = None, # noqa: ARG004
|
||||
) -> tuple[type[ToolApprovalWidget], dict[str, Any]]:
|
||||
file_path = tool_args.get("file_path", "")
|
||||
old_string = tool_args.get("old_string", "")
|
||||
@@ -127,6 +167,7 @@ _RENDERER_REGISTRY: dict[str, type[ToolRenderer]] = {
|
||||
"task": TaskRenderer,
|
||||
"write_file": WriteFileRenderer,
|
||||
"edit_file": EditFileRenderer,
|
||||
"delete": DeleteFileRenderer,
|
||||
}
|
||||
"""Registry mapping tool names to renderers
|
||||
|
||||
|
||||
@@ -24,6 +24,7 @@ from deepagents_code.agent import (
|
||||
DEFAULT_AGENT_NAME,
|
||||
_add_interrupt_on,
|
||||
_apply_inherited_pythonpath,
|
||||
_format_delete_description,
|
||||
_format_edit_file_description,
|
||||
_format_execute_description,
|
||||
_format_fetch_url_description,
|
||||
@@ -419,6 +420,30 @@ def test_format_edit_file_description_all_occurrences():
|
||||
assert "File:" not in description
|
||||
|
||||
|
||||
def test_format_delete_description() -> None:
|
||||
"""Test delete description for approval prompts."""
|
||||
tool_call = cast(
|
||||
"ToolCall",
|
||||
{"name": "delete", "args": {"file_path": "/path/to/file.py"}, "id": "call-5"},
|
||||
)
|
||||
|
||||
description = _format_delete_description(
|
||||
tool_call, cast("AgentState[Any]", None), cast("Runtime[Any]", None)
|
||||
)
|
||||
|
||||
assert "Action: Delete file or directory" in description
|
||||
|
||||
|
||||
def test_add_interrupt_on_gates_delete() -> None:
|
||||
"""The destructive delete tool is approval-gated like other write tools."""
|
||||
interrupt_map = _add_interrupt_on()
|
||||
|
||||
assert "delete" in interrupt_map
|
||||
assert interrupt_map["delete"]["allowed_decisions"] == ["approve", "reject"]
|
||||
assert interrupt_map["delete"]["description"] is _format_delete_description
|
||||
assert interrupt_map["delete"]["when"] is _should_interrupt_tool_call
|
||||
|
||||
|
||||
def test_format_web_search_description():
|
||||
"""Test web_search description formatting."""
|
||||
tool_call = cast(
|
||||
@@ -3412,12 +3437,17 @@ class TestResolvePtcOption:
|
||||
"""Write."""
|
||||
return ""
|
||||
|
||||
@tool
|
||||
def delete(path: str) -> str: # noqa: ARG001
|
||||
"""Delete."""
|
||||
return ""
|
||||
|
||||
@tool
|
||||
def grep(pattern: str) -> str: # noqa: ARG001
|
||||
"""Search."""
|
||||
return ""
|
||||
|
||||
return [read_file, write_file, grep]
|
||||
return [read_file, write_file, delete, grep]
|
||||
|
||||
def test_false_returns_none(self) -> None:
|
||||
from deepagents_code.agent import _resolve_ptc_option
|
||||
@@ -3474,7 +3504,7 @@ class TestResolvePtcOption:
|
||||
assert result is not None
|
||||
# `all` enumerates only the tools passed to `create_cli_agent`; SDK
|
||||
# runtime built-ins are injected later and are not enumerable here.
|
||||
assert sorted(result) == ["grep", "read_file", "write_file"]
|
||||
assert sorted(result) == ["delete", "grep", "read_file", "write_file"]
|
||||
|
||||
@staticmethod
|
||||
def _tools_with_task() -> list:
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import shutil
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
@@ -100,6 +101,27 @@ def test_tracker_records_edit_diff(tmp_path: Path) -> None:
|
||||
assert '+ return "hi"' in record.diff
|
||||
|
||||
|
||||
def test_tracker_records_delete_diff(tmp_path: Path) -> None:
|
||||
tracker = FileOpTracker(assistant_id=None)
|
||||
file_path = tmp_path / "old.txt"
|
||||
file_path.write_text("alpha\nbeta\n")
|
||||
|
||||
tracker.start_operation("delete", {"file_path": str(file_path)}, "delete-1")
|
||||
file_path.unlink()
|
||||
|
||||
message = ToolMessage(
|
||||
content=f"Deleted {file_path}", tool_call_id="delete-1", name="delete"
|
||||
)
|
||||
record = tracker.complete_with_message(message)
|
||||
|
||||
assert record is not None
|
||||
assert record.status == "success"
|
||||
assert record.metrics.lines_removed == 2
|
||||
assert record.diff is not None
|
||||
assert "-alpha" in record.diff
|
||||
assert "-beta" in record.diff
|
||||
|
||||
|
||||
def test_build_approval_preview_generates_diff(tmp_path: Path) -> None:
|
||||
target = tmp_path / "notes.txt"
|
||||
target.write_text("alpha\nbeta\n")
|
||||
@@ -118,3 +140,71 @@ def test_build_approval_preview_generates_diff(tmp_path: Path) -> None:
|
||||
assert preview is not None
|
||||
assert preview.diff is not None
|
||||
assert "+gamma" in preview.diff
|
||||
|
||||
|
||||
def test_build_delete_approval_preview_shows_removed_content(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
target = tmp_path / "notes.txt"
|
||||
target.write_text("alpha\nbeta\n")
|
||||
|
||||
preview = build_approval_preview(
|
||||
"delete",
|
||||
{"file_path": str(target)},
|
||||
assistant_id=None,
|
||||
)
|
||||
|
||||
assert preview is not None
|
||||
assert preview.title == "Delete notes.txt"
|
||||
assert "Action: Delete file or directory" in preview.details
|
||||
assert "Lines to delete: 2" in preview.details
|
||||
assert preview.diff is not None
|
||||
assert "-alpha" in preview.diff
|
||||
|
||||
|
||||
def test_tracker_records_directory_delete(tmp_path: Path) -> None:
|
||||
"""A recursive directory delete is tracked as a success without a diff."""
|
||||
target = tmp_path / "subdir"
|
||||
target.mkdir()
|
||||
(target / "child.txt").write_text("data\n")
|
||||
|
||||
tracker = FileOpTracker(assistant_id=None)
|
||||
tracker.start_operation("delete", {"file_path": str(target)}, "delete-dir")
|
||||
# Directory has no readable text content, so no before/after to diff.
|
||||
shutil.rmtree(target)
|
||||
|
||||
message = ToolMessage(
|
||||
content=f"Deleted {target}", tool_call_id="delete-dir", name="delete"
|
||||
)
|
||||
record = tracker.complete_with_message(message)
|
||||
|
||||
assert record is not None
|
||||
assert record.status == "success"
|
||||
assert record.metrics.lines_removed == 0
|
||||
assert not record.diff
|
||||
|
||||
|
||||
def test_build_delete_approval_preview_for_directory(tmp_path: Path) -> None:
|
||||
"""The delete preview flags directories instead of rendering a diff."""
|
||||
target = tmp_path / "subdir"
|
||||
target.mkdir()
|
||||
(target / "child.txt").write_text("data\n")
|
||||
|
||||
preview = build_approval_preview(
|
||||
"delete",
|
||||
{"file_path": str(target)},
|
||||
assistant_id=None,
|
||||
)
|
||||
|
||||
assert preview is not None
|
||||
assert preview.title == "Delete subdir"
|
||||
assert "Contents: directory or unreadable file" in preview.details
|
||||
assert preview.diff is None
|
||||
|
||||
|
||||
def test_build_delete_approval_preview_unresolvable_path() -> None:
|
||||
"""An empty path yields an explicit resolution error, not a blank preview."""
|
||||
preview = build_approval_preview("delete", {"file_path": ""}, assistant_id=None)
|
||||
|
||||
assert preview is not None
|
||||
assert preview.error == "Unable to resolve file path."
|
||||
|
||||
@@ -315,6 +315,118 @@ def test_file_created_with_block_passes_through(tmp_path) -> None:
|
||||
assert '- The user\'s preferred name is "Ada".' in path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_delete_guarded_file_with_block_is_rejected_before_tool_runs(tmp_path) -> None:
|
||||
"""Deleting a guarded memory file with a managed block is blocked."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
_managed_file(path, "Ada")
|
||||
middleware = ManagedMemoryGuardMiddleware([str(path)])
|
||||
handler = cast("Any", lambda _request: pytest.fail("delete should not run"))
|
||||
|
||||
result = middleware.wrap_tool_call(_request("delete", str(path)), handler)
|
||||
|
||||
assert isinstance(result, ToolMessage)
|
||||
assert result.status == "error"
|
||||
assert result.name == "delete"
|
||||
assert result.tool_call_id == "call-1"
|
||||
# The rejection must come from the delete-specific template, not the
|
||||
# write/edit restore message (which talks about "other changes kept").
|
||||
assert "must not be deleted" in result.content
|
||||
assert path.exists()
|
||||
assert '- The user\'s preferred name is "Ada".' in path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
async def test_async_delete_guarded_file_with_block_is_rejected(tmp_path) -> None:
|
||||
"""The async wrapper rejects guarded deletes like the sync path."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
_managed_file(path, "Ada")
|
||||
middleware = ManagedMemoryGuardMiddleware([str(path)])
|
||||
|
||||
async def handler(_request: ToolCallRequest) -> ToolMessage: # noqa: RUF029
|
||||
pytest.fail("delete should not run")
|
||||
|
||||
result = await middleware.awrap_tool_call(_request("delete", str(path)), handler)
|
||||
|
||||
assert isinstance(result, ToolMessage)
|
||||
assert result.status == "error"
|
||||
assert result.name == "delete"
|
||||
assert "must not be deleted" in result.content
|
||||
assert path.exists()
|
||||
|
||||
|
||||
def test_delete_parent_of_guarded_file_with_block_is_rejected(tmp_path) -> None:
|
||||
"""A recursive delete of a parent directory must not remove managed memory."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
_managed_file(path, "Ada")
|
||||
middleware = ManagedMemoryGuardMiddleware([str(path)])
|
||||
handler = cast("Any", lambda _request: pytest.fail("delete should not run"))
|
||||
|
||||
result = middleware.wrap_tool_call(_request("delete", str(path.parent)), handler)
|
||||
|
||||
assert isinstance(result, ToolMessage)
|
||||
assert result.status == "error"
|
||||
assert path.exists()
|
||||
|
||||
|
||||
def test_delete_guarded_file_without_block_is_allowed(tmp_path) -> None:
|
||||
"""A guarded path with no managed block is not delete-protected."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
path.parent.mkdir(parents=True)
|
||||
path.write_text("## Notes\n\nfreeform\n", encoding="utf-8")
|
||||
middleware = ManagedMemoryGuardMiddleware([str(path)])
|
||||
deleted = False
|
||||
|
||||
def handler(_request: ToolCallRequest) -> ToolMessage:
|
||||
nonlocal deleted
|
||||
path.unlink()
|
||||
deleted = True
|
||||
return ToolMessage(
|
||||
content=f"Deleted {path}", name="delete", tool_call_id="call-1"
|
||||
)
|
||||
|
||||
result = middleware.wrap_tool_call(_request("delete", str(path)), handler)
|
||||
|
||||
assert isinstance(result, ToolMessage)
|
||||
assert deleted is True
|
||||
assert not path.exists()
|
||||
|
||||
|
||||
def test_delete_unreadable_guarded_file_fails_closed(tmp_path) -> None:
|
||||
"""An existing-but-unreadable guarded file is not deleted (fail closed)."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
path.parent.mkdir(parents=True)
|
||||
# Invalid UTF-8 bytes make `_read` return None while the file still
|
||||
# exists, so the guard cannot confirm the file lacks a managed block.
|
||||
path.write_bytes(b"\xff\xfe not valid utf-8")
|
||||
middleware = ManagedMemoryGuardMiddleware([str(path)])
|
||||
handler = cast("Any", lambda _request: pytest.fail("delete should not run"))
|
||||
|
||||
result = middleware.wrap_tool_call(_request("delete", str(path)), handler)
|
||||
|
||||
assert isinstance(result, ToolMessage)
|
||||
assert result.status == "error"
|
||||
assert "must not be deleted" in result.content
|
||||
assert path.exists()
|
||||
|
||||
|
||||
def test_delete_missing_guarded_file_is_allowed(tmp_path) -> None:
|
||||
"""A guarded path that does not exist has nothing to protect, so it runs."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
middleware = ManagedMemoryGuardMiddleware([str(path)])
|
||||
ran = False
|
||||
|
||||
def handler(_request: ToolCallRequest) -> ToolMessage:
|
||||
nonlocal ran
|
||||
ran = True
|
||||
return ToolMessage(
|
||||
content=f"Deleted {path}", name="delete", tool_call_id="call-1"
|
||||
)
|
||||
|
||||
result = middleware.wrap_tool_call(_request("delete", str(path)), handler)
|
||||
|
||||
assert isinstance(result, ToolMessage)
|
||||
assert ran is True
|
||||
|
||||
|
||||
def test_error_message_propagates_call_metadata(tmp_path) -> None:
|
||||
"""The error result carries the originating tool name and call id."""
|
||||
path = tmp_path / "agent" / "AGENTS.md"
|
||||
|
||||
@@ -1644,6 +1644,12 @@ class TestToolCallMessageJsEvalArgs:
|
||||
|
||||
assert "js_eval" in _TOOLS_WITH_HEADER_INFO
|
||||
|
||||
def test_delete_in_tools_with_header_info(self) -> None:
|
||||
"""`delete` is registered so its path stays in the header only."""
|
||||
from deepagents_code.widgets.messages import _TOOLS_WITH_HEADER_INFO
|
||||
|
||||
assert "delete" in _TOOLS_WITH_HEADER_INFO
|
||||
|
||||
def test_single_line_code_not_expandable(self) -> None:
|
||||
"""One-line code is fully shown in the header — nothing to expand."""
|
||||
msg = ToolCallMessage("js_eval", {"code": "1 + 1"})
|
||||
|
||||
@@ -181,19 +181,25 @@ class TestFormatToolDisplay:
|
||||
|
||||
# --- file tools ---
|
||||
|
||||
@pytest.mark.parametrize("tool_name", ["read_file", "write_file", "edit_file"])
|
||||
@pytest.mark.parametrize(
|
||||
"tool_name", ["read_file", "write_file", "edit_file", "delete"]
|
||||
)
|
||||
def test_file_tool_with_file_path(self, tool_name: str) -> None:
|
||||
result = format_tool_display(tool_name, {"file_path": "/tmp/test.py"})
|
||||
assert result.startswith(_PREFIX)
|
||||
assert tool_name in result
|
||||
assert "test.py" in result
|
||||
|
||||
@pytest.mark.parametrize("tool_name", ["read_file", "write_file", "edit_file"])
|
||||
@pytest.mark.parametrize(
|
||||
"tool_name", ["read_file", "write_file", "edit_file", "delete"]
|
||||
)
|
||||
def test_file_tool_with_path_key(self, tool_name: str) -> None:
|
||||
result = format_tool_display(tool_name, {"path": "/tmp/test.py"})
|
||||
assert "test.py" in result
|
||||
|
||||
@pytest.mark.parametrize("tool_name", ["read_file", "write_file", "edit_file"])
|
||||
@pytest.mark.parametrize(
|
||||
"tool_name", ["read_file", "write_file", "edit_file", "delete"]
|
||||
)
|
||||
def test_file_tool_missing_path_falls_back_to_generic(self, tool_name: str) -> None:
|
||||
result = format_tool_display(tool_name, {})
|
||||
assert _PREFIX in result
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
from pathlib import Path
|
||||
|
||||
from deepagents_code.widgets.tool_renderers import get_renderer
|
||||
from deepagents_code.widgets.tool_widgets import (
|
||||
EditFileApprovalWidget,
|
||||
GenericApprovalWidget,
|
||||
)
|
||||
|
||||
|
||||
def test_delete_renderer_shows_removed_file_diff(tmp_path: Path) -> None:
|
||||
target = tmp_path / "old.txt"
|
||||
target.write_text("alpha\nbeta\n", encoding="utf-8")
|
||||
|
||||
widget_class, data = get_renderer("delete").get_approval_widget(
|
||||
{"file_path": str(target)}
|
||||
)
|
||||
|
||||
assert widget_class is EditFileApprovalWidget
|
||||
assert data["file_path"] == "old.txt"
|
||||
assert "-alpha" in data["diff_lines"]
|
||||
assert "-beta" in data["diff_lines"]
|
||||
|
||||
|
||||
def test_delete_renderer_flags_directories_without_diff(tmp_path: Path) -> None:
|
||||
target = tmp_path / "subdir"
|
||||
target.mkdir()
|
||||
(target / "child.txt").write_text("data\n", encoding="utf-8")
|
||||
|
||||
widget_class, data = get_renderer("delete").get_approval_widget(
|
||||
{"file_path": str(target)}
|
||||
)
|
||||
|
||||
assert widget_class is GenericApprovalWidget
|
||||
assert data["file_path"] == "subdir"
|
||||
assert "Contents: directory or unreadable file" in data["details"]
|
||||
|
||||
|
||||
def test_delete_renderer_surfaces_unresolvable_path_error() -> None:
|
||||
"""An empty path yields a resolution error shown in the approval widget."""
|
||||
widget_class, data = get_renderer("delete").get_approval_widget({"file_path": ""})
|
||||
|
||||
assert widget_class is GenericApprovalWidget
|
||||
assert data["error"] == "Unable to resolve file path."
|
||||
Reference in New Issue
Block a user