Coverage for core/langgraph/tools/read_file.py: 83%
75 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 05:04 +0000
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 05:04 +0000
1"""Per-tool module for the `read_file` filesystem tool.
3Owns the tool's model-facing definition, its instruction surface (none), and
4its realpath-bound execution inside the project root. The definition and the
5execution are the single source; the legacy sites delegate here.
6"""
7from __future__ import annotations
9import json
10from dataclasses import dataclass
11from pathlib import Path
12from typing import Any
14from ._paths import is_within
15from .contract import ToolContext, ToolResult
17READ_FILE_TOOL_NAME = "read_file"
18READ_FILE_MAX_BYTES = 200_000
19READ_FILE_MAX_LINES = 2000
21READ_FILE_DEFINITION: dict[str, Any] = {
22 "type": "function",
23 "function": {
24 "name": READ_FILE_TOOL_NAME,
25 "description": (
26 "Read a text file from the project root. The path is resolved "
27 "against the project root; any read that resolves outside the "
28 "project root is rejected. Content is truncated and flagged when "
29 "it exceeds the caps, and can be windowed by offset/limit (lines)."
30 ),
31 "parameters": {
32 "type": "object",
33 "properties": {
34 "path": {
35 "type": "string",
36 "description": "Path to the file, relative to the project root.",
37 },
38 "offset": {
39 "type": "integer",
40 "description": "Number of lines to skip (0-based). Optional.",
41 },
42 "limit": {
43 "type": "integer",
44 "description": "Max number of lines to return. Optional.",
45 },
46 },
47 "required": ["path"],
48 "additionalProperties": False,
49 },
50 },
51}
54@dataclass
55class ReadFileResult:
56 content: str
57 truncated: bool
58 error: str | None
60 def as_tool_message(self) -> str:
61 payload: dict[str, Any] = {
62 "name": READ_FILE_TOOL_NAME,
63 "ok": self.error is None,
64 "content": self.content,
65 "truncated": self.truncated,
66 }
67 if self.error is not None:
68 payload["error"] = self.error
69 return json.dumps(payload, ensure_ascii=False)
72def read_file(
73 project_root: Path,
74 path: str,
75 offset: int | None,
76 limit: int | None,
77) -> ReadFileResult:
78 """Realpath-bound read of a text file inside the project root (ADR-0012).
80 Every rejected read returns a clear error — no silent fallback. Truncation
81 over `READ_FILE_MAX_BYTES` / `READ_FILE_MAX_LINES` is explicit via
82 `truncated`.
83 """
84 if not path or not str(path).strip():
85 return ReadFileResult("", False, f"{READ_FILE_TOOL_NAME}: 'path' is required (ADR-0012)")
86 try:
87 root = project_root.resolve(strict=False)
88 except OSError as exc:
89 return ReadFileResult("", False, f"{READ_FILE_TOOL_NAME}: cannot resolve project root: {exc}")
90 try:
91 candidate = (root / path).resolve(strict=False)
92 except OSError as exc:
93 return ReadFileResult("", False, f"{READ_FILE_TOOL_NAME}: cannot resolve '{path}': {exc}")
94 if not is_within(root, candidate):
95 return ReadFileResult(
96 "",
97 False,
98 f"{READ_FILE_TOOL_NAME}: path '{path}' resolves outside project root (ADR-0012)",
99 )
100 if not candidate.is_file():
101 return ReadFileResult("", False, f"{READ_FILE_TOOL_NAME}: '{path}' is not a file or does not exist")
102 try:
103 text = candidate.read_text(encoding="utf-8")
104 except (OSError, UnicodeDecodeError) as exc:
105 return ReadFileResult("", False, f"{READ_FILE_TOOL_NAME}: cannot read '{path}': {exc}")
107 lines = text.splitlines()
108 total_lines = len(lines)
109 try:
110 start = int(offset) if offset is not None and int(offset) > 0 else 0
111 except (TypeError, ValueError):
112 start = 0
113 if limit is not None:
114 # A requested window is not an engine cap — no line-level truncation flag.
115 try:
116 max_lines = int(limit)
117 except (TypeError, ValueError):
118 max_lines = READ_FILE_MAX_LINES
119 truncated = False
120 end = start + max_lines if max_lines >= 0 else total_lines
121 else:
122 max_lines = READ_FILE_MAX_LINES
123 truncated = (start + max_lines) < total_lines
124 end = start + max_lines
125 content = "\n".join(lines[start:end])
127 if len(content.encode("utf-8")) > READ_FILE_MAX_BYTES:
128 content = content[:READ_FILE_MAX_BYTES]
129 truncated = True
130 return ReadFileResult(content, truncated, None)
133class _ReadFileTool:
134 name = READ_FILE_TOOL_NAME
136 def definition(self) -> dict[str, Any]:
137 return READ_FILE_DEFINITION
139 def instruction(self) -> str | None:
140 return None
142 def params(self) -> list[dict[str, Any]] | None:
143 return None
145 def execute(self, context: ToolContext) -> ToolResult:
146 if context.root is None:
147 return ToolResult(ok=False, content="", error=f"{READ_FILE_TOOL_NAME}: no project root")
148 res = read_file(context.root, context.args.get("path"), context.args.get("offset"), context.args.get("limit"))
149 return ToolResult(ok=res.error is None, content=res.content, error=res.error, state_patch={"truncated": res.truncated})
152TOOL = _ReadFileTool()