"""Tests for the triage analyzer (todos 3, 9, 10, 11). Pure, hardware-free.""" import json import logging from pathlib import Path from types import SimpleNamespace import pytest from stt_daemon import events from stt_daemon.assistant import triage from stt_daemon.assistant.service import AssistantService from stt_daemon.assistant.triage import FactLedger, TriageAnalyzer, load_questions from stt_daemon.config import Config from stt_daemon.tui.state import ViewState EXAMPLE_PATH = str(Path(__file__).resolve().parent.parent / "questions.example.json") def _cfg(path=EXAMPLE_PATH, **overrides): base = { "questions_file": path, "state_char_budget": 1200, "confidence_field": "answer_confidence", "escalate_threshold": 0.6, "choice_min_confidence": 0.85, } base.update(overrides) return SimpleNamespace(**base) def _write_questions(tmp_path, questions): path = tmp_path / "questions.json" path.write_text(json.dumps(questions), encoding="utf-8") return str(path) def _analyzer(tmp_path, questions=None, **overrides): if questions is None: return TriageAnalyzer(_cfg(**overrides)) return TriageAnalyzer(_cfg(_write_questions(tmp_path, questions), **overrides)) # --- todo 3: load_questions ------------------------------------------------- def test_load_example_order_preserved(): raw = json.loads(Path(EXAMPLE_PATH).read_text(encoding="utf-8")) questions = load_questions(EXAMPLE_PATH) assert questions is not None assert questions assert list(questions.keys()) == list(raw.keys()) for definition in questions.values(): assert definition["type"] in {"choice", "score", "noul"} assert isinstance(definition["instructions"], str) assert definition["instructions"].strip() def test_example_questions_match_design(): questions = load_questions(EXAMPLE_PATH) assert questions is not None assert list(questions.keys()) == [ "substantive_update", "opportunity", "blocker", "decision_status", "action_item", "responsibility_clarity", "deadline", "open_question", ] for qid, definition in questions.items(): assert isinstance(definition["instructions"], str) assert definition["instructions"].strip() if definition["type"] == "choice": criteria = definition["criteria"] options = list(criteria.keys()) if isinstance(criteria, dict) else list(criteria) assert "other" in options assert len(options) <= 7 if definition["type"] == "noul": assert isinstance(definition.get("instructions"), str) assert definition["instructions"].strip() def test_load_missing_file_returns_none_with_warning(caplog): with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"): assert load_questions("/nonexistent/questions.json") is None assert caplog.records, "expected a warning for the missing file" def test_load_bad_json_returns_none_with_warning(tmp_path, caplog): path = tmp_path / "questions.json" path.write_text("{not valid json", encoding="utf-8") with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"): assert load_questions(str(path)) is None assert caplog.records @pytest.mark.parametrize( "bad_entry", [ {"type": "mystery", "instructions": "do something"}, # unknown type {"type": "score", "instructions": "rate it", "criteria": {"0": "low"}}, # score dict { # noul criteria with a bad key "type": "noul", "instructions": "holds?", "criteria": {"true": "yes", "maybe": "hmm"}, }, "just a string", # non-object entry {"type": "choice", "instructions": "pick one", "criteria": []}, # empty choice {"type": "choice", "instructions": "pick one", "criteria": {}}, # empty choice dict {"type": "choice", "criteria": ["a", "b"]}, # missing instructions {"type": "score", "instructions": "rate it", "criteria": ["a", None]}, # None level {"type": "noul", "instructions": "holds?", "criteria": ["true"]}, # noul list ], ) def test_load_malformed_returns_none_with_warning(tmp_path, caplog, bad_entry): path = _write_questions( tmp_path, {"good": {"type": "noul", "instructions": "ok?"}, "bad": bad_entry} ) with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"): assert load_questions(str(path)) is None assert caplog.records # --- todo 9: normalize ------------------------------------------------------ def _score_questions(): return { "severity": { "type": "score", "instructions": "How severe is the issue?", "criteria": ["none", "low", "high", "critical"], }, "category": { "type": "choice", "instructions": "What is this about?", "criteria": {"support": "help needed", "sales": "buying"}, }, "churn_risk": {"type": "noul", "instructions": "User may leave."}, } def test_normalize_choice_exact(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) payload = { "model": "multilingual", "answers": { "category": { "type": "choice", "choice": "support", "confidence": 0.7, "answer_confidence": 0.4, } }, "usage": {"input_tokens": 123, "output_tokens": 0}, } result = analyzer.normalize(payload) assert result["order"] == ["category"] assert result["items"]["category"] == { "kind": "choice", "choice": "support", "confidence": 0.7, "answer_confidence": 0.4, "gate": 0.4, "flagged": True, } assert result["input_tokens"] == 123 assert result["escalated"] is True def test_normalize_score_with_legend(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) payload = { "answers": { "severity": { "type": "score", "score": 2.6, "legend": {"0": "none", "1": "low", "2": "high", "3": "critical"}, "confidence": 0.5, "answer_confidence": 0.6, } }, "usage": {"input_tokens": 7, "output_tokens": 0}, } item = analyzer.normalize(payload)["items"]["severity"] assert item["kind"] == "score" assert item["expected"] == 2.6 assert item["level"] == 3 assert item["label"] == "critical" assert item["confidence"] == 0.5 assert item["answer_confidence"] == 0.6 assert item["gate"] == 0.6 assert item["flagged"] is True def test_normalize_score_legend_fallback(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) payload = { "answers": { "severity": {"type": "score", "score": 1.2, "confidence": 0.9, "answer_confidence": 0.9} } } item = analyzer.normalize(payload)["items"]["severity"] assert item["level"] == 1 assert item["label"] == "low" # falls back to criteria[level] assert item["flagged"] is False # 1.2 < (4 - 1) * 0.5 def test_normalize_noul_and_bare_payload(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) bare = { "churn_risk": {"type": "noul", "noul": 0.9, "confidence": 0.9, "answer_confidence": 0.85} } result = analyzer.normalize(bare) assert result["order"] == ["churn_risk"] assert result["items"]["churn_risk"] == { "kind": "noul", "p": 0.9, "confidence": 0.9, "answer_confidence": 0.85, "gate": 0.85, "flagged": True, } assert result["input_tokens"] == 0 def test_normalize_skips_unknown_qid_and_malformed(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) payload = { "answers": { "ghost": {"type": "noul", "noul": 0.99, "confidence": 0.9, "answer_confidence": 0.9}, "category": {"type": "choice", "choice": 42}, # malformed: no confidences "churn_risk": "nonsense", # malformed: not a dict }, "usage": {"input_tokens": 3, "output_tokens": 0}, } result = analyzer.normalize(payload) assert result["order"] == [] assert result["items"] == {} assert result["escalated"] is False def test_normalize_ignores_side_channel_fields(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) payload = { "answers": { "category": { "type": "choice", "choice": "sales", "confidence": 0.95, "answer_confidence": 0.9, "action": {"act" + "_probability": 0.99}, } }, "usage": {"input_tokens": 1, "output_tokens": 0}, } result = analyzer.normalize(payload) assert result["items"]["category"]["flagged"] is False assert "act" + "_probability" not in json.dumps(result) source = Path(triage.__file__).read_text(encoding="utf-8") assert "act" + "_probability" not in source def test_normalize_gating_field_selectable(tmp_path): analyzer = _analyzer(tmp_path, _score_questions(), confidence_field="confidence") payload = { "answers": { "category": { "type": "choice", "choice": "sales", "confidence": 0.5, "answer_confidence": 0.99, } } } item = analyzer.normalize(payload)["items"]["category"] assert item["gate"] == 0.5 # gates on confidence, keeps both assert item["answer_confidence"] == 0.99 assert item["flagged"] is True # --- todo 9: condensation --------------------------------------------------- def test_condense_keeps_newest_chronological_no_numerals(tmp_path): analyzer = _analyzer(tmp_path, _score_questions(), state_char_budget=40) window = ["alpha first", "beta second", "gamma third newest here"] state = analyzer.build_state(window) assert "gamma third newest here" in state assert "alpha first" not in state # oldest dropped under the budget lines = state.splitlines() assert all(line.startswith("- ") for line in lines) assert not any(ch.isdigit() for ch in state) # no numerals injected # chronological restore: kept lines stay in original order kept = [u for u in window if u in state] assert [line[2:] for line in lines] == kept def test_condense_empty_window(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.build_state([]) == "" # --- todo 10: escalation + extraction --------------------------------------- def _signals(choice_gate=0.9, noul_p=0.1, score_expected=0.2): return { "order": ["category", "churn_risk", "severity"], "items": { "category": { "kind": "choice", "choice": "sales", "confidence": choice_gate, "answer_confidence": choice_gate, "gate": choice_gate, "flagged": choice_gate < 0.85, }, "churn_risk": { "kind": "noul", "p": noul_p, "confidence": noul_p, "answer_confidence": noul_p, "gate": noul_p, "flagged": noul_p >= 0.6, }, "severity": { "kind": "score", "expected": score_expected, "level": 0, "label": "none", "confidence": 0.9, "answer_confidence": 0.9, "gate": 0.9, "flagged": score_expected >= 1.5, }, }, "escalated": False, "input_tokens": 0, } def test_escalation_arms_fire_independently(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.should_escalate(_signals(choice_gate=0.5)) is True # choice arm assert analyzer.should_escalate(_signals(noul_p=0.8)) is True # noul arm assert analyzer.should_escalate(_signals(score_expected=2.0)) is True # score arm assert analyzer.should_escalate(_signals()) is False def test_normalize_escalated_matches_should_escalate(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) payload = { "answers": { "category": { "type": "choice", "choice": "sales", "confidence": 0.9, "answer_confidence": 0.9, }, "churn_risk": { "type": "noul", "noul": 0.1, "confidence": 0.9, "answer_confidence": 0.9, }, "severity": { "type": "score", "score": 0.1, "confidence": 0.9, "answer_confidence": 0.9, }, } } result = analyzer.normalize(payload) assert result["escalated"] == analyzer.should_escalate(result) is False # --- todo 5: brief prompt / parse / gate / finalize -------------------------- def test_build_brief_prompt_order_and_focus(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prompt = analyzer.build_brief_prompt( ["Alice: let's ship Friday", "Bob: agreed"], _signals(noul_p=0.8) ) assert prompt.startswith("New utterances:\n- Alice: let's ship Friday\n- Bob: agreed") assert "Focus:" in prompt assert "churn_risk" in prompt # signal named in words assert prompt.index("New utterances:") < prompt.index("Focus:") assert prompt.index("Focus:") < prompt.index("Return only the updated brief") # no JSON extraction keys or raw dict shapes leak into the brief prompt for key in ("summary", "opportunities", "action_items", "risks", "doubts"): assert key not in prompt assert "{" not in prompt and "}" not in prompt def test_build_brief_prompt_no_focus_without_signals(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) for signals in (None, {}, {"order": [], "items": {}}): prompt = analyzer.build_brief_prompt(["solo utterance"], signals) assert "solo utterance" in prompt assert "Focus:" not in prompt assert prompt.endswith("Return only the updated brief as plain text.") def test_build_ledger_prompt_inlines_ledger_narrative_and_deltas(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prompt = analyzer.build_ledger_prompt( "Summary: shipped\n\nOpportunities: expand", ["fact one", "fact two"], ["Alice: let's ship Friday"], _signals(noul_p=0.8), ) assert "- fact one" in prompt and "- fact two" in prompt assert "CURRENT LEDGER" in prompt assert "Summary: shipped" in prompt assert "NEW UTTERANCES:\n- Alice: let's ship Friday" in prompt assert prompt.index("CURRENT LEDGER") < prompt.index("CURRENT BRIEF:") assert prompt.index("CURRENT BRIEF:") < prompt.index("NEW UTTERANCES:") assert prompt.index("NEW UTTERANCES:") < prompt.index("Focus:") assert "churn_risk" in prompt assert "NEW FACTS:" in prompt and "BRIEF:" in prompt def test_build_ledger_prompt_empty_ledger_and_narrative(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prompt = analyzer.build_ledger_prompt("", [], ["solo"], None) assert "CURRENT LEDGER" in prompt and "(none yet)" in prompt assert "CURRENT BRIEF:\n(none yet)" in prompt assert "NEW UTTERANCES:\n- solo" in prompt assert "Focus:" not in prompt def test_extract_bullets_strips_new_marker_and_normalizes(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) brief = ( "Summary: things\n" "- top level bullet\n" " - indented child\n" "- **New:** tagged fact\n" "not a bullet\n" "\n" ) assert analyzer.extract_bullets(brief) == [ "top level bullet", "indented child", "tagged fact", ] def test_restore_missing_bullets_reappends_dropped_fact(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prior = ( "Summary: old\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" restored = analyzer.restore_missing_bullets(new, prior) assert "- Sense could find prey." in restored assert "- The budget was $305 million." in restored assert restored.count("- Sense could find prey.") == 1 def test_restore_missing_bullets_matches_reworded_bullet(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prior = "Prior Details:\n- **New:** Sense could find prey.\n" new = "Prior Details:\n- Sense could find prey.\n" restored = analyzer.restore_missing_bullets(new, prior) assert restored == new def test_restore_missing_bullets_restores_reworded_as_new_fact(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prior = "Prior Details:\n- The budget was $305 million.\n" new = "Prior Details:\n- Budget: $305 million (nearly Spider-Man 2).\n" restored = analyzer.restore_missing_bullets(new, prior) assert restored.count("$305 million") == 2 assert restored.rstrip().endswith("- The budget was $305 million.") def test_restore_missing_bullets_noop_without_prior_or_new(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) new = "Summary: only\nPrior Details:\n- keep me\n" assert analyzer.restore_missing_bullets(new, "") == new assert analyzer.restore_missing_bullets(new, None) == new assert analyzer.restore_missing_bullets("", new) == "" assert analyzer.restore_missing_bullets(None, new) is None def test_restore_missing_bullets_appends_section_when_absent(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) prior = "- orphan fact\n" new = "Summary: no prior details header\n" restored = analyzer.restore_missing_bullets(new, prior) assert "- orphan fact" in restored assert "Prior Details:" in restored def test_parse_brief_strips_fences_and_rejects_empty(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.parse_brief("```\nhello brief\n```") == "hello brief" assert analyzer.parse_brief("```markdown\nhello brief\n```") == "hello brief" assert analyzer.parse_brief("plain brief") == "plain brief" assert analyzer.parse_brief(" padded ") == "padded" assert analyzer.parse_brief("") is None assert analyzer.parse_brief(" ") is None assert analyzer.parse_brief(None) is None assert analyzer.parse_brief(42) is None assert analyzer.parse_brief("```") is None def test_parse_brief_unescapes_literal_backslash_n(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.parse_brief("## Title\\n\\nBody line") == "## Title\n\nBody line" assert analyzer.parse_brief("a\\tb") == "a\tb" assert analyzer.parse_brief("line1\nline2\\nline3") == "line1\nline2\\nline3" def test_parse_brief_truncates_not_rejects(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) parsed = analyzer.parse_brief("x" * 5000) assert parsed is not None assert parsed == "x" * 4000 assert len(parsed) == 4000 wrapped = "[update_brief(updated_brief='" + "x" * 5000 + "')]" assert analyzer.parse_brief(wrapped) == "x" * 4000 def test_parse_brief_honors_configured_cap(tmp_path): analyzer = _analyzer(tmp_path, _score_questions(), brief_max_chars=10) assert analyzer.parse_brief("x" * 50) == "x" * 10 def test_parse_brief_unwraps_single_quoted_tool_call_with_apostrophes(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) observed = "<|tool_call_start|>[update_brief(updated_brief='the company's game failed')]" assert analyzer.parse_brief(observed) == "the company's game failed" def test_parse_brief_unwraps_double_quoted_tool_call(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) wrapped = '[update_brief(updated_brief="the company\'s plan")]' assert analyzer.parse_brief(wrapped) == "the company's plan" def test_parse_brief_unwraps_trailing_end_marker(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) wrapped = "[update_brief(updated_brief='ship friday')]<|tool_call_end|>" assert analyzer.parse_brief(wrapped) == "ship friday" def test_parse_brief_unwraps_fenced_and_wrapped(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) wrapped = "```\n<|tool_call_start|>[update_brief(updated_brief='fenced payload')]\n```" assert analyzer.parse_brief(wrapped) == "fenced payload" def test_parse_brief_does_not_mangle_clean_markdown(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) clean = "Here's the [brief]: it's the company's plan - don't mangle [it]." assert analyzer.parse_brief(clean) == clean prose = "We should update_brief the notes, here's [why]." assert analyzer.parse_brief(prose) == prose def test_parse_brief_empty_tool_call_payload_is_none(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.parse_brief("[update_brief(updated_brief='')]") is None def test_parse_brief_unwraps_multi_arg_tool_call(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) wrapped = ( "[update_brief(updated_brief='## Opportunities\n- ship Friday', " "previous_brief='## old\n- y', mode='read')]" ) assert analyzer.parse_brief(wrapped) == "## Opportunities\n- ship Friday" def test_parse_brief_unwraps_multi_arg_with_markers(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) wrapped = ( "<|tool_call_start|>[update_brief(updated_brief='## Opportunities\n- ship Friday', " "previous_brief='## old\n- y', mode='read')]<|tool_call_end|>" ) assert analyzer.parse_brief(wrapped) == "## Opportunities\n- ship Friday" def test_parse_brief_unwraps_multi_arg_apostrophes_preserved(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) wrapped = ( "[update_brief(updated_brief='the company's game failed', " "previous_brief='## old', mode='read')]" ) assert analyzer.parse_brief(wrapped) == "the company's game failed" def test_parse_brief_unwraps_positional_tool_call(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.parse_brief("[update_brief('payload here', mode='read')]") == "payload here" def test_parse_brief_rejects_polluted_tool_call(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) polluted = "[update_brief(updated_brief='oops, previous_brief=', mode='read')]" assert analyzer.parse_brief(polluted) is None no_arg = "<|tool_call_start|>[update_brief(previous_brief=old, mode=read)]" assert analyzer.parse_brief(no_arg) is None def test_parse_brief_clean_mode_text_not_mangled(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) clean = "Set mode=read, then read [the brief] - it's fine (really)." assert analyzer.parse_brief(clean) == clean def test_parse_brief_rejects_marker_with_artifacts_but_no_call(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.parse_brief("<|tool_call_start|>previous_brief='x', mode='read'") is None assert analyzer.parse_brief("<|tool_call_start|>mode='read'") is None def _brief_signals(p=None): item = {"kind": "noul"} if p is not None: item["p"] = p return {"order": ["substantive_update"], "items": {"substantive_update": item}} def test_should_update_brief_gate(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.should_update_brief(_brief_signals(0.9)) is True assert analyzer.should_update_brief(_brief_signals(0.5)) is True # boundary assert analyzer.should_update_brief(_brief_signals(0.2)) is False # absent update / no numeric p / not a dict -> fail open assert analyzer.should_update_brief({"order": [], "items": {}}) is True assert analyzer.should_update_brief(_brief_signals(None)) is True assert analyzer.should_update_brief(None) is True assert analyzer.should_update_brief({"items": {"substantive_update": "nope"}}) is True def test_should_update_brief_threshold_configured(tmp_path): analyzer = _analyzer(tmp_path, _score_questions(), brief_gate_threshold=0.8) assert analyzer.should_update_brief(_brief_signals(0.6)) is False assert analyzer.should_update_brief(_brief_signals(0.8)) is True def test_finalize_brief_length_truncates_not_rejects(tmp_path, caplog): analyzer = _analyzer(tmp_path, _score_questions()) with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"): result = analyzer.finalize_brief("y" * 5000, "length") assert result is not None assert result == "y" * 4000 assert caplog.records # warned about the length cutoff but still accepted def test_finalize_brief_stop_and_empty(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.finalize_brief("a brief", "stop") == "a brief" assert analyzer.finalize_brief("", "stop") is None assert analyzer.finalize_brief("```\nfenced\n```", "stop") == "fenced" assert analyzer.finalize_brief(None, "length") is None # --- todo 11: cached state() ------------------------------------------------ def test_state_initial_loading_with_good_questions(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.state() == ("loading", "") def test_state_initial_no_questions_with_bad_path(tmp_path): missing = str(tmp_path / "missing.json") analyzer = TriageAnalyzer(_cfg(missing)) state, reason = analyzer.state() assert state == "no_questions" assert missing in reason def test_state_vocabulary_has_no_extra_member(): assert set(triage.TRIAGE_REASONS) == { "ok", "no_questions", "loading", "laya_missing", "disabled", "error", } assert "service" + "_down" not in triage.TRIAGE_REASONS def test_refresh_transitions(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) analyzer.refresh("ok") assert analyzer.state() == ("ok", "") analyzer.refresh("laya_missing", "run: uv sync --extra triage") assert analyzer.state() == ("laya_missing", "run: uv sync --extra triage") analyzer.refresh("error", "boom") assert analyzer.state() == ("error", "boom") def test_refresh_unknown_state_ignored(tmp_path, caplog): analyzer = _analyzer(tmp_path, _score_questions()) analyzer.refresh("ok") with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"): analyzer.refresh("bogus_state") assert analyzer.state() == ("ok", "") assert caplog.records def test_no_questions_state_cannot_be_overridden(tmp_path): analyzer = TriageAnalyzer(_cfg(str(tmp_path / "missing.json"))) assert analyzer.state()[0] == "no_questions" analyzer.refresh("ok") analyzer.refresh("error", "boom") assert analyzer.state()[0] == "no_questions" def test_state_does_not_touch_lane(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) class StrictLane: def __init__(self): self.calls = [] def __getattr__(self, name): def record(*args, **kwargs): self.calls.append(name) return record lane = StrictLane() for _ in range(3): analyzer.state() analyzer.refresh("ok") analyzer.state() assert lane.calls == [] # --- F3: end-to-end triage flow on fakes ------------------------------------ EXTRACT_JSON = ( '{"summary":"s","opportunities":[],"doubts":[],"action_items":[],"risks":[]}' ) class _Clock: """Injected clock: no sleeps, fully deterministic.""" 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 class _FakeLane: """Duck-typed LayaLane: canned-ready, records submits, counts drops.""" def __init__(self): self.submits: list[tuple] = [] self.drops = 0 def ready(self) -> bool: return True def state(self): return ("ok", "") def submit(self, state, questions, lang, generation): self.submits.append((state, questions, lang, generation)) return True def drop_pending(self): self.drops += 1 return 1 def _answers(noul_p=0.1): """A full laya-shaped answers payload for the example questions. `noul_p` drives every noul arm, including the `substantive_update` gate: below `brief_gate_threshold` (0.5) the turn is gated OUT (no brief job), at or above it the turn gates IN. """ answers = { "decision_status": { "type": "choice", "choice": "made", "confidence": 0.9, "answer_confidence": 0.9, }, "responsibility_clarity": { "type": "choice", "choice": "explicit", "confidence": 0.9, "answer_confidence": 0.9, }, } for qid in ( "substantive_update", "opportunity", "blocker", "action_item", "deadline", "open_question", ): answers[qid] = { "type": "noul", "noul": noul_p, "confidence": 0.9, "answer_confidence": 0.9, } return {"model": "multilingual", "answers": answers, "usage": {"input_tokens": 42}} def test_end_to_end_triage_flow(): """signals -> content -> cleared across service, analyzer, lane, and ViewState. The lane result is wired by calling `service._on_triage_result(gen, payload)` directly (the runner does exactly this in production); the brief job is drained with `job.run()` so no worker thread is started. The single-outstanding guard plus the shared `_pending_brief` collapse settled turns into ONE job consuming every buffered delta, and the `substantive_update` gate can reduce that to ZERO jobs. """ cfg = Config() cfg.assistant.turn_gap_s = 0.5 cfg.triage.enabled = True cfg.triage.questions_file = EXAMPLE_PATH clock = _Clock(1000.0) lane = _FakeLane() emitted: list[dict] = [] prompts: list[str] = [] def emit(type_, **fields): emitted.append({"type": type_, **fields}) analyzer = TriageAnalyzer(cfg.triage) def brief_fn(prompt, on_io=None, finalize=None): prompts.append(prompt) return "NEW FACTS:\nnone\n\nBRIEF:\nSummary: brief\n" service = AssistantService( cfg, _FakeSubscriber(), emit=emit, time_fn=clock, chat_fn=lambda text: "reply", triage=analyzer, laya_lane=lane, brief_fn=brief_fn, ) def triage_events(): return [e for e in emitted if e["type"] == "triage"] def brief_jobs(): return [j for j in service._inference._queue if j.name == "brief-update"] # 1. settled turn -> exactly one lane submit, lang + generation recorded assert service.toggle_chat() is True service.handle_event( {"type": events.TRANSCRIPT, "text": "hola, necesito ayuda", "lang": "es"} ) clock.advance(1.0) service.tick() assert len(lane.submits) == 1 state_text, questions, lang, gen1 = lane.submits[0] assert "hola, necesito ayuda" in state_text assert questions is analyzer.questions assert lang == "es" assert gen1 == 0 # 2. low-p turn -> signals only: the gate (p=0.1 < 0.5) dispatches ZERO jobs service._on_triage_result(gen1, _answers(noul_p=0.1)) assert [e["stage"] for e in triage_events()] == ["signals"] assert triage_events()[0]["triage"]["escalated"] is False assert service.last_triage is not None assert service.last_triage["escalated"] is False assert brief_jobs() == [] assert prompts == [] # 3. high-p turn -> second submit, then ONE brief job (the only one outstanding) service.handle_event( {"type": events.TRANSCRIPT, "text": "quiero cancelar mi suscripcion", "lang": "es"} ) clock.advance(1.0) service.tick() assert len(lane.submits) == 2 gen2 = lane.submits[1][3] service._on_triage_result(gen2, _answers(noul_p=0.9)) assert [e["stage"] for e in triage_events()] == ["signals", "signals"] assert triage_events()[1]["triage"]["escalated"] is True jobs = brief_jobs() assert len(jobs) == 1 # 4. run the brief job directly (deterministic, no worker thread): it consumes # both buffered deltas and emits ONE content with the brief merged over signals jobs[0].run() assert len(prompts) == 1 assert "hola, necesito ayuda" in prompts[0] assert "quiero cancelar mi suscripcion" in prompts[0] content_events = [e for e in triage_events() if e["stage"] == "content"] assert len(content_events) == 1 merged = content_events[0]["triage"] assert merged["brief"] == "Summary: brief" assert merged["escalated"] is True # signals preserved under the brief assert merged["order"] == triage_events()[1]["triage"]["order"] # 5. drive the emitted stream through a fresh ViewState view = ViewState() for event in emitted: view.ingest(event) analyzer.refresh("ok") # the lane callback does this in production view.ingest({"type": "status", **service.status()}) assert view.triage is not None assert view.triage["brief"] == "Summary: brief" lines = view.triage_lines() assert lines # 6. clear -> cleared event empties the panel and bumps the epoch baseline_epoch = view.triage_epoch already = len(emitted) service.clear() for event in emitted[already:]: view.ingest(event) assert view.triage is None assert view.triage_epoch > baseline_epoch # a stale result after clear must not re-populate the panel service._on_triage_result(gen2, _answers(noul_p=0.9)) assert [e["stage"] for e in triage_events()] == [ "signals", "signals", "content", "cleared", ] assert view.triage is None assert lane.drops == 1 # 7. disabled control: no triage/lane -> status disabled, clear emits nothing cfg2 = Config() clock2 = _Clock(2000.0) emitted2: list[dict] = [] def emit2(type_, **fields): emitted2.append({"type": type_, **fields}) plain = AssistantService(cfg2, _FakeSubscriber(), emit=emit2, time_fn=clock2) assert plain.status()["triage_state"] == "disabled" assert plain.clear() is True assert [e for e in emitted2 if e["type"] == "triage"] == [] def test_ledger_merge_dedupes_exact_and_punctuation_variants(): ledger = FactLedger() assert ledger.merge(["The budget was $305 million.", "Sense could find prey."]) == 2 assert ledger.merge(["sense could find prey", "The budget was $305 million"]) == 0 assert ledger.facts() == [ "The budget was $305 million.", "Sense could find prey.", ] def test_ledger_merge_keeps_near_match_with_different_number(): ledger = FactLedger() ledger.merge(["Sold 1.9 million units."]) assert ledger.merge(["Sold 10 million units."]) == 1 assert len(ledger) == 2 def test_ledger_merge_keeps_near_match_with_different_entity(): ledger = FactLedger() ledger.merge(["Marcus Smith directed the game."]) assert ledger.merge(["Mike Daly directed the game."]) == 1 assert len(ledger) == 2 def test_ledger_merge_ignores_empty_and_non_string(): ledger = FactLedger() assert ledger.merge(["", " ", None, 123]) == 0 assert ledger.render() == "" def test_ledger_render_has_prior_details_section(): ledger = FactLedger() ledger.merge(["fact one", "fact two"]) assert ledger.render() == "Prior Details:\n- fact one\n- fact two" def test_ledger_evicts_oldest_over_cap_and_never_new(): ledger = FactLedger(max_facts=2) ledger.merge(["alpha", "beta"]) ledger.merge(["gamma"]) assert ledger.facts() == ["beta", "gamma"] def test_ledger_truncates_overlong_fact(): ledger = FactLedger(fact_max_chars=10) ledger.merge(["x" * 50]) assert ledger.facts() == ["x" * 10] def test_ledger_clear_empties(): ledger = FactLedger() ledger.merge(["fact"]) ledger.clear() assert ledger.facts() == [] assert ledger.render() == "" def test_ledger_from_brief_harvests_bullets(): ledger = FactLedger.from_brief("Summary: s\n\nPrior Details:\n- one\n- two\n") assert ledger.facts() == ["one", "two"] def test_parse_ledger_reply_splits_narrative_and_facts(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) narrative, facts = analyzer.parse_ledger_reply( "NEW FACTS:\n- budget is $305m\n- sense finds prey\n\nBRIEF:\nSummary: ok" ) assert facts == ["budget is $305m", "sense finds prey"] assert narrative == "Summary: ok" def test_parse_ledger_reply_drops_trailing_fact_when_truncated(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) _, facts = analyzer.parse_ledger_reply( "NEW FACTS:\n- complete fact\n- half a fac", truncated=True ) assert facts == ["complete fact"] def test_parse_ledger_reply_missing_facts_block(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) narrative, facts = analyzer.parse_ledger_reply("BRIEF:\nSummary: only narrative") assert facts == [] assert narrative == "Summary: only narrative" def test_parse_ledger_reply_strips_model_prior_details(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) narrative, _ = analyzer.parse_ledger_reply( "NEW FACTS:\nnone\n\nBRIEF:\nSummary: s\n\nPrior Details:\n- model echo\n" ) assert "model echo" not in narrative assert "Summary: s" in narrative def test_parse_ledger_reply_none_on_unusable(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) assert analyzer.parse_ledger_reply("") is None assert analyzer.parse_ledger_reply("just some prose with no sections") is None def test_parse_facts_ignores_prose_and_handles_markers(tmp_path): analyzer = _analyzer(tmp_path, _score_questions()) block = "- bullet fact\n* star fact\n• dot fact\n1. numbered fact\nplain prose\nnone" assert analyzer.parse_facts(block) == [ "bullet fact", "star fact", "dot fact", "numbered fact", ] def _rich_questions(): return { "substantive_update": { "type": "noul", "instructions": "This recent conversation segment adds something a running brief " "should record.", }, "blocker": { "type": "noul", "instructions": "An obstacle, dependency, objection, or risk that could block " "or delay the work was raised.", }, "churn_risk": {"type": "noul", "instructions": "User may leave."}, "category": { "type": "choice", "instructions": "What is this about?", "criteria": {"support": "help needed", "sales": "buying"}, }, "severity": { "type": "score", "instructions": "How severe is the issue?", "criteria": ["none", "low", "high", "critical"], }, } def _rich_signals(**overrides): items = { "blocker": { "kind": "noul", "p": 0.91, "confidence": 0.9, "answer_confidence": 0.9, "gate": 0.9, "flagged": True, }, "churn_risk": { "kind": "noul", "p": 0.10, "confidence": 0.9, "answer_confidence": 0.9, "gate": 0.9, "flagged": False, }, "substantive_update": { "kind": "noul", "p": 0.10, "confidence": 0.9, "answer_confidence": 0.9, "gate": 0.9, "flagged": False, }, "category": { "kind": "choice", "choice": "support", "confidence": 0.4, "answer_confidence": 0.4, "gate": 0.4, "flagged": True, }, "severity": { "kind": "score", "expected": 2.6, "level": 3, "label": "critical", "confidence": 0.9, "answer_confidence": 0.9, "gate": 0.9, "flagged": True, }, } items.update(overrides.get("items", {})) order = overrides.get( "order", ["blocker", "churn_risk", "substantive_update", "category", "severity"]) return {"order": order, "items": items, "escalated": overrides.get("escalated", False), "input_tokens": 0} def test_signal_lines_filters_low_noul_but_keeps_gate_qid(tmp_path): analyzer = _analyzer(tmp_path, _rich_questions()) lines = analyzer._signal_lines(_rich_signals()) text = "\n".join(lines) assert not any("churn_risk" in line for line in lines) # p=0.10 filtered assert any("substantive_update" in line for line in lines) # gate qid kept assert any("blocker" in line for line in lines) # high-signal kept assert "p=0.91" in text and "flagged" in text assert "could block or delay" in text # instructions carried assert all(line.startswith("- ") for line in lines) # original order preserved among kept lines kept = [line for line in lines] assert [i for i, line in enumerate(kept) if "blocker" in line][0] < [ i for i, line in enumerate(kept) if "category" in line][0] def test_signal_lines_choice_carries_instructions_and_flag(tmp_path): analyzer = _analyzer(tmp_path, _rich_questions()) lines = analyzer._signal_lines(_rich_signals(order=["category"], escalated=False)) assert len(lines) == 1 assert "category (choice): selected support (conf 0.40, flagged as uncertain)" in lines[0] assert "What is this about?" in lines[0] assert "help needed" in lines[0] # criteria[selected] definition def test_signal_lines_escalated_marker_only_on_flagged(tmp_path): analyzer = _analyzer(tmp_path, _rich_questions()) lines = analyzer._signal_lines(_rich_signals( order=["blocker", "substantive_update"], escalated=True)) by_qid = {line.split(" ")[1]: line for line in lines} assert by_qid["blocker"].endswith(" [escalated]") assert not by_qid["substantive_update"].endswith(" [escalated]") plain = analyzer._signal_lines(_rich_signals( order=["blocker", "substantive_update"], escalated=False)) assert not any("[escalated]" in line for line in plain) def test_signal_lines_score_format_and_instruction_cap(tmp_path): questions = _rich_questions() questions["severity"]["instructions"] = "x " * 200 # force long instructions analyzer = _analyzer(tmp_path, questions) lines = analyzer._signal_lines(_rich_signals(order=["severity"])) assert len(lines) == 1 assert "severity (score): expected 2.60, level 3 (critical), flagged" in lines[0] assert len(lines[0]) <= len("- severity (score): expected 2.60, level 3 " "(critical), flagged — ") + 160 + len(" [escalated]") assert lines[0].rstrip().endswith("…") def test_signal_lines_never_raises_on_malformed(tmp_path): analyzer = _analyzer(tmp_path, _rich_questions()) assert analyzer._signal_lines(None) == [] # type: ignore[arg-type] assert analyzer._signal_lines({}) == [] # type: ignore[dict-item] assert analyzer._signal_lines({"order": "nope", "items": "nope"}) == [] # type: ignore[dict-item] assert analyzer._signal_lines({"order": ["ghost"], "items": {}}) == []