src/logger.py

"""Structured JSONL activity log + in-memory ring buffer for the future Gradio panel."""

from __future__ import annotations

import json
import re
import threading
from collections import deque
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Literal

_BUFFER_MAX = 50
_LOG_PATH = Path("logs/activity.jsonl")

_lock = threading.Lock()
_buffer: deque[dict[str, Any]] = deque(maxlen=_BUFFER_MAX)

_SENSITIVE_KEYS = frozenset({"pin", "password", "passwd", "secret", "token"})
_EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")
_PHONE_RE = re.compile(r"\b\+?\d[\d\s().-]{8,}\b")
_PIN_SEQ_RE = re.compile(r"\b\d{4}\b")


def redact(text: str | None) -> str | None:
    """Strip patterns that commonly carry credentials or direct identifiers."""

    if text is None:
        return None
    out = text
    out = _EMAIL_RE.sub("<email>", out)
    out = _PHONE_RE.sub("<phone>", out)
    # Avoid leaking numeric PINs while keeping longer order totals mostly intact.
    if len(out) <= 48:
        out = _PIN_SEQ_RE.sub("<pin>", out)
    return out


def _redact_mapping(shallow: dict[str, Any]) -> dict[str, Any]:
    red: dict[str, Any] = {}
    for k, v in shallow.items():
        lk = str(k).lower()
        if lk in _SENSITIVE_KEYS:
            red[k] = "<redacted>"
        elif lk == "email":
            red[k] = "<redacted>"
        else:
            red[k] = v
    return red


EventType = Literal["tool_call", "policy_decision", "llm_call", "error"]
Status = Literal["success", "error"]


def log_event(
    event_type: EventType,
    *,
    session_id: str,
    tool: str | None = None,
    decision: Literal["allow", "deny", "require_confirmation"] | None = None,
    deny_reason: str | None = None,
    latency_ms: int | None = None,
    status: Status | None = None,
    args_summary: str | None = None,
) -> None:
    deny_safe = redact(deny_reason)
    args_safe = redact(args_summary)

    row: dict[str, Any] = {
        "timestamp": datetime.now(tz=UTC).isoformat(),
        "event_type": event_type,
        "tool": tool,
        "decision": decision,
        "deny_reason": deny_safe,
        "latency_ms": latency_ms,
        "status": status,
        "session_id": session_id,
        "args_summary": args_safe,
    }
    line = json.dumps(row, ensure_ascii=False)
    _LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    with _lock:
        _buffer.append(row)
        with _LOG_PATH.open("a", encoding="utf-8") as fh:
            fh.write(line + "\n")


def get_recent_events(n: int = 50) -> list[dict[str, Any]]:
    with _lock:
        return list(_buffer)[-n:]


def summarize_tool_args(tool_name: str, args: dict[str, Any]) -> str:
    """Compact, policy-safe argument summary for logs (never includes PIN)."""

    scrubbed = _redact_mapping(args)
    if tool_name == "create_order":
        items = scrubbed.get("items")
        if isinstance(items, list):
            bits: list[str] = []
            for it in items[:6]:
                if isinstance(it, dict):
                    bits.append(
                        f"{it.get('sku')}x{it.get('quantity')}@{it.get('unit_price')}{it.get('currency', '')}"
                    )
                else:
                    bits.append(str(it))
            more = f"+{len(items) - 6} more" if len(items) > 6 else ""
            scrubbed = {**scrubbed, "items": ",".join(bits) + (f";{more}" if more else "")}
    try:
        raw = json.dumps(scrubbed, sort_keys=True, ensure_ascii=False, default=str)
    except TypeError:
        raw = str(scrubbed)
    return redact(raw) or raw