NexusOS/synapse/tools.py

"""Tools a playbook can call during chat.

Ollama drives the calling: `/api/chat` with a `tools` param returns
`message.tool_calls`, and this module is just the registry + dispatch.

Most tools READ local state (memory, history, documents, models). A few act:
`web_search`/`fetch_url` make outbound HTTP requests, and `remember` WRITES a
memory fact. The per-playbook allowlist (`PlaybookItem.tools`) is the security
boundary — an action tool only fires when a playbook explicitly lists it.
"""
from __future__ import annotations

import json
from typing import Awaitable, Callable

from .memory.store import store, MemoryItem
from .ollama_manager import get_ollama_manager


async def _search_memory(query: str = "", **_) -> str:
    q = (query or "").strip().lower()
    hits = [
        {"section": it.section, "text": it.text}
        for it in store.all()
        if not q
        or q in it.text.lower()
        or q in (it.section or "").lower()
        or any(q in t.lower() for t in it.tags)
    ]
    return json.dumps(hits[:20])


async def _search_history(query: str = "", **_) -> str:
    # Hybrid recall: semantic (embeddings) unioned with lexical, falls back to
    # lexical if embeddings are down. Same retrieval the chat endpoint uses.
    convs = await store.semantic_search_conversations(
        query or "", get_ollama_manager().embed, limit=3
    )
    return json.dumps([{"matches": c.get("matches", [])} for c in convs])


async def _list_models(**_) -> str:
    return json.dumps(await get_ollama_manager().list_models())


async def _search_documents(query: str = "", **_) -> str:
    hits = await store.search_documents(query or "", get_ollama_manager().embed, limit=3)
    return json.dumps([{"title": h["title"], "text": h["text"]} for h in hits])


async def _get_time(**_) -> str:
    from datetime import datetime
    return json.dumps({"now": datetime.now().isoformat(timespec="seconds")})


async def _web_search(query: str = "", **_) -> str:
    import asyncio as _a
    from .search import web_search
    res = await _a.to_thread(web_search, query or "", 4)
    return res or "(no results)"


_FETCH_MAX_REDIRECTS = 5


def _ip_is_blocked(ip: str) -> bool:
    """True if an address is one an outbound fetch has no business reaching:
    loopback, RFC1918/ULA private, link-local (incl. 169.254.169.254 cloud
    metadata), multicast, reserved, or unspecified. IPv4-mapped IPv6 is unwrapped
    first so ::ffff:127.0.0.1 can't sneak a loopback past the check."""
    import ipaddress
    try:
        addr = ipaddress.ip_address(ip.split("%")[0])  # drop any IPv6 zone id
    except ValueError:
        return True  # unparseable -> refuse rather than guess
    mapped = getattr(addr, "ipv4_mapped", None)
    if mapped is not None:
        addr = mapped
    return (
        addr.is_loopback or addr.is_private or addr.is_link_local
        or addr.is_multicast or addr.is_reserved or addr.is_unspecified
    )


def _ssrf_guard(host: str) -> str | None:
    """Resolve a hostname and return an error string if ANY of its A/AAAA
    records is a blocked address, else None. Checking every answer stops a name
    from smuggling one private record alongside a public one.

    ponytail: this validates then httpx re-resolves on connect, so a sub-second
    DNS-rebind could still slip a private address through the TOCTOU gap. That's
    an advanced attack against a playbook-gated, single-user tool; pin the
    connection to the resolved IP if this ever faces untrusted callers."""
    import socket
    if not host:
        return "missing host"
    try:
        infos = socket.getaddrinfo(host, None)
    except socket.gaierror as e:
        return f"cannot resolve host: {e}"
    ips = {info[4][0] for info in infos}
    if not ips:
        return "host did not resolve"
    blocked = [ip for ip in ips if _ip_is_blocked(ip)]
    if blocked:
        return f"refusing to fetch a private/loopback/link-local address ({', '.join(sorted(blocked))})"
    return None


