STT-Assistant/tests/test_assistant_service.py

"""Tests for the assistant service (Phase 2) using synthetic event streams."""

import itertools
import json
import re
import threading
import time
from typing import Any

from stt_daemon import events
from stt_daemon.assistant.inference import InferenceJob, InferenceQueue
from stt_daemon.assistant.service import EXECUTE_BUSY, AssistantService
from stt_daemon.assistant.triage import TriageAnalyzer
from stt_daemon.assistant.triage_dump import TriageDumper
from stt_daemon.config import Config

_sequence = itertools.count(1)

# A real analyzer (pure, stdlib-only helpers) backing the fake's retention
# guard so guard behaviour is exercised end to end.
_real_analyzer = TriageAnalyzer(Config().triage)


class Clock:
    def __init__(self, start=1000.0):
        self.t = float(start)

    def __call__(self):
        return self.t

    def advance(self, dt):
        self.t += float(dt)


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

    def close(self):
        pass


def _service():
    cfg = Config()
    clock = Clock()
    emitted: list[dict] = []

    def emit(type_, **fields):
        emitted.append({"type": type_, **fields})

    service = AssistantService(cfg, FakeSubscriber(), emit=emit, time_fn=clock)
    return service, emitted, clock


def _ev(type_, clock, **fields):
    event = {"v": 1, "type": type_, "seq": next(_sequence), "ts": clock()}
    event.update(fields)
    return event


def _transcript(clock, text, ts=None, session_dir=None):
    event = {
        "v": 1,
        "type": events.TRANSCRIPT,
        "seq": next(_sequence),
        "ts": clock() if ts is None else ts,
        "text": text,
    }
    if session_dir is not None:
        event["session_dir"] = session_dir
    return event


def test_ready_requires_matching_segment_done():
    service, _, clock = _service()

    service.handle_event(_ev(events.SPEECH_START, clock))
    assert service.ready() is False

    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
    assert service.ready() is False  # awaiting segment 1

    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))
    assert service.ready() is True


def test_stale_segment_done_does_not_satisfy_later_turn():
    service, _, clock = _service()

    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
    service.handle_event(_ev(events.SPEECH_START, clock))
    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=2))

    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))
    assert service.ready() is False  # still awaiting segment 2

    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=2, emitted=True))
    assert service.ready() is True


def test_gated_segment_done_clears_readiness():
    service, _, clock = _service()

    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=3))
    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=3, emitted=False))

    assert service.ready() is True


def test_speech_stop_without_seg_id_clears_readiness():
    """A discarded turn closes with speech_stop but no seg_id; readiness must recover."""
    service, _, clock = _service()

    service.handle_event(_ev(events.SPEECH_START, clock))
    assert service.ready() is False

    service.handle_event(_ev(events.SPEECH_STOP, clock))  # discarded: no seg_id
    assert service.speech_open is False
    assert service.awaited_seg_id is None
    assert service.ready() is True


def test_status_resets_stale_speech_state_on_reconnect():
    service, _, clock = _service()

    service.handle_event(_ev(events.SPEECH_START, clock))
    assert service.speech_open is True

    service.handle_event(_ev(events.STATUS, clock, capture="x", segments_done=0))
    assert service.speech_open is False
    assert service.awaited_seg_id is None
    assert service.ready() is True


def test_command_buffer_uses_arm_timestamp():
    service, _, clock = _service()
    clock.advance(10)
    assert service.arm() is True
    arm_ts = service.arm_ts
    assert arm_ts is not None

    # Replayed transcript from before arming is ignored.
    service.handle_event(_transcript(clock, "old", ts=arm_ts - 5))
    assert service.command_text == ""

    service.handle_event(_transcript(clock, "what time is it"))
    assert service.command_text == "what time is it"


def test_mutual_exclusion_between_command_and_chat():
    service, _, _ = _service()

    assert service.toggle_chat() is True
    assert service.arm() is False  # cannot arm in chat

    assert service.toggle_chat() is True  # back to idle
    assert service.arm() is True
    assert service.toggle_chat() is False  # cannot chat while armed


def test_arm_timeout_disarms_and_emits_error():
    service, emitted, clock = _service()
    service.arm()

    clock.advance(service.cfg.assistant.command_arm_timeout_s + 0.1)
    service.tick()

    assert service.modes.mode.value == "idle"
    assert service.command_text == ""
    assert any(e["type"] == "error" and e["code"] == "arm_timeout" for e in emitted)


def test_long_speech_does_not_disarm():
    service, emitted, clock = _service()
    service.arm()
    service.handle_event(_ev(events.SPEECH_START, clock))

    clock.advance(service.cfg.assistant.command_arm_timeout_s + 30.0)
    service.tick()

    assert service.modes.mode.value == "command_armed"
    assert not any(e["type"] == "error" for e in emitted)


def test_captured_command_survives_arm_timeout():
    service, emitted, clock = _service()
    service.arm()
    service.handle_event(_transcript(clock, "what time is it"))
    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))

    clock.advance(service.cfg.assistant.command_arm_timeout_s + 30.0)
    service.tick()

    assert service.modes.mode.value == "command_armed"
    assert service.command_text == "what time is it"
    assert not any(e["type"] == "error" for e in emitted)


def test_stt_stall_cancels_armed_command():
    service, emitted, clock = _service()
    service.handle_event(_ev(events.HEARTBEAT, clock, worker_stall_s=0.0))
    service.arm()

    clock.advance(service.cfg.assistant.stall_timeout_ms / 1000.0 + 0.5)
    service.tick()

    assert service.modes.mode.value == "idle"
    assert any(e["type"] == "error" and e["code"] == "stt_stall" for e in emitted)


def test_worker_stall_marks_stt_degraded():
    service, _, clock = _service()
    long_stall = service.cfg.assistant.worker_stall_ms / 1000.0 + 1.0

    service.handle_event(_ev(events.HEARTBEAT, clock, worker_stall_s=long_stall))
    assert service.stt_ok() is False
    assert service.ready() is False


def test_chat_idle_timeout_exits_chat():
    service, _, clock = _service()
    assert service.toggle_chat() is True

    clock.advance(service.cfg.assistant.chat_idle_timeout_s + 0.1)
    service.tick()

    assert service.modes.mode.value == "idle"


def test_chat_exit_phrase_exits_chat():
    service, _, clock = _service()
    assert service.toggle_chat() is True

    service.handle_event(_transcript(clock, "That's all."))

    assert service.modes.mode.value == "idle"
    assert service.chat_history == []


def test_viewer_detached_exits_chat_and_forgets_history():
    service, _, clock = _service()
    assert service.toggle_chat() is True
    service.handle_event(_transcript(clock, "hello"))

    assert service.viewer_detached() is True
    assert service.modes.mode.value == "idle"
    assert service.chat_history == []


def test_viewer_detached_is_noop_outside_chat():
    service, _, _ = _service()
    assert service.viewer_detached() is False
    assert service.modes.mode.value == "idle"


def test_exit_chat_invokes_chat_reset():
    cfg = Config()
    resets: list[int] = []
    service = AssistantService(
        cfg, FakeSubscriber(), time_fn=Clock(), chat_reset=lambda: resets.append(1)
    )
    service.toggle_chat()

    service.viewer_detached()

    assert resets == [1]


def test_chat_accumulates_turns():
    service, _, clock = _service()
    service.toggle_chat()

    for text in ("hello there", "how are you"):
        service.handle_event(_transcript(clock, text))

    assert service.chat_history == ["Unknown: hello there", "Unknown: how are you"]


def test_execute_requires_readiness():
    service, _, clock = _service()
    service.arm()
    service.handle_event(_transcript(clock, "do it"))

    service.handle_event(_ev(events.SPEECH_START, clock))
    assert service.execute() is None  # speech still open
    assert service.modes.mode.value == "command_armed"

    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
    assert service.execute() is None  # awaiting segment 1

    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))
    assert service.execute() == "do it"
    assert service.modes.mode.value == "executing"

    assert service.finish() is True
    assert service.modes.mode.value == "idle"


def test_execute_reports_busy_when_queue_rejects():
    service, emitted, clock = _service()

    # A zero-capacity lane rejects the priority command job on submit.
    service._inference = InferenceQueue(size=0)
    service.executor = object()
    service.arm()
    service.handle_event(_transcript(clock, "do it"))
    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))

    assert service.execute() is EXECUTE_BUSY
    assert service.modes.mode.value == "idle"
    assert not any(e.get("code") == "busy" for e in emitted)


def test_turn_ended_requires_quiet_gap():
    service, _, clock = _service()
    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))

    assert service.ready() is True
    assert service.turn_ended() is False

    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    assert service.turn_ended() is True


def test_emits_mode_and_command_text():
    service, emitted, clock = _service()
    service.arm()
    assert {"type": "mode", "mode": "command_armed"} in emitted

    service.handle_event(_transcript(clock, "hi"))
    assert any(e["type"] == "command_text" and e["text"] == "hi" for e in emitted)


def test_cancel_clears_command_buffer():
    service, _, clock = _service()
    service.arm()
    service.handle_event(_transcript(clock, "hi"))
    assert service.cancel() is True
    assert service.modes.mode.value == "idle"
    assert service.command_text == ""
    assert service.arm_ts is None


def test_pending_chat_is_not_lost_across_threads():
    """Transcripts appended on the reader thread while the ticker dispatches
    must all reach the model: the join-and-clear in _dispatch_chat is atomic,
    and the user bubble is emitted by the worker in _run_chat."""
    cfg = Config()
    cfg.assistant.turn_gap_s = 0.01
    dispatched: list[str] = []
    lock = threading.Lock()

    def emit(type_, **fields):
        if type_ == "chat_message" and fields.get("role") == "user":
            with lock:
                dispatched.append(fields["text"])

    service = AssistantService(cfg, FakeSubscriber(), emit=emit, chat_fn=lambda text: "ok")
    service.toggle_chat()
    service.start()
    try:
        total = 200
        producer = threading.Thread(
            target=lambda: [
                service.handle_event(
                    {
                        "v": 1,
                        "type": events.TRANSCRIPT,
                        "seq": i,
                        "ts": time.time(),
                        "text": f"w{i}",
                    }
                )
                for i in range(total)
            ],
            daemon=True,
        )
        producer.start()

        deadline = time.time() + 10.0
        while time.time() < deadline:
            service.tick()
            with lock:
                words = sum(len(text.split()) for text in dispatched)
                done = words == 2 * total
            if not producer.is_alive() and done:
                break
            time.sleep(0.002)
        producer.join(timeout=5)

        with lock:
            words = sum(len(text.split()) for text in dispatched)
        assert words == 2 * total
    finally:
        service.stop()


def _full_queue_service():
    service, emitted, clock = _service()
    service.toggle_chat()
    service._inference = InferenceQueue(size=1)
    service._inference.submit(
        InferenceJob(name="blocker", run=lambda: None, evictable=False)
    )
    return service, emitted, clock


def test_chat_refusal_rebuffers_and_emits_queue_full_once():
    """A refused chat submit re-buffers under lock and emits queue_full on flip."""
    service, emitted, _ = _full_queue_service()
    service._pending_chat = ["hello world"]

    service._dispatch_chat()

    assert service._pending_chat == ["hello world"]
    assert service._chat_queue_full is True
    errors = [e for e in emitted if e["type"] == "error" and e.get("code") == "queue_full"]
    assert len(errors) == 1
    assert [e for e in emitted if e["type"] == "chat_message"] == []

    service._dispatch_chat()

    assert service._pending_chat == ["hello world"]
    errors = [e for e in emitted if e["type"] == "error" and e.get("code") == "queue_full"]
    assert len(errors) == 1


def test_chat_dispatch_success_clears_latch_without_user_bubble():
    """Success clears the latch; the user bubble comes from _run_chat, not dispatch."""
    service, emitted, _ = _full_queue_service()
    service._pending_chat = ["hello world"]
    service._chat_queue_full = True

    service._inference.drop(lambda j: True)
    service._dispatch_chat()

    assert service._pending_chat == []
    assert service._chat_queue_full is False
    assert [e for e in emitted if e["type"] == "chat_message"] == []


def test_run_chat_emits_user_first_then_reply():
    service, emitted, _ = _service()
    service.chat_fn = lambda text: "reply!"

    service._run_chat("hello")

    assert emitted == [
        {"type": "chat_message", "role": "user", "text": "hello"},
        {"type": "chat_message", "role": "assistant", "text": "reply!"},
    ]


def test_chat_notify_emits_compaction_note():
    service, emitted, _ = _service()

    service.chat_notify("compacted", turns=4)

    assert {
        "type": "chat_message",
        "role": "note",
        "text": "Compacted 4 earlier turns into a summary.",
    } in emitted


def test_chat_notify_emits_truncated_error():
    service, emitted, _ = _service()

    service.chat_notify("truncated", message="cut off")

    assert {"type": "error", "code": "truncated", "message": "cut off"} in emitted


def test_clear_empties_chat_history():
    service, emitted, clock = _service()
    service.toggle_chat()

    for text in ("hello", "how are you"):
        service.handle_event(_transcript(clock, text))

    assert service.chat_history == ["Unknown: hello", "Unknown: how are you"]
    assert len(service._pending_chat) == 2

    assert service.clear() is True
    assert service.chat_history == []
    assert service._pending_chat == []
    assert {"type": "cleared"} in emitted


def test_clear_invokes_chat_reset():
    cfg = Config()
    resets: list[int] = []
    service = AssistantService(
        cfg, FakeSubscriber(), time_fn=Clock(), chat_reset=lambda: resets.append(1)
    )
    service.toggle_chat()

    service.clear()

    assert resets == [1]


def test_clear_does_not_change_mode():
    service, _, clock = _service()
    service.toggle_chat()
    assert service.modes.mode.value == "chat"

    service.clear()

    assert service.modes.mode.value == "chat"


def test_clear_always_returns_true():
    service, _, _ = _service()
    assert service.clear() is True

    service.toggle_chat()
    service.handle_event(_ev(events.TRANSCRIPT, Clock(), text="test"))
    assert service.clear() is True


def test_heartbeat_emits_inference_depth_and_in_flight():
    """Calling tick() emits heartbeat with inference_depth (int) and inference_in_flight (bool)."""
    service, emitted, clock = _service()
    service.tick()

    heartbeats = [e for e in emitted if e["type"] == "heartbeat"]
    assert len(heartbeats) == 1

    heartbeat = heartbeats[0]
    assert "inference_depth" in heartbeat
    assert isinstance(heartbeat["inference_depth"], int)
    assert "inference_in_flight" in heartbeat
    assert isinstance(heartbeat["inference_in_flight"], bool)


def test_status_contains_inference_depth_and_in_flight():
    """status() snapshot contains inference_depth (int) and inference_in_flight (bool)."""
    service, _, _ = _service()

    status = service.status()
    assert "inference_depth" in status
    assert isinstance(status["inference_depth"], int)
    assert "inference_in_flight" in status
    assert isinstance(status["inference_in_flight"], bool)


_AUTO = object()


class FakeTriage:
    def __init__(
        self,
        questions=_AUTO,
        state_pair=("ok", ""),
        escalate=False,
        parsed=_AUTO,
        normalized=_AUTO,
        gate=True,
    ):
        if questions is _AUTO:
            questions = {"urgency": {"type": "noul", "instructions": "how urgent"}}
        self._questions = questions
        self._state_pair = state_pair
        self._escalate = escalate
        self._parsed = parsed
        self._normalized = normalized
        self._gate = gate
        self.build_state_calls: list[tuple] = []
        self.normalize_calls: list = []
        self.prompt_calls: list[tuple] = []
        self.gate_calls: list = []

    @property
    def questions(self):
        return self._questions

    def state(self):
        return self._state_pair

    def build_state(self, window):
        self.build_state_calls.append(tuple(window))
        return "\n".join("- " + str(item) for item in (window or []))

    def normalize(self, payload):
        self.normalize_calls.append(payload)
        if self._normalized is not _AUTO:
            return self._normalized
        if not isinstance(payload, dict):
            return {}
        signals = {"order": [], "items": {}, "escalated": False, "input_tokens": 0}
        signals.update(payload)
        if self._escalate:
            signals["escalated"] = True
        return signals

    def build_brief_prompt(self, deltas, signals):
        self.prompt_calls.append(
            (tuple(deltas), dict(signals) if isinstance(signals, dict) else signals)
        )
        return "BRIEF-PROMPT"

    def build_ledger_prompt(self, narrative, ledger, deltas, signals):
        self.prompt_calls.append(
            (tuple(deltas), dict(signals) if isinstance(signals, dict) else signals)
        )
        ledger_text = "\n".join("- " + f for f in ledger) or "(none yet)"
        body = "\n".join("- " + d for d in deltas)
        return (
            f"CURRENT LEDGER:\n{ledger_text}\n\n"
            f"CURRENT BRIEF:\n{narrative or '(none yet)'}\n\n"
            f"NEW UTTERANCES:\n{body}"
        )

    def parse_ledger_reply(self, text, truncated=False):
        return _real_analyzer.parse_ledger_reply(text, truncated=truncated)

    def parse_facts(self, block):
        return _real_analyzer.parse_facts(block)

    def parse_brief(self, text):
        if self._parsed is _AUTO:
            return text if text else None
        return self._parsed

    def should_update_brief(self, signals):
        self.gate_calls.append(signals)
        return self._gate


class FakeBriefFile:
    """Recording stand-in for BriefFile (write/clear/set_directory never raise)."""

    def __init__(self):
        self.writes: list[str] = []
        self.clears = 0
        self.directories: list[str] = []

    def set_directory(self, directory):
        self.directories.append(directory)

    def write(self, text):
        self.writes.append(text)

    def clear(self):
        self.clears += 1


class FakeLane:
    def __init__(self, ready=True):
        self._ready = ready
        self.submits: list[tuple] = []
        self.drops = 0
        self.predict_gate = threading.Event()

    def ready(self):
        return self._ready

    def state(self):
        return ("ok", "") if self._ready else ("loading", "")

    def submit(self, state, questions, lang, generation):
        self.submits.append((state, questions, lang, generation))
        return True

    def drop_pending(self):
        self.drops += 1


class FakeDebugSink:
    def __init__(self):
        self.events: list[tuple] = []

    def emit(self, type_, **fields):
        self.events.append((type_, fields))


def _ok_reply(text):
    return "ok"


def _brief_reply(narrative="", facts=None):
    """Build a ledger-format reply the analyzer's parser can consume."""
    body = "\n".join("- " + fact for fact in (facts or [])) or "none"
    return f"NEW FACTS:\n{body}\n\nBRIEF:\n{narrative}"


def _triage_service(
    triage=None,
    lane=None,
    brief_fn=None,
    chat_fn: Any = _AUTO,
    brief_file=None,
    brief_reset=None,
    chat_reset=None,
    debug_sink=None,
):
    cfg = Config()
    cfg.assistant.turn_gap_s = 0.5
    clock = Clock()
    emitted: list[dict] = []

    def emit(type_, **fields):
        emitted.append({"type": type_, **fields})

    if triage is None:
        triage = FakeTriage()
    if lane is None:
        lane = FakeLane()
    if chat_fn is _AUTO:
        chat_fn = _ok_reply
    factory = (lambda: debug_sink) if debug_sink is not None else None
    service = AssistantService(
        cfg,
        FakeSubscriber(),
        emit=emit,
        time_fn=clock,
        triage=triage,
        laya_lane=lane,
        brief_fn=_wrap_brief_fn(brief_fn),
        brief_reset=brief_reset,
        brief_file=brief_file,
        chat_fn=chat_fn,
        chat_reset=chat_reset,
        debug_sink_factory=factory,
    )
    return service, emitted, clock, triage, lane


def _wrap_brief_fn(brief_fn):
    """Adapt a simple `prompt -> text` stub to the engine's call signature.

    A plain string return is treated as the volatile narrative half of a
    ledger reply, so existing stubs keep reading naturally.
    """
    if brief_fn is None:
        return None

    def wrapped(prompt, on_io=None, finalize=None):
        kwargs = {}
        if on_io is not None:
            kwargs["on_io"] = on_io
        try:
            result = brief_fn(prompt, **kwargs)
        except TypeError:
            result = brief_fn(prompt)
        if finalize is not None:
            result = finalize(result, "stop")
        is_ledger = "BRIEF:" in result or "NEW FACTS:" in result
        if isinstance(result, str) and result and not is_ledger:
            result = _brief_reply(narrative=result)
        return result

    return wrapped


def _triage_events(emitted):
    return [e for e in emitted if e["type"] == "triage"]


def _escalated_payload():
    return {"order": [], "items": {}, "escalated": True, "input_tokens": 3}


def test_triage_status_disabled_when_no_analyzer():
    service, _, _ = _service()

    status = service.status()

    assert status["triage_state"] == "disabled"
    assert status["triage_reason"] == ""
    assert status["triage"] is None


def test_triage_status_reflects_analyzer():
    service, _, _, _, _ = _triage_service(
        triage=FakeTriage(state_pair=("loading", "warming up"))
    )

    status = service.status()

    assert status["triage_state"] == "loading"
    assert status["triage_reason"] == "warming up"
    assert status["triage"] is None


def test_status_contains_questions_from_analyzer():
    questions = {
        "urgency": {"type": "noul", "instructions": "how urgent"},
        "sentiment": {"type": "score", "instructions": "tone", "criteria": ["low", "high"]},
    }
    service, _, _, triage, _ = _triage_service(triage=FakeTriage(questions=questions))

    status = service.status()

    assert status["questions"] == questions
    assert status["questions"] is triage.questions


def test_status_questions_none_without_analyzer():
    service, _, _ = _service()

    assert service.status()["questions"] is None


def test_status_event_carries_questions():
    service, emitted, _, triage, _ = _triage_service()

    service.tick()

    statuses = [e for e in emitted if e["type"] == "status"]
    assert len(statuses) == 1
    assert statuses[0]["questions"] is triage.questions


def test_triage_captures_transcript_lang():
    service, _, clock, _, _ = _triage_service()
    assert service._last_transcript_lang is None

    event = _transcript(clock, "hola")
    event["lang"] = "es"
    service.handle_event(event)
    assert service._last_transcript_lang == "es"

    service.handle_event(_transcript(clock, "hello"))
    assert service._last_transcript_lang is None


def test_maybe_triage_passes_lang_to_lane():
    service, _, clock, _, lane = _triage_service()
    event = _transcript(clock, "hola")
    event["lang"] = "es"
    service.handle_event(event)
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert len(lane.submits) == 1
    _, _, lang, _ = lane.submits[0]
    assert lang == "es"


def test_maybe_triage_runs_in_idle_mode():
    service, _, clock, triage, lane = _triage_service()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert len(lane.submits) == 1
    state, questions, lang, gen = lane.submits[0]
    assert state == "- hello"
    assert questions is triage.questions
    assert lang is None
    assert gen == 0
    assert triage.build_state_calls == [("hello",)]
    assert service.triage_history == ["hello"]


def test_maybe_triage_uses_transcript_buffer_not_chat_history():
    service, _, clock, triage, lane = _triage_service(chat_fn=None)
    service.chat_history = ["stale chat only"]
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert lane.submits == []
    assert triage.build_state_calls == []

    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert len(lane.submits) == 1
    assert triage.build_state_calls == [("hello",)]


def test_maybe_triage_requires_settled_turn():
    service, _, clock, _, lane = _triage_service()
    service.handle_event(_transcript(clock, "hello"))
    service.tick()

    assert lane.submits == []


def test_maybe_triage_periodic_fires_before_turn_settles():
    service, _, clock, _, lane = _triage_service()
    service.cfg.triage.periodic_s = 0.2
    service.handle_event(_transcript(clock, "hello"))
    assert service.turn_ended() is False

    clock.advance(0.25)  # due (0.2) yet still quieter than turn_gap_s (0.5)
    service.tick()

    assert service.turn_ended() is False
    assert len(lane.submits) == 1
    assert lane.submits[0][0] == "- hello"


def test_maybe_triage_periodic_disabled_ignores_accumulating_speech():
    service, _, clock, _, lane = _triage_service()
    service.cfg.triage.periodic_s = 0.0
    service.handle_event(_transcript(clock, "hello"))

    clock.advance(0.25)
    service.tick()

    assert service.turn_ended() is False
    assert lane.submits == []


def test_maybe_triage_periodic_rearms_after_submit():
    service, _, clock, _, lane = _triage_service()
    service.cfg.triage.periodic_s = 0.2
    service.handle_event(_transcript(clock, "first"))
    clock.advance(0.25)
    service.tick()
    assert len(lane.submits) == 1

    service.tick()
    service.tick()
    assert len(lane.submits) == 1  # identical window, no new speech

    service.handle_event(_transcript(clock, "second"))
    clock.advance(0.25)
    service.tick()
    assert len(lane.submits) == 2
    assert lane.submits[1][0] == "- first\n- second"


def test_triage_debug_emits_submit_and_result():
    sink = FakeDebugSink()
    service, _, clock, _, lane = _triage_service(debug_sink=sink)
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    submits = [event for event in sink.events if event[0] == "triage-submit"]
    assert submits == [
        ("triage-submit", {"trigger": "settled", "window": 1, "generation": 0})
    ]

    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, _escalated_payload())

    results = [event for event in sink.events if event[0] == "triage-result"]
    assert len(results) == 1
    assert results[0][1]["escalated"] is True
    assert results[0][1]["generation"] == 0


def test_triage_debug_marks_periodic_trigger():
    sink = FakeDebugSink()
    service, _, clock, _, lane = _triage_service(debug_sink=sink)
    service.cfg.triage.periodic_s = 0.2
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(0.25)
    service.tick()

    submits = [event for event in sink.events if event[0] == "triage-submit"]
    assert submits[0][1]["trigger"] == "periodic"


def test_triage_debug_emits_brief_dispatch():
    sink = FakeDebugSink()
    service, _, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: "RAW", debug_sink=sink
    )
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    brief = [event for event in sink.events if event[0] == "triage-brief"]
    assert brief == [("triage-brief", {"action": "dispatch", "generation": 0})]


def test_triage_debug_emits_brief_gate_skip():
    sink = FakeDebugSink()
    service, _, clock, _, lane = _triage_service(
        triage=FakeTriage(gate=False), brief_fn=lambda prompt: "RAW", debug_sink=sink
    )
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    brief = [event for event in sink.events if event[0] == "triage-brief"]
    assert brief == [
        ("triage-brief", {"action": "skip", "reason": "gate", "generation": 0})
    ]


def test_maybe_triage_skips_unready_lane():
    service, _, clock, _, lane = _triage_service(lane=FakeLane(ready=False))
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert lane.submits == []


def test_maybe_triage_skips_without_questions():
    service, _, clock, triage, lane = _triage_service(
        triage=FakeTriage(questions=None)
    )
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert lane.submits == []
    assert triage.build_state_calls == []


