STT-Assistant/src/stt_daemon/debug.py

"""Per-component health-signal log (signals only, never transcripts/audio).

Persists structured health signals to ``dir/YYYY-MM-DD.jsonl`` so the next
silent freeze is diagnosable from disk. No transcript text, no audio paths,
no prompt content -- numeric/boolean signal fields only.

Freeze-safety invariant: :class:`DebugSink` is a leaf lock. It never calls
back into the daemon, the assistant service, or any other component, so the
internal :class:`threading.Lock` cannot participate in an ABBA deadlock.
``emit()`` total-swallows every exception and latches a disabled circuit
breaker, because a raise in the daemon reader is misclassified as capture
failure (reopen churn/shutdown) and a raise in the worker is fatal.
"""

import json
import logging
import os
import threading
from datetime import datetime
from pathlib import Path

logger = logging.getLogger(__name__)

_VALID_PROCS = ("daemon", "assistant", "supervisor")
_FORBIDDEN_FIELDS = ("text", "audio", "prompt")


class DebugSink:
    """Append-only per-day JSONL signal log, safe to share across threads.

    State invariant: this sink owns no reference to any daemon/service
    object and performs no callbacks -- it is a leaf lock. The lock is held
    only for a day-rollover ``os.open`` (when needed), a single ``os.write``
    of the fully serialized line, and the byte-counter increment.
    """

    def __init__(self, dir, proc, max_bytes_per_day=50_000_000, now_fn=None):
        """Create the sink, making ``dir`` once.

        Args:
            dir: Destination directory for daily JSONL files.
            proc: One of ``daemon`` | ``assistant`` | ``supervisor``.
            max_bytes_per_day: Best-effort per-date cap; writes past the cap
                are skipped with a single warning (never crash, never rotate).
            now_fn: Injectable clock returning a ``datetime`` (tests only).

        Raises:
            ValueError: If ``proc`` is not a known component name.
        """
        if proc not in _VALID_PROCS:
            raise ValueError(f"unknown proc: {proc!r} (expected one of {_VALID_PROCS})")
        self._proc = proc
        self._dir = Path(dir)
        self._max = max_bytes_per_day
        self._now = now_fn or datetime.now
        self._lock = threading.Lock()
        self._fd = None
        self._date = None
        self._bytes = 0
        self._disabled = False
        self._cap_warned = False
        try:
            self._dir.mkdir(parents=True, exist_ok=True)
        except Exception as e:  # noqa: BLE001 - debug logging is best effort
            self._disabled = True
            logger.warning("debug log disabled (mkdir failed): %s", e)

    def _open_day(self, today):
        """Open today's file, closing the previous day's fd. Caller holds the lock."""
        if self._fd is not None:
            try:
                os.close(self._fd)
            except OSError:
                pass
            self._fd = None
        path = str(self._dir / f"{today}.jsonl")
        self._fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
        self._date = today
        self._bytes = 0
        self._cap_warned = False

    def emit(self, type, **fields):
        """Append one signal record. Never raises.

        The line is serialized BEFORE acquiring the lock; the lock covers
        only the day-rollover open (when needed), one ``os.write`` of the
        full line, and the byte-counter increment. The first I/O failure
        latches a disabled circuit breaker: it warns once and every
        subsequent emit is a no-op. ``text``/``audio``/``prompt`` fields are
        silently dropped (signals only).
        """
        try:
            if self._disabled:
                return
            for key in _FORBIDDEN_FIELDS:
                fields.pop(key, None)
            now = self._now()
            # Serialize before opening/locking: a failure here must not leave
            # a partial line in the file for the next record to append onto.
            record = {"ts": now.isoformat(), "proc": self._proc, "type": type, **fields}
            data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
            with self._lock:
                if self._disabled:
                    return
                today = now.strftime("%Y-%m-%d")
                if today != self._date:
                    self._open_day(today)
                if self._bytes + len(data) > self._max:
                    if not self._cap_warned:
                        self._cap_warned = True
                        logger.warning("debug log cap reached for %s; skipping", today)
                    return
                assert self._fd is not None
                os.write(self._fd, data)
                self._bytes += len(data)
        except Exception as e:  # noqa: BLE001 - debug logging must never break callers
            with self._lock:
                self._disabled = True
                if self._fd is not None:
                    try:
                        os.close(self._fd)
                    except OSError:
                        pass
                    self._fd = None
            logger.warning("debug log disabled after write failure: %s", e)

    def close(self):
        """Clean up (no-op; per-day fds are reclaimed on rollover/exit)."""