LLeMbas/tests/test_chat_tail.py

"""Turns that arrive without the page having asked for them.

A reply can begin outside a request: `jobs.wake` writes a completion turn and
calls `generation.ensure` when a background job finishes on an idle chat. The
browser has no way to hear about it -- the only stream here is per-message and
it is opened by the `sse-connect` on an incomplete assistant bubble, which is a
bubble the page does not have, because the reply that created it started
somewhere else. So the page polls, and this is that poll.

Most of what can go wrong is a duplicate or an avalanche: a cursor the server
cannot place must never be answered with the whole transcript, because the page
still holds every one of those bubbles.
"""

from __future__ import annotations

from html.parser import HTMLParser

import pytest
from fastapi.testclient import TestClient

from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Connection, Message, Model
from lembas.services import chat as chat_service
from lembas.services.crypto import encrypt


@pytest.fixture
def connection(db):
    connection = Connection(
        name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
    )
    db.add(connection)
    db.commit()
    db.add(Model(connection_id=connection.id, model_id="test-model"))
    db.commit()
    return connection


def _tail(client: TestClient, chat_id: str, after: str = ""):
    return client.get(f"/api/chats/{chat_id}/tail", params={"after": after} if after else {})


def _say(db, chat_id: str, role: str, content: str, **kwargs) -> Message:
    chat = db.get(Chat, chat_id)
    return chat_service.create_message(db, chat, role, content, **kwargs)


# --- The cursor ----------------------------------------------------------------
def test_nothing_new_is_a_204(client, db, registered, connection, make_chat):
    """A 204 and not an empty 200: htmx does not swap on a 204, where an empty
    body would still fire a swap and a settle on every open page every five
    seconds."""
    chat_id = make_chat()
    last = _say(db, chat_id, ROLE_USER, "hello")

    assert _tail(client, chat_id, last.id).status_code == 204


def test_a_turn_that_arrived_since_is_handed_back(client, db, registered, connection, make_chat):
    chat_id = make_chat()
    first = _say(db, chat_id, ROLE_USER, "hello")
    _say(db, chat_id, ROLE_ASSISTANT, "and a reply")

    response = _tail(client, chat_id, first.id)

    assert response.status_code == 200
    assert "and a reply" in response.text
    assert f"msg-{first.id}" not in response.text, "the cursor itself is not resent"


def test_no_after_returns_nothing(client, db, registered, connection, make_chat):
    """An empty `#thread` sends no cursor, and answering with the transcript
    would be the whole conversation appended to a page already showing it."""
    chat_id = make_chat()
    _say(db, chat_id, ROLE_USER, "hello")

    response = _tail(client, chat_id)

    assert response.status_code == 204
    assert "msg-" not in response.text


def test_an_unknown_after_does_not_dump_the_thread(
    client, db, registered, connection, make_chat
):
    """A rewind in another tab deletes the row the cursor names. A page whose
    history was rewritten underneath it is one only a reload can reconcile, and
    that is not this route's decision to make -- there may be a half-typed
    message in the box."""
    chat_id = make_chat()
    _say(db, chat_id, ROLE_USER, "hello")
    _say(db, chat_id, ROLE_ASSISTANT, "a reply")

    response = _tail(client, chat_id, "0" * 32)

    assert response.status_code == 204
    assert "msg-" not in response.text


def test_an_after_from_another_chat_returns_nothing(
    client, db, registered, connection, make_chat
):
    mine = make_chat()
    other = make_chat()
    elsewhere = _say(db, other, ROLE_USER, "in the other chat")
    _say(db, mine, ROLE_USER, "here")

    response = _tail(client, mine, elsewhere.id)

    assert response.status_code == 204
    assert "msg-" not in response.text


def test_another_readers_chat_is_not_found(client, db, registered, connection, make_chat):
    from lembas.db.models import User
    from lembas.security.passwords import hash_password

    stranger = User(
        email="stranger@example.com", name="Stranger", password_hash=hash_password("x" * 12)
    )
    db.add(stranger)
    db.commit()
    theirs = Chat(user_id=stranger.id)
    db.add(theirs)
    db.commit()
    last = _say(db, theirs.id, ROLE_USER, "private")

    assert _tail(client, theirs.id, last.id).status_code == 404


# --- What comes back -----------------------------------------------------------
def test_a_reply_that_started_outside_a_request_reaches_an_open_page(
    client, db, registered, connection, make_chat
):
    """The case the whole route exists for: `jobs.wake` wrote both rows and
    started a generation, and the page has neither. The assistant bubble has to
    arrive carrying its own `sse-connect`, because that shell is the only thing
    that opens a stream."""
    chat_id = make_chat()
    cursor = _say(db, chat_id, ROLE_ASSISTANT, "an earlier reply")
    _say(db, chat_id, ROLE_USER, "A background job you started has finished", machine=True)
    _say(db, chat_id, ROLE_ASSISTANT, "", complete_=False)

    response = _tail(client, chat_id, cursor.id)

    assert response.status_code == 200
    assert "A background job you started has finished" in response.text
    assert "sse-connect" in response.text
    assert "Background job" in response.text, "and not under the reader's name"


async def test_a_finished_job_reaches_the_page_without_a_reload(
    client, db, registered, connection, make_chat, monkeypatch
):
    """The complaint this was built for, end to end: `jobs.wake` on an idle chat,
    then the poll the open page would have made a moment later.

    Everything between is real -- the rows wake wrote, the cursor the page would
    have sent, the bubbles the route renders. Only `generation.ensure` is stubbed,
    since there is no upstream to answer.
    """
    from lembas.services import generation as generation_service
    from lembas.services.agent import jobs

    chat_id = make_chat()
    cursor = _say(db, chat_id, ROLE_ASSISTANT, "on it")
    monkeypatch.setattr(generation_service, "running_for", lambda _c: None)
    monkeypatch.setattr(generation_service, "ensure", lambda c, m: None)

    await jobs.wake(chat_id, "abc123abc123", "pytest -q", "done", 0, "1529 passed")

    response = _tail(client, chat_id, cursor.id)

    assert response.status_code == 200
    assert "1529 passed" in response.text, "the job's output arrived with it"
    assert "Background job" in response.text
    assert "msg__initial" not in response.text, "and never as something the reader sent"
    assert "sse-connect" in response.text, "the reply picks itself up from here"


def test_a_queued_turn_is_returned_and_carries_no_streaming_shell(
    client, db, registered, connection, make_chat
):
    """A completion waiting behind a running reply is exactly what the reader
    wants to watch arrive. It is safe to deliver early because the streaming
    shell requires the assistant role, so a queued user turn can never carry one
    -- and `_queue_frames` re-renders it in place when the reply ends."""
    chat_id = make_chat()
    cursor = _say(db, chat_id, ROLE_USER, "do the thing")
    _say(db, chat_id, ROLE_USER, "a job finished", queued=True, machine=True)

    response = _tail(client, chat_id, cursor.id)

    assert response.status_code == 200
    assert "a job finished" in response.text
    assert "sse-connect" not in response.text


def test_the_tail_renders_the_same_partial_the_thread_does(client, db, registered, connection):
    """Through `_render_bubble`, so a bubble that arrived late is the same bubble
    a reload would have drawn. Four handlers already render this template and a
    fifth that did its own thing is a fifth that forgets `template_flags`."""
    from pathlib import Path

    import lembas

    source = (Path(lembas.__file__).parent / "api/chats.py").read_text(encoding="utf-8")
    body = source[source.index("async def thread_tail") : source.index("async def post_message")]
    assert "_render_bubble" in body


# --- Being here counts as reading it -------------------------------------------
def test_polling_the_page_clears_the_unread_flag(client, db, registered, connection, make_chat):
    """`_persist` marks a reply unread whenever nobody is following it, which is
    true of a job-woken reply even with the reader watching -- so the toast
    announced the chat that was already on screen."""
    chat_id = make_chat()
    last = _say(db, chat_id, ROLE_USER, "hello")
    chat = db.get(Chat, chat_id)
    chat.unread = True
    chat.unread_notified = True
    db.commit()

    _tail(client, chat_id, last.id)

    db.expire_all()
    chat = db.get(Chat, chat_id)
    assert chat.unread is False
    assert chat.unread_notified is False


def test_the_flag_is_cleared_even_when_nothing_arrived(
    client, db, registered, connection, make_chat
):
    """The claim being made is that somebody is here, not that something came --
    and the empty answer is by far the common one."""
    chat_id = make_chat()
    last = _say(db, chat_id, ROLE_USER, "hello")
    chat = db.get(Chat, chat_id)
    chat.unread = True
    db.commit()

    assert _tail(client, chat_id, last.id).status_code == 204

    db.expire_all()
    assert db.get(Chat, chat_id).unread is False


# --- Where the poller sits -----------------------------------------------------
class _Ancestry(HTMLParser):
    """The open-tag stack above the element with a given id."""

    def __init__(self, wanted: str) -> None:
        super().__init__()
        self.wanted = wanted
        self.stack: list[tuple[str, dict[str, str]]] = []
        self.found: tuple[dict[str, str], list[tuple[str, dict[str, str]]]] | None = None

    def handle_starttag(self, tag, attrs):
        got = {key: (value or "") for key, value in attrs}
        if got.get("id") == self.wanted:
            self.found = (got, list(self.stack))
        if tag not in ("br", "img", "input", "hr", "meta", "link", "source", "use", "path"):
            self.stack.append((tag, got))

    def handle_endtag(self, tag):
        for index in range(len(self.stack) - 1, -1, -1):
            if self.stack[index][0] == tag:
                del self.stack[index:]
                return


def test_the_poller_is_outside_the_thread_and_outside_the_composer_form(
    client, db, registered, connection, make_chat
):
    """Both failures are silent and both have happened here before.

    Inside `#thread` it would be swapped away by the first rewind or compaction,
    which replace that container's contents, and never fire again. Inside the
    composer's form it would inherit `hx-target="#thread"` from an ancestor --
    the jobs chip did exactly that and blanked the transcript on every tick.
    """
    chat_id = make_chat()
    _say(db, chat_id, ROLE_USER, "hello")

    page = client.get(f"/chat/{chat_id}").text
    parser = _Ancestry("thread-tail")
    parser.feed(page)

    assert parser.found is not None, "the chat page has no tail poller"
    attrs, ancestors = parser.found

    ids = {got.get("id") for _tag, got in ancestors}
    assert "thread" not in ids, "a rewind would swap the poller away"
    assert not any(tag == "form" for tag, _got in ancestors), "it would inherit hx-target"

    assert attrs.get("hx-target") == "#thread"
    assert attrs.get("hx-swap") == "beforeend"
    assert attrs.get("hx-sync"), "two overlapping polls are two copies of one bubble"


def test_the_new_chat_screen_has_no_poller(client, db, registered, connection):
    """There is no row to poll: a chat is written together with its first
    message."""
    assert 'id="thread-tail"' not in client.get("/chat").text


def test_the_cursor_is_not_read_with_last_of_type():
    """`:last-of-type` is per-parent and `querySelector` returns the first match
    in document order, so on a compacted chat it answers with the last article
    inside `<details class="compacted">` rather than the newest message -- and
    the poll then asks about a message from the middle of the conversation and
    re-appends everything after it.

    Matched on the selector *as written in a selector string*, not on the words:
    the comment beside the code names `:last-of-type` in order to say why it is
    not being used, exactly as `data-prompt`'s comment names `hx-prompt`.
    """
    from pathlib import Path

    import lembas

    source = (Path(lembas.__file__).parent / "web/static/js/app.js").read_text(encoding="utf-8")
    assert ':last-of-type"' not in source
    assert ":last-of-type'" not in source
    assert 'querySelectorAll("article.msg")' in source


def test_the_poller_is_quiet_while_a_reply_is_streaming():
    """That reply delivers its own bubbles through the `done` frame, which is the
    only channel that gets the order right -- and it is the one window in which
    the transcript's order moves underneath us, since `_inject` restamps the
    placeholder."""
    from pathlib import Path

    import lembas

    source = (Path(lembas.__file__).parent / "web/static/js/app.js").read_text(encoding="utf-8")
    start = source.index("htmx:configRequest")
    block = source[start : source.index("htmx:beforeSwap", start)]
    assert "thread-tail" in block
    assert "sse-connect" in block
    assert "preventDefault" in block
    assert "parameters.after" in block


def test_a_bubble_the_page_already_has_is_never_swapped_in():
    """The race `hx-sync` cannot reach: the composer's POST committing between a
    tail request going out and its answer coming back. A duplicate here is not
    cosmetic -- it would carry a second `sse-connect` for one message."""
    from pathlib import Path

    import lembas

    source = (Path(lembas.__file__).parent / "web/static/js/app.js").read_text(encoding="utf-8")
    block = source[source.index("htmx:beforeSwap") :]
    assert "shouldSwap = false" in block
    assert "getElementById" in block


def test_no_template_hands_htmx_a_string_to_evaluate():
    """Every `hx-vals` in this project is static JSON rendered server-side. A
    `js:` one would be the only string htmx ever evaluated here, and it would die
    silently under any CSP a deployment added later -- the same family as the
    `hx-prompt` rule."""
    from pathlib import Path

    import lembas

    templates = Path(lembas.__file__).parent / "web/templates"
    offenders = [
        path.relative_to(templates)
        for path in templates.rglob("*.html")
        if 'hx-vals="js:' in path.read_text(encoding="utf-8")
        or "hx-vals='js:" in path.read_text(encoding="utf-8")
    ]
    assert not offenders, f"hx-vals with js: cannot cancel a request: {offenders}"


# --- Ordering ------------------------------------------------------------------
def test_rows_sharing_a_timestamp_do_not_stall_the_poll(
    client, db, registered, connection, make_chat
):
    """Under a bare `created_at >` a row sharing the cursor's microsecond is
    skipped forever -- returned never, passed never. The id clause is what makes
    the poll make progress instead of stopping on a row it can neither hand back
    nor step over."""
    chat_id = make_chat()
    cursor = _say(db, chat_id, ROLE_USER, "first")
    twin = _say(db, chat_id, ROLE_ASSISTANT, "same instant")
    twin.created_at = cursor.created_at
    db.commit()

    ordered = sorted([cursor.id, twin.id])
    response = _tail(client, chat_id, ordered[0])

    assert response.status_code == 200
    assert f"msg-{ordered[1]}" in response.text


def test_the_turns_come_back_in_the_order_they_happened(
    client, db, registered, connection, make_chat
):
    chat_id = make_chat()
    cursor = _say(db, chat_id, ROLE_USER, "first")
    _say(db, chat_id, ROLE_ASSISTANT, "second")
    _say(db, chat_id, ROLE_USER, "third")

    body = _tail(client, chat_id, cursor.id).text

    assert body.index("second") < body.index("third")