"""AwoooP Ansible controlled executor.

This service claims pending ``ansible_candidate_matched`` AOL rows, runs
``ansible-playbook --check --diff``, and for allowlisted low/medium/high risk
catalog rows performs controlled apply after the dry-run passes.
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import shutil
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import structlog
from sqlalchemy import text

from src.core.config import settings
from src.db.base import get_db_context
from src.services.awooop_ansible_audit_service import get_ansible_catalog_item

logger = structlog.get_logger(__name__)

_SAFE_HOST_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
_PLAYBOOK_PREFIX = Path("infra/ansible/playbooks")
_STDOUT_LIMIT = 20_000
_STDERR_LIMIT = 12_000
FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519")
REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts")


@dataclass(frozen=True)
class AnsibleCheckModeClaim:
    op_id: str
    source_candidate_op_id: str
    incident_id: str
    catalog_id: str
    playbook_path: str
    apply_playbook_path: str
    inventory_hosts: tuple[str, ...]
    risk_level: str
    input_payload: dict[str, Any]


@dataclass(frozen=True)
class AnsibleCommandSpec:
    command: list[str]
    cwd: Path
    env: dict[str, str]
    playbook_abs_path: Path
    inventory_abs_path: Path


@dataclass(frozen=True)
class AnsibleRunResult:
    returncode: int
    stdout: str
    stderr: str
    duration_ms: int
    timed_out: bool = False


def _tail(text_value: str, limit: int) -> str:
    if len(text_value) <= limit:
        return text_value
    return text_value[-limit:]


def _json_loads(value: Any) -> dict[str, Any]:
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
        except json.JSONDecodeError:
            return {}
        return parsed if isinstance(parsed, dict) else {}
    return {}


def _incident_id_from_payload(payload: dict[str, Any]) -> str:
    return str(payload.get("incident_id") or "").strip()


def _automation_operation_log_incident_id(value: str) -> int | None:
    # ADR-090 keeps external INC-* ids in input JSON; the DB column is BIGINT.
    normalized = str(value or "").strip()
    return int(normalized) if normalized.isdigit() else None


def _check_mode_ssh_key_path() -> Path:
    return Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH)


def _check_mode_known_hosts_path() -> Path:
    return Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH)


def _uses_repair_forced_command_transport(key_path: Path | None = None) -> bool:
    return (key_path or _check_mode_ssh_key_path()) == REPAIR_FORCED_COMMAND_KEY_PATH


def detect_ansible_transport_blockers(*values: Any) -> list[str]:
    combined = " ".join(str(value or "") for value in values)
    blockers: list[str] = []
    if "REPAIR_DENIED:invalid_command" in combined:
        blockers.append(FORCED_COMMAND_BLOCKER)
    return blockers


def _playbook_roots(module_path: Path | None = None) -> list[Path]:
    resolved_module_path = (module_path or Path(__file__)).resolve()
    return [
        Path("/app/infra/ansible"),
        Path.cwd() / "infra" / "ansible",
        *(parent / "infra" / "ansible" for parent in resolved_module_path.parents),
    ]


def _runtime_blockers(
    *,
    playbook_root: Path | None = None,
    check_mode_ssh_key_path: Path | None = None,
    check_mode_known_hosts_path: Path | None = None,
) -> list[str]:
    root = playbook_root or next((path for path in _playbook_roots() if path.exists()), None)
    ssh_key_path = check_mode_ssh_key_path or _check_mode_ssh_key_path()
    known_hosts_path = check_mode_known_hosts_path or _check_mode_known_hosts_path()
    blockers: list[str] = []
    if shutil.which("ansible-playbook") is None:
        blockers.append("ansible_playbook_binary_missing")
    if root is None:
        blockers.append("ansible_playbook_catalog_missing")
    elif not (root / "inventory" / "hosts.yml").exists():
        blockers.append("ansible_inventory_missing")
    if not ssh_key_path.is_file() or not os.access(ssh_key_path, os.R_OK):
        blockers.append("ansible_check_mode_ssh_key_missing")
    if not known_hosts_path.is_file() or not os.access(known_hosts_path, os.R_OK):
        blockers.append("ansible_check_mode_known_hosts_missing")
    return blockers


def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]:
    candidates = input_payload.get("executor_candidates")
    if not isinstance(candidates, list) or not candidates:
        raise ValueError("missing_executor_candidates")

    for candidate in candidates:
        if not isinstance(candidate, dict):
            continue
        catalog_id = str(candidate.get("catalog_id") or "")
        catalog_item = get_ansible_catalog_item(catalog_id)
        if not catalog_item:
            continue
        if catalog_item.get("supports_check_mode") is not True:
            continue
        catalog_playbook_path = str(catalog_item.get("playbook_path") or "")
        candidate_playbook_path = str(candidate.get("playbook_path") or catalog_playbook_path)
        check_mode_playbook_path = str(
            candidate.get("check_mode_playbook_path")
            or catalog_item.get("check_mode_playbook_path")
            or catalog_playbook_path
        )
        if candidate_playbook_path not in {catalog_playbook_path, check_mode_playbook_path}:
            continue
        inventory_hosts = candidate.get("inventory_hosts") or catalog_item.get("inventory_hosts") or []
        if (
            isinstance(inventory_hosts, list)
            and inventory_hosts
            and all(isinstance(host, str) and _SAFE_HOST_RE.fullmatch(host) for host in inventory_hosts)
        ):
            return {
                "catalog_id": catalog_id,
                "playbook_path": check_mode_playbook_path,
                "catalog_playbook_path": catalog_playbook_path,
                "source_candidate_playbook_path": candidate_playbook_path,
                "check_mode_playbook_path": check_mode_playbook_path,
                "inventory_hosts": tuple(inventory_hosts),
                "risk_level": str(candidate.get("risk_level") or catalog_item.get("risk_level") or ""),
                "auto_apply_enabled": bool(catalog_item.get("auto_apply_enabled") is True),
                "approval_required": bool(catalog_item.get("approval_required") is True),
                "break_glass_required": bool(catalog_item.get("break_glass_required") is True),
            }
    raise ValueError("no_safe_check_mode_candidate")


def _allowed_controlled_apply_risks() -> set[str]:
    raw = str(settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS or "")
    return {item.strip().lower() for item in raw.split(",") if item.strip()}


def _controlled_apply_allowed(candidate: dict[str, Any]) -> tuple[bool, str | None]:
    risk = str(candidate.get("risk_level") or "").strip().lower()
    if settings.ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY is not True:
        return False, "controlled_apply_disabled_by_config"
    if candidate.get("break_glass_required") is True:
        return False, "break_glass_required"
    if candidate.get("auto_apply_enabled") is not True:
        return False, "catalog_auto_apply_disabled"
    if candidate.get("approval_required") is True:
        return False, "catalog_still_requires_approval"
    if risk not in _allowed_controlled_apply_risks():
        return False, f"risk_level_not_allowed:{risk or 'unknown'}"
    return True, None


def build_ansible_check_mode_claim_input(
    *,
    source_candidate_op_id: str,
    candidate_input: dict[str, Any],
) -> dict[str, Any]:
    safe = _safe_candidate(candidate_input)
    incident_id = _incident_id_from_payload(candidate_input)
    controlled_apply_allowed, controlled_apply_blocker = _controlled_apply_allowed(safe)
    return {
        "incident_id": incident_id,
        "executor": "ansible",
        "execution_backend": "ansible",
        "execution_mode": "check_mode",
        "transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "check_mode": True,
        "diff": True,
        "apply_enabled": controlled_apply_allowed,
        "approval_required_before_apply": not controlled_apply_allowed,
        "controlled_apply_allowed": controlled_apply_allowed,
        "controlled_apply_blocker": controlled_apply_blocker,
        "source_candidate_op_id": source_candidate_op_id,
        "catalog_id": safe["catalog_id"],
        "playbook_path": safe["playbook_path"],
        "apply_playbook_path": safe["catalog_playbook_path"],
        "catalog_playbook_path": safe["catalog_playbook_path"],
        "source_candidate_playbook_path": safe["source_candidate_playbook_path"],
        "check_mode_playbook_path": safe["check_mode_playbook_path"],
        "inventory_hosts": list(safe["inventory_hosts"]),
        "risk_level": safe["risk_level"],
    }


def _resolve_playbook_path(playbook_root: Path, playbook_path: str) -> Path:
    relative = Path(playbook_path)
    if relative.is_absolute() or not str(relative).startswith(str(_PLAYBOOK_PREFIX)):
        raise ValueError("unsafe_playbook_path")
    repo_root = playbook_root.parent.parent
    resolved = (repo_root / relative).resolve()
    allowed_root = (repo_root / _PLAYBOOK_PREFIX).resolve()
    if allowed_root not in resolved.parents:
        raise ValueError("playbook_outside_catalog")
    if resolved.suffix not in {".yml", ".yaml"} or not resolved.exists():
        raise ValueError("playbook_not_found")
    return resolved


def _ansible_command_env(playbook_root: Path) -> dict[str, str]:
    roles_path = str((playbook_root / "roles").resolve())
    existing_roles_path = os.environ.get("ANSIBLE_ROLES_PATH")
    if existing_roles_path:
        roles_path = os.pathsep.join([roles_path, existing_roles_path])
    return {
        **os.environ,
        "ANSIBLE_HOST_KEY_CHECKING": "true",
        "ANSIBLE_RETRY_FILES_ENABLED": "false",
        "ANSIBLE_ROLES_PATH": roles_path,
    }


def build_ansible_check_mode_command(
    *,
    playbook_path: str,
    inventory_hosts: tuple[str, ...],
    playbook_root: Path | None = None,
    check_mode_ssh_key_path: Path | None = None,
    check_mode_known_hosts_path: Path | None = None,
) -> AnsibleCommandSpec:
    root = playbook_root or next((path for path in _playbook_roots() if path.exists()), None)
    if root is None:
        raise ValueError("ansible_playbook_catalog_missing")
    inventory_path = (root / "inventory" / "hosts.yml").resolve()
    if not inventory_path.exists():
        raise ValueError("ansible_inventory_missing")
    if not inventory_hosts or not all(_SAFE_HOST_RE.fullmatch(host) for host in inventory_hosts):
        raise ValueError("unsafe_inventory_hosts")

    playbook_abs = _resolve_playbook_path(root, playbook_path)
    ssh_key_path = check_mode_ssh_key_path or _check_mode_ssh_key_path()
    known_hosts_path = check_mode_known_hosts_path or _check_mode_known_hosts_path()
    ssh_common_args = (
        f"-o UserKnownHostsFile={known_hosts_path} "
        "-o IdentitiesOnly=yes -o BatchMode=yes"
    )
    extra_vars = {
        "ansible_ssh_private_key_file": str(ssh_key_path),
        "ansible_ssh_common_args": ssh_common_args,
    }
    command = [
        "ansible-playbook",
        "-i",
        str(inventory_path),
        str(playbook_abs),
        "--check",
        "--diff",
        "--limit",
        ",".join(inventory_hosts),
        "--extra-vars",
        json.dumps(extra_vars, ensure_ascii=False, separators=(",", ":")),
    ]
    env = _ansible_command_env(root)
    return AnsibleCommandSpec(
        command=command,
        cwd=root,
        env=env,
        playbook_abs_path=playbook_abs,
        inventory_abs_path=inventory_path,
    )


def build_ansible_apply_command(
    *,
    playbook_path: str,
    inventory_hosts: tuple[str, ...],
    playbook_root: Path | None = None,
    check_mode_ssh_key_path: Path | None = None,
    check_mode_known_hosts_path: Path | None = None,
) -> AnsibleCommandSpec:
    """Build the controlled apply command after check-mode has passed."""

    root = playbook_root or next((path for path in _playbook_roots() if path.exists()), None)
    if root is None:
        raise ValueError("ansible_playbook_catalog_missing")
    inventory_path = (root / "inventory" / "hosts.yml").resolve()
    if not inventory_path.exists():
        raise ValueError("ansible_inventory_missing")
    if not inventory_hosts or not all(_SAFE_HOST_RE.fullmatch(host) for host in inventory_hosts):
        raise ValueError("unsafe_inventory_hosts")

    playbook_abs = _resolve_playbook_path(root, playbook_path)
    ssh_key_path = check_mode_ssh_key_path or _check_mode_ssh_key_path()
    known_hosts_path = check_mode_known_hosts_path or _check_mode_known_hosts_path()
    ssh_common_args = (
        f"-o UserKnownHostsFile={known_hosts_path} "
        "-o IdentitiesOnly=yes -o BatchMode=yes"
    )
    extra_vars = {
        "ansible_ssh_private_key_file": str(ssh_key_path),
        "ansible_ssh_common_args": ssh_common_args,
    }
    command = [
        "ansible-playbook",
        "-i",
        str(inventory_path),
        str(playbook_abs),
        "--diff",
        "--limit",
        ",".join(inventory_hosts),
        "--extra-vars",
        json.dumps(extra_vars, ensure_ascii=False, separators=(",", ":")),
    ]
    env = _ansible_command_env(root)
    return AnsibleCommandSpec(
        command=command,
        cwd=root,
        env=env,
        playbook_abs_path=playbook_abs,
        inventory_abs_path=inventory_path,
    )


async def _run_ansible_command(spec: AnsibleCommandSpec, *, timeout_seconds: int) -> AnsibleRunResult:
    started = time.monotonic()
    process = await asyncio.create_subprocess_exec(
        *spec.command,
        cwd=str(spec.cwd),
        env=spec.env,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    timed_out = False
    try:
        stdout_bytes, stderr_bytes = await asyncio.wait_for(
            process.communicate(),
            timeout=timeout_seconds,
        )
    except TimeoutError:
        timed_out = True
        process.kill()
        stdout_bytes, stderr_bytes = await process.communicate()
    duration_ms = int((time.monotonic() - started) * 1000)
    return AnsibleRunResult(
        returncode=124 if timed_out else int(process.returncode or 0),
        stdout=stdout_bytes.decode("utf-8", "replace"),
        stderr=stderr_bytes.decode("utf-8", "replace"),
        duration_ms=duration_ms,
        timed_out=timed_out,
    )


def _build_result_payload(
    result: AnsibleRunResult,
    *,
    controlled_apply_allowed: bool = False,
    controlled_apply_blocker: str | None = None,
) -> tuple[str, dict[str, Any], dict[str, Any], str | None]:
    status = "success" if result.returncode == 0 else "failed"
    stdout_tail = _tail(result.stdout, _STDOUT_LIMIT)
    stderr_tail = _tail(result.stderr, _STDERR_LIMIT)
    output = {
        "executor": "ansible",
        "execution_mode": "check_mode",
        "transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "ssh_key_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH,
        "known_hosts_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH,
        "check_mode": True,
        "apply_enabled": controlled_apply_allowed,
        "approval_required_before_apply": not controlled_apply_allowed,
        "controlled_apply_allowed": controlled_apply_allowed,
        "controlled_apply_blocker": controlled_apply_blocker,
        "returncode": result.returncode,
        "timed_out": result.timed_out,
        "stdout_tail": stdout_tail,
        "stderr_tail": stderr_tail,
        "next_required_step": (
            "controlled_apply_queued"
            if result.returncode == 0 and controlled_apply_allowed
            else "ai_playbook_or_transport_repair_required"
            if result.returncode != 0
            else "controlled_apply_blocked"
        ),
    }
    dry_run_result = {
        "check_mode_executed": True,
        "apply_executed": False,
        "safe_to_apply_without_approval": controlled_apply_allowed,
        "controlled_apply_allowed": controlled_apply_allowed,
        "controlled_apply_blocker": controlled_apply_blocker,
        "transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "ssh_key_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH,
        "known_hosts_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH,
        "returncode": result.returncode,
        "timed_out": result.timed_out,
        "stdout_tail": stdout_tail,
        "stderr_tail": stderr_tail,
    }
    error = None if result.returncode == 0 else (stderr_tail or f"ansible_check_mode_failed_rc_{result.returncode}")
    return status, output, dry_run_result, error


def _build_apply_result_payload(result: AnsibleRunResult) -> tuple[str, dict[str, Any], dict[str, Any], str | None]:
    status = "success" if result.returncode == 0 else "failed"
    stdout_tail = _tail(result.stdout, _STDOUT_LIMIT)
    stderr_tail = _tail(result.stderr, _STDERR_LIMIT)
    output = {
        "executor": "ansible",
        "execution_mode": "controlled_apply",
        "transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "ssh_key_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH,
        "known_hosts_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH,
        "check_mode": False,
        "apply_enabled": True,
        "controlled_apply_executed": True,
        "returncode": result.returncode,
        "timed_out": result.timed_out,
        "stdout_tail": stdout_tail,
        "stderr_tail": stderr_tail,
        "next_required_step": (
            "post_apply_verifier_and_learning_writeback"
            if result.returncode == 0
            else "ai_rollback_or_playbook_repair_required"
        ),
    }
    dry_run_result = {
        "check_mode_executed_before_apply": True,
        "apply_executed": True,
        "controlled_apply_executed": True,
        "transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "ssh_key_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH,
        "known_hosts_path": settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH,
        "returncode": result.returncode,
        "timed_out": result.timed_out,
        "stdout_tail": stdout_tail,
        "stderr_tail": stderr_tail,
    }
    error = None if result.returncode == 0 else (stderr_tail or f"ansible_apply_failed_rc_{result.returncode}")
    return status, output, dry_run_result, error


def _build_auto_repair_execution_receipt(
    claim: AnsibleCheckModeClaim,
    result: AnsibleRunResult,
    *,
    apply_op_id: str,
) -> dict[str, Any]:
    success = result.returncode == 0
    return {
        "incident_id": claim.incident_id,
        "playbook_id": str(claim.catalog_id or "")[:36] or "ansible",
        "playbook_name": f"Ansible controlled apply: {claim.apply_playbook_path}"[:200],
        "success": success,
        "executed_steps": [
            f"candidate:{claim.source_candidate_op_id}",
            f"check_mode:{claim.op_id}",
            f"apply:{apply_op_id}",
            f"catalog:{claim.catalog_id}",
            f"returncode:{result.returncode}",
        ],
        "error_message": None if success else _tail(result.stderr or result.stdout, 2000),
        "triggered_by": "ansible_controlled_apply",
        "similarity_score": None,
        "risk_level": str(claim.risk_level or ""),
        "execution_time_ms": result.duration_ms,
    }


def _int_from_value(value: Any, *, default: int = 1) -> int:
    if value in (None, ""):
        return default
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def _claim_from_apply_operation_row(row: dict[str, Any]) -> tuple[AnsibleCheckModeClaim, AnsibleRunResult] | None:
    input_payload = _json_loads(row.get("input"))
    output_payload = _json_loads(row.get("output"))
    dry_run_result = _json_loads(row.get("dry_run_result"))
    catalog_id = str(input_payload.get("catalog_id") or row.get("catalog_id") or "")
    if not catalog_id:
        return None
    catalog_item = get_ansible_catalog_item(catalog_id) or {}
    incident_id = str(row.get("incident_id") or input_payload.get("incident_id") or "")
    if not incident_id:
        return None
    inventory_hosts = input_payload.get("inventory_hosts") or catalog_item.get("inventory_hosts") or []
    if not isinstance(inventory_hosts, list) or not inventory_hosts:
        return None
    apply_playbook_path = str(
        input_payload.get("apply_playbook_path")
        or input_payload.get("playbook_path")
        or row.get("playbook_path")
        or catalog_item.get("playbook_path")
        or ""
    )
    check_mode_playbook_path = str(
        input_payload.get("check_mode_playbook_path")
        or input_payload.get("playbook_path")
        or row.get("playbook_path")
        or apply_playbook_path
    )
    if not apply_playbook_path:
        return None
    returncode = _int_from_value(
        output_payload.get("returncode", dry_run_result.get("returncode")),
        default=0 if str(row.get("status") or "").lower() == "success" else 1,
    )
    claim = AnsibleCheckModeClaim(
        op_id=str(input_payload.get("check_mode_op_id") or row.get("parent_op_id") or ""),
        source_candidate_op_id=str(input_payload.get("source_candidate_op_id") or ""),
        incident_id=incident_id,
        catalog_id=catalog_id,
        playbook_path=check_mode_playbook_path,
        apply_playbook_path=apply_playbook_path,
        inventory_hosts=tuple(str(host) for host in inventory_hosts),
        risk_level=str(
            input_payload.get("risk_level")
            or row.get("risk_level")
            or catalog_item.get("risk_level")
            or ""
        ),
        input_payload=input_payload,
    )
    result = AnsibleRunResult(
        returncode=returncode,
        stdout=str(output_payload.get("stdout_tail") or ""),
        stderr=str(row.get("error") or output_payload.get("stderr_tail") or ""),
        duration_ms=_int_from_value(row.get("duration_ms"), default=0),
        timed_out=bool(output_payload.get("timed_out") or dry_run_result.get("timed_out") or False),
    )
    return claim, result


async def _record_auto_repair_execution_receipt(
    claim: AnsibleCheckModeClaim,
    result: AnsibleRunResult,
    *,
    apply_op_id: str,
    project_id: str,
) -> bool:
    receipt = _build_auto_repair_execution_receipt(
        claim,
        result,
        apply_op_id=apply_op_id,
    )
    try:
        async with get_db_context(project_id) as db:
            inserted = await db.execute(
                text("""
                    INSERT INTO auto_repair_executions (
                        incident_id,
                        playbook_id,
                        playbook_name,
                        success,
                        executed_steps,
                        error_message,
                        triggered_by,
                        similarity_score,
                        risk_level,
                        execution_time_ms
                    )
                    SELECT
                        CAST(:incident_id AS varchar(30)),
                        CAST(:playbook_id AS varchar(36)),
                        CAST(:playbook_name AS varchar(200)),
                        :success,
                        CAST(:executed_steps AS jsonb),
                        :error_message,
                        CAST(:triggered_by AS varchar(50)),
                        :similarity_score,
                        CAST(:risk_level AS varchar(20)),
                        :execution_time_ms
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM auto_repair_executions existing
                        WHERE existing.incident_id = CAST(:incident_id AS varchar(30))
                          AND existing.triggered_by = CAST(:triggered_by AS varchar(50))
                          AND existing.executed_steps::text LIKE CAST(:apply_op_id_needle AS text)
                    )
                    RETURNING id
                """),
                {
                    **receipt,
                    "executed_steps": json.dumps(receipt["executed_steps"], ensure_ascii=False),
                    "apply_op_id_needle": f"%{apply_op_id}%",
                },
            )
            return inserted.scalar() is not None
    except Exception as exc:
        logger.warning(
            "ansible_auto_repair_execution_receipt_failed",
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            error=str(exc),
        )
        return False


def _post_apply_verification_result(result: AnsibleRunResult) -> str:
    if result.timed_out:
        return "timeout"
    return "success" if result.returncode == 0 else "failed"


def _post_apply_km_path_type(apply_op_id: str) -> str:
    return f"ansible_apply_receipt:{str(apply_op_id or '')[:8]}"


def _post_apply_action_label(claim: AnsibleCheckModeClaim, *, apply_op_id: str) -> str:
    return (
        f"ansible_controlled_apply:{claim.catalog_id}:"
        f"{claim.apply_playbook_path}:apply_op={apply_op_id}"
    )


async def _record_learning_writeback_receipt(
    claim: AnsibleCheckModeClaim,
    result: AnsibleRunResult,
    *,
    apply_op_id: str,
    verification_result: str,
    action_label: str,
    project_id: str,
) -> bool:
    """Persist the post-verifier learning receipt after LearningService accepts it."""

    matched_playbook_id = str(claim.catalog_id or "")[:36] or None
    try:
        from src.services.learning_service import get_learning_service

        await get_learning_service().record_verification_result(
            incident_id=claim.incident_id,
            action_taken=action_label,
            verification_result=verification_result,
            matched_playbook_id=matched_playbook_id,
        )
    except Exception as exc:
        logger.warning(
            "ansible_post_apply_trust_learning_writeback_failed",
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            error=str(exc),
        )
        return False

    try:
        input_payload = {
            "schema_version": "ansible_learning_writeback_receipt_v1",
            "incident_id": claim.incident_id,
            "catalog_id": claim.catalog_id,
            "playbook_path": claim.apply_playbook_path,
            "apply_op_id": apply_op_id,
            "verification_result": verification_result,
            "matched_playbook_id": matched_playbook_id,
            "learning_repository": "repair_result",
            "playbook_trust_update_attempted": matched_playbook_id is not None,
            "stores_raw_logs": False,
            "stores_secret_values": False,
        }
        output_payload = {
            "learning_recorded": True,
            "success": verification_result == "success",
            "returncode": result.returncode,
            "timed_out": result.timed_out,
        }
        async with get_db_context(project_id) as db:
            inserted = await db.execute(
                text("""
                    INSERT INTO automation_operation_log (
                        operation_type, actor, status, incident_id,
                        input, output, dry_run_result,
                        parent_op_id, tags
                    )
                    SELECT
                        'ansible_learning_writeback_recorded',
                        'ansible_controlled_apply_worker',
                        'success',
                        :incident_db_id,
                        CAST(:input AS jsonb),
                        CAST(:output AS jsonb),
                        '{}'::jsonb,
                        CAST(:parent_op_id AS uuid),
                        :tags
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM automation_operation_log existing
                        WHERE existing.operation_type = 'ansible_learning_writeback_recorded'
                          AND existing.parent_op_id = CAST(:parent_op_id AS uuid)
                    )
                    RETURNING op_id
                """),
                {
                    "incident_db_id": _automation_operation_log_incident_id(claim.incident_id),
                    "input": json.dumps(input_payload, ensure_ascii=False),
                    "output": json.dumps(output_payload, ensure_ascii=False),
                    "parent_op_id": apply_op_id,
                    "tags": [
                        "ansible",
                        "controlled_apply",
                        "learning_writeback",
                        "playbook_trust",
                        "ai_agent_auto_execution",
                    ],
                },
            )
            return inserted.scalar() is not None
    except Exception as exc:
        logger.warning(
            "ansible_learning_writeback_receipt_failed",
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            error=str(exc),
        )
        return False


async def _record_post_apply_verifier_and_learning(
    claim: AnsibleCheckModeClaim,
    result: AnsibleRunResult,
    *,
    apply_op_id: str,
    project_id: str,
) -> dict[str, bool]:
    """Persist post-apply verifier evidence and KM learning for Ansible apply."""

    verification_result = _post_apply_verification_result(result)
    action_label = _post_apply_action_label(claim, apply_op_id=apply_op_id)
    post_state = {
        "schema_version": "ansible_apply_result_verifier_v1",
        "verifier": "ansible_apply_result_verifier",
        "apply_op_id": apply_op_id,
        "catalog_id": claim.catalog_id,
        "playbook_path": claim.apply_playbook_path,
        "returncode": result.returncode,
        "timed_out": result.timed_out,
        "action_taken": action_label,
        "next_required_step": (
            "km_playbook_trust_writeback"
            if verification_result == "success"
            else "rollback_or_playbook_repair_candidate"
        ),
    }
    evidence_summary = (
        "AI Agent Ansible controlled apply verifier: "
        f"incident={claim.incident_id}; catalog={claim.catalog_id}; "
        f"result={verification_result}; returncode={result.returncode}; apply_op={apply_op_id}"
    )
    status = {"verification": False, "learning": False, "trust_learning": False}

    try:
        async with get_db_context(project_id) as db:
            inserted = await db.execute(
                text("""
                    INSERT INTO incident_evidence (
                        id,
                        incident_id,
                        matched_playbook_id,
                        schema_version,
                        mcp_health,
                        collection_duration_ms,
                        sensors_attempted,
                        sensors_succeeded,
                        evidence_summary,
                        post_execution_state,
                        verification_result
                    )
                    SELECT
                        gen_random_uuid()::text,
                        CAST(:incident_id AS varchar(30)),
                        CAST(:matched_playbook_id AS varchar(36)),
                        'v1',
                        CAST(:mcp_health AS jsonb),
                        :collection_duration_ms,
                        1,
                        :sensors_succeeded,
                        :evidence_summary,
                        CAST(:post_execution_state AS jsonb),
                        CAST(:verification_result AS varchar(20))
                    WHERE NOT EXISTS (
                        SELECT 1
                        FROM incident_evidence existing
                        WHERE existing.incident_id = CAST(:incident_id AS varchar(30))
                          AND existing.post_execution_state ->> 'apply_op_id' = CAST(:apply_op_id AS text)
                    )
                    RETURNING id
                """),
                {
                    "incident_id": claim.incident_id,
                    "matched_playbook_id": str(claim.catalog_id or "")[:36],
                    "mcp_health": json.dumps({
                        "ansible_apply_result_verifier": verification_result == "success"
                    }, ensure_ascii=False),
                    "collection_duration_ms": max(0, int(result.duration_ms or 0)),
                    "sensors_succeeded": 1 if verification_result == "success" else 0,
                    "evidence_summary": evidence_summary,
                    "post_execution_state": json.dumps(post_state, ensure_ascii=False),
                    "verification_result": verification_result,
                    "apply_op_id": apply_op_id,
                },
            )
            status["verification"] = inserted.scalar() is not None
    except Exception as exc:
        logger.warning(
            "ansible_post_apply_verifier_receipt_failed",
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            error=str(exc),
        )

    try:
        from src.models.knowledge import (
            EntrySource,
            EntryStatus,
            EntryType,
            KnowledgeEntryCreate,
        )
        from src.repositories.knowledge_repository import KnowledgeDBRepository

        async with get_db_context(project_id) as db:
            path_type = _post_apply_km_path_type(apply_op_id)
            existing = await db.execute(
                text("""
                    SELECT id
                    FROM knowledge_entries
                    WHERE related_incident_id = :incident_id
                      AND path_type = :path_type
                    LIMIT 1
                """),
                {
                    "incident_id": claim.incident_id,
                    "path_type": path_type,
                },
            )
            if existing.scalar() is not None:
                status["learning"] = True
            else:
                repo = KnowledgeDBRepository(db)
                await repo.create(
                    KnowledgeEntryCreate(
                        title=f"AI 自動修復沉澱：{claim.incident_id}",
                        content=(
                            "AI Agent 已完成 Ansible controlled apply 並寫入驗證摘要。\n\n"
                            f"- Incident: {claim.incident_id}\n"
                            f"- Catalog: {claim.catalog_id}\n"
                            f"- PlayBook: {claim.apply_playbook_path}\n"
                            f"- Apply operation: {apply_op_id}\n"
                            f"- Verification result: {verification_result}\n"
                            f"- Return code: {result.returncode}\n"
                            f"- Next step: {post_state['next_required_step']}\n"
                        ),
                        entry_type=EntryType.INCIDENT_CASE,
                        category="AI自動化/Ansible受控修復",
                        tags=[
                            "ai_auto_repair",
                            "ansible_controlled_apply",
                            verification_result,
                            str(claim.catalog_id or ""),
                        ],
                        source=EntrySource.AI_EXTRACTED,
                        status=EntryStatus.REVIEW,
                        related_incident_id=claim.incident_id,
                        related_playbook_id=str(claim.catalog_id or "")[:36] or None,
                        path_type=path_type,
                        created_by="ai_agent_ansible_worker",
                    )
                )
                status["learning"] = True
    except Exception as exc:
        logger.warning(
            "ansible_post_apply_learning_writeback_failed",
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            error=str(exc),
        )
    status["trust_learning"] = await _record_learning_writeback_receipt(
        claim,
        result,
        apply_op_id=apply_op_id,
        verification_result=verification_result,
        action_label=action_label,
        project_id=project_id,
    )
    return status


async def _send_controlled_apply_telegram_receipt(
    claim: AnsibleCheckModeClaim,
    result: AnsibleRunResult,
    *,
    apply_op_id: str,
    writeback: dict[str, bool],
    project_id: str,
) -> bool:
    try:
        from src.services.telegram_gateway import get_telegram_gateway

        response = await get_telegram_gateway().send_controlled_apply_result_receipt(
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            playbook_path=claim.apply_playbook_path,
            verification_result=_post_apply_verification_result(result),
            returncode=result.returncode,
            duration_ms=result.duration_ms,
            verifier_written=bool(writeback.get("verification")),
            learning_written=bool(writeback.get("learning")),
            project_id=project_id,
        )
        return bool(response)
    except Exception as exc:
        logger.warning(
            "ansible_controlled_apply_telegram_receipt_failed",
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            apply_op_id=apply_op_id,
            error=str(exc),
        )
        return False


async def backfill_missing_auto_repair_execution_receipts_once(
    *,
    project_id: str = "awoooi",
    window_hours: int = 24,
    limit: int = 10,
) -> dict[str, Any]:
    """Create auto_repair_executions receipts for existing Ansible apply rows."""

    stats = {
        "scanned": 0,
        "written": 0,
        "verification_written": 0,
        "learning_written": 0,
        "trust_learning_written": 0,
        "skipped": 0,
        "error": None,
    }
    try:
        async with get_db_context(project_id) as db:
            await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
            result = await db.execute(
                text("""
                    SELECT
                        apply.op_id::text AS op_id,
                        apply.parent_op_id::text AS parent_op_id,
                        coalesce(apply.incident_id::text, apply.input ->> 'incident_id') AS incident_id,
                        apply.input,
                        apply.output,
                        apply.dry_run_result,
                        apply.error,
                        apply.duration_ms,
                        apply.status,
                        apply.created_at
                    FROM automation_operation_log apply
                    WHERE apply.operation_type = 'ansible_apply_executed'
                      AND apply.created_at >= NOW() - (:window_hours * INTERVAL '1 hour')
                      AND (
                          NOT EXISTS (
                              SELECT 1
                              FROM auto_repair_executions existing
                              WHERE existing.created_at >= NOW() - (:window_hours * INTERVAL '1 hour')
                                AND existing.triggered_by = 'ansible_controlled_apply'
                                AND existing.executed_steps::text LIKE '%' || apply.op_id::text || '%'
                          )
                          OR NOT EXISTS (
                              SELECT 1
                              FROM incident_evidence evidence
                              WHERE evidence.incident_id = coalesce(
                                  apply.incident_id::text,
                                  apply.input ->> 'incident_id'
                              )
                                AND evidence.post_execution_state ->> 'apply_op_id' = apply.op_id::text
                          )
                          OR NOT EXISTS (
                              SELECT 1
                              FROM knowledge_entries km
                              WHERE km.related_incident_id = coalesce(
                                  apply.incident_id::text,
                                  apply.input ->> 'incident_id'
                                )
                                AND km.path_type = 'ansible_apply_receipt:' || left(apply.op_id::text, 8)
                          )
                          OR NOT EXISTS (
                              SELECT 1
                              FROM automation_operation_log learning
                              WHERE learning.operation_type = 'ansible_learning_writeback_recorded'
                                AND learning.parent_op_id = apply.op_id
                          )
                      )
                    ORDER BY apply.created_at DESC
                    LIMIT :limit
                """),
                {"window_hours": max(1, window_hours), "limit": max(1, limit)},
            )
            rows = [dict(row) for row in result.mappings().all()]
        stats["scanned"] = len(rows)
        for row in rows:
            reconstructed = _claim_from_apply_operation_row(row)
            if reconstructed is None:
                stats["skipped"] += 1
                continue
            claim, result = reconstructed
            if await _record_auto_repair_execution_receipt(
                claim,
                result,
                apply_op_id=str(row.get("op_id") or ""),
                project_id=project_id,
            ):
                stats["written"] += 1
            else:
                stats["skipped"] += 1
            writeback = await _record_post_apply_verifier_and_learning(
                claim,
                result,
                apply_op_id=str(row.get("op_id") or ""),
                project_id=project_id,
            )
            if writeback.get("verification"):
                stats["verification_written"] += 1
            if writeback.get("learning"):
                stats["learning_written"] += 1
            if writeback.get("trust_learning"):
                stats["trust_learning_written"] += 1
    except Exception as exc:
        stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
        logger.warning("ansible_auto_repair_execution_receipt_backfill_failed", **stats)
    return stats


async def claim_pending_check_modes(
    *,
    project_id: str = "awoooi",
    limit: int = 1,
    candidate_max_age_hours: int | None = None,
) -> list[AnsibleCheckModeClaim]:
    """Claim pending Ansible candidates by inserting pending check-mode rows."""

    claims: list[AnsibleCheckModeClaim] = []
    max_age_hours = candidate_max_age_hours or settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS
    async with get_db_context(project_id) as db:
        result = await db.execute(
            text("""
                SELECT
                    candidate.op_id,
                    candidate.input
                FROM automation_operation_log candidate
                WHERE candidate.operation_type = 'ansible_candidate_matched'
                  AND candidate.status = 'dry_run'
                  AND candidate.input ->> 'executor' = 'ansible'
                  AND candidate.created_at >= NOW() - (:candidate_max_age_hours * INTERVAL '1 hour')
                  AND COALESCE((candidate.dry_run_result ->> 'check_mode_executed')::boolean, false) = false
                  AND NOT EXISTS (
                      SELECT 1
                      FROM automation_operation_log existing
                      WHERE existing.parent_op_id = candidate.op_id
                        AND existing.operation_type IN (
                            'ansible_check_mode_executed',
                            'ansible_execution_skipped'
                        )
                  )
                ORDER BY candidate.created_at ASC
                LIMIT :limit
                FOR UPDATE SKIP LOCKED
            """),
            {
                "limit": max(1, limit),
                "candidate_max_age_hours": max(1, max_age_hours),
            },
        )
        rows = result.mappings().all()
        for row in rows:
            source_op_id = str(row["op_id"])
            candidate_input = _json_loads(row["input"])
            try:
                claim_input = build_ansible_check_mode_claim_input(
                    source_candidate_op_id=source_op_id,
                    candidate_input=candidate_input,
                )
            except ValueError as exc:
                await _insert_skipped_candidate(
                    db,
                    source_candidate_op_id=source_op_id,
                    candidate_input=candidate_input,
                    reason=str(exc),
                )
                continue
            inserted = await db.execute(
                text("""
                    INSERT INTO automation_operation_log (
                        operation_type, actor, status, incident_id,
                        input, output, dry_run_result,
                        parent_op_id, tags
                    ) VALUES (
                        'ansible_check_mode_executed',
                        'ansible_check_mode_worker',
                        'pending',
                        :incident_db_id,
                        CAST(:input AS jsonb),
                        '{}'::jsonb,
                        CAST(:dry_run_result AS jsonb),
                        CAST(:parent_op_id AS uuid),
                        :tags
                    )
                    RETURNING op_id
                """),
                {
                    "incident_id": _incident_id_from_payload(claim_input),
                    "incident_db_id": _automation_operation_log_incident_id(
                        _incident_id_from_payload(claim_input)
                    ),
                    "input": json.dumps(claim_input, ensure_ascii=False),
                    "dry_run_result": json.dumps({
                        "check_mode_executed": False,
                        "apply_executed": False,
                        "claim_state": "claimed",
                        "controlled_apply_allowed": bool(
                            claim_input.get("controlled_apply_allowed")
                        ),
                        "controlled_apply_blocker": claim_input.get("controlled_apply_blocker"),
                    }, ensure_ascii=False),
                    "parent_op_id": source_op_id,
                    "tags": [
                        "ansible",
                        "check_mode",
                        "pending",
                        "controlled_apply_allowed"
                        if claim_input.get("controlled_apply_allowed")
                        else "controlled_apply_blocked",
                    ],
                },
            )
            op_id = str(inserted.scalar_one())
            claims.append(
                AnsibleCheckModeClaim(
                    op_id=op_id,
                    source_candidate_op_id=source_op_id,
                    incident_id=str(claim_input.get("incident_id") or ""),
                    catalog_id=str(claim_input["catalog_id"]),
                    playbook_path=str(claim_input["playbook_path"]),
                    apply_playbook_path=str(claim_input["apply_playbook_path"]),
                    inventory_hosts=tuple(str(host) for host in claim_input["inventory_hosts"]),
                    risk_level=str(claim_input.get("risk_level") or ""),
                    input_payload=claim_input,
                )
            )
    return claims


async def recent_ansible_transport_blockers(
    *,
    project_id: str = "awoooi",
    cooldown_seconds: int | None = None,
    include_repair_forced_command_blocker: bool | None = None,
) -> list[str]:
    """Return transport blockers observed from recent failed check-mode rows."""

    cooldown = cooldown_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_COOLDOWN_SECONDS
    include_forced_blocker = (
        _uses_repair_forced_command_transport()
        if include_repair_forced_command_blocker is None
        else include_repair_forced_command_blocker
    )
    async with get_db_context(project_id) as db:
        result = await db.execute(
            text("""
                SELECT
                    coalesce(output::text, '') AS output_text,
                    coalesce(dry_run_result::text, '') AS dry_run_text,
                    coalesce(error, '') AS error_text,
                    coalesce(stderr_feed_back, '') AS stderr_text
                FROM automation_operation_log
                WHERE operation_type = 'ansible_check_mode_executed'
                  AND status = 'failed'
                  AND created_at >= NOW() - (:cooldown_seconds * INTERVAL '1 second')
                ORDER BY created_at DESC
                LIMIT 20
            """),
            {"cooldown_seconds": max(60, cooldown)},
        )
        blockers: set[str] = set()
        for row in result.mappings().all():
            detected = detect_ansible_transport_blockers(
                row.get("output_text"),
                row.get("dry_run_text"),
                row.get("error_text"),
                row.get("stderr_text"),
            )
            blockers.update(
                blocker
                for blocker in detected
                if blocker != FORCED_COMMAND_BLOCKER or include_forced_blocker
            )
    return sorted(blockers)


async def _insert_skipped_candidate(
    db: Any,
    *,
    source_candidate_op_id: str,
    candidate_input: dict[str, Any],
    reason: str,
) -> None:
    input_payload = {
        "incident_id": _incident_id_from_payload(candidate_input),
        "executor": "ansible",
        "execution_backend": "ansible",
        "execution_mode": "check_mode",
        "transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "check_mode": True,
        "apply_enabled": False,
        "source_candidate_op_id": source_candidate_op_id,
        "not_used_reason": reason,
    }
    await db.execute(
        text("""
            INSERT INTO automation_operation_log (
                operation_type, actor, status, incident_id,
                input, output, dry_run_result,
                parent_op_id, tags
            ) VALUES (
                'ansible_execution_skipped',
                'ansible_check_mode_worker',
                'dry_run',
                :incident_db_id,
                CAST(:input AS jsonb),
                CAST(:output AS jsonb),
                CAST(:dry_run_result AS jsonb),
                CAST(:parent_op_id AS uuid),
                :tags
            )
        """),
        {
            "incident_id": _incident_id_from_payload(input_payload),
            "incident_db_id": _automation_operation_log_incident_id(
                _incident_id_from_payload(input_payload)
            ),
            "input": json.dumps(input_payload, ensure_ascii=False),
            "output": json.dumps({
                "not_used_reason": reason,
                "decision_effect": "skipped_before_runtime",
            }, ensure_ascii=False),
            "dry_run_result": json.dumps({
                "check_mode_executed": False,
                "apply_executed": False,
                "skipped": True,
                "reason": reason,
            }, ensure_ascii=False),
            "parent_op_id": source_candidate_op_id,
            "tags": ["ansible", "check_mode", "skipped", "apply_locked"],
        },
    )


async def finalize_check_mode_claim(
    claim: AnsibleCheckModeClaim,
    result: AnsibleRunResult,
    *,
    project_id: str = "awoooi",
) -> None:
    controlled_apply_allowed = bool(claim.input_payload.get("controlled_apply_allowed"))
    controlled_apply_blocker = (
        str(claim.input_payload.get("controlled_apply_blocker"))
        if claim.input_payload.get("controlled_apply_blocker")
        else None
    )
    status, output, dry_run_result, error = _build_result_payload(
        result,
        controlled_apply_allowed=controlled_apply_allowed,
        controlled_apply_blocker=controlled_apply_blocker,
    )
    async with get_db_context(project_id) as db:
        await db.execute(
            text("""
                UPDATE automation_operation_log
                SET status = :status,
                    output = CAST(:output AS jsonb),
                    dry_run_result = CAST(:dry_run_result AS jsonb),
                    error = :error,
                    duration_ms = :duration_ms,
                    stderr_feed_back = :stderr
                WHERE op_id = CAST(:op_id AS uuid)
            """),
            {
                "status": status,
                "output": json.dumps(output, ensure_ascii=False),
                "dry_run_result": json.dumps(dry_run_result, ensure_ascii=False),
                "error": _tail(error or "", 2000) or None,
                "duration_ms": result.duration_ms,
                "stderr": _tail(result.stderr, _STDERR_LIMIT),
                "op_id": claim.op_id,
            },
        )


async def run_claimed_check_mode(
    claim: AnsibleCheckModeClaim,
    *,
    timeout_seconds: int,
    project_id: str = "awoooi",
) -> AnsibleRunResult:
    try:
        spec = build_ansible_check_mode_command(
            playbook_path=claim.playbook_path,
            inventory_hosts=claim.inventory_hosts,
        )
        result = await _run_ansible_command(spec, timeout_seconds=timeout_seconds)
    except Exception as exc:
        result = AnsibleRunResult(
            returncode=1,
            stdout="",
            stderr=f"ansible_check_mode_runtime_error: {exc}",
            duration_ms=0,
        )
    await finalize_check_mode_claim(claim, result, project_id=project_id)
    logger.info(
        "ansible_check_mode_claim_completed",
        op_id=claim.op_id,
        source_candidate_op_id=claim.source_candidate_op_id,
        incident_id=claim.incident_id,
        catalog_id=claim.catalog_id,
        returncode=result.returncode,
        timed_out=result.timed_out,
    )
    return result


async def run_controlled_apply_for_claim(
    claim: AnsibleCheckModeClaim,
    *,
    timeout_seconds: int,
    project_id: str = "awoooi",
) -> AnsibleRunResult | None:
    """Execute apply for an allowlisted claim after check-mode succeeds."""

    if claim.input_payload.get("controlled_apply_allowed") is not True:
        logger.info(
            "ansible_controlled_apply_blocked",
            op_id=claim.op_id,
            source_candidate_op_id=claim.source_candidate_op_id,
            incident_id=claim.incident_id,
            catalog_id=claim.catalog_id,
            blocker=claim.input_payload.get("controlled_apply_blocker"),
        )
        return None

    async with get_db_context(project_id) as db:
        inserted = await db.execute(
            text("""
                INSERT INTO automation_operation_log (
                    operation_type, actor, status, incident_id,
                    input, output, dry_run_result,
                    parent_op_id, tags
                ) VALUES (
                    'ansible_apply_executed',
                    'ansible_controlled_apply_worker',
                    'pending',
                    :incident_db_id,
                    CAST(:input AS jsonb),
                    '{}'::jsonb,
                    CAST(:dry_run_result AS jsonb),
                    CAST(:parent_op_id AS uuid),
                    :tags
                )
                RETURNING op_id
            """),
            {
                "incident_id": claim.incident_id,
                "incident_db_id": _automation_operation_log_incident_id(claim.incident_id),
                "input": json.dumps({
                    **claim.input_payload,
                    "execution_mode": "controlled_apply",
                    "check_mode_op_id": claim.op_id,
                    "playbook_path": claim.apply_playbook_path,
                    "check_mode_playbook_path": claim.playbook_path,
                    "apply_playbook_path": claim.apply_playbook_path,
                    "check_mode": False,
                    "apply_enabled": True,
                    "approval_required_before_apply": False,
                    "controlled_apply_allowed": True,
                }, ensure_ascii=False),
                "dry_run_result": json.dumps({
                    "check_mode_executed_before_apply": True,
                    "apply_executed": False,
                    "claim_state": "claimed",
                    "source_check_mode_op_id": claim.op_id,
                }, ensure_ascii=False),
                "parent_op_id": claim.op_id,
                "tags": [
                    "ansible",
                    "controlled_apply",
                    str(claim.risk_level or "unknown").lower(),
                    "ai_agent_auto_execution",
                ],
            },
        )
        apply_op_id = str(inserted.scalar_one())

    try:
        spec = build_ansible_apply_command(
            playbook_path=claim.apply_playbook_path,
            inventory_hosts=claim.inventory_hosts,
        )
        result = await _run_ansible_command(spec, timeout_seconds=timeout_seconds)
    except Exception as exc:
        result = AnsibleRunResult(
            returncode=1,
            stdout="",
            stderr=f"ansible_controlled_apply_runtime_error: {exc}",
            duration_ms=0,
        )

    status, output, dry_run_result, error = _build_apply_result_payload(result)
    async with get_db_context(project_id) as db:
        await db.execute(
            text("""
                UPDATE automation_operation_log
                SET status = :status,
                    output = CAST(:output AS jsonb),
                    dry_run_result = CAST(:dry_run_result AS jsonb),
                    error = :error,
                    duration_ms = :duration_ms,
                    stderr_feed_back = :stderr
                WHERE op_id = CAST(:op_id AS uuid)
            """),
            {
                "status": status,
                "output": json.dumps(output, ensure_ascii=False),
                "dry_run_result": json.dumps(dry_run_result, ensure_ascii=False),
                "error": _tail(error or "", 2000) or None,
                "duration_ms": result.duration_ms,
                "stderr": _tail(result.stderr, _STDERR_LIMIT),
                "op_id": apply_op_id,
            },
        )
    receipt_written = await _record_auto_repair_execution_receipt(
        claim,
        result,
        apply_op_id=apply_op_id,
        project_id=project_id,
    )
    writeback = await _record_post_apply_verifier_and_learning(
        claim,
        result,
        apply_op_id=apply_op_id,
        project_id=project_id,
    )
    telegram_receipt_sent = await _send_controlled_apply_telegram_receipt(
        claim,
        result,
        apply_op_id=apply_op_id,
        writeback=writeback,
        project_id=project_id,
    )

    logger.info(
        "ansible_controlled_apply_completed",
        op_id=apply_op_id,
        check_mode_op_id=claim.op_id,
        source_candidate_op_id=claim.source_candidate_op_id,
        incident_id=claim.incident_id,
        catalog_id=claim.catalog_id,
        returncode=result.returncode,
        timed_out=result.timed_out,
        auto_repair_receipt_written=receipt_written,
        post_apply_verification_written=writeback.get("verification"),
        post_apply_learning_written=writeback.get("learning"),
        telegram_receipt_sent=telegram_receipt_sent,
    )
    return result


async def run_pending_check_modes_once(
    *,
    project_id: str = "awoooi",
    limit: int = 1,
    timeout_seconds: int | None = None,
) -> dict[str, Any]:
    blockers = _runtime_blockers()
    if blockers:
        logger.warning("ansible_check_mode_runtime_blocked", blockers=blockers)
        return {"claimed": 0, "completed": 0, "failed": 0, "blockers": blockers}
    transport_blockers = await recent_ansible_transport_blockers(project_id=project_id)
    if transport_blockers:
        logger.warning("ansible_check_mode_transport_blocked", blockers=transport_blockers)
        return {"claimed": 0, "completed": 0, "failed": 0, "blockers": transport_blockers}

    claims = await claim_pending_check_modes(
        project_id=project_id,
        limit=limit,
        candidate_max_age_hours=settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS,
    )
    completed = 0
    failed = 0
    apply_completed = 0
    apply_failed = 0
    apply_blocked = 0
    for claim in claims:
        result = await run_claimed_check_mode(
            claim,
            timeout_seconds=timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS,
            project_id=project_id,
        )
        completed += 1
        if result.returncode != 0:
            failed += 1
            continue
        apply_result = await run_controlled_apply_for_claim(
            claim,
            timeout_seconds=settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_TIMEOUT_SECONDS,
            project_id=project_id,
        )
        if apply_result is None:
            apply_blocked += 1
        else:
            apply_completed += 1
            if apply_result.returncode != 0:
                apply_failed += 1
    return {
        "claimed": len(claims),
        "completed": completed,
        "failed": failed,
        "apply_completed": apply_completed,
        "apply_failed": apply_failed,
        "apply_blocked": apply_blocked,
        "blockers": [],
    }
