fix(agent): separate registry-ready deploy marker blocker

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -134,6 +134,19 @@ def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[st
                 item["harbor_recovery_receipt_output_contract_count"]
                 for item in current_blocker_queue
             ),
+            "current_blocker_registry_v2_ready_count": sum(
+                1 for item in current_blocker_queue if item["registry_v2_ready"]
+            ),
+            "current_blocker_deploy_marker_readback_required_count": sum(
+                1
+                for item in current_blocker_queue
+                if item["deploy_marker_readback_required"]
+            ),
+            "current_blocker_cd_failure_after_registry_ready_count": sum(
+                1
+                for item in current_blocker_queue
+                if item["cd_failed_after_registry_ready"]
+            ),
             "runtime_dispatch_performed": False,
         },
         "active_blockers": active_blockers,
@@ -366,6 +379,34 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
         "node_load_classifier": node_load_classifier,
         "node_load_high": node_load_classifier == "high_load",
         "registry_v2_status": recovery.get("registry_v2_status"),
+        "registry_v2_ready": bool(recovery.get("registry_v2_ready") is True),
+        "registry_v2_status_classifier": str(
+            recovery.get("registry_v2_status_classifier") or ""
+        ),
+        "deployment_closure_state": str(
+            recovery.get("deployment_closure_state") or ""
+        ),
+        "deploy_marker_readback_required": bool(
+            recovery.get("deploy_marker_readback_required") is True
+        ),
+        "current_cd_run_id": str(recovery.get("current_cd_run_id") or ""),
+        "current_cd_run_status": str(recovery.get("current_cd_run_status") or ""),
+        "current_cd_commit_sha": str(recovery.get("current_cd_commit_sha") or ""),
+        "cd_failed_after_registry_ready": bool(
+            recovery.get("cd_failed_after_registry_ready") is True
+        ),
+        "harbor_110_repair_run_id": str(
+            recovery.get("harbor_110_repair_run_id") or ""
+        ),
+        "harbor_110_repair_run_status": str(
+            recovery.get("harbor_110_repair_run_status") or ""
+        ),
+        "harbor_110_repair_failed_after_registry_ready": bool(
+            recovery.get("harbor_110_repair_failed_after_registry_ready") is True
+        ),
+        "harbor_110_repair_failure_classifier": str(
+            recovery.get("harbor_110_repair_failure_classifier") or ""
+        ),
         "log_source_tags": log_source_tags,
         "log_source_tag_count": len(log_source_tags),
         "log_grouping_keys": list(_LOG_SOURCE_GROUPING_KEYS),
@@ -796,6 +837,42 @@ def _queue_readback_normalizer_contract() -> list[dict[str, Any]]:
             ],
             "learning_targets": ["km", "rag", "playbook", "mcp", "verifier"],
         },
