diff --git a/web/agent_platform_port.py b/web/agent_platform_port.py index 2579545..90a872e 100644 --- a/web/agent_platform_port.py +++ b/web/agent_platform_port.py @@ -33,12 +33,11 @@ from web.paperclip_api import ( pc_request, require_paperclip_db_url, ) +from web.agent_telemetry import instrument from web.paperclip_client import ( COMPANIES as PAPERCLIP_COMPANIES, accept_interaction as pc_accept_interaction, archive_project as pc_archive_project, - cancel_interaction as pc_cancel_interaction, - cancel_run as pc_cancel_run, create_project as pc_create_project, create_workflow_issue as pc_create_workflow_issue, get_agents_for_case as pc_get_agents_for_case, @@ -50,25 +49,75 @@ from web.paperclip_client import ( get_run_events as pc_get_run_events, get_run_log as pc_get_run_log, list_live_runs as pc_list_live_runs, - post_comment as pc_post_comment, - reap_stale_interactions as pc_reap_stale_interactions, reject_interaction as pc_reject_interaction, - reset_agent_session as pc_reset_agent_session, - reset_case_agents as pc_reset_case_agents, respond_to_interaction as pc_respond_to_interaction, restore_project as pc_restore_project, update_project_name as pc_update_project_name, - wake_analyst_for_appraiser_facts as pc_wake_analyst_for_appraiser_facts, - wake_analyst_for_argument_aggregation as pc_wake_analyst_for_argument_aggregation, - wake_analyst_for_protocol_analysis as pc_wake_analyst_for_protocol_analysis, get_generation_run_status as pc_get_generation_run_status, - wake_ceo_agent as pc_wake_ceo, - wake_ceo_for_action as pc_wake_ceo_for_action, - wake_ceo_for_feedback_fold as pc_wake_ceo_for_feedback_fold, - wake_curator_for_final as pc_wake_curator_for_final, - wake_for_precedent_extraction as pc_wake_for_precedent_extraction, + # ── raw imports: wrapped with telemetry below (state-affecting ops, #219) ── + cancel_interaction as _cancel_interaction, + cancel_run as _cancel_run, + post_comment as _post_comment, + reap_stale_interactions as _reap_stale_interactions, + reset_agent_session as _reset_agent_session, + reset_case_agents as _reset_case_agents, + wake_analyst_for_appraiser_facts as _wake_analyst_for_appraiser_facts, + wake_analyst_for_argument_aggregation as _wake_analyst_for_argument_aggregation, + wake_analyst_for_protocol_analysis as _wake_analyst_for_protocol_analysis, + wake_ceo_agent as _wake_ceo, + wake_ceo_for_action as _wake_ceo_for_action, + wake_ceo_for_feedback_fold as _wake_ceo_for_feedback_fold, + wake_curator_for_final as _wake_curator_for_final, + wake_for_precedent_extraction as _wake_for_precedent_extraction, ) +# ── telemetry-wrapped platform ops (#219 / docs/spec/X15) ─────────────────── +# Every state-affecting platform op the Port exposes emits one structured event +# through web.agent_telemetry, so recovery-loop behaviour (repeated wakeups, +# stranded dispositions) is observable in real time rather than reconstructed +# from the Paperclip DB. The wrappers preserve the public ``pc_*`` names and +# signatures — call sites in app.py are unchanged. Read-only observability ops +# (list_live_runs / get_run_log / get_issue_comments …) are intentionally NOT +# instrumented: they neither wake agents nor change disposition state. +pc_wake_ceo = instrument("agent.wakeup", role="ceo")(_wake_ceo) +pc_wake_ceo_for_action = instrument( + "agent.wakeup", role="ceo", keys=("case_number", "company_id", "action"), +)(_wake_ceo_for_action) +pc_wake_ceo_for_feedback_fold = instrument( + "agent.wakeup", role="ceo", keys=("feedback_id", "category", "block_id"), +)(_wake_ceo_for_feedback_fold) +pc_wake_curator_for_final = instrument( + "agent.wakeup", role="curator", keys=("case_number", "company_id", "task"), +)(_wake_curator_for_final) +pc_wake_for_precedent_extraction = instrument( + "agent.wakeup", role="ceo", keys=("case_law_id", "citation", "practice_area"), +)(_wake_for_precedent_extraction) +pc_wake_analyst_for_appraiser_facts = instrument( + "agent.wakeup", role="analyst", +)(_wake_analyst_for_appraiser_facts) +pc_wake_analyst_for_argument_aggregation = instrument( + "agent.wakeup", role="analyst", +)(_wake_analyst_for_argument_aggregation) +pc_wake_analyst_for_protocol_analysis = instrument( + "agent.wakeup", role="analyst", keys=("case_number", "company_id", "document_id"), +)(_wake_analyst_for_protocol_analysis) +pc_post_comment = instrument( + "agent.comment", keys=("issue_id", "company_id"), +)(_post_comment) +pc_cancel_interaction = instrument( + "interaction.cancelled", keys=("issue_id", "interaction_id"), +)(_cancel_interaction) +pc_reap_stale_interactions = instrument( + "interaction.reaped", result_keys=("cancelled",), +)(_reap_stale_interactions) +pc_cancel_run = instrument("run.cancelled", keys=("run_id",))(_cancel_run) +pc_reset_agent_session = instrument( + "agent.session_reset", keys=("agent_id",), +)(_reset_agent_session) +pc_reset_case_agents = instrument( + "agent.session_reset", keys=("case_number",), +)(_reset_case_agents) + # ── domain-named lifecycle aliases (preferred for new call sites) ─────────── archive_case_project = pc_archive_project restore_case_project = pc_restore_project diff --git a/web/agent_telemetry.py b/web/agent_telemetry.py new file mode 100644 index 0000000..7b8bca4 --- /dev/null +++ b/web/agent_telemetry.py @@ -0,0 +1,157 @@ +"""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 diff --git a/web/tests/test_agent_telemetry.py b/web/tests/test_agent_telemetry.py new file mode 100644 index 0000000..0e5844c --- /dev/null +++ b/web/tests/test_agent_telemetry.py @@ -0,0 +1,104 @@ +"""Tests for #219 — web.agent_telemetry structured agent-platform events. + +Verifies the emitter contract the recovery-loop observability relies on: +- ``emit`` stamps ts/v/event/outcome and drops ``None`` fields. +- ``instrument`` lifts standard + result keys, times the call, and derives the + outcome (ok / noop for ``{"ok": False}`` / error on exception). +- an exception is re-raised after the error event — the wrapped op is **never + silently swallowed** (constitution §6 / INV-G4). +- the decorator is transparent: wrapped functions stay coroutine functions. + +Pure-stdlib module (no Paperclip/web deps), so this runs without importorskip. +""" + +from __future__ import annotations + +import asyncio +import inspect +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # web/ + +from agent_telemetry import SCHEMA_VERSION, emit, instrument # noqa: E402 +import agent_telemetry # noqa: E402 + + +@pytest.fixture +def captured(monkeypatch): + """Intercept the sink; yield the list of emitted event dicts.""" + events: list[dict] = [] + monkeypatch.setattr(agent_telemetry, "_SINK", events.append) + return events + + +def test_emit_stamps_schema_and_drops_none(captured): + emit("x.test", role="ceo", case_number="8125-09-24", issue_id=None) + (e,) = captured + assert e["event"] == "x.test" + assert e["v"] == SCHEMA_VERSION + assert e["outcome"] == "ok" + assert e["ts"].endswith("Z") + assert e["case_number"] == "8125-09-24" + assert "issue_id" not in e # None fields dropped + + +def test_instrument_success_lifts_result_keys(captured): + @instrument("interaction.reaped", result_keys=("cancelled",)) + async def reap(): + return {"ok": True, "cancelled": 3} + + result = asyncio.run(reap()) + assert result == {"ok": True, "cancelled": 3} + (e,) = captured + assert e["event"] == "interaction.reaped" + assert e["outcome"] == "ok" + assert e["cancelled"] == 3 + assert "duration_ms" in e + + +def test_instrument_ok_false_is_noop(captured): + @instrument("interaction.cancelled", keys=("issue_id",)) + async def cancel(issue_id, interaction_id): + return {"ok": False, "error": "already resolved"} + + asyncio.run(cancel("iss-1", "int-9")) + (e,) = captured + assert e["outcome"] == "noop" + assert e["issue_id"] == "iss-1" + + +def test_instrument_reraises_after_error_event(captured): + @instrument("agent.wakeup", role="ceo") + async def wake(case_number, company_id=""): + raise RuntimeError("boom") + + with pytest.raises(RuntimeError, match="boom"): + asyncio.run(wake("1043-04-26", company_id="cmp")) + + (e,) = captured + assert e["outcome"] == "error" + assert "boom" in e["error"] + assert e["agent_role"] == "ceo" + assert e["case_number"] == "1043-04-26" + assert e["company_id"] == "cmp" + + +def test_decorator_is_transparent(): + @instrument("agent.wakeup", role="analyst") + async def wake(case_number, company_id=""): + return {"ok": True} + + assert inspect.iscoroutinefunction(wake) + assert wake.__name__ == "wake" + + +def test_emit_never_raises_on_sink_failure(monkeypatch): + def boom(_payload): + raise ValueError("sink down") + + monkeypatch.setattr(agent_telemetry, "_SINK", boom) + # Must not propagate — telemetry is best-effort, never fatal to the op. + emit("agent.wakeup", role="ceo")