fix(api): record stockplatform recovery receipt

apps/api/src/services/stockplatform_public_api_runtime_readback.py

@@ -9,14 +9,20 @@ from __future__ import annotations
 import json
 import urllib.error
 import urllib.request
+from collections.abc import Callable
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any
 
 from src.services.reboot_auto_recovery_slo_scorecard import (
     load_latest_reboot_auto_recovery_slo_scorecard,
 )
+from src.services.snapshot_paths import default_operations_dir
 
 _API_SCHEMA_VERSION = "stockplatform_public_api_runtime_readback_v1"
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_RECOVERY_RECEIPT_FILE = (
+    "stockplatform-public-api-runtime-recovery-control-receipt.snapshot.json"
+)
 _DEFAULT_BASE_URL = "https://stock.wooo.work"
 _DEFAULT_TIMEOUT_SECONDS = 4.0
 
@@ -33,9 +39,11 @@ def load_latest_stockplatform_public_api_runtime_readback(
     """Build a live public readback for StockPlatform web/API health."""
     http_probe = probe or _http_probe
     normalized_base_url = base_url.rstrip("/")
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
     committed_scorecard = load_latest_reboot_auto_recovery_slo_scorecard(
-        operations_dir
+        directory
     )
+    recovery_control_receipt = _load_recovery_control_receipt(directory)
     committed_stock = _dict(committed_scorecard.get("stockplatform_data_freshness"))
     endpoints = {
         "public_web_healthz": "/healthz",
@@ -57,6 +65,7 @@ def load_latest_stockplatform_public_api_runtime_readback(
         timeout_seconds=timeout_seconds,
         probes=probes,
         committed_stockplatform=committed_stock,
+        recovery_control_receipt=recovery_control_receipt,
     )
 
 
@@ -66,6 +75,7 @@ def _build_payload(
     timeout_seconds: float,
     probes: dict[str, dict[str, Any]],
     committed_stockplatform: dict[str, Any],
+    recovery_control_receipt: dict[str, Any],
 ) -> dict[str, Any]:
     web = _dict(probes.get("public_web_healthz"))
     api = _dict(probes.get("public_api_healthz"))
@@ -90,7 +100,10 @@ def _build_payload(
         ingestion_json=ingestion_json,
     )
     ready = all(checks.values())
-    recovery_control_path = _recovery_control_path_readback(runtime_ready=ready)
+    recovery_control_path = _recovery_control_path_readback(
+        runtime_ready=ready,
+        receipt=recovery_control_receipt,
+    )
     recovery_control_path_blockers = _strings(
         recovery_control_path.get("active_blockers")
     )
@@ -195,15 +208,42 @@ def _build_payload(
     }
 
 
-def _recovery_control_path_readback(*, runtime_ready: bool) -> dict[str, Any]:
+def _recovery_control_path_readback(
+    *,
+    runtime_ready: bool,
+    receipt: dict[str, Any],
+) -> dict[str, Any]:
     if runtime_ready:
         return {
             "status": "not_required_stockplatform_runtime_ready",
             "active_blockers": [],
+            "receipt": receipt,
             "safe_recovery_channels": [
                 "keep_monitoring_public_api_runtime_readback",
             ],
         }
+    if receipt:
+        active_blockers = _unique_strings(
+            _strings(receipt.get("active_blockers"))
+            or _strings(receipt.get("missing_receipts"))
+        )
+        return {
+            "status": str(
+                receipt.get("status")
+                or "blocked_recovery_control_path_receipt_incomplete"
+            ),
+            "active_blockers": active_blockers,
+            "receipt": receipt,
+            "provided_receipts": _strings(receipt.get("provided_receipts")),
+            "missing_receipts": _strings(receipt.get("missing_receipts")),
+            "safe_recovery_channels": _strings(
+                receipt.get("safe_recovery_channels")
+            ),
+            "forbidden_recovery_channels": _strings(
+                receipt.get("forbidden_recovery_channels")
+            ),
+            "required_receipts": _strings(receipt.get("required_receipts")),
+        }
     return {
         "status": "blocked_recovery_control_path_receipt_missing",
         "active_blockers": [
@@ -233,6 +273,23 @@ def _recovery_control_path_readback(*, runtime_ready: bool) -> dict[str, Any]:
     }
 
 
+def _load_recovery_control_receipt(operations_dir: Path) -> dict[str, Any]:
+    path = operations_dir / _RECOVERY_RECEIPT_FILE
+    if not path.exists():
+        return {}
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    if payload.get("schema_version") != (
+        "stockplatform_public_api_runtime_recovery_control_receipt_v1"
+    ):
+        raise ValueError(
+            f"{path}: unsupported schema_version={payload.get('schema_version')!r}"
+        )
+    return payload
+
+
 def _active_blockers(
     *,
     api: dict[str, Any],
@@ -314,7 +371,14 @@ def _http_probe(url: str, timeout_seconds: float) -> dict[str, Any]:
         body = exc.read(2048).decode("utf-8", "replace")
         return {"http_status": exc.code, "body": body, "error": ""}
     except Exception as exc:  # noqa: BLE001 - readback must fail closed.
-        return {"http_status": None, "body": "", "error": type(exc).__name__}
+        return {"http_status": None, "body": "", "error": _error_text(exc)}
+
+
+def _error_text(exc: Exception) -> str:
+    reason = getattr(exc, "reason", "")
+    if reason:
+        return f"{type(exc).__name__}: {reason}"
+    return type(exc).__name__
 
 
 def _dict(value: Any) -> dict[str, Any]: