"""The assistant's ability to change what it is: its own playbooks, its own runtime settings, and (in a source checkout only) its own source files. WHAT THIS IS NOT ---------------- Not a way to skip human review. Every function here is either read-only (the `preview_*`/`diff_text` functions, safe to call before anything is approved) or is only ever reached after the per-call approval round-trip in chat.py — and `edit_playbook`/`edit_settings`/`edit_source` are *always* gated that way, regardless of the global `action_tool_policy` setting (see `ALWAYS_ASK_TOOLS`). This module is the mechanism the approval actually acts on, layered: 1. Consent edit_* tools are ACTION tools with a hardcoded approval floor — chat.py pauses for Approve/Deny even when the global policy is "allow", so flipping that setting for an unrelated tool (web_search, say) can never silently unlock unattended self-modification too. 2. Real diff the human reviews a diff computed here, server-side, from what is actually on disk versus the model's proposed `new_content` — never a diff or description the model wrote itself. A model can misdescribe a change; it cannot make difflib misreport one. 3. Boundary edit_source is confined to one root (settings.project_root) via the same realpath + `Path.parents` check that closed a sibling-directory bypass in main.py's /icons/image, plus a denylist of dangerous subtrees inside that root (.git, the venv, node_modules, build output, runtime state). 4. Audit trail every applied source edit is committed to git (best-effort; a missing git binary or a non-repo root degrades the result to `commit: None`, it never blocks the write). This is a reversibility net, not a substitute for layer 1 — the approval already happened before anything is written. 5. Size caps MAX_FILE_CHARS/MAX_DIFF_CHARS bound what a single call can submit or what the approval UI has to render, mirroring code_run.py's MAX_SOURCE/MAX_OUTPUT. A file edit does not hot-reload the running process. Python does not re-import a changed module on its own, and the frontend's production build is the static `dist/` the backend serves — editing interface/web/src/*.jsx only affects a developer's own `npm run dev` session, if one happens to be running. Every successful edit_source result says so explicitly, because both the model and the human reviewing it will otherwise reasonably expect an instant effect that does not happen. """ from __future__ import annotations import difflib import os import subprocess from pathlib import Path from typing import Any from .memory.store import store from .nexus_config import settings from .playbooks.store import playbook_store from . import playbook_manager MAX_FILE_CHARS = 200_000 # chars of proposed file content accepted MAX_DIFF_CHARS = 20_000 # chars of diff text shown/returned, clipped like code_run.MAX_OUTPUT # Subdirectories of settings.project_root that edit_source must never touch, # even though they're inside the one allowed root. `data`/`runtime` land here # in a source checkout (see nexus_config.py DATA_DIR/RUNTIME_DIR) and are owned # by their own tools (edit_settings, the memory store), not raw file writes. _DENYLIST_SUBDIRS = frozenset({ ".git", "Promethean", "node_modules", "dist", "__pycache__", "runtime", "data", ".venv", "venv", }) # Tools that must always pause for human approval, regardless of the global # action_tool_policy setting. Shared by tools.py (ACTION_TOOLS membership) and # chat.py (the approval-gating condition) so there is one definition of the # floor, not two that could drift apart. ALWAYS_ASK_TOOLS = frozenset({"edit_playbook", "edit_settings", "edit_source"}) # Tool names whose approval payload gets a computed, human-readable preview # attached before the human ever sees the Approve/Deny prompt. PREVIEWABLE = frozenset({"edit_source", "edit_playbook", "edit_settings"}) WINDOWS = os.name == "nt" _NO_WINDOW = subprocess.CREATE_NO_WINDOW if WINDOWS else 0 class PathError(ValueError): pass def _resolve_source_path(rel_path: str) -> Path: """A path the model gave us, resolved and boundary-checked against settings.project_root. Raises PathError with a human-readable reason on any rejection — the same message is shown in the pre-approval preview and returned as the tool's error, so both audiences see exactly why.""" raw = (rel_path or "").strip() if not raw: raise PathError("path is required") # Cheap rejection before ever touching the filesystem: an absolute path or # a Windows drive prefix is never a legitimate "file in this project" path. if raw.startswith(("/", "\\")) or (len(raw) > 1 and raw[1] == ":"): raise PathError(f"path must be relative to the project root, not absolute: {raw!r}") root = Path(os.path.realpath(str(settings.project_root))) candidate = root / raw real = Path(os.path.realpath(str(candidate))) # Same real == root or root in real.parents pattern as # icons/compositor.py::_is_allowed_path — a bare str.startswith() here # would let a sibling directory that merely shares a prefix pass, exactly # the bug just fixed in main.py's /icons/image. if not (real == root or root in real.parents): raise PathError(f"path escapes the project root: {raw!r}") try: top = real.relative_to(root).parts[0] except (ValueError, IndexError): top = "" if top in _DENYLIST_SUBDIRS: raise PathError(f"{top}/ is off-limits to edit_source — use its own tool if there is one") return real def source_checkout_required() -> None: """Raise if this install has no live, editable source tree to write into. In a wheel install, synapse/ lives inside site-packages with no sibling repo — settings.project_root would just be the installed package dir, and there is nothing to commit into. Same precedent as nexusos_cli/ncp.py refusing to start the Vite dev server in a wheel install.""" if not settings.source_checkout: raise RuntimeError( "edit_source is unavailable — this is not a source checkout, so " "there is no live project tree to edit or commit into." ) def diff_text(path_display: str, before: str, after: str) -> str: lines = difflib.unified_diff( before.splitlines(keepends=True), after.splitlines(keepends=True), fromfile=path_display, tofile=path_display, ) text = "".join(lines) if len(text) <= MAX_DIFF_CHARS: return text return text[:MAX_DIFF_CHARS] + f"\n... [truncated at {MAX_DIFF_CHARS} characters, {len(text)} total]" def preview_source_edit(path: str, new_content: str) -> dict: """Read-only: never raises. Computes the real diff between what's on disk and the proposed new_content, so the approval UI shows ground truth before anything is written. Degrades to {"ok": False, "error": ...} on any rejection (bad path, wheel install, oversized content) rather than crashing the approval payload — the human still sees why it would fail.""" try: source_checkout_required() real = _resolve_source_path(path) new_content = new_content or "" if len(new_content) > MAX_FILE_CHARS: return {"ok": False, "error": f"new_content is over {MAX_FILE_CHARS} characters"} before = real.read_text(encoding="utf-8") if real.is_file() else "" display = str(real.relative_to(Path(os.path.realpath(str(settings.project_root))))) return { "ok": True, "path": display, "is_new_file": not real.is_file(), "diff": diff_text(display, before, new_content), } except (PathError, RuntimeError) as e: return {"ok": False, "error": str(e)} except Exception as e: return {"ok": False, "error": f"could not compute preview: {e}"} def _git_commit(real_path: Path, summary: str) -> str | None: """Best-effort audit-trail commit. Never raises, never undoes the write that already happened — a missing git binary or a project root that isn't a repo just means commit stays None.""" root = str(settings.project_root) try: rel = str(real_path.relative_to(Path(os.path.realpath(root)))) message = f"self-edit: {(summary or rel)[:180]}" subprocess.run( ["git", "add", "--", rel], cwd=root, check=True, capture_output=True, creationflags=_NO_WINDOW, ) subprocess.run( ["git", "commit", "-m", message, "--", rel], cwd=root, check=True, capture_output=True, creationflags=_NO_WINDOW, ) sha = subprocess.run( ["git", "rev-parse", "--short", "HEAD"], cwd=root, check=True, capture_output=True, text=True, creationflags=_NO_WINDOW, ) return sha.stdout.strip() or None except Exception: return None def apply_source_edit(path: str, new_content: str, summary: str = "") -> dict: """Write an approved edit_source call. Re-validates everything preview did — never trust that nothing changed between preview and approval — then writes, diffs against the pre-write content, and commits.""" try: source_checkout_required() real = _resolve_source_path(path) new_content = new_content or "" if len(new_content) > MAX_FILE_CHARS: return {"ok": False, "error": f"new_content is over {MAX_FILE_CHARS} characters"} except (PathError, RuntimeError) as e: return {"ok": False, "error": str(e)} before = real.read_text(encoding="utf-8") if real.is_file() else "" display = str(real.relative_to(Path(os.path.realpath(str(settings.project_root))))) real.parent.mkdir(parents=True, exist_ok=True) real.write_text(new_content, encoding="utf-8") return { "ok": True, "path": display, "diff": diff_text(display, before, new_content), "commit": _git_commit(real, summary), } def preview_playbook_edit(args: dict) -> dict: """Read-only before/after preview for edit_playbook. Mirrors the merge semantics of playbook_manager.persist_playbook(merge=True) without writing anything, so the approval UI shows exactly what will actually change.""" try: pb_id = str(args.get("id") or "") existing = playbook_store.get_playbook(pb_id) if pb_id else None make_active = bool(args.get("make_active")) fields = {} for key in ("title", "goal", "instructions", "tags", "tools", "model"): if key in args and args[key] is not None: fields[key] = args[key] elif existing is not None: fields[key] = getattr(existing, key) else: fields[key] = [] if key in ("tags", "tools") else "" if existing is None and not (fields["title"] and fields["goal"] and fields["instructions"]): return {"ok": False, "error": "title, goal, and instructions are required to create a new playbook"} is_new = existing is None becomes_main = make_active or (is_new and not playbook_store.all_playbooks()) return { "ok": True, "is_new": is_new, "before": existing.model_dump() if existing else None, "after": fields, "becomes_main_playbook": becomes_main, } except Exception as e: return {"ok": False, "error": f"could not compute preview: {e}"} def preview_settings_edit(args: dict) -> dict: """Read-only before/after preview for edit_settings, split into keys that will actually apply versus ones update_settings would silently ignore — the approval UI must never imply an unknown key will take effect.""" try: changes = args.get("changes") or {} current = store.get_settings() applied: dict[str, dict[str, Any]] = {} ignored_unknown: list[str] = [] for key, value in changes.items(): if key in store._SETTINGS_DEFAULTS: applied[key] = {"before": current.get(key), "after": value} else: ignored_unknown.append(key) return { "ok": True, "applied": applied, "ignored_unknown": ignored_unknown, "policy_change": "action_tool_policy" in applied, "system_prompt_change": "system_prompt" in applied, } except Exception as e: return {"ok": False, "error": f"could not compute preview: {e}"} def preview_for(name: str, args: dict) -> dict: """Single dispatch entry point chat.py calls to enrich an approval payload.""" if name == "edit_source": return preview_source_edit(args.get("path", ""), args.get("new_content", "")) if name == "edit_playbook": return preview_playbook_edit(args) if name == "edit_settings": return preview_settings_edit(args) return {"ok": False, "error": f"no previewer for {name}"}