STT-Assistant/tests/test_assistant_integration.py

"""Integration: the assistant service consuming a real publish-only bus."""

import json
import os
import shutil
import signal
import sys
import tempfile
import threading
import time
from types import SimpleNamespace

import pytest

from stt_daemon import events
from stt_daemon.assistant.brief_file import BriefFile
from stt_daemon.assistant.inference import InferenceJob, InferenceQueue
from stt_daemon.assistant.llm import ConversationEngine
from stt_daemon.assistant.runner import build_triage
from stt_daemon.assistant.service import AssistantService
from stt_daemon.bus import EventPublisher, EventSubscriber
from stt_daemon.config import Config


@pytest.fixture
def sock_path():
    directory = tempfile.mkdtemp(prefix="sttb-")
    try:
        yield os.path.join(directory, "s.sock")
    finally:
        shutil.rmtree(directory, ignore_errors=True)


def _wait(predicate, timeout=5.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if predicate():
            return True
        time.sleep(0.01)
    return predicate()


def _heartbeat_fields():
    return {
        "in_flight": False,
        "segments_done": 0,
        "worker_stall_s": 0.0,
        "queue_depth": 0,
        "frames_since_last": 1,
        "rms": 0.0,
        "capture": "test",
        "device_ok": True,
        "model_loaded": True,
        "uptime_s": 1.0,
        "last_transcript_age_s": 0.0,
    }


def test_service_consumes_real_bus(sock_path):
    cfg = Config()
    publisher = EventPublisher(sock_path, ring_size=10)
    publisher.start()

    subscriber = EventSubscriber(sock_path, reconnect_delay=0.05)
    emitted: list[dict] = []
    service = AssistantService(
        cfg,
        subscriber,
        emit=lambda type_, **fields: emitted.append({"type": type_, **fields}),
    )
    service.start()

    try:
        publisher.emit(events.HEARTBEAT, **_heartbeat_fields())
        assert _wait(lambda: service.stt_ok() is True)

        assert service.arm() is True
        assert _wait(lambda: any(e.get("mode") == "command_armed" for e in emitted))

        publisher.emit(events.SPEECH_START)
        assert _wait(lambda: service.speech_open is True)

        publisher.emit(events.SPEECH_STOP, seg_id=5)
        assert _wait(lambda: service.awaited_seg_id == 5)
        assert service.ready() is False

        publisher.emit(events.SEGMENT_DONE, seg_id=5, emitted=True)
        assert _wait(lambda: service.ready() is True)

        publisher.emit(events.TRANSCRIPT, seg_id=5, text="hello", lang="en", lang_prob=0.9)
        assert _wait(lambda: service.command_text == "hello")

        assert service.execute() == "hello"
        assert service.finish() is True
    finally:
        service.stop()
        publisher.close()


def _enabled_triage_cfg(tmp_path):
    cfg = Config()
    cfg.audit.enabled = False
    cfg.llm.enabled = False
    cfg.triage.enabled = True
    path = os.path.join(str(tmp_path), "questions.json")
    with open(path, "w", encoding="utf-8") as handle:
        json.dump({"urgent": {"type": "noul", "instructions": "is it urgent"}}, handle)
    cfg.triage.questions_file = path
    return cfg


class _FakeChatClient:
    def __init__(self):
        self.calls = []

    def chat(self, messages, tools=None, max_tokens=None, timeout_s=None, slot_id=None):
        self.calls.append(
            {
                "messages": messages,
                "tools": tools,
                "max_tokens": max_tokens,
                "timeout_s": timeout_s,
                "slot_id": slot_id,
            }
        )
        return SimpleNamespace(content="EXTRACTED", finish_reason="stop")


def test_build_triage_disabled_returns_nones():
    cfg = Config()
    assert cfg.triage.enabled is False
    assert build_triage(cfg, None) == (None, None, None, None)
    assert build_triage(cfg, _FakeChatClient()) == (None, None, None, None)
    assert "laya" not in sys.modules
    service = AssistantService(cfg, subscriber=object())
    assert service.status()["triage_state"] == "disabled"


def test_build_triage_enabled_constructs_without_client(tmp_path):
    cfg = _enabled_triage_cfg(tmp_path)
    before = len(threading.enumerate())
    analyzer, lane, brief_fn, brief_reset = build_triage(cfg, None)
    assert analyzer is not None
    assert lane is not None
    assert brief_fn is None
    assert brief_reset is None
    assert analyzer._cfg is cfg.triage
    assert lane._cfg is cfg.triage
    assert analyzer.state() == ("loading", "")
    assert len(threading.enumerate()) == before
    assert "laya" not in sys.modules


def test_build_triage_brief_fn_calls_client(tmp_path):
    cfg = _enabled_triage_cfg(tmp_path)
    client = _FakeChatClient()
    _, _, brief_fn, brief_reset = build_triage(cfg, client)
    assert brief_fn is not None
    assert brief_reset is not None
    assert callable(brief_fn)
    assert callable(brief_reset)
    assert brief_fn("PROMPT") == "EXTRACTED"
    assert len(client.calls) == 1
    call = client.calls[0]
    assert call["messages"] == [
        {"role": "system", "content": cfg.llm.triage_system_prompt},
        {"role": "user", "content": "PROMPT"},
    ]
    assert call["max_tokens"] == cfg.triage.extract_max_tokens
    assert call["timeout_s"] == cfg.triage.extract_timeout_s
    assert call["slot_id"] is None


def test_build_triage_brief_fn_slot_only_when_configured(tmp_path):
    cfg = _enabled_triage_cfg(tmp_path)
    cfg.triage.brief_slot_id = 2
    client = _FakeChatClient()
    _, _, brief_fn, _ = build_triage(cfg, client)
    assert brief_fn is not None
    brief_fn("PROMPT")
    assert client.calls[0]["slot_id"] == 2


def test_build_triage_on_state_refreshes_analyzer(tmp_path):
    from stt_daemon.assistant.laya_lane import LAYA_MISSING_HINT

    cfg = _enabled_triage_cfg(tmp_path)
    sentinel = object()
    analyzer, lane, _, _ = build_triage(cfg, None, on_result=sentinel)
    assert analyzer is not None
    assert lane is not None
    assert lane._on_result is sentinel
    assert callable(lane._on_state)
    lane._on_state("ok", "")
    assert analyzer.state() == ("ok", "")
    lane._on_state("laya_missing", LAYA_MISSING_HINT)
    assert analyzer.state() == ("laya_missing", LAYA_MISSING_HINT)


def _run_runner_main(monkeypatch, cfg):
    from stt_daemon.assistant import runner as runner_mod
    import stt_daemon.assistant.laya_lane as lane_mod

    calls = []
    made_lanes = []
    made_services = []

    class FakeLane:
        def __init__(self, triage_cfg, agent_factory=None, on_result=None, on_state=None):
            made_lanes.append(self)
            self.cfg = triage_cfg
            self.on_state = on_state
            self.on_result = on_result
            calls.append("lane.construct")

        def start(self):
            calls.append("lane.start")

        def stop(self):
            calls.append("lane.stop")

    class FakeSubscriber:
        def close(self):
            calls.append("subscriber.close")

    class FakeService:
        def __init__(self, cfg, subscriber, **kwargs):
            made_services.append(self)
            self.kwargs = kwargs
            calls.append("service.construct")

        def set_sinks(self, **kwargs):
            pass

        def chat_notify(self, *args, **kwargs):
            return None

        def _on_triage_result(self, generation, payload):
            calls.append("triage.result")

        def start(self):
            calls.append("service.start")

        def stop(self):
            calls.append("service.stop")

    class FakeControl:
        def __init__(self, cfg, service):
            self.service = service
            calls.append("control.construct")

        path = "/tmp/fake.sock"

        def start(self):
            calls.append("control.start")

        def broadcast(self, *args, **kwargs):
            pass

        def relay(self, event):
            pass

        def close(self):
            calls.append("control.close")

    class FakeEvent:
        def set(self):
            pass

        def wait(self, timeout=None):
            return True

    monkeypatch.setattr(lane_mod, "LayaLane", FakeLane)
    monkeypatch.setattr(runner_mod, "EventSubscriber", lambda *a, **k: FakeSubscriber())
    monkeypatch.setattr(runner_mod, "ControlServer", FakeControl)
    monkeypatch.setattr(runner_mod, "AssistantService", FakeService)
    monkeypatch.setattr(runner_mod, "load_config", lambda path=None: cfg)
    monkeypatch.setattr(threading, "Event", FakeEvent)
    monkeypatch.setattr(signal, "signal", lambda *a, **k: None)
    monkeypatch.setattr(sys, "argv", ["stt-assistant"])
    rc = runner_mod.main()
    return rc, calls, made_lanes, made_services


def test_runner_main_shutdown_order_lane_before_control(tmp_path, monkeypatch):
    cfg = _enabled_triage_cfg(tmp_path)
    rc, calls, made_lanes, made_services = _run_runner_main(monkeypatch, cfg)
    assert rc == 0
    assert len(made_lanes) == 1
    assert made_lanes[0].cfg is cfg.triage
    assert made_services[0].kwargs["laya_lane"] is made_lanes[0]
    assert made_services[0].kwargs["triage"] is not None
    assert made_services[0].kwargs["brief_fn"] is None
    assert made_services[0].kwargs["brief_reset"] is None
    assert calls.index("control.start") < calls.index("service.start")
    assert calls.index("service.start") < calls.index("lane.start")
    assert calls.index("service.stop") < calls.index("lane.stop")
    assert calls.index("lane.stop") < calls.index("control.close")
    made_lanes[0].on_result(3, {"answers": {}})
    assert "triage.result" in calls
    assert "laya" not in sys.modules


def test_runner_main_disabled_builds_no_lane(monkeypatch):
    cfg = Config()
    cfg.audit.enabled = False
    cfg.llm.enabled = False
    rc, calls, made_lanes, made_services = _run_runner_main(monkeypatch, cfg)
    assert rc == 0
    assert made_lanes == []
    assert "lane.construct" not in calls
    assert "lane.start" not in calls
    assert "lane.stop" not in calls
    assert made_services[0].kwargs["triage"] is None
    assert made_services[0].kwargs["laya_lane"] is None
    assert made_services[0].kwargs["brief_fn"] is None
    assert made_services[0].kwargs["brief_reset"] is None
    assert not [t for t in threading.enumerate() if t.name.startswith("laya-")]


def _patch_llm_client(monkeypatch, calls):
    from stt_daemon.assistant import runner as runner_mod

    class FakeLlmClient:
        def __init__(self, *args, **kwargs):
            pass

        def chat(self, messages, tools=None, max_tokens=None, timeout_s=None, **kwargs):
            calls.append({"messages": messages, "kwargs": kwargs})
            return SimpleNamespace(content="hi there", finish_reason="stop")

    monkeypatch.setattr(runner_mod, "LlmClient", FakeLlmClient)


def test_runner_wires_chat_engine_and_command_slot(monkeypatch):
    """Runner builds the chat conversation from ConversationEngine with the chat
    slot, and gives the command executor the configured command slot."""
    cfg = Config()
    cfg.audit.enabled = False
    cfg.llm.enabled = True
    cfg.llm.chat_slot_id = 7
    cfg.llm.command_slot_id = 3

    calls: list[dict] = []
    _patch_llm_client(monkeypatch, calls)
    rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)

    assert rc == 0
    kwargs = made_services[0].kwargs
    engine = kwargs["chat_fn"].__self__
    assert engine.system == cfg.llm.system_prompt_chat
    assert engine._slot_id == 7
    assert kwargs["executor"].slot_id == 3
    assert kwargs["chat_fn"]("hello") == "hi there"
    assert calls[0]["kwargs"] == {"slot_id": 7}


