path: root/test_llamachat.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-only
#
# llamachat - a small native chat client for a local llama.cpp router
# 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.
"""Self-checks for the non-GUI logic. Run: ./test_llamachat.py"""

import contextlib
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

# Some checks need PySide6, so reuse the launcher's venv discovery rather
# than requiring the venv interpreter to be named on the command line.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import llamachat_venv  # noqa: E402

llamachat_venv.reexec(__file__)

from llamachat import backend, config, db

PRESETS_SAMPLE = """\
version = 1

; A preset whose mmproj line is commented out -> not vision capable.
#[Disabled-Model]
#ngl = all
#mmproj = /models/disabled/mmproj.gguf

[vision-model]
ngl = all
ctx-size = 32768
mmproj = /models/vision/mmproj.gguf

[text-model]
ngl = all
ctx-size = 16384
; --- Multimodal ---
; WARNING: eats VRAM.
#mmproj = /models/text/mmproj.gguf

[no-ctx-model]
ngl = all
"""


def test_presets():
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "presets.ini"
        path.write_text(PRESETS_SAMPLE)
        presets = config.parse_presets(path)

        # A '#'-commented section must not appear at all.
        assert "Disabled-Model" not in presets, presets.keys()
        assert set(presets) == {"vision-model", "text-model", "no-ctx-model"}

        # mmproj present -> vision; commented out -> not vision.
        assert presets["vision-model"].vision is True
        assert presets["text-model"].vision is False
        assert presets["no-ctx-model"].vision is False

        assert presets["vision-model"].ctx_size == 32768
        assert presets["text-model"].ctx_size == 16384
        assert presets["no-ctx-model"].ctx_size == 4096  # default

        # 32768 tokens * 3.5 chars * 0.5 -> 57344 chars
        assert presets["vision-model"].char_budget(0.5, 3.5) == 57344

    # A missing file yields no presets rather than raising.
    assert config.parse_presets(Path("/nonexistent/presets.ini")) == {}
    print("ok  presets parsing")


def test_real_presets():
    """The user's actual file, if present, must classify as expected."""
    path = Path("/etc/llama-server/presets.ini")
    if not path.exists():
        print("skip real presets (file absent)")
        return
    presets = config.parse_presets(path)
    assert presets, "presets.ini exists but parsed to nothing"

    # Section names track whatever the user currently runs, so assert the
    # parsing properties rather than a list of names that goes stale on
    # every rename.
    for preset in presets.values():
        assert preset.name
        assert preset.ctx_size > 0
        assert isinstance(preset.vision, bool)

    # Whether a given section has vision is the user's choice and changes
    # when they edit the file; only the commented-out case is a parsing
    # claim, and PRESETS_SAMPLE covers that hermetically above.
    print(f"ok  real presets classification ({len(presets)} sections)")


def test_fts_query_escaping():
    # Bare punctuation and FTS keywords must not become query syntax.
    assert db._fts_query("hello world") == '"hello" "world"'
    assert db._fts_query("foo AND bar") == '"foo" "AND" "bar"'
    assert db._fts_query('say "hi"') == '"say" """hi"""'
    assert db._fts_query("-flag") == '"-flag"'
    assert db._fts_query("   ") == ""
    print("ok  fts query escaping")


def test_history_roundtrip():
    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "test.db")

        chat = history.create_session("chat", "vision-model", "About otters")
        history.add_message(chat, "user", "Tell me about otters please")
        history.add_message(chat, "assistant", "Otters are semiaquatic mammals")

        shot = history.create_session("oneshot", "text-model", "Capital city")
        history.add_message(shot, "user", "What is the capital of Italy")

        rows = history.messages(chat)
        assert len(rows) == 2
        assert rows[0]["role"] == "user"

        # Newest session first.
        recent = history.recent_sessions()
        assert len(recent) == 2
        assert recent[0]["id"] == shot

        # FTS finds content across sessions, and punctuation cannot break it.
        assert len(history.search("otters")) == 2
        assert len(history.search("capital")) == 1
        assert history.search("zebra") == []
        assert history.search('otters "AND') == []  # must not raise

        # Attachments survive with enough detail to reconstruct context.
        message_id = history.add_message(chat, "user", "look at this")
        history.add_attachment(
            message_id, "/home/u/pic.png", "image",
            mime="image/png", size=1234, sha256="abc", thumb=b"\xff\xd8jpeg",
        )
        saved = history.attachments(message_id)
        assert len(saved) == 1
        assert saved[0]["path"] == "/home/u/pic.png"
        assert saved[0]["kind"] == "image"
        assert saved[0]["thumb"] == b"\xff\xd8jpeg"

        # Editing a streamed message keeps the FTS index in step.
        streamed = history.add_message(chat, "assistant", "")
        history.update_message(streamed, "the platypus is unusual")
        assert len(history.search("platypus")) == 1

        # Deleting a session takes its messages out of search too.
        history.delete_session(chat)
        assert history.search("otters") == []
        assert len(history.recent_sessions()) == 1

        history.close()
    print("ok  history roundtrip")


def test_attachment_truncation():
    with tempfile.TemporaryDirectory() as tmp:
        big = Path(tmp) / "big.py"
        big.write_text("x" * 5000)

        att = backend.load_attachment(big, char_budget=1000)
        assert att.kind == "text"
        assert att.truncated is True
        assert len(att.text) == 1000
        assert att.size == 5000
        assert len(att.sha256) == 64

        small = Path(tmp) / "small.txt"
        small.write_text("hello")
        att = backend.load_attachment(small, char_budget=1000)
        assert att.truncated is False
        assert att.text == "hello"

        odd = Path(tmp) / "thing.bin"
        odd.write_bytes(b"\x00\x01")
        try:
            backend.load_attachment(odd, char_budget=1000)
        except backend.BackendError:
            pass
        else:
            raise AssertionError("unsupported type should raise")
    print("ok  attachment truncation")


def test_classify():
    assert backend.classify(Path("a.py")) == "text"
    assert backend.classify(Path("a.SlackBuild")) == "text"
    assert backend.classify(Path("a.png")) == "image"
    assert backend.classify(Path("a.jpg")) == "image"
    assert backend.classify(Path("a.so")) == "unknown"
    print("ok  file classification")


def test_sse_parsing():
    line = 'data: {"choices":[{"delta":{"content":"hi"}}]}'
    assert backend._parse_sse_line(line) == [("content", "hi")]

    # Reasoning arrives in its own field, which is what lets the UI keep
    # thinking and reply apart without parsing <think> tags.
    think = 'data: {"choices":[{"delta":{"reasoning_content":"hmm"}}]}'
    assert backend._parse_sse_line(think) == [("reasoning", "hmm")]

    assert backend._parse_sse_line("data: [DONE]") == backend.DONE
    assert backend._parse_sse_line(": keepalive") == []
    assert backend._parse_sse_line("") == []
    assert backend._parse_sse_line("data: {bad json") == []

    # An opening delta of {'role': 'assistant', 'content': None} carries
    # nothing to render and must not be mistaken for end-of-stream.
    opening = backend._parse_sse_line(
        'data: {"choices":[{"delta":{"role":"assistant","content":null}}]}'
    )
    assert opening == [], opening
    assert opening != backend.DONE

    # A delta carrying both fields is thinking, not reply. Classifying it as
    # content splices the tail of the reasoning onto the front of the reply.
    both = backend._parse_sse_line(
        'data: {"choices":[{"delta":'
        '{"reasoning_content":"still thinking","content":""}}]}'
    )
    assert both == [("reasoning", "still thinking")], both

    # An empty reasoning delta renders nothing and must not fall through to
    # the content branch.
    empty = backend._parse_sse_line(
        'data: {"choices":[{"delta":{"reasoning_content":""}}]}'
    )
    assert empty == [], empty

    # The final chunk of a tool round carries finish_reason and the usage
    # block together; both must survive, or the tool call is dropped.
    combined = backend._parse_sse_line(
        'data: {"choices":[{"index":0,"delta":{"content":"",'
        '"reasoning_content":null},"finish_reason":"tool_calls"}],'
        '"usage":{"prompt_tokens":388,"completion_tokens":74,'
        '"total_tokens":462}}'
    )
    assert combined == [
        (
            "usage",
            '{"prompt_tokens": 388, "completion_tokens": 74, "total_tokens": 462}',
        ),
        ("tool_finish", ""),
    ], combined
    print("ok  sse parsing")


def test_reasoning_storage():
    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "r.db")
        sid = history.create_session("chat", "m", "t")

        # A streamed reply is inserted empty, then filled in once done.
        mid = history.add_message(sid, "assistant", "")
        history.update_message(mid, "51", "the model's private thinking")

        row = history.messages(sid)[0]
        assert row["content"] == "51"
        assert row["reasoning"] == "the model's private thinking"

        # Reasoning must stay out of the search index, or thinking text
        # would drown out real hits.
        assert history.search("51") != []
        assert history.search("private") == []

        # Omitting the argument leaves stored reasoning untouched.
        history.update_message(mid, "52")
        assert history.messages(sid)[0]["reasoning"] == (
            "the model's private thinking"
        )
        history.close()
    print("ok  reasoning storage")


def test_migration_adds_reasoning():
    """A database created before the reasoning column must still open."""
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
            "INSERT INTO messages VALUES (1,1,'user','older message',0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        rows = history.messages(1)
        assert rows[0]["content"] == "older message"
        assert rows[0]["reasoning"] == ""  # backfilled by the migration

        mid = history.add_message(1, "assistant", "new")
        history.update_message(mid, "new", "fresh thinking")
        assert history.messages(1)[1]["reasoning"] == "fresh thinking"
        history.close()
    print("ok  reasoning column migration")


def test_user_content():
    with tempfile.TemporaryDirectory() as tmp:
        src = Path(tmp) / "code.py"
        src.write_text("print(1)")
        text_att = backend.load_attachment(src, 1000)

        # Text only -> a plain string, file inlined before the prompt.
        content = backend.build_user_content("explain", [text_att])
        assert isinstance(content, str)
        assert "print(1)" in content
        assert content.endswith("explain")

        # No attachments -> just the prompt.
        assert backend.build_user_content("hi", []) == "hi"

        # An image -> the multi-part array the vision API expects.
        img = backend.Attachment(
            path=Path("/x/a.png"), kind="image", mime="image/png",
            size=1, sha256="", data_url="data:image/png;base64,AAA",
        )
        content = backend.build_user_content("what is this", [img])
        assert isinstance(content, list)
        assert content[0]["type"] == "text"
        assert content[1]["type"] == "image_url"
        assert content[1]["image_url"]["url"].startswith("data:image/png")
    print("ok  user content assembly")


def test_markdown_rendering():
    """Replies render as markdown; markup inside them stays literal."""
    import os

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import _markdown_to_fragment

    app = QApplication.instance() or QApplication([])
    assert app is not None

    fragment = _markdown_to_fragment(
        "**bold** and *italic* and `code`\n\n"
        "- a\n- b\n\n"
        "| x | y |\n|---|---|\n| 1 | 2 |\n\n"
        "```py\ndef f():\n    pass\n```\n"
    )
    assert "font-weight:700" in fragment
    assert "font-style:italic" in fragment
    assert "<ul" in fragment
    assert "<table" in fragment
    assert "<pre" in fragment
    # Every <pre> is tinted, since Qt emits one per line of a code block.
    assert fragment.count("<pre") == fragment.count("background:rgba")
    # The document wrapper must not leak into the fragment.
    assert "<html" not in fragment and "<body" not in fragment

    # Markup in a reply is shown, never interpreted.
    hostile = _markdown_to_fragment(
        'Text <b>tag</b> <img src=x onerror=alert(1)> <a href="http://bad">l</a>'
    )
    assert "&lt;b&gt;" in hostile, hostile
    assert "font-weight:700" not in hostile
    # The URL may appear, but only as escaped text, never as a live anchor.
    assert "<a href" not in hostile
    assert "&lt;a href=&quot;http://bad&quot;&gt;" in hostile, hostile

    # A markdown link is a real anchor, so check a reply cannot forge one
    # aimed at the reasoning toggle.
    from llamachat.ui import REASONING_SCHEME

    forged = _markdown_to_fragment(f"[click]({REASONING_SCHEME}0)")
    assert f'href="{REASONING_SCHEME}' not in forged, forged
    assert 'href="blocked:' in forged, forged

    # Ordinary markdown links still work.
    link = _markdown_to_fragment("[site](http://example.com)")
    assert "http://example.com" in link

    # Empty input renders nothing rather than an empty document wrapper.
    assert _markdown_to_fragment("") == ""

    # Partial markdown arrives constantly while streaming and must not raise.
    for partial in ("```", "```py", "```py\ndef f(", "| a |", "**unclosed", "#"):
        _markdown_to_fragment(partial)
    print("ok  markdown rendering")


def test_unbalanced_backticks():
    """An odd backtick cannot reflow the rest of a reply as code."""
    import os

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import _close_fences, _markdown_to_fragment

    app = QApplication.instance() or QApplication([])
    assert app is not None

    # An unclosed fence gets one, so the tail is not swallowed.
    assert _close_fences("a\n```py\nx=1").endswith("\n```")
    # A closed fence is left exactly as it was.
    balanced = "a\n```py\nx=1\n```\ntail"
    assert _close_fences(balanced) == balanced
    # A longer opener needs a closer at least as long.
    assert _close_fences("````\nx").endswith("\n````")
    # Tildes are fences too.
    assert _close_fences("~~~\nx").endswith("\n~~~")
    # A lone inline tick is dropped rather than opening a run.
    assert "`" not in _close_fences("use ` here")
    # Balanced inline ticks survive untouched.
    assert _close_fences("a `b` c") == "a `b` c"

    # The real failure: reasoning text full of ticks must not become one
    # block that eats the prose after it.
    leaked = "` (wait)\nthen `find / -perm -4000` and more prose here"
    rendered = _markdown_to_fragment(leaked)
    assert "more prose here" in rendered
    assert "<pre" not in rendered, rendered
    print("ok  unbalanced backticks")


def test_code_block_wrapping():
    """A long code line wraps instead of widening the whole transcript."""
    import os

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtGui import QTextDocument
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import _markdown_to_fragment

    app = QApplication.instance() or QApplication([])
    assert app is not None

    # Qt's markdown stylesheet wraps p and li but not pre, and it is dropped
    # when only the body fragment is kept, so pre has to say so itself.
    fragment = _markdown_to_fragment("```\n" + "A" * 200 + "\n```\n")
    assert "white-space:pre-wrap" in fragment, fragment

    # The real check is the layout: no block may exceed the viewport.
    doc = QTextDocument()
    doc.setHtml(fragment)
    doc.setTextWidth(300)
    layout = doc.documentLayout()
    widest = 0.0
    block = doc.begin()
    while block.isValid():
        widest = max(widest, layout.blockBoundingRect(block).width())
        block = block.next()
    assert widest <= 300, widest
    print("ok  code block wrapping")


def test_code_block_copy_links():
    """Each fenced block gets a copy link addressing it by index."""
    import os
    import re as _re

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import (
        COPY_SCHEME, _close_fences, _code_blocks, _markdown_to_fragment,
    )

    app = QApplication.instance() or QApplication([])
    assert app is not None

    # Blocks are read from the source: Qt emits one <pre> per line, so the
    # text is not recoverable from the rendered fragment.
    assert _code_blocks("```py\nx=1\ny=2\n```\n") == ["x=1\ny=2"]
    assert _code_blocks("~~~\na\n~~~\n") == ["a"]
    assert _code_blocks("no code here") == []
    # A longer fence can hold shorter ones without ending the block.
    assert _code_blocks("````\n```\ninner\n```\n````\n") == ["```\ninner\n```"]

    # A link per block, numbered bubble.block so several replies coexist.
    markdown = "one\n\n```\nAAA\n```\n\ntwo\n\n```\nBBB\n```\n"
    fragment = _markdown_to_fragment(markdown, 3)
    assert _re.findall(r"x-llamachat-copy:([0-9.]+)", fragment) == ["3.0", "3.1"]

    # Icon only, and under its block: Qt cannot float it over the tint.
    assert "⧉" in fragment and "copy<" not in fragment
    assert fragment.rindex("⧉") > fragment.rindex("</pre>")

    # The indices the links carry must select the blocks the parser found.
    blocks = _code_blocks(_close_fences(markdown))
    assert blocks == ["AAA", "BBB"], blocks

    # A block still streaming is closed first, so it is copyable mid-reply.
    assert _code_blocks(_close_fences("```\nhalf")) == ["half"]

    # Without an index there are no links, so a bubble that cannot be
    # addressed does not emit a link that would not resolve.
    assert COPY_SCHEME not in _markdown_to_fragment(markdown)

    # A reply cannot forge one: the scheme is defused like the others.
    forged = _markdown_to_fragment(f"[copy]({COPY_SCHEME}0.0)", 0)
    assert f'href="{COPY_SCHEME}' not in forged, forged
    print("ok  code block copy links")


def test_code_block_copy_states():
    """The copy link dims, lights on hover, and ticks once used."""
    import os

    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat.ui import _PRE_SPACER, _markdown_to_fragment

    app = QApplication.instance() or QApplication([])
    assert app is not None

    markdown = "```py\nAAA\n```\n"

    def link_style(fragment: str, glyph: str) -> str:
        return fragment.split(glyph)[0].rsplit("<a", 1)[1]

    # Grey by default, so a transcript full of code is not a wall of
    # buttons. The pill is what makes it read as a control at all.
    plain = _markdown_to_fragment(markdown, 0)
    assert "rgba(127,127,127,0.28)" in link_style(plain, "⧉")

    # Qt rich text has no :hover, so hovering is a re-render with the
    # highlighted anchor passed back in.
    hovered = _markdown_to_fragment(markdown, 0, hovered="0.0")
    assert "palette(highlight)" in link_style(hovered, "⧉")

    # A different link being hovered must not light this one.
    other = _markdown_to_fragment(markdown, 0, hovered="0.1")
    assert "rgba(127,127,127,0.28)" in link_style(other, "⧉")

    # After copying, the icon becomes a tick on green, so the confirmation
    # is visible without a status message.
    copied = _markdown_to_fragment(markdown, 0, copied="0.0")
    assert "✓" in copied and "⧉" not in copied
    assert "#2e9e4f" in link_style(copied, "✓")

    # Qt draws no padding on <pre>, so a tinted spacer stands in above and
    # below; without it the tint sits flush against the code.
    assert plain.count(_PRE_SPACER) == 2
    print("ok  code block copy states")


def test_system_qt_theme_guard():
    """The plugin path is only borrowed when the Qt versions agree."""
    import os

    from llamachat.__main__ import _system_qt_version, _use_system_qt_theme

    # An explicit user setting is never overridden.
    saved = os.environ.get("QT_PLUGIN_PATH")
    try:
        os.environ["QT_PLUGIN_PATH"] = "/user/choice"
        _use_system_qt_theme()
        assert os.environ["QT_PLUGIN_PATH"] == "/user/choice"
    finally:
        if saved is None:
            os.environ.pop("QT_PLUGIN_PATH", None)
        else:
            os.environ["QT_PLUGIN_PATH"] = saved

    system = _system_qt_version()
    if not system:
        print("skip qt theme guard (no system Qt6 found)")
        return
    assert system.count(".") == 2, system

    from PySide6.QtCore import qVersion

    saved = os.environ.pop("QT_PLUGIN_PATH", None)
    try:
        _use_system_qt_theme()
        chosen = os.environ.get("QT_PLUGIN_PATH")
        if system == qVersion():
            assert chosen, "matching Qt versions should adopt system plugins"
            assert Path(chosen, "platformthemes").is_dir()
        else:
            assert chosen is None, (
                f"must not load {system} plugins into Qt {qVersion()}"
            )
    finally:
        if saved is None:
            os.environ.pop("QT_PLUGIN_PATH", None)
        else:
            os.environ["QT_PLUGIN_PATH"] = saved
    print("ok  system qt theme guard")


def test_prompt_store():
    """Prompts are files; a preset replaces the global one."""
    from llamachat import prompts

    with tempfile.TemporaryDirectory() as tmp:
        store = prompts.PromptStore(Path(tmp) / "prompts")
        assert store.names() == []

        store.ensure_default()
        assert store.names() == ["default"]
        assert store.path_for("default").exists()

        # ensure_default must not clobber an edited global prompt.
        store.save("default", "edited global")
        store.ensure_default()
        assert store.load("default").text == "edited global"

        store.save("coding", "you write code")
        store.save("terse", "be brief")
        # Global first, then presets alphabetically.
        assert store.names() == ["default", "coding", "terse"]

        # A preset replaces the global prompt rather than adding to it.
        assert store.resolve("coding") == "you write code"
        assert store.resolve("") == "edited global"
        assert store.resolve("default") == "edited global"

        # Sentinels.
        assert store.resolve(prompts.NONE) == ""
        assert store.resolve(prompts.CUSTOM, "one off") == "one off"

        # A preset deleted behind our back falls back rather than sending
        # nothing, which would silently change the model's behaviour.
        assert store.resolve("missing") == "edited global"

        assert store.delete("coding") is True
        assert store.names() == ["default", "terse"]
        # The global prompt is the fallback for everything, so it stays.
        assert store.delete("default") is False
        assert "default" in store.names()

        # Names become filenames, so path separators must not escape.
        assert prompts.safe_name("../../etc/passwd") == "etc-passwd"
        assert prompts.safe_name("/etc/passwd") == "etc-passwd"
        assert prompts.safe_name("my prompt!") == "my-prompt"
        for hostile in ("../escape", "a/b", "/tmp/x"):
            saved = store.save(hostile, "x")
            assert saved.path.parent == store.dir, saved.path
            assert saved.path.resolve().parent == store.dir.resolve()

        # A name that reduces to nothing is rejected rather than writing
        # a dotfile called ".md".
        for empty in ("..", "", "....", "///"):
            assert prompts.safe_name(empty) == ""
            assert store.load(empty) is None
            assert store.delete(empty) is False
            try:
                store.path_for(empty)
            except ValueError:
                pass
            else:
                raise AssertionError(f"{empty!r} should be rejected")
    print("ok  prompt store")


def test_token_estimate():
    """The meter's pre-send estimate tracks message size."""
    small = [{"role": "user", "content": "hi"}]
    large = [{"role": "user", "content": "word " * 1000}]

    assert backend.estimate_tokens([], 3.5) == 0
    assert backend.estimate_tokens(small, 3.5) < 20
    assert backend.estimate_tokens(large, 3.5) > 1000

    # More history means a bigger prompt.
    grown = large + [{"role": "assistant", "content": "reply " * 500}]
    assert backend.estimate_tokens(grown, 3.5) > backend.estimate_tokens(
        large, 3.5
    )

    # An image costs far more than its text part suggests.
    with_image = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "what is this"},
                {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAA"}},
            ],
        }
    ]
    assert backend.estimate_tokens(with_image, 3.5) > 500

    # A silly ratio must not divide by zero.
    assert backend.estimate_tokens(small, 0) > 0
    print("ok  token estimate")


def test_usage_parsing():
    """The final stream chunk carries the exact prompt token count."""
    line = (
        'data: {"choices":[],"usage":{"prompt_tokens":1234,'
        '"completion_tokens":56,"total_tokens":1290}}'
    )
    events = backend._parse_sse_line(line)
    assert [k for k, _ in events] == ["usage"], events
    kind, payload = events[0]
    assert kind == "usage"
    stats = json.loads(payload)
    assert stats["prompt_tokens"] == 1234
    assert stats["total_tokens"] == 1290

    # A usage-less chunk with empty choices is not mistaken for one.
    assert backend._parse_sse_line('data: {"choices":[]}') == []
    print("ok  usage parsing")


def test_session_prompt_storage():
    """A conversation remembers the prompt it was built with."""
    from llamachat import prompts

    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "p.db")

        sid = history.create_session(
            "chat", "m", "t", prompt_name="coding", prompt_custom=""
        )
        row = history.get_session(sid)
        assert row["prompt_name"] == "coding"
        assert row["prompt_custom"] == ""

        history.set_prompt(sid, prompts.CUSTOM, "just this once")
        row = history.get_session(sid)
        assert row["prompt_name"] == prompts.CUSTOM
        assert row["prompt_custom"] == "just this once"

        # Defaults keep older call sites working.
        plain = history.create_session("oneshot", "m", "t")
        assert history.get_session(plain)["prompt_name"] == ""
        history.close()
    print("ok  session prompt storage")


def test_prompt_column_migration():
    """A database predating the prompt columns must still open."""
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        row = history.get_session(1)
        assert row["title"] == "old"
        assert row["prompt_name"] == ""
        assert row["prompt_custom"] == ""
        history.set_prompt(1, "coding")
        assert history.get_session(1)["prompt_name"] == "coding"
        history.close()
    print("ok  prompt column migration")


def test_sidebar_toggle():
    """The history panel hides, restores its width, and persists."""
    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat import backend as _backend
    from llamachat import config as _config
    from llamachat import models as _models
    from llamachat.ui import ChatWindow

    app = QApplication.instance() or QApplication([])
    assert app is not None

    with tempfile.TemporaryDirectory() as tmp:
        cfg = _config.load(Path(tmp) / "config.toml")
        cfg.prompts_dir = Path(tmp) / "prompts"
        cfg.db_path = Path(tmp) / "t.db"
        # state_path already points inside tmp, since it is derived from the
        # config path, so these checks cannot touch the real saved layout.
        assert cfg.state_path.parent == Path(tmp), cfg.state_path

        history = db.History(cfg.db_path)
        client = _backend.MultiClient(cfg.providers, cfg.request_timeout)
        presets = _config.parse_presets(cfg.presets_path)
        store = _models.ModelStore(cfg.models_path)

        window = ChatWindow(cfg, history, client, presets, store)
        window.resize(1000, 700)

        # The shortcut must be registered on the window itself.
        shortcuts = [
            a.shortcut().toString()
            for a in window.actions()
            if not a.shortcut().isEmpty()
        ]
        assert "Ctrl+\\" in shortcuts, shortcuts

        window.show()
        assert window.sidebar_visible()
        assert window.sidebar_button.isChecked()

        window.toggle_sidebar()
        assert not window.sidebar_visible()
        assert not window.sidebar_button.isChecked()

        window.toggle_sidebar()
        assert window.sidebar_visible()

        # The width survives a hide/show rather than snapping to a default.
        window.splitter.setSizes([333, 667])
        app.processEvents()
        window.toggle_sidebar()
        window.toggle_sidebar()
        assert window.sidebar_width == 333, window.sidebar_width

        # Driving the button directly must take the same path.
        window.sidebar_button.setChecked(False)
        app.processEvents()
        assert not window.sidebar_visible()
        window.sidebar_button.setChecked(True)
        app.processEvents()
        assert window.sidebar_visible()
        assert window.sidebar_width == 333, window.sidebar_width

        # Hidden state and width must come back on the next start.
        window.sidebar_button.setChecked(False)
        window._save_layout()
        restored = ChatWindow(cfg, history, client, presets, store)
        assert not restored.sidebar_button.isChecked()
        assert restored.sidebar_width == 333, restored.sidebar_width

        restored.close()
        window.close()
        history.close()
    print("ok  sidebar toggle")


