STT-Assistant/tests/test_tui_app.py

"""Smoke test: the Textual app mounts, subscribes, and ingests broadcasts."""

import asyncio
import os
import shutil
import sys
import tempfile

import pytest

from stt_daemon.assistant.control import ControlServer
from stt_daemon.assistant.service import AssistantService
from stt_daemon.config import Config


class FakeSubscriber:
    def events(self):
        return iter(())

    def close(self):
        pass


@pytest.fixture
def server():
    directory = tempfile.mkdtemp(prefix="stta-")
    path = os.path.join(directory, "c.sock")
    cfg = Config()
    cfg.assistant.control_socket = path

    service = AssistantService(cfg, FakeSubscriber())
    control = ControlServer(cfg, service, path=path)
    service.set_sinks(emit=control.broadcast, relay=control.relay)
    control.start()

    yield path, service

    control.close()
    service.stop()
    shutil.rmtree(directory, ignore_errors=True)


def test_enter_sends_execute_and_e_opens_editor(server):
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)
    sent: list[tuple] = []

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.client.send = lambda type_, **fields: sent.append((type_, fields))

            await pilot.press("enter")
            await pilot.pause()
            await pilot.press("a")
            await pilot.pause()
            await pilot.press("escape")
            await pilot.pause()

    asyncio.run(scenario())
    assert ("command_execute", {}) in sent
    assert ("command_start", {}) in sent
    assert ("command_cancel", {}) in sent


def test_c_key_confirms_before_clearing(server):
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)
    sent: list[tuple] = []

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.client.send = lambda type_, **fields: sent.append((type_, fields))

            await pilot.press("c")
            await pilot.pause()
            assert not any(type_ == "clear" for type_, _ in sent), "clear sent before confirmation"

            await pilot.press("y")
            await pilot.pause()

    asyncio.run(scenario())
    assert ("clear", {}) in sent


def test_c_key_cancel_does_not_clear(server):
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)
    sent: list[tuple] = []

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.client.send = lambda type_, **fields: sent.append((type_, fields))

            await pilot.press("c")
            await pilot.pause()
            await pilot.press("n")
            await pilot.pause()

    asyncio.run(scenario())
    assert not any(type_ == "clear" for type_, _ in sent)


def test_c_key_escape_does_not_clear(server):
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)
    sent: list[tuple] = []

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.client.send = lambda type_, **fields: sent.append((type_, fields))

            await pilot.press("c")
            await pilot.pause()
            await pilot.press("escape")
            await pilot.pause()

    asyncio.run(scenario())
    assert not any(type_ == "clear" for type_, _ in sent)


def test_c_key_clears_rendered_output(server):
    """Pressing 'c' clears the conversation pane via the cleared event."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
            app.state.ingest(
                {
                    "type": "tool_result",
                    "tool": "get_datetime",
                    "ok": True,
                    "result": "2026-09-21",
                }
            )
            app._render()
            await pilot.pause()

            app.state.ingest({"type": "cleared"})
            app._render()
            await pilot.pause()

            log = app.query_one("#chat", RichLog)
            return list(log.lines)

    lines = asyncio.run(scenario())
    assert not any("hello" in str(line) for line in lines)
    assert not any("2026-09-21" in str(line) for line in lines)


def test_tui_renders_command_output(server):
    """A tool result must actually reach a widget - otherwise 'nothing happens'."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.state.ingest(
                {
                    "type": "tool_result",
                    "tool": "get_datetime",
                    "ok": True,
                    "result": "2026-09-21T01:00:00",
                }
            )
            app._render()
            await pilot.pause()
            log = app.query_one("#chat", RichLog)
            return list(log.lines)

    lines = asyncio.run(scenario())
    assert lines, "command output was not rendered anywhere"


