From 1f903e1b7cd4b8c07a45a3e17b376b659c671eb9 Mon Sep 17 00:00:00 2001 From: nzinci Date: Thu, 30 Jul 2026 22:24:25 +0300 Subject: [PATCH] Add sanitized fetch health for agents --- backend/routers/ai_intel.py | 8 +++ backend/services/fetch_health.py | 38 +++++++++++ backend/services/openclaw_channel.py | 5 ++ backend/tests/test_fetch_health.py | 65 ++++++++++++++++++- backend/tests/test_openclaw_route_security.py | 26 ++++++++ backend/tests/test_openclaw_skill_env.py | 23 +++++++ openclaw-skills/shadowbroker/SKILL.md | 1 + openclaw-skills/shadowbroker/sb_query.py | 5 ++ 8 files changed, 170 insertions(+), 1 deletion(-) diff --git a/backend/routers/ai_intel.py b/backend/routers/ai_intel.py index 2594bb4..60ea84d 100644 --- a/backend/routers/ai_intel.py +++ b/backend/routers/ai_intel.py @@ -1608,6 +1608,13 @@ async def agent_tool_manifest(request: Request): "parameters": {}, "returns": "{counts: {...}, available_layers: [...], non_empty_layers: [...], layer_aliases: {...}, last_updated, version}", }, + { + "name": "get_fetch_health", + "type": "read", + "description": "Get process-local outcomes for instrumented fetch and maintenance tasks. Reports counters, timestamps, latest condition, and duration without raw error text. This is not a data-freshness or upstream-availability check.", + "parameters": {}, + "returns": "{scope: 'process', persistent: false, observed_only: true, semantics: 'latest_recorded_task_outcome', tasks: {...}}", + }, { "name": "get_layer_slice", "type": "read", @@ -2431,6 +2438,7 @@ async def api_capabilities(request: Request): "get_telemetry": {"args": {}, "description": "All live fast-refresh data (flights, ships, sigint, earthquakes, weather, CCTV, etc)"}, "get_slow_telemetry": {"args": {}, "description": "Slow-refresh data (prediction markets, news, military bases, power plants, volcanoes, etc)"}, "get_summary": {"args": {}, "description": "Counts and discovery metadata for all live telemetry layers, including available layer names and common aliases."}, + "get_fetch_health": {"args": {}, "description": "Process-local outcomes for instrumented fetch and maintenance tasks, without raw error text. Condition reflects the latest recorded task outcome, not data freshness or upstream availability."}, "get_layer_slice": { "args": {"layers": "list[str]", "limit_per_layer": "int (optional, omit or <=0 for full layer)", "since_version": "int (optional)"}, "description": "Fetch only selected top-level layers. Accepts aliases such as gfw/global_fishing_watch → fishing_activity. If since_version matches current version, returns changed=false and no layer payload.", diff --git a/backend/services/fetch_health.py b/backend/services/fetch_health.py index f342aa7..588d749 100644 --- a/backend/services/fetch_health.py +++ b/backend/services/fetch_health.py @@ -92,3 +92,41 @@ def get_health_snapshot() -> Dict[str, Dict[str, Any]]: """Return a snapshot of current fetch health state.""" with _lock: return {k: dict(v) for k, v in _health.items()} + + +def get_agent_health_snapshot() -> Dict[str, Any]: + """Return process-local task outcomes without stored error details.""" + tasks = {} + for source, entry in get_health_snapshot().items(): + last_ok = entry.get("last_ok") + last_error = entry.get("last_error") + condition = "unknown" + if isinstance(last_ok, str) and last_ok and last_error is None: + condition = "healthy" + elif isinstance(last_error, str) and last_error and last_ok is None: + condition = "degraded" + elif ( + isinstance(last_ok, str) + and last_ok + and isinstance(last_error, str) + and last_error + ): + if last_ok > last_error: + condition = "healthy" + elif last_error > last_ok: + condition = "degraded" + tasks[source] = { + "condition": condition, + "ok_count": entry.get("ok_count"), + "error_count": entry.get("error_count"), + "last_ok": last_ok, + "last_error": last_error, + "last_duration_ms": entry.get("last_duration_ms"), + } + return { + "scope": "process", + "persistent": False, + "observed_only": True, + "semantics": "latest_recorded_task_outcome", + "tasks": tasks, + } diff --git a/backend/services/openclaw_channel.py b/backend/services/openclaw_channel.py index ad6d877..e723664 100644 --- a/backend/services/openclaw_channel.py +++ b/backend/services/openclaw_channel.py @@ -53,6 +53,7 @@ READ_COMMANDS = frozenset({ "get_telemetry", "get_slow_telemetry", "get_summary", + "get_fetch_health", "get_report", "get_layer_slice", "find_flights", @@ -746,6 +747,10 @@ def _dispatch_command(cmd: str, args: dict[str, Any]) -> dict[str, Any]: summary = get_telemetry_summary() return {"ok": True, "data": summary, "version": summary.get("version")} + if cmd == "get_fetch_health": + from services.fetch_health import get_agent_health_snapshot + return {"ok": True, "data": get_agent_health_snapshot()} + if cmd == "get_layer_slice": from services.telemetry import get_layer_slice layers = args.get("layers") or [] diff --git a/backend/tests/test_fetch_health.py b/backend/tests/test_fetch_health.py index 31c4516..837f45a 100644 --- a/backend/tests/test_fetch_health.py +++ b/backend/tests/test_fetch_health.py @@ -1,4 +1,23 @@ -from services.fetch_health import record_success, record_failure, get_health_snapshot +import json + +import pytest + +from services import fetch_health +from services.fetch_health import ( + get_agent_health_snapshot, + get_health_snapshot, + record_failure, + record_success, +) + + +@pytest.fixture(autouse=True) +def clear_fetch_health(): + with fetch_health._lock: + fetch_health._health.clear() + yield + with fetch_health._lock: + fetch_health._health.clear() def test_record_success_and_failure(): @@ -13,3 +32,47 @@ def test_record_success_and_failure(): assert entry["last_ok"] is not None assert entry["last_error"] is not None assert entry["last_duration_ms"] is not None + + +def test_agent_health_snapshot_is_sanitized(): + fake_secret = "RECOGNIZABLE_FAKE_SECRET_FOR_FETCH_HEALTH" + entry = { + "ok_count": 3, + "error_count": 2, + "last_ok": "2026-07-30T12:00:00.000000", + "last_error": "2026-07-30T11:00:00.000000", + "last_error_msg": f"upstream failed with {fake_secret}", + "last_duration_ms": 125.4, + "avg_duration_ms": 98.7, + "last_count": 42, + } + with fetch_health._lock: + fetch_health._health["latest_success"] = dict(entry) + fetch_health._health["latest_failure"] = { + **entry, + "last_ok": "2026-07-30T10:00:00.000000", + "last_error": "2026-07-30T13:00:00.000000", + } + + snapshot = get_agent_health_snapshot() + + assert {key: value for key, value in snapshot.items() if key != "tasks"} == { + "scope": "process", + "persistent": False, + "observed_only": True, + "semantics": "latest_recorded_task_outcome", + } + safe_fields = { + "condition", + "ok_count", + "error_count", + "last_ok", + "last_error", + "last_duration_ms", + } + assert set(snapshot["tasks"]["latest_success"]) == safe_fields + assert snapshot["tasks"]["latest_success"]["condition"] == "healthy" + assert snapshot["tasks"]["latest_failure"]["condition"] == "degraded" + serialized = json.dumps(snapshot) + assert "last_error_msg" not in serialized + assert fake_secret not in serialized diff --git a/backend/tests/test_openclaw_route_security.py b/backend/tests/test_openclaw_route_security.py index 8185ba7..7902d9a 100644 --- a/backend/tests/test_openclaw_route_security.py +++ b/backend/tests/test_openclaw_route_security.py @@ -130,6 +130,32 @@ class TestAuthenticatedRequestSucceeds: tool_names = {tool["name"] for tool in data["tools"]} assert set(data["available_commands"]).issubset(tool_names) + def test_fetch_health_read_command_dispatch_and_discovery(self, remote_client): + from services.openclaw_channel import READ_COMMANDS, _dispatch_command + + snapshot = { + "scope": "process", + "persistent": False, + "observed_only": True, + "semantics": "latest_recorded_task_outcome", + "tasks": {}, + } + with patch( + "services.fetch_health.get_agent_health_snapshot", return_value=snapshot + ): + result = _dispatch_command("get_fetch_health", {}) + + assert "get_fetch_health" in READ_COMMANDS + assert result == {"ok": True, "data": snapshot} + + headers = _sign("GET", "/api/ai/tools") + r = remote_client.get("/api/ai/tools", headers=headers) + assert r.status_code == 200, r.text + tool = next( + tool for tool in r.json()["tools"] if tool["name"] == "get_fetch_health" + ) + assert tool["parameters"] == {} + # --------------------------------------------------------------------------- # 2. Tampered body rejected (P1A body-binding at route layer) diff --git a/backend/tests/test_openclaw_skill_env.py b/backend/tests/test_openclaw_skill_env.py index 9318c44..b27c5e0 100644 --- a/backend/tests/test_openclaw_skill_env.py +++ b/backend/tests/test_openclaw_skill_env.py @@ -1,5 +1,6 @@ """Regression coverage for OpenClaw skill HMAC environment names.""" +import asyncio import importlib.util from pathlib import Path @@ -36,3 +37,25 @@ def test_openclaw_skill_accepts_legacy_key_as_hmac_secret_alias(monkeypatch): assert "X-SB-Signature" in headers assert "Authorization" not in headers assert "X-Admin-Key" not in headers + + +def test_openclaw_skill_get_fetch_health_unwraps_command_result(monkeypatch): + module = _load_sb_query(monkeypatch) + client = module.ShadowBrokerClient() + commands = [] + + async def send_command(cmd, args=None): + commands.append((cmd, args)) + return { + "result": { + "ok": True, + "data": {"scope": "process", "tasks": {}}, + } + } + + monkeypatch.setattr(client, "send_command", send_command) + + result = asyncio.run(client.get_fetch_health()) + + assert commands == [("get_fetch_health", None)] + assert result == {"scope": "process", "tasks": {}} diff --git a/openclaw-skills/shadowbroker/SKILL.md b/openclaw-skills/shadowbroker/SKILL.md index 3cf9b6e..84034ac 100644 --- a/openclaw-skills/shadowbroker/SKILL.md +++ b/openclaw-skills/shadowbroker/SKILL.md @@ -169,6 +169,7 @@ The channel operates over HMAC-authenticated HTTP with body-integrity binding: | `sb.stream_updates()` | SSE push: `layer_changed`, alerts, tasks | **Open first, keep open** — tells you exactly which layers updated | | `await sb.get_layer_slice(["ships", "gdelt"])` | Only the requested layers, with per-layer incremental | **Primary fetch method** — automatically skips layers you already have | | `await sb.send_command("get_summary")` | Lightweight counts-only summary | Discover what data exists before pulling anything | +| `await sb.get_fetch_health()` | Sanitized process-local outcomes for instrumented fetch and maintenance tasks | Results reset on backend restart; condition is the latest recorded task completion, not data freshness | | `await sb.ask("...")` | **Route + execute** | **Default** for natural-language reads | | `await sb.send_command("get_entity_profile", {...})` | **Preferred aircraft/vessel dossier** — identity, VIP tags, trail, route, ACARS, jamming/correlations, news | Tail, owner, callsign, MMSI | | `await sb.send_command("get_entity_trail", {...})` | Observed path + route + ACARS hints only | When you don't need full dossier | diff --git a/openclaw-skills/shadowbroker/sb_query.py b/openclaw-skills/shadowbroker/sb_query.py index 0b12e14..94ebfa8 100644 --- a/openclaw-skills/shadowbroker/sb_query.py +++ b/openclaw-skills/shadowbroker/sb_query.py @@ -294,6 +294,11 @@ class ShadowBrokerClient: return data if isinstance(data, dict) else {} return result + async def get_fetch_health(self) -> dict: + """Get sanitized process-local fetch task outcomes.""" + response = await self.send_command("get_fetch_health") + return self.unwrap_channel_result(response) + async def route_query( self, text: str,