from __future__ import annotations

import inspect
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace

from src.core.config import settings
from src.services.awooop_ansible_audit_service import (
    build_ansible_decision_audit_payload,
    build_ansible_truth,
    record_ansible_decision_audit,
)
from src.services.awooop_ansible_check_mode_service import (
    _automation_operation_log_incident_id,
    build_ansible_check_mode_claim_input,
    build_ansible_check_mode_command,
    claim_pending_check_modes,
    detect_ansible_transport_blockers,
    recent_ansible_transport_blockers,
)
from src.services.awooop_truth_chain_service import (
    _ansible_playbook_roots,
    _ansible_runtime_readiness,
    _automation_quality_score_bucket,
    _clean_row,
    _execution_backend_summary,
    _incident_fingerprints,
    _summarize_gateway_mcp,
    _truth_status,
    build_automation_quality,
    build_incident_reconciliation,
    fetch_truth_chain,
    fetch_automation_quality_summary,
    summarize_automation_quality_records,
)
from src.services.drift_repeat_state import (
    build_drift_fingerprint,
    build_drift_repeat_state,
)


def test_clean_row_parses_json_text_fields_for_gateway_visibility() -> None:
    row = {
        "gate_result": '{"schema_version":"legacy_mcp_bridge_v1","policy_enforced":false}',
        "source_envelope": '{"adapter":"legacy_telegram_gateway"}',
        "plain_text": '{"not":"parsed"}',
    }

    cleaned = _clean_row(row)

    assert cleaned["gate_result"]["schema_version"] == "legacy_mcp_bridge_v1"
    assert cleaned["gate_result"]["policy_enforced"] is False
    assert cleaned["source_envelope"]["adapter"] == "legacy_telegram_gateway"
    assert cleaned["plain_text"] == '{"not":"parsed"}'


def test_incident_fingerprints_reads_signal_labels() -> None:
    fingerprints = _incident_fingerprints({
        "incident_id": "INC-1",
        "signals": [
            {"labels": {"fingerprint": "fp-label"}},
            {"fingerprint": "fp-direct", "labels": {}},
            {"labels": {"fingerprint": "fp-label"}},
        ],
    })

    assert fingerprints == ["fp-direct", "fp-label"]


def test_fetch_truth_chain_can_match_inbound_provider_event_id() -> None:
    source = inspect.getsource(fetch_truth_chain)

    assert "provider_event_id = :source_id" in source


def test_quality_summary_includes_recent_ansible_operation_incidents() -> None:
    source = inspect.getsource(fetch_automation_quality_summary)

    assert "FROM automation_operation_log" in source
    assert "input ->> 'incident_id'" in source
    assert "JOIN incidents ON incidents.incident_id = source_ids.incident_id" in source
    assert "source_ids.recent_evidence_at DESC" in source


def test_quality_summary_uses_batched_truth_chain_inputs() -> None:
    source = inspect.getsource(fetch_automation_quality_summary)

    assert "fetch_truth_chain(" not in source
    assert "approval_records" in source
    assert "incident_evidence" in source
    assert "awooop_outbound_message" in source
    assert "_build_summary_quality_records" in source


def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() -> None:
    decision_source = inspect.getsource(record_ansible_decision_audit)
    claim_source = inspect.getsource(claim_pending_check_modes)

    assert "operation_type, actor, status, incident_id" in decision_source
    assert "coalesce(incident_id::text, input ->> 'incident_id')" in decision_source
    assert "operation_type, actor, status, incident_id" in claim_source
    assert "incident_db_id" in decision_source
    assert "incident_db_id" in claim_source
    assert "NULLIF(:incident_id, '')" not in decision_source
    assert "NULLIF(:incident_id, '')" not in claim_source
    assert _automation_operation_log_incident_id("INC-20260530-0E5C5C") is None
    assert _automation_operation_log_incident_id("12345") == 12345


def test_ansible_transport_cooldown_uses_asyncpg_safe_interval_parameter() -> None:
    source = inspect.getsource(recent_ansible_transport_blockers)

    assert ":cooldown_seconds * INTERVAL '1 second'" in source
    assert "CAST(:cooldown AS interval)" not in source
    assert "include_repair_forced_command_blocker" in source


def test_fetch_truth_chain_returns_inbound_redacted_envelope_fields() -> None:
    source = inspect.getsource(fetch_truth_chain)

    assert "content_redacted" in source
    assert "source_envelope" in source
    assert "source_refs,event_ids" in source
    assert "source_refs,incident_ids" in source
    assert "source_refs,sentry_issue_ids" in source
    assert "source_refs,signoz_alerts" in source


def test_truth_status_marks_no_action_approval_as_manual_required() -> None:
    status = _truth_status(
        incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
        approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}],
        automation_ops=[],
        drift=None,
        drift_repeat_count=0,
        gateway_mcp_total=0,
        legacy_mcp_total=8,
        outbound_visible_total=0,
    )

    assert status["current_stage"] == "manual_required"
    assert status["stage_status"] == "blocked"
    assert status["needs_human"] is True
    assert "approval_resolved_no_action_without_execution" in status["blockers"]
    assert "all_evidence_sensors_failed" in status["blockers"]
    assert "awooop_mcp_gateway_audit_empty" in status["blockers"]


def test_truth_status_marks_inbound_only_source_as_received() -> None:
    status = _truth_status(
        incident=None,
        approvals=[],
        evidence_rows=[],
        automation_ops=[],
        drift=None,
        drift_repeat_count=0,
        gateway_mcp_total=0,
        legacy_mcp_total=0,
        outbound_visible_total=0,
        inbound_visible_total=1,
    )

    assert status["current_stage"] == "inbound_received"
    assert status["stage_status"] == "observed"
    assert status["needs_human"] is False
    assert "awooop_mcp_gateway_audit_empty" in status["blockers"]


def test_truth_status_does_not_treat_no_action_audit_as_execution() -> None:
    status = _truth_status(
        incident={"incident_id": "INC-1", "status": "RESOLVED"},
        approvals=[{"status": "EXECUTION_SUCCESS", "action": "未知操作 | NO_ACTION"}],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}],
        automation_ops=[
            {
                "operation_type": "playbook_executed",
                "status": "success",
                "actor": "approval_execution",
                "output_reason": "NO_ACTION",
                "output_action": "未知操作 | NO_ACTION",
            },
            {
                "operation_type": "ansible_candidate_matched",
                "status": "dry_run",
                "output_not_used_reason": "Ansible check-mode is not wired yet",
            },
        ],
        drift=None,
        drift_repeat_count=0,
        gateway_mcp_total=8,
        legacy_mcp_total=8,
        outbound_visible_total=1,
    )

    assert status["current_stage"] == "manual_required"
    assert status["stage_status"] == "blocked"
    assert status["needs_human"] is True
    assert "approval_resolved_no_action_without_execution" in status["blockers"]


