@samitouri / QOSami_Agents_Zero-2 / commits / 32bf5889

Add deterministic sparse Agent Editor backend

Add dedicated Agent Editor APIs that read layered profile state and execute explicit change plans for metadata, prompts, model presets, tool policy, skill policy, and avatars. Keep writes sparse and user-scoped, preserve unknown files and keys, avoid the destructive legacy save path and all model calls, and cover profile lifecycle, validation, provenance, and reset behavior.

Alessandro committed Aug 5, 2026 at 10:54 UTC 32bf5889b022668d0215fb0667167c1999671274
9 files changed +2181
plugins/AGENTS.md
+1
@@ -69,6 +69,7 @@ Direct child DOX files:
69 | Child | Scope |
70 | --- | --- |
71 | [_a0_connector/AGENTS.md](_a0_connector/AGENTS.md) | HTTP and WebSocket connector integration with remote tools and runtime bridges. |
72 +| [_agent_editor/AGENTS.md](_agent_editor/AGENTS.md) | Deterministic sparse agent-profile editor API, helpers, and WebUI. |
73 | [_browser/AGENTS.md](_browser/AGENTS.md) | Playwright browser tool, helpers, viewer, and browser panel UI. |
74 | [_chat_branching/AGENTS.md](_chat_branching/AGENTS.md) | Chat branching from an existing message. |
75 | [_chat_compaction/AGENTS.md](_chat_compaction/AGENTS.md) | Full-chat compaction into a summary message. |
plugins/_agent_editor/AGENTS.md new
+38
@@ -0,0 +1,38 @@
1 +# Agent Editor Plugin DOX
2 +
3 +## Purpose
4 +
5 +- Own the deterministic Easy and Advanced Agent Editor API and WebUI workflow.
6 +
7 +## Ownership
8 +
9 +- `helpers/editor.py` composes existing profile, prompt, model, tool, and skill
10 + owners into editor state and sparse change plans.
11 +- `api/` owns authenticated/CSRF-protected state, save, and avatar routes.
12 +- `webui/` owns the Alpine store, modal surface, styling, and client-side draft.
13 +- `extensions/webui/` owns lifecycle registration on the shared modal stack and
14 + the global entry points used by existing WebUI surfaces.
15 +
16 +## Local Contracts
17 +
18 +- The editor performs zero model calls.
19 +- Writes are limited to `usr/agents/<profile-id>` and only to paths or config
20 + keys listed in the validated change plan.
21 +- Never call `helpers.subagents.save_agent_data`.
22 +- Authored profile definitions remain YAML; editor-written plugin configs remain
23 + JSON.
24 +- Profile config paths use `helpers.plugins.determine_plugin_asset_path` while
25 + remaining rooted in the editor's validated user-profile boundary.
26 +- Active project tool policy is resolved through the standard plugin asset
27 + provenance and shown as higher priority; the editor still writes only the
28 + user-profile scope.
29 +- Bundled `agents/` files are read-only.
30 +
31 +## Verification
32 +
33 +- Run Agent Editor, profile merge, tool policy, skill policy, API security, and
34 + WebUI tests, then verify the explicitly named bind-mounted runtime.
35 +
36 +## Child DOX Index
37 +
38 +No child DOX files.
plugins/_agent_editor/README.md new
+9
@@ -0,0 +1,9 @@
1 +# Agent Editor
2 +
3 +Agent Editor provides the deterministic Easy modal and Advanced workspace for
4 +Agent Zero profiles. It reads the existing layered profile architecture and
5 +writes only sparse user overrides under `usr/agents/<profile-id>`.
6 +
7 +The editor never invokes a model. Tool and skill controls are backed by the
8 +central runtime policy owners, and every save is previewed as exact file writes
9 +and deletions before the same validated plan is applied.
plugins/_agent_editor/api/agent_editor.py new
+79
@@ -0,0 +1,79 @@
1 +from __future__ import annotations
2 +
3 +from typing import Any
4 +
5 +from agent import AgentContext
6 +from helpers import projects
7 +from helpers.api import ApiHandler, Request, Response
8 +from plugins._agent_editor.helpers import editor
9 +
10 +
11 +class AgentEditor(ApiHandler):
12 + async def process(self, input: dict, request: Request) -> dict | Response:
13 + try:
14 + action = str(input.get("action") or "list").strip().lower()
15 + context = _context(input)
16 +
17 + if action == "list":
18 + return {"ok": True, "profiles": editor.list_profiles(context)}
19 + if action == "load":
20 + profile_id = editor.validate_profile_id(input.get("profile_id"))
21 + return {
22 + "ok": True,
23 + "state": editor.build_editor_state(profile_id, context),
24 + }
25 + if action in {"plan", "save"}:
26 + patch = input.get("patch")
27 + plan = editor.build_change_plan(patch, context)
28 + if action == "plan":
29 + return {"ok": True, **plan.response()}
30 + receipt = editor.apply_change_plan(plan)
31 + profile_id = editor.validate_profile_id(patch.get("profile_id"))
32 + return {
33 + "ok": True,
34 + **receipt,
35 + "effective_profile": editor.build_profile_state(profile_id, context),
36 + }
37 + if action in {"plan_remove_changes", "remove_changes"}:
38 + profile_id = editor.validate_profile_id(input.get("profile_id"))
39 + plan = editor.plan_remove_changes(
40 + profile_id,
41 + context,
42 + destructive=bool(input.get("destructive")),
43 + )
44 + if action == "plan_remove_changes":
45 + return {"ok": True, **plan.response()}
46 + return {"ok": True, **editor.apply_change_plan(plan)}
47 + if action == "delete_impact":
48 + return {
49 + "ok": True,
50 + "impact": editor.delete_impact(input.get("profile_id"), context),
51 + }
52 + if action in {"plan_delete", "delete"}:
53 + profile_id = editor.validate_profile_id(input.get("profile_id"))
54 + plan = editor.plan_delete_custom(profile_id, context)
55 + if action == "plan_delete":
56 + return {
57 + "ok": True,
58 + **plan.response(),
59 + "impact": editor.delete_impact(profile_id, context),
60 + }
61 + if input.get("confirm") is not True:
62 + raise ValueError("Deleting a custom agent requires confirmation.")
63 + return {"ok": True, **editor.apply_change_plan(plan)}
64 + raise ValueError(f"Unknown Agent Editor action: {action}")
65 + except ValueError as exc:
66 + return Response(status=400, response=str(exc), mimetype="text/plain")
67 +
68 +
69 +def _context(input: dict[str, Any]) -> Any:
70 + context_id = str(input.get("context_id") or "").strip()
71 + if context_id:
72 + context = AgentContext.get(context_id)
73 + if not context:
74 + raise ValueError("Chat context not found.")
75 + return context
76 + project_name = str(input.get("project_name") or "").strip()
77 + if project_name:
78 + project_name = projects.validate_project_name(project_name)
79 + return editor._EditorContext(project_name)
plugins/_agent_editor/api/agent_editor_avatar.py new
+36
@@ -0,0 +1,36 @@
1 +from __future__ import annotations
2 +
3 +from flask import send_file
4 +
5 +from helpers.api import ApiHandler, Request, Response
6 +from plugins._agent_editor.helpers import editor
7 +
8 +
9 +class AgentEditorAvatar(ApiHandler):
10 + @classmethod
11 + def get_methods(cls) -> list[str]:
12 + return ["GET", "POST"]
13 +
14 + async def process(self, input: dict, request: Request) -> dict | Response:
15 + try:
16 + if request.method == "POST":
17 + return {
18 + "ok": True,
19 + **editor.stage_avatar(request.files.get("avatar")),
20 + }
21 +
22 + profile_id = editor.validate_profile_id(request.args.get("profile_id"))
23 + project_name = str(request.args.get("project_name") or "").strip()
24 + if project_name:
25 + project_name = editor.projects.validate_project_name(project_name)
26 + path = editor.effective_avatar_path(
27 + profile_id,
28 + editor._EditorContext(project_name),
29 + )
30 + if not path or not path.is_file():
31 + return Response(status=404, response="Avatar not found.")
32 + response = send_file(path, mimetype="image/webp", conditional=True)
33 + response.headers["Cache-Control"] = "private, no-cache"
34 + return response
35 + except ValueError as exc:
36 + return Response(status=400, response=str(exc), mimetype="text/plain")
plugins/_agent_editor/helpers/__init__.py
plugins/_agent_editor/helpers/editor.py new
+1263
@@ -0,0 +1,1263 @@
1 +from __future__ import annotations
2 +
3 +from dataclasses import dataclass, field
4 +from io import BytesIO
5 +import json
6 +import os
7 +from pathlib import Path
8 +import re
9 +import tempfile
10 +import threading
11 +import time
12 +from types import SimpleNamespace
13 +from typing import Any
14 +from urllib.parse import urlencode
15 +from uuid import uuid4
16 +
17 +from helpers import cache, files, plugins, projects, skills, subagents, tool_policy
18 +from helpers import yaml as yaml_helper
19 +
20 +
21 +PROFILE_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9_-]{0,62}[a-z0-9])?$")
22 +COLOR_PATTERN = re.compile(r"^#[0-9A-Fa-f]{6}$")
23 +SPECIFICS_FILE = "agent.system.main.specifics.md"
24 +METADATA_KEYS = ("title", "description", "context", "avatar")
25 +MAX_PROMPT_BYTES = 512 * 1024
26 +MAX_PROMPTS_BYTES = 2 * 1024 * 1024
27 +MAX_AVATAR_BYTES = 8 * 1024 * 1024
28 +MAX_AVATAR_DIMENSION = 4096
29 +AVATAR_SIZE = 512
30 +RESERVED_PROFILE_IDS = {"_example"}
31 +NON_PROMPT_MARKDOWN = {"AGENTS.md"}
32 +USER_AGENTS_ROOT = Path(files.get_abs_path(subagents.USER_AGENTS_DIR))
33 +STAGED_AVATAR_ROOT = Path(files.get_abs_path("tmp", "agent-editor"))
34 +_POLICY_KEYS = ("mode", "default", "allowed", "blocked")
35 +_MUTATION_LOCK = threading.RLock()
36 +
37 +
38 +class _EditorContext:
39 + def __init__(self, project_name: str = "") -> None:
40 + self.project_name = project_name
41 +
42 + def get_data(self, key: str, recursive: bool = True):
43 + return self.project_name if key == projects.CONTEXT_DATA_KEY_PROJECT else None
44 +
45 +
46 +class EditorAgent:
47 + def __init__(self, profile_id: str, context: Any | None = None) -> None:
48 + self.config = SimpleNamespace(profile=profile_id)
49 + self.context = context or _EditorContext()
50 + self.data: dict[str, Any] = {}
51 +
52 + def get_data(self, key: str):
53 + return self.data.get(key)
54 +
55 + def read_prompt(self, filename: str, **kwargs: Any) -> str:
56 + path = files.find_file_in_dirs(filename, subagents.get_paths(self, "prompts"))
57 + return Path(path).read_text(encoding="utf-8")
58 +
59 +
60 +@dataclass(frozen=True)
61 +class ProfileLayer:
62 + kind: str
63 + metadata_path: Path | None
64 + data: dict[str, Any]
65 +
66 +
67 +@dataclass(frozen=True)
68 +class FileChange:
69 + action: str
70 + path: Path
71 + content: bytes | None = None
72 +
73 + @property
74 + def relative_path(self) -> str:
75 + return files.deabsolute_path(str(self.path)).replace(os.sep, "/")
76 +
77 +
78 +@dataclass
79 +class ChangePlan:
80 + changes: dict[Path, FileChange] = field(default_factory=dict)
81 + warnings: list[str] = field(default_factory=list)
82 + staged_tokens: set[str] = field(default_factory=set)
83 + profile_id: str = ""
84 + remove_empty_root: bool = False
85 +
86 + def write(self, path: Path, content: str | bytes) -> None:
87 + payload = content.encode("utf-8") if isinstance(content, str) else content
88 + if path.is_file() and path.read_bytes() == payload:
89 + self.changes.pop(path, None)
90 + return
91 + self.changes[path] = FileChange("write", path, payload)
92 +
93 + def delete(self, path: Path) -> None:
94 + if path.exists():
95 + self.changes[path] = FileChange("delete", path)
96 + else:
97 + self.changes.pop(path, None)
98 +
99 + def response(self) -> dict[str, Any]:
100 + ordered = sorted(self.changes.values(), key=lambda change: change.relative_path)
101 + return {
102 + "written": [
103 + change.relative_path for change in ordered if change.action == "write"
104 + ],
105 + "deleted": [
106 + change.relative_path for change in ordered if change.action == "delete"
107 + ],
108 + "warnings": list(self.warnings),
109 + }
110 +
111 +
112 +def validate_profile_id(profile_id: Any) -> str:
113 + value = str(profile_id or "").strip()
114 + if value in RESERVED_PROFILE_IDS or not PROFILE_ID_PATTERN.fullmatch(value):
115 + raise ValueError(
116 + "Profile ID must be 1–64 lowercase letters, numbers, hyphens, or underscores."
117 + )
118 + return value
119 +
120 +
121 +def profile_exists(profile_id: str, context: Any | None = None) -> bool:
122 + agent = EditorAgent(profile_id, context)
123 + if (Path(files.get_abs_path(subagents.DEFAULT_AGENTS_DIR)) / profile_id).is_dir():
124 + return True
125 + if (USER_AGENTS_ROOT / profile_id).is_dir():
126 + return True
127 + if any(Path(path).is_dir() for path in plugins.get_plugin_paths("agents", profile_id)):
128 + return True
129 + project_name = projects.get_context_project_name(agent.context) or ""
130 + return bool(
131 + project_name
132 + and Path(projects.get_project_meta(project_name, "agents", profile_id)).is_dir()
133 + )
134 +
135 +
136 +def list_profiles(context: Any | None = None) -> list[dict[str, Any]]:
137 + project_name = projects.get_context_project_name(context) if context else None
138 + resolved = subagents.get_agents_dict(project_name)
139 + names = set(resolved)
140 + for root in (Path(files.get_abs_path("agents")), USER_AGENTS_ROOT):
141 + if root.is_dir():
142 + names.update(path.name for path in root.iterdir() if path.is_dir())
143 +
144 + result: list[dict[str, Any]] = []
145 + for profile_id in sorted(names):
146 + if profile_id in RESERVED_PROFILE_IDS:
147 + continue
148 + state = metadata_state(profile_id, context)
149 + result.append(
150 + {
151 + "id": profile_id,
152 + "title": state["title"]["effective"] or profile_id,
153 + "description": state["description"]["effective"] or "",
154 + "origin": state["origin"],
155 + "origin_chain": state["origin_chain"],
156 + "built_in": state["built_in"],
157 + "has_user_overrides": state["has_user_overrides"],
158 + "avatar": state["avatar"]["effective"],
159 + "avatar_url": effective_avatar_url(profile_id, context),
160 + "project_override_active": state["project_override_active"],
161 + "enabled": bool(getattr(resolved.get(profile_id), "enabled", True)),
162 + "available": profile_exists(profile_id, context),
163 + }
164 + )
165 + return result
166 +
167 +
168 +def build_editor_state(
169 + profile_id: str,
170 + context: Any | None = None,
171 +) -> dict[str, Any]:
172 + profile_id = validate_profile_id(profile_id)
173 + agent = EditorAgent(profile_id, context)
174 + prompt_files = prompt_catalog(agent)
175 + tool_catalog = tool_policy.get_tool_catalog(agent)
176 + from plugins._model_config.helpers import model_config
177 +
178 + presets: list[dict[str, Any]] = []
179 + for preset in model_config.get_presets():
180 + name = str(preset.get("name") or "").strip()
181 + if not name:
182 + continue
183 + resolved = model_config.resolve_config_settings(
184 + {model_config.MODEL_PRESET_CONFIG_KEY: name}
185 + )
186 + presets.append(
187 + {
188 + "name": name,
189 + "main": _model_identity(resolved.get("chat_model")),
190 + "utility": _model_identity(resolved.get("utility_model")),
191 + "embedding": _model_identity(resolved.get("embedding_model")),
192 + }
193 + )
194 + model_path = _profile_config_path(profile_id, "_model_config")
195 + model_user = _read_mapping(model_path)
196 + selected = model_config.get_configured_preset_name(agent=agent)
197 +
198 + tool_path = _profile_config_path(profile_id, tool_policy.PLUGIN_NAME)
199 + skill_path = _profile_config_path(profile_id, skills.ACTIVE_SKILLS_PLUGIN_NAME)
200 + tool_user = _read_mapping(tool_path)
201 + project_name = projects.get_context_project_name(agent.context) or ""
202 + tool_source = (
203 + plugins.find_plugin_asset(
204 + tool_policy.PLUGIN_NAME,
205 + plugins.CONFIG_FILE_NAME,
206 + project_name=project_name,
207 + agent_profile=profile_id,
208 + )
209 + if project_name
210 + else None
211 + )
212 + skill_user = _read_mapping(skill_path)
213 + skill_policy = skills.normalize_visibility_policy(
214 + skill_user.get("visibility_policy")
215 + )
216 + effective_skill_policy = skills.get_visibility_policy(agent)
217 + skill_catalog = [
218 + {**item, "available": True}
219 + for item in skills.list_skill_catalog(agent=agent)
220 + ]
221 + known_skill_ids = {
222 + value.casefold()
223 + for item in skill_catalog
224 + for value in (
225 + str(item.get("name") or ""),
226 + str(item.get("path") or ""),
227 + Path(str(item.get("path") or "")).name,
228 + )
229 + if value
230 + }
231 + for skill_id in [*skill_policy["allowed"], *skill_policy["blocked"]]:
232 + if skill_id.casefold() in known_skill_ids:
233 + continue
234 + known_skill_ids.add(skill_id.casefold())
235 + allowed = skills.is_skill_allowed(effective_skill_policy, skill_id)
236 + skill_catalog.append(
237 + {
238 + "name": skill_id,
239 + "description": "",
240 + "path": skill_id,
241 + "origin": "Unavailable",
242 + "hidden": not allowed,
243 + "tags": [],
244 + "allowed_tools": [],
245 + "available": False,
246 + }
247 + )
248 + skill_catalog.sort(
249 + key=lambda item: (
250 + str(item.get("name") or "").casefold(),
251 + str(item.get("path") or ""),
252 + )
253 + )
254 + return {
255 + "profile": build_profile_state(profile_id, context),
256 + "prompts": prompt_files,
257 + "tools": {
258 + "policy": tool_policy.normalize_policy(tool_user),
259 + "has_override": any(key in tool_user for key in _POLICY_KEYS),
260 + "project_override_active": bool(
261 + tool_source and tool_source.get("project_name")
262 + ),
263 + "catalog": tool_catalog,
264 + },
265 + "skills": {
266 + "policy": skill_policy,
267 + "has_override": "visibility_policy" in skill_user,
268 + "catalog": skill_catalog,
269 + },
270 + "model_presets": presets,
271 + "model_preset": {
272 + "effective": selected,
273 + "override": model_user.get(model_config.MODEL_PRESET_CONFIG_KEY),
274 + "has_override": model_config.MODEL_PRESET_CONFIG_KEY in model_user,
275 + },
276 + }
277 +
278 +
279 +def build_profile_state(
280 + profile_id: str,
281 + context: Any | None = None,
282 +) -> dict[str, Any]:
283 + profile_id = validate_profile_id(profile_id)
284 + metadata = metadata_state(profile_id, context)
285 + return {
286 + "id": profile_id,
287 + "origin": metadata.pop("origin"),
288 + "origin_chain": metadata.pop("origin_chain"),
289 + "built_in": metadata.pop("built_in"),
290 + "has_user_overrides": metadata.pop("has_user_overrides"),
291 + "project_override_active": metadata.pop("project_override_active"),
292 + "metadata": metadata,
293 + "avatar_url": effective_avatar_url(profile_id, context),
294 + }
295 +
296 +
297 +def metadata_state(profile_id: str, context: Any | None = None) -> dict[str, Any]:
298 + layers = _metadata_layers(profile_id, context)
299 + user_layer = next((layer for layer in layers if layer.kind == "user"), None)
300 + lower_layers = [
301 + layer for layer in layers if layer.kind not in {"user", "project"}
302 + ]
303 + state: dict[str, Any] = {}
304 + for key in METADATA_KEYS:
305 + effective, source, source_kind = _layer_value(layers, key)
306 + inherited, inherited_source, _ = _layer_value(lower_layers, key)
307 + state[key] = {
308 + "effective": effective,
309 + "override": (
310 + user_layer.data.get(key)
311 + if user_layer and key in user_layer.data
312 + else None
313 + ),
314 + "has_override": bool(user_layer and key in user_layer.data),
315 + "source": _relative_source(source),
316 + "inherited": inherited,
317 + "inherited_source": _relative_source(inherited_source),
318 + "project_override_active": source_kind == "project",
319 + }
320 +
321 + user_dir = USER_AGENTS_ROOT / profile_id
322 + built_in = (Path(files.get_abs_path("agents")) / profile_id).is_dir()
323 + plugin_origin = any(layer.kind == "plugin" for layer in layers)
324 + state.update(
325 + {
326 + "origin": "Built-in" if built_in else "Plugin" if plugin_origin else "Custom",
327 + "origin_chain": list(dict.fromkeys(layer.kind for layer in layers)),
328 + "built_in": built_in,
329 + "has_user_overrides": user_dir.is_dir() and any(user_dir.iterdir()),
330 + "project_override_active": any(
331 + layer.kind == "project" for layer in layers
332 + ),
333 + }
334 + )
335 + return state
336 +
337 +
338 +def prompt_catalog(agent: EditorAgent) -> list[dict[str, Any]]:
339 + roots = [Path(path) for path in subagents.get_paths(agent, "prompts")]
340 + names = {
341 + path.name
342 + for root in roots
343 + if root.is_dir()
344 + for path in root.glob("*.md")
345 + if path.is_file() and path.name not in NON_PROMPT_MARKDOWN
346 + }
347 + names.add(SPECIFICS_FILE)
348 + user_prompt_dir = USER_AGENTS_ROOT / agent.config.profile / "prompts"
349 + project_name = projects.get_context_project_name(agent.context) or ""
350 + project_root = (
351 + Path(projects.get_project_meta(project_name)) if project_name else None
352 + )
353 + result: list[dict[str, Any]] = []
354 + for name in sorted(names, key=lambda value: (_prompt_group(value)[0], value)):
355 + occurrences: list[tuple[Path, str, str]] = []
356 + for root in roots:
357 + path = root / name
358 + if path.is_file():
359 + kind, label = _prompt_source(root, user_prompt_dir, project_root)
360 + occurrences.append((path, kind, label))
361 +
362 + effective = occurrences[0] if occurrences else None
363 + user = next((item for item in occurrences if item[1] == "user"), None)
364 + inherited = next(
365 + (
366 + item
367 + for item in occurrences
368 + if item[1] not in {"user", "project"}
369 + ),
370 + None,
371 + )
372 + effective_text, effective_error = _prompt_text(effective[0] if effective else None)
373 + user_text, user_error = _prompt_text(user[0] if user else None)
374 + inherited_text, inherited_error = _prompt_text(
375 + inherited[0] if inherited else None
376 + )
377 + group_number, group_label = _prompt_group(name)
378 + source_kind = effective[1] if effective else ""
379 + if effective_error or user_error or inherited_error:
380 + state = "Conflict"
381 + elif source_kind == "project":
382 + state = "Project override active"
383 + elif user:
384 + state = (
385 + "Overridden here (empty)"
386 + if user_text == ""
387 + else "Overridden here"
388 + )
389 + elif source_kind == "plugin":
390 + state = "Plugin-provided"
391 + elif effective:
392 + state = "Inherited"
393 + else:
394 + state = "Unavailable"
395 +
396 + preview = _expand_static_prompt(name, roots, 0, set())
397 +
398 + result.append(
399 + {
400 + "filename": name,
401 + "group": group_number,
402 + "group_label": group_label,
403 + "state": state,
404 + "effective": effective_text,
405 + "override": user_text if user else None,
406 + "has_override": bool(user),
407 + "inherited": inherited_text,
408 + "source": _relative_source(effective[0] if effective else None),
409 + "inherited_source": _relative_source(
410 + inherited[0] if inherited else None
411 + ),
412 + "source_chain": [
413 + label for _, _, label in reversed(occurrences)
414 + ],
415 + "project_override_active": source_kind == "project",
416 + "preview": preview,
417 + "error": effective_error or user_error or inherited_error,
418 + "dynamic_processor": any(
419 + (root / f"{Path(name).stem}.py").is_file() for root in roots
420 + ),
421 + }
422 + )
423 + return result
424 +
425 +
426 +def effective_avatar_path(profile_id: str, context: Any | None = None) -> Path | None:
427 + layers = _metadata_layers(profile_id, context)
428 + value, source, _ = _layer_value(layers, "avatar")
429 + if not isinstance(value, dict) or value.get("kind") != "image" or not source:
430 + return None
431 + relative = str(value.get("value") or "")
432 + candidate = (source.parent / relative).resolve()
433 + if not files.is_in_dir(str(candidate), str(source.parent)) or not candidate.is_file():
434 + return None
435 + return candidate
436 +
437 +
438 +def effective_avatar_url(profile_id: str, context: Any | None = None) -> str:
439 + path = effective_avatar_path(profile_id, context)
440 + if not path:
441 + return ""
442 + query = {
443 + "profile_id": profile_id,
444 + "v": str(path.stat().st_mtime_ns),
445 + }
446 + project_name = projects.get_context_project_name(context) if context else ""
447 + if project_name:
448 + query["project_name"] = project_name
449 + return "/api/plugins/_agent_editor/agent_editor_avatar?" + urlencode(query)
450 +
451 +
452 +def _metadata_layers(profile_id: str, context: Any | None) -> list[ProfileLayer]:
453 + agent = EditorAgent(profile_id, context)
454 + directories: list[tuple[str, Path]] = [
455 + (
456 + "profile",
457 + Path(files.get_abs_path("agents", profile_id)),
458 + )
459 + ]
460 + for directory in plugins.get_enabled_plugin_paths(agent, "agents", profile_id):
461 + directories.append(("plugin", Path(directory)))
462 + directories.append(("user", USER_AGENTS_ROOT / profile_id))
463 +
464 + project_name = projects.get_context_project_name(agent.context) or ""
465 + if project_name:
466 + directories.append(
467 + (
468 + "project",
469 + Path(projects.get_project_meta(project_name, "agents", profile_id)),
470 + )
471 + )
472 +
473 + return [
474 + layer
475 + for kind, directory in directories
476 + if (layer := _load_profile_layer(kind, directory)) is not None
477 + ]
478 +
479 +
480 +def _load_profile_layer(kind: str, directory: Path) -> ProfileLayer | None:
481 + if not directory.is_dir():
482 + return None
483 + yaml_path = directory / "agent.yaml"
484 + json_path = directory / "agent.json"
485 + path = yaml_path if yaml_path.is_file() else json_path if json_path.is_file() else None
486 + data = _read_mapping(path) if path else {}
487 + return ProfileLayer(kind, path, data)
488 +
489 +
490 +def _read_mapping(path: Path | None) -> dict[str, Any]:
491 + if not path or not path.is_file():
492 + return {}
493 + try:
494 + value = (
495 + json.loads(path.read_text(encoding="utf-8"))
496 + if path.suffix.lower() == ".json"
497 + else yaml_helper.loads(path.read_text(encoding="utf-8"))
498 + )
499 + except Exception:
500 + return {}
501 + return dict(value) if isinstance(value, dict) else {}
502 +
503 +
504 +def _layer_value(
505 + layers: list[ProfileLayer], key: str
506 +) -> tuple[Any, Path | None, str]:
507 + for layer in reversed(layers):
508 + if key in layer.data:
509 + return layer.data[key], layer.metadata_path, layer.kind
510 + return None, None, ""
511 +
512 +
513 +def _prompt_source(
514 + root: Path, user_prompt_dir: Path, project_root: Path | None
515 +) -> tuple[str, str]:
516 + if root.resolve() == user_prompt_dir.resolve():
517 + return "user", "Your override"
518 + if project_root and files.is_in_dir(str(root), str(project_root)):
519 + return "project", f"Project · {project_root.parent.name}"
520 + plugin = plugins.get_plugin_name_from_path(root)
521 + if plugin:
522 + return "plugin", f"Plugin · {plugin}"
523 + bundled_agents = Path(files.get_abs_path("agents"))
524 + if files.is_in_dir(str(root), str(bundled_agents)):
525 + return "profile", root.parent.name.replace("-", " ").title()
526 + if root.resolve() == Path(files.get_abs_path("prompts")).resolve():
527 + return "framework", "Framework"
528 + return "global", "Global"
529 +
530 +
531 +def _prompt_group(filename: str) -> tuple[str, str]:
532 + if filename == SPECIFICS_FILE:
533 + return "2.1", "Agent instructions"
534 + if filename == "agent.system.main.role.md":
535 + return "2.2", "Role"
536 + if filename == "agent.system.main.environment.md":
537 + return "2.3", "Environment"
538 + if filename.startswith("agent.system.main.communication"):
539 + return "2.4", "Communication"
540 + if filename == "agent.system.main.solving.md":
541 + return "2.5", "Problem solving"
542 + if filename == "agent.system.main.tips.md":
543 + return "2.6", "Tips"
544 + if filename.startswith("fw."):
545 + return "2.7", "Framework messages"
546 + if filename.startswith("agent.system.tool.") or "tool" in filename:
547 + return "2.8", "Tool instructions"
548 + if filename.startswith(("agent.context.", "agent.system.projects.", "agent.system.skills")):
549 + return "2.9", "Context, projects & skills"
550 + return "2.10", "Other"
551 +
552 +
553 +def _model_identity(value: Any) -> dict[str, str]:
554 + item = value if isinstance(value, dict) else {}
555 + return {
556 + "provider": str(item.get("provider") or ""),
557 + "name": str(item.get("name") or ""),
558 + }
559 +
560 +
561 +def _profile_config_path(profile_id: str, plugin_name: str) -> Path:
562 + return Path(
563 + plugins.determine_plugin_asset_path(
564 + plugin_name,
565 + "",
566 + profile_id,
567 + plugins.CONFIG_FILE_NAME,
568 + )
569 + )
570 +
571 +
572 +def _relative_source(path: Path | None) -> str:
573 + return files.deabsolute_path(str(path)).replace(os.sep, "/") if path else ""
574 +
575 +
576 +def _prompt_text(path: Path | None) -> tuple[str, str]:
577 + if not path:
578 + return "", ""
579 + try:
580 + return path.read_text(encoding="utf-8"), ""
581 + except (OSError, UnicodeError) as exc:
582 + return "", f"Prompt is not readable UTF-8: {exc}"
583 +
584 +
585 +_NAMED_INCLUDE = re.compile(r"{{\s*include\s*['\"](.*?)['\"]\s*}}")
586 +_ORIGINAL_INCLUDE = re.compile(r"{{\s*include\s+original\s*}}")
587 +
588 +
589 +def _expand_static_prompt(
590 + filename: str,
591 + roots: list[Path],
592 + start_index: int,
593 + seen: set[Path],
594 +) -> str:
595 + relative = Path(filename)
596 + if relative.is_absolute() or ".." in relative.parts:
597 + return ""
598 + match = next(
599 + (
600 + (index, root / relative)
601 + for index, root in enumerate(roots[start_index:], start=start_index)
602 + if (root / relative).is_file()
603 + ),
604 + None,
605 + )
606 + if not match:
607 + return ""
608 + index, path = match
609 + resolved = path.resolve()
610 + if resolved in seen:
611 + return "{{ include cycle }}"
612 + text, error = _prompt_text(path)
613 + if error:
614 + return ""
615 + branch_seen = {*seen, resolved}
616 + text = _ORIGINAL_INCLUDE.sub(
617 + lambda _match: _expand_static_prompt(
618 + filename,
619 + roots,
620 + index + 1,
621 + branch_seen,
622 + ),
623 + text,
624 + )
625 + return _NAMED_INCLUDE.sub(
626 + lambda include: _expand_static_prompt(
627 + include.group(1),
628 + roots,
629 + 0,
630 + branch_seen,
631 + ) or include.group(0),
632 + text,
633 + )
634 +def build_change_plan(
635 + patch: dict[str, Any],
636 + context: Any | None = None,
637 +) -> ChangePlan:
638 + if not isinstance(patch, dict):
639 + raise ValueError("The save patch must be an object.")
640 +
641 + profile_id = validate_profile_id(patch.get("profile_id"))
642 + creating = bool(patch.get("creating"))
643 + easy = str(patch.get("editor_mode") or "advanced").lower() == "easy"
644 + exists = profile_exists(profile_id, context)
645 + if creating and exists:
646 + raise ValueError(
647 + f"An agent with the profile ID `{profile_id}` already exists. "
648 + "Open it to edit its user overrides or choose another name."
649 + )
650 + if not creating and not exists:
651 + raise ValueError(f'Agent profile "{profile_id}" does not exist.')
652 +
653 + plan = ChangePlan(profile_id=profile_id)
654 + if "metadata" in patch:
655 + _plan_metadata(plan, patch["metadata"], context, creating=creating)
656 + if "prompts" in patch:
657 + _plan_prompts(
658 + plan,
659 + patch["prompts"],
660 + context,
661 + creating=creating,
662 + easy=easy,
663 + )
664 + if "model_preset" in patch:
665 + _plan_model_preset(plan, patch["model_preset"])
666 + if "tool_policy" in patch:
667 + _plan_tool_policy(plan, patch["tool_policy"])
668 + if "skill_policy" in patch:
669 + _plan_skill_policy(plan, patch["skill_policy"])
670 +
671 + if creating:
672 + metadata = _planned_mapping(plan, USER_AGENTS_ROOT / profile_id / "agent.yaml")
673 + if not str(metadata.get("title") or "").strip():
674 + raise ValueError("Agent name is required.")
675 + specifics = USER_AGENTS_ROOT / profile_id / "prompts" / SPECIFICS_FILE
676 + change = plan.changes.get(specifics)
677 + if not change or change.action != "write" or not (change.content or b"").strip():
678 + raise ValueError("Instructions are required for a new agent.")
679 +
680 + return plan
681 +
682 +
683 +def _plan_metadata(
684 + plan: ChangePlan,
685 + value: Any,
686 + context: Any | None,
687 + *,
688 + creating: bool,
689 +) -> None:
690 + section = _mapping(value, "metadata")
691 + set_values = _mapping(section.get("set", {}), "metadata.set")
692 + reset_values = _string_list(section.get("reset", []), "metadata.reset")
693 + unknown = (set(set_values) | set(reset_values)) - set(METADATA_KEYS)
694 + if unknown:
695 + raise ValueError(f"Unknown metadata field: {sorted(unknown)[0]}")
696 + if set(set_values).intersection(reset_values):
697 + raise ValueError("A metadata field cannot be set and reset in one save.")
698 +
699 + profile_id = plan.profile_id
700 + yaml_path = USER_AGENTS_ROOT / profile_id / "agent.yaml"
701 + legacy_path = USER_AGENTS_ROOT / profile_id / "agent.json"
702 + data = _read_mapping_strict(
703 + yaml_path if yaml_path.is_file() else legacy_path,
704 + "profile metadata",
705 + )
706 + state = metadata_state(profile_id, context)
707 + current_user_avatar = data.get("avatar")
708 +
709 + for key in reset_values:
710 + data.pop(key, None)
711 + if key == "avatar" and _editor_image_avatar(current_user_avatar):
712 + plan.delete(USER_AGENTS_ROOT / profile_id / "assets" / "avatar.webp")
713 +
714 + for key, raw in set_values.items():
715 + if key in {"title", "description", "context"}:
716 + if not isinstance(raw, str):
717 + raise ValueError(f"Metadata field {key} must be text.")
718 + normalized: Any = raw
719 + if key == "title" and not raw.strip():
720 + raise ValueError("Agent name is required.")
721 + else:
722 + normalized = _normalize_avatar(plan, raw)
723 + if normalized.get("kind") == "color" and _editor_image_avatar(
724 + current_user_avatar
725 + ):
726 + plan.delete(USER_AGENTS_ROOT / profile_id / "assets" / "avatar.webp")
727 +
728 + inherited = state[key]["inherited"]
729 + if not creating and normalized == inherited:
730 + data.pop(key, None)
731 + else:
732 + data[key] = normalized
733 +
734 + if data:
735 + plan.write(yaml_path, yaml_helper.dumps(data))
736 + else:
737 + plan.delete(yaml_path)
738 +
739 +
740 +def _plan_prompts(
741 + plan: ChangePlan,
742 + value: Any,
743 + context: Any | None,
744 + *,
745 + creating: bool,
746 + easy: bool,
747 +) -> None:
748 + section = _mapping(value, "prompts")
749 + set_values = _mapping(section.get("set", {}), "prompts.set")
750 + reset_values = _string_list(section.get("reset", []), "prompts.reset")
751 + if set(set_values).intersection(reset_values):
752 + raise ValueError("A prompt cannot be set and reset in one save.")
753 +
754 + agent = EditorAgent(plan.profile_id, context)
755 + catalog = {item["filename"]: item for item in prompt_catalog(agent)}
756 + for filename in [*set_values, *reset_values]:
757 + if filename not in catalog:
758 + raise ValueError(f'Prompt file "{filename}" is not in the editor catalog.')
759 +
760 + total = 0
761 + for filename, content in set_values.items():
762 + if not isinstance(content, str):
763 + raise ValueError(f'Prompt file "{filename}" must contain UTF-8 text.')
764 + if "\x00" in content:
765 + raise ValueError(f'Prompt file "{filename}" contains a NUL character.')
766 + payload_size = len(content.encode("utf-8"))
767 + if payload_size > MAX_PROMPT_BYTES:
768 + raise ValueError(f'Prompt file "{filename}" is too large.')
769 + total += payload_size
770 + if easy and filename == SPECIFICS_FILE and not content.strip():
771 + raise ValueError(
772 + "Instructions can’t be empty. To remove your changes, use "
773 + "Restore original instructions."
774 + )
775 + item = catalog[filename]
776 + path = USER_AGENTS_ROOT / plan.profile_id / "prompts" / filename
777 + if (
778 + not creating
779 + and item.get("inherited_source")
780 + and content == item.get("inherited", "")
781 + ):
782 + plan.delete(path)
783 + else:
784 + plan.write(path, content)
785 +
786 + if total > MAX_PROMPTS_BYTES:
787 + raise ValueError("The combined prompt changes are too large.")
788 + for filename in reset_values:
789 + plan.delete(USER_AGENTS_ROOT / plan.profile_id / "prompts" / filename)
790 +
791 +
792 +def _plan_model_preset(plan: ChangePlan, value: Any) -> None:
793 + section = _mapping(value, "model_preset")
794 + mode = str(section.get("mode") or "inherit").strip().lower()
795 + path = _profile_config_path(plan.profile_id, "_model_config")
796 + data = _read_mapping_strict(path, "model preset configuration")
797 +
798 + from plugins._model_config.helpers import model_config
799 +
800 + if mode == "inherit":
801 + data.pop(model_config.MODEL_PRESET_CONFIG_KEY, None)
802 + elif mode == "preset":
803 + requested = str(section.get("name") or "").strip()
804 + preset = model_config.resolve_preset(requested)
805 + if not preset:
806 + raise ValueError(f'Model preset "{requested}" does not exist.')
807 + data[model_config.MODEL_PRESET_CONFIG_KEY] = str(preset.get("name") or requested)
808 + else:
809 + raise ValueError("Model preset mode must be inherit or preset.")
810 + _plan_json_mapping(plan, path, data)
811 +
812 +
813 +def _plan_tool_policy(plan: ChangePlan, value: Any) -> None:
814 + section = _mapping(value, "tool_policy")
815 + mode = str(section.get("mode") or "inherit").strip().lower()
816 + path = _profile_config_path(plan.profile_id, tool_policy.PLUGIN_NAME)
817 + data = _read_mapping_strict(path, "tool policy configuration")
818 +
819 + if mode == "inherit":
820 + for key in _POLICY_KEYS:
821 + data.pop(key, None)
822 + else:
823 + if mode == "off":
824 + policy = {"mode": "custom", "default": "block", "allowed": [], "blocked": []}
825 + elif mode == "custom":
826 + policy = tool_policy.normalize_policy(section)
827 + allowed = set(policy["allowed"])
828 + blocked = set(policy["blocked"])
829 + if allowed.intersection(blocked):
830 + raise ValueError("A tool cannot be both allowed and blocked.")
831 + for tool_id in [*policy["allowed"], *policy["blocked"]]:
832 + if not _valid_tool_id(tool_id):
833 + raise ValueError(f'Invalid canonical tool ID "{tool_id}".')
834 + else:
835 + raise ValueError("Tool policy mode must be inherit, off, or custom.")
836 + data.update({key: policy[key] for key in _POLICY_KEYS})
837 + _plan_json_mapping(plan, path, data)
838 +
839 +
840 +def _plan_skill_policy(plan: ChangePlan, value: Any) -> None:
841 + section = _mapping(value, "skill_policy")
842 + mode = str(section.get("mode") or "inherit").strip().lower()
843 + path = _profile_config_path(plan.profile_id, skills.ACTIVE_SKILLS_PLUGIN_NAME)
844 + data = _read_mapping_strict(path, "skill policy configuration")
845 +
846 + if mode == "inherit":
847 + data.pop("visibility_policy", None)
848 + else:
849 + raw = (
850 + {"mode": "custom", "default": "block", "allowed": [], "blocked": []}
851 + if mode == "off"
852 + else section
853 + )
854 + if mode not in {"off", "custom"}:
855 + raise ValueError("Skill policy mode must be inherit, off, or custom.")
856 + policy = skills.normalize_visibility_policy(raw)
857 + if set(policy["allowed"]).intersection(policy["blocked"]):
858 + raise ValueError("A skill cannot be both allowed and blocked.")
859 + for skill_id in [*policy["allowed"], *policy["blocked"]]:
860 + if not skill_id or len(skill_id) > 512 or "\x00" in skill_id:
861 + raise ValueError("Invalid skill ID.")
862 + data["visibility_policy"] = policy
863 + _plan_json_mapping(plan, path, data)
864 +
865 +
866 +def _normalize_avatar(plan: ChangePlan, value: Any) -> dict[str, str]:
867 + avatar = _mapping(value, "metadata.set.avatar")
868 + kind = str(avatar.get("kind") or "").strip().lower()
869 + if kind == "color":
870 + color = str(avatar.get("value") or "").strip().upper()
871 + if not COLOR_PATTERN.fullmatch(color):
872 + raise ValueError("Avatar color must be a six-digit hex color.")
873 + return {"kind": "color", "value": color}
874 + if kind == "image":
875 + token = str(avatar.get("token") or "").strip()
876 + if token:
877 + source = staged_avatar_path(token)
878 + if not source.is_file():
879 + raise ValueError("The staged avatar has expired. Upload it again.")
880 + plan.write(
881 + USER_AGENTS_ROOT / plan.profile_id / "assets" / "avatar.webp",
882 + source.read_bytes(),
883 + )
884 + plan.staged_tokens.add(token)
885 + return {"kind": "image", "value": "assets/avatar.webp"}
886 + raise ValueError("Avatar must be a color or a staged image.")
887 +
888 +
889 +def _plan_json_mapping(plan: ChangePlan, path: Path, data: dict[str, Any]) -> None:
890 + if data:
891 + plan.write(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n")
892 + else:
893 + plan.delete(path)
894 +
895 +
896 +def _planned_mapping(plan: ChangePlan, path: Path) -> dict[str, Any]:
897 + change = plan.changes.get(path)
898 + if change and change.action == "delete":
899 + return {}
900 + if change and change.content is not None:
901 + value = yaml_helper.loads(change.content.decode("utf-8"))
902 + return dict(value) if isinstance(value, dict) else {}
903 + return _read_mapping(path)
904 +
905 +
906 +def _read_mapping_strict(path: Path, label: str) -> dict[str, Any]:
907 + if not path.is_file():
908 + return {}
909 + try:
910 + value = (
911 + json.loads(path.read_text(encoding="utf-8"))
912 + if path.suffix.lower() == ".json"
913 + else yaml_helper.loads(path.read_text(encoding="utf-8"))
914 + )
915 + except Exception as exc:
916 + raise ValueError(f"Existing {label} is invalid and was not changed: {exc}") from exc
917 + if not isinstance(value, dict):
918 + raise ValueError(f"Existing {label} must be an object and was not changed.")
919 + return dict(value)
920 +
921 +
922 +def _mapping(value: Any, label: str) -> dict[str, Any]:
923 + if not isinstance(value, dict):
924 + raise ValueError(f"{label} must be an object.")
925 + return dict(value)
926 +
927 +
928 +def _string_list(value: Any, label: str) -> list[str]:
929 + if not isinstance(value, list) or any(not isinstance(item, str) for item in value):
930 + raise ValueError(f"{label} must be an array of strings.")
931 + return list(dict.fromkeys(value))
932 +
933 +
934 +def _valid_tool_id(value: str) -> bool:
935 + return bool(
936 + re.fullmatch(
937 + r"(?:local:[^\s/:]+|(?:mcp|plugin):[^\s/:]+:[^\s/:]+)",
938 + value,
939 + )
940 + )
941 +
942 +
943 +def _editor_image_avatar(value: Any) -> bool:
944 + return (
945 + isinstance(value, dict)
946 + and value.get("kind") == "image"
947 + and value.get("value") == "assets/avatar.webp"
948 + )
949 +
950 +
951 +def apply_change_plan(plan: ChangePlan) -> dict[str, Any]:
952 + root = USER_AGENTS_ROOT / validate_profile_id(plan.profile_id)
953 + changes = sorted(plan.changes.values(), key=lambda item: str(item.path))
954 + _validate_plan_paths(root, changes)
955 + receipt = plan.response()
956 +
957 + with _MUTATION_LOCK:
958 + snapshots = {
959 + change.path: change.path.read_bytes() if change.path.is_file() else None
960 + for change in changes
961 + }
962 + staged: dict[Path, Path] = {}
963 + created_dirs: set[Path] = set()
964 + try:
965 + for change in changes:
966 + if change.action != "write":
967 + continue
968 + _ensure_parent(change.path.parent, created_dirs)
969 + staged[change.path] = _stage_bytes(
970 + change.path.parent,
971 + change.path.name,
972 + change.content or b"",
973 + )
974 +
975 + for change in changes:
976 + if change.action == "write":
977 + os.replace(staged.pop(change.path), change.path)
978 + _fsync_directory(change.path.parent)
979 + elif change.path.is_file() or change.path.is_symlink():
980 + change.path.unlink()
981 + _fsync_directory(change.path.parent)
982 +
983 + if plan.remove_empty_root:
984 + _prune_empty_directories(root)
985 + if root.is_dir() and not any(root.iterdir()):
986 + root.rmdir()
987 + except Exception:
988 + for temporary in staged.values():
989 + temporary.unlink(missing_ok=True)
990 + _restore_snapshots(snapshots, root)
991 + for directory in sorted(created_dirs, key=lambda path: len(path.parts), reverse=True):
992 + try:
993 + directory.rmdir()
994 + except OSError:
995 + pass
996 + raise
997 +
998 + for token in plan.staged_tokens:
999 + staged_avatar_path(token).unlink(missing_ok=True)
1000 + _invalidate_profile_caches()
1001 + return receipt
1002 +
1003 +
1004 +def plan_remove_changes(
1005 + profile_id: str,
1006 + context: Any | None = None,
1007 + *,
1008 + destructive: bool = False,
1009 +) -> ChangePlan:
1010 + profile_id = validate_profile_id(profile_id)
1011 + if not profile_exists(profile_id, context):
1012 + raise ValueError(f'Agent profile "{profile_id}" does not exist.')
1013 + if destructive:
1014 + return _full_delete_plan(profile_id)
1015 +
1016 + catalog = prompt_catalog(EditorAgent(profile_id, context))
1017 + prompt_resets = [
1018 + item["filename"]
1019 + for item in catalog
1020 + if item.get("has_override") and item.get("inherited_source")
1021 + ]
1022 + plan = build_change_plan(
1023 + {
1024 + "profile_id": profile_id,
1025 + "metadata": {"set": {}, "reset": list(METADATA_KEYS)},
1026 + "prompts": {"set": {}, "reset": prompt_resets},
1027 + "model_preset": {"mode": "inherit"},
1028 + "tool_policy": {"mode": "inherit"},
1029 + "skill_policy": {"mode": "inherit"},
1030 + },
1031 + context,
1032 + )
1033 + return plan
1034 +
1035 +
1036 +def plan_delete_custom(profile_id: str, context: Any | None = None) -> ChangePlan:
1037 + profile_id = validate_profile_id(profile_id)
1038 + state = metadata_state(profile_id, context)
1039 + if state["origin"] != "Custom":
1040 + raise ValueError("Built-in and plugin-provided agents cannot be deleted.")
1041 + return _full_delete_plan(profile_id)
1042 +
1043 +
1044 +def delete_impact(profile_id: str, context: Any | None = None) -> dict[str, Any]:
1045 + profile_id = validate_profile_id(profile_id)
1046 + state = metadata_state(profile_id, context)
1047 + root = USER_AGENTS_ROOT / profile_id
1048 + file_paths = [
1049 + _relative_source(path)
1050 + for path in sorted(root.rglob("*"))
1051 + if path.is_file() or path.is_symlink()
1052 + ] if root.is_dir() else []
1053 + references = _profile_reference_paths(profile_id)
1054 + sessions: list[str] = []
1055 + try:
1056 + from agent import AgentContext
1057 +
1058 + sessions = [
1059 + str(item.id)
1060 + for item in AgentContext.all()
1061 + if str(getattr(getattr(item.agent0, "config", None), "profile", ""))
1062 + == profile_id
1063 + ]
1064 + except Exception:
1065 + pass
1066 +
1067 + model_config = _read_mapping(_profile_config_path(profile_id, "_model_config"))
1068 + return {
1069 + "profile_id": profile_id,
1070 + "deletable": state["origin"] == "Custom",
1071 + "origin": state["origin"],
1072 + "files": file_paths,
1073 + "project_references": references,
1074 + "active_sessions": sessions,
1075 + "model_preset": str(model_config.get("model_preset") or ""),
1076 + "contains": {
1077 + name: (root / name).is_dir()
1078 + for name in ("tools", "extensions", "skills", "assets", "plugins")
1079 + },
1080 + }
1081 +
1082 +
1083 +def stage_avatar(upload: Any) -> dict[str, Any]:
1084 + if upload is None:
1085 + raise ValueError("Choose an image to upload.")
1086 + payload = upload.stream.read(MAX_AVATAR_BYTES + 1)
1087 + if len(payload) > MAX_AVATAR_BYTES:
1088 + raise ValueError("Avatar images must be 8 MB or smaller.")
1089 + if not payload:
1090 + raise ValueError("The uploaded image is empty.")
1091 +
1092 + from PIL import Image, ImageOps, UnidentifiedImageError
1093 +
1094 + try:
1095 + with Image.open(BytesIO(payload)) as source:
1096 + source_format = str(source.format or "").upper()
1097 + if source_format not in {"PNG", "JPEG", "WEBP"}:
1098 + raise ValueError("Avatar must be a PNG, JPEG, or WebP image.")
1099 + if max(source.size) > MAX_AVATAR_DIMENSION:
1100 + raise ValueError("Avatar dimensions must not exceed 4096 pixels.")
1101 + source.load()
1102 + normalized = ImageOps.exif_transpose(source)
1103 + mode = "RGBA" if "A" in normalized.getbands() else "RGB"
1104 + square = ImageOps.fit(
1105 + normalized.convert(mode),
1106 + (AVATAR_SIZE, AVATAR_SIZE),
1107 + method=Image.Resampling.LANCZOS,
1108 + )
1109 + output = BytesIO()
1110 + square.save(output, format="WEBP", quality=88, method=6)
1111 + except UnidentifiedImageError as exc:
1112 + raise ValueError("Avatar must be a valid PNG, JPEG, or WebP image.") from exc
1113 + except Image.DecompressionBombError as exc:
1114 + raise ValueError("Avatar dimensions are too large.") from exc
1115 +
1116 + _cleanup_staged_avatars()
1117 + STAGED_AVATAR_ROOT.mkdir(parents=True, exist_ok=True)
1118 + token = uuid4().hex
1119 + path = staged_avatar_path(token)
1120 + temporary = _stage_bytes(path.parent, path.name, output.getvalue())
1121 + os.replace(temporary, path)
1122 + return {"token": token}
1123 +
1124 +
1125 +def staged_avatar_path(token: str) -> Path:
1126 + if not re.fullmatch(r"[0-9a-f]{32}", str(token or "")):
1127 + raise ValueError("Invalid staged avatar token.")
1128 + return STAGED_AVATAR_ROOT / f"{token}.webp"
1129 +
1130 +
1131 +def _full_delete_plan(profile_id: str) -> ChangePlan:
1132 + root = USER_AGENTS_ROOT / profile_id
1133 + plan = ChangePlan(profile_id=profile_id, remove_empty_root=True)
1134 + if not root.is_dir():
1135 + return plan
1136 + for path in sorted(root.rglob("*")):
1137 + if path.is_symlink():
1138 + raise ValueError("Profile symlinks must be removed manually before deletion.")
1139 + if path.is_file():
1140 + plan.delete(path)
1141 + return plan
1142 +
1143 +
1144 +def _validate_plan_paths(root: Path, changes: list[FileChange]) -> None:
1145 + intended_root = root.absolute()
1146 + if root.is_symlink():
1147 + raise ValueError("The profile directory cannot be a symlink.")
1148 + for change in changes:
1149 + if change.action not in {"write", "delete"}:
1150 + raise ValueError("Invalid change-plan action.")
1151 + resolved = change.path.resolve(strict=False)
1152 + if not files.is_in_dir(str(resolved), str(intended_root)):
1153 + raise ValueError("A planned path is outside the user profile directory.")
1154 + cursor = change.path.parent
1155 + while files.is_in_dir(str(cursor), str(intended_root)):
1156 + if cursor.is_symlink():
1157 + raise ValueError("Profile paths cannot traverse symlinks.")
1158 + if cursor == root:
1159 + break
1160 + cursor = cursor.parent
1161 + if change.action == "delete" and change.path.exists() and not (
1162 + change.path.is_file() or change.path.is_symlink()
1163 + ):
1164 + raise ValueError("Change plans delete files, not directories.")
1165 +
1166 +
1167 +def _ensure_parent(parent: Path, created: set[Path]) -> None:
1168 + missing: list[Path] = []
1169 + cursor = parent
1170 + while not cursor.exists():
1171 + missing.append(cursor)
1172 + cursor = cursor.parent
1173 + for directory in reversed(missing):
1174 + directory.mkdir(exist_ok=True)
1175 + created.add(directory)
1176 +
1177 +
1178 +def _stage_bytes(directory: Path, name: str, content: bytes) -> Path:
1179 + target = directory / name
1180 + descriptor, temporary = tempfile.mkstemp(
1181 + prefix=f".{name}.", suffix=".tmp", dir=directory
1182 + )
1183 + path = Path(temporary)
1184 + try:
1185 + os.fchmod(
1186 + descriptor,
1187 + target.stat().st_mode & 0o777 if target.exists() else 0o644,
1188 + )
1189 + with os.fdopen(descriptor, "wb") as handle:
1190 + handle.write(content)
1191 + handle.flush()
1192 + os.fsync(handle.fileno())
1193 + return path
1194 + except Exception:
1195 + path.unlink(missing_ok=True)
1196 + raise
1197 +
1198 +
1199 +def _restore_snapshots(snapshots: dict[Path, bytes | None], root: Path) -> None:
1200 + for path, content in snapshots.items():
1201 + if content is None:
1202 + if path.is_file() or path.is_symlink():
1203 + path.unlink()
1204 + continue
1205 + path.parent.mkdir(parents=True, exist_ok=True)
1206 + temporary = _stage_bytes(path.parent, path.name, content)
1207 + os.replace(temporary, path)
1208 +
1209 +
1210 +def _prune_empty_directories(root: Path) -> None:
1211 + if not root.is_dir():
1212 + return
1213 + for directory in sorted(
1214 + (path for path in root.rglob("*") if path.is_dir()),
1215 + key=lambda path: len(path.parts),
1216 + reverse=True,
1217 + ):
1218 + try:
1219 + directory.rmdir()
1220 + except OSError:
1221 + pass
1222 +
1223 +
1224 +def _fsync_directory(path: Path) -> None:
1225 + descriptor = os.open(path, os.O_RDONLY)
1226 + try:
1227 + os.fsync(descriptor)
1228 + finally:
1229 + os.close(descriptor)
1230 +
1231 +
1232 +def _invalidate_profile_caches() -> None:
1233 + cache.clear(subagents.PATHS_CACHE_AREA)
1234 + plugins.clear_plugin_cache(
1235 + ["_agent_editor", "_model_config", tool_policy.PLUGIN_NAME, skills.ACTIVE_SKILLS_PLUGIN_NAME]
1236 + )
1237 +
1238 +
1239 +def _cleanup_staged_avatars() -> None:
1240 + if not STAGED_AVATAR_ROOT.is_dir():
1241 + return
1242 + cutoff = time.time() - 3600
1243 + for path in STAGED_AVATAR_ROOT.glob("*.webp"):
1244 + try:
1245 + if path.stat().st_mtime < cutoff:
1246 + path.unlink()
1247 + except OSError:
1248 + pass
1249 +
1250 +
1251 +def _profile_reference_paths(profile_id: str) -> list[str]:
1252 + roots = Path(files.get_abs_path("usr", "projects"))
1253 + if not roots.is_dir():
1254 + return []
1255 + references: list[str] = []
1256 + for suffix in ("*.json", "*.yaml", "*.yml"):
1257 + for path in roots.glob(f"*/.a0proj/**/{suffix}"):
1258 + try:
1259 + if profile_id in path.read_text(encoding="utf-8"):
1260 + references.append(_relative_source(path))
1261 + except (OSError, UnicodeError):
1262 + pass
1263 + return sorted(set(references))
plugins/_agent_editor/plugin.yaml new
+5
@@ -0,0 +1,5 @@
1 +name: _agent_editor
2 +title: Agent Editor
3 +description: Deterministic sparse editor for Agent Zero profiles.
4 +version: 1.0.0
5 +always_enabled: true
tests/test_agent_editor.py new
+750
@@ -0,0 +1,750 @@
1 +from __future__ import annotations
2 +
3 +from io import BytesIO
4 +import json
5 +from pathlib import Path
6 +import stat
7 +
8 +import pytest
9 +from werkzeug.datastructures import FileStorage
10 +
11 +from helpers import yaml as yaml_helper
12 +from plugins._agent_editor.api.agent_editor import AgentEditor
13 +from plugins._agent_editor.api.agent_editor_avatar import AgentEditorAvatar
14 +from plugins._agent_editor.helpers import editor
15 +
16 +
17 +@pytest.fixture
18 +def user_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
19 + root = tmp_path / "usr" / "agents"
20 + real_determine_path = editor.plugins.determine_plugin_asset_path
21 +
22 + def determine_plugin_asset_path(
23 + plugin_name: str,
24 + project_name: str,
25 + profile_id: str,
26 + *parts: str,
27 + ) -> str:
28 + if profile_id and not project_name:
29 + return str(
30 + root
31 + / profile_id
32 + / editor.files.PLUGINS_DIR
33 + / plugin_name
34 + / Path(*parts)
35 + )
36 + return real_determine_path(plugin_name, project_name, profile_id, *parts)
37 +
38 + monkeypatch.setattr(editor, "USER_AGENTS_ROOT", root)
39 + monkeypatch.setattr(editor, "STAGED_AVATAR_ROOT", tmp_path / "staged")
40 + monkeypatch.setattr(
41 + editor.plugins,
42 + "determine_plugin_asset_path",
43 + determine_plugin_asset_path,
44 + )
45 + monkeypatch.setattr(editor.plugins, "clear_plugin_cache", lambda _names: None)
46 + return root
47 +
48 +
49 +def _write_manual_files(root: Path) -> dict[Path, bytes]:
50 + manual_files = {
51 + root / "prompts" / "manual.md": b"prompt",
52 + root / "tools" / "manual.py": b"tool",
53 + root / "extensions" / "manual.py": b"extension",
54 + root / "skills" / "manual" / "SKILL.md": b"skill",
55 + root / "assets" / "manual.bin": b"asset",
56 + root / "plugins" / "manual" / "config.json": b"{}",
57 + root / "unknown.bin": b"unknown",
58 + }
59 + for path, payload in manual_files.items():
60 + path.parent.mkdir(parents=True, exist_ok=True)
61 + path.write_bytes(payload)
62 + return manual_files
63 +
64 +
65 +def test_new_easy_profile_writes_only_minimum_exact_files(user_root: Path) -> None:
66 + instructions = "Keep this exact. \n\nNo rewrite."
67 + plan = editor.build_change_plan(
68 + {
69 + "profile_id": "legal-research",
70 + "creating": True,
71 + "editor_mode": "easy",
72 + "metadata": {"set": {"title": "Legal Research"}, "reset": []},
73 + "prompts": {
74 + "set": {editor.SPECIFICS_FILE: instructions},
75 + "reset": [],
76 + },
77 + "tool_policy": {"mode": "inherit"},
78 + }
79 + )
80 +
81 + assert {path.relative_to(user_root).as_posix() for path in plan.changes} == {
82 + "legal-research/agent.yaml",
83 + f"legal-research/prompts/{editor.SPECIFICS_FILE}",
84 + }
85 + editor.apply_change_plan(plan)
86 +
87 + profile = user_root / "legal-research"
88 + assert yaml_helper.loads((profile / "agent.yaml").read_text()) == {
89 + "title": "Legal Research"
90 + }
91 + assert (profile / "prompts" / editor.SPECIFICS_FILE).read_text() == instructions
92 + assert not list(profile.rglob("*.json"))
93 + assert stat.S_IMODE((profile / "agent.yaml").stat().st_mode) == 0o644
94 + assert (
95 + stat.S_IMODE((profile / "prompts" / editor.SPECIFICS_FILE).stat().st_mode)
96 + == 0o644
97 + )
98 +
99 +
100 +def test_editor_lifecycle_needs_no_model_or_utility_configuration(
101 + user_root: Path,
102 + monkeypatch: pytest.MonkeyPatch,
103 +) -> None:
104 + import litellm
105 + from plugins._model_config.helpers import model_config
106 +
107 + def forbidden(*_args, **_kwargs):
108 + raise AssertionError("the Agent Editor attempted a model request")
109 +
110 + for name in ("completion", "acompletion", "embedding", "aembedding"):
111 + monkeypatch.setattr(litellm, name, forbidden, raising=False)
112 + monkeypatch.setattr(model_config, "get_presets", lambda: [{"name": "No utility"}])
113 + monkeypatch.setattr(
114 + model_config,
115 + "resolve_config_settings",
116 + lambda _settings: {
117 + "chat_model": {"provider": "offline", "name": "main"},
118 + "utility_model": {},
119 + "embedding_model": {},
120 + },
121 + )
122 + monkeypatch.setattr(
123 + model_config,
124 + "get_configured_preset_name",
125 + lambda **_kwargs: "No utility",
126 + )
127 + monkeypatch.setattr(editor.tool_policy, "get_tool_catalog", lambda _agent: [])
128 + monkeypatch.setattr(editor.skills, "list_skill_catalog", lambda agent=None: [])
129 +
130 + state = editor.build_editor_state("offline-editor")
131 + plan = editor.build_change_plan(
132 + {
133 + "profile_id": "offline-editor",
134 + "creating": True,
135 + "editor_mode": "easy",
136 + "metadata": {"set": {"title": "Offline Editor"}, "reset": []},
137 + "prompts": {"set": {editor.SPECIFICS_FILE: "Exact text."}, "reset": []},
138 + }
139 + )
140 + receipt = plan.response()
141 +
142 + assert state["model_presets"][0]["utility"] == {"provider": "", "name": ""}
143 + assert editor.apply_change_plan(plan) == receipt
144 +
145 +
146 +def test_state_catalog_is_complete_truthful_and_omits_internal_tools() -> None:
147 + state = editor.build_editor_state("researcher")
148 +
149 + assert all(
150 + preset[slot]["provider"] and preset[slot]["name"]
151 + for preset in state["model_presets"]
152 + for slot in ("main", "utility", "embedding")
153 + )
154 + assert {prompt["group"] for prompt in state["prompts"]} == {
155 + f"2.{index}" for index in range(1, 11)
156 + }
157 + assert all(
158 + {
159 + "effective",
160 + "override",
161 + "has_override",
162 + "source",
163 + "source_chain",
164 + "state",
165 + }.issubset(prompt)
166 + for prompt in state["prompts"]
167 + )
168 + assert "AGENTS.md" not in {prompt["filename"] for prompt in state["prompts"]}
169 + specifics = next(
170 + prompt for prompt in state["prompts"]
171 + if prompt["filename"] == editor.SPECIFICS_FILE
172 + )
173 + assert specifics["source_chain"] == ["Framework", "Researcher"]
174 + assert any(
175 + any(source.startswith("Plugin ·") for source in prompt["source_chain"])
176 + for prompt in state["prompts"]
177 + )
178 + assert not {
179 + item["name"] for item in state["tools"]["catalog"]
180 + }.intersection({"response", "vision_load"})
181 +
182 +
183 +def test_builtin_prompt_override_never_touches_bundled_profile(user_root: Path) -> None:
184 + bundled = Path("agents/researcher")
185 + before = {path: path.read_bytes() for path in bundled.rglob("*") if path.is_file()}
186 + content = "Only the user-layer instructions change."
187 + plan = editor.build_change_plan(
188 + {
189 + "profile_id": "researcher",
190 + "prompts": {
191 + "set": {editor.SPECIFICS_FILE: content},
192 + "reset": [],
193 + },
194 + }
195 + )
196 +
197 + assert list(plan.changes) == [
198 + user_root / "researcher" / "prompts" / editor.SPECIFICS_FILE
199 + ]
200 + editor.apply_change_plan(plan)
201 + assert all(path.read_bytes() == payload for path, payload in before.items())
202 +
203 + reset = editor.build_change_plan(
204 + {
205 + "profile_id": "researcher",
206 + "prompts": {"set": {}, "reset": [editor.SPECIFICS_FILE]},
207 + }
208 + )
209 + editor.apply_change_plan(reset)
210 + assert not (user_root / "researcher" / "prompts" / editor.SPECIFICS_FILE).exists()
211 +
212 +
213 +def test_unrelated_empty_user_directory_survives_save(user_root: Path) -> None:
214 + manual = user_root / "researcher" / "tools" / "reserved-for-manual-use"
215 + manual.mkdir(parents=True)
216 +
217 + plan = editor.build_change_plan(
218 + {
219 + "profile_id": "researcher",
220 + "prompts": {
221 + "set": {editor.SPECIFICS_FILE: "Sparse change only."},
222 + "reset": [],
223 + },
224 + }
225 + )
226 + editor.apply_change_plan(plan)
227 +
228 + assert manual.is_dir()
229 +
230 +
231 +def test_profile_collision_includes_disabled_plugin_profiles(
232 + user_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
233 +) -> None:
234 + plugin_profile = tmp_path / "disabled-plugin" / "agents" / "reserved-agent"
235 + plugin_profile.mkdir(parents=True)
236 + monkeypatch.setattr(
237 + editor.plugins,
238 + "get_plugin_paths",
239 + lambda *parts: [str(plugin_profile)] if parts == ("agents", "reserved-agent") else [],
240 + )
241 +
242 + assert editor.profile_exists("reserved-agent") is True
243 +
244 +
245 +def test_metadata_empty_values_and_unknown_keys_survive(user_root: Path) -> None:
246 + profile = user_root / "researcher"
247 + profile.mkdir(parents=True)
248 + metadata = profile / "agent.yaml"
249 + metadata.write_text("custom_key: keep\ndescription: old\n", encoding="utf-8")
250 +
251 + plan = editor.build_change_plan(
252 + {
253 + "profile_id": "researcher",
254 + "metadata": {
255 + "set": {"description": "", "context": ""},
256 + "reset": [],
257 + },
258 + }
259 + )
260 + editor.apply_change_plan(plan)
261 +
262 + saved = yaml_helper.loads(metadata.read_text())
263 + assert saved == {"custom_key": "keep", "description": "", "context": ""}
264 +
265 +
266 +def test_plugin_configs_preserve_unowned_keys_and_use_json(user_root: Path) -> None:
267 + profile = user_root / "researcher" / "plugins"
268 + model = profile / "_model_config" / "config.json"
269 + tools = profile / "_tool_access" / "config.json"
270 + skill = profile / "_skills" / "config.json"
271 + for path, value in (
272 + (model, {"manual": 1}),
273 + (tools, {"manual": 2}),
274 + (skill, {"active_skills": [{"name": "existing"}]}),
275 + ):
276 + path.parent.mkdir(parents=True, exist_ok=True)
277 + path.write_text(json.dumps(value), encoding="utf-8")
278 +
279 + preset = editor.build_editor_state("researcher")["model_presets"][0]["name"]
280 + plan = editor.build_change_plan(
281 + {
282 + "profile_id": "researcher",
283 + "model_preset": {"mode": "preset", "name": preset},
284 + "tool_policy": {
285 + "mode": "custom",
286 + "default": "block",
287 + "allowed": ["local:search_engine"],
288 + "blocked": ["local:shell"],
289 + },
290 + "skill_policy": {
291 + "mode": "custom",
292 + "default": "block",
293 + "allowed": ["a0-development"],
294 + "blocked": [],
295 + },
296 + }
297 + )
298 + editor.apply_change_plan(plan)
299 +
300 + assert json.loads(model.read_text())["manual"] == 1
301 + assert set(json.loads(tools.read_text())) >= {
302 + "manual", "mode", "default", "allowed", "blocked"
303 + }
304 + skill_data = json.loads(skill.read_text())
305 + assert skill_data["active_skills"] == [{"name": "existing"}]
306 + assert skill_data["visibility_policy"]["default"] == "block"
307 +
308 +
309 +def test_model_and_off_tool_choices_write_only_their_json_contracts(
310 + user_root: Path,
311 +) -> None:
312 + preset = editor.build_editor_state("researcher")["model_presets"][1]["name"]
313 + model_path = user_root / "researcher" / "plugins" / "_model_config" / "config.json"
314 + tool_path = user_root / "researcher" / "plugins" / "_tool_access" / "config.json"
315 +
316 + inherit = editor.build_change_plan(
317 + {"profile_id": "researcher", "model_preset": {"mode": "inherit"}}
318 + )
319 + assert inherit.changes == {}
320 +
321 + selected = editor.build_change_plan(
322 + {
323 + "profile_id": "researcher",
324 + "model_preset": {"mode": "preset", "name": preset},
325 + }
326 + )
327 + assert list(selected.changes) == [model_path]
328 + assert json.loads(selected.changes[model_path].content) == {"model_preset": preset}
329 +
330 + off = editor.build_change_plan(
331 + {"profile_id": "researcher", "tool_policy": {"mode": "off"}}
332 + )
333 + assert list(off.changes) == [tool_path]
334 + assert json.loads(off.changes[tool_path].content) == {
335 + "mode": "custom",
336 + "default": "block",
337 + "allowed": [],
338 + "blocked": [],
339 + }
340 +
341 +
342 +def test_project_tool_policy_is_visible_but_editor_plan_stays_in_user_profile(
343 + user_root: Path,
344 + monkeypatch: pytest.MonkeyPatch,
345 +) -> None:
346 + real_find = editor.plugins.find_plugin_asset
347 +
348 + def find_plugin_asset(plugin_name, *parts, **scope):
349 + if plugin_name == editor.tool_policy.PLUGIN_NAME:
350 + return {
351 + "path": "/project/.a0proj/plugins/_tool_access/config.json",
352 + "project_name": "demo",
353 + "agent_profile": "",
354 + }
355 + return real_find(plugin_name, *parts, **scope)
356 +
357 + monkeypatch.setattr(editor.plugins, "find_plugin_asset", find_plugin_asset)
358 + monkeypatch.setattr(editor.tool_policy, "get_tool_catalog", lambda _agent: [])
359 + monkeypatch.setattr(editor.skills, "list_skill_catalog", lambda agent=None: [])
360 + context = editor._EditorContext("demo")
361 +
362 + state = editor.build_editor_state("researcher", context)
363 + plan = editor.build_change_plan(
364 + {
365 + "profile_id": "researcher",
366 + "tool_policy": {"mode": "off"},
367 + },
368 + context,
369 + )
370 +
371 + assert state["tools"]["project_override_active"] is True
372 + assert list(plan.changes) == [
373 + user_root / "researcher" / "plugins" / "_tool_access" / "config.json"
374 + ]
375 +
376 +
377 +def test_unavailable_skill_policy_ids_are_retained_in_editor_state(
378 + user_root: Path,
379 + monkeypatch: pytest.MonkeyPatch,
380 +) -> None:
381 + config = user_root / "researcher" / "plugins" / "_skills" / "config.json"
382 + config.parent.mkdir(parents=True)
383 + config.write_text(
384 + json.dumps(
385 + {
386 + "visibility_policy": {
387 + "mode": "custom",
388 + "default": "allow",
389 + "allowed": [],
390 + "blocked": ["removed-skill"],
391 + }
392 + }
393 + ),
394 + encoding="utf-8",
395 + )
396 + monkeypatch.setattr(editor.skills, "list_skill_catalog", lambda agent=None: [])
397 + monkeypatch.setattr(
398 + editor.skills,
399 + "get_visibility_policy",
400 + lambda _agent: {
401 + "mode": "custom",
402 + "default": "allow",
403 + "allowed": [],
404 + "blocked": ["removed-skill"],
405 + },
406 + )
407 +
408 + state = editor.build_editor_state("researcher")
409 +
410 + assert state["skills"]["policy"]["blocked"] == ["removed-skill"]
411 + assert state["skills"]["catalog"] == [
412 + {
413 + "name": "removed-skill",
414 + "description": "",
415 + "path": "removed-skill",
416 + "origin": "Unavailable",
417 + "hidden": True,
418 + "tags": [],
419 + "allowed_tools": [],
420 + "available": False,
421 + }
422 + ]
423 +
424 +
425 +def test_display_title_change_keeps_profile_id_and_builtin_delete_is_rejected(
426 + user_root: Path,
427 +) -> None:
428 + with pytest.raises(ValueError, match="cannot be deleted"):
429 + editor.plan_delete_custom("researcher")
430 +
431 + plan = editor.build_change_plan(
432 + {
433 + "profile_id": "researcher",
434 + "metadata": {"set": {"title": "Renamed Display"}, "reset": []},
435 + }
436 + )
437 + assert list(plan.changes) == [user_root / "researcher" / "agent.yaml"]
438 + editor.apply_change_plan(plan)
439 +
440 + assert (user_root / "researcher" / "agent.yaml").is_file()
441 + assert not (user_root / "renamed-display").exists()
442 +
443 +
444 +def test_save_rolls_back_every_file_after_commit_failure(
445 + user_root: Path,
446 + monkeypatch: pytest.MonkeyPatch,
447 +) -> None:
448 + root = user_root / "rollback-agent"
449 + first = root / "agent.yaml"
450 + second = root / "prompts" / editor.SPECIFICS_FILE
451 + second.parent.mkdir(parents=True)
452 + first.write_bytes(b"title: Before\n")
453 + second.write_bytes(b"before")
454 + plan = editor.ChangePlan(profile_id="rollback-agent")
455 + plan.write(first, b"title: After\n")
456 + plan.write(second, b"after")
457 +
458 + original_replace = editor.os.replace
459 + calls = 0
460 + invalidations: list[bool] = []
461 + monkeypatch.setattr(
462 + editor,
463 + "_invalidate_profile_caches",
464 + lambda: invalidations.append(True),
465 + )
466 +
467 + def fail_second(source, destination):
468 + nonlocal calls
469 + calls += 1
470 + if calls == 2:
471 + raise OSError("simulated commit failure")
472 + return original_replace(source, destination)
473 +
474 + monkeypatch.setattr(editor.os, "replace", fail_second)
475 + with pytest.raises(OSError, match="simulated"):
476 + editor.apply_change_plan(plan)
477 +
478 + assert first.read_bytes() == b"title: Before\n"
479 + assert second.read_bytes() == b"before"
480 + assert invalidations == []
481 +
482 +
483 +def test_remove_my_changes_preserves_manual_and_unknown_files(
484 + user_root: Path,
485 + monkeypatch: pytest.MonkeyPatch,
486 +) -> None:
487 + root = user_root / "researcher"
488 + prompt = root / "prompts" / editor.SPECIFICS_FILE
489 + manual_files = _write_manual_files(root)
490 + prompt.parent.mkdir(parents=True, exist_ok=True)
491 + prompt.write_text("override", encoding="utf-8")
492 + (root / "agent.yaml").write_text(
493 + "title: Mine\nunknown_key: keep\n", encoding="utf-8"
494 + )
495 + tool_config = root / "plugins" / "_tool_access" / "config.json"
496 + tool_config.parent.mkdir(parents=True)
497 + tool_config.write_text(
498 + json.dumps({"mode": "custom", "default": "block", "manual": True}),
499 + encoding="utf-8",
500 + )
501 +
502 + real_catalog = editor.prompt_catalog
503 +
504 + def catalog(agent):
505 + items = real_catalog(agent)
506 + for item in items:
507 + if item["filename"] == editor.SPECIFICS_FILE:
508 + item.update({"has_override": True, "inherited_source": "agents/researcher"})
509 + return items
510 +
511 + monkeypatch.setattr(editor, "prompt_catalog", catalog)
512 + plan = editor.plan_remove_changes("researcher")
513 + assert set(manual_files).isdisjoint(plan.changes)
514 + editor.apply_change_plan(plan)
515 +
516 + assert all(path.read_bytes() == payload for path, payload in manual_files.items())
517 + assert yaml_helper.loads((root / "agent.yaml").read_text()) == {
518 + "unknown_key": "keep"
519 + }
520 + assert json.loads(tool_config.read_text()) == {"manual": True}
521 +
522 +
523 +def test_mixed_save_matches_plan_preserves_every_unrelated_family_and_refreshes_cache(
524 + user_root: Path,
525 + monkeypatch: pytest.MonkeyPatch,
526 +) -> None:
527 + root = user_root / "researcher"
528 + manual_files = _write_manual_files(root)
529 + preset = editor.build_editor_state("researcher")["model_presets"][1]["name"]
530 + cleared: list[object] = []
531 + monkeypatch.setattr(editor.cache, "clear", lambda area: cleared.append(area))
532 + monkeypatch.setattr(
533 + editor.plugins,
534 + "clear_plugin_cache",
535 + lambda names: cleared.append(tuple(names)),
536 + )
537 + plan = editor.build_change_plan(
538 + {
539 + "profile_id": "researcher",
540 + "metadata": {"set": {"description": "Scoped"}, "reset": []},
541 + "prompts": {
542 + "set": {editor.SPECIFICS_FILE: "Only this prompt."},
543 + "reset": [],
544 + },
545 + "model_preset": {"mode": "preset", "name": preset},
546 + }
547 + )
548 + expected = plan.response()
549 +
550 + assert {
551 + path.relative_to(user_root).as_posix()
552 + for path, change in plan.changes.items()
553 + if change.action == "write"
554 + } == {
555 + "researcher/agent.yaml",
556 + f"researcher/prompts/{editor.SPECIFICS_FILE}",
557 + "researcher/plugins/_model_config/config.json",
558 + }
559 + assert editor.apply_change_plan(plan) == expected
560 + assert cleared == [
561 + editor.subagents.PATHS_CACHE_AREA,
562 + ("_agent_editor", "_model_config", "_tool_access", "_skills"),
563 + ]
564 + assert all(path.read_bytes() == payload for path, payload in manual_files.items())
565 +
566 +
567 +def test_empty_prompt_override_and_project_precedence_are_distinct(
568 + user_root: Path,
569 + tmp_path: Path,
570 + monkeypatch: pytest.MonkeyPatch,
571 +) -> None:
572 + framework = tmp_path / "framework"
573 + user_prompts = user_root / "researcher" / "prompts"
574 + project_meta = tmp_path / "project" / ".a0proj"
575 + project_prompts = project_meta / "agents" / "researcher" / "prompts"
576 + for path, text in (
577 + (framework / editor.SPECIFICS_FILE, "framework"),
578 + (user_prompts / editor.SPECIFICS_FILE, ""),
579 + (project_prompts / editor.SPECIFICS_FILE, "project"),
580 + ):
581 + path.parent.mkdir(parents=True, exist_ok=True)
582 + path.write_text(text, encoding="utf-8")
583 + monkeypatch.setattr(
584 + editor.subagents,
585 + "get_paths",
586 + lambda _agent, *parts: (
587 + [str(user_prompts), str(framework)]
588 + if not editor.projects.get_context_project_name(_agent.context)
589 + else [str(project_prompts), str(user_prompts), str(framework)]
590 + ),
591 + )
592 + original_meta = editor.projects.get_project_meta
593 + monkeypatch.setattr(
594 + editor.projects,
595 + "get_project_meta",
596 + lambda name, *parts: str(project_meta.joinpath(*parts))
597 + if name == "acceptance-project"
598 + else original_meta(name, *parts),
599 + )
600 +
601 + empty = next(
602 + item
603 + for item in editor.prompt_catalog(editor.EditorAgent("researcher"))
604 + if item["filename"] == editor.SPECIFICS_FILE
605 + )
606 + project = next(
607 + item
608 + for item in editor.prompt_catalog(
609 + editor.EditorAgent(
610 + "researcher",
611 + editor._EditorContext("acceptance-project"),
612 + )
613 + )
614 + if item["filename"] == editor.SPECIFICS_FILE
615 + )
616 +
617 + assert empty["state"] == "Overridden here (empty)"
618 + assert empty["has_override"] is True
619 + assert empty["effective"] == ""
620 + assert project["state"] == "Project override active"
621 + assert project["project_override_active"] is True
622 + assert project["effective"] == "project"
623 + assert project["source_chain"][-2:] == ["Your override", "Project · project"]
624 +
625 + reset = editor.build_change_plan(
626 + {
627 + "profile_id": "researcher",
628 + "prompts": {"set": {}, "reset": [editor.SPECIFICS_FILE]},
629 + }
630 + )
631 + assert list(reset.changes) == [user_prompts / editor.SPECIFICS_FILE]
632 + assert next(iter(reset.changes.values())).action == "delete"
633 +
634 +
635 +def test_avatar_is_normalized_and_avatar_only_edit_is_sparse(user_root: Path) -> None:
636 + from PIL import Image
637 +
638 + source = BytesIO()
639 + Image.new("RGB", (800, 400), "red").save(source, format="PNG")
640 + upload = FileStorage(stream=BytesIO(source.getvalue()), filename="avatar.png")
641 + staged = editor.stage_avatar(upload)
642 + plan = editor.build_change_plan(
643 + {
644 + "profile_id": "researcher",
645 + "metadata": {
646 + "set": {"avatar": {"kind": "image", "token": staged["token"]}},
647 + "reset": [],
648 + },
649 + }
650 + )
651 +
652 + assert {path.relative_to(user_root).as_posix() for path in plan.changes} == {
653 + "researcher/agent.yaml",
654 + "researcher/assets/avatar.webp",
655 + }
656 + editor.apply_change_plan(plan)
657 + avatar = user_root / "researcher" / "assets" / "avatar.webp"
658 + with Image.open(avatar) as normalized:
659 + assert normalized.format == "WEBP"
660 + assert normalized.size == (editor.AVATAR_SIZE, editor.AVATAR_SIZE)
661 + assert not normalized.getexif()
662 +
663 +
664 +@pytest.mark.parametrize(
665 + ("relative_path", "patch", "label"),
666 + (
667 + (
668 + "agent.yaml",
669 + {"metadata": {"set": {"description": "new"}, "reset": []}},
670 + "profile metadata",
671 + ),
672 + (
673 + "plugins/_tool_access/config.json",
674 + {"tool_policy": {"mode": "off"}},
675 + "tool policy configuration",
676 + ),
677 + ),
678 +)
679 +def test_invalid_existing_authored_files_are_never_overwritten(
680 + user_root: Path,
681 + relative_path: str,
682 + patch: dict,
683 + label: str,
684 +) -> None:
685 + path = user_root / "researcher" / relative_path
686 + path.parent.mkdir(parents=True, exist_ok=True)
687 + original = b"{ definitely not valid\n"
688 + path.write_bytes(original)
689 +
690 + with pytest.raises(ValueError, match=label):
691 + editor.build_change_plan({"profile_id": "researcher", **patch})
692 +
693 + assert path.read_bytes() == original
694 +
695 +
696 +def test_editor_preview_raw_reads_markdown_without_running_dynamic_processor(
697 + tmp_path: Path,
698 + monkeypatch: pytest.MonkeyPatch,
699 +) -> None:
700 + prompt_root = tmp_path / "prompts"
701 + prompt_root.mkdir()
702 + (prompt_root / editor.SPECIFICS_FILE).write_text(
703 + "Raw {{value}}", encoding="utf-8"
704 + )
705 + (prompt_root / "agent.system.main.specifics.py").write_text(
706 + "raise RuntimeError('must not run')\n",
707 + encoding="utf-8",
708 + )
709 + monkeypatch.setattr(
710 + editor.subagents,
711 + "get_paths",
712 + lambda *_args, **_kwargs: [str(prompt_root)],
713 + )
714 + monkeypatch.setattr(
715 + editor.files,
716 + "read_prompt_file",
717 + lambda *_args, **_kwargs: pytest.fail("dynamic prompt loader ran"),
718 + )
719 +
720 + item = next(
721 + item
722 + for item in editor.prompt_catalog(editor.EditorAgent("researcher"))
723 + if item["filename"] == editor.SPECIFICS_FILE
724 + )
725 +
726 + assert item["effective"] == "Raw {{value}}"
727 + assert item["preview"] == "Raw {{value}}"
728 + assert item["dynamic_processor"] is True
729 +
730 +
731 +def test_editor_api_keeps_default_auth_and_csrf_protection() -> None:
732 + for handler in (AgentEditor, AgentEditorAvatar):
733 + assert handler.requires_auth() is True
734 + assert handler.requires_csrf() is True
735 +
736 +
737 +def test_backend_has_no_legacy_save_or_model_request_path() -> None:
738 + sources = "\n".join(
739 + path.read_text(encoding="utf-8")
740 + for path in (
741 + Path(editor.__file__),
742 + Path("plugins/_agent_editor/api/agent_editor.py"),
743 + Path("plugins/_agent_editor/api/agent_editor_avatar.py"),
744 + )
745 + )
746 +
747 + assert "save_agent_data" not in sources
748 + assert "call_llm" not in sources
749 + assert "call_utility_model" not in sources
750 + assert "litellm" not in sources