"""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-.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] = {} # Monotonic tie-breaker for same-millisecond collisions: the first # record for a millisecond keeps the exact `HHMMSS-mmm.json` shape, # later records in the same millisecond get a `-N` suffix so they # never overwrite the earlier file. self._name_lock = threading.Lock() self._last_stamp: str | None = None self._stamp_seq = 0 @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 _reserve_path(self, day_dir: Path, stamp: str) -> Path: """Pick a collision-free path for `stamp` (monotonic tie-breaker).""" key = f"{day_dir.name} {stamp}" with self._name_lock: if key == self._last_stamp: self._stamp_seq += 1 else: self._last_stamp = key self._stamp_seq = 0 seq = self._stamp_seq path = day_dir / (f"{stamp}.json" if seq == 0 else f"{stamp}-{seq}.json") while path.exists(): seq += 1 path = day_dir / f"{stamp}-{seq}.json" self._stamp_seq = seq return path 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) stamp = dt.strftime("%H%M%S") + "-" + dt.strftime("%f")[:3] final_path = self._reserve_path(day_dir, stamp) 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)