fix(reboot): overlay maintenance runtime readback

apps/api/src/services/reboot_auto_recovery_slo_scorecard.py

@@ -8,6 +8,7 @@ write StockPlatform data; it only reads the committed JSON scorecard.
 from __future__ import annotations
 
 import json
+import os
 from datetime import datetime
 from pathlib import Path
 from typing import Any
@@ -19,10 +20,25 @@ _DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
 _SCORECARD_FILE = "awoooi-reboot-auto-recovery-slo-scorecard.snapshot.json"
 _SOURCE_SCHEMA_VERSION = "awoooi_reboot_auto_recovery_slo_scorecard_v1"
 _API_SCHEMA_VERSION = "reboot_auto_recovery_slo_scorecard_readback_v1"
+_PUBLIC_MAINTENANCE_RUNTIME_FILE_ENV = (
+    "AWOOOI_PUBLIC_MAINTENANCE_FALLBACK_READBACK_FILE"
+)
+_PUBLIC_MAINTENANCE_RUNTIME_DIR_ENV = "AWOOOI_REBOOT_RECOVERY_LOG_DIR"
+_DEFAULT_REBOOT_RECOVERY_LOG_DIR = Path("/home/wooo/reboot-recovery")
+_PUBLIC_MAINTENANCE_RUNTIME_PATTERN = (
+    "reboot-auto-recovery-slo-*/public-maintenance-fallback.json"
+)
+_PUBLIC_MAINTENANCE_BLOCKERS = {
+    "public_maintenance_fallback_runtime_readback_missing",
+    "public_route_raw_5xx_without_maintenance_fallback",
+    "public_route_unreachable_without_external_l1_fallback",
+}
 
 
 def load_latest_reboot_auto_recovery_slo_scorecard(
     operations_dir: Path | None = None,
+    public_maintenance_runtime_path: Path | None = None,
+    public_maintenance_runtime_dir: Path | None = None,
 ) -> dict[str, Any]:
     """Load and validate the committed P0-006 reboot recovery scorecard."""
     directory = operations_dir or _DEFAULT_OPERATIONS_DIR
