test_triage.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812
  1. """Tests for the triage analyzer (todos 3, 9, 10, 11). Pure, hardware-free."""
  2. import json
  3. import logging
  4. from pathlib import Path
  5. from types import SimpleNamespace
  6. import pytest
  7. from stt_daemon import events
  8. from stt_daemon.assistant import triage
  9. from stt_daemon.assistant.service import AssistantService
  10. from stt_daemon.assistant.triage import TriageAnalyzer, load_questions
  11. from stt_daemon.config import Config
  12. from stt_daemon.tui.state import ViewState
  13. EXAMPLE_PATH = str(Path(__file__).resolve().parent.parent / "questions.example.json")
  14. def _cfg(path=EXAMPLE_PATH, **overrides):
  15. base = {
  16. "questions_file": path,
  17. "state_char_budget": 1200,
  18. "confidence_field": "answer_confidence",
  19. "escalate_threshold": 0.6,
  20. "choice_min_confidence": 0.85,
  21. }
  22. base.update(overrides)
  23. return SimpleNamespace(**base)
  24. def _write_questions(tmp_path, questions):
  25. path = tmp_path / "questions.json"
  26. path.write_text(json.dumps(questions), encoding="utf-8")
  27. return str(path)
  28. def _analyzer(tmp_path, questions=None, **overrides):
  29. if questions is None:
  30. return TriageAnalyzer(_cfg(**overrides))
  31. return TriageAnalyzer(_cfg(_write_questions(tmp_path, questions), **overrides))
  32. # --- todo 3: load_questions -------------------------------------------------
  33. def test_load_example_order_preserved():
  34. raw = json.loads(Path(EXAMPLE_PATH).read_text(encoding="utf-8"))
  35. questions = load_questions(EXAMPLE_PATH)
  36. assert questions is not None
  37. assert questions
  38. assert list(questions.keys()) == list(raw.keys())
  39. for definition in questions.values():
  40. assert definition["type"] in {"choice", "score", "noul"}
  41. assert isinstance(definition["instructions"], str)
  42. assert definition["instructions"].strip()
  43. def test_example_questions_match_design():
  44. questions = load_questions(EXAMPLE_PATH)
  45. assert questions is not None
  46. assert list(questions.keys()) == [
  47. "substantive_update",
  48. "opportunity",
  49. "blocker",
  50. "decision_status",
  51. "action_item",
  52. "responsibility_clarity",
  53. "deadline",
  54. "open_question",
  55. ]
  56. for qid, definition in questions.items():
  57. assert isinstance(definition["instructions"], str)
  58. assert definition["instructions"].strip()
  59. if definition["type"] == "choice":
  60. criteria = definition["criteria"]
  61. options = list(criteria.keys()) if isinstance(criteria, dict) else list(criteria)
  62. assert "other" in options
  63. assert len(options) <= 7
  64. if definition["type"] == "noul":
  65. assert isinstance(definition.get("instructions"), str)
  66. assert definition["instructions"].strip()
  67. def test_load_missing_file_returns_none_with_warning(caplog):
  68. with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"):
  69. assert load_questions("/nonexistent/questions.json") is None
  70. assert caplog.records, "expected a warning for the missing file"
  71. def test_load_bad_json_returns_none_with_warning(tmp_path, caplog):
  72. path = tmp_path / "questions.json"
  73. path.write_text("{not valid json", encoding="utf-8")
  74. with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"):
  75. assert load_questions(str(path)) is None
  76. assert caplog.records
  77. @pytest.mark.parametrize(
  78. "bad_entry",
  79. [
  80. {"type": "mystery", "instructions": "do something"}, # unknown type
  81. {"type": "score", "instructions": "rate it", "criteria": {"0": "low"}}, # score dict
  82. { # noul criteria with a bad key
  83. "type": "noul",
  84. "instructions": "holds?",
  85. "criteria": {"true": "yes", "maybe": "hmm"},
  86. },
  87. "just a string", # non-object entry
  88. {"type": "choice", "instructions": "pick one", "criteria": []}, # empty choice
  89. {"type": "choice", "instructions": "pick one", "criteria": {}}, # empty choice dict
  90. {"type": "choice", "criteria": ["a", "b"]}, # missing instructions
  91. {"type": "score", "instructions": "rate it", "criteria": ["a", None]}, # None level
  92. {"type": "noul", "instructions": "holds?", "criteria": ["true"]}, # noul list
  93. ],
  94. )
  95. def test_load_malformed_returns_none_with_warning(tmp_path, caplog, bad_entry):
  96. path = _write_questions(
  97. tmp_path, {"good": {"type": "noul", "instructions": "ok?"}, "bad": bad_entry}
  98. )
  99. with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"):
  100. assert load_questions(str(path)) is None
  101. assert caplog.records
  102. # --- todo 9: normalize ------------------------------------------------------
  103. def _score_questions():
  104. return {
  105. "severity": {
  106. "type": "score",
  107. "instructions": "How severe is the issue?",
  108. "criteria": ["none", "low", "high", "critical"],
  109. },
  110. "category": {
  111. "type": "choice",
  112. "instructions": "What is this about?",
  113. "criteria": {"support": "help needed", "sales": "buying"},
  114. },
  115. "churn_risk": {"type": "noul", "instructions": "User may leave."},
  116. }
  117. def test_normalize_choice_exact(tmp_path):
  118. analyzer = _analyzer(tmp_path, _score_questions())
  119. payload = {
  120. "model": "multilingual",
  121. "answers": {
  122. "category": {
  123. "type": "choice",
  124. "choice": "support",
  125. "confidence": 0.7,
  126. "answer_confidence": 0.4,
  127. }
  128. },
  129. "usage": {"input_tokens": 123, "output_tokens": 0},
  130. }
  131. result = analyzer.normalize(payload)
  132. assert result["order"] == ["category"]
  133. assert result["items"]["category"] == {
  134. "kind": "choice",
  135. "choice": "support",
  136. "confidence": 0.7,
  137. "answer_confidence": 0.4,
  138. "gate": 0.4,
  139. "flagged": True,
  140. }
  141. assert result["input_tokens"] == 123
  142. assert result["escalated"] is True
  143. def test_normalize_score_with_legend(tmp_path):
  144. analyzer = _analyzer(tmp_path, _score_questions())
  145. payload = {
  146. "answers": {
  147. "severity": {
  148. "type": "score",
  149. "score": 2.6,
  150. "legend": {"0": "none", "1": "low", "2": "high", "3": "critical"},
  151. "confidence": 0.5,
  152. "answer_confidence": 0.6,
  153. }
  154. },
  155. "usage": {"input_tokens": 7, "output_tokens": 0},
  156. }
  157. item = analyzer.normalize(payload)["items"]["severity"]
  158. assert item["kind"] == "score"
  159. assert item["expected"] == 2.6
  160. assert item["level"] == 3
  161. assert item["label"] == "critical"
  162. assert item["confidence"] == 0.5
  163. assert item["answer_confidence"] == 0.6
  164. assert item["gate"] == 0.6
  165. assert item["flagged"] is True
  166. def test_normalize_score_legend_fallback(tmp_path):
  167. analyzer = _analyzer(tmp_path, _score_questions())
  168. payload = {
  169. "answers": {
  170. "severity": {"type": "score", "score": 1.2, "confidence": 0.9, "answer_confidence": 0.9}
  171. }
  172. }
  173. item = analyzer.normalize(payload)["items"]["severity"]
  174. assert item["level"] == 1
  175. assert item["label"] == "low" # falls back to criteria[level]
  176. assert item["flagged"] is False # 1.2 < (4 - 1) * 0.5
  177. def test_normalize_noul_and_bare_payload(tmp_path):
  178. analyzer = _analyzer(tmp_path, _score_questions())
  179. bare = {
  180. "churn_risk": {"type": "noul", "noul": 0.9, "confidence": 0.9, "answer_confidence": 0.85}
  181. }
  182. result = analyzer.normalize(bare)
  183. assert result["order"] == ["churn_risk"]
  184. assert result["items"]["churn_risk"] == {
  185. "kind": "noul",
  186. "p": 0.9,
  187. "confidence": 0.9,
  188. "answer_confidence": 0.85,
  189. "gate": 0.85,
  190. "flagged": True,
  191. }
  192. assert result["input_tokens"] == 0
  193. def test_normalize_skips_unknown_qid_and_malformed(tmp_path):
  194. analyzer = _analyzer(tmp_path, _score_questions())
  195. payload = {
  196. "answers": {
  197. "ghost": {"type": "noul", "noul": 0.99, "confidence": 0.9, "answer_confidence": 0.9},
  198. "category": {"type": "choice", "choice": 42}, # malformed: no confidences
  199. "churn_risk": "nonsense", # malformed: not a dict
  200. },
  201. "usage": {"input_tokens": 3, "output_tokens": 0},
  202. }
  203. result = analyzer.normalize(payload)
  204. assert result["order"] == []
  205. assert result["items"] == {}
  206. assert result["escalated"] is False
  207. def test_normalize_ignores_side_channel_fields(tmp_path):
  208. analyzer = _analyzer(tmp_path, _score_questions())
  209. payload = {
  210. "answers": {
  211. "category": {
  212. "type": "choice",
  213. "choice": "sales",
  214. "confidence": 0.95,
  215. "answer_confidence": 0.9,
  216. "action": {"act" + "_probability": 0.99},
  217. }
  218. },
  219. "usage": {"input_tokens": 1, "output_tokens": 0},
  220. }
  221. result = analyzer.normalize(payload)
  222. assert result["items"]["category"]["flagged"] is False
  223. assert "act" + "_probability" not in json.dumps(result)
  224. source = Path(triage.__file__).read_text(encoding="utf-8")
  225. assert "act" + "_probability" not in source
  226. def test_normalize_gating_field_selectable(tmp_path):
  227. analyzer = _analyzer(tmp_path, _score_questions(), confidence_field="confidence")
  228. payload = {
  229. "answers": {
  230. "category": {
  231. "type": "choice",
  232. "choice": "sales",
  233. "confidence": 0.5,
  234. "answer_confidence": 0.99,
  235. }
  236. }
  237. }
  238. item = analyzer.normalize(payload)["items"]["category"]
  239. assert item["gate"] == 0.5 # gates on confidence, keeps both
  240. assert item["answer_confidence"] == 0.99
  241. assert item["flagged"] is True
  242. # --- todo 9: condensation ---------------------------------------------------
  243. def test_condense_keeps_newest_chronological_no_numerals(tmp_path):
  244. analyzer = _analyzer(tmp_path, _score_questions(), state_char_budget=40)
  245. window = ["alpha first", "beta second", "gamma third newest here"]
  246. state = analyzer.build_state(window)
  247. assert "gamma third newest here" in state
  248. assert "alpha first" not in state # oldest dropped under the budget
  249. lines = state.splitlines()
  250. assert all(line.startswith("- ") for line in lines)
  251. assert not any(ch.isdigit() for ch in state) # no numerals injected
  252. # chronological restore: kept lines stay in original order
  253. kept = [u for u in window if u in state]
  254. assert [line[2:] for line in lines] == kept
  255. def test_condense_empty_window(tmp_path):
  256. analyzer = _analyzer(tmp_path, _score_questions())
  257. assert analyzer.build_state([]) == ""
  258. # --- todo 10: escalation + extraction ---------------------------------------
  259. def _signals(choice_gate=0.9, noul_p=0.1, score_expected=0.2):
  260. return {
  261. "order": ["category", "churn_risk", "severity"],
  262. "items": {
  263. "category": {
  264. "kind": "choice",
  265. "choice": "sales",
  266. "confidence": choice_gate,
  267. "answer_confidence": choice_gate,
  268. "gate": choice_gate,
  269. "flagged": choice_gate < 0.85,
  270. },
  271. "churn_risk": {
  272. "kind": "noul",
  273. "p": noul_p,
  274. "confidence": noul_p,
  275. "answer_confidence": noul_p,
  276. "gate": noul_p,
  277. "flagged": noul_p >= 0.6,
  278. },
  279. "severity": {
  280. "kind": "score",
  281. "expected": score_expected,
  282. "level": 0,
  283. "label": "none",
  284. "confidence": 0.9,
  285. "answer_confidence": 0.9,
  286. "gate": 0.9,
  287. "flagged": score_expected >= 1.5,
  288. },
  289. },
  290. "escalated": False,
  291. "input_tokens": 0,
  292. }
  293. def test_escalation_arms_fire_independently(tmp_path):
  294. analyzer = _analyzer(tmp_path, _score_questions())
  295. assert analyzer.should_escalate(_signals(choice_gate=0.5)) is True # choice arm
  296. assert analyzer.should_escalate(_signals(noul_p=0.8)) is True # noul arm
  297. assert analyzer.should_escalate(_signals(score_expected=2.0)) is True # score arm
  298. assert analyzer.should_escalate(_signals()) is False
  299. def test_normalize_escalated_matches_should_escalate(tmp_path):
  300. analyzer = _analyzer(tmp_path, _score_questions())
  301. payload = {
  302. "answers": {
  303. "category": {
  304. "type": "choice",
  305. "choice": "sales",
  306. "confidence": 0.9,
  307. "answer_confidence": 0.9,
  308. },
  309. "churn_risk": {
  310. "type": "noul",
  311. "noul": 0.1,
  312. "confidence": 0.9,
  313. "answer_confidence": 0.9,
  314. },
  315. "severity": {
  316. "type": "score",
  317. "score": 0.1,
  318. "confidence": 0.9,
  319. "answer_confidence": 0.9,
  320. },
  321. }
  322. }
  323. result = analyzer.normalize(payload)
  324. assert result["escalated"] == analyzer.should_escalate(result) is False
  325. # --- todo 5: brief prompt / parse / gate / finalize --------------------------
  326. def test_build_brief_prompt_order_and_focus(tmp_path):
  327. analyzer = _analyzer(tmp_path, _score_questions())
  328. prompt = analyzer.build_brief_prompt(
  329. ["Alice: let's ship Friday", "Bob: agreed"], _signals(noul_p=0.8)
  330. )
  331. assert prompt.startswith("New utterances:\n- Alice: let's ship Friday\n- Bob: agreed")
  332. assert "Focus:" in prompt
  333. assert "churn_risk" in prompt # signal named in words
  334. assert prompt.index("New utterances:") < prompt.index("Focus:")
  335. assert prompt.index("Focus:") < prompt.index("Return only the updated brief")
  336. # no JSON extraction keys or raw dict shapes leak into the brief prompt
  337. for key in ("summary", "opportunities", "action_items", "risks", "doubts"):
  338. assert key not in prompt
  339. assert "{" not in prompt and "}" not in prompt
  340. def test_build_brief_prompt_no_focus_without_signals(tmp_path):
  341. analyzer = _analyzer(tmp_path, _score_questions())
  342. for signals in (None, {}, {"order": [], "items": {}}):
  343. prompt = analyzer.build_brief_prompt(["solo utterance"], signals)
  344. assert "solo utterance" in prompt
  345. assert "Focus:" not in prompt
  346. assert prompt.endswith("Return only the updated brief as plain text.")
  347. def test_parse_brief_strips_fences_and_rejects_empty(tmp_path):
  348. analyzer = _analyzer(tmp_path, _score_questions())
  349. assert analyzer.parse_brief("```\nhello brief\n```") == "hello brief"
  350. assert analyzer.parse_brief("```markdown\nhello brief\n```") == "hello brief"
  351. assert analyzer.parse_brief("plain brief") == "plain brief"
  352. assert analyzer.parse_brief(" padded ") == "padded"
  353. assert analyzer.parse_brief("") is None
  354. assert analyzer.parse_brief(" ") is None
  355. assert analyzer.parse_brief(None) is None
  356. assert analyzer.parse_brief(42) is None
  357. assert analyzer.parse_brief("```") is None
  358. def test_parse_brief_truncates_not_rejects(tmp_path):
  359. analyzer = _analyzer(tmp_path, _score_questions())
  360. parsed = analyzer.parse_brief("x" * 5000)
  361. assert parsed is not None
  362. assert parsed == "x" * 4000
  363. assert len(parsed) == 4000
  364. wrapped = "[update_brief(updated_brief='" + "x" * 5000 + "')]"
  365. assert analyzer.parse_brief(wrapped) == "x" * 4000
  366. def test_parse_brief_honors_configured_cap(tmp_path):
  367. analyzer = _analyzer(tmp_path, _score_questions(), brief_max_chars=10)
  368. assert analyzer.parse_brief("x" * 50) == "x" * 10
  369. def test_parse_brief_unwraps_single_quoted_tool_call_with_apostrophes(tmp_path):
  370. analyzer = _analyzer(tmp_path, _score_questions())
  371. observed = "<|tool_call_start|>[update_brief(updated_brief='the company's game failed')]"
  372. assert analyzer.parse_brief(observed) == "the company's game failed"
  373. def test_parse_brief_unwraps_double_quoted_tool_call(tmp_path):
  374. analyzer = _analyzer(tmp_path, _score_questions())
  375. wrapped = '[update_brief(updated_brief="the company\'s plan")]'
  376. assert analyzer.parse_brief(wrapped) == "the company's plan"
  377. def test_parse_brief_unwraps_trailing_end_marker(tmp_path):
  378. analyzer = _analyzer(tmp_path, _score_questions())
  379. wrapped = "[update_brief(updated_brief='ship friday')]<|tool_call_end|>"
  380. assert analyzer.parse_brief(wrapped) == "ship friday"
  381. def test_parse_brief_unwraps_fenced_and_wrapped(tmp_path):
  382. analyzer = _analyzer(tmp_path, _score_questions())
  383. wrapped = "```\n<|tool_call_start|>[update_brief(updated_brief='fenced payload')]\n```"
  384. assert analyzer.parse_brief(wrapped) == "fenced payload"
  385. def test_parse_brief_does_not_mangle_clean_markdown(tmp_path):
  386. analyzer = _analyzer(tmp_path, _score_questions())
  387. clean = "Here's the [brief]: it's the company's plan - don't mangle [it]."
  388. assert analyzer.parse_brief(clean) == clean
  389. prose = "We should update_brief the notes, here's [why]."
  390. assert analyzer.parse_brief(prose) == prose
  391. def test_parse_brief_empty_tool_call_payload_is_none(tmp_path):
  392. analyzer = _analyzer(tmp_path, _score_questions())
  393. assert analyzer.parse_brief("[update_brief(updated_brief='')]") is None
  394. def _brief_signals(p=None):
  395. item = {"kind": "noul"}
  396. if p is not None:
  397. item["p"] = p
  398. return {"order": ["substantive_update"], "items": {"substantive_update": item}}
  399. def test_should_update_brief_gate(tmp_path):
  400. analyzer = _analyzer(tmp_path, _score_questions())
  401. assert analyzer.should_update_brief(_brief_signals(0.9)) is True
  402. assert analyzer.should_update_brief(_brief_signals(0.5)) is True # boundary
  403. assert analyzer.should_update_brief(_brief_signals(0.2)) is False
  404. # absent update / no numeric p / not a dict -> fail open
  405. assert analyzer.should_update_brief({"order": [], "items": {}}) is True
  406. assert analyzer.should_update_brief(_brief_signals(None)) is True
  407. assert analyzer.should_update_brief(None) is True
  408. assert analyzer.should_update_brief({"items": {"substantive_update": "nope"}}) is True
  409. def test_should_update_brief_threshold_configured(tmp_path):
  410. analyzer = _analyzer(tmp_path, _score_questions(), brief_gate_threshold=0.8)
  411. assert analyzer.should_update_brief(_brief_signals(0.6)) is False
  412. assert analyzer.should_update_brief(_brief_signals(0.8)) is True
  413. def test_finalize_brief_length_truncates_not_rejects(tmp_path, caplog):
  414. analyzer = _analyzer(tmp_path, _score_questions())
  415. with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"):
  416. result = analyzer.finalize_brief("y" * 5000, "length")
  417. assert result is not None
  418. assert result == "y" * 4000
  419. assert caplog.records # warned about the length cutoff but still accepted
  420. def test_finalize_brief_stop_and_empty(tmp_path):
  421. analyzer = _analyzer(tmp_path, _score_questions())
  422. assert analyzer.finalize_brief("a brief", "stop") == "a brief"
  423. assert analyzer.finalize_brief("", "stop") is None
  424. assert analyzer.finalize_brief("```\nfenced\n```", "stop") == "fenced"
  425. assert analyzer.finalize_brief(None, "length") is None
  426. # --- todo 11: cached state() ------------------------------------------------
  427. def test_state_initial_loading_with_good_questions(tmp_path):
  428. analyzer = _analyzer(tmp_path, _score_questions())
  429. assert analyzer.state() == ("loading", "")
  430. def test_state_initial_no_questions_with_bad_path(tmp_path):
  431. missing = str(tmp_path / "missing.json")
  432. analyzer = TriageAnalyzer(_cfg(missing))
  433. state, reason = analyzer.state()
  434. assert state == "no_questions"
  435. assert missing in reason
  436. def test_state_vocabulary_has_no_extra_member():
  437. assert set(triage.TRIAGE_REASONS) == {
  438. "ok",
  439. "no_questions",
  440. "loading",
  441. "laya_missing",
  442. "disabled",
  443. "error",
  444. }
  445. assert "service" + "_down" not in triage.TRIAGE_REASONS
  446. def test_refresh_transitions(tmp_path):
  447. analyzer = _analyzer(tmp_path, _score_questions())
  448. analyzer.refresh("ok")
  449. assert analyzer.state() == ("ok", "")
  450. analyzer.refresh("laya_missing", "run: uv sync --extra triage")
  451. assert analyzer.state() == ("laya_missing", "run: uv sync --extra triage")
  452. analyzer.refresh("error", "boom")
  453. assert analyzer.state() == ("error", "boom")
  454. def test_refresh_unknown_state_ignored(tmp_path, caplog):
  455. analyzer = _analyzer(tmp_path, _score_questions())
  456. analyzer.refresh("ok")
  457. with caplog.at_level(logging.WARNING, logger="stt_daemon.assistant.triage"):
  458. analyzer.refresh("bogus_state")
  459. assert analyzer.state() == ("ok", "")
  460. assert caplog.records
  461. def test_no_questions_state_cannot_be_overridden(tmp_path):
  462. analyzer = TriageAnalyzer(_cfg(str(tmp_path / "missing.json")))
  463. assert analyzer.state()[0] == "no_questions"
  464. analyzer.refresh("ok")
  465. analyzer.refresh("error", "boom")
  466. assert analyzer.state()[0] == "no_questions"
  467. def test_state_does_not_touch_lane(tmp_path):
  468. analyzer = _analyzer(tmp_path, _score_questions())
  469. class StrictLane:
  470. def __init__(self):
  471. self.calls = []
  472. def __getattr__(self, name):
  473. def record(*args, **kwargs):
  474. self.calls.append(name)
  475. return record
  476. lane = StrictLane()
  477. for _ in range(3):
  478. analyzer.state()
  479. analyzer.refresh("ok")
  480. analyzer.state()
  481. assert lane.calls == []
  482. # --- F3: end-to-end triage flow on fakes ------------------------------------
  483. EXTRACT_JSON = (
  484. '{"summary":"s","opportunities":[],"doubts":[],"action_items":[],"risks":[]}'
  485. )
  486. class _Clock:
  487. """Injected clock: no sleeps, fully deterministic."""
  488. def __init__(self, start=1000.0):
  489. self.t = float(start)
  490. def __call__(self):
  491. return self.t
  492. def advance(self, dt):
  493. self.t += float(dt)
  494. class _FakeSubscriber:
  495. def events(self):
  496. return iter(())
  497. def close(self):
  498. pass
  499. class _FakeLane:
  500. """Duck-typed LayaLane: canned-ready, records submits, counts drops."""
  501. def __init__(self):
  502. self.submits: list[tuple] = []
  503. self.drops = 0
  504. def ready(self) -> bool:
  505. return True
  506. def state(self):
  507. return ("ok", "")
  508. def submit(self, state, questions, lang, generation):
  509. self.submits.append((state, questions, lang, generation))
  510. return True
  511. def drop_pending(self):
  512. self.drops += 1
  513. return 1
  514. def _answers(noul_p=0.1):
  515. """A full laya-shaped answers payload for the example questions.
  516. `noul_p` drives every noul arm, including the `substantive_update` gate:
  517. below `brief_gate_threshold` (0.5) the turn is gated OUT (no brief job),
  518. at or above it the turn gates IN.
  519. """
  520. answers = {
  521. "decision_status": {
  522. "type": "choice",
  523. "choice": "made",
  524. "confidence": 0.9,
  525. "answer_confidence": 0.9,
  526. },
  527. "responsibility_clarity": {
  528. "type": "choice",
  529. "choice": "explicit",
  530. "confidence": 0.9,
  531. "answer_confidence": 0.9,
  532. },
  533. }
  534. for qid in (
  535. "substantive_update",
  536. "opportunity",
  537. "blocker",
  538. "action_item",
  539. "deadline",
  540. "open_question",
  541. ):
  542. answers[qid] = {
  543. "type": "noul",
  544. "noul": noul_p,
  545. "confidence": 0.9,
  546. "answer_confidence": 0.9,
  547. }
  548. return {"model": "multilingual", "answers": answers, "usage": {"input_tokens": 42}}
  549. def test_end_to_end_triage_flow():
  550. """signals -> content -> cleared across service, analyzer, lane, and ViewState.
  551. The lane result is wired by calling `service._on_triage_result(gen, payload)`
  552. directly (the runner does exactly this in production); the brief job is
  553. drained with `job.run()` so no worker thread is started. The
  554. single-outstanding guard plus the shared `_pending_brief` collapse settled
  555. turns into ONE job consuming every buffered delta, and the
  556. `substantive_update` gate can reduce that to ZERO jobs.
  557. """
  558. cfg = Config()
  559. cfg.assistant.turn_gap_s = 0.5
  560. cfg.triage.enabled = True
  561. cfg.triage.questions_file = EXAMPLE_PATH
  562. clock = _Clock(1000.0)
  563. lane = _FakeLane()
  564. emitted: list[dict] = []
  565. prompts: list[str] = []
  566. def emit(type_, **fields):
  567. emitted.append({"type": type_, **fields})
  568. analyzer = TriageAnalyzer(cfg.triage)
  569. service = AssistantService(
  570. cfg,
  571. _FakeSubscriber(),
  572. emit=emit,
  573. time_fn=clock,
  574. chat_fn=lambda text: "reply",
  575. triage=analyzer,
  576. laya_lane=lane,
  577. brief_fn=lambda prompt: prompts.append(prompt) or "BRIEF-TEXT",
  578. )
  579. def triage_events():
  580. return [e for e in emitted if e["type"] == "triage"]
  581. def brief_jobs():
  582. return [j for j in service._inference._queue if j.name == "brief-update"]
  583. # 1. settled turn -> exactly one lane submit, lang + generation recorded
  584. assert service.toggle_chat() is True
  585. service.handle_event(
  586. {"type": events.TRANSCRIPT, "text": "hola, necesito ayuda", "lang": "es"}
  587. )
  588. clock.advance(1.0)
  589. service.tick()
  590. assert len(lane.submits) == 1
  591. state_text, questions, lang, gen1 = lane.submits[0]
  592. assert "hola, necesito ayuda" in state_text
  593. assert questions is analyzer.questions
  594. assert lang == "es"
  595. assert gen1 == 0
  596. # 2. low-p turn -> signals only: the gate (p=0.1 < 0.5) dispatches ZERO jobs
  597. service._on_triage_result(gen1, _answers(noul_p=0.1))
  598. assert [e["stage"] for e in triage_events()] == ["signals"]
  599. assert triage_events()[0]["triage"]["escalated"] is False
  600. assert service.last_triage is not None
  601. assert service.last_triage["escalated"] is False
  602. assert brief_jobs() == []
  603. assert prompts == []
  604. # 3. high-p turn -> second submit, then ONE brief job (the only one outstanding)
  605. service.handle_event(
  606. {"type": events.TRANSCRIPT, "text": "quiero cancelar mi suscripcion", "lang": "es"}
  607. )
  608. clock.advance(1.0)
  609. service.tick()
  610. assert len(lane.submits) == 2
  611. gen2 = lane.submits[1][3]
  612. service._on_triage_result(gen2, _answers(noul_p=0.9))
  613. assert [e["stage"] for e in triage_events()] == ["signals", "signals"]
  614. assert triage_events()[1]["triage"]["escalated"] is True
  615. jobs = brief_jobs()
  616. assert len(jobs) == 1
  617. # 4. run the brief job directly (deterministic, no worker thread): it consumes
  618. # both buffered deltas and emits ONE content with the brief merged over signals
  619. jobs[0].run()
  620. assert len(prompts) == 1
  621. assert "hola, necesito ayuda" in prompts[0]
  622. assert "quiero cancelar mi suscripcion" in prompts[0]
  623. content_events = [e for e in triage_events() if e["stage"] == "content"]
  624. assert len(content_events) == 1
  625. merged = content_events[0]["triage"]
  626. assert merged["brief"] == "BRIEF-TEXT"
  627. assert merged["escalated"] is True # signals preserved under the brief
  628. assert merged["order"] == triage_events()[1]["triage"]["order"]
  629. # 5. drive the emitted stream through a fresh ViewState
  630. view = ViewState()
  631. for event in emitted:
  632. view.ingest(event)
  633. analyzer.refresh("ok") # the lane callback does this in production
  634. view.ingest({"type": "status", **service.status()})
  635. assert view.triage is not None
  636. assert view.triage["brief"] == "BRIEF-TEXT"
  637. lines = view.triage_lines()
  638. assert lines
  639. # 6. clear -> cleared event empties the panel and bumps the epoch
  640. baseline_epoch = view.triage_epoch
  641. already = len(emitted)
  642. service.clear()
  643. for event in emitted[already:]:
  644. view.ingest(event)
  645. assert view.triage is None
  646. assert view.triage_epoch > baseline_epoch
  647. # a stale result after clear must not re-populate the panel
  648. service._on_triage_result(gen2, _answers(noul_p=0.9))
  649. assert [e["stage"] for e in triage_events()] == [
  650. "signals",
  651. "signals",
  652. "content",
  653. "cleared",
  654. ]
  655. assert view.triage is None
  656. assert lane.drops == 1
  657. # 7. disabled control: no triage/lane -> status disabled, clear emits nothing
  658. cfg2 = Config()
  659. clock2 = _Clock(2000.0)
  660. emitted2: list[dict] = []
  661. def emit2(type_, **fields):
  662. emitted2.append({"type": type_, **fields})
  663. plain = AssistantService(cfg2, _FakeSubscriber(), emit=emit2, time_fn=clock2)
  664. assert plain.status()["triage_state"] == "disabled"
  665. assert plain.clear() is True
  666. assert [e for e in emitted2 if e["type"] == "triage"] == []