Added triage dump feature

.gitignore

@@ -5,6 +5,7 @@ __pycache__/
 transcripts/
 audio/
 audit/
+triage/
 models/
 speakers/
 *.onnx

AGENTS.md

@@ -64,7 +64,7 @@ publish-only.
   `XDG_RUNTIME_DIR` (sockets), `HF_HOME` (Whisper cache, honored by faster-whisper/huggingface).
 - `[triage]` is the display-only signal panel, off by default (`enabled = false`). Its
   `questions_file` (`questions.json` in the cwd) is private user state and is gitignored, like
-  `config.toml`.
+  `config.toml`. `dump_io = true` writes one JSON per triage under `triage/` (also gitignored).
 - `[speaker]` is opt-in per-utterance speaker identification, off by default, backed by the
   `SpeakerConfig` dataclass (12th known section; 14 keys: `enabled`, `model_path`, `provider`,
   `threshold_id`, `threshold_new`, `min_enroll_s`, `rms_floor`, `margin`, `mint_cooldown_s`,
@@ -83,7 +83,7 @@ publish-only.
 
 ## Testing
 
-- `tests/` has 701 tests, all hardware- and network-free. Preserve that: inject fakes instead of
+- `tests/` has 733 tests, all hardware- and network-free. Preserve that: inject fakes instead of
   touching real `pactl`, audio devices, an LLM, or `sherpa_onnx` (the speaker lane uses a fake
   embedder; no ONNX in tests).
 - Key seams: `Daemon(...)` accepts `capture_factory`, `vad_factory`, `transcriber_factory`,
@@ -102,7 +102,7 @@ publish-only.
 ## Conventions & gotchas
 
 - Never commit gitignored runtime output: `config.toml`, `transcripts/`, `audio/`, `audit/`,
-  `speakers/` (biometric), `models/`, `.omo/`.
+  `triage/`, `speakers/` (biometric), `models/`, `.omo/`.
 - New assistant tools: register in `assistant/tools.py`. Model-supplied arguments are JSON-parsed and
   schema-validated before running; set `side_effect` and `requires_confirmation` for anything
   non-read-only.

README.md

@@ -143,7 +143,7 @@ See [docs/CLI.md](docs/CLI.md) for every subcommand and [docs/CONFIGURATION.md](
 - [docs/AUDIO-SETUP.md](docs/AUDIO-SETUP.md) — virtual sink creation and audio routing options.
 - [docs/CLI.md](docs/CLI.md) — `stt-daemon` subcommands and common overrides.
 - [docs/ASSISTANT.md](docs/ASSISTANT.md) — the assistant layer: modes, command flow, LLM, tools, audit.
-- [docs/TRIAGE.md](docs/TRIAGE.md) — the triage panel: Laya lane, escalation cascade, and questions file.
+- [docs/TRIAGE.md](docs/TRIAGE.md) — the triage panel: Laya lane, LLM extraction, and questions file.
 - [docs/SPEAKERS.md](docs/SPEAKERS.md) — speaker identification: enrollment, thresholds, store, and commands.
 - [docs/OUTPUT.md](docs/OUTPUT.md) — stdout, JSONL, and WAV output formats and fields.
 - [docs/TROUBLESHOOTING.md](docs/TROUBLESHOOTING.md) — common errors and how to fix them.

config.example.toml

@@ -106,6 +106,7 @@ extract_timeout_s    = 30.0
 device               = ""
 preload              = ["multilingual"]
 default_checkpoint   = "multilingual"
+dump_io              = false
 
 [speaker]
 enabled           = false

docs/ASSISTANT.md

@@ -159,12 +159,16 @@ three may be parallelized.
 
 On each settled turn in any mode (idle, chat, or command; `turn_ended()`), independent of chat,
 the lane classifies the last `window` transcribed utterances from the STT stream (not the chat
-history). Results flow through a **`signals -> content`** cascade: the signals stage
-always arrives first (per-category confidences), and when they escalate the assistant submits one
-`triage-extract` job to the existing LLM lane for a structured JSON extraction, which arrives as the
-content stage and is merged over the signals. The panel renders both; it is display-only and triggers
-no modes, commands, or tools. Leaving chat mode does not clear the panel or its buffer; an explicit
-`clear` does.
+history). Results flow through a `signals -> content` two-stage exchange: the signals stage always
+arrives first (per-category confidences), and the assistant then submits one `triage-extract` job to
+the existing LLM lane for a structured JSON extraction, which arrives as the content stage and is
+merged over the signals. Extraction runs after every Laya pass when `extract_enabled` is true and an
+LLM client is wired; on a Laya error it still runs with transcript state only. The deployment
+thresholds (`escalate_threshold`, `choice_min_confidence`) only drive the per-item `⚠` marker.
+The panel renders the last-triage clock (transcript time format/color) and both stages; it is
+display-only and triggers no modes, commands, or tools. Leaving chat mode does not clear the panel or
+its buffer; an explicit `clear` does. With `dump_io = true` each triage also writes one JSON file
+under `triage/` (see [TRIAGE.md](TRIAGE.md)).
 
 ## Speakers
 

docs/CONFIGURATION.md

@@ -130,6 +130,7 @@ extract_timeout_s    = 30.0                 # Extraction request timeout
 device               = ""                   # "" = auto -> None (Laya picks the multilingual checkpoint's default device)
 preload              = ["multilingual"]     # Fixed default: only the multilingual checkpoint is preloaded
 default_checkpoint   = "multilingual"       # Router default; multilingual is pinned for every call
+dump_io              = false                # Opt-in triage I/O dump: one JSON per triage under triage/YYYY-MM-DD/HHMMSS-mmm.json (Laya request/response + LLM request/response; best-effort, gitignored)
 
 [speaker]                                   # Speaker identification (sherpa-onnx embedding lane; off by default)
 enabled           = false                    # Master switch: false = no embedding lane, every record status "none"
@@ -197,9 +198,12 @@ and `default_checkpoint = "multilingual"`; the multilingual checkpoint is pinned
 (there is no `.en` English-only variant). `predict_max_tokens` is the Laya token budget (the
 multilingual checkpoint supports 1024); the condensed state shares it with the ~192-token
 question head, leaving ~830 tokens of room for state at the default. `confidence_field` selects which of Laya's two
-confidence values gates escalation, with `answer_confidence` (the calibrated `max(p)`) as the
-default. With `enabled = false` (the default) the lane is never built and the system behaves
-exactly as before.
+confidence values drives the per-item `flagged`/`⚠` marker, with `answer_confidence` (the
+calibrated `max(p)`) as the default. With `enabled = false` (the default) the lane is never built and the system behaves
+exactly as before. `dump_io = true` additionally writes one JSON file per triage under
+`triage/YYYY-MM-DD/HHMMSS-mmm.json` (local time, millisecond precision, same naming scheme as
+the audio files) containing the Laya request/response and the LLM request/response; it is
+off by default, best-effort (a dump failure never disturbs triage), and gitignored.
 
 `[llm].triage_system_prompt` is the strict JSON-only instruction used for the extraction cascade;
 it must return `{summary, opportunities[], doubts[], action_items[{what, why}], risks[]}` and

docs/DEVELOPMENT.md

@@ -55,6 +55,10 @@ src/stt_daemon/
 │   ├── executor.py      - NullExecutor / LlmExecutor (tool loop)
 │   ├── tools.py         - Tool registry + schema validation + get_datetime
 │   ├── llm.py           - llama.cpp client + chat session (block trimming)
+│   ├── inference.py     - Single-lane LLM queue (chat/command/triage-extract)
+│   ├── triage.py        - Triage analyzer (normalize, flag, extraction parse)
+│   ├── laya_lane.py     - In-process Laya lane (loader + single-flight worker)
+│   ├── triage_dump.py   - Opt-in per-triage I/O dump (`dump_io`)
 │   ├── audit.py         - Daily action audit log
 │   ├── runner.py        - `stt-assistant` process
 │   └── __main__.py      - Entry point for `python -m stt_daemon.assistant`
@@ -63,7 +67,7 @@ src/stt_daemon/
     ├── client.py        - Control-socket client
     └── app.py           - Textual app + `stt-tui` entry point
 
-tests/                    - 701 tests, no audio hardware or live LLM required
+tests/                    - 733 tests, no audio hardware or live LLM required
 
 scripts/
 ├── audio-sink.sh        - Thin wrapper over `stt-daemon setup|clear|list`

docs/OUTPUT.md

@@ -60,3 +60,8 @@ emitted and a warning is logged.
 > Point `audio_dir` at a volume with headroom and prune it yourself (e.g. a cron
 > job removing files older than N days), and leave `save_audio = false` unless you
 > actually need the clips for auditing.
+
+**Triage I/O dump (assistant-side)** — with `[triage].dump_io = true` the assistant writes one
+self-contained JSON per triage under `triage/YYYY-MM-DD/HHMMSS-mmm.json` (the Laya request/response
+plus the LLM request/response). It is separate from the daemon transcript output above, off by
+default, and best-effort. See [TRIAGE.md](TRIAGE.md#opt-in-io-dump-dump_io).

docs/TRIAGE.md

@@ -191,7 +191,8 @@ Because the multilingual checkpoint is pinned, every classification uses it. Mul
 runs at temperature 1.0 upstream, so its confidence scores are **uncalibrated out of the
 box**. The default thresholds (`escalate_threshold = 0.6`, `choice_min_confidence = 0.85`)
 are starting points, not measured values. Measure on your own transcripts before trusting
-the escalation behavior, and lower or raise the thresholds to match what you observe.
+the `⚠` flagging (the thresholds no longer gate extraction), and lower or raise the thresholds to
+match what you observe.
 
 The measurement lives in `tests/test_laya_lane.py::test_multilingual_pin_lang_and_resources`,
 gated by `pytest.importorskip("laya")`, so the default suite skips it green and the extra
@@ -228,11 +229,41 @@ file.
 | `device` | `""` | Empty string means auto, which maps to `None` so Laya picks the checkpoint's default device. Any other value is passed through. |
 | `preload` | `["multilingual"]` | Checkpoints preloaded at lane startup. The fixed default preloads the multilingual checkpoint only. |
 | `default_checkpoint` | `"multilingual"` | Router default and the per-request `model` for every call. Multilingual is pinned. |
+| `dump_io` | `false` | Opt-in I/O dump: one JSON file per triage under `triage/YYYY-MM-DD/HHMMSS-mmm.json` (see below). |
 
 `[llm].triage_system_prompt` is the strict JSON-only instruction used by the extraction
 cascade. It must return `{summary, opportunities[], doubts[], action_items[{what, why}],
 risks[]}` and nothing else.
 
+## Opt-in I/O dump (`dump_io`)
+
+With `[triage].dump_io = true` (default `false`) each triage writes exactly one
+self-contained JSON file under `triage/YYYY-MM-DD/HHMMSS-mmm.json` — local time with
+millisecond precision, the same naming scheme as the audio files
+(`audio/YYYY-MM-DD/HHMMSS-mmm-<reason>.wav`). The file holds all four I/O pieces of
+that triage: the Laya request, the Laya response, the LLM request (endpoint URL plus
+the full chat-completion body), and the LLM response (raw envelope), keyed by the
+same `ts` carried on the triage events:
+
+```json
+{
+  "ts": 1759073916.19,
+  "iso": "2026-09-28T09:38:36.412000-06:00",
+  "laya": {"request": {...}, "response": {...}, "error": null},
+  "llm": {"request": {"url": "...", "body": {...}}, "response": {...}, "error": null}
+}
+```
+
+When extraction does not run the file still gets written with only the `laya` half
+(`llm` is `null`); on a Laya error the file records `laya.response = null` with
+`laya.error = "predict failed"` plus the state-only LLM half. The directory is fixed
+as `triage/` (cwd-relative, like `transcripts/`/`audio/`/`audit/`) — there is no
+second config key for it. Files contain conversation text, so they are gitignored
+private debug output. Only the triage Laya predict and the triage extraction call are
+dumped (never the loader warmup predict or non-triage LLM calls), and dumping is
+best-effort: a dump failure is logged and swallowed, never disturbing triage
+behavior or events.
+
 ## Upgrading from a v5-draft config
 
 An earlier draft of this feature used an external service with its own transaction keys.
@@ -249,5 +280,5 @@ reads each field with a per-field `.get()`:
 | `max_len` | no successor; the 8192-token long-document capability is out of scope |
 
 Remove them from `config.toml` to avoid confusion, but they will not change behavior if left
-in place. There is also no `act_probability` anywhere; escalation uses the gating confidence
-selected by `confidence_field`.
+in place. There is also no `act_probability` anywhere; the per-item `flagged`/`⚠` marker uses the
+gating confidence selected by `confidence_field`.

src/stt_daemon/assistant/llm.py

@@ -54,6 +54,13 @@ def _default_opener(payload: dict, timeout: float) -> dict:
         return json.loads(response.read().decode("utf-8"))
 
 
+def _safe_io(cb, event) -> None:
+    try:
+      cb(event)
+    except Exception as e:  # noqa: BLE001 - a dump hook must not break chat
+      logger.warning("on_io callback failed: %s", e)
+
+
 class LlmClient:
     """Minimal OpenAI-compatible chat client (stdlib only)."""
 
@@ -74,6 +81,7 @@ class LlmClient:
         tools: list[dict] | None = None,
         max_tokens: int | None = None,
         timeout_s: float | None = None,
+        on_io: Callable[[dict], None] | None = None,
     ) -> LlmResponse:
         """Send one completion request and parse the reply.
 
@@ -82,6 +90,8 @@ class LlmClient:
             tools: Optional tool schemas.
             max_tokens: Per-request output cap; uses cfg.max_tokens when None.
             timeout_s: Per-request timeout; uses cfg.timeout_s when None.
+            on_io: Optional hook receiving `{"phase": "request", ...}` before
+                the call and `{"phase": "response", ...}` after it.
         """
         effective_max_tokens = self.cfg.max_tokens if max_tokens is None else max_tokens
         effective_timeout   = self.cfg.timeout_s  if timeout_s is None else timeout_s
@@ -97,13 +107,23 @@ class LlmClient:
             body["tools"] = tools
             body["tool_choice"] = "auto"
 
+        if on_io is not None:
+            _safe_io(
+                on_io, {"phase": "request", "url": self.cfg.endpoint, "body": body}
+            )
+
         try:
             payload = self._opener(
                 {"__url__": self.cfg.endpoint, "body": body}, effective_timeout
             )
         except (urllib.error.URLError, OSError, json.JSONDecodeError) as e:
+            if on_io is not None:
+                _safe_io(on_io, {"phase": "response", "payload": None, "error": str(e)})
             raise LlmError(str(e)) from e
 
+        if on_io is not None:
+            _safe_io(on_io, {"phase": "response", "payload": payload, "error": None})
+
         choices = payload.get("choices") or []
         if not choices:
             raise LlmError("response contained no choices")

src/stt_daemon/assistant/runner.py

@@ -30,7 +30,10 @@ def build_triage(cfg, client, on_result=None):
     )
     extract = None
     if client is not None:
-        def _extract(prompt):
+        def _extract(prompt, on_io=None):
+            kwargs = {}
+            if on_io is not None:
+                kwargs["on_io"] = on_io
             resp = client.chat(
                 [
                     {"role": "system", "content": cfg.llm.triage_system_prompt},
@@ -38,6 +41,7 @@ def build_triage(cfg, client, on_result=None):
                 ],
                 max_tokens=cfg.triage.extract_max_tokens,
                 timeout_s=cfg.triage.extract_timeout_s,
+                **kwargs,
             )
             return resp.content
 

src/stt_daemon/assistant/service.py

@@ -19,6 +19,7 @@ from ..config import Config
 from .executor import ExecutionResult
 from .inference import InferenceJob, InferenceQueue
 from .modes import Mode, ModeMachine
+from .triage_dump import TriageDumper
 
 logger = logging.getLogger(__name__)
 
@@ -63,7 +64,7 @@ class AssistantService:
         chat_reset: Callable[[], None] | None = None,
         triage: Any = None,
         laya_lane: Any = None,
-        triage_extract_fn: Callable[[str], str] | None = None,
+        triage_extract_fn: Callable[..., str] | None = None,
     ):
         """Create the service.
 
@@ -117,6 +118,8 @@ class AssistantService:
         self._last_triage_window: tuple = ()
         self._triage_generation = 0
         self._last_transcript_lang: str | None = None
+        self._triage_dump = TriageDumper(bool(getattr(cfg.triage, "dump_io", False)))
+        self._last_laya_request: dict | None = None
 
         # Health
         self._last_heartbeat_ts: float | None = None
@@ -678,7 +681,20 @@ class AssistantService:
         state = self._triage.build_state(window)
         if not state:
             return
-        self._laya_lane.submit(state, questions, self._last_transcript_lang, generation)
+        accepted = self._laya_lane.submit(
+            state, questions, self._last_transcript_lang, generation
+        )
+        if accepted and 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).
+            self._last_laya_request = {
+                "state": state,
+                "questions": questions,
+                "model": getattr(self.cfg.triage, "default_checkpoint", "multilingual"),
+                "lang": self._last_transcript_lang,
+                "max_len": int(getattr(self.cfg.triage, "predict_max_tokens", 1024)),
+            }
 
     def _on_triage_result(self, generation, payload) -> None:
         if self._triage is None:
@@ -688,49 +704,106 @@ class AssistantService:
                 return
             window = self._last_triage_window
         ts = self._now()
+        if self._triage_dump.enabled:
+            self._triage_dump.begin(
+                ts,
+                laya={
+                    "request": self._last_laya_request,
+                    "response": payload,
+                    "error": None if payload is not None else "predict failed",
+                },
+            )
         signals = None
         if payload is not None:
             signals = self._triage.normalize(payload)
             if not signals:
+                self._triage_dump.finalize(ts)
                 return
             with self._lock:
                 if generation != self._triage_generation:
+                    self._triage_dump.discard(ts)
                     return
                 self.last_triage = signals
             self._emit(TRIAGE_EVENT, stage="signals", triage=signals, ts=ts)
-        if self._triage_extract_fn is not None and getattr(
+        will_extract = self._triage_extract_fn is not None and getattr(
             self.cfg.triage, "extract_enabled", True
-        ):
+        )
+        if will_extract:
             job = InferenceJob(
                 name="triage-extract",
                 run=lambda: self._run_triage_extract(generation, window, signals, ts),
             )
-            self._inference.submit_if_room(job)
+            if not self._inference.submit_if_room(job):
+                self._triage_dump.finalize(ts)
+        else:
+            self._triage_dump.finalize(ts)
 
     def _run_triage_extract(self, generation, window, signals, ts=None) -> None:
         if self._triage is None or self._triage_extract_fn is None:
             return
+        dump = self._triage_dump.enabled
+        if ts is None:
+            ts = self._now()
         with self._lock:
             if generation != self._triage_generation:
+                if dump:
+                    self._triage_dump.discard(ts)
                 return
-        prompt = self._triage.build_extraction_prompt(window, signals)
+        io_hook: Callable[[dict], None] | None = None
+        if dump:
+            def _record_io(part: dict) -> None:
+                if part.get("phase") == "request":
+                    self._triage_dump.record(
+                        ts,
+                        llm={
+                            "request": {
+                                "url": part.get("url"),
+                                "body": part.get("body"),
+                            },
+                            "response": None,
+                            "error": None,
+                        },
+                    )
+                else:
+                    self._triage_dump.record(
+                        ts,
+                        llm={
+                            "request": None,
+                            "response": part.get("payload"),
+                            "error": part.get("error"),
+                        },
+                    )
+            io_hook = _record_io
         try:
-            text = self._triage_extract_fn(prompt)
-        except Exception as e:  # noqa: BLE001 - an extract failure must not kill the worker
-            logger.error("Triage extract error: %s", e)
-            return
-        with self._lock:
-            if generation != self._triage_generation:
+            prompt = self._triage.build_extraction_prompt(window, signals)
+            try:
+                if dump:
+                    text = self._triage_extract_fn(prompt, on_io=io_hook)
+                else:
+                    text = self._triage_extract_fn(prompt)
+            except Exception as e:  # noqa: BLE001 - an extract failure must not kill the worker
+                logger.error("Triage extract error: %s", e)
                 return
-        parsed = self._triage.parse_extraction(text)
-        if parsed is None:
-            return
-        with self._lock:
-            if generation != self._triage_generation:
+            with self._lock:
+                if generation != self._triage_generation:
+                    return
+            parsed = self._triage.parse_extraction(text)
+            if parsed is None:
                 return
-            merged = {**(signals or {}), "content": parsed}
-            self.last_triage = merged
-        self._emit(TRIAGE_EVENT, stage="content", triage=merged, ts=ts)
+            with self._lock:
+                if generation != self._triage_generation:
+                    return
+                merged = {**(signals or {}), "content": parsed}
+                self.last_triage = merged
+            self._emit(TRIAGE_EVENT, stage="content", triage=merged, ts=ts)
+        finally:
+            if dump:
+                with self._lock:
+                    stale = generation != self._triage_generation
+                if stale:
+                    self._triage_dump.discard(ts)
+                else:
+                    self._triage_dump.finalize(ts)
 
     def _reset_triage(self) -> None:
         if self._triage is None and self._laya_lane is None:
@@ -740,6 +813,7 @@ class AssistantService:
             self.last_triage = None
             self._last_triage_window = ()
             self.triage_history = []
+        self._triage_dump.clear()
         if self._laya_lane is not None:
             self._laya_lane.drop_pending()
         self._inference.drop(lambda j: j.name == "triage-extract")

src/stt_daemon/assistant/triage_dump.py

@@ -0,0 +1,132 @@
+"""Opt-in triage I/O dump: one JSON file per triage with Laya + LLM I/O.
+
+Enabled via `[triage].dump_io = true` (off by default). When enabled, each
+triage handled in `_on_triage_result` produces exactly one file under
+`triage/YYYY-MM-DD/HHMMSS-mmm.json` (local time, millisecond precision — the
+same naming scheme as `audio/YYYY-MM-DD/HHMMSS-mmm-<reason>.wav`).
+
+The directory `triage/` is fixed (cwd-relative, like `transcripts/`/`audio/`/
+`audit/`) and gitignored. Files contain conversation text (transcript state,
+Laya questions/answers, LLM prompts), so treat them as private debug output.
+
+Best-effort by design: a dump failure is logged and swallowed, never raised,
+so it can never disturb the triage flow.
+
+Merge semantics: `begin`/`record` merge one level deep for the `laya` and
+`llm` halves, and an incoming `None` never clobbers an already-recorded
+non-`None` value. That is what lets the service record the LLM request in the
+request phase and the LLM response in the response phase without the second
+call wiping the first.
+"""
+
+import json
+import logging
+import os
+import tempfile
+import threading
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class TriageDumper:
+    """Thread-safe, best-effort per-triage I/O recorder."""
+
+    def __init__(self, enabled: bool, directory: str = "triage") -> None:
+        self._enabled = bool(enabled)
+        self._directory = str(directory)
+        self._lock = threading.Lock()
+        self._pending: dict[float, dict] = {}
+
+    @property
+    def enabled(self) -> bool:
+        return self._enabled
+
+    def begin(self, ts: float, **parts: Any) -> None:
+        """Create (or merge into) the pending record keyed by `ts`."""
+        if not self._enabled:
+            return
+        with self._lock:
+            record = self._pending.get(ts)
+            if record is None:
+                record = {"ts": ts}
+                self._pending[ts] = record
+            self._merge(record, parts)
+
+    def record(self, ts: float, **parts: Any) -> None:
+        """Merge `parts` into the pending record keyed by `ts` (alias of begin)."""
+        self.begin(ts, **parts)
+
+    def finalize(self, ts: float) -> None:
+        """Pop the record for `ts` and write it to disk (no-op when unknown)."""
+        if not self._enabled:
+            return
+        with self._lock:
+            record = self._pending.pop(ts, None)
+        if record is None:
+            return
+        self._write(record)
+
+    def discard(self, ts: float) -> None:
+        """Drop the pending record for `ts` WITHOUT writing (no-op when unknown)."""
+        if not self._enabled:
+            return
+        with self._lock:
+            self._pending.pop(ts, None)
+
+    def clear(self) -> None:
+        """Drop all pending records."""
+        if not self._enabled:
+            return
+        with self._lock:
+            self._pending.clear()
+
+    @staticmethod
+    def _merge(record: dict, parts: dict) -> None:
+        for key, value in parts.items():
+            if (
+                key in ("laya", "llm")
+                and isinstance(value, dict)
+                and isinstance(record.get(key), dict)
+            ):
+                nested = record[key]
+                for sub_key, sub_value in value.items():
+                    if sub_value is None and nested.get(sub_key) is not None:
+                        continue
+                    nested[sub_key] = sub_value
+            else:
+                record[key] = value
+
+    def _write(self, record: dict) -> None:
+        try:
+            ts_raw = record.get("ts")
+            if not isinstance(ts_raw, (int, float)):
+                return
+            ts = float(ts_raw)
+            dt = datetime.fromtimestamp(ts)
+            day_dir = Path(self._directory) / dt.strftime("%Y-%m-%d")
+            day_dir.mkdir(parents=True, exist_ok=True)
+            filename = dt.strftime("%H%M%S") + "-" + dt.strftime("%f")[:3] + ".json"
+            final_path = day_dir / filename
+            payload = {
+                "ts": ts,
+                "iso": dt.astimezone().isoformat(),
+                "laya": record.get("laya"),
+                "llm": record.get("llm"),
+            }
+            text = json.dumps(payload, ensure_ascii=False, indent=2)
+            fd, tmp = tempfile.mkstemp(dir=str(day_dir), suffix=".tmp")
+            try:
+                with os.fdopen(fd, "w", encoding="utf-8") as f:
+                    f.write(text)
+                os.replace(tmp, final_path)
+            except OSError:
+                try:
+                    os.unlink(tmp)
+                except OSError:
+                    pass
+                raise
+        except Exception as e:  # noqa: BLE001 - dump is best-effort, never raises
+            logger.warning("triage I/O dump failed: %s", e)

src/stt_daemon/config.py

@@ -174,6 +174,7 @@ class TriageConfig:
     device: str = ""
     preload: list[str] = field(default_factory=lambda: ["multilingual"])
     default_checkpoint: str = "multilingual"
+    dump_io: bool = False
 
 
 @dataclass
@@ -501,6 +502,7 @@ def load_config(path: str | None = None) -> Config:
                 default_checkpoint=triage_data.get(
                     "default_checkpoint", cfg.triage.default_checkpoint
                 ),
+                dump_io=triage_data.get("dump_io", cfg.triage.dump_io),
             )
 
         # Speaker section

tests/test_assistant_service.py

@@ -1,6 +1,8 @@
 """Tests for the assistant service (Phase 2) using synthetic event streams."""
 
 import itertools
+import json
+import re
 import threading
 import time
 from typing import Any
@@ -8,6 +10,7 @@ from typing import Any
 from stt_daemon import events
 from stt_daemon.assistant.inference import InferenceQueue
 from stt_daemon.assistant.service import EXECUTE_BUSY, AssistantService
+from stt_daemon.assistant.triage_dump import TriageDumper
 from stt_daemon.config import Config
 
 _sequence = itertools.count(1)
@@ -1343,3 +1346,179 @@ def test_tick_heals_triage_panel_after_lane_ready():
         assert not any("loading" in line for line in lines)
     finally:
         lane.stop()
+
+
+def _dump_extract(prompt, on_io=None):
+    """Extract fake that mimics LlmClient.chat's on_io phases."""
+    if on_io is not None:
+        on_io({"phase": "request", "url": "http://llm/v1", "body": {"prompt": prompt}})
+        on_io({"phase": "response", "payload": {"content": "parsed"}, "error": None})
+    return "RAW"
+
+
+def _run_triage_flow(service, emitted, clock, lane, payload):
+    service.toggle_chat()
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+    _, _, _, gen = lane.submits[0]
+    service._on_triage_result(gen, payload)
+    return gen
+
+
+def _dump_files(tmp_path):
+    return list(tmp_path.rglob("*.json"))
+
+
+def test_dump_writes_full_laya_and_llm_io(tmp_path):
+    service, emitted, clock, triage, lane = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+
+    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())
+    jobs = [j for j in service._inference._queue if j.name == "triage-extract"]
+    assert len(jobs) == 1
+    jobs[0].run()
+
+    files = _dump_files(tmp_path)
+    assert len(files) == 1
+    assert re.fullmatch(r"\d{6}-\d{3}\.json", files[0].name)
+    data = json.loads(files[0].read_text(encoding="utf-8"))
+    assert data["ts"] == clock()
+    assert isinstance(data["iso"], str) and data["iso"]
+    assert data["laya"]["request"]["state"] == "- hello"
+    assert data["laya"]["request"]["questions"] == triage.questions
+    assert data["laya"]["request"]["model"] == "multilingual"
+    assert data["laya"]["request"]["max_len"] == 1024
+    assert data["laya"]["response"] == _escalated_payload()
+    assert data["laya"]["error"] is None
+    assert data["llm"]["request"]["url"] == "http://llm/v1"
+    assert data["llm"]["request"]["body"] == {"prompt": "EXTRACT"}
+    assert data["llm"]["response"] == {"content": "parsed"}
+    assert data["llm"]["error"] is None
+
+
+def test_dump_laya_error_keeps_state_only_llm_half(tmp_path):
+    service, emitted, clock, _, lane = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+
+    _run_triage_flow(service, emitted, clock, lane, None)
+    jobs = [j for j in service._inference._queue if j.name == "triage-extract"]
+    assert len(jobs) == 1
+    jobs[0].run()
+
+    files = _dump_files(tmp_path)
+    assert len(files) == 1
+    data = json.loads(files[0].read_text(encoding="utf-8"))
+    assert data["laya"]["response"] is None
+    assert data["laya"]["error"] == "predict failed"
+    assert data["llm"]["request"]["url"] == "http://llm/v1"
+    assert data["llm"]["response"] == {"content": "parsed"}
+
+
+def test_dump_without_extract_has_laya_only(tmp_path):
+    service, emitted, clock, _, lane = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+    service.cfg.triage.extract_enabled = False
+
+    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())
+
+    files = _dump_files(tmp_path)
+    assert len(files) == 1
+    data = json.loads(files[0].read_text(encoding="utf-8"))
+    assert data["laya"]["response"] == _escalated_payload()
+    assert data["llm"] is None
+
+
+def test_dump_queue_full_has_laya_only(tmp_path):
+    service, emitted, clock, _, lane = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+    service._inference = InferenceQueue(size=2)
+
+    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())
+
+    assert service._inference.depth() == 1
+    files = _dump_files(tmp_path)
+    assert len(files) == 1
+    data = json.loads(files[0].read_text(encoding="utf-8"))
+    assert data["laya"]["response"] == _escalated_payload()
+    assert data["llm"] is None
+
+
+def test_dump_disabled_writes_nothing(tmp_path):
+    service, emitted, clock, _, lane = _triage_service(
+        extract_fn=lambda prompt: "RAW"
+    )
+    assert service._triage_dump.enabled is False
+    service._triage_dump = TriageDumper(False, directory=str(tmp_path))
+
+    _run_triage_flow(service, emitted, clock, lane, _escalated_payload())
+    jobs = [j for j in service._inference._queue if j.name == "triage-extract"]
+    assert len(jobs) == 1
+    jobs[0].run()
+
+    assert _dump_files(tmp_path) == []
+
+
+def test_dump_stale_result_discards_without_file(tmp_path):
+    service, emitted, clock, triage, lane = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+    service.toggle_chat()
+    service.handle_event(_transcript(clock, "hello"))
+    clock.advance(service.cfg.assistant.turn_gap_s + 0.1)
+    service.tick()
+    _, _, _, gen = lane.submits[0]
+
+    real_normalize = triage.normalize
+
+    def stale_normalize(payload):
+        # Simulate a clear landing between the first check/begin and the re-check.
+        service._triage_generation += 1
+        return real_normalize(payload)
+
+    triage.normalize = stale_normalize
+    service._on_triage_result(gen, _escalated_payload())
+
+    assert _dump_files(tmp_path) == []
+    assert service._triage_dump._pending == {}
+
+
+def test_dump_stale_extract_discards_without_file(tmp_path):
+    service, _, clock, _, _ = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+    ts = clock()
+    service._triage_dump.begin(ts, laya={"request": {}, "response": {}, "error": None})
+    service._triage_generation += 1  # stale before the extract even starts
+
+    service._run_triage_extract(0, ("hello",), _escalated_payload(), ts)
+
+    assert _dump_files(tmp_path) == []
+    assert service._triage_dump._pending == {}
+
+
+def test_dump_extract_cleared_mid_flight_writes_nothing(tmp_path):
+    service, _, clock, _, _ = _triage_service(extract_fn=_dump_extract)
+    service._triage_dump = TriageDumper(True, directory=str(tmp_path))
+    service.cfg.triage.dump_io = True
+
+    def stale_extract(prompt, on_io=None):
+        # Simulate a clear landing mid-LLM-call (after the request phase fired).
+        service._triage_generation += 1
+        if on_io is not None:
+            on_io({"phase": "request", "url": "http://llm/v1", "body": {}})
+            on_io({"phase": "response", "payload": {"c": 1}, "error": None})
+        return "RAW"
+
+    service._triage_extract_fn = stale_extract
+    ts = clock()
+    service._triage_dump.begin(ts, laya={"request": {}, "response": {}, "error": None})
+
+    service._run_triage_extract(0, ("hello",), _escalated_payload(), ts)
+
+    assert _dump_files(tmp_path) == []
+    assert service._triage_dump._pending == {}

tests/test_config.py

@@ -579,6 +579,35 @@ default_checkpoint = "english"
     assert cfg.triage.default_checkpoint == "english"
 
 
+def test_triage_dump_io_defaults_to_false():
+    """[triage].dump_io is opt-in: False unless explicitly enabled."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        import os
+        old_cwd = os.getcwd()
+        try:
+            os.chdir(tmpdir)
+            cfg = load_config()
+        finally:
+            os.chdir(old_cwd)
+
+    assert cfg.triage.dump_io is False
+
+
+def test_triage_dump_io_is_parsed():
+    """[triage].dump_io = true is applied from TOML."""
+    config_toml = """
+[triage]
+dump_io = true
+"""
+    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.dump_io is True
+
+
 def test_unknown_triag_section_warns_and_keeps_defaults(caplog):
     """A typo'd [triag] section must warn and never raise; triage keeps defaults."""
     config_toml = """

tests/test_llm.py

@@ -584,3 +584,48 @@ def test_chat_session_truncated_partial_reply_is_appended_and_notified():
     assert session.send("hi") == "partial answer"
     assert session.history[-1] == {"role": "assistant", "content": "partial answer"}
     assert any(kind == "truncated" for kind, _ in notifications)
+
+
+def test_on_io_receives_request_then_response():
+    """chat(..., on_io=cb) reports the request (url+body) then the response payload."""
+    client, _ = _make_client(lambda _p: _reply(content="ok"))
+    seen: list[dict] = []
+
+    response = client.chat([{"role": "user", "content": "hi"}], on_io=seen.append)
+
+    assert response.content == "ok"
+    assert [event["phase"] for event in seen] == ["request", "response"]
+    assert seen[0]["url"] == LlmConfig().endpoint
+    assert seen[0]["body"]["messages"] == [{"role": "user", "content": "hi"}]
+    assert seen[0]["body"]["max_tokens"] == LlmConfig().max_tokens
+    assert seen[1]["payload"]["choices"][0]["message"]["content"] == "ok"
+    assert seen[1]["error"] is None
+
+
+def test_on_io_response_phase_reports_opener_error():
+    """An opener raising still yields a response phase with the error set."""
+    import urllib.error
+
+    def opener(_payload, _timeout):
+        raise urllib.error.URLError("down")
+
+    client = LlmClient(LlmConfig(), opener=opener)
+    seen: list[dict] = []
+
+    with pytest.raises(LlmError):
+        client.chat([{"role": "user", "content": "hi"}], on_io=seen.append)
+
+    assert [event["phase"] for event in seen] == ["request", "response"]
+    assert seen[1]["payload"] is None
+    assert "down" in seen[1]["error"]
+
+
+def test_raising_on_io_does_not_break_chat():
+    """A throwing on_io callback is swallowed and the reply still parses."""
+
+    def bad(_event):
+        raise RuntimeError("boom")
+
+    client, _ = _make_client(lambda _p: _reply(content="ok"))
+
+    assert client.chat([{"role": "user", "content": "hi"}], on_io=bad).content == "ok"

tests/test_triage_dump.py

@@ -0,0 +1,137 @@
+"""Tests for the opt-in triage I/O dump (TriageDumper)."""
+
+import json
+import re
+from datetime import datetime
+
+from stt_daemon.assistant.triage_dump import TriageDumper
+
+TS = 1759073916.19
+
+
+def _only_file(directory):
+    files = list(directory.rglob("*.json"))
+    assert len(files) == 1
+    return files[0]
+
+
+def test_enabled_writes_one_shaped_file(tmp_path):
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.begin(
+        TS,
+        laya={"request": {"state": "s"}, "response": {"ok": True}, "error": None},
+        llm={"request": {"url": "u", "body": {}}, "response": {"c": 1}, "error": None},
+    )
+    dumper.finalize(TS)
+
+    path = _only_file(tmp_path)
+    assert re.fullmatch(r"\d{6}-\d{3}\.json", path.name)
+    expected_day = datetime.fromtimestamp(TS).strftime("%Y-%m-%d")
+    assert path.parent.name == expected_day
+
+    data = json.loads(path.read_text(encoding="utf-8"))
+    assert data["ts"] == TS
+    assert isinstance(data["iso"], str) and data["iso"]
+    assert data["laya"] == {
+        "request": {"state": "s"},
+        "response": {"ok": True},
+        "error": None,
+    }
+    assert data["llm"] == {
+        "request": {"url": "u", "body": {}},
+        "response": {"c": 1},
+        "error": None,
+    }
+
+
+def test_disabled_writes_nothing(tmp_path):
+    dumper = TriageDumper(False, directory=str(tmp_path))
+    assert dumper.enabled is False
+    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
+    dumper.record(TS, llm={"request": {}, "response": {}, "error": None})
+    dumper.finalize(TS)
+    dumper.clear()
+
+    assert list(tmp_path.rglob("*")) == []
+
+
+def test_record_merges_without_dropping_earlier_parts(tmp_path):
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.begin(
+        TS, laya={"request": {"state": "s"}, "response": None, "error": None}
+    )
+    dumper.record(TS, llm={"request": {"url": "u"}, "response": None, "error": None})
+    dumper.record(TS, laya={"response": {"ok": True}, "error": None})
+    dumper.finalize(TS)
+
+    data = json.loads(_only_file(tmp_path).read_text(encoding="utf-8"))
+    assert data["laya"]["request"] == {"state": "s"}
+    assert data["laya"]["response"] == {"ok": True}
+    assert data["llm"]["request"] == {"url": "u"}
+
+
+def test_llm_response_phase_keeps_request(tmp_path):
+    """The response-phase record must not wipe the captured request."""
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
+    dumper.record(
+        TS,
+        llm={"request": {"url": "u", "body": {"m": 1}}, "response": None, "error": None},
+    )
+    dumper.record(TS, llm={"request": None, "response": {"c": 1}, "error": None})
+    dumper.finalize(TS)
+
+    data = json.loads(_only_file(tmp_path).read_text(encoding="utf-8"))
+    assert data["llm"]["request"] == {"url": "u", "body": {"m": 1}}
+    assert data["llm"]["response"] == {"c": 1}
+    assert data["llm"]["error"] is None
+
+
+def test_finalize_unknown_ts_is_noop(tmp_path):
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.finalize(1234567890.0)
+
+    assert list(tmp_path.rglob("*")) == []
+
+
+def test_clear_drops_pending(tmp_path):
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
+    dumper.clear()
+    dumper.finalize(TS)
+
+    assert list(tmp_path.rglob("*")) == []
+
+
+def test_write_failure_is_swallowed(tmp_path):
+    blocker = tmp_path / "blocker"
+    blocker.write_text("not a directory")
+    dumper = TriageDumper(True, directory=str(blocker / "sub"))
+
+    # Must not raise even though no directory can be created here.
+    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
+    dumper.finalize(TS)
+
+    assert blocker.is_file()
+
+
+def test_discard_drops_pending_without_writing(tmp_path):
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.begin(TS, laya={"request": {}, "response": {}, "error": None})
+    dumper.discard(TS)
+    dumper.finalize(TS)
+
+    assert list(tmp_path.rglob("*")) == []
+
+
+def test_discard_unknown_and_disabled_is_noop(tmp_path):
+    dumper = TriageDumper(True, directory=str(tmp_path))
+    dumper.discard(1234567890.0)
+
+    assert list(tmp_path.rglob("*")) == []
+
+    disabled = TriageDumper(False, directory=str(tmp_path))
+    disabled.discard(TS)
+    disabled.finalize(TS)
+
+    assert list(tmp_path.rglob("*")) == []