Spaces:
Running
Running
File size: 4,030 Bytes
a6a5d8e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | # SPDX-License-Identifier: Apache-2.0 · Doctrine v12 (additive). Yachay.
"""
status_feed — internal health -> public status_feed.json (fail-closed allow-list).
Reads Prometheus + active Alertmanager alerts + degradation receipts; emits ONLY the
szl.status_feed/v1 schema. Anything not explicitly mapped is dropped (never leaked).
Honest anti-cover-up: driven by the SAME Prometheus signals as the internal dashboard,
so the public page can never claim green while internally red. v11 LOCKED untouched.
"""
from datetime import datetime, timezone
# allow-list: internal flagship -> public component
PUBLIC_COMPONENT = {
"a11oy": "Governance & Brand", "amaru": "Memory / Cortex",
"sentra": "Immune / Policy", "vessels": "Maritime & Receipts",
"rosie": "Companion", "killinchu": "Drone Ops",
"lean-kernel": "Proof Kernel",
}
# Keys that MUST NEVER appear in the public feed (defense in depth; allow-list already
# drops them). We match on KEY NAMES (not a substring of the whole blob) so legitimate
# component copy like "Receipts"/"Companion" is never falsely flagged.
_NEVER_PUBLISH_KEYS = frozenset({
"provider", "model", "tripwire", "khipu_node", "digest",
"hostname", "ip", "secret", "token", "breaker", "circuit",
})
def _coarse_status(up: bool, degraded: bool, partial: bool) -> str:
if not up: return "major_outage"
if partial: return "partial_outage"
if degraded: return "degraded"
return "operational"
def build_feed(metrics: dict, alerts: list[dict]) -> dict:
components = []
for fl, comp in PUBLIC_COMPONENT.items():
up = metrics.get(f"szl_up::{fl}", 0) == 1
degraded = any(a for a in alerts
if a.get("flagship") == fl and a.get("impact") == "degraded")
components.append({
"name": comp,
"status": _coarse_status(up, degraded, partial=False),
"uptime_30d": round(metrics.get(f"szl_uptime_30d::{fl}", 0.0), 2),
})
# AI Responses component derived from router tiers (impact only, no provider names)
router_degraded = (metrics.get("szl_router_tier::T0_cache", 0)
+ metrics.get("szl_router_tier::T1_small", 0)) > 0
components.append({"name": "AI Responses",
"status": "degraded" if router_degraded else "operational",
"note": "Responses may be slower than usual." if router_degraded else None})
overall = "operational"
if any(c["status"] == "major_outage" for c in components): overall = "major_outage"
elif any(c["status"] == "partial_outage" for c in components): overall = "partial_outage"
elif any(c["status"] == "degraded" for c in components): overall = "degraded"
feed = {"schema": "szl.status_feed/v1",
"generated_at": datetime.now(timezone.utc).isoformat(),
"overall": overall, "components": components,
"active_incidents": _public_incidents(alerts),
"scheduled_maintenance": []}
_assert_no_leak(feed) # fail-closed: refuse to emit if any banned key present
return feed
def _public_incidents(alerts):
out = []
for a in alerts:
if not a.get("customer_impacting"): # only customer-impacting alerts go public
continue
out.append({"id": a["incident_id"], "title": a["public_title"], # pre-sanitized
"impact": a["impact"], "started_at": a["started_at"],
"latest_update": a["public_update"]})
return out
def _assert_no_leak(node) -> None:
"""Recursively assert no banned KEY appears anywhere in the feed (fail-closed)."""
if isinstance(node, dict):
for k, v in node.items():
if str(k).lower() in _NEVER_PUBLISH_KEYS:
raise RuntimeError(
f"status_feed leak guard tripped on key '{k}' — refusing to publish")
_assert_no_leak(v)
elif isinstance(node, (list, tuple)):
for item in node:
_assert_no_leak(item)
|