def test_shortcuts():
    """Ctrl+N, Ctrl+F and Escape behave as advertised."""
    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtCore import QEvent, Qt
    from PySide6.QtGui import QKeyEvent
    from PySide6.QtWidgets import QApplication

    from llamachat import backend as _backend
    from llamachat import config as _config
    from llamachat import models as _models
    from llamachat.ui import ChatWindow

    app = QApplication.instance() or QApplication([])

    with tempfile.TemporaryDirectory() as tmp:
        cfg = _config.load(Path(tmp) / "config.toml")
        cfg.prompts_dir = Path(tmp) / "prompts"
        cfg.db_path = Path(tmp) / "t.db"
        history = db.History(cfg.db_path)
        window = ChatWindow(
            cfg,
            history,
            _backend.MultiClient(cfg.providers, cfg.request_timeout),
            _config.parse_presets(cfg.presets_path),
            _models.ModelStore(cfg.models_path),
        )
        window.resize(1000, 700)
        window.show()
        window.raise_()
        window.activateWindow()
        # The offscreen platform only grants focus to the active window, and
        # a window left over from an earlier check can still hold it.
        window.setFocus()
        app.processEvents()

        registered = {
            a.shortcut().toString()
            for a in window.actions()
            if not a.shortcut().isEmpty()
        }
        for wanted in ("Ctrl+N", "Ctrl+F", "Ctrl+\\"):
            assert wanted in registered, (wanted, registered)

        # Ctrl+F puts the cursor in the search box. The offscreen platform
        # only grants real focus to one window per process, so check where
        # focus was directed rather than whether the platform granted it.
        window.input.setFocus()
        app.processEvents()
        window.focus_search()
        app.processEvents()
        assert window.focusWidget() is window.search_box, window.focusWidget()

        # The search box lives in the top bar, so hiding the history panel
        # must not take it away.
        window.sidebar_button.setChecked(False)
        app.processEvents()
        assert not window.sidebar_visible()
        assert window.search_box.isVisible()

        # Its results render in the panel, so focusing reveals the panel.
        window.focus_search()
        app.processEvents()
        assert window.sidebar_visible()

        def press_escape() -> None:
            window.keyPressEvent(
                QKeyEvent(QEvent.KeyPress, Qt.Key_Escape, Qt.NoModifier)
            )
            app.processEvents()

        # Escape backs out of the search box before hiding the window, so a
        # stray press while filtering does not dismiss everything.
        window.search_box.setFocus()
        window.search_box.setText("otters")
        app.processEvents()
        assert window.focusWidget() is window.search_box
        press_escape()
        assert window.search_box.text() == ""
        assert window.isVisible()
        press_escape()
        assert window.focusWidget() is window.input
        assert window.isVisible()
        press_escape()
        assert not window.isVisible()

        # Ctrl+N clears the conversation and puts the cursor in the input.
        window.show()
        window.raise_()
        window.activateWindow()
        # The offscreen platform only grants focus to the active window, and
        # a window left over from an earlier check can still hold it.
        window.setFocus()
        app.processEvents()
        session = history.create_session("chat", "m", "old")
        history.add_message(session, "user", "something")
        window.open_session(session)
        assert window.session_id == session
        window.search_box.setFocus()
        window.new_session()
        app.processEvents()
        assert window.session_id is None
        assert window.transcript.toPlainText().strip() == ""
        assert window.focusWidget() is window.input

        window.close()
        history.close()
    print("ok  shortcuts")


def test_status_dismiss():
    """Status messages show a dismiss button; clicking it hides them."""
    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
    from PySide6.QtWidgets import QApplication

    from llamachat import backend as _backend
    from llamachat import config as _config
    from llamachat import models as _models
    from llamachat.ui import ChatWindow

    app = QApplication.instance() or QApplication([])

    with tempfile.TemporaryDirectory() as tmp:
        cfg = _config.load(Path(tmp) / "config.toml")
        cfg.prompts_dir = Path(tmp) / "prompts"
        cfg.db_path = Path(tmp) / "t.db"
        history = db.History(cfg.db_path)
        window = ChatWindow(
            cfg,
            history,
            _backend.MultiClient(cfg.providers, cfg.request_timeout),
            _config.parse_presets(cfg.presets_path),
            _models.ModelStore(cfg.models_path),
        )
        window.show()
        app.processEvents()

        # Errors are dismissable: the ✕ button appears beside the message.
        window.show_status("boom", error=True)
        app.processEvents()
        assert not window.status_row.isHidden()
        assert window.status_label.text() == "boom"
        assert not window.status_dismiss.isHidden()

        window.status_dismiss.click()
        app.processEvents()
        assert window.status_row.isHidden()

        # Transient statuses are dismissable too.
        window.show_status("searching: otters…")
        app.processEvents()
        assert not window.status_row.isHidden()
        assert not window.status_dismiss.isHidden()

        window.status_dismiss.click()
        app.processEvents()
        assert window.status_row.isHidden()

        window.close()
        history.close()
    print("ok  status dismiss")


def test_version_matches_changelog():
    """The package version must be the newest release in the changelog."""
    import re

    import llamachat

    assert re.fullmatch(r"\d+\.\d+\.\d+", llamachat.__version__), (
        f"not semver: {llamachat.__version__}"
    )

    changelog = Path(__file__).resolve().parent / "CHANGELOG.md"
    if not changelog.exists():
        print("skip changelog check (file absent)")
        return

    released = re.findall(
        r"^## \[(\d+\.\d+\.\d+)\]", changelog.read_text(), re.MULTILINE
    )
    assert released, "changelog has no released versions"
    assert released[0] == llamachat.__version__, (
        f"__version__ is {llamachat.__version__} but the newest changelog "
        f"entry is {released[0]}"
    )
    print("ok  version matches changelog")


def test_venv_discovery():
    """The launcher must find a venv even when it shares the system binary."""
    saved = os.environ.pop("LLAMACHAT_PYTHON", None)
    try:
        found = llamachat_venv.find_interpreter()
        current = Path(sys.executable)
        other = [
            c for c in llamachat_venv.CANDIDATES if c.is_file() and c != current
        ]
        if other:
            # A venv built with --system-site-packages has a bin/python3 that
            # symlinks to the system interpreter. Comparing resolved paths
            # would discard it as "the interpreter we are already running".
            assert found is not None, (
                f"a candidate exists ({other[0]}) but none was selected"
            )
            assert found.is_file()
        else:
            # Already running as the only candidate; nothing to hand over to.
            assert found is None

        # An explicit override wins over the search order.
        os.environ["LLAMACHAT_PYTHON"] = sys.executable
        chosen = llamachat_venv.find_interpreter()
        # Only rejected because it is the interpreter already running.
        assert chosen is None or chosen == Path(sys.executable)

        os.environ["LLAMACHAT_PYTHON"] = "/nonexistent/python3"
        assert llamachat_venv.find_interpreter() is None
    finally:
        if saved is None:
            os.environ.pop("LLAMACHAT_PYTHON", None)
        else:
            os.environ["LLAMACHAT_PYTHON"] = saved

    # reexec must be a no-op once PySide6 is importable, or it would loop.
    if llamachat_venv.have_pyside():
        llamachat_venv.reexec(__file__)  # returns rather than exec'ing
    print("ok  venv discovery")


def test_config_defaults():
    cfg = config.load(Path("/nonexistent/config.toml"))
    assert cfg.base_url == "http://localhost:8181"
    assert cfg.presets_path == Path("/etc/llama-server/presets.ini")
    assert cfg.socket_path.name == "llamachat.sock"
    assert cfg.db_path.name == "history.db"

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "config.toml"
        path.write_text(
            'base_url = "http://localhost:9999/"\ndefault_model = "m"\n'
        )
        cfg = config.load(path)
        assert cfg.base_url == "http://localhost:9999"  # trailing / stripped
        assert cfg.default_model == "m"
        assert cfg.request_timeout == 300  # default kept
    print("ok  config defaults")


def test_provider_parsing():
    """Providers come from [providers.*]; a bare base_url synthesizes local."""
    from llamachat import providers

    # A modern config with two providers.
    parsed = providers.parse(
        {
            "providers": {
                "local": {"base_url": "http://localhost:8181/"},
                "together": {
                    "base_url": "https://api.example.org",
                    "api_key": "env:TEST_KEY_NAME",
                "filter": ["qwen", "deepseek"],
                "ctx_size": 32768,
                "price_in": 0.6,
                "price_out": 0.9,
                "thinking_budget": 8192,
                "replay_reasoning": True,
            },
            }
        }
    )
    assert set(parsed) == {"local", "together"}
    # Trailing slashes are stripped so URL joining stays predictable.
    assert parsed["local"].base_url == "http://localhost:8181"
    assert parsed["local"].api_key == ""
    assert parsed["together"].filter == ["qwen", "deepseek"]
    assert parsed["together"].ctx_size == 32768
    assert parsed["together"].price_in == 0.6
    assert parsed["together"].price_out == 0.9
    assert parsed["together"].thinking_budget == 8192
    assert parsed["together"].replay_reasoning is True
    # Off by default: replaying reasoning costs context and input tokens.
    assert parsed["local"].replay_reasoning is False

    # An old config: bare base_url, no providers table at all.
    legacy = providers.parse({"base_url": "http://localhost:8181"})
    assert set(legacy) == {"local"}
    assert legacy["local"].base_url == "http://localhost:8181"

    # Both present: the explicit entry wins over the bare key.
    both = providers.parse(
        {
            "base_url": "http://ignored.example.org",
            "providers": {"local": {"base_url": "http://explicit.example.org"}},
        }
    )
    assert both["local"].base_url == "http://explicit.example.org"

    # A provider with no base_url is skipped rather than half-configured.
    broken = providers.parse(
        {"providers": {"local": {"base_url": "http://x.example.org"},
                       "bad": {"api_key": "literal"}}}
    )
    assert set(broken) == {"local"}

    # Unset numbers stay None so "unknown" is distinguishable from zero.
    assert parsed["local"].ctx_size is None
    assert parsed["local"].price_in is None

    # nan and inf survive float() and would reach the cost arithmetic, where
    # they render as "$nan" or "$-inf" in the *priced* branch: the readout
    # inventing a figure in the one state built to admit it cannot say.
    # Unknown is the honest answer, exactly as in models.ini.
    nonfinite = providers.parse(
        {
            "providers": {
                "p": {
                    "base_url": "http://x.example.org",
                    "price_in": float("nan"),
                    "price_out": float("inf"),
                },
                "n": {
                    "base_url": "http://y.example.org",
                    "price_in": float("-inf"),
                    "ctx_size": 8192,
                },
            }
        }
    )
    assert nonfinite["p"].price_in is None
    assert nonfinite["p"].price_out is None
    # A bad price must not take the good ctx_size down with it.
    assert nonfinite["n"].price_in is None
    assert nonfinite["n"].ctx_size == 8192

    # A [providers.local] that omits base_url inherits the bare one rather
    # than shadowing the local provider out of existence.
    partial = providers.parse(
        {
            "base_url": "http://localhost:8181",
            "providers": {"local": {"api_key": "env:SOME_VAR"}},
        }
    )
    assert set(partial) == {"local"}
    assert partial["local"].base_url == "http://localhost:8181"
    # The explicit entry's own fields survive the merge.
    assert partial["local"].api_key == "env:SOME_VAR"

    # A filter given as a bare string is one needle, not four.
    stringy = providers.parse(
        {"providers": {"p": {"base_url": "http://x.example.org",
                             "filter": "qwen"}}}
    )
    assert stringy["p"].filter == ["qwen"]

    # Any other non-list shape is wrapped too, rather than iterated: a number
    # would raise, and a dict would silently degrade into its keys.
    odd = providers.parse(
        {"providers": {"n": {"base_url": "http://x.example.org", "filter": 5},
                       "d": {"base_url": "http://y.example.org",
                             "filter": {"a": 1}}}}
    )
    assert odd["n"].filter == ["5"]
    # The invariant is that the dict was wrapped whole, not iterated into its
    # keys. Asserting that rather than its repr, which is not ours to pin.
    assert odd["d"].filter != ["a"] and len(odd["d"].filter) == 1

    # A colon in a provider name would make every id built from it ambiguous,
    # so such a provider is skipped rather than silently routed to local.
    colonic = providers.parse(
        {"providers": {"local": {"base_url": "http://x.example.org"},
                       "a:b": {"base_url": "http://y.example.org"},
                       "": {"base_url": "http://z.example.org"}}}
    )
    assert set(colonic) == {"local"}
    print("ok  provider config parsing")


def test_provider_malformed_shapes():
    """Config shapes that are not tables are skipped, never raised on.

    Every case here is valid TOML a hand-editing user can write, and every
    one of them used to reach the GUI as a traceback before any window
    existed. The invariant is that the local provider survives all of them.
    """
    import io
    from contextlib import redirect_stderr

    from llamachat import providers

    # `providers` itself is not a table. Three shapes, and note they used to
    # raise two different exception types, which is why the guard is an
    # isinstance test and not a try/except.
    for bad in ("oops", ["a"], 5):
        fallen_back = providers.parse(
            {"base_url": "http://localhost:8181", "providers": bad}
        )
        assert set(fallen_back) == {"local"}, bad
        assert fallen_back["local"].base_url == "http://localhost:8181"

    # A single entry that is not a table. The list case is the one that
    # matters: [[providers.a]] is a plausible slip for [providers.a].
    for bad in (5, "x", ["x"], [{"base_url": "http://y.example.org"}]):
        mixed = providers.parse(
            {"providers": {"local": {"base_url": "http://x.example.org"},
                           "a": bad}}
        )
        assert set(mixed) == {"local"}, bad

    # A None entry, which TOML cannot produce but callers can, is still
    # skipped rather than crashing.
    assert providers.parse({"providers": {"a": None}}) == {}

    # Each skip says which provider it dropped and why, so a user running
    # from a terminal has something to act on.
    err = io.StringIO()
    with redirect_stderr(err):
        providers.parse(
            {
                "providers": {
                    "listy": [{"base_url": "http://x.example.org"}],
                    "urlless": {"api_key": "env:SOME_VAR"},
                    "a:b": {"base_url": "http://y.example.org"},
                }
            }
        )
    messages = err.getvalue()
    assert "listy" in messages and "urlless" in messages and "a:b" in messages
    assert "base_url" in messages  # the missing-URL case names what is missing
    # A list is the double-bracket slip, so the message names the fix. Any
    # other scalar was not written that way, and must not be told to change a
    # bracket it never had.
    assert "not [[providers.listy]]" in messages
    scalar = io.StringIO()
    with redirect_stderr(scalar):
        providers.parse({"providers": {"n": 5}})
    assert "[[" not in scalar.getvalue(), scalar.getvalue()

    # A malformed [[providers.local]] is the loudest case that needs saying,
    # not the quietest: the merge below it rebuilds local from the bare URL,
    # so the app comes up working and every field the user set is discarded
    # silently. The skip is reported at the merge site because the loop never
    # sees this entry.
    err = io.StringIO()
    with redirect_stderr(err):
        clobbered = providers.parse(
            {
                "base_url": "http://localhost:8181",
                "providers": {"local": [{"api_key": "env:SOME_VAR",
                                         "filter": ["qwen"],
                                         "ctx_size": 32768}]},
            }
        )
    assert "local" in err.getvalue()
    assert "not [[providers.local]]" in err.getvalue()
    # The local provider survives, which is the non-negotiable.
    assert clobbered["local"].base_url == "http://localhost:8181"
    # Pinning the loss rather than only the survival: these fields are gone,
    # and the warning above is the only thing that tells the user so.
    assert clobbered["local"].api_key == ""
    assert clobbered["local"].filter == []
    assert clobbered["local"].ctx_size is None

    # Without a bare base_url there is nothing to rebuild local from, so it
    # vanishes entirely. config.DEFAULTS always supplies one, which is what
    # keeps the real app safe; this pins that the safety net is that default
    # and not something parse() does on its own.
    assert providers.parse(
        {"providers": {"local": [{"base_url": "http://y.example.org"}]}}
    ) == {}
    assert "base_url" in config.DEFAULTS
    print("ok  malformed provider shapes are skipped, not raised")


