legal-ai/web/agent_telemetry.py

"""Agent-platform telemetry (TaskMaster #219) — structured, platform-agnostic events.

Emits one structured event per agent-platform operation (wakeup, disposition
comment, interaction reap/cancel, run cancel, session reset) so recovery-loop
behaviour is **observable in real time** instead of reconstructed forensically
from the Paperclip DB (``agent_wakeup_requests`` / ``heartbeat_runs``). See
[[reference_paperclip_recovery_loops]] / [[reference_recovery_loop_stranded_child]]
for the failure modes this makes visible.

**Platform-agnostic by design (INV-G12 / docs/spec/X15).** Events are named in
domain terms and carry no Paperclip-specific symbols — this module imports
*nothing* from the Paperclip shell. The Port (``web/agent_platform_port.py``)
calls :func:`instrument` / :func:`emit` around its operations, so the telemetry
semantics survive a platform swap.

**OTLP-ready, "logs first" (#219).** Each event is a flat dict with a stable
schema, written today as one JSON line on the ``agent.telemetry`` logger
(greppable in Coolify logs). Swapping :data:`_SINK` for an OTLP exporter — or
attaching an OTLP handler to that logger — is the later step, with **no
call-site changes**. Timestamps are stored in UTC (display-in-Israel is a
read-side concern; see [[feedback_israel_time]]).

Single path (INV-G2): the only telemetry emitter for the agent platform. Do not
add a parallel structured-log format for these events elsewhere — route through
:func:`emit`.
"""
from __future__ import annotations

import functools
import inspect
import json
import logging
import time
from collections.abc import Awaitable, Callable, Sequence
from datetime import datetime, timezone
from typing import Any, TypeVar

logger = logging.getLogger("agent.telemetry")

# Schema version — bump when the event shape changes so downstream consumers
# (dashboards, the #222 health taxonomy) can migrate deliberately.
SCHEMA_VERSION = 1

# Standard argument keys lifted onto every event when present on the wrapped call.
_STANDARD_KEYS: tuple[str, ...] = ("case_number", "issue_id", "company_id", "reason")

T = TypeVar("T")


def _now_iso() -> str:
    """UTC ISO-8601 with a trailing ``Z`` — store-UTC (feedback_israel_time)."""
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def _sink(payload: dict[str, Any]) -> None:
    """Where an event goes. Today: a JSON line on the ``agent.telemetry`` logger.

    Isolated so a future OTLP exporter is a one-line swap with no call-site
    churn (#219 "backend later").
    """
    logger.info(json.dumps(payload, ensure_ascii=False, default=str))


_SINK: Callable[[dict[str, Any]], None] = _sink


def emit(event: str, *, outcome: str = "ok", **fields: Any) -> None:
    """Emit one structured agent-platform event. Never raises.

    Telemetry must never break the operation it observes — any failure in the
    sink is swallowed to a warning. ``event`` is a dotted domain name
    (``agent.wakeup``, ``interaction.reaped``, ``run.cancelled`` …). Extra
    ``fields`` are merged flat into the event; ``None`` values are dropped to
    keep events terse.
    """
    payload: dict[str, Any] = {
        "ts": _now_iso(),
        "v": SCHEMA_VERSION,
        "event": event,
        "outcome": outcome,
    }
    for key, value in fields.items():
        if value is not None:
            payload[key] = value
    try:
        _SINK(payload)
    except Exception:  # pragma: no cover — telemetry is best-effort, never fatal
        logger.warning("agent.telemetry emit failed for event=%s", event, exc_info=True)


def _outcome_of(result: Any) -> str:
    """Derive an outcome from a platform call's return value.

    Our platform helpers return either a plain dict (Paperclip API JSON) or a
    ``{"ok": bool, ...}`` envelope. A falsy ``ok`` is a *no-op* (e.g. an
    interaction that was already resolved), distinct from a raised error.
    """
    if isinstance(result, dict) and result.get("ok") is False:
        return "noop"
    return "ok"


def _duration_ms(t0: float) -> int:
    return int((time.monotonic() - t0) * 1000)


def instrument(
    event: str,
    *,
    role: str | None = None,
    keys: Sequence[str] = _STANDARD_KEYS,
    result_keys: Sequence[str] = (),
    **static: Any,
) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
    """Decorate an async platform op so it emits a structured event per call.

    Fields are pulled from the call's bound arguments (``keys`` present on the
    signature), plus a static ``role`` and any ``static`` attributes, plus
    ``duration_ms`` and an ``outcome`` derived from the result (or ``error`` on
    exception). ``result_keys`` lifts named keys out of a dict return value onto
    the event (e.g. the ``cancelled`` count from the reaper).

    The emit is wrapped so a telemetry bug can never fail the underlying call.
    """

    def deco(fn: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
        sig = inspect.signature(fn)

        @functools.wraps(fn)
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            try:
                bound = sig.bind(*args, **kwargs)
                bound.apply_defaults()
                arguments = bound.arguments
            except TypeError:
                arguments = {}
            fields: dict[str, Any] = {k: arguments.get(k) for k in keys}
            if role is not None:
                fields["agent_role"] = role
            fields.update(static)

            t0 = time.monotonic()
            try:
                result = await fn(*args, **kwargs)
            except Exception as exc:
                emit(event, outcome="error", error=repr(exc), duration_ms=_duration_ms(t0), **fields)
                raise
            if isinstance(result, dict):
                for rk in result_keys:
                    if rk in result:
                        fields[rk] = result[rk]
            emit(event, outcome=_outcome_of(result), duration_ms=_duration_ms(t0), **fields)
            return result

        return wrapper

    return deco