@@ -36,10 +52,134 @@ def load_latest_reboot_auto_recovery_slo_scorecard(
         raise ValueError(f"{path}: unsupported schema_version={scorecard.get('schema_version')!r}")
 
     payload = _build_payload(scorecard, path)
+    public_maintenance_runtime = _load_latest_public_maintenance_runtime_readback(
+        runtime_path=public_maintenance_runtime_path,
+        runtime_dir=public_maintenance_runtime_dir,
+    )
+    if public_maintenance_runtime:
+        apply_public_maintenance_runtime_readback(payload, public_maintenance_runtime)
     _require_operation_boundaries(payload, str(path))
     return payload
 
 
+def _load_latest_public_maintenance_runtime_readback(
+    *,
+    runtime_path: Path | None = None,
+    runtime_dir: Path | None = None,
+) -> dict[str, Any]:
+    explicit_path = runtime_path or _path_from_env(_PUBLIC_MAINTENANCE_RUNTIME_FILE_ENV)
+    if explicit_path:
+        return _read_json_file(explicit_path)
+
+    directory = (
+        runtime_dir
+        or _path_from_env(_PUBLIC_MAINTENANCE_RUNTIME_DIR_ENV)
+        or _DEFAULT_REBOOT_RECOVERY_LOG_DIR
+    )
+    try:
+        candidates = [
+            path
+            for path in directory.glob(_PUBLIC_MAINTENANCE_RUNTIME_PATTERN)
+            if path.is_file()
+        ]
+    except OSError:
+        return {}
+    if not candidates:
+        return {}
+    latest = max(candidates, key=lambda path: path.stat().st_mtime)
+    return _read_json_file(latest)
+
+
+def _path_from_env(name: str) -> Path | None:
+    value = os.environ.get(name, "").strip()
+    return Path(value) if value else None
+
+
+def _read_json_file(path: Path) -> dict[str, Any]:
+    try:
+        with path.open(encoding="utf-8") as handle:
+            payload = json.load(handle)
+    except (FileNotFoundError, json.JSONDecodeError, OSError):
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def apply_public_maintenance_runtime_readback(
+    payload: dict[str, Any],
+    runtime_readback: dict[str, Any],
+) -> None:
+    """Overlay exporter public-route fallback evidence onto the scorecard."""
+    public_maintenance = _build_public_maintenance_fallback_readback(runtime_readback)
+    payload["public_maintenance_fallback"] = public_maintenance
+
+    active_blockers = [
+        blocker
+        for blocker in _strings(payload.get("active_blockers"))
+        if blocker not in _PUBLIC_MAINTENANCE_BLOCKERS
+    ]
+    active_blockers.extend(_strings(public_maintenance.get("blockers")))
+    active_blockers = _unique_strings(active_blockers)
+    payload["active_blockers"] = active_blockers
+
+    required_checks = _dict(payload.setdefault("required_checks", {}))
+    required_checks["public_maintenance_fallback_runtime_ready"] = (
+        public_maintenance.get("ready") is True
+    )
+    completed_check_count = sum(1 for value in required_checks.values() if value)
+    readiness_percent = _percent(
+        completed_check_count / max(len(required_checks), 1) * 100
+    )
+    active_blocker_count = len(active_blockers)
+    payload["active_blocker_count"] = active_blocker_count
+    payload["readiness_percent"] = readiness_percent
+
+    progress = _dict(payload.setdefault("reboot_sop_progress", {}))
+    progress["active_blockers"] = active_blockers
+    progress["active_blocker_count"] = active_blocker_count
+    progress["readiness_percent"] = readiness_percent
+    progress["current_phase"] = _reboot_sop_current_phase(
+        active_blockers,
+        payload.get("can_claim_all_services_recovered_within_target") is True,
+    )
+    progress["primary_blocker"] = _reboot_sop_primary_blocker(active_blockers)
+    payload["reboot_sop_progress"] = progress
+    payload["current_phase"] = progress["current_phase"]
+    payload["primary_blocker"] = progress["primary_blocker"]
+
+    readback = _dict(payload.setdefault("readback", {}))
+    readback["active_blocker_count"] = active_blocker_count
+    readback["readiness_percent"] = readiness_percent
+    readback["public_maintenance_runtime_readback_present"] = (
+        public_maintenance.get("runtime_readback_present") is True
+    )
+    readback["public_maintenance_fallback_ready"] = (
+        public_maintenance.get("ready") is True
+    )
+    readback["public_route_raw_5xx_without_fallback_count"] = _int(
+        public_maintenance.get("raw_5xx_without_fallback_count")
+    )
+
+    rollups = _dict(payload.setdefault("rollups", {}))
+    rollups["active_blocker_count"] = active_blocker_count
+    rollups["completed_check_count"] = completed_check_count
+    rollups["readiness_percent"] = readiness_percent
+    rollups["public_maintenance_runtime_readback_present"] = (
+        public_maintenance.get("runtime_readback_present") is True
+    )
+    rollups["public_maintenance_fallback_ready"] = (
+        public_maintenance.get("ready") is True
+    )
+    rollups["public_route_raw_5xx_without_fallback_count"] = _int(
+        public_maintenance.get("raw_5xx_without_fallback_count")
+    )
+    rollups["public_route_unreachable_without_external_fallback_count"] = _int(
+        public_maintenance.get("route_unreachable_without_external_fallback_count")
+    )
+    rollups["public_maintenance_fallback_route_count"] = _int(
+        public_maintenance.get("maintenance_fallback_route_count")
+    )
+
+
 def apply_stockplatform_runtime_readback(
     payload: dict[str, Any],
     runtime_readback: dict[str, Any],
@@ -870,6 +1010,8 @@ def _reboot_sop_current_phase(active_blockers: list[str], can_claim_slo: bool) -
         for blocker in active_blockers
     ):
         return "product_data_freshness_blocked"
+    if any(blocker in _PUBLIC_MAINTENANCE_BLOCKERS for blocker in active_blockers):
+        return "public_maintenance_fallback_blocked"
     if any("backup" in blocker for blocker in active_blockers):
         return "backup_readback_blocked"
     if "local_disk_free_below_minimum" in active_blockers:
@@ -900,6 +1042,9 @@ def _reboot_sop_primary_blocker(active_blockers: list[str]) -> str:
         "windows99_vmware_autostart_config_not_ready",
         "windows99_vmware_guest_power_not_ready",
         "windows99_update_no_auto_reboot_policy_not_ready",
+        "public_route_raw_5xx_without_maintenance_fallback",
+        "public_route_unreachable_without_external_l1_fallback",
+        "public_maintenance_fallback_runtime_readback_missing",
         "post_start_blocked_not_zero",
         "service_green_not_1",
         "host_188_service_green_not_1",