"""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_pinned_brief_prompt(self, current_brief, deltas, signals): self.prompt_calls.append( (tuple(deltas), dict(signals) if isinstance(signals, dict) else signals) ) body = "\n".join("- " + d for d in deltas) return ( f"CURRENT BRIEF:\n{current_brief or '(none yet)'}\n\nNEW UTTERANCES:\n{body}" ) def extract_bullets(self, brief): return _real_analyzer.extract_bullets(brief) def restore_missing_bullets(self, new_brief, prior_brief): return _real_analyzer.restore_missing_bullets(new_brief, prior_brief) 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 _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=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 _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_retention_guard_reappends_fact_dropped_by_model(): """A prior bullet the model omitted is deterministically restored.""" prior = ( "Summary: rolling\n\nPrior Details:\n" "- The budget was $305 million.\n" "- Sense could find prey.\n" ) new = "Summary: newer\n\nPrior Details:\n- Sense could find prey.\n" service, emitted, _, _, _ = _triage_service(brief_fn=lambda prompt: new) service.last_triage = {**_escalated_payload(), "brief": prior} service._pending_brief = ["Unknown: more"] 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 "- Sense could find prey." in brief assert "- The budget was $305 million." in brief assert brief.count("- Sense could find prey.") == 1 def test_retention_guard_leaves_complete_brief_unchanged(): """A brief that already carries every prior bullet is stored verbatim.""" prior = "Prior Details:\n- fact one\n- fact two\n" new = "Summary: newer\n\nPrior Details:\n- fact one\n- fact two\n" service, _, _, _, _ = _triage_service(brief_fn=lambda prompt: new) service.last_triage = {**_escalated_payload(), "brief": prior} service._pending_brief = ["Unknown: more"] service._brief_outstanding = True service._run_brief_update(0, _escalated_payload(), service._now(), service._brief_token) assert service.last_triage is not None assert service.last_triage["brief"] == new def test_pinned_prompt_carries_prior_brief_and_new_utterances(): """The brief request hands the model the pinned current brief + new deltas.""" seen: list[str] = [] service, _, _, _, _ = _triage_service( brief_fn=lambda prompt: seen.append(prompt) or "UPDATED" ) service.last_triage = {**_escalated_payload(), "brief": "PRIOR BRIEF TEXT"} 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 seen[0].startswith("CURRENT BRIEF:\nPRIOR BRIEF TEXT") 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): if not appended: appended.append(1) service.handle_event(_transcript(clock, "during call")) return "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): """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" 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 BRIEF:\n(none yet)\n\nNEW 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): # 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 "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): 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" 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 BRIEF:\nBRIEF\n\nNEW 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"]