text_editor: simplify RFC implementation
linuztx committed
Mar 2, 2026 at 17:04 UTC
1491d72db493f8cd4fbd8fe069b3671e87974c0c
3 files changed
+222
-308
plugins/text_editor/helpers/file_ops.py
+175
-67
@@ -1,55 +1,78 @@
1
"""
2
-Async file operations for the text_editor plugin. Routes I/O to the
3
-Docker container via rfc_text_editor._impl functions and applies
4
-token budgeting host-side.
2
+Pure file operations for the text_editor plugin.
3
+
4
+No agent/tool dependencies — only stdlib + tokens helper.
5
"""
6
7
-from dataclasses import dataclass
7
+import os
8
+import shutil
9
+import tempfile
10
+from typing import TypedDict
11
+
12
+from python.helpers import tokens
13
+
14
+_BINARY_PEEK = 8192
15
9
-from python.helpers import tokens, runtime
10
-from plugins.text_editor.helpers.rfc_text_editor import (
11
- is_binary_impl as _is_binary_impl,
12
- read_file_raw_impl as _read_file_raw_impl,
13
- write_file_impl as _write_file_impl,
14
- apply_patch_impl as _apply_patch_impl,
15
- file_info_impl as _file_info_impl,
16
-)
16
17
# ------------------------------------------------------------------
18
# Binary detection
19
# ------------------------------------------------------------------
20
22
-async def is_binary(path: str) -> bool:
21
+def is_binary(path: str) -> bool:
22
"""Detect binary file by checking for null bytes."""
24
- return await runtime.call_development_function(_is_binary_impl, path)
23
+ try:
24
+ with open(path, "rb") as f:
25
+ chunk = f.read(_BINARY_PEEK)
26
+ return b"\x00" in chunk
27
+ except OSError:
28
+ return False
29
30
31
# ------------------------------------------------------------------
32
# File metadata
33
# ------------------------------------------------------------------
34
31
-async def file_info(path: str) -> dict:
32
- """
33
- Get file metadata from the container.
34
-
35
- Returns dict with: exists, is_file, realpath, expanded, mtime.
36
- """
37
- return await runtime.call_development_function(_file_info_impl, path)
35
+class FileInfo(TypedDict):
36
+ exists: bool
37
+ is_file: bool
38
+ realpath: str
39
+ expanded: str
40
+ mtime: float | None
41
+
42
+
43
+def file_info(path: str) -> FileInfo:
44
+ """Return file metadata for mtime tracking and path resolution."""
45
+ path = os.path.expanduser(path)
46
+ rp = os.path.realpath(path)
47
+ exists = os.path.exists(path)
48
+ is_file = os.path.isfile(path)
49
+ mtime = None
50
+ if exists:
51
+ try:
52
+ mtime = os.path.getmtime(path)
53
+ except OSError:
54
+ pass
55
+ return FileInfo(
56
+ exists=exists,
57
+ is_file=is_file,
58
+ realpath=rp,
59
+ expanded=path,
60
+ mtime=mtime,
61
+ )
62
63
64
# ------------------------------------------------------------------
65
# Read
66
# ------------------------------------------------------------------
67
44
-@dataclass
45
-class ReadResult:
46
- content: str = ""
47
- total_lines: int = 0
48
- warnings: str = ""
49
- error: str = ""
68
+class ReadResult(TypedDict):
69
+ content: str
70
+ total_lines: int
71
+ warnings: str
72
+ error: str
73
74
52
-async def read_file(
75
+def read_file(
76
path: str,
77
line_from: int = 1,
78
line_to: int | None = None,
@@ -64,15 +87,30 @@ async def read_file(
87
line_from and line_to are both inclusive.
88
None line_to defaults to line_from + default_line_count - 1.
89
"""
67
- # I/O happens in the container via RFC; token budgeting stays host-side.
68
- raw = await runtime.call_development_function(_read_file_raw_impl, path)
90
+ path = os.path.expanduser(path)
91
+
92
+ if not os.path.isfile(path):
93
+ return ReadResult(
94
+ content="", total_lines=0, warnings="",
95
+ error="file not found",
96
+ )
97
70
- if raw["error"]:
71
- return ReadResult(error=raw["error"])
98
+ if is_binary(path):
99
+ return ReadResult(
100
+ content="", total_lines=0, warnings="",
101
+ error="file appears binary, use terminal instead",
102
+ )
103
73
- all_lines = raw["lines"]
74
- total_lines = raw["total_lines"]
104
+ try:
105
+ with open(path, "r", encoding="utf-8", errors="replace") as f:
106
+ all_lines = f.readlines()
107
+ except OSError as exc:
108
+ return ReadResult(
109
+ content="", total_lines=0, warnings="",
110
+ error=str(exc),
111
+ )
112
113
+ total_lines = len(all_lines)
114
line_from = max(line_from, 1)
115
if line_to is None:
116
line_to = line_from + default_line_count - 1
@@ -129,6 +167,7 @@ async def read_file(
167
content="\n".join(output_lines),
168
total_lines=total_lines,
169
warnings=warn_str,
170
+ error="",
171
)
172
173
@@ -136,33 +175,37 @@ async def read_file(
175
# Write
176
# ------------------------------------------------------------------
177
139
-@dataclass
140
-class WriteResult:
141
- total_lines: int = 0
142
- error: str = ""
178
+class WriteResult(TypedDict):
179
+ total_lines: int
180
+ error: str
181
182
145
-async def write_file(path: str, content: str | None) -> WriteResult:
183
+def write_file(path: str, content: str | None) -> WriteResult:
184
"""Create or overwrite a file."""
185
if content is None:
186
content = ""
149
- result = await runtime.call_development_function(
150
- _write_file_impl, path, content
187
+ path = os.path.expanduser(path)
188
+ try:
189
+ os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
190
+ with open(path, "w", encoding="utf-8") as f:
191
+ f.write(content)
192
+ except OSError as exc:
193
+ return WriteResult(total_lines=0, error=str(exc))
194
+
195
+ total = content.count("\n") + (
196
+ 1 if content and not content.endswith("\n") else 0
197
)
152
- if result["error"]:
153
- return WriteResult(error=result["error"])
154
- return WriteResult(total_lines=result["total_lines"])
198
+ return WriteResult(total_lines=total, error="")
199
200
201
# ------------------------------------------------------------------
202
# Patch
203
# ------------------------------------------------------------------
204
161
-@dataclass
162
-class PatchResult:
163
- total_lines: int = 0
164
- edit_count: int = 0
165
- error: str = ""
205
+class PatchResult(TypedDict):
206
+ total_lines: int
207
+ edit_count: int
208
+ error: str
209
210
211
def validate_edits(edits: list | None) -> tuple[list[dict], str]:
@@ -218,7 +261,7 @@ def validate_edits(edits: list | None) -> tuple[list[dict], str]:
261
return parsed, ""
262
263
221
-async def apply_patch(path: str, edits: list[dict]) -> int:
264
+def apply_patch(path: str, edits: list[dict]) -> int:
265
"""
266
Apply sorted, validated edits by streaming to a temp file.
267
@@ -226,29 +269,94 @@ async def apply_patch(path: str, edits: list[dict]) -> int:
269
Inserts have 'insert': True.
270
Returns total line count after patching.
271
"""
229
- result = await runtime.call_development_function(
230
- _apply_patch_impl, path, edits
231
- )
232
- if result["error"]:
233
- raise Exception(result["error"])
234
- return result["total_lines"]
235
-
272
+ # Ensure content always ends with newline to prevent line merging
273
+ for e in edits:
274
+ if e["content"] and not e["content"].endswith("\n"):
275
+ e["content"] += "\n"
276
237
-async def patch_file(path: str, edits: list | None) -> PatchResult:
277
+ dir_name = os.path.dirname(path) or "."
278
+ fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
279
+ try:
280
+ with (
281
+ open(path, "r", encoding="utf-8", errors="replace") as src,
282
+ os.fdopen(fd, "w", encoding="utf-8") as dst,
283
+ ):
284
+ edit_idx = 0
285
+ line_no = 1 # 1-based
286
+ total_written = 0
287
+
288
+ for raw_line in src:
289
+ # Process all inserts targeting this line first
290
+ while (
291
+ edit_idx < len(edits)
292
+ and edits[edit_idx]["insert"]
293
+ and edits[edit_idx]["from"] == line_no
294
+ ):
295
+ edit = edits[edit_idx]
296
+ if edit["content"]:
297
+ dst.write(edit["content"])
298
+ total_written += _count_content_lines(edit["content"])
299
+ edit_idx += 1
300
+
301
+ # Check if current line falls in a replace/delete range
302
+ if edit_idx < len(edits) and not edits[edit_idx]["insert"]:
303
+ edit = edits[edit_idx]
304
+ if edit["from"] <= line_no <= edit["to"]:
305
+ # Write replacement content once at range start
306
+ if line_no == edit["from"] and edit["content"]:
307
+ dst.write(edit["content"])
308
+ total_written += _count_content_lines(
309
+ edit["content"]
310
+ )
311
+ # Skip original line; advance edit at range end
312
+ if line_no == edit["to"]:
313
+ edit_idx += 1
314
+ line_no += 1
315
+ continue
316
+
317
+ dst.write(raw_line)
318
+ total_written += 1
319
+ line_no += 1
320
+
321
+ # Remaining edits past end of file
322
+ while edit_idx < len(edits):
323
+ edit = edits[edit_idx]
324
+ if edit["content"]:
325
+ dst.write(edit["content"])
326
+ total_written += _count_content_lines(edit["content"])
327
+ edit_idx += 1
328
+
329
+ shutil.move(tmp_path, path)
330
+ return total_written
331
+ except Exception:
332
+ if os.path.exists(tmp_path):
333
+ os.unlink(tmp_path)
334
+ raise
335
+
336
+
337
+def patch_file(path: str, edits: list | None) -> PatchResult:
338
"""Validate and apply edits to a file."""
239
- info = await runtime.call_development_function(
240
- _file_info_impl, path
241
- )
242
- if not info["is_file"]:
243
- return PatchResult(error="file not found")
339
+ path = os.path.expanduser(path)
340
+ if not os.path.isfile(path):
341
+ return PatchResult(total_lines=0, edit_count=0, error="file not found")
342
343
parsed, err = validate_edits(edits)
344
if err:
247
- return PatchResult(error=err)
345
+ return PatchResult(total_lines=0, edit_count=0, error=err)
346
347
try:
250
- total = await apply_patch(info["expanded"], parsed)
348
+ total = apply_patch(path, parsed)
349
except Exception as exc:
252
- return PatchResult(error=str(exc))
350
+ return PatchResult(total_lines=0, edit_count=0, error=str(exc))
351
254
- return PatchResult(total_lines=total, edit_count=len(parsed))
352
+ return PatchResult(total_lines=total, edit_count=len(parsed), error="")
353
+
354
+
355
+# ------------------------------------------------------------------
356
+# Internal
357
+# ------------------------------------------------------------------
358
+
359
+def _count_content_lines(content: str) -> int:
360
+ return content.count("\n") + (
361
+ 1 if content and not content.endswith("\n") else 0
362
+ )
plugins/text_editor/helpers/rfc_text_editor.py
deleted
-206
@@ -1,206 +0,0 @@
1
-"""
2
-Filesystem operations for the text_editor plugin, executed on the
3
-Docker container via RFC. Stdlib only — args and return values
4
-must be JSON-serializable.
5
-"""
6
-
7
-import os
8
-import shutil
9
-import tempfile
10
-
11
-_BINARY_PEEK = 8192
12
-
13
-
14
-# ------------------------------------------------------------------
15
-# Internal
16
-# ------------------------------------------------------------------
17
-
18
-def _count_content_lines(content: str) -> int:
19
- return content.count("\n") + (
20
- 1 if content and not content.endswith("\n") else 0
21
- )
22
-
23
-
24
-# ------------------------------------------------------------------
25
-# Binary detection
26
-# ------------------------------------------------------------------
27
-
28
-def is_binary_impl(path: str) -> bool:
29
- """Check for null bytes in the first 8 KiB."""
30
- try:
31
- with open(path, "rb") as f:
32
- chunk = f.read(_BINARY_PEEK)
33
- return b"\x00" in chunk
34
- except OSError:
35
- return False
36
-
37
-
38
-# ------------------------------------------------------------------
39
-# Read
40
-# ------------------------------------------------------------------
41
-
42
-def read_file_raw_impl(path: str) -> dict:
43
- """
44
- Read a text file and return raw lines plus metadata.
45
-
46
- Returns dict with:
47
- - lines: list[str] (raw lines including newlines)
48
- - total_lines: int
49
- - error: str (empty on success)
50
- """
51
- path = os.path.expanduser(path)
52
-
53
- if not os.path.isfile(path):
54
- return {"lines": [], "total_lines": 0, "error": "file not found"}
55
-
56
- # Binary check inline to avoid extra RFC round-trip
57
- try:
58
- with open(path, "rb") as f:
59
- chunk = f.read(_BINARY_PEEK)
60
- if b"\x00" in chunk:
61
- return {"lines": [], "total_lines": 0,
62
- "error": "file appears binary, use terminal instead"}
63
- except OSError:
64
- pass
65
-
66
- try:
67
- with open(path, "r", encoding="utf-8", errors="replace") as f:
68
- all_lines = f.readlines()
69
- except OSError as exc:
70
- return {"lines": [], "total_lines": 0, "error": str(exc)}
71
-
72
- return {"lines": all_lines, "total_lines": len(all_lines), "error": ""}
73
-
74
-
75
-# ------------------------------------------------------------------
76
-# Write
77
-# ------------------------------------------------------------------
78
-
79
-def write_file_impl(path: str, content: str) -> dict:
80
- """
81
- Write content to a file, creating parent dirs as needed.
82
-
83
- Returns dict with:
84
- - total_lines: int
85
- - error: str (empty on success)
86
- """
87
- path = os.path.expanduser(path)
88
- try:
89
- os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
90
- with open(path, "w", encoding="utf-8") as f:
91
- f.write(content)
92
- except OSError as exc:
93
- return {"total_lines": 0, "error": str(exc)}
94
-
95
- total = content.count("\n") + (
96
- 1 if content and not content.endswith("\n") else 0
97
- )
98
- return {"total_lines": total, "error": ""}
99
-
100
-
101
-# ------------------------------------------------------------------
102
-# Patch
103
-# ------------------------------------------------------------------
104
-
105
-def apply_patch_impl(path: str, edits: list) -> dict:
106
- """
107
- Apply sorted, validated edits to a file.
108
-
109
- Returns dict with:
110
- - total_lines: int
111
- - error: str (empty on success)
112
- """
113
- path = os.path.expanduser(path)
114
-
115
- # Ensure content always ends with newline to prevent line merging
116
- for e in edits:
117
- if e["content"] and not e["content"].endswith("\n"):
118
- e["content"] += "\n"
119
-
120
- dir_name = os.path.dirname(path) or "."
121
- fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
122
- try:
123
- with (
124
- open(path, "r", encoding="utf-8", errors="replace") as src,
125
- os.fdopen(fd, "w", encoding="utf-8") as dst,
126
- ):
127
- edit_idx = 0
128
- line_no = 1
129
- total_written = 0
130
-
131
- for raw_line in src:
132
- while (
133
- edit_idx < len(edits)
134
- and edits[edit_idx]["insert"]
135
- and edits[edit_idx]["from"] == line_no
136
- ):
137
- edit = edits[edit_idx]
138
- if edit["content"]:
139
- dst.write(edit["content"])
140
- total_written += _count_content_lines(edit["content"])
141
- edit_idx += 1
142
-
143
- if edit_idx < len(edits) and not edits[edit_idx]["insert"]:
144
- edit = edits[edit_idx]
145
- if edit["from"] <= line_no <= edit["to"]:
146
- if line_no == edit["from"] and edit["content"]:
147
- dst.write(edit["content"])
148
- total_written += _count_content_lines(
149
- edit["content"]
150
- )
151
- if line_no == edit["to"]:
152
- edit_idx += 1
153
- line_no += 1
154
- continue
155
-
156
- dst.write(raw_line)
157
- total_written += 1
158
- line_no += 1
159
-
160
- while edit_idx < len(edits):
161
- edit = edits[edit_idx]
162
- if edit["content"]:
163
- dst.write(edit["content"])
164
- total_written += _count_content_lines(edit["content"])
165
- edit_idx += 1
166
-
167
- shutil.move(tmp_path, path)
168
- return {"total_lines": total_written, "error": ""}
169
- except Exception as exc:
170
- if os.path.exists(tmp_path):
171
- os.unlink(tmp_path)
172
- return {"total_lines": 0, "error": str(exc)}
173
-
174
-
175
-# ------------------------------------------------------------------
176
-# File info
177
-# ------------------------------------------------------------------
178
-
179
-def file_info_impl(path: str) -> dict:
180
- """
181
- Return file metadata needed by the host for mtime tracking.
182
-
183
- Returns dict with:
184
- - exists: bool
185
- - is_file: bool
186
- - realpath: str
187
- - expanded: str (expanduser result)
188
- - mtime: float | None
189
- """
190
- path = os.path.expanduser(path)
191
- rp = os.path.realpath(path)
192
- exists = os.path.exists(path)
193
- is_file = os.path.isfile(path)
194
- mtime = None
195
- if exists:
196
- try:
197
- mtime = os.path.getmtime(path)
198
- except OSError:
199
- pass
200
- return {
201
- "exists": exists,
202
- "is_file": is_file,
203
- "realpath": rp,
204
- "expanded": path,
205
- "mtime": mtime,
206
- }
plugins/text_editor/tools/text_editor.py
+47
-35
@@ -1,7 +1,8 @@
1
from python.helpers.tool import Tool, Response
2
from python.helpers.extension import call_extensions
3
-from python.helpers import plugins
3
+from python.helpers import plugins, runtime
4
from plugins.text_editor.helpers.file_ops import (
5
+ FileInfo,
6
read_file,
7
write_file,
8
validate_edits,
@@ -41,7 +42,8 @@ class TextEditor(Tool):
42
raw_to = kwargs.get("line_to")
43
line_to = int(raw_to) if raw_to is not None else None
44
44
- result = await read_file(
45
+ result = await runtime.call_development_function(
46
+ read_file,
47
path,
48
line_from=line_from,
49
line_to=line_to,
@@ -50,14 +52,17 @@ class TextEditor(Tool):
52
max_total_read_tokens=cfg["max_total_read_tokens"],
53
)
54
53
- if result.error:
54
- return self._error("read", path, result.error)
55
+ if result["error"]:
56
+ return self._error("read", path, result["error"])
57
56
- info = await file_info(path)
57
- await _record_mtime(self.agent, info, result.total_lines)
58
+ info = await runtime.call_development_function(file_info, path)
59
+ _record_mtime(self.agent, info, result["total_lines"])
60
61
# Extension point
60
- ext_data = {"content": result.content, "warnings": result.warnings}
62
+ ext_data = {
63
+ "content": result["content"],
64
+ "warnings": result["warnings"],
65
+ }
66
await call_extensions(
67
"text_editor_read_after", agent=self.agent, data=ext_data
68
)
@@ -65,7 +70,7 @@ class TextEditor(Tool):
70
msg = self.agent.read_prompt(
71
"fw.text_editor.read_ok.md",
72
path=info["expanded"],
68
- total_lines=str(result.total_lines),
73
+ total_lines=str(result["total_lines"]),
74
warnings=ext_data["warnings"],
75
content=ext_data["content"],
76
)
@@ -74,7 +79,9 @@ class TextEditor(Tool):
79
# ------------------------------------------------------------------
80
# WRITE
81
# ------------------------------------------------------------------
77
- async def _write(self, path: str = "", content: str | None = "", **kwargs) -> Response:
82
+ async def _write(
83
+ self, path: str = "", content: str | None = "", **kwargs
84
+ ) -> Response:
85
if not path:
86
return self._error("write", path, "path is required")
87
@@ -84,25 +91,28 @@ class TextEditor(Tool):
91
"text_editor_write_before", agent=self.agent, data=ext_data
92
)
93
87
- result = await write_file(ext_data["path"], ext_data["content"])
94
+ result = await runtime.call_development_function(
95
+ write_file, ext_data["path"], ext_data["content"]
96
+ )
97
89
- if result.error:
90
- return self._error("write", path, result.error)
98
+ if result["error"]:
99
+ return self._error("write", path, result["error"])
100
101
# Extension point
102
await call_extensions(
103
"text_editor_write_after", agent=self.agent,
95
- data={"path": path, "total_lines": result.total_lines},
104
+ data={"path": path, "total_lines": result["total_lines"]},
105
)
106
98
- info = await file_info(path)
99
- await _record_mtime(self.agent, info, result.total_lines)
107
+ info = await runtime.call_development_function(file_info, path)
108
+ _record_mtime(self.agent, info, result["total_lines"])
109
110
cfg = _get_config(self.agent)
102
- read_result = await read_file(
111
+ read_result = await runtime.call_development_function(
112
+ read_file,
113
info["expanded"],
114
line_from=1,
105
- line_to=result.total_lines,
115
+ line_to=result["total_lines"],
116
max_line_tokens=cfg["max_line_tokens"],
117
max_total_read_tokens=cfg["max_total_read_tokens"],
118
)
@@ -110,8 +120,8 @@ class TextEditor(Tool):
120
msg = self.agent.read_prompt(
121
"fw.text_editor.write_ok.md",
122
path=info["expanded"],
113
- total_lines=str(result.total_lines),
114
- content=read_result.content,
123
+ total_lines=str(result["total_lines"]),
124
+ content=read_result["content"],
125
)
126
return Response(message=msg, break_loop=False)
127
@@ -122,7 +132,7 @@ class TextEditor(Tool):
132
if not path:
133
return self._error("patch", path, "path is required")
134
125
- info = await file_info(path)
135
+ info = await runtime.call_development_function(file_info, path)
136
if not info["is_file"]:
137
return self._error("patch", path, "file not found")
138
@@ -143,7 +153,9 @@ class TextEditor(Tool):
153
)
154
155
try:
146
- total_lines = await apply_patch(ext_data["path"], ext_data["edits"])
156
+ total_lines = await runtime.call_development_function(
157
+ apply_patch, ext_data["path"], ext_data["edits"]
158
+ )
159
except Exception as exc:
160
return self._error("patch", path, str(exc))
161
@@ -154,8 +166,12 @@ class TextEditor(Tool):
166
)
167
168
# Refresh file info after patch for updated mtime
157
- post_info = await file_info(expanded)
158
- _apply_patch_post(self.agent, post_info, total_lines, ext_data["edits"])
169
+ post_info = await runtime.call_development_function(
170
+ file_info, expanded
171
+ )
172
+ _apply_patch_post(
173
+ self.agent, post_info, total_lines, ext_data["edits"]
174
+ )
175
176
patch_content = await _read_patch_region(
177
expanded, ext_data["edits"], total_lines, _get_config(self.agent)
@@ -203,18 +219,18 @@ async def _read_patch_region(
219
max_to = max(e["to"] for e in edits)
220
end_line = max_to + added - removed + 3
221
206
- result = await read_file(
222
+ result = await runtime.call_development_function(
223
+ read_file,
224
path,
225
line_from=max(min_from - 1, 1),
226
line_to=min(end_line, total_lines),
227
max_line_tokens=cfg["max_line_tokens"],
228
max_total_read_tokens=cfg["max_total_read_tokens"],
229
)
213
- return result.content
230
+ return result["content"]
231
232
216
-async def _record_mtime(agent, info: dict, total_lines: int):
217
- """Record mtime using file_info dict from the container."""
233
+def _record_mtime(agent, info: FileInfo, total_lines: int):
234
mtimes = agent.data.setdefault(_MTIME_KEY, {})
235
if info["mtime"] is not None:
236
mtimes[info["realpath"]] = {
@@ -223,12 +239,6 @@ async def _record_mtime(agent, info: dict, total_lines: int):
239
}
240
241
226
-def _clear_mtime(agent, info: dict):
227
- mtimes = agent.data.get(_MTIME_KEY)
228
- if mtimes is not None:
229
- mtimes.pop(info["realpath"], None)
230
-
231
-
242
def _count_content_lines(content: str) -> int:
243
return content.count("\n") + (
244
1 if content and not content.endswith("\n") else 0
@@ -246,7 +256,9 @@ def _all_edits_in_place(edits: list[dict]) -> bool:
256
return True
257
258
249
-def _apply_patch_post(agent, info: dict, new_total: int, edits: list[dict]):
259
+def _apply_patch_post(
260
+ agent, info: FileInfo, new_total: int, edits: list[dict]
261
+):
262
mtimes = agent.data.setdefault(_MTIME_KEY, {})
263
real = info["realpath"]
264
@@ -272,7 +284,7 @@ def _apply_patch_post(agent, info: dict, new_total: int, edits: list[dict]):
284
mtimes[real] = {"mtime": 0, "total_lines": 0}
285
286
275
-def _check_mtime(agent, info: dict) -> str:
287
+def _check_mtime(agent, info: FileInfo) -> str:
288
mtimes = agent.data.get(_MTIME_KEY, {})
289
real = info["realpath"]
290
if real not in mtimes: