feat(agent): accept controlled cd lane readiness receipt

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -430,6 +430,16 @@ def _harbor_recovery_receipt_inputs() -> list[dict[str, Any]]:
             "purpose": "prove bounded repair-once result without daemon restart",
             **metadata_boundary,
         },
+        {
+            "input_id": "controlled_cd_lane_readiness_output",
+            "source": "check-awoooi-110-controlled-cd-lane-readiness.sh",
+            "required_when": "before_retrying_harbor_110_local_repair_queue",
+            "purpose": (
+                "classify awoooi-host lane guardrail blockers such as service, "
+                "registration, root restore-source, pressure, or label/config drift"
+            ),
+            **metadata_boundary,
+        },
         {
             "input_id": "public_registry_v2_http_status",
             "source": "public registry /v2/ verifier",

apps/api/src/services/harbor_registry_controlled_recovery_receipt.py

@@ -27,11 +27,17 @@ def validate_harbor_registry_controlled_recovery_receipt(
     ssh_local_output = _text(receipt_payload.get("ssh_local_repair_output"))
     watchdog_check_output = _text(receipt_payload.get("watchdog_check_output"))
     watchdog_repair_output = _text(receipt_payload.get("watchdog_repair_output"))
+    controlled_cd_lane_output = _text(
+        receipt_payload.get("controlled_cd_lane_readiness_output")
+    )
 
     ssh_diagnosis = _parse_ssh_publickey_diagnosis_output(ssh_diagnosis_output)
     ssh_local = _parse_ssh_local_repair_output(ssh_local_output)
     watchdog_check = _parse_watchdog_output(watchdog_check_output)
     watchdog_repair = _parse_watchdog_output(watchdog_repair_output)
+    controlled_cd_lane = _parse_controlled_cd_lane_readiness_output(
+        controlled_cd_lane_output
+    )
     verifier = _post_apply_verifier(receipt_payload)
     gitea_queue = _gitea_queue_readback(
         receipt_payload.get("gitea_actions_queue_readback")
@@ -45,6 +51,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         ssh_local=ssh_local,
         watchdog_check=watchdog_check,
         watchdog_repair=watchdog_repair,
+        controlled_cd_lane=controlled_cd_lane,
         verifier=verifier,
         gitea_queue=gitea_queue,
         deploy_marker=deploy_marker,
@@ -54,6 +61,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         ssh_local=ssh_local,
         watchdog_check=watchdog_check,
         watchdog_repair=watchdog_repair,
+        controlled_cd_lane=controlled_cd_lane,
         verifier=verifier,
         gitea_queue=gitea_queue,
         deploy_marker=deploy_marker,
@@ -64,6 +72,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
         ssh_local=ssh_local,
         watchdog_check=watchdog_check,
         watchdog_repair=watchdog_repair,
+        controlled_cd_lane=controlled_cd_lane,
         verifier=verifier,
     )
 
@@ -84,6 +93,9 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "ssh_local_repair_output": _text_stats(ssh_local_output),
             "watchdog_check_output": _text_stats(watchdog_check_output),
             "watchdog_repair_output": _text_stats(watchdog_repair_output),
+            "controlled_cd_lane_readiness_output": _text_stats(
+                controlled_cd_lane_output
+            ),
             "gitea_actions_queue_readback": {
                 "provided": gitea_queue["receipt_seen"],
                 "metadata_only": True,
@@ -101,6 +113,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "ssh_local_repair": ssh_local,
             "watchdog_check": watchdog_check,
             "watchdog_repair": watchdog_repair,
+            "controlled_cd_lane_readiness": controlled_cd_lane,
             "post_apply_verifier": verifier,
             "gitea_actions_queue": gitea_queue,
             "deploy_marker": deploy_marker,
@@ -117,6 +130,7 @@ def validate_harbor_registry_controlled_recovery_receipt(
                 "validate_non_secret_ssh_local_repair_receipt",
                 "validate_harbor_watchdog_check_receipt",
                 "validate_harbor_watchdog_repair_once_receipt",
+                "validate_110_controlled_cd_lane_readiness_receipt",
                 "validate_public_and_internal_registry_v2_verifier",
                 "km_rag_mcp_playbook_metadata_writeback",
                 "retry_gitea_cd_after_registry_v2_green",
@@ -170,6 +184,18 @@ def validate_harbor_registry_controlled_recovery_receipt(
             "watchdog_check_harbor_ready": watchdog_check["harbor_ready"],
             "watchdog_repair_receipt_seen": watchdog_repair["receipt_seen"],
             "watchdog_repair_harbor_ready": watchdog_repair["harbor_ready"],
+            "controlled_cd_lane_readiness_receipt_seen": controlled_cd_lane[
+                "receipt_seen"
+            ],
+            "controlled_cd_lane_ready": controlled_cd_lane[
+                "controlled_cd_lane_ready"
+            ],
+            "controlled_cd_lane_blocker_count": controlled_cd_lane[
+                "blocker_count"
+            ],
+            "controlled_cd_lane_safe_next_step": controlled_cd_lane[
+                "safe_next_step"
+            ],
             "post_apply_verifier_ready": verifier["registry_v2_ready"],
             "gitea_queue_readback_seen": gitea_queue["receipt_seen"],
             "gitea_queue_blocker_count": gitea_queue["blocker_count"],
@@ -260,6 +286,7 @@ def _local_console_phase_readback(
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
     watchdog_repair: dict[str, Any],
+    controlled_cd_lane: dict[str, Any],
     verifier: dict[str, Any],
 ) -> dict[str, Any]:
     phases = [
@@ -298,11 +325,11 @@ def _local_console_phase_readback(
         ),
         _phase(
             "verify_controlled_cd_lane",
-            _phase_status(
-                ready=verifier["registry_v2_ready"],
-                blocked_status="blocked_waiting_registry_v2_verifier_green",
+            _controlled_cd_lane_phase_status(
+                controlled_cd_lane=controlled_cd_lane,
+                verifier=verifier,
             ),
-            "post_apply_verifier",
+            "controlled_cd_lane_readiness",
         ),
     ]
     completed_statuses = {"ready", "skipped_not_required"}
@@ -381,6 +408,21 @@ def _harbor_repair_once_phase_status(
     return "blocked_waiting_harbor_watchdog_check_receipt"
 
 
+def _controlled_cd_lane_phase_status(
+    *,
+    controlled_cd_lane: dict[str, Any],
+    verifier: dict[str, Any],
+) -> str:
+    if controlled_cd_lane["receipt_seen"]:
+        if controlled_cd_lane["controlled_cd_lane_ready"]:
+            return "ready"
+        return "blocked_controlled_cd_lane_readiness_receipt_not_ready"
+    return _phase_status(
+        ready=verifier["registry_v2_ready"],
+        blocked_status="blocked_waiting_controlled_cd_lane_readiness_receipt",
+    )
+
+
 def _parse_ssh_publickey_diagnosis_output(output: str) -> dict[str, Any]:
     fields = _parse_key_values(output)
     marker_seen = "AWOOOI_110_SSH_PUBLICKEY_AUTH_DIAGNOSIS" in output
@@ -548,6 +590,40 @@ def _parse_watchdog_output(output: str) -> dict[str, Any]:
     }
 
 
+def _parse_controlled_cd_lane_readiness_output(output: str) -> dict[str, Any]:
+    fields = _parse_key_values(output)
+    marker_seen = "AWOOOI_110_CONTROLLED_CD_LANE_READY=" in output
+    blockers = _prefixed_blockers(
+        output,
+        prefix="controlled_cd_lane_readiness:",
+    )
+    warning_count = _int_or_none(fields.get("WARNING_COUNT")) or 0
+    blocker_count = _int_or_none(fields.get("BLOCKER_COUNT"))
+    if blocker_count is None:
+        blocker_count = len(blockers)
+    ready = _bool_from_field(fields.get("AWOOOI_110_CONTROLLED_CD_LANE_READY"))
+    return {
+        "receipt_seen": marker_seen,
+        "controlled_cd_lane_ready": bool(
+            marker_seen and ready and blocker_count == 0 and not blockers
+        ),
+        "config_ready": _bool_from_field(fields.get("CONFIG_READY")),
+        "binary_ready": _bool_from_field(fields.get("BINARY_READY")),
+        "registration_ready": _bool_from_field(fields.get("REGISTRATION_READY")),
+        "service_ready": _bool_from_field(fields.get("SERVICE_READY")),
+        "legacy_failclosed": _bool_from_field(fields.get("LEGACY_FAILCLOSED")),
+        "primary_lane_failclosed": _bool_from_field(
+            fields.get("PRIMARY_LANE_FAILCLOSED")
+        ),
+        "warning_count": warning_count,
+        "blocker_count": blocker_count,
+        "blockers": blockers,
+        "safe_next_step": str(fields.get("safe_next_step") or ""),
+        "metadata_only": True,
+        "raw_output_returned": False,
+    }
+
+
 def _post_apply_verifier(receipt_payload: dict[str, Any]) -> dict[str, Any]:
     public_status = _int_or_none(receipt_payload.get("public_registry_v2_http_status"))
     internal_status = _int_or_none(
@@ -977,6 +1053,7 @@ def _active_blockers(
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
     watchdog_repair: dict[str, Any],
+    controlled_cd_lane: dict[str, Any],
     verifier: dict[str, Any],
     gitea_queue: dict[str, Any],
     deploy_marker: dict[str, Any],
@@ -1010,6 +1087,7 @@ def _active_blockers(
         blockers.append("harbor_watchdog_repair_forbidden_action_seen")
     if watchdog_repair["receipt_seen"] and not watchdog_repair["harbor_ready"]:
         blockers.append("harbor_watchdog_repair_did_not_restore_local_v2")
+    blockers.extend(_strings(controlled_cd_lane.get("blockers")))
     if not verifier["public_registry_v2_ready"]:
         blockers.append("public_registry_v2_verifier_not_green")
     if not verifier["internal_registry_v2_ready"]:
@@ -1025,6 +1103,7 @@ def _status(
     ssh_local: dict[str, Any],
     watchdog_check: dict[str, Any],
     watchdog_repair: dict[str, Any],
+    controlled_cd_lane: dict[str, Any],
     verifier: dict[str, Any],
     gitea_queue: dict[str, Any],
     deploy_marker: dict[str, Any],
@@ -1044,6 +1123,10 @@ def _status(
         and deploy_marker["blocker_count"] > 0
     ):
         return "harbor_registry_recovery_receipt_verified_waiting_deploy_marker_readback"
+    if controlled_cd_lane["receipt_seen"] and not controlled_cd_lane[
+        "controlled_cd_lane_ready"
+    ]:
+        return "controlled_cd_lane_readiness_receipt_blocked"
     if watchdog_repair["receipt_seen"]:
         return "harbor_registry_repair_receipt_waiting_registry_v2_verifier"
     if watchdog_check["receipt_seen"] and watchdog_check["harbor_ready"]:
@@ -1064,6 +1147,8 @@ def _safe_next_step(*, status: str) -> str:
         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 == "controlled_cd_lane_readiness_receipt_blocked":
+        return "fix_controlled_cd_lane_guardrail_blockers_then_rerun_readiness_verifier"
     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":
@@ -1087,6 +1172,8 @@ def _current_apply_blocker(*, status: str) -> str:
         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 == "controlled_cd_lane_readiness_receipt_blocked":
+        return "controlled_cd_lane_readiness_required_for_awoooi_host_queue"
     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":
@@ -1132,6 +1219,18 @@ def _parse_key_values(output: str) -> dict[str, str]:
     return fields
 
 
+def _prefixed_blockers(output: str, *, prefix: str) -> list[str]:
+    blockers: list[str] = []
+    for raw_line in output.splitlines():
+        line = raw_line.strip()
+        if not line.startswith("BLOCKER "):
+            continue
+        blocker = line.split(" ", 1)[1].strip()
+        if blocker:
+            blockers.append(f"{prefix}{blocker}")
+    return _unique_strings(blockers)
+
+
 def _mode_from_marker_line(output: str) -> str:
     for raw_line in output.splitlines():
         if "AWOOOI_110_SSH_PUBLICKEY_AUTH_LOCAL_REPAIR" not in raw_line: