| 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 | import unicodedata |
| 15 | from urllib.parse import urlencode |
| 16 | from uuid import uuid4 |
| 17 | |
| 18 | from helpers import cache, files, plugins, projects, skills, subagents, tool_policy |
| 19 | from helpers import yaml as yaml_helper |
| 20 | |
| 21 | |
| 22 | PROFILE_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9_-]{0,62}[a-z0-9])?$") |
| 23 | COLOR_PATTERN = re.compile(r"^#[0-9A-Fa-f]{6}$") |
| 24 | SPECIFICS_FILE = "agent.system.main.specifics.md" |
| 25 | METADATA_KEYS = ("title", "description", "context", "avatar") |
| 26 | MAX_PROMPT_BYTES = 512 * 1024 |
| 27 | MAX_PROMPTS_BYTES = 2 * 1024 * 1024 |
| 28 | MAX_AVATAR_BYTES = 8 * 1024 * 1024 |
| 29 | MAX_AVATAR_DIMENSION = 4096 |
| 30 | AVATAR_SIZE = 512 |
| 31 | RESERVED_PROFILE_IDS = {"_example"} |
| 32 | NON_PROMPT_MARKDOWN = {"AGENTS.md"} |
| 33 | USER_AGENTS_ROOT = Path(files.get_abs_path(subagents.USER_AGENTS_DIR)) |
| 34 | STAGED_AVATAR_ROOT = Path(files.get_abs_path("tmp", "agent-editor")) |
| 35 | _TOOL_POLICY_KEYS = ("mode", "default", "mcp_default", "allowed", "blocked") |
| 36 | _MUTATION_LOCK = threading.RLock() |
| 37 | |
| 38 | |
| 39 | class _EditorContext: |
| 40 | def __init__(self, project_name: str = "") -> None: |
| 41 | self.project_name = project_name |
| 42 | |
| 43 | def get_data(self, key: str, recursive: bool = True): |
| 44 | return self.project_name if key == projects.CONTEXT_DATA_KEY_PROJECT else None |
| 45 | |
| 46 | |
| 47 | class EditorAgent: |
| 48 | def __init__(self, profile_id: str, context: Any | None = None) -> None: |
| 49 | self.config = SimpleNamespace(profile=profile_id) |
| 50 | self.context = context or _EditorContext() |
| 51 | self.data: dict[str, Any] = {} |
| 52 | |
| 53 | def get_data(self, key: str): |
| 54 | return self.data.get(key) |
| 55 | |
| 56 | def read_prompt(self, filename: str, **kwargs: Any) -> str: |
| 57 | path = files.find_file_in_dirs(filename, subagents.get_paths(self, "prompts")) |
| 58 | return Path(path).read_text(encoding="utf-8") |
| 59 | |
| 60 | |
| 61 | @dataclass(frozen=True) |
| 62 | class ProfileLayer: |
| 63 | kind: str |
| 64 | metadata_path: Path | None |
| 65 | data: dict[str, Any] |
| 66 | |
| 67 | |
| 68 | @dataclass(frozen=True) |
| 69 | class FileChange: |
| 70 | action: str |
| 71 | path: Path |
| 72 | content: bytes | None = None |
| 73 | |
| 74 | @property |
| 75 | def relative_path(self) -> str: |
| 76 | return files.deabsolute_path(str(self.path)).replace(os.sep, "/") |
| 77 | |
| 78 | |
| 79 | @dataclass |
| 80 | class ChangePlan: |
| 81 | changes: dict[Path, FileChange] = field(default_factory=dict) |
| 82 | warnings: list[str] = field(default_factory=list) |
| 83 | staged_tokens: set[str] = field(default_factory=set) |
| 84 | profile_id: str = "" |
| 85 | project_name: str = "" |
| 86 | creating: bool = False |
| 87 | remove_empty_root: bool = False |
| 88 | |
| 89 | def write(self, path: Path, content: str | bytes) -> None: |
| 90 | payload = content.encode("utf-8") if isinstance(content, str) else content |
| 91 | if path.is_file() and path.read_bytes() == payload: |
| 92 | self.changes.pop(path, None) |
| 93 | return |
| 94 | self.changes[path] = FileChange("write", path, payload) |
| 95 | |
| 96 | def delete(self, path: Path) -> None: |
| 97 | if path.exists(): |
| 98 | self.changes[path] = FileChange("delete", path) |
| 99 | else: |
| 100 | self.changes.pop(path, None) |
| 101 | |
| 102 | def response(self) -> dict[str, Any]: |
| 103 | ordered = sorted(self.changes.values(), key=lambda change: change.relative_path) |
| 104 | return { |
| 105 | "written": [ |
| 106 | change.relative_path for change in ordered if change.action == "write" |
| 107 | ], |
| 108 | "deleted": [ |
| 109 | change.relative_path for change in ordered if change.action == "delete" |
| 110 | ], |
| 111 | "warnings": list(self.warnings), |
| 112 | } |
| 113 | |
| 114 | |
| 115 | def validate_profile_id(profile_id: Any) -> str: |
| 116 | value = str(profile_id or "").strip() |
| 117 | if value in RESERVED_PROFILE_IDS or not PROFILE_ID_PATTERN.fullmatch(value): |
| 118 | raise ValueError( |
| 119 | "Profile ID must be 1–64 lowercase letters, numbers, hyphens, or underscores." |
| 120 | ) |
| 121 | return value |
| 122 | |
| 123 | |
| 124 | def profile_id_from_title(title: Any) -> str: |
| 125 | if not isinstance(title, str) or not title.strip(): |
| 126 | raise ValueError("Agent name is required.") |
| 127 | value = unicodedata.normalize("NFKD", title).encode("ascii", "ignore").decode() |
| 128 | value = re.sub(r"[^a-z0-9_-]+", "-", value.lower()) |
| 129 | value = re.sub(r"[-_]{2,}", "-", value).strip("-_")[:64].rstrip("-_") |
| 130 | if not value: |
| 131 | raise ValueError("Agent name must contain at least one letter or number.") |
| 132 | return validate_profile_id(value) |
| 133 | |
| 134 | |
| 135 | def save_easy_profile( |
| 136 | title: Any, |
| 137 | instructions: Any, |
| 138 | context: Any | None = None, |
| 139 | *, |
| 140 | tool_policy: Any | None = None, |
| 141 | ) -> tuple[str, dict[str, Any]]: |
| 142 | if not isinstance(instructions, str) or not instructions.strip(): |
| 143 | raise ValueError("Instructions are required for a new agent.") |
| 144 | profile_id = profile_id_from_title(title) |
| 145 | patch: dict[str, Any] = { |
| 146 | "profile_id": profile_id, |
| 147 | "creating": True, |
| 148 | "editor_mode": "easy", |
| 149 | "metadata": {"set": {"title": title}, "reset": []}, |
| 150 | "prompts": {"set": {SPECIFICS_FILE: instructions}, "reset": []}, |
| 151 | } |
| 152 | if tool_policy is not None: |
| 153 | patch["tool_policy"] = _mapping(tool_policy, "tool_policy") |
| 154 | return profile_id, apply_change_plan(build_change_plan(patch, context)) |
| 155 | |
| 156 | |
| 157 | def _context_project_name(context: Any | None) -> str: |
| 158 | return str(projects.get_context_project_name(context) or "") if context else "" |
| 159 | |
| 160 | |
| 161 | def _profile_root(profile_id: str, project_name: str = "") -> Path: |
| 162 | return ( |
| 163 | Path(projects.get_project_meta(project_name, "agents", profile_id)) |
| 164 | if project_name |
| 165 | else USER_AGENTS_ROOT / profile_id |
| 166 | ) |
| 167 | |
| 168 | |
| 169 | def _scope_owns_custom_profile(profile_id: str, project_name: str) -> bool: |
| 170 | if not _profile_root(profile_id, project_name).is_dir(): |
| 171 | return False |
| 172 | if (Path(files.get_abs_path(subagents.DEFAULT_AGENTS_DIR)) / profile_id).is_dir(): |
| 173 | return False |
| 174 | if any(Path(path).is_dir() for path in plugins.get_plugin_paths("agents", profile_id)): |
| 175 | return False |
| 176 | return not project_name or not (USER_AGENTS_ROOT / profile_id).is_dir() |
| 177 | |
| 178 | |
| 179 | def profile_exists(profile_id: str, context: Any | None = None) -> bool: |
| 180 | agent = EditorAgent(profile_id, context) |
| 181 | if (Path(files.get_abs_path(subagents.DEFAULT_AGENTS_DIR)) / profile_id).is_dir(): |
| 182 | return True |
| 183 | if (USER_AGENTS_ROOT / profile_id).is_dir(): |
| 184 | return True |
| 185 | if any(Path(path).is_dir() for path in plugins.get_plugin_paths("agents", profile_id)): |
| 186 | return True |
| 187 | project_name = _context_project_name(agent.context) |
| 188 | return bool( |
| 189 | project_name |
| 190 | and Path(projects.get_project_meta(project_name, "agents", profile_id)).is_dir() |
| 191 | ) |
| 192 | |
| 193 | |
| 194 | def list_profiles(context: Any | None = None) -> list[dict[str, Any]]: |
| 195 | project_name = _context_project_name(context) |
| 196 | resolved = subagents.get_agents_dict(project_name) |
| 197 | enabled = subagents.get_available_agents_dict(project_name) |
| 198 | names = set(resolved) |
| 199 | roots = [Path(files.get_abs_path("agents")), USER_AGENTS_ROOT] |
| 200 | if project_name: |
| 201 | roots.append(Path(projects.get_project_meta(project_name, "agents"))) |
| 202 | for root in roots: |
| 203 | if root.is_dir(): |
| 204 | names.update(path.name for path in root.iterdir() if path.is_dir()) |
| 205 | |
| 206 | result: list[dict[str, Any]] = [] |
| 207 | for profile_id in sorted(names): |
| 208 | if profile_id in RESERVED_PROFILE_IDS: |
| 209 | continue |
| 210 | state = metadata_state(profile_id, context) |
| 211 | result.append( |
| 212 | { |
| 213 | "id": profile_id, |
| 214 | "title": state["title"]["effective"] or profile_id, |
| 215 | "description": state["description"]["effective"] or "", |
| 216 | "origin": state["origin"], |
| 217 | "origin_chain": state["origin_chain"], |
| 218 | "built_in": state["built_in"], |
| 219 | "scope_has_overrides": _scope_has_overrides(profile_id, context), |
| 220 | "deletable": state["deletable"], |
| 221 | "avatar": state["avatar"]["effective"], |
| 222 | "avatar_url": effective_avatar_url(profile_id, context), |
| 223 | "enabled": profile_id in enabled, |
| 224 | "available": profile_exists(profile_id, context), |
| 225 | } |
| 226 | ) |
| 227 | return result |
| 228 | |
| 229 | |
| 230 | def build_editor_state( |
| 231 | profile_id: str, |
| 232 | context: Any | None = None, |
| 233 | ) -> dict[str, Any]: |
| 234 | profile_id = validate_profile_id(profile_id) |
| 235 | agent = EditorAgent(profile_id, context) |
| 236 | prompt_files = prompt_catalog(agent) |
| 237 | tool_catalog = tool_policy.get_tool_catalog(agent) |
| 238 | from plugins._model_config.helpers import model_config |
| 239 | |
| 240 | presets: list[dict[str, Any]] = [] |
| 241 | for preset in model_config.get_presets(): |
| 242 | name = str(preset.get("name") or "").strip() |
| 243 | if not name: |
| 244 | continue |
| 245 | resolved = model_config.resolve_config_settings( |
| 246 | {model_config.MODEL_PRESET_CONFIG_KEY: name} |
| 247 | ) |
| 248 | presets.append( |
| 249 | { |
| 250 | "name": name, |
| 251 | "main": _model_identity(resolved.get("chat_model")), |
| 252 | "utility": _model_identity(resolved.get("utility_model")), |
| 253 | "embedding": _model_identity(resolved.get("embedding_model")), |
| 254 | } |
| 255 | ) |
| 256 | project_name = _context_project_name(agent.context) |
| 257 | model_path = _profile_config_path(profile_id, "_model_config", project_name) |
| 258 | model_scope = _read_mapping(model_path) |
| 259 | selected = model_config.get_configured_preset_name(agent=agent) |
| 260 | |
| 261 | tool_path = _profile_config_path(profile_id, tool_policy.PLUGIN_NAME, project_name) |
| 262 | skill_path = _profile_config_path( |
| 263 | profile_id, |
| 264 | skills.ACTIVE_SKILLS_PLUGIN_NAME, |
| 265 | project_name, |
| 266 | ) |
| 267 | tool_scope = _read_mapping(tool_path) |
| 268 | skill_scope = _read_mapping(skill_path) |
| 269 | skill_policy = skills.normalize_visibility_policy( |
| 270 | skill_scope.get("visibility_policy") |
| 271 | ) |
| 272 | effective_skill_policy = skills.get_visibility_policy(agent) |
| 273 | skill_catalog = [ |
| 274 | {**item, "available": True} |
| 275 | for item in skills.list_skill_catalog(agent=agent) |
| 276 | ] |
| 277 | known_skill_ids = { |
| 278 | value.casefold() |
| 279 | for item in skill_catalog |
| 280 | for value in ( |
| 281 | str(item.get("name") or ""), |
| 282 | str(item.get("path") or ""), |
| 283 | Path(str(item.get("path") or "")).name, |
| 284 | ) |
| 285 | if value |
| 286 | } |
| 287 | for skill_id in [*skill_policy["allowed"], *skill_policy["blocked"]]: |
| 288 | if skill_id.casefold() in known_skill_ids: |
| 289 | continue |
| 290 | known_skill_ids.add(skill_id.casefold()) |
| 291 | allowed = skills.is_skill_allowed(effective_skill_policy, skill_id) |
| 292 | skill_catalog.append( |
| 293 | { |
| 294 | "name": skill_id, |
| 295 | "description": "", |
| 296 | "path": skill_id, |
| 297 | "origin": "Unavailable", |
| 298 | "hidden": not allowed, |
| 299 | "tags": [], |
| 300 | "allowed_tools": [], |
| 301 | "available": False, |
| 302 | } |
| 303 | ) |
| 304 | skill_catalog.sort( |
| 305 | key=lambda item: ( |
| 306 | str(item.get("name") or "").casefold(), |
| 307 | str(item.get("path") or ""), |
| 308 | ) |
| 309 | ) |
| 310 | return { |
| 311 | "profile": build_profile_state(profile_id, context), |
| 312 | "prompts": prompt_files, |
| 313 | "tools": { |
| 314 | "policy": tool_policy.normalize_policy(tool_scope), |
| 315 | "effective_policy": tool_policy.get_policy(agent), |
| 316 | "has_override": any(key in tool_scope for key in _TOOL_POLICY_KEYS), |
| 317 | "catalog": tool_catalog, |
| 318 | }, |
| 319 | "skills": { |
| 320 | "policy": skill_policy, |
| 321 | "effective_policy": effective_skill_policy, |
| 322 | "has_override": "visibility_policy" in skill_scope, |
| 323 | "catalog": skill_catalog, |
| 324 | }, |
| 325 | "model_presets": presets, |
| 326 | "model_preset": { |
| 327 | "effective": selected, |
| 328 | "override": model_scope.get(model_config.MODEL_PRESET_CONFIG_KEY), |
| 329 | "has_override": model_config.MODEL_PRESET_CONFIG_KEY in model_scope, |
| 330 | }, |
| 331 | } |
| 332 | |
| 333 | |
| 334 | def build_profile_state( |
| 335 | profile_id: str, |
| 336 | context: Any | None = None, |
| 337 | ) -> dict[str, Any]: |
| 338 | profile_id = validate_profile_id(profile_id) |
| 339 | metadata = metadata_state(profile_id, context) |
| 340 | return { |
| 341 | "id": profile_id, |
| 342 | "origin": metadata.pop("origin"), |
| 343 | "origin_chain": metadata.pop("origin_chain"), |
| 344 | "built_in": metadata.pop("built_in"), |
| 345 | "scope_has_overrides": _scope_has_overrides(profile_id, context), |
| 346 | "deletable": metadata.pop("deletable"), |
| 347 | "metadata": metadata, |
| 348 | "avatar_url": effective_avatar_url(profile_id, context), |
| 349 | } |
| 350 | |
| 351 | |
| 352 | def metadata_state(profile_id: str, context: Any | None = None) -> dict[str, Any]: |
| 353 | layers = _metadata_layers(profile_id, context) |
| 354 | project_name = _context_project_name(context) |
| 355 | scope_kind = "project" if project_name else "user" |
| 356 | scope_layer = next((layer for layer in layers if layer.kind == scope_kind), None) |
| 357 | lower_layers = [layer for layer in layers if layer.kind != scope_kind] |
| 358 | state: dict[str, Any] = {} |
| 359 | for key in METADATA_KEYS: |
| 360 | effective, source, _ = _layer_value(layers, key) |
| 361 | inherited, inherited_source, _ = _layer_value(lower_layers, key) |
| 362 | state[key] = { |
| 363 | "effective": effective, |
| 364 | "override": ( |
| 365 | scope_layer.data.get(key) |
| 366 | if scope_layer and key in scope_layer.data |
| 367 | else None |
| 368 | ), |
| 369 | "has_override": bool(scope_layer and key in scope_layer.data), |
| 370 | "source": _relative_source(source), |
| 371 | "inherited": inherited, |
| 372 | "inherited_source": _relative_source(inherited_source), |
| 373 | } |
| 374 | |
| 375 | built_in = (Path(files.get_abs_path("agents")) / profile_id).is_dir() |
| 376 | plugin_origin = any(layer.kind == "plugin" for layer in layers) |
| 377 | state.update( |
| 378 | { |
| 379 | "origin": "Built-in" if built_in else "Plugin" if plugin_origin else "Custom", |
| 380 | "origin_chain": list(dict.fromkeys(layer.kind for layer in layers)), |
| 381 | "built_in": built_in, |
| 382 | "deletable": _scope_owns_custom_profile(profile_id, project_name), |
| 383 | } |
| 384 | ) |
| 385 | return state |
| 386 | |
| 387 | |
| 388 | def _scope_has_overrides(profile_id: str, context: Any | None = None) -> bool: |
| 389 | project_name = _context_project_name(context) |
| 390 | root = _profile_root(profile_id, project_name) |
| 391 | metadata = _read_mapping( |
| 392 | root / "agent.yaml" if (root / "agent.yaml").is_file() else root / "agent.json" |
| 393 | ) |
| 394 | if any(key in metadata for key in METADATA_KEYS): |
| 395 | return True |
| 396 | |
| 397 | scope_prompts = root / "prompts" |
| 398 | if scope_prompts.is_dir(): |
| 399 | inherited_roots = [ |
| 400 | directory / "prompts" |
| 401 | for _, directory in _profile_directories(profile_id, context) |
| 402 | if directory.absolute() != root.absolute() |
| 403 | ] |
| 404 | inherited_roots.append(Path(files.get_abs_path("prompts"))) |
| 405 | if any( |
| 406 | path.name not in NON_PROMPT_MARKDOWN |
| 407 | and any((inherited / path.name).is_file() for inherited in inherited_roots) |
| 408 | for path in scope_prompts.glob("*.md") |
| 409 | if path.is_file() |
| 410 | ): |
| 411 | return True |
| 412 | |
| 413 | from plugins._model_config.helpers import model_config |
| 414 | |
| 415 | checks = ( |
| 416 | ( |
| 417 | _profile_config_path(profile_id, "_model_config", project_name), |
| 418 | (model_config.MODEL_PRESET_CONFIG_KEY,), |
| 419 | ), |
| 420 | ( |
| 421 | _profile_config_path(profile_id, tool_policy.PLUGIN_NAME, project_name), |
| 422 | _TOOL_POLICY_KEYS, |
| 423 | ), |
| 424 | ( |
| 425 | _profile_config_path( |
| 426 | profile_id, |
| 427 | skills.ACTIVE_SKILLS_PLUGIN_NAME, |
| 428 | project_name, |
| 429 | ), |
| 430 | ("visibility_policy",), |
| 431 | ), |
| 432 | ) |
| 433 | return any(any(key in _read_mapping(path) for key in keys) for path, keys in checks) |
| 434 | |
| 435 | |
| 436 | def prompt_catalog(agent: EditorAgent) -> list[dict[str, Any]]: |
| 437 | roots = [Path(path) for path in subagents.get_paths(agent, "prompts")] |
| 438 | names = { |
| 439 | path.name |
| 440 | for root in roots |
| 441 | if root.is_dir() |
| 442 | for path in root.glob("*.md") |
| 443 | if path.is_file() and path.name not in NON_PROMPT_MARKDOWN |
| 444 | } |
| 445 | names.add(SPECIFICS_FILE) |
| 446 | project_name = _context_project_name(agent.context) |
| 447 | scope_prompt_dir = _profile_root(agent.config.profile, project_name) / "prompts" |
| 448 | project_root = ( |
| 449 | Path(projects.get_project_meta(project_name)) if project_name else None |
| 450 | ) |
| 451 | result: list[dict[str, Any]] = [] |
| 452 | for name in sorted(names, key=lambda value: (_prompt_group(value)[0], value)): |
| 453 | occurrences: list[tuple[Path, str, str]] = [] |
| 454 | for root in roots: |
| 455 | path = root / name |
| 456 | if path.is_file(): |
| 457 | kind, label = _prompt_source(root, scope_prompt_dir, project_root) |
| 458 | occurrences.append((path, kind, label)) |
| 459 | |
| 460 | effective = occurrences[0] if occurrences else None |
| 461 | override = next((item for item in occurrences if item[1] == "scope"), None) |
| 462 | inherited = next( |
| 463 | ( |
| 464 | item |
| 465 | for item in occurrences |
| 466 | if item[1] != "scope" |
| 467 | ), |
| 468 | None, |
| 469 | ) |
| 470 | effective_text, effective_error = _prompt_text(effective[0] if effective else None) |
| 471 | override_text, override_error = _prompt_text(override[0] if override else None) |
| 472 | inherited_text, inherited_error = _prompt_text( |
| 473 | inherited[0] if inherited else None |
| 474 | ) |
| 475 | group_number, group_label = _prompt_group(name) |
| 476 | source_kind = effective[1] if effective else "" |
| 477 | if effective_error or override_error or inherited_error: |
| 478 | state = "Conflict" |
| 479 | elif override: |
| 480 | state = ( |
| 481 | "Overridden here (empty)" |
| 482 | if override_text == "" |
| 483 | else "Overridden here" |
| 484 | ) |
| 485 | elif source_kind == "plugin": |
| 486 | state = "Plugin-provided" |
| 487 | elif effective: |
| 488 | state = "Inherited" |
| 489 | else: |
| 490 | state = "Unavailable" |
| 491 | |
| 492 | preview = _expand_static_prompt(name, roots, 0, set()) |
| 493 | |
| 494 | result.append( |
| 495 | { |
| 496 | "filename": name, |
| 497 | "group": group_number, |
| 498 | "group_label": group_label, |
| 499 | "state": state, |
| 500 | "effective": effective_text, |
| 501 | "override": override_text if override else None, |
| 502 | "has_override": bool(override), |
| 503 | "inherited": inherited_text, |
| 504 | "source": _relative_source(effective[0] if effective else None), |
| 505 | "inherited_source": _relative_source( |
| 506 | inherited[0] if inherited else None |
| 507 | ), |
| 508 | "source_chain": [ |
| 509 | label for _, _, label in reversed(occurrences) |
| 510 | ], |
| 511 | "preview": preview, |
| 512 | "error": effective_error or override_error or inherited_error, |
| 513 | "dynamic_processor": any( |
| 514 | (root / f"{Path(name).stem}.py").is_file() for root in roots |
| 515 | ), |
| 516 | } |
| 517 | ) |
| 518 | return result |
| 519 | |
| 520 | |
| 521 | def effective_avatar_path(profile_id: str, context: Any | None = None) -> Path | None: |
| 522 | layers = _metadata_layers(profile_id, context) |
| 523 | value, source, _ = _layer_value(layers, "avatar") |
| 524 | if not isinstance(value, dict) or value.get("kind") != "image" or not source: |
| 525 | return None |
| 526 | relative = str(value.get("value") or "") |
| 527 | candidate = (source.parent / relative).resolve() |
| 528 | if not files.is_in_dir(str(candidate), str(source.parent)) or not candidate.is_file(): |
| 529 | return None |
| 530 | return candidate |
| 531 | |
| 532 | |
| 533 | def effective_avatar_url(profile_id: str, context: Any | None = None) -> str: |
| 534 | path = effective_avatar_path(profile_id, context) |
| 535 | if not path: |
| 536 | return "" |
| 537 | query = { |
| 538 | "profile_id": profile_id, |
| 539 | "v": str(path.stat().st_mtime_ns), |
| 540 | } |
| 541 | project_name = _context_project_name(context) |
| 542 | if project_name: |
| 543 | query["project_name"] = project_name |
| 544 | return "/api/plugins/_agent_editor/agent_editor_avatar?" + urlencode(query) |
| 545 | |
| 546 | |
| 547 | def _profile_directories( |
| 548 | profile_id: str, context: Any | None |
| 549 | ) -> list[tuple[str, Path]]: |
| 550 | agent = EditorAgent(profile_id, context) |
| 551 | directories: list[tuple[str, Path]] = [ |
| 552 | ( |
| 553 | "profile", |
| 554 | Path(files.get_abs_path("agents", profile_id)), |
| 555 | ) |
| 556 | ] |
| 557 | for directory in plugins.get_enabled_plugin_paths(agent, "agents", profile_id): |
| 558 | directories.append(("plugin", Path(directory))) |
| 559 | directories.append(("user", USER_AGENTS_ROOT / profile_id)) |
| 560 | |
| 561 | project_name = _context_project_name(agent.context) |
| 562 | if project_name: |
| 563 | directories.append( |
| 564 | ( |
| 565 | "project", |
| 566 | Path(projects.get_project_meta(project_name, "agents", profile_id)), |
| 567 | ) |
| 568 | ) |
| 569 | return directories |
| 570 | |
| 571 | |
| 572 | def _metadata_layers(profile_id: str, context: Any | None) -> list[ProfileLayer]: |
| 573 | return [ |
| 574 | layer |
| 575 | for kind, directory in _profile_directories(profile_id, context) |
| 576 | if (layer := _load_profile_layer(kind, directory)) is not None |
| 577 | ] |
| 578 | |
| 579 | |
| 580 | def _load_profile_layer(kind: str, directory: Path) -> ProfileLayer | None: |
| 581 | if not directory.is_dir(): |
| 582 | return None |
| 583 | yaml_path = directory / "agent.yaml" |
| 584 | json_path = directory / "agent.json" |
| 585 | path = yaml_path if yaml_path.is_file() else json_path if json_path.is_file() else None |
| 586 | data = _read_mapping(path) if path else {} |
| 587 | return ProfileLayer(kind, path, data) |
| 588 | |
| 589 | |
| 590 | def _read_mapping(path: Path | None) -> dict[str, Any]: |
| 591 | if not path or not path.is_file(): |
| 592 | return {} |
| 593 | try: |
| 594 | value = ( |
| 595 | json.loads(path.read_text(encoding="utf-8")) |
| 596 | if path.suffix.lower() == ".json" |
| 597 | else yaml_helper.loads(path.read_text(encoding="utf-8")) |
| 598 | ) |
| 599 | except Exception: |
| 600 | return {} |
| 601 | return dict(value) if isinstance(value, dict) else {} |
| 602 | |
| 603 | |
| 604 | def _layer_value( |
| 605 | layers: list[ProfileLayer], key: str |
| 606 | ) -> tuple[Any, Path | None, str]: |
| 607 | for layer in reversed(layers): |
| 608 | if key in layer.data: |
| 609 | return layer.data[key], layer.metadata_path, layer.kind |
| 610 | return None, None, "" |
| 611 | |
| 612 | |
| 613 | def _prompt_source( |
| 614 | root: Path, scope_prompt_dir: Path, project_root: Path | None |
| 615 | ) -> tuple[str, str]: |
| 616 | if root.resolve() == scope_prompt_dir.resolve(): |
| 617 | return "scope", "Your override" |
| 618 | if project_root and files.is_in_dir(str(root), str(project_root)): |
| 619 | return "project", f"Project · {project_root.parent.name}" |
| 620 | plugin = plugins.get_plugin_name_from_path(root) |
| 621 | if plugin: |
| 622 | return "plugin", f"Plugin · {plugin}" |
| 623 | bundled_agents = Path(files.get_abs_path("agents")) |
| 624 | if files.is_in_dir(str(root), str(bundled_agents)): |
| 625 | return "profile", root.parent.name.replace("-", " ").title() |
| 626 | if root.resolve() == Path(files.get_abs_path("prompts")).resolve(): |
| 627 | return "framework", "Framework" |
| 628 | return "global", "Global" |
| 629 | |
| 630 | |
| 631 | def _prompt_group(filename: str) -> tuple[str, str]: |
| 632 | if filename == SPECIFICS_FILE: |
| 633 | return "2.1", "Agent instructions" |
| 634 | if filename == "agent.system.main.role.md": |
| 635 | return "2.2", "Role" |
| 636 | if filename == "agent.system.main.environment.md": |
| 637 | return "2.3", "Environment" |
| 638 | if filename.startswith("agent.system.main.communication"): |
| 639 | return "2.4", "Communication" |
| 640 | if filename == "agent.system.main.solving.md": |
| 641 | return "2.5", "Problem solving" |
| 642 | if filename == "agent.system.main.tips.md": |
| 643 | return "2.6", "Tips" |
| 644 | if filename.startswith("fw."): |
| 645 | return "2.7", "Framework messages" |
| 646 | if filename.startswith("agent.system.tool.") or "tool" in filename: |
| 647 | return "2.8", "Tool instructions" |
| 648 | if filename.startswith(("agent.context.", "agent.system.projects.", "agent.system.skills")): |
| 649 | return "2.9", "Context, projects & skills" |
| 650 | return "2.10", "Other" |
| 651 | |
| 652 | |
| 653 | def _model_identity(value: Any) -> dict[str, str]: |
| 654 | item = value if isinstance(value, dict) else {} |
| 655 | return { |
| 656 | "provider": str(item.get("provider") or ""), |
| 657 | "name": str(item.get("name") or ""), |
| 658 | } |
| 659 | |
| 660 | |
| 661 | def _profile_config_path( |
| 662 | profile_id: str, |
| 663 | plugin_name: str, |
| 664 | project_name: str = "", |
| 665 | ) -> Path: |
| 666 | return Path( |
| 667 | plugins.determine_plugin_asset_path( |
| 668 | plugin_name, |
| 669 | project_name, |
| 670 | profile_id, |
| 671 | plugins.CONFIG_FILE_NAME, |
| 672 | ) |
| 673 | ) |
| 674 | |
| 675 | |
| 676 | def _relative_source(path: Path | None) -> str: |
| 677 | return files.deabsolute_path(str(path)).replace(os.sep, "/") if path else "" |
| 678 | |
| 679 | |
| 680 | def _prompt_text(path: Path | None) -> tuple[str, str]: |
| 681 | if not path: |
| 682 | return "", "" |
| 683 | try: |
| 684 | return path.read_text(encoding="utf-8"), "" |
| 685 | except (OSError, UnicodeError) as exc: |
| 686 | return "", f"Prompt is not readable UTF-8: {exc}" |
| 687 | |
| 688 | |
| 689 | _NAMED_INCLUDE = re.compile(r"{{\s*include\s*['\"](.*?)['\"]\s*}}") |
| 690 | _ORIGINAL_INCLUDE = re.compile(r"{{\s*include\s+original\s*}}") |
| 691 | |
| 692 | |
| 693 | def _expand_static_prompt( |
| 694 | filename: str, |
| 695 | roots: list[Path], |
| 696 | start_index: int, |
| 697 | seen: set[Path], |
| 698 | ) -> str: |
| 699 | relative = Path(filename) |
| 700 | if relative.is_absolute() or ".." in relative.parts: |
| 701 | return "" |
| 702 | match = next( |
| 703 | ( |
| 704 | (index, root / relative) |
| 705 | for index, root in enumerate(roots[start_index:], start=start_index) |
| 706 | if (root / relative).is_file() |
| 707 | ), |
| 708 | None, |
| 709 | ) |
| 710 | if not match: |
| 711 | return "" |
| 712 | index, path = match |
| 713 | resolved = path.resolve() |
| 714 | if resolved in seen: |
| 715 | return "{{ include cycle }}" |
| 716 | text, error = _prompt_text(path) |
| 717 | if error: |
| 718 | return "" |
| 719 | branch_seen = {*seen, resolved} |
| 720 | text = _ORIGINAL_INCLUDE.sub( |
| 721 | lambda _match: _expand_static_prompt( |
| 722 | filename, |
| 723 | roots, |
| 724 | index + 1, |
| 725 | branch_seen, |
| 726 | ), |
| 727 | text, |
| 728 | ) |
| 729 | return _NAMED_INCLUDE.sub( |
| 730 | lambda include: _expand_static_prompt( |
| 731 | include.group(1), |
| 732 | roots, |
| 733 | 0, |
| 734 | branch_seen, |
| 735 | ) or include.group(0), |
| 736 | text, |
| 737 | ) |
| 738 | def build_change_plan( |
| 739 | patch: dict[str, Any], |
| 740 | context: Any | None = None, |
| 741 | ) -> ChangePlan: |
| 742 | if not isinstance(patch, dict): |
| 743 | raise ValueError("The save patch must be an object.") |
| 744 | |
| 745 | profile_id = validate_profile_id(patch.get("profile_id")) |
| 746 | project_name = _context_project_name(context) |
| 747 | profile_root = _profile_root(profile_id, project_name) |
| 748 | creating = bool(patch.get("creating")) |
| 749 | easy = str(patch.get("editor_mode") or "advanced").lower() == "easy" |
| 750 | exists = profile_exists(profile_id, context) |
| 751 | if creating and exists: |
| 752 | raise ValueError( |
| 753 | f"An agent with the profile ID `{profile_id}` already exists. " |
| 754 | "Open it to edit its user overrides or choose another name." |
| 755 | ) |
| 756 | if not creating and not exists: |
| 757 | raise ValueError(f'Agent profile "{profile_id}" does not exist.') |
| 758 | |
| 759 | plan = ChangePlan( |
| 760 | profile_id=profile_id, |
| 761 | project_name=project_name, |
| 762 | creating=creating, |
| 763 | ) |
| 764 | if "metadata" in patch: |
| 765 | _plan_metadata(plan, patch["metadata"], context, creating=creating) |
| 766 | if "prompts" in patch: |
| 767 | _plan_prompts( |
| 768 | plan, |
| 769 | patch["prompts"], |
| 770 | context, |
| 771 | creating=creating, |
| 772 | easy=easy, |
| 773 | ) |
| 774 | if "model_preset" in patch: |
| 775 | _plan_model_preset(plan, patch["model_preset"]) |
| 776 | if "tool_policy" in patch: |
| 777 | _plan_tool_policy(plan, patch["tool_policy"]) |
| 778 | if "skill_policy" in patch: |
| 779 | _plan_skill_policy(plan, patch["skill_policy"]) |
| 780 | |
| 781 | if creating: |
| 782 | metadata = _planned_mapping(plan, profile_root / "agent.yaml") |
| 783 | if not str(metadata.get("title") or "").strip(): |
| 784 | raise ValueError("Agent name is required.") |
| 785 | specifics = profile_root / "prompts" / SPECIFICS_FILE |
| 786 | change = plan.changes.get(specifics) |
| 787 | if not change or change.action != "write" or not (change.content or b"").strip(): |
| 788 | raise ValueError("Instructions are required for a new agent.") |
| 789 | |
| 790 | return plan |
| 791 | |
| 792 | |
| 793 | def plan_profile_enabled( |
| 794 | profile_id: str, |
| 795 | enabled: bool, |
| 796 | context: Any | None = None, |
| 797 | ) -> ChangePlan: |
| 798 | profile_id = validate_profile_id(profile_id) |
| 799 | if _context_project_name(context): |
| 800 | raise ValueError("Project availability is stored in project settings.") |
| 801 | if not profile_exists(profile_id, context): |
| 802 | raise ValueError(f'Agent profile "{profile_id}" does not exist.') |
| 803 | |
| 804 | root = _profile_root(profile_id) |
| 805 | yaml_path = root / "agent.yaml" |
| 806 | legacy_path = root / "agent.json" |
| 807 | data = _read_mapping_strict( |
| 808 | yaml_path if yaml_path.is_file() else legacy_path, |
| 809 | "profile metadata", |
| 810 | ) |
| 811 | lower_layers = [ |
| 812 | layer for layer in _metadata_layers(profile_id, context) |
| 813 | if layer.kind != "user" |
| 814 | ] |
| 815 | inherited, _, _ = _layer_value(lower_layers, "enabled") |
| 816 | inherited = True if inherited is None else bool(inherited) |
| 817 | if enabled == inherited: |
| 818 | data.pop("enabled", None) |
| 819 | else: |
| 820 | data["enabled"] = enabled |
| 821 | |
| 822 | plan = ChangePlan(profile_id=profile_id) |
| 823 | if data: |
| 824 | plan.write(yaml_path, yaml_helper.dumps(data)) |
| 825 | else: |
| 826 | plan.delete(yaml_path) |
| 827 | return plan |
| 828 | |
| 829 | |
| 830 | def set_profile_enabled( |
| 831 | profile_id: str, |
| 832 | enabled: bool, |
| 833 | context: Any | None = None, |
| 834 | ) -> dict[str, Any]: |
| 835 | profile_id = validate_profile_id(profile_id) |
| 836 | project_name = _context_project_name(context) |
| 837 | if not profile_exists(profile_id, context): |
| 838 | raise ValueError(f'Agent profile "{profile_id}" does not exist.') |
| 839 | |
| 840 | with _MUTATION_LOCK: |
| 841 | available = subagents.get_available_agents_dict(project_name or None) |
| 842 | if not enabled and profile_id in available and len(available) == 1: |
| 843 | raise ValueError("At least one agent profile must remain available.") |
| 844 | |
| 845 | if project_name: |
| 846 | projects.set_project_subagent_enabled(project_name, profile_id, enabled) |
| 847 | receipt = { |
| 848 | "written": [ |
| 849 | _relative_source( |
| 850 | Path(projects.get_project_meta(project_name, "agents.json")) |
| 851 | ) |
| 852 | ], |
| 853 | "deleted": [], |
| 854 | "warnings": [], |
| 855 | } |
| 856 | else: |
| 857 | receipt = apply_change_plan( |
| 858 | plan_profile_enabled(profile_id, enabled, context) |
| 859 | ) |
| 860 | |
| 861 | if not enabled: |
| 862 | projects.reconcile_agent_profiles( |
| 863 | project_name or None, all_scopes=not project_name |
| 864 | ) |
| 865 | return receipt |
| 866 | |
| 867 | |
| 868 | def plan_duplicate_profile( |
| 869 | profile_id: str, |
| 870 | context: Any | None = None, |
| 871 | ) -> tuple[ChangePlan, str]: |
| 872 | profile_id = validate_profile_id(profile_id) |
| 873 | if not profile_exists(profile_id, context): |
| 874 | raise ValueError(f'Agent profile "{profile_id}" does not exist.') |
| 875 | |
| 876 | state = metadata_state(profile_id, context) |
| 877 | source_title = str(state["title"]["effective"] or profile_id).strip() or profile_id |
| 878 | index = 1 |
| 879 | while True: |
| 880 | suffix = f"-{index}" |
| 881 | stem = profile_id[: 64 - len(suffix)].rstrip("-_") |
| 882 | target_id = f"{stem}{suffix}" |
| 883 | if not profile_exists(target_id, context): |
| 884 | break |
| 885 | index += 1 |
| 886 | |
| 887 | target_title = f"{source_title} {index}" |
| 888 | project_name = _context_project_name(context) |
| 889 | target_root = _profile_root(target_id, project_name) |
| 890 | plan = ChangePlan( |
| 891 | profile_id=target_id, |
| 892 | project_name=project_name, |
| 893 | creating=True, |
| 894 | ) |
| 895 | |
| 896 | metadata: dict[str, Any] = {} |
| 897 | for layer in _metadata_layers(profile_id, context): |
| 898 | metadata.update(layer.data) |
| 899 | for key in ("name", "path", "origin", "prompts", "enabled"): |
| 900 | metadata.pop(key, None) |
| 901 | metadata["title"] = target_title |
| 902 | plan.write(target_root / "agent.yaml", yaml_helper.dumps(metadata)) |
| 903 | |
| 904 | skipped = {"agent.yaml", "agent.json", "AGENTS.md"} |
| 905 | for _, source_root in _profile_directories(profile_id, context): |
| 906 | if not source_root.is_dir(): |
| 907 | continue |
| 908 | if source_root.is_symlink(): |
| 909 | raise ValueError("Profile symlinks must be removed before duplication.") |
| 910 | for source in sorted(source_root.rglob("*")): |
| 911 | if source.is_symlink(): |
| 912 | raise ValueError("Profile symlinks must be removed before duplication.") |
| 913 | if source.is_file() and source.name not in skipped: |
| 914 | plan.write(target_root / source.relative_to(source_root), source.read_bytes()) |
| 915 | |
| 916 | return plan, target_title |
| 917 | |
| 918 | |
| 919 | def _plan_metadata( |
| 920 | plan: ChangePlan, |
| 921 | value: Any, |
| 922 | context: Any | None, |
| 923 | *, |
| 924 | creating: bool, |
| 925 | ) -> None: |
| 926 | section = _mapping(value, "metadata") |
| 927 | set_values = _mapping(section.get("set", {}), "metadata.set") |
| 928 | reset_values = _string_list(section.get("reset", []), "metadata.reset") |
| 929 | unknown = (set(set_values) | set(reset_values)) - set(METADATA_KEYS) |
| 930 | if unknown: |
| 931 | raise ValueError(f"Unknown metadata field: {sorted(unknown)[0]}") |
| 932 | if set(set_values).intersection(reset_values): |
| 933 | raise ValueError("A metadata field cannot be set and reset in one save.") |
| 934 | |
| 935 | profile_id = plan.profile_id |
| 936 | profile_root = _profile_root(profile_id, plan.project_name) |
| 937 | yaml_path = profile_root / "agent.yaml" |
| 938 | legacy_path = profile_root / "agent.json" |
| 939 | data = _read_mapping_strict( |
| 940 | yaml_path if yaml_path.is_file() else legacy_path, |
| 941 | "profile metadata", |
| 942 | ) |
| 943 | state = metadata_state(profile_id, context) |
| 944 | current_user_avatar = data.get("avatar") |
| 945 | |
| 946 | for key in reset_values: |
| 947 | data.pop(key, None) |
| 948 | if key == "avatar" and _editor_image_avatar(current_user_avatar): |
| 949 | plan.delete(profile_root / "assets" / "avatar.webp") |
| 950 | |
| 951 | for key, raw in set_values.items(): |
| 952 | if key in {"title", "description", "context"}: |
| 953 | if not isinstance(raw, str): |
| 954 | raise ValueError(f"Metadata field {key} must be text.") |
| 955 | normalized: Any = raw |
| 956 | if key == "title" and not raw.strip(): |
| 957 | raise ValueError("Agent name is required.") |
| 958 | else: |
| 959 | normalized = _normalize_avatar(plan, raw) |
| 960 | if normalized.get("kind") == "color" and _editor_image_avatar( |
| 961 | current_user_avatar |
| 962 | ): |
| 963 | plan.delete(profile_root / "assets" / "avatar.webp") |
| 964 | |
| 965 | inherited = state[key]["inherited"] |
| 966 | if not creating and normalized == inherited: |
| 967 | data.pop(key, None) |
| 968 | else: |
| 969 | data[key] = normalized |
| 970 | |
| 971 | if data: |
| 972 | plan.write(yaml_path, yaml_helper.dumps(data)) |
| 973 | else: |
| 974 | plan.delete(yaml_path) |
| 975 | |
| 976 | |
| 977 | def _plan_prompts( |
| 978 | plan: ChangePlan, |
| 979 | value: Any, |
| 980 | context: Any | None, |
| 981 | *, |
| 982 | creating: bool, |
| 983 | easy: bool, |
| 984 | ) -> None: |
| 985 | section = _mapping(value, "prompts") |
| 986 | set_values = _mapping(section.get("set", {}), "prompts.set") |
| 987 | reset_values = _string_list(section.get("reset", []), "prompts.reset") |
| 988 | if set(set_values).intersection(reset_values): |
| 989 | raise ValueError("A prompt cannot be set and reset in one save.") |
| 990 | |
| 991 | agent = EditorAgent(plan.profile_id, context) |
| 992 | catalog = {item["filename"]: item for item in prompt_catalog(agent)} |
| 993 | for filename in [*set_values, *reset_values]: |
| 994 | if filename not in catalog: |
| 995 | raise ValueError(f'Prompt file "{filename}" is not in the editor catalog.') |
| 996 | |
| 997 | total = 0 |
| 998 | for filename, content in set_values.items(): |
| 999 | if not isinstance(content, str): |
| 1000 | raise ValueError(f'Prompt file "{filename}" must contain UTF-8 text.') |
| 1001 | if "\x00" in content: |
| 1002 | raise ValueError(f'Prompt file "{filename}" contains a NUL character.') |
| 1003 | payload_size = len(content.encode("utf-8")) |
| 1004 | if payload_size > MAX_PROMPT_BYTES: |
| 1005 | raise ValueError(f'Prompt file "{filename}" is too large.') |
| 1006 | total += payload_size |
| 1007 | if easy and filename == SPECIFICS_FILE and not content.strip(): |
| 1008 | raise ValueError( |
| 1009 | "Instructions can’t be empty. To remove your changes, use " |
| 1010 | "Restore original instructions." |
| 1011 | ) |
| 1012 | item = catalog[filename] |
| 1013 | path = _profile_root(plan.profile_id, plan.project_name) / "prompts" / filename |
| 1014 | if ( |
| 1015 | not creating |
| 1016 | and item.get("inherited_source") |
| 1017 | and content == item.get("inherited", "") |
| 1018 | ): |
| 1019 | plan.delete(path) |
| 1020 | else: |
| 1021 | plan.write(path, content) |
| 1022 | |
| 1023 | if total > MAX_PROMPTS_BYTES: |
| 1024 | raise ValueError("The combined prompt changes are too large.") |
| 1025 | for filename in reset_values: |
| 1026 | plan.delete( |
| 1027 | _profile_root(plan.profile_id, plan.project_name) / "prompts" / filename |
| 1028 | ) |
| 1029 | |
| 1030 | |
| 1031 | def _plan_model_preset(plan: ChangePlan, value: Any) -> None: |
| 1032 | section = _mapping(value, "model_preset") |
| 1033 | mode = str(section.get("mode") or "inherit").strip().lower() |
| 1034 | path = _profile_config_path(plan.profile_id, "_model_config", plan.project_name) |
| 1035 | data = _read_mapping_strict(path, "model preset configuration") |
| 1036 | |
| 1037 | from plugins._model_config.helpers import model_config |
| 1038 | |
| 1039 | if mode == "inherit": |
| 1040 | data.pop(model_config.MODEL_PRESET_CONFIG_KEY, None) |
| 1041 | elif mode == "preset": |
| 1042 | requested = str(section.get("name") or "").strip() |
| 1043 | preset = model_config.resolve_preset(requested) |
| 1044 | if not preset: |
| 1045 | raise ValueError(f'Model preset "{requested}" does not exist.') |
| 1046 | data[model_config.MODEL_PRESET_CONFIG_KEY] = str(preset.get("name") or requested) |
| 1047 | else: |
| 1048 | raise ValueError("Model preset mode must be inherit or preset.") |
| 1049 | _plan_json_mapping(plan, path, data) |
| 1050 | |
| 1051 | |
| 1052 | def _plan_tool_policy(plan: ChangePlan, value: Any) -> None: |
| 1053 | section = _mapping(value, "tool_policy") |
| 1054 | mode = str(section.get("mode") or "inherit").strip().lower() |
| 1055 | path = _profile_config_path( |
| 1056 | plan.profile_id, |
| 1057 | tool_policy.PLUGIN_NAME, |
| 1058 | plan.project_name, |
| 1059 | ) |
| 1060 | data = _read_mapping_strict(path, "tool policy configuration") |
| 1061 | |
| 1062 | if mode == "inherit": |
| 1063 | for key in _TOOL_POLICY_KEYS: |
| 1064 | data.pop(key, None) |
| 1065 | else: |
| 1066 | if mode == "off": |
| 1067 | policy = { |
| 1068 | "mode": "custom", |
| 1069 | "default": "block", |
| 1070 | "mcp_default": "block", |
| 1071 | "allowed": [], |
| 1072 | "blocked": [], |
| 1073 | } |
| 1074 | elif mode == "custom": |
| 1075 | policy = tool_policy.normalize_policy(section) |
| 1076 | allowed = set(policy["allowed"]) |
| 1077 | blocked = set(policy["blocked"]) |
| 1078 | if allowed.intersection(blocked): |
| 1079 | raise ValueError("A tool cannot be both allowed and blocked.") |
| 1080 | for tool_id in [*policy["allowed"], *policy["blocked"]]: |
| 1081 | if not _valid_tool_id(tool_id): |
| 1082 | raise ValueError(f'Invalid canonical tool ID "{tool_id}".') |
| 1083 | else: |
| 1084 | raise ValueError("Tool policy mode must be inherit, off, or custom.") |
| 1085 | data.update({key: policy[key] for key in _TOOL_POLICY_KEYS}) |
| 1086 | _plan_json_mapping(plan, path, data) |
| 1087 | |
| 1088 | |
| 1089 | def _plan_skill_policy(plan: ChangePlan, value: Any) -> None: |
| 1090 | section = _mapping(value, "skill_policy") |
| 1091 | mode = str(section.get("mode") or "inherit").strip().lower() |
| 1092 | path = _profile_config_path( |
| 1093 | plan.profile_id, |
| 1094 | skills.ACTIVE_SKILLS_PLUGIN_NAME, |
| 1095 | plan.project_name, |
| 1096 | ) |
| 1097 | data = _read_mapping_strict(path, "skill policy configuration") |
| 1098 | |
| 1099 | if mode == "inherit": |
| 1100 | data.pop("visibility_policy", None) |
| 1101 | else: |
| 1102 | raw = ( |
| 1103 | {"mode": "custom", "default": "block", "allowed": [], "blocked": []} |
| 1104 | if mode == "off" |
| 1105 | else section |
| 1106 | ) |
| 1107 | if mode not in {"off", "custom"}: |
| 1108 | raise ValueError("Skill policy mode must be inherit, off, or custom.") |
| 1109 | policy = skills.normalize_visibility_policy(raw) |
| 1110 | if set(policy["allowed"]).intersection(policy["blocked"]): |
| 1111 | raise ValueError("A skill cannot be both allowed and blocked.") |
| 1112 | for skill_id in [*policy["allowed"], *policy["blocked"]]: |
| 1113 | if not skill_id or len(skill_id) > 512 or "\x00" in skill_id: |
| 1114 | raise ValueError("Invalid skill ID.") |
| 1115 | data["visibility_policy"] = policy |
| 1116 | _plan_json_mapping(plan, path, data) |
| 1117 | |
| 1118 | |
| 1119 | def _normalize_avatar(plan: ChangePlan, value: Any) -> dict[str, str]: |
| 1120 | avatar = _mapping(value, "metadata.set.avatar") |
| 1121 | kind = str(avatar.get("kind") or "").strip().lower() |
| 1122 | if kind == "color": |
| 1123 | color = str(avatar.get("value") or "").strip().upper() |
| 1124 | if not COLOR_PATTERN.fullmatch(color): |
| 1125 | raise ValueError("Avatar color must be a six-digit hex color.") |
| 1126 | return {"kind": "color", "value": color} |
| 1127 | if kind == "image": |
| 1128 | token = str(avatar.get("token") or "").strip() |
| 1129 | if token: |
| 1130 | source = staged_avatar_path(token) |
| 1131 | if not source.is_file(): |
| 1132 | raise ValueError("The staged avatar has expired. Upload it again.") |
| 1133 | plan.write( |
| 1134 | _profile_root(plan.profile_id, plan.project_name) |
| 1135 | / "assets" |
| 1136 | / "avatar.webp", |
| 1137 | source.read_bytes(), |
| 1138 | ) |
| 1139 | plan.staged_tokens.add(token) |
| 1140 | return {"kind": "image", "value": "assets/avatar.webp"} |
| 1141 | raise ValueError("Avatar must be a color or a staged image.") |
| 1142 | |
| 1143 | |
| 1144 | def _plan_json_mapping(plan: ChangePlan, path: Path, data: dict[str, Any]) -> None: |
| 1145 | if data: |
| 1146 | plan.write(path, json.dumps(data, ensure_ascii=False, indent=2) + "\n") |
| 1147 | else: |
| 1148 | plan.delete(path) |
| 1149 | |
| 1150 | |
| 1151 | def _planned_mapping(plan: ChangePlan, path: Path) -> dict[str, Any]: |
| 1152 | change = plan.changes.get(path) |
| 1153 | if change and change.action == "delete": |
| 1154 | return {} |
| 1155 | if change and change.content is not None: |
| 1156 | value = yaml_helper.loads(change.content.decode("utf-8")) |
| 1157 | return dict(value) if isinstance(value, dict) else {} |
| 1158 | return _read_mapping(path) |
| 1159 | |
| 1160 | |
| 1161 | def _read_mapping_strict(path: Path, label: str) -> dict[str, Any]: |
| 1162 | if not path.is_file(): |
| 1163 | return {} |
| 1164 | try: |
| 1165 | value = ( |
| 1166 | json.loads(path.read_text(encoding="utf-8")) |
| 1167 | if path.suffix.lower() == ".json" |
| 1168 | else yaml_helper.loads(path.read_text(encoding="utf-8")) |
| 1169 | ) |
| 1170 | except Exception as exc: |
| 1171 | raise ValueError(f"Existing {label} is invalid and was not changed: {exc}") from exc |
| 1172 | if not isinstance(value, dict): |
| 1173 | raise ValueError(f"Existing {label} must be an object and was not changed.") |
| 1174 | return dict(value) |
| 1175 | |
| 1176 | |
| 1177 | def _mapping(value: Any, label: str) -> dict[str, Any]: |
| 1178 | if not isinstance(value, dict): |
| 1179 | raise ValueError(f"{label} must be an object.") |
| 1180 | return dict(value) |
| 1181 | |
| 1182 | |
| 1183 | def _string_list(value: Any, label: str) -> list[str]: |
| 1184 | if not isinstance(value, list) or any(not isinstance(item, str) for item in value): |
| 1185 | raise ValueError(f"{label} must be an array of strings.") |
| 1186 | return list(dict.fromkeys(value)) |
| 1187 | |
| 1188 | |
| 1189 | def _valid_tool_id(value: str) -> bool: |
| 1190 | return bool( |
| 1191 | re.fullmatch( |
| 1192 | r"(?:local:[^\s/:]+|(?:mcp|plugin):[^\s/:]+:[^\s/:]+)", |
| 1193 | value, |
| 1194 | ) |
| 1195 | ) |
| 1196 | |
| 1197 | |
| 1198 | def _editor_image_avatar(value: Any) -> bool: |
| 1199 | return ( |
| 1200 | isinstance(value, dict) |
| 1201 | and value.get("kind") == "image" |
| 1202 | and value.get("value") == "assets/avatar.webp" |
| 1203 | ) |
| 1204 | |
| 1205 | |
| 1206 | def apply_change_plan(plan: ChangePlan) -> dict[str, Any]: |
| 1207 | project_name = ( |
| 1208 | projects.validate_project_name(plan.project_name) if plan.project_name else "" |
| 1209 | ) |
| 1210 | if project_name and not Path(projects.get_project_folder(project_name)).is_dir(): |
| 1211 | raise ValueError("Project not found.") |
| 1212 | root = _profile_root( |
| 1213 | validate_profile_id(plan.profile_id), |
| 1214 | project_name, |
| 1215 | ) |
| 1216 | changes = sorted(plan.changes.values(), key=lambda item: str(item.path)) |
| 1217 | _validate_plan_paths(root, changes) |
| 1218 | receipt = plan.response() |
| 1219 | |
| 1220 | with _MUTATION_LOCK: |
| 1221 | if plan.creating and profile_exists( |
| 1222 | plan.profile_id, |
| 1223 | _EditorContext(project_name), |
| 1224 | ): |
| 1225 | raise ValueError( |
| 1226 | f'Agent profile "{plan.profile_id}" was created before this save completed.' |
| 1227 | ) |
| 1228 | snapshots = { |
| 1229 | change.path: change.path.read_bytes() if change.path.is_file() else None |
| 1230 | for change in changes |
| 1231 | } |
| 1232 | staged: dict[Path, Path] = {} |
| 1233 | created_dirs: set[Path] = set() |
| 1234 | try: |
| 1235 | for change in changes: |
| 1236 | if change.action != "write": |
| 1237 | continue |
| 1238 | _ensure_parent(change.path.parent, created_dirs) |
| 1239 | staged[change.path] = _stage_bytes( |
| 1240 | change.path.parent, |
| 1241 | change.path.name, |
| 1242 | change.content or b"", |
| 1243 | ) |
| 1244 | |
| 1245 | for change in changes: |
| 1246 | if change.action == "write": |
| 1247 | os.replace(staged.pop(change.path), change.path) |
| 1248 | _fsync_directory(change.path.parent) |
| 1249 | elif change.path.is_file() or change.path.is_symlink(): |
| 1250 | change.path.unlink() |
| 1251 | _fsync_directory(change.path.parent) |
| 1252 | |
| 1253 | if plan.remove_empty_root: |
| 1254 | _prune_empty_directories(root) |
| 1255 | if root.is_dir() and not any(root.iterdir()): |
| 1256 | root.rmdir() |
| 1257 | except Exception: |
| 1258 | for temporary in staged.values(): |
| 1259 | temporary.unlink(missing_ok=True) |
| 1260 | _restore_snapshots(snapshots, root) |
| 1261 | for directory in sorted(created_dirs, key=lambda path: len(path.parts), reverse=True): |
| 1262 | try: |
| 1263 | directory.rmdir() |
| 1264 | except OSError: |
| 1265 | pass |
| 1266 | raise |
| 1267 | |
| 1268 | for token in plan.staged_tokens: |
| 1269 | staged_avatar_path(token).unlink(missing_ok=True) |
| 1270 | _invalidate_profile_caches() |
| 1271 | return receipt |
| 1272 | |
| 1273 | |
| 1274 | def plan_remove_changes( |
| 1275 | profile_id: str, |
| 1276 | context: Any | None = None, |
| 1277 | *, |
| 1278 | destructive: bool = False, |
| 1279 | ) -> ChangePlan: |
| 1280 | profile_id = validate_profile_id(profile_id) |
| 1281 | if not profile_exists(profile_id, context): |
| 1282 | raise ValueError(f'Agent profile "{profile_id}" does not exist.') |
| 1283 | if destructive: |
| 1284 | return _full_delete_plan(profile_id, context) |
| 1285 | |
| 1286 | catalog = prompt_catalog(EditorAgent(profile_id, context)) |
| 1287 | prompt_resets = [ |
| 1288 | item["filename"] |
| 1289 | for item in catalog |
| 1290 | if item.get("has_override") and item.get("inherited_source") |
| 1291 | ] |
| 1292 | plan = build_change_plan( |
| 1293 | { |
| 1294 | "profile_id": profile_id, |
| 1295 | "metadata": {"set": {}, "reset": list(METADATA_KEYS)}, |
| 1296 | "prompts": {"set": {}, "reset": prompt_resets}, |
| 1297 | "model_preset": {"mode": "inherit"}, |
| 1298 | "tool_policy": {"mode": "inherit"}, |
| 1299 | "skill_policy": {"mode": "inherit"}, |
| 1300 | }, |
| 1301 | context, |
| 1302 | ) |
| 1303 | return plan |
| 1304 | |
| 1305 | |
| 1306 | def plan_delete_custom(profile_id: str, context: Any | None = None) -> ChangePlan: |
| 1307 | profile_id = validate_profile_id(profile_id) |
| 1308 | state = metadata_state(profile_id, context) |
| 1309 | if not state["deletable"]: |
| 1310 | raise ValueError("Only custom agents created in this scope can be deleted.") |
| 1311 | return _full_delete_plan(profile_id, context) |
| 1312 | |
| 1313 | |
| 1314 | def delete_impact(profile_id: str, context: Any | None = None) -> dict[str, Any]: |
| 1315 | profile_id = validate_profile_id(profile_id) |
| 1316 | state = metadata_state(profile_id, context) |
| 1317 | project_name = _context_project_name(context) |
| 1318 | root = _profile_root(profile_id, project_name) |
| 1319 | file_paths = [ |
| 1320 | _relative_source(path) |
| 1321 | for path in sorted(root.rglob("*")) |
| 1322 | if path.is_file() or path.is_symlink() |
| 1323 | ] if root.is_dir() else [] |
| 1324 | references = _profile_reference_paths(profile_id, project_name) |
| 1325 | sessions: list[str] = [] |
| 1326 | try: |
| 1327 | from agent import AgentContext |
| 1328 | |
| 1329 | sessions = [ |
| 1330 | str(item.id) |
| 1331 | for item in AgentContext.all() |
| 1332 | if str(getattr(getattr(item.agent0, "config", None), "profile", "")) |
| 1333 | == profile_id |
| 1334 | and ( |
| 1335 | not project_name |
| 1336 | or projects.get_context_project_name(item) == project_name |
| 1337 | ) |
| 1338 | ] |
| 1339 | except Exception: |
| 1340 | pass |
| 1341 | |
| 1342 | model_config = _read_mapping( |
| 1343 | _profile_config_path(profile_id, "_model_config", project_name) |
| 1344 | ) |
| 1345 | return { |
| 1346 | "profile_id": profile_id, |
| 1347 | "deletable": state["deletable"], |
| 1348 | "origin": state["origin"], |
| 1349 | "project_name": project_name, |
| 1350 | "files": file_paths, |
| 1351 | "project_references": references, |
| 1352 | "active_sessions": sessions, |
| 1353 | "model_preset": str(model_config.get("model_preset") or ""), |
| 1354 | "contains": { |
| 1355 | name: (root / name).is_dir() |
| 1356 | for name in ("tools", "extensions", "skills", "assets", "plugins") |
| 1357 | }, |
| 1358 | } |
| 1359 | |
| 1360 | |
| 1361 | def stage_avatar(upload: Any) -> dict[str, Any]: |
| 1362 | if upload is None: |
| 1363 | raise ValueError("Choose an image to upload.") |
| 1364 | payload = upload.stream.read(MAX_AVATAR_BYTES + 1) |
| 1365 | if len(payload) > MAX_AVATAR_BYTES: |
| 1366 | raise ValueError("Avatar images must be 8 MB or smaller.") |
| 1367 | if not payload: |
| 1368 | raise ValueError("The uploaded image is empty.") |
| 1369 | |
| 1370 | from PIL import Image, ImageOps, UnidentifiedImageError |
| 1371 | |
| 1372 | try: |
| 1373 | with Image.open(BytesIO(payload)) as source: |
| 1374 | source_format = str(source.format or "").upper() |
| 1375 | if source_format not in {"PNG", "JPEG", "WEBP"}: |
| 1376 | raise ValueError("Avatar must be a PNG, JPEG, or WebP image.") |
| 1377 | if max(source.size) > MAX_AVATAR_DIMENSION: |
| 1378 | raise ValueError("Avatar dimensions must not exceed 4096 pixels.") |
| 1379 | source.load() |
| 1380 | normalized = ImageOps.exif_transpose(source) |
| 1381 | mode = "RGBA" if "A" in normalized.getbands() else "RGB" |
| 1382 | square = ImageOps.fit( |
| 1383 | normalized.convert(mode), |
| 1384 | (AVATAR_SIZE, AVATAR_SIZE), |
| 1385 | method=Image.Resampling.LANCZOS, |
| 1386 | ) |
| 1387 | output = BytesIO() |
| 1388 | square.save(output, format="WEBP", quality=88, method=6) |
| 1389 | except UnidentifiedImageError as exc: |
| 1390 | raise ValueError("Avatar must be a valid PNG, JPEG, or WebP image.") from exc |
| 1391 | except Image.DecompressionBombError as exc: |
| 1392 | raise ValueError("Avatar dimensions are too large.") from exc |
| 1393 | except OSError as exc: |
| 1394 | raise ValueError("Avatar must be a valid PNG, JPEG, or WebP image.") from exc |
| 1395 | |
| 1396 | _cleanup_staged_avatars() |
| 1397 | STAGED_AVATAR_ROOT.mkdir(parents=True, exist_ok=True) |
| 1398 | token = uuid4().hex |
| 1399 | path = staged_avatar_path(token) |
| 1400 | temporary = _stage_bytes(path.parent, path.name, output.getvalue()) |
| 1401 | os.replace(temporary, path) |
| 1402 | return {"token": token} |
| 1403 | |
| 1404 | |
| 1405 | def staged_avatar_path(token: str) -> Path: |
| 1406 | if not re.fullmatch(r"[0-9a-f]{32}", str(token or "")): |
| 1407 | raise ValueError("Invalid staged avatar token.") |
| 1408 | return STAGED_AVATAR_ROOT / f"{token}.webp" |
| 1409 | |
| 1410 | |
| 1411 | def _full_delete_plan( |
| 1412 | profile_id: str, |
| 1413 | context: Any | None = None, |
| 1414 | ) -> ChangePlan: |
| 1415 | project_name = _context_project_name(context) |
| 1416 | root = _profile_root(profile_id, project_name) |
| 1417 | plan = ChangePlan( |
| 1418 | profile_id=profile_id, |
| 1419 | project_name=project_name, |
| 1420 | remove_empty_root=True, |
| 1421 | ) |
| 1422 | if not root.is_dir(): |
| 1423 | return plan |
| 1424 | for path in sorted(root.rglob("*")): |
| 1425 | if path.is_symlink(): |
| 1426 | raise ValueError("Profile symlinks must be removed manually before deletion.") |
| 1427 | if path.is_file(): |
| 1428 | plan.delete(path) |
| 1429 | return plan |
| 1430 | |
| 1431 | |
| 1432 | def _validate_plan_paths(root: Path, changes: list[FileChange]) -> None: |
| 1433 | intended_root = root.absolute() |
| 1434 | if root.is_symlink(): |
| 1435 | raise ValueError("The profile directory cannot be a symlink.") |
| 1436 | for change in changes: |
| 1437 | if change.action not in {"write", "delete"}: |
| 1438 | raise ValueError("Invalid change-plan action.") |
| 1439 | resolved = change.path.resolve(strict=False) |
| 1440 | if not files.is_in_dir(str(resolved), str(intended_root)): |
| 1441 | raise ValueError("A planned path is outside the selected profile directory.") |
| 1442 | cursor = change.path.parent |
| 1443 | while files.is_in_dir(str(cursor), str(intended_root)): |
| 1444 | if cursor.is_symlink(): |
| 1445 | raise ValueError("Profile paths cannot traverse symlinks.") |
| 1446 | if cursor == root: |
| 1447 | break |
| 1448 | cursor = cursor.parent |
| 1449 | if change.action == "delete" and change.path.exists() and not ( |
| 1450 | change.path.is_file() or change.path.is_symlink() |
| 1451 | ): |
| 1452 | raise ValueError("Change plans delete files, not directories.") |
| 1453 | |
| 1454 | |
| 1455 | def _ensure_parent(parent: Path, created: set[Path]) -> None: |
| 1456 | missing: list[Path] = [] |
| 1457 | cursor = parent |
| 1458 | while not cursor.exists(): |
| 1459 | missing.append(cursor) |
| 1460 | cursor = cursor.parent |
| 1461 | for directory in reversed(missing): |
| 1462 | directory.mkdir(exist_ok=True) |
| 1463 | created.add(directory) |
| 1464 | |
| 1465 | |
| 1466 | def _stage_bytes(directory: Path, name: str, content: bytes) -> Path: |
| 1467 | target = directory / name |
| 1468 | descriptor, temporary = tempfile.mkstemp( |
| 1469 | prefix=f".{name}.", suffix=".tmp", dir=directory |
| 1470 | ) |
| 1471 | path = Path(temporary) |
| 1472 | try: |
| 1473 | os.fchmod( |
| 1474 | descriptor, |
| 1475 | target.stat().st_mode & 0o777 if target.exists() else 0o644, |
| 1476 | ) |
| 1477 | with os.fdopen(descriptor, "wb") as handle: |
| 1478 | handle.write(content) |
| 1479 | handle.flush() |
| 1480 | os.fsync(handle.fileno()) |
| 1481 | return path |
| 1482 | except Exception: |
| 1483 | path.unlink(missing_ok=True) |
| 1484 | raise |
| 1485 | |
| 1486 | |
| 1487 | def _restore_snapshots(snapshots: dict[Path, bytes | None], root: Path) -> None: |
| 1488 | for path, content in snapshots.items(): |
| 1489 | if content is None: |
| 1490 | if path.is_file() or path.is_symlink(): |
| 1491 | path.unlink() |
| 1492 | continue |
| 1493 | path.parent.mkdir(parents=True, exist_ok=True) |
| 1494 | temporary = _stage_bytes(path.parent, path.name, content) |
| 1495 | os.replace(temporary, path) |
| 1496 | |
| 1497 | |
| 1498 | def _prune_empty_directories(root: Path) -> None: |
| 1499 | if not root.is_dir(): |
| 1500 | return |
| 1501 | for directory in sorted( |
| 1502 | (path for path in root.rglob("*") if path.is_dir()), |
| 1503 | key=lambda path: len(path.parts), |
| 1504 | reverse=True, |
| 1505 | ): |
| 1506 | try: |
| 1507 | directory.rmdir() |
| 1508 | except OSError: |
| 1509 | pass |
| 1510 | |
| 1511 | |
| 1512 | def _fsync_directory(path: Path) -> None: |
| 1513 | descriptor = os.open(path, os.O_RDONLY) |
| 1514 | try: |
| 1515 | os.fsync(descriptor) |
| 1516 | finally: |
| 1517 | os.close(descriptor) |
| 1518 | |
| 1519 | |
| 1520 | def _invalidate_profile_caches() -> None: |
| 1521 | cache.clear(subagents.PATHS_CACHE_AREA) |
| 1522 | plugins.clear_plugin_cache( |
| 1523 | ["_agent_editor", "_model_config", tool_policy.PLUGIN_NAME, skills.ACTIVE_SKILLS_PLUGIN_NAME] |
| 1524 | ) |
| 1525 | |
| 1526 | |
| 1527 | def _cleanup_staged_avatars() -> None: |
| 1528 | if not STAGED_AVATAR_ROOT.is_dir(): |
| 1529 | return |
| 1530 | cutoff = time.time() - 3600 |
| 1531 | for path in STAGED_AVATAR_ROOT.glob("*.webp"): |
| 1532 | try: |
| 1533 | if path.stat().st_mtime < cutoff: |
| 1534 | path.unlink() |
| 1535 | except OSError: |
| 1536 | pass |
| 1537 | |
| 1538 | |
| 1539 | def _profile_reference_paths(profile_id: str, project_name: str = "") -> list[str]: |
| 1540 | root = ( |
| 1541 | Path(projects.get_project_meta(project_name)) |
| 1542 | if project_name |
| 1543 | else Path(files.get_abs_path("usr", "projects")) |
| 1544 | ) |
| 1545 | if not root.is_dir(): |
| 1546 | return [] |
| 1547 | references: list[str] = [] |
| 1548 | for suffix in ("*.json", "*.yaml", "*.yml"): |
| 1549 | pattern = f"**/{suffix}" if project_name else f"*/.a0proj/**/{suffix}" |
| 1550 | for path in root.glob(pattern): |
| 1551 | try: |
| 1552 | if profile_id in path.read_text(encoding="utf-8"): |
| 1553 | references.append(_relative_source(path)) |
| 1554 | except (OSError, UnicodeError): |
| 1555 | pass |
| 1556 | return sorted(set(references)) |