STT-Assistant/src/stt_daemon/assistant/triage_dump.py

"""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] = {}
        # 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)