def test_tui_clears_chat_pane_on_chat_exit(server):
    """Leaving chat must drop the rendered conversation but keep command output."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.state.ingest(
                {"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "kept"}
            )
            app.state.ingest({"type": "mode", "mode": "chat"})
            app.state.ingest({"type": "chat_message", "role": "user", "text": "gone"})
            app._render()
            await pilot.pause()

            app.state.ingest({"type": "mode", "mode": "idle"})
            app._render()
            await pilot.pause()
            return [str(line) for line in app.query_one("#chat", RichLog).lines]

    lines = asyncio.run(scenario())
    assert not any("gone" in line for line in lines)
    assert any("kept" in line for line in lines)


def test_tui_mounts_subscribes_and_ingests_broadcasts(server):
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            service.arm()  # broadcast to the subscribed viewer
            for _ in range(60):
                await pilot.pause()
                if app.state.mode == "command_armed":
                    break
                await asyncio.sleep(0.05)

    asyncio.run(scenario())
    assert app.state.mode == "command_armed"


def test_render_after_unmount_is_safe(server):
    """A timer tick can land during teardown; rendering must not raise."""
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()

    asyncio.run(scenario())
    app._render()  # must not raise NoMatches


def test_tui_records_connect_failure(tmp_path):
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    app = AssistantTUI(str(tmp_path / "missing.sock"))

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()

    asyncio.run(scenario())
    assert app.connect_error is not None
    assert "cannot connect" in app.connect_error


def test_border_titles_show_queue_and_inference_depths(server):
    """Border titles for transcript and chat show live segment and inference queue counts."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.state.ingest(
                {
                    "type": "heartbeat",
                    "queue_depth": 5,
                    "inference_depth": 3,
                    "inference_in_flight": False,
                }
            )
            app._render()
            await pilot.pause()

            transcript_title = app.query_one("#transcript", RichLog).border_title
            chat_title = app.query_one("#chat", RichLog).border_title

            assert "[seg 5]" in transcript_title
            assert "[inf 3]" in chat_title
            assert "busy" not in chat_title

            # Test with inference in flight
            app.state.ingest(
                {
                    "type": "heartbeat",
                    "inference_depth": 7,
                    "inference_in_flight": True,
                }
            )
            app._render()
            await pilot.pause()

            chat_title = app.query_one("#chat", RichLog).border_title
            assert "[inf 7" in chat_title
            assert "busy" in chat_title

    asyncio.run(scenario())


def test_main_exit_code_reflects_connect_failure(monkeypatch, tmp_path, capsys):
    pytest.importorskip("textual")
    from stt_daemon.tui import app as tui_app

    monkeypatch.setattr(
        tui_app.AssistantTUI,
        "run",
        lambda self: setattr(self, "connect_error", "boom"),
    )
    monkeypatch.setattr(sys, "argv", ["stt-tui", "--socket", str(tmp_path / "x.sock")])
    assert tui_app.main() == 1
    assert "boom" in capsys.readouterr().err

    monkeypatch.setattr(tui_app.AssistantTUI, "run", lambda self: None)
    assert tui_app.main() == 0


def test_triage_disabled_composes_no_triage_widget(server):
    """Default (triage_enabled=False): no #triage, no #main — tree as today."""
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            assert not app.query("#triage")
            assert not app.query("#main")
            # The rest of the tree is intact.
            app.query_one("#transcript")
            app.query_one("#status")
            app.query_one("#command")
            app.query_one("#chat")
            app.query_one("#edit")

    asyncio.run(scenario())


def test_triage_enabled_composes_split(server):
    """triage_enabled=True: #main wraps #transcript + #triage."""
    pytest.importorskip("textual")
    from textual.containers import Horizontal
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path, triage_enabled=True)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            main = app.query_one("#main", Horizontal)
            transcript = app.query_one("#transcript", RichLog)
            triage = app.query_one("#triage", RichLog)
            assert transcript.parent is main
            assert triage.parent is main
            assert triage.border_title == "triage"
            # Outside widgets stay outside #main.
            assert app.query_one("#status").parent is not main
            assert app.query_one("#chat").parent is not main

    asyncio.run(scenario())


