tools.py

"""The 7 tools + their JSON-schema definitions, in one file.

State that must survive across calls (paths read this session) lives at module
level. Every tool is a plain function returning a string; TOOL_FUNCS maps a name
to its function and TOOLS is the JSON-schema list handed to the model.
"""
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

# Files read this session — write_file refuses to clobber a file not read here.
READ_PATHS: set[str] = set()

MAX_OUTPUT = 30_000        # tool-result truncation ceiling (chars)
READ_LINE_CAP = 2000       # read_file line ceiling
GREP_LINE_CAP = 400        # grep output line ceiling

# --- spill-to-file (Experiment 1) --------------------------------------------
# Big tool results are 90% of context and re-sent every turn. Instead of keeping
# the whole dump in the message (fluff) or truncating it (missing), keep a tight
# PREVIEW in-context and spill the FULL result to a scratch file, returning a
# handle. The handle is a short, stable, append-only token (cache-friendly) and
# the body stays re-fetchable via read_file/grep (anti-missing). Gate: PIT_SPILL=0
# disables it (for the baseline arm of the experiment).
SPILL_CHARS = 4000                         # preview budget kept in-context
SPILL_DIR = Path(__file__).parent / ".scratch"


def spill_result(name: str, result: str) -> str:
    if (os.environ.get("PIT_SPILL") == "0" or not result
            or result.startswith("[error]") or len(result) <= SPILL_CHARS):
        return result
    try:
        SPILL_DIR.mkdir(exist_ok=True)
        n = len(list(SPILL_DIR.glob("*.txt"))) + 1
        path = SPILL_DIR / f"{name}_{n:03d}.txt"
        path.write_text(result, encoding="utf-8")
    except Exception:                        # noqa: BLE001
        return result                        # spill failed → keep full result
    preview = result[:SPILL_CHARS]
    return (preview + f"\n\n[+{len(result) - SPILL_CHARS:,} more chars elided. "
            f"FULL output saved to {path} — read_file or grep it for the rest.]")

# --- shells ------------------------------------------------------------------
_GITBASH = r"C:\Program Files\Git\bin\bash.exe"
BASH_EXE = _GITBASH if os.path.exists(_GITBASH) else "bash"

# --- deletion hard-block -----------------------------------------------------
# Match deletion verbs only in COMMAND POSITION (start of line, or right after a
# shell separator ; & | ` ( ), so "grep rm" or "warm" never trip it. Plus a few
# multi-word patterns (git clean / reset --hard, DROP TABLE) matched anywhere.
_DELETE_RE = re.compile(
    r"(?:^|[\n;&|`(])\s*(?:sudo\s+)?"
    r"(?:rm|rmdir|rd|del|remove-item|rimraf|mkfs\S*|format)\b"
    r"|\bgit\s+clean\b"
    r"|\bgit\s+reset\s+--hard\b"
    r"|\bdrop\s+table\b",
    re.IGNORECASE,
)
_DELETE_BLOCKED = (
    "[error] deletion blocked by harness policy — move to a junk/ folder or ask "
    "the user to delete manually."
)


def _truncate(s: str) -> str:
    if len(s) > MAX_OUTPUT:
        return s[:MAX_OUTPUT] + f"\n...[truncated {len(s) - MAX_OUTPUT} chars]"
    return s


# Runtime deletion guard: every spawned process gets guard/ on PYTHONPATH, so
# any Python it launches auto-imports sitecustomize.py, which blocks deletion
# APIs process-wide. Closes the "write a script with os.remove, then run it"
# hole that the command-string regexes can't see.
_GUARD_DIR = str(Path(__file__).parent / "guard")


def _run_shell(argv: list[str], timeout: int) -> str:
    timeout = max(1, min(int(timeout), 600))
    env = {**os.environ,
           "PYTHONPATH": _GUARD_DIR + os.pathsep + os.environ.get("PYTHONPATH", "")}
    try:
        p = subprocess.run(
            argv, capture_output=True, timeout=timeout,
            text=True, encoding="utf-8", errors="replace", env=env,
        )
    except subprocess.TimeoutExpired:
        return f"[error] command timed out after {timeout}s"
    except Exception as e:                       # noqa: BLE001
        return f"[error] failed to run: {e}"
    out = (p.stdout or "") + (p.stderr or "")
    out = out.strip() or "(no output)"
    if p.returncode != 0:
        out = f"[exit {p.returncode}]\n{out}"
    return _truncate(out)


# --- tools -------------------------------------------------------------------
def bash(command: str, timeout: int = 120) -> str:
    """Run a command in Git Bash."""
    if _DELETE_RE.search(command):
        return _DELETE_BLOCKED
    return _run_shell([BASH_EXE, "-lc", command], timeout)


def powershell(command: str, timeout: int = 120) -> str:
    """Run a command in Windows PowerShell (no profile)."""
    if _DELETE_RE.search(command):
        return _DELETE_BLOCKED
    return _run_shell(
        ["powershell.exe", "-NoProfile", "-Command", command], timeout
    )


def read_file(path: str, offset: int = 1, limit: int = READ_LINE_CAP) -> str:
    """Read a text file, returning `lineno\ttext` lines (1-based offset)."""
    p = Path(path)
    if not p.exists():
        return f"[error] file not found: {path}"
    if not p.is_file():
        return f"[error] not a file: {path}"
    try:
        text = p.read_text(encoding="utf-8", errors="replace")
    except Exception as e:                       # noqa: BLE001
        return f"[error] could not read {path}: {e}"
    READ_PATHS.add(str(p.resolve()).lower())
    lines = text.splitlines()
    start = max(1, int(offset))
    limit = max(1, min(int(limit), READ_LINE_CAP))
    chunk = lines[start - 1: start - 1 + limit]
    if not chunk:
        return f"(no lines: file has {len(lines)} lines, offset {start})"
    numbered = "\n".join(f"{start + i}\t{ln}" for i, ln in enumerate(chunk))
    tail = ""
    end = start - 1 + len(chunk)
    if end < len(lines):
        tail = f"\n...[{len(lines) - end} more lines; use offset={end + 1}]"
    return _truncate(numbered + tail)


_SHELL_SCRIPT_EXTS = {".sh", ".ps1", ".psm1", ".bat", ".cmd"}
_SCRIPT_BLOCKED = (
    "[error] blocked: this script contains file-deletion code (pit policy: "
    "never delete). Rewrite it to move files into a junk/ folder instead."
)


def _script_has_deletion(path: str, content: str) -> bool:
    """Deletion code is refused in executable scripts at write time — the
    runtime guard would stop it anyway, but failing early reads better."""
    ext = Path(path).suffix.lower()
    if ext == ".py":
        return bool(_PY_BLOCK_RE.search(content))
    if ext in _SHELL_SCRIPT_EXTS:
        return bool(_DELETE_RE.search(content))
    return False


def write_file(path: str, content: str) -> str:
    """Create or overwrite a file. Refuses to overwrite a file not read this
    session (guards against blind clobbers)."""
    p = Path(path)
    key = str(p.resolve()).lower()
    if p.exists() and key not in READ_PATHS:
        return (f"[error] {path} already exists and was not read this session. "
                "read_file it first to confirm before overwriting.")
    if _script_has_deletion(path, content):
        return _SCRIPT_BLOCKED
    try:
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
    except Exception as e:                       # noqa: BLE001
        return f"[error] write failed for {path}: {e}"
    READ_PATHS.add(key)                          # now safe to overwrite again
    return f"wrote {path} ({len(content.splitlines())} lines)"


def edit_file(path: str, old_text: str, new_text: str,
              replace_all: bool = False) -> str:
    """Exact-string replace. Fails loudly if old_text is missing or (without
    replace_all) not unique."""
    p = Path(path)
    if not p.is_file():
        return f"[error] file not found: {path}"
    try:
        content = p.read_text(encoding="utf-8")
    except Exception as e:                       # noqa: BLE001
        return f"[error] could not read {path}: {e}"
    count = content.count(old_text)
    if count == 0:
        hint = ""
        if old_text.strip() and old_text.strip() in content:
            hint = " (a whitespace-trimmed version exists — check indentation)"
        return f"[error] old_text not found in {path}{hint}"
    if count > 1 and not replace_all:
        return (f"[error] old_text appears {count} times in {path}; add context "
                "to make it unique or pass replace_all=true")
    new_content = content.replace(old_text, new_text)
    if _script_has_deletion(path, new_content):
        return _SCRIPT_BLOCKED
    try:
        p.write_text(new_content, encoding="utf-8")
    except Exception as e:                       # noqa: BLE001
        return f"[error] write failed for {path}: {e}"
    READ_PATHS.add(str(p.resolve()).lower())
    return f"edited {path} — replaced {count} occurrence(s)"


def grep(pattern: str, path: str = ".", glob: str = "") -> str:
    """Search file contents. Uses ripgrep if available, else `grep -rn`."""
    if shutil.which("rg"):
        argv = ["rg", "-n", "--no-heading", "--color", "never"]
        if glob:
            argv += ["-g", glob]
        argv += [pattern, path]
    else:
        argv = ["grep", "-rn"]
        if glob:
            argv += [f"--include={glob}"]
        argv += ["-e", pattern, path]
    out = _run_shell(argv, 60)
    lines = out.splitlines()
    if len(lines) > GREP_LINE_CAP:
        out = "\n".join(lines[:GREP_LINE_CAP]) + \
              f"\n...[{len(lines) - GREP_LINE_CAP} more matches]"
    return out


def _expand_braces(pattern: str) -> list[str]:
    """Expand a single {a,b,c} alternation (recursively) — pathlib.glob does not."""
    m = re.search(r"\{([^{}]*)\}", pattern)
    if not m:
        return [pattern]
    pre, post = pattern[:m.start()], pattern[m.end():]
    out = []
    for opt in m.group(1).split(","):
        out.extend(_expand_braces(pre + opt + post))
    return out


def glob(pattern: str, path: str = ".") -> str:
    """List paths matching a glob pattern, newest first (by mtime). Supports
    {a,b} brace alternation (e.g. **/*.{py,js})."""
    base = Path(path)
    matches, seen = [], set()
    try:
        for pat in _expand_braces(pattern):
            for p in base.glob(pat):
                if p.is_file() and p not in seen:
                    seen.add(p)
                    matches.append(p)
    except Exception as e:                       # noqa: BLE001
        return f"[error] bad glob: {e}"
    if not matches:
        return "(no matches)"
    matches.sort(key=lambda p: p.stat().st_mtime, reverse=True)
    return _truncate("\n".join(str(p) for p in matches[:500]))


# Python deletion/shell-escape guard for python_eval (bash guard doesn't see it)
_PY_BLOCK_RE = re.compile(
    r"os\.remove|os\.unlink|os\.rmdir|os\.removedirs|shutil\.rmtree|\.unlink\("
    r"|os\.system|subprocess|popen|\bpty\b|drop\s+table",
    re.IGNORECASE,
)


def python_eval(code: str, timeout: int = 120) -> str:
    """Run Python code directly with no shell (argv list) — avoids shell-quoting
    pain for data/sqlite inspection and quick computation. stdout+stderr returned.
    Deletion and shell-escape calls are blocked; use bash for real shell work."""
    if _PY_BLOCK_RE.search(code):
        return ("[error] blocked: python_eval forbids deletion/shell-escape calls "
                "(os.remove, shutil.rmtree, os.system, subprocess...). Use the bash "
                "tool for shell work, or move files to junk/ instead of deleting.")
    return _run_shell([sys.executable or "python", "-c", code], timeout)


# --- web -----------------------------------------------------------------------
def _fmt_hits(query, provider, hits):
    """hits = list of (title, url, snippet)."""
    lines = [f"Search: {query} (via {provider})\n"]
    for i, (title, url, snip) in enumerate(hits, 1):
        lines.append(f"{i}. {title}\n   {url}\n   {snip[:240]}\n")
    return _truncate("\n".join(lines))


def _tavily(query, n, httpx):
    key = os.environ.get("TAVILY_API_KEY")
    if not key:
        return None
    r = httpx.post("https://api.tavily.com/search",
                   json={"api_key": key, "query": query, "max_results": n,
                         "search_depth": "basic", "include_answer": False},
                   timeout=20)
    r.raise_for_status()
    res = r.json().get("results", [])
    return [(it.get("title", "(no title)"), it.get("url", ""),
             (it.get("content") or "").strip().replace("\n", " ")) for it in res]


def _brave(query, n, httpx):
    key = os.environ.get("BRAVE_API_KEY") or os.environ.get("BRAVE_SEARCH_API_KEY")
    if not key:
        return None
    r = httpx.get("https://api.search.brave.com/res/v1/web/search",
                  params={"q": query, "count": n},
                  headers={"X-Subscription-Token": key,
                           "Accept": "application/json"}, timeout=20)
    r.raise_for_status()
    res = (r.json().get("web") or {}).get("results", [])
    return [(it.get("title", "(no title)"), it.get("url", ""),
             (it.get("description") or "").strip()) for it in res]


def _serper(query, n, httpx):
    key = os.environ.get("SERPER_API_KEY")
    if not key:
        return None
    r = httpx.post("https://google.serper.dev/search",
                   json={"q": query, "num": n},
                   headers={"X-API-KEY": key,
                            "Content-Type": "application/json"}, timeout=20)
    r.raise_for_status()
    res = r.json().get("organic", [])
    return [(it.get("title", "(no title)"), it.get("link", ""),
             (it.get("snippet") or "").strip()) for it in res]


# Keyed providers, tried in order; whichever has a key + returns hits wins.
# DuckDuckGo (keyless) is the final fallback. "Use all the free ones."
_SEARCH_PROVIDERS = [("Tavily", _tavily), ("Brave", _brave), ("Serper", _serper)]


def web_search(query: str, max_results: int = 5) -> str:
    """Search the web across whatever free providers are keyed (Tavily / Brave /
    Serper), falling back to a keyless DuckDuckGo scrape. Numbered
    title/url/snippet results."""
    import httpx
    max_results = max(1, min(int(max_results), 10))
    for name, fn in _SEARCH_PROVIDERS:
        try:
            hits = fn(query, max_results, httpx)
        except Exception:                        # noqa: BLE001
            continue                             # provider errored → next one
        if hits:
            return _fmt_hits(query, name, hits[:max_results])
    try:
        r = httpx.get("https://html.duckduckgo.com/html/", params={"q": query},
                      headers={"User-Agent": "Mozilla/5.0"}, timeout=20)
        r.raise_for_status()
    except Exception as e:                       # noqa: BLE001
        return f"[error] web search failed: {e}"
    hits = re.findall(
        r'<a[^>]+class="result__a"[^>]+href="([^"]+)"[^>]*>([^<]+)</a>'
        r'.*?<a[^>]+class="result__snippet"[^>]*>(.*?)</a>',
        r.text, re.DOTALL)[:max_results]
    if not hits:
        return f"no results for '{query}'"
    import html as _html
    from urllib.parse import parse_qs, unquote, urlparse
    parsed = []
    for url, title, snip in hits:
        if "uddg=" in url:                       # unwrap DDG redirect links
            q = parse_qs(urlparse(_html.unescape(url)).query)
            url = unquote(q.get("uddg", [url])[0])
        snip = _html.unescape(re.sub(r"<[^>]+>", "", snip)).strip()
        parsed.append((_html.unescape(title), url, snip))
    return _fmt_hits(query, "DuckDuckGo, free", parsed)


def web_fetch(url: str, max_chars: int = 8000) -> str:
    """Fetch a URL and return its main text content (tags stripped;
    trafilatura-extracted if available)."""
    import httpx
    if not url.startswith(("http://", "https://")):
        return f"[error] URL must start with http:// or https://: {url}"
    try:
        r = httpx.get(url, timeout=20, follow_redirects=True,
                      headers={"User-Agent": "Mozilla/5.0 pit-agent"})
        r.raise_for_status()
        html = r.text
    except Exception as e:                       # noqa: BLE001
        return f"[error] could not fetch {url}: {e}"
    try:
        import trafilatura
        text = trafilatura.extract(html, include_comments=False,
                                   include_tables=True) or ""
    except ImportError:
        text = re.sub(r"<(script|style)[^>]*>.*?</\1>", "", html,
                      flags=re.DOTALL | re.IGNORECASE)
        text = re.sub(r"<[^>]+>", " ", text)
        text = re.sub(r"\s+", " ", text).strip()
    if not text.strip():
        return f"[warning] no text content extracted from {url}"
    max_chars = max(500, min(int(max_chars), MAX_OUTPUT))
    if len(text) > max_chars:
        text = text[:max_chars] + \
            f"\n\n[truncated — {len(text)} chars total, showing first {max_chars}]"
    return f"# {url}\n\n{text}"


def consult_oracle(question: str) -> str:
    """Escalate ONE hard question to a much smarter cloud model. Imported lazily
    so pit runs fine with no API key / no httpx-to-cloud."""
    import cloud
    return cloud.consult(question)


def delegate(agent: str, prompt: str, cwd: str = ".",
             yolo: bool = False, timeout: int = 600) -> str:
    """Hand a self-contained task to another CLI agent and capture its output.
    Imported lazily. See delegate.py."""
    import delegate as _d
    return _d.delegate(agent, prompt, cwd=cwd, yolo=yolo, timeout=timeout)


# --- kiwix (offline ZIM library) ----------------------------------------------
# Search/read the local Kiwix library (Wikipedia + reference wikis) — the
# offline stand-in for web_search/web_fetch. Reuses the kiwix-serve exe, port,
# and library.xml from ~/kiwix-rag/config.json (that project is the source of
# truth). OFF by default (needs kiwix-serve running); enable via --tools
# offline / /tools +kiwix — enabling auto-starts the server if it's down.
_KIWIX_CFG_PATH = Path.home() / "kiwix-rag" / "config.json"


def _kiwix_cfg() -> dict:
    try:
        import json
        cfg = json.loads(_KIWIX_CFG_PATH.read_text(encoding="utf-8"))
        return {"port": int(cfg.get("kiwix_port", 8099)),
                "exe": cfg.get("kiwix_serve_exe", ""),
                "library": cfg.get("library_path", "")}
    except Exception:                            # noqa: BLE001
        return {"port": 8099, "exe": "", "library": ""}


def _port_open(port: int) -> bool:
    import socket
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(0.4)
        return s.connect_ex(("127.0.0.1", port)) == 0


def ensure_kiwix(echo: bool = True) -> bool:
    """Reuse a running kiwix-serve, else spawn one from the kiwix-rag config."""
    import time
    cfg = _kiwix_cfg()
    if _port_open(cfg["port"]):
        return True
    if not (cfg["exe"] and os.path.exists(cfg["exe"])
            and os.path.exists(cfg["library"])):
        if echo:
            print("[kiwix] can't auto-start — exe or library.xml missing "
                  f"(expected via {_KIWIX_CFG_PATH}; run kiwix-rag once to build it)")
        return False
    subprocess.Popen(
        [cfg["exe"], "--port", str(cfg["port"]), "--address", "127.0.0.1",
         "--library", cfg["library"]],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    for _ in range(30):
        if _port_open(cfg["port"]):
            if echo:
                print(f"[kiwix] kiwix-serve up on :{cfg['port']}")
            return True
        time.sleep(0.5)
    if echo:
        print("[kiwix] server did not come up within 15s")
    return False


# HTML scraping regexes lifted from kiwix-rag/server.py (proven there).
_KX_LI = re.compile(r"<li>\s*<a href=\"([^\"]+)\">(.*?)</a>(.*?)</li>", re.S)
_KX_CITE = re.compile(r"<cite>(.*?)</cite>", re.S)
_KX_BOOK = re.compile(r'<div class="book-title">\s*from\s*(.*?)</div>', re.S)
_KX_P = re.compile(r"<p\b[^>]*>(.*?)</p>", re.S)
_KX_TAG = re.compile(r"<[^>]+>")
_KIWIX_DOWN = ("[error] kiwix-serve is not running and could not be started — "
               "check ~/kiwix-rag (bin + library.xml).")


def _kx_text(s: str) -> str:
    import html as _html
    return re.sub(r"\s+", " ", _html.unescape(_KX_TAG.sub(" ", s))).strip()


def kiwix_search(query: str, max_results: int = 6) -> str:
    import httpx
    cfg = _kiwix_cfg()
    if not _port_open(cfg["port"]) and not ensure_kiwix(echo=False):
        return _KIWIX_DOWN
    try:
        r = httpx.get(f"http://127.0.0.1:{cfg['port']}/search",
                      params={"pattern": query,
                              "pageLength": max(1, min(int(max_results), 20))},
                      timeout=20)
    except Exception as e:                       # noqa: BLE001
        return f"[error] kiwix search failed: {e}"
    out = []
    for href, title_raw, rest in _KX_LI.findall(r.text):
        cite = _KX_CITE.search(rest)
        book = _KX_BOOK.search(rest)
        bk = _kx_text(book.group(1)) if book else ""
        snip = _kx_text(cite.group(1)) if cite else ""
        out.append(f"{len(out)+1}. {_kx_text(title_raw)}"
                   + (f"  [{bk}]" if bk else "")
                   + (f"\n   {snip}" if snip else "")
                   + f"\n   href: {href}")
        if len(out) >= int(max_results):
            break
    if not out:
        return ("no hits. Kiwix full-text search is strict AND — retry with "
                "FEWER, more distinctive keywords (1-3 words).")
    return "\n".join(out)


def kiwix_read(href: str, max_chars: int = 8000) -> str:
    import httpx
    cfg = _kiwix_cfg()
    if not _port_open(cfg["port"]) and not ensure_kiwix(echo=False):
        return _KIWIX_DOWN
    if not href.startswith("/"):
        href = "/" + href
    try:
        r = httpx.get(f"http://127.0.0.1:{cfg['port']}{href}", timeout=20,
                      follow_redirects=True)
    except Exception as e:                       # noqa: BLE001
        return f"[error] kiwix fetch failed: {e}"
    paras = _KX_P.findall(r.text)
    text = "\n\n".join(t for p in paras if (t := _kx_text(p)))
    if len(text) < 160:                          # non-article page fallback
        body = re.sub(r"<(script|style)\b.*?</\1>", " ", r.text, flags=re.S)
        text = _kx_text(body)
    max_chars = max(500, min(int(max_chars), MAX_OUTPUT))
    if len(text) > max_chars:
        text = text[:max_chars] + \
            f"\n\n[truncated — {len(text)} chars total, showing first {max_chars}]"
    return f"# {href}\n\n{text}"


TOOL_FUNCS = {
    "bash": bash,
    "powershell": powershell,
    "python_eval": python_eval,
    "read_file": read_file,
    "write_file": write_file,
    "edit_file": edit_file,
    "grep": grep,
    "glob": glob,
    "web_search": web_search,
    "web_fetch": web_fetch,
    "consult_oracle": consult_oracle,
    "delegate": delegate,
    "kiwix_search": kiwix_search,
    "kiwix_read": kiwix_read,
}


def _fn(name, desc, props, required):
    return {"type": "function", "function": {
        "name": name, "description": desc,
        "parameters": {"type": "object", "properties": props,
                       "required": required}}}


_STR = {"type": "string"}
_INT = {"type": "integer"}
TOOLS = [
    _fn("bash", "Run a shell command in Git Bash on Windows. Deletion commands "
        "are blocked. Returns combined stdout+stderr.",
        {"command": _STR, "timeout": {**_INT, "description": "seconds, max 600"}},
        ["command"]),
    _fn("powershell", "Run a command in Windows PowerShell (-NoProfile). "
        "Deletion commands are blocked.",
        {"command": _STR, "timeout": {**_INT, "description": "seconds, max 600"}},
        ["command"]),
    _fn("python_eval", "Run Python code with NO shell (so no quoting headaches). "
        "Preferred for inspecting SQLite/JSON/CSV, quick math, and data checks. "
        "Returns stdout+stderr. Deletion/subprocess calls are blocked.",
        {"code": _STR, "timeout": {**_INT, "description": "seconds, max 600"}},
        ["code"]),
    _fn("read_file", "Read a text file. Returns numbered lines. Read a file "
        "before editing or overwriting it.",
        {"path": _STR,
         "offset": {**_INT, "description": "1-based first line (default 1)"},
         "limit": {**_INT, "description": "max lines (default 2000)"}},
        ["path"]),
    _fn("write_file", "Create a new file or overwrite one you have read this "
        "session. Refuses to clobber an unread existing file.",
        {"path": _STR, "content": _STR}, ["path", "content"]),
    _fn("edit_file", "Replace an exact substring in a file. Fails if old_text "
        "is absent, or non-unique unless replace_all is set.",
        {"path": _STR, "old_text": _STR, "new_text": _STR,
         "replace_all": {"type": "boolean"}},
        ["path", "old_text", "new_text"]),
    _fn("grep", "Search file contents by regex (ripgrep/grep). Optional path "
        "and glob filter.",
        {"pattern": _STR, "path": _STR, "glob": _STR}, ["pattern"]),
    _fn("glob", "List files matching a glob pattern (e.g. '**/*.py'), newest "
        "first.",
        {"pattern": _STR, "path": _STR}, ["pattern"]),
    _fn("web_search", "Search the web. Returns titles, urls, and snippets. Use "
        "for current events, fresh facts, docs, or anything needing up-to-date "
        "info you don't have.",
        {"query": _STR,
         "max_results": {**_INT, "description": "how many results (default 5)"}},
        ["query"]),
    _fn("web_fetch", "Fetch a web page and return its main text content (not "
        "raw HTML). Use after web_search to read a result, or to read a URL "
        "the user gave.",
        {"url": _STR,
         "max_chars": {**_INT, "description": "truncate to this (default 8000)"}},
        ["url"]),
    _fn("consult_oracle", "Ask a much smarter cloud model ONE well-formed "
        "question. EXPENSIVE — use only when genuinely stuck: you tried 2+ "
        "different approaches and failed, or the task needs deep reasoning/design "
        "judgment beyond you, or the user explicitly asked. Include ALL relevant "
        "code, errors, and context in the question — the oracle cannot see your "
        "session or files. Max 3 calls per session.",
        {"question": _STR}, ["question"]),
    _fn("recall", "Search what YOU already found earlier this session (your prior "
        "conclusions and tool results) by keyword. Use this INSTEAD of re-running "
        "a grep/read/search you've already done — if you looked something up a few "
        "turns ago, recall it rather than re-fetching it.",
        {"query": {**_STR, "description": "keywords for what you're trying to remember"}},
        ["query"]),
    _fn("delegate", "Hand a self-contained task to another CLI coding agent "
        "(claude, codex, or gemini) and get its output back — use to parallelize "
        "work, get a second implementation/opinion, or offload a chunk you've "
        "scoped. Give a COMPLETE standalone prompt: the agent starts fresh in "
        "`cwd` and sees none of this session. yolo=false (default) runs it in a "
        "safe read-only/plan mode; yolo=true lets it actually edit and run "
        "commands. NOTE: a delegated agent runs as its own process with its own "
        "permissions — pit's deletion guard does NOT protect its actions, so "
        "only use yolo=true when you trust the task.",
        {"agent": {**_STR, "description": "claude | codex | gemini"},
         "prompt": _STR,
         "cwd": {**_STR, "description": "working dir for the agent (default '.')"},
         "yolo": {"type": "boolean",
                  "description": "true = full auto edit/run; false = safe/read-only"},
         "timeout": {**_INT, "description": "seconds, max 1800 (default 600)"}},
        ["agent", "prompt"]),
    _fn("kiwix_search", "Search the OFFLINE Kiwix library (Wikipedia + reference "
        "wikis) for facts — no internet needed. When web tools are unavailable, "
        "this is your ONLY source of external information. Search is strict AND: "
        "use 1-3 distinctive keywords, not full questions. Returns titles, "
        "snippets, and hrefs for kiwix_read.",
        {"query": {**_STR, "description": "1-3 distinctive keywords"},
         "max_results": {**_INT, "description": "how many results (default 6)"}},
        ["query"]),
    _fn("kiwix_read", "Read an article from the offline Kiwix library. Pass an "
        "href from kiwix_search results (e.g. /content/<book>/<path>). Returns "
        "clean paragraph text.",
        {"href": _STR,
         "max_chars": {**_INT, "description": "truncate to this (default 8000)"}},
        ["href"]),
]

# recall is handled in loop.py (needs session history), so it's schema-only here.
# Gate it for the Exp 3 A/B: PIT_RECALL=0 removes it (baseline arm).
if os.environ.get("PIT_RECALL") == "0":
    TOOLS = [t for t in TOOLS if t["function"]["name"] != "recall"]


# --- tool profiles ------------------------------------------------------------
# Manually assign the toolset BEFORE a run (--tools <spec>) or at runtime
# (/tools <spec>). CACHE NOTE: the tools list is templated into the prompt
# prefix, so changing it MID-session invalidates the KV cache (one full
# re-prefill next turn) — same physics that killed in-place pruning (Exp 0).
# Assigning at launch is free.
_LOCAL = ["bash", "powershell", "python_eval", "read_file", "write_file",
          "edit_file", "grep", "glob", "recall"]
_NET = ["web_search", "web_fetch", "consult_oracle", "delegate"]
_KIWIX = ["kiwix_search", "kiwix_read"]
PROFILES = {
    "all":        _LOCAL + _NET,            # default — kiwix off (needs its server)
    "offline":    _LOCAL + _KIWIX,          # zero internet; kiwix = the info source
    "local":      _LOCAL,                   # pure local, no extra servers
    "everything": _LOCAL + _NET + _KIWIX,
}
ACTIVE: set[str] = set(PROFILES["all"])
ACTIVE_SPEC = "all"                          # for display


def active_tools() -> list:
    """The schema list actually handed to the model. Preserves TOOLS order, so
    a stable ACTIVE set = a stable, cache-friendly prefix."""
    return [t for t in TOOLS if t["function"]["name"] in ACTIVE]


def set_active(spec: str) -> str:
    """Set the active toolset. spec = profile name | explicit 'name,name' list
    | '+name,-name' modifiers (relative to current). Unknown names are reported,
    not silently dropped. Returns a status message."""
    global ACTIVE_SPEC
    all_names = {t["function"]["name"] for t in TOOLS}
    spec = spec.strip()
    if spec in PROFILES:
        ACTIVE.clear()
        ACTIVE.update(PROFILES[spec])
        ACTIVE_SPEC = spec
    else:
        parts = [p.strip() for p in spec.split(",") if p.strip()]
        unknown = [p.lstrip("+-") for p in parts
                   if p.lstrip("+-") not in all_names]
        if unknown:
            return (f"[tools] unknown: {', '.join(unknown)} — profiles: "
                    f"{', '.join(PROFILES)}; tools: {', '.join(sorted(all_names))}")
        if any(p[0] in "+-" for p in parts):     # modifier form
            for p in parts:
                (ACTIVE.discard if p[0] == "-" else ACTIVE.add)(p.lstrip("+-"))
        else:                                    # explicit list form
            ACTIVE.clear()
            ACTIVE.update(parts)
        ACTIVE_SPEC = "custom"
    if ACTIVE & set(_KIWIX):                     # enabling kiwix boots its server
        ensure_kiwix()
    return (f"[tools = {ACTIVE_SPEC}] "
            + ", ".join(t["function"]["name"] for t in active_tools()))

# SEAM: spawn(prompts[]) (parallel subagents) registers here later —
# add its TOOL_FUNCS entry + TOOLS schema; the loop needs no changes.