Chaim
5dbe970dbc
מוסיף web/agent_telemetry.py — emitter אגנוסטי-לפלטפורמה שפולט אירוע JSON
מובנה לכל פעולת-פלטפורמה משנת-מצב (wakeup/comment/interaction-reap/cancel/
run-cancel/session-reset). מטרה: להפוך דיבוג לולאות-recovery מפורנזיקה-ידנית
מול Paperclip DB (agent_wakeup_requests/heartbeat_runs) לנראוּת בזמן-אמת.
- הפליטה מחווטת ב-web/agent_platform_port.py בלבד (התפר היחיד) — כל pc_* משנה-מצב
נעטף ב-instrument(); השמות הציבוריים והחתימות נשמרים, app.py לא משתנה.
- אגנוסטי-לפלטפורמה (G12): agent_telemetry לא מייבא Paperclip; אירועים בשמות-דומיין
שורדים החלפת-פלטפורמה. מסלול-פליטה יחיד (G2).
- OTLP-ready ("logs first"): schema יציב + _SINK מבודד להחלפה עתידית בלי שינוי call-site.
- error מ-op נפלט כאירוע ואז נזרק-מחדש — אין בליעה שקטה (§6).
- store-UTC (feedback_israel_time). 6 בדיקות pytest (web/tests/test_agent_telemetry.py).
Invariants: מקיים G2 (מסלול-טלמטריה יחיד), G12/INV-PORT1 (פליטה מהשער, ליבה נקייה),
כלל-הנדסה §6 (אין בליעה שקטה). מזין את #222 (טקסונומיית-בריאות).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""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")
|