def test_unusable_config_exits():
    """A config that cannot load exits 1 with a message, not a traceback.

    Runs the real entry point in a child process rather than calling main()
    in-process. CONFIG_PATH is read from the environment at import time and
    baked into load()'s default argument, so rebinding the constant after
    import does nothing: only a fresh interpreter with a redirected
    XDG_CONFIG_HOME actually moves the file the app reads.
    """
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "llamachat" / "config.toml"
        path.parent.mkdir(parents=True)
        # A plain syntax error, the commonest way a hand-edited file breaks.
        path.write_text('base_url = "http://localhost:8181"\nsocket = = ""\n')

        env = dict(os.environ, XDG_CONFIG_HOME=tmp)
        # --ping is the cheapest path that still loads the config, and it
        # fails before it looks for a socket, so an instance actually running
        # on this machine cannot turn this into a false pass.
        proc = subprocess.run(
            [sys.executable, "-m", "llamachat", "--ping"],
            capture_output=True, text=True, env=env,
            cwd=str(Path(__file__).resolve().parent),
        )

    assert proc.returncode == 1, proc.returncode
    # A traceback here would mean the app died rather than reported.
    assert "Traceback" not in proc.stderr, proc.stderr
    # The path is what makes the message actionable: it says which file to fix.
    assert str(path) in proc.stderr, proc.stderr
    assert "unusable" in proc.stderr
    # tomllib names the line, and that detail survives into the report.
    assert "line" in proc.stderr, proc.stderr
    print("ok  unusable config exits without a traceback")


def test_model_ids_and_filtering():
    """Ids are provider:model, local stays bare, filters are substrings."""
    from llamachat import providers

    table = providers.parse(
        {
            "providers": {
                "local": {"base_url": "http://localhost:8181"},
                "together": {
                    "base_url": "https://api.example.org",
                    "filter": ["qwen", "deepseek"],
                },
                "unfiltered": {"base_url": "https://api2.example.org"},
            }
        }
    )

    # Local models carry no prefix, in the dropdown and in the database.
    assert providers.qualify("local", "gemma4") == "gemma4"
    assert providers.qualify("together", "Qwen/Qwen2.5") == "together:Qwen/Qwen2.5"

    # Splitting is the inverse, and only for providers that exist.
    assert providers.split("gemma4", table) == ("local", "gemma4")
    assert providers.split("together:Qwen/Qwen2.5", table) == (
        "together",
        "Qwen/Qwen2.5",
    )
    # An unknown prefix is part of the model name, not a provider. This is
    # what keeps a local model whose name contains a colon working.
    assert providers.split("weird:name", table) == ("local", "weird:name")
    # Only the first colon splits.
    assert providers.split("together:a:b", table) == ("together", "a:b")
    # Task 9 addresses a provider itself with an empty model name.
    assert providers.split("together:", table) == ("together", "")

    # Filtering is case-insensitive substring, any match wins.
    listed = [
        "Qwen/Qwen2.5-72B-Instruct-Turbo",
        "deepseek-ai/DeepSeek-V3",
        "meta-llama/Llama-3.3-70B",
    ]
    kept = providers.apply_filter(table["together"], listed)
    assert kept == [
        "Qwen/Qwen2.5-72B-Instruct-Turbo",
        "deepseek-ai/DeepSeek-V3",
    ]

    # No filter means everything.
    assert providers.apply_filter(table["unfiltered"], listed) == listed
    # The local provider is never filtered even if one is configured.
    table["local"].filter = ["nothing-matches-this"]
    assert providers.apply_filter(table["local"], listed) == listed
    # A filter matching nothing yields nothing, it does not fall back to all.
    table["together"].filter = ["zzz"]
    assert providers.apply_filter(table["together"], listed) == []
    # An empty needle is a typo rather than a request to hide everything, so
    # it means no filter. The opposite of the "zzz" case above, deliberately.
    table["together"].filter = [""]
    assert providers.apply_filter(table["together"], listed) == listed
    print("ok  model ids and filtering")


def test_key_resolution():
    """api_key is prefix-dispatched, resolved lazily and cached."""
    from llamachat import providers

    resolver = providers.KeyResolver()

    # No key configured: no Authorization header, and nothing is run.
    empty = providers.Provider(name="local", base_url="http://x.example.org")
    assert resolver.resolve(empty) == ""

    # A literal key is used as-is.
    literal = providers.Provider(
        name="p", base_url="http://x.example.org", api_key="sk-test-not-a-real-key"
    )
    assert resolver.resolve(literal) == "sk-test-not-a-real-key"

    # env: reads the environment.
    os.environ["LLAMACHAT_TEST_KEY"] = "from-env"
    env = providers.Provider(
        name="e", base_url="http://x.example.org",
        api_key="env:LLAMACHAT_TEST_KEY",
    )
    assert resolver.resolve(env) == "from-env"
    del os.environ["LLAMACHAT_TEST_KEY"]

    # A missing env var is an error naming the provider, not a silent "".
    missing = providers.Provider(
        name="gone", base_url="http://x.example.org",
        api_key="env:LLAMACHAT_ABSENT_VAR",
    )
    try:
        resolver.resolve(missing)
        assert False, "a missing env var must raise"
    except providers.KeyResolutionError as exc:
        assert "gone" in str(exc)

    # pass: shells out. Substitute the runner rather than requiring gpg.
    calls = []

    def fake_run(cmd, timeout):
        calls.append((cmd, timeout))
        return "line-one\nline-two\n"

    passed = providers.Provider(
        name="together", base_url="http://x.example.org",
        api_key="pass:api/together",
    )
    cached = providers.KeyResolver(runner=fake_run)
    assert cached.resolve(passed) == "line-one"      # first line only
    assert calls[0][0] == ["pass", "show", "api/together"]
    assert calls[0][1] == providers.KEY_TIMEOUT

    # Cached: a second resolve must not shell out again.
    assert cached.resolve(passed) == "line-one"
    assert len(calls) == 1

    # But the cache follows the spec, not just the name. A reloaded config
    # that points the same provider at a different entry must re-resolve,
    # otherwise correcting a wrong entry appears to do nothing.
    moved = providers.Provider(
        name="together", base_url="http://x.example.org",
        api_key="pass:api/together-corrected",
    )
    assert cached.resolve(moved) == "line-one"
    assert len(calls) == 2
    assert calls[1][0] == ["pass", "show", "api/together-corrected"]

    # A failing pass is reported, naming the provider.
    def boom(cmd, timeout):
        raise OSError("pass: entry not found")

    try:
        providers.KeyResolver(runner=boom).resolve(passed)
        assert False, "a failing pass must raise"
    except providers.KeyResolutionError as exc:
        assert "together" in str(exc)

    # A timeout points at the pinentry never appearing, which is the silent
    # case, rather than at one the user can already see.
    def slow(cmd, timeout):
        raise subprocess.TimeoutExpired(cmd, timeout, output="partial-secret")

    try:
        providers.KeyResolver(runner=slow).resolve(passed)
        assert False, "a pass timeout must raise"
    except providers.KeyResolutionError as exc:
        assert "gpg-agent" in str(exc)
        # The partial stdout a timeout captures must never reach the message.
        assert "partial-secret" not in str(exc)
        # `from None` suppresses the chained-traceback display. It does not
        # clear __context__, and the chained traceback would not have shown
        # the secret anyway, so this pins tidiness, not secret hygiene.
        assert exc.__suppress_context__ and exc.__cause__ is None

    # A non-zero exit quotes gpg's stderr, which is the only useful part, and
    # never stdout, which is where the secret would be.
    def refused(cmd, timeout):
        raise subprocess.CalledProcessError(
            2, cmd, output="sk-test-not-a-real-key\n",
            stderr="gpg: decryption failed: No secret key\n",
        )

    try:
        providers.KeyResolver(runner=refused).resolve(passed)
        assert False, "a non-zero pass exit must raise"
    except providers.KeyResolutionError as exc:
        assert "No secret key" in str(exc)
        assert "sk-test-not-a-real-key" not in str(exc)

    # An empty or absent stderr falls back to the exit status rather than
    # reporting a blank reason.
    for blank in ("", None):
        def quiet(cmd, timeout, _s=blank):
            raise subprocess.CalledProcessError(3, cmd, output="", stderr=_s)

        try:
            providers.KeyResolver(runner=quiet).resolve(passed)
            assert False, "a non-zero pass exit must raise"
        except providers.KeyResolutionError as exc:
            assert "exit status 3" in str(exc)
            assert exc.__suppress_context__ and exc.__cause__ is None

    # A missing `pass` binary names the binary, not just "No such file".
    def absent(cmd, timeout):
        raise FileNotFoundError(2, "No such file or directory", "pass")

    try:
        providers.KeyResolver(runner=absent).resolve(passed)
        assert False, "a missing pass binary must raise"
    except providers.KeyResolutionError as exc:
        assert "not installed" in str(exc)
        assert "together" in str(exc)

    # Empty output is a failure too: an empty key would 401 confusingly.
    try:
        providers.KeyResolver(runner=lambda cmd, timeout: "  \n").resolve(passed)
        assert False, "empty pass output must raise"
    except providers.KeyResolutionError:
        pass
    print("ok  api key resolution")


def test_config_providers():
    """config.load exposes the provider table and the models.ini path."""
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "config.toml"

        # A legacy config: bare base_url only.
        path.write_text('base_url = "http://localhost:9999"\n')
        cfg = config.load(path)
        assert set(cfg.providers) == {"local"}
        assert cfg.providers["local"].base_url == "http://localhost:9999"
        # base_url stays populated: existing code still reads it.
        assert cfg.base_url == "http://localhost:9999"
        assert cfg.models_path == path.parent / "models.ini"

        # A config with an explicit cloud provider.
        path.write_text(
            'base_url = "http://localhost:9999"\n'
            "\n"
            "[providers.together]\n"
            'base_url = "https://api.example.org"\n'
            'api_key = "pass:api/together"\n'
            'filter = ["qwen"]\n'
            "ctx_size = 32768\n"
            "price_in = 0.6\n"
            "price_out = 0.9\n"
            "thinking_budget = 8192\n"
        )
        cfg = config.load(path)
        assert set(cfg.providers) == {"local", "together"}
        assert cfg.providers["together"].api_key == "pass:api/together"
        assert cfg.providers["together"].filter == ["qwen"]
        assert cfg.providers["together"].price_out == 0.9
        assert cfg.providers["together"].thinking_budget == 8192

        # A config with no base_url and no providers still loads, with the
        # built-in default synthesizing local.
        path.write_text("request_timeout = 60\n")
        cfg = config.load(path)
        assert set(cfg.providers) == {"local"}
        assert cfg.providers["local"].base_url == config.DEFAULTS["base_url"]
    print("ok  config provider table")


def test_replays_reasoning():
    """reasoning replay is per-provider, gated on search being on."""
    from llamachat import providers

    table = providers.parse(
        {
            "providers": {
                "local": {"base_url": "http://localhost:8181"},
                "deepseek": {"base_url": "https://api.deepseek.com",
                             "replay_reasoning": True},
                "siliconflow": {"base_url": "https://api.siliconflow.com"},
            }
        }
    )

    # Only the opted-in provider replays, and only when tools are sent.
    assert providers.replays_reasoning("deepseek:m", table, True) is True
    assert providers.replays_reasoning("deepseek:m", table, False) is False
    assert providers.replays_reasoning("siliconflow:m", table, True) is False
    assert providers.replays_reasoning("localmodel", table, True) is False

    # An unknown prefix resolves to local, which never replays.
    assert providers.replays_reasoning("unconfigured:m", table, True) is False
    print("ok  replays reasoning")


def test_models_store():
    """models.ini round-trips per-model metadata and the cancel record."""
    from llamachat import models

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "models.ini"
        store = models.ModelStore(path)

        # Nothing recorded yet.
        assert store.get("together:Qwen/Qwen2.5") is None
        assert store.was_offered("together:Qwen/Qwen2.5") is False

        store.save(
            "together:Qwen/Qwen2.5",
            models.ModelInfo(
                ctx_size=32768, vision=False, price_in=1.2, price_out=1.2
            ),
        )
        # A cancelled dialog records that it was offered, nothing more.
        store.mark_skipped("together:Llama-Vision-Free")

        # Reread from disk, not from memory: this is the round trip.
        fresh = models.ModelStore(path)
        info = fresh.get("together:Qwen/Qwen2.5")
        assert info.ctx_size == 32768
        assert info.vision is False
        assert info.price_in == 1.2
        assert info.price_out == 1.2
        assert fresh.was_offered("together:Qwen/Qwen2.5") is True

        assert fresh.get("together:Llama-Vision-Free") is None
        assert fresh.was_offered("together:Llama-Vision-Free") is True

        # Partial entries are legal: prices may be left blank.
        fresh.save("together:cheap", models.ModelInfo(ctx_size=8192))
        again = models.ModelStore(path)
        partial = again.get("together:cheap")
        assert partial.ctx_size == 8192
        assert partial.price_in is None
        assert partial.vision is None

        # A model id with a colon must survive being an ini section name.
        again.save("together:org/name:v2", models.ModelInfo(ctx_size=4096))
        assert models.ModelStore(path).get("together:org/name:v2").ctx_size == 4096

        # False must survive as False, not degrade to None: "no vision" is a
        # real answer that shadows a provider default, unlike "unknown".
        # Pins the lowercase wire format too, which Task 15 documents for
        # hand-editing. (Collapsing the bool branch in save() would still
        # pass, since _get_bool lowercases: this guards the meaning, not
        # that one branch.)
        again.save("together:novision", models.ModelInfo(vision=False))
        assert models.ModelStore(path).get("together:novision").vision is False
        assert "vision = false" in path.read_text(encoding="utf-8")

        # Cancelling a dialog over a model we already know must not erase it,
        # nor mark it skipped: the marker means "no real keys", so a section
        # holding both would be a state no reader is written to expect.
        again.mark_skipped("together:Qwen/Qwen2.5")
        assert models.ModelStore(path).get("together:Qwen/Qwen2.5").ctx_size == 32768
        assert models.SKIPPED not in path.read_text(encoding="utf-8").split(
            "[together:Qwen/Qwen2.5]"
        )[1].split("[")[0]

        # A model literally named DEFAULT must not write a [DEFAULT] section:
        # a stock-configparser reader, which Task 15 invites by documenting
        # this file, would read it as inherited defaults for every model.
        again.save("DEFAULT", models.ModelInfo(ctx_size=2048))
        assert "[DEFAULT]" not in path.read_text(encoding="utf-8")
        escaped = models.ModelStore(path)
        assert escaped.get("DEFAULT").ctx_size == 2048
        assert escaped.was_offered("DEFAULT") is True
        # The escape must not swallow a neighbouring id.
        assert escaped.was_offered("DEFAULTS") is False

        # A hand-edited file must degrade, not raise: "32k" is not an int.
        # The [DEFAULT] value is deliberately a *valid* int, so this catches
        # the leak itself rather than an unparseable value hiding it.
        path.write_text(
            "[DEFAULT]\nctx_size = 999\n\n"
            "[together:junk]\nctx_size = 32k\nprice_in = free\n\n"
            "[together:empty]\n",
            encoding="utf-8",
        )
        edited = models.ModelStore(path)
        assert edited.get("together:junk") is None
        assert edited.was_offered("together:junk") is True
        # Would be ctx_size 999, inherited from [DEFAULT], if the store used
        # configparser's real default section.
        assert edited.get("together:empty") is None

        # nan and inf parse cleanly through float(), so they would reach the
        # cost arithmetic and render as "$nan". Unknown is the honest answer.
        path.write_text(
            "[together:nan]\nprice_in = nan\nprice_out = inf\n\n"
            "[together:neg]\nprice_in = -inf\nctx_size = 8192\n",
            encoding="utf-8",
        )
        weird = models.ModelStore(path)
        assert weird.get("together:nan") is None
        # A bad price must not take the good ctx_size down with it.
        assert weird.get("together:neg").price_in is None
        assert weird.get("together:neg").ctx_size == 8192

        # A file that is not ini at all reads as empty rather than raising.
        path.write_text("this is not an ini file\n", encoding="utf-8")
        assert models.ModelStore(path).was_offered("together:junk") is False

        # Nor may a binary file take the app down at startup: that raises
        # UnicodeDecodeError, which is not a configparser.Error.
        path.write_bytes(b"\xff\xfe\x00not utf-8 at all\x00")
        assert models.ModelStore(path).was_offered("together:junk") is False
    print("ok  models.ini storage")


def test_metadata_and_cost():
    """models.ini beats provider defaults beats unknown; cost sums per model."""
    from llamachat import models, providers

    table = providers.parse(
        {
            "providers": {
                "local": {"base_url": "http://localhost:8181"},
                "together": {
                    "base_url": "https://api.example.org",
                    "api_key": "env:X",
                    "ctx_size": 32768,
                    "price_in": 0.6,
                    "price_out": 0.9,
                },
                "free": {"base_url": "https://api3.example.org"},
            }
        }
    )

    with tempfile.TemporaryDirectory() as tmp:
        store = models.ModelStore(Path(tmp) / "models.ini")
        store.save(
            "together:specific",
            models.ModelInfo(ctx_size=8192, vision=True, price_in=5.0),
        )

        # Layer 1: models.ini wins where it has a value.
        info = models.resolve("together:specific", table, store)
        assert info.ctx_size == 8192
        assert info.vision is True
        assert info.price_in == 5.0
        # Layer 2 fills the gap models.ini left: price_out was never set.
        assert info.price_out == 0.9

        # Layer 2 alone for a model with no models.ini entry.
        other = models.resolve("together:other", table, store)
        assert other.ctx_size == 32768
        assert other.price_in == 0.6
        assert other.vision is None       # layer 3: still unknown

        # Layer 3 throughout for a provider that configured nothing.
        bare = models.resolve("free:anything", table, store)
        assert bare.ctx_size is None
        assert bare.price_in is None

        # A local model resolves against the local provider rather than
        # falling through to an unrelated one: an unprefixed id and an
        # unconfigured prefix both split to local, and local configured
        # nothing here, so nothing may be inherited.
        assert models.resolve("gemma4", table, store).ctx_size is None
        # A models.ini entry is keyed by the full id, so a local model's
        # own metadata must still come back when no provider supplies any.
        store.save("gemma4", models.ModelInfo(ctx_size=4096))
        assert models.resolve("gemma4", table, store).ctx_size == 4096

        # A config with cloud providers but no local one is legal, and
        # split() still answers "local" for a bare id, so resolve() looks up
        # a provider that is not in the table. The stored metadata must
        # survive that, and nothing may be inherited from an unrelated
        # provider: charging a local model together's prices would invent
        # money that was never spent.
        cloud_only = providers.parse(
            {
                "providers": {
                    "together": {
                        "base_url": "https://api.example.org",
                        "api_key": "env:X",
                        "ctx_size": 32768,
                        "price_in": 0.6,
                    }
                }
            }
        )
        orphan = models.resolve("gemma4", cloud_only, store)
        assert orphan.ctx_size == 4096
        assert orphan.price_in is None
        assert models.is_billable("gemma4", cloud_only) is False
        assert models.conversation_cost(
            [{"model": "gemma4", "prompt_tokens": 1_000_000,
              "completion_tokens": 1_000_000}], cloud_only, store
        ) == 0.0

        # Cost: prompt at the input rate, completion at the output rate.
        #
        # The exact == below is safe for these particular prices, not
        # because the arithmetic is exact in general: n * p / n round-trips
        # to p for 0.6 and 5.0, and 0.6 + 5.0 is exactly 5.6, the same way
        # 0.1 + 0.2 is famously not 0.3. Adding a price here and asserting
        # its exact total can fail in the last bits and look like a costing
        # bug when it is only float representation. Use
        # abs(cost - expected) < 1e-9 for any price you add.
        rows = [
            {"model": "together:other", "prompt_tokens": 1_000_000,
             "completion_tokens": 1_000_000},
            # A pre-migration row: no counts, no model. Contributes zero.
            {"model": None, "prompt_tokens": None, "completion_tokens": None},
            # Counts but no model, which is what a reply interrupted before
            # it recorded its model leaves behind. There is no rate to apply,
            # so it must contribute zero rather than borrow another row's.
            {"model": None, "prompt_tokens": 1_000_000,
             "completion_tokens": 1_000_000},
        ]
        assert models.conversation_cost(rows, table, store) == 1.5

        # A mixed conversation prices each reply at what produced it.
        mixed = [
            {"model": "together:other", "prompt_tokens": 1_000_000,
             "completion_tokens": 0},
            {"model": "together:specific", "prompt_tokens": 1_000_000,
             "completion_tokens": 0},
        ]
        assert models.conversation_cost(mixed, table, store) == 5.6

        # An unpriced model contributes nothing rather than guessing.
        assert models.conversation_cost(
            [{"model": "free:anything", "prompt_tokens": 1_000_000,
              "completion_tokens": 0}], table, store
        ) == 0.0

        # A model priced on only one side still charges that side, rather
        # than being all-or-nothing: together:specific has no price_out of
        # its own but inherits one, so this uses a store-only provider.
        assert models.conversation_cost(
            [{"model": "free:half", "prompt_tokens": 1_000_000,
              "completion_tokens": 1_000_000}], table, store
        ) == 0.0
        store.save("free:half", models.ModelInfo(price_in=2.0))
        assert models.conversation_cost(
            [{"model": "free:half", "prompt_tokens": 1_000_000,
              "completion_tokens": 1_000_000}], table, store
        ) == 2.0
        # The other side alone, so neither rate is quietly gated on the
        # other being known.
        store.save("free:outonly", models.ModelInfo(price_out=3.0))
        assert models.conversation_cost(
            [{"model": "free:outonly", "prompt_tokens": 1_000_000,
              "completion_tokens": 1_000_000}], table, store
        ) == 3.0
        assert models.is_priced("free:outonly", table, store) is True

        # The projection prices input only: the reply's length is unknown
        # until it arrives, so guessing it would overstate every turn.
        assert models.projected_cost(
            1_000_000, "together:other", table, store
        ) == 0.6
        assert models.projected_cost(
            1_000_000, "free:anything", table, store
        ) == 0.0

        # Whether a model can be priced at all decides ? versus blank.
        assert models.is_priced("together:other", table, store) is True
        assert models.is_priced("free:anything", table, store) is False

        # A non-finite price is one no cost can be computed from, so the
        # answer is "not priced" rather than a True that sends the readout
        # down the priced branch to show "$nan". Both of today's writers
        # filter these out already; this states is_priced's own predicate
        # completely, for Task 10's dialog and Task 15's hand-editing.
        poisoned = providers.parse(
            {"providers": {"p": {"base_url": "http://x.example.org",
                                 "api_key": "env:X"}}}
        )
        poisoned["p"].price_in = float("nan")
        poisoned["p"].price_out = float("-inf")
        assert models.is_priced("p:x", poisoned, store) is False
        # Local is free, never unpriced.
        assert models.is_billable("gemma4", table) is False
        assert models.is_billable("free:anything", table) is False  # no api_key
        assert models.is_billable("together:other", table) is True

        # A local router behind an authenticating proxy is a legal config,
        # and it is still free. Without this the assertion above passes
        # only because the local provider happens to carry no api_key,
        # which is what the is_local test actually exists to cover.
        keyed_local = providers.parse(
            {
                "providers": {
                    "local": {
                        "base_url": "http://localhost:8181",
                        "api_key": "env:X",
                        "price_in": 9.0,
                    }
                }
            }
        )
        assert models.is_billable("gemma4", keyed_local) is False
        # A modelless row must stay free even when the local provider it
        # would otherwise resolve to carries prices, which is what makes
        # the empty-id guard in message_cost() load-bearing rather than
        # decorative.
        assert models.conversation_cost(
            [{"model": None, "prompt_tokens": 1_000_000,
              "completion_tokens": 1_000_000}], keyed_local, store
        ) == 0.0

        # Formatting: cents matter, so three decimals below a dollar.
        assert models.format_cost(0.0) == "$0.000"
        assert models.format_cost(1.5) == "$1.50"
        assert models.format_cost(12.345) == "$12.35"
    print("ok  metadata resolution and cost")


def test_token_column_migration():
    """A pre-token database opens, and new rows record counts and model."""
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
            "INSERT INTO messages VALUES (1,1,'assistant','older reply',0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        rows = history.messages(1)
        # The pre-migration row survives and reads as unknown, not as zero.
        assert rows[0]["content"] == "older reply"
        assert rows[0]["prompt_tokens"] is None
        assert rows[0]["completion_tokens"] is None
        assert rows[0]["model"] is None

        mid = history.add_message(1, "assistant", "")
        history.update_message(
            mid, "new reply",
            prompt_tokens=1200, completion_tokens=340,
            model="together:Qwen/Qwen2.5",
        )
        fresh = history.messages(1)[1]
        assert fresh["prompt_tokens"] == 1200
        assert fresh["completion_tokens"] == 340
        assert fresh["model"] == "together:Qwen/Qwen2.5"

        # Omitting them leaves stored values alone, as with reasoning.
        history.update_message(mid, "edited")
        kept = history.messages(1)[1]
        assert kept["content"] == "edited"
        assert kept["prompt_tokens"] == 1200
        assert kept["model"] == "together:Qwen/Qwen2.5"
        history.close()

    # A fresh database must end up with the same columns as a migrated one,
    # or cost would read back on one path and raise on the other. This
    # passes via SCHEMA or via _migrate() indifferently, which is the point:
    # both paths run on every open and either one alone suffices.
    from llamachat import models, providers

    with tempfile.TemporaryDirectory() as tmp:
        fresh_db = db.History(Path(tmp) / "new.db")
        columns = [
            row["name"]
            for row in fresh_db.conn.execute("PRAGMA table_info(messages)")
        ]
        assert {"prompt_tokens", "completion_tokens", "model"} <= set(columns)
        # SCHEMA puts created_at last while ALTER TABLE appends after it, so
        # the two paths hold the same columns in a different order. That is
        # tolerable only because every read goes by name: a positional read
        # of a message row would be right on one path and wrong on the other.
        assert columns[-1] == "created_at"
        sid = fresh_db.create_session("chat", "m", "t")
        mid = fresh_db.add_message(sid, "assistant", "")
        fresh_db.update_message(
            mid, "hi", prompt_tokens=7, completion_tokens=3,
            model="together:Qwen/Qwen2.5",
        )
        row = fresh_db.messages(sid)[0]
        # A stored row prices straight through models.message_cost(), which
        # proves the three column names are exactly the ones it reads.
        table = providers.parse(
            {
                "providers": {
                    "together": {
                        "base_url": "https://api.example.org",
                        "api_key": "env:X",
                        "price_in": 1_000_000.0,
                        "price_out": 1_000_000.0,
                    }
                }
            }
        )
        store = models.ModelStore(Path(tmp) / "models.ini")
        assert models.conversation_cost([row], table, store) == 10.0

        # Reopening runs _migrate() again over columns that already exist.
        fresh_db.close()
        again = db.History(Path(tmp) / "new.db")
        assert again.messages(sid)[0]["prompt_tokens"] == 7
        again.close()
    print("ok  token column migration")