def test_maybe_triage_dedupes_identical_window():
    service, _, clock, triage, lane = _triage_service()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert len(lane.submits) == 1
    state, questions, lang, gen = lane.submits[0]
    assert state == "- hello"
    assert questions is triage.questions
    assert lang is None
    assert gen == 0

    service.tick()
    service.tick()
    assert len(lane.submits) == 1

    service.handle_event(_transcript(clock, "second"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 2
    assert lane.submits[1][0] == "- hello\n- second"


def test_second_transcript_produces_second_submit():
    service, _, clock, _, lane = _triage_service()
    service.handle_event(_transcript(clock, "first"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 1

    service.handle_event(_transcript(clock, "second"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()

    assert len(lane.submits) == 2
    assert lane.submits[1][0] == "- first\n- second"


def test_tick_under_50ms_with_blocked_lane():
    service, _, clock, _, lane = _triage_service()
    assert not lane.predict_gate.is_set()
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)

    start = time.perf_counter()
    service.tick()
    elapsed = time.perf_counter() - start

    assert len(lane.submits) == 1
    assert elapsed < 0.05


def test_on_triage_result_emits_signals():
    service, emitted, clock, _, lane = _triage_service()
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    payload = {
        "order": ["urgency"],
        "items": {"urgency": {"kind": "noul"}},
        "escalated": False,
        "input_tokens": 12,
    }

    service._on_triage_result(gen, payload)

    staged = _triage_events(emitted)
    assert len(staged) == 1
    assert staged[0]["stage"] == "signals"
    assert staged[0]["triage"] == service.last_triage
    assert staged[0]["triage"]["escalated"] is False
    assert staged[0]["triage"]["input_tokens"] == 12
    assert service.status()["triage"] == service.last_triage


def test_triage_events_carry_the_server_timestamp():
    service, emitted, clock, _, lane = _triage_service(brief_fn=lambda prompt: "RAW")
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    expected_ts = clock()

    service._on_triage_result(gen, _escalated_payload())

    signals = [e for e in _triage_events(emitted) if e.get("stage") == "signals"]
    assert len(signals) == 1
    assert isinstance(signals[0]["ts"], (int, float))
    assert signals[0]["ts"] == expected_ts

    job = [j for j in service._inference._queue if j.name == "brief-update"][0]
    job.run()

    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
    assert len(contents) == 1
    assert isinstance(contents[0]["ts"], (int, float))
    assert contents[0]["ts"] == expected_ts  # content reuses the signals timestamp


def test_on_triage_result_suppresses_stale_generation():
    service, emitted, clock, _, lane = _triage_service()
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    service.clear()

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is None
    assert [e for e in _triage_events(emitted) if e.get("stage") == "signals"] == []

    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 2


def test_pending_brief_capped_by_char_budget():
    """The buffer keeps the newest deltas within budget, dropping the oldest."""
    service, _, clock, _, _ = _triage_service(brief_fn=lambda prompt: "RAW")
    service.cfg.triage.brief_pending_char_budget = 30

    service.handle_event(_transcript(clock, "aaa"))
    service.handle_event(_transcript(clock, "bbbb"))
    assert service._pending_brief == ["Unknown: aaa", "Unknown: bbbb"]

    service.handle_event(_transcript(clock, "ccccc"))

    assert service._pending_brief == ["Unknown: bbbb", "Unknown: ccccc"]
    assert service.triage_history == ["aaa", "bbbb", "ccccc"]


def test_pending_brief_untouched_without_triage():
    service, _, clock = _service()
    service.handle_event(_transcript(clock, "hello"))

    assert service._pending_brief == []
    assert service.triage_history == ["hello"]


def test_on_triage_result_accepts_non_chat_mode():
    service, emitted, clock, _, lane = _triage_service()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    window = service._last_triage_window

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is not None
    assert service._last_triage_window == window
    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]


def test_on_triage_result_ignores_none_payload():
    service, emitted, clock, _, lane = _triage_service()
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, None)

    assert service.last_triage is None
    assert _triage_events(emitted) == []


def test_none_payload_dispatches_nothing_and_keeps_deltas():
    """A Laya predict failure never dispatches: no job, no events, deltas stay."""
    calls: list[str] = []
    service, emitted, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: calls.append(prompt) or "RAW"
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    assert service._pending_brief == ["Unknown: hello"]

    service._on_triage_result(gen, None)

    assert service.last_triage is None
    assert [e.get("stage") for e in _triage_events(emitted)] == []
    assert [j.name for j in service._inference._queue] == ["chat"]
    assert calls == []
    assert service._pending_brief == ["Unknown: hello"]
    assert service._brief_outstanding is False


def test_none_payload_without_brief_fn_enqueues_nothing():
    service, emitted, clock, _, lane = _triage_service()
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, None)

    assert service.last_triage is None
    assert _triage_events(emitted) == []
    assert [j.name for j in service._inference._queue] == ["chat"]
    assert service._pending_brief == ["Unknown: hello"]


def test_on_triage_result_ignores_empty_signals():
    service, emitted, clock, _, lane = _triage_service(
        triage=FakeTriage(normalized={})
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is None
    assert _triage_events(emitted) == []


def test_substantive_turn_dispatches_one_brief_job():
    """A substantive settled turn calls the gate once and queues one brief job."""
    service, emitted, clock, triage, lane = _triage_service(
        triage=FakeTriage(escalate=True), brief_fn=lambda prompt: "RAW"
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "we might churn"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(
        gen, {"order": [], "items": {}, "escalated": False, "input_tokens": 0}
    )

    assert service.last_triage is not None
    assert service.last_triage["escalated"] is True
    assert service._last_brief_signals == service.last_triage
    assert service._inference.depth() == 2
    assert [j.name for j in service._inference._queue] == ["chat", "brief-update"]
    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]
    assert len(triage.gate_calls) == 1
    assert service._brief_outstanding is True
    assert service._brief_token == 1


def test_brief_dispatches_without_escalation():
    service, _, clock, _, lane = _triage_service(brief_fn=lambda prompt: "RAW")
    service.toggle_chat()
    service.handle_event(_transcript(clock, "all calm"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(
        gen, {"order": [], "items": {}, "escalated": False, "input_tokens": 0}
    )

    assert service.last_triage is not None
    assert service._inference.depth() == 2
    assert [j.name for j in service._inference._queue] == ["chat", "brief-update"]


def test_gated_out_turn_skips_brief():
    """A gated-out turn stores signals but never calls brief_fn."""
    calls: list[str] = []
    service, emitted, clock, triage, lane = _triage_service(
        triage=FakeTriage(gate=False),
        brief_fn=lambda prompt: calls.append(prompt) or "RAW",
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "small talk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is not None
    assert "brief" not in service.last_triage  # no prior brief to retain
    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]
    assert calls == []
    assert [j.name for j in service._inference._queue] == ["chat"]
    assert service._brief_outstanding is False
    assert len(triage.gate_calls) == 1


def test_gated_out_turn_retains_previous_brief():
    """A gated-out settled turn keeps the last accepted brief in the panel."""
    service, emitted, clock, triage, lane = _triage_service(
        triage=FakeTriage(gate=False), brief_fn=lambda prompt: "RAW"
    )
    service.toggle_chat()
    signals = _escalated_payload()
    service._last_brief_signals = signals
    token = _seed_brief(service, "Unknown: first")
    service._run_brief_update(0, signals, service._now(), token)
    assert service.last_triage is not None
    assert service.last_triage["brief"] == "RAW"
    emitted.clear()

    service.handle_event(_transcript(clock, "small talk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is not None
    assert service.last_triage["brief"] == "RAW"
    assert "brief" not in service._last_brief_signals  # raw signals stay raw
    staged = [e for e in _triage_events(emitted) if e.get("stage") == "signals"]
    assert len(staged) == 1
    assert staged[0]["triage"] == service.last_triage
    assert staged[0]["triage"]["brief"] == "RAW"
    assert len(triage.gate_calls) == 1


def test_gated_out_turn_retains_brief_when_extract_disabled():
    """`extract_enabled = false` skips the dispatch but still retains the brief."""
    service, emitted, clock, _, lane = _triage_service(
        triage=FakeTriage(gate=False), brief_fn=lambda prompt: "RAW"
    )
    service.toggle_chat()
    signals = _escalated_payload()
    service._last_brief_signals = signals
    token = _seed_brief(service, "Unknown: first")
    service._run_brief_update(0, signals, service._now(), token)
    assert service.last_triage is not None
    assert service.last_triage["brief"] == "RAW"
    service.cfg.triage.extract_enabled = False
    emitted.clear()

    service.handle_event(_transcript(clock, "small talk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is not None
    assert service.last_triage["brief"] == "RAW"
    assert [j.name for j in service._inference._queue] == ["chat"]
    staged = [e for e in _triage_events(emitted) if e.get("stage") == "signals"]
    assert staged[0]["triage"]["brief"] == "RAW"


def test_gated_out_turn_retains_brief_without_brief_fn():
    """No brief_fn means no dispatch, but a retained brief is still carried."""
    service, emitted, clock, _, lane = _triage_service(triage=FakeTriage(gate=False))
    service.toggle_chat()
    service.last_triage = {**_escalated_payload(), "brief": "KEEP"}
    emitted.clear()

    service.handle_event(_transcript(clock, "small talk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage["brief"] == "KEEP"
    staged = [e for e in _triage_events(emitted) if e.get("stage") == "signals"]
    assert staged[0]["triage"]["brief"] == "KEEP"


def test_later_accepted_update_replaces_retained_brief():
    """A later accepted update overwrites the retained brief text."""
    texts = iter(["OLD", "NEW"])
    service, emitted, _, _, _ = _triage_service(brief_fn=lambda prompt: next(texts))
    service.toggle_chat()
    signals = _escalated_payload()
    service._last_brief_signals = signals
    token = _seed_brief(service, "Unknown: first")
    service._run_brief_update(0, signals, service._now(), token)
    assert service.last_triage is not None
    assert service.last_triage["brief"] == "OLD"

    token = _seed_brief(service, "Unknown: second")
    service._run_brief_update(0, signals, service._now(), token)

    assert service.last_triage is not None
    assert service.last_triage["brief"] == "NEW"
    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
    assert contents[-1]["triage"]["brief"] == "NEW"

    service.clear()
    assert service.last_triage is None


def test_no_brief_when_disabled_in_config():
    calls: list[str] = []
    service, _, clock, _, lane = _triage_service(
        triage=FakeTriage(escalate=True),
        brief_fn=lambda prompt: calls.append(prompt) or "RAW",
    )
    service.cfg.triage.extract_enabled = False
    service.toggle_chat()
    service.handle_event(_transcript(clock, "churn risk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is not None
    assert service.last_triage["escalated"] is True
    assert calls == []
    assert [j.name for j in service._inference._queue] == ["chat"]


def test_no_brief_without_brief_fn():
    service, _, clock, _, lane = _triage_service(triage=FakeTriage(escalate=True))
    service.toggle_chat()
    service.handle_event(_transcript(clock, "churn risk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service.last_triage is not None
    assert [j.name for j in service._inference._queue] == ["chat"]
    assert service._brief_outstanding is False


def test_brief_refused_when_queue_full():
    """A refused submit leaves deltas buffered and the outstanding flag clear."""
    service, emitted, clock, _, lane = _triage_service(
        triage=FakeTriage(escalate=True), brief_fn=lambda prompt: "RAW"
    )
    service._inference = InferenceQueue(size=2)
    service.toggle_chat()
    service.handle_event(_transcript(clock, "churn risk"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert service._inference.depth() == 1
    _, _, _, gen = lane.submits[0]

    service._on_triage_result(gen, _escalated_payload())

    assert service._inference.depth() == 1
    assert service._inference._queue[0].name == "chat"
    assert service.last_triage is not None
    assert service.last_triage["escalated"] is True
    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]
    assert service._pending_brief == ["Unknown: churn risk"]
    assert service._brief_outstanding is False


def test_second_result_while_outstanding_is_refused():
    """At most one brief job is outstanding: the second turn only stores signals."""
    service, emitted, clock, _, lane = _triage_service(brief_fn=lambda prompt: "RAW")
    service.toggle_chat()
    service.handle_event(_transcript(clock, "first"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen0 = lane.submits[0]
    service._on_triage_result(gen0, _escalated_payload())
    assert [j.name for j in service._inference._queue] == ["chat", "brief-update"]

    service.handle_event(_transcript(clock, "second"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen1 = lane.submits[1]
    service._on_triage_result(gen1, _escalated_payload())

    jobs = [j for j in service._inference._queue if j.name == "brief-update"]
    assert len(jobs) == 1
    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals", "signals"]
    assert service._brief_outstanding is True
    assert service._pending_brief == ["Unknown: first", "Unknown: second"]


def _seed_brief(service, *deltas):
    service._pending_brief = list(deltas)
    service._brief_outstanding = True
    return service._brief_token


def test_ledger_records_fact_and_survives_model_reemit():
    """A recorded fact persists even when a later reply omits it."""
    first = _brief_reply("Summary: rolling", ["The budget was $305 million."])
    second = _brief_reply("Summary: newer", [])
    replies = iter([first, second])
    service, _, _, _, _ = _triage_service(brief_fn=lambda prompt: next(replies))

    service._pending_brief = ["Unknown: one"]
    service._brief_outstanding = True
    service._run_brief_update(0, _escalated_payload(), service._now(), service._brief_token)

    service._pending_brief = ["Unknown: two"]
    service._brief_outstanding = True
    service._run_brief_update(0, _escalated_payload(), service._now(), service._brief_token)

    assert service.last_triage is not None
    brief = service.last_triage["brief"]
    assert "The budget was $305 million." in brief
    assert service._fact_ledger.facts() == ["The budget was $305 million."]


def test_ledger_dedupes_repeated_fact():
    """The same fact proposed twice is recorded once."""
    reply = _brief_reply("Summary: s", ["Sense could find prey.", "sense could find prey"])
    service, _, _, _, _ = _triage_service(brief_fn=lambda prompt: reply)
    service._pending_brief = ["Unknown: one"]
    service._brief_outstanding = True

    service._run_brief_update(0, _escalated_payload(), service._now(), service._brief_token)

    assert service._fact_ledger.facts() == ["Sense could find prey."]


def test_ledger_prompt_carries_ledger_and_narrative():
    """The brief request hands the model the ledger, narrative, and new deltas."""
    seen: list[str] = []
    service, _, _, _, _ = _triage_service(
        brief_fn=lambda prompt: seen.append(prompt) or _brief_reply("UPDATED")
    )
    service._fact_ledger.merge(["PRIOR FACT"])
    service._last_brief_narrative = "PRIOR NARRATIVE"
    service._pending_brief = ["Unknown: hello there"]
    service._brief_outstanding = True

    service._run_brief_update(0, _escalated_payload(), service._now(), service._brief_token)

    assert len(seen) == 1
    assert "CURRENT LEDGER:" in seen[0]
    assert "- PRIOR FACT" in seen[0]
    assert "CURRENT BRIEF:\nPRIOR NARRATIVE" in seen[0]
    assert "NEW UTTERANCES:\n- Unknown: hello there" in seen[0]


def test_brief_merges_content_and_writes_file():
    brief_file = FakeBriefFile()
    service, emitted, _, triage, _ = _triage_service(
        brief_fn=lambda prompt: "the brief", brief_file=brief_file
    )
    service.toggle_chat()
    signals = _escalated_payload()
    service._last_brief_signals = signals
    token = _seed_brief(service, "Unknown: hello")
    ts = service._now()

    service._run_brief_update(0, signals, ts, token)

    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
    assert len(contents) == 1
    assert contents[0]["triage"] == {**signals, "brief": "the brief"}
    assert service.last_triage == contents[0]["triage"]
    assert triage.prompt_calls == [(("Unknown: hello",), signals)]
    assert brief_file.writes == ["the brief"]
    assert service._pending_brief == []
    assert service._brief_outstanding is False

def test_empty_brief_requeues_deltas():
    """A `""` reply keeps signals, re-buffers the deltas, and clears the flag."""
    service, emitted, _, _, _ = _triage_service(brief_fn=lambda prompt: "")
    service.toggle_chat()
    signals = _escalated_payload()
    service.last_triage = signals
    service._last_brief_signals = signals
    token = _seed_brief(service, "Unknown: hello")
    ts = service._now()

    service._run_brief_update(0, signals, ts, token)

    assert [e for e in _triage_events(emitted) if e.get("stage") == "content"] == []
    assert service.last_triage == signals
    assert "brief" not in service.last_triage
    assert service._pending_brief == ["Unknown: hello"]
    assert service._brief_outstanding is False


def test_brief_error_requeues_and_clears_flag():
    """A raising brief_fn leaves the flag clear and the deltas buffered."""

    def boom(prompt):
        raise RuntimeError("llm down")

    service, emitted, _, _, _ = _triage_service(brief_fn=boom)
    service.toggle_chat()
    token = _seed_brief(service, "Unknown: hello")
    ts = service._now()

    service._run_brief_update(0, _escalated_payload(), ts, token)

    assert _triage_events(emitted) == []
    assert service.last_triage is None
    assert service._pending_brief == ["Unknown: hello"]
    assert service._brief_outstanding is False


def test_brief_empty_buffer_returns_early():
    """An empty buffer returns early: no call, no event, flag cleared."""
    calls: list[str] = []
    service, emitted, _, _, _ = _triage_service(
        brief_fn=lambda prompt: calls.append(prompt) or "RAW"
    )
    service.toggle_chat()
    service._brief_outstanding = True
    token = service._brief_token
    ts = service._now()
    assert service._pending_brief == []

    service._run_brief_update(0, _escalated_payload(), ts, token)

    assert calls == []
    assert _triage_events(emitted) == []
    assert service._brief_outstanding is False


def test_brief_suppressed_after_reset():
    service, emitted, _, _, _ = _triage_service(brief_fn=lambda prompt: "RAW")
    service.toggle_chat()
    service._pending_brief = ["Unknown: hello"]
    service._reset_triage()
    emitted.clear()

    service._run_brief_update(0, _escalated_payload(), service._now(), 0)

    assert _triage_events(emitted) == []
    assert service.last_triage is None
    assert service._pending_brief == ["Unknown: hello"]


def test_brief_works_after_leaving_chat():
    seen = []
    service, emitted, _, _, _ = _triage_service(
        brief_fn=lambda prompt: seen.append(prompt) or "RAW"
    )
    service.toggle_chat()
    assert service.modes.exit_chat() is True
    token = _seed_brief(service, "Unknown: hello")
    signals = _escalated_payload()
    service._last_brief_signals = signals

    service._run_brief_update(0, signals, service._now(), token)

    assert seen != []
    assert [e.get("stage") for e in _triage_events(emitted)] == ["content"]
    assert service.last_triage is not None
    assert service.last_triage["brief"] == "RAW"


def test_deltas_arriving_during_call_survive_and_flush():
    """Deltas appended mid-call are not consumed; the finally flushes them."""
    service, emitted, clock, _, _ = _triage_service()
    service.toggle_chat()
    signals = _escalated_payload()
    service._last_brief_signals = signals
    token = _seed_brief(service, "Unknown: first")

    appended: list[int] = []

    def _brief(prompt, on_io=None, finalize=None):
        if not appended:
            appended.append(1)
            service.handle_event(_transcript(clock, "during call"))
        return _brief_reply("first brief")

    service._brief_fn = _brief
    service._run_brief_update(0, signals, clock(), token)

    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
    assert len(contents) == 1
    assert contents[0]["triage"]["brief"] == "first brief"
    flushed = [j for j in service._inference._queue if j.name == "brief-update"]
    assert len(flushed) == 1
    assert service._brief_outstanding is True
    assert service._pending_brief == ["Unknown: during call"]

    flushed[0].run()

    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
    assert len(contents) == 2
    assert contents[1]["triage"]["brief"] == "first brief"
    assert service._pending_brief == []
    assert service._brief_outstanding is False


def test_in_flight_brief_after_clear_emits_and_writes_nothing():
    """Stale in-flight job: no event, no file write, no requeue, flag untouched."""
    brief_file = FakeBriefFile()
    service, emitted, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: "RAW", brief_file=brief_file
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, _escalated_payload())
    job = [j for j in service._inference._queue if j.name == "brief-update"][0]
    assert service._brief_outstanding is True

    service.clear()
    assert service._brief_outstanding is False
    token_after_clear = service._brief_token
    emitted.clear()

    job.run()

    assert [e for e in _triage_events(emitted)] == []
    assert brief_file.writes == []
    assert service._pending_brief == []
    assert service.last_triage is None
    assert service._brief_token == token_after_clear
    assert service._brief_outstanding is False


def test_clear_resets_triage():
    resets: list[int] = []
    chats: list[int] = []
    brief_file = FakeBriefFile()
    service, emitted, clock, _, lane = _triage_service(
        triage=FakeTriage(escalate=True),
        brief_fn=lambda prompt: "RAW",
        brief_file=brief_file,
        brief_reset=lambda: resets.append(1),
        chat_reset=lambda: chats.append(1),
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "churn"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, _escalated_payload())
    assert service._inference.depth() == 2
    assert service._pending_brief == ["Unknown: churn"]
    assert service._brief_outstanding is True
    token_before = service._brief_token

    assert service.clear() is True

    assert service._triage_generation == 1
    assert service.last_triage is None
    assert service._last_triage_window == ()
    assert service.triage_history == []
    assert service._pending_brief == []
    assert service._brief_token == token_before + 1
    assert service._brief_outstanding is False
    assert service._last_brief_signals is None
    assert resets == [1]
    assert brief_file.clears == 1
    assert chats == [1]
    assert lane.drops == 1
    assert service._inference.depth() == 0
    assert list(service._inference._queue) == []
    cleared = [e for e in _triage_events(emitted) if e.get("stage") == "cleared"]
    assert len(cleared) == 1
    assert cleared[0]["triage"] is None
    assert {"type": "cleared"} in emitted

    signals_before = len(
        [e for e in _triage_events(emitted) if e.get("stage") == "signals"]
    )
    service._on_triage_result(gen, _escalated_payload())
    assert service.last_triage is None
    assert (
        len([e for e in _triage_events(emitted) if e.get("stage") == "signals"])
        == signals_before
    )

    submits_before = len(lane.submits)
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == submits_before


def test_exit_chat_does_not_reset_triage():
    chats: list[int] = []
    bresets: list[int] = []
    brief_file = FakeBriefFile()
    service, emitted, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: "RAW",
        chat_reset=lambda: chats.append(1),
        brief_reset=lambda: bresets.append(1),
        brief_file=brief_file,
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 1
    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, _escalated_payload())
    assert service.last_triage is not None
    assert service._triage_generation == 0

    assert service.toggle_chat() is True

    assert service.modes.mode.value == "idle"
    assert service._triage_generation == 0
    assert service.last_triage is not None
    assert service._last_triage_window == ("hello",)
    assert service.triage_history == ["hello"]
    assert service._pending_brief == ["Unknown: hello"]
    assert service._last_brief_signals is not None
    assert service._brief_outstanding is True
    assert lane.drops == 0
    cleared = [e for e in _triage_events(emitted) if e.get("stage") == "cleared"]
    assert cleared == []
    assert chats == [1]
    assert bresets == []
    assert brief_file.clears == 0


def test_toggle_chat_enter_does_not_touch_triage():
    """Entering chat resets only the chat buffers; the brief engine is independent."""
    bresets: list[int] = []
    brief_file = FakeBriefFile()
    service, emitted, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: "RAW",
        brief_reset=lambda: bresets.append(1),
        brief_file=brief_file,
    )
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, _escalated_payload())
    triage_before = service.last_triage
    pending_before = list(service._pending_brief)

    assert service.toggle_chat() is True

    assert service.modes.mode.value == "chat"
    assert service._triage_generation == 0
    assert service.last_triage is triage_before
    assert service._last_triage_window == ("hello",)
    assert service.triage_history == ["hello"]
    assert service._pending_brief == pending_before
    assert service._last_brief_signals is not None
    assert service._brief_outstanding is True
    assert lane.drops == 0
    assert bresets == []
    assert brief_file.clears == 0
    cleared = [e for e in _triage_events(emitted) if e.get("stage") == "cleared"]
    assert cleared == []


def test_clear_without_triage_emits_no_triage_event():
    service, emitted, _ = _service()
    brief_file = FakeBriefFile()
    bresets: list[int] = []
    service._brief_file = brief_file
    service._brief_reset = lambda: bresets.append(1)

    service.clear()

    assert {"type": "cleared"} in emitted
    assert _triage_events(emitted) == []
    assert brief_file.clears == 1
    assert bresets == []


def test_stale_brief_after_clear_does_not_clear_newer_dispatch_flag():
    """A stale in-flight job must not clear a newer dispatch's outstanding flag."""
    brief_file = FakeBriefFile()
    service, emitted, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: "RAW", brief_file=brief_file
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, _escalated_payload())
    stale_job = [j for j in service._inference._queue if j.name == "brief-update"][0]
    assert service._brief_outstanding is True

    service.clear()

    # A fresh turn dispatches a new brief AFTER clear bumped the token.
    service.handle_event(_transcript(clock, "fresh"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen2 = lane.submits[1]
    service._on_triage_result(gen2, _escalated_payload())
    assert service._brief_outstanding is True
    newer_token = service._brief_token
    pending_before = list(service._pending_brief)
    triage_before = service.last_triage
    emitted.clear()

    stale_job.run()

    assert _triage_events(emitted) == []
    assert brief_file.writes == []
    assert service._pending_brief == pending_before
    assert service._brief_outstanding is True
    assert service._brief_token == newer_token
    assert service.last_triage is triage_before


def test_clear_resets_chat_latch():
    service, emitted, _ = _service()
    service.toggle_chat()
    service._chat_queue_full = True
    service._pending_chat = ["stuck text"]

    service.clear()

    assert service._chat_queue_full is False
    assert service._pending_chat == []
    assert {"type": "cleared"} in emitted


def test_triage_end_to_end_signals_content_cleared():
    """Two settled turns collapse into ONE brief job (single-outstanding guard).

    The second `_on_triage_result` is refused as `"outstanding"`, so the queue
    holds one `brief-update` that consumes both deltas and emits one `content`.
    The panel keeps the FIRST turn's signals until the next settled turn (lag).
    """
    prompts: list[str] = []
    service, emitted, clock, triage, lane = _triage_service(
        brief_fn=lambda prompt: prompts.append(prompt) or "the brief"
    )
    service.toggle_chat()

    service.handle_event(_transcript(clock, "first thought"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 1
    _, _, _, gen0 = lane.submits[0]
    service._on_triage_result(
        gen0, {"order": ["urgency"], "items": {}, "escalated": False, "input_tokens": 1}
    )

    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals"]
    assert prompts == []
    assert service.last_triage is not None
    assert service.last_triage["escalated"] is False
    assert [j.name for j in service._inference._queue] == ["chat", "brief-update"]

    triage._escalate = True
    service.handle_event(_transcript(clock, "we might churn"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 2
    _, _, _, gen1 = lane.submits[1]
    service._on_triage_result(gen1, _escalated_payload())

    assert [e.get("stage") for e in _triage_events(emitted)] == ["signals", "signals"]
    assert [j.name for j in service._inference._queue] == [
        "chat",
        "brief-update",
        "chat",
    ]
    for job in [j for j in service._inference._queue if j.name == "brief-update"]:
        job.run()

    contents = [e for e in _triage_events(emitted) if e.get("stage") == "content"]
    assert len(contents) == 1
    assert len(prompts) == 1
    deltas, _ = triage.prompt_calls[0]
    assert deltas == ("Unknown: first thought", "Unknown: we might churn")
    assert contents[0]["triage"]["escalated"] is False
    assert contents[0]["triage"]["brief"] == "the brief"
    assert service.last_triage == contents[0]["triage"]
    assert service.status()["triage"] == contents[0]["triage"]

    service.clear()

    cleared = [e for e in _triage_events(emitted) if e.get("stage") == "cleared"]
    assert len(cleared) == 1
    assert cleared[0]["triage"] is None
    assert service.last_triage is None
    service._on_triage_result(gen1, _escalated_payload())
    assert [e.get("stage") for e in _triage_events(emitted)] == [
        "signals",
        "signals",
        "content",
        "cleared",
    ]

    assert service.toggle_chat() is True
    service.handle_event(_transcript(clock, "idle thought"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    assert len(lane.submits) == 3
    assert lane.submits[2][0].endswith("- idle thought")


def test_tick_emits_status_on_triage_state_change():
    service, emitted, _, triage, _ = _triage_service(
        triage=FakeTriage(state_pair=("loading", "warming up"))
    )

    service.tick()
    statuses = [e for e in emitted if e["type"] == "status"]
    assert len(statuses) == 1
    assert statuses[0]["triage_state"] == "loading"
    assert statuses[0]["triage_reason"] == "warming up"
    assert "triage" in statuses[0]

    triage._state_pair = ("ok", "")
    service.tick()
    statuses = [e for e in emitted if e["type"] == "status"]
    assert len(statuses) == 2
    assert statuses[1]["triage_state"] == "ok"
    assert statuses[1]["triage_reason"] == ""
    assert "triage" in statuses[1]

    service.tick()
    assert len([e for e in emitted if e["type"] == "status"]) == 2


def test_no_status_emitted_when_triage_disabled():
    service, emitted, _ = _service()

    service.tick()
    service.tick()

    assert [e for e in emitted if e["type"] == "status"] == []


def test_tick_heals_triage_panel_after_lane_ready():
    import time as _time
    from pathlib import Path

    from stt_daemon.assistant.laya_lane import LayaLane
    from stt_daemon.assistant.triage import TriageAnalyzer
    from stt_daemon.tui.state import ViewState

    class _FakeRouter:
        def preload(self, names):
            return self

        def predict(self, state, questions, model=None, lang=None, **kwargs):
            if "_warmup" in questions:
                return {"model": model, "answers": {}, "usage": {}, "routing": {}}
            return {
                "model": model,
                "answers": {
                    "frustration": {
                        "noul": 0.1,
                        "confidence": 0.9,
                        "answer_confidence": 0.9,
                    },
                    "urgency": {
                        "choice": "low",
                        "confidence": 0.9,
                        "answer_confidence": 0.9,
                    },
                },
                "usage": {"input_tokens": 5},
            }

    def _factory(**kwargs):
        return _FakeRouter()

    cfg = Config()
    cfg.assistant.turn_gap_s = 0.5
    cfg.triage.questions_file = str(
        Path(__file__).resolve().parent.parent / "questions.example.json"
    )
    clock = Clock()
    emitted: list[dict] = []

    def emit(type_, **fields):
        emitted.append({"type": type_, **fields})

    analyzer = TriageAnalyzer(cfg.triage)
    assert analyzer.questions, "example questions must load"
    holder: dict = {}
    lane = LayaLane(
        cfg.triage,
        agent_factory=_factory,
        on_result=lambda gen, payload: holder["svc"]._on_triage_result(gen, payload),
        on_state=lambda state, reason: analyzer.refresh(state, reason),
    )
    service = AssistantService(
        cfg,
        FakeSubscriber(),
        emit=emit,
        time_fn=clock,
        triage=analyzer,
        laya_lane=lane,
        chat_fn=_ok_reply,
    )
    holder["svc"] = service
    try:
        lane.start()
        deadline = _time.monotonic() + 5.0
        while not lane.ready() and _time.monotonic() < deadline:
            _time.sleep(0.005)
        assert lane.ready(), "lane did not become ready"
        assert analyzer.state() == ("ok", "")

        service.tick()
        statuses = [e for e in emitted if e["type"] == "status"]
        assert statuses, "tick emitted no status after lane ready"
        assert statuses[-1]["triage_state"] == "ok"

        service.handle_event(_transcript(clock, "hello"))
        clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
        service.tick()

        deadline = _time.monotonic() + 5.0
        while service.last_triage is None and _time.monotonic() < deadline:
            _time.sleep(0.005)
        assert service.last_triage is not None, "no triage signals from lane"

        view = ViewState()
        for event in emitted:
            view.ingest(event)
        assert view.triage_state == "ok"
        lines = view.triage_lines()
        assert lines
        assert not any("unavailable" in line for line in lines)
        assert not any("loading" in line for line in lines)
    finally:
        lane.stop()


def _dump_brief(prompt, on_io=None, finalize=None):
    """Brief fake that mimics LlmClient.chat's on_io phases."""
    if on_io is not None:
        on_io({"phase": "request", "url": "http://llm/v1", "body": {"prompt": prompt}})
        on_io({"phase": "response", "payload": {"content": "parsed"}, "error": None})
    return _brief_reply("BRIEF")


def _run_triage_flow(service, emitted, clock, lane, payload):
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]
    service._on_triage_result(gen, payload)
    return gen


def _dump_files(tmp_path):
    return list(tmp_path.rglob("*.json"))


def test_dump_writes_full_laya_and_llm_io(tmp_path):
    service, emitted, clock, triage, lane = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True

    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())
    jobs = [j for j in service._inference._queue if j.name == "brief-update"]
    assert len(jobs) == 1
    jobs[0].run()

    files = _dump_files(tmp_path)
    assert len(files) == 1
    assert re.fullmatch(r"\d{6}-\d{3}\.json", files[0].name)
    data = json.loads(files[0].read_text(encoding="utf-8"))
    assert data["ts"] == clock()
    assert isinstance(data["iso"], str) and data["iso"]
    assert data["laya"]["request"]["state"] == "- hello"
    assert data["laya"]["request"]["questions"] == triage.questions
    assert data["laya"]["request"]["model"] == "multilingual"
    assert data["laya"]["request"]["max_len"] == 1024
    assert data["laya"]["response"] == _escalated_payload()
    assert data["laya"]["error"] is None
    assert data["llm"]["request"]["url"] == "http://llm/v1"
    assert data["llm"]["request"]["body"] == {
        "prompt": "CURRENT LEDGER:\n(none yet)\n\nCURRENT BRIEF:\n(none yet)\n\n"
        "NEW UTTERANCES:\n- Unknown: hello"
    }
    assert data["llm"]["response"] == {"content": "parsed"}
    assert data["llm"]["error"] is None
    assert service._triage_dump._pending == {}


def test_dump_laya_error_writes_laya_only_and_keeps_deltas(tmp_path):
    """A Laya predict failure finalizes the Laya half with `llm: null`.

    Nothing is dispatched, so no LLM half is ever recorded and the buffered
    deltas survive for the next settled turn.
    """
    service, emitted, clock, _, lane = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True

    _run_triage_flow(service, emitted, clock, lane, None)

    assert [j.name for j in service._inference._queue] == ["chat"]
    assert service._pending_brief == ["Unknown: hello"]

    files = _dump_files(tmp_path)
    assert len(files) == 1
    data = json.loads(files[0].read_text(encoding="utf-8"))
    assert data["laya"]["response"] is None
    assert data["laya"]["error"] == "predict failed"
    assert data["llm"] is None
    assert service._triage_dump._pending == {}


def test_dump_without_brief_has_laya_only(tmp_path):
    service, emitted, clock, _, lane = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True
    service.cfg.triage.extract_enabled = False

    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())

    files = _dump_files(tmp_path)
    assert len(files) == 1
    data = json.loads(files[0].read_text(encoding="utf-8"))
    assert data["laya"]["response"] == _escalated_payload()
    assert data["llm"] is None
    assert service._triage_dump._pending == {}


def test_dump_queue_full_has_laya_only(tmp_path):
    service, emitted, clock, _, lane = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True
    service._inference = InferenceQueue(size=2)

    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())

    assert service._inference.depth() == 1
    assert [j.name for j in service._inference._queue] == ["chat"]
    assert service._pending_brief == ["Unknown: hello"]
    files = _dump_files(tmp_path)
    assert len(files) == 1
    data = json.loads(files[0].read_text(encoding="utf-8"))
    assert data["laya"]["response"] == _escalated_payload()
    assert data["llm"] is None
    assert service._triage_dump._pending == {}


def test_dump_disabled_writes_nothing(tmp_path):
    service, emitted, clock, _, lane = _triage_service(
        brief_fn=lambda prompt: "RAW"
    )
    assert service._triage_dump.enabled is False
    service._triage_dump = TriageDumper(False, directory=str(tmp_path))

    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())
    jobs = [j for j in service._inference._queue if j.name == "brief-update"]
    assert len(jobs) == 1
    jobs[0].run()

    assert _dump_files(tmp_path) == []


def test_dump_stale_result_discards_without_file(tmp_path):
    service, emitted, clock, triage, lane = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen = lane.submits[0]

    real_normalize = triage.normalize

    def stale_normalize(payload):
        # Simulate a clear landing between the first check/begin and the re-check.
        service._triage_generation += 1
        return real_normalize(payload)

    triage.normalize = stale_normalize
    service._on_triage_result(gen, _escalated_payload())

    assert _dump_files(tmp_path) == []
    assert service._triage_dump._pending == {}


def test_dump_stale_brief_discards_without_file(tmp_path):
    service, _, clock, _, _ = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True
    ts = clock()
    service._triage_dump.begin(ts, laya={"request": {}, "response": {}, "error": None})
    service._pending_brief = ["Unknown: hello"]
    service._triage_generation += 1  # stale before the brief even starts

    service._run_brief_update(0, _escalated_payload(), ts, service._brief_token)

    assert _dump_files(tmp_path) == []
    assert service._triage_dump._pending == {}
    assert service._pending_brief == ["Unknown: hello"]


def test_dump_brief_cleared_mid_flight_writes_nothing(tmp_path):
    service, _, clock, _, _ = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True

    def stale_brief(prompt, on_io=None, finalize=None):
        # Simulate a clear landing mid-LLM-call (after the request phase fired).
        service._triage_generation += 1
        if on_io is not None:
            on_io({"phase": "request", "url": "http://llm/v1", "body": {}})
            on_io({"phase": "response", "payload": {"c": 1}, "error": None})
        return _brief_reply("RAW")

    service._brief_fn = stale_brief
    service._pending_brief = ["Unknown: hello"]
    service._brief_outstanding = True
    token = service._brief_token
    ts = clock()
    service._triage_dump.begin(ts, laya={"request": {}, "response": {}, "error": None})

    service._run_brief_update(0, _escalated_payload(), ts, token)

    assert _dump_files(tmp_path) == []
    assert service._triage_dump._pending == {}
    assert service._pending_brief == []
    assert service._brief_outstanding is False


def test_dump_gated_out_writes_laya_only_and_no_pending(tmp_path):
    """A turn the gate leaves out writes `llm: null` and leaks no pending record."""
    service, emitted, clock, triage, lane = _triage_service(
        triage=FakeTriage(escalate=True, gate=False), brief_fn=_dump_brief
    )
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True

    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())

    assert [j.name for j in service._inference._queue] == ["chat"]
    assert len(triage.gate_calls) == 1
    files = _dump_files(tmp_path)
    assert len(files) == 1
    data = json.loads(files[0].read_text(encoding="utf-8"))
    assert data["laya"]["response"] == _escalated_payload()
    assert data["llm"] is None
    assert service._triage_dump._pending == {}


def test_dump_coalesced_update_writes_laya_only_then_llm_only_flush(tmp_path):
    """A coalesced turn writes a Laya-only record; the later flush a separate llm-only one."""
    service, emitted, clock, _, lane = _triage_service(brief_fn=_dump_brief)
    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
    service.cfg.triage.dump_io = True

    service.toggle_chat()
    service.handle_event(_transcript(clock, "first"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen0 = lane.submits[0]
    service._on_triage_result(gen0, _escalated_payload())

    service.handle_event(_transcript(clock, "second"))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
    service.tick()
    _, _, _, gen1 = lane.submits[1]
    service._on_triage_result(gen1, _escalated_payload())
    assert service._brief_outstanding is True

    appended: list[int] = []

    def _brief(prompt, on_io=None, finalize=None):
        if not appended:
            appended.append(1)
            service.handle_event(_transcript(clock, "during call"))
        if on_io is not None:
            on_io({"phase": "request", "url": "http://llm/v1", "body": {"prompt": prompt}})
            on_io({"phase": "response", "payload": {"content": "parsed"}, "error": None})
        return _brief_reply("BRIEF")

    service._brief_fn = _brief
    clock.advance(1.0)

    job = [j for j in service._inference._queue if j.name == "brief-update"][0]
    service._inference._queue.remove(job)
    job.run()
    flushed = [j for j in service._inference._queue if j.name == "brief-update"]
    assert len(flushed) == 1
    flushed[0].run()

    assert service._triage_dump._pending == {}
    records = [
        json.loads(path.read_text(encoding="utf-8")) for path in _dump_files(tmp_path)
    ]
    assert len(records) == 3
    coalesced = [r for r in records if r["laya"] is not None and r["llm"] is None]
    flushed_records = [r for r in records if r["laya"] is None and r["llm"] is not None]
    both = [r for r in records if r["laya"] is not None and r["llm"] is not None]
    assert len(coalesced) == 1
    assert len(flushed_records) == 1
    assert len(both) == 1
    assert coalesced[0]["ts"] != flushed_records[0]["ts"]
    assert flushed_records[0]["llm"]["request"]["body"] == {
        "prompt": "CURRENT LEDGER:\n(none yet)\n\nCURRENT BRIEF:\nBRIEF\n\n"
        "NEW UTTERANCES:\n- Unknown: during call"
    }


def test_chat_client_failure_emits_llm_error_and_service_survives():
    """Adversarial (a): a chat client raising on `chat` surfaces as an `llm_error`
    event and the service keeps functioning afterwards."""
    from stt_daemon.assistant.llm import ConversationEngine, LlmError

    class BoomClient:
        def chat(self, messages, max_tokens=None, timeout_s=None, **kwargs):
            raise LlmError("boom")

    engine = ConversationEngine(Config().llm, BoomClient(), system_prompt="SYS")
    service, emitted, clock = _service()
    service.chat_fn = engine.send
    assert service.toggle_chat() is True

    service._run_chat("hello")

    errors = [e for e in emitted if e["type"] == "error" and e.get("code") == "llm_error"]
    assert errors and "boom" in errors[-1]["message"]
    assert service.toggle_chat() is True
    assert service.modes.mode.value == "idle"


# ---------------------------------------------------------------------------
# Conversation session adoption (SESSION/TRANSCRIPT session_dir).
# ---------------------------------------------------------------------------


def _session_event(clock, session_dir):
    event = _ev(events.SESSION, clock)
    event["session_id"] = session_dir.rsplit("/", 1)[-1]
    event["session_dir"] = session_dir
    return event


def _ring(service, clock, *items):
    """Feed a daemon-seeded replay window: status -> items -> ring_end."""
    service.handle_event(_ev(events.STATUS, clock))
    for item in items:
        service.handle_event(item)
    service.handle_event(_ev(events.RING_END, clock))


def _cleared(emitted):
    return [e for e in _triage_events(emitted) if e.get("stage") == "cleared"]


def test_status_includes_session_dir_without_brief_file():
    service, _, clock = _service()
    assert service.status()["session_dir"] is None

    service.handle_event(_session_event(clock, "transcripts/2026-09-30-12-47-29"))

    assert service.status()["session_dir"] == "transcripts/2026-09-30-12-47-29"


def test_live_session_event_sets_session_and_records_directory():
    brief_file = FakeBriefFile()
    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
    session = "transcripts/2026-09-30-12-47-29"

    service.handle_event(_session_event(clock, session))

    assert service.status()["session_dir"] == session
    assert brief_file.directories == [session]
    assert service._triage_generation == 0


def test_live_transcript_session_sets_session_without_wiping_utterance():
    brief_file = FakeBriefFile()
    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
    session = "transcripts/s1"

    service.handle_event(_transcript(clock, "hello", session_dir=session))

    assert service.status()["session_dir"] == session
    assert brief_file.directories == [session]
    assert service.triage_history == ["hello"]
    assert service._pending_brief == ["Unknown: hello"]
    assert service._triage_generation == 0


def test_brief_write_lands_under_session_dir(tmp_path):
    from stt_daemon.assistant.brief_file import BriefFile

    brief_file = BriefFile(True)
    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
    session = str(tmp_path / "transcripts" / "s1")

    service.handle_event(_transcript(clock, "hello", session_dir=session))
    brief_file.write("the brief")

    path = tmp_path / "transcripts" / "s1" / "brief.md"
    assert path.read_text(encoding="utf-8") == "the brief"


def test_live_session_change_resets_triage_and_keeps_chat():
    brief_file = FakeBriefFile()
    resets: list[int] = []
    service, emitted, clock, _, lane = _triage_service(
        brief_file=brief_file, brief_reset=lambda: resets.append(1)
    )
    service.toggle_chat()
    service.handle_event(_transcript(clock, "hello"))
    service.handle_event(_session_event(clock, "transcripts/a"))
    service.last_triage = {"signals": 1}
    service._pending_brief = ["stale"]
    emitted.clear()

    service.handle_event(_session_event(clock, "transcripts/b"))

    assert service.status()["session_dir"] == "transcripts/b"
    assert service.last_triage is None
    assert service._pending_brief == []
    assert service._triage_generation == 1
    assert service.chat_history == ["Unknown: hello"]
    assert resets == [1]
    assert lane.drops == 1
    assert brief_file.directories == ["transcripts/a", "transcripts/b"]
    assert len(_cleared(emitted)) == 1


def test_sessionless_transcript_does_not_reset():
    service, _, clock, _, _ = _triage_service()
    service.handle_event(_session_event(clock, "transcripts/a"))
    service.last_triage = {"signals": 1}

    service.handle_event(_transcript(clock, "hello"))

    assert service._triage_generation == 0
    assert service.last_triage == {"signals": 1}
    assert service.status()["session_dir"] == "transcripts/a"


def test_same_dir_readoption_does_not_reset():
    brief_file = FakeBriefFile()
    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
    session = "transcripts/a"
    service.handle_event(_session_event(clock, session))
    service.last_triage = {"signals": 1}

    service.handle_event(_transcript(clock, "again", session_dir=session))

    assert service._triage_generation == 0
    assert service.last_triage == {"signals": 1}
    assert brief_file.directories == [session]


def test_session_change_without_brief_file_does_not_raise():
    service, _, clock, _, _ = _triage_service(brief_file=None)

    service.handle_event(_session_event(clock, "transcripts/a"))
    service.handle_event(_session_event(clock, "transcripts/b"))

    assert service.status()["session_dir"] == "transcripts/b"
    assert service._triage_generation == 1


def test_replay_single_session_adopts_without_reset():
    brief_file = FakeBriefFile()
    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
    session = "transcripts/a"

    _ring(service, clock, _transcript(clock, "hello", session_dir=session))

    assert service.status()["session_dir"] == session
    assert brief_file.directories == [session]
    assert service._triage_generation == 0
    assert _cleared(emitted) == []
    assert service.triage_history == ["hello"]


def test_replay_two_sessions_commits_newest_and_resets_once():
    brief_file = FakeBriefFile()
    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
    a, b = "transcripts/a", "transcripts/b"

    _ring(
        service,
        clock,
        _session_event(clock, a),
        _transcript(clock, "old turn", session_dir=a),
        _session_event(clock, b),
        _transcript(clock, "new turn", session_dir=b),
    )

    assert service.status()["session_dir"] == b
    assert brief_file.directories == [b]
    assert service._triage_generation == 1
    assert service.last_triage is None
    assert service._pending_brief == []
    assert len(_cleared(emitted)) == 1


def test_reconnect_tailing_current_session_is_noop():
    brief_file = FakeBriefFile()
    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
    session = "transcripts/a"
    service.handle_event(_session_event(clock, session))
    service.last_triage = {"signals": 1}
    emitted.clear()

    _ring(service, clock, _transcript(clock, "replayed", session_dir=session))

    assert service.status()["session_dir"] == session
    assert brief_file.directories == [session]
    assert service._triage_generation == 0
    assert service.last_triage == {"signals": 1}
    assert _cleared(emitted) == []


def test_reconnect_older_to_current_resets_once_without_rewind():
    brief_file = FakeBriefFile()
    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
    a, b = "transcripts/a", "transcripts/b"
    service.handle_event(_session_event(clock, a))
    service.last_triage = {"signals": 1}
    emitted.clear()

    _ring(
        service,
        clock,
        _session_event(clock, a),
        _transcript(clock, "old", session_dir=a),
        _session_event(clock, b),
        _transcript(clock, "new", session_dir=b),
    )

    assert service.status()["session_dir"] == b
    assert brief_file.directories == [a, b]
    assert service._triage_generation == 1
    assert service.last_triage is None
    assert len(_cleared(emitted)) == 1


def test_reconnect_only_newer_session_resets_once():
    brief_file = FakeBriefFile()
    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
    a, b = "transcripts/a", "transcripts/b"
    service.handle_event(_session_event(clock, a))
    service.last_triage = {"signals": 1}

    _ring(service, clock, _transcript(clock, "new", session_dir=b))

    assert service.status()["session_dir"] == b
    assert service._triage_generation == 1
    assert service.last_triage is None
    assert brief_file.directories == [a, b]


def test_reconnect_span_older_current_clears_buffered_older_turns():
    """A RING_END whose ring spanned an older session resets even when the
    newest session already matches the adopted directory (no re-set_directory)."""
    brief_file = FakeBriefFile()
    service, emitted, clock, _, _ = _triage_service(brief_file=brief_file)
    a, b = "transcripts/a", "transcripts/b"
    _ring(service, clock, _session_event(clock, a))
    assert service._triage_generation == 0
    service.handle_event(_transcript(clock, "old turn", session_dir=a))
    assert service._pending_brief == ["Unknown: old turn"]

    service.handle_event(_session_event(clock, b))
    assert service._triage_generation == 1
    service.handle_event(_transcript(clock, "new turn", session_dir=b))
    assert service._pending_brief == ["Unknown: new turn"]
    emitted.clear()

    _ring(
        service,
        clock,
        _session_event(clock, a),
        _transcript(clock, "old turn", session_dir=a),
        _session_event(clock, b),
        _transcript(clock, "new turn", session_dir=b),
    )

    assert service.status()["session_dir"] == b
    assert brief_file.directories == [a, b]
    assert service._pending_brief == []
    assert service._triage_generation == 2
    assert len(_cleared(emitted)) == 1


def test_daemon_restart_new_session_resets_once():
    brief_file = FakeBriefFile()
    service, _, clock, _, _ = _triage_service(brief_file=brief_file)
    service.handle_event(_session_event(clock, "transcripts/a"))
    service.last_triage = {"signals": 1}

    service.handle_event(_session_event(clock, "transcripts/b"))

    assert service.status()["session_dir"] == "transcripts/b"
    assert service._triage_generation == 1
    assert service.last_triage is None
    assert brief_file.directories == ["transcripts/a", "transcripts/b"]