def test_runner_wires_slotless_engine_when_slots_unset(monkeypatch):
    """Unset slot config leaves both lanes slot-free (getattr fallback)."""
    cfg = Config()
    cfg.audit.enabled = False
    cfg.llm.enabled = True
    assert cfg.llm.chat_slot_id is None
    assert cfg.llm.command_slot_id is None

    calls: list[dict] = []
    _patch_llm_client(monkeypatch, calls)
    rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)

    assert rc == 0
    kwargs = made_services[0].kwargs
    assert kwargs["executor"].slot_id is None
    assert kwargs["chat_fn"].__self__._slot_id is None
    assert kwargs["chat_fn"]("hello") == "hi there"
    assert calls[0]["kwargs"] == {}


def test_runner_wires_brief_engine_slot_when_configured(monkeypatch, tmp_path):
    """The triage brief engine sends `id_slot` only when `brief_slot_id` is set."""
    cfg = _enabled_triage_cfg(tmp_path)
    cfg.llm.enabled = True
    cfg.triage.brief_slot_id = 5

    calls: list[dict] = []
    _patch_llm_client(monkeypatch, calls)
    rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)

    assert rc == 0
    brief_fn = made_services[0].kwargs["brief_fn"]
    assert brief_fn is not None
    assert brief_fn("BRIEF-PROMPT") == "hi there"
    assert calls[0]["kwargs"] == {"slot_id": 5}