def test_usage_columns():
    """Provider, raw usage and the GUI estimate persist per reply."""
    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "u.db")
        sid = history.create_session("chat", "m", "t")
        mid = history.add_message(sid, "assistant", "")
        usage = '{"prompt_tokens":100,"completion_tokens":20,"total_tokens":120}'
        history.update_message(
            mid, "done",
            prompt_tokens=100, completion_tokens=20,
            model="siliconflow:zai-org/GLM-5.2", provider="siliconflow",
            usage_json=usage, reported_cost_usd=0.000212,
        )
        row = history.messages(sid)[0]
        assert row["provider"] == "siliconflow"
        assert row["usage_json"] == usage
        assert row["reported_cost_usd"] == 0.000212

        # None means keep, exactly like the token columns.
        history.update_message(mid, "edited")
        kept = history.messages(sid)[0]
        assert kept["provider"] == "siliconflow"
        assert kept["usage_json"] == usage
        assert kept["reported_cost_usd"] == 0.000212
        history.close()
    print("ok  usage columns")


def test_usage_accumulation():
    """Every round of a searched turn reaches usage_all, not only the last.

    Each search round is its own billed API call, so dropping the earlier
    rounds' usage would under-report what a searched turn cost.
    """
    from llamachat import ui

    class FakeClient:
        def client_for(self, model):
            return self

        def wire_name(self, model):
            return model

        def stream_chat(self, wire, messages, search_cfg=None):
            # One usage block per round: a tool round, then the final answer.
            yield ("usage", '{"prompt_tokens": 100, "completion_tokens": 5,'
                            ' "total_tokens": 105}')
            yield ("usage", '{"prompt_tokens": 140, "completion_tokens": 60,'
                            ' "total_tokens": 200}')

    worker = ui.StreamWorker(FakeClient(), "m", [])
    got: list[str] = []
    worker.usage_all.connect(got.append)
    worker.run()

    assert len(got) == 1, got
    blocks = json.loads(got[0])
    assert [b["prompt_tokens"] for b in blocks] == [100, 140]
    assert [b["completion_tokens"] for b in blocks] == [5, 60]

    # A turn with no usage chunk at all (no include_usage support) yields
    # an empty array, which the window stores as NULL.
    class SilentClient(FakeClient):
        def stream_chat(self, wire, messages, search_cfg=None):
            yield ("content", "hi")
            return

    worker2 = ui.StreamWorker(SilentClient(), "m", [])
    got2: list[str] = []
    worker2.usage_all.connect(got2.append)
    worker2.run()
    assert got2 == ["[]"], got2
    print("ok  usage accumulation")


def test_usage_column_migration():
    """A pre-provider database opens, and the new columns read as NULL."""
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " prompt_tokens INTEGER, completion_tokens INTEGER, model TEXT,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
            "INSERT INTO messages VALUES (1,1,'assistant','older reply',"
            "1200,340,'m',0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        # The pre-migration row reads as unknown, never as an empty string
        # or a fabricated zero.
        old = history.messages(1)[0]
        assert old["provider"] is None
        assert old["usage_json"] is None
        assert old["reported_cost_usd"] is None

        mid = history.add_message(1, "assistant", "")
        history.update_message(
            mid, "new reply", prompt_tokens=10, completion_tokens=5,
            model="q:r", provider="q", usage_json='{"total_tokens":15}',
        )
        fresh = history.messages(1)[1]
        assert fresh["provider"] == "q"
        assert fresh["usage_json"] == '{"total_tokens":15}'
        assert fresh["reported_cost_usd"] is None
        history.close()
    print("ok  usage column migration")


def test_client_auth_header():
    """A client with a key sends Bearer auth; one without sends no header."""
    sent = {}

    class _HttpxResponse:
        """Enough of an httpx response for Client.models().

        The existing _FakeResponse in this file wraps bytes for urlopen and
        has neither .json() nor .raise_for_status(), so it cannot stand in
        for an httpx call.
        """

        status_code = 200

        def __init__(self, payload):
            self._payload = payload

        def raise_for_status(self):
            return None

        def json(self):
            return self._payload

    def _recorder(url, timeout=None, headers=None):
        sent["url"] = url
        sent["headers"] = headers or {}
        return _HttpxResponse({"data": [{"id": "m1"}]})

    import httpx
    with _patched(httpx, "get", _recorder):
        assert backend.Client("http://x.example.org").models() == ["m1"]
        assert "Authorization" not in sent["headers"]

        backend.Client(
            "http://x.example.org", api_key="sk-test-not-a-real-key"
        ).models()
        assert sent["headers"]["Authorization"] == "Bearer sk-test-not-a-real-key"

    # A provider that returns a bare array instead of {"data": [...]}
    # must still parse: some OpenAI-compatible endpoints skip the envelope.
    def _bare_list(url, timeout=None, headers=None):
        return _HttpxResponse([{"id": "bare"}])
    with _patched(httpx, "get", _bare_list):
        assert backend.Client("http://x.example.org").models() == ["bare"]

    # The streaming path is the one that carries every real turn, and no
    # other test reaches it: the search tests all stub _stream_once out.
    class _Stream:
        status_code = 200

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def iter_lines(self):
            return iter(['data: {"choices":[{"delta":{"content":"hi"}}]}'])

    def _stream_recorder(method, url, json=None, timeout=None, headers=None):
        sent["headers"] = headers or {}
        return _Stream()

    with _patched(httpx, "stream", _stream_recorder):
        client = backend.Client("http://x.example.org")
        assert list(client._stream_once("m", [], tools=None)) == [("content", "hi")]
        assert "Authorization" not in sent["headers"]

        keyed = backend.Client(
            "http://x.example.org", api_key="sk-test-not-a-real-key"
        )
        list(keyed._stream_once("m", [], tools=None))
        assert sent["headers"]["Authorization"] == "Bearer sk-test-not-a-real-key"

    # The key is not on the client's repr, which reaches logs and tracebacks.
    assert "sk-test-not-a-real-key" not in repr(keyed)
    print("ok  client authorization header")


def test_debug_log():
    """The diagnostic log is off unless $LLAMACHAT_DEBUG_LOG names a path."""
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "debug.log"
        saved = os.environ.get("LLAMACHAT_DEBUG_LOG")
        try:
            # Unset -> a no-op that creates nothing to rotate.
            os.environ.pop("LLAMACHAT_DEBUG_LOG", None)
            backend._debug_log("must not be written")
            assert not path.exists()

            # Set -> appends timestamped records.
            os.environ["LLAMACHAT_DEBUG_LOG"] = str(path)
            backend._debug_log("first")
            backend._debug_log("second")
            lines = path.read_text(encoding="utf-8").strip().splitlines()
            assert len(lines) == 2, lines
            assert lines[0].startswith("[") and "first" in lines[0]
            assert "second" in lines[1]
        finally:
            if saved is None:
                os.environ.pop("LLAMACHAT_DEBUG_LOG", None)
            else:
                os.environ["LLAMACHAT_DEBUG_LOG"] = saved
    print("ok  debug log")


def test_stream_body_for_cloud():
    """Cloud providers get max_tokens and thinking_budget; local does not."""
    captured = {}

    class _Stream:
        status_code = 200

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def iter_lines(self):
            return iter(
                ['data: {"choices":[{"delta":{"content":"hi"}}]}']
            )

    def _recorder(method, url, json=None, timeout=None, headers=None):
        captured["json"] = json
        return _Stream()

    import httpx
    with _patched(httpx, "stream", _recorder):
        # Local client: no max_tokens, no thinking_budget.
        local = backend.Client("http://localhost:8181", is_local=True)
        list(local._stream_once("m", [], tools=None))
        assert "max_tokens" not in captured["json"]
        assert "thinking_budget" not in captured["json"]

        # Cloud client with thinking_budget set.
        cloud = backend.Client(
            "http://api.example.org",
            is_local=False,
            thinking_budget=8192,
        )
        list(cloud._stream_once("m", [], tools=None))
        assert captured["json"]["max_tokens"] == 32768
        assert captured["json"]["thinking_budget"] == 8192

        # Cloud client without thinking_budget: max_tokens still sent.
        cloud_no_budget = backend.Client(
            "http://api.example.org", is_local=False
        )
        list(cloud_no_budget._stream_once("m", [], tools=None))
        assert captured["json"]["max_tokens"] == 32768
        assert "thinking_budget" not in captured["json"]
    print("ok  stream body for cloud providers")


def test_multi_client():
    """Models fan out across providers; requests route by model id."""
    from llamachat import providers

    table = providers.parse(
        {
            "providers": {
                "local": {"base_url": "http://localhost:8181"},
                "together": {
                    "base_url": "https://api.example.org",
                    "api_key": "env:MULTI_TEST_KEY",
                    "filter": ["qwen"],
                },
                "down": {"base_url": "https://dead.example.org"},
            }
        }
    )
    os.environ["MULTI_TEST_KEY"] = "sk-test-not-a-real-key"

    listings = {
        "http://localhost:8181": ["gemma4", "qwen3.5-9b"],
        "https://api.example.org": [
            "Qwen/Qwen2.5-72B",
            "meta-llama/Llama-3.3-70B",
        ],
    }
    built = []

    class _StubClient:
        def __init__(self, base_url, timeout=300, api_key="", **kwargs):
            self.base_url = base_url
            self.api_key = api_key
            built.append(self)

        def models(self, list_timeout=30):
            if self.base_url not in listings:
                raise backend.BackendError(f"cannot reach {self.base_url}")
            return listings[self.base_url]

    multi = backend.MultiClient(
        table, timeout=300, resolver=providers.KeyResolver(),
        client_factory=_StubClient,
    )
    listed, problems = multi.models()

    # Local models stay bare, cloud ones are prefixed, and the filter cut
    # the Llama model out of together's listing.
    assert listed == ["gemma4", "qwen3.5-9b", "together:Qwen/Qwen2.5-72B"]

    # The unreachable provider is reported, and did not break the rest.
    assert any("down" in p for p in problems)

    # Routing: the client for a cloud model carries that provider's key.
    client = multi.client_for("together:Qwen/Qwen2.5-72B")
    assert client.base_url == "https://api.example.org"
    assert client.api_key == "sk-test-not-a-real-key"

    # And a local model gets the local client with no key at all.
    local = multi.client_for("gemma4")
    assert local.base_url == "http://localhost:8181"
    assert local.api_key == ""

    # The bare model name is what goes on the wire, not the prefixed id.
    assert multi.wire_name("together:Qwen/Qwen2.5-72B") == "Qwen/Qwen2.5-72B"
    assert multi.wire_name("gemma4") == "gemma4"

    # A filter that matches nothing is reported by name with counts.
    table["together"].filter = ["zzz"]
    empty = backend.MultiClient(
        table, timeout=300, resolver=providers.KeyResolver(),
        client_factory=_StubClient,
    )
    _, notes = empty.models()
    assert any("together: 0 of 2" in n for n in notes)

    del os.environ["MULTI_TEST_KEY"]
    print("ok  multi-provider client")


def test_model_dialog_values():
    """The dialog's field text converts to ModelInfo, blanks meaning unknown."""
    from llamachat import models, modeldialog

    # Everything filled in.
    info = modeldialog.to_info(
        ctx_text="32768", vision=True, in_text="1.2", out_text="0.9"
    )
    assert info.ctx_size == 32768
    assert info.vision is True
    assert info.price_in == 1.2
    assert info.price_out == 0.9

    # Blank prices are legal and mean unpriced, not free.
    blank = modeldialog.to_info(
        ctx_text="8192", vision=False, in_text="", out_text="  "
    )
    assert blank.ctx_size == 8192
    assert blank.price_in is None
    assert blank.price_out is None

    # Garbage reads as unknown rather than crashing the dialog.
    junk = modeldialog.to_info(
        ctx_text="not a number", vision=False, in_text="free", out_text=""
    )
    assert junk.ctx_size is None
    assert junk.price_in is None

    # Vision is tri-state through a binary checkbox: an unchecked box with an
    # unknown prefill stays unknown, instead of writing False, which would
    # shadow a provider-level vision=True through models.resolve's pick(). A
    # known prefill (True or False) left unchecked is a real "no".
    assert modeldialog.to_info(
        ctx_text="", vision=True, in_text="", out_text=""
    ).vision is True
    assert modeldialog.to_info(
        ctx_text="", vision=False, in_text="", out_text=""
    ).vision is None
    assert modeldialog.to_info(
        ctx_text="", vision=False, in_text="", out_text="", vision_prefill=True
    ).vision is False
    assert modeldialog.to_info(
        ctx_text="", vision=False, in_text="", out_text="", vision_prefill=False
    ).vision is False

    # Prefill is the inverse: unknown becomes an empty field.
    assert modeldialog.to_fields(models.ModelInfo()) == ("", False, "", "")
    assert modeldialog.to_fields(
        models.ModelInfo(ctx_size=4096, vision=True, price_in=0.5)
    ) == ("4096", True, "0.5", "")
    print("ok  model dialog value conversion")


def test_cost_label_text():
    """The label distinguishes free, unpriced, and a real figure."""
    from llamachat import ui

    # A local model costs nothing, so the label says nothing.
    assert ui.cost_text(spent=0.0, projected=0.0, billable=False, priced=False) == ""

    # A cloud model whose price was never entered: ? rather than blank, so
    # it cannot be mistaken for free.
    assert ui.cost_text(
        spent=0.0, projected=0.0, billable=True, priced=False
    ) == "?"

    # Spent so far, with nothing composed yet.
    assert ui.cost_text(
        spent=0.043, projected=0.0, billable=True, priced=True
    ) == "$0.043"

    # Spent plus what sending the draft would add, kept visually separate.
    assert ui.cost_text(
        spent=0.043, projected=0.011, billable=True, priced=True
    ) == "$0.043 +$0.011"

    # A fresh conversation on a priced model still shows the projection.
    assert ui.cost_text(
        spent=0.0, projected=0.002, billable=True, priced=True
    ) == "$0.000 +$0.002"
    print("ok  cost label text")


class _FakeResponse:
    """Enough of an http.client response for urlopen's context manager."""

    def __init__(self, body: bytes):
        self._body = body

    def read(self):
        return self._body

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False


