SoniCoder v2: Optimized code agent with Gemini CLI / Claude Code patterns, MCP support, multiple 1B models

app.py

@@ -1,40 +1,97 @@
-"""SoniCoder — entry point.
-
-Uses MiniCPM5-1B (text) or MiniCPM-V-4.6 (vision+text) for local inference.
-No external APIs required.
-Supports generating fullstack applications in any language.
-Can push generated projects to HuggingFace Hub.
-Web search via Google scraping (no API keys needed).
-Gradio app support for Python.
-Image understanding with MiniCPM-V-4.6.
-
-Project structure:
-    code/
-    ├── config/constants.py         App constants, model configs, system prompt
-    ├── model/loader.py             Dual model loading & switching
-    ├── model/inference.py          Streaming inference (text + VLM)
-    ├── execution/code_extractor.py Code extraction & language normalization
-    ├── execution/python_runner.py  Sandboxed Python execution
-    ├── execution/gradio_runner.py  Gradio app subprocess runner
-    ├── websearch/google_scraper.py Web search scraping (no API)
-    ├── huggingface/push.py         HuggingFace Hub push & ZIP packaging
-    ├── server/chat_helpers.py      Chat history & prompt building
-    └── server/routes.py            FastAPI / Gradio server routes
+"""SoniCoder v2 — Entry point.
+
+Optimized AI code writer agent with:
+- Event-driven agent protocol (from Gemini CLI)
+- ToolBuilder/ToolInvocation pattern (from Gemini CLI)
+- MCP support (from both Gemini CLI & Claude Code)
+- Multiple smart 1B models (Qwen2.5-Coder-1.5B, DeepSeek-Coder-1.3B, etc.)
+- Policy/permission engine (from Gemini CLI)
+- Context management
+- Professional web UI
+
+Usage:
+    python app.py
 """
 
 from __future__ import annotations
 
+import json
 import logging
+from pathlib import Path
 
-from code.model.loader import start_background_load
-from code.server.routes import get_app
+# Ensure workspace exists
+from sonicoder.config import WORKSPACE_ROOT, CONFIG_DIR
+WORKSPACE_ROOT.mkdir(parents=True, exist_ok=True)
+CONFIG_DIR.mkdir(parents=True, exist_ok=True)
 
-logging.basicConfig(level=logging.INFO)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%H:%M:%S",
+)
 logger = logging.getLogger(__name__)
 
-# Start loading default model in background
-start_background_load()
 
-# Launch the server
-application = get_app()
-application.launch(show_error=True)
+def main():
+    logger.info("SoniCoder v2 starting...")
+
+    # Start loading default model in background
+    from sonicoder.model.loader import start_background_load
+    start_background_load()
+    logger.info("Background model loading started")
+
+    # Set up MCP servers from config
+    from sonicoder.mcp import get_mcp_manager
+    from sonicoder.config import load_settings
+    settings = load_settings()
+    if settings.mcp_servers:
+        mcp = get_mcp_manager()
+        mcp.configure(settings.mcp_servers)
+        results = mcp.connect_all()
+        for name, error in results.items():
+            if error:
+                logger.warning(f"MCP server '{name}' failed: {error}")
+            else:
+                logger.info(f"MCP server '{name}' connected")
+
+    # Create the Gradio app
+    from sonicoder.server import get_app, api_get_config
+    application = get_app()
+
+    # Inject runtime config into index.html and serve via custom route
+    config_json = api_get_config()
+    config_data = json.loads(config_json)
+    config_str = json.dumps(config_data, ensure_ascii=False)
+
+    index_path = Path(__file__).parent / "static" / "index.html"
+    if index_path.exists():
+        html = index_path.read_text(encoding="utf-8")
+        html = html.replace(
+            "__RUNTIME_CONFIG__ || {}",
+            config_str + " || {}"
+        )
+
+        # Register a custom root route on the underlying FastAPI app
+        from fastapi import Response
+        from fastapi.responses import HTMLResponse
+
+        if hasattr(application, "app") and hasattr(application.app, "get"):
+            # Gradio 4.x: application.app is the FastAPI app (or Starlette)
+            fastapi_app = application.app
+            if hasattr(fastapi_app, "routes"):
+                @fastapi_app.get("/", response_class=HTMLResponse, include_in_schema=False)
+                async def custom_root():
+                    return Response(content=html, media_type="text/html")
+                logger.info("Custom root route registered")
+
+    logger.info("Launching web interface on http://0.0.0.0:7860")
+    application.launch(
+        show_error=True,
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False,
+    )
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -1,15 +1,17 @@
-gradio[oauth]==6.14.0
+# SoniCoder v2 — Optimized AI Code Agent
+# Architecture inspired by Gemini CLI and Claude Code
+
+gradio>=4.0.0
 transformers>=4.45.0
 torch>=2.1.0
 accelerate>=0.25.0
 huggingface_hub>=0.20.0
-matplotlib>=3.8
 requests>=2.31.0
 beautifulsoup4>=4.12.0
 Pillow>=10.0
-torchvision>=0.16.0
-# New deps for agent features (most are stdlib in 3.11+)
-# (No new external deps required — agent/skills/hooks/commands use stdlib only)
-# `gradio[oauth]` extras pull in authlib + itsdangerous for the
-# /login/huggingface + /login/callback OAuth flow. Required for the
-# Sign-in-with-HF button to work in the Deploy tab.
+
+# Optional: MCP support
+# mcp>=1.0.0
+
+# Optional: VLM support
+# torchvision>=0.16.0

sonicoder/__init__.py

@@ -0,0 +1,12 @@
+"""SoniCoder v2 — Optimized AI Code Writer Agent.
+
+Architecture inspired by Gemini CLI and Claude Code.
+- Event-driven agent protocol
+- ToolBuilder/ToolInvocation pattern
+- MCP (Model Context Protocol) support
+- Multiple smart 1B models
+- Policy/permission engine
+- Context window management with compression
+- Plugin system
+"""
+__version__ = "2.0.0"

sonicoder/agent/__init__.py

@@ -0,0 +1,474 @@
+"""Agent loop — the core agentic turn cycle.
+
+Implements the full agent loop inspired by both Gemini CLI and Claude Code:
+1. Build context (system prompt + tool schemas + history)
+2. Call model
+3. Parse tool calls from response
+4. Check policy
+5. Execute tools
+6. Feed results back to model
+7. Repeat until done or max iterations
+
+Tool call format: ```tool\nname\nkey: value\nkey: value\n```
+This is simpler than JSON function calling and works well with 1B models.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import time
+from typing import Any, Generator
+
+from ..config import Settings, get_model_config
+from ..inference import call_model_stream, TOOL_CALL_RE, THINKING_BLOCK_RE
+from ..tools import ToolRegistry, ToolResult, get_tool_registry
+from ..tools.fs import BUILTIN_TOOLS as FS_TOOLS
+from ..tools.bash import BUILTIN_TOOLS as BASH_TOOLS
+from ..tools.web import BUILTIN_TOOLS as WEB_TOOLS
+from ..tools.todos import BUILTIN_TOOLS as TODO_TOOLS
+from .protocol import AgentProtocol, AgentEvent, AgentEventType
+from ..policy import get_policy_engine, ApprovalDecision
+from ..mcp import get_mcp_manager
+
+logger = logging.getLogger(__name__)
+
+# ── System Prompt ──────────────────────────────────────────────────────
+
+SYSTEM_PROMPT = """You are SoniCoder v2, an expert AI coding agent. You write clean, well-structured code and help users build projects.
+
+## Capabilities
+- Read, write, and edit files in the workspace
+- Execute shell commands (bash)
+- Search the web for information
+- Manage task lists
+- Use MCP tools when available
+
+## Rules
+1. Always use tools to interact with the file system. Never guess file contents.
+2. When creating multi-file projects, write files one at a time with write_file.
+3. When editing existing files, use edit_file (search & replace) instead of rewriting.
+4. Run commands to verify your code works (e.g., python file.py, npm test).
+5. Think step-by-step. Plan before coding complex features.
+6. Use the todo tool to track multi-step tasks.
+7. If a tool fails, read the error and try to fix it.
+8. When done, summarize what you built.
+
+## File Output Format
+You can write code in two ways:
+1. Use the write_file tool for each file (preferred)
+2. Use fenced code blocks with language tags for display
+
+## Tool Call Format
+To use a tool, output a fenced code block with "tool" language:
+```tool
+tool_name
+key: value
+key2: value2
+```
+
+Example:
+```tool
+write_file
+path: hello.py
+content: print("Hello, World!")
+```
+
+## Multi-File Projects
+For projects with multiple files, write each file separately using write_file.
+Start with the main entry point, then supporting files.
+
+## Best Practices
+- Include error handling in your code
+- Add comments for complex logic
+- Use meaningful variable names
+- Follow the language's conventions and idioms
+- Test your code after writing it
+"""
+
+
+# ── Simple YAML Parser (no PyYAML dependency) ──────────────────────────
+
+def _parse_yaml_block(text: str) -> dict[str, Any]:
+    """Parse a simple YAML-like block into a dictionary.
+
+    Supports: key: value, key: | (block scalar), - item (lists)
+    """
+    result: dict[str, Any] = {}
+    current_key: str | None = None
+    current_lines: list[str] = []
+    in_block = False
+
+    for line in text.strip().splitlines():
+        # Check for list items
+        if line.strip().startswith("- "):
+            if in_block and current_key:
+                result[current_key] = "\n".join(current_lines).strip()
+                current_lines = []
+                in_block = False
+            item = line.strip()[2:].strip()
+            if current_key:
+                result.setdefault(current_key, [])
+                if isinstance(result[current_key], str):
+                    result[current_key] = [result[current_key]]
+                result[current_key].append(item)
+            continue
+
+        # Check for block scalar
+        if line.endswith("|"):
+            current_key = line[:-1].strip().rstrip(":")
+            in_block = True
+            current_lines = []
+            continue
+
+        # Check for key: value
+        match = re.match(r"^(\w[\w\s]*?)\s*:\s*(.*)", line)
+        if match and not in_block:
+            # Save previous block
+            if current_key and in_block and current_lines:
+                result[current_key] = "\n".join(current_lines).strip()
+
+            key = match.group(1).strip()
+            value = match.group(2).strip()
+            current_key = key
+
+            # Type coercion
+            if value.lower() in ("true", "yes"):
+                result[key] = True
+            elif value.lower() in ("false", "no"):
+                result[key] = False
+            elif value.isdigit():
+                result[key] = int(value)
+            else:
+                try:
+                    result[key] = float(value)
+                except ValueError:
+                    result[key] = value
+            in_block = False
+            current_lines = []
+            continue
+
+        if in_block and current_key:
+            current_lines.append(line)
+
+    # Save remaining block
+    if in_block and current_key and current_lines:
+        result[current_key] = "\n".join(current_lines).strip()
+
+    return result
+
+
+def _parse_tool_calls(text: str) -> list[dict[str, Any]]:
+    """Parse all tool calls from model output.
+
+    Format:
+    ```tool
+    tool_name
+    key: value
+    ```
+    """
+    calls = []
+    for match in TOOL_CALL_RE.finditer(text):
+        block = match.group(1).strip()
+        lines = block.split("\n", 1)
+
+        tool_name = lines[0].strip()
+        args_raw = lines[1] if len(lines) > 1 else ""
+        args = _parse_yaml_block(args_raw)
+
+        calls.append({
+            "tool_name": tool_name,
+            "args": args,
+            "raw": block,
+        })
+
+    return calls
+
+
+# ── Agent Session ──────────────────────────────────────────────────────
+
+class AgentSession(AgentProtocol):
+    """Full agent session with tool execution loop.
+
+    This is the main class that runs the agentic loop:
+    user message → model → tool calls → execute → feed back → model → ... → done
+    """
+
+    def __init__(
+        self,
+        settings: Settings | None = None,
+        tool_registry: ToolRegistry | None = None,
+    ):
+        super().__init__()
+        self._settings = settings or Settings()
+        self._registry = tool_registry or get_tool_registry()
+        self._policy = get_policy_engine()
+        self._mcp = get_mcp_manager()
+        self._history: list[dict[str, str]] = []
+        self._total_tokens = 0
+        self._tool_calls = 0
+        self._start_time = 0.0
+
+    def run(
+        self,
+        message: str,
+        session_id: str = "",
+        image_url: str | None = None,
+    ) -> Generator[dict, None, None]:
+        """Run the agent loop. Yields event dicts for streaming to the client."""
+        self._start_time = time.time()
+        self._tool_calls = 0
+
+        # Initialize
+        stream_id = self.send(message)
+        self._emit(AgentEventType.STATUS, {"status": "starting", "message": "Starting agent..."})
+
+        # Add user message to history
+        self._history.append({"role": "user", "content": message})
+
+        # Build tool schemas
+        tool_schemas = self._build_tool_schemas()
+
+        # Build full system prompt
+        system_prompt = self._build_system_prompt(tool_schemas)
+
+        # Agent loop
+        max_iterations = self._settings.max_iterations
+        model_config = get_model_config(self._settings.default_model)
+
+        for iteration in range(max_iterations):
+            if self.is_aborted:
+                self._emit(AgentEventType.COMPLETE, {"reason": "aborted"})
+                yield {"type": "complete", "reason": "aborted"}
+                return
+
+            self._emit(AgentEventType.STATUS, {
+                "status": "thinking",
+                "iteration": iteration + 1,
+                "max_iterations": max_iterations,
+                "message": f"Thinking (iteration {iteration + 1}/{max_iterations})...",
+            })
+
+            # Call model
+            try:
+                response_text = ""
+                for chunk in call_model_stream(
+                    messages=self._history,
+                    system_prompt=system_prompt,
+                    max_new_tokens=self._settings.max_tokens,
+                    temperature=self._settings.temperature,
+                    image_url=image_url,
+                ):
+                    response_text += chunk
+                    # Strip thinking blocks for display
+                    display = THINKING_BLOCK_RE.sub("", response_text)
+                    # Strip tool calls for display
+                    display_clean = TOOL_CALL_RE.sub("", display).strip()
+                    yield {"type": "streaming", "content": display_clean, "full_text": response_text}
+
+            except Exception as e:
+                self._emit(AgentEventType.ERROR, {"error": str(e)})
+                yield {"type": "error", "error": str(e)}
+                return
+
+            # Add assistant response to history
+            self._history.append({"role": "assistant", "content": response_text})
+
+            # Parse tool calls
+            tool_calls = _parse_tool_calls(response_text)
+
+            if not tool_calls:
+                # No tool calls — agent is done
+                display = THINKING_BLOCK_RE.sub("", response_text).strip()
+                self._emit(AgentEventType.MESSAGE, {"content": display})
+                self._emit(AgentEventType.COMPLETE, {"reason": "done"})
+                elapsed = time.time() - self._start_time
+                yield {
+                    "type": "complete",
+                    "reason": "done",
+                    "content": display,
+                    "iterations": iteration + 1,
+                    "tool_calls": self._tool_calls,
+                    "elapsed_seconds": round(elapsed, 1),
+                }
+                return
+
+            # Process tool calls
+            for tc in tool_calls:
+                if self.is_aborted:
+                    yield {"type": "complete", "reason": "aborted"}
+                    return
+
+                tool_name = tc["tool_name"]
+                args = tc["args"]
+
+                self._emit(AgentEventType.TOOL_REQUEST, {
+                    "tool": tool_name,
+                    "args": args,
+                })
+
+                yield {
+                    "type": "tool_call",
+                    "tool": tool_name,
+                    "args": args,
+                }
+
+                # Check policy
+                policy_result = self._policy.check(tool_name, args)
+                if policy_result.decision == ApprovalDecision.DENY:
+                    result = ToolResult(
+                        success=False,
+                        error=f"Blocked by policy: {policy_result.reason}",
+                    )
+                else:
+                    # Execute tool
+                    result = self._execute_tool(tool_name, args, session_id)
+
+                self._tool_calls += 1
+                self._emit(AgentEventType.TOOL_RESPONSE, {
+                    "tool": tool_name,
+                    "result": result.to_dict(),
+                })
+
+                yield {
+                    "type": "tool_result",
+                    "tool": tool_name,
+                    "result": result.to_dict(),
+                }
+
+                # Feed result back to model
+                result_content = result.to_json()
+                self._history.append({
+                    "role": "tool",
+                    "content": f"[{tool_name}] {result_content}",
+                })
+
+        # Max iterations reached
+        self._emit(AgentEventType.COMPLETE, {"reason": "max_iterations"})
+        yield {
+            "type": "complete",
+            "reason": "max_iterations",
+            "iterations": max_iterations,
+            "tool_calls": self._tool_calls,
+            "elapsed_seconds": round(time.time() - self._start_time, 1),
+        }
+
+    def _execute_tool(
+        self, tool_name: str, args: dict[str, Any], session_id: str
+    ) -> ToolResult:
+        """Execute a single tool call."""
+        # Check if it's an MCP tool
+        if tool_name.startswith("mcp_"):
+            return self._execute_mcp_tool(tool_name, args)
+
+        # Look up in registry
+        tool_builder = self._registry.get_tool(tool_name)
+        if not tool_builder:
+            return ToolResult(success=False, error=f"Unknown tool: {tool_name}")
+
+        # Validate
+        errors = tool_builder.validate(args)
+        if errors:
+            return ToolResult(success=False, error=f"Validation error: {'; '.join(errors)}")
+
+        # Create invocation
+        invocation = tool_builder.create_invocation(args)
+
+        # Inject session_id for todo tool
+        if tool_name == "todo":
+            invocation.session_id = session_id
+
+        # Execute
+        try:
+            return invocation.execute()
+        except Exception as e:
+            logger.error(f"Tool execution error ({tool_name}): {e}")
+            return ToolResult(success=False, error=f"Tool error: {e}")
+
+    def _execute_mcp_tool(self, tool_name: str, args: dict[str, Any]) -> ToolResult:
+        """Execute an MCP tool."""
+        bridge = self._mcp.get_tool(tool_name)
+        if not bridge:
+            return ToolResult(success=False, error=f"MCP tool not found: {tool_name}")
+
+        try:
+            result = self._mcp.call_tool(bridge.server_name, bridge.tool_name, args)
+            if "error" in result:
+                return ToolResult(success=False, error=result["error"])
+            return ToolResult(success=True, output=result.get("output", str(result)))
+        except Exception as e:
+            return ToolResult(success=False, error=f"MCP error: {e}")
+
+    def _build_tool_schemas(self) -> str:
+        """Build tool schema documentation for the system prompt."""
+        active = self._registry.get_active_tools()
+        mcp_tools = self._mcp.get_tools()
+
+        schema_parts = []
+        for name, tool in active.items():
+            schema_parts.append(f"### {name}\n{tool.description}")
+            if hasattr(tool, "get_schema"):
+                params = tool.get_schema()
+                props = params.get("properties", {})
+                required = params.get("required", [])
+                if props:
+                    schema_parts.append("Parameters:")
+                    for pname, pinfo in props.items():
+                        req = " (required)" if pname in required else ""
+                        ptype = pinfo.get("type", "any")
+                        desc = pinfo.get("description", "")
+                        schema_parts.append(f"  - {pname}: {ptype}{req} — {desc}")
+            schema_parts.append("")
+
+        # Add MCP tools
+        for name, bridge in mcp_tools.items():
+            schema_parts.append(f"### {name}\n{bridge.description}")
+            params = bridge._schema.get("inputSchema", {}).get("properties", {})
+            for pname, pinfo in params.items():
+                desc = pinfo.get("description", "")
+                schema_parts.append(f"  - {pname}: {desc}")
+            schema_parts.append("")
+
+        if not schema_parts:
+            return "No tools available."
+
+        return "## Available Tools\n\n" + "\n".join(schema_parts)
+
+    def _build_system_prompt(self, tool_schemas: str) -> str:
+        """Build the full system prompt."""
+        parts = [SYSTEM_PROMPT, "\n", tool_schemas]
+
+        # Add model-specific tips
+        model_key = self._settings.default_model
+        if "qwen" in model_key.lower():
+            parts.append(
+                "\n## Tips for Qwen Models\n"
+                "- Be explicit and structured in your responses\n"
+                "- Use complete file paths in tool calls\n"
+                "- Break complex tasks into clear steps"
+            )
+        elif "deepseek" in model_key.lower():
+            parts.append(
+                "\n## Tips for DeepSeek Models\n"
+                "- Think carefully before acting\n"
+                "- Use comments to explain complex code\n"
+                "- Verify code correctness with test commands"
+            )
+
+        return "\n".join(parts)
+
+    def clear_history(self) -> None:
+        """Clear conversation history."""
+        self._history.clear()
+        self._total_tokens = 0
+        self._tool_calls = 0
+
+
+# ── Register Built-in Tools ────────────────────────────────────────────
+
+def register_builtin_tools(registry: ToolRegistry) -> None:
+    """Register all built-in tools into the registry."""
+    for tools_module in [FS_TOOLS, BASH_TOOLS, WEB_TOOLS, TODO_TOOLS]:
+        for tool in tools_module:
+            registry.register(tool)

sonicoder/agent/protocol.py

@@ -0,0 +1,106 @@
+"""Agent protocol — event-driven architecture from Gemini CLI.
+
+The agent is purely event-driven. All communication flows through typed events.
+This is the core architectural pattern borrowed from Gemini CLI's AgentProtocol.
+"""
+
+from __future__ import annotations
+
+import time
+import uuid
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Callable
+
+
+# ── Event Types ────────────────────────────────────────────────────────
+
+class AgentEventType(Enum):
+    """All possible agent events — discriminated union pattern from Gemini CLI."""
+    INITIALIZE = "initialize"
+    STATUS = "status"
+    MESSAGE = "message"            # Agent reasoning/text output
+    TOOL_REQUEST = "tool_request"  # Agent wants to call a tool
+    TOOL_RESPONSE = "tool_response"  # Tool result returned
+    STREAMING = "streaming"        # Incremental token
+    COMPLETE = "complete"          # Agent finished (success)
+    ERROR = "error"                # Agent finished (error)
+    USAGE = "usage"                # Token usage stats
+
+
+@dataclass
+class AgentEvent:
+    """A single event in the agent event stream."""
+    type: AgentEventType
+    data: dict[str, Any] = field(default_factory=dict)
+    timestamp: float = field(default_factory=time.time)
+    event_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "type": self.type.value,
+            "data": self.data,
+            "timestamp": self.timestamp,
+            "event_id": self.event_id,
+        }
+
+
+# ── Agent Protocol Interface ───────────────────────────────────────────
+
+class AgentProtocol:
+    """Event-driven agent interface.
+
+    Inspired by Gemini CLI's AgentProtocol:
+    - send(): Submit a user message
+    - subscribe(): Listen for events
+    - abort(): Cancel current operation
+    """
+
+    def __init__(self):
+        self._subscribers: list[Callable[[AgentEvent], None]] = []
+        self._events: list[AgentEvent] = []
+        self._aborted = False
+        self._stream_id: str | None = None
+
+    def send(self, message: str, **kwargs) -> str:
+        """Submit a user message. Returns stream_id."""
+        self._aborted = False
+        self._stream_id = str(uuid.uuid4())[:12]
+        self._emit(AgentEventType.INITIALIZE, {
+            "stream_id": self._stream_id,
+            "message": message,
+        })
+        return self._stream_id
+
+    def subscribe(self, callback: Callable[[AgentEvent], None]) -> Callable:
+        """Subscribe to events. Returns unsubscribe function."""
+        self._subscribers.append(callback)
+        def unsubscribe():
+            self._subscribers.remove(callback)
+        return unsubscribe
+
+    def abort(self) -> None:
+        """Cancel the current operation."""
+        self._aborted = True
+
+    @property
+    def is_aborted(self) -> bool:
+        return self._aborted
+
+    def _emit(self, event_type: AgentEventType, data: dict[str, Any] | None = None) -> AgentEvent:
+        """Emit an event to all subscribers."""
+        event = AgentEvent(type=event_type, data=data or {})
+        self._events.append(event)
+        for sub in self._subscribers:
+            try:
+                sub(event)
+            except Exception:
+                pass
+        return event
+
+    @property
+    def events(self) -> list[AgentEvent]:
+        return list(self._events)
+
+    def clear_history(self) -> None:
+        self._events.clear()

sonicoder/config/__init__.py

@@ -0,0 +1,225 @@
+"""Configuration system — hierarchical settings like Claude Code."""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+# ── Defaults ──────────────────────────────────────────────────────────
+
+WORKSPACE_ROOT = Path(os.environ.get("SONICODER_WORKSPACE", "./workspace"))
+CONFIG_DIR = Path(os.environ.get("SONICODER_CONFIG", ".sonicoder"))
+
+DEFAULT_TEMPERATURE = 0.35
+DEFAULT_MAX_TOKENS = 4096
+DEFAULT_MAX_ITERATIONS = 12
+AGENT_LOOP_TIMEOUT = 300  # seconds
+BASH_TIMEOUT = 120
+PYTHON_TIMEOUT = 30
+
+
+@dataclass
+class ModelConfig:
+    """Configuration for a single model."""
+    name: str
+    model_id: str
+    model_type: str  # "text" | "vlm"
+    size_gb: float
+    description: str
+    dtype: str = "float16"
+    trust_remote_code: bool = True
+    device: str = "auto"
+
+
+@dataclass
+class MCPServerConfig:
+    """Configuration for an MCP server."""
+    name: str
+    transport: str  # "stdio" | "sse"
+    command: str | None = None
+    args: list[str] = field(default_factory=list)
+    env: dict[str, str] = field(default_factory=dict)
+    url: str | None = None
+    enabled: bool = True
+
+
+@dataclass
+class PolicyRule:
+    """A policy rule for tool approval."""
+    tool_name: str  # supports "*" wildcard
+    args_pattern: str | None = None
+    approval: str = "ask"  # "allow" | "ask" | "deny"
+    priority: int = 0
+    description: str = ""
+
+
+@dataclass
+class Settings:
+    """Hierarchical settings — cascades from global → project → local."""
+    # Model
+    default_model: str = "qwen25-coder-1.5b"
+    temperature: float = DEFAULT_TEMPERATURE
+    max_tokens: int = DEFAULT_MAX_TOKENS
+
+    # Agent
+    max_iterations: int = DEFAULT_MAX_ITERATIONS
+    agent_timeout: int = AGENT_LOOP_TIMEOUT
+
+    # Tools
+    enabled_tools: list[str] = field(default_factory=lambda: ["*"])
+    disabled_tools: list[str] = field(default_factory=list)
+
+    # Safety
+    sandbox_enabled: bool = True
+    bash_timeout: int = BASH_TIMEOUT
+    python_timeout: int = PYTHON_TIMEOUT
+
+    # MCP
+    mcp_servers: list[MCPServerConfig] = field(default_factory=list)
+
+    # Policy
+    policy_rules: list[PolicyRule] = field(default_factory=list)
+
+    # Context
+    max_context_tokens: int = 32768
+    auto_compact: bool = True
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> Settings:
+        """Create settings from a dictionary, ignoring unknown fields."""
+        known_fields = {f.name for f in cls.__dataclass_fields__.values()}
+        filtered = {k: v for k, v in data.items() if k in known_fields}
+        return cls(**filtered)
+
+    def merge(self, other: Settings) -> Settings:
+        """Merge another settings object (other takes precedence for non-default values)."""
+        result = Settings(
+            default_model=other.default_model or self.default_model,
+            temperature=other.temperature if other.temperature != DEFAULT_TEMPERATURE else self.temperature,
+            max_tokens=other.max_tokens if other.max_tokens != DEFAULT_MAX_TOKENS else self.max_tokens,
+            max_iterations=other.max_iterations if other.max_iterations != DEFAULT_MAX_ITERATIONS else self.max_iterations,
+            agent_timeout=other.agent_timeout if other.agent_timeout != AGENT_LOOP_TIMEOUT else self.agent_timeout,
+            sandbox_enabled=other.sandbox_enabled,
+            bash_timeout=other.bash_timeout,
+            python_timeout=other.python_timeout,
+            auto_compact=other.auto_compact,
+        )
+        # Merge lists
+        result.enabled_tools = other.enabled_tools if other.enabled_tools != ["*"] else self.enabled_tools
+        result.disabled_tools = list(set(self.disabled_tools + other.disabled_tools))
+        result.mcp_servers = self.mcp_servers + other.mcp_servers
+        result.policy_rules = other.policy_rules + self.policy_rules
+        return result
+
+
+# ── Model Registry ────────────────────────────────────────────────────
+
+MODEL_REGISTRY: dict[str, ModelConfig] = {
+    "qwen25-coder-1.5b": ModelConfig(
+        name="Qwen2.5-Coder-1.5B",
+        model_id="Qwen/Qwen2.5-Coder-1.5B-Instruct",
+        model_type="text",
+        size_gb=2.9,
+        description="Best-in-class 1.5B code model. Excellent at Python, JS, TypeScript, and multi-file generation.",
+    ),
+    "deepseek-coder-1.3b": ModelConfig(
+        name="DeepSeek-Coder-1.3B",
+        model_id="deepseek-ai/deepseek-coder-1.3b-instruct",
+        model_type="text",
+        size_gb=2.6,
+        description="Strong code understanding and generation. Good at multi-language coding tasks.",
+    ),
+    "qwen25-1.5b": ModelConfig(
+        name="Qwen2.5-1.5B-Instruct",
+        model_id="Qwen/Qwen2.5-1.5B-Instruct",
+        model_type="text",
+        size_gb=2.9,
+        description="General-purpose 1.5B model. Good for reasoning, analysis, and non-code tasks.",
+    ),
+    "smolvlm-2.2b": ModelConfig(
+        name="SmolVLM-2.2B",
+        model_id="HuggingFaceTB/SmolVLM2-2.2B-Instruct",
+        model_type="vlm",
+        size_gb=4.4,
+        description="Vision-language model. Can understand screenshots, diagrams, and generate code from images.",
+    ),
+    "minicpm5-1b": ModelConfig(
+        name="MiniCPM5-1B",
+        model_id="openbmb/MiniCPM5-1B",
+        model_type="text",
+        size_gb=2.17,
+        description="Compact text model. Fast inference, decent code generation.",
+    ),
+    "tinyllama-1.1b": ModelConfig(
+        name="TinyLlama-1.1B-Chat",
+        model_id="TinyLlama/TinyLlama-1.1B-Chat-v1.0",
+        model_type="text",
+        size_gb=2.2,
+        description="Ultra-compact chat model. Good for simple tasks, fastest inference.",
+    ),
+}
+
+
+def get_available_models() -> dict[str, ModelConfig]:
+    """Return all registered models."""
+    return MODEL_REGISTRY.copy()
+
+
+def get_model_config(model_key: str) -> ModelConfig | None:
+    """Get a model config by key."""
+    return MODEL_REGISTRY.get(model_key)
+
+
+def load_settings(project_dir: Path | None = None) -> Settings:
+    """Load settings with hierarchical cascading.
+
+    Priority (highest → lowest):
+    1. Environment variables
+    2. .sonicoder/settings.local.json (local overrides, gitignored)
+    3. .sonicoder/settings.json (project settings)
+    4. Defaults
+    """
+    settings = Settings()
+
+    search_dirs = [Path.cwd()]
+    if project_dir:
+        search_dirs.insert(0, project_dir)
+
+    # Load project-level settings
+    for d in search_dirs:
+        local_path = d / CONFIG_DIR / "settings.local.json"
+        proj_path = d / CONFIG_DIR / "settings.json"
+
+        for path in [local_path, proj_path]:
+            if path.exists():
+                try:
+                    data = json.loads(path.read_text(encoding="utf-8"))
+                    # Parse MCP server configs
+                    if "mcp_servers" in data:
+                        data["mcp_servers"] = [
+                            MCPServerConfig(**s) for s in data["mcp_servers"]
+                        ]
+                    if "policy_rules" in data:
+                        data["policy_rules"] = [
+                            PolicyRule(**r) for r in data["policy_rules"]
+                        ]
+                    settings = settings.merge(Settings.from_dict(data))
+                except (json.JSONDecodeError, TypeError) as e:
+                    print(f"[config] Warning: Failed to parse {path}: {e}")
+
+    # Environment variable overrides
+    env_model = os.environ.get("SONICODER_MODEL")
+    if env_model and env_model in MODEL_REGISTRY:
+        settings.default_model = env_model
+
+    env_temp = os.environ.get("SONICODER_TEMPERATURE")
+    if env_temp:
+        try:
+            settings.temperature = float(env_temp)
+        except ValueError:
+            pass
+
+    return settings

sonicoder/context/__init__.py