def test_truth_status_marks_open_incident_after_successful_execution() -> None:
    status = _truth_status(
        incident={"incident_id": "INC-OPEN", "status": "INVESTIGATING"},
        approvals=[{"status": "EXECUTION_SUCCESS", "action": "kubectl rollout restart deployment/app"}],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}],
        automation_ops=[],
        drift=None,
        drift_repeat_count=0,
        gateway_mcp_total=3,
        legacy_mcp_total=2,
        outbound_visible_total=1,
        auto_repair_executions=[{"success": True}],
    )

    assert status["current_stage"] == "execution_succeeded"
    assert status["stage_status"] == "success"
    assert status["needs_human"] is True
    assert "incident_open_after_successful_execution" in status["blockers"]


def test_truth_status_marks_repeated_pending_drift_as_human_needed() -> None:
    status = _truth_status(
        incident=None,
        approvals=[],
        evidence_rows=[],
        automation_ops=[],
        drift={
            "report_id": "7f858956",
            "status": "pending",
            "interpretation": {"confidence": 0.0},
        },
        drift_repeat_count=12,
        gateway_mcp_total=0,
        legacy_mcp_total=0,
        outbound_visible_total=0,
    )

    assert status["current_stage"] == "dedup_or_repeat_updated"
    assert status["stage_status"] == "pending"
    assert status["needs_human"] is True
    assert "drift_report_pending_without_resolution" in status["blockers"]
    assert "drift_ai_confidence_zero" in status["blockers"]


def test_gateway_summary_surfaces_first_class_approval_execution() -> None:
    summary = _summarize_gateway_mcp([
        {
            "agent_id": "approval_executor",
            "tool_name": "ssh_docker_restart",
            "result_status": "failed",
            "block_gate": None,
            "gate_result": {
                "schema_version": "awooop_mcp_gateway_audit_v1",
                "gateway_path": "awooop_mcp_gateway",
                "policy_enforced": True,
                "required_scope": "write",
                "is_shadow": False,
                "gate5_approval": True,
            },
        }
    ])

    assert summary["total"] == 1
    assert summary["first_class_total"] == 1
    assert summary["legacy_bridge_total"] == 0
    assert summary["policy_enforced_total"] == 1
    assert summary["approval_executor_total"] == 1
    assert summary["stage"] == "provider_failed_after_gateway"
    assert summary["stage_status"] == "failed"
    assert summary["needs_human"] is True
    assert summary["by_agent"][0]["agent_id"] == "approval_executor"
    assert summary["by_tool"][0]["tool_name"] == "ssh_docker_restart"
    assert summary["by_scope"][0]["required_scope"] == "write"
    assert summary["by_scope"][0]["failed"] == 1


def _drift_item(
    *,
    resource_name: str = "awoooi-api",
    field_path: str = "spec.template.spec.containers[0].image",
    actual_value: str = "api:hotfix",
) -> dict:
    return {
        "resource_kind": "Deployment",
        "resource_name": resource_name,
        "namespace": "awoooi-prod",
        "field_path": field_path,
        "git_value": "api:main",
        "actual_value": actual_value,
        "drift_level": "high",
        "is_allowlisted": False,
    }


def test_drift_fingerprint_is_stable_across_item_order() -> None:
    item_a = _drift_item(resource_name="awoooi-api")
    item_b = _drift_item(
        resource_name="awoooi-worker",
        field_path="spec.template.spec.serviceAccountName",
        actual_value="awoooi-executor",
    )

    first = build_drift_fingerprint("awoooi-prod", [item_a, item_b])
    second = build_drift_fingerprint("awoooi-prod", [item_b, item_a])
    changed = build_drift_fingerprint(
        "awoooi-prod",
        [item_a, {**item_b, "actual_value": "different-service-account"}],
    )

    assert first == second
    assert first.startswith("dfp_")
    assert first != changed


def test_drift_repeat_state_counts_matching_fingerprint_only() -> None:
    now = datetime(2026, 5, 13, 1, 0, tzinfo=UTC)
    report = {
        "report_id": "drift-now",
        "namespace": "awoooi-prod",
        "status": "pending",
        "scanned_at": now,
        "created_at": now,
        "items": [_drift_item()],
    }
    recent = [
        {
            **report,
            "report_id": "drift-prev",
            "scanned_at": now - timedelta(hours=1),
            "created_at": now - timedelta(hours=1),
        },
        {
            **report,
            "report_id": "drift-different",
            "scanned_at": now - timedelta(hours=2),
            "created_at": now - timedelta(hours=2),
            "items": [_drift_item(actual_value="api:other")],
        },
        {
            **report,
            "report_id": "drift-old",
            "scanned_at": now - timedelta(hours=13),
            "created_at": now - timedelta(hours=13),
        },
    ]

    repeat_state = build_drift_repeat_state(report, recent)

    assert repeat_state["schema_version"] == "drift_repeat_state_v1"
    assert repeat_state["fingerprint"].startswith("dfp_")
    assert repeat_state["matching_strategy"] == "namespace_and_stable_items_v1"
    assert repeat_state["occurrences_12h"] == 2
    assert repeat_state["operator_stage"] == "pending_human"
    assert [row["report_id"] for row in repeat_state["reports"]] == [
        "drift-now",
        "drift-prev",
    ]


def test_drift_repeat_state_can_group_semantic_shape_without_values() -> None:
    now = datetime(2026, 5, 19, 1, 0, tzinfo=UTC)
    report = {
        "report_id": "drift-now",
        "namespace": "awoooi-prod",
        "status": "pending",
        "scanned_at": now,
        "created_at": now,
        "items": [_drift_item(actual_value=["vol-a", "vol-b"])],
    }
    recent = [
        {
            **report,
            "report_id": "drift-prev",
            "scanned_at": now - timedelta(hours=1),
            "created_at": now - timedelta(hours=1),
            "items": [_drift_item(actual_value=["vol-a", "vol-b", "vol-c"])],
        },
    ]

    strict_state = build_drift_repeat_state(report, recent)
    semantic_state = build_drift_repeat_state(
        report,
        recent,
        include_values=False,
    )

    assert strict_state["occurrences_12h"] == 1
    assert semantic_state["matching_strategy"] == "namespace_resource_field_level_v2"
    assert semantic_state["occurrences_12h"] == 2
    assert semantic_state["fingerprint"] != semantic_state["strict_fingerprint"]


def test_reconciliation_blocks_open_incident_after_no_action_approval() -> None:
    reconciliation = build_incident_reconciliation(
        incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
        approvals=[
            {
                "id": "approval-1",
                "status": "APPROVED",
                "action": "未知操作 | NO_ACTION",
                "resolved_at": "2026-05-13T01:00:00+00:00",
            }
        ],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}],
        automation_ops=[],
        timeline_events=[],
    )

    codes = {row["code"] for row in reconciliation["mismatches"]}
    assert reconciliation["schema_version"] == "incident_reconciliation_v1"
    assert reconciliation["consistency_status"] == "blocked"
    assert reconciliation["operator_next_state"] == "manual_required"
    assert reconciliation["facts"]["incident_closed"] is False
    assert reconciliation["facts"]["automation_operation_records"] == 0
    assert "incident_open_after_approval_resolved" in codes
    assert "approval_approved_without_execution_record" in codes
    assert "approval_no_action_without_execution" in codes
    assert "evidence_all_sensors_failed" in codes
    assert "timeline_missing_for_approval" in codes


def test_reconciliation_accepts_approval_with_raw_timeline_event() -> None:
    reconciliation = build_incident_reconciliation(
        incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
        approvals=[
            {
                "id": "approval-1",
                "status": "PENDING",
                "action": "kubectl rollout restart deployment/api",
            }
        ],
        evidence_rows=[],
        automation_ops=[],
        timeline_events=[
            {
                "event_type": "human",
                "status": "warning",
                "approval_id": "approval-1",
            }
        ],
    )

    codes = {row["code"] for row in reconciliation["mismatches"]}
    assert "timeline_missing_for_approval" not in codes
    assert reconciliation["facts"]["timeline_events"] == 1


def test_reconciliation_counts_auto_repair_execution_as_real_execution() -> None:
    reconciliation = build_incident_reconciliation(
        incident={"incident_id": "INC-2", "status": "INVESTIGATING"},
        approvals=[
            {
                "id": "approval-2",
                "status": "APPROVED",
                "action": "未知操作 | NO_ACTION",
                "resolved_at": "2026-05-13T01:00:00+00:00",
            }
        ],
        evidence_rows=[
            {
                "sensors_attempted": 8,
                "sensors_succeeded": 6,
                "verification_result": "degraded",
            }
        ],
        automation_ops=[],
        auto_repair_executions=[
            {
                "id": "repair-1",
                "success": True,
                "playbook_id": "PB-1",
            }
        ],
        timeline_events=[{"event_type": "executor", "status": "success"}],
    )

    codes = {row["code"] for row in reconciliation["mismatches"]}
    assert reconciliation["consistency_status"] == "blocked"
    assert reconciliation["facts"]["auto_repair_execution_records"] == 1
    assert reconciliation["facts"]["successful_auto_repair_records"] == 1
    assert reconciliation["facts"]["effective_execution_records"] == 1
    assert "incident_open_after_successful_execution" in codes
    assert "verification_degraded_after_auto_repair" in codes
    assert "approval_approved_without_execution_record" not in codes
    assert "approval_no_action_without_execution" not in codes


def test_automation_quality_marks_no_action_without_execution() -> None:
    quality = build_automation_quality(
        incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
        approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}],
        automation_ops=[],
        auto_repair_executions=[],
        gateway_mcp_summary={"total": 8},
        legacy_mcp_summary={"total": 8},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[],
        timeline_events=[],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["schema_version"] == "automation_quality_v1"
    assert quality["verdict"] == "manual_required_no_action"
    assert quality["facts"]["auto_repair_execution_records"] == 0
    assert gates["execution_recorded"] == "missing"
    assert gates["verification_recorded"] == "not_applicable"
    assert "execution_recorded" in quality["blockers"]


def test_automation_quality_marks_rejected_approval_closed_without_execution() -> None:
    quality = build_automation_quality(
        incident={"incident_id": "INC-REJECTED", "status": "INVESTIGATING"},
        approvals=[{"status": "REJECTED", "action": "kubectl rollout restart deployment/api"}],
        evidence_rows=[],
        automation_ops=[],
        auto_repair_executions=[],
        gateway_mcp_summary={"total": 1},
        legacy_mcp_summary={"total": 1},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[],
        timeline_events=[],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["verdict"] == "approval_rejected_no_execution"
    assert gates["approval_state"] == "passed"
    assert gates["execution_recorded"] == "missing"


def test_automation_quality_marks_expired_approval_for_manual_review() -> None:
    quality = build_automation_quality(
        incident={"incident_id": "INC-EXPIRED", "status": "INVESTIGATING"},
        approvals=[{"status": "EXPIRED", "action": "kubectl rollout restart deployment/api"}],
        evidence_rows=[],
        automation_ops=[],
        auto_repair_executions=[],
        gateway_mcp_summary={"total": 1},
        legacy_mcp_summary={"total": 1},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[],
        timeline_events=[],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["verdict"] == "approval_expired_manual_review"
    assert gates["approval_state"] == "warning"
    assert "approval_state" not in quality["blockers"]


def test_truth_status_marks_rejected_approval_as_closed() -> None:
    status = _truth_status(
        incident={"incident_id": "INC-REJECTED", "status": "INVESTIGATING"},
        approvals=[{"status": "REJECTED", "action": "kubectl rollout restart deployment/api"}],
        evidence_rows=[],
        automation_ops=[],
        drift=None,
        drift_repeat_count=0,
        gateway_mcp_total=1,
        legacy_mcp_total=1,
        outbound_visible_total=1,
        auto_repair_executions=[],
    )

    assert status["current_stage"] == "approval_rejected"
    assert status["stage_status"] == "closed"
    assert status["needs_human"] is False


def test_truth_status_marks_expired_approval_as_manual_needed() -> None:
    status = _truth_status(
        incident={"incident_id": "INC-EXPIRED", "status": "INVESTIGATING"},
        approvals=[{"status": "EXPIRED", "action": "kubectl rollout restart deployment/api"}],
        evidence_rows=[],
        automation_ops=[],
        drift=None,
        drift_repeat_count=0,
        gateway_mcp_total=1,
        legacy_mcp_total=1,
        outbound_visible_total=1,
        auto_repair_executions=[],
    )

    assert status["current_stage"] == "approval_expired"
    assert status["stage_status"] == "expired"
    assert status["needs_human"] is True
    assert "approval_expired_without_operator_decision" in status["blockers"]


def test_reconciliation_ignores_no_action_audit_rows_as_execution() -> None:
    reconciliation = build_incident_reconciliation(
        incident={"incident_id": "INC-NOOP", "status": "INVESTIGATING"},
        approvals=[
            {
                "id": "approval-noop",
                "status": "APPROVED",
                "action": "未知操作 | NO_ACTION",
                "resolved_at": "2026-05-31T01:00:00+00:00",
            }
        ],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}],
        automation_ops=[
            {
                "operation_type": "playbook_executed",
                "status": "success",
                "actor": "approval_execution",
                "output_reason": "NO_ACTION",
                "output_action": "未知操作 | NO_ACTION",
            }
        ],
        auto_repair_executions=[],
        timeline_events=[{"event_type": "executor", "status": "success"}],
    )

    codes = {row["code"] for row in reconciliation["mismatches"]}
    assert reconciliation["facts"]["executed_operation_records"] == 0
    assert reconciliation["facts"]["effective_execution_records"] == 0
    assert "approval_approved_without_execution_record" in codes
    assert "approval_no_action_without_execution" in codes
    assert "incident_open_after_successful_execution" not in codes


def test_automation_quality_ignores_no_action_audit_rows_as_execution() -> None:
    quality = build_automation_quality(
        incident={"incident_id": "INC-1", "status": "RESOLVED"},
        approvals=[{"status": "EXECUTION_SUCCESS", "action": "未知操作 | NO_ACTION"}],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 6}],
        automation_ops=[
            {
                "operation_type": "playbook_executed",
                "status": "success",
                "actor": "approval_execution",
                "output_reason": "NO_ACTION",
                "output_action": "未知操作 | NO_ACTION",
            },
            {
                "operation_type": "ansible_candidate_matched",
                "status": "dry_run",
                "output_not_used_reason": "Ansible check-mode is not wired yet",
            },
        ],
        auto_repair_executions=[],
        gateway_mcp_summary={"total": 8},
        legacy_mcp_summary={"total": 8},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[{"id": "km-1"}],
        timeline_events=[{"id": "tl-1"}],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["verdict"] == "manual_required_no_action"
    assert quality["facts"]["automation_operation_records"] == 2
    assert quality["facts"]["effective_execution_records"] == 0
    assert quality["facts"]["noop_operation_records"] == 1
    assert quality["facts"]["audit_only_operation_records"] == 1
    assert gates["execution_recorded"] == "missing"
    assert gates["verification_recorded"] == "not_applicable"


def test_automation_quality_uses_approval_metadata_to_suppress_diagnostic_ops() -> None:
    quality = build_automation_quality(
        incident={"incident_id": "INC-DIAG", "status": "RESOLVED"},
        approvals=[
            {
                "status": "EXECUTION_SUCCESS",
                "action": "ssh 192.168.0.110 'df -h /data/minio'",
                "extra_metadata": {
                    "execution_kind": "diagnostic",
                    "repair_executed": False,
                },
            }
        ],
        evidence_rows=[
            {
                "sensors_attempted": 8,
                "sensors_succeeded": 6,
                "verification_result": "degraded",
            }
        ],
        automation_ops=[
            {
                "operation_type": "playbook_executed",
                "status": "success",
                "actor": "approval_execution",
                "output_action": "ssh 192.168.0.110 'df -h /data/minio'",
            }
        ],
        auto_repair_executions=[],
        gateway_mcp_summary={"total": 8},
        legacy_mcp_summary={"total": 8},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[{"id": "km-1"}],
        timeline_events=[{"id": "tl-1"}],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["verdict"] == "manual_required_diagnostic_only"
    assert quality["facts"]["automation_operation_records"] == 1
    assert quality["facts"]["effective_execution_records"] == 0
    assert quality["facts"]["approval_repair_suppressed"] is True
    assert gates["execution_recorded"] == "missing"
    assert gates["verification_recorded"] == "not_applicable"


def test_automation_quality_marks_verified_auto_repair() -> None:
    quality = build_automation_quality(
        incident={
            "incident_id": "INC-2",
            "status": "RESOLVED",
            "verification_result": "success",
        },
        approvals=[],
        evidence_rows=[{"sensors_attempted": 3, "sensors_succeeded": 3}],
        automation_ops=[],
        auto_repair_executions=[
            {
                "id": "repair-1",
                "success": True,
                "playbook_id": "pb-1",
            }
        ],
        gateway_mcp_summary={"total": 3},
        legacy_mcp_summary={"total": 3},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[{"id": "km-1"}],
        timeline_events=[{"id": "tl-1"}],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["verdict"] == "auto_repaired_verified"
    assert quality["facts"]["verification_result"] == "success"
    assert quality["score"] == 100
    assert gates["auto_repair_recorded"] == "passed"
    assert gates["verification_recorded"] == "passed"
    assert quality["blockers"] == []


def test_automation_quality_marks_degraded_auto_repair_verification() -> None:
    quality = build_automation_quality(
        incident={
            "incident_id": "INC-3",
            "status": "INVESTIGATING",
        },
        approvals=[{"status": "APPROVED", "action": "未知操作 | NO_ACTION"}],
        evidence_rows=[
            {
                "sensors_attempted": 8,
                "sensors_succeeded": 6,
                "verification_result": "degraded",
            }
        ],
        automation_ops=[],
        auto_repair_executions=[
            {
                "id": "repair-1",
                "success": True,
                "playbook_id": "PB-1",
            }
        ],
        gateway_mcp_summary={"total": 3},
        legacy_mcp_summary={"total": 3},
        outbound_rows=[{"message_id": "m1"}],
        km_entries=[],
        timeline_events=[{"id": "tl-1"}],
    )

    gates = {row["name"]: row["status"] for row in quality["gates"]}
    assert quality["verdict"] == "auto_repaired_verification_degraded"
    assert quality["facts"]["auto_repair_execution_records"] == 1
    assert quality["facts"]["verification_result"] == "degraded"
    assert gates["execution_recorded"] == "passed"
    assert gates["verification_recorded"] == "warning"


def test_automation_quality_score_buckets_are_stable() -> None:
    assert _automation_quality_score_bucket(100) == "green"
    assert _automation_quality_score_bucket(85) == "green"
    assert _automation_quality_score_bucket(84) == "yellow"
    assert _automation_quality_score_bucket(60) == "yellow"
    assert _automation_quality_score_bucket(59) == "red"


def test_automation_quality_summary_denies_full_claim_when_unverified() -> None:
    summary = summarize_automation_quality_records(
        project_id="awoooi",
        window_hours=24,
        limit=200,
        records=[
            {
                "incident": {
                    "incident_id": "INC-OK",
                    "alertname": "container recovered",
                    "severity": "P4",
                    "status": "RESOLVED",
                    "created_at": "2026-05-13T01:00:00+00:00",
                },
                "truth_status": {
                    "current_stage": "execution_succeeded",
                    "stage_status": "success",
                    "needs_human": False,
                },
                "automation_quality": {
                    "applicable": True,
                    "verdict": "auto_repaired_verified",
                    "score": 100,
                    "facts": {
                        "automation_operation_records": 1,
                        "effective_execution_records": 1,
                        "noop_operation_records": 0,
                        "audit_only_operation_records": 0,
                        "auto_repair_execution_records": 1,
                    },
                    "gates": [
                        {"name": "verification_recorded", "status": "passed"},
                    ],
                    "blockers": [],
                },
                "execution": {
                    "automation_operation_log": [
                        {
                            "operation_type": "playbook_executed",
                            "status": "success",
                        }
                    ],
                    "auto_repair_executions": [{"success": True}],
                    "ansible": {
                        "considered": False,
                        "records": [],
                        "candidate_catalog": {"candidates": []},
                    },
                },
            },
            {
                "incident": {
                    "incident_id": "INC-GAP",
                    "alertname": "low risk action",
                    "severity": "P4",
                    "status": "INVESTIGATING",
                    "created_at": "2026-05-13T02:00:00+00:00",
                },
                "truth_status": {
                    "current_stage": "execution_succeeded",
                    "stage_status": "success",
                    "needs_human": False,
                },
                "automation_quality": {
                    "applicable": True,
                    "verdict": "execution_unverified",
                    "score": 65,
                    "facts": {
                        "automation_operation_records": 2,
                        "effective_execution_records": 0,
                        "noop_operation_records": 0,
                        "audit_only_operation_records": 2,
                        "auto_repair_execution_records": 0,
                    },
                    "gates": [
                        {"name": "verification_recorded", "status": "missing"},
                        {"name": "learning_recorded", "status": "missing"},
                    ],
                    "blockers": ["verification_recorded", "learning_recorded"],
                },
                "execution": {
                    "automation_operation_log": [
                        {
                            "operation_type": "ansible_candidate_matched",
                            "status": "dry_run",
                        },
                        {
                            "operation_type": "ansible_check_mode_executed",
                            "status": "dry_run",
                        },
                    ],
                    "auto_repair_executions": [],
                    "ansible": {
                        "considered": True,
                        "records": [
                            {
                                "operation_type": "ansible_candidate_matched",
                                "status": "dry_run",
                            },
                            {
                                "operation_type": "ansible_check_mode_executed",
                                "status": "dry_run",
                            },
                        ],
                        "candidate_catalog": {
                            "candidates": [
                                {"catalog_id": "ansible:188-ai-web"},
                                {"catalog_id": "ansible:nginx-sync"},
                            ]
                        },
                    },
                },
            },
        ],
    )

    assert summary["schema_version"] == "automation_quality_summary_v1"
    assert summary["incident_total"] == 2
    assert summary["evaluated_total"] == 2
    assert summary["verified_auto_repair_total"] == 1
    assert summary["score_buckets"] == {"green": 1, "yellow": 1, "red": 0}
    assert summary["production_claim"]["can_claim_full_auto_repair"] is False
    assert summary["production_claim"]["reason"] == "some_incidents_are_not_auto_repaired_verified"
    assert {row["verdict"]: row["total"] for row in summary["by_verdict"]} == {
        "auto_repaired_verified": 1,
        "execution_unverified": 1,
    }
    assert {row["gate"]: row["total"] for row in summary["gate_failures"]} == {
        "learning_recorded": 1,
        "verification_recorded": 1,
    }
    assert summary["execution_backend_summary"] == {
        "operation_records_total": 3,
        "effective_execution_records_total": 1,
        "noop_operation_records_total": 0,
        "audit_only_operation_records_total": 2,
        "auto_repair_execution_records_total": 1,
        "ansible_considered_total": 1,
        "ansible_audit_record_total": 2,
        "ansible_candidate_total": 2,
        "ansible_check_mode_total": 1,
        "ansible_apply_total": 0,
        "ansible_rollback_total": 0,
        "ansible_pending_check_mode_total": 1,
    }
    assert summary["ansible_runtime"]["playbook_root_present"] is True
    assert summary["ansible_runtime"]["inventory_present"] is True
    assert summary["ansible_runtime"]["playbook_count"] >= 1
    assert "ansible_playbook_binary_present" in summary["ansible_runtime"]
    assert summary["examples"][1]["incident_id"] == "INC-GAP"
    assert summary["examples"][1]["score_bucket"] == "yellow"


def test_automation_quality_summary_builds_operator_flow_gates() -> None:
    def quality(
        *,
        incident_id: str,
        verdict: str,
        score: int,
        gates: list[dict[str, str]],
        blockers: list[str] | None = None,
    ) -> dict[str, object]:
        return {
            "incident": {
                "incident_id": incident_id,
                "alertname": "container restart loop",
                "severity": "P3",
                "status": "INVESTIGATING",
            },
            "truth_status": {
                "current_stage": "execution_succeeded",
                "stage_status": "success",
                "needs_human": verdict != "auto_repaired_verified",
            },
            "automation_quality": {
                "applicable": True,
                "verdict": verdict,
                "score": score,
                "facts": {
                    "automation_operation_records": 1,
                    "effective_execution_records": 1 if verdict == "auto_repaired_verified" else 0,
                    "auto_repair_execution_records": 1 if verdict == "auto_repaired_verified" else 0,
                },
                "gates": gates,
                "blockers": blockers or [],
            },
            "execution": {
                "automation_operation_log": [],
                "auto_repair_executions": [],
                "ansible": {"considered": False, "records": [], "candidate_catalog": {"candidates": []}},
            },
        }

    passed_gates = [
        {"name": "source_persisted", "status": "passed"},
        {"name": "outbound_recorded", "status": "passed"},
        {"name": "evidence_collected", "status": "passed"},
        {"name": "mcp_gateway_observed", "status": "passed"},
        {"name": "approval_state", "status": "not_applicable"},
        {"name": "execution_recorded", "status": "passed"},
        {"name": "auto_repair_recorded", "status": "passed"},
        {"name": "verification_recorded", "status": "passed"},
        {"name": "learning_recorded", "status": "passed"},
        {"name": "timeline_recorded", "status": "passed"},
    ]
    blocked_gates = [
        {"name": "source_persisted", "status": "passed"},
        {"name": "outbound_recorded", "status": "passed"},
        {"name": "evidence_collected", "status": "missing"},
        {"name": "mcp_gateway_observed", "status": "missing"},
        {"name": "approval_state", "status": "warning"},
        {"name": "execution_recorded", "status": "missing"},
        {"name": "auto_repair_recorded", "status": "missing"},
        {"name": "verification_recorded", "status": "not_applicable"},
        {"name": "learning_recorded", "status": "not_applicable"},
        {"name": "timeline_recorded", "status": "missing"},
    ]

    summary = summarize_automation_quality_records(
        project_id="awoooi",
        window_hours=24,
        limit=2,
        records=[
            quality(
                incident_id="INC-PASS",
                verdict="auto_repaired_verified",
                score=100,
                gates=passed_gates,
            ),
            quality(
                incident_id="INC-BLOCKED",
                verdict="manual_required_no_action",
                score=40,
                gates=blocked_gates,
                blockers=["mcp_gateway_observed", "execution_recorded"],
            ),
        ],
    )

    flow = summary["automation_flow_gates"]
    gates = {row["gate"]: row for row in flow["gates"]}

    assert flow["schema_version"] == "automation_flow_gate_summary_v1"
    assert flow["overall_status"] == "blocked"
    assert gates["alert_intake"]["status"] == "passed"
    assert gates["mcp_investigation"]["missing_total"] == 1
    assert gates["approval_policy"]["warning_total"] == 1
    assert gates["execution_recorded"]["missing_total"] == 1
    assert gates["verification_recorded"]["missing_total"] == 1
    assert gates["knowledge_recorded"]["missing_total"] == 1
    assert gates["operator_visible"]["missing_total"] == 1
    assert gates["mcp_investigation"]["examples"][0]["incident_id"] == "INC-BLOCKED"


def test_ansible_runtime_readiness_reports_check_mode_blockers() -> None:
    readiness = _ansible_runtime_readiness()

    assert readiness["playbook_root_present"] is True
    assert readiness["inventory_present"] is True
    assert readiness["playbook_count"] >= 1
    assert isinstance(readiness["can_run_check_mode"], bool)
    assert isinstance(readiness["blockers"], list)
    assert "check_mode_transport_profile" in readiness
    assert "check_mode_ssh_key_readable" in readiness
    assert "check_mode_known_hosts_readable" in readiness
    assert "repair_ssh_key_readable" in readiness
    assert "repair_known_hosts_readable" in readiness


def test_ansible_runtime_readiness_requires_check_mode_ssh_material(tmp_path: Path) -> None:
    missing_key = tmp_path / "missing-id-ed25519"
    missing_known_hosts = tmp_path / "missing-known-hosts"

    readiness = _ansible_runtime_readiness(
        check_mode_ssh_key_path=missing_key,
        check_mode_known_hosts_path=missing_known_hosts,
    )

    assert readiness["check_mode_ssh_key_present"] is False
    assert readiness["check_mode_ssh_key_readable"] is False
    assert readiness["check_mode_known_hosts_present"] is False
    assert readiness["check_mode_known_hosts_readable"] is False
    assert "ansible_check_mode_ssh_key_missing" in readiness["blockers"]
    assert "ansible_check_mode_known_hosts_missing" in readiness["blockers"]


def test_ansible_runtime_readiness_accepts_readable_check_mode_ssh_material(tmp_path: Path) -> None:
    key_path = tmp_path / "id_ed25519"
    known_hosts_path = tmp_path / "known_hosts"
    key_path.write_text("test-key", encoding="utf-8")
    known_hosts_path.write_text("192.168.0.110 ssh-ed25519 AAAATEST", encoding="utf-8")
    key_path.chmod(0o400)
    known_hosts_path.chmod(0o400)

    readiness = _ansible_runtime_readiness(
        check_mode_ssh_key_path=key_path,
        check_mode_known_hosts_path=known_hosts_path,
    )

    assert readiness["check_mode_ssh_key_present"] is True
    assert readiness["check_mode_ssh_key_readable"] is True
    assert readiness["check_mode_known_hosts_present"] is True
    assert readiness["check_mode_known_hosts_readable"] is True
    assert "ansible_check_mode_ssh_key_missing" not in readiness["blockers"]
    assert "ansible_check_mode_known_hosts_missing" not in readiness["blockers"]


def test_ansible_playbook_roots_supports_flat_container_module_path() -> None:
    roots = _ansible_playbook_roots(Path("/app/src/services/awooop_truth_chain_service.py"))

    assert Path("/app/infra/ansible") in roots
    assert Path("/app/src/infra/ansible") in roots


def test_reconciliation_marks_consistent_resolved_execution() -> None:
    reconciliation = build_incident_reconciliation(
        incident={"incident_id": "INC-2", "status": "RESOLVED"},
        approvals=[
            {
                "id": "approval-2",
                "status": "APPROVED",
                "action": "restart service",
                "resolved_at": "2026-05-13T01:00:00+00:00",
            }
        ],
        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 7}],
        automation_ops=[{"status": "success"}],
        timeline_events=[{"event_type": "executor", "status": "success"}],
    )

    assert reconciliation["consistency_status"] == "consistent"
    assert reconciliation["operator_next_state"] == "continue"
    assert reconciliation["mismatches"] == []


def test_ansible_truth_surfaces_audited_check_mode_record() -> None:
    truth = build_ansible_truth(
        [
            {
                "op_id": "op-ansible-1",
                "operation_type": "ansible_check_mode_executed",
                "status": "dry_run",
                "actor": "platform_operator",
                "input_catalog_id": "ansible:188-momo-backup-user",
                "input_execution_mode": "check_mode",
                "input_playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "input_check_mode": "true",
                "dry_run_result": {
                    "changed": 1,
                    "check_mode_executed": True,
                    "apply_executed": False,
                    "returncode": 0,
                },
                "tags": ["ansible", "check_mode"],
                "created_at": "2026-05-12T22:00:00+08:00",
            }
        ],
        incident={"incident_id": "INC-1", "alertname": "momo pg_backup failed on 188"},
        drift=None,
    )

    assert truth["considered"] is True
    assert truth["not_used_reason"] is None
    assert truth["records"][0]["playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
    assert truth["records"][0]["check_mode"] == "true"
    assert truth["records"][0]["catalog_id"] == "ansible:188-momo-backup-user"
    assert truth["records"][0]["execution_mode"] == "check_mode"
    assert truth["records"][0]["returncode"] == 0
    assert truth["records"][0]["dry_run_result"]["changed"] == 1
    assert truth["summary"]["check_mode_total"] == 1
    assert truth["summary"]["apply_total"] == 0
    assert truth["summary"]["latest_catalog_id"] == "ansible:188-momo-backup-user"
    assert truth["summary"]["latest_returncode"] == 0
    assert "ansible_check_mode_executed" in truth["audit_contract"]["operation_types"]
    assert truth["candidate_catalog"]["decision_effect"] == "none"
    assert truth["candidate_catalog"]["candidates"][0]["catalog_id"] == "ansible:188-momo-backup-user"
    assert truth["candidate_catalog"]["candidates"][0]["auto_apply_enabled"] is False
    assert (
        truth["candidate_catalog"]["candidates"][0]["check_mode_playbook_path"]
        == "infra/ansible/playbooks/188-momo-backup-user.yml"
    )


def test_ansible_truth_keeps_catalog_hint_separate_from_runtime_use() -> None:
    truth = build_ansible_truth(
        [],
        incident={"incident_id": "INC-2", "alertname": "nginx 502 upstream timeout"},
        drift=None,
    )

    assert truth["considered"] is False
    assert truth["records"] == []
    assert truth["not_used_reason"].startswith("no automation_operation_log row")
    assert truth["candidate_catalog"]["candidates"][0]["catalog_id"] == "ansible:nginx-sync"
    assert truth["candidate_catalog"]["candidates"][0]["approval_required"] is True
    assert truth["candidate_catalog"]["decision_effect"] == "none"


def test_ansible_decision_audit_payload_is_dry_run_only() -> None:
    incident = SimpleNamespace(
        incident_id="INC-DOCKER",
        project_id="awoooi",
        alert_category="infrastructure",
        notification_type="TYPE-3",
        severity=SimpleNamespace(value="P3"),
        affected_services=["bitan-pharmacy-bitan-1"],
        signals=[
            SimpleNamespace(
                alert_name="DockerContainerUnhealthy",
                labels={"alertname": "DockerContainerUnhealthy", "container": "bitan-pharmacy-bitan-1"},
                annotations={},
            )
        ],
    )

    payload = build_ansible_decision_audit_payload(
        incident=incident,
        proposal_data={"source": "expert_system", "risk_level": "low", "action": "NO_ACTION"},
        decision_path="manual_approval",
        not_used_reason="manual approval required; Ansible check-mode is not wired yet",
    )

    assert payload is not None
    assert payload["operation_type"] == "ansible_candidate_matched"
    assert payload["status"] == "dry_run"
    assert payload["input"]["executor"] == "ansible"
    assert payload["input"]["check_mode"] is True
    assert payload["input"]["apply_enabled"] is False
    assert payload["input"]["approval_required"] is True
    assert payload["input"]["executor_candidates"]
    assert payload["output"]["decision_effect"] == "audit_only"
    assert payload["dry_run_result"]["check_mode_executed"] is False


def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags() -> None:
    incident = SimpleNamespace(
        incident_id="INC-MOMO",
        project_id="awoooi",
        alert_category="database",
        notification_type="TYPE-3",
        severity=SimpleNamespace(value="P3"),
        affected_services=["momo"],
        signals=[
            SimpleNamespace(
                alert_name="MomoPostgresBackupFailed",
                labels={"alertname": "MomoPostgresBackupFailed", "instance": "188"},
                annotations={},
            )
        ],
    )

    payload = build_ansible_decision_audit_payload(
        incident=incident,
        proposal_data={"source": "expert_system", "risk_level": "low"},
        decision_path="manual_approval",
        not_used_reason="candidate audit",
    )

    candidate = payload["input"]["executor_candidates"][0]
    assert candidate["catalog_id"] == "ansible:188-momo-backup-user"
    assert candidate["supports_check_mode"] is True
    assert candidate["playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
    assert candidate["check_mode_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
    assert candidate["auto_apply_enabled"] is False
    assert candidate["approval_required"] is True
    assert candidate["risk_level"] == "low"


def test_ansible_check_mode_claim_input_keeps_apply_locked() -> None:
    candidate_input = {
        "incident_id": "INC-MOMO",
        "executor": "ansible",
        "executor_candidates": [
            {
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "inventory_hosts": ["host_188"],
                "risk_level": "medium",
            }
        ],
    }

    claim = build_ansible_check_mode_claim_input(
        source_candidate_op_id="00000000-0000-0000-0000-000000000001",
        candidate_input=candidate_input,
    )

    assert claim["execution_mode"] == "check_mode"
    assert claim["check_mode"] is True
    assert claim["diff"] is True
    assert claim["apply_enabled"] is False
    assert claim["transport_profile"]
    assert claim["approval_required_before_apply"] is True
    assert claim["catalog_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
    assert claim["source_candidate_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml"
    assert claim["check_mode_playbook_path"] == "infra/ansible/playbooks/188-ai-web-readonly.yml"
    assert claim["playbook_path"] == "infra/ansible/playbooks/188-ai-web-readonly.yml"


def test_ansible_check_mode_claim_input_accepts_momo_backup_user_catalog() -> None:
    candidate_input = {
        "incident_id": "INC-MOMO",
        "executor": "ansible",
        "executor_candidates": [
            {
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "inventory_hosts": ["host_188"],
                "risk_level": "low",
            }
        ],
    }

    claim = build_ansible_check_mode_claim_input(
        source_candidate_op_id="00000000-0000-0000-0000-000000000003",
        candidate_input=candidate_input,
    )

    assert claim["execution_mode"] == "check_mode"
    assert claim["apply_enabled"] is False
    assert claim["catalog_id"] == "ansible:188-momo-backup-user"
    assert claim["risk_level"] == "low"
    assert claim["catalog_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
    assert claim["source_candidate_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
    assert claim["check_mode_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
    assert claim["playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"


def test_ansible_check_mode_claim_rejects_non_check_mode_catalog() -> None:
    candidate_input = {
        "incident_id": "INC-SSH",
        "executor": "ansible",
        "executor_candidates": [
            {
                "catalog_id": "ansible:restore-password-auth",
                "playbook_path": "infra/ansible/playbooks/restore-password-auth.yml",
                "inventory_hosts": ["host_188"],
                "risk_level": "high",
            }
        ],
    }

    try:
        build_ansible_check_mode_claim_input(
            source_candidate_op_id="00000000-0000-0000-0000-000000000002",
            candidate_input=candidate_input,
        )
    except ValueError as exc:
        assert str(exc) == "no_safe_check_mode_candidate"
    else:
        raise AssertionError("non-check-mode catalog should be rejected")


def test_ansible_check_mode_command_uses_check_diff_and_selected_ssh_transport(tmp_path: Path) -> None:
    playbook_root = tmp_path / "infra" / "ansible"
    playbook_dir = playbook_root / "playbooks"
    inventory_dir = playbook_root / "inventory"
    playbook_dir.mkdir(parents=True)
    inventory_dir.mkdir(parents=True)
    (playbook_dir / "188-ai-web.yml").write_text("---\n- hosts: host_188\n  tasks: []\n")
    (inventory_dir / "hosts.yml").write_text("all: {}\n")
    repair_key = tmp_path / "id_ed25519"
    known_hosts = tmp_path / "known_hosts"
    repair_key.write_text("key")
    known_hosts.write_text("host key")

    spec = build_ansible_check_mode_command(
        playbook_path="infra/ansible/playbooks/188-ai-web.yml",
        inventory_hosts=("host_188",),
        playbook_root=playbook_root,
        check_mode_ssh_key_path=repair_key,
        check_mode_known_hosts_path=known_hosts,
    )

    assert "--check" in spec.command
    assert "--diff" in spec.command
    assert "--limit" in spec.command
    assert "host_188" in spec.command
    assert "ansible_ssh_private_key_file" in spec.command[-1]
    assert str(repair_key) in spec.command[-1]
    assert str(known_hosts) in spec.command[-1]
    assert "apply" not in " ".join(spec.command)


def test_ansible_claim_query_limits_recent_candidate_backlog() -> None:
    source = inspect.getsource(claim_pending_check_modes)

    assert "candidate.created_at >= NOW() - (:candidate_max_age_hours * INTERVAL '1 hour')" in source
    assert "AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS" in source


def test_ansible_transport_blocker_detects_repair_forced_command_denial() -> None:
    blockers = detect_ansible_transport_blockers(
        "fatal: host unreachable REPAIR_DENIED:invalid_command",
    )

    assert blockers == ["ansible_repair_ssh_forced_command_denies_ansible_bootstrap"]


def test_execution_backend_summary_subtracts_completed_check_mode_parent() -> None:
    summary = _execution_backend_summary([
        {
            "execution": {
                "ansible": {
                    "considered": True,
                    "candidate_catalog": {"candidates": [{"catalog_id": "ansible:188-ai-web"}]},
                    "records": [
                        {
                            "op_id": "candidate-1",
                            "operation_type": "ansible_candidate_matched",
                            "status": "dry_run",
                        },
                        {
                            "op_id": "check-1",
                            "parent_op_id": "candidate-1",
                            "operation_type": "ansible_check_mode_executed",
                            "status": "success",
                        },
                    ],
                },
                "automation_operation_log": [],
                "auto_repair_executions": [],
            },
            "automation_quality": {"facts": {}},
        }
    ])

    assert summary["ansible_check_mode_total"] == 1
    assert summary["ansible_pending_check_mode_total"] == 0


def test_quality_summary_marks_forced_command_denial_as_runtime_blocker(monkeypatch) -> None:
    monkeypatch.setattr(
        settings,
        "AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH",
        "/etc/repair-ssh/id_ed25519",
    )
    summary = summarize_automation_quality_records(
        project_id="awoooi",
        window_hours=24,
        limit=20,
        records=[
            {
                "incident": {"incident_id": "INC-1", "alertname": "DockerContainerUnhealthy"},
                "truth_status": {},
                "automation_quality": {"applicable": True, "score": 50, "verdict": "observed"},
                "execution": {
                    "automation_operation_log": [],
                    "auto_repair_executions": [],
                    "ansible": {
                        "considered": True,
                        "candidate_catalog": {"candidates": [{"catalog_id": "ansible:110-devops"}]},
                        "records": [
                            {
                                "op_id": "check-1",
                                "operation_type": "ansible_check_mode_executed",
                                "status": "failed",
                                "dry_run_result": {
                                    "stdout_tail": "REPAIR_DENIED:invalid_command",
                                },
                            }
                        ],
                    },
                },
            }
        ],
    )

    assert summary["ansible_runtime"]["can_run_check_mode"] is False
    assert (
        "ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
        in summary["ansible_runtime"]["blockers"]
    )


def test_quality_summary_keeps_repair_forced_blocker_historical_for_ssh_mcp(monkeypatch) -> None:
    monkeypatch.setattr(
        settings,
        "AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH",
        "/run/secrets/ssh_mcp_key",
    )
    summary = summarize_automation_quality_records(
        project_id="awoooi",
        window_hours=24,
        limit=20,
        records=[
            {
                "incident": {"incident_id": "INC-1", "alertname": "DockerContainerUnhealthy"},
                "truth_status": {},
                "automation_quality": {"applicable": True, "score": 50, "verdict": "observed"},
                "execution": {
                    "automation_operation_log": [],
                    "auto_repair_executions": [],
                    "ansible": {
                        "considered": True,
                        "candidate_catalog": {"candidates": [{"catalog_id": "ansible:110-devops"}]},
                        "records": [
                            {
                                "op_id": "check-1",
                                "operation_type": "ansible_check_mode_executed",
                                "status": "failed",
                                "dry_run_result": {
                                    "stdout_tail": "REPAIR_DENIED:invalid_command",
                                },
                            }
                        ],
                    },
                },
            }
        ],
    )

    assert (
        "ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
        in summary["ansible_runtime"]["historical_transport_blockers"]
    )
    assert (
        "ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
        not in summary["ansible_runtime"]["blockers"]
    )
