fix(agent): align log loop blocker with harbor remote control

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -405,25 +405,27 @@ def _harbor_recovery_receipt_inputs() -> list[dict[str, Any]]:
             "required_when": "before_and_after_110_controlled_recovery",
             "purpose": (
                 "classify cd run status, harbor repair waiting state, "
-                "awoooi-host no-matching runner, and stale job payloads"
+                "remote control channel availability, no-matching runners, "
+                "and stale job payloads"
             ),
             **metadata_boundary,
         },
         {
-            "input_id": "ssh_publickey_diagnosis_output",
-            "source": "diagnose-110-ssh-publickey-auth.sh",
-            "required_when": "diagnose_ssh_publickey_phase",
+            "input_id": "remote_control_channel_readback",
+            "source": "read-public-gitea-actions-queue.py --json plus bounded 110 ssh probe",
+            "required_when": "diagnose_remote_control_channel_phase",
             "purpose": (
-                "classify TCP banner, node load, and publickey auth timeout "
-                "without reading authorized_keys or asking for passwords"
+                "classify 110 TCP/SSH reachability, bounded timeout, node load, "
+                "and remote-control blocker without reading key material or "
+                "asking for passwords"
             ),
             **metadata_boundary,
         },
         {
-            "input_id": "ssh_local_repair_output",
+            "input_id": "local_console_recovery_output",
             "source": "recover-110-control-path-and-harbor-local.sh --check/apply",
-            "required_when": "after_diagnose_ssh_publickey_phase",
-            "purpose": "prove 110 ssh control-channel metadata and permissions",
+            "required_when": "after_remote_control_channel_diagnosis_phase",
+            "purpose": "prove 110 ssh control path, Harbor, and local console repair results",
             **metadata_boundary,
         },
         {
@@ -596,6 +598,14 @@ def _queue_readback_normalizer_contract() -> list[dict[str, Any]]:
             ],
             "learning_targets": ["km", "rag", "playbook", "mcp", "verifier"],
         },
+        {
+            "field_id": "latest_visible_harbor_110_repair_remote_control_channel_unavailable",
+            "purpose": "classify whether 110 rejects or times out the bounded Harbor repair control channel",
+            "writes_blockers": [
+                "gitea_queue_harbor_110_remote_control_channel_unavailable",
+            ],
+            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
+        },
         {
             "field_id": "current_cd_workflow_runner_readiness",
             "purpose": "classify whether current CD is waiting on a non110 runner label",

apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py

@@ -220,9 +220,11 @@ def _current_blocker_recovery(receipt: dict[str, Any]) -> dict[str, Any] | None:
         ),
         "post_apply_verifier": str(classification.get("post_apply_verifier") or ""),
         "safe_next_step": str(classification.get("safe_next_step") or ""),
-        "runtime_apply_required_on_110_local_console": (
-            blocker_id == "harbor_110_repair_no_matching_runner"
-        ),
+        "runtime_apply_required_on_110_local_console": blocker_id
+        in {
+            "harbor_110_repair_no_matching_runner",
+            "harbor_110_remote_control_channel_unavailable",
+        },
         "external_control_path_blocker": str(
             classification.get("ssh_auth_classification") or ""
         ),

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -49,7 +49,7 @@ def _assert_log_controlled_writeback_executor(payload: dict):
     assert len(context["current_blocker_execution_queue"]) == 1
     current_queue = context["current_blocker_execution_queue"][0]
     assert current_queue["queue_item_id"] == (
-        "current-p0-blocker::harbor_110_repair_no_matching_runner"
+        "current-p0-blocker::harbor_110_remote_control_channel_unavailable"
     )
     assert current_queue["controlled_recovery_package"] == (
         "recover-110-control-path-and-harbor-local.sh --check"
@@ -60,11 +60,11 @@ def _assert_log_controlled_writeback_executor(payload: dict):
     assert [phase["phase_id"] for phase in current_queue[
         "controlled_local_console_execution_plan"
     ]] == [
-        "diagnose_ssh_publickey",
+        "diagnose_remote_control_channel",
         "preflight_control_path_and_harbor",
-        "repair_ssh_metadata_if_check_confirms_metadata_drift",
+        "repair_ssh_service_or_metadata_if_check_confirms_drift",
         "repair_harbor_once_if_v2_still_502",
-        "verify_controlled_cd_lane",
+        "verify_harbor_queue_and_controlled_cd_lane",
     ]
     assert current_queue["post_recovery_readback_commands"] == [
         "read-public-gitea-actions-queue.py --json",
@@ -75,7 +75,7 @@ def _assert_log_controlled_writeback_executor(payload: dict):
         "forbidden_runtime_actions"
     ]
     assert current_queue["external_control_path_blocker"] == (
-        "publickey_offer_timeout"
+        "bounded_ssh_timeout"
     )
     assert current_queue["node_load_classifier"] == "high_load"
     assert current_queue["node_load_high"] is True

apps/api/tests/test_ai_agent_log_controlled_writeback_executor_readback_api.py

@@ -106,9 +106,11 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
         )
         assert batch["current_blocker_recovery_count"] == 1
         assert batch["current_blocker_recoveries"][0]["blocker_id"] == (
-            "harbor_110_repair_no_matching_runner"
+            "harbor_110_remote_control_channel_unavailable"
+        )
+        assert batch["current_blocker_recoveries"][0]["runner_label"] == (
+            "awoooi-non110-host"
         )
