|
@@ -13,6 +13,9 @@ from types import SimpleNamespace
|
|
|
import pytest
|
|
import pytest
|
|
|
|
|
|
|
|
from stt_daemon import events
|
|
from stt_daemon import events
|
|
|
|
|
+from stt_daemon.assistant.brief_file import BriefFile
|
|
|
|
|
+from stt_daemon.assistant.inference import InferenceJob, InferenceQueue
|
|
|
|
|
+from stt_daemon.assistant.llm import ConversationEngine
|
|
|
from stt_daemon.assistant.runner import build_triage
|
|
from stt_daemon.assistant.runner import build_triage
|
|
|
from stt_daemon.assistant.service import AssistantService
|
|
from stt_daemon.assistant.service import AssistantService
|
|
|
from stt_daemon.bus import EventPublisher, EventSubscriber
|
|
from stt_daemon.bus import EventPublisher, EventSubscriber
|
|
@@ -110,23 +113,24 @@ class _FakeChatClient:
|
|
|
def __init__(self):
|
|
def __init__(self):
|
|
|
self.calls = []
|
|
self.calls = []
|
|
|
|
|
|
|
|
- def chat(self, messages, tools=None, max_tokens=None, timeout_s=None):
|
|
|
|
|
|
|
+ def chat(self, messages, tools=None, max_tokens=None, timeout_s=None, slot_id=None):
|
|
|
self.calls.append(
|
|
self.calls.append(
|
|
|
{
|
|
{
|
|
|
"messages": messages,
|
|
"messages": messages,
|
|
|
"tools": tools,
|
|
"tools": tools,
|
|
|
"max_tokens": max_tokens,
|
|
"max_tokens": max_tokens,
|
|
|
"timeout_s": timeout_s,
|
|
"timeout_s": timeout_s,
|
|
|
|
|
+ "slot_id": slot_id,
|
|
|
}
|
|
}
|
|
|
)
|
|
)
|
|
|
- return SimpleNamespace(content="EXTRACTED")
|
|
|
|
|
|
|
+ return SimpleNamespace(content="EXTRACTED", finish_reason="stop")
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_build_triage_disabled_returns_nones():
|
|
def test_build_triage_disabled_returns_nones():
|
|
|
cfg = Config()
|
|
cfg = Config()
|
|
|
assert cfg.triage.enabled is False
|
|
assert cfg.triage.enabled is False
|
|
|
- assert build_triage(cfg, None) == (None, None, None)
|
|
|
|
|
- assert build_triage(cfg, _FakeChatClient()) == (None, None, None)
|
|
|
|
|
|
|
+ assert build_triage(cfg, None) == (None, None, None, None)
|
|
|
|
|
+ assert build_triage(cfg, _FakeChatClient()) == (None, None, None, None)
|
|
|
assert "laya" not in sys.modules
|
|
assert "laya" not in sys.modules
|
|
|
service = AssistantService(cfg, subscriber=object())
|
|
service = AssistantService(cfg, subscriber=object())
|
|
|
assert service.status()["triage_state"] == "disabled"
|
|
assert service.status()["triage_state"] == "disabled"
|
|
@@ -135,10 +139,11 @@ def test_build_triage_disabled_returns_nones():
|
|
|
def test_build_triage_enabled_constructs_without_client(tmp_path):
|
|
def test_build_triage_enabled_constructs_without_client(tmp_path):
|
|
|
cfg = _enabled_triage_cfg(tmp_path)
|
|
cfg = _enabled_triage_cfg(tmp_path)
|
|
|
before = len(threading.enumerate())
|
|
before = len(threading.enumerate())
|
|
|
- analyzer, lane, extract_fn = build_triage(cfg, None)
|
|
|
|
|
|
|
+ analyzer, lane, brief_fn, brief_reset = build_triage(cfg, None)
|
|
|
assert analyzer is not None
|
|
assert analyzer is not None
|
|
|
assert lane is not None
|
|
assert lane is not None
|
|
|
- assert extract_fn is None
|
|
|
|
|
|
|
+ assert brief_fn is None
|
|
|
|
|
+ assert brief_reset is None
|
|
|
assert analyzer._cfg is cfg.triage
|
|
assert analyzer._cfg is cfg.triage
|
|
|
assert lane._cfg is cfg.triage
|
|
assert lane._cfg is cfg.triage
|
|
|
assert analyzer.state() == ("loading", "")
|
|
assert analyzer.state() == ("loading", "")
|
|
@@ -146,12 +151,15 @@ def test_build_triage_enabled_constructs_without_client(tmp_path):
|
|
|
assert "laya" not in sys.modules
|
|
assert "laya" not in sys.modules
|
|
|
|
|
|
|
|
|
|
|
|
|
-def test_build_triage_extract_fn_calls_client(tmp_path):
|
|
|
|
|
|
|
+def test_build_triage_brief_fn_calls_client(tmp_path):
|
|
|
cfg = _enabled_triage_cfg(tmp_path)
|
|
cfg = _enabled_triage_cfg(tmp_path)
|
|
|
client = _FakeChatClient()
|
|
client = _FakeChatClient()
|
|
|
- _, _, extract_fn = build_triage(cfg, client)
|
|
|
|
|
- assert extract_fn is not None
|
|
|
|
|
- assert extract_fn("PROMPT") == "EXTRACTED"
|
|
|
|
|
|
|
+ _, _, brief_fn, brief_reset = build_triage(cfg, client)
|
|
|
|
|
+ assert brief_fn is not None
|
|
|
|
|
+ assert brief_reset is not None
|
|
|
|
|
+ assert callable(brief_fn)
|
|
|
|
|
+ assert callable(brief_reset)
|
|
|
|
|
+ assert brief_fn("PROMPT") == "EXTRACTED"
|
|
|
assert len(client.calls) == 1
|
|
assert len(client.calls) == 1
|
|
|
call = client.calls[0]
|
|
call = client.calls[0]
|
|
|
assert call["messages"] == [
|
|
assert call["messages"] == [
|
|
@@ -160,6 +168,17 @@ def test_build_triage_extract_fn_calls_client(tmp_path):
|
|
|
]
|
|
]
|
|
|
assert call["max_tokens"] == cfg.triage.extract_max_tokens
|
|
assert call["max_tokens"] == cfg.triage.extract_max_tokens
|
|
|
assert call["timeout_s"] == cfg.triage.extract_timeout_s
|
|
assert call["timeout_s"] == cfg.triage.extract_timeout_s
|
|
|
|
|
+ assert call["slot_id"] is None
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_build_triage_brief_fn_slot_only_when_configured(tmp_path):
|
|
|
|
|
+ cfg = _enabled_triage_cfg(tmp_path)
|
|
|
|
|
+ cfg.triage.brief_slot_id = 2
|
|
|
|
|
+ client = _FakeChatClient()
|
|
|
|
|
+ _, _, brief_fn, _ = build_triage(cfg, client)
|
|
|
|
|
+ assert brief_fn is not None
|
|
|
|
|
+ brief_fn("PROMPT")
|
|
|
|
|
+ assert client.calls[0]["slot_id"] == 2
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_build_triage_on_state_refreshes_analyzer(tmp_path):
|
|
def test_build_triage_on_state_refreshes_analyzer(tmp_path):
|
|
@@ -167,7 +186,7 @@ def test_build_triage_on_state_refreshes_analyzer(tmp_path):
|
|
|
|
|
|
|
|
cfg = _enabled_triage_cfg(tmp_path)
|
|
cfg = _enabled_triage_cfg(tmp_path)
|
|
|
sentinel = object()
|
|
sentinel = object()
|
|
|
- analyzer, lane, _ = build_triage(cfg, None, on_result=sentinel)
|
|
|
|
|
|
|
+ analyzer, lane, _, _ = build_triage(cfg, None, on_result=sentinel)
|
|
|
assert analyzer is not None
|
|
assert analyzer is not None
|
|
|
assert lane is not None
|
|
assert lane is not None
|
|
|
assert lane._on_result is sentinel
|
|
assert lane._on_result is sentinel
|
|
@@ -271,7 +290,8 @@ def test_runner_main_shutdown_order_lane_before_control(tmp_path, monkeypatch):
|
|
|
assert made_lanes[0].cfg is cfg.triage
|
|
assert made_lanes[0].cfg is cfg.triage
|
|
|
assert made_services[0].kwargs["laya_lane"] is made_lanes[0]
|
|
assert made_services[0].kwargs["laya_lane"] is made_lanes[0]
|
|
|
assert made_services[0].kwargs["triage"] is not None
|
|
assert made_services[0].kwargs["triage"] is not None
|
|
|
- assert made_services[0].kwargs["triage_extract_fn"] is None
|
|
|
|
|
|
|
+ assert made_services[0].kwargs["brief_fn"] is None
|
|
|
|
|
+ assert made_services[0].kwargs["brief_reset"] is None
|
|
|
assert calls.index("control.start") < calls.index("service.start")
|
|
assert calls.index("control.start") < calls.index("service.start")
|
|
|
assert calls.index("service.start") < calls.index("lane.start")
|
|
assert calls.index("service.start") < calls.index("lane.start")
|
|
|
assert calls.index("service.stop") < calls.index("lane.stop")
|
|
assert calls.index("service.stop") < calls.index("lane.stop")
|
|
@@ -293,5 +313,553 @@ def test_runner_main_disabled_builds_no_lane(monkeypatch):
|
|
|
assert "lane.stop" not in calls
|
|
assert "lane.stop" not in calls
|
|
|
assert made_services[0].kwargs["triage"] is None
|
|
assert made_services[0].kwargs["triage"] is None
|
|
|
assert made_services[0].kwargs["laya_lane"] is None
|
|
assert made_services[0].kwargs["laya_lane"] is None
|
|
|
- assert made_services[0].kwargs["triage_extract_fn"] is None
|
|
|
|
|
|
|
+ assert made_services[0].kwargs["brief_fn"] is None
|
|
|
|
|
+ assert made_services[0].kwargs["brief_reset"] is None
|
|
|
assert not [t for t in threading.enumerate() if t.name.startswith("laya-")]
|
|
assert not [t for t in threading.enumerate() if t.name.startswith("laya-")]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _patch_llm_client(monkeypatch, calls):
|
|
|
|
|
+ from stt_daemon.assistant import runner as runner_mod
|
|
|
|
|
+
|
|
|
|
|
+ class FakeLlmClient:
|
|
|
|
|
+ def __init__(self, *args, **kwargs):
|
|
|
|
|
+ pass
|
|
|
|
|
+
|
|
|
|
|
+ def chat(self, messages, tools=None, max_tokens=None, timeout_s=None, **kwargs):
|
|
|
|
|
+ calls.append({"messages": messages, "kwargs": kwargs})
|
|
|
|
|
+ return SimpleNamespace(content="hi there", finish_reason="stop")
|
|
|
|
|
+
|
|
|
|
|
+ monkeypatch.setattr(runner_mod, "LlmClient", FakeLlmClient)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_runner_wires_chat_engine_and_command_slot(monkeypatch):
|
|
|
|
|
+ """Runner builds the chat conversation from ConversationEngine with the chat
|
|
|
|
|
+ slot, and gives the command executor the configured command slot."""
|
|
|
|
|
+ cfg = Config()
|
|
|
|
|
+ cfg.audit.enabled = False
|
|
|
|
|
+ cfg.llm.enabled = True
|
|
|
|
|
+ cfg.llm.chat_slot_id = 7
|
|
|
|
|
+ cfg.llm.command_slot_id = 3
|
|
|
|
|
+
|
|
|
|
|
+ calls: list[dict] = []
|
|
|
|
|
+ _patch_llm_client(monkeypatch, calls)
|
|
|
|
|
+ rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
|
|
|
|
|
+
|
|
|
|
|
+ assert rc == 0
|
|
|
|
|
+ kwargs = made_services[0].kwargs
|
|
|
|
|
+ engine = kwargs["chat_fn"].__self__
|
|
|
|
|
+ assert engine.system == cfg.llm.system_prompt_chat
|
|
|
|
|
+ assert engine._slot_id == 7
|
|
|
|
|
+ assert kwargs["executor"].slot_id == 3
|
|
|
|
|
+ assert kwargs["chat_fn"]("hello") == "hi there"
|
|
|
|
|
+ assert calls[0]["kwargs"] == {"slot_id": 7}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_runner_wires_slotless_engine_when_slots_unset(monkeypatch):
|
|
|
|
|
+ """Unset slot config leaves both lanes slot-free (getattr fallback)."""
|
|
|
|
|
+ cfg = Config()
|
|
|
|
|
+ cfg.audit.enabled = False
|
|
|
|
|
+ cfg.llm.enabled = True
|
|
|
|
|
+ assert cfg.llm.chat_slot_id is None
|
|
|
|
|
+ assert cfg.llm.command_slot_id is None
|
|
|
|
|
+
|
|
|
|
|
+ calls: list[dict] = []
|
|
|
|
|
+ _patch_llm_client(monkeypatch, calls)
|
|
|
|
|
+ rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
|
|
|
|
|
+
|
|
|
|
|
+ assert rc == 0
|
|
|
|
|
+ kwargs = made_services[0].kwargs
|
|
|
|
|
+ assert kwargs["executor"].slot_id is None
|
|
|
|
|
+ assert kwargs["chat_fn"].__self__._slot_id is None
|
|
|
|
|
+ assert kwargs["chat_fn"]("hello") == "hi there"
|
|
|
|
|
+ assert calls[0]["kwargs"] == {}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_runner_wires_brief_engine_slot_when_configured(monkeypatch, tmp_path):
|
|
|
|
|
+ """The triage brief engine sends `id_slot` only when `brief_slot_id` is set."""
|
|
|
|
|
+ cfg = _enabled_triage_cfg(tmp_path)
|
|
|
|
|
+ cfg.llm.enabled = True
|
|
|
|
|
+ cfg.triage.brief_slot_id = 5
|
|
|
|
|
+
|
|
|
|
|
+ calls: list[dict] = []
|
|
|
|
|
+ _patch_llm_client(monkeypatch, calls)
|
|
|
|
|
+ rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
|
|
|
|
|
+
|
|
|
|
|
+ assert rc == 0
|
|
|
|
|
+ brief_fn = made_services[0].kwargs["brief_fn"]
|
|
|
|
|
+ assert brief_fn is not None
|
|
|
|
|
+ assert brief_fn("BRIEF-PROMPT") == "hi there"
|
|
|
|
|
+ assert calls[0]["kwargs"] == {"slot_id": 5}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_runner_wires_slotless_brief_engine_when_slot_unset(monkeypatch, tmp_path):
|
|
|
|
|
+ """An unset `brief_slot_id` leaves the brief engine slot-free (`id_slot` absent)."""
|
|
|
|
|
+ cfg = _enabled_triage_cfg(tmp_path)
|
|
|
|
|
+ cfg.llm.enabled = True
|
|
|
|
|
+ assert cfg.triage.brief_slot_id is None
|
|
|
|
|
+
|
|
|
|
|
+ calls: list[dict] = []
|
|
|
|
|
+ _patch_llm_client(monkeypatch, calls)
|
|
|
|
|
+ rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
|
|
|
|
|
+
|
|
|
|
|
+ assert rc == 0
|
|
|
|
|
+ brief_fn = made_services[0].kwargs["brief_fn"]
|
|
|
|
|
+ assert brief_fn is not None
|
|
|
|
|
+ assert brief_fn("BRIEF-PROMPT") == "hi there"
|
|
|
|
|
+ assert calls[0]["kwargs"] == {}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_runner_brief_file_writes_when_enabled(monkeypatch, tmp_path):
|
|
|
|
|
+ """`[brief].write=true` makes the runner pass a live, durable writer."""
|
|
|
|
|
+ from stt_daemon.assistant.brief_file import BriefFile
|
|
|
|
|
+
|
|
|
|
|
+ cfg = _enabled_triage_cfg(tmp_path)
|
|
|
|
|
+ cfg.brief.write = True
|
|
|
|
|
+ cfg.brief.dir = str(tmp_path / "brief")
|
|
|
|
|
+ cfg.brief.filename = "b.md"
|
|
|
|
|
+
|
|
|
|
|
+ rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
|
|
|
|
|
+
|
|
|
|
|
+ assert rc == 0
|
|
|
|
|
+ brief_file = made_services[0].kwargs["brief_file"]
|
|
|
|
|
+ assert isinstance(brief_file, BriefFile)
|
|
|
|
|
+ assert brief_file.write_enabled is True
|
|
|
|
|
+ brief_file.write("latest brief")
|
|
|
|
|
+ path = os.path.join(cfg.brief.dir, "b.md")
|
|
|
|
|
+ with open(path, encoding="utf-8") as handle:
|
|
|
|
|
+ assert handle.read() == "latest brief"
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_runner_brief_file_disabled_writes_nothing(monkeypatch, tmp_path):
|
|
|
|
|
+ """The default (`[brief].write=false`) writer is present but creates no file."""
|
|
|
|
|
+ from stt_daemon.assistant.brief_file import BriefFile
|
|
|
|
|
+
|
|
|
|
|
+ cfg = _enabled_triage_cfg(tmp_path)
|
|
|
|
|
+ cfg.brief.write = False
|
|
|
|
|
+ cfg.brief.dir = str(tmp_path / "brief")
|
|
|
|
|
+
|
|
|
|
|
+ rc, _, _, made_services = _run_runner_main(monkeypatch, cfg)
|
|
|
|
|
+
|
|
|
|
|
+ assert rc == 0
|
|
|
|
|
+ brief_file = made_services[0].kwargs["brief_file"]
|
|
|
|
|
+ assert isinstance(brief_file, BriefFile)
|
|
|
|
|
+ assert brief_file.write_enabled is False
|
|
|
|
|
+ brief_file.write("nope")
|
|
|
|
|
+ assert not os.path.exists(os.path.join(cfg.brief.dir, cfg.brief.filename))
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
|
|
+# End-to-end dual-engine coverage (chat + triage on ONE inference queue).
|
|
|
|
|
+#
|
|
|
|
|
+# These tests build the REAL `AssistantService` plus the real `build_triage`
|
|
|
|
|
+# wiring (real `TriageAnalyzer`, real `ConversationEngine` brief engine, real
|
|
|
|
|
+# `InferenceQueue`, real `BriefFile`) and only fake the LLM client and the bus.
|
|
|
|
|
+# Laya results are delivered as fake payloads through the service's documented
|
|
|
|
|
+# `_on_triage_result` seam; the queued jobs are then driven for real via
|
|
|
|
|
+# `job.run()` (the queue's worker body).
|
|
|
|
|
+# ---------------------------------------------------------------------------
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class _Clock:
|
|
|
|
|
+ """Deterministic clock for the integration harness."""
|
|
|
|
|
+
|
|
|
|
|
+ def __init__(self, start: float = 1000.0) -> None:
|
|
|
|
|
+ self.t = float(start)
|
|
|
|
|
+
|
|
|
|
|
+ def __call__(self) -> float:
|
|
|
|
|
+ return self.t
|
|
|
|
|
+
|
|
|
|
|
+ def advance(self, dt: float) -> None:
|
|
|
|
|
+ self.t += float(dt)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class _SilentSubscriber:
|
|
|
|
|
+ def events(self):
|
|
|
|
|
+ return iter(())
|
|
|
|
|
+
|
|
|
|
|
+ def close(self) -> None:
|
|
|
|
|
+ pass
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class _EngineClient:
|
|
|
|
|
+ """Recording LLM client: brief replies differ from chat replies.
|
|
|
|
|
+
|
|
|
|
|
+ Accepts `slot_id` like the real client and always returns a `finish_reason`
|
|
|
|
|
+ (the brief engine reads it). `brief_error` makes the brief call raise, which
|
|
|
|
|
+ the real `_run_brief_update` must absorb.
|
|
|
|
|
+ """
|
|
|
|
|
+
|
|
|
|
|
+ def __init__(
|
|
|
|
|
+ self, *, brief_reply: str = "BRIEF", brief_error: bool = False,
|
|
|
|
|
+ chat_reply: str = "chat reply",
|
|
|
|
|
+ ) -> None:
|
|
|
|
|
+ self.calls: list[dict] = []
|
|
|
|
|
+ self.brief_reply = brief_reply
|
|
|
|
|
+ self.brief_error = brief_error
|
|
|
|
|
+ self.chat_reply = chat_reply
|
|
|
|
|
+ self.brief_system: str | None = None
|
|
|
|
|
+
|
|
|
|
|
+ def chat(self, messages, tools=None, max_tokens=None, timeout_s=None, slot_id=None):
|
|
|
|
|
+ system = messages[0].get("content") if messages else ""
|
|
|
|
|
+ is_brief = self.brief_system is not None and system == self.brief_system
|
|
|
|
|
+ self.calls.append(
|
|
|
|
|
+ {
|
|
|
|
|
+ "is_brief": is_brief,
|
|
|
|
|
+ "system": system,
|
|
|
|
|
+ "user": messages[-1].get("content") if messages else "",
|
|
|
|
|
+ "slot_id": slot_id,
|
|
|
|
|
+ "max_tokens": max_tokens,
|
|
|
|
|
+ "timeout_s": timeout_s,
|
|
|
|
|
+ }
|
|
|
|
|
+ )
|
|
|
|
|
+ if is_brief and self.brief_error:
|
|
|
|
|
+ raise RuntimeError("llm down")
|
|
|
|
|
+ content = self.brief_reply if is_brief else self.chat_reply
|
|
|
|
|
+ return SimpleNamespace(content=content, finish_reason="stop")
|
|
|
|
|
+
|
|
|
|
|
+ def brief_calls(self) -> list[dict]:
|
|
|
|
|
+ return [call for call in self.calls if call["is_brief"]]
|
|
|
|
|
+
|
|
|
|
|
+ def chat_calls(self) -> list[dict]:
|
|
|
|
|
+ return [call for call in self.calls if not call["is_brief"]]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _laya_payload() -> dict:
|
|
|
|
|
+ """A realistic fake Laya result for the `urgent` noul question."""
|
|
|
|
|
+ return {"answers": {"urgent": {"noul": 0.9, "confidence": 0.9, "answer_confidence": 0.9}}}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _engine_service(
|
|
|
|
|
+ tmp_path,
|
|
|
|
|
+ client,
|
|
|
|
|
+ *,
|
|
|
|
|
+ chat_slot=None,
|
|
|
|
|
+ brief_slot=None,
|
|
|
|
|
+ brief_write: bool = False,
|
|
|
|
|
+):
|
|
|
|
|
+ """Build the real service + `build_triage` wiring over a fake client."""
|
|
|
|
|
+ cfg = _enabled_triage_cfg(tmp_path)
|
|
|
|
|
+ cfg.llm.enabled = True
|
|
|
|
|
+ cfg.assistant.turn_gap_s = 0.5
|
|
|
|
|
+ cfg.triage.brief_gate_enabled = False # deterministic: dispatch every turn
|
|
|
|
|
+ cfg.triage.brief_slot_id = brief_slot
|
|
|
|
|
+ cfg.llm.chat_slot_id = chat_slot
|
|
|
|
|
+ cfg.brief.write = brief_write
|
|
|
|
|
+ cfg.brief.dir = str(tmp_path / "brief")
|
|
|
|
|
+ cfg.brief.filename = "brief.md"
|
|
|
|
|
+ client.brief_system = cfg.llm.triage_system_prompt
|
|
|
|
|
+
|
|
|
|
|
+ analyzer, lane, brief_fn, brief_reset = build_triage(cfg, client)
|
|
|
|
|
+ chat_engine = ConversationEngine(
|
|
|
|
|
+ cfg.llm,
|
|
|
|
|
+ client,
|
|
|
|
|
+ system_prompt=cfg.llm.system_prompt_chat,
|
|
|
|
|
+ slot_id=chat_slot,
|
|
|
|
|
+ )
|
|
|
|
|
+ emitted: list[dict] = []
|
|
|
|
|
+ clock = _Clock()
|
|
|
|
|
+ brief_file = BriefFile(cfg.brief.write, cfg.brief.dir, cfg.brief.filename)
|
|
|
|
|
+ service = AssistantService(
|
|
|
|
|
+ cfg,
|
|
|
|
|
+ _SilentSubscriber(),
|
|
|
|
|
+ emit=lambda type_, **fields: emitted.append({"type": type_, **fields}),
|
|
|
|
|
+ time_fn=clock,
|
|
|
|
|
+ triage=analyzer,
|
|
|
|
|
+ laya_lane=lane,
|
|
|
|
|
+ brief_fn=brief_fn,
|
|
|
|
|
+ brief_reset=brief_reset,
|
|
|
|
|
+ brief_file=brief_file,
|
|
|
|
|
+ chat_fn=chat_engine.send,
|
|
|
|
|
+ chat_reset=chat_engine.reset,
|
|
|
|
|
+ )
|
|
|
|
|
+ return SimpleNamespace(
|
|
|
|
|
+ service=service,
|
|
|
|
|
+ emitted=emitted,
|
|
|
|
|
+ clock=clock,
|
|
|
|
|
+ analyzer=analyzer,
|
|
|
|
|
+ lane=lane,
|
|
|
|
|
+ chat_engine=chat_engine,
|
|
|
|
|
+ brief_file=brief_file,
|
|
|
|
|
+ cfg=cfg,
|
|
|
|
|
+ client=client,
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _transcript(clock, text):
|
|
|
|
|
+ return {"v": 1, "type": events.TRANSCRIPT, "seq": 1, "ts": clock(), "text": text}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _triage_events(emitted):
|
|
|
|
|
+ return [event for event in emitted if event["type"] == "triage"]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _stages(emitted):
|
|
|
|
|
+ return [event.get("stage") for event in _triage_events(emitted)]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _brief_jobs(service):
|
|
|
|
|
+ return [job for job in service._inference._queue if job.name == "brief-update"]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _chat_jobs(service):
|
|
|
|
|
+ return [job for job in service._inference._queue if job.name == "chat"]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _run_queued(service, job):
|
|
|
|
|
+ """Pop a queued job (mirroring the worker) and run it."""
|
|
|
|
|
+ service._inference._queue.remove(job)
|
|
|
|
|
+ job.run()
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _settle(service, clock, text):
|
|
|
|
|
+ service.handle_event(_transcript(clock, text))
|
|
|
|
|
+ clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _deliver(service, payload):
|
|
|
|
|
+ service._on_triage_result(service._triage_generation, payload)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_two_engines_emit_signals_then_brief_and_chat_works(tmp_path):
|
|
|
|
|
+ """(a) signals then content.brief; (b) chat works while triage is active;
|
|
|
|
|
+ (c) each engine carries id_slot only when set."""
|
|
|
|
|
+ client = _EngineClient(brief_reply="E2E BRIEF")
|
|
|
|
|
+ h = _engine_service(tmp_path, client, chat_slot=7, brief_slot=5)
|
|
|
|
|
+
|
|
|
|
|
+ assert h.service.toggle_chat() is True
|
|
|
|
|
+ _settle(h.service, h.clock, "first thought")
|
|
|
|
|
+ h.service.tick() # dispatches the chat turn through the real queue
|
|
|
|
|
+
|
|
|
|
|
+ _deliver(h.service, _laya_payload())
|
|
|
|
|
+ assert _stages(h.emitted) == ["signals"]
|
|
|
|
|
+ assert h.service.last_triage is not None
|
|
|
|
|
+ assert "brief" not in h.service.last_triage
|
|
|
|
|
+
|
|
|
|
|
+ chat_jobs = _chat_jobs(h.service)
|
|
|
|
|
+ assert len(chat_jobs) == 1
|
|
|
|
|
+ _run_queued(h.service, chat_jobs[0]) # chat works while triage is active
|
|
|
|
|
+ assert h.service.last_triage is not None # triage still owns its state
|
|
|
|
|
+ assert any(
|
|
|
|
|
+ event["type"] == "chat_message"
|
|
|
|
|
+ and event.get("role") == "assistant"
|
|
|
|
|
+ and event.get("text") == "chat reply"
|
|
|
|
|
+ for event in h.emitted
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ brief_jobs = _brief_jobs(h.service)
|
|
|
|
|
+ assert len(brief_jobs) == 1
|
|
|
|
|
+ _run_queued(h.service, brief_jobs[0])
|
|
|
|
|
+
|
|
|
|
|
+ assert _stages(h.emitted) == ["signals", "content"]
|
|
|
|
|
+ contents = [event for event in _triage_events(h.emitted) if event.get("stage") == "content"]
|
|
|
|
|
+ assert contents[0]["triage"]["brief"] == "E2E BRIEF"
|
|
|
|
|
+ assert h.service.last_triage["brief"] == "E2E BRIEF"
|
|
|
|
|
+
|
|
|
|
|
+ assert [call["slot_id"] for call in client.chat_calls()] == [7]
|
|
|
|
|
+ assert [call["slot_id"] for call in client.brief_calls()] == [5]
|
|
|
|
|
+ assert client.brief_calls()[0]["user"].startswith("New utterances:")
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_slotless_engines_omit_id_slot(tmp_path):
|
|
|
|
|
+ """(c) unset slots leave `id_slot` off both engines' calls."""
|
|
|
|
|
+ client = _EngineClient(brief_reply="BRIEF")
|
|
|
|
|
+ h = _engine_service(tmp_path, client, chat_slot=None, brief_slot=None)
|
|
|
|
|
+
|
|
|
|
|
+ assert h.service.toggle_chat() is True
|
|
|
|
|
+ _settle(h.service, h.clock, "hello")
|
|
|
|
|
+ h.service.tick()
|
|
|
|
|
+ _deliver(h.service, _laya_payload())
|
|
|
|
|
+ _run_queued(h.service, _chat_jobs(h.service)[0])
|
|
|
|
|
+ _run_queued(h.service, _brief_jobs(h.service)[0])
|
|
|
|
|
+
|
|
|
|
|
+ assert [call["slot_id"] for call in client.chat_calls()] == [None]
|
|
|
|
|
+ assert [call["slot_id"] for call in client.brief_calls()] == [None]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_refused_brief_keeps_deltas_and_next_dispatch_delivers(tmp_path):
|
|
|
|
|
+ """(d) a full queue refuses the brief without losing pending deltas, and the
|
|
|
|
|
+ next dispatch delivers them all."""
|
|
|
|
|
+ client = _EngineClient(brief_reply="BRIEF")
|
|
|
|
|
+ h = _engine_service(tmp_path, client)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+ service._inference = InferenceQueue(size=2)
|
|
|
|
|
+ service._inference.submit(InferenceJob(name="blocker", run=lambda: None, evictable=False))
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+
|
|
|
|
|
+ # Refused: no brief job admitted, but the delta is buffered and the flag clear.
|
|
|
|
|
+ assert _brief_jobs(service) == []
|
|
|
|
|
+ assert service._pending_brief == ["Unknown: first thought"]
|
|
|
|
|
+ assert service._brief_outstanding is False
|
|
|
|
|
+ assert _stages(h.emitted) == ["signals"]
|
|
|
|
|
+
|
|
|
|
|
+ # Free room; the next settled turn re-dispatches and consumes both deltas.
|
|
|
|
|
+ service._inference._queue.clear()
|
|
|
|
|
+ _settle(service, h.clock, "second thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ jobs = _brief_jobs(service)
|
|
|
|
|
+ assert len(jobs) == 1
|
|
|
|
|
+ assert service._pending_brief == ["Unknown: first thought", "Unknown: second thought"]
|
|
|
|
|
+
|
|
|
|
|
+ _run_queued(service, jobs[0])
|
|
|
|
|
+
|
|
|
|
|
+ assert _stages(h.emitted) == ["signals", "signals", "content"]
|
|
|
|
|
+ prompt = client.brief_calls()[-1]["user"]
|
|
|
|
|
+ assert "- Unknown: first thought" in prompt
|
|
|
|
|
+ assert "- Unknown: second thought" in prompt
|
|
|
|
|
+ assert service._pending_brief == []
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_empty_brief_requeues_deltas_and_clears_flag(tmp_path):
|
|
|
|
|
+ """(e) a `""` brief re-queues the deltas and clears the outstanding flag."""
|
|
|
|
|
+ client = _EngineClient(brief_reply="")
|
|
|
|
|
+ h = _engine_service(tmp_path, client)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ _run_queued(service, _brief_jobs(service)[0])
|
|
|
|
|
+
|
|
|
|
|
+ assert _stages(h.emitted) == ["signals"] # no content emitted
|
|
|
|
|
+ assert service._pending_brief == ["Unknown: first thought"]
|
|
|
|
|
+ assert service._brief_outstanding is False
|
|
|
|
|
+ assert _brief_jobs(service) == [] # requeued, not flushed
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_raising_brief_requeues_deltas_and_clears_flag(tmp_path):
|
|
|
|
|
+ """(e) a raising brief leaves no stranded flag and re-buffers the deltas."""
|
|
|
|
|
+ client = _EngineClient(brief_error=True)
|
|
|
|
|
+ h = _engine_service(tmp_path, client)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ _run_queued(service, _brief_jobs(service)[0])
|
|
|
|
|
+
|
|
|
|
|
+ assert _stages(h.emitted) == ["signals"]
|
|
|
|
|
+ assert service._pending_brief == ["Unknown: first thought"]
|
|
|
|
|
+ assert service._brief_outstanding is False
|
|
|
|
|
+ assert _brief_jobs(service) == []
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_none_laya_payload_does_not_dispatch(tmp_path):
|
|
|
|
|
+ """(f) a None Laya payload stores nothing and never dispatches the brief."""
|
|
|
|
|
+ client = _EngineClient()
|
|
|
|
|
+ h = _engine_service(tmp_path, client)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, None)
|
|
|
|
|
+
|
|
|
|
|
+ assert _triage_events(h.emitted) == []
|
|
|
|
|
+ assert service.last_triage is None
|
|
|
|
|
+ assert _brief_jobs(service) == []
|
|
|
|
|
+ assert service._pending_brief == ["Unknown: first thought"]
|
|
|
|
|
+ assert client.calls == []
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_clear_resets_both_engines_and_stale_job_cannot_clobber(tmp_path):
|
|
|
|
|
+ """(g) clear resets chat + triage, suppresses stale emits, and a stale job
|
|
|
|
|
+ cannot clear a newer dispatch's flag."""
|
|
|
|
|
+ client = _EngineClient()
|
|
|
|
|
+ h = _engine_service(tmp_path, client)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+
|
|
|
|
|
+ assert service.toggle_chat() is True
|
|
|
|
|
+ _settle(service, h.clock, "hello")
|
|
|
|
|
+ service.tick()
|
|
|
|
|
+ _run_queued(service, _chat_jobs(service)[0])
|
|
|
|
|
+ assert h.chat_engine.history != []
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ stale_job = _brief_jobs(service)[0]
|
|
|
|
|
+ assert service._brief_outstanding is True
|
|
|
|
|
+
|
|
|
|
|
+ service.clear()
|
|
|
|
|
+
|
|
|
|
|
+ # Both engines reset.
|
|
|
|
|
+ assert h.chat_engine.history == []
|
|
|
|
|
+ assert service.chat_history == []
|
|
|
|
|
+ assert service._triage_generation == 1
|
|
|
|
|
+ assert service.last_triage is None
|
|
|
|
|
+ assert service.triage_history == []
|
|
|
|
|
+ assert service._pending_brief == []
|
|
|
|
|
+ assert service._brief_outstanding is False
|
|
|
|
|
+ assert service._last_brief_signals is None
|
|
|
|
|
+ assert _brief_jobs(service) == []
|
|
|
|
|
+ assert "cleared" in _stages(h.emitted)
|
|
|
|
|
+ assert {"type": "cleared"} in h.emitted
|
|
|
|
|
+
|
|
|
|
|
+ h.emitted.clear()
|
|
|
|
|
+ stale_job.run() # clear() already dropped it from the queue
|
|
|
|
|
+ assert _triage_events(h.emitted) == []
|
|
|
|
|
+ assert service._pending_brief == []
|
|
|
|
|
+ assert service.last_triage is None
|
|
|
|
|
+
|
|
|
|
|
+ # A newer dispatch after clear owns the flag; the stale job cannot clear it.
|
|
|
|
|
+ _settle(service, h.clock, "fresh")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ newer_job = _brief_jobs(service)[0]
|
|
|
|
|
+ newer_token = service._brief_token
|
|
|
|
|
+ pending_after = list(service._pending_brief)
|
|
|
|
|
+ triage_after = service.last_triage
|
|
|
|
|
+ assert service._brief_outstanding is True
|
|
|
|
|
+
|
|
|
|
|
+ h.emitted.clear()
|
|
|
|
|
+ stale_job.run()
|
|
|
|
|
+ assert _triage_events(h.emitted) == []
|
|
|
|
|
+ assert service._brief_outstanding is True
|
|
|
|
|
+ assert service._brief_token == newer_token
|
|
|
|
|
+ assert service._pending_brief == pending_after
|
|
|
|
|
+ assert service.last_triage is triage_after
|
|
|
|
|
+
|
|
|
|
|
+ _run_queued(service, newer_job)
|
|
|
|
|
+ assert _stages(h.emitted) == ["content"]
|
|
|
|
|
+ assert service._brief_outstanding is False
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_durable_brief_file_holds_latest_and_clears(tmp_path):
|
|
|
|
|
+ """(h) with `[brief].write=true` the file holds the exact latest brief and
|
|
|
|
|
+ is empty after `clear`."""
|
|
|
|
|
+ client = _EngineClient(brief_reply="FIRST BRIEF")
|
|
|
|
|
+ h = _engine_service(tmp_path, client, brief_write=True)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+ path = os.path.join(h.cfg.brief.dir, h.cfg.brief.filename)
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ _run_queued(service, _brief_jobs(service)[0])
|
|
|
|
|
+ with open(path, encoding="utf-8") as handle:
|
|
|
|
|
+ assert handle.read() == "FIRST BRIEF"
|
|
|
|
|
+
|
|
|
|
|
+ client.brief_reply = "LATEST BRIEF"
|
|
|
|
|
+ _settle(service, h.clock, "second thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ _run_queued(service, _brief_jobs(service)[0])
|
|
|
|
|
+ with open(path, encoding="utf-8") as handle:
|
|
|
|
|
+ assert handle.read() == "LATEST BRIEF"
|
|
|
|
|
+ assert service.last_triage["brief"] == "LATEST BRIEF"
|
|
|
|
|
+
|
|
|
|
|
+ service.clear()
|
|
|
|
|
+ assert os.path.exists(path)
|
|
|
|
|
+ with open(path, encoding="utf-8") as handle:
|
|
|
|
|
+ assert handle.read() == ""
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_disabled_brief_file_absent_but_panel_updates(tmp_path):
|
|
|
|
|
+ """(h) with `[brief].write=false` no file appears while the panel brief still
|
|
|
|
|
+ updates."""
|
|
|
|
|
+ client = _EngineClient(brief_reply="PANEL BRIEF")
|
|
|
|
|
+ h = _engine_service(tmp_path, client, brief_write=False)
|
|
|
|
|
+ service = h.service
|
|
|
|
|
+ path = os.path.join(h.cfg.brief.dir, h.cfg.brief.filename)
|
|
|
|
|
+
|
|
|
|
|
+ _settle(service, h.clock, "first thought")
|
|
|
|
|
+ _deliver(service, _laya_payload())
|
|
|
|
|
+ _run_queued(service, _brief_jobs(service)[0])
|
|
|
|
|
+
|
|
|
|
|
+ assert h.brief_file.write_enabled is False
|
|
|
|
|
+ assert not os.path.exists(path)
|
|
|
|
|
+ assert service.last_triage["brief"] == "PANEL BRIEF"
|