"""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)."""