feat: add bounded HTTP response snapshots to telemetry

.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

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)

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
 

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.

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

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(

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

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