Merge origin/dev into main

src/agent_tools/__init__.py

@@ -22,6 +22,9 @@ from .subprocess_tools import BashTool, PythonTool
 from .web_tools import WebSearchTool, WebFetchTool
 from .filesystem_tools import ReadFileTool, WriteFileTool, EditFileTool, LsTool, GlobTool, GrepTool, GetWorkspaceTool
 from .document_tools import CreateDocumentTool, UpdateDocumentTool, EditDocumentTool, SuggestDocumentTool, ManageDocumentTool
+from .model_interaction_tools import ChatWithModelTool, AskTeacherTool, ListModelsTool
+from .bg_job_tools import ManageBgJobsTool
+from .session_tools import CreateSessionTool, ListSessionsTool, SendToSessionTool, ManageSessionTool
 
 TOOL_HANDLERS = {
     "bash": BashTool().execute,
@@ -40,6 +43,14 @@ TOOL_HANDLERS = {
     "suggest_document": SuggestDocumentTool().execute,
     "manage_documents": ManageDocumentTool().execute,
     "get_workspace": GetWorkspaceTool().execute,
+    "chat_with_model": ChatWithModelTool().execute,
+    "ask_teacher": AskTeacherTool().execute,
+    "list_models": ListModelsTool().execute,
+    "manage_bg_jobs": ManageBgJobsTool().execute,
+    "create_session": CreateSessionTool().execute,
+    "list_sessions": ListSessionsTool().execute,
+    "send_to_session": SendToSessionTool().execute,
+    "manage_session": ManageSessionTool().execute,
 }
 
 # ---------------------------------------------------------------------------
@@ -52,7 +63,7 @@ PYTHON_TIMEOUT = 30
 
 # Tool types that trigger execution
 TOOL_TAGS = {"bash", "python", "web_search", "web_fetch", "read_file", "write_file", "edit_file",
-             "grep", "glob", "ls", "get_workspace",
+             "grep", "glob", "ls", "get_workspace", "manage_bg_jobs",
              "create_document", "update_document", "edit_document",
              "search_chats",
              "chat_with_model", "create_session", "list_sessions",

src/agent_tools/bg_job_tools.py

@@ -0,0 +1,98 @@
+"""Agent tool to inspect and control detached background `bash` jobs.
+
+`bash` blocks prefixed with a `#!bg` marker run detached via `src.bg_jobs`; the
+agent is auto-re-invoked with the output when they finish. This tool covers the
+gaps in that flow: list the jobs in the current chat, read a still-running job's
+output on demand, and kill a runaway job instead of waiting out its max-runtime.
+
+Registry tool (`TOOL_HANDLERS["manage_bg_jobs"]`). Jobs are scoped to the chat
+that launched them, so every action requires the caller's `session_id` and a job
+from another session is treated as not found.
+"""
+
+import json
+import time
+from typing import Any, Dict, List
+
+_LIST_ACTIONS = {"list", "ls", "jobs"}
+_OUTPUT_ACTIONS = {"output", "get", "read", "tail", "status", "show"}
+_KILL_ACTIONS = {"kill", "stop", "cancel", "terminate"}
+
+
+def _age(rec: Dict[str, Any]) -> str:
+    start = rec.get("started_at")
+    if not start:
+        return "?"
+    secs = int(time.time() - start)
+    if secs < 60:
+        return f"{secs}s"
+    if secs < 3600:
+        return f"{secs // 60}m"
+    return f"{secs // 3600}h{(secs % 3600) // 60}m"
+
+
+def _status_label(rec: Dict[str, Any]) -> str:
+    status = rec.get("status", "?")
+    if rec.get("killed"):
+        return "killed"
+    if rec.get("timed_out"):
+        return "timed out"
+    if rec.get("died"):
+        return "died"
+    if status in ("done", "failed"):
+        return f"{status} (exit {rec.get('exit_code')})"
+    return status
+
+
+def _row(rec: Dict[str, Any]) -> str:
+    cmd = (rec.get("command") or "").strip().splitlines()[0][:80]
+    return f"[{rec.get('id')}] {_status_label(rec)} | {_age(rec)} | {cmd}"
+
+
+class ManageBgJobsTool:
+    async def execute(self, content: str, ctx: dict) -> dict:
+        from src import bg_jobs
+
+        session_id = ctx.get("session_id")
+        raw = (content or "").strip()
+        try:
+            args = json.loads(raw) if raw else {}
+        except (ValueError, TypeError):
+            args = {}
+        if not isinstance(args, dict):
+            args = {}
+        action = str(args.get("action", "list")).strip().lower()
+        job_id = str(args.get("job_id") or args.get("id") or "").strip()
+
+        if not session_id:
+            return {"error": "manage_bg_jobs: no active chat session; background jobs are scoped to a chat.", "exit_code": 1}
+
+        if action in _LIST_ACTIONS:
+            jobs: List[Dict[str, Any]] = bg_jobs.list_for_session(session_id)
+            if not jobs:
+                return {"output": "No background jobs in this chat.", "exit_code": 0}
+            jobs.sort(key=lambda r: r.get("started_at") or 0, reverse=True)
+            lines = "\n".join(_row(r) for r in jobs)
+            return {"output": f"{len(jobs)} background job(s):\n{lines}", "exit_code": 0}
+
+        if action in _OUTPUT_ACTIONS or action in _KILL_ACTIONS:
+            if not job_id:
+                return {"error": f"manage_bg_jobs: action '{action}' requires a job_id (see action='list').", "exit_code": 1}
+            rec = bg_jobs.get(job_id)
+            # Scope: only the chat that launched a job may see or control it.
+            if rec is None or rec.get("session_id") != session_id:
+                return {"error": f"manage_bg_jobs: no background job '{job_id}' in this chat.", "exit_code": 1}
+
+            if action in _KILL_ACTIONS:
+                if rec.get("status") != "running":
+                    return {"output": f"Job `{job_id}` already {_status_label(rec)}; nothing to kill.", "exit_code": 0}
+                killed = bg_jobs.kill(job_id)
+                return {"output": f"Killed background job `{job_id}` ({(killed or {}).get('command', '').splitlines()[0][:80]}).", "exit_code": 0}
+
+            out = rec.get("output") or "(no output yet)"
+            return {
+                "output": f"Job `{job_id}` [{_status_label(rec)}, {_age(rec)}]\nCommand: {rec.get('command')}\n\nOutput:\n{out}",
+                "exit_code": 0,
+            }
+
+        return {"error": f"manage_bg_jobs: unknown action '{action}'. Use list, output, or kill.", "exit_code": 1}

src/agent_tools/filesystem_tools.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import os
+import re
 import difflib
 import fnmatch
 import shutil
@@ -16,6 +17,31 @@ _CODENAV_SKIP_DIRS = frozenset({
 _CODENAV_MAX_HITS = 200
 _CODENAV_MAX_LINE = 400
 
+
+def _glob_to_regex(pat: str) -> "re.Pattern":
+    """Translate a forward-slash glob (**, *, ?) into a compiled regex.
+    `**/` matches zero or more complete directories.
+    `*` matches within a single path segment (does not cross /).
+    """
+    i, n, out = 0, len(pat), []
+    while i < n:
+        if pat[i : i + 3] == "**/":
+            out.append("(?:[^/]+/)*")
+            i += 3
+        elif pat[i : i + 2] == "**":
+            out.append(".*")
+            i += 2
+        elif pat[i] == "*":
+            out.append("[^/]*")
+            i += 1
+        elif pat[i] == "?":
+            out.append("[^/]")
+            i += 1
+        else:
+            out.append(re.escape(pat[i]))
+            i += 1
+    return re.compile("".join(out))
+
 def _unified_diff(old: str, new: str, path: str) -> Optional[Dict[str, Any]]:
     if old == new:
         return None
@@ -259,23 +285,38 @@ class GlobTool:
             return {"error": f"glob: {e}", "exit_code": 1}
 
         def _glob():
-            from pathlib import Path
-            base = Path(root)
-            if not base.is_dir():
+            base = os.path.abspath(root)
+            if not os.path.isdir(base):
                 return None, f"glob: {root}: not a directory"
+            norm_pat = pattern.replace("\\", "/")
+            # Fast path: literal pattern (no wildcards) → direct path lookup.
+            if not any(c in norm_pat for c in "*?["):
+                cand = os.path.normpath(os.path.join(base, norm_pat))
+                if os.path.exists(cand):
+                    return [cand], None
+                # Literal not at exact path — fall through to walk so
+                # e.g. "foo.py" still matches at any depth (like rglob).
+            # Compile glob to regex: * stays within one segment, **/ spans dirs.
+            regex = _glob_to_regex(norm_pat)
             matched = []
+            cap = _CODENAV_MAX_HITS * 5
             try:
-                for p in base.rglob(pattern):
-                    if set(p.relative_to(base).parts) & _CODENAV_SKIP_DIRS:
-                        continue
-                    try:
-                        mtime = p.stat().st_mtime
-                    except OSError:
-                        mtime = 0
-                    matched.append((mtime, str(p)))
-                    if len(matched) > _CODENAV_MAX_HITS * 5:
+                for dp, dns, fns in os.walk(base):
+                    # Prune skipped dirs before descending (unlike rglob which
+                    # descends first then filters — fatal on large node_modules).
+                    dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS]
+                    for name in fns + dns:
+                        full = os.path.join(dp, name)
+                        rel = os.path.relpath(full, base).replace(os.sep, "/")
+                        if regex.fullmatch(rel) or regex.fullmatch(name):
+                            try:
+                                mtime = os.stat(full).st_mtime
+                            except OSError:
+                                mtime = 0
+                            matched.append((mtime, full))
+                    if len(matched) > cap:
                         break
-            except (OSError, ValueError) as _e:
+            except OSError as _e:
                 return None, f"glob: {_e}"
             matched.sort(key=lambda t: t[0], reverse=True)
             return [pth for _, pth in matched[:_CODENAV_MAX_HITS]], None

src/agent_tools/model_interaction_tools.py

@@ -0,0 +1,208 @@
+"""model_interaction_tools.py - agent tools for talking to other models.
+
+Owns the model-interaction tool implementations (chat_with_model, ask_teacher,
+list_models) and their handler classes, registered in ``TOOL_HANDLERS``. Part
+of the tool -> registry migration (#3629): the implementations were moved here
+out of ``src.ai_interaction`` so dispatch flows through the registry instead of
+the elif chain / dispatch_ai_tool in tool_execution.py.
+
+Shared helpers that still live in ``src.ai_interaction`` and are used by tools
+not yet migrated (``_resolve_model``, ``AI_CHAT_TIMEOUT``) are imported lazily
+inside the functions to avoid an import cycle at module load.
+"""
+import logging
+from typing import Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+_TEACHER_SYSTEM_PROMPT = (
+    "You are a senior AI mentor. A less capable model is stuck on a problem and asking for help. "
+    "Provide clear, actionable guidance:\n"
+    "1. Brief analysis of the problem\n"
+    "2. Recommended approach (step by step)\n"
+    "3. Key things to watch out for\n\n"
+    "Be concise and practical. No preamble."
+)
+
+
+async def chat_with_model(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """Send a message to a specific model and return its response.
+
+    Content format:
+      Line 1: model_name (or model_name@endpoint_name)
+      Line 2+: the message to send
+    """
+    from src.ai_interaction import _resolve_model, AI_CHAT_TIMEOUT
+    from src.llm_core import llm_call_async
+
+    lines = content.strip().split("\n", 1)
+    if not lines or not lines[0].strip():
+        return {"error": "First line must be the model name"}
+
+    model_spec = lines[0].strip()
+    message = lines[1].strip() if len(lines) > 1 else ""
+    if not message:
+        return {"error": "No message provided (line 2+ is the message)"}
+
+    try:
+        url, model, headers = _resolve_model(model_spec, owner=owner)
+    except ValueError as e:
+        return {"error": str(e)}
+
+    try:
+        response = await llm_call_async(
+            url, model,
+            [{"role": "user", "content": message}],
+            headers=headers,
+            timeout=AI_CHAT_TIMEOUT,
+        )
+        # Truncate very long responses
+        if len(response) > 10000:
+            response = response[:10000] + "\n... (truncated)"
+        return {"model": model, "response": response}
+    except Exception as e:
+        logger.error(f"chat_with_model failed: {e}")
+        return {"error": f"Failed to get response from {model_spec}: {e}"}
+
+
+async def ask_teacher(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """Ask a more capable model for help.
+
+    Content format:
+      Line 1: model_name (or 'auto')
+      Line 2+: the problem description
+    """
+    from src.ai_interaction import _resolve_model, AI_CHAT_TIMEOUT
+    from src.llm_core import llm_call_async
+    from src.settings import get_setting
+
+    lines = content.strip().split("\n", 1)
+    model_spec = lines[0].strip() if lines else "auto"
+    problem = lines[1].strip() if len(lines) > 1 else ""
+
+    if not problem:
+        return {"error": "No problem description provided"}
+
+    if model_spec.lower() in ("auto", ""):
+        model_spec = get_setting("teacher_model", "")
+        if not model_spec:
+            return {"error": "No teacher model configured. Specify a model name or set teacher_model in settings."}
+
+    try:
+        url, model, headers = _resolve_model(model_spec, owner=owner)
+    except ValueError as e:
+        return {"error": str(e)}
+
+    try:
+        response = await llm_call_async(
+            url, model,
+            [
+                {"role": "system", "content": _TEACHER_SYSTEM_PROMPT},
+                {"role": "user", "content": f"Problem:\n{problem}"},
+            ],
+            headers=headers,
+            timeout=AI_CHAT_TIMEOUT,
+        )
+        if len(response) > 8000:
+            response = response[:8000] + "\n... (truncated)"
+        return {"model": model, "response": response, "teacher": True}
+    except Exception as e:
+        logger.error(f"ask_teacher failed: {e}")
+        return {"error": f"Teacher call failed ({model_spec}): {e}"}
+
+
+async def list_models(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """List all available models across configured endpoints.
+
+    Content = optional filter keyword.
+    """
+    import json
+    import httpx
+    from src.database import SessionLocal, ModelEndpoint
+    from src.llm_core import _detect_provider, ANTHROPIC_MODELS
+    from src.auth_helpers import owner_filter
+    from src.endpoint_resolver import resolve_endpoint_runtime, build_headers, build_models_url
+
+    keyword = content.strip().lower() if content.strip() else None
+
+    db = SessionLocal()
+    try:
+        query = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True)
+        if owner:
+            query = owner_filter(query, ModelEndpoint, owner)
+        endpoints = query.all()
+        if not endpoints:
+            return {"results": "No enabled model endpoints configured."}
+
+        result_lines = []
+        total_models = 0
+
+        for ep in endpoints:
+            try:
+                base, api_key = resolve_endpoint_runtime(ep, owner=owner)
+            except Exception:
+                continue
+            provider = _detect_provider(base)
+            headers = build_headers(api_key, base)
+
+            model_ids = []
+            if provider == "anthropic":
+                model_ids = list(ANTHROPIC_MODELS)
+            else:
+                try:
+                    models_url = build_models_url(base)
+                    if models_url:
+                        r = httpx.get(models_url, headers=headers, timeout=5)
+                        r.raise_for_status()
+                        data = r.json()
+                        model_ids = [m.get("id") for m in (data.get("data") or []) if m.get("id")]
+                        if not model_ids:
+                            model_ids = [
+                                m.get("name") or m.get("model")
+                                for m in (data.get("models") or [])
+                                if m.get("name") or m.get("model")
+                            ]
+                    else:
+                        model_ids = json.loads(ep.cached_models or "[]")
+                except Exception:
+                    model_ids = ["(endpoint offline)"]
+
+            if keyword:
+                model_ids = [m for m in model_ids if keyword in m.lower() or keyword in (ep.name or "").lower()]
+
+            if model_ids:
+                result_lines.append(f"\n**{ep.name or base}** ({provider}):")
+                for mid in model_ids:
+                    result_lines.append(f"  - `{mid}`")
+                    total_models += 1
+
+        if not result_lines:
+            return {"results": "No models found" + (f" matching '{keyword}'" if keyword else "") + "."}
+
+        header = f"Available models ({total_models} total):"
+        return {"results": header + "\n".join(result_lines)}
+    except Exception as e:
+        logger.error(f"list_models failed: {e}")
+        return {"error": str(e)}
+    finally:
+        db.close()
+
+
+# ---------------------------------------------------------------------------
+# Handler classes registered in TOOL_HANDLERS
+# ---------------------------------------------------------------------------
+
+class ChatWithModelTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await chat_with_model(content, ctx.get("session_id"), owner=ctx.get("owner"))
+
+
+class AskTeacherTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await ask_teacher(content, ctx.get("session_id"), owner=ctx.get("owner"))
+
+
+class ListModelsTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await list_models(content, ctx.get("session_id"), owner=ctx.get("owner"))

src/agent_tools/session_tools.py

@@ -0,0 +1,464 @@
+"""session_tools.py - agent tools for AI-to-AI session management.
+
+Owns create_session, list_sessions, send_to_session and manage_session, moved
+out of src.ai_interaction as part of the tool -> registry migration (#3629), and
+their handler classes registered in TOOL_HANDLERS.
+
+The session manager is a runtime-set singleton in src.ai_interaction, so each
+function fetches it via get_session_manager() (imported here); _resolve_model and
+AI_CHAT_TIMEOUT are reused from there too.
+"""
+import json
+import logging
+import uuid
+from typing import Dict, Optional
+
+from src.ai_interaction import get_session_manager, _resolve_model, AI_CHAT_TIMEOUT
+
+logger = logging.getLogger(__name__)
+
+
+async def create_session(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """Create a new chat session.
+
+    Content format:
+      Line 1: session name
+      Line 2: model_name (or model_name@endpoint_name)
+    """
+    _session_manager = get_session_manager()
+    if not _session_manager:
+        return {"error": "Session manager not available"}
+
+    lines = content.strip().split("\n")
+    if len(lines) < 2:
+        return {"error": "Need 2 lines: session name, then model spec"}
+
+    name = lines[0].strip()
+    model_spec = lines[1].strip()
+
+    if not name:
+        return {"error": "Session name cannot be empty"}
+
+    try:
+        url, model, headers = _resolve_model(model_spec, owner=owner)
+    except ValueError as e:
+        return {"error": str(e)}
+
+    sid = str(uuid.uuid4())[:8]
+    try:
+        _session_manager.create_session(
+            session_id=sid,
+            name=name,
+            endpoint_url=url,
+            model=model,
+            rag=False,
+            owner=owner,
+        )
+        # Store headers on session for future calls
+        sess = _session_manager.get_session(sid)
+        if sess and headers:
+            sess.headers = headers
+        try:
+            from src.event_bus import fire_event
+            fire_event("session_created", owner)
+        except Exception:
+            logger.debug("session_created event dispatch failed", exc_info=True)
+
+        return {"session_id": sid, "name": name, "model": model, "endpoint_url": url}
+    except Exception as e:
+        logger.error(f"create_session failed: {e}")
+        return {"error": f"Failed to create session: {e}"}
+
+async def list_sessions(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """List sessions sorted by most-recently-active first.
+
+    Output includes a relative "last active" timestamp per row so the
+    agent can answer "open my last chat" without guessing from titles.
+    The most-recent session is always first in the list.
+
+    Content = optional filter keyword (matches session name).
+    """
+    _session_manager = get_session_manager()
+    if not _session_manager:
+        return {"error": "Session manager not available"}
+
+    keyword = content.strip().lower() if content.strip() else None
+
+    try:
+        from core.database import SessionLocal, Session as DbSession
+        from datetime import datetime, timezone
+
+        # Pull every session's last_accessed from the DB so we can sort
+        # by recency. In-memory sessions hold name + model + msg_count;
+        # the DB row holds the timestamps.
+        db = SessionLocal()
+        try:
+            db_rows = {r.id: r for r in db.query(DbSession).all()}
+        finally:
+            db.close()
+
+        # SECURITY: scope to the caller's sessions. Passing None returned
+        # every user's sessions, which the agent tool then exposed via the
+        # "list my chats" reply.
+        sessions = _session_manager.get_sessions_for_user(owner)
+        rows = []
+        for sid, sess in sessions.items():
+            if keyword and keyword not in (sess.name or "").lower():
+                continue
+            db_row = db_rows.get(sid)
+            # Prefer last_accessed; fall back to updated_at, then created_at.
+            ts = None
+            if db_row:
+                ts = getattr(db_row, 'last_accessed', None) or getattr(db_row, 'updated_at', None) or getattr(db_row, 'created_at', None)
+            rows.append((ts, sid, sess))
+
+        # Sort by timestamp DESC; rows without a timestamp sink to the bottom.
+        rows.sort(key=lambda r: r[0] or datetime.min, reverse=True)
+
+        def _rel(ts):
+            if not ts:
+                return 'never'
+            now = datetime.utcnow()
+            try:
+                if ts.tzinfo is not None:
+                    now = datetime.now(timezone.utc)
+                diff = (now - ts).total_seconds()
+            except Exception:
+                return 'unknown'
+            if diff < 60: return 'just now'
+            if diff < 3600: return f'{int(diff / 60)}m ago'
+            if diff < 86400: return f'{int(diff / 3600)}h ago'
+            if diff < 86400 * 7: return f'{int(diff / 86400)}d ago'
+            return ts.strftime('%Y-%m-%d')
+
+        lines = []
+        for i, (ts, sid, sess) in enumerate(rows):
+            if i >= 50:
+                lines.append(f"... and {len(rows) - 50} more (showing first 50)")
+                break
+            safe_name = (sess.name or "Untitled").replace("[", "\\[").replace("]", "\\]")
+            msg_count = getattr(sess, "message_count", 0) or 0
+            model = getattr(sess, "model", "unknown")
+            marker = " ← most recent" if i == 0 else ""
+            lines.append(f"- **[{safe_name}](#session-{sid})** (id: `{sid}`, model: {model}, {msg_count} msgs, last active {_rel(ts)}){marker}")
+
+        if not lines:
+            return {"results": "No sessions found" + (f" matching '{keyword}'" if keyword else "") + "."}
+
+        return {
+            "results": (
+                f"Found {len(rows)} session(s), sorted most-recent first:\n"
+                + "\n".join(lines)
+                + "\n\nAssistant: when replying to the user, preserve the chat-title markdown links exactly as shown, e.g. `[Chat](#session-id)`. Do not rewrite this as a plain, non-clickable table."
+            )
+        }
+    except Exception as e:
+        logger.error(f"list_sessions failed: {e}")
+        return {"error": str(e)}
+
+async def send_to_session(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """Send a message to an existing session and get a response.
+
+    Content format:
+      Line 1: session_id
+      Line 2+: message
+    """
+    _session_manager = get_session_manager()
+    from src.llm_core import llm_call_async
+    from core.models import ChatMessage
+
+    if not _session_manager:
+        return {"error": "Session manager not available"}
+
+    lines = content.strip().split("\n", 1)
+    if len(lines) < 2:
+        return {"error": "Need 2 lines: session_id, then message"}
+
+    target_sid = lines[0].strip()
+    message = lines[1].strip()
+
+    sess = _session_manager.get_session(target_sid)
+    if not sess:
+        return {"error": f"Session '{target_sid}' not found"}
+
+    # Owner-scope: reject access to another user's session
+    if owner and getattr(sess, "owner", None) and sess.owner != owner:
+        return {"error": f"Session '{target_sid}' not found"}
+
+    if not message:
+        return {"error": "No message provided"}
+
+    try:
+        # Build context from session history
+        context = sess.get_context_messages()
+        context.append({"role": "user", "content": message})
+
+        response = await llm_call_async(
+            sess.endpoint_url, sess.model, context,
+            headers=sess.headers,
+            timeout=AI_CHAT_TIMEOUT,
+        )
+
+        # Save both messages to session
+        sess.add_message(ChatMessage("user", message))
+        sess.add_message(ChatMessage("assistant", response))
+
+        # Truncate for tool output
+        if len(response) > 10000:
+            response = response[:10000] + "\n... (truncated)"
+
+        return {
+            "session_id": target_sid,
+            "session_name": sess.name,
+            "response": response,
+        }
+    except Exception as e:
+        logger.error(f"send_to_session failed: {e}")
+        return {"error": f"Failed to send to session: {e}"}
+
+async def manage_session(content: str, session_id: Optional[str] = None, owner: Optional[str] = None) -> Dict:
+    """Manage sessions: rename, archive, delete, important, truncate, fork.
+
+    Content format:
+      Line 1: action (rename|archive|unarchive|delete|important|unimportant|truncate|fork)
+      Line 2: target session_id (or "current" to use the active session)
+      Line 3+: action-specific params (e.g. new name for rename, keep_count for truncate)
+    """
+    _session_manager = get_session_manager()
+    if not _session_manager:
+        return {"error": "Session manager not available"}
+
+    from src.database import SessionLocal, Session as DbSession
+
+    # Accept BOTH the structured JSON args the tool schema advertises
+    # ({action, session_id, value}) AND the legacy line-based format
+    # (line1=action, line2=session_id, line3=value). Native function-calling
+    # models send JSON; fenced-block callers send lines. Previously only the
+    # line format was parsed, so a model that followed the schema (JSON) got
+    # "Need at least 2 lines" / "Rename needs line 3" and couldn't drive it.
+    _raw = (content or "").strip()
+    action = ""
+    target_sid = ""
+    value = None      # the action param: new name (rename) / keep_count (truncate, fork)
+    _list_filter = ""
+    _parsed = None
+    if _raw.startswith("{"):
+        try:
+            _parsed = json.loads(_raw)
+        except Exception:
+            _parsed = None
+    if isinstance(_parsed, dict):
+        action = str(_parsed.get("action") or "").strip().lower()
+        target_sid = str(_parsed.get("session_id") or _parsed.get("session") or _parsed.get("id") or "").strip()
+        _v = _parsed.get("value")
+        if _v is None:
+            _v = (_parsed.get("name") or _parsed.get("new_name")
+                  or _parsed.get("title") or _parsed.get("keep_count"))
+        value = None if _v is None else str(_v).strip()
+        _list_filter = str(_parsed.get("filter") or "").strip()
+    else:
+        lines = _raw.split("\n")
+        if not lines or not lines[0].strip():
+            return {"error": "Missing action (rename|archive|delete|important|truncate|fork|list|switch)"}
+        action = lines[0].strip().lower()
+        target_sid = lines[1].strip() if len(lines) >= 2 else ""
+        value = lines[2].strip() if len(lines) >= 3 else None
+        _list_filter = "\n".join(lines[1:]).strip()
+
+    if not action:
+        return {"error": "Missing action (rename|archive|delete|important|truncate|fork|list|switch)"}
+
+    # `list` alias - dispatch to list_sessions so the agent's natural
+    # first guess (every other manage_* tool has a `list` action) works.
+    if action == "list":
+        return await list_sessions(_list_filter, session_id, owner=owner)
+
+    if not target_sid:
+        return {"error": "Need a session_id (or 'current' for the active chat)"}
+
+    # Allow "current" to refer to the active session
+    if target_sid.lower() == "current" and session_id:
+        target_sid = session_id
+
+    # `switch` / `open` / `select` / `view` - the agent reaches for
+    # these when the user asks to "open" or "switch to" a session.
+    # There's no server-side way to make the browser navigate, so we
+    # just return a clickable anchor link the user can click. The
+    # frontend's chat-history click delegate routes `#session-<id>`
+    # to selectSession(). The agent's reply naturally embeds this
+    # result so the user sees a single clickable line.
+    def _session_query(db):
+        query = db.query(DbSession).filter(DbSession.id == target_sid)
+        if owner is not None:
+            query = query.filter(DbSession.owner == owner)
+        return query
+
+    if action in ("switch", "open", "select", "view"):
+        db = SessionLocal()
+        try:
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            name = db_sess.name or target_sid
+        finally:
+            db.close()
+        return {
+            "action": action,
+            "session_id": target_sid,
+            "name": name,
+            "results": f"[{name}](#session-{target_sid}) - click to open.",
+        }
+
+    db = SessionLocal()
+    try:
+        if action == "rename":
+            if not value:
+                return {"error": "rename needs a new name (the `value` arg, or line 3 in the legacy format)"}
+            new_name = value
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            db_sess.name = new_name
+            db.commit()
+            _session_manager.update_session_name(target_sid, new_name)
+            return {"action": "rename", "session_id": target_sid, "name": new_name,
+                    "results": f"Session renamed to '{new_name}'"}
+
+        elif action == "archive":
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            db_sess.archived = True
+            db.commit()
+            return {"action": "archive", "session_id": target_sid,
+                    "results": f"Session '{db_sess.name}' archived"}
+
+        elif action == "unarchive":
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            db_sess.archived = False
+            db.commit()
+            return {"action": "unarchive", "session_id": target_sid,
+                    "results": f"Session '{db_sess.name}' unarchived"}
+
+        elif action == "delete":
+            if target_sid == session_id:
+                return {"error": "Cannot delete the current session while chatting in it. Delete other sessions first."}
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Refusing to delete an unknown chat id; use the exact id from list_sessions."}
+            if db_sess and db_sess.is_important:
+                return {"error": f"Session '{db_sess.name}' is starred/favorited. Unstar it first before deleting."}
+            try:
+                ok = _session_manager.delete_session(target_sid)
+                if not ok:
+                    return {"error": f"Session '{target_sid}' was not deleted because it no longer exists."}
+                return {"action": "delete", "session_id": target_sid,
+                        "results": f"Session '{db_sess.name or target_sid}' deleted"}
+            except Exception as e:
+                return {"error": f"Failed to delete session: {e}"}
+
+        elif action in ("important", "unimportant"):
+            is_important = action == "important"
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            # Prevent AI from unstarring sessions - only the user can do that manually
+            if not is_important and db_sess.is_important:
+                return {"error": f"Session '{db_sess.name}' is starred by the user. Only the user can unstar sessions manually."}
+            db_sess.is_important = is_important
+            db.commit()
+            status = "marked as important" if is_important else "unmarked as important"
+            return {"action": action, "session_id": target_sid,
+                    "results": f"Session '{db_sess.name}' {status}"}
+
+        elif action == "truncate":
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            keep_count = 10
+            if value:
+                try:
+                    keep_count = int(value)
+                except ValueError:
+                    pass
+            success = _session_manager.truncate_messages(target_sid, keep_count)
+            if success:
+                return {"action": "truncate", "session_id": target_sid,
+                        "results": f"Session truncated to last {keep_count} messages"}
+            return {"error": f"Failed to truncate session '{target_sid}'"}
+
+        elif action == "fork":
+            db_sess = _session_query(db).first()
+            if not db_sess:
+                return {"error": f"Session '{target_sid}' not found. Use list_sessions and pass the exact id it returned."}
+            keep_count = 0  # 0 = all messages
+            if value:
+                try:
+                    keep_count = int(value)
+                except ValueError:
+                    pass
+
+            source = _session_manager.get_session(target_sid)
+            if not source:
+                return {"error": f"Session '{target_sid}' not found"}
+
+            new_sid = str(uuid.uuid4())[:8]
+            _session_manager.create_session(
+                session_id=new_sid,
+                name=f"Fork: {source.name}",
+                endpoint_url=source.endpoint_url,
+                model=source.model,
+                rag=False,
+                owner=owner,
+            )
+            # Copy messages
+            history = source.get_context_messages()
+            if keep_count > 0:
+                history = history[:keep_count]
+            from core.models import ChatMessage as InMemoryMsg
+            new_sess = _session_manager.get_session(new_sid)
+            for msg in history:
+                new_sess.add_message(InMemoryMsg(msg["role"], msg["content"]))
+            try:
+                from src.event_bus import fire_event
+                fire_event("session_created", owner)
+            except Exception:
+                logger.debug("session_created event dispatch failed", exc_info=True)
+
+            return {"action": "fork", "session_id": new_sid,
+                    "source_session": target_sid, "messages_copied": len(history),
+                    "results": f"Forked session '{source.name}' -> new session {new_sid} ({len(history)} messages)"}
+
+        else:
+            return {"error": f"Unknown action '{action}'. Use: list, switch, rename, archive, unarchive, delete, important, unimportant, truncate, fork"}
+    except Exception as e:
+        logger.error(f"manage_session failed: {e}")
+        return {"error": str(e)}
+    finally:
+        db.close()
+
+
+# ---------------------------------------------------------------------------
+# Handler classes registered in TOOL_HANDLERS
+# ---------------------------------------------------------------------------
+
+class CreateSessionTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await create_session(content, ctx.get("session_id"), owner=ctx.get("owner"))
+
+
+class ListSessionsTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await list_sessions(content, ctx.get("session_id"), owner=ctx.get("owner"))
+
+
+class SendToSessionTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await send_to_session(content, ctx.get("session_id"), owner=ctx.get("owner"))
+
+
+class ManageSessionTool:
+    async def execute(self, content: str, ctx: dict) -> Dict:
+        return await manage_session(content, ctx.get("session_id"), owner=ctx.get("owner"))

src/agent_tools/web_tools.py

@@ -79,13 +79,23 @@ class WebSearchTool:
 class WebFetchTool:
     async def execute(self, content: str, ctx: dict) -> dict:
         from src.search.content import fetch_webpage_content
+        from src.constants import WEB_FETCH_HARD_MAX_BYTES
         raw = content.strip()
         url = ""
+        max_bytes = None
         if raw.startswith("{"):
             try:
                 parsed = json.loads(raw)
                 if isinstance(parsed, dict):
                     url = str(parsed.get("url") or "").strip()
+                    # Download-budget override (#3812): "full": true raises the
+                    # budget to the hard cap; an explicit max_bytes is clamped
+                    # to the hard cap downstream. Default stays the soft cap.
+                    if parsed.get("full") is True:
+                        max_bytes = WEB_FETCH_HARD_MAX_BYTES
+                    mb = parsed.get("max_bytes")
+                    if isinstance(mb, int) and mb > 0:
+                        max_bytes = mb
             except json.JSONDecodeError:
                 url = ""
         if not url:
@@ -100,7 +110,7 @@ class WebFetchTool:
         loop = asyncio.get_running_loop()
         try:
             result = await asyncio.wait_for(
-                loop.run_in_executor(None, lambda: fetch_webpage_content(url, timeout=10)),
+                loop.run_in_executor(None, lambda: fetch_webpage_content(url, timeout=10, max_bytes=max_bytes)),
                 timeout=30,
             )
         except asyncio.TimeoutError:
@@ -116,8 +126,28 @@ class WebFetchTool:
                 return {"error": f"web_fetch: {url}: {err}", "exit_code": 1}
             return {"error": f"web_fetch: {url}: no readable text content (not HTML, or the page needs JS/login)", "exit_code": 1}
 
+        # Tell the model when the download budget cut the body short and how
+        # to get the rest, instead of silently presenting a partial page as
+        # the whole thing.
+        size_note = ""
+        if result.get("truncated"):
+            fetched = result.get("fetched_bytes") or 0
+            total = result.get("total_bytes")
+            total_txt = f" of {total:,} bytes" if total else ""
+            size_note = (
+                f"[partial content: download stopped at {fetched:,} bytes{total_txt}. "
+                f'Re-call with {{"url": "{url}", "full": true}} to fetch up to '
+                f"{WEB_FETCH_HARD_MAX_BYTES:,} bytes.]\n\n"
+            )
+
+        # The notice must lead the output so the MAX_OUTPUT_CHARS trim below can
+        # never drop it. The title is untrusted, uncapped page content, so a
+        # giant title ahead of the notice could push it out of range; keep the
+        # notice first and cap the title as a second guard.
+        if len(title) > 300:
+            title = title[:300] + "..."
         header = (f"# {title}\n" if title else "") + f"Source: {url}\n\n"
-        output = header + text
+        output = size_note + header + text
         if len(output) > MAX_OUTPUT_CHARS:
             output = output[:MAX_OUTPUT_CHARS] + "\n\n[...truncated]"
         return {"output": output, "exit_code": 0}