#!/usr/bin/env python3
# desktop-assistant: local text assistant daemon.
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 2 as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
# Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""Questions arrive as JSON lines on a Unix socket, the loaded llama-server
model answers with tools, replies stream back on the same socket, and every
finished turn is kept in a SQLite history with full-text search.

Run: python3 assistant.py        (--test runs the self-check, --defaults
prints the built-in config as JSON, for a settings UI, --cloud-models NAME
prints that [[cloud]] provider's model ids as JSON)
"""

import base64
import glob
import hashlib
import http.client
import json
import mimetypes
import os
import re
import shlex
import signal
import socket
import sqlite3
import subprocess
import sys
import threading
import time
import tomllib
import traceback
import wave
import urllib.error
import urllib.parse
import urllib.request

CONFIG_DIR = os.path.join(os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config"),
                          "desktop-assistant")
DB_PATH = os.path.join(os.environ.get("XDG_DATA_HOME") or os.path.expanduser("~/.local/share"),
                       "desktop-assistant", "history.db")
DEFAULTS = {
    "llm": "http://localhost:8181/v1",
    "system": ("You are a desktop assistant. Answer in the language the user writes, "
               "Italian or English. Markdown is rendered: use lists, tables and fenced "
               "code blocks when they help, and keep answers as short as the question "
               "allows. Facts about the user's computer, devices, accounts, mail, "
               "calendar or money come only from a tool: call one, and if none can "
               "tell, say you do not know. Never guess them."),
    "tool_rounds": 3,
    "temperature": None,                  # None: the llama-server preset's
    "thinking": True,                     # False skips the model's reasoning
    "cloud": [],                          # [[cloud]] providers, see check_cloud
    "env": {},
}
OPEN = {"env"}                            # free-form tables: replaced whole, not merged


def check_cloud(v):
    # [[cloud]] tables. A provider's name prefixes its models' ids
    # ("name:model"), so it is unique and holds no colon. models may be
    # empty: the settings page saves a provider before fetching its list.
    if not isinstance(v, list) or not all(isinstance(p, dict) for p in v):
        raise SystemExit("config: cloud must be [[cloud]] tables")
    seen = set()
    for p in v:
        name = p.get("name")
        if set(p) != {"name", "url", "key_cmd", "models"}:
            raise SystemExit(f"config: cloud {name!r} needs exactly name, url, key_cmd, models")
        if (not all(isinstance(p[k], str) and p[k] for k in ("name", "url", "key_cmd"))
                or not isinstance(p["models"], list)
                or not all(isinstance(m, str) for m in p["models"])):
            raise SystemExit(f"config: cloud {name!r}: name, url and key_cmd are strings, "
                             "models a list of strings")
        if ":" in name or name in seen:
            raise SystemExit(f"config: cloud name {name!r} is repeated or has a colon")
        seen.add(name)
    return v


def merge(base, over, where=""):
    # A typo must fail loudly, not fall back to the default in silence.
    out = dict(base)
    for k, v in over.items():
        if k not in base:
            raise SystemExit(f"config: unknown key {where}{k}")
        if isinstance(base[k], dict) and k not in OPEN:
            if not isinstance(v, dict):
                raise SystemExit(f"config: {where}{k} must be a table")
            v = merge(base[k], v, f"{where}{k}.")
        if k == "cloud":
            v = check_cloud(v)
        out[k] = v
    return out


def load_config(path):
    try:
        with open(path, "rb") as f:
            return merge(DEFAULTS, tomllib.load(f))
    except FileNotFoundError:
        return DEFAULTS


def load_tools(d):
    tools = {}
    for path in sorted(glob.glob(os.path.join(d, "*.toml"))):
        with open(path, "rb") as f:
            t = tomllib.load(f)
        if not isinstance(t.get("description"), str) or not t.get("command"):
            raise SystemExit(f"{path}: needs description and command")
        if not isinstance(t.get("cloud", False), bool):
            raise SystemExit(f"{path}: cloud must be true or false")
        tools[os.path.basename(path)[:-5]] = t
    return tools


def schema(tools):
    return [{"type": "function", "function": {
        "name": name, "description": t["description"],
        "parameters": {"type": "object", "properties": {
            p: {"type": s.get("type", "string"), "description": s.get("description", "")}
            for p, s in t.get("params", {}).items()}}}}
        for name, t in tools.items()]


CFG = load_config(os.path.join(CONFIG_DIR, "config.toml"))
TOOLS_DIR = os.path.join(CONFIG_DIR, "tools")
TOOLS = load_tools(TOOLS_DIR)
SOCK = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "desktop-assistant.sock")


def log(*a):
    print(time.strftime("%H:%M:%S"), *a, flush=True)


# History: one SQLite file, written only by this process. turns.messages is
# what the model is sent for that turn; user, reply and tools are what the
# canvas draws. The FTS index follows turns through triggers, and a
# conversation delete cascades into turns and so into the index.
SCHEMA = """
CREATE TABLE IF NOT EXISTS conversations(id INTEGER PRIMARY KEY, title TEXT,
                                         created INTEGER, updated INTEGER);
CREATE TABLE IF NOT EXISTS turns(id INTEGER PRIMARY KEY,
    conv INTEGER REFERENCES conversations ON DELETE CASCADE,
    seq INTEGER, user TEXT, reply TEXT, tools TEXT, messages TEXT);
CREATE INDEX IF NOT EXISTS turns_conv ON turns(conv, seq);
CREATE VIRTUAL TABLE IF NOT EXISTS turns_fts USING fts5(user, reply, content='turns',
    content_rowid='id', tokenize='unicode61 remove_diacritics 2');
CREATE TRIGGER IF NOT EXISTS turns_ai AFTER INSERT ON turns BEGIN
    INSERT INTO turns_fts(rowid, user, reply) VALUES (new.id, new.user, new.reply);
END;
CREATE TRIGGER IF NOT EXISTS turns_ad AFTER DELETE ON turns BEGIN
    INSERT INTO turns_fts(turns_fts, rowid, user, reply)
        VALUES ('delete', old.id, old.user, old.reply);
END;
"""


def db_open(path):
    if path != ":memory:":
        os.makedirs(os.path.dirname(path), exist_ok=True)
        os.chmod(os.path.dirname(path), 0o700)   # also when something else made it
    old = os.umask(0o177)                 # 0600; SQLite gives the WAL files the db's mode
    try:
        conn = sqlite3.connect(path, check_same_thread=False)
    finally:
        os.umask(old)
    if path != ":memory:":
        os.chmod(path, 0o600)             # an existing file keeps its old mode otherwise
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    conn.executescript(SCHEMA)
    # Added with attachments; a database from before them gains the column.
    if "files" not in {r[1] for r in conn.execute("PRAGMA table_info(turns)")}:
        conn.execute("ALTER TABLE turns ADD COLUMN files TEXT")
    # Added with cloud models: the model a conversation is asked with, NULL
    # for whatever the router has loaded.
    if "model" not in {r[1] for r in conn.execute("PRAGMA table_info(conversations)")}:
        conn.execute("ALTER TABLE conversations ADD COLUMN model TEXT")
    return conn


def save_turn(conn, conv, user, reply, tools, messages, files=(), model=None):
    # Returns the conversation id: the row is made at its first finished turn,
    # so a New that is never used leaves nothing behind. A question that is
    # only files is titled by their names.
    now = int(time.time())
    title = " ".join(user.split()[:6]) or ", ".join(f["name"] for f in files)
    with conn:
        if conv is None:
            conv = conn.execute("INSERT INTO conversations(title, created, updated, model) "
                                "VALUES (?, ?, ?, ?)", (title, now, now, model)).lastrowid
        else:
            conn.execute("UPDATE conversations SET updated = ?, model = ? WHERE id = ?",
                         (now, model, conv))
        seq = conn.execute("SELECT COALESCE(MAX(seq), 0) + 1 FROM turns WHERE conv = ?",
                           (conv,)).fetchone()[0]
        conn.execute("INSERT INTO turns(conv, seq, user, reply, tools, messages, files) "
                     "VALUES (?, ?, ?, ?, ?, ?, ?)",
                     (conv, seq, user, reply, json.dumps(tools), json.dumps(messages),
                      json.dumps(list(files))))
    return conv


def list_convs(conn):
    return [{"id": i, "title": t, "updated": u} for i, t, u in conn.execute(
        "SELECT id, title, updated FROM conversations ORDER BY updated DESC, id DESC")]


def load_conv(conn, conv):
    # (model history, shown turns), or None for an id that is not there.
    rows = conn.execute("SELECT user, reply, tools, messages, files FROM turns "
                        "WHERE conv = ? ORDER BY seq", (conv,)).fetchall()
    if not rows:
        return None
    return ([json.loads(m) for _, _, _, m, _ in rows],
            [{"user": u, "reply": r, "tools": json.loads(t), "files": json.loads(f or "[]")}
             for u, r, t, _, f in rows])


def search(conn, q):
    # The query is one quoted phrase with a prefix star on its last word, so
    # FTS syntax typed by the user is matched as text and cannot widen it.
    # ponytail: rank per turn, first hit per conversation; 30 results.
    q = q.replace("\0", "").strip()         # NUL ends the FTS string early
    if not q:
        return []
    seen, out = set(), []
    for conv, title, snip in conn.execute(
            "SELECT t.conv, c.title, snippet(turns_fts, -1, '**', '**', '…', 12) "
            "FROM turns_fts JOIN turns t ON t.id = turns_fts.rowid "
            "JOIN conversations c ON c.id = t.conv "
            "WHERE turns_fts MATCH ? ORDER BY rank LIMIT 300",
            ('"' + q.replace('"', '""') + '"*',)):
        if conv not in seen:
            seen.add(conv)
            out.append({"id": conv, "title": title, "snippet": snip})
    return out[:30]


def delete_conv(conn, conv):
    with conn:
        conn.execute("DELETE FROM conversations WHERE id = ?", (conv,))


# Attachments. Paths come from the canvas; the daemon runs as the user and
# reads them itself. Text, and a PDF's text, go inline and are stored that
# way, so a reopened conversation still has them. Images are stored as a
# path and hash and read again when sent, so the history does not grow by
# every picture, and a file changed since is reported gone rather than
# shown as something it no longer is.
CODE_SUFFIXES = {
    ".py", ".c", ".h", ".cpp", ".hpp", ".rs", ".go", ".js", ".ts", ".jsx",
    ".tsx", ".sh", ".bash", ".zsh", ".lua", ".rb", ".pl", ".php", ".java",
    ".kt", ".swift", ".sql", ".toml", ".yaml", ".yml", ".json", ".ini",
    ".cfg", ".conf", ".md", ".rst", ".txt", ".csv", ".xml", ".html", ".css",
    ".diff", ".patch", ".slackbuild", ".info", ".desktop", ".service", ".qml",
}
IMAGE_TYPES = {"image/png", "image/jpeg", "image/gif", "image/webp"}
MAX_FILES = 8
MAX_IMAGE = 20 << 20
MAX_PAGE_IMAGES = 4                       # scanned PDF pages sent as images
PAGE_PX = 1600                            # longest side of a page render
# ~25 tokens a second: 60 s stays under the 2048 ubatch, which an image
# crossing took the model down (AGENTS.md). Audio past it is untried.
MAX_AUDIO_S = 60
HEAR = ("Transcribe this audio exactly, in the language spoken (Italian or English). "
        "Output only the transcript.")
TITLE = ("Write a title of at most six words for this conversation, in the language "
         "it is in. Output only the title.")


class Refused(Exception):
    """A file or request the user has to change; logged without a trace."""


def kind_of(path):
    mime = mimetypes.guess_type(path)[0]
    if mime in IMAGE_TYPES:
        return "image"
    if mime == "application/pdf":
        return "pdf"
    if (mime or "").startswith("text/") or os.path.splitext(path)[1].lower() in CODE_SUFFIXES:
        return "text"
    if mime is None and looks_like_text(path):    # no telling name: a script, a Makefile
        return "text"
    return "other"


def looks_like_text(path):
    # The first 4 KB: no NUL and valid UTF-8 (a character cut at the end of
    # the window is allowed).
    try:
        with open(path, "rb") as f:
            head = f.read(4096)
    except OSError:
        return False
    if b"\0" in head:
        return False
    try:
        head.decode("utf-8")
    except UnicodeDecodeError as e:
        return len(head) == 4096 and e.reason == "unexpected end of data"
    return True


def sha256(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def pdf_pages(path):
    # ponytail: the whole text is read, then cut to the share; a page cap
    # (-l) if huge PDFs ever matter.
    try:
        out = subprocess.run(["pdftotext", "-layout", path, "-"], capture_output=True,
                             text=True, timeout=60)
    except subprocess.TimeoutExpired:
        raise Refused(f"cannot read {os.path.basename(path)}: timed out") from None
    except OSError as e:                  # poppler missing
        raise Refused(f"cannot read {os.path.basename(path)}: {e.strerror}") from None
    if out.returncode:
        last = out.stderr.strip().splitlines()[-1:] or ["pdftotext failed"]
        raise Refused(f"cannot read {os.path.basename(path)}: {last[0]}")
    pages = out.stdout.split("\f")
    return pages[:-1] if out.stdout.endswith("\f") else pages   # each page ends in \f


def attach(paths, budget):
    # (text blocks, image references, chips) for the files; Refused says
    # what is wrong with the first bad one. Text files and PDFs share the
    # budget of characters equally.
    if len(paths) > MAX_FILES:
        raise Refused(f"at most {MAX_FILES} files")
    loaded = []
    for p in paths:
        # A newline in a name would close the file block early.
        name = " ".join((os.path.basename(p) or p).splitlines())
        if not (os.path.isabs(p) and os.path.isfile(p) and os.access(p, os.R_OK)):
            raise Refused(f"cannot read {name}")
        kind = kind_of(p)
        if kind == "other":
            raise Refused(f"{name}: chat supports text, images and PDF")
        if kind == "image" and os.path.getsize(p) > MAX_IMAGE:
            raise Refused(f"{name}: images over 20 MB are refused")
        loaded.append((p, name, kind))
    share = budget // max(1, sum(k != "image" for _, _, k in loaded))
    blocks, refs, chips = [], [], []
    for p, name, kind in loaded:
        chip = {"name": name, "kind": kind, "truncated": False}
        chips.append(chip)
        if kind == "image":
            refs.append({"type": "image_ref", "name": name, "path": p, "sha256": sha256(p)})
            continue
        if kind == "pdf":
            pages = pdf_pages(p)
            chip["pages"] = len(pages)
            body = "\n\n".join(f"[page {i}]\n{t.rstrip()}" for i, t in enumerate(pages, 1))
            # A page with no text is a scan: the model reads it as an image.
            scans = [i for i, t in enumerate(pages, 1) if not t.strip()]
            blank, unseen = scans[:MAX_PAGE_IMAGES], scans[MAX_PAGE_IMAGES:]
            digest = sha256(p) if blank else ""
            refs += [{"type": "image_ref", "name": f"{name} page {n}", "path": p,
                      "sha256": digest, "page": n} for n in blank]
        else:
            with open(p, "rb") as f:          # no more than the share could use
                body = f.read(share * 4 + 4).decode("utf-8", "replace")
        if len(body) > share:
            body, chip["truncated"] = body[:share], True
        if kind == "pdf" and unseen:          # the model must not think it saw them all
            body += f"\n[scanned pages not shown: {', '.join(map(str, unseen))}]"
            chip["truncated"] = True
        note = " (truncated)" if chip["truncated"] else ""
        blocks.append(f"--- file: {name}{note} ---\n{body}\n--- end file ---")
    return blocks, refs, chips


def resolve(part, images=True):
    # An image reference back into an image the model can see, or a note
    # that it is gone (moved, deleted, changed since it was sent) or that the
    # model now loaded cannot see it, which would otherwise fail the request.
    if part.get("type") != "image_ref":
        return part
    if not images:
        return {"type": "text", "text": f"[image {part['name']}: this model cannot see images]"}
    gone = {"type": "text", "text": f"[image {part['name']} is no longer available]"}
    try:
        if "page" in part:
            if sha256(part["path"]) != part["sha256"]:
                return gone
            # 150 dpi, lowered for a large page so its longest side stays
            # within PAGE_PX: unbounded, a 3000 pt page rendered 6250x6250 at
            # 171 MB. Never raised: -scale-to also upscales, and a 600x300 pt
            # scan sent as 2048x1024 with a second image took llama-server's
            # model process down.
            n = str(part["page"])
            info = subprocess.run(["pdfinfo", "-f", n, "-l", n, part["path"]],
                                  capture_output=True, text=True, timeout=30).stdout
            size = re.search(r"size:\s+([\d.]+) x ([\d.]+)", info)
            longest = max(map(float, size.groups())) if size else 842.0   # A4 if unreported
            dpi = min(150.0, PAGE_PX * 72 / longest)
            out = subprocess.run(["pdftoppm", "-f", n, "-l", n, "-r", f"{dpi:.2f}", "-png",
                                  part["path"]], capture_output=True, timeout=60)
            if out.returncode or not out.stdout or len(out.stdout) > MAX_IMAGE:
                return gone
            data, mime = out.stdout, "image/png"
        else:
            with open(part["path"], "rb") as f:   # once: the bytes hashed are the bytes sent
                data = f.read()
            if hashlib.sha256(data).hexdigest() != part["sha256"]:
                return gone
            mime = mimetypes.guess_type(part["path"])[0]
    except (OSError, subprocess.TimeoutExpired):
        return gone
    return {"type": "image_url",
            "image_url": {"url": f"data:{mime};base64,{base64.b64encode(data).decode()}"}}


def wire(messages, images=True):
    # What is sent: the stored messages with their image references resolved.
    return [{**m, "content": [resolve(p, images) for p in m["content"]]}
            if isinstance(m.get("content"), list) else m for m in messages]


# The canvas's feed: JSON lines on a Unix socket, both ways. Every client
# gets every event. A client that stalls for 2 s is dropped rather than
# holding up the others.
clients, clients_lock = [], threading.Lock()
now_state = {"state": "idle"}             # repeated in hello for a late client
ctx = {"model": None, "max": 0, "used": 0}
# One lock for the conversation, the turn counter and the database. Emits
# that depend on the turn happen under it, so a stopped reply cannot send a
# token after its done.
lock = threading.RLock()
cur = {"id": None, "history": [], "turns": [], "pending": None, "model": None}
inflight = {"conn": None}   # the HTTP connection the current reply streams over
writing = {"on": False}     # a write in flight, which interrupt() also ends
turn = 0          # bumped by every ask, stop, new and open: older replies are stale
DB = None


def emit(event):
    if event["type"] == "state":
        now_state.clear()
        now_state.update({k: v for k, v in event.items() if k != "type"})
    line = (json.dumps(event) + "\n").encode()
    # ponytail: sends run under the caller's lock, so one stalled client holds
    # everyone up for its 2 s timeout, once, before it is dropped; per-client
    # send queues if that is ever felt.
    with clients_lock:
        for c in clients[:]:
            try:
                c.sendall(line)
            except OSError:               # gone, or stalled past its timeout
                clients.remove(c)
                c.close()


def hello():
    return {"type": "hello", **now_state, "model": ctx["model"], "ctx_used": ctx["used"],
            "ctx_max": ctx["max"], "id": cur["id"], "turns": cur["turns"],
            "pending": cur["pending"], "chat_model": cur["model"]}


def drop(c):
    with clients_lock:
        if c in clients:
            clients.remove(c)
    c.close()


def reader(c):
    # A timed-out recv is only an idle client; the socket stays usable.
    buf = b""
    while True:
        try:
            data = c.recv(65536)
        except TimeoutError:
            continue
        except OSError:
            break
        if not data:
            break
        buf += data
        if len(buf) > 1 << 20:            # a megabyte with no newline is not a command
            log("client line over 1 MB, dropped")
            emit({"type": "error", "message": "line too long"})
            buf = b""
            continue
        *lines, buf = buf.split(b"\n")
        for line in lines:
            if line.strip():
                handle(line)
    drop(c)


def serve(path=SOCK):
    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        probe.connect(path)
    except OSError:
        pass                              # missing, or left over from a crash
    else:
        # Unlinking a live socket would orphan the running daemon's clients.
        probe.close()
        raise SystemExit(f"another assistant is serving {path}")
    probe.close()
    try:
        os.unlink(path)
    except FileNotFoundError:
        pass
    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    old = os.umask(0o177)                 # 0600: only this user talks to it
    try:
        srv.bind(path)
    finally:
        os.umask(old)
    srv.listen()

    def accept():
        while True:
            try:
                c, _ = srv.accept()
            except OSError as e:          # EMFILE and friends: log, back off, keep serving
                log(f"accept failed: {e}")
                time.sleep(1)
                continue
            c.settimeout(2)
            # Under the lock: no event can fall between hello and the client
            # joining the list.
            with lock:
                try:
                    c.sendall((json.dumps(hello()) + "\n").encode())
                except OSError:
                    c.close()
                    continue
                with clients_lock:
                    clients.append(c)
            threading.Thread(target=reader, args=(c,), daemon=True).start()
    threading.Thread(target=accept, daemon=True).start()


def interrupt():
    # Caller holds lock. A reply in flight ends now: done marks it stopped,
    # its partial text is not saved, its thread notices the turn and returns.
    # Its connection is shut too: llama-server stops only when the client
    # hangs up, and a reply waiting for its first chunk through a long
    # prompt pass would otherwise keep the GPU busy until that chunk.
    global turn
    turn += 1
    if writing["on"]:                     # a write has no pending turn to close
        writing["on"] = False
        # Whatever ended it (stop, an ask, a new conversation): the canvas
        # waits for an outcome, so it gets one.
        emit({"type": "error", "message": "stopped", "turn": True})
        emit({"type": "state", "state": "idle"})
    conn = inflight["conn"]
    if conn is not None and conn.sock is not None:
        try:
            conn.sock.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass
    if cur["pending"] is not None:
        cur["pending"] = None
        emit({"type": "done", "id": cur["id"], "stopped": True})
        emit({"type": "state", "state": "idle"})


def start_new():
    interrupt()
    cur.update(id=None, history=[], turns=[], model=None)


def handle(raw):
    try:
        cmd = json.loads(raw)
        kind = cmd["type"]
    except (ValueError, TypeError, KeyError):
        log(f"bad command: {raw[:200]!r}")
        emit({"type": "error", "message": "bad command"})
        return
    # Ids are integers: SQLite would match "1" to row 1 while Python's
    # comparison with the current id would not, and a missing id would
    # equal an unsaved conversation's None.
    bad = ("bad id" if kind in ("open", "delete") and type(cmd.get("id")) is not int
           else "bad text" if kind == "ask" and not isinstance(cmd.get("text"), str)
           else "bad files" if kind == "ask" and not (
               isinstance(cmd.get("files", []), list)
               and all(isinstance(f, str) for f in cmd.get("files", [])))
           else "bad audio" if kind == "ask" and "audio" in cmd and not (
               isinstance(cmd["audio"], str) and not cmd["text"].strip())
           else "bad model" if kind == "ask" and "model" in cmd and not (
               isinstance(cmd["model"], str) and cmd["model"])
           else "bad query" if kind == "search" and not isinstance(cmd.get("q"), str)
           else "bad write" if kind == "write" and not (
               isinstance(cmd.get("files"), list) and cmd["files"]
               and all(isinstance(f, str) for f in cmd["files"])
               and isinstance(cmd.get("prompt"), str)
               and isinstance(cmd.get("dir"), str) and os.path.isabs(cmd["dir"])
               and isinstance(cmd.get("name"), str) and cmd["name"].strip(". ")
               and "/" not in cmd["name"])
           else None)
    if bad:
        # A refused write ends the canvas's wait for it, so it carries turn.
        emit({"type": "error", "message": bad, **({"turn": True} if kind == "write" else {})})
        return
    try:
        with lock:
            command(kind, cmd)
    except Exception:
        # A failed command must not kill the reader: the client would stay
        # connected, receiving events, with nothing reading what it sends.
        log(f"{kind} failed:\n" + traceback.format_exc().rstrip())
        emit({"type": "error", "message": f"{kind} failed"})


def command(kind, cmd):
    # Caller holds lock; arguments are already checked.
    if kind == "write":
        interrupt()
        threading.Thread(target=write, daemon=True,
                         args=(turn, cmd["files"], cmd["prompt"], cmd["dir"], cmd["name"])).start()
    elif kind == "ask":
        text, files, audio = cmd["text"].strip(), cmd.get("files", []), cmd.get("audio")
        if text or files or audio:
            interrupt()
            # A pick applies from this ask on and stays with the chat. One
            # that cannot be asked (no key) is refused here, so the chat keeps
            # the model it is stored with.
            model = cmd.get("model", cur["model"])
            try:
                route(model)
            except Refused as e:
                emit({"type": "error", "message": str(e), "turn": True})
                return
            cur["model"] = model
            threading.Thread(target=reply, args=(turn, text, files, audio, cur["model"]),
                             daemon=True).start()
    elif kind == "stop":
        interrupt()
    elif kind == "new":
        start_new()
        emit({"type": "conversation", "id": None, "turns": [], "model": None})
    elif kind == "list":
        emit({"type": "conversations", "items": list_convs(DB)})
    elif kind == "open":
        got = load_conv(DB, cmd["id"])
        if got is None:
            emit({"type": "error", "message": "no such conversation"})
            return
        interrupt()
        model = DB.execute("SELECT model FROM conversations WHERE id = ?",
                           (cmd["id"],)).fetchone()[0]
        cur.update(id=cmd["id"], history=got[0], turns=got[1], model=model)
        emit({"type": "conversation", "id": cur["id"], "turns": cur["turns"], "model": model})
    elif kind == "search":
        emit({"type": "results", "items": search(DB, cmd["q"])})
    elif kind == "delete":
        delete_conv(DB, cmd["id"])
        if cmd["id"] == cur["id"]:
            start_new()
            emit({"type": "conversation", "id": None, "turns": [], "model": None})
        emit({"type": "conversations", "items": list_convs(DB)})
    elif kind == "models":
        # Off the lock: the router can take seconds to answer.
        threading.Thread(target=send_models, daemon=True).start()
    else:
        log(f"unknown command: {kind!r}")
        emit({"type": "error", "message": f"unknown command {kind}"})


def context_size(model):
    # The router can swap models, so ask again whenever the id changes.
    if ctx["model"] != model:
        url = f"{CFG['llm'].removesuffix('/v1')}/props?model={urllib.parse.quote(model)}"
        try:
            with urllib.request.urlopen(url, timeout=5) as r:
                ctx["max"] = json.load(r)["default_generation_settings"]["n_ctx"]
        except (OSError, KeyError, ValueError):
            ctx["max"] = 0
        ctx["model"] = model
    return ctx["max"]


def run_tool(tools, name, arguments, d=TOOLS_DIR):
    # Whatever goes wrong becomes the tool's answer, so the model can say it
    # could not find out; a tool never takes the assistant down.
    t = tools.get(name)
    if t is None:
        return f"error: no tool named {name}"
    try:
        args = json.loads(arguments or "{}")
    except json.JSONDecodeError:
        args = None
    if not isinstance(args, dict):
        return "error: arguments are not valid JSON"
    vals = {p: "" if args.get(p) is None else str(args[p]) for p in t.get("params", {})}
    # No shell: each {name} fills in inside one argv element.
    argv = [re.sub(r"\{(\w+)\}", lambda m: vals.get(m.group(1), ""), a) for a in t["command"]]
    if "/" not in argv[0] and os.path.exists(os.path.join(d, argv[0])):
        argv[0] = os.path.join(d, argv[0])
    env = {**os.environ, **{k: str(v) for k, v in CFG["env"].items()}}
    try:
        # Own process group: a timeout kills the script's children too (curl),
        # which would otherwise keep the pipes open and block us.
        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                             env=env, start_new_session=True)
    except OSError as e:
        return f"error: cannot run {argv[0]}: {e.strerror}"
    try:
        out, err = p.communicate(timeout=t.get("timeout", 10))
    except subprocess.TimeoutExpired:
        os.killpg(p.pid, signal.SIGKILL)
        p.communicate()
        return "error: timed out"
    if p.returncode:
        last = err.strip().splitlines()[-1:]
        return ("error: " + (last[0] if last else f"exit status {p.returncode}"))[:2000]
    return out.strip()[:2000]


def collect_calls(calls, fragments):
    # A streamed tool call arrives in pieces keyed by index: id and name in
    # the first, the JSON arguments spread over the rest.
    for f in fragments:
        c = calls.setdefault(f["index"], {"id": "", "name": "", "arguments": ""})
        fn = f.get("function") or {}
        c["id"] = f.get("id") or c["id"]
        c["name"] = fn.get("name") or c["name"]
        c["arguments"] += fn.get("arguments") or ""


# Cloud providers, named in [[cloud]]. A key is key_cmd's stdout, read once
# at startup; a provider whose key_cmd fails is left out, so its models are
# not offered and a chat still on one is refused.
PROVIDERS = {}                            # name -> {"url", "key", "models"}


def provider_key(p):
    # Only the exit status is reported: a secret tool's stderr is not worth
    # the risk of it holding the secret.
    try:
        r = subprocess.run(shlex.split(p["key_cmd"]), capture_output=True, text=True,
                           stdin=subprocess.DEVNULL, timeout=10)
    except (OSError, ValueError, subprocess.TimeoutExpired) as e:
        raise RuntimeError(f"cloud {p['name']}: key_cmd failed ({type(e).__name__})") from None
    if r.returncode or not r.stdout.strip():
        raise RuntimeError(f"cloud {p['name']}: key_cmd failed (exit {r.returncode})")
    # The first line, as pass keeps it, with the entry's other lines after.
    # Anything not one clean token would go into a header, where http.client
    # refuses it with the whole value in the message.
    key = r.stdout.strip().splitlines()[0]
    if any(c.isspace() or not c.isprintable() for c in key):
        raise RuntimeError(f"cloud {p['name']}: key_cmd did not print a key on its first line")
    return key


def load_providers(cloud):
    out = {}
    for p in cloud:
        try:
            out[p["name"]] = {"url": p["url"].rstrip("/"), "key": provider_key(p),
                              "models": p["models"]}
        except RuntimeError as e:
            log(e)
    return out


def route(model):
    # (base url, key, the id the server knows, provider or None). A prefix
    # is a provider only when one by that name is configured, so a local id
    # with a colon stays local.
    name, sep, rest = (model or "").partition(":")
    if sep and name in PROVIDERS:
        if not rest:
            raise Refused(f"cloud {name}: no model named")
        p = PROVIDERS[name]
        return p["url"], p["key"], rest, name
    if sep and any(p["name"] == name for p in CFG["cloud"]):
        raise Refused(f"cloud {name}: no key (key_cmd failed at startup)")
    return CFG["llm"], None, model, None


def cloud_models(name):
    # For the settings page (--cloud-models): the provider's model ids,
    # fetched here so the key never reaches it.
    p = next((p for p in CFG["cloud"] if p["name"] == name), None)
    if p is None:
        raise SystemExit(f"no cloud provider named {name}")
    try:
        key = provider_key(p)
    except RuntimeError as e:
        raise SystemExit(str(e)) from None
    req = urllib.request.Request(p["url"].rstrip("/") + "/models",
                                 headers={"Authorization": f"Bearer {key}"})
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return sorted(m["id"] for m in json.load(r)["data"])
    except urllib.error.HTTPError as e:
        raise SystemExit(f"cloud {name} {e.code}: "
                         f"{e.read().decode(errors='replace')[:300]}") from None
    except (OSError, KeyError, TypeError, ValueError) as e:
        raise SystemExit(f"cloud {name}: {e}") from None


def send_models():
    # The canvas's model dropdown: the router's models, then every cloud
    # model whose provider's key loaded.
    try:
        with urllib.request.urlopen(CFG["llm"] + "/models", timeout=10) as r:
            local = [{"id": m["id"], "status": m["status"]["value"], "cloud": False}
                     for m in json.load(r)["data"]]
    except (OSError, KeyError, TypeError, ValueError):
        local = []                        # router down: the cloud ones still show
    emit({"type": "models", "items": local + [
        {"id": f"{n}:{m}", "provider": n, "cloud": True}
        for n, p in PROVIDERS.items() for m in p["models"]]})


def model_inputs(model=None):
    # (id, inputs) of the model a chat uses: its pick, else whatever the
    # router has resident (asking for another would swap it out under other
    # clients). inputs is None when not reported, as for every cloud model.
    if model and route(model)[3]:
        return model, None
    with urllib.request.urlopen(CFG["llm"] + "/models", timeout=10) as r:
        models = json.load(r)["data"]
    m = (next((m for m in models if m["id"] == model), None)
         or next((m for m in models if m["status"]["value"] != "unloaded"), models[0]))
    if model and m["id"] != model:
        log(f"no local model {model}; using {m['id']}")
    return m["id"], (m.get("architecture") or {}).get("input_modalities")


def llm_model():
    return model_inputs()[0]


def ask(messages, calls, tools=None, mine=None, think=None, meter=True, model=None):
    # Yields ("think", text) and ("token", text) as they stream; tool-call
    # fragments go into calls. mine is the reply's turn: a stop shuts the
    # connection from the other thread (interrupt), which ends the reading.
    # model is the id model_inputs gave, None for the router's loaded one.
    url, key, mid, cloud = route(model or llm_model())
    # Only the chat's own replies move the meter: a title or a write still
    # running for another chat must not overwrite it.
    if meter and cloud and ctx["model"] != model:
        ctx.update(model=model, max=0, used=0)   # no /props: tokens used only
    elif meter and not cloud:
        context_size(mid)
    body = {"model": mid, "messages": messages, "stream": True,
            "stream_options": {"include_usage": True}}
    if not cloud:
        # llama.cpp's own field; strict providers refuse unknown ones.
        body["chat_template_kwargs"] = {
            "enable_thinking": CFG["thinking"] if think is None else think}
    if tools:
        body["tools"] = tools
    if CFG["temperature"] is not None:
        body["temperature"] = CFG["temperature"]
    # http.client rather than urllib: it exposes the socket a stop must shut.
    u = urllib.parse.urlsplit(url + "/chat/completions")
    conn = (http.client.HTTPSConnection if u.scheme == "https" else http.client.HTTPConnection)(
        u.hostname, u.port, timeout=300)
    with lock:
        if mine is not None and turn != mine:
            return                        # stopped before the request went out
        inflight["conn"] = conn
    try:
        yield from stream(conn, u.path, body, calls, meter, key, cloud or "llama-server")
    except OSError:
        if mine is not None and turn != mine:
            return                        # the stop hung up under the read
        raise
    finally:
        with lock:
            if inflight["conn"] is conn:
                inflight["conn"] = None
        conn.close()


def stream(conn, path, body, calls, meter=True, key=None, who="llama-server"):
    headers = {"Content-Type": "application/json"}
    if key:
        headers["Authorization"] = f"Bearer {key}"
    conn.request("POST", path, json.dumps(body).encode(), headers)
    r = conn.getresponse()
    if r.status >= 400:
        # The body says why (a full context, a bad request, a spent quota);
        # the status alone does not.
        why = r.read().decode(errors="replace")
        if key:
            why = why.replace(key, "***")   # some providers echo the key back
        raise RuntimeError(f"{who} {r.status}: {why[:2000]}")
    with r:
        streamed, other = False, b""
        for line in r:
            if not line.startswith(b"data: "):
                # The router answers a crashed model with a plain line, e.g.
                # "proxy error: Failed to read connection", not a stream.
                other = other or line.strip()
                continue
            streamed = True
            if line.strip() == b"data: [DONE]":
                continue
            chunk = json.loads(line[6:])
            if "error" in chunk:          # a failure mid-stream arrives as a chunk
                err = chunk["error"]
                raise RuntimeError(err.get("message", str(err)) if isinstance(err, dict) else str(err))
            if chunk.get("usage") and meter:
                # How full the model's context is: history, tool results and
                # tool definitions all count.
                ctx["used"] = chunk["usage"]["prompt_tokens"]
                # The model too: the router can have swapped it since hello,
                # and a client showing the hello's name would show a stale one.
                emit({"type": "ctx", "used": ctx["used"], "max": ctx["max"], "model": ctx["model"]})
            choices = chunk["choices"]
            if not choices:
                continue
            delta = choices[0]["delta"]
            collect_calls(calls, delta.get("tool_calls") or [])
            # llama-server sends reasoning_content, OpenRouter reasoning.
            think = delta.get("reasoning_content") or delta.get("reasoning")
            if think:
                yield "think", think
            if delta.get("content"):
                yield "token", delta["content"]
        if not streamed:
            raise RuntimeError(other.decode(errors="replace")[:300] or f"{who} sent nothing")


def say(mine, kind, value, done=False):
    # Sends one piece of the reply, unless the reply went stale; False tells
    # the reply thread to give up.
    with lock:
        if turn != mine:
            return False
        p = cur["pending"]
        if kind == "think":
            p["think"] += value
            emit({"type": "think", "delta": value})
        elif kind == "token":
            p["reply"] += value
            emit({"type": "token", "delta": value})
        else:
            if not done:
                p["tools"].append(value)
            emit({"type": "tool", "name": value, "done": done})
            emit({"type": "state", "state": "thinking"} if done
                 else {"type": "state", "state": "tool", "tool": value})
        return True


def free_name(d, name):
    # The first of name, "stem (1).ext", "stem (2).ext" not taken in d.
    stem, ext = os.path.splitext(name)
    path, n = os.path.join(d, name), 1
    while os.path.lexists(path):
        path, n = os.path.join(d, f"{stem} ({n}){ext}"), n + 1
    return path


def unfence(text):
    # Models often wrap a whole document in one code fence; the file wants
    # what is inside it.
    lines = text.strip().split("\n")
    if (len(lines) >= 2 and lines[0].startswith("```") and lines[-1].strip() == "```"
            and not any(l.startswith("```") for l in lines[1:-1])):   # one fence, not two blocks
        return "\n".join(lines[1:-1]) + "\n"
    return text


def write(mine, files, prompt, d, name):
    # A one-off request whose reply goes to a file, for the canvas's To .md:
    # no history in, nothing kept in the conversation, no thinking, no tools.
    tmp = None
    try:
        # No history goes with a write, so half the whole context is its share.
        blocks, refs, _ = attach(files, ctx["max"] // 2 * 3 if ctx["max"] else 24000)
        # Always the local model, whatever the chat uses: a write has no chat.
        model, inputs = model_inputs()
        images = inputs is None or "image" in inputs
        if refs and not images:
            raise Refused(f"{model} cannot see images")
        with lock:
            if turn != mine:
                return
            writing["on"] = True
            emit({"type": "state", "state": "thinking"})
        body = "\n\n".join(blocks + [prompt])
        msgs = [{"role": "user", "content": [{"type": "text", "text": body}] + refs if refs else body}]
        text, shown = "", 0.0
        # meter=False: the chat's context meter is not about this request.
        for kind, piece in ask(wire(msgs, images), {}, None, mine, think=False, meter=False):
            if turn != mine:
                return
            if kind == "token":
                text += piece
                if time.time() - shown >= 1:
                    shown = time.time()
                    emit({"type": "progress", "chars": len(text)})
        if turn != mine:                  # stopped: the hang-up ended the stream
            return
        if not text.strip():
            raise Refused("the model gave an empty reply; nothing written")
        os.makedirs(d, exist_ok=True)
        tmp = os.path.join(d, f".write-{os.getpid()}-{mine}.tmp")
        with open(tmp, "w", encoding="utf-8") as f:
            f.write(unfence(text))
        with lock:
            if turn != mine:
                return
            path = free_name(d, name)
            os.rename(tmp, path)
            tmp = None
            writing["on"] = False
            emit({"type": "written", "path": path})
            emit({"type": "state", "state": "idle"})
        log(f"wrote {path}")
    except Exception as e:
        log(f"refused: {e}" if isinstance(e, Refused)
            else "write failed:\n" + traceback.format_exc().rstrip())
        with lock:
            if turn == mine:
                writing["on"] = False
                emit({"type": "error", "message": str(e) or type(e).__name__, "turn": True})
                emit({"type": "state", "state": "idle"})
    finally:
        if tmp is not None and os.path.exists(tmp):
            os.remove(tmp)


def transcribe(mine, path):
    # The model hears the clip. Asked to answer it directly, it replied in
    # Estonian or Spanish 5 times in 6; the words first, then a typed ask
    # (AGENTS.md, "Audio input"). History and search need the words anyway.
    if not os.path.isabs(path):
        raise Refused("audio path must be absolute")
    try:
        with wave.open(path) as w:
            rate = w.getframerate() * w.getnchannels() * w.getsampwidth()
        # From the size: a recording stopped by a signal has a header whose
        # length field was never filled in.
        secs = (os.path.getsize(path) - 44) / rate
    except (OSError, EOFError, wave.Error, ZeroDivisionError):
        raise Refused(f"not a readable WAV: {os.path.basename(path)}") from None
    if secs > MAX_AUDIO_S:
        raise Refused("recording too long")
    model, inputs = model_inputs()
    if inputs is not None and "audio" not in inputs:
        raise Refused(f"{model} cannot hear audio")
    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    with lock:
        if turn != mine:
            return None
        # A turn in flight from here: a stop ends it with done, as a reply.
        cur["pending"] = {"user": "", "reply": "", "think": "", "tools": [], "files": []}
        emit({"type": "state", "state": "transcribing"})
    msgs = [{"role": "user", "content": [
        {"type": "text", "text": HEAR},
        {"type": "input_audio", "input_audio": {"data": b64, "format": "wav"}}]}]
    text = "".join(p for k, p in ask(msgs, {}, None, mine, think=False, meter=False)
                   if k == "token").strip()
    with lock:
        if turn != mine:
            return None
    if not text:
        raise Refused("heard nothing")
    log(f"heard: {text}")
    return text


def reply(mine, text, files=(), audio=None, model=None):
    try:
        if audio:
            # Always the local model: the audio never leaves, only the words.
            text = transcribe(mine, audio)
            if text is None:
                return
        answer(mine, text, files, model)
    except Exception as e:
        # A refused file is the user's to fix, not a bug: one line, no trace.
        log(f"refused: {e}" if isinstance(e, Refused)
            else "reply failed:\n" + traceback.format_exc().rstrip())
        with lock:
            if turn == mine:
                cur["pending"] = None
                # turn: this error ended the reply. Command errors carry no
                # turn and must not end a reply another client is watching.
                emit({"type": "error", "message": str(e) or type(e).__name__, "turn": True})
                emit({"type": "state", "state": "idle"})


def answer(mine, text, files=(), chosen=None):
    blocks, refs, chips = [], [], []
    if files:
        # Half of what the context has free, at about 3 characters a token.
        free = (ctx["max"] - ctx["used"]) // 2 * 3 if ctx["max"] else 24000
        blocks, refs, chips = attach(files, max(free, 2000))
    # A server that does not report inputs is left to decide for itself.
    model, inputs = model_inputs(chosen)
    images = inputs is None or "image" in inputs
    # A cloud chat is offered only the tools marked cloud = true: the others
    # would send mail, money or calendar to the provider. One of them called
    # anyway (copied from history) answers "no tool named".
    offered = ({n: t for n, t in TOOLS.items() if t.get("cloud") is True}
               if route(model)[3] else TOOLS)
    offer = schema(offered) or None
    if refs and not images:
        raise Refused(f"{model} cannot see images")
    with lock:
        if turn != mine:
            return
        history = list(cur["history"])
        cur["pending"] = {"user": text, "reply": "", "think": "", "tools": [], "files": chips}
        emit({"type": "user", "text": text, "files": chips})
        emit({"type": "state", "state": "thinking"})
    log(f"you: {text}" + "".join(f" [{c['name']}]" for c in chips))
    body = "\n\n".join(blocks + [text] if text else blocks)
    question = {"role": "user", "content": [{"type": "text", "text": body}] + refs if refs else body}
    # The model has no clock: without the date it makes one up for tools.
    # Also stamped on the question itself, which history does not keep: with
    # the time only in the system prompt, "what time is it?" asked again an
    # hour later copied the old answer from history. Stamping every message
    # in history fixed that but lost tool arguments (account filter 0/6).
    now = time.strftime("%A %d %B %Y, %H:%M")
    # ponytail: last 10 turns, summarise if long sessions need memory
    msgs = ([{"role": "system", "content": f"{CFG['system']} It is {now}."}]
            + [m for past in history[-10:] for m in past]
            + [{"role": "user", "content": [{"type": "text", "text": f"[{now}] {body}"}] + refs
                if refs else f"[{now}] {body}"}])
    first = len(msgs)                            # this turn's tool messages start here
    # Resolved once per answer: images re-read and pages re-rendered on every
    # tool round would cost seconds between rounds, and a file changed
    # mid-answer would flip from image to gone.
    sent = wire(msgs, images)
    said, used = "", []
    # A stop shuts this reply's connection (interrupt), so llama-server's slot
    # frees even mid prompt pass. ponytail: a running tool still runs out.
    for rnd in range(CFG["tool_rounds"] + 1):
        calls, content = {}, ""
        # The last round offers no tools, which forces an answer.
        for kind, piece in ask(sent, calls, offer if rnd < CFG["tool_rounds"] else None,
                               mine, model=model):
            if not say(mine, kind, piece):
                log("interrupted")
                return
            if kind == "token":
                content += piece
        said += content
        if not calls:
            break
        call = {"role": "assistant", "content": content, "tool_calls": [
            {"id": c["id"], "type": "function",
             "function": {"name": c["name"], "arguments": c["arguments"]}}
            for c in calls.values()]}
        msgs.append(call)
        sent.append(call)
        # ponytail: a stop does not kill a running tool, its result is dropped
        for c in calls.values():
            if not say(mine, "tool", c["name"]):
                log("interrupted")
                return
            used.append(c["name"])
            result = run_tool(offered, c["name"], c["arguments"])
            log(f"tool {c['name']}({c['arguments']}): {result[:120]!r}")
            msgs.append({"role": "tool", "tool_call_id": c["id"], "content": result})
            sent.append(msgs[-1])
            if not say(mine, "tool", c["name"], done=True):
                log("interrupted")
                return
    # History keeps the tool calls: with only what was said, the model saw
    # facts answered without tools and invented the next ones (no tool call
    # 9/10 on a replayed session, 0/10 with the calls kept). Results are
    # replaced by a placeholder, for context: same 10/10, and a follow-up on
    # an old result calls the tool again (10/10). A stopped exchange is
    # dropped whole, so user and assistant still alternate.
    past = [{**m, "content": "(shown to the user at the time)"} if m["role"] == "tool" else m
            for m in msgs[first:]]
    entry = [question] + past + [{"role": "assistant", "content": said}]
    with lock:
        if turn != mine:
            log("interrupted")
            return
        fresh = cur["id"] is None
        cur["id"] = save_turn(DB, cur["id"], text, said, used, entry, chips, chosen)
        conv = cur["id"]
        cur["history"].append(entry)
        cur["turns"].append({"user": text, "reply": said, "tools": used, "files": chips})
        cur["pending"] = None
        emit({"type": "done", "id": cur["id"]})
        emit({"type": "state", "state": "idle"})
    log("assistant:", said)
    if fresh:
        retitle(mine, conv, text, said, model)


def retitle(mine, conv, user, said, model=None):
    # The first six words stand until the model names the conversation. Asked
    # after done, under the reply's turn: the next ask, stop, new or open
    # hangs it up, so it never holds a reply back, and the six words stay.
    try:
        msgs = [{"role": "user",
                 "content": f"{TITLE}\n\nUser: {user[:2000]}\n\nAssistant: {said[:2000]}"}]
        text = "".join(p for k, p in ask(msgs, {}, None, mine, think=False, meter=False,
                                         model=model)
                       if k == "token")
        line = next((l for l in text.splitlines() if l.strip()), "")
        title = line.strip().strip("\"'*#").strip()[:80]
        with lock:
            if turn != mine or not title:
                return
            with DB:
                DB.execute("UPDATE conversations SET title = ? WHERE id = ?", (title, conv))
        log(f"titled: {title}")
    except Exception:                     # the reply is saved; a title is not worth an error
        log("title failed:\n" + traceback.format_exc().rstrip())


def main():
    global DB
    # SIGTERM runs the finally below, so the socket is not left behind.
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    served = False
    try:
        DB = db_open(DB_PATH)
        PROVIDERS.update(load_providers(CFG["cloud"]))
        try:
            context_size(llm_model())     # so the first hello has it
        except Exception:
            pass                          # llama-server down: asked again per request
        serve()
        served = True
        log(f"ready on {SOCK}")
        while True:
            signal.pause()
    except (KeyboardInterrupt, SystemExit) as e:
        log(f"exit: {type(e).__name__} {e}".rstrip())
        if not served:
            raise SystemExit(1)           # refused or failed before serving: not a clean stop
    except BaseException:
        log("crashed:\n" + traceback.format_exc().rstrip())
        raise
    finally:
        if served and os.path.exists(SOCK):   # never another daemon's socket
            os.remove(SOCK)


def test():
    import contextlib
    import http.server
    import io
    import tempfile

    c = merge(DEFAULTS, {"tool_rounds": 1, "env": {"K": "v"}})
    assert c["tool_rounds"] == 1 and c["env"] == {"K": "v"}
    assert DEFAULTS["tool_rounds"] == 3                                 # defaults untouched
    for bad in ({"card": "X"}, {"whisper": {"port": 1}}):       # voice-era keys are gone
        try:
            merge(DEFAULTS, bad)
            assert False, bad
        except SystemExit as e:
            assert "config:" in str(e)
    good = {"name": "or", "url": "https://llm.example.org/v1", "key_cmd": "echo k",
            "models": ["a/b"]}
    assert merge(DEFAULTS, {"cloud": [good]})["cloud"] == [good]
    assert merge(DEFAULTS, {"cloud": [{**good, "models": []}]})   # before the first fetch
    assert DEFAULTS["cloud"] == []
    for bad in ({"cloud": good}, {"cloud": [{**good, "extra": 1}]},
                {"cloud": [{**good, "name": "a:b"}]}, {"cloud": [good, good]},
                {"cloud": [{**good, "models": "a/b"}]}, {"cloud": [{**good, "url": ""}]}):
        try:
            merge(DEFAULTS, bad)
            assert False, bad
        except SystemExit as e:
            assert "config: cloud" in str(e), (bad, e)
    with tempfile.TemporaryDirectory() as d:
        with open(os.path.join(d, "x.toml"), "w") as f:
            f.write('description = "d"\ncommand = ["date"]\ncloud = "yes"\n')
        try:
            load_tools(d)
            assert False
        except SystemExit as e:
            assert "cloud must be true or false" in str(e)
    with tempfile.TemporaryDirectory() as d:
        with open(os.path.join(d, "echo.sh"), "w") as f:
            f.write('#!/bin/sh\nprintf "%s|" "$@"\n')
        os.chmod(os.path.join(d, "echo.sh"), 0o755)
        with open(os.path.join(d, "now.toml"), "w") as f:
            f.write('description = "d"\ncommand = ["date"]\n'
                    '[params.x]\ntype = "string"\ndescription = "px"\n')
        assert load_tools(d) == {"now": {"description": "d", "command": ["date"],
                                         "params": {"x": {"type": "string", "description": "px"}}}}
        assert schema(load_tools(d)) == [{"type": "function", "function": {
            "name": "now", "description": "d", "parameters": {"type": "object", "properties": {
                "x": {"type": "string", "description": "px"}}}}}]
        tools = {"echo": {"command": ["echo.sh", "{a}", "x{b}y"], "params": {"a": {}, "b": {}}},
                 "slow": {"command": ["sh", "-c", "sleep 5; true"], "timeout": 0.3},
                 "fail": {"command": ["sh", "-c", "echo oops >&2; exit 3"]},
                 "gone": {"command": ["no-such-binary-here"]}}
        # A value is one argv element, never a shell command or a second argument.
        assert run_tool(tools, "echo", '{"a": "x; rm -rf ~", "b": 2}', d) == "x; rm -rf ~|x2y|"
        assert run_tool(tools, "echo", "{}", d) == "|xy|"
        t0 = time.time()
        assert run_tool(tools, "slow", "{}", d) == "error: timed out"
        assert time.time() - t0 < 2                 # the grandchild sleep was killed too
        assert run_tool(tools, "fail", "{}", d) == "error: oops"
        assert run_tool(tools, "gone", "{}", d).startswith("error: cannot run")
        assert run_tool(tools, "nope", "{}", d) == "error: no tool named nope"
        assert run_tool(tools, "echo", "{bad", d) == "error: arguments are not valid JSON"
        assert run_tool(tools, "echo", "[1]", d) == "error: arguments are not valid JSON"
    calls = {}
    collect_calls(calls, [{"index": 0, "id": "c1", "type": "function",
                           "function": {"name": "weather", "arguments": "{"}}])
    for piece in ['"place', '":', '"', "Rome", '"', "}"]:
        collect_calls(calls, [{"index": 0, "function": {"arguments": piece}}])
    collect_calls(calls, [{"index": 1, "id": "c2", "function": {"name": "now", "arguments": "{}"}}])
    assert calls == {0: {"id": "c1", "name": "weather", "arguments": '{"place":"Rome"}'},
                     1: {"id": "c2", "name": "now", "arguments": "{}"}}

    # Storage.
    db = db_open(":memory:")
    a = save_turn(db, None, "che tempo fa in città oggi pomeriggio?", "Piove.", ["weather"],
                  [{"role": "user", "content": "q"}])
    assert save_turn(db, a, "e domani?", "Sole.", [], [{"role": "user", "content": "q2"}]) == a
    b = save_turn(db, None, "how do I list files", "Use `ls`.", [], [])
    assert [x["id"] for x in list_convs(db)] == [b, a]
    assert list_convs(db)[1]["title"] == "che tempo fa in città oggi"   # first six words
    hist, shown = load_conv(db, a)
    assert hist == [[{"role": "user", "content": "q"}], [{"role": "user", "content": "q2"}]]
    assert shown == [{"user": "che tempo fa in città oggi pomeriggio?", "reply": "Piove.",
                      "tools": ["weather"], "files": []},
                     {"user": "e domani?", "reply": "Sole.", "tools": [], "files": []}]
    assert load_conv(db, 999) is None
    m = save_turn(db, None, "q", "r", [], [], model="fake:x")
    pick = lambda: db.execute("SELECT model FROM conversations WHERE id = ?", (m,)).fetchone()[0]
    assert pick() == "fake:x"
    save_turn(db, m, "q", "r", [], [], model="local")
    assert pick() == "local"
    save_turn(db, m, "q", "r", [], [])
    assert pick() is None                            # an ask with no choice: the loaded one
    delete_conv(db, m)
    hits = search(db, "citta")                       # no accent typed, accented word found
    assert [h["id"] for h in hits] == [a] and "**città**" in hits[0]["snippet"]
    assert [h["id"] for h in search(db, "cit")] == [a]                 # prefix, as you type
    assert [h["id"] for h in search(db, "piove")] == [a]               # replies are indexed
    for q in ('"', "*", 'a"b', "NOT", "x OR y", "ls)", "", "a\0b"):    # FTS syntax is text
        search(db, q)
    assert search(db, "piove OR ls") == []          # unquoted, this would match both
    delete_conv(db, a)
    assert load_conv(db, a) is None and search(db, "citta") == []
    # Raises if the index still holds the deleted turns: the cascade must
    # have fired the delete trigger.
    db.execute("INSERT INTO turns_fts(turns_fts, rank) VALUES ('integrity-check', 1)")
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "sub", "h.db")
        os.makedirs(os.path.dirname(path), mode=0o755)
        open(path, "w").close()
        os.chmod(path, 0o644)                       # made earlier by something else
        db_open(path).close()
        assert os.stat(os.path.dirname(path)).st_mode & 0o777 == 0o700
        assert os.stat(path).st_mode & 0o777 == 0o600

    # Attachments.
    import struct
    import zlib

    def tiny_pdf(path, pages):
        # One Helvetica line per page; "" is a page with no text, a scan. A
        # page may be (text, width, height) in points.
        objs = ["<< /Type /Catalog /Pages 2 0 R >>",
                "<< /Type /Pages /Kids [%s] /Count %d >>"
                % (" ".join(f"{4 + 2 * i} 0 R" for i in range(len(pages))), len(pages)),
                "<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>"]
        for i, t in enumerate(pages):
            t, w, h = t if isinstance(t, tuple) else (t, 612, 792)
            stream = f"BT /F1 24 Tf 72 700 Td ({t}) Tj ET" if t else ""
            objs += [f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 {w} {h}] /Resources "
                     f"<< /Font << /F1 3 0 R >> >> /Contents {5 + 2 * i} 0 R >>",
                     f"<< /Length {len(stream)} >>\nstream\n{stream}\nendstream"]
        out, offsets = b"%PDF-1.4\n", []
        for k, o in enumerate(objs, 1):
            offsets.append(len(out))
            out += f"{k} 0 obj\n{o}\nendobj\n".encode()
        xref = len(out)
        out += (f"xref\n0 {len(objs) + 1}\n0000000000 65535 f \n".encode()
                + b"".join(f"{o:010d} 00000 n \n".encode() for o in offsets)
                + f"trailer\n<< /Size {len(objs) + 1} /Root 1 0 R >>\n"
                  f"startxref\n{xref}\n%%EOF\n".encode())
        with open(path, "wb") as f:
            f.write(out)

    def tiny_png(path, rgb=b"\xff\x00\x00"):
        def chunk(t, d):
            return struct.pack(">I", len(d)) + t + d + struct.pack(">I", zlib.crc32(t + d))
        with open(path, "wb") as f:
            f.write(b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 2, 0, 0, 0))
                    + chunk(b"IDAT", zlib.compress(b"\x00" + rgb)) + chunk(b"IEND", b""))

    files_dir = tempfile.TemporaryDirectory()
    fd = files_dir.name
    code, png, pdf, other = (os.path.join(fd, n) for n in ("a.py", "red.png", "doc.pdf", "x.bin"))
    with open(code, "w") as f:
        f.write("print('hi')\n" * 50)
    tiny_png(png)
    tiny_pdf(pdf, ["hello pdf", ("", 3000, 3000)])
    with open(other, "wb") as f:
        f.write(b"\x00\x01")
    assert [kind_of(p) for p in (code, png, pdf, other, "/x/README.md", "/x/b.SlackBuild")] == \
        ["text", "image", "pdf", "other", "text", "text"]
    # No suffix: the content decides. A script or plain UTF-8 is text; a NUL
    # byte, or bytes that are not UTF-8, are not.
    for name, data, want in (("run", b"#!/bin/bash\necho hi\n", "text"),
                             ("notes", "caff\u00e8 alle 9\n".encode(), "text"),
                             ("blob", b"\x7fELF\x00\x01", "other"),
                             ("latin", "caff\u00e8".encode("latin-1"), "other")):
        with open(os.path.join(fd, name), "wb") as f:
            f.write(data)
        assert kind_of(os.path.join(fd, name)) == want, name
    assert kind_of(os.path.join(fd, "missing")) == "other"
    with open(os.path.join(fd, "long"), "wb") as f:      # a character cut by the window
        f.write(b"a" * 4095 + "\u00e8 tail".encode())
    assert kind_of(os.path.join(fd, "long")) == "text"
    blocks, refs, chips = attach([code, pdf, png], 100000)
    assert blocks[0].startswith("--- file: a.py ---\nprint('hi')") and "(truncated)" not in blocks[0]
    assert "[page 1]\nhello pdf" in blocks[1] and "[page 2]" in blocks[1]
    assert chips == [{"name": "a.py", "kind": "text", "truncated": False},
                     {"name": "doc.pdf", "kind": "pdf", "truncated": False, "pages": 2},
                     {"name": "red.png", "kind": "image", "truncated": False}]
    # The blank page is a scan: it goes as an image, rendered when sent.
    assert [r["name"] for r in refs] == ["doc.pdf page 2", "red.png"] and refs[0]["page"] == 2
    sent = [resolve(r) for r in refs]
    assert all(x["type"] == "image_url" for x in sent)
    assert sent[1]["image_url"]["url"].startswith("data:image/png;base64,")
    page = base64.b64decode(sent[0]["image_url"]["url"].split(",", 1)[1])
    # Page 2, the square one, bounded to PAGE_PX: page 1 would not be square.
    w, h = struct.unpack(">II", page[16:24])
    assert page[:4] == b"\x89PNG" and w == h and PAGE_PX - 2 <= w <= PAGE_PX, (w, h)
    scans = os.path.join(fd, "scans.pdf")
    tiny_pdf(scans, ["cover"] + [("", 300, 150)] * 6)
    b, r, c = attach([scans], 100000)
    assert len(r) == 4 and c[0]["truncated"] and "[scanned pages not shown: 6, 7]" in b[0]
    small = base64.b64decode(resolve(r[0])["image_url"]["url"].split(",", 1)[1])
    # A small page stays at 150 dpi, not raised to the bound: 300 pt is 625 px.
    assert struct.unpack(">II", small[16:24]) == (625, 313), struct.unpack(">II", small[16:24])
    with open(os.path.join(fd, "exact.txt"), "w") as f:
        f.write("x" * 24)
    with open(os.path.join(fd, "over.txt"), "w") as f:
        f.write("x" * 25)
    for n, cut in (("exact.txt", False), ("over.txt", True)):
        b, _, c = attach([os.path.join(fd, n)], 24)
        assert c[0]["truncated"] is cut and b[0].split("\n")[1] == "x" * 24, n
    with open(os.path.join(fd, "a\nb.txt"), "w") as f:
        f.write("x")
    b, _, c = attach([os.path.join(fd, "a\nb.txt")], 100)
    assert c[0]["name"] == "a b.txt" and b[0].startswith("--- file: a b.txt ---\n")
    small = attach([code], 24)[0][0]                 # the share is the cut: two lines
    assert "(truncated)" in small and small.count("print") == 2
    for bad, why in ((["a.py"], "cannot read a.py"), ([fd + "/nope.txt"], "cannot read nope.txt"),
                     ([fd], "cannot read"), ([other], "x.bin: chat supports text, images and PDF"),
                     ([code] * 9, "at most 8 files")):
        try:
            attach(bad, 1000)
            assert False, bad
        except Refused as e:
            assert why in str(e), (bad, e)
    # Stored as a reference: survives a save and reload, gone once changed.
    a = save_turn(db, None, "", "A red pixel.", [], [{"role": "user", "content": refs[1:]}],
                  chips[2:])
    hist, shown = load_conv(db, a)
    assert shown[0]["files"] == chips[2:] and list_convs(db)[0]["title"] == "red.png"
    assert wire(hist[0])[0]["content"][0]["type"] == "image_url"
    assert wire(hist[0], images=False)[0]["content"][0] == {   # a text-only model now
        "type": "text", "text": "[image red.png: this model cannot see images]"}
    tiny_png(png, b"\x00\xff\x00")
    assert wire(hist[0])[0]["content"][0] == {
        "type": "text", "text": "[image red.png is no longer available]"}
    os.remove(png)
    assert resolve(refs[1])["type"] == "text"        # deleted reads the same
    # An old database gains the files column.
    with tempfile.TemporaryDirectory() as d:
        old = sqlite3.connect(os.path.join(d, "h.db"))
        old.executescript(SCHEMA)
        old.close()
        assert "files" in {r[1] for r in db_open(os.path.join(d, "h.db"))
                           .execute("PRAGMA table_info(turns)")}
        assert "model" in {r[1] for r in db_open(os.path.join(d, "h.db"))
                           .execute("PRAGMA table_info(conversations)")}

    # Protocol, against a fake llama-server.
    global DB, TOOLS
    script = {"replies": [], "delay": 0, "inputs": ["text", "image"]}

    class Fake(http.server.BaseHTTPRequestHandler):
        def log_message(self, *a):
            pass

        def do_GET(self):
            script["auth"] = self.headers.get("Authorization")
            body = ({"data": [{"id": "m", "status": {"value": "loaded"},
                               "architecture": {"input_modalities": script["inputs"]}}]}
                    if self.path.endswith("/models")
                    else {"default_generation_settings": {"n_ctx": 4096}})
            self.send_response(200)
            self.end_headers()
            self.wfile.write(json.dumps(body).encode())

        def do_POST(self):
            script["auth"] = self.headers.get("Authorization")
            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
            if str(body["messages"][0]["content"]).startswith(TITLE):
                # Answered apart, so a title never takes a reply meant for an ask.
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"data: " + json.dumps({"choices": [{"delta": {
                    "content": script.get("title", "")}}]}).encode() + b"\n\ndata: [DONE]\n\n")
                return
            script.get("paths", []).append(self.path)
            script["body"] = body
            script.get("bodies", []).append(script["body"])
            if script.get("hold"):
                # A long prompt pass: nothing is sent, so only the client
                # hanging up ends it. Records how soon that happened.
                import select
                t0 = time.time()
                while time.time() - t0 < 5:
                    if select.select([self.connection], [], [], 0.05)[0]:
                        if not self.connection.recv(1, socket.MSG_PEEK):
                            script["hung_up"] = time.time()
                            return
            self.send_response(200)
            self.end_headers()
            try:
                reply = script["replies"].pop(0)
                for delta in reply:
                    self.wfile.write(delta.encode() if isinstance(delta, str) else
                                     b"data: " + json.dumps({"choices": [{"delta": delta}]})
                                     .encode() + b"\n\n")
                    self.wfile.flush()
                    time.sleep(script["delay"])
                # A plain line stands alone, as the router sends it.
                if not any(isinstance(d, str) and not d.startswith("data:") for d in reply):
                    self.wfile.write(b'data: {"choices": [], "usage": {"prompt_tokens": 42}}\n\n'
                                     b"data: [DONE]\n\n")
            except BrokenPipeError:              # a stopped reply hangs up mid-stream
                pass

    srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fake)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    CFG["llm"] = f"http://127.0.0.1:{srv.server_address[1]}/v1"
    # Cloud providers: the key from key_cmd, models fetched with it. The
    # fake answers /cloud/v1 like /v1, so the path tells them apart.
    base = f"http://127.0.0.1:{srv.server_address[1]}"
    CFG["cloud"] = [{"name": "fake", "url": base + "/cloud/v1/", "key_cmd": "echo sk-test",
                     "models": ["vendor/x:free"]},
                    {"name": "broken", "url": base + "/v1", "key_cmd": "false", "models": ["y"]}]
    PROVIDERS.update(load_providers(CFG["cloud"]))
    assert set(PROVIDERS) == {"fake"} and PROVIDERS["fake"]["key"] == "sk-test"
    assert route("fake:vendor/x:free") == (base + "/cloud/v1", "sk-test", "vendor/x:free", "fake")
    assert route("Gemma:q4") == (CFG["llm"], None, "Gemma:q4", None)   # no such provider
    try:
        route("broken:y")
        assert False
    except Refused as e:
        assert "no key" in str(e)
    try:
        route("fake:")
        assert False
    except Refused as e:
        assert "no model" in str(e)
    # pass prints the key on the first line and the entry's other lines after.
    more = [{"name": n, "url": base, "key_cmd": c, "models": []}
            for n, c in (("pass", "printf 'sk-x\\nlogin: y\\n'"), ("spaced", "printf 'a b'"))]
    assert {n: p["key"] for n, p in load_providers(more).items()} == {"pass": "sk-x"}
    assert cloud_models("fake") == ["m"] and script["auth"] == "Bearer sk-test"
    for name, why in (("broken", "key_cmd failed"), ("nope", "no cloud provider")):
        try:
            cloud_models(name)
            assert False, name
        except SystemExit as e:
            assert why in str(e), (name, e)
    DB = db_open(":memory:")
    TOOLS = {"now": {"description": "d", "command": ["sh", "-c", "echo noon"]}}
    with tempfile.TemporaryDirectory() as d:
        path = os.path.join(d, "s.sock")
        serve(path)
        assert os.stat(path).st_mode & 0o777 == 0o600
        cl = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        cl.connect(path)
        cl.settimeout(5)
        f = cl.makefile("rb")

        def send(**cmd):
            cl.sendall((json.dumps(cmd) + "\n").encode())

        def until(kind):
            seen = []
            while True:
                e = json.loads(f.readline())
                seen.append(e)
                if e["type"] == kind:
                    return seen

        h = json.loads(f.readline())
        assert h["type"] == "hello" and h["state"] == "idle" and h["turns"] == []
        assert h["pending"] is None and "ctx_max" in h
        assert h["chat_model"] is None
        # Thinking, a tool round, then the answer.
        script["replies"] = [
            [{"reasoning_content": "hmm"},
             {"tool_calls": [{"index": 0, "id": "c1", "function": {"name": "now",
                                                                   "arguments": "{}"}}]}],
            [{"content": "It is "}, {"content": "noon."}]]
        script["title"] = '"Ora attuale"\n'
        send(type="ask", text="what time is it")
        got = [(e["type"], e.get("delta") or e.get("name") or e.get("state") or e.get("text"))
               for e in until("done") if e["type"] != "ctx"]
        assert got == [("user", "what time is it"), ("state", "thinking"), ("think", "hmm"),
                       ("tool", "now"), ("state", "tool"), ("tool", "now"),
                       ("state", "thinking"), ("token", "It is "), ("token", "noon."),
                       ("done", None)], got
        assert until("state")[-1]["state"] == "idle"
        assert script["body"]["chat_template_kwargs"]["enable_thinking"] is True
        conv = cur["id"]
        for _ in range(40):                          # the title comes after done
            if list_convs(DB)[0]["title"] == "Ora attuale":
                break
            time.sleep(0.05)
        assert list_convs(DB)[0]["title"] == "Ora attuale", list_convs(DB)
        script["title"] = ""
        assert load_conv(DB, conv)[1] == [{"user": "what time is it", "reply": "It is noon.",
                                           "tools": ["now"], "files": []}]
        # The tool result went to the model but history keeps a placeholder.
        assert load_conv(DB, conv)[0][0][2] == {"role": "tool", "tool_call_id": "c1",
                                                "content": "(shown to the user at the time)"}
        # A plain line instead of a stream (the router, when the model died)
        # is an error carrying that line, not an empty reply.
        script["replies"] = [["proxy error: Failed to read connection\n"]]
        send(type="ask", text="anyone?")
        e = until("error")[-1]
        assert e["message"] == "proxy error: Failed to read connection" and e["turn"] is True
        # An error chunk mid-stream reaches the client with its message.
        script["replies"] = [['data: {"error": {"message": "context full"}}\n\n']]
        CFG["thinking"] = False
        send(type="ask", text="too much")
        e = until("error")[-1]
        assert e["message"] == "context full" and e["turn"] is True
        assert script["body"]["chat_template_kwargs"]["enable_thinking"] is False
        CFG["thinking"] = True
        # Stop while the server is still reading the prompt: the daemon hangs
        # up at once rather than at the first chunk, which frees the GPU.
        script.update(hold=True, hung_up=None)
        script["replies"] = [[]]
        send(type="ask", text="a long prompt")
        until("state")                               # thinking: the request is out
        time.sleep(0.3)
        stopped = time.time()
        send(type="stop")
        assert until("done")[-1].get("stopped") is True
        for _ in range(40):
            if script["hung_up"]:
                break
            time.sleep(0.05)
        assert script["hung_up"] and script["hung_up"] - stopped < 1, script["hung_up"]
        script["hold"] = False
        # Stop mid-stream: no token after done, nothing saved.
        script["replies"] = [[{"content": f"w{i} "} for i in range(50)]]
        script["delay"] = 0.02
        send(type="ask", text="long one")
        until("token")
        c3 = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        c3.connect(path)                         # mid-reply: hello carries the partial
        h3 = json.loads(c3.makefile("rb").readline())
        assert h3["pending"]["user"] == "long one" and h3["pending"]["reply"].startswith("w0")
        c3.close()
        send(type="stop")
        seen = until("done")
        assert seen[-1].get("stopped") is True
        time.sleep(0.3)                              # the reply thread notices and quits
        send(type="list")
        late = until("conversations")
        assert not [e for e in late if e["type"] == "token"], late
        assert len(load_conv(DB, conv)[1]) == 1
        script["delay"] = 0
        # History browsing.
        send(type="new")
        assert until("conversation")[-1] == {"type": "conversation", "id": None, "turns": [],
                                             "model": None}
        send(type="list")
        assert [x["id"] for x in until("conversations")[-1]["items"]] == [conv]
        send(type="search", q="tim")
        assert until("results")[-1]["items"][0]["id"] == conv
        send(type="open", id=conv)
        assert until("conversation")[-1]["turns"][0]["reply"] == "It is noon."
        send(type="open", id=999)
        assert until("error")[-1]["message"] == "no such conversation"
        cl.sendall(b"not json\n")
        assert until("error")[-1]["message"] == "bad command"
        for cmd, why in (({"type": "open", "id": [1]}, "bad id"),
                         ({"type": "open", "id": str(conv)}, "bad id"),
                         ({"type": "delete"}, "bad id"),
                         ({"type": "ask", "text": None}, "bad text"),
                         ({"type": "search", "q": 3}, "bad query")):
            send(**cmd)
            e = until("error")[-1]
            assert e["message"] == why and "turn" not in e, cmd      # a command error
        assert cur["id"] == conv                    # none of them touched the conversation
        real, DB = DB, None                          # every query now raises
        send(type="list")
        assert until("error")[-1]["message"] == "list failed"
        DB = real
        send(type="list")                            # the reader survived it
        assert [x["id"] for x in until("conversations")[-1]["items"]] == [conv]
        try:
            serve(path)                              # a second daemon on a live socket
            assert False
        except SystemExit as e:
            assert "another assistant" in str(e)
        send(type="list")                            # and the first one still serves
        until("conversations")
        send(type="delete", id=conv)
        seen = until("conversations")
        assert {"type": "conversation", "id": None, "turns": [], "model": None} in seen   # was current
        assert seen[-1]["items"] == []
        # Files: the text inline, the image resolved, chips on the turn.
        tiny_png(png)
        script["replies"] = [[{"content": "Red."}]]
        send(type="ask", text="what is it?", files=[pdf, png])
        user = until("user")[-1]
        assert user["files"][1] == {"name": "red.png", "kind": "image", "truncated": False}
        until("done")
        q = script["body"]["messages"][-1]["content"]
        assert "hello pdf" in q[0]["text"] and q[0]["text"].endswith("what is it?")
        assert [x["type"] for x in q] == ["text", "image_url", "image_url"]
        assert cur["turns"][-1]["files"] == user["files"]
        # Only files, no text.
        script["replies"] = [[{"content": "Code."}]]
        send(type="ask", text="", files=[code])
        assert until("user")[-1]["text"] == ""
        until("done")
        # A model without vision is refused before anything is sent.
        script["inputs"] = ["text"]
        send(type="ask", text="see?", files=[png])
        e = until("error")[-1]
        assert e["message"] == "m cannot see images" and e["turn"] is True
        script["inputs"] = ["text", "image"]
        send(type="ask", text="x", files=[other])
        assert "chat supports" in until("error")[-1]["message"]
        send(type="ask", text="x", files="not a list")
        assert until("error")[-1]["message"] == "bad files"
        # Audio: the model transcribes, then the words are asked as text.
        import wave

        def tiny_wav(p, secs, rate=100):
            with wave.open(p, "wb") as w:
                w.setnchannels(1)
                w.setsampwidth(2)
                w.setframerate(rate)
                w.writeframes(b"\0\0" * int(secs * rate))
        wav, long = os.path.join(fd, "q.wav"), os.path.join(fd, "long.wav")
        tiny_wav(wav, 3)
        tiny_wav(long, 61)
        script["inputs"] = ["text", "image", "audio"]
        script["bodies"] = []
        script["replies"] = [[{"content": " Che ore sono? "}], [{"content": "Noon."}]]
        send(type="ask", text="", audio=wav)
        seen = until("user")
        assert {"type": "state", "state": "transcribing"} in seen
        assert seen[-1]["text"] == "Che ore sono?"
        until("done")
        heard, asked = script["bodies"]
        with open(wav, "rb") as fh:
            b64 = base64.b64encode(fh.read()).decode()
        assert heard["messages"][0]["content"][1] == {
            "type": "input_audio", "input_audio": {"data": b64, "format": "wav"}}
        assert "tools" not in heard and heard["chat_template_kwargs"]["enable_thinking"] is False
        assert asked["messages"][-1]["content"].endswith("] Che ore sono?")
        assert cur["turns"][-1]["user"] == "Che ore sono?"
        script["replies"] = [[{"content": " "}]]
        send(type="ask", text="", audio=wav)
        assert until("error")[-1]["message"] == "heard nothing"
        assert until("state")[-1]["state"] == "idle"
        for p, msg in ((long, "recording too long"), ("q.wav", "audio path must be absolute"),
                       (pdf, "not a readable WAV")):
            send(type="ask", text="", audio=p)
            e = until("error")[-1]
            assert msg in e["message"] and e["turn"] is True, (p, e)
        send(type="ask", text="hi", audio=wav)
        assert until("error")[-1]["message"] == "bad audio"
        # A stop while transcribing ends the turn like any reply.
        script.update(hold=True, hung_up=None)
        script["replies"] = [[]]
        send(type="ask", text="", audio=wav)
        until("state")
        send(type="stop")
        seen = until("state")
        assert seen[-1]["state"] == "idle" and seen[-2]["stopped"] is True, seen
        script["hold"] = False
        script["inputs"] = ["text", "image"]
        send(type="ask", text="", audio=wav)
        assert until("error")[-1]["message"] == "m cannot hear audio"
        # write: the reply goes to a file, not the conversation.
        before = len(list_convs(DB))
        with open(os.path.join(fd, "doc.md"), "w") as fh:     # f is the socket's
            fh.write("taken")
        script["replies"] = [['data: {"choices": [{"delta": {"content": "```markdown\\n# Title\\n\\nbody\\n"}}]}\n\n',
                              {"content": "```"}]]
        send(type="write", files=[pdf], prompt="As Markdown.", dir=fd, name="doc.md")
        w = until("written")[-1]
        assert w["path"] == os.path.join(fd, "doc (1).md"), w
        with open(w["path"]) as fh:
            assert fh.read() == "# Title\n\nbody\n"       # the one fence unwrapped
        assert script["body"]["chat_template_kwargs"]["enable_thinking"] is False
        assert "tools" not in script["body"] and script["body"]["messages"][-1]["role"] == "user"
        assert len(list_convs(DB)) == before                 # nothing in History
        assert until("state")[-1]["state"] == "idle"
        # A stopped write leaves nothing, and the state goes back to idle.
        script.update(hold=True, hung_up=None)
        script["replies"] = [[]]
        send(type="write", files=[pdf], prompt="As Markdown.", dir=fd, name="held.md")
        until("state")
        send(type="stop")
        seen = until("state")
        assert seen[-1]["state"] == "idle"
        assert {"type": "error", "message": "stopped", "turn": True} in seen   # an outcome
        time.sleep(0.3)
        assert not os.path.exists(os.path.join(fd, "held.md"))
        assert not [n for n in os.listdir(fd) if n.startswith(".write-")]
        script["hold"] = False
        for bad in ({"dir": "relative", "name": "x.md"}, {"dir": fd, "name": "a/b.md"},
                    {"dir": fd, "name": ""}):
            send(type="write", files=[pdf], prompt="p", **bad)
            e = until("error")[-1]
            assert e["message"] == "bad write" and e["turn"] is True, bad
        assert unfence("```\na\n```\nmid\n```\nb\n```") == "```\na\n```\nmid\n```\nb\n```"
        script["replies"] = [[{"content": "  "}]]
        send(type="write", files=[pdf], prompt="p", dir=fd, name="empty.md")
        assert "empty reply" in until("error")[-1]["message"]
        assert not os.path.exists(os.path.join(fd, "empty.md"))
        # Cloud: the chat's model picks the provider, and only cloud tools
        # go along. A call to another one, copied from history, gets no
        # output. The key reaches the provider and nothing else.
        saved = TOOLS
        TOOLS = {"now": {"description": "d", "command": ["sh", "-c", "echo noon"], "cloud": True},
                 "mail": {"description": "m", "command": ["sh", "-c", "echo secret subject"]}}
        logged = io.StringIO()
        with contextlib.redirect_stdout(logged):
            send(type="new")
            until("conversation")
            script["bodies"], script["paths"] = [], []
            script["replies"] = [
                [{"reasoning": "hm"},
                 {"tool_calls": [{"index": 0, "id": "k1", "function": {"name": "mail",
                                                                       "arguments": "{}"}}]}],
                [{"content": "No mail tool."}]]
            send(type="ask", text="any mail?", model="fake:vendor/x:free")
            seen = until("done")
            first, second = script["bodies"]
            assert script["paths"] == ["/cloud/v1/chat/completions"] * 2
            assert script["auth"] == "Bearer sk-test"
            assert first["model"] == "vendor/x:free" and "chat_template_kwargs" not in first
            assert [t["function"]["name"] for t in first["tools"]] == ["now"]
            assert second["messages"][-1] == {"role": "tool", "tool_call_id": "k1",
                                              "content": "error: no tool named mail"}
            assert ("think", "hm") in [(e["type"], e.get("delta")) for e in seen]
            meters = [e for e in seen if e["type"] == "ctx"]
            assert meters[-1]["model"] == "fake:vendor/x:free" and meters[-1]["max"] == 0
            assert "sk-test" not in json.dumps(seen)
            conv = cur["id"]
            assert DB.execute("SELECT model FROM conversations WHERE id = ?",
                              (conv,)).fetchone()[0] == "fake:vendor/x:free"
            # The next ask with no model stays on the chat's.
            script["replies"] = [[{"content": "Still cloud."}]]
            send(type="ask", text="again")
            until("done")
            assert script["paths"][-1] == "/cloud/v1/chat/completions"
            # Reopened, the chat reports its model.
            send(type="new")
            assert until("conversation")[-1]["model"] is None
            send(type="open", id=conv)
            assert until("conversation")[-1]["model"] == "fake:vendor/x:free"
            # A local pick goes to the router with every tool and the
            # thinking switch.
            script["replies"] = [[{"content": "Local."}]]
            send(type="ask", text="local now", model="m")
            until("done")
            assert script["paths"][-1] == "/v1/chat/completions"
            assert script["body"]["model"] == "m" and "chat_template_kwargs" in script["body"]
            assert [t["function"]["name"] for t in script["body"]["tools"]] == ["now", "mail"]
            # The dropdown's list: the router's models, then the cloud ones
            # whose key loaded.
            send(type="models")
            items = until("models")[-1]["items"]
            assert {"id": "m", "status": "loaded", "cloud": False} in items
            assert {"id": "fake:vendor/x:free", "provider": "fake", "cloud": True} in items
            assert not [i for i in items if i["id"].startswith("broken:")]
            send(type="ask", text="x", model="")
            assert until("error")[-1]["message"] == "bad model"
            send(type="ask", text="x", model="broken:y")
            e = until("error")[-1]
            assert "no key" in e["message"] and e["turn"] is True
            assert cur["model"] == "m"                   # the chat keeps its model
            send(type="new")
            until("conversation")
        TOOLS = saved
        assert "sk-test" not in logged.getvalue(), logged.getvalue()
        # llama-server down: an error, not a dead thread.
        CFG["llm"] = "http://127.0.0.1:9/v1"
        send(type="ask", text="anyone?")
        assert until("error")[-1]["turn"] is True
        assert until("state")[-1]["state"] == "idle" and cur["pending"] is None
        # A second client gets hello with the conversation so far.
        c2 = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        c2.connect(path)
        assert json.loads(c2.makefile("rb").readline())["type"] == "hello"
        c2.close()
        cl.close()
    srv.shutdown()

    print("ok")


if __name__ == "__main__":
    if sys.argv[1:] == ["--test"]:
        test()
    elif sys.argv[1:] == ["--defaults"]:
        print(json.dumps(DEFAULTS))
    elif len(sys.argv) == 3 and sys.argv[1] == "--cloud-models":
        print(json.dumps(cloud_models(sys.argv[2])))
    else:
        main()