"""GET /api/cortex-os/runtime-health - Cortex OS Runtime Health slice. Public API surface used: api.register_route, api.logger. No forced internal dependencies. This module does not import Hermes internals. """ from __future__ import annotations import json import re from typing import Any ROUTE_PATH = "/api/cortex-os/runtime-health" ROUTE_METHOD = "GET" CONTRACT_ID = "runtime-health/v0.1" CHECKED_AT = "2026-05-29T00:00:00Z" SCHEMA_VERSION = "0.1.0" HOST_RUNTIME_ID = "webui" HOST_ADAPTER_ID = "hermes" HOST_ADAPTER_KIND = "development_host" _FORBIDDEN_TEXT = re.compile( r"(https?://|/home/|workspaces/|\b\d{2,5}\b|token|secret|cookie|traceback|exception|\.env)", re.IGNORECASE, ) def register(api: Any) -> None: """Wire the read-only Runtime Health route.""" log = api.logger("svrnty.routes.cortex_os_runtime_health") api.register_route(ROUTE_PATH, ROUTE_METHOD, _handle_runtime_health) log.info("cortex os runtime health endpoint registered") def _handle_runtime_health(handler: Any, parsed: Any) -> bool: """Handler signature matches the plugin loader contract.""" if getattr(handler, "command", ROUTE_METHOD) != ROUTE_METHOD: _write_json(handler, 405, _error_envelope("method_not_allowed", "read only route")) return True if getattr(parsed, "query", ""): _write_json(handler, 400, _error_envelope("query_not_allowed", "query targets are not accepted")) return True _write_json(handler, 200, {"ok": True, "result": runtime_health_payload(), "error": None}) return True def runtime_health_payload(host_signals: dict[str, Any] | None = None) -> dict[str, Any]: """Return the host-neutral Runtime Health envelope.""" signals = _summarize_host_signals(host_signals or {}) status = _derive_status(signals) return { "schema_version": SCHEMA_VERSION, "cortex_os_contract_id": CONTRACT_ID, "host_runtime_id": HOST_RUNTIME_ID, "host_adapter_id": HOST_ADAPTER_ID, "host_adapter_kind": HOST_ADAPTER_KIND, "checked_at": CHECKED_AT, "status": status, "readiness": "not_configured", "signals": signals, "display_summary": _display_summary_for(status), "redactions": [ {"target_path": "signals", "reason_code": "no_raw_host_payload", "label": "raw host payloads omitted"}, {"target_path": "signals.summary", "reason_code": "path_redacted", "label": "raw paths redacted"}, {"target_path": "signals.summary", "reason_code": "secret_redacted", "label": "secrets redacted"}, { "target_path": "source_trace", "reason_code": "request_response_redacted", "label": "request and response details omitted", }, ], "warnings": [ { "code": "deterministic_host_surface_inputs_only", "message": "Runtime Health uses deterministic declared host-surface inputs only.", } ], "errors": [], "source_trace": [ { "source_id": "runtime-health-contract", "source_kind": "sot_document", "label": "Runtime Health Contract", }, { "source_id": "hermes-host-adapter-contract", "source_kind": "host_adapter_contract", "label": "Hermes Host Adapter Contract", }, { "source_id": "hermes-runtime-health-slice", "source_kind": "validator", "label": "Hermes Runtime Health Slice", }, ], "authority": { "runtime_coding": False, "hermes_source_edits": False, "hermes_host_adapter_authority_map": False, "hermes_host_adapter_implementation": False, "webui_plugin_implementation": False, "local_json_api_route_files": False, "local_json_api_route_handlers": False, "browser_source": False, "host_runtime_start": False, "runtime_process_behavior": False, "runtime_state_mutation": False, "product_ui_implementation": False, "display_source": False, "tool_callable_authority": False, "mcp_server_runtime_behavior": False, "mcp_tool_exposure": False, "profile_exposure_broadening": False, "memory_domain_access": False, "delegated_memory_grants": False, "sharing": False, "installer_automation": False, "source_import": False, "forced_internal_upstream_dependency": False, "live_smoke_execution": False, "product_readiness_claim": False, }, } def _summarize_host_signals(host_signals: dict[str, Any]) -> list[dict[str, str]]: names = ["health", "agent_health", "dashboard_status"] if not host_signals: return [ {"signal_id": name, "state": "unknown", "summary": "not configured", "redacted": False} for name in names ] return [ _summarize_signal(name, host_signals) for name in names ] def _summarize_signal(name: str, host_signals: dict[str, Any]) -> dict[str, Any]: summary, redacted = _bounded_text(host_signals.get(f"{name}_detail", "declared surface")) return { "signal_id": name, "state": _clean_status(host_signals.get(name, "unknown")), "summary": summary, "redacted": redacted, } def _derive_status(signals: list[dict[str, str]]) -> str: statuses = {signal["state"] for signal in signals} if "unavailable" in statuses: return "unavailable" if "degraded" in statuses: return "degraded" if statuses == {"healthy"}: return "healthy" return "unknown" def _clean_status(value: Any) -> str: text = str(value).strip().lower() return text if text in {"healthy", "degraded", "unavailable", "unknown"} else "unknown" def _display_summary_for(status: str) -> dict[str, str]: if status == "healthy": return { "headline": "Runtime Health signals are healthy.", "detail": "Declared Runtime Health signals are healthy.", "severity": "ok", } if status == "degraded": return { "headline": "Runtime Health signals are degraded.", "detail": "One or more declared Runtime Health signals are degraded.", "severity": "warning", } if status == "unavailable": return { "headline": "Runtime Health signals are unavailable.", "detail": "One or more declared Runtime Health signals are unavailable.", "severity": "error", } return { "headline": "Runtime Health is not configured.", "detail": "Runtime Health has not been live-probed in this slice.", "severity": "neutral", } def _bounded_text(value: Any) -> tuple[str, bool]: text = str(value).strip().replace("\n", " ") if not text: return "redacted", True if _FORBIDDEN_TEXT.search(text): return "redacted", True return text[:160], False def _error_envelope(code: str, message: str) -> dict[str, Any]: return {"ok": False, "result": None, "error": {"code": code, "message": message}} def _write_json(handler: Any, status_code: int, payload: dict[str, Any]) -> None: body = json.dumps(payload, sort_keys=True).encode("utf-8") handler.send_response(status_code) handler.send_header("Content-Type", "application/json; charset=utf-8") handler.send_header("Content-Length", str(len(body))) handler.send_header("Cache-Control", "no-store") handler.end_headers() handler.wfile.write(body)