fix(reboot): expose runtime scorecard freshness

apps/api/src/services/delivery_closure_workbench.py

@@ -7,7 +7,9 @@ need to fan out across five separate endpoints or duplicate blocker math.
 
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from src.services.awoooi_gitea_onboarding_warning_step_runtime_enablement_gate import (
     load_latest_awoooi_gitea_onboarding_warning_step_runtime_enablement_gate,
@@ -1322,14 +1324,28 @@ def build_delivery_closure_workbench(
         _source_status("runtime_surface", runtime),
         _source_status("backup_dr", backup),
     ]
-    generated_candidates = [
+    source_generated_candidates = [
         source["generated_at"] for source in source_statuses if source["generated_at"]
     ]
+    snapshot_generated_candidates = [
+        source["snapshot_generated_at"]
+        for source in source_statuses
+        if source["snapshot_generated_at"]
+    ]
     loaded_source_count = sum(1 for source in source_statuses if source["loaded"])
     high_risk_blocker_count = sum(_int(lane["blocker_count"]) for lane in lanes)
     average_completion = _percent(
         sum(_int(lane["completion_percent"]) for lane in lanes) / max(len(lanes), 1)
     )
+    source_generated_at = (
+        max(source_generated_candidates) if source_generated_candidates else ""
+    )
+    snapshot_generated_at = (
+        max(snapshot_generated_candidates)
+        if snapshot_generated_candidates
+        else source_generated_at
+    )
+    runtime_readback_generated_at = _taipei_now_iso()
     next_focus = [
         {
             "lane_id": lane["id"],
@@ -1350,12 +1366,19 @@ def build_delivery_closure_workbench(
 
     return {
         "schema_version": _SCHEMA_VERSION,
-        "generated_at": max(generated_candidates) if generated_candidates else "",
+        "generated_at": runtime_readback_generated_at,
+        "runtime_readback_generated_at": runtime_readback_generated_at,
+        "source_generated_at": source_generated_at,
+        "snapshot_generated_at": snapshot_generated_at,
         "status": delivery_status,
         "safe_next_step": current_p0_safe_next_step,
         "active_blockers": current_p0_active_blockers,
         "readback": {
             "workbench_status": delivery_status,
+            "generated_at": runtime_readback_generated_at,
+            "runtime_readback_generated_at": runtime_readback_generated_at,
+            "source_generated_at": source_generated_at,
+            "snapshot_generated_at": snapshot_generated_at,
             "current_p0_workplan_id": current_p0_workplan_id,
             "current_p0_source_id": "reboot_auto_recovery_slo_scorecard",
             "current_p0_status": str(reboot_slo.get("status") or ""),
@@ -1442,8 +1465,14 @@ def build_delivery_closure_workbench(
             "secret_values_collected": False,
             "workflow_trigger_authorized": False,
             "runtime_write_authorized": False,
+            "runtime_readback_generated_at_present": True,
+            "source_generated_at_present": bool(source_generated_at),
         },
         "summary": {
+            "generated_at": runtime_readback_generated_at,
+            "runtime_readback_generated_at": runtime_readback_generated_at,
+            "source_generated_at": source_generated_at,
+            "snapshot_generated_at": snapshot_generated_at,
             "source_count": len(source_statuses),
             "loaded_source_count": loaded_source_count,
             "average_completion_percent": average_completion,
@@ -2266,6 +2295,10 @@ def _source_status(source_id: str, payload: dict[str, Any]) -> dict[str, Any]:
         "loaded": not source_missing,
         "schema_version": str(payload.get("schema_version") or ""),
         "generated_at": str(payload.get("generated_at") or ""),
+        "runtime_readback_generated_at": str(
+            payload.get("runtime_readback_generated_at") or ""
+        ),
+        "snapshot_generated_at": str(payload.get("snapshot_generated_at") or ""),
         "missing_reason": str(payload.get("missing_reason") or "")
         if source_missing
         else "",
@@ -2302,6 +2335,10 @@ def _strings(value: Any) -> list[str]:
     return [str(item) for item in value if item is not None]
 
 
+def _taipei_now_iso() -> str:
+    return datetime.now(ZoneInfo("Asia/Taipei")).isoformat(timespec="seconds")
+
+
 def _first_string(value: Any) -> str:
     if isinstance(value, list) and value:
         return str(value[0])

apps/api/src/services/reboot_auto_recovery_slo_scorecard.py

@@ -8,8 +8,10 @@ write StockPlatform data; it only reads the committed JSON scorecard.
 from __future__ import annotations
 
 import json
+from datetime import datetime
 from pathlib import Path
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from src.services.snapshot_paths import default_operations_dir
 
@@ -80,6 +82,8 @@ def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
     )
     if can_claim_slo and not active_blockers:
         readiness_percent = 100
+    snapshot_generated_at = str(scorecard.get("generated_at") or "")
+    runtime_readback_generated_at = _taipei_now_iso()
     latest_verify_metric = _dict(scorecard.get("latest_verify_only_metric"))
     active_blocker_count = len(active_blockers)
     observed_host_count = len(_strings(host_boot_detection.get("observed_hosts")))
@@ -228,10 +232,13 @@ def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
         "windows99_rdp_console_reachable": (
             windows99_management.get("rdp_console_reachable") is True
         ),
+        "runtime_readback_generated_at_present": True,
     }
     return {
         "schema_version": _API_SCHEMA_VERSION,
-        "generated_at": str(scorecard.get("generated_at") or ""),
+        "generated_at": runtime_readback_generated_at,
+        "runtime_readback_generated_at": runtime_readback_generated_at,
+        "snapshot_generated_at": snapshot_generated_at,
         "priority": "P0-006",
         "scope": "reboot_auto_recovery_slo_scorecard",
         "status": str(scorecard.get("status") or "unknown"),
@@ -277,6 +284,9 @@ def _build_payload(scorecard: dict[str, Any], path: Path) -> dict[str, Any]:
             "workplan_id": "P0-006",
             "workplan_title": "主機重啟自動偵測、自動觸發與 10 分鐘恢復 SLO",
             "source_scorecard_ref": f"docs/operations/{path.name}",
+            "generated_at": runtime_readback_generated_at,
+            "runtime_readback_generated_at": runtime_readback_generated_at,
+            "snapshot_generated_at": snapshot_generated_at,
             "target_minutes": _int(scorecard.get("target_minutes")),
             "current_phase": sop_progress["current_phase"],
             "eta_or_wait_reason": sop_progress["eta_or_wait_reason"],
@@ -734,3 +744,7 @@ def _strings(value: Any) -> list[str]:
     if not isinstance(value, list):
         return []
     return [str(item) for item in value if item is not None]
+
+
+def _taipei_now_iso() -> str:
+    return datetime.now(ZoneInfo("Asia/Taipei")).isoformat(timespec="seconds")

apps/api/tests/test_delivery_closure_workbench_api.py

@@ -316,9 +316,18 @@ def test_delivery_closure_workbench_exposes_p0_006_reboot_slo_lane():
 
 def _assert_delivery_workbench_shape(data: dict):
     assert data["schema_version"] == "delivery_closure_workbench_v1"
+    assert data["snapshot_generated_at"]
+    assert data["source_generated_at"]
+    assert data["generated_at"] != data["snapshot_generated_at"]
+    assert data["generated_at"].endswith("+08:00")
+    assert data["runtime_readback_generated_at"] == data["generated_at"]
     assert data["safe_next_step"] == REBOOT_SAFE_NEXT_STEP
     assert data["active_blockers"] == REBOOT_ACTIVE_BLOCKERS
     assert data["readback"]["workbench_status"] == "blocked_delivery_actions_required"
+    assert data["readback"]["generated_at"] == data["generated_at"]
+    assert data["readback"]["runtime_readback_generated_at"] == data["generated_at"]
+    assert data["readback"]["source_generated_at"] == data["source_generated_at"]
+    assert data["readback"]["snapshot_generated_at"] == data["snapshot_generated_at"]
     assert data["readback"]["current_p0_workplan_id"] == "P0-006"
     assert data["readback"]["current_p0_source_id"] == (
         "reboot_auto_recovery_slo_scorecard"
@@ -384,7 +393,18 @@ def _assert_delivery_workbench_shape(data: dict):
     assert data["rollups"]["secret_values_collected"] is False
     assert data["rollups"]["workflow_trigger_authorized"] is False
     assert data["rollups"]["runtime_write_authorized"] is False
+    assert data["rollups"]["runtime_readback_generated_at_present"] is True
+    assert data["rollups"]["source_generated_at_present"] is True
+    source_statuses = {source["id"]: source for source in data["source_statuses"]}
+    reboot_source = source_statuses["reboot_auto_recovery_slo_scorecard"]
+    assert reboot_source["snapshot_generated_at"] == "2026-07-02T15:08:44+08:00"
+    assert reboot_source["runtime_readback_generated_at"] == reboot_source["generated_at"]
+    assert reboot_source["runtime_readback_generated_at"].endswith("+08:00")
     assert data["summary"]["source_count"] == 9
+    assert data["summary"]["generated_at"] == data["generated_at"]
+    assert data["summary"]["runtime_readback_generated_at"] == data["generated_at"]
+    assert data["summary"]["source_generated_at"] == data["source_generated_at"]
+    assert data["summary"]["snapshot_generated_at"] == data["snapshot_generated_at"]
     assert data["summary"]["loaded_source_count"] == 9
     assert data["summary"]["runtime_execution_authorized"] is False
     assert data["summary"]["remote_write_authorized"] is False

apps/api/tests/test_reboot_auto_recovery_slo_scorecard_api.py

@@ -75,6 +75,10 @@ def test_reboot_auto_recovery_drill_preflight_endpoint_returns_readback():
 
 def _assert_reboot_slo_payload(payload: dict):
     assert payload["schema_version"] == "reboot_auto_recovery_slo_scorecard_readback_v1"
+    assert payload["snapshot_generated_at"] == "2026-07-02T15:08:44+08:00"
+    assert payload["generated_at"] != payload["snapshot_generated_at"]
+    assert payload["generated_at"].endswith("+08:00")
+    assert payload["runtime_readback_generated_at"] == payload["generated_at"]
     assert payload["priority"] == "P0-006"
     assert payload["status"] == "blocked_reboot_auto_recovery_slo_not_ready"
     assert payload["current_phase"] == "host_boot_detection_blocked"
@@ -115,6 +119,13 @@ def _assert_reboot_slo_payload(payload: dict):
         "windows99_management_channel_probe_source_present"
     ] is True
     assert payload["readback"]["workplan_id"] == "P0-006"
+    assert payload["readback"]["generated_at"] == payload["generated_at"]
+    assert payload["readback"]["runtime_readback_generated_at"] == payload[
+        "generated_at"
+    ]
+    assert payload["readback"]["snapshot_generated_at"] == payload[
+        "snapshot_generated_at"
+    ]
     assert payload["readback"]["target_minutes"] == 10
     assert payload["readback"]["current_phase"] == "host_boot_detection_blocked"
     assert payload["readback"]["eta_or_wait_reason"] == (
@@ -188,6 +199,7 @@ def _assert_reboot_slo_payload(payload: dict):
     assert payload["rollups"]["windows99_host_reachable"] is True
     assert payload["rollups"]["windows99_remote_execution_channel_ready"] is False
     assert payload["rollups"]["windows99_ssh_batch_status"] == "permission_denied"
+    assert payload["rollups"]["runtime_readback_generated_at_present"] is True
     assert payload["rollups"]["windows99_rdp_console_reachable"] is True
     assert payload["rollups"]["stockplatform_final_retry_window_passed"] is False
     assert (