Merge pull request 'feat(telemetry): אירועי-סוכן מובנים בשער-הפלטפורמה — נראוּת לולאות-recovery (#219)' (#404) from worktree-agent-telemetry into main

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,
+    # ── raw imports: wrapped with telemetry below (state-affecting ops, #219) ──
-    wake_ceo_for_action as pc_wake_ceo_for_action,
+    cancel_interaction as _cancel_interaction,
-    wake_ceo_for_feedback_fold as pc_wake_ceo_for_feedback_fold,
+    cancel_run as _cancel_run,
-    wake_curator_for_final as pc_wake_curator_for_final,
+    post_comment as _post_comment,
-    wake_for_precedent_extraction as pc_wake_for_precedent_extraction,
+    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

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

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