fix(reboot): overlay runtime slo scorecard artifact

apps/api/src/services/reboot_auto_recovery_slo_scorecard.py

@@ -24,10 +24,13 @@ _PUBLIC_MAINTENANCE_RUNTIME_FILE_ENV = (
     "AWOOOI_PUBLIC_MAINTENANCE_FALLBACK_READBACK_FILE"
 )
 _PUBLIC_MAINTENANCE_RUNTIME_DIR_ENV = "AWOOOI_REBOOT_RECOVERY_LOG_DIR"
+_RUNTIME_SCORECARD_FILE_ENV = "AWOOOI_REBOOT_RECOVERY_SLO_SCORECARD_FILE"
+_RUNTIME_SCORECARD_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"
 )
+_RUNTIME_SCORECARD_PATTERN = "reboot-auto-recovery-slo-*/scorecard.json"
 _PUBLIC_MAINTENANCE_BLOCKERS = {
     "public_maintenance_fallback_runtime_readback_missing",
     "public_route_raw_5xx_without_maintenance_fallback",
@@ -39,6 +42,8 @@ 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,
+    runtime_scorecard_path: Path | None = None,
+    runtime_scorecard_dir: Path | None = None,
 ) -> dict[str, Any]:
     """Load and validate the committed P0-006 reboot recovery scorecard."""
     directory = operations_dir or _DEFAULT_OPERATIONS_DIR
@@ -51,7 +56,21 @@ def load_latest_reboot_auto_recovery_slo_scorecard(
     if scorecard.get("schema_version") != _SOURCE_SCHEMA_VERSION:
         raise ValueError(f"{path}: unsupported schema_version={scorecard.get('schema_version')!r}")
 
-    payload = _build_payload(scorecard, path)
+    source_scorecard = scorecard
+    runtime_scorecard, runtime_scorecard_source = _load_latest_runtime_scorecard(
+        runtime_path=runtime_scorecard_path,
+        runtime_dir=runtime_scorecard_dir,
+    )
+    if runtime_scorecard:
+        source_scorecard = runtime_scorecard
+
+    payload = _build_payload(source_scorecard, path)
+    _annotate_runtime_scorecard_readback(
+        payload=payload,
+        committed_scorecard=scorecard,
+        runtime_scorecard=runtime_scorecard,
+        runtime_scorecard_source=runtime_scorecard_source,
+    )
     public_maintenance_runtime = _load_latest_public_maintenance_runtime_readback(
         runtime_path=public_maintenance_runtime_path,
         runtime_dir=public_maintenance_runtime_dir,
@@ -62,6 +81,70 @@ def load_latest_reboot_auto_recovery_slo_scorecard(
     return payload
 
 
+def _load_latest_runtime_scorecard(
+    *,
+    runtime_path: Path | None = None,
+    runtime_dir: Path | None = None,
+) -> tuple[dict[str, Any], Path | None]:
+    explicit_path = runtime_path or _path_from_env(_RUNTIME_SCORECARD_FILE_ENV)
+    if explicit_path:
+        payload = _read_json_file(explicit_path)
+        return _runtime_scorecard_if_supported(payload), explicit_path
+
+    directory = (
+        runtime_dir
+        or _path_from_env(_RUNTIME_SCORECARD_DIR_ENV)
+        or _DEFAULT_REBOOT_RECOVERY_LOG_DIR
+    )
+    try:
+        candidates = [
+            path
+            for path in directory.glob(_RUNTIME_SCORECARD_PATTERN)
+            if path.is_file()
+        ]
+    except OSError:
+        return {}, None
+    if not candidates:
+        return {}, None
+    latest = max(candidates, key=lambda path: path.stat().st_mtime)
+    payload = _read_json_file(latest)
+    return _runtime_scorecard_if_supported(payload), latest
+
+
+def _runtime_scorecard_if_supported(payload: dict[str, Any]) -> dict[str, Any]:
+    if payload.get("schema_version") != _SOURCE_SCHEMA_VERSION:
+        return {}
+    return payload
+
+
+def _annotate_runtime_scorecard_readback(
+    *,
+    payload: dict[str, Any],
+    committed_scorecard: dict[str, Any],
+    runtime_scorecard: dict[str, Any],
+    runtime_scorecard_source: Path | None,
+) -> None:
+    runtime_present = bool(runtime_scorecard)
+    committed_generated_at = str(committed_scorecard.get("generated_at") or "")
+    runtime_generated_at = str(runtime_scorecard.get("generated_at") or "")
+    runtime_source_ref = str(runtime_scorecard_source or "")
+
+    payload["snapshot_generated_at"] = committed_generated_at
+    payload["runtime_scorecard_readback_present"] = runtime_present
+    payload["runtime_scorecard_generated_at"] = runtime_generated_at
+    payload["runtime_scorecard_source_ref"] = runtime_source_ref
+
+    readback = _dict(payload.setdefault("readback", {}))
+    readback["snapshot_generated_at"] = committed_generated_at
+    readback["runtime_scorecard_readback_present"] = runtime_present
+    readback["runtime_scorecard_generated_at"] = runtime_generated_at
+    readback["runtime_scorecard_source_ref"] = runtime_source_ref
+
+    rollups = _dict(payload.setdefault("rollups", {}))
+    rollups["runtime_scorecard_readback_present"] = runtime_present
+    rollups["runtime_scorecard_generated_at"] = runtime_generated_at
+
+
 def _load_latest_public_maintenance_runtime_readback(
     *,
     runtime_path: Path | None = None,