@@ -0,0 +1,154 @@
+"""Context window manager — conversation history with compression.
+
+Inspired by Gemini CLI's graph-based context pipeline and Claude Code's auto-compaction.
+For 1B models, context window is small (~32K tokens), so compression is critical.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ContextNode:
+    """A single node in the context graph."""
+    role: str  # "user" | "assistant" | "tool" | "system"
+    content: str
+    tokens: int = 0
+    compressed: bool = False
+    protected: bool = False  # Can't be GC'd
+    turn_index: int = 0
+
+
+class ContextManager:
+    """Manages conversation context within a token budget.
+
+    Strategies (inspired by both Gemini CLI and Claude Code):
+    1. Clear old tool outputs first (biggest token consumers)
+    2. Compress old conversation turns
+    3. Protect recent turns and system messages
+    4. Lazy tool schema loading (deferred definitions like Claude Code)
+    """
+
+    def __init__(self, max_tokens: int = 32768, auto_compact: bool = True):
+        self.max_tokens = max_tokens
+        self.auto_compact = auto_compact
+        self._nodes: list[ContextNode] = []
+        self._turn_count = 0
+
+    def add_message(self, role: str, content: str, protected: bool = False) -> None:
+        """Add a message to the context."""
+        tokens = self._estimate_tokens(content)
+        node = ContextNode(
+            role=role,
+            content=content,
+            tokens=tokens,
+            protected=protected,
+            turn_index=self._turn_count,
+        )
+        self._nodes.append(node)
+        if role == "user":
+            self._turn_count += 1
+
+    def get_messages(self) -> list[dict[str, str]]:
+        """Get all messages for the LLM, with auto-compaction if needed."""
+        if self.auto_compact:
+            self._maybe_compact()
+
+        return [
+            {"role": n.role, "content": n.content}
+            for n in self._nodes
+            if n.content  # Skip empty nodes
+        ]
+
+    def total_tokens(self) -> int:
+        """Estimate total tokens in context."""
+        return sum(n.tokens for n in self._nodes)
+
+    def clear(self) -> None:
+        """Clear all context."""
+        self._nodes.clear()
+        self._turn_count = 0
+
+    def _maybe_compact(self) -> None:
+        """Auto-compact if over budget."""
+        total = self.total_tokens()
+        if total <= self.max_tokens:
+            return
+
+        logger.info(f"Context over budget ({total} > {self.max_tokens}). Compacting...")
+
+        # Phase 1: Remove old tool results (biggest savings)
+        budget_remaining = self.max_tokens * 0.8  # Target 80% utilization
+        for node in self._nodes:
+            if self.total_tokens() <= budget_remaining:
+                break
+            if node.role == "tool" and not node.protected:
+                # Compress tool output
+                if len(node.content) > 200:
+                    node.content = node.content[:200] + f"\n... [truncated, was {len(node.content)} chars]"
+                    node.tokens = self._estimate_tokens(node.content)
+                    node.compressed = True
+
+        # Phase 2: Summarize old conversation turns
+        if self.total_tokens() > self.max_tokens:
+            # Find oldest non-protected assistant/user pairs to compress
+            pairs = self._find_compactable_pairs()
+            for user_node, assistant_node in pairs:
+                if self.total_tokens() <= budget_remaining:
+                    break
+                # Replace with summary
+                summary = self._summarize_pair(user_node, assistant_node)
+                user_node.content = summary
+                user_node.tokens = self._estimate_tokens(summary)
+                user_node.compressed = True
+                # Remove assistant node
+                self._nodes.remove(assistant_node)
+
+        # Phase 3: If still over budget, remove oldest non-protected nodes
+        if self.total_tokens() > self.max_tokens:
+            to_remove = []
+            for node in self._nodes:
+                if not node.protected and not node.compressed:
+                    to_remove.append(node)
+                    if self.total_tokens() - sum(n.tokens for n in to_remove) <= budget_remaining:
+                        break
+            for node in to_remove:
+                self._nodes.remove(node)
+
+        logger.info(f"Compaction done. Context now ~{self.total_tokens()} tokens")
+
+    def _find_compactable_pairs(self) -> list[tuple[ContextNode, ContextNode]]:
+        """Find user+assistant pairs that can be compressed."""
+        pairs = []
+        i = 0
+        while i < len(self._nodes) - 1:
+            if (self._nodes[i].role == "user" and
+                self._nodes[i + 1].role == "assistant" and
+                not self._nodes[i].protected):
+                pairs.append((self._nodes[i], self._nodes[i + 1]))
+                i += 2
+            else:
+                i += 1
+        return pairs
+
+    def _summarize_pair(self, user: ContextNode, assistant: ContextNode) -> str:
+        """Create a brief summary of a conversation pair."""
+        u_preview = user.content[:100].replace("\n", " ")
+        a_preview = assistant.content[:150].replace("\n", " ")
+        return f"[Earlier: User asked about '{u_preview}...' and assistant responded with '{a_preview}...']"
+
+    @staticmethod
+    def _estimate_tokens(text: str) -> int:
+        """Rough token estimation (1 token ~ 4 chars for code, ~3.5 for English)."""
+        if not text:
+            return 0
+        # Detect if mostly code
+        code_chars = sum(1 for c in text if c in " \t{}[]();:=<>+-*/!&|\\")
+        ratio = code_chars / max(len(text), 1)
+        chars_per_token = 3.5 + ratio * 1.5  # 3.5 for English, 5 for code
+        return int(len(text) / chars_per_token)

sonicoder/mcp/__init__.py

@@ -0,0 +1,336 @@
+"""MCP (Model Context Protocol) support — full client implementation.
+
+Architecture from Gemini CLI's McpClientManager:
+- Manages lifecycle of multiple MCP servers
+- Bridges MCP tools into the ToolRegistry as ToolBuilder instances
+- Supports stdio and SSE transports
+- Lazy initialization (connects on first use)
+
+Based on the MCP Python SDK.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import threading
+from typing import Any, Callable
+
+from ..config import MCPServerConfig
+
+logger = logging.getLogger(__name__)
+
+
+# ── MCP Tool Bridge ────────────────────────────────────────────────────
+
+class MCPToolBridge:
+    """Wraps an MCP tool as a ToolBuilder, making it indistinguishable from built-in tools.
+
+    This is the key insight from Gemini CLI: MCP tools use the same ToolBuilder
+    interface so they get the same policy, confirmation, and scheduling treatment.
+    """
+
+    def __init__(self, server_name: str, tool_name: str, tool_schema: dict, call_fn: Callable):
+        self.server_name = server_name
+        self.tool_name = tool_name
+        self._schema = tool_schema
+        self._call_fn = call_fn
+
+        # Generate MCP-prefixed name
+        self.name = f"mcp_{server_name}_{tool_name}"
+        self.description = tool_schema.get("description", f"MCP tool: {tool_name}")
+        self.kind = "other"
+        self.is_read_only = tool_schema.get("annotations", {}).get("readOnlyHint", False)
+
+    def get_tool_definition(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "description": self.description,
+            "parameters": self._schema.get("inputSchema", {"type": "object", "properties": {}}),
+            "kind": self.kind,
+            "mcp_server": self.server_name,
+            "mcp_tool": self.tool_name,
+        }
+
+    def call(self, arguments: dict[str, Any]) -> dict[str, Any]:
+        """Execute the MCP tool call."""
+        return self._call_fn(self.tool_name, arguments)
+
+
+# ── MCP Client Manager ─────────────────────────────────────────────────
+
+class MCPClientManager:
+    """Manages multiple MCP server connections.
+
+    Lifecycle: configure → connect (lazy) → use → disconnect
+    """
+
+    def __init__(self):
+        self._configs: dict[str, MCPServerConfig] = {}
+        self._clients: dict[str, Any] = {}  # server_name → MCP client/session
+        self._tools: dict[str, MCPToolBridge] = {}
+        self._connected: set[str] = set()
+        self._lock = threading.Lock()
+
+    def configure(self, configs: list[MCPServerConfig]) -> None:
+        """Set MCP server configurations."""
+        self._configs = {c.name: c for c in configs if c.enabled}
+
+    def add_server(self, config: MCPServerConfig) -> None:
+        """Add or update a single MCP server config."""
+        self._configs[config.name] = config
+        # If already connected to this server, disconnect and reconnect
+        if config.name in self._connected:
+            self._disconnect(config.name)
+
+    def remove_server(self, name: str) -> None:
+        """Remove an MCP server."""
+        self._configs.pop(name, None)
+        self._disconnect(name)
+
+    def connect_all(self) -> dict[str, str | None]:
+        """Connect to all configured MCP servers. Returns {name: error_or_None}."""
+        results = {}
+        for name, config in self._configs.items():
+            if config.enabled:
+                try:
+                    error = self._connect(name, config)
+                    results[name] = error
+                except Exception as e:
+                    results[name] = str(e)
+                    logger.error(f"MCP connect failed for {name}: {e}")
+        return results
+
+    def disconnect_all(self) -> None:
+        """Disconnect all MCP servers."""
+        for name in list(self._connected):
+            self._disconnect(name)
+
+    def get_tools(self) -> dict[str, MCPToolBridge]:
+        """Get all discovered MCP tools."""
+        return dict(self._tools)
+
+    def get_tool(self, name: str) -> MCPToolBridge | None:
+        """Get a specific MCP tool by its full name (mcp_server_tool)."""
+        return self._tools.get(name)
+
+    def call_tool(self, server_name: str, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+        """Call an MCP tool on a specific server."""
+        client = self._clients.get(server_name)
+        if not client:
+            # Try lazy connect
+            config = self._configs.get(server_name)
+            if not config:
+                return {"error": f"MCP server '{server_name}' not configured"}
+            error = self._connect(server_name, config)
+            if error:
+                return {"error": f"Failed to connect to '{server_name}': {error}"}
+            client = self._clients.get(server_name)
+            if not client:
+                return {"error": f"MCP server '{server_name}' connection failed"}
+
+        try:
+            if hasattr(client, "call_tool"):
+                result = client.call_tool(tool_name, arguments)
+                if hasattr(result, "content"):
+                    # Extract text content from MCP result
+                    texts = []
+                    for item in result.content:
+                        if hasattr(item, "text"):
+                            texts.append(item.text)
+                        else:
+                            texts.append(str(item))
+                    return {"output": "\n".join(texts)}
+                return {"output": str(result)}
+            return {"output": str(client)}
+        except Exception as e:
+            return {"error": f"MCP tool call failed: {e}"}
+
+    def list_resources(self, server_name: str | None = None) -> list[dict]:
+        """List available MCP resources."""
+        resources = []
+        servers = [server_name] if server_name else list(self._connected)
+        for name in servers:
+            client = self._clients.get(name)
+            if client and hasattr(client, "list_resources"):
+                try:
+                    result = client.list_resources()
+                    for r in (result.resources if hasattr(result, "resources") else []):
+                        resources.append({
+                            "server": name,
+                            "uri": getattr(r, "uri", str(r)),
+                            "name": getattr(r, "name", ""),
+                            "description": getattr(r, "description", ""),
+                        })
+                except Exception as e:
+                    logger.warning(f"Failed to list resources from {name}: {e}")
+        return resources
+
+    def get_status(self) -> dict[str, Any]:
+        """Get status of all MCP servers."""
+        status = {}
+        for name, config in self._configs.items():
+            status[name] = {
+                "enabled": config.enabled,
+                "connected": name in self._connected,
+                "transport": config.transport,
+                "tool_count": sum(1 for t in self._tools.values() if t.server_name == name),
+            }
+        return status
+
+    # ── Internal ───────────────────────────────────────────────────────
+
+    def _connect(self, name: str, config: MCPServerConfig) -> str | None:
+        """Connect to a single MCP server. Returns error message or None."""
+        if name in self._connected:
+            return None
+
+        try:
+            # Try to import MCP SDK
+            try:
+                from mcp import ClientSession, StdioServerParameters
+                from mcp.client.stdio import stdio_client
+            except ImportError:
+                logger.info("MCP SDK not installed. Skipping MCP server %s", name)
+                return "mcp SDK not installed (pip install mcp)"
+
+            if config.transport == "stdio":
+                server_params = StdioServerParameters(
+                    command=config.command or "",
+                    args=config.args,
+                    env=config.env or None,
+                )
+
+                # Run async connect in a new thread with event loop
+                client_session, error = self._run_async_connect(name, server_params)
+                if error or not client_session:
+                    return error or f"Failed to connect to {name}"
+
+                self._clients[name] = client_session
+                self._connected.add(name)
+
+                # Discover tools
+                self._discover_tools(name, client_session)
+                logger.info(f"MCP server '{name}' connected with {sum(1 for t in self._tools.values() if t.server_name == name)} tools")
+                return None
+
+            elif config.transport == "sse":
+                return f"SSE transport not yet supported for {name}"
+
+            return f"Unknown transport: {config.transport}"
+
+        except Exception as e:
+            logger.error(f"MCP connect error for {name}: {e}")
+            return str(e)
+
+    def _run_async_connect(self, name: str, server_params):
+        """Run async MCP connection in a sync context."""
+        from mcp import ClientSession, StdioServerParameters
+        from mcp.client.stdio import stdio_client
+
+        result = {"session": None, "error": None}
+
+        async def _connect_async():
+            try:
+                async with stdio_client(server_params) as (read, write):
+                    async with ClientSession(read, write) as session:
+                        await session.initialize()
+                        result["session"] = _AsyncSessionWrapper(session)
+                        # Keep the context managers alive by blocking
+                        await asyncio.Future()  # Block forever
+            except Exception as e:
+                result["error"] = str(e)
+
+        loop = asyncio.new_event_loop()
+        thread = threading.Thread(target=loop.run_until_complete, args=(_connect_async(),), daemon=True)
+        thread.start()
+        thread.join(timeout=10)
+
+        if result["error"] and not result["session"]:
+            return None, result["error"]
+        if result["session"]:
+            return result["session"], None
+        return None, "Connection timeout"
+
+    def _discover_tools(self, server_name: str, client) -> None:
+        """Discover and register tools from an MCP server."""
+        # Remove old tools for this server
+        old_prefix = f"mcp_{server_name}_"
+        for k in list(self._tools.keys()):
+            if k.startswith(old_prefix):
+                del self._tools[k]
+
+        # Discover new tools (sync wrapper)
+        try:
+            tools_result = client.list_tools() if hasattr(client, "list_tools") else None
+            tools = tools_result.tools if hasattr(tools_result, "tools") else []
+
+            for tool in tools:
+                schema = {
+                    "description": getattr(tool, "description", ""),
+                    "inputSchema": getattr(tool, "inputSchema", {"type": "object", "properties": {}}),
+                    "annotations": getattr(tool, "annotations", {}),
+                }
+                tool_name = getattr(tool, "name", "unknown")
+
+                bridge = MCPToolBridge(
+                    server_name=server_name,
+                    tool_name=tool_name,
+                    tool_schema=schema,
+                    call_fn=lambda sn=server_name, tn=tool_name, args=None: self.call_tool(sn, tn, args or {}),
+                )
+                self._tools[bridge.name] = bridge
+                logger.debug(f"Discovered MCP tool: {bridge.name}")
+
+        except Exception as e:
+            logger.warning(f"Failed to discover tools from {server_name}: {e}")
+
+    def _disconnect(self, name: str) -> None:
+        """Disconnect from an MCP server."""
+        self._clients.pop(name, None)
+        self._connected.discard(name)
+        # Remove tools
+        prefix = f"mcp_{name}_"
+        for k in list(self._tools.keys()):
+            if k.startswith(prefix):
+                del self._tools[k]
+
+
+class _AsyncSessionWrapper:
+    """Wraps an MCP ClientSession for sync access."""
+
+    def __init__(self, session):
+        self._session = session
+
+    def __getattr__(self, name):
+        attr = getattr(self._session, name)
+        if asyncio.iscoroutinefunction(attr):
+            def sync_wrapper(*args, **kwargs):
+                try:
+                    loop = asyncio.get_event_loop()
+                    if loop.is_running():
+                        # Can't await in running loop, return a placeholder
+                        import concurrent.futures
+                        with concurrent.futures.ThreadPoolExecutor() as pool:
+                            future = pool.submit(asyncio.run, attr(*args, **kwargs))
+                            return future.result(timeout=30)
+                    else:
+                        return loop.run_until_complete(attr(*args, **kwargs))
+                except RuntimeError:
+                    return asyncio.run(attr(*args, **kwargs))
+            return sync_wrapper
+        return attr
+
+
+# ── Global Manager ─────────────────────────────────────────────────────
+
+_global_manager: MCPClientManager | None = None
+
+
+def get_mcp_manager() -> MCPClientManager:
+    """Get or create the global MCP client manager."""
+    global _global_manager
+    if _global_manager is None:
+        _global_manager = MCPClientManager()
+    return _global_manager

sonicoder/model/inference.py

@@ -0,0 +1,225 @@
+"""Streaming inference engine — supports both text and VLM models.
+
+Optimized for 1B models with proper streaming via TextIteratorStreamer.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+import threading
+from typing import Any, Generator
+
+from transformers import TextIteratorStreamer
+
+from ..config import MODEL_REGISTRY
+from .loader import get_model_and_tokenizer, get_model_status
+
+logger = logging.getLogger(__name__)
+
+# ── Regex Patterns ─────────────────────────────────────────────────────
+
+THINKING_BLOCK_RE = re.compile(r"<think>(.*?)</think>", re.DOTALL)
+CODE_BLOCK_RE = re.compile(r"```(\w*)\n(.*?)```", re.DOTALL)
+FILE_BLOCK_RE = re.compile(r"@@FILE:(.+?)@@\n(.*?)@@END@@", re.DOTALL)
+TOOL_CALL_RE = re.compile(r"```tool\s*\n(.*?)```", re.DOTALL)
+
+
+def _build_text_prompt(messages: list[dict[str, str]], system_prompt: str) -> str:
+    """Build a chat-style prompt for text models.
+
+    Uses a simple but effective template that works well with Qwen and DeepSeek models.
+    """
+    parts = [f"System: {system_prompt}\n"]
+
+    for msg in messages:
+        role = msg.get("role", "user")
+        content = msg.get("content", "")
+        if role == "system":
+            parts.append(f"System: {content}\n")
+        elif role == "user":
+            parts.append(f"User: {content}\n")
+        elif role == "assistant":
+            parts.append(f"Assistant: {content}\n")
+        elif role == "tool":
+            parts.append(f"Tool Result: {content}\n")
+
+    parts.append("Assistant:")
+    return "\n".join(parts)
+
+
+def call_model_stream(
+    messages: list[dict[str, str]],
+    system_prompt: str = "",
+    max_new_tokens: int = 4096,
+    temperature: float = 0.35,
+    image_url: str | None = None,
+) -> Generator[str, None, None]:
+    """Stream tokens from the model.
+
+    Yields partial text strings as they're generated.
+    Falls back to non-streaming if TextIteratorStreamer isn't available.
+    """
+    status = get_model_status()
+    if status["status"] != "ready":
+        raise RuntimeError(f"Model not ready: {status}")
+
+    model_key = status["model_key"]
+    config = MODEL_REGISTRY[model_key]
+
+    if config.model_type == "vlm" and image_url:
+        yield from _call_vlm_stream(messages, system_prompt, max_new_tokens, temperature, image_url)
+    else:
+        yield from _call_text_stream(messages, system_prompt, max_new_tokens, temperature)
+
+
+def _call_text_stream(
+    messages: list[dict[str, str]],
+    system_prompt: str,
+    max_new_tokens: int,
+    temperature: float,
+) -> Generator[str, None, None]:
+    """Stream from a text-only model."""
+    model, tokenizer = get_model_and_tokenizer()
+
+    prompt = _build_text_prompt(messages, system_prompt)
+    inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=32768)
+    inputs = {k: v.to(model.device) for k, v in inputs.items()}
+
+    # Try streaming first
+    try:
+        streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
+
+        generation_kwargs = {
+            **inputs,
+            "max_new_tokens": max_new_tokens,
+            "temperature": temperature,
+            "do_sample": temperature > 0,
+            "top_p": 0.9,
+            "top_k": 50,
+            "repetition_penalty": 1.1,
+            "streamer": streamer,
+            "pad_token_id": tokenizer.eos_token_id,
+        }
+
+        thread = threading.Thread(
+            target=model.generate, kwargs=generation_kwargs, daemon=True
+        )
+        thread.start()
+
+        for token in streamer:
+            if token:
+                yield token
+
+        thread.join(timeout=60)
+
+    except Exception as e:
+        logger.warning(f"Streaming failed, falling back to non-streaming: {e}")
+        # Non-streaming fallback
+        with torch.no_grad():
+            outputs = model.generate(
+                **inputs,
+                max_new_tokens=max_new_tokens,
+                temperature=temperature,
+                do_sample=temperature > 0,
+                top_p=0.9,
+                top_k=50,
+                repetition_penalty=1.1,
+                pad_token_id=tokenizer.eos_token_id,
+            )
+        text = tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
+        yield text
+
+
+def _call_vlm_stream(
+    messages: list[dict[str, str]],
+    system_prompt: str,
+    max_new_tokens: int,
+    temperature: float,
+    image_url: str,
+) -> Generator[str, None, None]:
+    """Stream from a vision-language model."""
+    import torch
+    from PIL import Image
+    import requests
+
+    model, processor = get_model_and_tokenizer()
+
+    # Build messages for VLM
+    vlm_messages = []
+    if system_prompt:
+        vlm_messages.append({"role": "system", "content": system_prompt})
+
+    for msg in messages:
+        if msg["role"] == "user" and msg is messages[-1]:
+            # Add image to last user message
+            vlm_messages.append({
+                "role": "user",
+                "content": [
+                    {"type": "image", "url": image_url},
+                    {"type": "text", "text": msg["content"]},
+                ],
+            })
+        else:
+            vlm_messages.append(msg)
+
+    # Apply chat template
+    try:
+        prompt = processor.apply_chat_template(
+            vlm_messages, tokenize=False, add_generation_prompt=True
+        )
+    except Exception:
+        prompt = _build_text_prompt(messages, system_prompt)
+
+    # Process inputs
+    image = None
+    if image_url:
+        try:
+            if image_url.startswith(("http://", "https://")):
+                resp = requests.get(image_url, timeout=10)
+                image = Image.open(__import__("io").BytesIO(resp.content)).convert("RGB")
+            else:
+                image = Image.open(image_url).convert("RGB")
+        except Exception as e:
+            logger.warning(f"Failed to load image: {e}")
+
+    inputs = processor(text=prompt, images=image, return_tensors="pt", truncation=True)
+    inputs = {k: v.to(model.device) for k, v in inputs.items()}
+
+    # Non-streaming for VLM (most 1B VLMs don't support TextIteratorStreamer well)
+    with torch.no_grad():
+        outputs = model.generate(
+            **inputs,
+            max_new_tokens=max_new_tokens,
+            temperature=temperature,
+            do_sample=temperature > 0,
+            pad_token_id=processor.tokenizer.eos_token_id if hasattr(processor, "tokenizer") else 2,
+        )
+
+    # Decode
+    if hasattr(processor, "tokenizer"):
+        text = processor.tokenizer.decode(
+            outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True
+        )
+    elif hasattr(processor, "decode"):
+        text = processor.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
+    else:
+        text = str(outputs[0])
+
+    yield text
+
+
+def call_model_sync(
+    messages: list[dict[str, str]],
+    system_prompt: str = "",
+    max_new_tokens: int = 4096,
+    temperature: float = 0.35,
+    image_url: str | None = None,
+) -> str:
+    """Non-streaming call — collects all tokens and returns the full text."""
+    chunks = list(call_model_stream(messages, system_prompt, max_new_tokens, temperature, image_url))
+    return "".join(chunks)
+
+
+# Import torch here to avoid issues
+import torch

sonicoder/model/loader.py

@@ -0,0 +1,181 @@
+"""Model loader — multi-model management with hot swapping.
+
+Supports multiple smart 1B models. Only one model loaded at a time
+to conserve memory (important for HF Spaces free tier).
+"""
+
+from __future__ import annotations
+
+import gc
+import logging
+import threading
+from typing import Any, Callable
+
+import torch
+from transformers import AutoConfig
+
+from ..config import MODEL_REGISTRY, ModelConfig
+
+logger = logging.getLogger(__name__)
+
+# ── Module State ───────────────────────────────────────────────────────
+
+_model = None
+_processor = None
+_tokenizer = None
+_model_key: str | None = None
+_model_loaded = False
+_model_loading = False
+_load_error: str | None = None
+_lock = threading.Lock()
+_on_status_change: list[Callable] = []
+
+
+def _get_device() -> str:
+    """Auto-detect best device."""
+    if torch.cuda.is_available():
+        return "cuda"
+    if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
+        return "mps"
+    return "cpu"
+
+
+def _get_dtype(device: str) -> torch.dtype:
+    """Select dtype based on device."""
+    return torch.float16 if device == "cuda" else torch.float32
+
+
+def _unload_model() -> None:
+    """Unload current model and free memory."""
+    global _model, _processor, _tokenizer, _model_loaded
+    if _model is not None:
+        del _model
+        _model = None
+    if _processor is not None:
+        del _processor
+        _processor = None
+    if _tokenizer is not None:
+        del _tokenizer
+        _tokenizer = None
+    _model_loaded = False
+    gc.collect()
+    if torch.cuda.is_available():
+        torch.cuda.empty_cache()
+
+
+def load_model(model_key: str) -> None:
+    """Load a model by key. Blocks until loaded."""
+    global _model, _processor, _tokenizer, _model_key, _model_loaded, _model_loading, _load_error
+
+    config_entry = MODEL_REGISTRY.get(model_key)
+    if not config_entry:
+        raise ValueError(f"Unknown model: {model_key}. Available: {list(MODEL_REGISTRY.keys())}")
+
+    with _lock:
+        if _model_key == model_key and _model_loaded:
+            return  # Already loaded
+
+        _model_loading = True
+        _load_error = None
+        _notify_status()
+
+        try:
+            _unload_model()
+
+            device = _get_device()
+            dtype = _get_dtype(device)
+            logger.info(f"Loading {config_entry.name} ({config_entry.model_id}) on {device} ({dtype})...")
+
+            model_kwargs = {
+                "trust_remote_code": config_entry.trust_remote_code,
+                "low_cpu_mem_usage": True,
+                "torch_dtype": dtype,
+            }
+
+            if device == "cuda":
+                model_kwargs["device_map"] = "auto"
+            else:
+                model_kwargs["device_map"] = None
+
+            if config_entry.model_type == "vlm":
+                from transformers import AutoModelForImageTextToText, AutoProcessor
+                _model = AutoModelForImageTextToText.from_pretrained(
+                    config_entry.model_id, **model_kwargs
+                )
+                _processor = AutoProcessor.from_pretrained(
+                    config_entry.model_id, trust_remote_code=True
+                )
+                _tokenizer = None
+            else:
+                from transformers import AutoModelForCausalLM, AutoTokenizer
+                _model = AutoModelForCausalLM.from_pretrained(
+                    config_entry.model_id, **model_kwargs
+                )
+                _tokenizer = AutoTokenizer.from_pretrained(
+                    config_entry.model_id, trust_remote_code=True
+                )
+                _processor = None
+
+            if not device == "cuda" and hasattr(_model, "to"):
+                _model = _model.to(device)
+
+            _model.eval()
+            _model_key = model_key
+            _model_loaded = True
+            logger.info(f"Model loaded: {config_entry.name}")
+
+        except Exception as e:
+            _load_error = str(e)
+            logger.error(f"Failed to load model: {e}")
+            raise
+        finally:
+            _model_loading = False
+            _notify_status()
+
+
+def start_background_load(model_key: str | None = None) -> None:
+    """Start loading a model in a background thread."""
+    key = model_key or "qwen25-coder-1.5b"
+    t = threading.Thread(target=load_model, args=(key,), daemon=True)
+    t.start()
+
+
+def switch_model(model_key: str) -> None:
+    """Switch to a different model (loads in background)."""
+    start_background_load(model_key)
+
+
+def get_model_status() -> dict[str, Any]:
+    """Get current model status."""
+    config = MODEL_REGISTRY.get(_model_key) if _model_key else None
+    status = "error" if _load_error else ("ready" if _model_loaded else ("loading" if _model_loading else "unknown"))
+    return {
+        "status": status,
+        "model_key": _model_key,
+        "model_name": config.name if config else None,
+        "model_id": config.model_id if config else None,
+        "model_type": config.model_type if config else None,
+        "error": _load_error,
+    }
+
+
+def get_model_and_tokenizer() -> tuple:
+    """Get the current model and tokenizer/processor. Returns (model, tokenizer_or_processor)."""
+    if not _model_loaded or _model is None:
+        raise RuntimeError("Model not loaded")
+    return _model, _tokenizer or _processor
+
+
+def on_status_change(callback: Callable) -> None:
+    """Register a callback for model status changes."""
+    _on_status_change.append(callback)
+
+
+def _notify_status() -> None:
+    """Notify all status change listeners."""
+    status = get_model_status()
+    for cb in _on_status_change:
+        try:
+            cb(status)
+        except Exception:
+            pass

sonicoder/policy/__init__.py

@@ -0,0 +1,245 @@
+"""Policy engine — rule-based tool approval system.
+
+Inspired by Gemini CLI's PolicyEngine with priority-sorted rules
+and pattern matching. Also incorporates Claude Code's allow/ask/deny model.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import re
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+
+class ApprovalDecision(Enum):
+    ALLOW = "allow"
+    ASK = "ask"
+    DENY = "deny"
+
+
+@dataclass
+class PolicyCheckResult:
+    """Result of a policy check."""
+    decision: ApprovalDecision
+    reason: str = ""
+    matched_rule: str | None = None
+
+
+class PolicyEngine:
+    """Rule-based policy engine with priority ordering.
+
+    Rules are evaluated in priority order (highest first).
+    First matching rule wins.
+    Default: ASK for interactive, DENY for non-interactive.
+    """
+
+    # Built-in safety rules (always evaluated, highest priority)
+    BUILTIN_RULES = [
+        {
+            "tool_name": "bash",
+            "args_pattern": r"rm\s+-rf\s+(/|~|\*)",
+            "approval": "deny",
+            "priority": 1000,
+            "description": "Block recursive delete of root/home",
+        },
+        {
+            "tool_name": "bash",
+            "args_pattern": r"(shutdown|reboot|halt|poweroff)",
+            "approval": "deny",
+            "priority": 1000,
+            "description": "Block system shutdown commands",
+        },
+        {
+            "tool_name": "bash",
+            "args_pattern": r"(mkfs|fdisk|parted|dd\s+if=/dev/zero)",
+            "approval": "deny",
+            "priority": 1000,
+            "description": "Block disk destruction commands",
+        },
+        {
+            "tool_name": "bash",
+            "args_pattern": r"(chmod\s+-R\s+777|chown\s+-R)",
+            "approval": "deny",
+            "priority": 900,
+            "description": "Block dangerous permission changes",
+        },
+        {
+            "tool_name": "write_file",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -100,
+            "description": "Allow file writes in sandboxed workspace",
+        },
+        {
+            "tool_name": "edit_file",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -100,
+            "description": "Allow file edits in sandboxed workspace",
+        },
+        {
+            "tool_name": "read_file",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow file reads",
+        },
+        {
+            "tool_name": "glob",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow glob searches",
+        },
+        {
+            "tool_name": "grep",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow grep searches",
+        },
+        {
+            "tool_name": "list_dir",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow directory listing",
+        },
+        {
+            "tool_name": "web_search",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow web search",
+        },
+        {
+            "tool_name": "web_fetch",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow web fetch",
+        },
+        {
+            "tool_name": "todo",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow todo management",
+        },
+        {
+            "tool_name": "snapshot_workspace",
+            "args_pattern": None,
+            "approval": "allow",
+            "priority": -200,
+            "description": "Allow workspace snapshot",
+        },
+    ]
+
+    def __init__(self, interactive: bool = True):
+        self._rules: list[dict] = list(self.BUILTIN_RULES)
+        self._interactive = interactive
+        self._approval_mode: str = "default"  # default | auto_edit | yolo
+
+    @property
+    def approval_mode(self) -> str:
+        return self._approval_mode
+
+    @approval_mode.setter
+    def approval_mode(self, mode: str) -> None:
+        if mode in ("default", "auto_edit", "yolo"):
+            self._approval_mode = mode
+
+    def add_rule(self, rule: dict) -> None:
+        """Add a custom policy rule."""
+        rule.setdefault("priority", 0)
+        rule.setdefault("approval", "ask")
+        self._rules.append(rule)
+        self._rules.sort(key=lambda r: r["priority"], reverse=True)
+
+    def remove_rule(self, description: str) -> bool:
+        """Remove a rule by description."""
+        for i, r in enumerate(self._rules):
+            if r.get("description") == description:
+                self._rules.pop(i)
+                return True
+        return False
+
+    def set_rules(self, rules: list[dict]) -> None:
+        """Replace all custom rules (keeps built-in rules)."""
+        self._rules = list(self.BUILTIN_RULES)
+        for r in rules:
+            self.add_rule(r)
+
+    def check(self, tool_name: str, args: dict[str, Any] | None = None) -> PolicyCheckResult:
+        """Check if a tool call is allowed.
+
+        Returns the decision and reason from the highest-priority matching rule.
+        """
+        args_str = str(args) if args else ""
+
+        for rule in self._rules:
+            # Check tool name match (supports wildcards)
+            if not self._match_tool_name(tool_name, rule["tool_name"]):
+                continue
+
+            # Check args pattern if specified
+            if rule.get("args_pattern"):
+                try:
+                    if not re.search(rule["args_pattern"], args_str, re.IGNORECASE):
+                        continue
+                except re.error:
+                    continue
+
+            # Match found — return this rule's decision
+            approval = rule["approval"]
+
+            # YOLO mode: upgrade DENY to ASK (except built-in high-priority)
+            if self._approval_mode == "yolo" and rule["priority"] < 900:
+                if approval == "deny":
+                    approval = "ask"
+
+            # Auto-edit mode: auto-approve file writes/edits
+            if self._approval_mode == "auto_edit" and tool_name in ("write_file", "edit_file"):
+                approval = "allow"
+
+            return PolicyCheckResult(
+                decision=ApprovalDecision(approval),
+                reason=rule.get("description", f"Rule matched: {rule['tool_name']}"),
+                matched_rule=rule.get("description"),
+            )
+
+        # Default: ASK for interactive, DENY for non-interactive
+        default = ApprovalDecision.ASK if self._interactive else ApprovalDecision.DENY
+        return PolicyCheckResult(
+            decision=default,
+            reason=f"No matching rule for {tool_name} (default: {default.value})",
+        )
+
+    def _match_tool_name(self, name: str, pattern: str) -> bool:
+        """Match tool name against pattern (supports * wildcards)."""
+        if pattern == "*":
+            return True
+        if "*" in pattern:
+            # Convert glob to regex
+            regex = pattern.replace("*", ".*")
+            return bool(re.match(f"^{regex}$", name))
+        return name == pattern
+
+    def get_rules(self) -> list[dict]:
+        """Get all rules for display."""
+        return list(self._rules)
+
+
+# ── Global Engine ───────────────────────────────────────────────────────
+
+_global_engine: PolicyEngine | None = None
+
+
+def get_policy_engine() -> PolicyEngine:
+    """Get or create the global policy engine."""
+    global _global_engine
+    if _global_engine is None:
+        _global_engine = PolicyEngine()
+    return _global_engine

sonicoder/server/__init__.py

@@ -0,0 +1,381 @@
+"""Server routes — FastAPI + Gradio API endpoints.
+
+All HTTP and Gradio API endpoints for the application.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from pathlib import Path
+from typing import Any, Generator
+
+from ..config import (
+    Settings, get_available_models, load_settings,
+    WORKSPACE_ROOT, CONFIG_DIR,
+)
+from ..model.loader import (
+    get_model_status, start_background_load, switch_model,
+)
+from ..tools import get_tool_registry, ToolRegistry
+from ..tools.fs import BUILTIN_TOOLS as FS_TOOLS
+from ..tools.bash import BUILTIN_TOOLS as BASH_TOOLS
+from ..tools.web import BUILTIN_TOOLS as WEB_TOOLS
+from ..tools.todos import BUILTIN_TOOLS as TODO_TOOLS
+from ..agent import AgentSession, register_builtin_tools
+from ..mcp import get_mcp_manager
+from ..policy import get_policy_engine
+
+logger = logging.getLogger(__name__)
+
+# ── Global State ───────────────────────────────────────────────────────
+
+_settings: Settings | None = None
+_registry: ToolRegistry | None = None
+_sessions: dict[str, AgentSession] = {}
+
+
+def _get_settings() -> Settings:
+    global _settings
+    if _settings is None:
+        _settings = load_settings()
+    return _settings
+
+
+def _get_registry() -> ToolRegistry:
+    global _registry
+    if _registry is None:
+        _registry = get_tool_registry()
+        register_builtin_tools(_registry)
+        # Apply filters
+        settings = _get_settings()
+        _registry.set_filters(settings.enabled_tools, settings.disabled_tools)
+    return _registry
+
+
+def _get_or_create_session(session_id: str = "") -> tuple[str, AgentSession]:
+    """Get existing session or create new one."""
+    if not session_id:
+        session_id = str(uuid.uuid4())[:12]
+    if session_id not in _sessions:
+        _sessions[session_id] = AgentSession(
+            settings=_get_settings(),
+            tool_registry=_get_registry(),
+        )
+    return session_id, _sessions[session_id]
+
+
+# ── Gradio API Functions ──────────────────────────────────────────────
+
+def api_chat(
+    message: str,
+    session_id: str = "",
+    image_url: str | None = None,
+) -> Generator[str, None, None]:
+    """Main chat/agent endpoint — streams events as JSON lines."""
+    if not message or not message.strip():
+        yield json.dumps({"type": "error", "error": "Empty message"})
+        return
+
+    sid, session = _get_or_create_session(session_id)
+
+    try:
+        for event in session.run(message, session_id=sid, image_url=image_url):
+            yield json.dumps(event, ensure_ascii=False)
+    except Exception as e:
+        logger.error(f"Chat error: {e}", exc_info=True)
+        yield json.dumps({"type": "error", "error": str(e)})
+
+
+def api_chat_sync(message: str, session_id: str = "") -> str:
+    """Non-streaming chat for simple calls."""
+    events = list(api_chat(message, session_id))
+    if not events:
+        return json.dumps({"type": "error", "error": "No response"})
+    return events[-1]  # Return last event (complete)
+
+
+def api_model_status() -> str:
+    """Get model loading status."""
+    return json.dumps(get_model_status())
+
+
+def api_switch_model(model_key: str) -> str:
+    """Switch to a different model."""
+    switch_model(model_key)
+    return json.dumps({"status": "switching", "model": model_key})
+
+
+def api_list_models() -> str:
+    """List all available models."""
+    models = get_available_models()
+    current = get_model_status().get("model_key", "")
+    result = []
+    for key, config in models.items():
+        result.append({
+            "key": key,
+            "name": config.name,
+            "model_type": config.model_type,
+            "size_gb": config.size_gb,
+            "description": config.description,
+            "current": key == current,
+        })
+    return json.dumps(result, ensure_ascii=False)
+
+
+def api_workspace_tree(max_depth: int = 3) -> str:
+    """Get workspace directory tree."""
+    import os
+    tree = []
+
+    def walk(path: Path, prefix: str = "", depth: int = 0):
+        if depth >= max_depth:
+            return
+        try:
+            items = sorted(path.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower()))
+        except PermissionError:
+            return
+
+        skip = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build", ".sonicoder"}
+        for i, item in enumerate(items):
+            if item.name.startswith(".") or item.name in skip:
+                continue
+            is_last = i == len(items) - 1
+            connector = "└── " if is_last else "├── "
+            if item.is_dir():
+                tree.append(f"{prefix}{connector}{item.name}/")
+                extension = "    " if is_last else "│   "
+                walk(item, prefix + extension, depth + 1)
+            else:
+                tree.append(f"{prefix}{connector}{item.name}")
+
+    if WORKSPACE_ROOT.exists():
+        walk(WORKSPACE_ROOT)
+
+    return json.dumps({
+        "tree": "\n".join(tree) if tree else "(empty workspace)",
+        "root": str(WORKSPACE_ROOT),
+    })
+
+
+def api_workspace_read(path: str, offset: int = 1, limit: int = 500) -> str:
+    """Read a workspace file."""
+    from ..tools.fs import _resolve_safe
+    try:
+        resolved = _resolve_safe(path)
+        if not resolved.exists():
+            return json.dumps({"error": f"File not found: {path}"})
+        content = resolved.read_text(encoding="utf-8", errors="replace")
+        lines = content.splitlines()
+        total = len(lines)
+        start = max(0, offset - 1)
+        end = min(total, start + limit)
+        selected = lines[start:end]
+        numbered = [f"  {i + start + 1:>6}\t{line}" for i, line in enumerate(selected)]
+        return json.dumps({
+            "content": "\n".join(numbered),
+            "total_lines": total,
+            "truncated": end < total,
+        })
+    except Exception as e:
+        return json.dumps({"error": str(e)})
+
+
+def api_workspace_write(path: str, content: str) -> str:
+    """Write a workspace file."""
+    from ..tools.fs import _resolve_safe
+    try:
+        resolved = _resolve_safe(path)
+        resolved.parent.mkdir(parents=True, exist_ok=True)
+        resolved.write_text(content, encoding="utf-8")
+        return json.dumps({"success": True, "path": path})
+    except Exception as e:
+        return json.dumps({"error": str(e)})
+
+
+def api_workspace_bash(command: str) -> str:
+    """Execute a bash command in workspace."""
+    from ..tools.bash import BashBuilder
+    builder = BashBuilder()
+    errors = builder.validate({"command": command})
+    if errors:
+        return json.dumps({"error": "; ".join(errors)})
+    invocation = builder.create_invocation({"command": command})
+    result = invocation.execute()
+    return json.dumps(result.to_dict())
+
+
+def api_web_search(query: str) -> str:
+    """Search the web."""
+    from ..tools.web import WebSearchBuilder
+    builder = WebSearchBuilder()
+    invocation = builder.create_invocation({"query": query})
+    result = invocation.execute()
+    return json.dumps(result.to_dict(), ensure_ascii=False)
+
+
+def api_mcp_status() -> str:
+    """Get MCP server status."""
+    manager = get_mcp_manager()
+    return json.dumps(manager.get_status())
+
+
+def api_policy_status() -> str:
+    """Get policy engine status."""
+    engine = get_policy_engine()
+    return json.dumps({
+        "mode": engine.approval_mode,
+        "rules": engine.get_rules(),
+    })
+
+
+def api_reset_session(session_id: str = "") -> str:
+    """Reset a chat session."""
+    if session_id in _sessions:
+        _sessions[session_id].clear_history()
+    return json.dumps({"success": True})
+
+
+def api_get_config() -> str:
+    """Get runtime configuration for the frontend."""
+    models = get_available_models()
+    model_list = []
+    for key, config in models.items():
+        model_list.append({
+            "key": key,
+            "name": config.name,
+            "model_type": config.model_type,
+            "size_gb": config.size_gb,
+            "description": config.description,
+        })
+
+    return json.dumps({
+        "models": model_list,
+        "default_model": _get_settings().default_model,
+        "workspace": str(WORKSPACE_ROOT),
+        "version": "2.0.0",
+    }, ensure_ascii=False)
+
+
+# ── Create Gradio App ──────────────────────────────────────────────────
+
+def get_app():
+    """Create and return the Gradio application."""
+    import gradio as gr
+    from fastapi.responses import FileResponse
+    from pathlib import Path
+
+    # Create the Gradio app
+    app = gr.Blocks(
+        title="SoniCoder v2 — AI Code Agent",
+        theme=gr.themes.Soft(),
+    )
+
+    with app:
+        # Hidden components for API
+        api_chat_output = gr.Textbox(visible=False)
+        api_status_output = gr.Textbox(visible=False)
+        api_json_output = gr.Textbox(visible=False)
+
+        # API endpoints
+        gr.Interface(
+            fn=api_chat,
+            inputs=[
+                gr.Textbox(label="message"),
+                gr.Textbox(label="session_id", default=""),
+                gr.Textbox(label="image_url", default=""),
+            ],
+            outputs=[api_chat_output],
+            api_name="chat",
+        )
+
+        gr.Interface(
+            fn=api_model_status,
+            inputs=[],
+            outputs=[api_json_output],
+            api_name="model_status",
+        )
+
+        gr.Interface(
+            fn=api_switch_model,
+            inputs=[gr.Textbox(label="model_key")],
+            outputs=[api_json_output],
+            api_name="switch_model",
+        )
+
+        gr.Interface(
+            fn=api_list_models,
+            inputs=[],
+            outputs=[api_json_output],
+            api_name="list_models",
+        )
+
+        gr.Interface(
+            fn=api_workspace_tree,
+            inputs=[gr.Number(label="max_depth", value=3)],
+            outputs=[api_json_output],
+            api_name="workspace_tree",
+        )
+
+        gr.Interface(
+            fn=api_workspace_read,
+            inputs=[
+                gr.Textbox(label="path"),
+                gr.Number(label="offset", value=1),
+                gr.Number(label="limit", value=500),
+            ],
+            outputs=[api_json_output],
+            api_name="workspace_read",
+        )
+
+        gr.Interface(
+            fn=api_workspace_write,
+            inputs=[gr.Textbox(label="path"), gr.Textbox(label="content", lines=10)],
+            outputs=[api_json_output],
+            api_name="workspace_write",
+        )
+
+        gr.Interface(
+            fn=api_workspace_bash,
+            inputs=[gr.Textbox(label="command")],
+            outputs=[api_json_output],
+            api_name="workspace_bash",
+        )
+
+        gr.Interface(
+            fn=api_web_search,
+            inputs=[gr.Textbox(label="query")],
+            outputs=[api_json_output],
+            api_name="web_search",
+        )
+
+        gr.Interface(
+            fn=api_mcp_status,
+            inputs=[],
+            outputs=[api_json_output],
+            api_name="mcp_status",
+        )
+
+        gr.Interface(
+            fn=api_policy_status,
+            inputs=[],
+            outputs=[api_json_output],
+            api_name="policy_status",
+        )
+
+        gr.Interface(
+            fn=api_reset_session,
+            inputs=[gr.Textbox(label="session_id", default="")],
+            outputs=[api_json_output],
+            api_name="reset_session",
+        )
+
+        gr.Interface(
+            fn=api_get_config,
+            inputs=[],
+            outputs=[api_json_output],
+            api_name="get_config",
+        )
+
+    return app

sonicoder/tools/__init__.py

@@ -0,0 +1,215 @@
+"""Tool system — ToolBuilder/ToolInvocation pattern from Gemini CLI.
+
+Every tool follows:
+  1. Define a class extending ToolBuilder
+  2. Implement get_schema(), validate(), create_invocation()
+  3. The invocation's execute() does the actual work
+  4. Register in ToolRegistry
+"""
+
+from __future__ import annotations
+
+import json
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Callable
+
+
+# ── Tool Kinds ────────────────────────────────────────────────────────
+
+class ToolKind(Enum):
+    READ = "read"
+    WRITE = "write"
+    EDIT = "edit"
+    SHELL = "shell"
+    OTHER = "other"
+
+
+# ── Result Types ──────────────────────────────────────────────────────
+
+@dataclass
+class ToolResult:
+    """Result of a tool execution."""
+    success: bool
+    output: str = ""
+    error: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        d: dict[str, Any] = {"success": self.success}
+        if self.output:
+            d["output"] = self.output
+        if self.error:
+            d["error"] = self.error
+        if self.metadata:
+            d["metadata"] = self.metadata
+        return d
+
+    def to_json(self) -> str:
+        return json.dumps(self.to_dict(), ensure_ascii=False)
+
+
+# ── Tool Builder (Schema + Validation + Invocation Factory) ───────────
+
+class ToolBuilder(ABC):
+    """Base class for all tools — Gemini CLI pattern.
+
+    Tools are defined by:
+    - name: LLM-visible identifier
+    - description: What the tool does
+    - kind: READ/WRITE/EDIT/SHELL/OTHER
+    - get_schema(): JSON Schema for parameters
+    - validate(): Parameter validation
+    - create_invocation(): Factory for ToolInvocation
+    """
+
+    name: str = ""
+    description: str = ""
+    kind: ToolKind = ToolKind.OTHER
+    is_read_only: bool = False
+
+    @abstractmethod
+    def get_schema(self) -> dict[str, Any]:
+        """Return JSON Schema for this tool's parameters."""
+        ...
+
+    def get_display_name(self) -> str:
+        return self.name.replace("_", " ").title()
+
+    def get_tool_definition(self) -> dict[str, Any]:
+        """Full tool definition for the LLM."""
+        return {
+            "name": self.name,
+            "description": self.description,
+            "parameters": self.get_schema(),
+            "kind": self.kind.value,
+        }
+
+    def validate(self, params: dict[str, Any]) -> list[str]:
+        """Validate params. Returns list of error messages (empty = valid)."""
+        return []
+
+    @abstractmethod
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        """Create an invocation with validated params."""
+        ...
+
+
+# ── Tool Invocation (Ready-to-Execute) ────────────────────────────────
+
+class ToolInvocation(ABC):
+    """A validated, ready-to-execute tool call — Gemini CLI pattern."""
+
+    def __init__(self, params: dict[str, Any]):
+        self.params = params
+
+    def get_description(self) -> str:
+        """Human-readable description of this invocation."""
+        return f"Execute {self.__class__.__name__}"
+
+    def get_affected_paths(self) -> list[str]:
+        """Return file paths affected by this invocation."""
+        return []
+
+    @abstractmethod
+    def execute(self) -> ToolResult:
+        """Execute the tool and return a result."""
+        ...
+
+
+# ── Tool Registry ─────────────────────────────────────────────────────
+
+class ToolRegistry:
+    """Central registry for all tools — manages discovery, filtering, and lookup.
+
+    Inspired by Gemini CLI's ToolRegistry with enable/disable support.
+    """
+
+    def __init__(self):
+        self._tools: dict[str, ToolBuilder] = {}
+        self._mcp_tools: dict[str, ToolBuilder] = {}
+        self._enabled_patterns: list[str] = ["*"]
+        self._disabled_patterns: list[str] = []
+
+    def register(self, tool: ToolBuilder) -> None:
+        """Register a built-in tool."""
+        self._tools[tool.name] = tool
+
+    def register_mcp(self, tool: ToolBuilder) -> None:
+        """Register an MCP-discovered tool (prefixed with mcp_)."""
+        self._mcp_tools[tool.name] = tool
+
+    def unregister_mcp(self, server_name: str) -> None:
+        """Unregister all MCP tools from a specific server."""
+        prefix = f"mcp_{server_name}_"
+        to_remove = [k for k in self._mcp_tools if k.startswith(prefix)]
+        for k in to_remove:
+            del self._mcp_tools[k]
+
+    def get_tool(self, name: str) -> ToolBuilder | None:
+        """Look up a tool by name."""
+        if name in self._mcp_tools:
+            return self._mcp_tools[name]
+        return self._tools.get(name)
+
+    def get_all_tools(self) -> dict[str, ToolBuilder]:
+        """Get all registered tools (built-in + MCP)."""
+        return {**self._tools, **self._mcp_tools}
+
+    def get_active_tools(self) -> dict[str, ToolBuilder]:
+        """Get tools that pass the enable/disable filters."""
+        all_tools = self.get_all_tools()
+        active = {}
+        for name, tool in all_tools.items():
+            if self._is_tool_enabled(name):
+                active[name] = tool
+        return active
+
+    def get_tool_schemas(self) -> list[dict[str, Any]]:
+        """Get schemas for all active tools (for LLM context)."""
+        active = self.get_active_tools()
+        return [tool.get_tool_definition() for tool in active.values()]
+
+    def set_filters(self, enabled: list[str], disabled: list[str]) -> None:
+        """Set enable/disable patterns."""
+        self._enabled_patterns = enabled
+        self._disabled_patterns = disabled
+
+    def _is_tool_enabled(self, name: str) -> bool:
+        """Check if a tool passes the filters."""
+        # Check explicit disable first
+        for pattern in self._disabled_patterns:
+            if self._match_pattern(name, pattern):
+                return False
+        # Check enable patterns
+        if "*" in self._enabled_patterns:
+            return True
+        for pattern in self._enabled_patterns:
+            if self._match_pattern(name, pattern):
+                return True
+        return False
+
+    @staticmethod
+    def _match_pattern(name: str, pattern: str) -> bool:
+        """Simple wildcard matching: * at start/end."""
+        if pattern == "*":
+            return True
+        if pattern.endswith("*"):
+            return name.startswith(pattern[:-1])
+        if pattern.startswith("*"):
+            return name.endswith(pattern[1:])
+        return name == pattern
+
+
+# ── Global Registry ───────────────────────────────────────────────────
+
+_global_registry: ToolRegistry | None = None
+
+
+def get_tool_registry() -> ToolRegistry:
+    """Get or create the global tool registry."""
+    global _global_registry
+    if _global_registry is None:
+        _global_registry = ToolRegistry()
+    return _global_registry

sonicoder/tools/bash.py

@@ -0,0 +1,143 @@
+"""Bash/shell execution tool — sandboxed subprocess execution.
+
+Inspired by Claude Code's Bash tool and Gemini CLI's run_shell_command.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import threading
+from typing import Any
+
+from . import ToolBuilder, ToolInvocation, ToolKind, ToolResult
+
+# ── Security ───────────────────────────────────────────────────────────
+
+_BLOCKED_PATTERNS = [
+    "rm -rf /", "rm -rf ~", "rm -rf /*", "mkfs", "dd if=/dev/zero",
+    "shutdown", "reboot", "halt", "poweroff", ":(){ :|:& };:",
+    "chmod -R 777 /", "chown -R", "> /dev/sd", "mkswap",
+    "fdisk", "parted", "wipefs",
+]
+
+_SENSITIVE_ENV_VARS = [
+    "HF_TOKEN", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GOOGLE_API_KEY",
+    "GEMINI_API_KEY", "AWS_SECRET_ACCESS_KEY", "AWS_ACCESS_KEY_ID",
+    "AZURE_API_KEY", "DEEPSEEK_API_KEY",
+]
+
+
+def _is_blocked(command: str) -> str | None:
+    """Check if command matches a blocked pattern. Returns reason or None."""
+    cmd_lower = command.strip().lower()
+    for pattern in _BLOCKED_PATTERNS:
+        if pattern in cmd_lower:
+            return f"Blocked: dangerous command pattern '{pattern}'"
+    return None
+
+
+def _scrub_env() -> dict[str, str]:
+    """Return a clean environment with sensitive vars removed."""
+    env = os.environ.copy()
+    for var in _SENSITIVE_ENV_VARS:
+        env.pop(var, None)
+    return env
+
+
+# ── Bash Tool ──────────────────────────────────────────────────────────
+
+class BashBuilder(ToolBuilder):
+    name = "bash"
+    description = (
+        "Execute a shell command in the workspace. Supports any bash command. "
+        "Commands are sandboxed and sensitive environment variables are scrubbed. "
+        "Timeout: 120 seconds by default."
+    )
+    kind = ToolKind.SHELL
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "command": {"type": "string", "description": "Shell command to execute"},
+                "timeout": {"type": "integer", "description": "Timeout in seconds (max 300)", "default": 120},
+            },
+            "required": ["command"],
+        }
+
+    def validate(self, params: dict[str, Any]) -> list[str]:
+        errors = []
+        if not params.get("command"):
+            errors.append("command is required")
+        blocked = _is_blocked(params.get("command", ""))
+        if blocked:
+            errors.append(blocked)
+        return errors
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return BashInvocation(params)
+
+
+class BashInvocation(ToolInvocation):
+    MAX_OUTPUT = 50000
+
+    def get_description(self) -> str:
+        cmd = self.params["command"][:80]
+        return f"Bash: {cmd}"
+
+    def execute(self) -> ToolResult:
+        command = self.params["command"]
+        timeout = min(self.params.get("timeout", 120) or 120, 300)
+
+        # Security check
+        blocked = _is_blocked(command)
+        if blocked:
+            return ToolResult(success=False, error=blocked)
+
+        workspace = os.environ.get("SONICODER_WORKSPACE", "./workspace")
+        env = _scrub_env()
+        env["SONICODER_WORKSPACE"] = workspace
+
+        try:
+            proc = subprocess.run(
+                ["bash", "-c", command],
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                cwd=workspace,
+                env=env,
+            )
+
+            stdout = proc.stdout or ""
+            stderr = proc.stderr or ""
+            combined = stdout
+            if stderr:
+                combined += ("\n--- stderr ---\n" + stderr) if stdout else stderr
+
+            # Truncate large output
+            if len(combined) > self.MAX_OUTPUT:
+                combined = combined[:self.MAX_OUTPUT] + f"\n... [truncated, total {len(proc.stdout + proc.stderr)} chars]"
+
+            if proc.returncode == 0:
+                return ToolResult(
+                    success=True,
+                    output=combined or "(no output)",
+                    metadata={"exit_code": 0},
+                )
+            else:
+                return ToolResult(
+                    success=False,
+                    output=combined,
+                    error=f"Exit code: {proc.returncode}",
+                    metadata={"exit_code": proc.returncode},
+                )
+        except subprocess.TimeoutExpired:
+            return ToolResult(success=False, error=f"Command timed out after {timeout}s")
+        except Exception as e:
+            return ToolResult(success=False, error=f"Execution error: {e}")
+
+
+# ── Register ───────────────────────────────────────────────────────────
+
+BUILTIN_TOOLS = [BashBuilder()]

sonicoder/tools/fs.py

@@ -0,0 +1,508 @@
+"""File system tools — read, write, edit, glob, grep, list_dir.
+
+Inspired by Claude Code's FileRead/FileWrite/FileEdit and Gemini CLI's file tools.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+from . import ToolBuilder, ToolInvocation, ToolKind, ToolResult
+
+WORKSPACE = Path(os.environ.get("SONICODER_WORKSPACE", "./workspace")).resolve()
+
+
+def _resolve_safe(path: str) -> Path:
+    """Resolve a path and ensure it stays within the workspace sandbox."""
+    if not path:
+        path = "."
+    p = Path(path)
+    if not p.is_absolute():
+        p = WORKSPACE / p
+    try:
+        p = p.resolve()
+    except (OSError, ValueError):
+        p = WORKSPACE / path
+    # Security: must be under workspace
+    try:
+        p.relative_to(WORKSPACE)
+    except ValueError:
+        raise ValueError(f"Path '{path}' escapes workspace")
+    return p
+
+
+# ── Read File ──────────────────────────────────────────────────────────
+
+class ReadFileBuilder(ToolBuilder):
+    name = "read_file"
+    description = (
+        "Read file contents with line numbers. Supports offset/limit for pagination. "
+        "Returns content, line count, and truncation info."
+    )
+    kind = ToolKind.READ
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "description": "File path relative to workspace root"},
+                "offset": {"type": "integer", "description": "Start line (1-indexed)", "default": 1},
+                "limit": {"type": "integer", "description": "Max lines to return", "default": 500},
+            },
+            "required": ["path"],
+        }
+
+    def validate(self, params: dict[str, Any]) -> list[str]:
+        errors = []
+        if not params.get("path"):
+            errors.append("path is required")
+        return errors
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return ReadFileInvocation(params)
+
+
+class ReadFileInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        return f"Read file: {self.params['path']}"
+
+    def get_affected_paths(self) -> list[str]:
+        return [self.params.get("path", "")]
+
+    def execute(self) -> ToolResult:
+        try:
+            path = _resolve_safe(self.params["path"])
+            if not path.exists():
+                return ToolResult(success=False, error=f"File not found: {self.params['path']}")
+            if not path.is_file():
+                return ToolResult(success=False, error=f"Not a file: {self.params['path']}")
+
+            offset = self.params.get("offset", 1) or 1
+            limit = self.params.get("limit", 500) or 500
+
+            lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
+            total = len(lines)
+            start = max(0, offset - 1)
+            end = min(total, start + limit)
+            selected = lines[start:end]
+
+            # Add line numbers
+            numbered = [f"  {i + start + 1:>6}\t{line}" for i, line in enumerate(selected)]
+            content = "\n".join(numbered)
+            truncated = end < total
+
+            meta = {"total_lines": total, "returned_lines": len(selected), "truncated": truncated}
+            if truncated:
+                meta["next_offset"] = end + 1
+
+            return ToolResult(success=True, output=content, metadata=meta)
+        except ValueError as e:
+            return ToolResult(success=False, error=str(e))
+        except Exception as e:
+            return ToolResult(success=False, error=f"Read error: {e}")
+
+
+# ── Write File ─────────────────────────────────────────────────────────
+
+class WriteFileBuilder(ToolBuilder):
+    name = "write_file"
+    description = (
+        "Write content to a file. Creates parent directories automatically. "
+        "Overwrites existing file content entirely."
+    )
+    kind = ToolKind.WRITE
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "description": "File path relative to workspace root"},
+                "content": {"type": "string", "description": "Full file content to write"},
+            },
+            "required": ["path", "content"],
+        }
+
+    def validate(self, params: dict[str, Any]) -> list[str]:
+        errors = []
+        if not params.get("path"):
+            errors.append("path is required")
+        if "content" not in params:
+            errors.append("content is required")
+        return errors
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return WriteFileInvocation(params)
+
+
+class WriteFileInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        return f"Write file: {self.params['path']}"
+
+    def get_affected_paths(self) -> list[str]:
+        return [self.params.get("path", "")]
+
+    def execute(self) -> ToolResult:
+        try:
+            path = _resolve_safe(self.params["path"])
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(self.params["content"], encoding="utf-8")
+            line_count = self.params["content"].count("\n") + 1
+            return ToolResult(
+                success=True,
+                output=f"Written {line_count} lines to {self.params['path']}",
+                metadata={"path": str(path), "lines": line_count},
+            )
+        except ValueError as e:
+            return ToolResult(success=False, error=str(e))
+        except Exception as e:
+            return ToolResult(success=False, error=f"Write error: {e}")
+
+
+# ── Edit File (Search & Replace) ──────────────────────────────────────
+
+class EditFileBuilder(ToolBuilder):
+    name = "edit_file"
+    description = (
+        "Edit a file by finding exact old_string and replacing with new_string. "
+        "Claude Code pattern: exact match, not line-based diffs. "
+        "Set replace_all=true to replace every occurrence."
+    )
+    kind = ToolKind.EDIT
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "description": "File path relative to workspace root"},
+                "old_string": {"type": "string", "description": "Exact text to find"},
+                "new_string": {"type": "string", "description": "Replacement text"},
+                "replace_all": {"type": "boolean", "description": "Replace all occurrences", "default": False},
+            },
+            "required": ["path", "old_string", "new_string"],
+        }
+
+    def validate(self, params: dict[str, Any]) -> list[str]:
+        errors = []
+        if not params.get("path"):
+            errors.append("path is required")
+        if not params.get("old_string"):
+            errors.append("old_string is required")
+        if "new_string" not in params:
+            errors.append("new_string is required")
+        if params.get("old_string") == params.get("new_string"):
+            errors.append("old_string and new_string must differ")
+        return errors
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return EditFileInvocation(params)
+
+
+class EditFileInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        p = self.params["path"]
+        old = self.params["old_string"][:50].replace("\n", "\\n")
+        return f"Edit {p}: '{old}...'"
+
+    def get_affected_paths(self) -> list[str]:
+        return [self.params.get("path", "")]
+
+    def execute(self) -> ToolResult:
+        try:
+            path = _resolve_safe(self.params["path"])
+            if not path.exists():
+                return ToolResult(success=False, error=f"File not found: {self.params['path']}")
+
+            content = path.read_text(encoding="utf-8")
+            old = self.params["old_string"]
+            new = self.params["new_string"]
+            replace_all = self.params.get("replace_all", False)
+
+            if old not in content:
+                # Fuzzy hint
+                return ToolResult(
+                    success=False,
+                    error=f"old_string not found in {self.params['path']}. "
+                          f"Make sure the text matches the file exactly (including whitespace).",
+                )
+
+            count = content.count(old)
+            if replace_all:
+                new_content = content.replace(old, new)
+            else:
+                if count > 1:
+                    return ToolResult(
+                        success=False,
+                        error=f"Found {count} occurrences. Use replace_all=true to replace all, "
+                              f"or make old_string more specific.",
+                    )
+                new_content = content.replace(old, new, 1)
+
+            path.write_text(new_content, encoding="utf-8")
+            return ToolResult(
+                success=True,
+                output=f"Replaced {count if replace_all else 1} occurrence(s) in {self.params['path']}",
+                metadata={"path": str(path), "replacements": count if replace_all else 1},
+            )
+        except ValueError as e:
+            return ToolResult(success=False, error=str(e))
+        except Exception as e:
+            return ToolResult(success=False, error=f"Edit error: {e}")
+
+
+# ── Glob ───────────────────────────────────────────────────────────────
+
+class GlobBuilder(ToolBuilder):
+    name = "glob"
+    description = (
+        "Find files matching a glob pattern (e.g., '**/*.py', 'src/**/*.ts'). "
+        "Searches recursively within the workspace."
+    )
+    kind = ToolKind.READ
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "pattern": {"type": "string", "description": "Glob pattern (e.g., '**/*.py')"},
+                "path": {"type": "string", "description": "Subdirectory to search in", "default": "."},
+            },
+            "required": ["pattern"],
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return GlobInvocation(params)
+
+
+class GlobInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        return f"Glob: {self.params['pattern']}"
+
+    def execute(self) -> ToolResult:
+        try:
+            base = _resolve_safe(self.params.get("path", "."))
+            pattern = self.params["pattern"]
+            if not base.exists():
+                return ToolResult(success=True, output="Directory not found", metadata={"matches": []})
+
+            matches = []
+            for root, dirs, files in os.walk(base):
+                # Skip hidden and common heavy dirs
+                dirs[:] = [d for d in dirs if not d.startswith(".") and d not in
+                           ("node_modules", "__pycache__", ".venv", "venv", "dist", "build", ".git")]
+                for f in files:
+                    full = os.path.join(root, f)
+                    rel = os.path.relpath(full, WORKSPACE)
+                    if fnmatch.fnmatch(rel, pattern) or fnmatch.fnmatch(f, pattern):
+                        matches.append(rel)
+
+            matches.sort()
+            output = "\n".join(matches) if matches else "No matches found."
+            return ToolResult(success=True, output=output, metadata={"matches": matches, "count": len(matches)})
+        except ValueError as e:
+            return ToolResult(success=False, error=str(e))
+        except Exception as e:
+            return ToolResult(success=False, error=f"Glob error: {e}")
+
+
+# ── Grep ───────────────────────────────────────────────────────────────
+
+class GrepBuilder(ToolBuilder):
+    name = "grep"
+    description = (
+        "Search file contents using regex. Returns matching lines with file paths and line numbers. "
+        "Like ripgrep but simpler."
+    )
+    kind = ToolKind.READ
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "pattern": {"type": "string", "description": "Regex pattern to search for"},
+                "path": {"type": "string", "description": "Subdirectory to search in", "default": "."},
+                "include": {"type": "string", "description": "File glob to include (e.g., '*.py')"},
+                "ignore_case": {"type": "boolean", "description": "Case insensitive", "default": False},
+                "max_results": {"type": "integer", "description": "Max matches to return", "default": 50},
+            },
+            "required": ["pattern"],
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return GrepInvocation(params)
+
+
+class GrepInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        return f"Grep: {self.params['pattern']}"
+
+    def execute(self) -> ToolResult:
+        try:
+            base = _resolve_safe(self.params.get("path", "."))
+            pattern = self.params["pattern"]
+            include = self.params.get("include", "*")
+            ignore_case = self.params.get("ignore_case", False)
+            max_results = self.params.get("max_results", 50) or 50
+
+            flags = re.IGNORECASE if ignore_case else 0
+            regex = re.compile(pattern, flags)
+            results = []
+            count = 0
+
+            for root, dirs, files in os.walk(base):
+                dirs[:] = [d for d in dirs if not d.startswith(".") and d not in
+                           ("node_modules", "__pycache__", ".venv", "venv", "dist", "build", ".git")]
+                for f in files:
+                    if not fnmatch.fnmatch(f, include):
+                        continue
+                    try:
+                        filepath = os.path.join(root, f)
+                        rel = os.path.relpath(filepath, WORKSPACE)
+                        lines = filepath.read_text(encoding="utf-8", errors="replace").splitlines()
+                        for i, line in enumerate(lines, 1):
+                            if regex.search(line):
+                                results.append(f"{rel}:{i}: {line.strip()}")
+                                count += 1
+                                if count >= max_results:
+                                    output = "\n".join(results)
+                                    output += f"\n\n... truncated at {max_results} matches"
+                                    return ToolResult(
+                                        success=True, output=output,
+                                        metadata={"matches": results, "count": count, "truncated": True},
+                                    )
+                    except (OSError, UnicodeDecodeError):
+                        continue
+
+            output = "\n".join(results) if results else "No matches found."
+            return ToolResult(
+                success=True, output=output,
+                metadata={"matches": results, "count": count, "truncated": False},
+            )
+        except ValueError as e:
+            return ToolResult(success=False, error=str(e))
+        except re.error as e:
+            return ToolResult(success=False, error=f"Invalid regex: {e}")
+        except Exception as e:
+            return ToolResult(success=False, error=f"Grep error: {e}")
+
+
+# ── List Directory ─────────────────────────────────────────────────────
+
+class ListDirBuilder(ToolBuilder):
+    name = "list_dir"
+    description = "List files and directories. Returns sorted listing with type indicators."
+    kind = ToolKind.READ
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "path": {"type": "string", "description": "Directory path (default: workspace root)", "default": "."},
+            },
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return ListDirInvocation(params)
+
+
+class ListDirInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        return f"List dir: {self.params.get('path', '.')}"
+
+    def execute(self) -> ToolResult:
+        try:
+            path = _resolve_safe(self.params.get("path", "."))
+            if not path.exists():
+                return ToolResult(success=False, error=f"Directory not found: {self.params.get('path', '.')}")
+            if not path.is_dir():
+                return ToolResult(success=False, error=f"Not a directory: {self.params.get('path', '.')}")
+
+            entries = []
+            for item in sorted(path.iterdir()):
+                if item.name.startswith("."):
+                    continue
+                if item.is_dir():
+                    entries.append(f"  [DIR]  {item.name}/")
+                else:
+                    size = item.stat().st_size
+                    if size > 1024 * 1024:
+                        size_str = f"{size / 1024 / 1024:.1f}MB"
+                    elif size > 1024:
+                        size_str = f"{size / 1024:.1f}KB"
+                    else:
+                        size_str = f"{size}B"
+                    entries.append(f"  [FILE] {item.name}  ({size_str})")
+
+            output = "\n".join(entries) if entries else "(empty directory)"
+            return ToolResult(success=True, output=output, metadata={"entries": len(entries)})
+        except ValueError as e:
+            return ToolResult(success=False, error=str(e))
+        except Exception as e:
+            return ToolResult(success=False, error=f"List error: {e}")
+
+
+# ── Workspace Snapshot ─────────────────────────────────────────────────
+
+class SnapshotBuilder(ToolBuilder):
+    name = "snapshot_workspace"
+    description = "Get a snapshot of all workspace files and their contents. Useful for getting full project state."
+    kind = ToolKind.READ
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {},
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return SnapshotInvocation(params)
+
+
+class SnapshotInvocation(ToolInvocation):
+    SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build",
+                 ".sonicoder"}
+
+    def execute(self) -> ToolResult:
+        try:
+            files = {}
+            for root, dirs, filenames in os.walk(WORKSPACE):
+                dirs[:] = [d for d in dirs if d not in self.SKIP_DIRS and not d.startswith(".")]
+                for f in filenames:
+                    if f.startswith("."):
+                        continue
+                    fp = os.path.join(root, f)
+                    rel = os.path.relpath(fp, WORKSPACE)
+                    try:
+                        files[rel] = fp if os.path.getsize(fp) > 500000 else \
+                            Path(fp).read_text(encoding="utf-8", errors="replace")
+                    except (OSError, UnicodeDecodeError):
+                        files[rel] = f"[binary file, {os.path.getsize(fp)} bytes]"
+
+            output = "\n".join(f"  {k}" for k in sorted(files.keys()))
+            return ToolResult(
+                success=True, output=output,
+                metadata={"files": files, "count": len(files)},
+            )
+        except Exception as e:
+            return ToolResult(success=False, error=f"Snapshot error: {e}")
+
+
+# ── Register All ───────────────────────────────────────────────────────
+
+BUILTIN_TOOLS = [
+    ReadFileBuilder(),
+    WriteFileBuilder(),
+    EditFileBuilder(),
+    GlobBuilder(),
+    GrepBuilder(),
+    ListDirBuilder(),
+    SnapshotBuilder(),
+]

