Coverage for core/langgraph/tools/files_delete.py: 95%
114 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-28 00:19 +0000
1"""Per-tool module for the `files-delete` deterministic builtin.
3Owns the tool's model-facing definition, its instruction surface (none), and
4its all-or-nothing workspace deletion (`tracked → git rm`, `untracked →
5unlink`, ADR-0012). The definition and the execution are the single source;
6the legacy builtin site delegates here.
7"""
8from __future__ import annotations
10import json
11import subprocess
12from dataclasses import dataclass
13from pathlib import Path
14from typing import Any
16from .contract import ToolContext, ToolResult
18try:
19 from _files_output import (
20 FilesOutputError,
21 _resolve_path,
22 resolve_workspace_root,
23 )
24except ImportError: # pragma: no cover — main.py run (qualified imports)
25 from langgraph._files_output import (
26 FilesOutputError,
27 _resolve_path,
28 resolve_workspace_root,
29 )
31FILES_DELETE_TOOL_NAME = "files-delete"
33FILES_DELETE_DEFINITION: dict[str, Any] = {
34 "type": "function",
35 "function": {
36 "name": FILES_DELETE_TOOL_NAME,
37 "description": (
38 "Delete one or more workspace-relative files. Accepts a single "
39 "path string or a list of them; tracked files are removed via "
40 "`git rm` (deletion staged), untracked files via filesystem "
41 "unlink. Validation is atomic: one invalid path (traversal, "
42 ".git target, absolute path, missing file, non-string) rejects "
43 "the whole call and nothing is deleted (ADR-0012)."
44 ),
45 "parameters": {
46 "type": "object",
47 "properties": {
48 "paths": {
49 "description": (
50 "A single workspace-relative path string or a list "
51 "of them."
52 ),
53 "anyOf": [
54 {"type": "string"},
55 {
56 "type": "array",
57 "items": {"type": "string"},
58 },
59 ],
60 },
61 },
62 "required": ["paths"],
63 "additionalProperties": False,
64 },
65 },
66}
69class _FilesDeleteError(Exception):
70 def __init__(self, message: str, details: dict[str, Any] | None = None) -> None:
71 super().__init__(message)
72 self.message = message
73 self.details = details
76@dataclass
77class GitResult:
78 error: str | None
79 details: dict[str, Any] | None = None
82@dataclass
83class FilesDeleteResult:
84 deleted: list[str]
85 count: int
86 error: str | None
87 details: dict[str, Any] | None
89 def as_tool_message(self) -> str:
90 payload: dict[str, Any] = {
91 "name": FILES_DELETE_TOOL_NAME,
92 "ok": self.error is None,
93 "deleted": self.deleted,
94 "count": self.count,
95 }
96 if self.error is not None:
97 payload["error"] = self.error
98 return json.dumps(payload, ensure_ascii=False)
101def _normalize_paths(paths: Any) -> list:
102 if isinstance(paths, str):
103 return [paths]
104 if not isinstance(paths, list) or not paths:
105 raise _FilesDeleteError(
106 "files-delete: 'params.paths' is required — a single "
107 "workspace-relative path string or a list of them (ADR-0012)"
108 )
109 return paths
112def _resolve_delete_path(root: Path, raw_path: str, ref: str) -> Path:
113 try:
114 return _resolve_path(root, raw_path, root, ref)
115 except FilesOutputError as exc:
116 raise _FilesDeleteError(str(exc)) from exc
119def _prepare_path(root: Path, raw_path: Any, index: int) -> tuple[Path, str, str]:
120 ref = f"path {index}"
121 if not isinstance(raw_path, str):
122 raise _FilesDeleteError(
123 f"files-delete: {ref} must be a string path, "
124 f"got {type(raw_path).__name__} (ADR-0012)"
125 )
126 if not raw_path.strip():
127 raise _FilesDeleteError(
128 f"files-delete: {ref} is empty or whitespace-only — "
129 "a path is required (ADR-0012)"
130 )
131 if Path(raw_path).is_absolute():
132 raise _FilesDeleteError(
133 f"files-delete: {ref} '{raw_path}' is absolute — only "
134 "workspace-relative paths are allowed (ADR-0012)"
135 )
136 resolved = _resolve_delete_path(root, raw_path, ref)
137 if not resolved.is_file():
138 raise _FilesDeleteError(
139 f"files-delete: {ref} '{raw_path}' is missing or not a "
140 "file — nothing deleted (ADR-0012)"
141 )
142 return resolved, raw_path, ref
145def _is_tracked(workspace_root: Path, relative: str) -> bool:
146 try:
147 proc = subprocess.run(
148 ["git", "-c", "safe.directory=*", "ls-files", "--", relative],
149 cwd=workspace_root,
150 capture_output=True,
151 text=True,
152 check=False,
153 )
154 except OSError as exc:
155 raise _FilesDeleteError(
156 f"files-delete: 'git ls-files -- {relative}' failed: {exc} (ADR-0012)"
157 ) from exc
158 return proc.returncode == 0 and bool(proc.stdout.strip())
161def git_rm(workspace_root: Path, relative: str) -> GitResult:
162 """Remove a tracked file and stage the deletion (`git rm --force` semantics).
164 Never raises: a subprocess failure is returned as an error-carrying result.
165 """
166 try:
167 subprocess.run(
168 ["git", "-c", "safe.directory=*", "rm", "--force", "--", relative],
169 cwd=workspace_root,
170 capture_output=True,
171 text=True,
172 check=True,
173 )
174 except subprocess.CalledProcessError as exc:
175 return GitResult(
176 f"files-delete: 'git rm {relative}' failed (exit {exc.returncode}): "
177 f"{(exc.stderr or exc.stdout or '').strip()} (ADR-0012)",
178 None,
179 )
180 except FileNotFoundError as exc:
181 return GitResult(
182 "files-delete: git is not available — cannot stage the deletion "
183 f"of tracked file '{relative}' (ADR-0012)",
184 None,
185 )
186 except OSError as exc:
187 return GitResult(
188 f"files-delete: 'git rm {relative}' failed: {exc} (ADR-0012)",
189 None,
190 )
191 return GitResult(None, None)
194def _git_rm(workspace_root: Path, relative: str) -> None:
195 result = git_rm(workspace_root, relative)
196 if result.error is not None:
197 raise _FilesDeleteError(result.error, result.details)
200def _unlink(resolved: Path, raw_path: str, ref: str) -> None:
201 try:
202 resolved.unlink()
203 except OSError as exc:
204 raise _FilesDeleteError(
205 f"files-delete: {ref} failed to delete '{raw_path}': "
206 f"{exc} (ADR-0012)"
207 ) from exc
210def _delete_paths(paths: Any, workspace_root: Path) -> list[str]:
211 normalized = _normalize_paths(paths)
212 root = workspace_root.resolve(strict=False)
213 prepared = [
214 _prepare_path(root, raw_path, index)
215 for index, raw_path in enumerate(normalized)
216 ]
217 deleted: list[str] = []
218 for resolved, raw_path, ref in prepared:
219 if _is_tracked(workspace_root, raw_path):
220 _git_rm(workspace_root, raw_path)
221 else:
222 _unlink(resolved, raw_path, ref)
223 deleted.append(raw_path)
224 return deleted
227def files_delete(
228 paths: Any, workspace_root: Path, step_id: str
229) -> FilesDeleteResult:
230 """Delete workspace-relative files all-or-nothing (ADR-0012).
232 Every path is validated before any deletion; a single invalid path aborts
233 the whole call. Every rejected call returns an explicit error; it never
234 raises.
235 """
236 try:
237 deleted = _delete_paths(paths, workspace_root)
238 except _FilesDeleteError as exc:
239 return FilesDeleteResult([], 0, exc.message, exc.details)
240 return FilesDeleteResult(deleted, len(deleted), None, None)
243class _FilesDeleteTool:
244 name = FILES_DELETE_TOOL_NAME
246 def definition(self) -> dict[str, Any]:
247 return FILES_DELETE_DEFINITION
249 def instruction(self) -> str | None:
250 return None
252 def params(self) -> list[dict[str, Any]] | None:
253 return None
255 def execute(self, context: ToolContext) -> ToolResult:
256 step = context.step or {}
257 step_id = step["id"]
258 paths = (context.args or {}).get("paths")
259 workspace_root = resolve_workspace_root(context.state or {})
260 res = files_delete(paths, workspace_root, step_id)
261 if res.error is not None:
262 return ToolResult(ok=False, content=res.as_tool_message(), error=res.error)
263 return ToolResult(
264 ok=True,
265 content=res.as_tool_message(),
266 state_patch={step_id: {"deleted": res.deleted, "count": res.count}},
267 )
270TOOL = _FilesDeleteTool()