feat(agent): add p0 harbor blocker loop sample

apps/api/src/services/ai_agent_log_intelligence_integration_readback.py

@@ -146,6 +146,15 @@ def load_latest_ai_agent_log_intelligence_integration_readback(
     missing_lane_count = len(lanes) - ready_lane_count
     runtime_sample = _load_runtime_sample(root)
     runtime_sample_present = runtime_sample is not None
+    runtime_samples = runtime_sample.get("samples", []) if runtime_sample else []
+    current_p0_blocker_sample_present = _sample_present(
+        runtime_samples,
+        "p0_110_harbor_runner_control_path_sample",
+    )
+    controlled_recovery_package_sample_present = _sample_event_present(
+        runtime_samples,
+        "local_recovery_package_packaged",
+    )
     label_field_count = len(
         {field for group in _LABEL_TAXONOMY for field in group["required_fields"]}
     )
@@ -199,6 +208,12 @@ def load_latest_ai_agent_log_intelligence_integration_readback(
             "runtime_sample_count": len(runtime_sample.get("samples", []))
             if runtime_sample
             else 0,
+            "current_p0_blocker_runtime_sample_present": (
+                current_p0_blocker_sample_present
+            ),
+            "controlled_recovery_package_sample_present": (
+                controlled_recovery_package_sample_present
+            ),
             "km_rag_feedback_receipt_ready": _feedback_ready(
                 runtime_sample,
                 "km_rag_feedback_receipt_ready",
@@ -298,6 +313,21 @@ def _feedback_ready(runtime_sample: dict[str, Any] | None, field: str) -> bool:
     return (runtime_sample.get("feedback_receipt_readiness") or {}).get(field) is True
 
 
+def _sample_present(samples: Any, sample_id: str) -> bool:
+    return any(
+        isinstance(sample, dict) and sample.get("sample_id") == sample_id
+        for sample in samples
+    )
+
+
+def _sample_event_present(samples: Any, event: str) -> bool:
+    return any(
+        isinstance(sample, dict)
+        and event in {str(item) for item in sample.get("observed_events", [])}
+        for sample in samples
+    )
+
+
 def _build_lane(root: Path, lane: dict[str, Any]) -> dict[str, Any]:
     evidence_refs = [str(ref) for ref in lane["required_refs"]]
     missing_refs = [ref for ref in evidence_refs if not _ref_exists(root, ref)]

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -26,11 +26,11 @@ def _assert_log_controlled_writeback_executor(payload: dict):
     assert policy["rollback_required"] is True
     assert policy["post_apply_verifier_required"] is True
 
-    assert payload["rollups"]["source_writeback_plan_count"] == 12
+    assert payload["rollups"]["source_writeback_plan_count"] == 18
     assert payload["rollups"]["execution_batch_count"] == 6
     assert payload["rollups"]["ready_execution_batch_count"] == 6
-    assert payload["rollups"]["target_selector_count"] == 12
-    assert payload["rollups"]["source_of_truth_diff_count"] == 12
+    assert payload["rollups"]["target_selector_count"] == 18
+    assert payload["rollups"]["source_of_truth_diff_count"] == 18
     assert payload["rollups"]["controlled_executor_dispatch_ready"] is True
     assert payload["rollups"]["controlled_apply_enabled_by_policy"] is True
     assert payload["rollups"]["runtime_dispatch_performed"] is False

apps/api/tests/test_ai_agent_log_controlled_writeback_plan_readback_api.py

@@ -48,20 +48,20 @@ def _assert_controlled_writeback_plan(payload: dict):
     assert controlled_apply["critical_break_glass_required"] is True
 
     assert payload["rollups"]["target_count"] == 6
-    assert payload["rollups"]["candidate_receipt_count"] == 12
-    assert payload["rollups"]["writeback_plan_count"] == 12
-    assert payload["rollups"]["ready_writeback_plan_count"] == 12
-    assert payload["rollups"]["check_mode_plan_count"] == 12
-    assert payload["rollups"]["rollback_plan_count"] == 12
-    assert payload["rollups"]["post_apply_verifier_plan_count"] == 12
+    assert payload["rollups"]["candidate_receipt_count"] == 18
+    assert payload["rollups"]["writeback_plan_count"] == 18
+    assert payload["rollups"]["ready_writeback_plan_count"] == 18
+    assert payload["rollups"]["check_mode_plan_count"] == 18
+    assert payload["rollups"]["rollback_plan_count"] == 18
+    assert payload["rollups"]["post_apply_verifier_plan_count"] == 18
     assert payload["rollups"]["controlled_writeback_plan_ready"] is True
     assert payload["rollups"]["runtime_write_performed"] is False
 
     target_rollups = {item["target"]: item for item in payload["target_rollups"]}
     assert set(target_rollups) == {"km", "rag", "playbook", "mcp", "verifier", "ai_agent"}
     for rollup in target_rollups.values():
-        assert rollup["plan_count"] == 2
-        assert rollup["ready_plan_count"] == 2
+        assert rollup["plan_count"] == 3
+        assert rollup["ready_plan_count"] == 3
         assert rollup["check_mode_ready"] is True
         assert rollup["rollback_ready"] is True
         assert rollup["post_apply_verifier_ready"] is True
@@ -74,10 +74,20 @@ def _assert_controlled_writeback_plan(payload: dict):
         selector = plan["target_selector"]
         assert selector["project_id"] == "awoooi"
         assert selector["product"] == "awoooi"
-        assert selector["service"] in {"awoooi-api", "awoooi-worker"}
-        assert selector["package"] == "apps/api"
-        assert selector["tool"] == "kubectl_logs_readback"
-        assert selector["source_system"] == "k8s_logs"
+        assert selector["service"] in {
+            "awoooi-api",
+            "awoooi-worker",
+            "awoooi-110-control-path",
+        }
+        assert selector["package"] in {"apps/api", "ops/runner"}
+        assert selector["tool"] in {
+            "kubectl_logs_readback",
+            "read_public_gitea_queue_and_ssh_publickey_classifier",
+        }
+        assert selector["source_system"] in {
+            "k8s_logs",
+            "gitea_queue_and_ssh_classifier_readback",
+        }
         assert selector["target_surface"]
         assert plan["source_of_truth_diff"]["current_state"] == (
             "metadata_only_feedback_not_written"
@@ -101,6 +111,15 @@ def _assert_controlled_writeback_plan(payload: dict):
             "post-write-verifier://ai-agent-log-feedback/"
         )
 
+    p0_plans = [
+        plan
+        for plan in plans
+        if plan["target_selector"]["source_sample_id"]
+        == "p0_110_harbor_runner_control_path_sample"
+    ]
+    assert len(p0_plans) == 6
+    assert {plan["target"] for plan in p0_plans} == set(target_rollups)
+
     boundaries = payload["operation_boundaries"]
     assert boundaries["plan_readback_only"] is True
     assert boundaries["km_write_performed"] is False

apps/api/tests/test_ai_agent_log_intelligence_integration_readback_api.py

@@ -52,7 +52,9 @@ def _assert_log_intelligence_payload(payload: dict):
     assert payload["rollups"]["missing_lane_count"] == 0
     assert payload["rollups"]["readiness_percent"] == 100
     assert payload["rollups"]["runtime_e2e_log_sample_readback_present"] is True
-    assert payload["rollups"]["runtime_sample_count"] == 2
+    assert payload["rollups"]["runtime_sample_count"] == 3
+    assert payload["rollups"]["current_p0_blocker_runtime_sample_present"] is True
+    assert payload["rollups"]["controlled_recovery_package_sample_present"] is True
     assert payload["rollups"]["km_rag_feedback_receipt_ready"] is True
     assert payload["rollups"]["playbook_feedback_receipt_ready"] is True
     assert payload["rollups"]["mcp_audit_feedback_receipt_ready"] is True
@@ -62,7 +64,7 @@ def _assert_log_intelligence_payload(payload: dict):
     assert payload["runtime_e2e_verifier"]["schema_version"] == (
         "ai_agent_log_intelligence_runtime_sample_readback_v1"
     )
-    assert payload["runtime_e2e_verifier"]["label_receipt"]["runtime_sample_count"] == 2
+    assert payload["runtime_e2e_verifier"]["label_receipt"]["runtime_sample_count"] == 3
     assert payload["runtime_e2e_verifier"]["operation_boundaries"][
         "runtime_log_sample_read_performed"
     ] is True
@@ -95,6 +97,26 @@ def _assert_log_intelligence_payload(payload: dict):
     assert all(lane["runtime_write_enabled"] is False for lane in lanes.values())
     assert all(not lane["missing_refs"] for lane in lanes.values())
 
+    samples = {
+        sample["sample_id"]: sample
+        for sample in payload["runtime_e2e_verifier"]["samples"]
+    }
+    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"
+    )
+    assert p0_sample["classification"]["ssh_auth_classification"] == (
+        "publickey_offer_timeout"
+    )
+    assert p0_sample["classification"]["controlled_recovery_package"] == (
+        "recover-110-control-path-and-harbor-local.sh --check"
+    )
+    assert p0_sample["classification"]["post_apply_verifier"] == (
+        "check-awoooi-110-controlled-cd-lane-readiness.sh"
+    )
+    assert p0_sample["raw_log_payload_persisted"] is False
+
     label_groups = {
         group["label_group"]: set(group["required_fields"])
         for group in payload["label_taxonomy"]