def _fake_urlopen(body, capture=None):
    """A urlopen replacement returning `body`, or raising it when an error."""

    def opener(request, timeout=None):
        if capture is not None:
            capture.append(request.full_url)
        if isinstance(body, Exception):
            raise body
        return _FakeResponse(body)

    return opener


def _with_urlopen(body, capture=None):
    """Swap search's urlopen for a fake. Returns the original to restore."""
    import urllib.request

    original = urllib.request.urlopen
    urllib.request.urlopen = _fake_urlopen(body, capture)
    return original


@contextlib.contextmanager
def _patched(obj, name, value):
    """Swap an attribute for the duration of the block, restored even on error."""
    original = getattr(obj, name)
    setattr(obj, name, value)
    try:
        yield
    finally:
        setattr(obj, name, original)


def test_search_tool_schema():
    from llamachat import search

    schema = search.TOOL_SCHEMA
    assert schema["type"] == "function"
    function = schema["function"]
    assert function["name"] == "web_search"
    assert function["description"]
    params = function["parameters"]
    assert params["type"] == "object"
    assert params["required"] == ["query"]
    assert params["properties"]["query"]["type"] == "string"

    # Disabled search must not put a tools key on the wire at all, or a
    # model that ignores it still pays for the tokens.
    sent = []
    client = backend.Client("http://x")
    client._stream_once = lambda model, messages, tools: (
        sent.append(tools) or iter([("content", "hi")])
    )
    out = list(client.stream_chat("m", [], backend.SearchConfig(enabled=False)))
    assert out == [("content", "hi")]
    assert sent == [None], sent

    # Two searches per turn, which a follow-up call now survives: the final
    # round's tool result says the tool is gone, and quantized KV cache, the
    # other half of the empty-reply bug, is a server-side setting.
    assert config.DEFAULTS["max_searches"] == 2
    assert config.load(Path("/nonexistent/config.toml")).max_searches == 2

    # A configured-but-urlless setup resolves to disabled at config load.
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "config.toml"
        path.write_text('search_enabled = true\nsearch_url = ""\n')
        assert config.load(path).search_enabled is False

        path.write_text(
            'search_enabled = true\nsearch_url = "http://searx.local:8888/"\n'
        )
        cfg = config.load(path)
        assert cfg.search_enabled is True
        assert cfg.search_url == "http://searx.local:8888"  # trailing / gone
    print("ok  search tool schema")


def test_search_results_sanitising():
    import urllib.request

    from llamachat import search
    from llamachat.ui import SEARCH_SCHEME, _markdown_to_fragment, _search_html

    payload = json.dumps(
        {
            "results": [
                {
                    "title": "First",
                    "url": "https://example.com/a",
                    "content": "x" * 500,
                    # Fields the model has no business seeing.
                    "engine": "duckduckgo",
                    "score": 1.5,
                    "positions": [1],
                },
                {"title": "Second", "url": "https://example.com/b", "content": "s"},
                {"title": "Third", "url": "https://example.com/c", "content": "t"},
            ]
        }
    ).encode()

    original = _with_urlopen(payload)
    try:
        results = search.search("http://searx", "q", count=2, snippet_chars=100)
    finally:
        urllib.request.urlopen = original

    # count caps the list; only three fields survive; snippets truncate.
    assert len(results) == 2, results
    assert set(results[0]) == {"title", "url", "content"}, results[0]
    assert len(results[0]["content"]) == 100
    assert "engine" not in results[0]

    # Result text is escaped on display, so markup in a snippet stays text.
    hostile = [
        {
            "query": "q",
            "error": "",
            "results": [
                {
                    "title": "<script>alert(1)</script>",
                    "url": "https://example.com/x",
                    "content": f"click [here]({SEARCH_SCHEME}0)",
                }
            ],
        }
    ]
    rendered = _search_html(hostile, 0, expanded=True)
    assert "<script>" not in rendered, rendered
    assert "&lt;script&gt;" in rendered
    # The scheme appears exactly once, as this block's own toggle. The copy
    # inside the snippet stayed literal text rather than becoming an href.
    assert rendered.count(f'href="{SEARCH_SCHEME}') == 1, rendered

    # A reply forging the scheme as a markdown link gets it defused.
    forged = _markdown_to_fragment(f"[expand]({SEARCH_SCHEME}0)")
    assert SEARCH_SCHEME not in forged, forged
    assert "blocked:" in forged
    print("ok  search result sanitising")


def test_tool_call_accumulation():
    from llamachat import search

    calls: dict = {}
    # The id and name arrive first, the arguments in pieces after it.
    search.accumulate(calls, [
        {"index": 0, "id": "call_1", "function": {"name": "web_search", "arguments": ""}}
    ])
    search.accumulate(calls, [{"index": 0, "function": {"arguments": '{"que'}}])
    search.accumulate(calls, [{"index": 0, "function": {"arguments": 'ry": "kern'}}])
    search.accumulate(calls, [{"index": 0, "function": {"arguments": 'el"}'}}])

    assert list(calls) == [0]
    assert calls[0]["id"] == "call_1"
    assert calls[0]["name"] == "web_search"
    assert search.parse_query(calls[0]["arguments"]) == "kernel"

    # Two calls in one round stay apart, keyed by index.
    pair: dict = {}
    search.accumulate(pair, [
        {"index": 0, "id": "a", "function": {"name": "web_search", "arguments": '{"query":"one"}'}},
        {"index": 1, "id": "b", "function": {"name": "web_search", "arguments": '{"query":"two"}'}},
    ])
    assert search.parse_query(pair[0]["arguments"]) == "one"
    assert search.parse_query(pair[1]["arguments"]) == "two"

    # Unusable arguments yield no query rather than an exception.
    assert search.parse_query("{not json") == ""
    assert search.parse_query("[]") == ""
    assert search.parse_query('{"query": 7}') == ""
    assert search.parse_query("") == ""
    print("ok  tool call accumulation")


def _tool_round(query: str = "q"):
    """One streamed round that asks for a search."""
    return [
        (
            "tool_calls",
            json.dumps(
                [
                    {
                        "index": 0,
                        "id": "call_1",
                        "function": {
                            "name": "web_search",
                            "arguments": json.dumps({"query": query}),
                        },
                    }
                ]
            ),
        ),
        ("tool_finish", ""),
    ]


def test_search_loop_cap():
    import urllib.request

    from llamachat import search

    sent_tools = []
    client = backend.Client("http://x")

    def always_tool_calls(model, messages, tools):
        sent_tools.append(tools)
        yield from _tool_round()

    client._stream_once = always_tool_calls
    original = _with_urlopen(json.dumps({"results": []}).encode())
    try:
        cfg = backend.SearchConfig(
            enabled=True, url="http://searx", max_searches=2
        )
        list(client.stream_chat("m", [{"role": "user", "content": "hi"}], cfg))
    finally:
        urllib.request.urlopen = original

    # max_searches rounds offer the tool, then one final round without it.
    assert len(sent_tools) == 3, sent_tools
    assert sent_tools[0] == [search.TOOL_SCHEMA]
    assert sent_tools[1] == [search.TOOL_SCHEMA]
    assert sent_tools[-1] is None, "the final round must withdraw the tool"

    # A round that never asks for a tool ends the turn immediately.
    quiet = []

    def no_tools(model, messages, tools):
        quiet.append(tools)
        yield ("content", "done")

    client._stream_once = no_tools
    out = list(client.stream_chat("m", [], backend.SearchConfig(
        enabled=True, url="http://searx")))
    assert out == [("content", "done")]
    assert len(quiet) == 1, quiet
    print("ok  search loop cap")


def test_search_failure_paths():
    import urllib.error
    import urllib.request

    from llamachat import search

    cases = [
        (TimeoutError("timed out"), "Cannot reach SearXNG"),
        (urllib.error.URLError("Connection refused"), "Cannot reach SearXNG"),
        (b"<html>not json</html>", "did not return JSON"),
    ]
    for body, expected in cases:
        original = _with_urlopen(body)
        try:
            failed = ""
            try:
                search.search("http://searx", "q")
            except search.SearchError as exc:
                failed = str(exc)
        finally:
            urllib.request.urlopen = original
        assert expected in failed, (body, failed)

    # Zero results with healthy engines is a success with an empty list:
    # the web really had nothing.
    original = _with_urlopen(json.dumps({"results": []}).encode())
    try:
        assert search.search("http://searx", "q") == []
    finally:
        urllib.request.urlopen = original

    # Zero results *because* every engine was rate-limited or CAPTCHA'd is
    # a failed search. Reporting it as "no results" would tell the model
    # the web is empty and invite an answer from stale training data.
    dead = json.dumps(
        {
            "results": [],
            "unresponsive_engines": [
                ["duckduckgo", "CAPTCHA"],
                ["brave", "Suspended: too many requests"],
            ],
        }
    ).encode()
    original = _with_urlopen(dead)
    try:
        failed = ""
        try:
            search.search("http://searx", "q")
        except search.SearchError as exc:
            failed = str(exc)
    finally:
        urllib.request.urlopen = original
    assert "every search engine failed" in failed, failed
    assert "duckduckgo: CAPTCHA" in failed, failed

    # Engines that failed while others still answered are not an error:
    # partial results are results.
    partial = json.dumps(
        {
            "results": [{"title": "T", "url": "https://e.com", "content": "c"}],
            "unresponsive_engines": [["brave", "timeout"]],
        }
    ).encode()
    original = _with_urlopen(partial)
    try:
        assert len(search.search("http://searx", "q")) == 1
    finally:
        urllib.request.urlopen = original

    # A malformed unresponsive_engines field must not crash the summary.
    assert search._unresponsive({"unresponsive_engines": "nonsense"}) == ""
    assert search._unresponsive({"unresponsive_engines": [["solo"]]}) == "solo"
    assert search._unresponsive({}) == ""

    # An unconfigured URL fails before any request is attempted.
    try:
        search.search("", "q")
        raise AssertionError("empty url must raise")
    except search.SearchError:
        pass

    # Every failure still completes the turn and still tells the model.
    client = backend.Client("http://x")
    rounds = [_tool_round("kernel"), [("content", "answered anyway")]]
    seen_messages = []

    def scripted(model, messages, tools):
        seen_messages.append([dict(m) for m in messages])
        yield from rounds[min(len(seen_messages) - 1, len(rounds) - 1)]

    client._stream_once = scripted
    original = _with_urlopen(urllib.error.URLError("Connection refused"))
    try:
        cfg = backend.SearchConfig(enabled=True, url="http://searx")
        out = list(client.stream_chat("m", [{"role": "user", "content": "hi"}], cfg))
    finally:
        urllib.request.urlopen = original

    assert ("content", "answered anyway") in out, out
    starts = [p for k, p in out if k == "search_start"]
    dones = [json.loads(p) for k, p in out if k == "search_done"]
    assert starts == ["kernel"], starts
    assert dones and dones[0]["error"], dones
    assert dones[0]["results"] == []

    # The second request carries the assistant tool call and a tool reply
    # naming the failure, so the model knows the search did not happen.
    second = seen_messages[1]
    assert second[-2]["role"] == "assistant"
    assert second[-2]["tool_calls"][0]["id"] == "call_1"
    tool_msg = second[-1]
    assert tool_msg["role"] == "tool"
    assert tool_msg["tool_call_id"] == "call_1"
    assert search.RESULT_PREFIX in tool_msg["content"]
    assert "error" in json.loads(
        tool_msg["content"][len(search.RESULT_PREFIX):].strip()
    )

    # A malformed tool call is answered rather than left dangling, or the
    # next request would be rejected for an unanswered call.
    bad = {"id": "call_9", "name": "web_search", "arguments": "{not json"}
    convo: list = []
    emitted = list(client._run_search(convo, bad, backend.SearchConfig(
        enabled=True, url="http://searx")))
    assert emitted == [], emitted  # nothing searched, nothing displayed
    assert convo[-1]["role"] == "tool"
    assert convo[-1]["tool_call_id"] == "call_9"
    print("ok  search failure paths")


def test_reasoning_content_preserved():
    """A tool-call round replays its reasoning_content verbatim.

    DeepSeek V3.2+/V4 and GLM-4.7+ on SiliconFlow emit chain-of-thought as
    reasoning_content and require it sent back unchanged on the assistant
    tool-call message. Dropping it breaks their multi-step tool flow, which
    surfaces as a turn that stops at the thinking with no answer.
    """
    import urllib.request

    from llamachat import search

    client = backend.Client("http://x")
    seen_messages = []

    def scripted(model, messages, tools):
        seen_messages.append([dict(m) for m in messages])
        if len(seen_messages) == 1:
            yield ("reasoning", "I should check the current version.")
            yield from _tool_round("latest kernel")
        else:
            yield ("content", "answered")

    client._stream_once = scripted
    original = _with_urlopen(json.dumps({"results": []}).encode())
    try:
        cfg = backend.SearchConfig(
            enabled=True, url="http://searx", max_searches=1
        )
        list(client.stream_chat("m", [{"role": "user", "content": "hi"}], cfg))
    finally:
        urllib.request.urlopen = original

    # The second request's assistant message must carry the thinking back.
    second = seen_messages[1]
    assistant = next(m for m in second if m["role"] == "assistant" and m.get("tool_calls"))
    assert assistant["reasoning_content"] == "I should check the current version.", assistant

    # The reasoning still reaches the UI unchanged.
    assert len(seen_messages) == 2, seen_messages
    print("ok  reasoning content preserved")


