fix(recovery): propagate 110 high load blocker

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -103,6 +103,11 @@ def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[st
                 for item in current_blocker_queue
                 if item["external_control_path_blocker"]
             ),
+            "current_blocker_control_path_pressure_blocked_count": sum(
+                1
+                for item in current_blocker_queue
+                if item["control_path_pressure_blocker"]
+            ),
             "current_blocker_local_recovery_package_count": sum(
                 1 for item in current_blocker_queue if item["controlled_recovery_package"]
             ),
@@ -322,6 +327,10 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
     blocker_id = str(recovery.get("blocker_id") or "")
     source_sample_id = str(recovery.get("source_sample_id") or "")
     external_blocker = str(recovery.get("external_control_path_blocker") or "")
+    node_load_classifier = str(recovery.get("node_load_classifier") or "")
+    control_path_pressure_blocker = (
+        "node_load_high" if node_load_classifier == "high_load" else ""
+    )
     return {
         "queue_item_id": f"current-p0-blocker::{blocker_id}",
         "source_sample_id": source_sample_id,
@@ -330,6 +339,8 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
         "status": "controlled_recovery_packaged_waiting_control_path_readback",
         "risk_tier": str(recovery.get("risk_tier") or "unknown"),
         "runner_label": str(recovery.get("runner_label") or ""),
+        "node_load_classifier": node_load_classifier,
+        "node_load_high": node_load_classifier == "high_load",
         "registry_v2_status": recovery.get("registry_v2_status"),
         "controlled_recovery_package": str(
             recovery.get("controlled_recovery_package") or ""
@@ -357,6 +368,7 @@ def _current_blocker_queue_item(recovery: dict[str, Any]) -> dict[str, Any]:
             recovery.get("runtime_apply_required_on_110_local_console") is True
         ),
         "external_control_path_blocker": external_blocker,
+        "control_path_pressure_blocker": control_path_pressure_blocker,
         "control_channel_readback_required": bool(external_blocker),
         "metadata_only_writeback_ready": True,
         "runtime_dispatch_performed": False,

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -544,6 +544,8 @@ def apply_ai_loop_current_blocker_execution_queue(
     ]
     forbidden_runtime_actions = _strings(first_item.get("forbidden_runtime_actions"))
     external_blocker = str(first_item.get("external_control_path_blocker") or "")
+    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 "")
 
     state = _dict(payload.setdefault("mainline_execution_state", {}))
@@ -557,6 +559,10 @@ 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_control_path_blocker"] = external_blocker
+    state["ai_loop_current_blocker_control_path_pressure_blocker"] = (
+        pressure_blocker
+    )
+    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_learning_target_count"] = len(learning_targets)
     state["ai_loop_current_blocker_local_console_phase_count"] = len(
@@ -582,6 +588,7 @@ def apply_ai_loop_current_blocker_execution_queue(
         + ["ai_loop_current_blocker_execution_queue"]
         + ([blocker_id] if blocker_id else [])
         + ([external_blocker] if external_blocker else [])
+        + ([pressure_blocker] if pressure_blocker else [])
     )
 
     for item in _list(payload.get("in_progress_or_blocked_in_priority_order")):
@@ -609,6 +616,12 @@ def apply_ai_loop_current_blocker_execution_queue(
         )
         evidence["ai_loop_current_blocker_runtime_write_gate"] = runtime_write_gate
         evidence["ai_loop_current_blocker_control_path_blocker"] = external_blocker
+        evidence["ai_loop_current_blocker_control_path_pressure_blocker"] = (
+            pressure_blocker
+        )
+        evidence["ai_loop_current_blocker_node_load_classifier"] = (
+            node_load_classifier
+        )
         evidence["ai_loop_current_blocker_learning_writeback_targets"] = (
             learning_targets
         )
@@ -698,6 +711,10 @@ def apply_ai_loop_current_blocker_execution_queue(
     summary["ai_loop_current_blocker_execution_queue_count"] = len(queue)
     summary["ai_loop_current_blocker_id"] = blocker_id
     summary["ai_loop_current_blocker_control_path_blocker"] = external_blocker
+    summary["ai_loop_current_blocker_control_path_pressure_blocker"] = (
+        pressure_blocker
+    )
+    summary["ai_loop_current_blocker_node_load_classifier"] = node_load_classifier
     summary["ai_loop_current_blocker_controlled_recovery_package"] = (
         controlled_recovery_package
     )

apps/api/src/services/harbor_registry_controlled_recovery_receipt.py

@@ -267,9 +267,9 @@ def _local_console_phase_readback(
         ),
         _phase(
             "preflight_control_path_and_harbor",
-            _phase_status(
-                ready=watchdog_check["receipt_seen"],
-                blocked_status="blocked_waiting_harbor_watchdog_check_receipt",
+            _preflight_control_path_phase_status(
+                ssh_diagnosis=ssh_diagnosis,
+                watchdog_check=watchdog_check,
             ),
             "watchdog_check",
         ),
@@ -328,6 +328,18 @@ def _phase_status(*, ready: bool, blocked_status: str) -> str:
     return "ready" if ready else blocked_status
 
 
+def _preflight_control_path_phase_status(
+    *,
+    ssh_diagnosis: dict[str, Any],
+    watchdog_check: dict[str, Any],
+) -> str:
+    if watchdog_check["receipt_seen"]:
+        return "ready"
+    if ssh_diagnosis["receipt_seen"] and ssh_diagnosis["node_high_load_seen"]:
+        return "blocked_waiting_110_load_to_normalize_before_harbor_preflight"
+    return "blocked_waiting_harbor_watchdog_check_receipt"
+
+
 def _ssh_metadata_phase_status(
     *,
     ssh_diagnosis: dict[str, Any],
@@ -394,14 +406,15 @@ def _parse_ssh_publickey_diagnosis_output(output: str) -> dict[str, Any]:
         and item["classifier"] == "systemctl_show_timeout"
         for item in systemd_units
     )
+    node_load_classifier = str(fields.get("NODE_LOAD_CLASSIFIER") or "")
+    node_high_load_seen = node_load_classifier == "high_load"
     ssh_port_tcp_open = str(fields.get("SSH_PORT") or "") == "tcp_open"
     node_exporter_ok = str(fields.get("NODE_EXPORTER") or "") == "ok"
-    node_high_load_seen = str(fields.get("NODE_LOAD_CLASSIFIER") or "") == "high_load"
     return {
         "receipt_seen": marker_seen,
         "target": str(fields.get("TARGET") or ""),
         "node_exporter_ok": node_exporter_ok,
-        "node_load_classifier": str(fields.get("NODE_LOAD_CLASSIFIER") or ""),
+        "node_load_classifier": node_load_classifier,
         "node_load1_per_cpu": _float_or_none(fields.get("NODE_LOAD1_PER_CPU")),
         "node_high_load_seen": node_high_load_seen,
         "ssh_port_tcp_open": ssh_port_tcp_open,