feat: Add plan mode to the chat agent (#638)

* feat: Add plan mode to the chat agent

Adds a plan mode: the agent investigates read-only, proposes a checklist, and
waits for approval before changing anything. On approval it runs with full
tools and checks items off as it goes. Enforcement reuses the existing
disabled_tools gate.

Includes a slash command: `/plan [on|off]` (and `/toggle plan`) to flip the
plan toggle from the chat input.

- src/tool_security.py, src/mcp_manager.py: read-only allowlist (tools + MCP).
- src/agent_loop.py, routes/chat_routes.py: union the disabled set, prepend the
  plan directive, force agent mode.
- static/: plan toggle pill, Approve & Run, dockable plan window, task-list
  checkboxes, and the /plan slash command.
- tests/test_plan_mode.py.

* Plan mode: persistent re-referenceable plan + agent write-back

Three improvements so a long plan survives a weak model and stays in reach:

1. Re-reference the plan (out-of-context fix). On the execution turn the frontend
   sends the approved checklist back (`approved_plan`); the backend pins it as a
   top-of-context `## ACTIVE PLAN` system note (kept by the context trimmer), so
   the agent can always re-read the plan instead of losing the thread on a long
   run. New `build_active_plan_note()` (unit-tested).

2. Re-open / dock the plan anytime. The plan checklist is stored per-session
   (localStorage). When a plan exists, the plan-mode button opens a small menu
   ("Show plan" / "Plan mode: On/Off") that re-opens the side-dockable plan
   window — so it can stay docked while the agent works. The window live-refreshes
   as the plan changes.

3. Agent write-back: new `update_plan` tool. The agent calls it to tick steps
   `- [x]` after finishing them, or to revise steps when the user asks. Marker
   tool (no I/O) → `plan_update` SSE event → the stored plan + docked window
   update live. The ACTIVE PLAN note instructs the agent to use it.

Backend: src/agent_loop.py (param + pin + note builder + emit + prompt blurb),
src/tool_execution.py (update_plan handler), routes/chat_routes.py (parse
`approved_plan`, relay `plan_update`), registration in tool_schemas / agent_tools
/ tool_index (always-available, not admin-gated).
Frontend: static/js/chat.js (plan store, send `approved_plan`, handle
`plan_update`, capture restated checklists), static/app.js (plan-button menu),
static/js/planWindow.js (`isPlanWindowOpen`), static/js/storage.js (PLAN key).
Tests: tests/test_plan_mode.py (plan-note), tests/test_update_plan_tool.py.

* Plan mode: drop bash/python, rely on read-only discovery tools

Shell can mutate (write files, hit the network) and can't be constrained to
read-only at the tool layer, so plan mode no longer relies on a prompt to keep
it well-behaved — bash/python are removed from the read-only allowlist and added
to the fail-closed block set. Discovery is covered by the dedicated read-only
tools (read_file, grep, glob, ls) instead.

Rewrites the plan-mode directive to state shell is disabled and lists the
available read-only tools positively. Addresses review feedback on #638.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Comment: note _MCP_READONLY_VERBS are prefixes not whole words

Clarifies that entries like "summar" are intentional stems matched via
startswith (covers summarise/summarize/summary), not typos. Addresses review
feedback on #638.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Plan mode: clarify why gating inverts the allowlist into a denylist

Rename _PLAN_MODE_FALLBACK_BLOCK -> _PLAN_MODE_KNOWN_MUTATORS and rewrite the
comments. The tool gate is a denylist (disabled_tools); plan mode's policy is an
allowlist, so it returns the inverse (all known tool names minus the allowlist).
The static mutator set is a backstop for the schema-derived name list, which
misses XML-only tools and can fail to import. Addresses review feedback on #638.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* Plan mode: stop hardcoding the read-only tool list in the directive

The model is already shown its available (read-only) tools by _assemble_prompt,
which removes every disabled tool. Enumerating them again in the directive only
duplicated that list and would drift as tools change. Point at the tools listed
below instead. Addresses review feedback on #638.

src/agent_loop.py

@@ -19,7 +19,7 @@ from src.llm_core import stream_llm, stream_llm_with_fallback, _is_ollama_native
 from src.model_context import estimate_tokens
 from src.settings import get_setting
 from src.prompt_security import untrusted_context_message
-from src.tool_security import blocked_tools_for_owner
+from src.tool_security import blocked_tools_for_owner, plan_mode_disabled_tools
 from src.agent_tools import (
     parse_tool_blocks,
     strip_tool_blocks,
@@ -336,6 +336,7 @@ If the user asks for a reminder/alarm before the event, pass `reminder_minutes`
     "pipeline": "- ```pipeline``` — Run a multi-step AI pipeline. Args (JSON) with ordered steps, each specifying a model and prompt. Use for complex workflows.",
     "ui_control": "- ```ui_control``` — Control the UI: toggle tools on/off, OPEN PANELS, open email reply drafts, switch models, change themes. Commands: `toggle <name> on/off` (names: bash/shell, web/search, research, incognito, document_editor/documents), `open_panel <name>` (panels: documents, gallery, email, sessions, notes, memories/brain, skills, settings, cookbook), `open_email_reply <uid> <folder> <reply|reply-all|ai-reply>` (opens an email compose document, does NOT send), `set_mode agent/chat`, `switch_model <name>`, `set_theme <preset>`, `create_theme <name> <bg> <fg> <panel> <border> <accent>` (optional key=val for advanced colors AND background effects: bgPattern=<none|dots|synapse|rain|constellations|perlin-flow|petals|sparkles|embers>, bgEffectColor=#RRGGBB, bgEffectIntensity=<num>, bgEffectSize=<num>, frosted=true|false). \"open documents\" / \"open library\" / \"show gallery\" / \"open inbox\" / \"open notes\" / \"open cookbook\" all map to `open_panel <name>`. Theme presets: dark, light, midnight, paper, cyberpunk, retrowave, forest, ocean, ume, copper, terminal, organs, lavender, gpt, claude, cute.",
     "ask_user": "- ```ask_user``` — Ask the user a multiple-choice question when the task is genuinely ambiguous and the answer changes what you do next (pick an approach, confirm an assumption, choose a target). Args (JSON): {\"question\": \"...\", \"options\": [{\"label\": \"...\", \"description\": \"...\"?}, ...], \"multi\": false?}. 2-6 options. The user gets clickable buttons; calling this ENDS your turn and their choice comes back as your next message. Prefer sensible defaults — only ask when you truly can't proceed well without their input.",
+    "update_plan": "- ```update_plan``` — While executing an approved plan, write the plan back: tick steps done or revise them. Args (JSON): {\"plan\": \"- [x] done step\\n- [ ] next step\"}. Always pass the COMPLETE checklist, not a diff. Call it after finishing each step (mark it `- [x]`) and whenever the user asks to change the plan. The user's docked plan window updates live. Does nothing if there's no active plan.",
     "list_served_models": "- ```list_served_models``` — Show what the Cookbook (LLM-serving subsystem) is currently running. NO args. Use this for ANY 'what's running' / 'what's serving' / 'show my cookbook' / 'is anything up' query. DO NOT shell out (`ps aux`, `docker ps`, etc.) — this tool is the source of truth. Failed serve tasks include recent logs plus diagnosis/retry suggestions; use those suggestions to call `serve_model` again with an adjusted command when appropriate.",
     "stop_served_model": "- ```stop_served_model``` — Stop a running model server. Args (JSON): {\"session_id\": \"<from list_served_models>\"}. Use for 'kill my cookbook' / 'stop the model' / 'shut down vLLM'.",
     "tail_serve_output": "- ```tail_serve_output``` — Read the actual tmux stderr/traceback of a CURRENTLY failing cookbook task. Args (JSON): {\"session_id\": \"<from list_served_models>\", \"tail\": 150?}. **Use ONLY after** you just launched something via `serve_model` AND `list_served_models` reports YOUR new task as `crashed`/`error`. DO NOT use it on old stopped/completed download tasks (they're historical noise — won't predict whether a new launch succeeds). DO NOT call it before launching a fresh attempt. When you do call it, bump `tail` to 400+ only if the visible error references 'see root cause above'.",
@@ -1372,6 +1373,53 @@ def _empty_response_fallback(
     return _error_msg, f'data: {json.dumps({"delta": _error_msg})}\n\n'
 
 
+PLAN_MODE_DIRECTIVE = (
+    "## PLAN MODE — OVERRIDES EVERYTHING ELSE BELOW\n"
+    "You are in PLAN MODE. Your ONLY job this turn is to PROPOSE a plan. You have "
+    "NOT done anything yet. Do NOT claim you created, wrote, ran, sent, or changed "
+    "anything — that would be a lie.\n"
+    "\n"
+    "ABSOLUTE RULE — DO NOT MUTATE ANYTHING. Every write/state-changing tool, "
+    "including the shell (`bash`/`python`), is disabled this turn and will be "
+    "rejected — only read-only tools remain available. Use the read-only tools "
+    "listed below (read files, search code, browse the project, web lookups) to "
+    "ground the plan. If the task is 'write a file', your plan is to DESCRIBE "
+    "writing it — you do NOT write it now.\n"
+    "\n"
+    "OUTPUT: present the plan as a GitHub-style checklist, one concrete step per line:\n"
+    "- [ ] first action you will take once approved\n"
+    "- [ ] next action\n"
+    "Each item = one concrete action (file to create/edit, command to run, side "
+    "effect). Do not execute. Do not end with 'Done' or anything implying the work "
+    "is finished. End your turn with the checklist."
+)
+
+
+def build_active_plan_note(approved_plan: str) -> str:
+    """System note that pins an approved plan during execution.
+
+    Sent back by the frontend each turn so a long plan on a weak model survives
+    history truncation — the agent can always re-read it. Returns "" for empty
+    input.
+    """
+    if not approved_plan or not approved_plan.strip():
+        return ""
+    return (
+        "## ACTIVE PLAN (approved — execute this)\n"
+        "You are executing a plan the user already approved. THE FULL PLAN IS "
+        "BELOW — it is always provided here every turn. Do NOT say you lost it, "
+        "and do NOT look for it in tasks, notes, memory, files, or the API; just "
+        "read it below. Work through it IN ORDER. After finishing each step, call "
+        "the `update_plan` tool with the full checklist and that step marked "
+        "`- [x]` so progress stays visible in the user's plan window. If the user "
+        "asks to change the plan, call `update_plan` with the revised checklist. "
+        "Do the next unchecked item until all are done. Do not skip, reorder, or "
+        "invent steps; if a step is genuinely impossible, say so and stop.\n\n"
+        "Current plan:\n"
+        + approved_plan.strip()
+    )
+
+
 async def stream_agent_loop(
     endpoint_url: str,
     model: str,
@@ -1390,6 +1438,8 @@ async def stream_agent_loop(
     relevant_tools: Optional[Set[str]] = None,
     fallbacks: Optional[List[tuple]] = None,
     workspace: Optional[str] = None,
+    plan_mode: bool = False,
+    approved_plan: Optional[str] = None,
     _is_teacher_run: bool = False,
 ) -> AsyncGenerator[str, None]:
     """Streaming agent loop generator.
@@ -1413,6 +1463,13 @@ async def stream_agent_loop(
         # public/non-admin users rather than trying to enumerate every tool.
         mcp_mgr = None
 
+    if plan_mode:
+        # Plan mode: investigate read-only, propose a plan, don't execute. The
+        # route also unions the read-only-disabled set, but enforce here too so
+        # the loop is safe regardless of caller. MCP stays available but is
+        # filtered to read-only tools below (after the disabled map is loaded).
+        disabled_tools.update(plan_mode_disabled_tools())
+
     _t0 = time.time()
     _needs_admin = _detect_admin_intent(messages)
     _last_user = _extract_last_user_message(messages)
@@ -1420,6 +1477,13 @@ async def stream_agent_loop(
     # not just the latest message, so short follow-ups don't drop just-used tools.
     _retrieval_query = _recent_context_for_retrieval(messages) or _last_user
     _mcp_disabled_map = _load_mcp_disabled_map() if mcp_mgr else {}
+    if plan_mode and mcp_mgr:
+        # Allow read-only MCP tools to investigate, block write/unknown ones:
+        # hide them from the schemas AND reject them at runtime by qualified name.
+        _mcp_block_map, _mcp_block_q = mcp_mgr.plan_mode_blocked_mcp()
+        for _sid, _names in _mcp_block_map.items():
+            _mcp_disabled_map.setdefault(_sid, set()).update(_names)
+        disabled_tools.update(_mcp_block_q)
     prep_timings["request_setup"] = time.time() - _t0
 
     # RAG-based tool selection: retrieve relevant tools for this query.
@@ -1577,6 +1641,27 @@ async def stream_agent_loop(
         else:
             messages.insert(0, {"role": "system", "content": _ws_note})
         logger.info("[workspace] active for this turn: %s", workspace)
+    if plan_mode:
+        # Steer the model to investigate-then-propose. Hard tool gating handles
+        # every write path except shell; this directive is what keeps the
+        # intentionally-allowed bash/python read-only, so it must DOMINATE. Put
+        # it at the very TOP of the system prompt (the base prompt is large and
+        # action-oriented — appending buried it, and small models ignored it).
+        if messages and messages[0].get("role") == "system":
+            messages[0]["content"] = PLAN_MODE_DIRECTIVE + "\n\n" + (messages[0].get("content") or "")
+        else:
+            messages.insert(0, {"role": "system", "content": PLAN_MODE_DIRECTIVE})
+    elif approved_plan and approved_plan.strip():
+        # EXECUTING an approved plan. Pin the checklist as a top-of-context
+        # system note so a long plan on a weak model survives history
+        # truncation — the agent can always re-read the plan instead of losing
+        # the thread. (The first system message is kept by the context trimmer.)
+        _plan_note = build_active_plan_note(approved_plan)
+        if messages and messages[0].get("role") == "system":
+            messages[0]["content"] = _plan_note + "\n\n" + (messages[0].get("content") or "")
+        else:
+            messages.insert(0, {"role": "system", "content": _plan_note})
+        logger.info("[plan] pinned approved plan (%d chars) for execution turn", len(approved_plan))
     prep_timings["prompt_build"] = time.time() - _t2
 
     _t3 = time.time()
@@ -2287,6 +2372,14 @@ async def stream_agent_loop(
                 )
                 _awaiting_user = True
 
+            # update_plan: agent wrote back to the plan (ticked a step / revised).
+            # Push it to the frontend so the stored plan + docked window update
+            # live. Does NOT end the turn — the agent keeps working.
+            if "plan_update" in result:
+                yield (
+                    f'data: {json.dumps({"type": "plan_update", "data": result["plan_update"]})}\n\n'
+                )
+
             # Build output for frontend tool bubble.
             # Document tools get a short summary — content goes to the editor panel.
             output_text = ""

src/agent_tools.py

@@ -34,7 +34,7 @@ TOOL_TAGS = {"bash", "python", "web_search", "web_fetch", "read_file", "write_fi
              "send_to_session",
              "pipeline",
              "manage_session", "manage_memory", "list_models",
-             "ui_control", "generate_image", "ask_user",
+             "ui_control", "generate_image", "ask_user", "update_plan",
              "manage_tasks", "api_call", "ask_teacher", "manage_skills",
              "suggest_document",
              "manage_endpoints", "manage_mcp", "manage_webhooks",

src/mcp_manager.py

@@ -9,7 +9,7 @@ import json
 import logging
 import os
 import re
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 logger = logging.getLogger(__name__)
 
@@ -90,6 +90,44 @@ def _format_mcp_params(input_schema: Any) -> str:
     return hint
 
 
+# Tool-name prefixes that denote a read-only/inspection operation. Used to
+# classify MCP tools for plan mode when the server provides no readOnlyHint.
+# These are PREFIXES, not whole words (matched via str.startswith below), so a
+# stem like "summar" intentionally covers "summarise"/"summarize"/"summary".
+_MCP_READONLY_VERBS = (
+    "list", "get", "read", "search", "fetch", "query", "find", "describe",
+    "show", "view", "lookup", "count", "status", "info", "inspect", "summar",
+)
+
+
+def mcp_tool_is_readonly(tool: Dict) -> bool:
+    """Classify an MCP tool as safe (non-mutating) for plan mode.
+
+    Prefer the server's own annotations (readOnlyHint / destructiveHint). When
+    absent, fall back to a tool-name verb heuristic, and FAIL CLOSED (treat as
+    write) for anything that doesn't clearly read — plan mode must not run a
+    write tool just because its intent is ambiguous.
+    """
+    ann = tool.get("annotations")
+    # annotations may be a dict or a pydantic model
+    read_hint = None
+    destructive = None
+    if ann is not None:
+        if isinstance(ann, dict):
+            read_hint = ann.get("readOnlyHint")
+            destructive = ann.get("destructiveHint")
+        else:
+            read_hint = getattr(ann, "readOnlyHint", None)
+            destructive = getattr(ann, "destructiveHint", None)
+    if read_hint is True:
+        return True
+    if read_hint is False or destructive is True:
+        return False
+    # No usable hint — heuristic on the tool name's leading verb.
+    name = (tool.get("name") or "").lower()
+    return name.startswith(_MCP_READONLY_VERBS)
+
+
 class McpManager:
     """Manages MCP server connections and tool routing."""
 
@@ -170,6 +208,10 @@ class McpManager:
                     "name": tool.name,
                     "description": tool.description or "",
                     "input_schema": tool.inputSchema if hasattr(tool, 'inputSchema') else {},
+                    # MCP tool annotations (readOnlyHint / destructiveHint) drive
+                    # plan-mode read-only gating. Absent on many servers, so we
+                    # fall back to a name heuristic in mcp_tool_is_readonly().
+                    "annotations": getattr(tool, 'annotations', None),
                 })
 
             self._sessions[server_id] = session
@@ -227,6 +269,10 @@ class McpManager:
                     "name": tool.name,
                     "description": tool.description or "",
                     "input_schema": tool.inputSchema if hasattr(tool, 'inputSchema') else {},
+                    # MCP tool annotations (readOnlyHint / destructiveHint) drive
+                    # plan-mode read-only gating. Absent on many servers, so we
+                    # fall back to a name heuristic in mcp_tool_is_readonly().
+                    "annotations": getattr(tool, 'annotations', None),
                 })
 
             self._sessions[server_id] = session
@@ -537,6 +583,24 @@ class McpManager:
                 })
         return result
 
+    def plan_mode_blocked_mcp(self) -> Tuple[Dict[str, Set[str]], Set[str]]:
+        """Plan mode: block every MCP tool that isn't clearly read-only.
+
+        Returns (disabled_map, qualified_names):
+          - disabled_map: {server_id: {tool_name, ...}} to hide write tools from
+            the prompt/schemas (merged into the existing mcp_disabled_map).
+          - qualified_names: {"mcp__<server>__<tool>", ...} for runtime rejection
+            in execute_tool_block (which matches the qualified name).
+        """
+        disabled_map: Dict[str, Set[str]] = {}
+        qualified: Set[str] = set()
+        for server_id, tools in self._tools.items():
+            for tool in tools:
+                if not mcp_tool_is_readonly(tool):
+                    disabled_map.setdefault(server_id, set()).add(tool["name"])
+                    qualified.add(f"mcp__{server_id}__{tool['name']}")
+        return disabled_map, qualified
+
     def is_builtin(self, server_id: str) -> bool:
         """Check if a server is a built-in (auto-registered) server."""
         return server_id.startswith("builtin_") or server_id in {

src/tool_execution.py

@@ -1263,6 +1263,41 @@ async def execute_tool_block(
         logger.info("Tool executed: %s (%d options, multi=%s)", desc, len(options), multi)
         return desc, result
 
+    # update_plan: the agent writes back to the active plan — tick an item done
+    # or revise steps (e.g. when the user asks to change something). Pure UI
+    # marker: returns a `plan_update` payload the agent loop turns into a
+    # `plan_update` SSE event; the frontend replaces the stored plan and refreshes
+    # the docked plan window. Does NOT end the turn.
+    if tool == "update_plan":
+        import json as _json
+        raw = (content or "").strip()
+        plan = ""
+        try:
+            parsed = _json.loads(raw) if raw else {}
+        except (ValueError, TypeError):
+            parsed = {}
+        if isinstance(parsed, dict) and parsed.get("plan"):
+            plan = str(parsed.get("plan", "")).strip()
+        else:
+            # Plain-string call (raw checklist) or JSON without a usable `plan`.
+            plan = raw
+        if not plan:
+            return "update_plan: invalid", {
+                "error": "update_plan needs a non-empty `plan` (the full updated checklist as markdown).",
+                "exit_code": 1,
+            }
+        plan = plan[:8192]
+        done = plan.count("- [x]") + plan.count("- [X]")
+        total = done + plan.count("- [ ]")
+        desc = f"update_plan: {done}/{total} done" if total else "update_plan"
+        result = {
+            "plan_update": {"plan": plan},
+            "output": f"Plan updated ({done}/{total} steps complete)." if total else "Plan updated.",
+            "exit_code": 0,
+        }
+        logger.info("Tool executed: %s", desc)
+        return desc, result
+
     # Background execution: a `bash` block whose first line is the `#!bg`
     # marker runs DETACHED — returns a job id immediately so the chat stream
     # isn't held open for a multi-minute install/ffmpeg/download. The always-on

src/tool_index.py

@@ -55,6 +55,8 @@ ALWAYS_AVAILABLE = frozenset({
     # Ask the user a multiple-choice question for a decision/clarification.
     # Always reachable so the agent can pause and ask at any point.
     "ask_user",
+    # Write back to the active plan (tick steps done / revise) during execution.
+    "update_plan",
 })
 
 # Tools that the Personal Assistant always has access to during scheduled
@@ -115,6 +117,7 @@ BUILTIN_TOOL_DESCRIPTIONS: Dict[str, str] = {
     "send_to_session": "Send a message to another chat. Cross-chat communication.",
     "search_chats": "Search through chat history across all sessions.",
     "ask_user": "Ask the user a multiple-choice question to get a decision or clarification. Use this when the task is genuinely ambiguous and the answer changes what you do next — pick between approaches, confirm an assumption, choose among options — instead of guessing. Provide a clear `question` and 2-6 `options` (each with a short `label`, optional `description`). Calling this ENDS your turn: the user sees clickable buttons and their choice arrives as your next message. Don't use it for things you can decide from context or sensible defaults, or for irreversible-action confirmation if a dedicated flow exists.",
+    "update_plan": "Write back to the ACTIVE PLAN while executing an approved plan: mark steps done or revise them. After finishing a step call this with the full checklist and that step marked done; when the user asks to change the plan call it with the revised checklist. Always pass the COMPLETE markdown checklist (`- [ ]` / `- [x]`), not a diff. The user's docked plan window updates live. No effect when there is no active plan.",
     "ui_control": "Control the UI and toggle tools on/off. Use this to turn off / turn on / disable / enable individual tools and features: shell (bash), search (web), research, browser, documents, incognito. Open panels (documents library, gallery, email inbox, sessions, notes, memories/brain, skills, settings, cookbook) via `open_panel <name>`. Use `open_email_reply <uid> <folder> reply` to open an email reply draft document without sending. Also switches between chat/agent modes, changes the current model, and applies/creates themes.",
     "list_email_accounts": "List configured email accounts and default status. Use before reading or sending mail when the user mentions Gmail, work mail, custom domain mail, another mailbox, or asks to compare/check multiple inboxes.",
     "list_emails": "List emails for a folder/account, newest first, including read messages by default. Shows subject, sender, date, UID, account, and AI summary. Check inbox, find emails needing replies. Supports account from list_email_accounts for Gmail/work/custom mailboxes. For last/latest/newest email, use max_results=1 and unread_only=false.",

src/tool_schemas.py

@@ -474,6 +474,20 @@ FUNCTION_TOOL_SCHEMAS = [
             }
         }
     },
+    {
+        "type": "function",
+        "function": {
+            "name": "update_plan",
+            "description": "Write back to the ACTIVE PLAN: mark steps done or revise them. Use this while executing an approved plan — after you finish a step, call update_plan with the full checklist and that step marked `- [x]`; when the user asks to change the plan, call it with the revised checklist. The user's docked plan window updates live. Pass the COMPLETE checklist every time (not a diff). No effect if there is no active plan.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "plan": {"type": "string", "description": "The full updated plan as a GitHub-style markdown checklist — one step per line, `- [ ]` for pending and `- [x]` for done. Always send the whole list."}
+                },
+                "required": ["plan"]
+            }
+        }
+    },
     {
         "type": "function",
         "function": {

src/tool_security.py

@@ -51,6 +51,101 @@ NON_ADMIN_BLOCKED_TOOLS = {
 }
 
 
+# Plan mode: the agent may investigate but must not mutate anything. Only these
+# read-only/inspection tools stay enabled; everything else (writes, sends,
+# manage_*, model serving, MCP, etc.) is blocked. Allowlist rather than blocklist
+# so any newly added tool defaults to BLOCKED in plan mode — fail safe.
+#
+# bash/python are deliberately NOT here: the shell can mutate (write files, hit
+# the network) and can't be constrained to read-only at the tool layer, so plan
+# mode blocks it outright rather than relying on a prompt to keep it well-behaved.
+# Code/file discovery is covered by the dedicated read-only tools below
+# (read_file, grep, glob, ls) instead of freestyle shell.
+PLAN_MODE_READONLY_TOOLS = {
+    "read_file",
+    "grep",
+    "glob",
+    "ls",
+    "web_search",
+    "web_fetch",
+    "search_chats",
+    "list_models",
+    "list_sessions",
+    "list_emails",
+    "read_email",
+    "list_served_models",
+    "list_downloads",
+    "list_cached_models",
+    "search_hf_models",
+    "list_serve_presets",
+    "list_cookbook_servers",
+    "resolve_contact",
+    "chat_with_model",
+    "ask_teacher",
+}
+
+
+# The agent's tool gate is a DENYLIST: execute_tool_block blocks any tool whose
+# name is in `disabled_tools`. Plan mode's policy is the opposite — an allowlist
+# (PLAN_MODE_READONLY_TOOLS). To apply an allowlist through a denylist, plan mode
+# returns the inverse: every known tool name minus the allowlist.
+#
+# Known tool names come from FUNCTION_TOOL_SCHEMAS, but that source is imperfect:
+# some tools are only XML-invocable (e.g. manage_notes, generate_image) and never
+# appear there, and the import can fail outright. Either gap would drop a mutating
+# tool from the subtraction and silently leave it enabled. This set is the static
+# backstop for both: union it in so known mutators are always subtracted, and so a
+# failed import still blocks them (fail closed, never open). Only mutators belong
+# here — read-only tools are covered by the allowlist. Keep in sync when adding
+# new mutating tools.
+_PLAN_MODE_KNOWN_MUTATORS = {
+    "write_file", "create_document", "edit_document", "update_document",
+    "suggest_document", "manage_documents", "create_session", "manage_session",
+    "send_to_session", "pipeline", "manage_memory", "manage_skills",
+    "manage_tasks", "manage_notes", "manage_endpoints", "manage_mcp",
+    "manage_webhooks", "manage_tokens", "manage_settings", "manage_contact",
+    "manage_calendar", "api_call", "app_api", "ui_control",
+    "send_email", "reply_to_email", "bulk_email", "delete_email",
+    "archive_email", "mark_email_read", "download_model", "serve_model",
+    "stop_served_model", "cancel_download", "adopt_served_model", "serve_preset",
+    "generate_image", "edit_image", "trigger_research", "manage_research",
+    # Shell is never read-only-safe; block it explicitly so it stays out of plan
+    # mode even if the schema list fails to load.
+    "bash", "python",
+}
+
+
+def plan_mode_disabled_tools() -> Set[str]:
+    """Tool names to add to the denylist in plan mode.
+
+    Plan mode allows only PLAN_MODE_READONLY_TOOLS. The gate is a denylist, so
+    return the inverse: every known tool name minus the allowlist. Known names
+    come from the function-tool schemas, backstopped by _PLAN_MODE_KNOWN_MUTATORS
+    (see above) so XML-only tools and a failed schema import can't leave a mutator
+    enabled. MCP tools are handled separately — the loop drops the MCP manager
+    entirely in plan mode."""
+    try:
+        # agent_tools / tool_parsing / tool_schemas form a mutually-circular
+        # cluster that only resolves cleanly when entered via agent_tools.
+        # Import it first so the lazy schema import works even from a cold
+        # import (e.g. tests) — not just after the app has wired everything up.
+        import src.agent_tools  # noqa: F401
+        from src.tool_schemas import FUNCTION_TOOL_SCHEMAS
+
+        all_names = {
+            (t.get("function") or {}).get("name")
+            for t in FUNCTION_TOOL_SCHEMAS
+        }
+        all_names.discard(None)
+    except Exception as exc:
+        logger.warning("Unable to load tool schemas for plan-mode gating: %s", exc)
+        all_names = set()
+    # Subtract the allowlist from all known tool names (schema-derived plus the
+    # static mutator backstop). Fail closed: if the schema import failed above,
+    # the backstop alone still blocks known mutators.
+    return (all_names | _PLAN_MODE_KNOWN_MUTATORS) - PLAN_MODE_READONLY_TOOLS
+
+
 def is_public_blocked_tool(tool_name: Optional[str]) -> bool:
     """Return True when a non-admin/public user must not execute this tool.