def test_runner_wires_slotless_brief_engine_when_slot_unset(monkeypatch, tmp_path):
    """An unset `brief_slot_id` leaves the brief engine slot-free (`id_slot` absent)."""
    cfg = _enabled_triage_cfg(tmp_path)
    cfg.llm.enabled = True
    assert cfg.triage.brief_slot_id is None

    calls: list[dict] = []
    _patch_llm_client(monkeypatch, calls)
    rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)

    assert rc == 0
    brief_fn = made_services[0].kwargs["brief_fn"]
    assert brief_fn is not None
    assert brief_fn("BRIEF-PROMPT") == "hi there"
    assert calls[0]["kwargs"] == {}


def test_runner_brief_file_writes_when_enabled(monkeypatch, tmp_path):
    """`[brief].write=true` makes the runner pass a live, durable writer."""
    from stt_daemon.assistant.brief_file import BriefFile

    cfg = _enabled_triage_cfg(tmp_path)
    cfg.brief.write = True

    rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)

    assert rc == 0
    brief_file = made_services[0].kwargs["brief_file"]
    assert isinstance(brief_file, BriefFile)
    assert brief_file.write_enabled is True
    session_dir = str(tmp_path / "transcripts" / "s1")
    brief_file.set_directory(session_dir)
    brief_file.write("latest brief")
    path = os.path.join(session_dir, "brief.md")
    with open(path, encoding="utf-8") as handle:
        assert handle.read() == "latest brief"


