@samitouri / QOSami_Agents_Zero-2 / commits / 4c2bc3d7

Add context-based patch_text support to text_editor

Introduces patch_text editing for the Docker-local text_editor, sharing request validation and freshness-state logic with text_editor_remote while preserving legacy line-number edits. Adds anchored context patching, safer state handling after context edits, updated model guidance, live remote wrapper reuse, and focused regression coverage for chained patches and Python replacement cases.

Alessandro committed Apr 21, 2026 at 14:56 UTC 4c2bc3d783a27f18da75b8f8c12380f142913fa9
12 files changed +1254 -253
plugins/_a0_connector/extensions/python/message_loop_prompts_after/_79_include_text_editor_remote.py
+28 -6
@@ -27,7 +27,8 @@ class IncludeTextEditorRemote(Extension):
27 access_mode = "Read&Write (legacy/unknown)"
28 write_guidance = (
29 "- Writes and patches are expected to be available, but this CLI did not "
30 - "advertise an explicit F3 access mode."
30 + "advertise an explicit F3 access mode.\n"
31 + "- Prefer `patch_text` for context-anchored edits when supported."
32 )
33 write_examples = """```json
34 {
@@ -46,9 +47,7 @@ class IncludeTextEditorRemote(Extension):
47 "tool_args": {
48 "op": "patch",
49 "path": "/path/on/remote/machine/file.py",
49 - "edits": [
50 - {"from": 5, "to": 5, "content": " if x == 2:\\n"}
51 - ]
50 + "patch_text": "*** Begin Patch\\n*** Update File: /path/on/remote/machine/file.py\\n@@ def main():\\n+ setup()\\n*** End Patch"
51 }
52 }
53 ```"""
@@ -56,8 +55,9 @@ class IncludeTextEditorRemote(Extension):
55 access_mode = "Read&Write"
56 write_guidance = (
57 "- Use `write` only when replacing or creating the full file is the right operation.\n"
59 - "- Use `patch` for surgical line-range edits. Keep the edit set tight and based on the latest remote read.\n"
60 - "- Freshness-aware patching may reject stale edits. If a patch requires a reread, read the file again and then retry with updated ranges."
58 + "- Use `patch` with `patch_text` for context-anchored edits, especially after inserts/deletes or when line numbers may have shifted.\n"
59 + "- Use `patch` with `edits` only for surgical line-range edits based on the latest remote read.\n"
60 + "- Freshness-aware line patching may reject stale edits. If a line patch requires a reread, read the file again and then retry with updated ranges."
61 )
62 write_examples = """```json
63 {
@@ -70,6 +70,28 @@ class IncludeTextEditorRemote(Extension):
70 }
71 ```
72
73 +```json
74 +{
75 + "tool_name": "text_editor_remote",
76 + "tool_args": {
77 + "op": "patch",
78 + "path": "/path/on/remote/machine/file.py",
79 + "patch_text": "*** Begin Patch\\n*** Update File: /path/on/remote/machine/file.py\\n@@ def main():\\n+ setup()\\n*** End Patch"
80 + }
81 +}
82 +```
83 +
84 +```json
85 +{
86 + "tool_name": "text_editor_remote",
87 + "tool_args": {
88 + "op": "patch",
89 + "path": "/path/on/remote/machine/file.py",
90 + "patch_text": "*** Begin Patch\\n*** Update File: /path/on/remote/machine/file.py\\n@@ def main():\\n- old_helper()\\n+ new_helper()\\n*** End Patch"
91 + }
92 +}
93 +```
94 +
95 ```json
96 {
97 "tool_name": "text_editor_remote",
plugins/_a0_connector/helpers/text_editor_freshness.py
+17 -145
@@ -1,159 +1,31 @@
1 from __future__ import annotations
2
3 -from typing import Any, TypedDict
3 +from typing import Any
4
5 -
6 -_FRESHNESS_KEY = "_a0_connector_text_editor_remote_mtimes"
7 -
8 -
9 -class FileMetadata(TypedDict):
10 - realpath: str
11 - mtime: float | None
12 - total_lines: int
13 -
14 -
15 -def coerce_file_metadata(file_data: Any) -> FileMetadata | None:
16 - if not isinstance(file_data, dict):
17 - return None
18 -
19 - realpath = str(file_data.get("realpath", "")).strip()
20 - if not realpath:
21 - return None
22 -
23 - try:
24 - total_lines = int(file_data.get("total_lines", 0))
25 - except (TypeError, ValueError):
26 - return None
27 -
28 - raw_mtime = file_data.get("mtime")
29 - if raw_mtime is None:
30 - mtime: float | None = None
31 - else:
32 - try:
33 - mtime = float(raw_mtime)
34 - except (TypeError, ValueError):
35 - mtime = None
36 -
37 - return FileMetadata(
38 - realpath=realpath,
39 - mtime=mtime,
40 - total_lines=max(total_lines, 0),
41 - )
5 +from plugins._text_editor.helpers.patch_state import (
6 + REMOTE_FRESHNESS_KEY as _FRESHNESS_KEY,
7 + FileMetadata,
8 + apply_patch_post_state as _apply_patch_post_state,
9 + check_patch_freshness as _check_patch_freshness,
10 + coerce_file_metadata,
11 + mark_file_state_stale as _mark_file_state_stale,
12 + record_file_state as _record_file_state,
13 +)
14
15
16 def record_file_state(agent, file_data: Any) -> None:
45 - file_meta = coerce_file_metadata(file_data)
46 - if file_meta is None or file_meta["mtime"] is None:
47 - return
48 -
49 - freshness = agent.data.setdefault(_FRESHNESS_KEY, {})
50 - freshness[file_meta["realpath"]] = {
51 - "mtime": file_meta["mtime"],
52 - "total_lines": file_meta["total_lines"],
53 - }
17 + _record_file_state(agent, file_data, key=_FRESHNESS_KEY)
18
19
20 def mark_file_state_stale(agent, file_data: Any) -> None:
57 - file_meta = coerce_file_metadata(file_data)
58 - if file_meta is None:
59 - return
60 -
61 - freshness = agent.data.setdefault(_FRESHNESS_KEY, {})
62 - freshness[file_meta["realpath"]] = {"mtime": 0, "total_lines": 0}
21 + _mark_file_state_stale(agent, file_data, key=_FRESHNESS_KEY)
22
23
24 def check_patch_freshness(agent, file_data: Any) -> str | None:
66 - file_meta = coerce_file_metadata(file_data)
67 - if file_meta is None:
68 - return "patch_need_read"
69 -
70 - freshness = agent.data.get(_FRESHNESS_KEY, {})
71 - realpath = file_meta["realpath"]
72 - if realpath not in freshness:
73 - return "patch_need_read"
74 -
75 - stored = freshness[realpath]
76 - mtime = stored.get("mtime") if isinstance(stored, dict) else stored
77 - if mtime is None:
78 - freshness.pop(realpath, None)
79 - return "patch_need_read"
80 -
81 - current = file_meta["mtime"]
82 - if current is None:
83 - return None
84 - if current != mtime:
85 - return "patch_stale_read"
86 - return None
87 -
88 -
89 -def apply_patch_post_state(agent, file_data: Any, edits: list[Any] | None) -> None:
90 - file_meta = coerce_file_metadata(file_data)
91 - if file_meta is None:
92 - return
93 -
94 - freshness = agent.data.setdefault(_FRESHNESS_KEY, {})
95 - realpath = file_meta["realpath"]
96 -
97 - if not _all_edits_in_place(edits):
98 - freshness[realpath] = {"mtime": 0, "total_lines": 0}
99 - return
100 -
101 - stored = freshness.get(realpath)
102 - if not isinstance(stored, dict) or "total_lines" not in stored:
103 - freshness[realpath] = {"mtime": 0, "total_lines": 0}
104 - return
105 -
106 - if file_meta["total_lines"] != int(stored["total_lines"]):
107 - freshness[realpath] = {"mtime": 0, "total_lines": 0}
108 - return
109 -
110 - if file_meta["mtime"] is None:
111 - freshness[realpath] = {"mtime": 0, "total_lines": 0}
112 - return
113 -
114 - freshness[realpath] = {
115 - "mtime": file_meta["mtime"],
116 - "total_lines": file_meta["total_lines"],
117 - }
118 -
119 -
120 -def _all_edits_in_place(edits: list[Any] | None) -> bool:
121 - if not isinstance(edits, list):
122 - return False
123 -
124 - for edit in edits:
125 - if not isinstance(edit, dict):
126 - return False
127 -
128 - try:
129 - start = int(edit.get("from", 0) or 0)
130 - except (TypeError, ValueError):
131 - return False
132 - if start < 1:
133 - return False
134 -
135 - raw_to = edit.get("to")
136 - if raw_to is None:
137 - return False
138 -
139 - try:
140 - end = int(raw_to)
141 - except (TypeError, ValueError):
142 - return False
143 - if end < start:
144 - return False
145 -
146 - removed = end - start + 1
147 - added = _count_content_lines(edit.get("content"))
148 - if removed != added:
149 - return False
150 -
151 - return True
152 -
25 + return _check_patch_freshness(agent, file_data, key=_FRESHNESS_KEY)
26
154 -def _count_content_lines(content: Any) -> int:
155 - if not content:
156 - return 0
27
158 - text = str(content)
159 - return text.count("\n") + (1 if not text.endswith("\n") else 0)
28 +def apply_patch_post_state(
29 + agent, file_data: Any, edits: list[Any] | None
30 +) -> None:
31 + _apply_patch_post_state(agent, file_data, edits, key=_FRESHNESS_KEY)
plugins/_a0_connector/prompts/agent.extras.text_editor_remote.md
+5 -1
@@ -5,7 +5,11 @@ Current access mode: `{{access_mode}}`
5
6 - Use `text_editor_remote` when the user asks you to edit files on their local machine while connected via the CLI.
7 - Paths are evaluated on the remote CLI machine's filesystem, not on the Agent Zero server.
8 -- Prefer `read` before `patch` so you have current line numbers and freshness metadata.
8 +- Prefer `patch_text` for edits that can be located by surrounding code context.
9 +- For `patch_text` inserts, use one `@@ existing line` anchor followed directly by `+new line`.
10 +- For `patch_text` replacements, use `@@ before target` then `-old`/`+new`, or `@@ old target` then the same `-old`/`+new`.
11 +- Do not repeat the same old line as both context and deletion in one replacement hunk.
12 +- Prefer `read` before line-number `edits` so you have current line numbers and freshness metadata.
13 - `read` is always the safest first step for inspecting the local file.
14 {{write_guidance}}
15
plugins/_a0_connector/prompts/agent.system.tool.text_editor_remote.md
+7 -1
@@ -12,8 +12,14 @@ subscribed CLI, so the base system prompt stays small when remote editing is not
12 ## Operations
13 - `read`: optional `line_from`, `line_to`
14 - `write`: requires `content`
15 -- `patch`: requires `edits`
15 +- `patch`: requires either `patch_text` or `edits`
16
17 ## Notes
18 - Paths are evaluated on the **remote machine's filesystem**, not the Agent Zero server.
19 - The transport uses `connector_file_op` and `connector_file_op_result` with a shared `op_id`.
20 +- `patch_text` uses context chunks and does not require fresh line numbers.
21 +- `patch_text` line rules: `@@ existing line` anchors the hunk; `+new` inserts after the anchor when there are no context or delete lines; `-old` then `+new` replaces the next matching old line after the anchor, or the anchor line itself when `@@` is the old target line.
22 +- For replacements, do not repeat the same old line as both a space-context line and a `-old` line.
23 +- Every non-header content line in `patch_text` must start with exactly one prefix: space for kept context, `+` for added content, or `-` for removed content. Do not emit raw unprefixed content lines.
24 +- Do not stack multiple `@@` lines for one insert. Use one anchor, then the `+` lines to insert.
25 +- `edits` uses 1-based line ranges and may require rereading after line-count changes.
plugins/_a0_connector/tools/text_editor_remote.py
+27 -4
@@ -22,6 +22,7 @@ from plugins._a0_connector.helpers.ws_runtime import (
22 store_pending_file_op,
23 subscribed_sids_for_context,
24 )
25 +from plugins._text_editor.helpers.patch_request import parse_patch_request
26
27
28 FILE_OP_TIMEOUT = 30.0
@@ -70,19 +71,41 @@ class TextEditorRemote(Tool):
71 result = await self._execute_file_op(op, path, content=content)
72 self._record_success_state(result)
73 else:
73 - edits = self.args.get("edits")
74 - if not edits:
74 + patch_request, err = parse_patch_request(
75 + self.args.get("edits"),
76 + self.args.get("patch_text"),
77 + both_error="provide either edits or patch_text for patch, not both",
78 + )
79 + if err:
80 return Response(
76 - message="edits is required for patch",
81 + message=err,
82 break_loop=False,
83 )
79 - result = await self._execute_patch(path, edits)
84 + if patch_request and patch_request.mode == "patch_text":
85 + result = await self._execute_context_patch(
86 + path, patch_request.patch_text
87 + )
88 + else:
89 + result = await self._execute_patch(
90 + path,
91 + patch_request.edits if patch_request else self.args.get("edits"),
92 + )
93
94 return Response(
95 message=self._extract_result(result, op, path),
96 break_loop=False,
97 )
98
99 + async def _execute_context_patch(self, path: str, patch_text: str) -> dict[str, Any]:
100 + patch_result = await self._execute_file_op("patch", path, patch_text=patch_text)
101 + if not self._result_ok(patch_result):
102 + return patch_result
103 +
104 + patch_file = self._extract_file_metadata(patch_result)
105 + if patch_file is not None:
106 + mark_file_state_stale(self.agent, patch_file)
107 + return patch_result
108 +
109 async def _execute_patch(self, path: str, edits: Any) -> dict[str, Any]:
110 stat_result = await self._execute_file_op("stat", path)
111 if self._is_unsupported_cli_freshness(stat_result):
plugins/_text_editor/helpers/context_patch.py new
+255
@@ -0,0 +1,255 @@
1 +from __future__ import annotations
2 +
3 +from dataclasses import dataclass
4 +from typing import Iterable
5 +
6 +
7 +class ContextPatchError(ValueError):
8 + """Raised when a context patch cannot be parsed or applied safely."""
9 +
10 +
11 +@dataclass(frozen=True)
12 +class ContextPatchApplyResult:
13 + content: str
14 + line_from: int
15 + line_to: int
16 + hunk_count: int
17 +
18 +
19 +@dataclass(frozen=True)
20 +class _PatchLine:
21 + kind: str
22 + text: str
23 +
24 +
25 +@dataclass(frozen=True)
26 +class _Hunk:
27 + anchor: str
28 + lines: tuple[_PatchLine, ...]
29 +
30 +
31 +@dataclass(frozen=True)
32 +class _HunkApplyResult:
33 + cursor: int
34 + line_from: int
35 + line_to: int
36 +
37 +
38 +_FILE_HEADERS = (
39 + "*** Update File:",
40 + "*** Add File:",
41 + "*** Delete File:",
42 + "*** End Patch",
43 +)
44 +
45 +
46 +def apply_context_patch(content: str, patch_text: str) -> str:
47 + """Apply a PseudoPatch-inspired context patch to one text file."""
48 + return apply_context_patch_with_metadata(content, patch_text).content
49 +
50 +
51 +def apply_context_patch_with_metadata(
52 + content: str, patch_text: str
53 +) -> ContextPatchApplyResult:
54 + """Apply a context patch and report the touched line range."""
55 + body = _extract_single_file_body(patch_text)
56 + hunks = _parse_hunks(body)
57 + if not hunks:
58 + raise ContextPatchError("patch_text must contain at least one update hunk")
59 +
60 + lines = content.split("\n")
61 + cursor = 0
62 + line_from: int | None = None
63 + line_to = 1
64 + for hunk in hunks:
65 + result = _apply_hunk(lines, hunk, cursor)
66 + cursor = result.cursor
67 + line_from = (
68 + result.line_from if line_from is None
69 + else min(line_from, result.line_from)
70 + )
71 + line_to = max(line_to, result.line_to)
72 +
73 + return ContextPatchApplyResult(
74 + content="\n".join(lines),
75 + line_from=line_from or 1,
76 + line_to=line_to,
77 + hunk_count=len(hunks),
78 + )
79 +
80 +
81 +def _extract_single_file_body(patch_text: str) -> list[str]:
82 + raw_lines = [line.rstrip("\r") for line in str(patch_text).splitlines()]
83 + lines = _trim_outer_blank_lines(raw_lines)
84 + if not lines:
85 + raise ContextPatchError("patch_text is required")
86 +
87 + if not lines[0].startswith("*** Begin Patch"):
88 + return lines
89 + if len(lines) < 2 or lines[-1] != "*** End Patch":
90 + raise ContextPatchError("patch_text missing *** End Patch")
91 +
92 + body: list[str] = []
93 + in_update = False
94 + update_count = 0
95 + for line in lines[1:-1]:
96 + if line.startswith("*** Update File:"):
97 + if update_count:
98 + raise ContextPatchError(
99 + "patch_text may update only one file per operation"
100 + )
101 + in_update = True
102 + update_count += 1
103 + continue
104 + if line.startswith("*** Move to:"):
105 + raise ContextPatchError("patch_text does not support file moves")
106 + if line.startswith(("*** Add File:", "*** Delete File:")):
107 + raise ContextPatchError("patch_text supports update hunks only")
108 + if in_update:
109 + body.append(line)
110 +
111 + if not update_count:
112 + raise ContextPatchError("patch_text must include an update file block")
113 + return body
114 +
115 +
116 +def _trim_outer_blank_lines(lines: list[str]) -> list[str]:
117 + start = 0
118 + end = len(lines)
119 + while start < end and not lines[start].strip():
120 + start += 1
121 + while end > start and not lines[end - 1].strip():
122 + end -= 1
123 + return lines[start:end]
124 +
125 +
126 +def _parse_hunks(lines: list[str]) -> list[_Hunk]:
127 + hunks: list[_Hunk] = []
128 + anchor = ""
129 + current: list[_PatchLine] = []
130 +
131 + def finish_current() -> None:
132 + nonlocal anchor, current
133 + if current:
134 + hunks.append(_Hunk(anchor=anchor, lines=tuple(current)))
135 + current = []
136 + anchor = ""
137 +
138 + for line in lines:
139 + if line.startswith(_FILE_HEADERS):
140 + finish_current()
141 + break
142 + if line.startswith("@@"):
143 + finish_current()
144 + anchor = line[2:].strip()
145 + continue
146 + if line.startswith("***"):
147 + raise ContextPatchError(f"invalid patch control line: {line}")
148 +
149 + if line == "":
150 + current.append(_PatchLine(" ", ""))
151 + continue
152 + if line[0] not in {" ", "+", "-"}:
153 + raise ContextPatchError(f"invalid patch line prefix: {line}")
154 + current.append(_PatchLine(line[0], line[1:]))
155 +
156 + finish_current()
157 + return hunks
158 +
159 +
160 +def _apply_hunk(
161 + lines: list[str], hunk: _Hunk, cursor: int
162 +) -> _HunkApplyResult:
163 + old_lines = [line.text for line in hunk.lines if line.kind in {" ", "-"}]
164 + new_lines = [line.text for line in hunk.lines if line.kind in {" ", "+"}]
165 +
166 + if old_lines == new_lines:
167 + raise ContextPatchError("patch hunk does not change content")
168 + if not old_lines:
169 + if not hunk.anchor:
170 + raise ContextPatchError("insert-only patch hunk needs an @@ anchor")
171 + insert_at = _find_anchor(lines, hunk.anchor, cursor)
172 + lines[insert_at:insert_at] = new_lines
173 + return _HunkApplyResult(
174 + cursor=insert_at + len(new_lines),
175 + line_from=insert_at + 1,
176 + line_to=insert_at + max(len(new_lines), 1),
177 + )
178 +
179 + start = cursor
180 + if hunk.anchor:
181 + start = _find_anchor(lines, hunk.anchor, cursor)
182 +
183 + try:
184 + match_index = _find_context(lines, old_lines, start, anchored=bool(hunk.anchor))
185 + except ContextPatchError:
186 + if not hunk.anchor:
187 + raise
188 + # Models often anchor on the line they want to replace. Preserve the
189 + # insert-after-anchor rule, but allow replacement context to start at
190 + # the anchor line when it is not found after the anchor.
191 + match_index = _find_context(lines, old_lines, start - 1, anchored=True)
192 + lines[match_index : match_index + len(old_lines)] = new_lines
193 + return _HunkApplyResult(
194 + cursor=match_index + len(new_lines),
195 + line_from=match_index + 1,
196 + line_to=match_index + max(len(new_lines), 1),
197 + )
198 +
199 +
200 +def _find_anchor(lines: list[str], anchor: str, start: int) -> int:
201 + for index in range(start, len(lines)):
202 + if lines[index] == anchor:
203 + return index + 1
204 + stripped_anchor = anchor.strip()
205 + for index in range(start, len(lines)):
206 + if lines[index].strip() == stripped_anchor:
207 + return index + 1
208 + raise ContextPatchError(f"anchor not found: {anchor}")
209 +
210 +
211 +def _find_context(
212 + lines: list[str],
213 + context: list[str],
214 + start: int,
215 + *,
216 + anchored: bool,
217 +) -> int:
218 + matches = _matching_indexes(lines, context, start, mode="exact")
219 + if not matches:
220 + matches = _matching_indexes(lines, context, start, mode="rstrip")
221 +
222 + if not matches:
223 + preview = "\n".join(context[:5])
224 + raise ContextPatchError(f"context not found:\n{preview}")
225 + if len(matches) > 1 and not anchored:
226 + preview = "\n".join(context[:5])
227 + raise ContextPatchError(
228 + "context matched multiple locations; add an @@ anchor or more context:\n"
229 + f"{preview}"
230 + )
231 + return matches[0]
232 +
233 +
234 +def _matching_indexes(
235 + lines: list[str],
236 + context: list[str],
237 + start: int,
238 + *,
239 + mode: str,
240 +) -> list[int]:
241 + if len(lines) < len(context):
242 + return []
243 +
244 + matches: list[int] = []
245 + for index in range(max(start, 0), len(lines) - len(context) + 1):
246 + candidate = lines[index : index + len(context)]
247 + if _lines_equal(candidate, context, mode=mode):
248 + matches.append(index)
249 + return matches
250 +
251 +
252 +def _lines_equal(left: Iterable[str], right: Iterable[str], *, mode: str) -> bool:
253 + if mode == "rstrip":
254 + return [item.rstrip() for item in left] == [item.rstrip() for item in right]
255 + return list(left) == list(right)
plugins/_text_editor/helpers/file_ops.py
+39
@@ -10,6 +10,9 @@ import tempfile
10 from typing import TypedDict
11
12 from helpers import tokens
13 +from plugins._text_editor.helpers.context_patch import (
14 + apply_context_patch_with_metadata,
15 +)
16
17 _BINARY_PEEK = 8192
18
@@ -208,6 +211,13 @@ class PatchResult(TypedDict):
211 error: str
212
213
214 +class ContextPatchFileResult(TypedDict):
215 + total_lines: int
216 + hunk_count: int
217 + line_from: int
218 + line_to: int
219 +
220 +
221 def validate_edits(edits: list | None) -> tuple[list[dict], str]:
222 """
223 Normalise and validate an edits array.
@@ -352,6 +362,35 @@ def patch_file(path: str, edits: list | None) -> PatchResult:
362 return PatchResult(total_lines=total, edit_count=len(parsed), error="")
363
364
365 +def apply_context_patch_file(path: str, patch_text: str) -> ContextPatchFileResult:
366 + """Apply a context patch to an existing text file."""
367 + path = os.path.expanduser(path)
368 + if not os.path.isfile(path):
369 + raise FileNotFoundError("file not found")
370 +
371 + with open(path, "r", encoding="utf-8", errors="replace") as src:
372 + content = src.read()
373 +
374 + result = apply_context_patch_with_metadata(content, patch_text)
375 + dir_name = os.path.dirname(path) or "."
376 + fd, tmp_path = tempfile.mkstemp(dir=dir_name, suffix=".tmp")
377 + try:
378 + with os.fdopen(fd, "w", encoding="utf-8") as dst:
379 + dst.write(result.content)
380 + shutil.move(tmp_path, path)
381 + except Exception:
382 + if os.path.exists(tmp_path):
383 + os.unlink(tmp_path)
384 + raise
385 +
386 + return ContextPatchFileResult(
387 + total_lines=_count_content_lines(result.content),
388 + hunk_count=result.hunk_count,
389 + line_from=result.line_from,
390 + line_to=result.line_to,
391 + )
392 +
393 +
394 # ------------------------------------------------------------------
395 # Internal
396 # ------------------------------------------------------------------
plugins/_text_editor/helpers/patch_request.py new
+37
@@ -0,0 +1,37 @@
1 +from __future__ import annotations
2 +
3 +from dataclasses import dataclass
4 +from typing import Any, Literal
5 +
6 +
7 +PatchMode = Literal["edits", "patch_text"]
8 +
9 +
10 +@dataclass(frozen=True)
11 +class PatchRequest:
12 + mode: PatchMode
13 + edits: Any = None
14 + patch_text: str = ""
15 +
16 +
17 +def parse_patch_request(
18 + edits: Any,
19 + patch_text: Any,
20 + *,
21 + both_error: str = "provide either edits or patch_text, not both",
22 + missing_error: str = "edits or patch_text is required for patch",
23 +) -> tuple[PatchRequest | None, str]:
24 + """Validate the mutually-exclusive patch request shape."""
25 + if edits is not None and patch_text is not None:
26 + return None, both_error
27 +
28 + if patch_text is not None:
29 + text = str(patch_text)
30 + if not text.strip():
31 + return None, "patch_text must not be empty"
32 + return PatchRequest(mode="patch_text", patch_text=text), ""
33 +
34 + if not edits:
35 + return None, missing_error
36 +
37 + return PatchRequest(mode="edits", edits=edits), ""
plugins/_text_editor/helpers/patch_state.py new
+187
@@ -0,0 +1,187 @@
1 +from __future__ import annotations
2 +
3 +from typing import Any, TypedDict
4 +
5 +
6 +LOCAL_FRESHNESS_KEY = "_text_editor_mtimes"
7 +REMOTE_FRESHNESS_KEY = "_a0_connector_text_editor_remote_mtimes"
8 +
9 +
10 +class FileMetadata(TypedDict):
11 + realpath: str
12 + mtime: float | None
13 + total_lines: int
14 +
15 +
16 +def coerce_file_metadata(
17 + file_data: Any, *, total_lines: int | None = None
18 +) -> FileMetadata | None:
19 + if not isinstance(file_data, dict):
20 + return None
21 +
22 + realpath = str(file_data.get("realpath", "")).strip()
23 + if not realpath:
24 + return None
25 +
26 + try:
27 + line_count = (
28 + int(total_lines)
29 + if total_lines is not None
30 + else int(file_data.get("total_lines", 0))
31 + )
32 + except (TypeError, ValueError):
33 + return None
34 +
35 + raw_mtime = file_data.get("mtime")
36 + if raw_mtime is None:
37 + mtime: float | None = None
38 + else:
39 + try:
40 + mtime = float(raw_mtime)
41 + except (TypeError, ValueError):
42 + mtime = None
43 +
44 + return FileMetadata(
45 + realpath=realpath,
46 + mtime=mtime,
47 + total_lines=max(line_count, 0),
48 + )
49 +
50 +
51 +def record_file_state(
52 + agent,
53 + file_data: Any,
54 + *,
55 + key: str,
56 + total_lines: int | None = None,
57 +) -> None:
58 + file_meta = coerce_file_metadata(file_data, total_lines=total_lines)
59 + if file_meta is None or file_meta["mtime"] is None:
60 + return
61 +
62 + freshness = agent.data.setdefault(key, {})
63 + freshness[file_meta["realpath"]] = {
64 + "mtime": file_meta["mtime"],
65 + "total_lines": file_meta["total_lines"],
66 + }
67 +
68 +
69 +def mark_file_state_stale(
70 + agent,
71 + file_data: Any,
72 + *,
73 + key: str,
74 + total_lines: int | None = None,
75 +) -> None:
76 + file_meta = coerce_file_metadata(file_data, total_lines=total_lines)
77 + if file_meta is None:
78 + return
79 +
80 + freshness = agent.data.setdefault(key, {})
81 + freshness[file_meta["realpath"]] = {"mtime": 0, "total_lines": 0}
82 +
83 +
84 +def check_patch_freshness(agent, file_data: Any, *, key: str) -> str | None:
85 + file_meta = coerce_file_metadata(file_data)
86 + if file_meta is None:
87 + return "patch_need_read"
88 +
89 + freshness = agent.data.get(key, {})
90 + realpath = file_meta["realpath"]
91 + if realpath not in freshness:
92 + return "patch_need_read"
93 +
94 + stored = freshness[realpath]
95 + mtime = stored.get("mtime") if isinstance(stored, dict) else stored
96 + if mtime is None:
97 + freshness.pop(realpath, None)
98 + return "patch_need_read"
99 +
100 + current = file_meta["mtime"]
101 + if current is None:
102 + return None
103 + if current != mtime:
104 + return "patch_stale_read"
105 + return None
106 +
107 +
108 +def apply_patch_post_state(
109 + agent,
110 + file_data: Any,
111 + edits: list[Any] | None,
112 + *,
113 + key: str,
114 + total_lines: int | None = None,
115 +) -> None:
116 + file_meta = coerce_file_metadata(file_data, total_lines=total_lines)
117 + if file_meta is None:
118 + return
119 +
120 + freshness = agent.data.setdefault(key, {})
121 + realpath = file_meta["realpath"]
122 +
123 + if not all_edits_in_place(edits):
124 + freshness[realpath] = {"mtime": 0, "total_lines": 0}
125 + return
126 +
127 + stored = freshness.get(realpath)
128 + if not isinstance(stored, dict) or "total_lines" not in stored:
129 + freshness[realpath] = {"mtime": 0, "total_lines": 0}
130 + return
131 +
132 + if file_meta["total_lines"] != int(stored["total_lines"]):
133 + freshness[realpath] = {"mtime": 0, "total_lines": 0}
134 + return
135 +
136 + if file_meta["mtime"] is None:
137 + freshness[realpath] = {"mtime": 0, "total_lines": 0}
138 + return
139 +
140 + freshness[realpath] = {
141 + "mtime": file_meta["mtime"],
142 + "total_lines": file_meta["total_lines"],
143 + }
144 +
145 +
146 +def all_edits_in_place(edits: list[Any] | None) -> bool:
147 + if not isinstance(edits, list):
148 + return False
149 +
150 + for edit in edits:
151 + if not isinstance(edit, dict):
152 + return False
153 + if edit.get("insert"):
154 + return False
155 +
156 + try:
157 + start = int(edit.get("from", 0) or 0)
158 + except (TypeError, ValueError):
159 + return False
160 + if start < 1:
161 + return False
162 +
163 + raw_to = edit.get("to")
164 + if raw_to is None:
165 + return False
166 +
167 + try:
168 + end = int(raw_to)
169 + except (TypeError, ValueError):
170 + return False
171 + if end < start:
172 + return False
173 +
174 + removed = end - start + 1
175 + added = count_content_lines(edit.get("content"))
176 + if removed != added:
177 + return False
178 +
179 + return True
180 +
181 +
182 +def count_content_lines(content: Any) -> int:
183 + if not content:
184 + return 0
185 +
186 + text = str(content)
187 + return text.count("\n") + (1 if not text.endswith("\n") else 0)
plugins/_text_editor/prompts/agent.system.tool.text_editor.md
+13 -16
@@ -38,19 +38,19 @@ usage:
38 ~~~
39
40 #### text_editor:patch
41 -line edits on existing file
42 -args path edits [{from to content}]
43 -from to inclusive \n in content
44 -{from:2 to:2 content:"x\n"} replace line
45 -{from:1 to:3 content:"x\n"} replace range
46 -{from:2 to:2} delete (no content)
47 -{from:2 content:"x\n"} insert before (omit to)
48 -use original line numbers from read
49 -dont adjust for shifts no overlapping edits
41 +edit existing file. prefer patch_text; use edits only right after read for tiny line edits
42 +args path plus exactly one of: patch_text string OR edits [{from to content}]
43 +patch_text uses current file content, no prior read required
44 +patch_text update-only forms:
45 +- insert after anchor: @@ exact existing line then +new lines
46 +- replace: use @@ line before target then -old +new, or @@ old target line then -same old target line +new
47 +- do not repeat the same old line as both a space-context line and a -removed line
48 +- context lines start with space, removals with -, additions with +
49 +- use enough unique context; add @@ anchor when repeated text exists
50 +edits legacy line mode: from/to inclusive, original line numbers from read, no overlaps
51 +edits examples: {from:2 to:2 content:"x\n"} replace; {from:2 to:2} delete; {from:2 content:"x\n"} insert before
52 +for edits, re-read after insert/delete or line-count-changing replace
53 ensure valid syntax in content (all braces brackets tags closed)
51 -only replace exact lines needed dont include surrounding unchanged lines
52 -re-read when insert delete or N≠M replace else patch again ok
53 -large changes write over multiple patches
54 usage:
55 ~~~json
56 {
@@ -58,10 +58,7 @@ usage:
58 "tool_name": "text_editor:patch",
59 "tool_args": {
60 "path": "/path/file.py",
61 - "edits": [
62 - {"from": 1, "content": "import sys\n"},
63 - {"from": 5, "to": 5, "content": " if x == 2:\n"}
64 - ]
61 + "patch_text": "*** Begin Patch\n*** Update File: file.py\n@@ def run():\n+ print('ready')\n*** End Patch"
62 }
63 }
64 ~~~
plugins/_text_editor/tools/text_editor.py
+141 -80
@@ -7,12 +7,21 @@ from plugins._text_editor.helpers.file_ops import (
7 write_file,
8 validate_edits,
9 apply_patch,
10 + apply_context_patch_file,
11 file_info,
12 )
13 +from plugins._text_editor.helpers.patch_request import parse_patch_request
14 +from plugins._text_editor.helpers.patch_state import (
15 + LOCAL_FRESHNESS_KEY,
16 + apply_patch_post_state,
17 + check_patch_freshness,
18 + mark_file_state_stale,
19 + record_file_state,
20 +)
21
22 # Key used in agent.data to store file state for patch validation
23 # Value: {path: {"mtime": float, "total_lines": int}}
15 -_MTIME_KEY = "_text_editor_mtimes"
24 +_MTIME_KEY = LOCAL_FRESHNESS_KEY
25
26
27
@@ -56,7 +65,12 @@ class TextEditor(Tool):
65 return self._error("read", path, result["error"])
66
67 info = await runtime.call_development_function(file_info, path)
59 - _record_mtime(self.agent, info, result["total_lines"])
68 + record_file_state(
69 + self.agent,
70 + info,
71 + key=_MTIME_KEY,
72 + total_lines=result["total_lines"],
73 + )
74
75 # Extension point
76 ext_data = {
@@ -105,7 +119,12 @@ class TextEditor(Tool):
119 )
120
121 info = await runtime.call_development_function(file_info, path)
108 - _record_mtime(self.agent, info, result["total_lines"])
122 + record_file_state(
123 + self.agent,
124 + info,
125 + key=_MTIME_KEY,
126 + total_lines=result["total_lines"],
127 + )
128
129 cfg = _get_config(self.agent)
130 read_result = await runtime.call_development_function(
@@ -128,9 +147,18 @@ class TextEditor(Tool):
147 # ------------------------------------------------------------------
148 # PATCH
149 # ------------------------------------------------------------------
131 - async def _patch(self, path: str = "", edits=None, **kwargs) -> Response:
150 + async def _patch(
151 + self, path: str = "", edits=None, patch_text=None, **kwargs
152 + ) -> Response:
153 if not path:
154 return self._error("patch", path, "path is required")
155 + patch_request, err = parse_patch_request(
156 + edits,
157 + patch_text,
158 + missing_error="",
159 + )
160 + if err:
161 + return self._error("patch", path, err)
162
163 info = await runtime.call_development_function(file_info, path)
164 if not info["is_file"]:
@@ -138,9 +166,28 @@ class TextEditor(Tool):
166
167 expanded = info["expanded"]
168
141 - stale_err = _check_mtime(self.agent, info)
142 - if stale_err:
143 - return self._error("patch", path, stale_err)
169 + if patch_request and patch_request.mode == "patch_text":
170 + return await self._patch_context(
171 + path, expanded, patch_request.patch_text
172 + )
173 +
174 + return await self._patch_edits(
175 + path,
176 + expanded,
177 + info,
178 + patch_request.edits if patch_request else edits,
179 + )
180 +
181 + async def _patch_edits(
182 + self, path: str, expanded: str, info: FileInfo, edits
183 + ) -> Response:
184 + freshness_code = check_patch_freshness(self.agent, info, key=_MTIME_KEY)
185 + if freshness_code:
186 + return self._error(
187 + "patch",
188 + path,
189 + _freshness_error_message(self.agent, info, freshness_code),
190 + )
191
192 parsed, err = validate_edits(edits)
193 if err:
@@ -169,8 +216,12 @@ class TextEditor(Tool):
216 post_info = await runtime.call_development_function(
217 file_info, expanded
218 )
172 - _apply_patch_post(
173 - self.agent, post_info, total_lines, ext_data["edits"]
219 + apply_patch_post_state(
220 + self.agent,
221 + post_info,
222 + ext_data["edits"],
223 + key=_MTIME_KEY,
224 + total_lines=total_lines,
225 )
226
227 patch_content = await _read_patch_region(
@@ -186,6 +237,64 @@ class TextEditor(Tool):
237 )
238 return Response(message=msg, break_loop=False)
239
240 + async def _patch_context(
241 + self, path: str, expanded: str, patch_text
242 + ) -> Response:
243 + patch_text = str(patch_text)
244 + if not patch_text.strip():
245 + return self._error("patch", path, "patch_text must not be empty")
246 +
247 + # Extension point
248 + ext_data = {
249 + "path": expanded,
250 + "patch_text": patch_text,
251 + "edits": [],
252 + "mode": "patch_text",
253 + }
254 + await call_extensions_async(
255 + "text_editor_patch_before", agent=self.agent, data=ext_data
256 + )
257 +
258 + try:
259 + result = await runtime.call_development_function(
260 + apply_context_patch_file,
261 + ext_data["path"],
262 + ext_data["patch_text"],
263 + )
264 + except Exception as exc:
265 + return self._error("patch", path, str(exc))
266 +
267 + total_lines = result["total_lines"]
268 +
269 + # Extension point
270 + await call_extensions_async(
271 + "text_editor_patch_after", agent=self.agent,
272 + data={
273 + "path": ext_data["path"],
274 + "total_lines": total_lines,
275 + "hunk_count": result["hunk_count"],
276 + "mode": "patch_text",
277 + },
278 + )
279 +
280 + post_info = await runtime.call_development_function(
281 + file_info, ext_data["path"]
282 + )
283 + mark_file_state_stale(self.agent, post_info, key=_MTIME_KEY)
284 +
285 + patch_content = await _read_context_patch_region(
286 + ext_data["path"], result, _get_config(self.agent)
287 + )
288 +
289 + msg = self.agent.read_prompt(
290 + "fw.text_editor.patch_ok.md",
291 + path=ext_data["path"],
292 + edit_count=str(result["hunk_count"]),
293 + total_lines=str(total_lines),
294 + content=patch_content,
295 + )
296 + return Response(message=msg, break_loop=False)
297 +
298 # ------------------------------------------------------------------
299 # Shared error helper
300 # ------------------------------------------------------------------
@@ -230,82 +339,34 @@ async def _read_patch_region(
339 return result["content"]
340
341
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"]] = {
237 - "mtime": info["mtime"],
238 - "total_lines": total_lines,
239 - }
342 +async def _read_context_patch_region(
343 + path: str, result: dict, cfg: dict
344 +) -> str:
345 + total_lines = int(result["total_lines"])
346 + if total_lines <= 0:
347 + return ""
348
349 + line_from = min(max(int(result["line_from"]), 1), total_lines)
350 + line_to = min(max(int(result["line_to"]), line_from) + 3, total_lines)
351
242 -def _count_content_lines(content: str) -> int:
243 - return content.count("\n") + (
244 - 1 if content and not content.endswith("\n") else 0
352 + read_result = await runtime.call_development_function(
353 + read_file,
354 + path,
355 + line_from=max(line_from - 1, 1),
356 + line_to=line_to,
357 + max_line_tokens=cfg["max_line_tokens"],
358 + max_total_read_tokens=cfg["max_total_read_tokens"],
359 )
360 + return read_result["content"]
361
362
248 -def _all_edits_in_place(edits: list[dict]) -> bool:
249 - for e in edits:
250 - if e.get("insert"):
251 - return False
252 - removed = max(e["to"] - e["from"] + 1, 0)
253 - added = _count_content_lines(e.get("content", "") or "")
254 - if removed != added:
255 - return False
256 - return True
257 -
258 -
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 -
265 - if not _all_edits_in_place(edits):
266 - # Line count changed — mark stale so next patch gets
267 - # "file changed since last read" instead of "line numbers unknown"
268 - mtimes[real] = {"mtime": 0, "total_lines": 0}
269 - return
270 -
271 - stored = mtimes.get(real)
272 - if not isinstance(stored, dict) or "total_lines" not in stored:
273 - mtimes[real] = {"mtime": 0, "total_lines": 0}
274 - return
275 - if new_total != stored["total_lines"]:
276 - mtimes[real] = {"mtime": 0, "total_lines": 0}
277 - return
278 - if info["mtime"] is not None:
279 - mtimes[real] = {
280 - "mtime": info["mtime"],
281 - "total_lines": new_total,
282 - }
283 - else:
284 - mtimes[real] = {"mtime": 0, "total_lines": 0}
285 -
286 -
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:
291 - return agent.read_prompt(
292 - "fw.text_editor.patch_need_read.md", path=info["expanded"]
293 - )
294 - stored = mtimes[real]
295 - mtime = stored.get("mtime") if isinstance(stored, dict) else stored
296 - if mtime is None:
297 - mtimes.pop(real, None)
298 - return agent.read_prompt(
299 - "fw.text_editor.patch_need_read.md", path=info["expanded"]
300 - )
301 - current = info["mtime"]
302 - if current is None:
303 - return ""
304 - if current != mtime:
305 - return agent.read_prompt(
306 - "fw.text_editor.patch_stale_read.md", path=info["expanded"]
307 - )
308 - return ""
363 +def _freshness_error_message(agent, info: FileInfo, code: str) -> str:
364 + prompt = (
365 + "fw.text_editor.patch_stale_read.md"
366 + if code == "patch_stale_read"
367 + else "fw.text_editor.patch_need_read.md"
368 + )
369 + return agent.read_prompt(prompt, path=info["expanded"])
370
371 # ------------------------------------------------------------------
372 # Config
tests/test_text_editor_context_patch.py new
+498
@@ -0,0 +1,498 @@
1 +import asyncio
2 +import importlib
3 +import inspect
4 +import os
5 +import sys
6 +import types
7 +from dataclasses import dataclass
8 +from pathlib import Path
9 +
10 +import pytest
11 +
12 +PROJECT_ROOT = Path(__file__).resolve().parents[1]
13 +if str(PROJECT_ROOT) not in sys.path:
14 + sys.path.insert(0, str(PROJECT_ROOT))
15 +
16 +from plugins._text_editor.helpers.context_patch import ContextPatchError
17 +from plugins._text_editor.helpers.file_ops import apply_context_patch_file
18 +from plugins._text_editor.helpers.patch_request import parse_patch_request
19 +from plugins._text_editor.helpers.patch_state import (
20 + LOCAL_FRESHNESS_KEY,
21 + REMOTE_FRESHNESS_KEY,
22 + apply_patch_post_state,
23 + check_patch_freshness,
24 + mark_file_state_stale,
25 + record_file_state,
26 +)
27 +
28 +
29 +def test_context_patch_chains_after_line_shift(tmp_path: Path) -> None:
30 + target = tmp_path / "sample.txt"
31 + target.write_text("alpha\nbeta\ngamma\n", encoding="utf-8")
32 +
33 + first = apply_context_patch_file(
34 + str(target),
35 + (
36 + "*** Begin Patch\n"
37 + "*** Update File: sample.txt\n"
38 + "@@ alpha\n"
39 + "+inserted\n"
40 + "*** End Patch"
41 + ),
42 + )
43 + second = apply_context_patch_file(
44 + str(target),
45 + (
46 + "*** Begin Patch\n"
47 + "*** Update File: sample.txt\n"
48 + " beta\n"
49 + "-gamma\n"
50 + "+gamma-updated\n"
51 + "*** End Patch"
52 + ),
53 + )
54 +
55 + assert first["total_lines"] == 4
56 + assert first["hunk_count"] == 1
57 + assert second["total_lines"] == 4
58 + assert target.read_text(encoding="utf-8") == (
59 + "alpha\ninserted\nbeta\ngamma-updated\n"
60 + )
61 +
62 +
63 +def test_context_patch_inserts_after_anchor(tmp_path: Path) -> None:
64 + target = tmp_path / "sample.txt"
65 + target.write_text("alpha\nbeta\n", encoding="utf-8")
66 +
67 + result = apply_context_patch_file(
68 + str(target),
69 + (
70 + "*** Begin Patch\n"
71 + "*** Update File: sample.txt\n"
72 + "@@ alpha\n"
73 + "+inserted\n"
74 + "*** End Patch"
75 + ),
76 + )
77 +
78 + assert result["line_from"] == 2
79 + assert result["line_to"] == 2
80 + assert target.read_text(encoding="utf-8") == "alpha\ninserted\nbeta\n"
81 +
82 +
83 +def test_context_patch_replaces_matching_context(tmp_path: Path) -> None:
84 + target = tmp_path / "sample.txt"
85 + target.write_text("alpha\nbeta\ngamma\n", encoding="utf-8")
86 +
87 + result = apply_context_patch_file(
88 + str(target),
89 + (
90 + "*** Begin Patch\n"
91 + "*** Update File: sample.txt\n"
92 + " beta\n"
93 + "-gamma\n"
94 + "+delta\n"
95 + "*** End Patch"
96 + ),
97 + )
98 +
99 + assert result["line_from"] == 2
100 + assert target.read_text(encoding="utf-8") == "alpha\nbeta\ndelta\n"
101 +
102 +
103 +def test_context_patch_replaces_when_anchor_is_target_line(
104 + tmp_path: Path,
105 +) -> None:
106 + target = tmp_path / "sample.py"
107 + target.write_text(
108 + (
109 + "def main():\n"
110 + " print(greet(\"Agent Zero\"))\n"
111 + "\n"
112 + "\n"
113 + "if __name__ == \"__main__\":\n"
114 + " main()\n"
115 + ),
116 + encoding="utf-8",
117 + )
118 +
119 + result = apply_context_patch_file(
120 + str(target),
121 + (
122 + "*** Begin Patch\n"
123 + "*** Update File: sample.py\n"
124 + "@@ print(greet(\"Agent Zero\"))\n"
125 + "- print(greet(\"Agent Zero\"))\n"
126 + "+ print(greet(\"Agent Zero\").upper())\n"
127 + "*** End Patch"
128 + ),
129 + )
130 +
131 + assert result["line_from"] == 2
132 + assert target.read_text(encoding="utf-8") == (
133 + "def main():\n"
134 + " print(greet(\"Agent Zero\").upper())\n"
135 + "\n"
136 + "\n"
137 + "if __name__ == \"__main__\":\n"
138 + " main()\n"
139 + )
140 +
141 +
142 +def test_context_patch_rejects_ambiguous_unanchored_context(
143 + tmp_path: Path,
144 +) -> None:
145 + target = tmp_path / "sample.txt"
146 + target.write_text("same\nold\nsame\nold\n", encoding="utf-8")
147 +
148 + with pytest.raises(ContextPatchError, match="matched multiple locations"):
149 + apply_context_patch_file(
150 + str(target),
151 + (
152 + "*** Begin Patch\n"
153 + "*** Update File: sample.txt\n"
154 + " same\n"
155 + "-old\n"
156 + "+new\n"
157 + "*** End Patch"
158 + ),
159 + )
160 +
161 +
162 +@pytest.mark.parametrize(
163 + "patch_text, expected",
164 + [
165 + (
166 + "*** Begin Patch\n*** Add File: sample.txt\n+x\n*** End Patch",
167 + "supports update hunks only",
168 + ),
169 + (
170 + "*** Begin Patch\n*** Delete File: sample.txt\n*** End Patch",
171 + "supports update hunks only",
172 + ),
173 + (
174 + (
175 + "*** Begin Patch\n"
176 + "*** Update File: sample.txt\n"
177 + "*** Move to: other.txt\n"
178 + "*** End Patch"
179 + ),
180 + "does not support file moves",
181 + ),
182 + (
183 + (
184 + "*** Begin Patch\n"
185 + "*** Update File: sample.txt\n"
186 + "@@ alpha\n"
187 + "+one\n"
188 + "*** Update File: other.txt\n"
189 + "@@ beta\n"
190 + "+two\n"
191 + "*** End Patch"
192 + ),
193 + "may update only one file",
194 + ),
195 + ],
196 +)
197 +def test_context_patch_rejects_unsupported_file_operations(
198 + tmp_path: Path, patch_text: str, expected: str
199 +) -> None:
200 + target = tmp_path / "sample.txt"
201 + target.write_text("alpha\nbeta\n", encoding="utf-8")
202 +
203 + with pytest.raises(ContextPatchError, match=expected):
204 + apply_context_patch_file(str(target), patch_text)
205 +
206 +
207 +def test_patch_request_rejects_edits_and_patch_text_together() -> None:
208 + request, err = parse_patch_request(
209 + [{"from": 1, "to": 1, "content": "x\n"}],
210 + "@@ alpha\n+beta",
211 + )
212 +
213 + assert request is None
214 + assert err == "provide either edits or patch_text, not both"
215 +
216 +
217 +def test_patch_request_rejects_empty_patch_text() -> None:
218 + request, err = parse_patch_request(None, " \n")
219 +
220 + assert request is None
221 + assert err == "patch_text must not be empty"
222 +
223 +
224 +def test_patch_state_records_and_checks_fresh_file_state() -> None:
225 + agent = _FakeAgent()
226 + file_data = {"realpath": "/tmp/sample.txt", "mtime": 1.0, "total_lines": 3}
227 +
228 + record_file_state(agent, file_data, key=LOCAL_FRESHNESS_KEY)
229 +
230 + assert check_patch_freshness(agent, file_data, key=LOCAL_FRESHNESS_KEY) is None
231 + assert check_patch_freshness(
232 + agent,
233 + {"realpath": "/tmp/sample.txt", "mtime": 2.0, "total_lines": 3},
234 + key=LOCAL_FRESHNESS_KEY,
235 + ) == "patch_stale_read"
236 +
237 +
238 +def test_patch_state_marks_context_patches_stale() -> None:
239 + agent = _FakeAgent()
240 + file_data = {"realpath": "/tmp/sample.txt", "mtime": 1.0, "total_lines": 3}
241 +
242 + record_file_state(agent, file_data, key=LOCAL_FRESHNESS_KEY)
243 + mark_file_state_stale(agent, file_data, key=LOCAL_FRESHNESS_KEY)
244 +
245 + assert agent.data[LOCAL_FRESHNESS_KEY]["/tmp/sample.txt"] == {
246 + "mtime": 0,
247 + "total_lines": 0,
248 + }
249 +
250 +
251 +def test_patch_state_line_preserving_edits_can_chain() -> None:
252 + agent = _FakeAgent()
253 + initial = {"realpath": "/tmp/sample.txt", "mtime": 1.0, "total_lines": 3}
254 + patched = {"realpath": "/tmp/sample.txt", "mtime": 2.0, "total_lines": 3}
255 + edits = [{"from": 2, "to": 2, "content": "line-2a\n"}]
256 +
257 + record_file_state(agent, initial, key=LOCAL_FRESHNESS_KEY)
258 + apply_patch_post_state(agent, patched, edits, key=LOCAL_FRESHNESS_KEY)
259 +
260 + assert agent.data[LOCAL_FRESHNESS_KEY]["/tmp/sample.txt"] == {
261 + "mtime": 2.0,
262 + "total_lines": 3,
263 + }
264 + assert check_patch_freshness(agent, patched, key=LOCAL_FRESHNESS_KEY) is None
265 +
266 +
267 +def test_patch_state_line_count_changes_force_reread() -> None:
268 + agent = _FakeAgent()
269 + initial = {"realpath": "/tmp/sample.txt", "mtime": 1.0, "total_lines": 3}
270 + patched = {"realpath": "/tmp/sample.txt", "mtime": 2.0, "total_lines": 4}
271 + edits = [{"from": 2, "content": "inserted\n"}]
272 +
273 + record_file_state(agent, initial, key=LOCAL_FRESHNESS_KEY)
274 + apply_patch_post_state(agent, patched, edits, key=LOCAL_FRESHNESS_KEY)
275 +
276 + assert agent.data[LOCAL_FRESHNESS_KEY]["/tmp/sample.txt"] == {
277 + "mtime": 0,
278 + "total_lines": 0,
279 + }
280 +
281 +
282 +def test_patch_state_uses_separate_local_and_remote_keys() -> None:
283 + agent = _FakeAgent()
284 + file_data = {"realpath": "/tmp/sample.txt", "mtime": 1.0, "total_lines": 3}
285 +
286 + record_file_state(agent, file_data, key=LOCAL_FRESHNESS_KEY)
287 + mark_file_state_stale(agent, file_data, key=REMOTE_FRESHNESS_KEY)
288 +
289 + assert agent.data[LOCAL_FRESHNESS_KEY]["/tmp/sample.txt"] == {
290 + "mtime": 1.0,
291 + "total_lines": 3,
292 + }
293 + assert agent.data[REMOTE_FRESHNESS_KEY]["/tmp/sample.txt"] == {
294 + "mtime": 0,
295 + "total_lines": 0,
296 + }
297 +
298 +
299 +@dataclass
300 +class _FakeResponse:
301 + message: str
302 + break_loop: bool
303 + additional: dict | None = None
304 +
305 +
306 +class _FakeTool:
307 + def __init__(
308 + self,
309 + agent,
310 + name: str = "text_editor",
311 + method: str = "patch",
312 + args: dict | None = None,
313 + message: str = "",
314 + loop_data=None,
315 + **kwargs,
316 + ) -> None:
317 + self.agent = agent
318 + self.name = name
319 + self.method = method
320 + self.args = args or {}
321 + self.message = message
322 + self.loop_data = loop_data
323 +
324 +
325 +class _FakeAgent:
326 + def __init__(self) -> None:
327 + self.data = {}
328 +
329 + def read_prompt(self, name: str, **kwargs) -> str:
330 + if name.endswith("patch_ok.md"):
331 + return (
332 + f"{kwargs['path']} patched {kwargs['edit_count']} edits applied "
333 + f"{kwargs['total_lines']} lines now\n>>>\n{kwargs['content']}\n<<<"
334 + )
335 + if name.endswith("patch_need_read.md"):
336 + return f"must read {kwargs['path']} first"
337 + if name.endswith("patch_stale_read.md"):
338 + return f"stale read for {kwargs['path']}"
339 + return f"error patching {kwargs.get('path')}: {kwargs.get('error')}"
340 +
341 +
342 +def _load_text_editor_tool(monkeypatch: pytest.MonkeyPatch):
343 + calls: list[tuple[str, dict | None]] = []
344 +
345 + tool_stub = types.ModuleType("helpers.tool")
346 + tool_stub.Tool = _FakeTool
347 + tool_stub.Response = _FakeResponse
348 +
349 + extension_stub = types.ModuleType("helpers.extension")
350 +
351 + async def call_extensions_async(name: str, *args, **kwargs):
352 + calls.append((name, kwargs.get("data")))
353 +
354 + extension_stub.call_extensions_async = call_extensions_async
355 +
356 + plugins_stub = types.ModuleType("helpers.plugins")
357 + plugins_stub.get_plugin_config = lambda *args, **kwargs: {}
358 +
359 + runtime_stub = types.ModuleType("helpers.runtime")
360 +
361 + async def call_development_function(func, *args, **kwargs):
362 + result = func(*args, **kwargs)
363 + if inspect.isawaitable(result):
364 + return await result
365 + return result
366 +
367 + runtime_stub.call_development_function = call_development_function
368 +
369 + monkeypatch.setitem(sys.modules, "helpers.tool", tool_stub)
370 + monkeypatch.setitem(sys.modules, "helpers.extension", extension_stub)
371 + monkeypatch.setitem(sys.modules, "helpers.plugins", plugins_stub)
372 + monkeypatch.setitem(sys.modules, "helpers.runtime", runtime_stub)
373 + sys.modules.pop("plugins._text_editor.tools.text_editor", None)
374 + module = importlib.import_module("plugins._text_editor.tools.text_editor")
375 + return module, calls
376 +
377 +
378 +def test_text_editor_patch_text_does_not_require_prior_read(
379 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch
380 +) -> None:
381 + module, calls = _load_text_editor_tool(monkeypatch)
382 + target = tmp_path / "sample.txt"
383 + target.write_text("line-1\nline-2\nline-3\n", encoding="utf-8")
384 + agent = _FakeAgent()
385 + tool = module.TextEditor(agent, "text_editor", "patch", {}, "", None)
386 +
387 + response = asyncio.run(
388 + tool._patch(
389 + path=str(target),
390 + patch_text=(
391 + "*** Begin Patch\n"
392 + "*** Update File: sample.txt\n"
393 + "@@ line-1\n"
394 + "+inserted\n"
395 + "*** End Patch"
396 + ),
397 + )
398 + )
399 +
400 + assert "patched 1 edits applied 4 lines now" in response.message
401 + assert "inserted" in response.message
402 + assert target.read_text(encoding="utf-8") == (
403 + "line-1\ninserted\nline-2\nline-3\n"
404 + )
405 + realpath = os.path.realpath(target)
406 + assert agent.data[module._MTIME_KEY][realpath] == {
407 + "mtime": 0,
408 + "total_lines": 0,
409 + }
410 + assert calls[0] == (
411 + "text_editor_patch_before",
412 + {
413 + "path": str(target),
414 + "patch_text": (
415 + "*** Begin Patch\n"
416 + "*** Update File: sample.txt\n"
417 + "@@ line-1\n"
418 + "+inserted\n"
419 + "*** End Patch"
420 + ),
421 + "edits": [],
422 + "mode": "patch_text",
423 + },
424 + )
425 + assert calls[1][0] == "text_editor_patch_after"
426 + assert calls[1][1]["mode"] == "patch_text"
427 +
428 +
429 +def test_text_editor_patch_text_rejects_simultaneous_edits(
430 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch
431 +) -> None:
432 + module, _calls = _load_text_editor_tool(monkeypatch)
433 + target = tmp_path / "sample.txt"
434 + target.write_text("line-1\n", encoding="utf-8")
435 + tool = module.TextEditor(_FakeAgent(), "text_editor", "patch", {}, "", None)
436 +
437 + response = asyncio.run(
438 + tool._patch(
439 + path=str(target),
440 + edits=[{"from": 1, "to": 1, "content": "updated\n"}],
441 + patch_text="@@ line-1\n+inserted",
442 + )
443 + )
444 +
445 + assert "provide either edits or patch_text" in response.message
446 + assert target.read_text(encoding="utf-8") == "line-1\n"
447 +
448 +
449 +def test_text_editor_patch_text_marks_existing_line_state_stale(
450 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch
451 +) -> None:
452 + module, _calls = _load_text_editor_tool(monkeypatch)
453 + target = tmp_path / "sample.txt"
454 + target.write_text("line-1\nline-2\n", encoding="utf-8")
455 + realpath = os.path.realpath(target)
456 + agent = _FakeAgent()
457 + agent.data[module._MTIME_KEY] = {
458 + realpath: {"mtime": os.path.getmtime(target), "total_lines": 2}
459 + }
460 + tool = module.TextEditor(agent, "text_editor", "patch", {}, "", None)
461 +
462 + asyncio.run(
463 + tool._patch(
464 + path=str(target),
465 + patch_text=(
466 + "*** Begin Patch\n"
467 + "*** Update File: sample.txt\n"
468 + "@@ line-1\n"
469 + "+inserted\n"
470 + "*** End Patch"
471 + ),
472 + )
473 + )
474 +
475 + assert agent.data[module._MTIME_KEY][realpath] == {
476 + "mtime": 0,
477 + "total_lines": 0,
478 + }
479 +
480 +
481 +def test_text_editor_line_edits_still_require_prior_read(
482 + tmp_path: Path, monkeypatch: pytest.MonkeyPatch
483 +) -> None:
484 + module, calls = _load_text_editor_tool(monkeypatch)
485 + target = tmp_path / "sample.txt"
486 + target.write_text("line-1\n", encoding="utf-8")
487 + tool = module.TextEditor(_FakeAgent(), "text_editor", "patch", {}, "", None)
488 +
489 + response = asyncio.run(
490 + tool._patch(
491 + path=str(target),
492 + edits=[{"from": 1, "to": 1, "content": "updated\n"}],
493 + )
494 + )
495 +
496 + assert "must read" in response.message
497 + assert target.read_text(encoding="utf-8") == "line-1\n"
498 + assert calls == []