debug_log.py

"""Debug logging system — structured logs to stderr + in-memory buffer.

Every critical function logs its entry, exit, duration, and errors. Logs are
written to:
   1. stderr (HF Space logs capture this automatically)
   2. In-memory ring buffer (for /api/debug/logs endpoint)
   3. SSE subscriber queues (for /api/debug/stream real-time streaming)

View logs in the browser via GET /api/debug/logs?cat=glm&tail=50
Stream logs in real-time via GET /api/debug/stream (SSE, auth-gated)
(gated by the rotation token or CRITIQUE_TOKEN env var).
"""
from __future__ import annotations

import datetime
import json
import os
import queue
import sys
from collections import deque
from typing import Any

# --- config -------------------------------------------------------------------
_MAX_MEMORY_LOGS = 5000
_STDERR_ENABLED = os.environ.get("DOOMALAYSOCREATE_LOG", "1").strip() not in ("0", "false", "no", "")

# --- in-memory ring buffer ----------------------------------------------------
_memory_logs: deque[dict] = deque(maxlen=_MAX_MEMORY_LOGS)
_categories: set[str] = set()

# --- SSE subscriber queues ----------------------------------------------------
_sse_queues: list[queue.Queue] = []


def subscribe() -> queue.Queue:
    """Register an SSE subscriber queue. Returns a Queue that receives new log records."""
    q: queue.Queue = queue.Queue()
    _sse_queues.append(q)
    return q


def unsubscribe(q: queue.Queue) -> None:
    """Remove a subscriber queue."""
    try:
        _sse_queues.remove(q)
    except ValueError:
        pass


def _notify_subscribers(record: dict) -> None:
    """Push a log record to all SSE subscriber queues (best-effort)."""
    dead: list[queue.Queue] = []
    for q in _sse_queues:
        try:
            q.put_nowait(record)
        except Exception:
            dead.append(q)
    for q in dead:
        try:
            _sse_queues.remove(q)
        except ValueError:
            pass

# --- sensitive field patterns for auto-redaction ----------------------------
_SENSITIVE_PATTERNS = [
    (r"\b(token|secret|key|password|auth)\s*[:=]\S+", "[REDACTED]"),
    (r"\b(Bearer\s+)[A-Za-z0-9-_.]+", r"\1[REDACTED]"),
    (r"\b(api[_-]?key)\s*[:=]\S+", r"\1=[REDACTED]"),
]


def _iso_ts() -> str:
    now = datetime.datetime.now(datetime.timezone.utc)
    return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond // 1000:03d}Z"


def _redact(msg: str) -> str:
    import re
    for pattern, replacement in _SENSITIVE_PATTERNS:
        msg = re.sub(pattern, replacement, msg, flags=re.IGNORECASE)
    return msg


def log_event(category: str, **fields: Any) -> None:
    """Fire-and-forget: never raises. Observability must not break a request."""
    try:
        record = {
            "ts": _iso_ts(),
            "cat": category,
            "fields": {k: _redact(str(v)) for k, v in fields.items()},
        }
        _memory_logs.append(record)
        _categories.add(category)
        if _STDERR_ENABLED:
            print(json.dumps(record, ensure_ascii=False), file=sys.stderr, flush=True)
        _notify_subscribers(record)
    except Exception:
        pass  # never break the calling code


def log_entry(level: str = "INFO", cat: str = "", fn: str = "",
              msg: str = "", data: dict | None = None,
              ms: float | None = None) -> None:
    """Log a structured entry in the frontend-compatible format.

    This is the primary logging call used by the debug screen. Parameters
    match the LogEntry interface in DebugScreen.tsx:
      - level: ERROR / WARN / INFO / DEBUG
      - cat:   category (startup, glm, auth, agent, conscious, errors, http)
      - fn:    calling function name
      - msg:   human-readable message
      - data:  optional structured payload (shown as expandable JSON)
      - ms:    optional elapsed milliseconds for timed operations
    """
    try:
        record = {
            "ts": _iso_ts(),
            "level": level,
            "cat": cat,
            "fn": fn,
            "msg": msg,
            "data": data or {},
        }
        if ms is not None:
            record["ms"] = ms
        _memory_logs.append(record)
        _categories.add(cat)
        if _STDERR_ENABLED:
            print(json.dumps(record, ensure_ascii=False), file=sys.stderr, flush=True)
        _notify_subscribers(record)
    except Exception:
        pass


def get_recent_logs(category: str | None = None,
                    tail: int = 100,
                    min_level: str | None = None) -> list[dict]:
    """Return recent log entries, optionally filtered by category and level."""
    all_logs = list(_memory_logs)
    if category:
        all_logs = [r for r in all_logs if r.get("cat") == category]
    if min_level:
        _LEVEL_WEIGHT = {"ERROR": 40, "WARN": 30, "INFO": 20, "DEBUG": 10}
        threshold = _LEVEL_WEIGHT.get(min_level.upper(), 0)
        all_logs = [r for r in all_logs if _LEVEL_WEIGHT.get(r.get("level", "INFO"), 0) >= threshold]
    return all_logs[-tail:]


def list_log_categories() -> list[str]:
    return sorted(_categories)


def clear_logs(category: str | None = None) -> None:
    """Clear the in-memory ring buffer (optionally by category)."""
    global _memory_logs, _categories
    if category:
        _memory_logs = deque(
            (r for r in _memory_logs if r.get("cat") != category),
            maxlen=_MAX_MEMORY_LOGS,
        )
        remaining = set(r.get("cat", "") for r in _memory_logs)
        _categories = remaining
    else:
        _memory_logs.clear()
        _categories.clear()