Auto-update config.json with new scalar fields from config.example.json on startup

On each startup, if config.json exists alongside config.example.json, any scalar
fields (strings, ints, booleans) present in the example but absent in the running
config are added automatically and the file is rewritten atomically. Existing values
and dict/list sections are never touched. Added fields are logged at INFO level.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

CLAUDE.md

@@ -0,0 +1,56 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Running the bot
+
+```bash
+pip install -r requirements.txt
+python main.py
+```
+
+Logs go to `bot.log` by default (rotating, 10 MB, 3 backups). Override with `LOG_FILE=<path>`.
+
+## Configuration
+
+Copy `config.example.json` to `config.json` and fill in credentials, or use environment variables (`.env.example` shows the env-var form). Both sources are merged at startup — env vars take precedence.
+
+Key env vars: `TG_API_ID`, `TG_API_HASH`, `SESSION_DIR`, `ACCOUNTS_JSON` (JSON list) or `ACCOUNT_SESSIONS` (comma-separated session names), `AI_CONFIG` (full AI block as JSON), `BOT_CONFIG` (path to config file), `BOT_DOTENV_PATH` (path to .env).
+
+## Architecture
+
+### Boot path
+`main.py` → `load_app_config()` → `AccountManager` → one Pyrogram (`kurigram`) `Client` per account → `register_common_handlers()` → `pyrogram.idle()`.
+
+### Command system (`app/commands/`)
+- **Dispatcher** (`dispatcher.py`): parses `.command` (owner prefix) or `..command` (shared/guest prefix) from message text, resolves to a `CommandDefinition`, runs pre-checks, calls handler.
+- **Registry** (`core.py`): `CommandRegistry` holds `CommandDefinition` objects. `DEFAULT_REGISTRY` is the singleton used at runtime; commands self-register via decorator.
+- **Built-in commands** (`builtins.py`): `.ping`, `.status`, `.help`, `.rights`/`.acl`, `.allow`, `.deny` (access), `.echo` (shared), `.ai` (shared). Add new commands here using the `@command_registry.command(...)` decorator pattern.
+- **Check system** (`core.py`): `CommandCheck` functions (e.g. `owner_only_check()`, `shared_access_check()`, `require_reply_check()`, `require_args_check()`) gate handler execution. Chain them with `.check(...)` or `.owner_only()` / `.require_reply()` / `.require_args()` on the builder.
+- **ACL** (`acl.py`): `RuntimeAclStore` persists per-session `<session>.acl.json` allow/deny rules. Only shared (`.is_shared`) commands can be granted to guests.
+
+### AI system (`app/ai/`)
+- **AIService** (`service.py`): OpenAI-compatible client (via `openai` SDK + `httpx`) that supports multiple named providers and model aliases defined in config. History collection reads `context_messages` prior messages from the chat. Reactions (🤔/👍/👎) are used for status signaling.
+- **Parser** (`parser.py`): `.ai [N] [--img] [--model-alias] prompt` — leading integer overrides context window, `--img` enables image understanding, other `--flags` are treated as model aliases.
+- **Prompts** (`prompts.py`): `PromptBook` loads `prompts/ai/system.md` + all `prompts/ai/parts/*.md` (sorted), joined with double newline. Edit those files to tune AI behavior without touching code.
+- **Image understanding** (`image_understand.py`): Spawns the MiniMax MCP subprocess (`uvx minimax-coding-plan-mcp`) and communicates via JSON-RPC over stdio (LSP-style `Content-Length` framing). Images are downloaded to a temp dir, analyzed one by one, and injected as a system message before the final completion.
+
+### Key types
+- `CommandContext` — per-invocation context passed to every handler; holds `client`, `manager`, `account`, `message`, parsed `args`/`argv`, and convenience properties (`actor_id`, `owner_id`, `is_owner`, `replied_user`).
+- `AccountState` — runtime state per account (client, ACL store, status, `me` object).
+- `AIInvocation` — parsed AI call parameters (context_messages, model_flags, use_image_understanding, prompt).
+
+## Adding a new command
+
+```python
+@command_registry.command(
+    "mycommand",
+    "Short description",
+    usage=".mycommand <arg>",
+    category="general",
+).shared()          # omit for owner-only; add .require_args() / .require_reply() as needed
+async def mycommand(ctx: "CommandContext") -> None:
+    await ctx.reply(ctx.args)
+```
+
+Shared commands use `..mycommand` for guests and `.mycommand` for the owner. Owner-only commands use `.mycommand` only.

app/ai/image_understand.py

@@ -187,6 +187,31 @@ class MinimaxImageUnderstandClient:
         self._stdin = None
         self._stdout = None
 
+    async def list_tools(self) -> list[dict]:
+        if self._process is None:
+            await self.start()
+        timeout = self.config.timeout if self.config.timeout is not None else 30.0
+        result = await self._request("tools/list", {}, timeout=timeout)
+        if isinstance(result, dict):
+            tools = result.get("tools")
+            if isinstance(tools, list):
+                return tools
+        return []
+
+    async def call_tool(self, name: str, arguments: dict) -> str:
+        if self._process is None:
+            await self.start()
+        timeout = self.config.timeout if self.config.timeout is not None else 120.0
+        result = await self._request(
+            "tools/call",
+            {"name": name, "arguments": arguments},
+            timeout=timeout,
+        )
+        text = self._extract_text(result)
+        if not text:
+            raise MCPError(f"MCP tool '{name}' returned an empty response")
+        return text
+
     async def understand_image(self, *, image_path: Path, prompt: str) -> str:
         if self._process is None:
             await self.start()

app/ai/service.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 import inspect
+import json
 import logging
 from pathlib import Path
 import tempfile
@@ -18,6 +19,7 @@ from app.ai.image_understand import (
 )
 from app.ai.parser import AIInvocation
 from app.ai.prompts import PromptBook
+from app.ai.tools import FILE_TOOL_DISPATCH, FILE_TOOL_SCHEMAS, mcp_tools_to_openai_schemas
 from app.config import AIConfig, AIModelConfig, AIProviderConfig
 
 if TYPE_CHECKING:
@@ -66,6 +68,8 @@ class AIService:
         self.prompt_book = PromptBook(config.prompt.root)
         self._clients: dict[str, AsyncOpenAI] = {}
         self._system_prompt: str | None = None
+        self._mcp_tool_schemas: list[dict] | None = None
+        self._mcp_tools_fetched: bool = False
 
     @property
     def enabled(self) -> bool:
@@ -287,8 +291,8 @@ class AIService:
         prompt: str,
         history: list[HistoryMessage],
         image_notes: list[str],
-    ) -> list[dict[str, str]]:
+    ) -> list[dict[str, Any]]:
-        messages: list[dict[str, str]] = [{"role": "system", "content": self.load_system_prompt()}]
+        messages: list[dict[str, Any]] = [{"role": "system", "content": self.load_system_prompt()}]
         if image_notes:
             messages.append({"role": "system", "content": self._build_image_summary_message(image_notes)})
         for item in history:
@@ -298,6 +302,60 @@ class AIService:
         messages.append({"role": "user", "content": prompt})
         return messages
 
+    async def _fetch_mcp_tool_schemas(self) -> list[dict]:
+        if self._mcp_tools_fetched:
+            return self._mcp_tool_schemas or []
+        self._mcp_tools_fetched = True
+        if self.config.image_understand is None:
+            self._mcp_tool_schemas = []
+            return []
+        try:
+            async with MinimaxImageUnderstandClient(self.config.image_understand) as mcp_client:
+                raw = await mcp_client.list_tools()
+            self._mcp_tool_schemas = mcp_tools_to_openai_schemas(raw)
+        except Exception:
+            log.warning("Failed to fetch MCP tool schemas; MiniMax tools will be unavailable")
+            self._mcp_tool_schemas = []
+        return self._mcp_tool_schemas or []
+
+    async def _build_tool_list(self) -> list[dict]:
+        mcp_schemas = await self._fetch_mcp_tool_schemas()
+        return FILE_TOOL_SCHEMAS + mcp_schemas
+
+    def _get_allowed_roots(self) -> tuple[Path, ...]:
+        return (self.config.workspace_dir, self.config.prompt.root)
+
+    async def _execute_tool_call(
+        self,
+        tool_call: Any,
+        mcp_client: MinimaxImageUnderstandClient | None,
+        mcp_names: set[str],
+        allowed_roots: tuple[Path, ...],
+        progress: ProgressCallback | None,
+    ) -> str:
+        name = getattr(getattr(tool_call, "function", None), "name", None) or ""
+        raw_args = getattr(getattr(tool_call, "function", None), "arguments", None) or "{}"
+
+        try:
+            arguments: dict = json.loads(raw_args)
+        except json.JSONDecodeError as exc:
+            return f"Tool {name!r}: invalid JSON arguments: {exc}"
+
+        try:
+            if name in FILE_TOOL_DISPATCH:
+                result = await FILE_TOOL_DISPATCH[name](arguments, allowed_roots)
+            elif name in mcp_names and mcp_client is not None:
+                result = await mcp_client.call_tool(name, arguments)
+            else:
+                result = f"Unknown tool: {name!r}"
+        except (ValueError, OSError) as exc:
+            result = f"Tool {name!r} error: {exc}"
+        except MCPError as exc:
+            result = f"Tool {name!r} MCP error: {exc}"
+
+        await self._maybe_call_progress(progress, f"🛠 Tool {name}: {result[:80]}")
+        return result
+
     async def generate(
         self,
         ctx: "CommandContext",
@@ -327,21 +385,91 @@ class AIService:
             image_notes=image_notes,
         )
 
+        tools = await self._build_tool_list()
+        allowed_roots = self._get_allowed_roots()
+        max_rounds = self.config.tool_max_rounds
+        mcp_names: set[str] = {
+            s["function"]["name"]
+            for s in (self._mcp_tool_schemas or [])
+        }
+
+        mcp_client: MinimaxImageUnderstandClient | None = None
+
         try:
-            response = await client.chat.completions.create(
+            for round_num in range(max_rounds + 1):
-                model=model.model.model,
+                kwargs: dict[str, Any] = {
-                messages=messages,
+                    "model": model.model.model,
-            )
+                    "messages": messages,
+                }
+                if tools:
+                    kwargs["tools"] = tools
+
+                try:
+                    response = await client.chat.completions.create(**kwargs)
                 except Exception as exc:
                     log.exception("AI request failed for model %s", model.alias)
                     raise RuntimeError(f"AI request failed: {exc}") from exc
 
+                choice = response.choices[0]
+                finish_reason = getattr(choice, "finish_reason", None)
+
+                if finish_reason != "tool_calls":
                     text = self._extract_response_text(response)
                     if not text:
                         raise RuntimeError("AI model returned an empty response")
-
                     return AIResponse(text=text, model=model.model, provider=model.provider)
 
+                if round_num >= max_rounds:
+                    raise RuntimeError(
+                        f"Agentic loop exceeded the maximum of {max_rounds} tool rounds"
+                    )
+
+                tool_calls = getattr(choice.message, "tool_calls", None) or []
+                if not tool_calls:
+                    raise RuntimeError("finish_reason was tool_calls but no tool_calls present")
+
+                messages.append({
+                    "role": "assistant",
+                    "content": getattr(choice.message, "content", None),
+                    "tool_calls": [
+                        {
+                            "id": tc.id,
+                            "type": "function",
+                            "function": {
+                                "name": tc.function.name,
+                                "arguments": tc.function.arguments,
+                            },
+                        }
+                        for tc in tool_calls
+                    ],
+                })
+
+                needs_mcp = any(
+                    getattr(getattr(tc, "function", None), "name", "") in mcp_names
+                    for tc in tool_calls
+                )
+                if needs_mcp and mcp_client is None and self.config.image_understand is not None:
+                    mcp_client = MinimaxImageUnderstandClient(self.config.image_understand)
+                    await mcp_client.start()
+
+                for tc in tool_calls:
+                    tc_name = getattr(getattr(tc, "function", None), "name", "?")
+                    await self._maybe_call_progress(progress, f"🛠 Calling tool: {tc_name}")
+                    result_text = await self._execute_tool_call(
+                        tc, mcp_client, mcp_names, allowed_roots, progress
+                    )
+                    messages.append({
+                        "role": "tool",
+                        "tool_call_id": tc.id,
+                        "content": result_text,
+                    })
+
+            raise RuntimeError("Agentic loop terminated without a response")
+
+        finally:
+            if mcp_client is not None:
+                await mcp_client.close()
+
     def _extract_response_text(self, response: Any) -> str:
         choices = getattr(response, "choices", None) or []
         if not choices:

app/ai/tools.py

@@ -0,0 +1,347 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Awaitable, Callable
+
+
+def _resolve_sandboxed_path(raw_path: str, allowed_roots: tuple[Path, ...]) -> Path:
+    if raw_path.startswith("/"):
+        raise ValueError(f"Absolute paths are not allowed: {raw_path!r}")
+    for root in allowed_roots:
+        resolved_root = root.resolve()
+        candidate = (root / raw_path).resolve()
+        if candidate == resolved_root or str(candidate).startswith(str(resolved_root) + "/"):
+            return candidate
+    raise ValueError(
+        f"Path {raw_path!r} is outside the allowed directories "
+        f"({', '.join(str(r) for r in allowed_roots)})"
+    )
+
+
+async def tool_write_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    content = str(arguments.get("content", ""))
+    if not raw_path:
+        raise ValueError("write_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        return f"Written {len(content.encode('utf-8'))} bytes to {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"write_file failed: {exc.strerror}") from exc
+
+
+async def tool_read_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    if not raw_path:
+        raise ValueError("read_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if not path.exists():
+            raise ValueError(f"read_file: file not found: {raw_path!r}")
+        if not path.is_file():
+            raise ValueError(f"read_file: not a file: {raw_path!r}")
+        size = path.stat().st_size
+        if size > 100 * 1024:
+            raise ValueError(f"read_file: file exceeds the 100 KB read limit ({size} bytes)")
+        try:
+            return path.read_text(encoding="utf-8")
+        except UnicodeDecodeError:
+            return "File is binary (not readable as text)"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"read_file failed: {exc.strerror}") from exc
+
+
+async def tool_create_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    content = str(arguments.get("content", ""))
+    if not raw_path:
+        raise ValueError("create_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if path.exists():
+            raise ValueError(f"create_file: file already exists: {raw_path!r}")
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        return f"Created {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"create_file failed: {exc.strerror}") from exc
+
+
+async def tool_modify_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    operation = str(arguments.get("operation", "")).strip().lower()
+    if not raw_path:
+        raise ValueError("modify_file: 'path' is required")
+    if operation not in ("append", "replace"):
+        raise ValueError("modify_file: 'operation' must be 'append' or 'replace'")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if operation == "append":
+            text = str(arguments.get("text", ""))
+            existing = path.read_text(encoding="utf-8") if path.exists() else ""
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(existing + text, encoding="utf-8")
+            return f"Appended {len(text.encode('utf-8'))} bytes to {raw_path}"
+        else:
+            old_text = arguments.get("old_text")
+            new_text = arguments.get("new_text")
+            if old_text is None or new_text is None:
+                raise ValueError("modify_file replace: 'old_text' and 'new_text' are required")
+            old_text = str(old_text)
+            new_text = str(new_text)
+            if not path.exists():
+                raise ValueError(f"modify_file: file not found: {raw_path!r}")
+            content = path.read_text(encoding="utf-8")
+            if old_text not in content:
+                raise ValueError(f"modify_file replace: 'old_text' not found in {raw_path!r}")
+            path.write_text(content.replace(old_text, new_text, 1), encoding="utf-8")
+            return f"Replaced text in {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"modify_file failed: {exc.strerror}") from exc
+
+
+async def tool_delete_file(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    if not raw_path:
+        raise ValueError("delete_file: 'path' is required")
+    try:
+        path = _resolve_sandboxed_path(raw_path, allowed_roots)
+        if not path.exists():
+            raise ValueError(f"delete_file: file not found: {raw_path!r}")
+        path.unlink()
+        return f"Deleted {raw_path}"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"delete_file failed: {exc.strerror}") from exc
+
+
+async def tool_list_files(arguments: dict, allowed_roots: tuple[Path, ...]) -> str:
+    raw_path = str(arguments.get("path", "")).strip()
+    try:
+        if raw_path:
+            base = _resolve_sandboxed_path(raw_path, allowed_roots)
+        else:
+            # list all allowed roots combined
+            entries: list[str] = []
+            for root in allowed_roots:
+                if not root.exists():
+                    continue
+                resolved_root = root.resolve()
+                entries.extend(_collect_files(root, resolved_root, max_depth=3))
+            return "\n".join(sorted(entries)) or "(empty)"
+
+        if not base.exists():
+            return "(empty)"
+        if not base.is_dir():
+            raise ValueError(f"list_files: not a directory: {raw_path!r}")
+        resolved_base = base.resolve()
+        entries = _collect_files(base, resolved_base, max_depth=3)
+        return "\n".join(sorted(entries)) or "(empty)"
+    except ValueError:
+        raise
+    except OSError as exc:
+        raise ValueError(f"list_files failed: {exc.strerror}") from exc
+
+
+def _collect_files(directory: Path, resolved_root: Path, *, max_depth: int) -> list[str]:
+    result: list[str] = []
+    _walk(directory, resolved_root, directory, max_depth, 0, result)
+    return result
+
+
+def _walk(
+    current: Path,
+    resolved_root: Path,
+    base: Path,
+    max_depth: int,
+    depth: int,
+    result: list[str],
+) -> None:
+    if depth > max_depth:
+        return
+    try:
+        for entry in current.iterdir():
+            resolved_entry = entry.resolve()
+            # skip symlinks that escape the root
+            if not str(resolved_entry).startswith(str(resolved_root)):
+                continue
+            if entry.is_file():
+                result.append(str(entry.relative_to(base)))
+            elif entry.is_dir():
+                _walk(entry, resolved_root, base, max_depth, depth + 1, result)
+    except OSError:
+        pass
+
+
+ToolExecutor = Callable[[dict, tuple[Path, ...]], Awaitable[str]]
+
+FILE_TOOL_DISPATCH: dict[str, ToolExecutor] = {
+    "write_file": tool_write_file,
+    "read_file": tool_read_file,
+    "create_file": tool_create_file,
+    "modify_file": tool_modify_file,
+    "delete_file": tool_delete_file,
+    "list_files": tool_list_files,
+}
+
+FILE_TOOL_SCHEMAS: list[dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": (
+                "Create or overwrite a file. Allowed path prefixes: workspace/ or prompts/. "
+                "Example path: workspace/notes.txt or prompts/ai/parts/extra.md"
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                    "content": {"type": "string", "description": "File content to write"},
+                },
+                "required": ["path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "read_file",
+            "description": "Read the content of a file (max 100 KB). Allowed path prefixes: workspace/ or prompts/.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                },
+                "required": ["path"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "create_file",
+            "description": (
+                "Create a new file. Fails if the file already exists. "
+                "Allowed path prefixes: workspace/ or prompts/."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                    "content": {"type": "string", "description": "Initial file content"},
+                },
+                "required": ["path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "modify_file",
+            "description": (
+                "Modify an existing file. Two operations: "
+                "'append' adds text at end (requires 'text'); "
+                "'replace' substitutes first occurrence of old_text with new_text "
+                "(requires 'old_text' and 'new_text', fails if old_text not found). "
+                "Allowed path prefixes: workspace/ or prompts/."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                    "operation": {
+                        "type": "string",
+                        "enum": ["append", "replace"],
+                        "description": "Operation to perform",
+                    },
+                    "text": {
+                        "type": "string",
+                        "description": "Text to append (used when operation=append)",
+                    },
+                    "old_text": {
+                        "type": "string",
+                        "description": "Text to find (used when operation=replace)",
+                    },
+                    "new_text": {
+                        "type": "string",
+                        "description": "Replacement text (used when operation=replace)",
+                    },
+                },
+                "required": ["path", "operation"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "delete_file",
+            "description": "Delete a file. Allowed path prefixes: workspace/ or prompts/.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Relative file path"},
+                },
+                "required": ["path"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "list_files",
+            "description": (
+                "List files recursively (up to 3 levels deep). "
+                "Pass an empty path or omit to list all allowed directories. "
+                "Allowed path prefixes: workspace/ or prompts/."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {
+                        "type": "string",
+                        "description": "Directory path to list (relative). Leave empty to list all.",
+                        "default": "",
+                    },
+                },
+                "required": [],
+            },
+        },
+    },
+]
+
+
+def mcp_tools_to_openai_schemas(mcp_tools: list[dict]) -> list[dict]:
+    schemas: list[dict] = []
+    for tool in mcp_tools:
+        if not isinstance(tool, dict):
+            continue
+        name = tool.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+        description = str(tool.get("description", "")).strip() or f"MCP tool: {name}"
+        input_schema = tool.get("inputSchema") or tool.get("input_schema")
+        if not isinstance(input_schema, dict):
+            input_schema = {"type": "object", "properties": {}, "required": []}
+        schemas.append(
+            {
+                "type": "function",
+                "function": {
+                    "name": name,
+                    "description": description,
+                    "parameters": input_schema,
+                },
+            }
+        )
+    return schemas

app/config.py

@@ -1,11 +1,15 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
+import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
+log = logging.getLogger(__name__)
+
 try:
     from dotenv import load_dotenv
 except ModuleNotFoundError:  # pragma: no cover - optional dependency fallback
@@ -71,6 +75,8 @@ class AIConfig:
     image_understand: AIImageUnderstandConfig | None = None
     providers: dict[str, AIProviderConfig] = field(default_factory=dict)
     models: dict[str, AIModelConfig] = field(default_factory=dict)
+    workspace_dir: Path = field(default_factory=lambda: Path("workspace"))
+    tool_max_rounds: int = 10
 
 
 @dataclass(slots=True)
@@ -86,6 +92,35 @@ def _read_json_file(path: Path) -> dict[str, Any]:
     return json.loads(path.read_text(encoding="utf-8"))
 
 
+def _write_json_file(path: Path, data: dict[str, Any]) -> None:
+    serialized = json.dumps(data, ensure_ascii=False, indent=2)
+    with tempfile.NamedTemporaryFile(
+        "w",
+        encoding="utf-8",
+        delete=False,
+        dir=path.parent,
+        prefix=path.name,
+        suffix=".tmp",
+    ) as handle:
+        handle.write(serialized)
+        temp_path = Path(handle.name)
+    temp_path.replace(path)
+
+
+def _merge_example_defaults(target: dict[str, Any], source: dict[str, Any], path: str = "") -> list[str]:
+    """Add scalar keys present in source but absent in target; recurse into existing dicts."""
+    added: list[str] = []
+    for key, value in source.items():
+        full_path = f"{path}.{key}" if path else key
+        if key not in target:
+            if not isinstance(value, (dict, list)):
+                target[key] = value
+                added.append(full_path)
+        elif isinstance(target[key], dict) and isinstance(value, dict):
+            added.extend(_merge_example_defaults(target[key], value, full_path))
+    return added
+
+
 def _parse_int(value: Any, *, field_name: str) -> int:
     try:
         return int(value)
@@ -289,6 +324,15 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
                 f"ai.models.{alias}.provider references unknown provider: {model_config.provider}"
             )
 