def test_runner_brief_file_disabled_writes_nothing(monkeypatch, tmp_path):
    """The default (`[brief].write=false`) writer is present but creates no file."""
    from stt_daemon.assistant.brief_file import BriefFile

    cfg = _enabled_triage_cfg(tmp_path)
    cfg.brief.write = False

    rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)

    assert rc == 0
    brief_file = made_services[0].kwargs["brief_file"]
    assert isinstance(brief_file, BriefFile)
    assert brief_file.write_enabled is False
    session_dir = str(tmp_path / "transcripts" / "s1")
    brief_file.set_directory(session_dir)
    brief_file.write("nope")
    assert not os.path.exists(os.path.join(session_dir, "brief.md"))


# ---------------------------------------------------------------------------
# End-to-end dual-engine coverage (chat + triage on ONE inference queue).
#
# These tests build the REAL `AssistantService` plus the real `build_triage`
# wiring (real `TriageAnalyzer`, real `ConversationEngine` brief engine, real
# `InferenceQueue`, real `BriefFile`) and only fake the LLM client and the bus.
# Laya results are delivered as fake payloads through the service's documented
# `_on_triage_result` seam; the queued jobs are then driven for real via
# `job.run()` (the queue's worker body).
# ---------------------------------------------------------------------------


class _Clock:
    """Deterministic clock for the integration harness."""

    def __init__(self, start: float = 1000.0) -> None:
        self.t = float(start)

    def __call__(self) -> float:
        return self.t

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


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

    def close(self) -> None:
        pass


class _EngineClient:
    """Recording LLM client: brief replies differ from chat replies.

    Accepts `slot_id` like the real client and always returns a `finish_reason`
    (the brief engine reads it). `brief_error` makes the brief call raise, which
    the real `_run_brief_update` must absorb.
    """

    def __init__(
        self, *, brief_reply: str = "BRIEF", brief_error: bool = False,
        chat_reply: str = "chat reply",
    ) -> None:
        self.calls: list[dict] = []
        self.brief_reply = brief_reply
        self.brief_error = brief_error
        self.chat_reply = chat_reply
        self.brief_system: str | None = None

    def chat(self, messages, tools=None, max_tokens=None, timeout_s=None, slot_id=None):
        system = messages[0].get("content") if messages else ""
        is_brief = self.brief_system is not None and system == self.brief_system
        self.calls.append(
            {
                "is_brief": is_brief,
                "system": system,
                "user": messages[-1].get("content") if messages else "",
                "slot_id": slot_id,
                "max_tokens": max_tokens,
                "timeout_s": timeout_s,
            }
        )
        if is_brief and self.brief_error:
            raise RuntimeError("llm down")
        content = self.brief_reply if is_brief else self.chat_reply
        return SimpleNamespace(content=content, finish_reason="stop")

    def brief_calls(self) -> list[dict]:
        return [call for call in self.calls if call["is_brief"]]

    def chat_calls(self) -> list[dict]:
        return [call for call in self.calls if not call["is_brief"]]


def _laya_payload() -> dict:
    """A realistic fake Laya result for the `urgent` noul question."""
    return {"answers": {"urgent": {"noul": 0.9, "confidence": 0.9, "answer_confidence": 0.9}}}


def _engine_service(
    tmp_path,
    client,
    *,
    chat_slot=None,
    brief_slot=None,
    brief_write: bool = False,
):
    """Build the real service + `build_triage` wiring over a fake client."""
    cfg = _enabled_triage_cfg(tmp_path)
    cfg.llm.enabled = True
    cfg.assistant.turn_gap_s = 0.5
    cfg.triage.brief_gate_enabled = False  # deterministic: dispatch every turn
    cfg.triage.brief_slot_id = brief_slot
    cfg.llm.chat_slot_id = chat_slot
    cfg.brief.write = brief_write
    client.brief_system = cfg.llm.triage_system_prompt

    analyzer, lane, brief_fn, brief_reset = build_triage(cfg, client)
    chat_engine = ConversationEngine(
        cfg.llm,
        client,
        system_prompt=cfg.llm.system_prompt_chat,
        slot_id=chat_slot,
    )
    emitted: list[dict] = []
    clock = _Clock()
    session_dir = str(tmp_path / "transcripts" / "s1")
    brief_file = BriefFile(cfg.brief.write)
    service = AssistantService(
        cfg,
        _SilentSubscriber(),
        emit=lambda type_, **fields: emitted.append({"type": type_, **fields}),
        time_fn=clock,
        triage=analyzer,
        laya_lane=lane,
        brief_fn=brief_fn,
        brief_reset=brief_reset,
        brief_file=brief_file,
        chat_fn=chat_engine.send,
        chat_reset=chat_engine.reset,
    )
    return SimpleNamespace(
        service=service,
        emitted=emitted,
        clock=clock,
        analyzer=analyzer,
        lane=lane,
        chat_engine=chat_engine,
        brief_file=brief_file,
        session_dir=session_dir,
        cfg=cfg,
        client=client,
    )


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


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


def _stages(emitted):
    return [event.get("stage") for event in _triage_events(emitted)]


def _brief_jobs(service):
    return [job for job in service._inference._queue if job.name == "brief-update"]


def _chat_jobs(service):
    return [job for job in service._inference._queue if job.name == "chat"]


def _run_queued(service, job):
    """Pop a queued job (mirroring the worker) and run it."""
    service._inference._queue.remove(job)
    job.run()


def _settle(service, clock, text, session_dir=None):
    service.handle_event(_transcript(clock, text, session_dir))
    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)


def _deliver(service, payload):
    service._on_triage_result(service._triage_generation, payload)


def test_two_engines_emit_signals_then_brief_and_chat_works(tmp_path):
    """(a) signals then content.brief; (b) chat works while triage is active;
    (c) each engine carries id_slot only when set."""
    client = _EngineClient(brief_reply="E2E BRIEF")
    h = _engine_service(tmp_path, client, chat_slot=7, brief_slot=5)

    assert h.service.toggle_chat() is True
    _settle(h.service, h.clock, "first thought")
    h.service.tick()  # dispatches the chat turn through the real queue

    _deliver(h.service, _laya_payload())
    assert _stages(h.emitted) == ["signals"]
    assert h.service.last_triage is not None
    assert "brief" not in h.service.last_triage

    chat_jobs = _chat_jobs(h.service)
    assert len(chat_jobs) == 1
    _run_queued(h.service, chat_jobs[0])  # chat works while triage is active
    assert h.service.last_triage is not None  # triage still owns its state
    assert any(
        event["type"] == "chat_message"
        and event.get("role") == "assistant"
        and event.get("text") == "chat reply"
        for event in h.emitted
    )

    brief_jobs = _brief_jobs(h.service)
    assert len(brief_jobs) == 1
    _run_queued(h.service, brief_jobs[0])

    assert _stages(h.emitted) == ["signals", "content"]
    contents = [event for event in _triage_events(h.emitted) if event.get("stage") == "content"]
    assert contents[0]["triage"]["brief"] == "E2E BRIEF"
    assert h.service.last_triage["brief"] == "E2E BRIEF"

    assert [call["slot_id"] for call in client.chat_calls()] == [7]
    assert [call["slot_id"] for call in client.brief_calls()] == [5]
    assert client.brief_calls()[0]["user"].startswith("CURRENT BRIEF:\n(none yet)")


