STT-Assistant/tests/test_tui_state.py

"""Tests for the TUI view-state reducer."""

from stt_daemon.tui.state import MAX_TRANSCRIPTS, ViewState


def test_mode_event_updates_mode_and_style():
    state = ViewState()
    state.ingest({"type": "mode", "mode": "command_armed"})
    assert state.mode == "command_armed"
    assert state.mode_style() == "yellow"


def test_initial_health_is_not_optimistic():
    """Before the first heartbeat arrives, health must read as unknown/down."""
    state = ViewState()
    assert state.health_ok() is False
    state.ingest({"type": "heartbeat", "stt_ok": True, "device_ok": True})
    assert state.health_ok() is True


def test_heartbeat_self_corrects_missed_mode():
    state = ViewState()
    state.ingest(
        {
            "type": "heartbeat",
            "mode": "chat",
            "stt_ok": True,
            "in_flight": False,
            "worker_stall_s": 0.0,
            "queue_depth": 2,
            "device_ok": True,
        }
    )
    assert state.mode == "chat"
    assert state.queue_depth == 2


def test_heartbeat_ingests_inference_depth_and_in_flight():
    state = ViewState()
    state.ingest(
        {
            "type": "heartbeat",
            "inference_depth": 3,
            "inference_in_flight": True,
        }
    )
    assert state.inference_depth == 3
    assert state.inference_in_flight is True


def test_heartbeat_omitting_inference_fields_preserves_prior_values():
    state = ViewState()
    state.inference_depth = 5
    state.inference_in_flight = True
    state.ingest(
        {
            "type": "heartbeat",
            "mode": "idle",
            "stt_ok": True,
        }
    )
    assert state.inference_depth == 5
    assert state.inference_in_flight is True


def test_health_red_when_stt_not_ok():
    state = ViewState()
    state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": False})
    assert state.health_ok() is False


def test_health_red_when_device_down():
    state = ViewState()
    state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": True, "device_ok": False})
    assert state.health_ok() is False


def test_worker_stall_recorded_and_degrades():
    state = ViewState()
    state.ingest({"type": "heartbeat", "stt_ok": False, "worker_stall_s": 16.5})
    assert state.worker_stall_s == 16.5
    assert state.health_ok() is False


def test_command_text_and_edited_flag():
    state = ViewState()
    state.ingest({"type": "command_text", "text": "hello", "edited": True, "phase": "ready"})
    assert state.command_text == "hello"
    assert state.command_edited is True
    assert state.phase == "ready"


def test_status_populates_fields():
    state = ViewState()
    state.ingest(
        {
            "type": "status",
            "mode": "command_armed",
            "ready": True,
            "phase": "ready",
            "command_text": "x",
            "stt_ok": True,
        }
    )
    assert state.mode == "command_armed"
    assert state.ready is True
    assert state.command_text == "x"


def test_transcripts_are_bounded():
    state = ViewState()
    for i in range(MAX_TRANSCRIPTS + 25):
        state.ingest({"type": "transcript", "text": f"t{i}", "lang": "en"})

    assert len(state.transcripts) == MAX_TRANSCRIPTS
    assert state.transcripts[-1]["text"] == f"t{MAX_TRANSCRIPTS + 24}"
    last_block = state.transcript_lines()[-1].plain
    assert f"t{MAX_TRANSCRIPTS + 24}" in last_block
    assert "EN" in last_block


def test_transcript_block_header_has_clock_lang_duration_and_reason():
    state = ViewState()
    state.ingest(
        {
            "type": "transcript",
            "ts": 1759073916.19,
            "lang": "es",
            "lang_prob": 0.99,
            "duration_s": 7.83,
            "end_reason": "end_of_turn",
            "text": "hola mundo",
        }
    )
    header, body = state.transcript_lines()[0].plain.split("\n")
    assert header[2] == ":" and header[5] == ":" and len(header[:8]) == 8
    assert "ES" in header
    assert "7.8s" in header
    assert "end_of_turn" in header
    assert body.strip() == "hola mundo"


def test_transcript_time_language_and_text_colors_are_disjoint():
    state = ViewState()
    for lang in ("es", "en", "fr", "de", "pt", "it", "zz"):
        state.ingest(
            {
                "type": "transcript",
                "ts": 1.0,
                "lang": lang,
                "duration_s": 1.0,
                "end_reason": "pause",
                "text": "x",
            }
        )

    palette = {
        str(span.style).split()[-1]
        for block in state.transcript_lines()
        for span in block.spans
    }
    time_color = str(state.transcript_lines()[0].spans[0].style).split()[-1]
    lang_color = str(state.transcript_lines()[0].spans[1].style).split()[-1]
    text_colors = {
        str(block.spans[-1].style).split()[-1] for block in state.transcript_lines()
    }
    assert not ({time_color, lang_color} & text_colors), palette
    assert time_color != lang_color


def test_transcripts_total_keeps_counting_past_the_cap():
    """The bounded buffer drops old lines; the total must still advance."""
    state = ViewState()
    for i in range(MAX_TRANSCRIPTS + 25):
        state.ingest({"type": "transcript", "text": f"t{i}", "lang": "en"})

    assert state.transcripts_total == MAX_TRANSCRIPTS + 25


def test_command_output_is_collected_for_display():
    state = ViewState()
    state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
    state.ingest(
        {"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "2026-09-21T01:00:00"}
    )
    state.ingest({"type": "error", "code": "not_ready", "message": "command is not ready"})

    lines = state.lower_lines()
    assert any("tool get_datetime" in line for line in lines)
    assert any("ok:" in line and "2026-09-21" in line for line in lines)
    assert any("not_ready" in line for line in lines)


def test_chat_turns_also_appear_in_lower_log():
    state = ViewState()
    state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
    assert state.lower_lines() == ["user: hello"]


def test_lower_log_is_bounded():
    from stt_daemon.tui.state import MAX_LOWER

    state = ViewState()
    for i in range(MAX_LOWER + 20):
        state.ingest({"type": "error", "code": f"e{i}", "message": "m"})
    assert len(state.lower_lines()) == MAX_LOWER


def test_speech_activity_is_tracked():
    state = ViewState()
    assert state.speech_open is False

    state.ingest({"type": "speech_start"})
    assert state.speech_open is True

    state.ingest({"type": "speech_stop"})
    assert state.speech_open is False

    state.ingest({"type": "status", "speech_open": True})
    assert state.speech_open is True


def test_chat_and_tool_events_recorded():
    state = ViewState()
    state.ingest({"type": "chat_message", "role": "user", "text": "hi"})
    state.ingest({"type": "chat_message", "role": "assistant", "text": "hello"})
    state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
    state.ingest({"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "x"})
    state.ingest({"type": "error", "code": "no_viewer", "message": "m"})

    assert state.chat_lines() == ["user: hi", "assistant: hello"]
    assert state.tool is not None and state.tool["tool"] == "get_datetime"
    assert state.tool_result is not None and state.tool_result["ok"] is True
    assert state.last_error is not None and state.last_error["code"] == "no_viewer"


def test_leaving_chat_clears_chat_pane_but_keeps_command_output():
    state = ViewState()
    state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
    state.ingest({"type": "mode", "mode": "chat"})
    state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
    state.ingest({"type": "chat_message", "role": "assistant", "text": "hi"})
    assert state.chat_lines() == ["user: hello", "assistant: hi"]

    epoch = state.chat_epoch
    state.ingest({"type": "mode", "mode": "idle"})

    assert state.chat_lines() == []
    assert state.lower_lines() == ["-> tool get_datetime"]
    assert state.chat_epoch == epoch + 1


def test_heartbeat_mode_change_clears_chat():
    state = ViewState()
    state.ingest({"type": "mode", "mode": "chat"})
    state.ingest({"type": "chat_message", "role": "user", "text": "x"})

    state.ingest({"type": "heartbeat", "mode": "idle", "stt_ok": True})

    assert state.chat_lines() == []
    assert state.lower_lines() == []


def test_compaction_note_renders_and_clears_on_chat_exit():
    state = ViewState()
    state.ingest({"type": "mode", "mode": "chat"})
    state.ingest(
        {
            "type": "chat_message",
            "role": "note",
            "text": "Compacted 4 earlier turns into a summary.",
        }
    )

    assert "note: Compacted 4 earlier turns into a summary." in state.lower_lines()

    state.ingest({"type": "mode", "mode": "idle"})

    assert state.lower_lines() == []


def test_entering_chat_does_not_clear_rendered_log():
    state = ViewState()
    state.ingest({"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "x"})
    epoch = state.chat_epoch

    state.ingest({"type": "mode", "mode": "chat"})

    assert state.chat_epoch == epoch
    assert any("ok:" in line for line in state.lower_lines())


def test_cleared_event_wipes_output_and_increments_epoch():
    state = ViewState()
    state.ingest({"type": "chat_message", "role": "user", "text": "hello"})
    state.ingest({"type": "chat_message", "role": "assistant", "text": "hi"})
    state.ingest({"type": "tool_call", "tool": "get_datetime", "args": {}})
    state.ingest(
        {
            "type": "tool_result",
            "tool": "get_datetime",
            "ok": True,
            "result": "2026-09-21T01:00:00",
        }
    )
    state.ingest({"type": "error", "code": "test_error", "message": "test message"})
    state.ingest({"type": "transcript", "text": "preserve this", "lang": "en"})

    epoch = state.chat_epoch
    assert len(state.chat_lines()) == 2
    assert len(state.lower_lines()) >= 4
    assert state.tool is not None
    assert state.tool_result is not None
    assert state.last_error is not None
    assert len(state.transcripts) == 1

    state.ingest({"type": "cleared"})

    assert state.chat_lines() == []
    assert state.lower_lines() == []
    assert state.tool is None
    assert state.tool_result is None
    assert state.last_error is None
    assert state.chat_epoch == epoch + 1
    assert len(state.transcripts) == 1
    block = state.transcript_lines()[0].plain
    assert "preserve this" in block
    assert "EN" in block


def _signals_triage():
    return {
        "order": ["q_choice", "q_noul", "q_score", "q_unknown"],
        "items": {
            "q_choice": {
                "kind": "choice",
                "choice": "yes",
                "answer_confidence": 0.91,
                "flagged": True,
            },
            "q_noul": {"kind": "noul", "p": 0.75, "flagged": False},
            "q_score": {"kind": "score", "level": 3, "label": "high", "flagged": True},
            "q_unknown": {"kind": "mystery", "flagged": False},
        },
        "escalated": True,
        "input_tokens": 42,
    }


def test_triage_signals_event_stores_and_bumps_epoch():
    state = ViewState()
    epoch = state.triage_epoch
    state.ingest({"type": "triage", "stage": "signals", "triage": _signals_triage()})
    assert state.triage == _signals_triage()
    assert state.triage_epoch == epoch + 1


def test_triage_content_event_replaces_triage_and_bumps_epoch():
    state = ViewState()
    state.ingest({"type": "triage", "stage": "signals", "triage": _signals_triage()})
    epoch = state.triage_epoch
    merged = {
        **_signals_triage(),
        "content": {"summary": "s", "opportunities": [], "risks": [], "action_items": []},
    }
    state.ingest({"type": "triage", "stage": "content", "triage": merged})
    assert state.triage == merged
    assert state.triage_epoch == epoch + 1


def test_triage_cleared_event_empties_and_bumps_epoch():
    state = ViewState()
    state.ingest({"type": "triage", "stage": "signals", "triage": _signals_triage()})
    epoch = state.triage_epoch
    state.ingest({"type": "triage", "stage": "cleared"})
    assert state.triage is None
    assert state.triage_epoch == epoch + 1


def test_status_adopts_triage_fields():
    state = ViewState()
    epoch = state.triage_epoch
    state.ingest(
        {
            "type": "status",
            "triage": _signals_triage(),
            "triage_state": "ok",
            "triage_reason": "",
        }
    )
    assert state.triage_state == "ok"
    assert state.triage == _signals_triage()
    assert state.triage_epoch == epoch + 1


def test_status_without_triage_fields_bumps_nothing():
    state = ViewState()
    state.ingest({"type": "status", "mode": "idle", "ready": True})
    assert state.triage_epoch == 0
    assert state.triage_state == "disabled"


def test_triage_lines_choice_noul_score_and_flag_marks():
    state = ViewState()
    state.triage = _signals_triage()
    state.triage_state = "ok"
    lines = state.triage_lines()
    assert lines[0] == "q_choice: yes (0.91) ⚠"
    assert lines[1].startswith("███████    q_noul: 0.75")
    assert lines[2].startswith("███") and "q_score: high ⚠" in lines[2]
    assert lines[3] == "q_unknown"


def test_triage_lines_content_block_appended():
    state = ViewState()
    state.triage = {
        "order": ["q1"],
        "items": {"q1": {"kind": "choice", "choice": "no", "answer_confidence": 0.9}},
        "content": {
            "summary": "user wants action",
            "opportunities": ["opp1", "opp2", "opp3", "opp4"],
            "risks": ["risk1"],
            "action_items": [{"what": "call", "why": "urgent"}],
        },
    }
    state.triage_state = "ok"
    lines = state.triage_lines()
    assert "" in lines
    assert "summary: user wants action" in lines
    assert "opportunitie: opp1" in lines
    assert "opp4" not in " ".join(lines)
    assert "risk: risk1" in lines
    assert "action_item: {'what': 'call', 'why': 'urgent'}" in lines


def test_triage_lines_non_ok_state_renders_reason():
    state = ViewState()
    state.triage_state = "loading"
    state.triage_reason = "model warming up"
    assert state.triage_lines() == ["triage: unavailable — model warming up"]


def test_triage_lines_unknown_reason_falls_back_generically():
    state = ViewState()
    state.triage_state = "service_down"
    assert state.triage_lines() == ["triage: unavailable — service_down"]


def test_triage_lines_ok_without_data_renders_no_data():
    state = ViewState()
    state.triage_state = "ok"
    assert state.triage_lines() == ["triage: unavailable — no data"]


def test_triage_lines_malformed_payload_never_raises():
    state = ViewState()
    state.triage_state = "ok"
    state.triage = {
        "order": ["a", "b", "c"],
        "items": {
            "a": {"kind": "choice", "flagged": True},
            "b": {"kind": "noul", "p": "not-a-number", "flagged": True},
            "c": {"kind": "score", "level": None},
        },
    }
    lines = state.triage_lines()
    assert lines[0] == "a: ? (?) ⚠"
    assert "b: ? ⚠" in lines[1]
    assert "c: ?" in lines[2]


def test_triage_lines_empty_order_renders_no_signals():
    state = ViewState()
    state.triage_state = "ok"
    state.triage = {"order": [], "items": {}}
    assert state.triage_lines() == ["triage: no signals"]


def test_clear_output_resets_triage_and_bumps_epoch():
    state = ViewState()
    state.ingest({"type": "triage", "stage": "signals", "triage": _signals_triage()})
    epoch = state.triage_epoch
    state.clear_output()
    assert state.triage is None
    assert state.triage_epoch == epoch + 1


def test_status_adopted_triage_survives_then_cleared_event_empties_panel():
    state = ViewState()
    state.ingest(
        {"type": "status", "triage": _signals_triage(), "triage_state": "ok", "triage_reason": ""}
    )
    assert state.triage_lines()[0].startswith("q_choice:")
    state.ingest({"type": "triage", "stage": "cleared"})
    assert state.triage is None


def test_lone_signals_event_renders_without_status_snapshot():
    state = ViewState()
    state.ingest(
        {
            "type": "triage",
            "stage": "signals",
            "triage": {
                "order": ["category"],
                "items": {
                    "category": {"kind": "choice", "choice": "question", "answer_confidence": 0.91}
                },
            },
        }
    )
    assert state.triage_state == "ok"
    assert state.triage_lines() == ["category: question (0.91)"]


def test_lone_content_event_renders_without_status_snapshot():
    state = ViewState()
    state.ingest(
        {
            "type": "triage",
            "stage": "content",
            "triage": {
                "order": ["category"],
                "items": {
                    "category": {"kind": "choice", "choice": "question", "answer_confidence": 0.91}
                },
                "content": {"summary": "wants action"},
            },
        }
    )
    assert state.triage_state == "ok"
    lines = state.triage_lines()
    assert lines[0] == "category: question (0.91)"
    assert "summary: wants action" in lines


def test_cleared_event_after_lone_signals_leaves_state_usable():
    state = ViewState()
    state.ingest({"type": "triage", "stage": "signals", "triage": _signals_triage()})
    state.ingest({"type": "triage", "stage": "cleared"})
    assert state.triage is None
    assert state.triage_lines() == ["triage: unavailable — no data"]