awoooi/apps/api/src/services/awooop_ansible_audit_service.py

"""AwoooP Ansible audit helpers.

This module exposes the Ansible audit contract and repo-known playbook catalog.
Catalog rows are the first hard boundary for AI Agent controlled apply: only
allowlisted playbooks with check-mode support can move from dry-run to apply.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
from typing import Any
from uuid import UUID, uuid4

import structlog
from sqlalchemy import text

from src.db.base import get_db_context
from src.services.controlled_alert_target_router import (
    resolve_controlled_alert_target,
    resolve_typed_incident_target,
)

logger = structlog.get_logger(__name__)

_TYPED_ROUTE_GENERATION_PREFIX = "typed-route-v2"
_RECURRENCE_REQUIRED_PROPOSAL_SOURCES = frozenset(
    {
        "alert_webhook_controlled_router",
        "truth_chain_candidate_backfill",
        "iwooos_wazuh_manager_posture_scheduler",
        "iwooos_wazuh_alert_ingress_scheduler",
    }
)


def _source_fingerprint(incident: dict[str, Any], proposal: dict[str, Any]) -> str:
    """Return the current alert fingerprint without inventing one."""

    explicit = str(proposal.get("source_fingerprint") or "").strip()
    if explicit:
        return explicit
    for signal in incident.get("signals") or []:
        if not isinstance(signal, dict):
            continue
        labels = signal.get("labels")
        if not isinstance(labels, dict):
            continue
        fingerprint = str(
            labels.get("fingerprint") or labels.get("alert_fingerprint") or ""
        ).strip()
        if fingerprint:
            return fingerprint
    return ""


def _source_recurrence_receipt(
    incident: dict[str, Any],
    proposal: dict[str, Any],
) -> dict[str, Any]:
    """Build a fail-closed receipt proving this is not a historical-only replay."""

    explicit_fingerprint = str(
        proposal.get("source_fingerprint") or ""
    ).strip()
    fingerprint = explicit_fingerprint or _source_fingerprint(incident, proposal)
    proposal_source = str(proposal.get("source") or "").strip()
    occurrence_id = str(proposal.get("source_occurrence_id") or "").strip()
    verified = bool(
        explicit_fingerprint
        and occurrence_id
        and proposal.get("source_recurrence_verified") is True
    )
    return {
        "schema_version": "alert_source_recurrence_receipt_v1",
        "verified": verified,
        "fingerprint": fingerprint or None,
        "occurrence_id": occurrence_id or None,
        "observed_at": str(proposal.get("source_received_at") or "") or None,
        "identity_anchor": str(
            proposal.get("source_recurrence_anchor") or ""
        )
        or None,
        "canonical_asset_id": str(
            proposal.get("source_canonical_asset_id") or ""
        )
        or None,
        "source": (
            "current_alert_webhook"
            if proposal_source == "alert_webhook_controlled_router"
            else str(proposal.get("source_recurrence_source") or "") or None
        ),
        "durable_readback_verified": (
            proposal.get("source_recurrence_durable_readback_verified") is True
        ),
    }


def build_typed_target_route_generation(
    *,
    typed_target_route: dict[str, Any],
    executor_candidates: list[dict[str, Any]],
) -> str:
    """Hash only semantic target scope, not deploy SHA or volatile evidence."""

    identity = {
        "schema_version": str(typed_target_route.get("schema_version") or ""),
        "resolution_status": str(
            typed_target_route.get("resolution_status") or ""
        ),
        "canonical_asset_id": str(
            typed_target_route.get("canonical_asset_id") or ""
        ),
        "typed_domain": str(typed_target_route.get("target_kind") or ""),
        "executor": str(typed_target_route.get("executor") or ""),
        "cross_domain_fallback_allowed": typed_target_route.get(
            "cross_domain_fallback_allowed"
        ),
        "candidates": sorted(
            (
                {
                    "catalog_id": str(row.get("catalog_id") or ""),
                    "catalog_revision": str(row.get("catalog_revision") or ""),
                    "canonical_asset_id": str(
                        row.get("canonical_asset_id") or ""
                    ),
                    "typed_domain": str(row.get("typed_domain") or ""),
                    "inventory_hosts": sorted(
                        str(host) for host in row.get("inventory_hosts") or []
                    ),
                }
                for row in executor_candidates
            ),
            key=lambda row: (
                row["catalog_id"],
                row["catalog_revision"],
                row["inventory_hosts"],
            ),
        ),
    }
    canonical = json.dumps(
        identity,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:24]
    return f"{_TYPED_ROUTE_GENERATION_PREFIX}:{digest}"


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


ANSIBLE_OPERATION_TYPES = frozenset({
    "ansible_candidate_matched",
    "ansible_check_mode_executed",
    "ansible_apply_executed",
    "ansible_learning_writeback_recorded",
    "ansible_rollback_executed",
    "ansible_execution_skipped",
})


def verified_ansible_candidate_terminal_sql(candidate_alias: str) -> str:
    """Return the shared SQL predicate for a safely terminal candidate."""

    if re.fullmatch(r"[a-z_][a-z0-9_]*", candidate_alias) is None:
        raise ValueError("invalid_candidate_sql_alias")
    candidate = candidate_alias
    incident_id = (
        f"coalesce({candidate}.incident_id::text, "
        f"{candidate}.input ->> 'incident_id')"
    )
    automation_run_id = (
        f"coalesce(nullif({candidate}.input ->> 'automation_run_id', ''), "
        f"{candidate}.op_id::text)"
    )
    template = """
        EXISTS (
            SELECT 1
            FROM automation_operation_log direct_terminal
            WHERE direct_terminal.parent_op_id = __CANDIDATE__.op_id
              AND direct_terminal.operation_type = 'ansible_execution_skipped'
              AND direct_terminal.status = 'dry_run'
              AND direct_terminal.dry_run_result ->> 'skipped' = 'true'
              AND coalesce(
                    direct_terminal.dry_run_result ->> 'apply_executed',
                    'false'
                  ) = 'false'
        )
        OR EXISTS (
            SELECT 1
            FROM automation_operation_log check_mode
            JOIN automation_operation_log apply
              ON apply.parent_op_id = check_mode.op_id
             AND apply.operation_type = 'ansible_apply_executed'
             AND apply.status = 'failed'
            JOIN automation_operation_log replay
              ON replay.parent_op_id = apply.op_id
             AND replay.operation_type = 'ansible_execution_skipped'
             AND replay.status IN ('success', 'failed')
             AND replay.input ->> 'execution_mode'
                    = 'controlled_retry_check_mode_replay'
             AND replay.output ->> 'runtime_apply_executed' = 'false'
            WHERE check_mode.parent_op_id = __CANDIDATE__.op_id
              AND check_mode.operation_type = 'ansible_check_mode_executed'
              AND check_mode.status = 'success'
              AND check_mode.input ->> 'automation_run_id'
                    = __AUTOMATION_RUN_ID__
              AND coalesce(
                    check_mode.incident_id::text,
                    check_mode.input ->> 'incident_id'
                  ) = __INCIDENT_ID__
              AND apply.input ->> 'automation_run_id' = __AUTOMATION_RUN_ID__
              AND coalesce(
                    apply.incident_id::text,
                    apply.input ->> 'incident_id'
                  ) = __INCIDENT_ID__
              AND replay.input ->> 'automation_run_id' = __AUTOMATION_RUN_ID__
              AND coalesce(
                    replay.incident_id::text,
                    replay.input ->> 'incident_id'
                  ) = __INCIDENT_ID__
              AND replay.output ->> 'terminal_disposition' = CASE
                    WHEN replay.status = 'success'
                    THEN 'no_write_replay_passed_waiting_repair_candidate'
                    ELSE 'no_write_replay_failed_waiting_playbook_or_transport_repair'
                  END
              AND EXISTS (
                    SELECT 1
                    FROM incident_evidence verifier
                    WHERE verifier.incident_id = __INCIDENT_ID__
                      AND verifier.post_execution_state ->> 'apply_op_id'
                            = apply.op_id::text
                      AND verifier.verification_result IN ('failed', 'timeout')
              )
              AND EXISTS (
                    SELECT 1
                    FROM jsonb_array_elements(
                        coalesce(
                            apply.input -> 'runtime_stage_receipts',
                            '[]'::jsonb
                        )
                    ) receipt(value)
                    WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback'
                      AND receipt.value ->> 'automation_run_id'
                            = __AUTOMATION_RUN_ID__
                      AND receipt.value #>> '{detail,failed_apply_op_id}'
                            = apply.op_id::text
                      AND receipt.value #>> '{detail,retry_check_mode_op_id}'
                            = replay.op_id::text
                      AND receipt.value #>> '{detail,terminal_type}'
                            = replay.output ->> 'terminal_disposition'
                      AND receipt.value #>> '{detail,verified_no_write_terminal}'
                            = 'true'
                      AND receipt.value #>> '{detail,runtime_apply_executed}'
                            = 'false'
                      AND receipt.value ->> 'durable_receipt' = 'true'
              )
              AND EXISTS (
                    SELECT 1
                    FROM jsonb_array_elements(
                        coalesce(
                            apply.input -> 'runtime_stage_receipts',
                            '[]'::jsonb
                        )
                    ) closure_receipt(value)
                    WHERE closure_receipt.value ->> 'stage_id'
                            = 'incident_closure'
                      AND closure_receipt.value ->> 'automation_run_id'
                            = __AUTOMATION_RUN_ID__
                      AND closure_receipt.value #>> '{detail,apply_op_id}'
                            = apply.op_id::text
                      AND closure_receipt.value #>> '{detail,retry_op_id}'
                            = replay.op_id::text
                      AND closure_receipt.value #>> '{detail,terminal_type}'
                            = replay.output ->> 'terminal_disposition'
                      AND closure_receipt.value #>>
                            '{detail,no_write_terminal}' = 'true'
                      AND closure_receipt.value #>>
                            '{detail,proposal_executed}' = 'true'
                      AND closure_receipt.value #>>
                            '{detail,execution_success}' = 'false'
                      AND closure_receipt.value #>>
                            '{detail,telegram_receipt_acknowledged}' = 'true'
                      AND closure_receipt.value #>>
                            '{detail,repository_readback_verified}' = 'true'
                      AND closure_receipt.value ->> 'durable_receipt' = 'true'
              )
              AND EXISTS (
                    SELECT 1
                    FROM alert_operation_log lifecycle
                    WHERE lifecycle.incident_id = __INCIDENT_ID__
                      AND lifecycle.event_type::text = 'EXECUTION_COMPLETED'
                      AND lifecycle.success IS FALSE
                      AND lifecycle.context ->> 'apply_op_id'
                            = apply.op_id::text
                      AND lifecycle.context ->> 'automation_run_id'
                            = __AUTOMATION_RUN_ID__
              )
        )
        OR EXISTS (
            SELECT 1
            FROM automation_operation_log failed_check
            JOIN automation_operation_log replay_check
              ON replay_check.parent_op_id = failed_check.parent_op_id
             AND replay_check.operation_type = 'ansible_check_mode_executed'
             AND replay_check.status IN ('success', 'failed')
             AND replay_check.input ->> 'execution_mode'
                    = 'stdin_boundary_check_mode_replay'
             AND failed_check.op_id = CAST(
                    NULLIF(
                        replay_check.input
                            ->> 'replay_of_check_mode_op_id',
                        ''
                    ) AS uuid
                 )
            JOIN automation_operation_log terminal
              ON terminal.parent_op_id = replay_check.op_id
             AND terminal.operation_type = 'ansible_execution_skipped'
             AND terminal.status IN ('success', 'failed')
             AND terminal.input ->> 'execution_mode'
                    = 'stdin_boundary_check_mode_replay_terminal'
             AND terminal.output ->> 'runtime_apply_executed' = 'false'
            WHERE failed_check.parent_op_id = __CANDIDATE__.op_id
              AND failed_check.operation_type = 'ansible_check_mode_executed'
              AND failed_check.status = 'failed'
              AND failed_check.input ->> 'automation_run_id'
                    = __AUTOMATION_RUN_ID__
              AND coalesce(
                    failed_check.incident_id::text,
                    failed_check.input ->> 'incident_id'
                  ) = __INCIDENT_ID__
              AND replay_check.input ->> 'automation_run_id'
                    = __AUTOMATION_RUN_ID__
              AND coalesce(
                    replay_check.incident_id::text,
                    replay_check.input ->> 'incident_id'
                  ) = __INCIDENT_ID__
              AND terminal.input ->> 'automation_run_id'
                    = __AUTOMATION_RUN_ID__
              AND terminal.input ->> 'retry_of_check_mode_op_id'
                    = failed_check.op_id::text
              AND coalesce(
                    terminal.incident_id::text,
                    terminal.input ->> 'incident_id'
                  ) = __INCIDENT_ID__
              AND (
                    SELECT COUNT(DISTINCT receipt.value ->> 'stage_id')
                    FROM jsonb_array_elements(
                        coalesce(
                            terminal.input -> 'runtime_stage_receipts',
                            '[]'::jsonb
                        )
                    ) receipt(value)
                    WHERE receipt.value ->> 'stage_id' IN (
                        'mcp_context',
                        'service_log_evidence',
                        'normalized_asset_identity',
                        'source_truth_diff',
                        'risk_policy_decision',
                        'executor_log_projection',
                        'retry_or_rollback',
                        'km_playbook_writeback',
                        'rag_writeback',
                        'playbook_trust',
                        'timeline_projection'
                    )
                      AND receipt.value ->> 'automation_run_id'
                            = __AUTOMATION_RUN_ID__
                      AND receipt.value ->> 'durable_receipt' = 'true'
                  ) = 11
              AND EXISTS (
                    SELECT 1
                    FROM jsonb_array_elements(
                        coalesce(
                            terminal.input -> 'runtime_stage_receipts',
                            '[]'::jsonb
                        )
                    ) receipt(value)
                    WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback'
                      AND receipt.value #>>
                            '{detail,retry_terminal_op_id}'
                            = terminal.op_id::text
                      AND receipt.value #>>
                            '{detail,verified_no_write_terminal}' = 'true'
                      AND receipt.value #>>
                            '{detail,runtime_apply_executed}' = 'false'
                      AND receipt.value #>>
                            '{detail,repository_readback_verified}' = 'true'
                      AND receipt.value ->> 'durable_receipt' = 'true'
              )
              AND EXISTS (
                    SELECT 1
                    FROM jsonb_array_elements(
                        coalesce(
                            terminal.input -> 'runtime_stage_receipts',
                            '[]'::jsonb
                        )
                    ) receipt(value)
                    WHERE receipt.value ->> 'stage_id' = 'incident_closure'
                      AND receipt.value #>> '{detail,terminal_op_id}'
                            = terminal.op_id::text
                      AND receipt.value #>> '{detail,no_write_terminal}'
                            = 'true'
                      AND receipt.value #>>
                            '{detail,telegram_receipt_acknowledged}' = 'true'
                      AND receipt.value #>>
                            '{detail,learning_writeback_acknowledged}' = 'true'
                      AND receipt.value #>>
                            '{detail,incident_projection_readback_verified}'
                            = 'true'
                      AND receipt.value #>>
                            '{detail,projection_readback_verified}' = 'true'
                      AND receipt.value #>>
                            '{detail,repository_readback_verified}' = 'true'
                      AND receipt.value ->> 'durable_receipt' = 'true'
              )
              AND EXISTS (
                    SELECT 1
                    FROM awooop_outbound_message outbound
                    WHERE outbound.channel_type = 'telegram'
                      AND (
                            (
                                outbound.send_status = 'sent'
                                AND outbound.provider_message_id IS NOT NULL
                            )
                            OR (
                                outbound.send_status = 'shadow'
                                AND outbound.provider_message_id IS NULL
                                AND outbound.source_envelope #>>
                                    '{notification_policy,disposition}'
                                    = 'suppressed'
                            )
                      )
                      AND outbound.source_envelope #>>
                            '{callback_reply,action}'
                            = 'controlled_apply_result'
                      AND outbound.source_envelope #>>
                            '{callback_reply,execution_kind}'
                            = 'no_write_replay'
                      AND outbound.source_envelope #>>
                            '{callback_reply,automation_run_id}'
                            = __AUTOMATION_RUN_ID__
                      AND outbound.source_envelope #>>
                            '{callback_reply,incident_id}' = __INCIDENT_ID__
                      AND outbound.source_envelope #>>
                            '{callback_reply,apply_op_id}'
                            = terminal.op_id::text
              )
              AND EXISTS (
                    SELECT 1
                    FROM alert_operation_log lifecycle
                    WHERE lifecycle.incident_id = __INCIDENT_ID__
                      AND lifecycle.event_type::text = 'EXECUTION_COMPLETED'
                      AND lifecycle.success IS FALSE
                      AND lifecycle.context ->> 'apply_op_id'
                            = terminal.op_id::text
                      AND lifecycle.context ->> 'automation_run_id'
                            = __AUTOMATION_RUN_ID__
              )
              AND EXISTS (
                    SELECT 1
                    FROM alert_operation_log lifecycle
                    WHERE lifecycle.incident_id = __INCIDENT_ID__
                      AND (
                            lifecycle.event_type::text
                                = 'TELEGRAM_RESULT_SENT'
                            OR (
                                lifecycle.event_type::text
                                    = 'NOTIFICATION_CLASSIFIED'
                                AND lifecycle.action_detail
                                    = 'stdin_boundary_no_write_result_receipt_suppressed'
                            )
                      )
                      AND lifecycle.context ->> 'apply_op_id'
                            = terminal.op_id::text
                      AND lifecycle.context ->> 'automation_run_id'
                            = __AUTOMATION_RUN_ID__
              )
        )
    """
    return (
        template.replace("__CANDIDATE__", candidate)
        .replace("__INCIDENT_ID__", incident_id)
        .replace("__AUTOMATION_RUN_ID__", automation_run_id)
    )

_CATALOG: tuple[dict[str, Any], ...] = (
    {
        "catalog_id": "ansible:awoooi-auto-repair-canary",
        "playbook_path": "infra/ansible/playbooks/awoooi-auto-repair-canary.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/awoooi-auto-repair-canary.yml",
        "inventory_hosts": ["host_121"],
        "domains": ["awoooi", "auto_repair", "k3s", "controlled_canary"],
        "keywords": [
            "awooopautorepaircanaryt16",
            "awoooi-auto-repair-canary",
            "auto repair canary",
            "controlled canary",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "low",
    },
    {
        "catalog_id": "ansible:wazuh-manager-posture-readback",
        "playbook_path": "infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
        "inventory_hosts": ["host_112"],
        "domains": ["wazuh", "siem", "manager_posture", "security"],
        "keywords": [
            "wazuh",
            "wazuhmanager",
            "wazuh-manager",
            "siem",
            "fim",
            "security manager",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "low",
        "no_write_only": True,
    },
    {
        "catalog_id": "ansible:wazuh-alertmanager-integration",
        "playbook_path": (
            "infra/ansible/playbooks/wazuh-alertmanager-integration.yml"
        ),
        "check_mode_playbook_path": (
            "infra/ansible/playbooks/wazuh-alertmanager-integration.yml"
        ),
        "inventory_hosts": ["host_112"],
        "domains": [
            "wazuh",
            "siem",
            "security_event_ingress",
            "alertmanager",
        ],
        "keywords": [
            "wazuhalertmanagerintegrationconvergence",
            "wazuh alertmanager integration convergence",
            "custom-awoooi-alertmanager",
            "security event ingress",
        ],
        "activation_keywords": [
            "wazuhalertmanagerintegrationconvergence",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "high",
        "catalog_revision": "2026-07-15-wazuh-alert-ingress-v2",
    },
    {
        "catalog_id": "ansible:110-alertmanager-delivery-recovery",
        "playbook_path": (
            "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml"
        ),
        "check_mode_playbook_path": (
            "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml"
        ),
        "inventory_hosts": ["host_110"],
        "domains": ["alert_chain_health", "alertmanager", "docker_container"],
        "keywords": [
            "alertchainbrokenalertmanager",
            "alertchainbroken_alertmanager",
            "alertmanager delivery",
        ],
        "activation_keywords": ["alertchainbrokenalertmanager"],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "medium",
        "canonical_asset_id": "container:host_110:alertmanager",
        "catalog_revision": "2026-07-16-alertmanager-exact-container-v1",
    },
    {
        "catalog_id": "ansible:111-ollama-fallback",
        "playbook_path": "infra/ansible/playbooks/111-ollama-fallback.yml",
        "check_mode_playbook_path": (
            "infra/ansible/playbooks/111-ollama-fallback.yml"
        ),
        "inventory_hosts": ["host_111", "host_120", "host_121"],
        "domains": ["ai_provider", "ollama", "host_launchagent"],
        "keywords": [
            "ollamalocalunavailable",
            "ollama111unavailable",
            "ollama-local",
            "ollama 111",
        ],
        "activation_keywords": [
            "ollamalocalunavailable",
            "ollama111unavailable",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "medium",
        "canonical_asset_id": "ai-provider:ollama_local",
        "catalog_revision": "2026-07-16-host111-launchagent-k3s-path-v2",
    },
    {
        "catalog_id": "ansible:110-host-pressure-readonly",
        "playbook_path": "infra/ansible/playbooks/110-host-pressure-readonly.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/110-host-pressure-readonly.yml",
        "inventory_hosts": ["host_110"],
        "domains": ["host_pressure", "gitea", "runner", "postgres", "systemd_control_plane"],
        "keywords": [
            "host110sustainedmoderatepressure",
            "host-pressure-controller",
            "sustained pressure",
            "host_resource",
            "gitea_service",
            "systemd_control_plane",
        ],
        "activation_keywords": [
            "host110sustainedmoderatepressure",
            "host-pressure-controller",
            "sustained pressure",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": False,
        "approval_required": False,
        "risk_level": "low",
        "no_write_only": True,
        "catalog_revision": "2026-07-14-host110-inputs-v2",
    },
    {
        "catalog_id": "ansible:110-disk-pressure",
        "playbook_path": (
            "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml"
        ),
        "check_mode_playbook_path": (
            "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml"
        ),
        "inventory_hosts": ["host_110"],
        "domains": ["disk_pressure", "docker_artifacts", "host_110"],
        "keywords": [
            "hostoutofdiskspace",
            "hostdiskusagehigh",
            "hostdiskusagecritical",
            "hostdiskwillfillin24hours",
            "disk_full",
            "node-exporter-110",
            "192.168.0.110",
        ],
        "activation_keyword_groups": [
            [
                "hostoutofdiskspace",
                "hostdiskusagehigh",
                "hostdiskusagecritical",
                "hostdiskwillfillin24hours",
            ],
            ["node-exporter-110", "192.168.0.110", "host_110", "host110"],
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "medium",
        "catalog_revision": "2026-07-14-disk-pressure-bounded-v1",
    },
    {
        "catalog_id": "ansible:188-disk-pressure",
        "playbook_path": (
            "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml"
        ),
        "check_mode_playbook_path": (
            "infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml"
        ),
        "inventory_hosts": ["host_188"],
        "domains": ["disk_pressure", "docker_artifacts", "host_188"],
        "keywords": [
            "hostoutofdiskspace",
            "hostdiskusagehigh",
            "hostdiskusagecritical",
            "hostdiskwillfillin24hours",
            "disk_full",
            "node-exporter-188",
            "192.168.0.188",
        ],
        "activation_keyword_groups": [
            [
                "hostoutofdiskspace",
                "hostdiskusagehigh",
                "hostdiskusagecritical",
                "hostdiskwillfillin24hours",
            ],
            ["node-exporter-188", "192.168.0.188", "host_188", "host188"],
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "medium",
        "catalog_revision": "2026-07-14-disk-pressure-bounded-v1",
    },
    {
        "catalog_id": "ansible:110-devops",
        "playbook_path": "infra/ansible/playbooks/110-node-exporter-repair.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/110-node-exporter-repair.yml",
        "inventory_hosts": ["host_110"],
        "domains": ["monitoring", "prometheus", "node_exporter", "host_110"],
        "keywords": [
            "node-exporter-110",
            "node exporter 110",
            "nodeexporterdown",
            "node_exporter",
            "node-exporter",
        ],
        "activation_keyword_groups": [
            [
                "nodeexporterdown",
                "nodeexporterscrapedown",
                "node_exporter_down",
            ],
            ["node-exporter-110", "192.168.0.110", "host_110", "host110"],
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "low",
        "catalog_revision": "2026-07-14-node-exporter-bounded-v5",
    },
    {
        "catalog_id": "ansible:110-sentry-profiling-consumer-recovery",
        "playbook_path": (
            "infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml"
        ),
        "check_mode_playbook_path": (
            "infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml"
        ),
        "inventory_hosts": ["host_110"],
        "domains": ["docker_container", "sentry", "host_110"],
        "keywords": [
            "dockercontainerunhealthy",
            "sentry-self-hosted-snuba-profiling-functions-consumer-1",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "medium",
        "canonical_asset_id": "service:sentry",
        "catalog_revision": "2026-07-15-sentry-exact-container-v1",
    },
    {
        "catalog_id": "ansible:110-devops-full-convergence",
        "playbook_path": "infra/ansible/playbooks/110-devops.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/110-devops.yml",
        "inventory_hosts": ["host_110"],
        "domains": [
            "swap",
            "harbor",
            "sentry",
            "gitea",
            "langfuse",
            "bitan",
            "runner",
            "keepalived",
            "nginx",
            "backup",
        ],
        "keywords": [
            "110-devops-full-convergence",
            "harbor",
            "sentry",
            "langfuse",
            "bitan",
            "keepalived",
            "github-runner",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": False,
        "approval_required": False,
        "risk_level": "high",
        "no_write_only": True,
        "catalog_revision": "2026-07-14-full-convergence-quarantined-v1",
    },
    {
        "catalog_id": "ansible:188-momo-backup-user",
        "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
        "inventory_hosts": ["host_188"],
        "domains": ["momo_backup", "postgresql", "cron", "awooop_notification"],
        "keywords": [
            "188",
            "momo",
            "momopostgresbackupfailed",
            "postgres",
            "postgresql",
            "backup",
            "momo pg_backup",
            "momo postgres backup",
            "pg_backup",
            "momo-pg-backup",
            "cron",
            "crontab",
        ],
        "activation_keywords": [
            "momo",
            "momopostgresbackupfailed",
            "backup",
            "momo pg_backup",
            "momo postgres backup",
            "pg_backup",
            "momo-pg-backup",
            "cron",
            "crontab",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "low",
    },
    {
        "catalog_id": "ansible:188-ai-web",
        "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/188-ai-web-readonly.yml",
        "inventory_hosts": ["host_188"],
        "domains": ["docker", "momo_backup", "signoz", "minio", "litellm", "n8n", "open_webui", "nginx"],
        "keywords": [
            "188",
            "docker",
            "container",
            "dockercontainerunhealthy",
            "momo",
            "backup",
            "postgresql",
            "pg_backup",
            "signoz",
            "minio",
            "litellm",
            "n8n",
            "open-webui",
            "openwebui",
            "docker-registry",
        ],
        "activation_keywords": [
            "dockercontainerunhealthy",
            "signoz",
            "minio",
            "litellm",
            "n8n",
            "open-webui",
            "openwebui",
            "docker-registry",
        ],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "medium",
    },
    {
        "catalog_id": "ansible:nginx-sync",
        "playbook_path": "infra/ansible/playbooks/nginx-sync.yml",
        "check_mode_playbook_path": "infra/ansible/playbooks/nginx-sync-readonly.yml",
        "inventory_hosts": ["host_188"],
        "domains": ["nginx", "proxy", "tls"],
        "keywords": ["nginx", "proxy", "tls", "cert", "502", "upstream"],
        "supports_check_mode": True,
        "auto_apply_enabled": True,
        "approval_required": False,
        "risk_level": "high",
    },
    {
        "catalog_id": "ansible:restore-password-auth",
        "playbook_path": "infra/ansible/playbooks/restore-password-auth.yml",
        "inventory_hosts": ["host_110", "host_120", "host_121", "host_188"],
        "domains": ["ssh", "password_auth"],
        "keywords": ["ssh", "passwordauthentication", "password auth", "login", "auth"],
        "supports_check_mode": False,
        "auto_apply_enabled": False,
        "approval_required": True,
        "risk_level": "high",
        "break_glass_required": True,
    },
)


def get_ansible_catalog_item(catalog_id: str) -> dict[str, Any] | None:
    """Return one repo-known Ansible catalog item without exposing mutability."""

    for item in _CATALOG:
        if item["catalog_id"] == catalog_id:
            return dict(item)
    return None


def list_ansible_catalog() -> tuple[dict[str, Any], ...]:
    """Return public-safe copies of the controlled Ansible catalog rows."""

    return tuple(dict(row) for row in _CATALOG)


def _get(row: dict[str, Any], key: str) -> Any:
    return row.get(key)


def _tags(row: dict[str, Any]) -> list[str]:
    raw = _get(row, "tags")
    if isinstance(raw, list):
        return [str(item).lower() for item in raw]
    if isinstance(raw, str):
        return [part.strip().lower() for part in raw.split(",") if part.strip()]
    return []


def _first_present(row: dict[str, Any], keys: tuple[str, ...]) -> Any:
    for key in keys:
        value = _get(row, key)
        if value not in (None, ""):
            return value
    return None


def _json_object(row: dict[str, Any], key: str) -> dict[str, Any]:
    raw = _get(row, key)
    if isinstance(raw, dict):
        return raw
    if isinstance(raw, str) and raw.strip():
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            return {}
        return parsed if isinstance(parsed, dict) else {}
    return {}


def _json_value(row: dict[str, Any], container_key: str, value_key: str) -> Any:
    return _json_object(row, container_key).get(value_key)


def _first_present_with_json(row: dict[str, Any], keys: tuple[str, ...], json_keys: tuple[tuple[str, str], ...] = ()) -> Any:
    value = _first_present(row, keys)
    if value not in (None, ""):
        return value
    for container_key, value_key in json_keys:
        value = _json_value(row, container_key, value_key)
        if value not in (None, ""):
            return value
    return None


def _bool_or_none(value: Any) -> bool | None:
    if isinstance(value, bool):
        return value
    if value in (None, ""):
        return None
    normalized = str(value).strip().lower()
    if normalized in {"1", "true", "yes", "y", "on"}:
        return True
    if normalized in {"0", "false", "no", "n", "off"}:
        return False
    return None


def _int_or_none(value: Any) -> int | None:
    if value in (None, ""):
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _is_controlled_apply_record(row: dict[str, Any]) -> bool:
    if not row:
        return False
    if row.get("approval_source"):
        return True
    if str(row.get("execution_mode") or "").strip().lower() == "controlled_apply":
        return True
    if "controlled_apply" in _tags(row):
        return True
    return str(row.get("actor") or "").strip().lower() == "ansible_controlled_apply_worker"


def _is_ansible_operation(row: dict[str, Any]) -> bool:
    operation_type = str(_get(row, "operation_type") or "").lower()
    if operation_type in ANSIBLE_OPERATION_TYPES:
        return True
    if "ansible" in _tags(row):
        return True
    executor = str(
        _first_present(
            row,
            (
                "input_executor",
                "input_execution_backend",
                "output_executor",
                "output_execution_backend",
            ),
        )
        or ""
    ).lower()
    if executor == "ansible":
        return True
    playbook_path = str(
        _first_present(row, ("input_playbook_path", "output_playbook_path", "input_ansible_playbook_path", "output_ansible_playbook_path"))
        or ""
    ).lower()
    return "infra/ansible/" in playbook_path or playbook_path.endswith(".yml") and "ansible" in playbook_path


def _ansible_record(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "op_id": _get(row, "op_id"),
        "parent_op_id": _get(row, "parent_op_id"),
        "operation_type": _get(row, "operation_type"),
        "status": _get(row, "status"),
        "actor": _get(row, "actor"),
        "catalog_id": _first_present(row, ("input_catalog_id", "output_catalog_id")),
        "playbook_id": _first_present(row, ("input_playbook_id", "output_playbook_id")),
        "playbook_path": _first_present(
            row,
            ("input_playbook_path", "output_playbook_path", "input_ansible_playbook_path", "output_ansible_playbook_path"),
        ),
        "execution_mode": _first_present(row, ("input_execution_mode", "output_execution_mode")),
        "check_mode": _first_present(row, ("input_check_mode", "output_check_mode")),
        "check_mode_executed": _first_present_with_json(
            row,
            ("input_check_mode_executed", "output_check_mode_executed", "dry_run_check_mode_executed"),
            (("dry_run_result", "check_mode_executed"),),
        ),
        "apply_enabled": _first_present(row, ("input_apply_enabled", "output_apply_enabled")),
        "apply_executed": _first_present_with_json(
            row,
            ("input_apply_executed", "output_apply_executed", "dry_run_apply_executed"),
            (("dry_run_result", "apply_executed"),),
        ),
        "approval_source": _first_present(row, ("input_approval_source", "output_approval_source")),
        "returncode": _first_present_with_json(
            row,
            ("output_returncode", "input_returncode", "dry_run_returncode"),
            (("dry_run_result", "returncode"),),
        ),
        "not_used_reason": _first_present(row, ("input_not_used_reason", "output_not_used_reason")),
        "dry_run_result": _get(row, "dry_run_result"),
        "error": _get(row, "error"),
        "duration_ms": _get(row, "duration_ms"),
        "tags": _get(row, "tags"),
        "created_at": _get(row, "created_at"),
    }


def summarize_ansible_execution(records: list[dict[str, Any]]) -> dict[str, Any]:
    """Summarize durable Ansible audit rows for Telegram and Operator UI."""
    typed_records = [row for row in records if isinstance(row, dict)]
    check_mode_total = 0
    apply_total = 0
    rollback_total = 0
    pending_check_mode_total = 0
    applied_success_total = 0

    terminal_check_mode_parent_ids = {
        str(row.get("parent_op_id"))
        for row in typed_records
        if str(row.get("operation_type") or "") in {
            "ansible_check_mode_executed",
            "ansible_execution_skipped",
        }
        and row.get("parent_op_id")
    }

    for row in typed_records:
        operation_type = str(row.get("operation_type") or "")
        status = str(row.get("status") or "").lower()
        if operation_type == "ansible_check_mode_executed" and status != "pending":
            check_mode_total += 1
        elif operation_type == "ansible_apply_executed":
            apply_total += 1
            if status in {"success", "completed", "executed", "applied"}:
                applied_success_total += 1
        elif operation_type == "ansible_rollback_executed":
            rollback_total += 1
        elif (
            operation_type == "ansible_candidate_matched"
            and str(row.get("op_id")) not in terminal_check_mode_parent_ids
        ):
            pending_check_mode_total += 1

    latest_record = typed_records[0] if typed_records else {}
    latest_apply = next(
        (row for row in typed_records if row.get("operation_type") == "ansible_apply_executed"),
        {},
    )
    latest_check = next(
        (row for row in typed_records if row.get("operation_type") == "ansible_check_mode_executed"),
        {},
    )
    focused = latest_apply or latest_check or latest_record
    returncode = _int_or_none(focused.get("returncode"))
    approval_source = focused.get("approval_source")

    return {
        "check_mode_total": check_mode_total,
        "apply_total": apply_total,
        "rollback_total": rollback_total,
        "pending_check_mode_total": pending_check_mode_total,
        "applied_success_total": applied_success_total,
        "applied": applied_success_total > 0,
        "controlled_apply": bool(latest_apply) and _is_controlled_apply_record(latest_apply),
        "latest_operation_type": focused.get("operation_type"),
        "latest_status": focused.get("status"),
        "latest_catalog_id": focused.get("catalog_id"),
        "latest_playbook_path": focused.get("playbook_path"),
        "latest_execution_mode": focused.get("execution_mode"),
        "latest_check_mode": _bool_or_none(focused.get("check_mode")),
        "latest_check_mode_executed": _bool_or_none(focused.get("check_mode_executed")),
        "latest_apply_enabled": _bool_or_none(focused.get("apply_enabled")),
        "latest_apply_executed": _bool_or_none(focused.get("apply_executed")),
        "latest_returncode": returncode if returncode is not None else focused.get("returncode"),
        "approval_source": approval_source,
        "latest_actor": focused.get("actor"),
        "latest_op_id": focused.get("op_id"),
        "latest_parent_op_id": focused.get("parent_op_id"),
    }


def _flatten_text(value: Any, pieces: list[str], remaining: int = 80) -> int:
    if remaining <= 0 or value is None:
        return remaining
    if isinstance(value, dict):
        for key, item in value.items():
            remaining = _flatten_text(key, pieces, remaining)
            remaining = _flatten_text(item, pieces, remaining)
            if remaining <= 0:
                break
        return remaining
    if isinstance(value, list):
        for item in value:
            remaining = _flatten_text(item, pieces, remaining)
            if remaining <= 0:
                break
        return remaining
    pieces.append(str(value).lower())
    return remaining - 1


def _source_haystack(incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> str:
    pieces: list[str] = []
    _flatten_text(incident, pieces)
    _flatten_text(drift, pieces)
    return " ".join(pieces)


def _domain_controlled_executor_route(
    incident: dict[str, Any] | None,
) -> dict[str, Any] | None:
    """Keep host control-plane alerts out of the generic Ansible catalog."""

    if not isinstance(incident, dict):
        return None
    signals = [
        signal
        for signal in incident.get("signals") or []
        if isinstance(signal, dict)
    ]
    first_signal = signals[0] if signals else {}
    labels = (
        first_signal.get("labels")
        if isinstance(first_signal.get("labels"), dict)
        else {}
    )
    alertname = str(
        incident.get("alertname")
        or first_signal.get("alert_name")
        or labels.get("alertname")
        or ""
    )
    affected_services = [
        str(value)
        for value in incident.get("affected_services") or []
        if str(value).strip()
    ]
    target_resource = str(
        affected_services[0]
        if affected_services
        else labels.get("component")
        or labels.get("target_resource")
        or ""
    )
    namespace = str(labels.get("namespace") or "")
    return resolve_controlled_alert_target(
        alertname=alertname,
        target_resource=target_resource,
        namespace=namespace,
    )


def _catalog_hints(incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> dict[str, Any]:
    typed_route = resolve_typed_incident_target(incident)
    allowed_catalog_ids = {
        str(value)
        for value in typed_route.get("allowed_catalog_ids") or []
        if str(value)
    }
    candidates: list[dict[str, Any]] = []
    unmatched: list[str] = []
    for item in _CATALOG:
        catalog_id = str(item["catalog_id"])
        if catalog_id not in allowed_catalog_ids:
            unmatched.append(catalog_id)
            continue
        public_item = {
            key: value
            for key, value in item.items()
            if key
            in {
                "catalog_id",
                "playbook_path",
                "inventory_hosts",
                "domains",
                "supports_check_mode",
                "check_mode_playbook_path",
                "auto_apply_enabled",
                "approval_required",
                "risk_level",
                "no_write_only",
                "catalog_revision",
            }
        }
        allowed_hosts = {
            str(value)
            for value in typed_route.get("allowed_inventory_hosts") or []
            if str(value)
        }
        catalog_hosts = {
            str(value)
            for value in item.get("inventory_hosts") or []
            if str(value)
        }
        if not catalog_hosts or catalog_hosts != allowed_hosts:
            unmatched.append(catalog_id)
            continue
        candidates.append(
            {
                **public_item,
                "match_score": 100,
                "matched_keywords": ["typed_domain_exact_match"],
                "semantic_match": True,
                "matched_activation_keywords": [],
                "matched_activation_keyword_groups": [],
                "canonical_asset_id": typed_route.get("canonical_asset_id"),
                "typed_domain": typed_route.get("target_kind"),
                "target_scope_validated": True,
            }
        )

    candidates.sort(key=lambda row: str(row["catalog_id"]))
    resolution_status = str(typed_route.get("resolution_status") or "")
    if resolution_status == "asset_identity_unresolved":
        decision_effect = "create_asset_drift_work_item"
        not_used_reason = "asset_identity_unresolved_no_fallback"
    elif typed_route.get("executor") != "host_ansible_executor":
        decision_effect = "delegate_to_domain_executor"
        not_used_reason = "typed_non_ansible_domain_route"
    elif not candidates:
        decision_effect = "create_domain_playbook_work_item"
        not_used_reason = "typed_domain_has_no_exact_bounded_playbook"
    else:
        decision_effect = "bounded_domain_candidate"
        not_used_reason = None

    result = {
        "match_mode": "typed_domain_router_v2",
        "decision_effect": decision_effect,
        "available_count": len(_CATALOG),
        "candidates": candidates,
        "unmatched_catalog_ids": sorted(set(unmatched)),
        "typed_target_route": typed_route,
        "cross_domain_fallback_allowed": False,
    }
    if not_used_reason:
        result["not_used_reason"] = not_used_reason
    if decision_effect == "delegate_to_domain_executor":
        result["controlled_executor"] = typed_route
    if resolution_status == "asset_identity_unresolved":
        result["drift_work_item_id"] = typed_route.get("drift_work_item_id")
    return result


def _legacy_catalog_hints(
    incident: dict[str, Any] | None,
    drift: dict[str, Any] | None,
) -> dict[str, Any]:
    """Retained for offline comparison only; runtime routing uses v2 above."""

    controlled_route = _domain_controlled_executor_route(incident)
    if controlled_route is not None:
        return {
            "match_mode": "domain_controlled_executor_route_v1",
            "decision_effect": "delegate_to_domain_executor",
            "available_count": len(_CATALOG),
            "candidates": [],
            "unmatched_catalog_ids": [str(item["catalog_id"]) for item in _CATALOG],
            "controlled_executor": controlled_route,
            "not_used_reason": "non_kubernetes_control_plane_route",
        }

    haystack = _source_haystack(incident, drift)
    candidates: list[dict[str, Any]] = []
    unmatched: list[str] = []
    for item in _CATALOG:
        matched = [keyword for keyword in item["keywords"] if keyword in haystack]
        activation_keywords = item.get("activation_keywords") or []
        matched_activation_keywords = [
            keyword for keyword in activation_keywords if keyword in haystack
        ]
        activation_keyword_groups = item.get("activation_keyword_groups") or []
        matched_activation_keyword_groups = [
            [keyword for keyword in group if keyword in haystack]
            for group in activation_keyword_groups
        ]
        semantic_match = (
            (not activation_keywords or bool(matched_activation_keywords))
            and all(matched_activation_keyword_groups)
        )
        public_item = {
            key: value
            for key, value in item.items()
            if key
            in {
                "catalog_id",
                "playbook_path",
                "inventory_hosts",
                "domains",
                "supports_check_mode",
                "check_mode_playbook_path",
                "auto_apply_enabled",
                "approval_required",
                "risk_level",
                "no_write_only",
            }
        }
        if matched and semantic_match:
            candidates.append({
                **public_item,
                "match_score": len(matched),
                "matched_keywords": matched,
                "semantic_match": True,
                "matched_activation_keywords": matched_activation_keywords,
                "matched_activation_keyword_groups": (
                    matched_activation_keyword_groups
                ),
            })
        else:
            unmatched.append(item["catalog_id"])
    candidates.sort(key=lambda row: (-int(row["match_score"]), str(row["catalog_id"])))
    return {
        "match_mode": "static_catalog_keyword_hint_v1",
        "decision_effect": "none",
        "available_count": len(_CATALOG),
        "candidates": candidates,
        "unmatched_catalog_ids": unmatched,
    }


def get_ansible_catalog_candidates(
    incident: dict[str, Any] | None,
    drift: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Return current semantically matched catalog rows for replay routing."""

    hints = _catalog_hints(incident, drift)
    candidates = hints.get("candidates")
    return [dict(row) for row in candidates or [] if isinstance(row, dict)]


