"""AI Agent LOG controlled writeback executor readback.

Turns the verified LOG controlled writeback plan into executor-ready batches
and AI Agent consumption context. This endpoint-level readback opens the
controlled-apply route for low / medium / high metadata writeback decisions,
while keeping the function itself side-effect free: it does not dispatch an
executor, write KM, index RAG, update PlayBook trust, call MCP tools, trigger
workflows, or persist raw log payloads.
"""

from __future__ import annotations

from typing import Any

from src.services.ai_agent_log_controlled_writeback_plan_readback import (
    load_latest_ai_agent_log_controlled_writeback_plan_readback,
)

_SCHEMA_VERSION = "ai_agent_log_controlled_writeback_executor_readback_v1"
_PLAN_READY_STATUS = "controlled_writeback_plan_ready"
_TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
_EXECUTOR_ROUTE = "ai_agent_metadata_writeback_executor"
_LOG_SOURCE_GROUPING_KEYS = (
    "project_id",
    "product",
    "site_or_route",
    "service",
    "package",
    "tool",
    "source_system",
    "runtime_component",
    "signal_lane",
    "evidence_boundary",
)


def load_latest_ai_agent_log_controlled_writeback_executor_readback() -> dict[str, Any]:
    """Return executor-ready LOG writeback batches for AI Agent consumption."""
    plan = load_latest_ai_agent_log_controlled_writeback_plan_readback()
    writeback_plans = _writeback_plans(plan)
    plan_ready = (
        plan.get("status") == _PLAN_READY_STATUS
        and plan.get("active_blockers") == []
        and (plan.get("rollups") or {}).get("controlled_writeback_plan_ready") is True
    )
    execution_batches = _execution_batches(writeback_plans)
    active_blockers = _active_blockers(
        plan_ready=plan_ready,
        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,
        "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
        "scope": "ai_agent_log_controlled_writeback_executor",
        "status": (
            "controlled_writeback_executor_ready"
            if not active_blockers
            else "blocked_waiting_controlled_writeback_executor_inputs"
        ),
        "readback": {
            "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-EXECUTOR",
            "workplan_title": "LOG feedback controlled writeback executor and AI Agent consumption readback",
            "source_schema_version": plan.get("schema_version"),
            "source_status": plan.get("status"),
            "safe_next_step": "dispatch_controlled_metadata_writeback_batches_then_post_apply_verify",
        },
        "executor_policy": {
            "executor_route": _EXECUTOR_ROUTE,
            "low_medium_high_controlled_apply_enabled": True,
            "owner_review_required_for_low_medium_high": False,
            "critical_break_glass_required": True,
            "target_selector_required": True,
            "source_of_truth_diff_required": True,
            "check_mode_required": True,
            "rollback_required": True,
            "post_apply_verifier_required": True,
        },
        "execution_batches": execution_batches,
        "agent_consumption_context": agent_context,
        "rollups": {
            "source_writeback_plan_count": len(writeback_plans),
            "execution_batch_count": len(execution_batches),
            "ready_execution_batch_count": sum(
                1
                for batch in execution_batches
                if batch["status"] == "ready_for_controlled_executor_dispatch"
            ),
            "target_count": len(_TARGETS),
            "target_selector_count": sum(
                batch["target_selector_count"] for batch in execution_batches
            ),
            "source_of_truth_diff_count": sum(
                batch["source_of_truth_diff_count"] for batch in execution_batches
            ),
            "check_mode_ready_count": sum(
                1 for batch in execution_batches if batch["check_mode"]["enabled"] is True
            ),
            "rollback_ready_count": sum(
                1 for batch in execution_batches if batch["rollback"]["required"] is True
            ),
            "post_apply_verifier_ready_count": sum(
                1
                for batch in execution_batches
                if batch["post_apply_verifier"]["required"] is True
            ),
            "log_source_tagging_contract_count": len(_log_source_tagging_contract()),
            "log_source_grouping_key_count": len(_LOG_SOURCE_GROUPING_KEYS),
            "metadata_only_log_source_tagging_ready": True,
            "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_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"]
            ),
            "current_blocker_harbor_recovery_receipt_input_count": sum(
                item["harbor_recovery_receipt_input_count"]
                for item in current_blocker_queue
            ),
            "current_blocker_harbor_recovery_receipt_output_contract_count": sum(
                item["harbor_recovery_receipt_output_contract_count"]
                for item in current_blocker_queue
            ),
            "current_blocker_registry_v2_ready_count": sum(
                1 for item in current_blocker_queue if item["registry_v2_ready"]
            ),
            "current_blocker_deploy_marker_readback_required_count": sum(
                1
                for item in current_blocker_queue
                if item["deploy_marker_readback_required"]
            ),
            "current_blocker_cd_failure_after_registry_ready_count": sum(
                1
                for item in current_blocker_queue
                if item["cd_failed_after_registry_ready"]
            ),
            "runtime_dispatch_performed": False,
        },
        "active_blockers": active_blockers,
        "operation_boundaries": {
            "executor_readback_only": True,
            "controlled_apply_enabled_by_policy": True,
            "executor_dispatch_performed": False,
            "km_write_performed": False,
            "rag_index_write_performed": False,
            "playbook_trust_write_performed": False,
            "mcp_tool_call_performed": False,
            "agent_runtime_action_performed": False,
            "workflow_trigger_performed": False,
            "raw_log_payload_persisted": False,
            "secret_value_collection_allowed": False,
            "github_api_used": False,
        },
    }


def _writeback_plans(plan: dict[str, Any]) -> list[dict[str, Any]]:
    plans = plan.get("writeback_plans")
    if not isinstance(plans, list):
        return []
    return [item for item in plans if isinstance(item, dict)]


def _execution_batches(writeback_plans: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [
        _execution_batch(target, _plans_for_target(writeback_plans, target))
        for target in _TARGETS
    ]


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
    ]
    verifier_refs = [
        str((plan.get("post_apply_verifier") or {}).get("verifier_ref") or "")
        for plan in plans
    ]
    ready = _batch_ready(plans)
    return {
        "batch_id": f"log-feedback-controlled-writeback::{target}",
        "target": target,
        "target_surface": _target_surface(target),
        "risk_tier": risk_tier,
        "executor_route": _EXECUTOR_ROUTE,
        "status": (
            "ready_for_controlled_executor_dispatch"
            if ready
            else "blocked_waiting_batch_controls"
        ),
        "apply_mode": "controlled_apply",
        "dispatch_enabled_by_policy": True,
        "plan_count": len(plans),
        "plan_ids": [str(plan.get("plan_id") or "") for plan in plans],
        "receipt_ids": [str(plan.get("receipt_id") or "") for plan in plans],
        "target_selector_count": len(target_selectors),
        "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,
            "checks": [
                "resolve_target_selectors",
                "compare_source_of_truth_diffs",
                "verify_metadata_only_redaction",
                "verify_rollback_refs",
                "verify_post_apply_verifier_refs",
            ],
        },
        "rollback": {
            "required": True,
            "rollback_refs": rollback_refs,
            "strategy": "mark_receipts_superseded_and_remove_target_bindings",
        },
        "post_apply_verifier": {
            "required": True,
            "verifier_refs": verifier_refs,
            "canonical_readback": (
                "/api/v1/agents/agent-log-controlled-writeback-executor-readback"
            ),
        },
        "runtime_dispatch_performed": False,
    }


def _batch_ready(plans: list[dict[str, Any]]) -> bool:
    if not plans:
        return False
    for plan in plans:
        if plan.get("status") != "controlled_apply_ready":
            return False
        if plan.get("write_enabled_by_plan") is not False:
            return False
        if not plan.get("target_selector"):
            return False
        if not plan.get("source_of_truth_diff"):
            return False
        if (plan.get("check_mode") or {}).get("enabled") is not True:
            return False
        if (plan.get("rollback") or {}).get("required") is not True:
            return False
        if (plan.get("post_apply_verifier") or {}).get("required") is not True:
            return False
    return True


def _agent_consumption_context(execution_batches: list[dict[str, Any]]) -> dict[str, Any]:
    ready_batches = [
        batch
        for batch in execution_batches
        if batch["status"] == "ready_for_controlled_executor_dispatch"
    ]
    return {
        "context_id": "ai-agent-log-controlled-writeback-consumption-v1",
        "consumable_by": [
            "ai_agent_autonomous_runtime_control",
            "awooop_work_items",
            "alert_triage_loop",
            "km_rag_playbook_learning_loop",
            "mcp_audit_context_loop",
        ],
        "log_source_tagging_contract": _log_source_tagging_contract(),
        "log_source_tagging_contract_count": len(_log_source_tagging_contract()),
        "log_source_grouping_keys": list(_LOG_SOURCE_GROUPING_KEYS),
        "log_source_grouping_key_count": len(_LOG_SOURCE_GROUPING_KEYS),
        "evidence_chain": [
            "/api/v1/agents/agent-log-intelligence-integration-readback",
            "/api/v1/agents/agent-log-feedback-receipt-dry-run",
            "/api/v1/agents/agent-log-post-write-verifier-dry-run",
            "/api/v1/agents/agent-log-controlled-writeback-plan-readback",
        ],
        "next_action_queue": [
            {
                "batch_id": batch["batch_id"],
                "target": batch["target"],
                "executor_route": batch["executor_route"],
                "apply_mode": batch["apply_mode"],
                "plan_count": batch["plan_count"],
                "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",
            "playbook",
            "mcp",
            "verifier",
            "ai_agent",
        ],
        "raw_payload_required": False,
    }


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 "")
    node_load_classifier = str(recovery.get("node_load_classifier") or "")
    control_path_pressure_blocker = (
        "node_load_high" if node_load_classifier == "high_load" else ""
    )
    log_source_tags = _current_blocker_log_source_tags(recovery)
    safe_next_action = _current_blocker_safe_next_action(
        recovery,
        control_path_pressure_blocker=control_path_pressure_blocker,
    )
    queue_status = "controlled_recovery_packaged_waiting_control_path_readback"
    if blocker_id == "controlled_cd_lane_guardrails_blocked":
        queue_status = (
            "controlled_recovery_packaged_waiting_controlled_cd_lane_guardrails_readback"
        )
    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": queue_status,
        "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"),
        "registry_v2_ready": bool(recovery.get("registry_v2_ready") is True),
        "registry_v2_status_classifier": str(
            recovery.get("registry_v2_status_classifier") or ""
        ),
        "deployment_closure_state": str(
            recovery.get("deployment_closure_state") or ""
        ),
        "deploy_marker_readback_required": bool(
            recovery.get("deploy_marker_readback_required") is True
        ),
        "current_cd_run_id": str(recovery.get("current_cd_run_id") or ""),
        "current_cd_run_status": str(recovery.get("current_cd_run_status") or ""),
        "current_cd_commit_sha": str(recovery.get("current_cd_commit_sha") or ""),
        "cd_failed_after_registry_ready": bool(
            recovery.get("cd_failed_after_registry_ready") is True
        ),
        "harbor_110_repair_run_id": str(
            recovery.get("harbor_110_repair_run_id") or ""
        ),
        "harbor_110_repair_run_status": str(
            recovery.get("harbor_110_repair_run_status") or ""
        ),
        "harbor_110_repair_failed_after_registry_ready": bool(
            recovery.get("harbor_110_repair_failed_after_registry_ready") is True
        ),
        "harbor_110_repair_failure_classifier": str(
            recovery.get("harbor_110_repair_failure_classifier") or ""
        ),
        "log_source_tags": log_source_tags,
        "log_source_tag_count": len(log_source_tags),
        "log_grouping_keys": list(_LOG_SOURCE_GROUPING_KEYS),
        "controlled_recovery_package": str(
            recovery.get("controlled_recovery_package") or ""
        ),
        "post_apply_verifier": str(recovery.get("post_apply_verifier") or ""),
        "harbor_recovery_receipt_endpoint": (
            "/api/v1/agents/harbor-registry-controlled-recovery-receipt"
        ),
        "harbor_recovery_receipt_inputs": _harbor_recovery_receipt_inputs(),
        "harbor_recovery_receipt_input_count": len(
            _harbor_recovery_receipt_inputs()
        ),
        "harbor_recovery_receipt_output_contract": (
            _harbor_recovery_receipt_output_contract()
        ),
        "harbor_recovery_receipt_output_contract_count": len(
            _harbor_recovery_receipt_output_contract()
        ),
        "queue_readback_normalizer_contract": _queue_readback_normalizer_contract(),
        "queue_readback_normalizer_contract_count": len(
            _queue_readback_normalizer_contract()
        ),
        "controlled_local_console_execution_plan": _list_of_dicts(
            recovery.get("controlled_local_console_execution_plan")
        ),
        "post_recovery_readback_commands": _list_of_strings(
            recovery.get("post_recovery_readback_commands")
        ),
        "forbidden_runtime_actions": _list_of_strings(
            recovery.get("forbidden_runtime_actions")
        ),
        "safe_next_step": str(recovery.get("safe_next_step") or ""),
        "safe_next_action_id": safe_next_action["action_id"],
        "safe_next_action_stage": safe_next_action["stage"],
        "safe_next_action": safe_next_action["action"],
        "safe_next_action_command": safe_next_action["command"],
        "safe_next_action_post_verifier": safe_next_action["post_verifier"],
        "safe_next_action_requires_local_console": safe_next_action[
            "requires_local_console"
        ],
        "safe_next_action_blocker_fields": safe_next_action["blocker_fields"],
        "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_path_pressure_blocker": control_path_pressure_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 _current_blocker_safe_next_action(
    recovery: dict[str, Any],
    *,
    control_path_pressure_blocker: str,
) -> dict[str, Any]:
    blocker_id = str(recovery.get("blocker_id") or "")
    external_blocker = str(recovery.get("external_control_path_blocker") or "")
    deploy_marker_required = recovery.get("deploy_marker_readback_required") is True
    cd_failed_after_registry_ready = (
        recovery.get("cd_failed_after_registry_ready") is True
    )
    harbor_repair_failed_after_registry_ready = (
        recovery.get("harbor_110_repair_failed_after_registry_ready") is True
    )

    if blocker_id == "harbor_110_remote_ssh_publickey_auth_stalled":
        return {
            "action_id": (
                "run_110_local_ssh_session_control_path_recovery_then_verify_cd_and_deploy_marker_readback"
            ),
            "stage": "local_console_control_path_receipt_required",
            "action": (
                "Run 110 local-console SSH/session check, then verify registry, "
                "queue, CD, and deploy-marker closure."
            ),
            "command": "recover-110-control-path-and-harbor-local.sh --check",
            "post_verifier": (
                "read-public-gitea-actions-queue.py --json && "
                "check-awoooi-110-controlled-cd-lane-readiness.sh"
            ),
            "requires_local_console": True,
            "blocker_fields": [
                "latest_visible_harbor_110_repair_remote_ssh_publickey_auth_stalled",
                external_blocker,
            ],
        }

    if blocker_id == "controlled_cd_lane_guardrails_blocked":
        return {
            "action_id": (
                "restore_controlled_cd_lane_guardrails_then_rerun_readiness_receipt"
            ),
            "stage": "controlled_cd_lane_guardrails_receipt_required",
            "action": (
                "Verify the 110 controlled CD lane guardrails: config, ELF "
                "binary, registration metadata, systemd limits, fail-closed "
                "legacy runners, host pressure, and receipt writeback."
            ),
            "command": "check-awoooi-110-controlled-cd-lane-readiness.sh",
            "post_verifier": (
                "harbor-registry-controlled-recovery-receipt validator && "
                "priority work-order readback"
            ),
            "requires_local_console": False,
            "blocker_fields": [
                "controlled_cd_lane_config_missing",
                "controlled_cd_lane_binary_not_ready",
                "controlled_cd_lane_registration_missing",
                "controlled_cd_lane_service_not_active",
                "controlled_cd_lane_pressure_guard",
            ],
        }

    if control_path_pressure_blocker:
        return {
            "action_id": "wait_host_pressure_gate_then_rerun_110_control_path_readback",
            "stage": "host_pressure_gate_wait",
            "action": (
                "Keep host pressure protection fail-closed, then rerun 110 "
                "control-path and queue readbacks."
            ),
            "command": "awoooi-wait-host-web-build-pressure.sh",
            "post_verifier": "read-public-gitea-actions-queue.py --json",
            "requires_local_console": False,
            "blocker_fields": [control_path_pressure_blocker],
        }

    if (
        deploy_marker_required
        or cd_failed_after_registry_ready
        or harbor_repair_failed_after_registry_ready
    ):
        return {
            "action_id": "verify_registry_ready_then_close_cd_deploy_marker_readback",
            "stage": "deploy_marker_closure_after_registry_ready",
            "action": (
                "Keep registry 200/401 readiness separate from CD/deploy-marker "
                "closure, then verify production priority readback."
            ),
            "command": "read-public-gitea-actions-queue.py --json",
            "post_verifier": "awoooi production deploy marker readback",
            "requires_local_console": False,
            "blocker_fields": [
                "deploy_marker_readback_required_after_registry_ready",
                "current_cd_failure_after_registry_ready",
                "harbor_110_repair_failure_after_registry_ready",
            ],
        }

    return {
        "action_id": "continue_ai_loop_current_blocker_readback",
        "stage": "ai_loop_current_blocker_readback",
        "action": "Continue metadata-only current blocker readback and writeback.",
        "command": "read-public-gitea-actions-queue.py --json",
        "post_verifier": "priority work-order readback",
        "requires_local_console": False,
        "blocker_fields": [external_blocker] if external_blocker else [],
    }


def _log_source_tagging_contract() -> list[dict[str, Any]]:
    metadata_boundary = {
        "metadata_only": True,
        "raw_output_allowed": False,
        "secret_value_allowed": False,
        "writeback_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
    }
    return [
        {
            "tag_key": "project_id",
            "source_fields": ["target_selector.project_id", "receipt.project_id"],
            "purpose": "group all LOG evidence by project before KM/RAG/PlayBook writeback",
            **metadata_boundary,
        },
        {
            "tag_key": "product",
            "source_fields": ["target_selector.product", "receipt.product"],
            "purpose": "group LOG evidence by product surface across AWOOOI/AwoooP/IwoooS",
            **metadata_boundary,
        },
        {
            "tag_key": "site_or_route",
            "source_fields": [
                "target_selector.service",
                "post_recovery_readback_commands",
                "harbor_recovery_receipt_inputs.source",
            ],
            "purpose": "group website, API, registry, and route verifier signals",
            **metadata_boundary,
        },
        {
            "tag_key": "service",
            "source_fields": ["target_selector.service", "receipt.service"],
            "purpose": "group service runtime logs and queue evidence by service",
            **metadata_boundary,
        },
        {
            "tag_key": "package",
            "source_fields": ["target_selector.package", "receipt.package"],
            "purpose": "group code package or repo path evidence without storing raw paths",
            **metadata_boundary,
        },
        {
            "tag_key": "tool",
            "source_fields": ["target_selector.tool", "receipt.tool"],
            "purpose": "group LOG evidence by tool, script, verifier, or classifier",
            **metadata_boundary,
        },
        {
            "tag_key": "source_system",
            "source_fields": ["target_selector.source_system", "receipt.source_system"],
            "purpose": "group raw signal origin such as k8s logs, Gitea queue, registry verifier, or runner readback",
            **metadata_boundary,
        },
        {
            "tag_key": "runtime_component",
            "source_fields": [
                "harbor_recovery_receipt_inputs.input_id",
                "harbor_recovery_receipt_output_contract.output_id",
                "queue_readback_normalizer_contract.field_id",
            ],
            "purpose": "group runtime components such as runner, registry, SSH control path, CD, verifier, and deploy marker",
            **metadata_boundary,
        },
        {
            "tag_key": "signal_lane",
            "source_fields": [
                "classification.current_blocker",
                "classification.ssh_auth_classification",
                "classification.risk_tier",
            ],
            "purpose": "group evidence into AI repair, rollback, verifier, security, host resource, and controlled apply lanes",
            **metadata_boundary,
        },
        {
            "tag_key": "evidence_boundary",
            "source_fields": [
                "redaction_state",
                "metadata_only",
                "raw_payload_required",
                "secret_value_allowed",
            ],
            "purpose": "keep learning evidence metadata-only and prevent raw LOG or secret ingestion",
            **metadata_boundary,
        },
    ]


def _current_blocker_log_source_tags(recovery: dict[str, Any]) -> list[dict[str, Any]]:
    tag_values = {
        "project_id": str(recovery.get("project_id") or ""),
        "product": str(recovery.get("product") or ""),
        "site_or_route": str(recovery.get("service") or ""),
        "service": str(recovery.get("service") or ""),
        "package": str(recovery.get("package") or ""),
        "tool": str(recovery.get("tool") or ""),
        "source_system": str(recovery.get("source_system") or ""),
        "runtime_component": "harbor_registry_110_control_path",
        "signal_lane": str(recovery.get("blocker_id") or ""),
        "evidence_boundary": "metadata_only_no_raw_payload",
    }
    return [
        {
            "tag_key": key,
            "tag_value": tag_values[key],
            "metadata_only": True,
            "raw_output_allowed": False,
            "secret_value_allowed": False,
            "writeback_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        }
        for key in _LOG_SOURCE_GROUPING_KEYS
    ]


def _harbor_recovery_receipt_inputs() -> list[dict[str, Any]]:
    metadata_boundary = {
        "metadata_only": True,
        "raw_output_allowed": False,
        "secret_value_allowed": False,
    }
    return [
        {
            "input_id": "gitea_actions_queue_readback",
            "source": "read-public-gitea-actions-queue.py --json",
            "required_when": "before_and_after_110_controlled_recovery",
            "purpose": (
                "classify cd run status, harbor repair waiting state, "
                "remote control channel availability, no-matching runners, "
                "and stale job payloads"
            ),
            **metadata_boundary,
        },
        {
            "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 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",
            "source": "repair-110-ssh-publickey-auth-local.sh --check/apply",
            "required_when": "after_remote_control_channel_diagnosis_phase",
            "purpose": (
                "prove non-secret 110 account, authorized_keys metadata, sshd "
                "effective config, and SSH reload result before Harbor retry"
            ),
            **metadata_boundary,
        },
        {
            "input_id": "local_console_recovery_output",
            "source": "recover-110-control-path-and-harbor-local.sh --check/apply",
            "required_when": "after_remote_control_channel_diagnosis_phase",
            "purpose": "prove 110 ssh control path, Harbor, and local console repair results",
            **metadata_boundary,
        },
        {
            "input_id": "watchdog_check_output",
            "source": "harbor-watchdog.sh --check",
            "required_when": "before_harbor_repair_once_phase",
            "purpose": "prove check-mode harbor local registry /v2/ state",
            **metadata_boundary,
        },
        {
            "input_id": "watchdog_repair_output",
            "source": "harbor-watchdog.sh --repair-once",
            "required_when": "only_if_check_mode_proves_v2_unhealthy",
            "purpose": "prove bounded repair-once result without daemon restart",
            **metadata_boundary,
        },
        {
            "input_id": "controlled_cd_lane_readiness_output",
            "source": "check-awoooi-110-controlled-cd-lane-readiness.sh",
            "required_when": "before_retrying_harbor_110_local_repair_queue",
            "purpose": (
                "classify awoooi-host lane guardrail blockers such as service, "
                "registration, root restore-source, pressure, or label/config drift"
            ),
            **metadata_boundary,
        },
        {
            "input_id": "non110_runner_readiness_output",
            "source": "check-awoooi-non110-runner-readiness.sh",
            "required_when": "before_retrying_current_cd_or_harbor_repair_queue",
            "purpose": (
                "classify non110 controlled runner config, binary, registration, "
                "service, active service, autostart path, rollback, pressure, "
                "and safe-next-step blockers without reading .runner content"
            ),
            **metadata_boundary,
        },
        {
            "input_id": "public_registry_v2_http_status",
            "source": "public registry /v2/ verifier",
            "required_when": "post_apply_verifier",
            "purpose": "prove external registry route is 200/401 before CD retry",
            **metadata_boundary,
        },
        {
            "input_id": "internal_registry_v2_http_status",
            "source": "internal 110 registry /v2/ verifier",
            "required_when": "post_apply_verifier",
            "purpose": "prove local registry route is 200/401 before CD retry",
            **metadata_boundary,
        },
        {
            "input_id": "deploy_marker_readback",
            "source": "awoooi production deploy marker readback verifier",
            "required_when": "after_registry_v2_ready_and_cd_retry",
            "purpose": (
                "prove latest CD success, production image tag sha, and "
                "priority readback HTTP status before closing the P0 blocker"
            ),
            "expected_schema": "awoooi_production_deploy_readback_blocker_v1",
            **metadata_boundary,
        },
    ]


def _harbor_recovery_receipt_output_contract() -> list[dict[str, Any]]:
    metadata_boundary = {
        "metadata_only": True,
        "raw_output_allowed": False,
        "secret_value_allowed": False,
        "writeback_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
    }
    return [
        {
            "output_id": "control_path_readiness",
            "source": "harbor-registry-controlled-recovery-receipt.readback.control_path_readiness",
            "required_when": "after_receipt_validation_before_km_rag_mcp_playbook_writeback",
            "purpose": (
                "normalize 110 SSH, node load, awoooi-host queue, registry, "
                "and safe-next-action signals into AI learning evidence"
            ),
            **metadata_boundary,
        },
        {
            "output_id": "ssh_local_repair",
            "source": "harbor-registry-controlled-recovery-receipt.readback.ssh_local_repair",
            "required_when": "after_110_local_ssh_publickey_auth_repair_receipt",
            "purpose": (
                "write back non-secret 110 target user, authorized_keys, sshd "
                "publickey, AuthorizedKeysFile, and reload readiness signals"
            ),
            **metadata_boundary,
        },
        {
            "output_id": "controlled_cd_lane_readiness",
            "source": "harbor-registry-controlled-recovery-receipt.readback.controlled_cd_lane_readiness",
            "required_when": "before_retrying_harbor_110_local_repair_queue",
            "purpose": (
                "write back controlled lane registration, service, fail-closed, "
                "blocker count, and safe-next-step readiness"
            ),
            **metadata_boundary,
        },
        {
            "output_id": "non110_runner_readiness",
            "source": "harbor-registry-controlled-recovery-receipt.readback.non110_runner_readiness",
            "required_when": "before_retrying_current_cd_or_harbor_repair_queue",
            "purpose": (
                "write back non110 runner registration, service, active service, "
                "autostart path, blocker count, and safe-next-step readiness"
            ),
            **metadata_boundary,
        },
        {
            "output_id": "gitea_actions_queue",
            "source": "harbor-registry-controlled-recovery-receipt.readback.gitea_actions_queue",
            "required_when": "after_public_gitea_queue_readback",
            "purpose": (
                "write back normalized CD job, Harbor repair job, and "
                "awoooi-host no-matching runner classifiers"
            ),
            **metadata_boundary,
        },
        {
            "output_id": "local_console_phase_readback",
            "source": "harbor-registry-controlled-recovery-receipt.local_console_phase_readback",
            "required_when": "after_receipt_validation",
            "purpose": (
                "write back ordered diagnose, preflight, repair, verifier phase "
                "status without raw console output"
            ),
            **metadata_boundary,
        },
        {
            "output_id": "post_apply_verifier",
            "source": "harbor-registry-controlled-recovery-receipt.readback.post_apply_verifier",
            "required_when": "after_registry_v2_verifier",
            "purpose": "write back public and internal registry /v2/ readiness",
            **metadata_boundary,
        },
        {
            "output_id": "deploy_marker",
            "source": "harbor-registry-controlled-recovery-receipt.readback.deploy_marker",
            "required_when": "after_registry_v2_ready_and_cd_retry",
            "purpose": (
                "write back deploy marker, production image match, and latest "
                "CD status before closing the P0 blocker"
            ),
            **metadata_boundary,
        },
    ]


def _queue_readback_normalizer_contract() -> list[dict[str, Any]]:
    return [
        {
            "field_id": "cd_run_jobs_payload_classifier",
            "purpose": "classify stale or cross-run CD jobs API payloads",
            "writes_blockers": [
                "gitea_queue_cd_jobs_head_sha_mismatch",
                "gitea_queue_cd_jobs_run_id_mismatch",
                "gitea_queue_cd_jobs_stale_or_mismatched",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "ai_agent"],
        },
        {
            "field_id": "harbor_110_repair_jobs_payload_classifier",
            "purpose": "classify Harbor repair jobs API payload workflow drift",
            "writes_blockers": [
                "gitea_queue_harbor_110_repair_jobs_cross_workflow_mismatch",
                "gitea_queue_harbor_110_repair_jobs_stale_or_mismatched",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "ai_agent"],
        },
        {
            "field_id": "latest_visible_harbor_110_repair_no_matching_runner_label",
            "purpose": "classify whether the awoooi-host controlled repair lane is unavailable",
            "writes_blockers": [
                "gitea_queue_harbor_110_repair_no_matching_runner",
            ],
            "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": "latest_visible_harbor_110_repair_remote_ssh_publickey_auth_stalled",
            "purpose": "classify whether 110 accepts TCP reachability but stalls during SSH publickey userauth",
            "writes_blockers": [
                "gitea_queue_harbor_110_remote_ssh_publickey_auth_stalled",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "latest_visible_harbor_110_repair_remote_ssh_publickey_offer_timeout",
            "purpose": "classify whether 110 times out after the client offers a public key but before key acceptance",
            "writes_blockers": [
                "gitea_queue_harbor_110_remote_ssh_publickey_offer_timeout",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "latest_visible_harbor_110_repair_remote_ssh_server_accepts_key_then_session_timeout",
            "purpose": "classify whether 110 accepts the SSH key and then times out during session, PAM, account, or shell setup",
            "writes_blockers": [
                "gitea_queue_harbor_110_remote_ssh_server_accepts_key_then_session_timeout",
            ],
            "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",
            "writes_blockers": [
                "gitea_queue_current_cd_no_matching_runner",
                "gitea_queue_current_cd_waiting_for_runner_or_queue",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier"],
        },
        {
            "field_id": "current_cd_deploy_marker_closure",
            "purpose": (
                "classify latest visible CD status, commit sha, and deploy-marker "
                "closure after registry /v2/ has returned a ready 200/401 state"
            ),
            "writes_blockers": [
                "deploy_marker_readback_required_after_registry_ready",
                "current_cd_failure_after_registry_ready",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "registry_v2_ready_state",
            "purpose": (
                "distinguish registry /v2/ ready 200/401 evidence from stale "
                "Harbor 502 repair blockers"
            ),
            "writes_blockers": [
                "public_registry_v2_verifier_not_green",
                "internal_registry_v2_verifier_not_green",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "harbor_110_repair_failure_after_registry_ready",
            "purpose": (
                "classify Harbor 110 repair failure as SSH/control-path closure "
                "work when registry /v2/ is already ready"
            ),
            "writes_blockers": [
                "harbor_110_remote_ssh_publickey_auth_stalled",
                "harbor_110_repair_failure_after_registry_ready",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "harbor_110_repair_visible_running_jobs_api_stale",
            "purpose": (
                "classify when the visible Harbor repair run is still Running "
                "but the jobs API payload belongs to another workflow or stale run"
            ),
            "writes_blockers": [
                "gitea_queue_harbor_110_repair_visible_running_jobs_api_stale",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "current_cd_waiting_behind_harbor_110_repair_running",
            "purpose": (
                "classify when current CD is waiting behind an in-flight or "
                "stale Harbor 110 repair readback"
            ),
            "writes_blockers": [
                "gitea_queue_current_cd_waiting_behind_harbor_110_repair_running",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "safe_next_action_id",
            "purpose": (
                "publish one machine-readable next action from queue classifiers "
                "so AI Loop does not treat stale jobs API payloads, runner waits, "
                "or 110 SSH/session blockers as generic manual triage"
            ),
            "writes_blockers": [
                "gitea_queue_safe_next_action_missing",
                "gitea_queue_safe_next_action_requires_local_console",
                "gitea_queue_safe_next_action_stale_payload_quarantined",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
        },
        {
            "field_id": "controlled_profile_no_matching_runner_labels",
            "purpose": (
                "classify all controlled workflow labels with no matching online runner"
            ),
            "writes_blockers": [
                "gitea_queue_controlled_profile_no_matching_runner_labels",
            ],
            "learning_targets": ["km", "rag", "playbook", "mcp", "verifier"],
        },
    ]


def _list_of_dicts(value: Any) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        return []
    return [dict(item) for item in value if isinstance(item, dict)]


def _list_of_strings(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    return [str(item) for item in value if str(item)]


def _active_blockers(
    *,
    plan_ready: bool,
    writeback_plans: list[dict[str, Any]],
    execution_batches: list[dict[str, Any]],
) -> list[str]:
    blockers = []
    if not plan_ready:
        blockers.append("controlled_writeback_plan_not_ready")
    if not writeback_plans:
        blockers.append("source_writeback_plans_missing")
    for target in _TARGETS:
        batch = next(
            (item for item in execution_batches if item.get("target") == target),
            None,
        )
        if not batch:
            blockers.append(f"{target}_execution_batch_missing")
            continue
        if batch["status"] != "ready_for_controlled_executor_dispatch":
            blockers.append(f"{target}_execution_batch_not_ready")
        if batch["dispatch_enabled_by_policy"] is not True:
            blockers.append(f"{target}_dispatch_policy_not_enabled")
        if batch["runtime_dispatch_performed"] is not False:
            blockers.append(f"{target}_runtime_dispatch_already_performed")
    return _unique(blockers)


def _plans_for_target(plans: list[dict[str, Any]], target: str) -> list[dict[str, Any]]:
    return [plan for plan in plans if plan.get("target") == target]


def _target_surface(target: str) -> str:
    return {
        "km": "knowledge_memory",
        "rag": "rag_chunk_index",
        "playbook": "playbook_trust_learning",
        "mcp": "mcp_audit_context",
        "verifier": "post_apply_verifier_feedback",
        "ai_agent": "agent_decision_context",
    }.get(target, "unknown")


def _unique(values: list[str]) -> list[str]:
    seen = set()
    result = []
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        result.append(value)
    return result
