feat(agent): validate harbor deploy marker receipt

apps/api/src/services/harbor_registry_controlled_recovery_receipt.py

@@ -32,6 +32,9 @@ def validate_harbor_registry_controlled_recovery_receipt(
     gitea_queue = _gitea_queue_readback(
         receipt_payload.get("gitea_actions_queue_readback")
     )
+    deploy_marker = _deploy_marker_readback(
+        receipt_payload.get("deploy_marker_readback")
+    )
 
     active_blockers = _active_blockers(
         ssh_local=ssh_local,
@@ -39,6 +42,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         watchdog_repair=watchdog_repair,
         verifier=verifier,
         gitea_queue=gitea_queue,
+        deploy_marker=deploy_marker,
     )
     status = _status(
         ssh_local=ssh_local,
@@ -46,6 +50,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         watchdog_repair=watchdog_repair,
         verifier=verifier,
         gitea_queue=gitea_queue,
+        deploy_marker=deploy_marker,
         active_blockers=active_blockers,
     )
     local_console_phase_readback = _local_console_phase_readback(
@@ -75,6 +80,11 @@ def validate_harbor_registry_controlled_recovery_receipt(
                 "raw_output_returned": False,
                 "top_visible_runs_returned": False,
             },
+            "deploy_marker_readback": {
+                "provided": deploy_marker["receipt_seen"],
+                "metadata_only": True,
+                "raw_output_returned": False,
+            },
         },
         "readback": {
             "ssh_local_repair": ssh_local,
@@ -82,6 +92,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "watchdog_repair": watchdog_repair,
             "post_apply_verifier": verifier,
             "gitea_actions_queue": gitea_queue,
+            "deploy_marker": deploy_marker,
         },
         "local_console_phase_readback": local_console_phase_readback,
         "controlled_apply_policy": {
@@ -135,6 +146,15 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "gitea_queue_harbor_110_repair_run_status": gitea_queue[
                 "latest_visible_harbor_110_repair_run_status"
             ],
+            "deploy_marker_readback_seen": deploy_marker["receipt_seen"],
+            "deploy_marker_verified": deploy_marker["deploy_marker_verified"],
+            "deploy_marker_blocker_count": deploy_marker["blocker_count"],
+            "deploy_marker_latest_cd_run_status": deploy_marker[
+                "latest_visible_cd_run_status"
+            ],
+            "deploy_marker_production_image_matches_expected": deploy_marker[
+                "production_image_matches_expected"
+            ],
             "metadata_writeback_contract_ready": True,
             "local_console_phase_count": local_console_phase_readback["phase_count"],
             "local_console_completed_phase_count": local_console_phase_readback[
@@ -466,6 +486,134 @@ def _gitea_queue_blockers(
     return blockers
 
 
+def _deploy_marker_readback(value: Any) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        return {
+            "receipt_seen": False,
+            "schema_version": "",
+            "status": "not_provided",
+            "expected_commit_sha": "",
+            "deploy_marker_commit_sha": "",
+            "production_image_tag_sha": "",
+            "latest_visible_cd_run_id": "",
+            "latest_visible_cd_run_status": "",
+            "production_preflight_http_status": None,
+            "production_workbench_http_status": None,
+            "production_image_matches_expected": False,
+            "production_readback_http_green": False,
+            "deploy_marker_verified": False,
+            "operation_boundary_violation": False,
+            "blockers": [],
+            "blocker_count": 0,
+            "metadata_only": True,
+            "raw_output_returned": False,
+        }
+
+    readback = _dict(value.get("readback"))
+    rollups = _dict(value.get("rollups"))
+    operation_boundaries = _dict(value.get("operation_boundaries"))
+    expected_sha = str(
+        value.get("expected_commit_sha")
+        or readback.get("expected_commit_sha")
+        or readback.get("observed_source_control_main_sha")
+        or readback.get("latest_visible_cd_run_commit_sha")
+        or ""
+    )
+    production_image_sha = str(
+        readback.get("production_image_tag_sha")
+        or readback.get("runtime_build_commit_sha")
+        or ""
+    )
+    production_image_matches = bool(
+        expected_sha
+        and production_image_sha
+        and production_image_sha == expected_sha
+    ) or bool(rollups.get("production_image_tag_matches_main") is True)
+    preflight_status = _int_or_none(readback.get("production_preflight_http_status"))
+    workbench_status = _int_or_none(readback.get("production_workbench_http_status"))
+    http_green = preflight_status == 200 and workbench_status == 200
+    latest_cd_status = str(readback.get("latest_visible_cd_run_status") or "")
+    cd_success = latest_cd_status in {"", "Success", "success"}
+    boundary_violation = any(
+        operation_boundaries.get(flag) is True
+        for flag in (
+            "deploy_trigger_allowed",
+            "workflow_modification_allowed",
+            "gitea_api_write_allowed",
+            "host_write_allowed",
+            "k8s_write_allowed",
+            "secret_read_allowed",
+            "github_api_allowed",
+            "raw_session_or_sqlite_read_allowed",
+        )
+    )
+    status = str(value.get("status") or "unknown")
+    verified = bool(
+        status == "closure_verified"
+        and production_image_matches
+        and http_green
+        and cd_success
+        and not boundary_violation
+    )
+    blockers = _deploy_marker_blockers(
+        verified=verified,
+        status=status,
+        production_image_matches=production_image_matches,
+        http_green=http_green,
+        cd_success=cd_success,
+        boundary_violation=boundary_violation,
+    )
+    return {
+        "receipt_seen": True,
+        "schema_version": str(value.get("schema_version") or ""),
+        "status": status,
+        "expected_commit_sha": expected_sha,
+        "deploy_marker_commit_sha": str(
+            readback.get("deploy_marker_commit_sha") or ""
+        ),
+        "production_image_tag_sha": production_image_sha,
+        "latest_visible_cd_run_id": str(
+            readback.get("latest_visible_cd_run_id") or ""
+        ),
+        "latest_visible_cd_run_status": latest_cd_status,
+        "production_preflight_http_status": preflight_status,
+        "production_workbench_http_status": workbench_status,
+        "production_image_matches_expected": production_image_matches,
+        "production_readback_http_green": http_green,
+        "deploy_marker_verified": verified,
+        "operation_boundary_violation": boundary_violation,
+        "blockers": blockers,
+        "blocker_count": len(blockers),
+        "metadata_only": True,
+        "raw_output_returned": False,
+    }
+
+
+def _deploy_marker_blockers(
+    *,
+    verified: bool,
+    status: str,
+    production_image_matches: bool,
+    http_green: bool,
+    cd_success: bool,
+    boundary_violation: bool,
+) -> list[str]:
+    if verified:
+        return []
+    blockers: list[str] = []
+    if status != "closure_verified":
+        blockers.append("deploy_marker_closure_not_verified")
+    if not production_image_matches:
+        blockers.append("deploy_marker_production_image_not_current_main")
+    if not http_green:
+        blockers.append("deploy_marker_production_readback_http_not_green")
+    if not cd_success:
+        blockers.append("deploy_marker_latest_cd_run_not_success")
+    if boundary_violation:
+        blockers.append("deploy_marker_readback_boundary_violation")
+    return blockers
+
+
 def _active_blockers(
     *,
     ssh_local: dict[str, Any],
@@ -473,6 +621,7 @@ def _active_blockers(
     watchdog_repair: dict[str, Any],
     verifier: dict[str, Any],
     gitea_queue: dict[str, Any],
+    deploy_marker: dict[str, Any],
 ) -> list[str]:
     blockers: list[str] = []
     if ssh_local["receipt_seen"] and not ssh_local["control_channel_metadata_ready"]:
@@ -492,6 +641,7 @@ def _active_blockers(
     if not verifier["internal_registry_v2_ready"]:
         blockers.append("internal_registry_v2_verifier_not_green")
     blockers.extend(_strings(gitea_queue.get("blockers")))
+    blockers.extend(_strings(deploy_marker.get("blockers")))
     return _unique_strings(blockers)
 
 
@@ -502,6 +652,7 @@ def _status(
     watchdog_repair: dict[str, Any],
     verifier: dict[str, Any],
     gitea_queue: dict[str, Any],
+    deploy_marker: dict[str, Any],
     active_blockers: list[str],
 ) -> str:
     if verifier["registry_v2_ready"] and not active_blockers:
@@ -512,6 +663,12 @@ def _status(
         and gitea_queue["blocker_count"] > 0
     ):
         return "harbor_registry_recovery_receipt_verified_waiting_gitea_queue_clearance"
+    if (
+        verifier["registry_v2_ready"]
+        and deploy_marker["receipt_seen"]
+        and deploy_marker["blocker_count"] > 0
+    ):
+        return "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback"
     if watchdog_repair["receipt_seen"]:
         return "harbor_registry_repair_receipt_waiting_registry_v2_verifier"
     if watchdog_check["receipt_seen"] and watchdog_check["harbor_ready"]:
@@ -528,6 +685,8 @@ def _safe_next_step(*, status: str) -> str:
         return "retry_gitea_cd_then_verify_deploy_marker_and_priority_readback"
     if status == "harbor_registry_recovery_receipt_verified_waiting_gitea_queue_clearance":
         return "clear_harbor_110_runner_queue_then_retry_cd_marker_readback"
+    if status == "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback":
+        return "rerun_gitea_cd_then_verify_deploy_marker_and_priority_readback"
     if status == "harbor_registry_repair_receipt_waiting_registry_v2_verifier":
         return "rerun_public_and_internal_registry_v2_verifier_before_cd_retry"
     if status == "harbor_local_registry_ready_waiting_public_registry_v2_verifier":
@@ -544,6 +703,8 @@ def _current_apply_blocker(*, status: str) -> str:
         return "receipt_verified_waiting_cd_marker_readback"
     if status == "harbor_registry_recovery_receipt_verified_waiting_gitea_queue_clearance":
         return "gitea_queue_clearance_required_after_registry_receipt"
+    if status == "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback":
+        return "deploy_marker_readback_required_after_registry_receipt"
     if status == "harbor_watchdog_check_unhealthy_waiting_repair_once_receipt":
         return "repair_once_receipt_required_after_unhealthy_check"
     if status == "ssh_local_repair_receipt_waiting_harbor_watchdog_check":