+    workspace_dir_raw = raw_ai.get("workspace_dir", "workspace")
+    workspace_dir = Path(str(workspace_dir_raw)).expanduser()
+    if not workspace_dir.is_absolute():
+        workspace_dir = (root / workspace_dir).resolve()
+
+    tool_max_rounds = _parse_int(raw_ai.get("tool_max_rounds", 10), field_name="ai.tool_max_rounds")
+    if tool_max_rounds < 1:
+        raise ValueError("ai.tool_max_rounds must be >= 1")
+
     return AIConfig(
         default_model=default_model,
         context_messages=context_messages,
@@ -296,6 +340,8 @@ def _parse_ai_config(raw_ai: Any, *, root: Path) -> AIConfig | None:
         image_understand=image_understand,
         providers=providers,
         models=models,
+        workspace_dir=workspace_dir,
+        tool_max_rounds=tool_max_rounds,
     )
 
 
@@ -337,6 +383,16 @@ def load_app_config(root: Path) -> AppConfig:
     raw: dict[str, Any] = {}
     if candidate.exists():
         raw = _read_json_file(candidate)
+        example_path = root / "config.example.json"
+        if example_path.exists():
+            try:
+                example_raw = _read_json_file(example_path)
+                added = _merge_example_defaults(raw, example_raw)
+                if added:
+                    _write_json_file(candidate, raw)
+                    log.info("Auto-updated %s with new fields: %s", candidate.name, ", ".join(added))
+            except Exception as exc:
+                log.warning("Could not auto-update config from example: %s", exc)
 
     env_api_id = os.environ.get("TG_API_ID")
     env_api_hash = os.environ.get("TG_API_HASH")

config.example.json

@@ -15,6 +15,8 @@
   "ai": {
     "default_model": "mm",
     "context_messages": 12,
+    "workspace_dir": "workspace",
+    "tool_max_rounds": 10,
     "prompt": {
       "root": "prompts/ai",
       "system_file": "system.md",