def test_triage_enabled_renders_signals_on_epoch_change(server):
    """A triage event + status event must actually write rendered lines into #triage."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path, triage_enabled=True)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app._timer.stop()  # no background drain: server status must not clobber state
            app._drain()
            app.state.ingest(
                {
                    "type": "triage",
                    "stage": "signals",
                    "triage": {
                        "order": ["category"],
                        "items": {
                            "category": {
                                "kind": "choice",
                                "choice": "question",
                                "answer_confidence": 0.91,
                            }
                        },
                    },
                }
            )
            app.state.ingest(
                {"type": "status", "triage_state": "ok", "triage_reason": ""}
            )
            app._render()
            await pilot.pause()
            return [str(line) for line in app.query_one("#triage", RichLog).lines]

    lines = asyncio.run(scenario())
    rendered = "\n".join(lines)
    assert "category: question (0.91)" in rendered


def test_triage_enabled_loading_shows_reason_line_never_blank(server):
    """loading/unknown triage_state renders the reason line — never a blank panel."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path, triage_enabled=True)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app._timer.stop()  # no background drain: server status must not clobber state
            app._drain()
            app.state.ingest(
                {
                    "type": "status",
                    "triage_state": "loading",
                    "triage_reason": "model loading",
                }
            )
            app._render()
            await pilot.pause()
            loading_lines = [str(line) for line in app.query_one("#triage", RichLog).lines]

            # Unknown state falls back to a generic reason line, no crash.
            app.state.ingest(
                {"type": "status", "triage_state": "weird_state", "triage_reason": ""}
            )
            app._render()
            await pilot.pause()
            unknown_lines = [str(line) for line in app.query_one("#triage", RichLog).lines]
            return loading_lines, unknown_lines

    loading_lines, unknown_lines = asyncio.run(scenario())
    assert loading_lines, "triage panel was blank in loading state"
    assert any("triage: unavailable" in line for line in loading_lines)
    assert any("model loading" in line for line in loading_lines)
    assert unknown_lines, "triage panel was blank for unknown state"
    assert any("triage: unavailable" in line for line in unknown_lines)
    assert any("weird_state" in line for line in unknown_lines)


def test_triage_epoch_change_clears_before_rewrite(server):
    """Epoch bump clears #triage first: no duplicated stale lines."""
    pytest.importorskip("textual")
    from textual.widgets import RichLog

    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path, triage_enabled=True)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app._timer.stop()  # no background drain: server status must not clobber state
            app._drain()
            app.state.ingest(
                {
                    "type": "triage",
                    "stage": "signals",
                    "triage": {"order": ["q1"], "items": {"q1": {"kind": "choice", "choice": "a"}}},
                }
            )
            app.state.ingest({"type": "status", "triage_state": "ok"})
            app._render()
            await pilot.pause()

            # New epoch: cleared stage empties the payload, panel must not keep old lines.
            app.state.ingest({"type": "triage", "stage": "cleared"})
            app._render()
            await pilot.pause()
            return [str(line) for line in app.query_one("#triage", RichLog).lines]

    lines = asyncio.run(scenario())
    rendered = "\n".join(lines)
    assert "q1" not in rendered, "stale triage lines survived an epoch change"
    assert "triage: unavailable — no data" in rendered


def test_triage_disabled_never_writes_triage(server):
    """Disabled: triage events are ingested but #triage is never touched (it doesn't exist)."""
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)

    async def scenario():
        async with app.run_test() as pilot:
            await pilot.pause()
            app.state.ingest({"type": "triage", "stage": "signals", "triage": {}})
            app.state.ingest({"type": "status", "triage_state": "ok"})
            app._render()  # must not raise NoMatches for #triage
            await pilot.pause()
            assert not app.query("#triage")

    asyncio.run(scenario())


def test_default_ctor_has_no_triage_arg(server):
    """AssistantTUI(path) stays valid: triage_enabled defaults to False."""
    pytest.importorskip("textual")
    from stt_daemon.tui.app import AssistantTUI

    path, _service = server
    app = AssistantTUI(path)
    assert app.triage_enabled is False
    assert app._seen_triage_epoch == 0