diff --git a/.project/logs/202604071545__telemetry-http-snapshots.md b/.project/logs/202604071545__telemetry-http-snapshots.md new file mode 100644 index 0000000..86621cd --- /dev/null +++ b/.project/logs/202604071545__telemetry-http-snapshots.md @@ -0,0 +1,24 @@ +# Telemetry HTTP Snapshots + +## Scope + +- add bounded structured HTTP response snapshots to action telemetry +- keep the feature generic in public MASE +- document the persisted telemetry surfaces and snapshot limits + +## Changed Surfaces + +- `services/agent-launcher/app/http_snapshot.py` +- `services/agent-launcher/app/config.py` +- `services/agent-launcher/app/executor.py` +- `services/agent-launcher/tests/unit/test_http_snapshot.py` +- `services/agent-launcher/tests/integration/test_heartbeat_endpoint_multiround.py` +- `docs/telemetry.md` +- `docs/run-and-inspect.md` +- `README.md` + +## Notes + +- snapshots are bounded and redacted, not raw body dumps +- controller storage/export path already carries payload JSON, so no controller schema change was needed +- this is intended to support later environment-specific projections such as feed score/rank analysis diff --git a/README.md b/README.md index 3b6ed8b..b505413 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,7 @@ Public-facing docs for the extractable OSS surface now live under `docs/`: - [`docs/concepts.md`](docs/concepts.md) - [`docs/quickstart.md`](docs/quickstart.md) - [`docs/openrouter.md`](docs/openrouter.md) +- [`docs/telemetry.md`](docs/telemetry.md) - [`docs/runtime-contract.md`](docs/runtime-contract.md) - [`docs/environment-contract.md`](docs/environment-contract.md) - [`docs/create-environment.md`](docs/create-environment.md) diff --git a/docs/run-and-inspect.md b/docs/run-and-inspect.md index 55d82e2..52377e4 100644 --- a/docs/run-and-inspect.md +++ b/docs/run-and-inspect.md @@ -35,6 +35,16 @@ Useful API endpoints: - `GET /api/v1/runs/{run_id}/events` - `GET /api/v1/runs/{run_id}/metrics` - `GET /api/v1/runs/{run_id}/scheduler/status` +- `GET /api/v1/telemetry/events/{run_id}` +- `GET /api/v1/telemetry/metrics/{run_id}` + +`action_attempt` telemetry rows now persist two response surfaces for HTTP actions: + +- `response_preview`: short redacted string preview +- `response_snapshot`: bounded structured JSON snapshot for JSON-like bodies + +The snapshot is generic and size-limited. Environments can later derive domain-specific analyses +from it without requiring raw full-body dumps in telemetry. ## Operate diff --git a/docs/telemetry.md b/docs/telemetry.md new file mode 100644 index 0000000..0241814 --- /dev/null +++ b/docs/telemetry.md @@ -0,0 +1,58 @@ +# Telemetry + +MASE persists run-scoped telemetry at two levels: + +- `events`: general run/system events +- `agent_action_events`: per-action rows emitted by the runtime + +The most useful starting point is `action_attempt` telemetry. Those rows already include: + +- action identity (`action_type`, `action_name`, `action_key`) +- request metadata (`method`, `path`, `status_code`, `request_id`) +- timing and success/failure fields +- a bounded `response_preview` + +For JSON-like HTTP responses, MASE now also stores: + +- `response_snapshot` +- `response_snapshot_meta` + +The snapshot is intentionally bounded rather than a raw body dump: + +- nested depth is capped +- dict keys and list items are capped +- long strings are truncated +- sensitive-looking values are redacted + +This makes telemetry usable for downstream analysis without turning the controller database into a +full packet capture. + +## Querying + +Useful endpoints: + +- `GET /api/v1/telemetry/events/{run_id}` +- `GET /api/v1/telemetry/metrics/{run_id}` +- `GET /api/v1/runs/{run_id}/events` + +## Tuning + +Agent-launcher snapshot limits are configurable via environment variables: + +- `AGENT_LAUNCHER_HTTP_RESPONSE_SNAPSHOT_ENABLED` +- `AGENT_LAUNCHER_HTTP_RESPONSE_SNAPSHOT_MAX_DEPTH` +- `AGENT_LAUNCHER_HTTP_RESPONSE_SNAPSHOT_MAX_DICT_KEYS` +- `AGENT_LAUNCHER_HTTP_RESPONSE_SNAPSHOT_MAX_LIST_ITEMS` +- `AGENT_LAUNCHER_HTTP_RESPONSE_SNAPSHOT_MAX_STRING_CHARS` +- `AGENT_LAUNCHER_HTTP_RESPONSE_SNAPSHOT_MAX_TOTAL_NODES` + +## Extension Pattern + +The generic snapshot layer should remain environment-agnostic. + +If an environment needs domain-specific analysis, the recommended pattern is: + +1. capture a bounded generic `response_snapshot` in MASE +2. derive environment-specific projections in the environment repo or downstream analysis pipeline + +That keeps core telemetry reusable while still supporting richer experiment-specific metrics. diff --git a/services/agent-launcher/app/config.py b/services/agent-launcher/app/config.py index 325045a..a7f8e86 100644 --- a/services/agent-launcher/app/config.py +++ b/services/agent-launcher/app/config.py @@ -33,6 +33,12 @@ class Settings(BaseSettings): # Execution Limits max_actions_per_heartbeat: int = 10 http_timeout: int = 30 + http_response_snapshot_enabled: bool = True + http_response_snapshot_max_depth: int = 4 + http_response_snapshot_max_dict_keys: int = 24 + http_response_snapshot_max_list_items: int = 64 + http_response_snapshot_max_string_chars: int = 120 + http_response_snapshot_max_total_nodes: int = 2048 # Service Configuration environment_url: Optional[str] = None diff --git a/services/agent-launcher/app/executor.py b/services/agent-launcher/app/executor.py index 9bd3484..48697f4 100644 --- a/services/agent-launcher/app/executor.py +++ b/services/agent-launcher/app/executor.py @@ -11,6 +11,7 @@ from .action_parser import Action, ActionType, HTTPMethod from .agent_fs import AgentFilesystem from .config import settings +from .http_snapshot import build_response_snapshot from .telemetry_client import ActionCategory, record_action from .scheduler import heartbeat_scheduler @@ -1138,6 +1139,8 @@ async def _record_action_telemetry( response_preview = None response_body_chars = None + response_snapshot = None + response_snapshot_meta = None response_payload = result.get("response") if isinstance(response_payload, dict): body = response_payload.get("body") @@ -1150,6 +1153,15 @@ async def _record_action_telemetry( response_body_chars = len(response_preview) if len(response_preview) > 700: response_preview = f"{response_preview[:697]}..." + response_snapshot, response_snapshot_meta = build_response_snapshot( + body, + enabled=settings.http_response_snapshot_enabled, + max_depth=settings.http_response_snapshot_max_depth, + max_dict_keys=settings.http_response_snapshot_max_dict_keys, + max_list_items=settings.http_response_snapshot_max_list_items, + max_string_chars=settings.http_response_snapshot_max_string_chars, + max_total_nodes=settings.http_response_snapshot_max_total_nodes, + ) action_url = None if hasattr(action, "url"): @@ -1174,6 +1186,8 @@ async def _record_action_telemetry( "error_code": error_code if not success else None, "response_preview": response_preview, "response_body_chars": response_body_chars, + "response_snapshot": response_snapshot, + "response_snapshot_meta": response_snapshot_meta, } await record_action( diff --git a/services/agent-launcher/app/http_snapshot.py b/services/agent-launcher/app/http_snapshot.py new file mode 100644 index 0000000..91671ff --- /dev/null +++ b/services/agent-launcher/app/http_snapshot.py @@ -0,0 +1,172 @@ +"""Bounded structured snapshots for HTTP response telemetry.""" + +from __future__ import annotations + +import re +from typing import Any, Dict, Optional, Tuple + + +SENSITIVE_KEY_PATTERN = re.compile( + r"(api[_-]?key|token|authorization|secret|password|cookie|session|credential|bearer)", + flags=re.IGNORECASE, +) +SENSITIVE_TEXT_PATTERNS = ( + re.compile(r"Bearer\s+[A-Za-z0-9\-._~+/]+=*", flags=re.IGNORECASE), + re.compile( + r"(?i)(api[_-]?key|token|authorization|secret|password|cookie|session|credential)\s*[:=]\s*[^\s,;]+" + ), +) +SNAPSHOT_SCHEMA = "mase.http_response_snapshot.v1" +TRUNCATED_MARKER = "__mase_truncated__" +SUMMARY_MARKER = "__mase_summary__" + + +def _sanitize_text(text: str) -> Tuple[str, bool]: + value = str(text or "") + redacted = False + for pattern in SENSITIVE_TEXT_PATTERNS: + updated = pattern.sub(lambda _match: "***", value) + if updated != value: + redacted = True + value = updated + return value, redacted + + +def _summarize_container(value: Any, reason: str) -> Dict[str, Any]: + size = len(value) if isinstance(value, (dict, list)) else None + kind = "object" if isinstance(value, dict) else "array" if isinstance(value, list) else type(value).__name__ + summary: Dict[str, Any] = {"type": kind, "reason": reason} + if size is not None: + summary["size"] = size + return {SUMMARY_MARKER: summary} + + +def _snapshot_value( + value: Any, + *, + depth: int, + key_hint: Optional[str], + stats: Dict[str, Any], + limits: Dict[str, int], +) -> Any: + stats["nodes"] += 1 + if stats["nodes"] > limits["max_total_nodes"]: + stats["truncated"] = True + return _summarize_container(value, "max_total_nodes") + + if key_hint and SENSITIVE_KEY_PATTERN.search(key_hint): + stats["redacted"] = True + return "***" + + if isinstance(value, str): + text, redacted = _sanitize_text(value) + if redacted: + stats["redacted"] = True + if len(text) > limits["max_string_chars"]: + stats["truncated"] = True + return text[: limits["max_string_chars"]] + "..." + return text + + if isinstance(value, (int, float, bool)) or value is None: + return value + + if depth >= limits["max_depth"]: + stats["truncated"] = True + return _summarize_container(value, "max_depth") + + if isinstance(value, dict): + snapshot: Dict[str, Any] = {} + items = list(value.items()) + for index, (raw_key, raw_value) in enumerate(items): + if index >= limits["max_dict_keys"]: + stats["truncated"] = True + snapshot[TRUNCATED_MARKER] = { + "reason": "max_dict_keys", + "remaining": len(items) - limits["max_dict_keys"], + } + break + key = str(raw_key) + snapshot[key] = _snapshot_value( + raw_value, + depth=depth + 1, + key_hint=key, + stats=stats, + limits=limits, + ) + return snapshot + + if isinstance(value, list): + snapshot = [] + for index, item in enumerate(value): + if index >= limits["max_list_items"]: + stats["truncated"] = True + snapshot.append( + { + TRUNCATED_MARKER: { + "reason": "max_list_items", + "remaining": len(value) - limits["max_list_items"], + } + } + ) + break + snapshot.append( + _snapshot_value( + item, + depth=depth + 1, + key_hint=None, + stats=stats, + limits=limits, + ) + ) + return snapshot + + text, redacted = _sanitize_text(str(value)) + if redacted: + stats["redacted"] = True + if len(text) > limits["max_string_chars"]: + stats["truncated"] = True + return text[: limits["max_string_chars"]] + "..." + return text + + +def build_response_snapshot( + body: Any, + *, + enabled: bool, + max_depth: int, + max_dict_keys: int, + max_list_items: int, + max_string_chars: int, + max_total_nodes: int, +) -> Tuple[Optional[Any], Optional[Dict[str, Any]]]: + """Return a bounded structured snapshot for JSON-like HTTP bodies.""" + if not enabled or not isinstance(body, (dict, list)): + return None, None + + limits = { + "max_depth": max(1, int(max_depth)), + "max_dict_keys": max(1, int(max_dict_keys)), + "max_list_items": max(1, int(max_list_items)), + "max_string_chars": max(16, int(max_string_chars)), + "max_total_nodes": max(32, int(max_total_nodes)), + } + stats: Dict[str, Any] = {"nodes": 0, "truncated": False, "redacted": False} + snapshot = _snapshot_value( + body, + depth=0, + key_hint=None, + stats=stats, + limits=limits, + ) + + meta = { + "schema": SNAPSHOT_SCHEMA, + "body_type": "object" if isinstance(body, dict) else "array", + "truncated": bool(stats["truncated"]), + "redacted": bool(stats["redacted"]), + "nodes_captured": int(stats["nodes"]), + "limits": limits, + "top_level_items_original": len(body), + "top_level_items_captured": len(snapshot) if isinstance(snapshot, (dict, list)) else None, + } + return snapshot, meta diff --git a/services/agent-launcher/tests/integration/test_heartbeat_endpoint_multiround.py b/services/agent-launcher/tests/integration/test_heartbeat_endpoint_multiround.py index 334e80e..d1ff44e 100644 --- a/services/agent-launcher/tests/integration/test_heartbeat_endpoint_multiround.py +++ b/services/agent-launcher/tests/integration/test_heartbeat_endpoint_multiround.py @@ -487,6 +487,7 @@ async def test_heartbeat_endpoint_emits_prompt_llm_and_action_telemetry( ) -> None: emitted_action_types = [] emitted_event_types = [] + action_attempt_payloads = [] async def _capture_record_action(*args, **kwargs): action_type = kwargs.get("action_type") @@ -496,6 +497,8 @@ async def _capture_record_action(*args, **kwargs): event_type = payload.get("event_type") if event_type: emitted_event_types.append(str(event_type)) + if event_type == "action_attempt": + action_attempt_payloads.append(payload) monkeypatch.setattr(launcher_module, "record_action", _capture_record_action) monkeypatch.setattr(executor_module, "record_action", _capture_record_action) @@ -521,6 +524,30 @@ async def _local_dummy_messages(self, messages, tools=None): _local_dummy_messages, ) + async def _fake_execute_direct_http(self, action, method, headers, body): + return { + "action": "http", + "success": True, + "action_type": "http_get", + "action_name": "local contract read", + "action_key": "http_get:local contract read", + "method": method.value, + "path": action.url, + "status_code": 200, + "request_id": "req-test", + "response": { + "status_code": 200, + "headers": {"content-type": "application/json"}, + "body": { + "contract": "ok", + "items": [{"id": "p1", "score": 3}, {"id": "p2", "score": 5}], + }, + }, + "timestamp": "2026-01-01T00:00:00Z", + } + + monkeypatch.setattr(executor_module.ActionExecutor, "_execute_direct_http", _fake_execute_direct_http) + transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: create = await client.post( @@ -550,6 +577,12 @@ async def _local_dummy_messages(self, messages, tools=None): assert "prompt_part" in emitted_action_types assert "llm_io" in emitted_action_types assert "action_attempt" in emitted_event_types + assert action_attempt_payloads + assert any(item.get("response_snapshot") is not None for item in action_attempt_payloads) + assert any( + (item.get("response_snapshot_meta") or {}).get("schema") == "mase.http_response_snapshot.v1" + for item in action_attempt_payloads + ) class LegacyHeartbeatResponse(BaseModel): diff --git a/services/agent-launcher/tests/unit/test_http_snapshot.py b/services/agent-launcher/tests/unit/test_http_snapshot.py new file mode 100644 index 0000000..f709593 --- /dev/null +++ b/services/agent-launcher/tests/unit/test_http_snapshot.py @@ -0,0 +1,53 @@ +"""Unit tests for bounded HTTP response telemetry snapshots.""" + +from app.http_snapshot import SNAPSHOT_SCHEMA, TRUNCATED_MARKER, build_response_snapshot + + +def test_build_response_snapshot_redacts_sensitive_keys_and_truncates_strings() -> None: + snapshot, meta = build_response_snapshot( + { + "posts": [ + { + "id": "p1", + "score": 7, + "title": "A" * 80, + "api_token": "TOPSECRET", + } + ] + }, + enabled=True, + max_depth=4, + max_dict_keys=8, + max_list_items=8, + max_string_chars=24, + max_total_nodes=128, + ) + + assert meta is not None + assert meta["schema"] == SNAPSHOT_SCHEMA + assert meta["redacted"] is True + assert meta["truncated"] is True + assert snapshot["posts"][0]["id"] == "p1" + assert snapshot["posts"][0]["score"] == 7 + assert snapshot["posts"][0]["api_token"] == "***" + assert snapshot["posts"][0]["title"].endswith("...") + assert len(snapshot["posts"][0]["title"]) == 27 + + +def test_build_response_snapshot_limits_large_lists() -> None: + snapshot, meta = build_response_snapshot( + {"items": [{"id": f"p{i}", "score": i} for i in range(5)]}, + enabled=True, + max_depth=4, + max_dict_keys=8, + max_list_items=3, + max_string_chars=40, + max_total_nodes=128, + ) + + assert meta is not None + assert meta["truncated"] is True + items = snapshot["items"] + assert len(items) == 4 + assert items[-1][TRUNCATED_MARKER]["reason"] == "max_list_items" + assert items[-1][TRUNCATED_MARKER]["remaining"] == 2