def test_tool_call_finish_with_usage():
    """finish_reason riding the usage chunk still triggers the search.

    DeepSeek's include_usage stream puts finish_reason: "tool_calls" on the
    same SSE line as the usage block. The old single-event parser returned
    that line as "usage" and never saw the finish, so the tool call was
    dropped and the turn ended at the thinking with no search.
    """
    import httpx
    import urllib.request

    def sse(obj):
        return "data: " + json.dumps(obj)

    lines = [
        sse({"choices": [{"delta": {"reasoning_content": "hmm"}}]}),
        sse({"choices": [{"delta": {"tool_calls": [
            {
                "index": 0,
                "id": "call_1",
                "type": "function",
                "function": {"name": "web_search", "arguments": '{"query":"kernel"}'},
            }
        ]}}]}),
        sse({
            "choices": [{
                "index": 0,
                "delta": {"content": "", "reasoning_content": None},
                "finish_reason": "tool_calls",
            }],
            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
        }),
        "data: [DONE]",
    ]

    class _Stream:
        status_code = 200

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def iter_lines(self):
            return iter(lines)

    def _fake_stream(method, url, json=None, timeout=None, headers=None):
        return _Stream()

    client = backend.Client("http://x")
    original = _with_urlopen(json.dumps({"results": []}).encode())
    try:
        with _patched(httpx, "stream", _fake_stream):
            cfg = backend.SearchConfig(enabled=True, url="http://searx", max_searches=1)
            out = list(client.stream_chat("m", [{"role": "user", "content": "hi"}], cfg))
    finally:
        urllib.request.urlopen = original

    kinds = [k for k, _ in out]
    assert "search_start" in kinds, kinds
    assert "search_done" in kinds, kinds
    assert ("reasoning", "hmm") in out, out
    print("ok  tool call finish with usage")


def test_search_storage():
    import sqlite3

    with tempfile.TemporaryDirectory() as tmp:
        history = db.History(Path(tmp) / "s.db")
        sid = history.create_session("chat", "m", "t")
        mid = history.add_message(sid, "assistant", "")

        records = [
            {
                "query": "latest kernel",
                "results": [
                    {
                        "title": "Kernel",
                        "url": "https://kernel.org",
                        "content": "snippet",
                    }
                ],
                "error": "",
            }
        ]
        history.update_message(mid, "7.1.5", "thinking", json.dumps(records))

        row = history.messages(sid)[0]
        assert row["content"] == "7.1.5"
        assert row["reasoning"] == "thinking"
        assert json.loads(row["searches"]) == records

        # Snippets stay out of the FTS index: web text the user never wrote
        # must not compete with their own messages.
        assert history.search("7.1.5") != []
        assert history.search("snippet") == []

        # Omitting the argument leaves stored searches untouched.
        history.update_message(mid, "7.1.6")
        assert json.loads(history.messages(sid)[0]["searches"]) == records

        # A message with no searches stores NULL rather than an empty list.
        plain = history.add_message(sid, "assistant", "")
        history.update_message(plain, "no search", "", None)
        assert history.messages(sid)[1]["searches"] is None
        history.close()

    # A database predating the column must still open and read as no
    # searches rather than raising.
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "old.db"
        conn = sqlite3.connect(path)
        conn.executescript(
            "CREATE TABLE sessions (id INTEGER PRIMARY KEY, mode TEXT,"
            " title TEXT, model TEXT, created_at INTEGER, updated_at INTEGER);"
            "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id INTEGER,"
            " role TEXT NOT NULL, content TEXT NOT NULL,"
            " created_at INTEGER NOT NULL);"
            "INSERT INTO sessions VALUES (1,'chat','old','m',0,0);"
            "INSERT INTO messages VALUES (1,1,'assistant','older',0);"
        )
        conn.commit()
        conn.close()

        history = db.History(path)
        from llamachat.ui import _column, _searches

        row = history.messages(1)[0]
        assert row["searches"] is None
        assert _searches(_column(row, "searches")) == []
        history.close()
    print("ok  search storage")


def test_date_note():
    import datetime

    from llamachat import ui
    from llamachat.ui import _date_note

    fixed = datetime.datetime(2026, 7, 31, 14, 30)
    note = _date_note(fixed)

    # The real date must be stated, or the model falls back on its cutoff.
    assert "Friday, 31 July 2026" in note, note
    # And it must be told not to date its own queries, which is the bug
    # that poisons results before they are even fetched.
    assert "year in a search query" in note, note

    # Composed onto the chosen prompt rather than replacing it, so a preset
    # keeps its instructions.
    class _Cfg:
        search_enabled = True

    class _Win:
        cfg = _Cfg()
        system_prompt_text = staticmethod(lambda: "Be terse.")
        _system_messages = ui.ChatWindow._system_messages

    msgs = _Win._system_messages(_Win())
    assert len(msgs) == 1
    assert msgs[0]["role"] == "system"
    assert msgs[0]["content"].startswith("Be terse.")
    assert "2026" in msgs[0]["content"]

    # With no system prompt selected, the date still goes: the model has no
    # clock either way.
    class _Bare(_Win):
        system_prompt_text = staticmethod(lambda: "")

    bare = _Bare._system_messages(_Bare())
    assert len(bare) == 1
    assert "Today's date is" in bare[0]["content"]

    # Search off means no date note and no system message at all.
    class _Off(_Bare):
        class cfg:
            search_enabled = False

    assert _Off._system_messages(_Off()) == []
    print("ok  date note")


def test_search_html():
    from llamachat.ui import SEARCH_SCHEME, _search_html, _searches

    assert _search_html([], 0, False) == ""

    ok = [
        {
            "query": "latest kernel",
            "error": "",
            "results": [
                {
                    "title": "Kernel.org",
                    "url": "https://kernel.org",
                    "content": "The Linux Kernel Archives",
                },
                {
                    "title": "Wikipedia",
                    "url": "https://en.wikipedia.org/wiki/Linux",
                    "content": "An operating system kernel",
                },
            ],
        }
    ]

    collapsed = _search_html(ok, 3, expanded=False)
    assert "▸" in collapsed
    assert "searched: latest kernel (2 results)" in collapsed
    assert f'href="{SEARCH_SCHEME}3"' in collapsed
    # Collapsed shows the summary only, never the sources.
    assert "kernel.org" not in collapsed

    expanded = _search_html(ok, 3, expanded=True)
    assert "▾" in expanded
    assert "Kernel.org" in expanded
    assert 'href="https://kernel.org"' in expanded
    assert "The Linux Kernel Archives" in expanded

    # A failure says so, and names the reason.
    failed = _search_html(
        [{"query": "kernel", "results": [], "error": "Cannot reach SearXNG"}],
        0,
        expanded=False,
    )
    assert "search failed: kernel" in failed
    assert "Cannot reach SearXNG" in failed

    # A non-web URL is shown but never becomes a clickable anchor.
    sneaky = _search_html(
        [
            {
                "query": "q",
                "error": "",
                "results": [
                    {"title": "T", "url": "file:///etc/passwd", "content": "c"}
                ],
            }
        ],
        0,
        expanded=True,
    )
    assert 'href="file:' not in sneaky, sneaky
    assert "file:///etc/passwd" in sneaky

    # The stored column decodes back into what rendering expects, and junk
    # in that column degrades to no block rather than raising.
    assert _searches(json.dumps(ok)) == ok
    assert _searches("") == []
    assert _searches("{not json") == []
    assert _searches('{"query": "not a list"}') == []
    print("ok  search html")


def test_final_round_note():
    """The last search result tells the model it has no searches left.

    Withdrawing the tool schema is invisible to the model: it tries a second
    search anyway and emits it as literal <tool_call> text, or trails off
    inside the thinking block, either way leaving the reply empty. Saying so
    in the tool result is what actually ends the turn with an answer. The
    note cannot be a trailing system message: this model's chat template
    raises "System message must be at the beginning".
    """
    from llamachat import search

    plain = search.tool_message("call_1", [{"title": "T", "url": "u", "content": "c"}])
    final = search.tool_message(
        "call_1", [{"title": "T", "url": "u", "content": "c"}], last=True
    )
    assert plain["role"] == "tool" and final["role"] == "tool"
    assert search.NO_MORE_SEARCHES not in plain["content"]
    assert search.NO_MORE_SEARCHES in final["content"]
    # The results themselves survive the note being appended.
    for message in (plain, final):
        assert '"title": "T"' in message["content"] or '"title":"T"' in message["content"]
    # A failed final search still gets the note, or the model retries.
    failed = search.tool_message("call_1", None, error="boom", last=True)
    assert "boom" in failed["content"]
    assert search.NO_MORE_SEARCHES in failed["content"]
    print("ok  final round note")


def test_title_cleaning():
    clean = backend.clean_title

    assert clean("Kernel build failure") == "Kernel build failure"
    # Quotes, trailing punctuation and a leading label all come off.
    assert clean('"Kernel build failure"') == "Kernel build failure"
    assert clean("Title: Kernel build failure.") == "Kernel build failure"
    assert clean("**Kernel build failure**") == "Kernel build failure"
    # Thinking that leaks into the body is dropped, answer kept.
    assert clean("<think>hmm, short</think>Kernel build") == "Kernel build"
    # An unclosed think block leaves nothing usable rather than a stray tag.
    assert clean("<think>still reasoning") == ""
    # Chatter before the title: the last non-empty line wins.
    assert clean("Sure, here you go:\nKernel build failure") == "Kernel build failure"
    # Too long is cut on a word boundary, not mid-word.
    long = clean("word " * 40)
    assert len(long) <= 60, long
    assert not long.endswith("wor"), long
    # Nothing usable yields nothing, so the caller keeps its fallback.
    assert clean("") == ""
    assert clean("   \n\n ") == ""
    assert clean('""') == ""
    print("ok  title cleaning")


def test_title_request():
    """Client.complete posts a non-streaming request and returns the text."""
    seen = {}

    class FakeResponse:
        status_code = 200
        text = ""

        def raise_for_status(self):
            pass

        def json(self):
            return {"choices": [{"message": {"content": " A Title "}}]}

    def fake_post(url, json=None, timeout=None, headers=None):
        seen["url"] = url
        seen["body"] = json
        seen["headers"] = headers or {}
        return FakeResponse()

    import httpx

    with _patched(httpx, "post", fake_post):
        client = backend.Client("http://router:8181")
        out = client.complete("m", [{"role": "user", "content": "hi"}], max_tokens=32)

    assert out == " A Title ", out
    assert seen["url"] == "http://router:8181/v1/chat/completions"
    assert seen["body"]["model"] == "m"
    assert seen["body"]["stream"] is False
    assert seen["body"]["max_tokens"] == 32
    # No tools are offered: a title turn must not trigger a search.
    assert "tools" not in seen["body"]
    # Thinking off, or a reasoning model spends the budget and returns "".
    assert seen["body"]["chat_template_kwargs"] == {"enable_thinking": False}
    # The local router needs no key, so the side errand carries no header.
    assert "Authorization" not in seen["headers"]

    # A keyed provider authenticates on this path too, not only on models().
    # It runs after the body assertions above because it overwrites `seen`.
    with _patched(httpx, "post", fake_post):
        keyed = backend.Client("http://router:8181", api_key="sk-test-not-a-real-key")
        assert keyed.complete("m", []) == " A Title "
    assert seen["headers"]["Authorization"] == "Bearer sk-test-not-a-real-key"

    # A reply with no choices is empty rather than an exception.
    class Empty(FakeResponse):
        def json(self):
            return {}

    with _patched(httpx, "post", lambda url, json=None, timeout=None, headers=None: Empty()):
        assert backend.Client("http://x").complete("m", []) == ""
    print("ok  title request")


def test_needs_title():
    """Which finished turns are worth asking the model to name.

    The trigger is a session still carrying its placeholder title, not the
    turn number: a first reply that comes back empty (the follow-up
    <tool_call> bug) must not forfeit titling for the whole session, and a
    one-shot window keeps appending to the same session until New is
    pressed.
    """
    needs = backend.needs_title
    q = "how do I build a custom kernel on slackware?"

    assert needs(q, q, "Fetch the sources.") is True
    # Nothing to title from: the empty-reply turn is skipped, not consumed.
    assert needs(q, q, "") is False
    assert needs(q, q, "   \n ") is False
    # Already named by the model, so leave it alone.
    assert needs("Building a custom kernel", q, "Fetch the sources.") is False
    # A long question is stored truncated; that still counts as untouched.
    long_q = "x" * 200
    assert needs(long_q[:60], long_q, "an answer") is True
    # A title the user or model set that happens to be short is not a
    # placeholder, even though it is under the cut-off.
    assert needs("Kernels", long_q, "an answer") is False
    print("ok  needs title")


def test_title_prompt():
    """The title request carries the exchange and asks for a short title."""
    messages = backend.title_messages("how do I build a kernel?", "Run make.")
    assert messages[-1]["role"] == "user"
    blob = json.dumps(messages)
    assert "how do I build a kernel?" in blob
    assert "Run make." in blob
    # A huge exchange is trimmed so titling never costs a full context.
    big = backend.title_messages("x" * 10000, "y" * 10000)
    assert len(json.dumps(big)) < 6000, len(json.dumps(big))
    print("ok  title prompt")


if __name__ == "__main__":
    test_presets()
    test_real_presets()
    test_fts_query_escaping()
    test_history_roundtrip()
    test_attachment_truncation()
    test_classify()
    test_sse_parsing()
    test_reasoning_storage()
    test_migration_adds_reasoning()
    test_user_content()
    test_prompt_store()
    test_token_estimate()
    test_usage_parsing()
    test_session_prompt_storage()
    test_prompt_column_migration()
    test_markdown_rendering()
    test_unbalanced_backticks()
    test_sidebar_toggle()
    test_shortcuts()
    test_status_dismiss()
    test_system_qt_theme_guard()
    test_version_matches_changelog()
    test_venv_discovery()
    test_config_defaults()
    test_provider_parsing()
    test_provider_malformed_shapes()
    test_unusable_config_exits()
    test_model_ids_and_filtering()
    test_key_resolution()
    test_config_providers()
    test_replays_reasoning()
    test_models_store()
    test_metadata_and_cost()
    test_token_column_migration()
    test_usage_columns()
    test_usage_accumulation()
    test_usage_column_migration()
    test_client_auth_header()
    test_debug_log()
    test_stream_body_for_cloud()
    test_multi_client()
    test_model_dialog_values()
    test_cost_label_text()
    test_search_tool_schema()
    test_search_results_sanitising()
    test_tool_call_accumulation()
    test_search_loop_cap()
    test_search_failure_paths()
    test_reasoning_content_preserved()
    test_tool_call_finish_with_usage()
    test_search_storage()
    test_date_note()
    test_search_html()
    test_final_round_note()
    test_title_cleaning()
    test_title_request()
    test_needs_title()
    test_title_prompt()
    test_code_block_wrapping()
    test_code_block_copy_links()
    test_code_block_copy_states()
    print("\nall checks passed")