fix(agent): publish queue safe next action

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -368,6 +368,10 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
         "node_load_high" if node_load_classifier == "high_load" else ""
     )
     log_source_tags = _current_blocker_log_source_tags(recovery)
+    safe_next_action = _current_blocker_safe_next_action(
+        recovery,
+        control_path_pressure_blocker=control_path_pressure_blocker,
+    )
     return {
         "queue_item_id": f"current-p0-blocker::{blocker_id}",
         "source_sample_id": source_sample_id,
@@ -441,6 +445,15 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
             recovery.get("forbidden_runtime_actions")
         ),
         "safe_next_step": str(recovery.get("safe_next_step") or ""),
+        "safe_next_action_id": safe_next_action["action_id"],
+        "safe_next_action_stage": safe_next_action["stage"],
+        "safe_next_action": safe_next_action["action"],
+        "safe_next_action_command": safe_next_action["command"],
+        "safe_next_action_post_verifier": safe_next_action["post_verifier"],
+        "safe_next_action_requires_local_console": safe_next_action[
+            "requires_local_console"
+        ],
+        "safe_next_action_blocker_fields": safe_next_action["blocker_fields"],
         "runtime_write_gate": "controlled_after_110_local_console_preflight",
         "runtime_apply_required_on_110_local_console": bool(
             recovery.get("runtime_apply_required_on_110_local_console") is True
@@ -456,6 +469,90 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _current_blocker_safe_next_action(
+    recovery: dict[str, Any],
+    *,
+    control_path_pressure_blocker: str,
+) -> dict[str, Any]:
+    blocker_id = str(recovery.get("blocker_id") or "")
+    external_blocker = str(recovery.get("external_control_path_blocker") or "")
+    deploy_marker_required = recovery.get("deploy_marker_readback_required") is True
+    cd_failed_after_registry_ready = (
+        recovery.get("cd_failed_after_registry_ready") is True
+    )
+    harbor_repair_failed_after_registry_ready = (
+        recovery.get("harbor_110_repair_failed_after_registry_ready") is True
+    )
+
+    if blocker_id == "harbor_110_remote_ssh_publickey_auth_stalled":
+        return {
+            "action_id": (
+                "run_110_local_ssh_session_control_path_recovery_then_verify_cd_and_deploy_marker_readback"
+            ),
+            "stage": "local_console_control_path_receipt_required",
+            "action": (
+                "Run 110 local-console SSH/session check, then verify registry, "
+                "queue, CD, and deploy-marker closure."
+            ),
+            "command": "recover-110-control-path-and-harbor-local.sh --check",
+            "post_verifier": (
+                "read-public-gitea-actions-queue.py --json && "
+                "check-awoooi-110-controlled-cd-lane-readiness.sh"
+            ),
+            "requires_local_console": True,
+            "blocker_fields": [
+                "latest_visible_harbor_110_repair_remote_ssh_publickey_auth_stalled",
+                external_blocker,
+            ],
+        }
+
+    if control_path_pressure_blocker:
+        return {
+            "action_id": "wait_host_pressure_gate_then_rerun_110_control_path_readback",
+            "stage": "host_pressure_gate_wait",
+            "action": (
+                "Keep host pressure protection fail-closed, then rerun 110 "
+                "control-path and queue readbacks."
+            ),
+            "command": "awoooi-wait-host-web-build-pressure.sh",
+            "post_verifier": "read-public-gitea-actions-queue.py --json",
+            "requires_local_console": False,
+            "blocker_fields": [control_path_pressure_blocker],
+        }
+
+    if (
+        deploy_marker_required
+        or cd_failed_after_registry_ready
+        or harbor_repair_failed_after_registry_ready
+    ):
+        return {
+            "action_id": "verify_registry_ready_then_close_cd_deploy_marker_readback",
+            "stage": "deploy_marker_closure_after_registry_ready",
+            "action": (
+                "Keep registry 200/401 readiness separate from CD/deploy-marker "
+                "closure, then verify production priority readback."
+            ),
+            "command": "read-public-gitea-actions-queue.py --json",
+            "post_verifier": "awoooi production deploy marker readback",
+            "requires_local_console": False,
+            "blocker_fields": [
+                "deploy_marker_readback_required_after_registry_ready",
+                "current_cd_failure_after_registry_ready",
+                "harbor_110_repair_failure_after_registry_ready",
+            ],
+        }
+
+    return {
+        "action_id": "continue_ai_loop_current_blocker_readback",
+        "stage": "ai_loop_current_blocker_readback",
+        "action": "Continue metadata-only current blocker readback and writeback.",
+        "command": "read-public-gitea-actions-queue.py --json",
+        "post_verifier": "priority work-order readback",
+        "requires_local_console": False,
+        "blocker_fields": [external_blocker] if external_blocker else [],
+    }
+
+
 def _log_source_tagging_contract() -> list[dict[str, Any]]:
     metadata_boundary = {
         "metadata_only": True,
@@ -895,6 +992,20 @@ def _queue_readback_normalizer_contract() -> list[dict[str, Any]]:
             ],
             "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
         },
+        {
+            "field_id": "safe_next_action_id",
+            "purpose": (
+                "publish one machine-readable next action from queue classifiers "
+                "so AI Loop does not treat stale jobs API payloads, runner waits, "
+                "or 110 SSH/session blockers as generic manual triage"
+            ),
+            "writes_blockers": [
+                "gitea_queue_safe_next_action_missing",
+                "gitea_queue_safe_next_action_requires_local_console",
+                "gitea_queue_safe_next_action_stale_payload_quarantined",
+            ],
+            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+        },
         {
             "field_id": "controlled_profile_no_matching_runner_labels",
             "purpose": (

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -579,6 +579,19 @@ 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 "")
+    safe_next_action_id = str(first_item.get("safe_next_action_id") or "")
+    safe_next_action_stage = str(first_item.get("safe_next_action_stage") or "")
+    safe_next_action = str(first_item.get("safe_next_action") or "")
+    safe_next_action_command = str(first_item.get("safe_next_action_command") or "")
+    safe_next_action_post_verifier = str(
+        first_item.get("safe_next_action_post_verifier") or ""
+    )
+    safe_next_action_requires_local_console = bool(
+        first_item.get("safe_next_action_requires_local_console") is True
+    )
+    safe_next_action_blocker_fields = _strings(
+        first_item.get("safe_next_action_blocker_fields")
+    )
     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 ""
@@ -648,6 +661,16 @@ def apply_ai_loop_current_blocker_execution_queue(
     )
     state["ai_loop_current_blocker_node_load_classifier"] = node_load_classifier
     state["ai_loop_current_blocker_runtime_write_gate"] = runtime_write_gate
+    state["ai_loop_current_blocker_safe_next_action_id"] = safe_next_action_id
+    state["ai_loop_current_blocker_safe_next_action_stage"] = (
+        safe_next_action_stage
+    )
+    state["ai_loop_current_blocker_safe_next_action_requires_local_console"] = (
+        safe_next_action_requires_local_console
+    )
+    state["ai_loop_current_blocker_safe_next_action_blocker_fields"] = (
+        safe_next_action_blocker_fields
+    )
     state["ai_loop_current_blocker_learning_target_count"] = len(learning_targets)
     state["ai_loop_current_blocker_local_console_phase_count"] = len(
         local_console_plan
@@ -760,6 +783,25 @@ def apply_ai_loop_current_blocker_execution_queue(
             post_apply_verifier
         )
         evidence["ai_loop_current_blocker_runtime_write_gate"] = runtime_write_gate
+        evidence["ai_loop_current_blocker_safe_next_action_id"] = (
+            safe_next_action_id
+        )
+        evidence["ai_loop_current_blocker_safe_next_action_stage"] = (
+            safe_next_action_stage
+        )
+        evidence["ai_loop_current_blocker_safe_next_action"] = safe_next_action
+        evidence["ai_loop_current_blocker_safe_next_action_command"] = (
+            safe_next_action_command
+        )
+        evidence["ai_loop_current_blocker_safe_next_action_post_verifier"] = (
+            safe_next_action_post_verifier
+        )
+        evidence["ai_loop_current_blocker_safe_next_action_requires_local_console"] = (
+            safe_next_action_requires_local_console
+        )
+        evidence["ai_loop_current_blocker_safe_next_action_blocker_fields"] = (
+            safe_next_action_blocker_fields
+        )
         evidence["ai_loop_current_blocker_control_path_blocker"] = external_blocker
         evidence["ai_loop_current_blocker_control_path_pressure_blocker"] = (
             pressure_blocker
@@ -937,6 +979,16 @@ def apply_ai_loop_current_blocker_execution_queue(
         controlled_recovery_package
     )
     summary["ai_loop_current_blocker_post_apply_verifier"] = post_apply_verifier
+    summary["ai_loop_current_blocker_safe_next_action_id"] = safe_next_action_id
+    summary["ai_loop_current_blocker_safe_next_action_stage"] = (
+        safe_next_action_stage
+    )
+    summary["ai_loop_current_blocker_safe_next_action_requires_local_console"] = (
+        safe_next_action_requires_local_console
+    )
+    summary["ai_loop_current_blocker_safe_next_action_blocker_fields"] = (
+        safe_next_action_blocker_fields
+    )
     summary["ai_loop_current_blocker_learning_target_count"] = len(learning_targets)
     summary["ai_loop_current_blocker_local_console_phase_count"] = len(
         local_console_plan