-        assert batch["current_blocker_recoveries"][0]["runner_label"] == "awoooi-host"
         assert batch["runtime_dispatch_performed"] is False
 
     context = payload["agent_consumption_context"]
@@ -121,12 +123,12 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
     current_queue = context["current_blocker_execution_queue"]
     assert len(current_queue) == 1
     assert current_queue[0]["queue_item_id"] == (
-        "current-p0-blocker::harbor_110_repair_no_matching_runner"
+        "current-p0-blocker::harbor_110_remote_control_channel_unavailable"
     )
     assert current_queue[0]["status"] == (
         "controlled_recovery_packaged_waiting_control_path_readback"
     )
-    assert current_queue[0]["runner_label"] == "awoooi-host"
+    assert current_queue[0]["runner_label"] == "awoooi-non110-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
@@ -145,8 +147,8 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
         for item in current_queue[0]["harbor_recovery_receipt_inputs"]
     ] == [
         "gitea_actions_queue_readback",
-        "ssh_publickey_diagnosis_output",
-        "ssh_local_repair_output",
+        "remote_control_channel_readback",
+        "local_console_recovery_output",
         "watchdog_check_output",
         "watchdog_repair_output",
         "controlled_cd_lane_readiness_output",
@@ -185,7 +187,7 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
         ]
         for item in current_queue[0]["harbor_recovery_receipt_output_contract"]
     )
-    assert current_queue[0]["queue_readback_normalizer_contract_count"] == 5
+    assert current_queue[0]["queue_readback_normalizer_contract_count"] == 6
     assert [
         item["field_id"]
         for item in current_queue[0]["queue_readback_normalizer_contract"]
@@ -193,6 +195,7 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
         "cd_run_jobs_payload_classifier",
         "harbor_110_repair_jobs_payload_classifier",
         "latest_visible_harbor_110_repair_no_matching_runner_label",
+        "latest_visible_harbor_110_repair_remote_control_channel_unavailable",
         "current_cd_workflow_runner_readiness",
         "controlled_profile_no_matching_runner_labels",
     ]
@@ -208,11 +211,11 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
     assert [phase["phase_id"] for phase in current_queue[0][
         "controlled_local_console_execution_plan"
     ]] == [
-        "diagnose_ssh_publickey",
+        "diagnose_remote_control_channel",
         "preflight_control_path_and_harbor",
-        "repair_ssh_metadata_if_check_confirms_metadata_drift",
+        "repair_ssh_service_or_metadata_if_check_confirms_drift",
         "repair_harbor_once_if_v2_still_502",
-        "verify_controlled_cd_lane",
+        "verify_harbor_queue_and_controlled_cd_lane",
     ]
     assert current_queue[0]["controlled_local_console_execution_plan"][0][
         "mode"
@@ -233,7 +236,7 @@ def _assert_executor_readback(payload: dict, *, public_endpoint: bool = False):
     )
     assert current_queue[0]["runtime_apply_required_on_110_local_console"] is True
     assert current_queue[0]["external_control_path_blocker"] == (
-        "publickey_offer_timeout"
+        "bounded_ssh_timeout"
     )
     assert current_queue[0]["control_path_pressure_blocker"] == "node_load_high"
     assert current_queue[0]["control_channel_readback_required"] is True

apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py