async def _fetch_url(url: str = "", **_) -> str:
    import re
    import httpx
    from urllib.parse import urlparse, urljoin
    url = (url or "").strip()
    if not url.startswith(("http://", "https://")):
        return json.dumps({"error": "url must start with http:// or https://"})
    # SSRF guard: validate the host of the initial URL AND every redirect hop
    # against the private/loopback/link-local block-list before connecting, so a
    # granted fetch_url can't be steered at 127.0.0.1:11434, cloud metadata, or
    # LAN hosts — and a public URL can't 302 its way there either.
    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=False) as c:
            for _ in range(_FETCH_MAX_REDIRECTS + 1):
                parsed = urlparse(url)
                if parsed.scheme not in ("http", "https"):
                    return json.dumps({"error": "only http(s) URLs are allowed"})
                err = _ssrf_guard(parsed.hostname or "")
                if err:
                    return json.dumps({"error": f"blocked: {err}"})
                r = await c.get(url, headers={"User-Agent": "NexusOS/1.0"})
                location = r.headers.get("location")
                if r.is_redirect and location:
                    url = urljoin(url, location)
                    continue
                r.raise_for_status()
                html = r.text
                break
            else:
                return json.dumps({"error": "too many redirects"})
    except Exception as e:
        return json.dumps({"error": f"fetch failed: {e}"})
    text = re.sub(r"(?is)<(script|style).*?</\1>", " ", html)
    text = re.sub(r"(?s)<[^>]+>", " ", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text[:4000]


async def _remember(text: str = "", section: str = "General", **_) -> str:
    """WRITE tool: persist a memory fact. First action tool — allowlist-gated."""
    import uuid as _uuid
    text = (text or "").strip()
    if not text:
        return json.dumps({"error": "text is required"})
    store.add(MemoryItem(id=str(_uuid.uuid4()), section=(section or "General"), text=text))
    return json.dumps({"saved": text, "section": section or "General"})


# Canvas drawing APIs a real visualization must use — resizing width/height alone
# clears the buffer and draws nothing (a failure mode small models hit often).
_CANVAS_DRAW_APIS = (
    "fillrect", "strokerect", "filltext", "stroketext", "lineto", "arc(",
    "beziercurveto", "quadraticcurveto", "fill(", "stroke(", "putimagedata",
    "drawimage", "ellips(",
)

# Rejection threshold for a preview stage. Tiny 40×40 tiles (a recurring
# small-model collapse, often copied from earlier demos) can't show a sequence.
#
# These numbers are a threshold, never advice: every message that mentions a
# size quotes _STAGE_W/_STAGE_H instead. Weak models copy the first dimensions
# they read, so a message saying "at least 320x200, prefer 480x280" reliably
# produces 320x200 — three separate transcripts landed on exactly the minimum,
# including one that had a 480x280 example in front of it. Name one size.
_MIN_CANVAS_W = 320
_MIN_CANVAS_H = 200

# The size to ask for, and the only one any message should mention.
_STAGE_W = 480
_STAGE_H = 280

# Domain-agnostic interactive shell returned on reject as a *pattern* to adapt —
# not a finished demo for any particular algorithm. The model must implement
# generate() for the user's request (or ask them to clarify first).
_INTERACTIVE_SCAFFOLD_HTML = """<!DOCTYPE html>
<html><head><meta charset="utf-8"><style>
body{margin:0;font:14px/1.4 system-ui,sans-serif;background:#111;color:#eee;padding:12px}
.row{display:flex;gap:8px;align-items:center;margin-bottom:8px;flex-wrap:wrap}
input,button{font:inherit;padding:6px 10px}
canvas{display:block;width:480px;max-width:100%;height:auto;background:#1a1a1a;border:1px solid #333}
</style></head><body>
<div class="row">
  <label>n <input id="n" type="number" min="1" value="20"></label>
  <button id="go">Plot</button>
  <span id="meta"></span>
</div>
<canvas id="c" width="480" height="280"></canvas>
<script>
const canvas = document.getElementById('c');
const ctx = canvas.getContext('2d');

/** Return an array of numbers (or {x,y} points) for THIS demo. */
function generate(n) {
  // TODO: implement the user's algorithm / data here. Do not leave empty.
  const seq = [];
  for (let i = 0; i < n; i++) seq.push(i); // placeholder — replace
  return seq;
}

function plot(seq) {
  if (!seq || seq.length < 2) return;
  const vals = seq.map(v => (typeof v === 'number' ? v : v.y));
  const max = Math.max(1, ...vals);
  const w = canvas.width, h = canvas.height, pad = 16;
  ctx.clearRect(0, 0, w, h);
  ctx.strokeStyle = '#3b82f6';
  ctx.lineWidth = 2;
  ctx.beginPath();
  seq.forEach((v, i) => {
    const x = pad + i * ((w - 2 * pad) / Math.max(1, seq.length - 1));
    const y = h - pad - ((typeof v === 'number' ? v : v.y) / max) * (h - 2 * pad);
    if (i === 0) ctx.moveTo(x, y); else ctx.lineTo(x, y);
  });
  ctx.stroke();
  document.getElementById('meta').textContent = seq.length + ' points · max ' + max;
}

function go() {
  plot(generate(+document.getElementById('n').value || 20));
}
document.getElementById('go').onclick = go;
go();
</script></body></html>"""


def _wants_data_visual(purpose: str = "", title: str = "", markup: str = "") -> bool:
    """True when the submission claims to be a chart/plot/interactive visual."""
    blob = f"{purpose} {title} {markup}".lower()
    return any(k in blob for k in (
        "plot", "chart", "graph", "visual", "sequence", "orbit", "interactive",
        "demo", "canvas", "diagram", "animation", "simulate", "conjecture",
    ))


# Component counterpart to _INTERACTIVE_SCAFFOLD_HTML, handed back when a
# jsx/tsx submission is rejected. Same contract: a pattern to adapt, not a demo
# to paste. No imports — the preview puts hooks and h/render in scope already,
# and there is no module loader in the sandbox to satisfy an import anyway.
_INTERACTIVE_SCAFFOLD_JSX = """export default function App() {
  const canvasRef = useRef(null);
  const [n, setN] = useState(20);

  /** Return an array of numbers for THIS demo. */
  function generate(count) {
    // TODO: implement the user's algorithm / data here. Do not leave empty.
    const seq = [];
    for (let i = 0; i < count; i++) seq.push(i); // placeholder — replace
    return seq;
  }

  useEffect(() => {
    const canvas = canvasRef.current;
    const ctx = canvas.getContext('2d');
    const seq = generate(n);
    const max = Math.max(1, ...seq);
    const w = canvas.width, h = canvas.height, pad = 16;
    ctx.clearRect(0, 0, w, h);
    ctx.strokeStyle = '#3b82f6';
    ctx.lineWidth = 2;
    ctx.beginPath();
    seq.forEach((v, i) => {
      const x = pad + i * ((w - 2 * pad) / Math.max(1, seq.length - 1));
      const y = h - pad - (v / max) * (h - 2 * pad);
      if (i === 0) ctx.moveTo(x, y); else ctx.lineTo(x, y);
    });
    ctx.stroke();
  }, [n]);

  return (
    <div style={{ font: '14px system-ui', background: '#111', color: '#eee', padding: 12 }}>
      <label>n <input type="number" value={n} onInput={(e) => setN(+e.target.value || 1)} /></label>
      <canvas ref={canvasRef} width="480" height="280" style={{ display: 'block', background: '#1a1a1a' }} />
    </div>
  );
}"""


def _wants_chart(purpose: str = "", title: str = "", markup: str = "") -> bool:
    """True only when the submission claims to draw *data* — a narrower test
    than _wants_data_visual, which also counts "interactive" and "demo".

    That wider net is right for an HTML fence, where an interactive demo with no
    canvas is usually a model writing prose and calling it a visualization. It
    is wrong for a component fence: a JSX counter or form is interactive through
    its own elements and state, and demanding a <canvas> of it would reject the
    most ordinary thing JSX is for."""
    blob = f"{purpose} {title} {markup}".lower()
    return any(k in blob for k in (
        "plot", "chart", "graph", "visualiz", "diagram", "sequence", "orbit",
        "histogram", "scatter",
    ))


def _decorative_svg_not_plot(markup: str) -> bool:
    """True when SVG is present but doesn't encode a multi-point data chart."""
    import re
    lower = markup.lower()
    if "<svg" not in lower:
        return False
    # <defs> holds definitions, not output — nothing in it is drawn unless a
    # <use>/fill references it. A long path parked in there was passing as proof
    # of a real chart while the preview rendered an empty box.
    if "<defs" in lower and not re.search(r"<use\b|url\(#", lower):
        lower = re.sub(r"<defs\b.*?</defs\s*>", " ", lower, flags=re.S)
    rich_poly = bool(re.search(
        r"<polyline\b[^>]*\bpoints\s*=\s*[\"'][^\"']{40,}", lower,
    ))
    rich_path = bool(re.search(
        r"<path\b[^>]*\bd\s*=\s*[\"'][^\"']{40,}", lower,
    ))
    builds_plot = bool(
        re.search(r"createelementns\s*\(", lower)
        and ("polyline" in lower or "path" in lower or "line" in lower)
        and any(k in lower for k in ("foreach", "for (", "for(", "while(", "while ("))
        and any(k in lower for k in ("seq", "points", "push(", "data"))
    )
    canvas_plot = "getcontext" in lower and any(a in lower for a in _CANVAS_DRAW_APIS)
    return not (rich_poly or rich_path or builds_plot or canvas_plot)


def _critique_shared(markup: str) -> list[str]:
    """Checks that hold for every preview language."""
    import re
    issues: list[str] = []
    lower = markup.lower()

    if re.search(r"""(?i)(?:src|href)\s*=\s*['"]https?://""", markup):
        issues.append(
            "Remove external http(s) URLs — the sandboxed preview blocks them. "
            "Inline CSS/JS; use data: URIs for images/fonts."
        )

    # Model talking about the chat UI instead of building the visual.
    if any(p in lower for p in (
        "preview/code", "render_preview", "live preview/code",
        "paste the returned", "fenced block",
    )):
        issues.append(
            "Do not describe the chat Preview UI — submit only the visualization "
            "markup (canvas/SVG that plots data)."
        )

    return issues


def _critique_svg(markup: str, wants_plot: bool) -> list[str]:
    import re
    issues: list[str] = []
    lower = markup.lower()

    if "<svg" not in lower:
        issues.append("SVG markup must include an <svg> root element.")
    root = re.search(r"<svg\b[^>]*>", markup, re.I)
    if root:
        tag = root.group(0)
        wm = re.search(r'\bwidth\s*=\s*["\']?(\d+)', tag, re.I)
        hm = re.search(r'\bheight\s*=\s*["\']?(\d+)', tag, re.I)
        if wm and int(wm.group(1)) < _MIN_CANVAS_W:
            issues.append(
                f'SVG width is {wm.group(1)}px — too small to read. Use '
                f'width="{_STAGE_W}" height="{_STAGE_H}".'
            )
        if hm and int(hm.group(1)) < _MIN_CANVAS_H:
            issues.append(
                f'SVG height is {hm.group(1)}px — too small. Use height="{_STAGE_H}".'
            )
    if len(re.sub(r"\s+", "", markup)) < 60:
        issues.append(
            "SVG is too empty — add shapes (path/rect/circle/line/text) that "
            "actually illustrate the idea."
        )
    if wants_plot and _decorative_svg_not_plot(markup):
        issues.append(
            "This SVG is decorative (gradient/rect/single line), not a data "
            "chart. Build a <polyline>/<path> from many computed points, or "
            "prefer a <canvas> with getContext + lineTo over an array."
        )
    return issues


def _critique_html(markup: str, wants_plot: bool) -> list[str]:
    import re
    issues: list[str] = []
    lower = markup.lower()

    if re.search(r"(?:width|height)\s*:\s*40px", markup, re.I):
        issues.append(
            "Do not use 40x40 CSS tiles — that is not a visualization. Size the "
            f"stage {_STAGE_W}x{_STAGE_H}px."
        )

    if "<pre" in lower and ("<html" in lower or "<!doctype" in lower):
        issues.append(
            "Do not nest another HTML document inside <pre>. Put one interactive "
            "<canvas> (or <svg>) in the body and draw there."
        )

    has_canvas = "<canvas" in lower
    has_svg = "<svg" in lower
    issues += _critique_prose(markup)

    if wants_plot and not has_canvas and not has_svg:
        issues.append(
            "For chart/plot/interactive demos include a <canvas> or <svg> that "
            "draws the data — prose alone is rejected."
        )
    if wants_plot and has_svg and not has_canvas and _decorative_svg_not_plot(markup):
        issues.append(
            "SVG stage present but it does not plot data (no multi-point "
            "polyline/path, no JS that builds one from an array). Prefer "
            f'<canvas width="{_STAGE_W}" height="{_STAGE_H}"> + getContext + lineTo.'
        )
    issues += _critique_canvas(markup)

    if len(re.sub(r"\s+", "", markup)) < 40:
        issues.append("markup is too short to be a useful preview.")

    return issues


def _critique_prose(markup: str) -> list[str]:
    """Reject an explanation dressed up as a visualization — paragraphs, a list,
    maybe a button that reveals more text, and nothing that draws.

    Shared by html and jsx: a component returning four <p> elements is the same
    non-answer as a page of them, and for a while jsx was accepted precisely
    because this check lived only on the html side."""
    import re
    lower = markup.lower()
    if "<canvas" in lower or "<svg" in lower:
        return []
    prose_tags = len(re.findall(r"<(?:p|li|h[1-6]|ul|ol)\b", lower))
    toggle_only = prose_tags >= 3 and (
        "display" in lower or "toggle" in lower or "<button" in lower
    )
    if prose_tags >= 4 or toggle_only:
        return [
            "This is an explanation, not a visualization. Draw the data on a "
            f'<canvas width="{_STAGE_W}" height="{_STAGE_H}"> (or an <svg> chart) '
            "— not paragraphs, lists, or a button that only reveals more text."
        ]
    return []


def _critique_canvas(markup: str) -> list[str]:
    """Checks for markup that has a <canvas> in it, wherever that markup came
    from — a plain HTML body or the JSX that renders one."""
    import re
    issues: list[str] = []
    lower = markup.lower()

    if "<canvas" in lower:
        if "getcontext" not in lower:
            issues.append(
                "Canvas is present but never gets a 2D context — call "
                "canvas.getContext('2d') and draw with it."
            )
        if not any(api in lower for api in _CANVAS_DRAW_APIS):
            issues.append(
                "Canvas never draws anything — plot each step with "
                "fillRect/stroke/lineTo/arc/fillText (etc.). Do not only assign "
                "canvas.width/height inside a loop; that clears the canvas."
            )

        for tag in re.findall(r"<canvas\b[^>]*>", markup, re.I):
            wm = re.search(r'\bwidth\s*=\s*["\']?(\d+)', tag, re.I)
            hm = re.search(r'\bheight\s*=\s*["\']?(\d+)', tag, re.I)
            if wm and int(wm.group(1)) < _MIN_CANVAS_W:
                issues.append(
                    f'Canvas width="{wm.group(1)}" is too small — use width="{_STAGE_W}" '
                    f'height="{_STAGE_H}", then map each data value to (x, y) pixels.'
                )
            if hm and int(hm.group(1)) < _MIN_CANVAS_H:
                issues.append(
                    f'Canvas height="{hm.group(1)}" is too small — use height="{_STAGE_H}".'
                )
            if re.search(r'\bwidth\s*=\s*["\']?100%', tag, re.I):
                issues.append(
                    "Use numeric canvas width/height attributes (e.g. width=\"480\"), "
                    "not percentages — the bitmap size must be explicit."
                )

        full_blit = bool(re.search(
            r"(?:fillrect|clearrect)\s*\(\s*0\s*,\s*0\s*,\s*\d+\s*,\s*\d+\s*\)",
            lower,
        ))
        plots_points = bool(
            re.search(r"lineto\s*\(", lower)
            or re.search(r"fillrect\s*\(\s*(?!0\s*,\s*0)", lower)
            or re.search(r"filltext\s*\(", lower)
            or re.search(r"arc\s*\(", lower)
            or re.search(r"strokerect\s*\(\s*(?!0\s*,\s*0)", lower)
        )
        if full_blit and not plots_points:
            issues.append(
                "You are only fillRect/clearRect(0,0,W,H) — that paints the whole "
                "canvas, not the data. Collect values into an array, then for each "
                "index i draw at x=i*step, y=height - value*scale (lineTo or "
                "fillRect(x, y, barW, barH))."
            )

    return issues


def _critique_jsx(markup: str, wants_plot: bool) -> list[str]:
    """A JSX/TSX fence is one self-contained component. It is transformed and
    mounted in the browser (interface/web/src/preview/), so the checks here are
    the things that transform cannot recover from or would mount into nothing."""
    import re
    issues: list[str] = []
    lower = markup.lower()

    # Something has to be mounted: an explicit default export, a component named
    # App, or some capitalized declaration to fall back to.
    has_component = bool(
        re.search(r"\bexport\s+default\b", markup)
        or re.search(r"\bfunction\s+[A-Z]\w*", markup)
        or re.search(r"\b(?:const|let|var)\s+[A-Z]\w*\s*=", markup)
        or re.search(r"\bclass\s+[A-Z]\w*", markup)
    )
    if not has_component:
        issues.append(
            "No component to mount — define one with a capitalized name "
            "(e.g. `function App() { ... }`) or `export default` it."
        )

    if "<" not in markup or not re.search(r"<[A-Za-z>]", markup):
        issues.append(
            "No JSX found — the component must return elements "
            "(e.g. `return <div>…</div>;`)."
        )

    # Imports are stripped before the code runs: there is no module loader and
    # no network in the sandbox. React/Preact itself is already in scope.
    for module in re.findall(r"""\bfrom\s+['"]([^'"]+)['"]""", markup):
        if module.split("/")[0] not in ("react", "react-dom", "preact"):
            issues.append(
                f"Cannot import '{module}' — the preview has no module loader and "
                "no network. Inline what you need; React/Preact hooks are already "
                "in scope without importing."
            )

    if wants_plot and "<canvas" not in lower and "<svg" not in lower:
        issues.append(
            "For chart/plot/interactive demos render a <canvas> or <svg> that "
            "draws the data — prose alone is rejected."
        )

    issues += _critique_prose(markup)
    issues += _critique_canvas(markup)

    if len(re.sub(r"\s+", "", markup)) < 40:
        issues.append("markup is too short to be a useful preview.")

    return issues


# The one place that says which languages the render window supports. Each entry
# owns that language's validation; the tool schema's `lang` enum, the dispatch in
# _critique_render, and the capability line in the system prompt are all derived
# from these keys rather than repeating them.
#
# The frontend keeps its own matching registry (PREVIEW_LANGS in
# interface/web/src/Markdown.jsx) because the two sides need different things per
# language - this side validates, that side renders - and neither should depend
# on the other at runtime. tests/test_tools.py asserts the key sets stay equal,
# so drift fails the check gate instead of silently degrading to a plain code
# block in the chat.
PREVIEW_LANGS: dict[str, dict] = {
    "html": {
        "summary": "self-contained HTML document",
        "critique": _critique_html,
        # HTML also covers ordinary interactive UIs (forms, calculators, DOM
        # demos). Only require a drawing surface when the request specifically
        # claims to be a chart/plot/data visualization.
        "wants_visual": _wants_chart,
        "scaffold": _INTERACTIVE_SCAFFOLD_HTML,
    },
    "svg": {
        "summary": "standalone SVG image",
        "critique": _critique_svg,
        "wants_visual": _wants_data_visual,
        "scaffold": _INTERACTIVE_SCAFFOLD_HTML,
    },
    "jsx": {
        "summary": "single Preact/React component (JSX)",
        "critique": _critique_jsx,
        "wants_visual": _wants_chart,
        "scaffold": _INTERACTIVE_SCAFFOLD_JSX,
    },
    "tsx": {
        "summary": "single Preact/React component (TypeScript JSX)",
        "critique": _critique_jsx,
        "wants_visual": _wants_chart,
        "scaffold": _INTERACTIVE_SCAFFOLD_JSX,
    },
}


def _scaffold_for(lang: str) -> str:
    """The starting pattern handed back on reject. Per-language: answering a
    rejected component with a full HTML document tells the model to write the
    wrong thing entirely."""
    entry = PREVIEW_LANGS.get(lang)
    return (entry["scaffold"] if entry else _INTERACTIVE_SCAFFOLD_HTML).strip()


def _lang_prose() -> str:
    """'html or svg' — the supported languages as a phrase for prompts/errors."""
    names = list(PREVIEW_LANGS)
    if len(names) < 2:
        return names[0] if names else ""
    return f"{', '.join(names[:-1])} or {names[-1]}"


def _critique_render(lang: str, markup: str, purpose: str = "", title: str = "") -> list[str]:
    """Cheap static checks so render_preview rejects empty/fake visuals before
    the model pastes them into the chat as a 'working' demo.

    Scope: things that RUN but are not what was asked for — a 40x40 stage, a
    decorative gradient standing in for a chart, prose with no drawing in it, a
    canvas that only paints itself one colour. These fail silently no matter
    what, so static checks are the only thing that can catch them.

    Not in scope: code that throws. The preview reports its own runtime errors
    now (the bootstrap in interface/web/src/Markdown.jsx), so guessing at them
    here bought nothing and cost accuracy. Three checks were removed once that
    landed, each verified against the real error channel first:

        const canvas = el.getContext('2d') ... ctx.lineTo()
            -> "ReferenceError: ctx is not defined  (line 4)"
        function collatz() ... coll(27)
            -> "ReferenceError: coll is not defined  (line 5)"
        document.createElementNS('line')
            -> "TypeError: ... 2 arguments required, but only 1 present.  (line 3)"

    The real messages are better than the regexes were: they carry a line
    number, and they catch *any* undefined name rather than the two spellings
    someone thought to anticipate. Resist re-adding a static check for anything
    that already throws."""
    entry = PREVIEW_LANGS.get(lang)
    if entry is None:
        return [f"unsupported preview language {lang!r} — use {_lang_prose()}."]
    # What counts as "claimed a visual" differs by language: see _wants_chart.
    wants_plot = entry["wants_visual"](purpose, title, markup)
    return _critique_shared(markup) + entry["critique"](markup, wants_plot)


def _with_scaffold(payload: dict, lang: str, attempt: int) -> dict:
    """Attach the starting pattern, but only from the second rejection on.

    A complete, styled, runnable document handed to a struggling model does not
    get adapted — it gets pasted, and then it persists in the conversation and
    comes back as retrieved context for the next request, carrying its example
    domain with it. Transcripts show exactly that: a scaffold's CSS reappearing
    verbatim in an answer to an unrelated prompt, in a conversation where this
    tool was never even called. So the first rejection says only what is wrong;
    the pattern appears once that has not been enough."""
    if attempt < 1:
        return payload
    return {
        **payload,
        "scaffold": _scaffold_for(lang),
        "scaffold_note": (
            "A pattern to adapt, not an answer to paste. Replace generate() with "
            "the logic this request actually needs; keep nothing you do not use."
        ),
    }


async def _render_preview(
    lang: str = "html",
    title: str = "",
    markup: str = "",
    purpose: str = "",
    _attempt: int = 0,
    **_,
) -> str:
    """Validate + package a live-preview fence. Read-only: nothing is executed
    server-side; the chat UI renders the returned fence in a sandboxed iframe.

    `_attempt` is supplied by the tool loop, not by the model — it is how many
    times this call has already been rejected in the current turn."""
    lang = (lang or "html").strip().lower()
    markup = (markup or "").strip()
    title = (title or "").strip()
    purpose = (purpose or "").strip()

    if lang not in PREVIEW_LANGS:
        return json.dumps({"ok": False, "error": f"lang must be {_lang_prose()}"})
    if not markup:
        return json.dumps(_with_scaffold({
            "ok": False,
            "error": (
                f"markup is required — send the complete {lang} for the visual you "
                "want, with all CSS and JS inline and no external URLs."
            ),
        }, lang, _attempt))

    issues = _critique_render(lang, markup, purpose=purpose, title=title)
    if issues:
        return json.dumps(_with_scaffold({
            "ok": False,
            "issues": issues,
            "hint": (
                "Fix these and call render_preview again, building the thing that "
                f"was actually asked for. Draw on a {_STAGE_W}x{_STAGE_H} stage; "
                "compute the values into an array first, then plot them point by "
                "point with lineTo/fillRect(x,y,w,h); add <input>/<button> controls "
                "when it should be interactive."
            ),
            "purpose": purpose or None,
        }, lang, _attempt))

    fence = f"```{lang}\n{markup}\n```"
    return json.dumps({
        "ok": True,
        "title": title or None,
        "purpose": purpose or None,
        "instruction": (
            "Write a short intro, then paste this fenced block exactly as it is. "
            "Do not wrap it in a second fence, resize it, or rewrite the code."
        ),
        "fence": fence,
    })


# name -> (schema, callable). Schema is the OpenAI/Ollama function-tool format.
REGISTRY: dict[str, tuple[dict, Callable[..., Awaitable[str]]]] = {
    "search_memory": (
        {
            "type": "function",
            "function": {
                "name": "search_memory",
                "description": "Search the user's persistent memory facts. Empty query returns all facts.",
                "parameters": {
                    "type": "object",
                    "properties": {"query": {"type": "string", "description": "text to match"}},
                },
            },
        },
        _search_memory,
    ),
    "search_history": (
        {
            "type": "function",
            "function": {
                "name": "search_history",
                "description": "Search past conversations for exchanges containing the query text.",
                "parameters": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            },
        },
        _search_history,
    ),
    "list_models": (
        {
            "type": "function",
            "function": {
                "name": "list_models",
                "description": "List the locally installed Ollama models.",
                "parameters": {"type": "object", "properties": {}},
            },
        },
        _list_models,
    ),
    "search_documents": (
        {
            "type": "function",
            "function": {
                "name": "search_documents",
                "description": "Search the user's uploaded documents for relevant passages.",
                "parameters": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            },
        },
        _search_documents,
    ),
    "get_time": (
        {
            "type": "function",
            "function": {
                "name": "get_time",
                "description": "Get the current local date and time.",
                "parameters": {"type": "object", "properties": {}},
            },
        },
        _get_time,
    ),
    "render_preview": (
        {
            "type": "function",
            "function": {
                "name": "render_preview",
                # Written as instructions TO you, imperative and short. Earlier
                # versions narrated what "the user" wants and listed numbered
                # requirements; weak models echoed that narration back as their
                # reply — asking the user to clarify an already-clear request,
                # in the third person, instead of building anything. Keep this
                # terse, keep it second-person, and add nothing the model can
                # recite in place of acting.
                "description": (
                    f"Build a working visual — chart, plot, diagram, interactive demo — "
                    f"as self-contained {_lang_prose()} and send it here to check. "
                    f"Draw on a {_STAGE_W}x{_STAGE_H} stage. Compute your values into an "
                    "array, then plot them point by point (canvas: getContext, then "
                    "lineTo/fillRect(x,y,w,h)/arc per point). Add <input>/<button> "
                    "controls if it should be interactive. Inline all CSS and JS; the "
                    "preview is sandboxed with no network, so external URLs will not "
                    "load. Build what was asked for, not a similar demo you know better. "
                    "Rejected: fix what `issues` lists and send it again. "
                    "Accepted: paste the returned `fence` into your reply unchanged."
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "lang": {
                            "type": "string",
                            "enum": list(PREVIEW_LANGS),
                            "description": (
                                "Preview language tag for the fenced block: "
                                + "; ".join(
                                    f"{name} ({spec['summary']})"
                                    for name, spec in PREVIEW_LANGS.items()
                                )
                            ),
                        },
                        "title": {
                            "type": "string",
                            "description": "Short label for the visual.",
                        },
                        "purpose": {
                            "type": "string",
                            "description": "One sentence: what this visual shows.",
                        },
                        "markup": {
                            "type": "string",
                            "description": (
                                "Full self-contained HTML document or SVG. Inline all "
                                "CSS/JS. No external script/style/img URLs."
                            ),
                        },
                    },
                    "required": ["lang", "markup"],
                },
            },
        },
        _render_preview,
    ),
    "web_search": (
        {
            "type": "function",
            "function": {
                "name": "web_search",
                "description": "Search the web (DuckDuckGo) and return the top result snippets.",
                "parameters": {
                    "type": "object",
                    "properties": {"query": {"type": "string"}},
                    "required": ["query"],
                },
            },
        },
        _web_search,
    ),
    "fetch_url": (
        {
            "type": "function",
            "function": {
                "name": "fetch_url",
                "description": "Fetch a web page and return its visible text (truncated).",
                "parameters": {
                    "type": "object",
                    "properties": {"url": {"type": "string", "description": "http(s) URL"}},
                    "required": ["url"],
                },
            },
        },
        _fetch_url,
    ),
    "remember": (
        {
            "type": "function",
            "function": {
                "name": "remember",
                "description": "Save a durable fact to the user's persistent memory.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "text": {"type": "string", "description": "the fact to remember"},
                        "section": {"type": "string", "description": "optional category, e.g. Health"},
                    },
                    "required": ["text"],
                },
            },
        },
        _remember,
    ),
}


# Tools that act (write local state or reach the network). These require an
# explicit consent gate (settings.allow_action_tools) on top of the per-playbook
# allowlist — a playbook granting one isn't enough on its own.
ACTION_TOOLS = frozenset({"web_search", "fetch_url", "remember"})

# Always advertised when the user asks for a visual (see wants_render_preview).
# Not playbook-gated — the render window is a standing UI capability.
STANDING_TOOLS = frozenset({"render_preview"})

# User-message cues that justify running the (slow, non-stream) tool loop with
# render_preview. Kept narrow so ordinary chat isn't blocked behind a tool turn.
_RENDER_HINTS = (
    "visual", "visualize", "visualization", "chart", "graph", "diagram",
    "canvas", "plot", "interactive", "animation", "render_preview",
    "render preview", "svg", "draw me", "live preview",
    "demonstrate", "demo", "html demo", "html snippet", "html file",
    # Ways of asking for something that reacts to the pointer. "interactive"
    # alone missed "mouse-over sensitive", and with it the whole feature.
    "hover", "mouse", "drag", "click on", "real-time", "realtime",
    "simulation", "simulate", "particle", "animate",
    # Every language the render window can display. Naming one is asking for a
    # preview, and this way a language added to PREVIEW_LANGS starts hinting
    # for itself instead of being unreachable until someone edits this tuple -
    # which is exactly what happened to jsx/tsx.
) + tuple(PREVIEW_LANGS)


def wants_render_preview(message: str) -> bool:
    """True when this turn should advertise render_preview / enter the tool loop."""
    lower = (message or "").lower()
    return any(h in lower for h in _RENDER_HINTS)


def is_action(name: str) -> bool:
    return name in ACTION_TOOLS


def schemas_for(names: list[str], allow_actions: bool = True) -> list[dict]:
    """Tool schemas for a playbook's allowlist; unknown names are dropped.
    When allow_actions is False, action tools are withheld so the model can't
    even call them."""
    return [
        REGISTRY[n][0] for n in (names or [])
        if n in REGISTRY and (allow_actions or not is_action(n))
    ]


def standing_schemas() -> list[dict]:
    """Schemas that ship with visual turns (currently just render_preview)."""
    return schemas_for(sorted(STANDING_TOOLS), allow_actions=True)


async def dispatch(name: str, args: dict | None) -> str:
    """Run a tool by name. Never raises — returns an error string on failure."""
    entry = REGISTRY.get(name)
    if not entry:
        return json.dumps({"error": f"unknown tool: {name}"})
    try:
        return await entry[1](**(args or {}))
    except Exception as e:  # a broken tool must not kill the chat loop
        return json.dumps({"error": f"{name} failed: {e}"})