+        {
+            "field_id": "current_cd_deploy_marker_closure",
+            "purpose": (
+                "classify latest visible CD status, commit sha, and deploy-marker "
+                "closure after registry /v2/ has returned a ready 200/401 state"
+            ),
+            "writes_blockers": [
+                "deploy_marker_readback_required_after_registry_ready",
+                "current_cd_failure_after_registry_ready",
+            ],
+            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+        },
+        {
+            "field_id": "registry_v2_ready_state",
+            "purpose": (
+                "distinguish registry /v2/ ready 200/401 evidence from stale "
+                "Harbor 502 repair blockers"
+            ),
+            "writes_blockers": [
+                "public_registry_v2_verifier_not_green",
+                "internal_registry_v2_verifier_not_green",
+            ],
+            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+        },
+        {
+            "field_id": "harbor_110_repair_failure_after_registry_ready",
+            "purpose": (
+                "classify Harbor 110 repair failure as SSH/control-path closure "
+                "work when registry /v2/ is already ready"
+            ),
+            "writes_blockers": [
+                "harbor_110_remote_ssh_publickey_auth_stalled",
+                "harbor_110_repair_failure_after_registry_ready",
+            ],
+            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+        },
         {
             "field_id": "harbor_110_repair_visible_running_jobs_api_stale",
             "purpose": (

apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py

@@ -204,6 +204,18 @@ def _current_blocker_recovery(receipt: dict[str, Any]) -> dict[str, Any] | None:
     if not blocker_id:
         return None
     source_sample_id = str(receipt.get("source_sample_id") or "")
+    registry_v2_status = classification.get("registry_v2_status")
+    registry_v2_ready = _registry_v2_ready(registry_v2_status)
+    current_cd_status = str(classification.get("current_cd_run_status") or "")
+    harbor_repair_status = str(
+        classification.get("harbor_110_repair_run_status") or ""
+    )
+    cd_failed_after_registry_ready = bool(
+        registry_v2_ready and current_cd_status in {"Failure", "Blocked", "Canceled"}
+    )
+    harbor_repair_failed_after_registry_ready = bool(
+        registry_v2_ready and harbor_repair_status in {"Failure", "Blocked"}
+    )
     recovery = {
         "source_sample_id": source_sample_id,
         "project_id": str(receipt.get("project_id") or ""),
@@ -220,7 +232,41 @@ def _current_blocker_recovery(receipt: dict[str, Any]) -> dict[str, Any] | None:
             classification.get("ssh_auth_classification") or ""
         ),
         "node_load_classifier": str(classification.get("node_load_classifier") or ""),
-        "registry_v2_status": classification.get("registry_v2_status"),
+        "registry_v2_status": registry_v2_status,
+        "registry_v2_ready": registry_v2_ready,
+        "registry_v2_status_classifier": _registry_v2_status_classifier(
+            registry_v2_status
+        ),
+        "deployment_closure_state": _deployment_closure_state(
+            registry_v2_ready=registry_v2_ready,
+            current_cd_status=current_cd_status,
+            harbor_repair_status=harbor_repair_status,
+            fallback=str(classification.get("deployment_closure_state") or ""),
+        ),
+        "deploy_marker_readback_required": bool(
+            registry_v2_ready
+            and (
+                cd_failed_after_registry_ready
+                or harbor_repair_failed_after_registry_ready
+                or classification.get("deploy_marker_readback_required") is True
+            )
+        ),
+        "current_cd_run_id": str(classification.get("current_cd_run_id") or ""),
+        "current_cd_run_status": current_cd_status,
+        "current_cd_commit_sha": str(
+            classification.get("current_cd_commit_sha") or ""
+        ),
+        "cd_failed_after_registry_ready": cd_failed_after_registry_ready,
+        "harbor_110_repair_run_id": str(
+            classification.get("harbor_110_repair_run_id") or ""
+        ),
+        "harbor_110_repair_run_status": harbor_repair_status,
+        "harbor_110_repair_failed_after_registry_ready": (
+            harbor_repair_failed_after_registry_ready
+        ),
+        "harbor_110_repair_failure_classifier": str(
+            classification.get("harbor_110_repair_failure_classifier") or ""
+        ),
         "controlled_recovery_package": str(
             classification.get("controlled_recovery_package") or ""
         ),
@@ -249,6 +295,43 @@ def _current_blocker_recovery(receipt: dict[str, Any]) -> dict[str, Any] | None:
     return recovery
 
 
+def _registry_v2_ready(value: Any) -> bool:
+    try:
+        return int(str(value)) in {200, 401}
+    except (TypeError, ValueError):
+        return False
+
+
+def _registry_v2_status_classifier(value: Any) -> str:
+    try:
+        status = int(str(value))
+    except (TypeError, ValueError):
+        return "unknown"
+    if status == 200:
+        return "ready_http_200"
+    if status == 401:
+        return "ready_auth_required_http_401"
+    if status:
+        return f"blocked_http_{status}"
+    return "unknown"
+
+
+def _deployment_closure_state(
+    *,
+    registry_v2_ready: bool,
+    current_cd_status: str,
+    harbor_repair_status: str,
+    fallback: str,
+) -> str:
+    if registry_v2_ready and current_cd_status in {"Failure", "Blocked", "Canceled"}:
+        return "blocked_latest_visible_cd_failure_after_registry_ready"
+    if registry_v2_ready and harbor_repair_status in {"Failure", "Blocked"}:
+        return "blocked_harbor_110_repair_failure_after_registry_ready"
+    if registry_v2_ready:
+        return fallback or "registry_ready_waiting_deploy_marker_readback"
+    return fallback or "registry_v2_not_ready"
+
+
 def _list_of_dicts(value: Any) -> list[dict[str, Any]]:
     if not isinstance(value, list):
         return []

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -513,6 +513,7 @@ def apply_ai_loop_current_blocker_execution_queue(
     if not _ai_loop_current_blocker_can_override(
         status=str(payload.get("status") or ""),
         blocker_id=blocker_id,
+        queue_item=first_item,
     ):
         return
 
@@ -578,6 +579,30 @@ def apply_ai_loop_current_blocker_execution_queue(
     pressure_blocker = str(first_item.get("control_path_pressure_blocker") or "")
     node_load_classifier = str(first_item.get("node_load_classifier") or "")
     runtime_write_gate = str(first_item.get("runtime_write_gate") or "")
+    registry_v2_ready = bool(first_item.get("registry_v2_ready") is True)
+    registry_v2_status_classifier = str(
+        first_item.get("registry_v2_status_classifier") or ""
+    )
+    deployment_closure_state = str(first_item.get("deployment_closure_state") or "")
+    deploy_marker_readback_required = bool(
+        first_item.get("deploy_marker_readback_required") is True
+    )
+    current_cd_run_id = str(first_item.get("current_cd_run_id") or "")
+    current_cd_run_status = str(first_item.get("current_cd_run_status") or "")
+    current_cd_commit_sha = str(first_item.get("current_cd_commit_sha") or "")
+    cd_failed_after_registry_ready = bool(
+        first_item.get("cd_failed_after_registry_ready") is True
+    )
+    harbor_110_repair_run_id = str(first_item.get("harbor_110_repair_run_id") or "")
+    harbor_110_repair_run_status = str(
+        first_item.get("harbor_110_repair_run_status") or ""
+    )
+    harbor_110_repair_failure_classifier = str(
+        first_item.get("harbor_110_repair_failure_classifier") or ""
+    )
+    harbor_110_repair_failed_after_registry_ready = bool(
+        first_item.get("harbor_110_repair_failed_after_registry_ready") is True
+    )
 
     state = _dict(payload.setdefault("mainline_execution_state", {}))
     state["active_p0_state"] = "blocked_ai_loop_current_blocker_execution_queue"
@@ -589,6 +614,34 @@ def apply_ai_loop_current_blocker_execution_queue(
     )
     state["ai_loop_current_blocker_execution_queue_count"] = len(queue)
     state["ai_loop_current_blocker_id"] = blocker_id
+    state["ai_loop_current_blocker_registry_v2_ready"] = registry_v2_ready
+    state["ai_loop_current_blocker_registry_v2_status_classifier"] = (
+        registry_v2_status_classifier
+    )
+    state["ai_loop_current_blocker_deployment_closure_state"] = (
+        deployment_closure_state
+    )
+    state["ai_loop_current_blocker_deploy_marker_readback_required"] = (
+        deploy_marker_readback_required
+    )
+    state["ai_loop_current_blocker_current_cd_run_id"] = current_cd_run_id
+    state["ai_loop_current_blocker_current_cd_run_status"] = current_cd_run_status
+    state["ai_loop_current_blocker_current_cd_commit_sha"] = current_cd_commit_sha
+    state["ai_loop_current_blocker_cd_failed_after_registry_ready"] = (
+        cd_failed_after_registry_ready
+    )
+    state["ai_loop_current_blocker_harbor_110_repair_run_id"] = (
+        harbor_110_repair_run_id
+    )
+    state["ai_loop_current_blocker_harbor_110_repair_run_status"] = (
+        harbor_110_repair_run_status
+    )
+    state["ai_loop_current_blocker_harbor_110_repair_failure_classifier"] = (
+        harbor_110_repair_failure_classifier
+    )
+    state[
+        "ai_loop_current_blocker_harbor_110_repair_failed_after_registry_ready"
+    ] = harbor_110_repair_failed_after_registry_ready
     state["ai_loop_current_blocker_control_path_blocker"] = external_blocker
     state["ai_loop_current_blocker_control_path_pressure_blocker"] = (
         pressure_blocker
@@ -639,6 +692,16 @@ def apply_ai_loop_current_blocker_execution_queue(
         + ([blocker_id] if blocker_id else [])
         + ([external_blocker] if external_blocker else [])
         + ([pressure_blocker] if pressure_blocker else [])
+        + (
+            ["deploy_marker_readback_required_after_registry_ready"]
+            if deploy_marker_readback_required
+            else []
+        )
+        + (
+            ["current_cd_failure_after_registry_ready"]
+            if cd_failed_after_registry_ready
+            else []
+        )
     )
 
     for item in _list(payload.get("in_progress_or_blocked_in_priority_order")):
@@ -658,6 +721,38 @@ def apply_ai_loop_current_blocker_execution_queue(
         evidence["ai_loop_current_blocker_registry_v2_status"] = first_item.get(
             "registry_v2_status"
         )
+        evidence["ai_loop_current_blocker_registry_v2_ready"] = registry_v2_ready
+        evidence["ai_loop_current_blocker_registry_v2_status_classifier"] = (
+            registry_v2_status_classifier
+        )
+        evidence["ai_loop_current_blocker_deployment_closure_state"] = (
+            deployment_closure_state
+        )
+        evidence["ai_loop_current_blocker_deploy_marker_readback_required"] = (
+            deploy_marker_readback_required
+        )
+        evidence["ai_loop_current_blocker_current_cd_run_id"] = current_cd_run_id
+        evidence["ai_loop_current_blocker_current_cd_run_status"] = (
+            current_cd_run_status
+        )
+        evidence["ai_loop_current_blocker_current_cd_commit_sha"] = (
+            current_cd_commit_sha
+        )
+        evidence["ai_loop_current_blocker_cd_failed_after_registry_ready"] = (
+            cd_failed_after_registry_ready
+        )
+        evidence["ai_loop_current_blocker_harbor_110_repair_run_id"] = (
+            harbor_110_repair_run_id
+        )
+        evidence["ai_loop_current_blocker_harbor_110_repair_run_status"] = (
+            harbor_110_repair_run_status
+        )
+        evidence["ai_loop_current_blocker_harbor_110_repair_failure_classifier"] = (
+            harbor_110_repair_failure_classifier
+        )
+        evidence[
+            "ai_loop_current_blocker_harbor_110_repair_failed_after_registry_ready"
+        ] = harbor_110_repair_failed_after_registry_ready
         evidence["ai_loop_current_blocker_controlled_recovery_package"] = (
             controlled_recovery_package
         )
@@ -762,31 +857,77 @@ def apply_ai_loop_current_blocker_execution_queue(
         )
 
     payload["status"] = "p0_006_blocked_ai_loop_current_blocker_execution_queue"
-    payload["next_execution_order"] = [
-        (
-            "P0-006-AI-LOOP-CURRENT-BLOCKER-EXECUTION-QUEUE: execute the "
-            f"{blocker_id} queue item through 110 control-path readback, "
-            f"{len(local_console_plan)} ordered local-console phases, post-recovery "
-            "readback commands, Harbor receipt inputs and outputs, and "
-            "metadata-only KM/RAG/MCP/PlayBook writeback from normalized queue "
-            "classifiers and control-path readiness classifiers."
-        ),
-        (
-            "P0-006-HARBOR-REGISTRY-CONTROLLED-RECOVERY-PREFLIGHT: after the "
-            "AI Loop queue item verifies 110 control path, rerun Harbor "
-            "watchdog check-mode / repair-once and registry /v2/ readback."
-        ),
-        (
-            "P0-006-CD-DEPLOY-MARKER-READBACK: after registry /v2/ is 200/401, "
-            "let the next Gitea CD run build/push/deploy and then verify "
-            "delivery-closure-workbench, priority work order, and production "
-            "marker advance."
-        ),
-    ]
+    if registry_v2_ready:
+        payload["next_execution_order"] = [
+            (
+                "P0-006-AI-LOOP-CURRENT-BLOCKER-EXECUTION-QUEUE: registry /v2/ "
+                f"is already {first_item.get('registry_v2_status')} "
+                f"({registry_v2_status_classifier}); do not reopen Harbor 502 "
+                f"as the main blocker. Execute {blocker_id} through 110 "
+                "SSH/session control-path readback, ordered local-console phases, "
+                "post-recovery queue readbacks, and metadata-only KM/RAG/MCP/"
+                "PlayBook writeback."
+            ),
+            (
+                "P0-006-CD-DEPLOY-MARKER-READBACK: close the latest visible CD "
+                f"{current_cd_run_id or 'unknown'} status "
+                f"{current_cd_run_status or 'unknown'} and verify deploy marker / "
+                "production image / priority API before claiming runtime closure."
+            ),
+            (
+                "P0-006-HARBOR-REGISTRY-CONTROLLED-RECOVERY-PREFLIGHT: only "
+                "rerun Harbor watchdog repair if registry /v2/ regresses below "
+                "200/401; otherwise keep focus on 110 SSH control-path and "
+                "deploy-marker closure."
+            ),
+        ]
+    else:
+        payload["next_execution_order"] = [
+            (
+                "P0-006-AI-LOOP-CURRENT-BLOCKER-EXECUTION-QUEUE: execute the "
+                f"{blocker_id} queue item through 110 control-path readback, "
+                f"{len(local_console_plan)} ordered local-console phases, "
+                "post-recovery readback commands, Harbor receipt inputs and "
+                "outputs, and metadata-only KM/RAG/MCP/PlayBook writeback from "
+                "normalized queue classifiers and control-path readiness "
+                "classifiers."
+            ),
+            (
+                "P0-006-HARBOR-REGISTRY-CONTROLLED-RECOVERY-PREFLIGHT: after the "
+                "AI Loop queue item verifies 110 control path, rerun Harbor "
+                "watchdog check-mode / repair-once and registry /v2/ readback."
+            ),
+            (
+                "P0-006-CD-DEPLOY-MARKER-READBACK: after registry /v2/ is "
+                "200/401, let the next Gitea CD run build/push/deploy and then "
+                "verify delivery-closure-workbench, priority work order, and "
+                "production marker advance."
+            ),
+        ]
     _refresh_rollups_after_stockplatform_overlay(payload, state)
     summary = _dict(payload.setdefault("summary", {}))
     summary["ai_loop_current_blocker_execution_queue_count"] = len(queue)
     summary["ai_loop_current_blocker_id"] = blocker_id
+    summary["ai_loop_current_blocker_registry_v2_ready"] = registry_v2_ready
+    summary["ai_loop_current_blocker_registry_v2_status_classifier"] = (
+        registry_v2_status_classifier
+    )
+    summary["ai_loop_current_blocker_deployment_closure_state"] = (
+        deployment_closure_state
+    )
+    summary["ai_loop_current_blocker_deploy_marker_readback_required"] = (
+        deploy_marker_readback_required
+    )
+    summary["ai_loop_current_blocker_current_cd_run_status"] = current_cd_run_status
+    summary["ai_loop_current_blocker_cd_failed_after_registry_ready"] = (
+        cd_failed_after_registry_ready
+    )
+    summary["ai_loop_current_blocker_harbor_110_repair_run_status"] = (
+        harbor_110_repair_run_status
+    )
+    summary["ai_loop_current_blocker_harbor_110_repair_failure_classifier"] = (
+        harbor_110_repair_failure_classifier
+    )
     summary["ai_loop_current_blocker_control_path_blocker"] = external_blocker
     summary["ai_loop_current_blocker_control_path_pressure_blocker"] = (
         pressure_blocker
@@ -836,12 +977,23 @@ def apply_ai_loop_current_blocker_execution_queue(
     )
 
 
-def _ai_loop_current_blocker_can_override(*, status: str, blocker_id: str) -> bool:
+def _ai_loop_current_blocker_can_override(
+    *,
+    status: str,
+    blocker_id: str,
+    queue_item: dict[str, Any],
+) -> bool:
     if status == "p0_006_blocked_harbor_registry_controlled_recovery_preflight":
         return True
     if status == "p0_006_blocked_stockplatform_public_api_runtime_drift":
         return blocker_id.startswith("harbor_110_")
-    return False
+    if not blocker_id.startswith("harbor_110_"):
+        return False
+    return bool(
+        queue_item.get("deploy_marker_readback_required") is True
+        or queue_item.get("cd_failed_after_registry_ready") is True
+        or queue_item.get("harbor_110_repair_failed_after_registry_ready") is True
+    )
 
 
 def _harbor_registry_next_state(*, status: str, candidate_packaged: bool) -> str: