feat(agent): expose current blocker execution queue

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -3018,6 +3018,11 @@ def _attach_runtime_receipt_readback(
     log_executor_queue = log_executor_context.get("next_action_queue")
     if not isinstance(log_executor_queue, list):
         log_executor_queue = []
+    log_executor_current_blocker_queue = log_executor_context.get(
+        "current_blocker_execution_queue"
+    )
+    if not isinstance(log_executor_current_blocker_queue, list):
+        log_executor_current_blocker_queue = []
     log_executor_blockers = log_executor.get("active_blockers")
     if not isinstance(log_executor_blockers, list):
         log_executor_blockers = []
@@ -3140,6 +3145,19 @@ def _attach_runtime_receipt_readback(
         "live_log_controlled_writeback_next_action_queue_count": len(
             log_executor_queue
         ),
+        "live_log_controlled_writeback_current_blocker_queue_count": len(
+            log_executor_current_blocker_queue
+        ),
+        "live_log_controlled_writeback_current_blocker_control_path_blocked_count": (
+            _int_value(
+                log_executor_rollups.get("current_blocker_control_path_blocked_count")
+            )
+        ),
+        "live_log_controlled_writeback_current_blocker_local_recovery_package_count": (
+            _int_value(
+                log_executor_rollups.get("current_blocker_local_recovery_package_count")
+            )
+        ),
         "live_log_controlled_writeback_dispatch_count": _int_value(
             log_dispatch_summary.get("total")
         ),

apps/api/src/services/ai_agent_log_controlled_writeback_executor_readback.py

@@ -37,6 +37,8 @@ def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[st
         writeback_plans=writeback_plans,
         execution_batches=execution_batches,
     )
+    agent_context = _agent_consumption_context(execution_batches)
+    current_blocker_queue = agent_context["current_blocker_execution_queue"]
 
     return {
         "schema_version": _SCHEMA_VERSION,
@@ -66,7 +68,7 @@ def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[st
             "post_apply_verifier_required": True,
         },
         "execution_batches": execution_batches,
-        "agent_consumption_context": _agent_consumption_context(execution_batches),
+        "agent_consumption_context": agent_context,
         "rollups": {
             "source_writeback_plan_count": len(writeback_plans),
             "execution_batch_count": len(execution_batches),
@@ -95,6 +97,15 @@ def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[st
             ),
             "controlled_executor_dispatch_ready": not active_blockers,
             "controlled_apply_enabled_by_policy": True,
+            "current_blocker_execution_queue_count": len(current_blocker_queue),
+            "current_blocker_control_path_blocked_count": sum(
+                1
+                for item in current_blocker_queue
+                if item["external_control_path_blocker"]
+            ),
+            "current_blocker_local_recovery_package_count": sum(
+                1 for item in current_blocker_queue if item["controlled_recovery_package"]
+            ),
             "runtime_dispatch_performed": False,
         },
         "active_blockers": active_blockers,
@@ -133,6 +144,7 @@ def _execution_batch(target: str, plans: list[dict[str, Any]]) -> dict[str, Any]
     risk_tier = "medium" if target in {"km", "rag", "playbook"} else "low"
     target_selectors = [plan.get("target_selector") or {} for plan in plans]
     diffs = [plan.get("source_of_truth_diff") or {} for plan in plans]
+    current_blocker_recoveries = _current_blocker_recoveries(plans)
     rollback_refs = [
         str((plan.get("rollback") or {}).get("rollback_ref") or "")
         for plan in plans
@@ -162,6 +174,8 @@ def _execution_batch(target: str, plans: list[dict[str, Any]]) -> dict[str, Any]
         "target_selectors": target_selectors,
         "source_of_truth_diff_count": len(diffs),
         "source_of_truth_diffs": diffs,
+        "current_blocker_recovery_count": len(current_blocker_recoveries),
+        "current_blocker_recoveries": current_blocker_recoveries,
         "check_mode": {
             "enabled": True,
             "required": True,
@@ -241,9 +255,15 @@ def _agent_consumption_context(execution_batches: list[dict[str, Any]]) -> dict[
                 "check_mode_required": batch["check_mode"]["required"],
                 "rollback_required": batch["rollback"]["required"],
                 "post_apply_verifier_required": batch["post_apply_verifier"]["required"],
+                "current_blocker_recovery_count": batch[
+                    "current_blocker_recovery_count"
+                ],
             }
             for batch in ready_batches
         ],
+        "current_blocker_execution_queue": _current_blocker_execution_queue(
+            ready_batches
+        ),
         "learning_feedback_targets": [
             "km",
             "rag",
@@ -256,6 +276,76 @@ def _agent_consumption_context(execution_batches: list[dict[str, Any]]) -> dict[
     }
 
 
+def _current_blocker_recoveries(plans: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    recoveries = []
+    for plan in plans:
+        recovery = plan.get("current_blocker_recovery")
+        if not isinstance(recovery, dict):
+            continue
+        recoveries.append(dict(recovery))
+    return recoveries
+
+
+def _current_blocker_execution_queue(
+    ready_batches: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    queue_by_key: dict[tuple[str, str], dict[str, Any]] = {}
+    for batch in ready_batches:
+        target = str(batch.get("target") or "")
+        batch_id = str(batch.get("batch_id") or "")
+        for recovery in batch.get("current_blocker_recoveries") or []:
+            if not isinstance(recovery, dict):
+                continue
+            source_sample_id = str(recovery.get("source_sample_id") or "")
+            blocker_id = str(recovery.get("blocker_id") or "")
+            if not source_sample_id or not blocker_id:
+                continue
+            key = (source_sample_id, blocker_id)
+            queue_item = queue_by_key.setdefault(
+                key,
+                _current_blocker_queue_item(recovery),
+            )
+            queue_item["target_batches"] = _unique(
+                queue_item["target_batches"] + [batch_id]
+            )
+            queue_item["learning_writeback_targets"] = _unique(
+                queue_item["learning_writeback_targets"] + [target]
+            )
+    return list(queue_by_key.values())
+
+
+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 "")
+    return {
+        "queue_item_id": f"current-p0-blocker::{blocker_id}",
+        "source_sample_id": source_sample_id,
+        "incident_id": str(recovery.get("incident_id") or ""),
+        "blocker_id": blocker_id,
+        "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 ""),
+        "registry_v2_status": recovery.get("registry_v2_status"),
+        "controlled_recovery_package": str(
+            recovery.get("controlled_recovery_package") or ""
+        ),
+        "post_apply_verifier": str(recovery.get("post_apply_verifier") or ""),
+        "safe_next_step": str(recovery.get("safe_next_step") or ""),
+        "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
+        ),
+        "external_control_path_blocker": external_blocker,
+        "control_channel_readback_required": bool(external_blocker),
+        "metadata_only_writeback_ready": True,
+        "runtime_dispatch_performed": False,
+        "raw_payload_required": False,
+        "target_batches": [],
+        "learning_writeback_targets": [],
+    }
+
+
 def _active_blockers(
     *,
     plan_ready: bool,

apps/api/src/services/ai_agent_log_controlled_writeback_plan_readback.py

@@ -85,6 +85,18 @@ def load_latest_ai_agent_log_controlled_writeback_plan_readback() -> dict[str, A
             "ready_writeback_plan_count": sum(
                 1 for plan in writeback_plans if plan["status"] == "controlled_apply_ready"
             ),
+            "current_blocker_recovery_plan_count": sum(
+                1 for plan in writeback_plans if plan.get("current_blocker_recovery")
+            ),
+            "current_blocker_recovery_unique_count": len(
+                _unique(
+                    [
+                        str((plan.get("current_blocker_recovery") or {}).get("blocker_id"))
+                        for plan in writeback_plans
+                        if plan.get("current_blocker_recovery")
+                    ]
+                )
+            ),
             "check_mode_plan_count": sum(
                 1 for plan in writeback_plans if plan["check_mode"]["enabled"] is True
             ),
@@ -129,59 +141,95 @@ def _writeback_plans(receipts: list[dict[str, Any]]) -> list[dict[str, Any]]:
         source_sample_id = str(receipt.get("source_sample_id") or "")
         receipt_id = str(receipt.get("receipt_id") or "")
         plan_id = f"log-feedback-writeback::{target}::{source_sample_id}"
-        plans.append(
-            {
-                "plan_id": plan_id,
-                "receipt_id": receipt_id,
-                "target": target,
-                "status": "controlled_apply_ready",
-                "risk_tier": "medium" if target in {"km", "rag", "playbook"} else "low",
-                "target_selector": {
-                    "project_id": receipt.get("project_id"),
-                    "product": receipt.get("product"),
-                    "service": receipt.get("service"),
-                    "package": receipt.get("package"),
-                    "tool": receipt.get("tool"),
-                    "source_system": receipt.get("source_system"),
-                    "source_sample_id": source_sample_id,
-                    "target_surface": _target_surface(target),
-                },
-                "source_of_truth_diff": {
-                    "current_state": "metadata_only_feedback_not_written",
-                    "desired_state": "metadata_only_feedback_receipt_bound_to_target",
-                    "delta_kind": _delta_kind(target),
-                    "raw_payload_included": False,
-                    "redaction_state": receipt.get("redaction_state"),
-                    "observed_event_count": receipt.get("observed_event_count"),
-                    "observed_field_count": receipt.get("observed_field_count"),
-                },
-                "check_mode": {
-                    "enabled": True,
-                    "checks": [
-                        "target_selector_resolves_single_surface",
-                        "metadata_only_redaction_state",
-                        "raw_log_payload_absent",
-                        "post_apply_verifier_ref_present",
-                    ],
-                },
-                "rollback": {
-                    "required": True,
-                    "rollback_ref": f"rollback://ai-agent-log-feedback/{target}/{source_sample_id}",
-                    "strategy": "mark_receipt_superseded_and_remove_target_binding",
-                },
-                "post_apply_verifier": {
-                    "required": True,
-                    "verifier_ref": f"post-write-verifier://ai-agent-log-feedback/{target}/{source_sample_id}",
-                    "canonical_readback": (
-                        "/api/v1/agents/agent-log-controlled-writeback-plan-readback"
-                    ),
-                },
-                "write_enabled_by_plan": False,
-            }
-        )
+        current_blocker_recovery = _current_blocker_recovery(receipt)
+        plan = {
+            "plan_id": plan_id,
+            "receipt_id": receipt_id,
+            "target": target,
+            "status": "controlled_apply_ready",
+            "risk_tier": "medium" if target in {"km", "rag", "playbook"} else "low",
+            "target_selector": {
+                "project_id": receipt.get("project_id"),
+                "product": receipt.get("product"),
+                "service": receipt.get("service"),
+                "package": receipt.get("package"),
+                "tool": receipt.get("tool"),
+                "source_system": receipt.get("source_system"),
+                "source_sample_id": source_sample_id,
+                "target_surface": _target_surface(target),
+            },
+            "source_of_truth_diff": {
+                "current_state": "metadata_only_feedback_not_written",
+                "desired_state": "metadata_only_feedback_receipt_bound_to_target",
+                "delta_kind": _delta_kind(target),
+                "raw_payload_included": False,
+                "redaction_state": receipt.get("redaction_state"),
+                "observed_event_count": receipt.get("observed_event_count"),
+                "observed_field_count": receipt.get("observed_field_count"),
+            },
+            "check_mode": {
+                "enabled": True,
+                "checks": [
+                    "target_selector_resolves_single_surface",
+                    "metadata_only_redaction_state",
+                    "raw_log_payload_absent",
+                    "post_apply_verifier_ref_present",
+                ],
+            },
+            "rollback": {
+                "required": True,
+                "rollback_ref": f"rollback://ai-agent-log-feedback/{target}/{source_sample_id}",
+                "strategy": "mark_receipt_superseded_and_remove_target_binding",
+            },
+            "post_apply_verifier": {
+                "required": True,
+                "verifier_ref": f"post-write-verifier://ai-agent-log-feedback/{target}/{source_sample_id}",
+                "canonical_readback": (
+                    "/api/v1/agents/agent-log-controlled-writeback-plan-readback"
+                ),
+            },
+            "write_enabled_by_plan": False,
+        }
+        if current_blocker_recovery:
+            plan["current_blocker_recovery"] = current_blocker_recovery
+        plans.append(plan)
     return plans
 
 
+def _current_blocker_recovery(receipt: dict[str, Any]) -> dict[str, Any] | None:
+    classification = receipt.get("classification")
+    if not isinstance(classification, dict):
+        return None
+    blocker_id = str(classification.get("current_blocker") or "")
+    if not blocker_id:
+        return None
+    source_sample_id = str(receipt.get("source_sample_id") or "")
+    return {
+        "source_sample_id": source_sample_id,
+        "incident_id": str(classification.get("incident_id") or ""),
+        "blocker_id": blocker_id,
+        "risk_tier": str(classification.get("risk_tier") or "unknown"),
+        "runner_label": str(classification.get("runner_label") or ""),
+        "ssh_auth_classification": str(
+            classification.get("ssh_auth_classification") or ""
+        ),
+        "node_load_classifier": str(classification.get("node_load_classifier") or ""),
+        "registry_v2_status": classification.get("registry_v2_status"),
+        "controlled_recovery_package": str(
+            classification.get("controlled_recovery_package") or ""
+        ),
+        "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"
+        ),
+        "external_control_path_blocker": str(
+            classification.get("ssh_auth_classification") or ""
+        ),
+        "metadata_only": True,
+    }
+
+
 def _target_rollups(plans: list[dict[str, Any]]) -> list[dict[str, Any]]:
     return [
         {

apps/api/src/services/ai_agent_log_feedback_receipt_dry_run.py

@@ -167,6 +167,9 @@ def _sample_rows(runtime_sample: dict[str, Any]) -> list[dict[str, Any]]:
     for sample in samples:
         if not isinstance(sample, dict):
             continue
+        classification = sample.get("classification")
+        if not isinstance(classification, dict):
+            classification = {}
         rows.append(
             {
                 "sample_id": str(sample.get("sample_id") or ""),
@@ -179,6 +182,7 @@ def _sample_rows(runtime_sample: dict[str, Any]) -> list[dict[str, Any]]:
                 "redaction_state": str(sample.get("redaction_state") or ""),
                 "observed_events": [str(item) for item in sample.get("observed_events") or []],
                 "observed_fields": [str(item) for item in sample.get("observed_fields") or []],
+                "classification": dict(classification),
             }
         )
     return rows
@@ -202,6 +206,7 @@ def _candidate_receipts(samples: list[dict[str, Any]]) -> list[dict[str, Any]]:
                     "redaction_state": sample["redaction_state"],
                     "observed_event_count": len(sample["observed_events"]),
                     "observed_field_count": len(sample["observed_fields"]),
+                    "classification": sample["classification"],
                     "dry_run_status": "candidate_ready",
                     "write_enabled": False,
                     "raw_log_payload_persisted": False,