sonicoder/tools/todos.py

@@ -0,0 +1,109 @@
+"""Todo management tool — per-session task tracking.
+
+Inspired by Claude Code's TodoWrite/TaskCreate.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+from . import ToolBuilder, ToolInvocation, ToolKind, ToolResult
+
+# ── In-Memory Store ────────────────────────────────────────────────────
+
+_store: dict[str, list[dict]] = {}
+_lock = threading.Lock()
+
+
+def _get_todos(session_id: str) -> list[dict]:
+    with _lock:
+        return _store.setdefault(session_id, [])
+
+
+# ── Todo Tool ──────────────────────────────────────────────────────────
+
+class TodoBuilder(ToolBuilder):
+    name = "todo"
+    description = (
+        "Manage a task todo list. Actions: 'read' (list all), 'write' (create), "
+        "'update' (change status/content), 'clear' (remove all). "
+        "Each todo has: id, content, status (pending/in_progress/completed)."
+    )
+    kind = ToolKind.OTHER
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "action": {
+                    "type": "string",
+                    "enum": ["read", "write", "update", "clear"],
+                    "description": "Action to perform",
+                },
+                "id": {"type": "string", "description": "Todo ID (for update)"},
+                "content": {"type": "string", "description": "Todo content (for write/update)"},
+                "status": {"type": "string", "description": "Status: pending/in_progress/completed"},
+            },
+            "required": ["action"],
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return TodoInvocation(params, session_id="")
+
+
+class TodoInvocation(ToolInvocation):
+    def __init__(self, params: dict[str, Any], session_id: str = ""):
+        super().__init__(params)
+        self.session_id = session_id
+
+    def get_description(self) -> str:
+        return f"Todo: {self.params['action']}"
+
+    def execute(self) -> ToolResult:
+        action = self.params["action"]
+        todos = _get_todos(self.session_id)
+
+        if action == "read":
+            if not todos:
+                return ToolResult(success=True, output="No todos.")
+            lines = []
+            for t in todos:
+                status_icon = {"pending": "[ ]", "in_progress": "[>]", "completed": "[x]"}.get(t["status"], "[ ]")
+                lines.append(f"  {status_icon} {t['id']}: {t['content']}")
+            return ToolResult(success=True, output="\n".join(lines), metadata={"todos": todos})
+
+        elif action == "write":
+            content = self.params.get("content", "")
+            if not content:
+                return ToolResult(success=False, error="content is required for write action")
+            new_id = str(len(todos) + 1)
+            todo = {"id": new_id, "content": content, "status": "pending"}
+            todos.append(todo)
+            return ToolResult(success=True, output=f"Added todo {new_id}: {content}", metadata={"todo": todo})
+
+        elif action == "update":
+            tid = self.params.get("id", "")
+            if not tid:
+                return ToolResult(success=False, error="id is required for update action")
+            for t in todos:
+                if t["id"] == tid:
+                    if "content" in self.params and self.params["content"]:
+                        t["content"] = self.params["content"]
+                    if "status" in self.params and self.params["status"]:
+                        if self.params["status"] not in ("pending", "in_progress", "completed"):
+                            return ToolResult(success=False, error="Invalid status")
+                        t["status"] = self.params["status"]
+                    return ToolResult(success=True, output=f"Updated todo {tid}", metadata={"todo": t})
+            return ToolResult(success=False, error=f"Todo {tid} not found")
+
+        elif action == "clear":
+            todos.clear()
+            return ToolResult(success=True, output="All todos cleared.")
+
+        return ToolResult(success=False, error=f"Unknown action: {action}")
+
+
+# ── Register ───────────────────────────────────────────────────────────
+
+BUILTIN_TOOLS = [TodoBuilder()]

sonicoder/tools/web.py

@@ -0,0 +1,163 @@
+"""Web tools — search and fetch.
+
+Inspired by Claude Code's WebSearch/WebFetch and Gemini CLI's google_web_search/web_fetch.
+No API keys needed — uses HTML scraping like original SoniCoder.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+from urllib.parse import quote_plus, urljoin
+
+import requests
+from bs4 import BeautifulSoup
+
+from . import ToolBuilder, ToolInvocation, ToolKind, ToolResult
+
+HEADERS = {
+    "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+                  "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+}
+
+
+# ── Web Search ─────────────────────────────────────────────────────────
+
+class WebSearchBuilder(ToolBuilder):
+    name = "web_search"
+    description = (
+        "Search the web using DuckDuckGo. No API key needed. "
+        "Returns numbered results with titles, URLs, and snippets."
+    )
+    kind = ToolKind.OTHER
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "Search query"},
+                "max_results": {"type": "integer", "description": "Max results (default 8)", "default": 8},
+            },
+            "required": ["query"],
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return WebSearchInvocation(params)
+
+
+class WebSearchInvocation(ToolInvocation):
+    def get_description(self) -> str:
+        return f"Web search: {self.params['query'][:60]}"
+
+    def execute(self) -> ToolResult:
+        query = self.params["query"]
+        max_results = self.params.get("max_results", 8) or 8
+
+        try:
+            # DuckDuckGo HTML search
+            url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
+            resp = requests.get(url, headers=HEADERS, timeout=15)
+            resp.raise_for_status()
+
+            soup = BeautifulSoup(resp.text, "html.parser")
+            results = []
+
+            for result in soup.select(".result"):
+                title_el = result.select_one(".result__title a, .result__a")
+                snippet_el = result.select_one(".result__snippet")
+                if title_el:
+                    title = title_el.get_text(strip=True)
+                    href = title_el.get("href", "")
+                    # Clean DuckDuckGo redirect URL
+                    if "uddg=" in href:
+                        import urllib.parse
+                        parsed = urllib.parse.parse_qs(urllib.parse.urlparse(href).query)
+                        href = parsed.get("uddg", [href])[0]
+                    snippet = snippet_el.get_text(strip=True) if snippet_el else ""
+                    results.append(f"{len(results) + 1}. {title}\n   {href}\n   {snippet}")
+                    if len(results) >= max_results:
+                        break
+
+            if not results:
+                return ToolResult(success=True, output="No results found.", metadata={"results": []})
+
+            output = "\n\n".join(results)
+            return ToolResult(
+                success=True, output=output,
+                metadata={"results": results, "count": len(results)},
+            )
+        except requests.RequestException as e:
+            return ToolResult(success=False, error=f"Search failed: {e}")
+        except Exception as e:
+            return ToolResult(success=False, error=f"Search error: {e}")
+
+
+# ── Web Fetch ──────────────────────────────────────────────────────────
+
+class WebFetchBuilder(ToolBuilder):
+    name = "web_fetch"
+    description = (
+        "Fetch and extract text content from a URL. "
+        "Returns the main text content (strips HTML tags, scripts, styles)."
+    )
+    kind = ToolKind.OTHER
+    is_read_only = True
+
+    def get_schema(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "url": {"type": "string", "description": "URL to fetch"},
+            },
+            "required": ["url"],
+        }
+
+    def create_invocation(self, params: dict[str, Any]) -> ToolInvocation:
+        return WebFetchInvocation(params)
+
+
+class WebFetchInvocation(ToolInvocation):
+    MAX_CONTENT = 30000
+
+    def get_description(self) -> str:
+        return f"Fetch: {self.params['url'][:80]}"
+
+    def execute(self) -> ToolResult:
+        url = self.params["url"]
+        if not url.startswith(("http://", "https://")):
+            url = "https://" + url
+
+        try:
+            resp = requests.get(url, headers=HEADERS, timeout=20)
+            resp.raise_for_status()
+
+            soup = BeautifulSoup(resp.text, "html.parser")
+
+            # Remove scripts, styles, nav, footer
+            for tag in soup.select("script, style, nav, footer, header, aside, iframe"):
+                tag.decompose()
+
+            # Extract text
+            text = soup.get_text(separator="\n", strip=True)
+
+            # Clean up whitespace
+            lines = [line.strip() for line in text.splitlines() if line.strip()]
+            text = "\n".join(lines)
+
+            if len(text) > self.MAX_CONTENT:
+                text = text[:self.MAX_CONTENT] + f"\n... [truncated at {self.MAX_CONTENT} chars]"
+
+            return ToolResult(
+                success=True, output=text,
+                metadata={"url": url, "chars": len(text), "title": soup.title.string if soup.title else ""},
+            )
+        except requests.RequestException as e:
+            return ToolResult(success=False, error=f"Fetch failed: {e}")
+        except Exception as e:
+            return ToolResult(success=False, error=f"Fetch error: {e}")
+
+
+# ── Register ───────────────────────────────────────────────────────────
+
+BUILTIN_TOOLS = [WebSearchBuilder(), WebFetchBuilder()]

static/index.html

@@ -0,0 +1,568 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="UTF-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0">
+<title>SoniCoder v2 — AI Code Agent</title>
+<style>
+  :root {
+    --bg-primary: #0f0f13;
+    --bg-secondary: #16161d;
+    --bg-tertiary: #1e1e2a;
+    --bg-hover: #252535;
+    --text-primary: #e4e4f0;
+    --text-secondary: #9393a8;
+    --text-muted: #6b6b80;
+    --accent: #7c6ef0;
+    --accent-hover: #6b5de0;
+    --accent-light: rgba(124,110,240,0.12);
+    --success: #4ade80;
+    --warning: #fbbf24;
+    --error: #f87171;
+    --info: #60a5fa;
+    --border: #2a2a3a;
+    --code-bg: #1a1a28;
+    --radius: 12px;
+    --radius-sm: 8px;
+    --shadow: 0 4px 24px rgba(0,0,0,0.4);
+  }
+  * { margin:0; padding:0; box-sizing:border-box; }
+  body { font-family: 'Inter', -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif; background:var(--bg-primary); color:var(--text-primary); height:100vh; overflow:hidden; }
+  
+  /* Layout */
+  .app { display:flex; height:100vh; }
+  .sidebar { width:280px; background:var(--bg-secondary); border-right:1px solid var(--border); display:flex; flex-direction:column; flex-shrink:0; }
+  .main { flex:1; display:flex; flex-direction:column; min-width:0; }
+  .workspace-panel { width:300px; background:var(--bg-secondary); border-left:1px solid var(--border); display:flex; flex-direction:column; flex-shrink:0; transition:width 0.3s; overflow:hidden; }
+  .workspace-panel.collapsed { width:0; border-left:none; }
+
+  /* Sidebar */
+  .sidebar-header { padding:20px; border-bottom:1px solid var(--border); }
+  .logo { font-size:20px; font-weight:700; background:linear-gradient(135deg, #7c6ef0, #a78bfa, #60a5fa); -webkit-background-clip:text; -webkit-text-fill-color:transparent; }
+  .logo-sub { font-size:11px; color:var(--text-muted); margin-top:2px; letter-spacing:1px; text-transform:uppercase; }
+  .model-selector { padding:16px 20px; border-bottom:1px solid var(--border); }
+  .model-selector label { font-size:11px; color:var(--text-muted); text-transform:uppercase; letter-spacing:0.5px; display:block; margin-bottom:8px; }
+  .model-selector select { width:100%; padding:8px 12px; background:var(--bg-tertiary); color:var(--text-primary); border:1px solid var(--border); border-radius:var(--radius-sm); font-size:13px; cursor:pointer; outline:none; }
+  .model-selector select:focus { border-color:var(--accent); }
+  .model-status { padding:8px 20px; border-bottom:1px solid var(--border); }
+  .status-dot { display:inline-block; width:8px; height:8px; border-radius:50%; margin-right:6px; }
+  .status-dot.ready { background:var(--success); box-shadow:0 0 6px var(--success); }
+  .status-dot.loading { background:var(--warning); animation:pulse 1.5s infinite; }
+  .status-dot.error { background:var(--error); }
+  .status-dot.unknown { background:var(--text-muted); }
+  @keyframes pulse { 0%,100%{opacity:1} 50%{opacity:0.4} }
+  .status-text { font-size:12px; color:var(--text-secondary); }
+  
+  .sidebar-section { padding:16px 20px; }
+  .sidebar-section h3 { font-size:11px; color:var(--text-muted); text-transform:uppercase; letter-spacing:0.5px; margin-bottom:12px; }
+  .sidebar-info { font-size:12px; color:var(--text-secondary); line-height:1.6; }
+  .sidebar-info code { background:var(--bg-tertiary); padding:2px 6px; border-radius:4px; font-size:11px; color:var(--accent); }
+
+  .new-chat-btn { margin:16px 20px; padding:10px; background:var(--accent); color:white; border:none; border-radius:var(--radius-sm); font-size:13px; font-weight:600; cursor:pointer; transition:all 0.2s; }
+  .new-chat-btn:hover { background:var(--accent-hover); transform:translateY(-1px); }
+
+  /* Chat Area */
+  .chat-header { padding:16px 24px; border-bottom:1px solid var(--border); display:flex; align-items:center; justify-content:space-between; }
+  .chat-header h2 { font-size:15px; font-weight:600; }
+  .header-actions { display:flex; gap:8px; }
+  .header-btn { padding:6px 12px; background:var(--bg-tertiary); color:var(--text-secondary); border:1px solid var(--border); border-radius:var(--radius-sm); font-size:12px; cursor:pointer; transition:all 0.2s; }
+  .header-btn:hover { background:var(--bg-hover); color:var(--text-primary); }
+
+  .chat-messages { flex:1; overflow-y:auto; padding:24px; scroll-behavior:smooth; }
+  .chat-messages::-webkit-scrollbar { width:6px; }
+  .chat-messages::-webkit-scrollbar-track { background:transparent; }
+  .chat-messages::-webkit-scrollbar-thumb { background:var(--border); border-radius:3px; }
+
+  .message { margin-bottom:24px; animation:fadeIn 0.3s ease; }
+  @keyframes fadeIn { from{opacity:0;transform:translateY(8px)} to{opacity:1;transform:translateY(0)} }
+  .message-header { display:flex; align-items:center; gap:8px; margin-bottom:8px; }
+  .message-avatar { width:28px; height:28px; border-radius:8px; display:flex; align-items:center; justify-content:center; font-size:13px; font-weight:700; }
+  .message-avatar.user { background:var(--accent); color:white; }
+  .message-avatar.agent { background:var(--bg-tertiary); color:var(--accent); border:1px solid var(--border); }
+  .message-role { font-size:13px; font-weight:600; }
+  .message-time { font-size:11px; color:var(--text-muted); }
+
+  .message-body { padding-left:36px; line-height:1.7; font-size:14px; color:var(--text-primary); }
+  .message-body p { margin-bottom:8px; }
+  .message-body code { background:var(--code-bg); padding:2px 6px; border-radius:4px; font-size:13px; font-family:'JetBrains Mono','Fira Code',monospace; }
+  .message-body pre { background:var(--code-bg); padding:16px; border-radius:var(--radius-sm); overflow-x:auto; margin:12px 0; border:1px solid var(--border); }
+  .message-body pre code { background:none; padding:0; font-size:13px; line-height:1.5; }
+  .message-body ul, .message-body ol { padding-left:20px; margin:8px 0; }
+  .message-body li { margin:4px 0; }
+
+  /* Tool Calls */
+  .tool-call { background:var(--bg-tertiary); border:1px solid var(--border); border-radius:var(--radius-sm); margin:12px 0; padding:12px 16px; }
+  .tool-call-header { display:flex; align-items:center; gap:8px; margin-bottom:8px; }
+  .tool-icon { font-size:14px; }
+  .tool-name { font-size:13px; font-weight:600; color:var(--accent); }
+  .tool-status { font-size:11px; padding:2px 8px; border-radius:10px; font-weight:600; }
+  .tool-status.success { background:rgba(74,222,128,0.12); color:var(--success); }
+  .tool-status.error { background:rgba(248,113,113,0.12); color:var(--error); }
+  .tool-status.running { background:rgba(251,191,36,0.12); color:var(--warning); }
+  .tool-args { font-size:12px; color:var(--text-muted); margin-bottom:8px; }
+  .tool-result { font-size:12px; background:var(--bg-primary); border-radius:6px; padding:8px 12px; max-height:300px; overflow-y:auto; white-space:pre-wrap; word-break:break-all; font-family:'JetBrains Mono',monospace; line-height:1.5; color:var(--text-secondary); }
+  .tool-result::-webkit-scrollbar { width:4px; }
+  .tool-result::-webkit-scrollbar-thumb { background:var(--border); border-radius:2px; }
+
+  /* Iteration Badge */
+  .iteration-badge { display:inline-flex; align-items:center; gap:4px; font-size:11px; color:var(--text-muted); background:var(--bg-tertiary); padding:2px 10px; border-radius:10px; margin-bottom:8px; }
+
+  /* Input Area */
+  .chat-input { padding:16px 24px 24px; border-top:1px solid var(--border); }
+  .input-wrapper { display:flex; align-items:flex-end; gap:12px; background:var(--bg-tertiary); border:1px solid var(--border); border-radius:var(--radius); padding:12px 16px; transition:border-color 0.2s; }
+  .input-wrapper:focus-within { border-color:var(--accent); }
+  .input-wrapper textarea { flex:1; background:none; border:none; outline:none; color:var(--text-primary); font-size:14px; font-family:inherit; resize:none; max-height:200px; line-height:1.5; }
+  .input-wrapper textarea::placeholder { color:var(--text-muted); }
+  .send-btn { padding:8px 20px; background:var(--accent); color:white; border:none; border-radius:var(--radius-sm); font-size:13px; font-weight:600; cursor:pointer; transition:all 0.2s; white-space:nowrap; }
+  .send-btn:hover { background:var(--accent-hover); }
+  .send-btn:disabled { opacity:0.5; cursor:not-allowed; }
+  .stop-btn { padding:8px 20px; background:var(--error); color:white; border:none; border-radius:var(--radius-sm); font-size:13px; font-weight:600; cursor:pointer; }
+  .input-hint { font-size:11px; color:var(--text-muted); margin-top:8px; text-align:center; }
+
+  /* Workspace Panel */
+  .ws-header { padding:16px; border-bottom:1px solid var(--border); display:flex; align-items:center; justify-content:space-between; }
+  .ws-header h3 { font-size:13px; font-weight:600; }
+  .ws-tree { flex:1; overflow-y:auto; padding:12px 16px; font-family:'JetBrains Mono',monospace; font-size:12px; line-height:1.8; color:var(--text-secondary); white-space:pre; }
+  .ws-tree::-webkit-scrollbar { width:4px; }
+  .ws-tree::-webkit-scrollbar-thumb { background:var(--border); border-radius:2px; }
+
+  /* Typing indicator */
+  .typing-indicator { display:flex; gap:4px; padding:8px 0; }
+  .typing-dot { width:6px; height:6px; border-radius:50%; background:var(--accent); animation:typingBounce 1.4s infinite; }
+  .typing-dot:nth-child(2) { animation-delay:0.2s; }
+  .typing-dot:nth-child(3) { animation-delay:0.4s; }
+  @keyframes typingBounce { 0%,60%,100%{transform:translateY(0)} 30%{transform:translateY(-6px)} }
+
+  /* Example prompts */
+  .examples { padding:16px 20px; }
+  .example-btn { display:block; width:100%; text-align:left; padding:10px 12px; background:var(--bg-tertiary); color:var(--text-secondary); border:1px solid var(--border); border-radius:var(--radius-sm); font-size:12px; margin-bottom:8px; cursor:pointer; transition:all 0.2s; line-height:1.4; }
+  .example-btn:hover { background:var(--bg-hover); color:var(--text-primary); border-color:var(--accent); }
+
+  /* Responsive */
+  @media (max-width:900px) {
+    .sidebar { display:none; }
+    .workspace-panel { display:none; }
+  }
+</style>
+</head>
+<body>
+<div class="app">
+  <!-- Sidebar -->
+  <aside class="sidebar">
+    <div class="sidebar-header">
+      <div class="logo">SoniCoder v2</div>
+      <div class="logo-sub">AI Code Agent</div>
+    </div>
+
+    <div class="model-selector">
+      <label>Model</label>
+      <select id="modelSelect"></select>
+    </div>
+
+    <div class="model-status">
+      <span class="status-dot unknown" id="statusDot"></span>
+      <span class="status-text" id="statusText">Initializing...</span>
+    </div>
+
+    <button class="new-chat-btn" onclick="newChat()">+ New Chat</button>
+
+    <div class="sidebar-section">
+      <h3>Quick Start</h3>
+      <div class="sidebar-info">
+        Ask me to build anything — a web app, API, script, or full-stack project. I'll create the files and verify they work.
+      </div>
+    </div>
+
+    <div class="examples" id="examplesPanel">
+      <div class="example-btn" onclick="useExample(this)">Build a Flask REST API with user CRUD endpoints</div>
+      <div class="example-btn" onclick="useExample(this)">Create a React todo app with localStorage persistence</div>
+      <div class="example-btn" onclick="useExample(this)">Write a Python web scraper with error handling</div>
+      <div class="example-btn" onclick="useExample(this)">Build a CLI tool that converts Markdown to HTML</div>
+      <div class="example-btn" onclick="useExample(this)">Create a FastAPI backend with JWT authentication</div>
+    </div>
+  </aside>
+
+  <!-- Main Chat -->
+  <div class="main">
+    <div class="chat-header">
+      <h2>Chat</h2>
+      <div class="header-actions">
+        <button class="header-btn" onclick="toggleWorkspace()">Workspace</button>
+        <button class="header-btn" onclick="refreshWorkspace()">Refresh</button>
+      </div>
+    </div>
+
+    <div class="chat-messages" id="chatMessages">
+      <div class="message" style="text-align:center; padding:60px 20px;">
+        <div style="font-size:28px; margin-bottom:12px; opacity:0.6;">SoniCoder v2</div>
+        <div style="color:var(--text-secondary); font-size:14px; max-width:500px; margin:0 auto; line-height:1.7;">
+          AI-powered code agent that writes, runs, and debugs code. Built with patterns from Gemini CLI and Claude Code. Supports MCP tools and multiple 1B models.
+        </div>
+      </div>
+    </div>
+
+    <div class="chat-input">
+      <div class="input-wrapper">
+        <textarea id="chatInput" placeholder="Describe what you want to build..." rows="1" onkeydown="handleKeyDown(event)" oninput="autoResize(this)"></textarea>
+        <button class="send-btn" id="sendBtn" onclick="sendMessage()">Send</button>
+      </div>
+      <div class="input-hint">Press Enter to send, Shift+Enter for new line</div>
+    </div>
+  </div>
+
+  <!-- Workspace Panel -->
+  <div class="workspace-panel" id="workspacePanel">
+    <div class="ws-header">
+      <h3>Workspace</h3>
+    </div>
+    <div class="ws-tree" id="wsTree">Loading...</div>
+  </div>
+</div>
+
+<script>
+const __CONFIG__ = __RUNTIME_CONFIG__ || {};
+
+// ── State ──
+let sessionId = '';
+let isStreaming = false;
+let abortController = null;
+let currentToolCalls = {};
+
+// ── Init ──
+document.addEventListener('DOMContentLoaded', () => {
+  loadConfig();
+  pollModelStatus();
+  refreshWorkspace();
+});
+
+async function loadConfig() {
+  try {
+    const resp = await fetch('/api/get_config');
+    const config = await resp.json();
+    const parsed = typeof config === 'string' ? JSON.parse(config) : config;
+
+    const select = document.getElementById('modelSelect');
+    select.innerHTML = '';
+    (parsed.models || []).forEach(m => {
+      const opt = document.createElement('option');
+      opt.value = m.key;
+      opt.textContent = `${m.name} (${m.size_gb}GB)`;
+      if (m.key === parsed.default_model) opt.selected = true;
+      select.appendChild(opt);
+    });
+
+    select.addEventListener('change', () => switchModel(select.value));
+  } catch(e) { console.error('Config load failed:', e); }
+}
+
+async function pollModelStatus() {
+  try {
+    const resp = await fetch('/api/model_status');
+    const status = JSON.parse(await resp.json());
+    const dot = document.getElementById('statusDot');
+    const text = document.getElementById('statusText');
+
+    dot.className = 'status-dot ' + (status.status || 'unknown');
+    const names = { ready:'Ready', loading:'Loading...', error:'Error', unknown:'Unknown' };
+    text.textContent = (status.model_name || names[status.status] || 'Unknown');
+
+    if (status.status === 'loading') {
+      setTimeout(pollModelStatus, 3000);
+    }
+  } catch(e) { setTimeout(pollModelStatus, 5000); }
+}
+
+async function switchModel(key) {
+  try {
+    await fetch('/api/switch_model', { method:'POST', body: new URLSearchParams({data: JSON.stringify([key])}), headers:{'Content-Type':'application/x-www-form-urlencoded'} });
+    document.getElementById('statusDot').className = 'status-dot loading';
+    document.getElementById('statusText').textContent = 'Switching...';
+    setTimeout(pollModelStatus, 2000);
+  } catch(e) { console.error(e); }
+}
+
+async function refreshWorkspace() {
+  try {
+    const resp = await fetch('/api/workspace_tree', { method:'POST', body: new URLSearchParams({data: JSON.stringify([3])}), headers:{'Content-Type':'application/x-www-form-urlencoded'} });
+    const data = JSON.parse(await resp.json());
+    document.getElementById('wsTree').textContent = data.tree || '(empty)';
+  } catch(e) { console.error(e); }
+}
+
+function toggleWorkspace() {
+  document.getElementById('workspacePanel').classList.toggle('collapsed');
+}
+
+function newChat() {
+  if (sessionId) {
+    fetch('/api/reset_session', { method:'POST', body: new URLSearchParams({data: JSON.stringify([sessionId])}), headers:{'Content-Type':'application/x-www-form-urlencoded'} });
+  }
+  sessionId = '';
+  document.getElementById('chatMessages').innerHTML = `
+    <div class="message" style="text-align:center; padding:60px 20px;">
+      <div style="font-size:28px; margin-bottom:12px; opacity:0.6;">SoniCoder v2</div>
+      <div style="color:var(--text-secondary); font-size:14px;">New session started. What would you like to build?</div>
+    </div>`;
+}
+
+function useExample(btn) {
+  document.getElementById('chatInput').value = btn.textContent.trim();
+  sendMessage();
+}
+
+function handleKeyDown(e) {
+  if (e.key === 'Enter' && !e.shiftKey) {
+    e.preventDefault();
+    sendMessage();
+  }
+}
+
+function autoResize(el) {
+  el.style.height = 'auto';
+  el.style.height = Math.min(el.scrollHeight, 200) + 'px';
+}
+
+// ── Message Rendering ──
+function appendMessage(role, content, extra={}) {
+  const container = document.getElementById('chatMessages');
+  const div = document.createElement('div');
+  div.className = 'message';
+  div.id = extra.msgId || '';
+
+  const time = new Date().toLocaleTimeString([], {hour:'2-digit', minute:'2-digit'});
+  const avatarClass = role === 'user' ? 'user' : 'agent';
+  const avatarText = role === 'user' ? 'Y' : 'SC';
+
+  div.innerHTML = `
+    <div class="message-header">
+      <div class="message-avatar ${avatarClass}">${avatarText}</div>
+      <span class="message-role">${role === 'user' ? 'You' : 'SoniCoder'}</span>
+      <span class="message-time">${time}</span>
+      ${extra.iteration ? `<span class="iteration-badge">Iteration ${extra.iteration}</span>` : ''}
+    </div>
+    <div class="message-body">${formatContent(content)}</div>`;
+
+  container.appendChild(div);
+  container.scrollTop = container.scrollHeight;
+  return div;
+}
+
+function formatContent(text) {
+  if (!text) return '';
+  // Code blocks
+  text = text.replace(/```(\w*)\n([\s\S]*?)```/g, (_, lang, code) => {
+    return `<pre><code class="language-${lang}">${escapeHtml(code.trim())}</code></pre>`;
+  });
+  // Inline code
+  text = text.replace(/`([^`]+)`/g, '<code>$1</code>');
+  // Bold
+  text = text.replace(/\*\*([^*]+)\*\*/g, '<strong>$1</strong>');
+  // Lists
+  text = text.replace(/^[-•] (.+)$/gm, '<li>$1</li>');
+  text = text.replace(/(<li>.*<\/li>)/s, '<ul>$1</ul>');
+  // Line breaks
+  text = text.replace(/\n/g, '<br>');
+  return text;
+}
+
+function escapeHtml(text) {
+  const d = document.createElement('div');
+  d.textContent = text;
+  return d.innerHTML;
+}
+
+function addToolCall(toolName, args) {
+  const container = document.getElementById('chatMessages');
+  const id = 'tc_' + Date.now();
+  const div = document.createElement('div');
+  div.className = 'tool-call';
+  div.id = id;
+  div.innerHTML = `
+    <div class="tool-call-header">
+      <span class="tool-icon">⚡</span>
+      <span class="tool-name">${escapeHtml(toolName)}</span>
+      <span class="tool-status running">Running</span>
+    </div>
+    <div class="tool-args">${formatToolArgs(args)}</div>
+    <div class="tool-result" style="display:none;"></div>`;
+  container.appendChild(div);
+  container.scrollTop = container.scrollHeight;
+  currentToolCalls[id] = div;
+  return id;
+}
+
+function updateToolCall(id, success, output) {
+  const div = currentToolCalls[id];
+  if (!div) return;
+  const status = div.querySelector('.tool-status');
+  const result = div.querySelector('.tool-result');
+  status.className = 'tool-status ' + (success ? 'success' : 'error');
+  status.textContent = success ? 'Done' : 'Error';
+  if (output) {
+    result.textContent = output;
+    result.style.display = 'block';
+  }
+  document.getElementById('chatMessages').scrollTop = document.getElementById('chatMessages').scrollHeight;
+}
+
+function formatToolArgs(args) {
+  if (!args) return '';
+  const lines = Object.entries(args).map(([k, v]) => {
+    const val = typeof v === 'string' ? (v.length > 60 ? v.slice(0, 60) + '...' : v) : JSON.stringify(v);
+    return `${k}: ${val}`;
+  });
+  return escapeHtml(lines.join('\n'));
+}
+
+// ── Streaming ──
+let currentStreamDiv = null;
+let currentStreamContent = '';
+
+function startStreaming() {
+  currentStreamContent = '';
+  const container = document.getElementById('chatMessages');
+  // Remove welcome message if present
+  const welcome = container.querySelector('.message[style*="text-align:center"]');
+  if (welcome) welcome.remove();
+
+  currentStreamDiv = appendMessage('agent', '', {});
+  const body = currentStreamDiv.querySelector('.message-body');
+  body.innerHTML = '<div class="typing-indicator"><div class="typing-dot"></div><div class="typing-dot"></div><div class="typing-dot"></div></div>';
+}
+
+function updateStreaming(text) {
+  if (!currentStreamDiv) return;
+  const body = currentStreamDiv.querySelector('.message-body');
+  body.innerHTML = formatContent(text) || '<div class="typing-indicator"><div class="typing-dot"></div><div class="typing-dot"></div><div class="typing-dot"></div></div>';
+  document.getElementById('chatMessages').scrollTop = document.getElementById('chatMessages').scrollHeight;
+}
+
+function finalizeStream(text) {
+  if (!currentStreamDiv) return;
+  const body = currentStreamDiv.querySelector('.message-body');
+  body.innerHTML = formatContent(text);
+  currentStreamDiv = null;
+}
+
+// ── Send Message ──
+async function sendMessage() {
+  const input = document.getElementById('chatInput');
+  const message = input.value.trim();
+  if (!message || isStreaming) return;
+
+  input.value = '';
+  input.style.height = 'auto';
+  isStreaming = true;
+
+  const sendBtn = document.getElementById('sendBtn');
+  sendBtn.textContent = 'Stop';
+  sendBtn.className = 'stop-btn';
+  sendBtn.onclick = stopStreaming;
+
+  appendMessage('user', message);
+  startStreaming();
+
+  try {
+    const resp = await fetch('/api/chat', {
+      method: 'POST',
+      body: new URLSearchParams({
+        data: JSON.stringify([message, sessionId, '']),
+      }),
+      headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
+    });
+
+    const reader = resp.body.getReader();
+    const decoder = new TextDecoder();
+    let buffer = '';
+
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+
+      buffer += decoder.decode(value, { stream: true });
+      const lines = buffer.split('\n');
+      buffer = lines.pop() || '';
+
+      for (const line of lines) {
+        if (!line.trim()) continue;
+        try {
+          const event = JSON.parse(line);
+
+          switch (event.type) {
+            case 'streaming':
+              const display = (event.content || '').replace(/```tool[\s\S]*?```/g, '').trim();
+              updateStreaming(display);
+              break;
+
+            case 'tool_call':
+              addToolCall(event.tool, event.args);
+              break;
+
+            case 'tool_result':
+              // Find the latest tool call for this tool name
+              const tcIds = Object.keys(currentToolCalls).reverse();
+              for (const tcId of tcIds) {
+                const tcDiv = currentToolCalls[tcId];
+                if (tcDiv && tcDiv.querySelector('.tool-status.running')) {
+                  updateToolCall(tcId, event.result.success, event.result.output || event.result.error);
+                  break;
+                }
+              }
+              break;
+
+            case 'status':
+              if (event.iteration) {
+                const header = currentStreamDiv?.querySelector('.message-header');
+                if (header) {
+                  let badge = header.querySelector('.iteration-badge');
+                  if (!badge) {
+                    badge = document.createElement('span');
+                    badge.className = 'iteration-badge';
+                    header.appendChild(badge);
+                  }
+                  badge.textContent = `Iteration ${event.iteration}/${event.max_iterations}`;
+                }
+              }
+              break;
+
+            case 'complete':
+              const finalText = event.content || currentStreamContent;
+              finalizeStream(finalText || 'Done.');
+              if (event.iterations) {
+                setTimeout(refreshWorkspace, 500);
+              }
+              break;
+
+            case 'error':
+              finalizeStream(`Error: ${event.error}`);
+              break;
+          }
+        } catch (e) { /* skip malformed lines */ }
+      }
+    }
+  } catch (e) {
+    if (e.name !== 'AbortError') {
+      finalizeStream(`Connection error: ${e.message}`);
+    }
+  }
+
+  isStreaming = false;
+  sendBtn.textContent = 'Send';
+  sendBtn.className = 'send-btn';
+  sendBtn.onclick = sendMessage;
+}
+
+function stopStreaming() {
+  // The streaming will naturally end when we close
+  isStreaming = false;
+  const sendBtn = document.getElementById('sendBtn');
+  sendBtn.textContent = 'Send';
+  sendBtn.className = 'send-btn';
+  sendBtn.onclick = sendMessage;
+  if (currentStreamDiv) {
+    finalizeStream(currentStreamContent || '(stopped)');
+  }
+}
+</script>
+</body>
+</html>