From f0c24dcd939b208f6949bbffea5e47904b8192e5 Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 1 Jul 2026 07:48:52 +0800 Subject: [PATCH] fix(recovery): propagate 110 high load blocker --- ..._controlled_writeback_executor_readback.py | 12 +++++++++ .../awoooi_priority_work_order_readback.py | 17 ++++++++++++ ...or_registry_controlled_recovery_receipt.py | 23 ++++++++++++---- ...est_ai_agent_autonomous_runtime_control.py | 7 +++++ ...trolled_writeback_executor_readback_api.py | 7 +++++ ...awoooi_priority_work_order_readback_api.py | 14 ++++++++++ ...or_registry_controlled_recovery_receipt.py | 27 +++++++++++++++++++ ...ence-runtime-sample-readback.snapshot.json | 3 ++- 8 files changed, 104 insertions(+), 6 deletions(-) diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py index 909e2a1d5..9041b5255 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py +++ b/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, diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py index 52ab35641..8be595b01 100644 --- a/apps/api/src/services/awoooi_priority_work_order_readback.py +++ b/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 ) diff --git a/apps/api/src/services/harbor_registry_controlled_recovery_receipt.py b/apps/api/src/services/harbor_registry_controlled_recovery_receipt.py index a789fdb23..1f1bfe2c7 100644 --- a/apps/api/src/services/harbor_registry_controlled_recovery_receipt.py +++ b/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, diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index 80b51de4c..690e2f024 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -35,6 +35,10 @@ def _assert_log_controlled_writeback_executor(payload: dict): assert payload["rollups"]["controlled_apply_enabled_by_policy"] is True assert payload["rollups"]["current_blocker_execution_queue_count"] == 1 assert payload["rollups"]["current_blocker_control_path_blocked_count"] == 1 + assert ( + payload["rollups"]["current_blocker_control_path_pressure_blocked_count"] + == 1 + ) assert payload["rollups"]["current_blocker_local_recovery_package_count"] == 1 assert payload["rollups"]["runtime_dispatch_performed"] is False @@ -73,6 +77,9 @@ def _assert_log_controlled_writeback_executor(payload: dict): assert current_queue["external_control_path_blocker"] == ( "publickey_offer_timeout" ) + assert current_queue["node_load_classifier"] == "high_load" + assert current_queue["node_load_high"] is True + assert current_queue["control_path_pressure_blocker"] == "node_load_high" assert context["raw_payload_required"] is False boundaries = payload["operation_boundaries"] diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py index 4192632e4..7b5c89356 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py @@ -65,6 +65,10 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False): assert payload["rollups"]["controlled_apply_enabled_by_policy"] is True assert payload["rollups"]["current_blocker_execution_queue_count"] == 1 assert payload["rollups"]["current_blocker_control_path_blocked_count"] == 1 + assert ( + payload["rollups"]["current_blocker_control_path_pressure_blocked_count"] + == 1 + ) assert payload["rollups"]["current_blocker_local_recovery_package_count"] == 1 assert ( payload["rollups"]["current_blocker_harbor_recovery_receipt_input_count"] @@ -117,6 +121,8 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False): "controlled_recovery_packaged_waiting_control_path_readback" ) assert current_queue[0]["runner_label"] == "awoooi-host" + assert current_queue[0]["node_load_classifier"] == "high_load" + assert current_queue[0]["node_load_high"] is True assert current_queue[0]["registry_v2_status"] == 502 assert current_queue[0]["controlled_recovery_package"] == ( "recover-110-control-path-and-harbor-local.sh --check" @@ -180,6 +186,7 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False): assert current_queue[0]["external_control_path_blocker"] == ( "publickey_offer_timeout" ) + assert current_queue[0]["control_path_pressure_blocker"] == "node_load_high" assert current_queue[0]["control_channel_readback_required"] is True assert set(current_queue[0]["learning_writeback_targets"]) == set(batches) assert len(current_queue[0]["target_batches"]) == 6 diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py index 208488b95..f0f56970d 100644 --- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py +++ b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py @@ -263,6 +263,10 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu assert state["ai_loop_current_blocker_control_path_blocker"] == ( "publickey_offer_timeout" ) + assert state["ai_loop_current_blocker_control_path_pressure_blocker"] == ( + "node_load_high" + ) + assert state["ai_loop_current_blocker_node_load_classifier"] == "high_load" assert evidence["ai_loop_current_blocker_execution_queue_count"] == 1 assert evidence["ai_loop_current_blocker_id"] == ( "harbor_110_repair_no_matching_runner" @@ -281,6 +285,10 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu assert evidence["ai_loop_current_blocker_control_path_blocker"] == ( "publickey_offer_timeout" ) + assert evidence["ai_loop_current_blocker_control_path_pressure_blocker"] == ( + "node_load_high" + ) + assert evidence["ai_loop_current_blocker_node_load_classifier"] == "high_load" assert set(evidence["ai_loop_current_blocker_learning_writeback_targets"]) == { "km", "rag", @@ -350,6 +358,12 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu assert payload["summary"]["ai_loop_current_blocker_id"] == ( "harbor_110_repair_no_matching_runner" ) + assert payload["summary"]["ai_loop_current_blocker_control_path_pressure_blocker"] == ( + "node_load_high" + ) + assert payload["summary"]["ai_loop_current_blocker_node_load_classifier"] == ( + "high_load" + ) assert payload["summary"]["ai_loop_current_blocker_local_console_phase_count"] == 5 assert payload["summary"][ "ai_loop_current_blocker_post_recovery_readback_command_count" diff --git a/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py b/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py index 2ff63c6be..c4eb293fb 100644 --- a/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py +++ b/apps/api/tests/test_harbor_registry_controlled_recovery_receipt.py @@ -198,6 +198,33 @@ def test_harbor_recovery_receipt_classifies_publickey_offer_timeout() -> None: ] == "blocked_waiting_ssh_metadata_repair_receipt_after_publickey_timeout" +def test_harbor_recovery_receipt_classifies_110_high_load() -> None: + diagnosis_output = _ssh_publickey_diagnosis_output().replace( + "NODE_LOAD_CLASSIFIER=load_not_high", + "NODE_LOAD_CLASSIFIER=high_load", + ) + + payload = validate_harbor_registry_controlled_recovery_receipt( + { + "ssh_publickey_diagnosis_output": diagnosis_output, + } + ) + + assert "ssh_publickey_node_high_load_on_110" in payload["active_blockers"] + diagnosis = payload["readback"]["ssh_publickey_diagnosis"] + assert diagnosis["node_load_classifier"] == "high_load" + assert diagnosis["node_high_load_seen"] is True + phases = { + phase["phase_id"]: phase + for phase in payload["local_console_phase_readback"]["phases"] + } + assert phases["preflight_control_path_and_harbor"]["status"] == ( + "blocked_waiting_110_load_to_normalize_before_harbor_preflight" + ) + assert payload["rollups"]["ssh_publickey_node_load_classifier"] == "high_load" + assert payload["rollups"]["ssh_publickey_node_high_load_seen"] is True + + def test_harbor_recovery_receipt_surfaces_gitea_queue_blockers() -> None: payload = validate_harbor_registry_controlled_recovery_receipt( { diff --git a/docs/operations/ai-agent-log-intelligence-runtime-sample-readback.snapshot.json b/docs/operations/ai-agent-log-intelligence-runtime-sample-readback.snapshot.json index 3d29b5850..d606c7940 100644 --- a/docs/operations/ai-agent-log-intelligence-runtime-sample-readback.snapshot.json +++ b/docs/operations/ai-agent-log-intelligence-runtime-sample-readback.snapshot.json @@ -84,6 +84,7 @@ "harbor_registry_public_route_unavailable", "harbor_110_repair_no_matching_runner", "ssh_publickey_offer_timeout", + "node_load_high", "controlled_lane_verifier_packaged", "local_recovery_package_packaged" ], @@ -105,7 +106,7 @@ "runner_label": "awoooi-host", "current_blocker": "harbor_110_repair_no_matching_runner", "ssh_auth_classification": "publickey_offer_timeout", - "node_load_classifier": "load_not_high", + "node_load_classifier": "high_load", "registry_v2_status": 502, "controlled_recovery_package": "recover-110-control-path-and-harbor-local.sh --check", "post_apply_verifier": "check-awoooi-110-controlled-cd-lane-readiness.sh",