def test_slotless_engines_omit_id_slot(tmp_path):
    """(c) unset slots leave `id_slot` off both engines' calls."""
    client = _EngineClient(brief_reply="BRIEF")
    h = _engine_service(tmp_path, client, chat_slot=None, brief_slot=None)

    assert h.service.toggle_chat() is True
    _settle(h.service, h.clock, "hello")
    h.service.tick()
    _deliver(h.service, _laya_payload())
    _run_queued(h.service, _chat_jobs(h.service)[0])
    _run_queued(h.service, _brief_jobs(h.service)[0])

    assert [call["slot_id"] for call in client.chat_calls()] == [None]
    assert [call["slot_id"] for call in client.brief_calls()] == [None]


def test_refused_brief_keeps_deltas_and_next_dispatch_delivers(tmp_path):
    """(d) a full queue refuses the brief without losing pending deltas, and the
    next dispatch delivers them all."""
    client = _EngineClient(brief_reply="BRIEF")
    h = _engine_service(tmp_path, client)
    service = h.service
    service._inference = InferenceQueue(size=2)
    service._inference.submit(InferenceJob(name="blocker", run=lambda: None, evictable=False))

    _settle(service, h.clock, "first thought")
    _deliver(service, _laya_payload())

    # Refused: no brief job admitted, but the delta is buffered and the flag clear.
    assert _brief_jobs(service) == []
    assert service._pending_brief == ["Unknown: first thought"]
    assert service._brief_outstanding is False
    assert _stages(h.emitted) == ["signals"]

    # Free room; the next settled turn re-dispatches and consumes both deltas.
    service._inference._queue.clear()
    _settle(service, h.clock, "second thought")
    _deliver(service, _laya_payload())
    jobs = _brief_jobs(service)
    assert len(jobs) == 1
    assert service._pending_brief == ["Unknown: first thought", "Unknown: second thought"]

    _run_queued(service, jobs[0])

    assert _stages(h.emitted) == ["signals", "signals", "content"]
    prompt = client.brief_calls()[-1]["user"]
    assert "- Unknown: first thought" in prompt
    assert "- Unknown: second thought" in prompt
    assert service._pending_brief == []


def test_empty_brief_requeues_deltas_and_clears_flag(tmp_path):
    """(e) a `""` brief re-queues the deltas and clears the outstanding flag."""
    client = _EngineClient(brief_reply="")
    h = _engine_service(tmp_path, client)
    service = h.service

    _settle(service, h.clock, "first thought")
    _deliver(service, _laya_payload())
    _run_queued(service, _brief_jobs(service)[0])

    assert _stages(h.emitted) == ["signals"]  # no content emitted
    assert service._pending_brief == ["Unknown: first thought"]
    assert service._brief_outstanding is False
    assert _brief_jobs(service) == []  # requeued, not flushed


def test_raising_brief_requeues_deltas_and_clears_flag(tmp_path):
    """(e) a raising brief leaves no stranded flag and re-buffers the deltas."""
    client = _EngineClient(brief_error=True)
    h = _engine_service(tmp_path, client)
    service = h.service

    _settle(service, h.clock, "first thought")
    _deliver(service, _laya_payload())
    _run_queued(service, _brief_jobs(service)[0])

    assert _stages(h.emitted) == ["signals"]
    assert service._pending_brief == ["Unknown: first thought"]
    assert service._brief_outstanding is False
    assert _brief_jobs(service) == []


