@samitouri / QOSami_Agents_Zero / commits / f86d1c55

add connector stale-read protection to remote patching

Add _text_editor-style freshness checks to the _a0_connector remote text editor flow. - add local freshness helpers for remote file metadata and patch-state tracking - require a prior read or write before allowing remote patch operations - run remote patches through stat -> stale check -> patch using private websocket plumbing - store freshness state in agent.data keyed by CLI-reported realpath - reuse fw.text_editor patch_need_read and patch_stale_read prompt behavior - refresh stored state after line-preserving patches and mark it stale after insert/delete or line-count changes - return a clear compatibility error when the connected CLI does not support internal stat This keeps the existing edits schema and human-facing success messages unchanged, and does not change remote tree publishing behavior. Bump plugin version to match CLI Connector.

Alessandro committed Apr 15, 2026 at 00:06 UTC f86d1c555cfa667297c0a27ba02b1256d0a93744
3 files changed +312 -38
plugins/_a0_connector/helpers/text_editor_freshness.py new
+159
@@ -0,0 +1,159 @@
1 +from __future__ import annotations
2 +
3 +from typing import Any, TypedDict
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 + )
42 +
43 +
44 +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 + }
54 +
55 +
56 +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}
63 +
64 +
65 +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 +
153 +
154 +def _count_content_lines(content: Any) -> int:
155 + if not content:
156 + return 0
157 +
158 + text = str(content)
159 + return text.count("\n") + (1 if not text.endswith("\n") else 0)
plugins/_a0_connector/plugin.yaml
+1 -1
@@ -1,7 +1,7 @@
1 name: _a0_connector
2 title: A0 Connector
3 description: Current Agent Zero connector plugin for HTTP plus /ws integration, using session auth and handler activation through auth.handlers.
4 -version: 0.1.0
4 +version: 1.4
5 settings_sections:
6 - external
7 - developer
plugins/_a0_connector/tools/text_editor_remote.py
+152 -37
@@ -1,4 +1,4 @@
1 -"""text_editor_remote tool — edit files on the CLI machine via `/ws`."""
1 +"""text_editor_remote tool - edit files on the CLI machine via `/ws`."""
2 from __future__ import annotations
3
4 import asyncio
@@ -9,6 +9,13 @@ from helpers.tool import Response, Tool
9 from helpers.ws import NAMESPACE
10 from helpers.ws_manager import ConnectionNotFoundError, get_shared_ws_manager
11
12 +from plugins._a0_connector.helpers.text_editor_freshness import (
13 + apply_patch_post_state,
14 + check_patch_freshness,
15 + coerce_file_metadata,
16 + mark_file_state_stale,
17 + record_file_state,
18 +)
19 from plugins._a0_connector.helpers.ws_runtime import (
20 clear_pending_file_op,
21 select_target_sid,
@@ -18,6 +25,10 @@ from plugins._a0_connector.helpers.ws_runtime import (
25
26 FILE_OP_TIMEOUT = 30.0
27 FILE_OP_EVENT = "connector_file_op"
28 +UNSUPPORTED_FRESHNESS_ERROR = (
29 + "text_editor_remote: the connected CLI is too old for freshness-aware patching. "
30 + "Upgrade the CLI and try again."
31 +)
32
33
34 class TextEditorRemote(Tool):
@@ -40,29 +51,14 @@ class TextEditorRemote(Tool):
51 if not path:
52 return Response(message="path is required", break_loop=False)
53
43 - context_id = self.agent.context.id
44 - sid = select_target_sid(context_id)
45 - if not sid:
46 - return Response(
47 - message=(
48 - "text_editor_remote: no CLI client connected to this context. "
49 - "Make sure the CLI is connected and subscribed."
50 - ),
51 - break_loop=False,
52 - )
53 -
54 - op_id = str(uuid.uuid4())
55 - payload: dict[str, Any] = {
56 - "op_id": op_id,
57 - "op": op,
58 - "path": path,
59 - "context_id": context_id,
60 - }
54 if op == "read":
55 + payload: dict[str, Any] = {}
56 if self.args.get("line_from"):
57 payload["line_from"] = int(self.args["line_from"])
58 if self.args.get("line_to"):
59 payload["line_to"] = int(self.args["line_to"])
60 + result = await self._execute_file_op(op, path, **payload)
61 + self._record_success_state(result)
62 elif op == "write":
63 content = self.args.get("content")
64 if content is None:
@@ -70,7 +66,8 @@ class TextEditorRemote(Tool):
66 message="content is required for write",
67 break_loop=False,
68 )
73 - payload["content"] = content
69 + result = await self._execute_file_op(op, path, content=content)
70 + self._record_success_state(result)
71 else:
72 edits = self.args.get("edits")
73 if not edits:
@@ -78,7 +75,74 @@ class TextEditorRemote(Tool):
75 message="edits is required for patch",
76 break_loop=False,
77 )
81 - payload["edits"] = edits
78 + result = await self._execute_patch(path, edits)
79 +
80 + return Response(
81 + message=self._extract_result(result, op, path),
82 + break_loop=False,
83 + )
84 +
85 + async def _execute_patch(self, path: str, edits: Any) -> dict[str, Any]:
86 + stat_result = await self._execute_file_op("stat", path)
87 + if self._is_unsupported_cli_freshness(stat_result):
88 + return self._freshness_error(
89 + "unsupported_cli_freshness",
90 + UNSUPPORTED_FRESHNESS_ERROR,
91 + )
92 + if not self._result_ok(stat_result):
93 + return stat_result
94 +
95 + stat_file = self._extract_file_metadata(stat_result)
96 + if stat_file is None:
97 + return self._freshness_error(
98 + "unsupported_cli_freshness",
99 + UNSUPPORTED_FRESHNESS_ERROR,
100 + )
101 +
102 + freshness_code = check_patch_freshness(self.agent, stat_file)
103 + if freshness_code:
104 + return self._freshness_error(freshness_code)
105 +
106 + patch_result = await self._execute_file_op("patch", path, edits=edits)
107 + if not self._result_ok(patch_result):
108 + return patch_result
109 +
110 + patch_file = self._extract_file_metadata(patch_result)
111 + if patch_file is None:
112 + mark_file_state_stale(self.agent, stat_file)
113 + else:
114 + apply_patch_post_state(
115 + self.agent,
116 + patch_file,
117 + edits if isinstance(edits, list) else [],
118 + )
119 + return patch_result
120 +
121 + async def _execute_file_op(
122 + self,
123 + op: str,
124 + path: str,
125 + **payload_extra: Any,
126 + ) -> dict[str, Any]:
127 + context_id = self.agent.context.id
128 + sid = select_target_sid(context_id)
129 + if not sid:
130 + return {
131 + "ok": False,
132 + "error": (
133 + "text_editor_remote: no CLI client connected to this context. "
134 + "Make sure the CLI is connected and subscribed."
135 + ),
136 + }
137 +
138 + op_id = str(uuid.uuid4())
139 + payload: dict[str, Any] = {
140 + "op_id": op_id,
141 + "op": op,
142 + "path": path,
143 + "context_id": context_id,
144 + }
145 + payload.update(payload_extra)
146
147 loop = asyncio.get_running_loop()
148 future: asyncio.Future[dict[str, Any]] = loop.create_future()
@@ -101,35 +165,73 @@ class TextEditorRemote(Tool):
165 result = await asyncio.wait_for(future, timeout=FILE_OP_TIMEOUT)
166 except ConnectionNotFoundError:
167 clear_pending_file_op(op_id)
104 - return Response(
105 - message=(
168 + return {
169 + "op_id": op_id,
170 + "ok": False,
171 + "error": (
172 "text_editor_remote: the selected CLI client disconnected before "
173 "the file operation could be delivered"
174 ),
109 - break_loop=False,
110 - )
175 + }
176 except asyncio.TimeoutError:
177 clear_pending_file_op(op_id)
113 - return Response(
114 - message=(
178 + return {
179 + "op_id": op_id,
180 + "ok": False,
181 + "error": (
182 f"text_editor_remote: timed out waiting for CLI to respond "
183 f"to {op} on {path!r}"
184 ),
118 - break_loop=False,
119 - )
185 + }
186 except Exception as exc:
187 clear_pending_file_op(op_id)
122 - return Response(
123 - message=f"text_editor_remote: error sending file_op: {exc}",
124 - break_loop=False,
125 - )
188 + return {
189 + "op_id": op_id,
190 + "ok": False,
191 + "error": f"text_editor_remote: error sending file_op: {exc}",
192 + }
193 finally:
194 clear_pending_file_op(op_id)
195
129 - return Response(
130 - message=self._extract_result(result, op, path),
131 - break_loop=False,
132 - )
196 + if isinstance(result, dict):
197 + return result
198 +
199 + return {
200 + "op_id": op_id,
201 + "ok": False,
202 + "error": f"Unexpected response format from CLI: {result!r}",
203 + }
204 +
205 + def _record_success_state(self, result: Any) -> None:
206 + if not self._result_ok(result):
207 + return
208 +
209 + file_meta = self._extract_file_metadata(result)
210 + if file_meta is not None:
211 + record_file_state(self.agent, file_meta)
212 +
213 + def _extract_file_metadata(self, result: Any) -> dict[str, Any] | None:
214 + if not isinstance(result, dict):
215 + return None
216 +
217 + data = result.get("result")
218 + if not isinstance(data, dict):
219 + return None
220 +
221 + return coerce_file_metadata(data.get("file"))
222 +
223 + def _freshness_error(self, code: str, error: str = "") -> dict[str, Any]:
224 + return {"ok": False, "code": code, "error": error}
225 +
226 + def _result_ok(self, result: Any) -> bool:
227 + return isinstance(result, dict) and bool(result.get("ok"))
228 +
229 + def _is_unsupported_cli_freshness(self, result: Any) -> bool:
230 + if not isinstance(result, dict) or bool(result.get("ok")):
231 + return False
232 +
233 + error = str(result.get("error") or "").strip().lower()
234 + return "unknown op: stat" in error
235
236 def _extract_result(self, result: Any, op: str, path: str) -> str:
237 if not isinstance(result, dict):
@@ -138,8 +240,21 @@ class TextEditorRemote(Tool):
240 ok = bool(result.get("ok"))
241 data = result.get("result")
242 error = result.get("error")
243 + code = str(result.get("code") or "").strip().lower()
244
245 if not ok:
246 + if code == "patch_need_read":
247 + return self.agent.read_prompt(
248 + "fw.text_editor.patch_need_read.md",
249 + path=path,
250 + )
251 + if code == "patch_stale_read":
252 + return self.agent.read_prompt(
253 + "fw.text_editor.patch_stale_read.md",
254 + path=path,
255 + )
256 + if code == "unsupported_cli_freshness":
257 + return str(error or UNSUPPORTED_FRESHNESS_ERROR)
258 return f"Error ({op} {path!r}): {error or 'Unknown error'}"
259
260 if not isinstance(data, dict):