Added periodic trigger for accumulating speech and triage debug signals

config.example.toml

@@ -99,6 +99,7 @@ enabled                   = false
 questions_file            = "questions.json"
 window                    = 500
 state_char_budget         = 3000
+periodic_s                = 30.0
 predict_max_tokens        = 1024
 confidence_field          = "answer_confidence"
 escalate_threshold        = 0.6

docs/CONFIGURATION.md

@@ -123,6 +123,7 @@ enabled              = false                # Master gate: false = no lane, no p
 questions_file       = "questions.json"     # Laya questions JSON in the cwd; missing/invalid -> no_questions
 window               = 500                    # Utterances available per classification; the char budget is the real limiter
 state_char_budget    = 3000                 # Char cap (~750 tokens) for the condensed state sent to Laya
+periodic_s           = 30.0                 # Accumulating-speech cadence (s): while utterances keep arriving without a settled turn, re-run triage every N seconds; <= 0 disables
 predict_max_tokens   = 1024                 # Max tokens sent to Laya; the multilingual checkpoint supports 1024; state shares this budget with the ~192-token question head (~830-token state room)
 confidence_field     = "answer_confidence"  # Gating confidence: "answer_confidence" (calibrated max(p)) | "confidence"
 escalate_threshold   = 0.6                  # noul answers at or above this p are flagged (⚠ marker only)
@@ -209,8 +210,12 @@ chat, the assistant classifies the last `window` transcribed utterances (from th
 not the chat history) with a local Laya **multilingual** checkpoint on a dedicated
 in-process lane, then, when `extract_enabled` is true, asks the `[llm]` lane on a dedicated
 brief conversation to fold the turn into a living plain-text brief (on a Laya error the prompt
-carries transcript state only). `extract_enabled` stays the brief engine's master switch: with
-it false the panel still classifies but no brief update is requested.
+carries transcript state only). A settled turn means speech stopped plus quiet for
+`[assistant].turn_gap_s`; because a long unbroken monologue never settles, `periodic_s`
+(default 30 seconds) additionally re-runs classification every N seconds while new utterances
+keep arriving (a repeated window is still skipped), so the panel stays current mid-stream. Set
+`periodic_s = 0` to require a settled turn again. `extract_enabled` stays the brief engine's
+master switch: with it false the panel still classifies but no brief update is requested.
 `device = ""` means auto: it maps to `None`, so Laya picks the
 default device for the multilingual checkpoint. `preload = ["multilingual"]` is the fixed default
 and `default_checkpoint = "multilingual"`; the multilingual checkpoint is pinned for every call
@@ -256,7 +261,9 @@ object per line, `ensure_ascii=False`) so the next silent freeze is diagnosable
 Signals only — no transcript text, no audio, no prompt content: RMS/queue/timing fields such
 as `device_ok`, `worker_stall_s`, `queue_depth`, `inference_depth`, plus anomaly lines for
 capture reopens, inference-queue `queue-full` / `eviction-drop`, `llm-timeout`, worker
-`worker-stall`, triage `triage-brief-failure`, and supervisor/child exit lines. With
+`worker-stall`, triage lifecycle (`triage-submit` / `triage-skip` / `triage-result` /
+`triage-brief` with a `trigger`/`reason`, plus `triage-brief-failure`), and supervisor/child
+exit lines. With
 `enabled = false` (the default) no sink is built and the suite stays disk-clean; with
 `heartbeat = false` only anomalies are written. `max_bytes_per_day` is a best-effort guard:
 writes past the cap are skipped with one warning (never crash, never rotate). `debug/` lives

docs/TRIAGE.md

@@ -43,7 +43,7 @@ disk, and it is a no-op when `[triage].enabled = false`.
 ## How the brief works
 
 ```
-settled turn (any mode)
+settled turn (any mode), or every periodic_s while speech keeps accumulating
     -> newest transcribed-utterance window (window entries, condensed under state_char_budget)
     -> Laya multilingual classification (in-process lane)
     -> signals (emitted; flagged per should_escalate)
@@ -57,8 +57,12 @@ settled turn (any mode)
 1. **Trigger.** After the chat dispatch block in the assistant tick, triage runs on any
    settled turn in any mode (idle, chat, or command), independent of chat. It runs when
    the triage components are wired and the turn has ended (`turn_ended()`: speech stopped
-   plus quiet for `turn_gap_s`). A repeated window is skipped, so an identical set of
-   utterances never re-gates.
+   plus quiet for `turn_gap_s`). Because continuous speech never produces a settled turn,
+   triage also re-runs every `periodic_s` seconds (default `30`) while new utterances keep
+   arriving: the timer arms on the first transcript of a burst and re-arms on every accepted
+   submit, so a long monologue is classified roughly every `periodic_s` instead of only when
+   it stops. Set `periodic_s = 0` to disable the periodic trigger and require a settled turn.
+   A repeated window is skipped, so an identical set of utterances never re-gates.
 2. **Window.** The window is the newest `window` transcribed utterances from the STT
    stream, not the chat conversation history. `build_state` condenses it newest-first
    under `state_char_budget` and restores chronological order, with `- ` prefixed lines.
@@ -273,6 +277,7 @@ file.
 | `questions_file` | `"questions.json"` | Laya questions JSON in the working directory. Missing or invalid means `no_questions`. |
 | `window` | `500` | Recent transcribed utterances available per classification; the char budget is the real limiter. |
 | `state_char_budget` | `3000` | Character cap (~750 tokens) for the condensed state sent to Laya. |
+| `periodic_s` | `30.0` | Accumulating-speech cadence in seconds: while new utterances keep arriving without a settled turn, triage re-runs every `periodic_s`; `<= 0` disables the periodic trigger. |
 | `predict_max_tokens` | `1024` | Max tokens sent to Laya; the multilingual checkpoint supports 1024; state shares this budget with the ~192-token question head (~830-token state room). |
 | `confidence_field` | `"answer_confidence"` | Which Laya confidence feeds the per-item `flagged`/`⚠` marker: `answer_confidence` (calibrated `max(p)`) or `confidence` (normalized entropy). |
 | `escalate_threshold` | `0.6` | A `noul` answer at or above this `p` is flagged (`⚠` marker only). |

docs/TROUBLESHOOTING.md

@@ -149,6 +149,12 @@ Read the tail in this order:
    `uv run stt-daemon up --with-tui` and check which child exited first.
 5. **Assistant `queue-full` / `eviction-drop` / `llm-timeout` / `triage-brief-failure`**:
    backpressure or LLM-side stalls on the assistant lane, not a capture problem.
+6. **Triage lifecycle `triage-submit` / `triage-skip` / `triage-result` / `triage-brief`**:
+   why triage did or did not run. `triage-submit.trigger` is `settled` or `periodic` (a long
+   monologue fires `periodic`), `triage-skip.reason` is `empty_state` / `lane_refused`, and
+   `triage-brief.action` is `dispatch`, `refuse` (`stale` / `outstanding`), or `skip`
+   (`gate` / `extract_disabled` / `no_brief_fn`). A run that only ever shows `triage-submit`
+   with no `triage-result` means the Laya predict is still loading or stalled.
 
 When the system is healthy again, clear the log manually (`rm debug/*`); there is no
 automatic retention job. Keep `debug/` on local storage — a slow/NFS/FUSE mount can block

src/stt_daemon/assistant/service.py

@@ -160,6 +160,11 @@ class AssistantService:
         self._last_triage_state: tuple[str, str] | None = None
         self.last_triage: dict | None = None
         self._last_triage_window: tuple = ()
+        # Periodic cadence for accumulating speech (no settled turn): armed to
+        # `now + periodic_s` on the first transcript of a burst and re-armed on
+        # every accepted submit. None while disarmed (periodic_s <= 0 or after a
+        # reset), so a long monologue still triages every `periodic_s`.
+        self._triage_next_due: float | None = None
         self._triage_generation = 0
         self._last_transcript_lang: str | None = None
         self._triage_dump = TriageDumper(bool(getattr(cfg.triage, "dump_io", False)))
@@ -800,6 +805,9 @@ class AssistantService:
             with self._lock:
                 self._pending_brief.append(f"{self._speaker_display(event)}: {text}")
                 self._cap_pending_brief_locked()
+                periodic_s = self._periodic_s()
+                if periodic_s > 0 and self._triage_next_due is None:
+                    self._triage_next_due = now + periodic_s
 
         if self.modes.mode is Mode.COMMAND_ARMED:
             ts = float(event.get("ts", now))
@@ -904,28 +912,54 @@ class AssistantService:
         if reply:
             self._emit(CHAT_MESSAGE_EVENT, role="assistant", text=reply)
 
+    def _periodic_s(self) -> float:
+        """Accumulating-speech cadence; <= 0 disables the periodic trigger."""
+        try:
+            return float(getattr(self.cfg.triage, "periodic_s", 0.0))
+        except (TypeError, ValueError):
+            return 0.0
+
     def _maybe_triage(self) -> None:
-        """Classify the recent transcribed utterances on any settled turn (independent of chat)."""
-        if not self.turn_ended():
-            return
+        """Classify recent utterances on a settled turn, or periodically while speech grows."""
         questions = self._triage.questions
         if not questions:
             return
         if not self._laya_lane.ready():
             return
+        now = self._now()
+        settled = self.turn_ended()
         with self._lock:
+            periodic_s = self._periodic_s()
+            periodic_due = (
+                periodic_s > 0
+                and self._triage_next_due is not None
+                and now >= self._triage_next_due
+            )
+            if not settled and not periodic_due:
+                return
             window = tuple(self.triage_history[-int(self.cfg.triage.window):])
             if not window or window == self._last_triage_window:
                 return
             generation = self._triage_generation
             self._last_triage_window = window
+        trigger = "settled" if settled else "periodic"
         state = self._triage.build_state(window)
         if not state:
+            self._debug_emit("triage-skip", reason="empty_state", generation=generation)
             return
         accepted = self._laya_lane.submit(
             state, questions, self._last_transcript_lang, generation
         )
-        if accepted and self._triage_dump.enabled:
+        if not accepted:
+            self._debug_emit("triage-skip", reason="lane_refused", generation=generation)
+            return
+        with self._lock:
+            if periodic_s > 0:
+                self._triage_next_due = now + periodic_s
+        self._debug_emit(
+            "triage-submit", trigger=trigger, window=len(window), generation=generation
+        )
+        if self._triage_dump.enabled:
             # Newest-wins caveat: the lane is single-flight and does not echo
             # a request id, so under two submits within one predict the paired
             # request is the most recent submit (acceptable for debug I/O).
@@ -954,10 +988,12 @@ class AssistantService:
                 },
             )
         if payload is None:
+            self._debug_emit("triage-result", error="predict_failed", generation=generation)
             self._triage_dump.finalize(ts)
             return
         signals = self._triage.normalize(payload)
         if not signals:
+            self._debug_emit("triage-result", error="empty_signals", generation=generation)
             self._triage_dump.finalize(ts)
             return
         with self._lock:
@@ -970,6 +1006,12 @@ class AssistantService:
             self.last_triage = merged
             self._last_brief_signals = signals
             self._emit(TRIAGE_EVENT, stage="signals", triage=merged, ts=ts)
+        self._debug_emit(
+            "triage-result",
+            items=len(signals.get("order", [])),
+            escalated=bool(signals.get("escalated")),
+            generation=generation,
+        )
         cfg = self.cfg.triage
         gated_in = not getattr(cfg, "brief_gate_enabled", True) or (
             self._triage.should_update_brief(signals)
@@ -981,6 +1023,15 @@ class AssistantService:
         ):
             self._dispatch_brief(signals, ts, generation)
         else:
+            if self._brief_fn is None:
+                reason = "no_brief_fn"
+            elif not getattr(cfg, "extract_enabled", True):
+                reason = "extract_disabled"
+            else:
+                reason = "gate"
+            self._debug_emit(
+                "triage-brief", action="skip", reason=reason, generation=generation
+            )
             self._triage_dump.finalize(ts)
 
     def _dispatch_brief(self, signals, ts, generation) -> bool:
@@ -1005,6 +1056,9 @@ class AssistantService:
                     run=lambda: self._run_brief_update(generation, signals, ts, token),
                 )
         if refusal is not None:
+            self._debug_emit(
+                "triage-brief", action="refuse", reason=refusal, generation=generation
+            )
             self._triage_dump.finalize(ts)
             return False
         assert job is not None and token is not None
@@ -1015,6 +1069,7 @@ class AssistantService:
             self._triage_dump.finalize(ts)
             self._debug_emit("queue-full", job="brief-update")
             return False
+        self._debug_emit("triage-brief", action="dispatch", generation=generation)
         return True
 
     def _run_brief_update(self, generation, signals, ts=None, token=None) -> None:
@@ -1127,6 +1182,7 @@ class AssistantService:
         self._triage_generation += 1
         self.last_triage = None
         self._last_triage_window = ()
+        self._triage_next_due = None
         self.triage_history = []
         self._brief_token += 1
         self._brief_outstanding = False

src/stt_daemon/config.py

@@ -179,6 +179,7 @@ class TriageConfig:
     questions_file: str = "questions.json"
     window: int = 500
     state_char_budget: int = 3000
+    periodic_s: float = 30.0
     predict_max_tokens: int = 1024
     confidence_field: str = "answer_confidence"
     escalate_threshold: float = 0.6
@@ -534,6 +535,7 @@ def load_config(path: str | None = None) -> Config:
                 state_char_budget=triage_data.get(
                     "state_char_budget", cfg.triage.state_char_budget
                 ),
+                periodic_s=triage_data.get("periodic_s", cfg.triage.periodic_s),
                 predict_max_tokens=triage_data.get(
                     "predict_max_tokens", cfg.triage.predict_max_tokens
                 ),

tests/test_assistant_service.py

@@ -647,6 +647,14 @@ class FakeLane:
         self.drops += 1
 
 
+class FakeDebugSink:
+    def __init__(self):
+        self.events: list[tuple] = []
+
+    def emit(self, type_, **fields):
+        self.events.append((type_, fields))
+
+
 def _ok_reply(text):
     return "ok"
 
@@ -659,6 +667,7 @@ def _triage_service(
     brief_file=None,
     brief_reset=None,
     chat_reset=None,
+    debug_sink=None,
 ):
     cfg = Config()
     cfg.assistant.turn_gap_s = 0.5
@@ -674,6 +683,7 @@ def _triage_service(
         lane = FakeLane()
     if chat_fn is _AUTO:
         chat_fn = _ok_reply
+    factory = (lambda: debug_sink) if debug_sink is not None else None
     service = AssistantService(
         cfg,
         FakeSubscriber(),
@@ -686,6 +696,7 @@ def _triage_service(
         brief_file=brief_file,
         chat_fn=chat_fn,
         chat_reset=chat_reset,
+        debug_sink_factory=factory,
     )
     return service, emitted, clock, triage, lane
 
@@ -816,6 +827,118 @@ def test_maybe_triage_requires_settled_turn():
     assert lane.submits == []
 
 
+def test_maybe_triage_periodic_fires_before_turn_settles():
+    service, _, clock, _, lane = _triage_service()
+    service.cfg.triage.periodic_s = 0.2
+    service.handle_event(_transcript(clock, "hello"))
+    assert service.turn_ended() is False
+
+    clock.advance(0.25)  # due (0.2) yet still quieter than turn_gap_s (0.5)
+    service.tick()
+
+    assert service.turn_ended() is False
+    assert len(lane.submits) == 1
+    assert lane.submits[0][0] == "- hello"
+
+
+def test_maybe_triage_periodic_disabled_ignores_accumulating_speech():
+    service, _, clock, _, lane = _triage_service()
+    service.cfg.triage.periodic_s = 0.0
+    service.handle_event(_transcript(clock, "hello"))
+
+    clock.advance(0.25)
+    service.tick()
+
+    assert service.turn_ended() is False
+    assert lane.submits == []
+
+
+def test_maybe_triage_periodic_rearms_after_submit():
+    service, _, clock, _, lane = _triage_service()
+    service.cfg.triage.periodic_s = 0.2
+    service.handle_event(_transcript(clock, "first"))
+    clock.advance(0.25)
+    service.tick()
+    assert len(lane.submits) == 1
+
+    service.tick()
+    service.tick()
+    assert len(lane.submits) == 1  # identical window, no new speech
+
+    service.handle_event(_transcript(clock, "second"))
+    clock.advance(0.25)
+    service.tick()
+    assert len(lane.submits) == 2
+    assert lane.submits[1][0] == "- first\n- second"
+
+
+def test_triage_debug_emits_submit_and_result():
+    sink = FakeDebugSink()
+    service, _, clock, _, lane = _triage_service(debug_sink=sink)
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+
+    submits = [event for event in sink.events if event[0] == "triage-submit"]
+    assert submits == [
+        ("triage-submit", {"trigger": "settled", "window": 1, "generation": 0})
+    ]
+
+    _, _, _, gen = lane.submits[0]
+    service._on_triage_result(gen, _escalated_payload())
+
+    results = [event for event in sink.events if event[0] == "triage-result"]
+    assert len(results) == 1
+    assert results[0][1]["escalated"] is True
+    assert results[0][1]["generation"] == 0
+
+
+def test_triage_debug_marks_periodic_trigger():
+    sink = FakeDebugSink()
+    service, _, clock, _, lane = _triage_service(debug_sink=sink)
+    service.cfg.triage.periodic_s = 0.2
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(0.25)
+    service.tick()
+
+    submits = [event for event in sink.events if event[0] == "triage-submit"]
+    assert submits[0][1]["trigger"] == "periodic"
+
+
+def test_triage_debug_emits_brief_dispatch():
+    sink = FakeDebugSink()
+    service, _, clock, _, lane = _triage_service(
+        brief_fn=lambda prompt: "RAW", debug_sink=sink
+    )
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+    _, _, _, gen = lane.submits[0]
+
+    service._on_triage_result(gen, _escalated_payload())
+
+    brief = [event for event in sink.events if event[0] == "triage-brief"]
+    assert brief == [("triage-brief", {"action": "dispatch", "generation": 0})]
+
+
+def test_triage_debug_emits_brief_gate_skip():
+    sink = FakeDebugSink()
+    service, _, clock, _, lane = _triage_service(
+        triage=FakeTriage(gate=False), brief_fn=lambda prompt: "RAW", debug_sink=sink
+    )
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+    _, _, _, gen = lane.submits[0]
+
+    service._on_triage_result(gen, _escalated_payload())
+
+    brief = [event for event in sink.events if event[0] == "triage-brief"]
+    assert brief == [
+        ("triage-brief", {"action": "skip", "reason": "gate", "generation": 0})
+    ]
+
+
 def test_maybe_triage_skips_unready_lane():
     service, _, clock, _, lane = _triage_service(lane=FakeLane(ready=False))
     service.handle_event(_transcript(clock, "hello"))

tests/test_config.py

@@ -557,6 +557,7 @@ def test_triage_defaults():
     assert cfg.triage.questions_file == "questions.json"
     assert cfg.triage.window == 500
     assert cfg.triage.state_char_budget == 3000
+    assert cfg.triage.periodic_s == 30.0
     assert cfg.triage.predict_max_tokens == 1024
     assert cfg.triage.confidence_field == "answer_confidence"
     assert cfg.triage.escalate_threshold == 0.6
@@ -730,6 +731,21 @@ brief_pending_char_budget = 777
     assert cfg.triage.brief_pending_char_budget == 777
 
 
+def test_triage_periodic_key_is_parsed():
+    """The [triage] periodic cadence is applied from TOML."""
+    config_toml = """
+[triage]
+periodic_s = 12.5
+"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        config_path = Path(tmpdir) / "config.toml"
+        config_path.write_text(config_toml)
+
+        cfg = load_config(str(config_path))
+
+    assert cfg.triage.periodic_s == 12.5
+
+
 def test_brief_defaults():
     """The [brief] section defaults to no on-disk writing."""
     with tempfile.TemporaryDirectory() as tmpdir: