feat(agent): add harbor recovery phase receipts

apps/api/src/services/harbor_registry_controlled_recovery_receipt.py

@@ -43,6 +43,12 @@ def validate_harbor_registry_controlled_recovery_receipt(
         verifier=verifier,
         active_blockers=active_blockers,
     )
+    local_console_phase_readback = _local_console_phase_readback(
+        ssh_local=ssh_local,
+        watchdog_check=watchdog_check,
+        watchdog_repair=watchdog_repair,
+        verifier=verifier,
+    )
 
     return {
         "schema_version": _SCHEMA_VERSION,
@@ -65,6 +71,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "watchdog_repair": watchdog_repair,
             "post_apply_verifier": verifier,
         },
+        "local_console_phase_readback": local_console_phase_readback,
         "controlled_apply_policy": {
             "risk_level": "high",
             "break_glass_required": False,
@@ -103,6 +110,13 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "watchdog_repair_harbor_ready": watchdog_repair["harbor_ready"],
             "post_apply_verifier_ready": verifier["registry_v2_ready"],
             "metadata_writeback_contract_ready": True,
+            "local_console_phase_count": local_console_phase_readback["phase_count"],
+            "local_console_completed_phase_count": local_console_phase_readback[
+                "completed_phase_count"
+            ],
+            "local_console_blocked_phase_count": local_console_phase_readback[
+                "blocked_phase_count"
+            ],
         },
         "operation_boundaries": {
             "raw_output_returned": False,
@@ -122,6 +136,111 @@ def validate_harbor_registry_controlled_recovery_receipt(
     }
 
 
+def _local_console_phase_readback(
+    *,
+    ssh_local: dict[str, Any],
+    watchdog_check: dict[str, Any],
+    watchdog_repair: dict[str, Any],
+    verifier: dict[str, Any],
+) -> dict[str, Any]:
+    phases = [
+        _phase(
+            "diagnose_ssh_publickey",
+            _phase_status(
+                ready=ssh_local["receipt_seen"],
+                blocked_status="blocked_waiting_ssh_publickey_diagnosis_receipt",
+            ),
+            "ssh_local_repair",
+        ),
+        _phase(
+            "preflight_control_path_and_harbor",
+            _phase_status(
+                ready=watchdog_check["receipt_seen"],
+                blocked_status="blocked_waiting_harbor_watchdog_check_receipt",
+            ),
+            "watchdog_check",
+        ),
+        _phase(
+            "repair_ssh_metadata_if_check_confirms_metadata_drift",
+            _ssh_metadata_phase_status(ssh_local=ssh_local, watchdog_check=watchdog_check),
+            "ssh_local_repair",
+        ),
+        _phase(
+            "repair_harbor_once_if_v2_still_502",
+            _harbor_repair_once_phase_status(
+                watchdog_check=watchdog_check,
+                watchdog_repair=watchdog_repair,
+            ),
+            "watchdog_repair",
+        ),
+        _phase(
+            "verify_controlled_cd_lane",
+            _phase_status(
+                ready=verifier["registry_v2_ready"],
+                blocked_status="blocked_waiting_registry_v2_verifier_green",
+            ),
+            "post_apply_verifier",
+        ),
+    ]
+    completed_statuses = {"ready", "skipped_not_required"}
+    return {
+        "phase_count": len(phases),
+        "completed_phase_count": sum(
+            1 for phase in phases if phase["status"] in completed_statuses
+        ),
+        "blocked_phase_count": sum(
+            1 for phase in phases if phase["status"] not in completed_statuses
+        ),
+        "phase_ids": [phase["phase_id"] for phase in phases],
+        "phases": phases,
+        "metadata_only": True,
+        "raw_output_returned": False,
+    }
+
+
+def _phase(phase_id: str, status: str, evidence_key: str) -> dict[str, Any]:
+    return {
+        "phase_id": phase_id,
+        "status": status,
+        "evidence_key": evidence_key,
+        "raw_output_returned": False,
+    }
+
+
+def _phase_status(*, ready: bool, blocked_status: str) -> str:
+    return "ready" if ready else blocked_status
+
+
+def _ssh_metadata_phase_status(
+    *,
+    ssh_local: dict[str, Any],
+    watchdog_check: dict[str, Any],
+) -> str:
+    if ssh_local["receipt_seen"]:
+        if ssh_local["control_channel_metadata_ready"]:
+            return "ready"
+        return "blocked_ssh_metadata_repair_receipt_not_ready"
+    if watchdog_check["receipt_seen"]:
+        return "skipped_not_required"
+    return "blocked_waiting_ssh_metadata_or_harbor_preflight_receipt"
+
+
+def _harbor_repair_once_phase_status(
+    *,
+    watchdog_check: dict[str, Any],
+    watchdog_repair: dict[str, Any],
+) -> str:
+    if watchdog_check["receipt_seen"] and watchdog_check["harbor_ready"]:
+        return "skipped_not_required"
+    if watchdog_repair["receipt_seen"] and watchdog_repair["harbor_ready"]:
+        return "ready"
+    if watchdog_repair["receipt_seen"]:
+        return "blocked_harbor_repair_once_receipt_not_green"
+    if watchdog_check["receipt_seen"]:
+        return "blocked_waiting_harbor_repair_once_receipt"
+    return "blocked_waiting_harbor_watchdog_check_receipt"
+
+
 def _parse_ssh_local_repair_output(output: str) -> dict[str, Any]:
     fields = _parse_key_values(output)
     marker_seen = "AWOOOI_110_SSH_PUBLICKEY_AUTH_LOCAL_REPAIR" in output