def build_ansible_truth(
    automation_ops: list[dict[str, Any]],
    *,
    incident: dict[str, Any] | None,
    drift: dict[str, Any] | None,
) -> dict[str, Any]:
    """Build the truth-chain Ansible section from audited facts and catalog hints."""

    records = [_ansible_record(row) for row in automation_ops if _is_ansible_operation(row)]
    summary = summarize_ansible_execution(records)
    candidate_catalog = _catalog_hints(incident, drift)
    return {
        "considered": bool(records),
        "records": records,
        "summary": summary,
        "audit_contract": {
            "schema_version": "ansible_executor_audit_v1",
            "operation_types": sorted(ANSIBLE_OPERATION_TYPES),
            "required_audit_fields": [
                "incident_id",
                "operation_type",
                "status",
                "actor",
                "input.executor",
                "input.catalog_id",
                "input.execution_mode",
                "input.playbook_path",
                "input.check_mode",
                "input.approval_source",
                "output.returncode",
                "output.not_used_reason",
                "dry_run_result",
            ],
            "default_execution_mode": "catalog/dry-run audit only until approval execution is explicitly wired",
        },
        "candidate_catalog": candidate_catalog,
        "not_used_reason": (
            None
            if records
            else candidate_catalog.get("not_used_reason")
            or "no automation_operation_log row with Ansible operation type, tag, or executor backend for this source"
        ),
    }


def _incident_public_dict(incident: Any) -> dict[str, Any]:
    if incident is None:
        return {}
    if isinstance(incident, dict):
        return incident
    severity = getattr(incident, "severity", None)
    signals_payload: list[dict[str, Any]] = []
    for signal in getattr(incident, "signals", None) or []:
        signals_payload.append({
            "alert_name": getattr(signal, "alert_name", None),
            "labels": getattr(signal, "labels", None) or {},
            "annotations": getattr(signal, "annotations", None) or {},
        })
    return {
        "incident_id": getattr(incident, "incident_id", None),
        "project_id": getattr(incident, "project_id", None),
        "status": getattr(getattr(incident, "status", None), "value", None),
        "alertname": getattr(incident, "alertname", None),
        "alert_category": getattr(incident, "alert_category", None),
        "notification_type": getattr(incident, "notification_type", None),
        "severity": getattr(severity, "value", severity),
        "affected_services": getattr(incident, "affected_services", None) or [],
        "signals": signals_payload,
    }


_CANONICAL_INCIDENT_ID_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,30}$")
_CANONICAL_PROJECT_ID_RE = re.compile(r"^[A-Za-z0-9_.:-]{1,64}$")


def _canonical_incident_status(value: Any) -> str:
    normalized = str(getattr(value, "value", value) or "").strip().upper()
    return (
        normalized
        if normalized
        in {"INVESTIGATING", "MITIGATING", "RESOLVED", "CLOSED", "ESCALATED"}
        else "INVESTIGATING"
    )


def _canonical_incident_severity(value: Any) -> str:
    normalized = str(getattr(value, "value", value) or "").strip().upper()
    return {
        "CRITICAL": "P0",
        "HIGH": "P1",
        "WARNING": "P2",
        "MEDIUM": "P2",
        "LOW": "P3",
        "INFO": "P3",
    }.get(
        normalized,
        normalized if normalized in {"P0", "P1", "P2", "P3"} else "P2",
    )


def _canonical_incident_ledger_params(
    incident_payload: dict[str, Any],
    *,
    automation_run_id: str,
) -> dict[str, Any] | None:
    incident_id = str(incident_payload.get("incident_id") or "").strip()
    project_id = str(incident_payload.get("project_id") or "awoooi").strip()
    if (
        _CANONICAL_INCIDENT_ID_RE.fullmatch(incident_id) is None
        or _CANONICAL_PROJECT_ID_RE.fullmatch(project_id) is None
    ):
        return None

    signals = [
        signal
        for signal in incident_payload.get("signals") or []
        if isinstance(signal, dict)
    ][:20]
    affected_services = [
        str(value)[:100]
        for value in incident_payload.get("affected_services") or []
        if str(value).strip()
    ][:50]
    notification_type = str(
        incident_payload.get("notification_type") or ""
    ).strip()
    if len(notification_type) > 10:
        notification_type = ""
    decision_chain = {
        "schema_version": "ansible_candidate_incident_ingress_v1",
        "trace_id": str(
            incident_payload.get("trace_id") or automation_run_id
        ),
        "run_id": str(incident_payload.get("run_id") or automation_run_id),
        "work_item_id": str(incident_payload.get("work_item_id") or ""),
        "source_receipt_ref": str(
            incident_payload.get("source_receipt_ref") or ""
        ),
        "asset_scope_aliases": [
            str(value)[:100]
            for value in incident_payload.get("asset_scope_aliases") or []
            if str(value).strip()
        ][:20],
        "normalized_asset_identity": True,
        "candidate_runtime_apply_executed": False,
    }
    return {
        "incident_id": incident_id,
        "project_id": project_id,
        "status": _canonical_incident_status(incident_payload.get("status")),
        "severity": _canonical_incident_severity(
            incident_payload.get("severity")
        ),
        "signals": json.dumps(signals, ensure_ascii=False),
        "affected_services": json.dumps(
            affected_services,
            ensure_ascii=False,
        ),
        "decision_chain": json.dumps(decision_chain, ensure_ascii=False),
        "alertname": str(incident_payload.get("alertname") or "")[:100]
        or None,
        "notification_type": notification_type or None,
        "alert_category": str(
            incident_payload.get("alert_category") or ""
        )[:50]
        or None,
    }


async def _ensure_ansible_incident_ledger_with_db(
    db: Any,
    incident_payload: dict[str, Any],
    *,
    automation_run_id: str,
) -> dict[str, bool]:
    params = _canonical_incident_ledger_params(
        incident_payload,
        automation_run_id=automation_run_id,
    )
    if params is None:
        return {"ready": False, "inserted": False}
    result = await db.execute(
        text("""
            WITH inserted AS (
                INSERT INTO incidents (
                    incident_id,
                    project_id,
                    status,
                    severity,
                    signals,
                    affected_services,
                    decision_chain,
                    proposal_ids,
                    alertname,
                    notification_type,
                    alert_category,
                    created_at,
                    updated_at,
                    ttl_days,
                    vectorized
                ) VALUES (
                    :incident_id,
                    :project_id,
                    CAST(:status AS incidentstatus),
                    CAST(:severity AS severity),
                    CAST(:signals AS json),
                    CAST(:affected_services AS json),
                    CAST(:decision_chain AS json),
                    CAST('[]' AS json),
                    :alertname,
                    :notification_type,
                    :alert_category,
                    NOW(),
                    NOW(),
                    7,
                    FALSE
                )
                ON CONFLICT (incident_id) DO NOTHING
                RETURNING incident_id
            )
            SELECT
                EXISTS (SELECT 1 FROM inserted) AS inserted,
                EXISTS (
                    SELECT 1
                    FROM incidents incident
                    WHERE incident.incident_id = :incident_id
                      AND incident.project_id = :project_id
                ) AS ready
        """),
        params,
    )
    row = result.mappings().one_or_none()
    return {
        "ready": bool(row and row.get("ready") is True),
        "inserted": bool(row and row.get("inserted") is True),
    }


async def ensure_ansible_incident_ledger(
    *,
    incident: Any,
    automation_run_id: str,
    project_id: str | None = None,
) -> dict[str, bool]:
    """Ensure a controlled candidate has a durable canonical incident row."""

    incident_payload = dict(_incident_public_dict(incident))
    if project_id:
        incident_payload["project_id"] = project_id
    effective_project_id = str(
        incident_payload.get("project_id") or "awoooi"
    )
    async with get_db_context(effective_project_id) as db:
        return await _ensure_ansible_incident_ledger_with_db(
            db,
            incident_payload,
            automation_run_id=automation_run_id,
        )


def build_ansible_decision_audit_payload(
    *,
    incident: Any,
    proposal_data: dict[str, Any],
    decision_path: str,
    not_used_reason: str,
) -> dict[str, Any] | None:
    """Return an AOL payload when Ansible has catalog candidates for a decision."""

    incident_payload = _incident_public_dict(incident)
    hints = _catalog_hints(incident_payload, None)
    candidates = hints.get("candidates") or []
    if not candidates:
        return None

    incident_id = str(incident_payload.get("incident_id") or "")
    project_id = str(incident_payload.get("project_id") or "awoooi")
    affected_services = [
        str(value)
        for value in incident_payload.get("affected_services") or []
        if str(value).strip()
    ]
    namespaces = sorted({
        str((signal.get("labels") or {}).get("namespace") or "").strip()
        for signal in incident_payload.get("signals") or []
        if isinstance(signal, dict)
        and str((signal.get("labels") or {}).get("namespace") or "").strip()
    })
    proposal_risk_level = str(proposal_data.get("risk_level") or "").strip().lower()
    execution_priority = proposal_data.get("execution_priority", 100)
    if not isinstance(execution_priority, int) or isinstance(execution_priority, bool):
        execution_priority = 100
    execution_priority = max(0, min(execution_priority, 100))
    typed_target_route = hints["typed_target_route"]
    executor_candidates = [
        {
            "catalog_id": row["catalog_id"],
            "playbook_path": row["playbook_path"],
            "check_mode_playbook_path": row.get("check_mode_playbook_path"),
            "inventory_hosts": row["inventory_hosts"],
            "supports_check_mode": row["supports_check_mode"],
            "auto_apply_enabled": row["auto_apply_enabled"],
            "approval_required": row["approval_required"],
            "risk_level": row["risk_level"],
            "catalog_revision": row.get("catalog_revision"),
            "canonical_asset_id": row.get("canonical_asset_id"),
            "typed_domain": row.get("typed_domain"),
            "target_scope_validated": row.get("target_scope_validated"),
            "match_score": row["match_score"],
            "matched_keywords": row["matched_keywords"],
            "semantic_match": row["semantic_match"],
            "matched_activation_keywords": row[
                "matched_activation_keywords"
            ],
        }
        for row in candidates[:5]
    ]
    target_route_generation = build_typed_target_route_generation(
        typed_target_route=typed_target_route,
        executor_candidates=executor_candidates,
    )
    source_recurrence = _source_recurrence_receipt(
        incident_payload,
        proposal_data,
    )
    occurrence_identity = "|".join(
        str(source_recurrence.get(key) or "")
        for key in ("fingerprint", "occurrence_id", "observed_at")
    )
    occurrence_generation = hashlib.sha256(
        occurrence_identity.encode("utf-8")
    ).hexdigest()[:20]
    idempotency_key = (
        f"ansible-candidate:{project_id}:{incident_id}:"
        f"{target_route_generation}:{occurrence_generation}"
    )
    input_payload = {
        "incident_id": incident_id,
        "project_id": project_id,
        "trace_id": str(incident_payload.get("trace_id") or ""),
        "run_id": str(incident_payload.get("run_id") or ""),
        "work_item_id": str(incident_payload.get("work_item_id") or ""),
        "run_namespace": str(incident_payload.get("run_namespace") or ""),
        "source_receipt_ref": str(incident_payload.get("source_receipt_ref") or ""),
        "asset_scope_aliases": [
            str(alias)
            for alias in incident_payload.get("asset_scope_aliases") or []
            if isinstance(alias, str) and alias
        ][:20],
        "executor": "ansible",
        "execution_backend": "ansible",
        "single_writer_executor": "awoooi-ansible-executor-broker",
        "execution_priority": execution_priority,
        "decision_path": decision_path,
        "idempotency_key": idempotency_key,
        "candidate_lane_lock_key": f"ansible-candidate:{project_id}:{incident_id}",
        "target_route_generation": target_route_generation,
        "source_occurrence_generation": occurrence_generation,
        "source_recurrence": source_recurrence,
        "router_source_sha": os.getenv("AWOOOI_BUILD_COMMIT_SHA", "")
        .strip()
        .lower(),
        "check_mode": True,
        "apply_enabled": False,
        "approval_required": proposal_risk_level == "critical",
        "candidate_catalog_schema": hints["match_mode"],
        "typed_target_route": typed_target_route,
        "cross_domain_fallback_allowed": False,
        "executor_candidates": executor_candidates,
        "proposal_source": proposal_data.get("source", ""),
        "approval_id": str(proposal_data.get("approval_id") or ""),
        "proposal_risk_level": proposal_risk_level,
        "proposal_action_preview": str(
            proposal_data.get("action")
            or proposal_data.get("kubectl_command")
            or ""
        )[:240],
        "target_selector": {
            "schema_version": "ai_decision_target_selector_v2",
            "project_id": project_id,
            "incident_id": incident_id,
            "canonical_asset_id": typed_target_route.get("canonical_asset_id"),
            "typed_domain": typed_target_route.get("target_kind"),
            "resolution_status": typed_target_route.get("resolution_status"),
            "affected_services": affected_services,
            "namespaces": namespaces,
            "catalog_ids": [str(row["catalog_id"]) for row in candidates[:5]],
            "inventory_hosts": sorted({
                str(host)
                for row in candidates[:5]
                for host in row.get("inventory_hosts") or []
            }),
            "cross_domain_fallback_allowed": False,
        },
        "source_truth_diff": {
            "required_before_apply": True,
            "producer": "awooop_ansible_check_mode_worker",
            "receipt_status": "pending_check_mode",
        },
        "risk_policy_decision": {
            "risk_level": proposal_risk_level,
            "controlled_apply_candidate": proposal_risk_level in {"low", "medium", "high"},
            "critical_break_glass_required": proposal_risk_level == "critical",
        },
    }
    controlled_queue = decision_path == "repair_candidate_controlled_queue"
    output_payload = {
        "not_used_reason": not_used_reason,
        "decision_effect": "check_mode_queue_ready" if controlled_queue else "audit_only",
        "next_required_step": (
            "awooop_ansible_check_mode_worker_claims_candidate"
            if controlled_queue
            else "wire approval_execution to Ansible check-mode before apply"
        ),
    }
    return {
        "operation_type": "ansible_candidate_matched",
        "status": "dry_run",
        "input": input_payload,
        "output": output_payload,
        "dry_run_result": {
            "check_mode_executed": False,
            "candidate_count": len(candidates),
            "reason": not_used_reason,
        },
        "tags": ["ansible", "decision", "candidate", "check_mode_pending"],
    }


async def preflight_ansible_candidate_generation(
    *,
    incident: Any,
    proposal_data: dict[str, Any],
    decision_path: str,
    not_used_reason: str,
    project_id: str = "awoooi",
) -> dict[str, Any]:
    """Avoid expensive evidence collection for an already-current generation."""

    payload = build_ansible_decision_audit_payload(
        incident=incident,
        proposal_data=proposal_data,
        decision_path=decision_path,
        not_used_reason=not_used_reason,
    )
    if payload is None:
        return {
            "status": "no_allowlisted_ansible_candidate",
            "should_collect_evidence": False,
        }
    incident_id = str(payload["input"].get("incident_id") or "")
    generation = str(
        payload["input"].get("target_route_generation") or ""
    )
    occurrence = str(
        payload["input"].get("source_occurrence_generation") or ""
    )
    source_recurrence = payload["input"].get("source_recurrence") or {}
    proposal_source = str(payload["input"].get("proposal_source") or "")
    if (
        proposal_source in _RECURRENCE_REQUIRED_PROPOSAL_SOURCES
        and source_recurrence.get("verified") is not True
    ):
        return {
            "status": "current_source_recurrence_not_verified",
            "should_collect_evidence": False,
            "target_route_generation": generation,
        }
    terminal_sql = verified_ansible_candidate_terminal_sql("candidate")
    try:
        async with get_db_context(project_id) as db:
            result = await db.execute(
                text(f"""
                    SELECT
                        candidate.op_id::text AS op_id,
                        candidate.input,
                        ({terminal_sql}) AS is_terminal,
                        EXISTS (
                            SELECT 1
                            FROM automation_operation_log child
                            WHERE child.parent_op_id = candidate.op_id
                              AND child.status IN ('pending', 'running')
                        ) OR EXISTS (
                            SELECT 1
                            FROM automation_operation_log check_mode
                            JOIN automation_operation_log apply
                              ON apply.parent_op_id = check_mode.op_id
                             AND apply.operation_type = 'ansible_apply_executed'
                             AND apply.status IN ('pending', 'running')
                            WHERE check_mode.parent_op_id = candidate.op_id
                              AND check_mode.operation_type
                                    = 'ansible_check_mode_executed'
                        ) AS has_pending_child,
                        EXISTS (
                            SELECT 1
                            FROM automation_operation_log check_mode
                            JOIN automation_operation_log apply
                              ON apply.parent_op_id = check_mode.op_id
                             AND apply.operation_type = 'ansible_apply_executed'
                            WHERE check_mode.parent_op_id = candidate.op_id
                              AND check_mode.operation_type
                                    = 'ansible_check_mode_executed'
                        ) AS has_apply
                    FROM automation_operation_log candidate
                    WHERE candidate.operation_type = 'ansible_candidate_matched'
                      AND coalesce(
                            candidate.incident_id::text,
                            candidate.input ->> 'incident_id'
                          ) = :incident_id
                      AND candidate.input ->> 'executor' = 'ansible'
                      AND candidate.input ->> 'decision_path'
                            = 'repair_candidate_controlled_queue'
                    ORDER BY candidate.created_at DESC, candidate.op_id DESC
                    LIMIT 1
                """),
                {"incident_id": incident_id},
            )
            latest = result.mappings().first()
    except Exception as exc:
        logger.warning(
            "ansible_candidate_generation_preflight_failed",
            incident_id=incident_id,
            error_type=type(exc).__name__,
        )
        return {
            "status": "generation_preflight_unavailable",
            "should_collect_evidence": False,
        }

    if latest is None:
        return {
            "status": "missing_candidate_generation",
            "should_collect_evidence": True,
            "target_route_generation": generation,
        }
    latest_input = _json_object(dict(latest), "input")
    same_generation = str(
        latest_input.get("target_route_generation") or ""
    ) == generation
    same_occurrence = str(
        latest_input.get("source_occurrence_generation") or ""
    ) == occurrence
    is_terminal = latest.get("is_terminal") is True
    if same_generation and (not is_terminal or same_occurrence):
        return {
            "status": "existing_current_generation",
            "should_collect_evidence": False,
            "target_route_generation": generation,
            "existing_candidate_op_id": latest.get("op_id"),
        }
    if not is_terminal and latest.get("has_apply") is True:
        return {
            "status": "generation_reconciliation_blocked_has_apply",
            "should_collect_evidence": False,
            "target_route_generation": generation,
            "existing_candidate_op_id": latest.get("op_id"),
        }
    if not is_terminal and latest.get("has_pending_child") is True:
        return {
            "status": "generation_reconciliation_deferred_pending_child",
            "should_collect_evidence": False,
            "target_route_generation": generation,
            "existing_candidate_op_id": latest.get("op_id"),
        }
    return {
        "status": (
            "fresh_occurrence_same_generation"
            if same_generation
            else "generation_reconciliation_required"
        ),
        "should_collect_evidence": True,
        "target_route_generation": generation,
        "existing_candidate_op_id": latest.get("op_id"),
    }


async def record_ansible_decision_audit(
    *,
    incident: Any,
    proposal_data: dict[str, Any],
    decision_path: str,
    not_used_reason: str,
    automation_run_id: str | None = None,
) -> bool:
    """Write a best-effort Ansible candidate audit row for one decision."""

    payload = build_ansible_decision_audit_payload(
        incident=incident,
        proposal_data=proposal_data,
        decision_path=decision_path,
        not_used_reason=not_used_reason,
    )
    if payload is None:
        return False

    incident_id = payload["input"]["incident_id"]
    project_id = str(payload["input"].get("project_id") or "awoooi")
    try:
        candidate_op_id = str(UUID(automation_run_id)) if automation_run_id else str(uuid4())
    except (TypeError, ValueError):
        candidate_op_id = str(uuid4())
    payload["input"]["automation_run_id"] = candidate_op_id
    incident_payload = dict(_incident_public_dict(incident))
    incident_payload["project_id"] = project_id
    terminal_candidate_sql = verified_ansible_candidate_terminal_sql("candidate")
    target_route_generation = str(
        payload["input"].get("target_route_generation") or ""
    )
    source_recurrence = payload["input"].get("source_recurrence") or {}
    proposal_source = str(payload["input"].get("proposal_source") or "")
    if (
        proposal_source in _RECURRENCE_REQUIRED_PROPOSAL_SOURCES
        and source_recurrence.get("verified") is not True
    ):
        logger.warning(
            "ansible_candidate_write_blocked",
            incident_id=incident_id,
            reason="current_source_recurrence_not_verified",
        )
        return False
    try:
        async with get_db_context(str(project_id)) as db:
            await db.execute(
                text("""
                    SELECT pg_advisory_xact_lock(
                        hashtextextended(:idempotency_key, 0)
                    )
                """),
                {
                    "idempotency_key": payload["input"][
                        "candidate_lane_lock_key"
                    ]
                },
            )
            incident_ledger = await _ensure_ansible_incident_ledger_with_db(
                db,
                incident_payload,
                automation_run_id=candidate_op_id,
            )
            if incident_ledger["ready"] is not True:
                logger.warning(
                    "ansible_candidate_incident_ledger_missing",
                    incident_id=incident_id,
                    project_id=project_id,
                    automation_run_id=candidate_op_id,
                )
                return False
            latest_result = await db.execute(
                text(f"""
                    SELECT
                        candidate.op_id::text AS op_id,
                        candidate.input,
                        ({terminal_candidate_sql}) AS is_terminal,
                        EXISTS (
                            SELECT 1
                            FROM automation_operation_log child
                            WHERE child.parent_op_id = candidate.op_id
                              AND child.status IN ('pending', 'running')
                        ) OR EXISTS (
                            SELECT 1
                            FROM automation_operation_log check_mode
                            JOIN automation_operation_log apply
                              ON apply.parent_op_id = check_mode.op_id
                             AND apply.operation_type = 'ansible_apply_executed'
                             AND apply.status IN ('pending', 'running')
                            WHERE check_mode.parent_op_id = candidate.op_id
                              AND check_mode.operation_type
                                    = 'ansible_check_mode_executed'
                        ) AS has_pending_child,
                        EXISTS (
                            SELECT 1
                            FROM automation_operation_log check_mode
                            JOIN automation_operation_log apply
                              ON apply.parent_op_id = check_mode.op_id
                             AND apply.operation_type = 'ansible_apply_executed'
                            WHERE check_mode.parent_op_id = candidate.op_id
                              AND check_mode.operation_type
                                    = 'ansible_check_mode_executed'
                        ) AS has_apply
                    FROM automation_operation_log candidate
                    WHERE candidate.operation_type = 'ansible_candidate_matched'
                      AND coalesce(
                          candidate.incident_id::text,
                          candidate.input ->> 'incident_id'
                      ) = :incident_id
                      AND candidate.input ->> 'executor' = 'ansible'
                      AND candidate.input ->> 'decision_path'
                          = 'repair_candidate_controlled_queue'
                    ORDER BY candidate.created_at DESC, candidate.op_id DESC
                    LIMIT 1
                """),
                {"incident_id": incident_id},
            )
            latest = latest_result.mappings().first()
            if latest is not None:
                latest_input = _json_object(dict(latest), "input")
                latest_generation = str(
                    latest_input.get("target_route_generation") or ""
                )
                latest_is_terminal = latest.get("is_terminal") is True
                same_generation = latest_generation == target_route_generation
                same_occurrence = str(
                    latest_input.get("source_occurrence_generation") or ""
                ) == str(
                    payload["input"].get("source_occurrence_generation") or ""
                )
                if same_generation and (
                    not latest_is_terminal or same_occurrence
                ):
                    return False
                if source_recurrence.get("verified") is not True:
                    logger.warning(
                        "ansible_typed_route_supersession_blocked",
                        incident_id=incident_id,
                        reason="current_source_recurrence_not_verified",
                        previous_candidate_op_id=latest.get("op_id"),
                    )
                    return False
                if not latest_is_terminal and not same_generation:
                    if latest.get("has_apply") is True:
                        logger.warning(
                            "ansible_typed_route_supersession_blocked",
                            incident_id=incident_id,
                            reason="previous_candidate_has_apply",
                            previous_candidate_op_id=latest.get("op_id"),
                        )
                        return False
                    if latest.get("has_pending_child") is True:
                        logger.info(
                            "ansible_typed_route_supersession_deferred",
                            incident_id=incident_id,
                            reason="previous_candidate_has_pending_child",
                            previous_candidate_op_id=latest.get("op_id"),
                        )
                        return False

                    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',
                                'typed_route_reconciler',
                                '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_db_id": _automation_operation_log_incident_id(
                                incident_id
                            ),
                            "input": json.dumps(
                                {
                                    "incident_id": incident_id,
                                    "automation_run_id": str(
                                        latest_input.get("automation_run_id")
                                        or latest.get("op_id")
                                        or ""
                                    ),
                                    "execution_mode": "typed_route_supersession",
                                    "previous_target_route_generation": (
                                        latest_generation or None
                                    ),
                                    "superseded_by_target_route_generation": (
                                        target_route_generation
                                    ),
                                    "superseded_by_candidate_op_id": candidate_op_id,
                                    "source_recurrence": source_recurrence,
                                    "cross_domain_fallback_allowed": False,
                                },
                                ensure_ascii=False,
                            ),
                            "output": json.dumps(
                                {
                                    "reason": "typed_route_superseded",
                                    "terminal_disposition": (
                                        "typed_route_superseded_no_write"
                                    ),
                                    "runtime_apply_executed": False,
                                },
                                ensure_ascii=False,
                            ),
                            "dry_run_result": json.dumps(
                                {
                                    "skipped": True,
                                    "apply_executed": False,
                                    "runtime_apply_executed": False,
                                    "supersession_recorded": True,
                                },
                                ensure_ascii=False,
                            ),
                            "parent_op_id": str(latest.get("op_id") or ""),
                            "tags": [
                                "ansible",
                                "no_write",
                                "typed_route_superseded",
                            ],
                        },
                    )
            inserted = await db.execute(
                text("""
                    INSERT INTO automation_operation_log (
                        op_id, operation_type, actor, status, incident_id,
                        input, output, dry_run_result, tags
                    ) VALUES (
                        CAST(:op_id AS uuid),
                        :operation_type,
                        'decision_manager',
                        :status,
                        :incident_db_id,
                        CAST(:input AS jsonb),
                        CAST(:output AS jsonb),
                        CAST(:dry_run_result AS jsonb),
                        :tags
                    )
                    ON CONFLICT (op_id) DO NOTHING
                    RETURNING op_id::text
                """),
                {
                    "operation_type": payload["operation_type"],
                    "op_id": candidate_op_id,
                    "status": payload["status"],
                    "incident_id": incident_id,
                    "incident_db_id": _automation_operation_log_incident_id(incident_id),
                    "input": json.dumps(payload["input"], ensure_ascii=False),
                    "output": json.dumps(payload["output"], ensure_ascii=False),
                    "dry_run_result": json.dumps(payload["dry_run_result"], ensure_ascii=False),
                    "tags": payload["tags"],
                },
            )
            return inserted.scalar() is not None
    except Exception as exc:
        logger.warning(
            "ansible_decision_audit_write_failed",
            incident_id=incident_id,
            error=str(exc),
        )
        return False