def test_none_laya_payload_does_not_dispatch(tmp_path):
    """(f) a None Laya payload stores nothing and never dispatches the brief."""
    client = _EngineClient()
    h = _engine_service(tmp_path, client)
    service = h.service

    _settle(service, h.clock, "first thought")
    _deliver(service, None)

    assert _triage_events(h.emitted) == []
    assert service.last_triage is None
    assert _brief_jobs(service) == []
    assert service._pending_brief == ["Unknown: first thought"]
    assert client.calls == []


def test_clear_resets_both_engines_and_stale_job_cannot_clobber(tmp_path):
    """(g) clear resets chat + triage, suppresses stale emits, and a stale job
    cannot clear a newer dispatch's flag."""
    client = _EngineClient()
    h = _engine_service(tmp_path, client)
    service = h.service

    assert service.toggle_chat() is True
    _settle(service, h.clock, "hello")
    service.tick()
    _run_queued(service, _chat_jobs(service)[0])
    assert h.chat_engine.history != []

    _settle(service, h.clock, "first thought")
    _deliver(service, _laya_payload())
    stale_job = _brief_jobs(service)[0]
    assert service._brief_outstanding is True

    service.clear()

    # Both engines reset.
    assert h.chat_engine.history == []
    assert service.chat_history == []
    assert service._triage_generation == 1
    assert service.last_triage is None
    assert service.triage_history == []
    assert service._pending_brief == []
    assert service._brief_outstanding is False
    assert service._last_brief_signals is None
    assert _brief_jobs(service) == []
    assert "cleared" in _stages(h.emitted)
    assert {"type": "cleared"} in h.emitted

    h.emitted.clear()
    stale_job.run()  # clear() already dropped it from the queue
    assert _triage_events(h.emitted) == []
    assert service._pending_brief == []
    assert service.last_triage is None

    # A newer dispatch after clear owns the flag; the stale job cannot clear it.
    _settle(service, h.clock, "fresh")
    _deliver(service, _laya_payload())
    newer_job = _brief_jobs(service)[0]
    newer_token = service._brief_token
    pending_after = list(service._pending_brief)
    triage_after = service.last_triage
    assert service._brief_outstanding is True

    h.emitted.clear()
    stale_job.run()
    assert _triage_events(h.emitted) == []
    assert service._brief_outstanding is True
    assert service._brief_token == newer_token
    assert service._pending_brief == pending_after
    assert service.last_triage is triage_after

    _run_queued(service, newer_job)
    assert _stages(h.emitted) == ["content"]
    assert service._brief_outstanding is False


def test_durable_brief_file_holds_latest_and_clears(tmp_path):
    """(h) with `[brief].write=true` the file holds the exact latest brief and
    is empty after `clear`."""
    client = _EngineClient(brief_reply="FIRST BRIEF")
    h = _engine_service(tmp_path, client, brief_write=True)
    service = h.service
    path = os.path.join(h.session_dir, "brief.md")

    _settle(service, h.clock, "first thought", h.session_dir)
    _deliver(service, _laya_payload())
    _run_queued(service, _brief_jobs(service)[0])
    with open(path, encoding="utf-8") as handle:
        assert handle.read() == "FIRST BRIEF"

    client.brief_reply = "LATEST BRIEF"
    _settle(service, h.clock, "second thought", h.session_dir)
    _deliver(service, _laya_payload())
    _run_queued(service, _brief_jobs(service)[0])
    with open(path, encoding="utf-8") as handle:
        assert handle.read() == "LATEST BRIEF"
    assert service.last_triage["brief"] == "LATEST BRIEF"

    service.clear()
    assert os.path.exists(path)
    with open(path, encoding="utf-8") as handle:
        assert handle.read() == ""


def test_disabled_brief_file_absent_but_panel_updates(tmp_path):
    """(h) with `[brief].write=false` no file appears while the panel brief still
    updates."""
    client = _EngineClient(brief_reply="PANEL BRIEF")
    h = _engine_service(tmp_path, client, brief_write=False)
    service = h.service
    path = os.path.join(h.session_dir, "brief.md")

    _settle(service, h.clock, "first thought", h.session_dir)
    _deliver(service, _laya_payload())
    _run_queued(service, _brief_jobs(service)[0])

    assert h.brief_file.write_enabled is False
    assert not os.path.exists(path)
    assert service.last_triage["brief"] == "PANEL BRIEF"