@@ -85,11 +85,11 @@ def _assert_controlled_writeback_plan(payload: dict, *, public_endpoint: bool =
         assert selector["package"] in {"apps/api", "ops/runner"}
         assert selector["tool"] in {
             "kubectl_logs_readback",
-            "read_public_gitea_queue_and_ssh_publickey_classifier",
+            "read_public_gitea_queue_and_remote_control_classifier",
         }
         assert selector["source_system"] in {
             "k8s_logs",
-            "gitea_queue_and_ssh_classifier_readback",
+            "gitea_queue_and_remote_control_classifier_readback",
         }
         assert selector["target_surface"]
         assert plan["source_of_truth_diff"]["current_state"] == (
@@ -124,7 +124,7 @@ def _assert_controlled_writeback_plan(payload: dict, *, public_endpoint: bool =
     assert {plan["target"] for plan in p0_plans} == set(target_rollups)
     assert {
         plan["current_blocker_recovery"]["blocker_id"] for plan in p0_plans
-    } == {"harbor_110_repair_no_matching_runner"}
+    } == {"harbor_110_remote_control_channel_unavailable"}
     assert {
         plan["current_blocker_recovery"]["controlled_recovery_package"]
         for plan in p0_plans
@@ -141,11 +141,11 @@ def _assert_controlled_writeback_plan(payload: dict, *, public_endpoint: bool =
             ]
         ]
         == [
-            "diagnose_ssh_publickey",
+            "diagnose_remote_control_channel",
             "preflight_control_path_and_harbor",
-            "repair_ssh_metadata_if_check_confirms_metadata_drift",
+            "repair_ssh_service_or_metadata_if_check_confirms_drift",
             "repair_harbor_once_if_v2_still_502",
-            "verify_controlled_cd_lane",
+            "verify_harbor_queue_and_controlled_cd_lane",
         ]
         for plan in p0_plans
     )
@@ -167,7 +167,7 @@ def _assert_controlled_writeback_plan(payload: dict, *, public_endpoint: bool =
     assert {
         plan["current_blocker_recovery"]["external_control_path_blocker"]
         for plan in p0_plans
-    } == {"publickey_offer_timeout"}
+    } == {"bounded_ssh_timeout"}
 
     boundaries = payload["operation_boundaries"]
     assert boundaries["plan_readback_only"] is True

apps/api/tests/test_ai_agent_log_feedback_receipt_dry_run_api.py

@@ -93,7 +93,7 @@ def _assert_feedback_dry_run_payload(payload: dict):
     }
     assert {
         receipt["classification"]["current_blocker"] for receipt in p0_receipts
-    } == {"harbor_110_repair_no_matching_runner"}
+    } == {"harbor_110_remote_control_channel_unavailable"}
     assert {
         receipt["classification"]["controlled_recovery_package"]
         for receipt in p0_receipts

apps/api/tests/test_ai_agent_log_intelligence_integration_readback_api.py

@@ -104,11 +104,14 @@ def _assert_log_intelligence_payload(payload: dict):
     p0_sample = samples["p0_110_harbor_runner_control_path_sample"]
     assert p0_sample["service"] == "awoooi-110-control-path"
     assert p0_sample["classification"]["current_blocker"] == (
-        "harbor_110_repair_no_matching_runner"
+        "harbor_110_remote_control_channel_unavailable"
     )
     assert p0_sample["classification"]["ssh_auth_classification"] == (
-        "publickey_offer_timeout"
+        "bounded_ssh_timeout"
     )
+    assert p0_sample["classification"]["remote_control_channel"] == "unavailable"
+    assert p0_sample["classification"]["bounded_ssh_timeout_seen"] is True
+    assert p0_sample["classification"]["remote_ssh_reachable"] is False
     assert p0_sample["classification"]["controlled_recovery_package"] == (
         "recover-110-control-path-and-harbor-local.sh --check"
     )
@@ -119,11 +122,11 @@ def _assert_log_intelligence_payload(payload: dict):
         "controlled_local_console_execution_plan"
     ]
     assert [phase["phase_id"] for phase in recovery_plan] == [
-        "diagnose_ssh_publickey",
+        "diagnose_remote_control_channel",
         "preflight_control_path_and_harbor",
-        "repair_ssh_metadata_if_check_confirms_metadata_drift",
+        "repair_ssh_service_or_metadata_if_check_confirms_drift",
         "repair_harbor_once_if_v2_still_502",
-        "verify_controlled_cd_lane",
+        "verify_harbor_queue_and_controlled_cd_lane",
     ]
     assert recovery_plan[0]["mode"] == "read_only"
     assert recovery_plan[2]["mode"] == "controlled_apply"

apps/api/tests/test_awoooi_priority_work_order_readback_api.py

@@ -258,10 +258,10 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
     )
     assert state["ai_loop_current_blocker_execution_queue_count"] == 1
     assert state["ai_loop_current_blocker_id"] == (
-        "harbor_110_repair_no_matching_runner"
+        "harbor_110_remote_control_channel_unavailable"
     )
     assert state["ai_loop_current_blocker_control_path_blocker"] == (
-        "publickey_offer_timeout"
+        "bounded_ssh_timeout"
     )
     assert state["ai_loop_current_blocker_control_path_pressure_blocker"] == (
         "node_load_high"
@@ -269,9 +269,9 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
     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"
+        "harbor_110_remote_control_channel_unavailable"
     )
-    assert evidence["ai_loop_current_blocker_runner_label"] == "awoooi-host"
+    assert evidence["ai_loop_current_blocker_runner_label"] == "awoooi-non110-host"
     assert evidence["ai_loop_current_blocker_registry_v2_status"] == 502
     assert evidence["ai_loop_current_blocker_controlled_recovery_package"] == (
         "recover-110-control-path-and-harbor-local.sh --check"
@@ -283,7 +283,7 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
         "controlled_after_110_local_console_preflight"
     )
     assert evidence["ai_loop_current_blocker_control_path_blocker"] == (
-        "publickey_offer_timeout"
+        "bounded_ssh_timeout"
     )
     assert evidence["ai_loop_current_blocker_control_path_pressure_blocker"] == (
         "node_load_high"
@@ -300,11 +300,11 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
     assert len(evidence["ai_loop_current_blocker_target_batches"]) == 6
     assert evidence["ai_loop_current_blocker_local_console_phase_count"] == 5
     assert evidence["ai_loop_current_blocker_local_console_phase_ids"] == [
-        "diagnose_ssh_publickey",
+        "diagnose_remote_control_channel",
         "preflight_control_path_and_harbor",
-        "repair_ssh_metadata_if_check_confirms_metadata_drift",
+        "repair_ssh_service_or_metadata_if_check_confirms_drift",
         "repair_harbor_once_if_v2_still_502",
-        "verify_controlled_cd_lane",
+        "verify_harbor_queue_and_controlled_cd_lane",
     ]
     assert evidence["ai_loop_current_blocker_local_console_execution_plan"][0][
         "mode"
@@ -332,8 +332,8 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
     assert evidence["ai_loop_current_blocker_harbor_recovery_receipt_input_count"] == 10
     assert evidence["ai_loop_current_blocker_harbor_recovery_receipt_input_ids"] == [
         "gitea_actions_queue_readback",
-        "ssh_publickey_diagnosis_output",
-        "ssh_local_repair_output",
+        "remote_control_channel_readback",
+        "local_console_recovery_output",
         "watchdog_check_output",
         "watchdog_repair_output",
         "controlled_cd_lane_readiness_output",
@@ -367,6 +367,7 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
         "cd_run_jobs_payload_classifier",
         "harbor_110_repair_jobs_payload_classifier",
         "latest_visible_harbor_110_repair_no_matching_runner_label",
+        "latest_visible_harbor_110_repair_remote_control_channel_unavailable",
         "current_cd_workflow_runner_readiness",
         "controlled_profile_no_matching_runner_labels",
     ]
@@ -390,7 +391,7 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
     assert "Do not restart Docker daemon" in in_progress["professional_fix"]["action"]
     assert payload["summary"]["ai_loop_current_blocker_execution_queue_count"] == 1
     assert payload["summary"]["ai_loop_current_blocker_id"] == (
-        "harbor_110_repair_no_matching_runner"
+        "harbor_110_remote_control_channel_unavailable"
     )
     assert payload["summary"]["ai_loop_current_blocker_control_path_pressure_blocker"] == (
         "node_load_high"
@@ -427,7 +428,7 @@ def test_awoooi_priority_work_order_readback_overlays_ai_loop_current_blocker_qu
         payload["summary"][
             "ai_loop_current_blocker_queue_readback_normalizer_contract_count"
         ]
-        == 5
+        == 6
     )
     assert payload["summary"][
         "ai_loop_current_blocker_queue_readback_normalizer_field_ids"