#!/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)
"""

import glob
import json
import os
import re
import signal
import socket
import sqlite3
import subprocess
import sys
import threading
import time
import tomllib
import traceback
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
    "env": {},
}
OPEN = {"env"}                            # free-form tables: replaced whole, not merged


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}.")
        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")
        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)
TOOL_SCHEMA = schema(TOOLS) or None
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)
    return conn


def save_turn(conn, conv, user, reply, tools, messages):
    # Returns the conversation id: the row is made at its first finished turn,
    # so a New that is never used leaves nothing behind.
    now = int(time.time())
    with conn:
        if conv is None:
            conv = conn.execute("INSERT INTO conversations(title, created, updated) "
                                "VALUES (?, ?, ?)",
                                (" ".join(user.split()[:6]), now, now)).lastrowid
        else:
            conn.execute("UPDATE conversations SET updated = ? WHERE id = ?", (now, 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) "
                     "VALUES (?, ?, ?, ?, ?, ?)",
                     (conv, seq, user, reply, json.dumps(tools), json.dumps(messages)))
    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 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)} for u, r, t, _ 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,))


# 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}
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"]}


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.
    global turn
    turn += 1
    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=[])


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 query" if kind == "search" and not isinstance(cmd.get("q"), str)
           else None)
    if bad:
        emit({"type": "error", "message": bad})
        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 == "ask":
        text = cmd["text"].strip()
        if text:
            interrupt()
            threading.Thread(target=reply, args=(turn, text), daemon=True).start()
    elif kind == "stop":
        interrupt()
    elif kind == "new":
        start_new()
        emit({"type": "conversation", "id": None, "turns": []})
    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()
        cur.update(id=cmd["id"], history=got[0], turns=got[1])
        emit({"type": "conversation", "id": cur["id"], "turns": cur["turns"]})
    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": []})
        emit({"type": "conversations", "items": list_convs(DB)})
    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 ""


def llm_model():
    # The router keeps one model resident; asking for another would swap it
    # out under other clients, so use whatever is not unloaded.
    with urllib.request.urlopen(CFG["llm"] + "/models", timeout=10) as r:
        models = json.load(r)["data"]
    return next((m["id"] for m in models if m["status"]["value"] != "unloaded"),
                models[0]["id"])


def ask(messages, calls, tools=None):
    # Yields ("think", text) and ("token", text) as they stream; tool-call
    # fragments go into calls.
    model = llm_model()
    context_size(model)
    body = {"model": model, "messages": messages, "stream": True,
            "stream_options": {"include_usage": True},
            "chat_template_kwargs": {"enable_thinking": True}}
    if tools:
        body["tools"] = tools
    if CFG["temperature"] is not None:
        body["temperature"] = CFG["temperature"]
    req = urllib.request.Request(CFG["llm"] + "/chat/completions", json.dumps(body).encode(),
                                 {"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=300) as r:
        for line in r:
            if not line.startswith(b"data: ") or line.strip() == b"data: [DONE]":
                continue
            chunk = json.loads(line[6:])
            if chunk.get("usage"):
                # How full the model's context is: history, tool results and
                # tool definitions all count.
                ctx["used"] = chunk["usage"]["prompt_tokens"]
                emit({"type": "ctx", "used": ctx["used"], "max": ctx["max"]})
            choices = chunk["choices"]
            if not choices:
                continue
            delta = choices[0]["delta"]
            collect_calls(calls, delta.get("tool_calls") or [])
            if delta.get("reasoning_content"):
                yield "think", delta["reasoning_content"]
            if delta.get("content"):
                yield "token", delta["content"]


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 reply(mine, text):
    try:
        answer(mine, text)
    except Exception as e:
        log("reply failed:\n" + traceback.format_exc().rstrip())
        with lock:
            if turn == mine:
                cur["pending"] = None
                emit({"type": "error", "message": str(e) or type(e).__name__})
                emit({"type": "state", "state": "idle"})


def answer(mine, text):
    with lock:
        if turn != mine:
            return
        history = list(cur["history"])
        cur["pending"] = {"user": text, "reply": "", "think": "", "tools": []}
        emit({"type": "user", "text": text})
        emit({"type": "state", "state": "thinking"})
    log(f"you: {text}")
    question = {"role": "user", "content": text}
    # 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": f"[{now}] {text}"}])
    first = len(msgs)                            # this turn's tool messages start here
    said, used = "", []
    # ponytail: a stopped reply notices at its next chunk, so llama-server's
    # slot stays busy until then and the next question queues behind it.
    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(msgs, calls, TOOL_SCHEMA if rnd < CFG["tool_rounds"] else None):
            if not say(mine, kind, piece):
                log("interrupted")
                return
            if kind == "token":
                content += piece
        said += content
        if not calls:
            break
        msgs.append({"role": "assistant", "content": content, "tool_calls": [
            {"id": c["id"], "type": "function",
             "function": {"name": c["name"], "arguments": c["arguments"]}}
            for c in calls.values()]})
        # 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(TOOLS, 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})
            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
        cur["id"] = save_turn(DB, cur["id"], text, said, used, entry)
        cur["history"].append(entry)
        cur["turns"].append({"user": text, "reply": said, "tools": used})
        cur["pending"] = None
        emit({"type": "done", "id": cur["id"]})
        emit({"type": "state", "state": "idle"})
    log("assistant:", said)


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)
        try:
            context_size(llm_model())     # so the first hello has it
        except OSError:
            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())
    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 http.server
    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)
    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"]},
                     {"user": "e domani?", "reply": "Sole.", "tools": []}]
    assert load_conv(db, 999) is None
    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

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

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

        def do_GET(self):
            body = ({"data": [{"id": "m", "status": {"value": "loaded"}}]}
                    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):
            self.rfile.read(int(self.headers["Content-Length"]))
            self.send_response(200)
            self.end_headers()
            try:
                for delta in script["replies"].pop(0):
                    self.wfile.write(b"data: " + json.dumps({"choices": [{"delta": delta}]})
                                     .encode() + b"\n\n")
                    self.wfile.flush()
                    time.sleep(script["delay"])
                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"
    DB = db_open(":memory:")
    TOOLS = {"now": {"description": "d", "command": ["sh", "-c", "echo noon"]}}
    TOOL_SCHEMA = schema(TOOLS)
    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
        # 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."}]]
        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"
        conv = cur["id"]
        assert load_conv(DB, conv)[1] == [{"user": "what time is it", "reply": "It is noon.",
                                           "tools": ["now"]}]
        # 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)"}
        # 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": []}
        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)
            assert until("error")[-1]["message"] == why, cmd
        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": []} in seen   # was current
        assert seen[-1]["items"] == []
        # 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]["type"] == "error"
        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__":
    test() if sys.argv[1:] == ["--test"] else main()