from __future__ import annotations import inspect import os from datetime import UTC, datetime, timedelta from pathlib import Path from types import SimpleNamespace import pytest from src.core.config import settings from src.jobs.awooop_ansible_candidate_backfill_job import ( enqueue_ai_decision_ansible_candidate, ) 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 ( _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE, AnsibleCheckModeClaim, AnsibleRunResult, _append_runtime_stage_receipts_to_apply, _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, _claim_from_apply_operation_row, _claim_from_stale_check_mode_row, _execution_capability_timeout_seconds, _expire_stale_ansible_execution_capabilities, _issue_ansible_execution_capability, _load_open_failed_apply_retry_row, _load_pre_decision_context_runtime_stage_receipts, _post_apply_km_path_type, _post_apply_verification_result, _record_auto_repair_execution_receipt, _record_learning_writeback_receipt, _record_retry_runtime_stage_receipt, _record_runtime_stage_receipts, _record_timeline_projection_receipt, _resolve_execution_capability_operation_type, _revoke_ansible_execution_capability, _run_ansible_command, _send_controlled_apply_telegram_receipt, backfill_missing_auto_repair_execution_receipts_once, build_ansible_apply_command, build_ansible_check_mode_claim_input, build_ansible_check_mode_command, build_ansible_context_runtime_stage_receipts, build_ansible_post_apply_runtime_stage_receipts, build_ansible_pre_apply_runtime_stage_receipts, build_ansible_timeline_runtime_stage_receipt, claim_catalog_drift_failed_check_modes, claim_pending_check_modes, claim_stale_pending_check_modes, detect_ansible_transport_blockers, finalize_check_mode_claim, recent_ansible_transport_blockers, run_controlled_apply_for_claim, run_failed_apply_check_mode_replay_once, run_pending_check_modes_once, ) from src.services.awooop_ansible_learning_writeback import ( _ensure_rag_chunk_fallback, _playbook_trust_readback, canonical_ansible_playbook_id, ensure_ansible_rag_writeback, record_ansible_playbook_trust_writeback, resolve_ansible_playbook_identity, ) from src.services.awooop_truth_chain_service import ( _ansible_playbook_roots, _ansible_runtime_readiness, _automation_quality_score_bucket, _clean_row, _execution_backend_summary, _fetch_inbound_conversation_event_rows, _incident_fingerprints, _summarize_gateway_mcp, _truth_status, build_automation_quality, build_incident_reconciliation, fetch_automation_quality_summary, fetch_truth_chain, 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_inbound_conversation_event_rows) 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) helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows) assert "content_redacted" in source assert "source_envelope" in helper_source assert "source_refs,event_ids" in helper_source assert "source_refs,incident_ids" in helper_source assert "source_refs,sentry_issue_ids" in helper_source assert "source_refs,signoz_alerts" in helper_source def test_fetch_truth_chain_splits_inbound_lookup_into_index_friendly_branches() -> None: source = inspect.getsource(fetch_truth_chain) helper_source = inspect.getsource(_fetch_inbound_conversation_event_rows) assert "await _fetch_inbound_conversation_event_rows(" in source assert "run_id::text = :source_id" in helper_source assert "provider_event_id = :source_id" in helper_source assert "source_refs,event_ids" in helper_source assert "source_refs,incident_ids" in helper_source assert "content_preview ILIKE :source_needle" in helper_source assert "len(rows_by_id) < branch_limit" in helper_source assert "OR content_preview ILIKE :source_needle" not in helper_source assert "OR coalesce(source_envelope #>" not in helper_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_keeps_verified_controlled_apply_autonomous_when_incident_open() -> None: status = _truth_status( incident={ "incident_id": "INC-OPEN-VERIFIED", "status": "INVESTIGATING", "verification_result": "success", }, approvals=[{"status": "EXECUTION_SUCCESS", "action": "ansible controlled apply"}], 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 False 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_ai_retry() -> 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_ai_retry" 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 True assert ( truth["candidate_catalog"]["candidates"][0]["check_mode_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml" ) def test_ansible_truth_marks_worker_apply_as_controlled_apply() -> None: truth = build_ansible_truth( [ { "op_id": "apply-op-1", "operation_type": "ansible_apply_executed", "status": "success", "actor": "ansible_controlled_apply_worker", "input_catalog_id": "ansible:188-momo-backup-user", "input_execution_mode": "controlled_apply", "input_playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml", "input_apply_enabled": "true", "output_apply_executed": "true", "output_returncode": "0", "tags": ["ansible", "controlled_apply", "low", "ai_agent_auto_execution"], }, { "op_id": "check-op-1", "operation_type": "ansible_check_mode_executed", "status": "success", "actor": "ansible_check_mode_worker", "input_catalog_id": "ansible:188-momo-backup-user", "input_execution_mode": "check_mode", "input_playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml", "dry_run_result": {"check_mode_executed": True, "apply_executed": False}, "tags": ["ansible", "check_mode", "controlled_apply_allowed"], }, ], incident={"incident_id": "INC-MOMO", "alertname": "MomoPostgresBackupFailed"}, drift=None, ) summary = truth["summary"] assert summary["check_mode_total"] == 1 assert summary["apply_total"] == 1 assert summary["applied"] is True assert summary["controlled_apply"] is True assert summary["latest_actor"] == "ansible_controlled_apply_worker" assert summary["latest_execution_mode"] == "controlled_apply" assert summary["latest_apply_executed"] is True 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]["check_mode_playbook_path"] == ( "infra/ansible/playbooks/nginx-sync-readonly.yml" ) assert truth["candidate_catalog"]["candidates"][0]["approval_required"] is False 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_marks_repair_candidate_queue_claimable() -> None: incident = SimpleNamespace( incident_id="INC-NODE-188", project_id="awoooi", alert_category="infrastructure", notification_type="TYPE-3", severity=SimpleNamespace(value="P3"), affected_services=["node-exporter-188"], signals=[ SimpleNamespace( alert_name="NodeExporterDown", labels={"alertname": "NodeExporterDown", "instance": "node-exporter-188"}, annotations={}, ) ], ) payload = build_ansible_decision_audit_payload( incident=incident, proposal_data={ "source": "repair_candidate_controlled_queue", "risk_level": "medium", "action": "systemctl restart node-exporter-188", }, decision_path="repair_candidate_controlled_queue", not_used_reason="repair candidate controlled queue ready", ) assert payload is not None assert payload["operation_type"] == "ansible_candidate_matched" assert payload["status"] == "dry_run" assert payload["output"]["decision_effect"] == "check_mode_queue_ready" assert payload["output"]["next_required_step"] == ( "awooop_ansible_check_mode_worker_claims_candidate" ) assert payload["input"]["executor_candidates"] assert payload["dry_run_result"]["check_mode_executed"] is False assert payload["input"]["project_id"] == "awoooi" def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags( monkeypatch, ) -> None: monkeypatch.setenv("AWOOOI_BUILD_COMMIT_SHA", "source-sha-123") 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 True assert candidate["approval_required"] is False assert candidate["risk_level"] == "low" assert payload["input"]["router_source_sha"] == "source-sha-123" def test_ansible_check_mode_claim_input_authorizes_controlled_apply() -> None: candidate_input = { "incident_id": "INC-MOMO", "automation_run_id": "untrusted-upstream-run-id", "executor": "ansible", "decision_path": "repair_candidate_controlled_queue", "idempotency_key": "ansible-candidate:awoooi:INC-MOMO", "router_source_sha": "source-sha-123", "target_selector": {"catalog_ids": ["ansible:188-ai-web"]}, "source_truth_diff": {"required_before_apply": True}, "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["automation_run_id"] == "00000000-0000-0000-0000-000000000001" assert claim["check_mode"] is True assert claim["diff"] is True assert claim["apply_enabled"] is True assert claim["transport_profile"] assert claim["approval_required_before_apply"] is False assert claim["controlled_apply_allowed"] is True assert claim["controlled_apply_blocker"] is None assert claim["single_writer_executor"] == "awoooi-ansible-executor-broker" assert claim["candidate_idempotency_key"] == ( "ansible-candidate:awoooi:INC-MOMO" ) assert claim["apply_idempotency_key"] == ( "ansible-apply:00000000-0000-0000-0000-000000000001:" "ansible:188-ai-web" ) assert claim["router_source_sha"] == "source-sha-123" assert claim["target_selector"] == { "catalog_ids": ["ansible:188-ai-web"] } assert claim["source_truth_diff"]["required_before_apply"] is True assert claim["catalog_playbook_path"] == "infra/ansible/playbooks/188-ai-web.yml" assert claim["apply_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["automation_run_id"] == "00000000-0000-0000-0000-000000000003" assert claim["apply_enabled"] is True assert claim["controlled_apply_allowed"] is True assert claim["approval_required_before_apply"] 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["apply_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) assert str((playbook_root / "roles").resolve()) in spec.env[ "ANSIBLE_ROLES_PATH" ].split(os.pathsep) def test_ansible_apply_command_uses_controlled_apply_without_check(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_apply_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" not 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 str((playbook_root / "roles").resolve()) in spec.env[ "ANSIBLE_ROLES_PATH" ].split(os.pathsep) def test_ansible_controlled_apply_builds_auto_repair_receipt() -> None: claim = AnsibleCheckModeClaim( op_id="check-op-1", source_candidate_op_id="candidate-op-1", incident_id="INC-20260627-NODE110", catalog_id="ansible:110-devops", playbook_path="infra/ansible/playbooks/110-devops.yml", apply_playbook_path="infra/ansible/playbooks/110-devops.yml", inventory_hosts=("host_110",), risk_level="medium", input_payload={"controlled_apply_allowed": True}, ) result = AnsibleRunResult( returncode=0, stdout="ok", stderr="", duration_ms=1234, post_verifier_passed=True, ) receipt = _build_auto_repair_execution_receipt( claim, result, apply_op_id="apply-op-1", ) assert receipt["incident_id"] == "INC-20260627-NODE110" assert receipt["playbook_id"] == "ansible:110-devops" assert receipt["success"] is True assert receipt["triggered_by"] == "ansible_controlled_apply" assert "automation_run_id:candidate-op-1" in receipt["executed_steps"] assert "apply:apply-op-1" in receipt["executed_steps"] assert receipt["risk_level"] == "medium" assert receipt["execution_time_ms"] == 1234 def test_ansible_auto_repair_receipt_never_trusts_executor_only() -> None: claim = AnsibleCheckModeClaim( op_id="check-op-1", source_candidate_op_id="candidate-op-1", incident_id="INC-20260627-NODE110", catalog_id="ansible:110-devops", playbook_path="infra/ansible/playbooks/110-devops.yml", apply_playbook_path="infra/ansible/playbooks/110-devops.yml", inventory_hosts=("host_110",), risk_level="medium", input_payload={"controlled_apply_allowed": True}, ) executor_only = AnsibleRunResult( returncode=0, stdout="ok", stderr="", duration_ms=1234, ) receipt = _build_auto_repair_execution_receipt( claim, executor_only, apply_op_id="apply-op-1", ) assert receipt["success"] is False assert receipt["error_message"] == "independent_post_verifier_failed" assert "post_verifier_passed:false" in receipt["executed_steps"] def test_ansible_apply_operation_row_can_backfill_auto_repair_receipt() -> None: reconstructed = _claim_from_apply_operation_row({ "op_id": "apply-op-1", "parent_op_id": "check-op-1", "incident_id": "INC-20260627-NODE110", "status": "success", "input": { "incident_id": "INC-20260627-NODE110", "catalog_id": "ansible:110-devops", "source_candidate_op_id": "candidate-op-1", "check_mode_op_id": "check-op-1", "playbook_path": "infra/ansible/playbooks/110-devops.yml", "apply_playbook_path": "infra/ansible/playbooks/110-devops.yml", "inventory_hosts": ["host_110"], "risk_level": "medium", }, "output": {"returncode": 0, "stdout_tail": "ok"}, "dry_run_result": {"apply_executed": True}, "duration_ms": 456, }) assert reconstructed is not None claim, result = reconstructed assert claim.incident_id == "INC-20260627-NODE110" assert claim.catalog_id == "ansible:110-devops" assert claim.apply_playbook_path == "infra/ansible/playbooks/110-devops.yml" assert result.returncode == 0 assert result.duration_ms == 456 def test_ansible_apply_operation_row_reconstructs_from_input_without_physical_columns() -> None: reconstructed = _claim_from_apply_operation_row({ "op_id": "apply-op-2", "parent_op_id": "check-op-2", "incident_id": "INC-20260629-231F8E", "status": "success", "input": { "incident_id": "INC-20260629-231F8E", "catalog_id": "ansible:188-momo-backup-user", "source_candidate_op_id": "candidate-op-2", "check_mode_op_id": "check-op-2", "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml", "risk_level": "low", }, "output": {"returncode": 0, "stdout_tail": "ok"}, "dry_run_result": {"apply_executed": True}, "duration_ms": 7727, }) assert reconstructed is not None claim, result = reconstructed assert claim.catalog_id == "ansible:188-momo-backup-user" assert claim.inventory_hosts == ("host_188",) assert claim.apply_playbook_path == "infra/ansible/playbooks/188-momo-backup-user.yml" assert claim.risk_level == "low" assert result.returncode == 0 def test_stale_check_mode_row_is_revalidated_with_canonical_run_id() -> None: source_candidate_op_id = "00000000-0000-0000-0000-000000000091" claim = _claim_from_stale_check_mode_row({ "op_id": "00000000-0000-0000-0000-000000000092", "parent_op_id": source_candidate_op_id, "incident_id": "INC-20260709-46478E", "input": { "automation_run_id": "untrusted-old-run-id", "incident_id": "INC-20260709-46478E", "source_candidate_op_id": source_candidate_op_id, "catalog_id": "ansible:188-momo-backup-user", "catalog_playbook_path": ( "infra/ansible/playbooks/188-momo-backup-user.yml" ), "inventory_hosts": ["host_188"], "risk_level": "low", }, }) assert claim is not None assert claim.op_id == "00000000-0000-0000-0000-000000000092" assert claim.source_candidate_op_id == source_candidate_op_id assert claim.input_payload["automation_run_id"] == source_candidate_op_id assert claim.input_payload["controlled_apply_allowed"] is True assert claim.playbook_path == "infra/ansible/playbooks/188-momo-backup-user.yml" def test_failed_nginx_check_mode_revalidates_to_readonly_catalog_path() -> None: source_candidate_op_id = "00000000-0000-0000-0000-000000000093" claim = _claim_from_stale_check_mode_row({ "op_id": "00000000-0000-0000-0000-000000000094", "parent_op_id": source_candidate_op_id, "incident_id": "INC-20260710-F87642", "input": { "incident_id": "INC-20260710-F87642", "source_candidate_op_id": source_candidate_op_id, "catalog_id": "ansible:nginx-sync", "catalog_playbook_path": "infra/ansible/playbooks/nginx-sync.yml", "check_mode_playbook_path": "infra/ansible/playbooks/nginx-sync.yml", "inventory_hosts": ["host_110", "host_188"], "risk_level": "high", }, }) assert claim is not None assert claim.playbook_path == "infra/ansible/playbooks/nginx-sync-readonly.yml" assert claim.apply_playbook_path == "infra/ansible/playbooks/nginx-sync.yml" def test_stale_check_mode_reclaim_uses_lease_lock_and_current_policy() -> None: source = inspect.getsource(claim_stale_pending_check_modes) run_source = inspect.getsource(run_pending_check_modes_once) assert "FOR UPDATE SKIP LOCKED" in source assert "reclaimed_at" in source assert "ORDER BY check_mode.created_at DESC" in source assert "stale_pending_reclaim_rejected_by_current_policy" in source assert "_claim_from_stale_check_mode_row" in source assert "claim_stale_pending_check_modes" in run_source assert "effective_timeout_seconds + 120" in run_source def test_ansible_execution_broker_issues_expires_and_revokes_bounded_capabilities() -> None: issue_source = inspect.getsource(_issue_ansible_execution_capability) expire_source = inspect.getsource(_expire_stale_ansible_execution_capabilities) revoke_source = inspect.getsource(_revoke_ansible_execution_capability) resolver_source = inspect.getsource(_resolve_execution_capability_operation_type) run_source = inspect.getsource(run_pending_check_modes_once) assert "execution_capability_catalog_not_allowlisted" in issue_source assert "execution_capability_inventory_outside_allowlist" in issue_source assert "ansible_executor_capability_issued" in issue_source assert "execution_capability_active_lease_exists" in issue_source assert "WHERE NOT EXISTS" in issue_source assert "raw_credential_exposed_to_requester" in issue_source assert "expires_at" in issue_source assert "ansible_executor_capability_expired" in expire_source assert "FOR UPDATE SKIP LOCKED" in expire_source assert "ansible_executor_capability_revoked" in revoke_source assert "further_execution_allowed" in revoke_source assert "automation_operation_log_type_valid" in resolver_source assert "semantic_operation_type" in resolver_source assert _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE == "remediation_executed" assert "_issue_ansible_execution_capability" in run_source assert "_revoke_ansible_execution_capability" in run_source assert "_expire_stale_ansible_execution_capabilities" in run_source def test_ansible_execution_capability_enforces_scope_mode_and_expiry() -> None: now = datetime(2026, 7, 11, 1, 0, tzinfo=UTC) run_id = "00000000-0000-0000-0000-000000000201" check_op_id = "00000000-0000-0000-0000-000000000202" scope = { "catalog_id": "ansible:110-devops", "playbook_path": "infra/ansible/playbooks/110-devops.yml", "apply_playbook_path": "infra/ansible/playbooks/110-devops.yml", "inventory_hosts": ["host_110"], "risk_level": "medium", "allowed_execution_modes": ["check_mode", "controlled_apply"], } def _claim(*, capability_scope: dict, expires_at: datetime) -> AnsibleCheckModeClaim: return AnsibleCheckModeClaim( op_id=check_op_id, source_candidate_op_id=run_id, incident_id="INC-CAPABILITY", catalog_id="ansible:110-devops", playbook_path="infra/ansible/playbooks/110-devops.yml", apply_playbook_path="infra/ansible/playbooks/110-devops.yml", inventory_hosts=("host_110",), risk_level="medium", input_payload={ "automation_run_id": run_id, "controlled_apply_allowed": True, "execution_capability": { "capability_op_id": ( "00000000-0000-0000-0000-000000000203" ), "automation_run_id": run_id, "check_mode_op_id": check_op_id, "issued_at": (now - timedelta(seconds=1)).isoformat(), "expires_at": expires_at.isoformat(), "scope": capability_scope, }, }, ) claim = _claim( capability_scope=scope, expires_at=now + timedelta(seconds=100), ) assert _execution_capability_timeout_seconds( claim, execution_mode="controlled_apply", requested_timeout_seconds=180, now=now, ) == 95 with pytest.raises(ValueError, match="execution_capability_expired"): _execution_capability_timeout_seconds( _claim(capability_scope=scope, expires_at=now), execution_mode="check_mode", requested_timeout_seconds=180, now=now, ) with pytest.raises(ValueError, match="execution_capability_mode_not_allowed"): _execution_capability_timeout_seconds( _claim( capability_scope={ **scope, "allowed_execution_modes": ["check_mode"], }, expires_at=now + timedelta(seconds=100), ), execution_mode="controlled_apply", requested_timeout_seconds=180, now=now, ) with pytest.raises(ValueError, match="execution_capability_scope_mismatch"): _execution_capability_timeout_seconds( _claim( capability_scope={**scope, "inventory_hosts": ["host_188"]}, expires_at=now + timedelta(seconds=100), ), execution_mode="check_mode", requested_timeout_seconds=180, now=now, ) def test_failed_check_mode_catalog_drift_replays_once() -> None: source = inspect.getsource(claim_catalog_drift_failed_check_modes) run_source = inspect.getsource(run_pending_check_modes_once) assert "replay_of_check_mode_op_id" in source assert "catalog_drift_replay" in source assert "FOR UPDATE SKIP LOCKED" in source assert "claim_catalog_drift_failed_check_modes" in run_source @pytest.mark.asyncio async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.services import awooop_ansible_check_mode_service as service fresh_claim_attempted = False async def no_expired_capabilities(**_kwargs): return 0 async def no_transport_blockers(**_kwargs): return [] async def no_stale_claims(**_kwargs): return [] async def no_failed_apply_retry(**_kwargs): return { "scanned": 0, "replayed": 0, "check_mode_passed": 0, "check_mode_failed": 0, "runtime_stage_receipt_written": 0, "blockers": [], "error": None, } async def failed_catalog_replay(**_kwargs): raise RuntimeError("catalog replay query canceled") async def fresh_claims_continue(**_kwargs): nonlocal fresh_claim_attempted fresh_claim_attempted = True return [] monkeypatch.setattr( service, "_expire_stale_ansible_execution_capabilities", no_expired_capabilities, ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service, "recent_ansible_transport_blockers", no_transport_blockers, ) monkeypatch.setattr(service, "claim_stale_pending_check_modes", no_stale_claims) monkeypatch.setattr( service, "run_failed_apply_check_mode_replay_once", no_failed_apply_retry, ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", failed_catalog_replay, ) monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claims_continue) result = await service.run_pending_check_modes_once(limit=1) assert fresh_claim_attempted is True assert result["claimed"] == 0 assert result["catalog_replayed"] == 0 assert result["catalog_replay_error"] == "RuntimeError" @pytest.mark.asyncio async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_claims( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.services import awooop_ansible_check_mode_service as service claim_attempted = False async def no_expired_capabilities(**_kwargs): return 0 async def no_transport_blockers(**_kwargs): return [] async def replay_open_retry(**_kwargs): return { "scanned": 1, "replayed": 1, "check_mode_passed": 1, "check_mode_failed": 0, "runtime_stage_receipt_written": 1, "blockers": [], "error": None, } async def candidate_claim_must_not_run(**_kwargs): nonlocal claim_attempted claim_attempted = True return [] monkeypatch.setattr( service, "_expire_stale_ansible_execution_capabilities", no_expired_capabilities, ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service, "recent_ansible_transport_blockers", no_transport_blockers, ) monkeypatch.setattr( service, "run_failed_apply_check_mode_replay_once", replay_open_retry, ) monkeypatch.setattr( service, "claim_stale_pending_check_modes", candidate_claim_must_not_run, ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", candidate_claim_must_not_run, ) monkeypatch.setattr( service, "claim_pending_check_modes", candidate_claim_must_not_run, ) result = await service.run_pending_check_modes_once(limit=1) assert claim_attempted is False assert result["claimed"] == 0 assert result["failed_apply_retry_scanned"] == 1 assert result["failed_apply_retry_replayed"] == 1 assert result["failed_apply_retry_check_mode_passed"] == 1 assert result["failed_apply_retry_stage_receipt_written"] == 1 assert result["failed_apply_retry_priority_tick"] is True def test_check_mode_worker_error_backoff_is_short_and_bounded() -> None: from src.jobs import awooop_ansible_check_mode_job as job delay = job._error_backoff_seconds() assert job._ERROR_BACKOFF_MIN_SECONDS <= delay <= job._ERROR_BACKOFF_MAX_SECONDS assert job._ERROR_BACKOFF_MAX_SECONDS < 60 def test_ansible_apply_receipt_backfill_queries_existing_apply_rows() -> None: source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once) assert "operation_type = 'ansible_apply_executed'" in source assert "auto_repair_executions existing" in source assert "executed_steps::text LIKE" in source assert "apply.catalog_id" not in source assert "apply.playbook_path" not in source assert "apply.risk_level" not in source def test_ansible_auto_repair_receipt_insert_casts_asyncpg_parameters() -> None: source = inspect.getsource(_record_auto_repair_execution_receipt) assert "CAST(:incident_id AS varchar(30))" in source assert "CAST(:playbook_id AS varchar(36))" in source assert "CAST(:triggered_by AS varchar(50))" in source assert "CAST(:apply_op_id_needle AS text)" in source def test_ansible_apply_receipt_backfill_includes_verifier_and_km_gaps() -> None: source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once) assert "FROM incident_evidence evidence" in source assert "evidence.post_execution_state ->> 'apply_op_id' = apply.op_id::text" in source assert "FROM knowledge_entries km" in source assert "km.path_type = 'ansible_apply_receipt:' || left(apply.op_id::text, 8)" in source assert "ansible_learning_writeback_recorded" in source assert "_record_post_apply_verifier_and_learning" in source assert "runtime_stage_receipts" in source assert "_record_runtime_stage_receipts" in source assert "playbook_trust" in source assert "rag_writeback" in source assert "INTERVAL '5 minutes'" in source def test_ansible_learning_writeback_uses_canonical_durable_service() -> None: source = inspect.getsource(_record_learning_writeback_receipt) durable_source = inspect.getsource(record_ansible_playbook_trust_writeback) receipt_source = inspect.getsource(_playbook_trust_readback) assert "record_ansible_playbook_trust_writeback" in source assert "resolve_ansible_playbook_identity" in durable_source assert "FOR UPDATE" in durable_source assert "await db.refresh(playbook)" in durable_source assert '"operation_receipt_readback_verified": True' in receipt_source assert "operation_readback" in durable_source assert 'operation_output.get("learning_recorded") is True' in durable_source assert 'operation_output.get("trust_updated") is True' in durable_source assert '"durable_write_acknowledged": True' in receipt_source def test_ansible_catalog_ids_map_to_stable_playbook_primary_keys() -> None: expected = { "ansible:110-devops": "PB-ANSIBLE-110-DEVOPS", "ansible:188-momo-backup-user": "PB-ANSIBLE-188-MOMO-BACKUP-USER", "ansible:188-ai-web": "PB-ANSIBLE-188-AI-WEB", "ansible:nginx-sync": "PB-ANSIBLE-NGINX-SYNC", "ansible:restore-password-auth": ( "PB-ANSIBLE-RESTORE-PASSWORD-AUTH" ), } assert { catalog_id: canonical_ansible_playbook_id(catalog_id) for catalog_id in expected } == expected assert all(len(playbook_id) <= 32 for playbook_id in expected.values()) for catalog_id, playbook_id in expected.items(): identity = resolve_ansible_playbook_identity(catalog_id) assert identity is not None assert identity["schema_version"] == "ansible_playbook_identity_v1" assert identity["canonical_playbook_id"] == playbook_id assert len(identity["identity_fingerprint"]) == 64 assert resolve_ansible_playbook_identity("ansible:not-allowlisted") is None @pytest.mark.asyncio async def test_ansible_learning_writeback_rejects_catalog_path_mismatch() -> None: receipt = await record_ansible_playbook_trust_writeback( project_id="awoooi", automation_run_id="run-path-mismatch", incident_id="INC-PATH-MISMATCH", incident_db_id=None, catalog_id="ansible:188-momo-backup-user", playbook_path="infra/ansible/playbooks/not-allowlisted.yml", apply_op_id="11111111-1111-4111-8111-111111111111", verification_result="success", ) assert receipt is None def test_ansible_rag_writeback_requires_embedding_readback() -> None: source = inspect.getsource(ensure_ansible_rag_writeback) fallback_source = inspect.getsource(_ensure_rag_chunk_fallback) assert "ensure_entry_embedding" in source assert "embedding IS NOT NULL" in source assert "durable_write_acknowledged" in source assert "_ensure_rag_chunk_fallback" in source assert "index_document" in fallback_source assert "count_chunks_by_source_id" in fallback_source assert "rag_chunk_count" in fallback_source assert "chunk_index_verified" in fallback_source assert "raw_log_payload_stored" in fallback_source assert "secret_value_stored" in fallback_source assert "ansible_rag_km_embedding_copy_failed" in fallback_source @pytest.mark.asyncio async def test_ansible_rag_fallback_keeps_verified_chunks_when_km_copy_fails( monkeypatch, ) -> None: from src.repositories import rag_chunk_repository from src.services import awooop_ansible_learning_writeback as learning_writeback requested_projects: list[str | None] = [] async def count_chunks( _source_id: str, project_id: str | None = None, ) -> int: requested_projects.append(project_id) return 2 class FailingCopyDb: async def execute(self, *_args, **_kwargs): raise RuntimeError("copy unavailable") class FailingCopyContext: async def __aenter__(self): return FailingCopyDb() async def __aexit__(self, *_args): return False monkeypatch.setattr( rag_chunk_repository, "count_chunks_by_source_id", count_chunks, ) monkeypatch.setattr( learning_writeback, "get_db_context", lambda _project_id: FailingCopyContext(), ) receipt = await _ensure_rag_chunk_fallback( project_id="awoooi", incident_id="INC-RAG-FALLBACK", apply_op_id="apply-op-1", entry_id="km-entry-1", title="Verified repair", content="Durable public-safe learning summary", ) assert requested_projects == ["awoooi"] assert receipt is not None assert receipt["embedding_persisted"] is False assert receipt["source_table"] == "rag_chunks" assert receipt["rag_chunk_count"] == 2 assert receipt["chunk_index_verified"] is True assert receipt["durable_write_acknowledged"] is True def test_ansible_post_apply_km_writeback_is_idempotent_for_learning_backfill() -> None: from src.services.awooop_ansible_check_mode_service import ( _record_post_apply_verifier_and_learning, ) source = inspect.getsource(_record_post_apply_verifier_and_learning) assert "SELECT id" in source assert "FROM knowledge_entries" in source assert "path_type = :path_type" in source assert "KnowledgeDBRepository" in source assert "stage_id=\"rag_writeback\"" in source assert "stage_id=\"playbook_trust\"" in source assert "stage_id=\"km_playbook_writeback\"" in source assert "repository_readback_verified" in source assert "km_row_version" in source assert '"knowledge_entries:"' in source assert '"rag_chunks:"' in source def test_ansible_learning_writeback_operation_type_has_schema_migration() -> None: migrations_dir = Path(__file__).resolve().parents[1] / "migrations" migration = Path( migrations_dir / "adr090e_ansible_learning_writeback_operation_type.sql" ).read_text() down = Path( migrations_dir / "adr090e_ansible_learning_writeback_operation_type_down.sql" ).read_text() assert "ansible_learning_writeback_recorded" in migration assert "automation_operation_log_type_valid" in migration assert "DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid" in migration assert "cannot remove ansible_learning_writeback_recorded" in down def test_log_controlled_writeback_dispatch_operation_type_has_schema_migration() -> None: migrations_dir = Path(__file__).resolve().parents[1] / "migrations" migration = Path( migrations_dir / "adr090f_log_controlled_writeback_dispatch_operation_type.sql" ).read_text() down = Path( migrations_dir / "adr090f_log_controlled_writeback_dispatch_operation_type_down.sql" ).read_text() assert "log_controlled_writeback_dispatched" in migration assert "automation_operation_log_type_valid" in migration assert "DROP CONSTRAINT IF EXISTS automation_operation_log_type_valid" in migration assert "cannot remove log_controlled_writeback_dispatched" in down def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_not() -> None: live_source = inspect.getsource(run_controlled_apply_for_claim) backfill_source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once) assert "_send_controlled_apply_telegram_receipt" in live_source assert "telegram_receipt_sent" in live_source assert "_send_controlled_apply_telegram_receipt" not in backfill_source assert inspect.iscoroutinefunction(_send_controlled_apply_telegram_receipt) def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None: source = inspect.getsource(run_failed_apply_check_mode_replay_once) preflight_source = inspect.getsource(_load_open_failed_apply_retry_row) assert "FOR UPDATE SKIP LOCKED" in preflight_source assert "controlled_retry_check_mode_replay" in source assert "build_ansible_check_mode_command" in source assert "verifier.incident_id = coalesce(" in preflight_source assert "apply.input ->> 'incident_id'" in preflight_source assert "controlled_apply_allowed=True" in source assert '"approval_required_before_apply": False' in source assert '"owner_review_required": False' in source assert '"runtime_apply_executed": False' in source assert "queue_ai_playbook_or_transport_repair_candidate" in source assert "retry_requires_repair_and_new_apply_gate" not in source assert "_record_retry_runtime_stage_receipt" in source assert "NOT EXISTS" in preflight_source assert "failed_apply_retry_already_claimed" in source assert "run_controlled_apply_for_claim" not in source receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt) append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply) assert "_runtime_stage_receipt" in receipt_source assert 'stage_id="retry_or_rollback"' in receipt_source assert "jsonb_array_elements" in append_source assert "ansible_controlled_retry_terminal_v2" in receipt_source assert "ansible_retry_terminal_receipt_not_written_unverified" in receipt_source assert "runtime_apply_executed" in receipt_source def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> None: finalize_source = inspect.getsource(finalize_check_mode_claim) apply_source = inspect.getsource(run_controlled_apply_for_claim) assert "'{automation_run_id}'" in finalize_source assert '"automation_run_id": str(' in apply_source @pytest.mark.asyncio async def test_ai_decision_handoff_queues_evidence_backed_single_writer_candidate() -> None: incident = SimpleNamespace( incident_id="INC-AIA-P0-002", project_id="awoooi", severity=SimpleNamespace(value="low"), alertname="MomoPostgresBackupFailed", alert_category="backup_failure", notification_type="TYPE-3", affected_services=["momo-postgres"], signals=[ SimpleNamespace( alert_name="MomoPostgresBackupFailed", labels={"alertname": "MomoPostgresBackupFailed", "instance": "188"}, annotations={}, ) ], ) recorded: list[dict] = [] async def collect(**_kwargs): return SimpleNamespace(snapshot_id="evidence-1") async def verify(**_kwargs): return True async def record(**kwargs): recorded.append(kwargs) return True handoff = await enqueue_ai_decision_ansible_candidate( incident=incident, proposal_data={ "source": "decision_manager", "risk_level": "low", "action": "systemctl restart momo-backup", }, recorder=record, evidence_collector=collect, evidence_verifier=verify, ) assert handoff["status"] == "controlled_check_mode_queued" assert handoff["queued"] is True assert handoff["side_effect_performed"] is False assert handoff["single_writer_executor"] == "awoooi-ansible-executor-broker" assert handoff["source_truth_diff"]["required_before_apply"] is True assert handoff["risk_policy_decision"]["risk_level"] == "low" assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue" @pytest.mark.asyncio async def test_ai_decision_handoff_blocks_critical_before_evidence_or_queue() -> None: calls = 0 async def must_not_run(**_kwargs): nonlocal calls calls += 1 return True handoff = await enqueue_ai_decision_ansible_candidate( incident=SimpleNamespace(incident_id="INC-CRITICAL"), proposal_data={"risk_level": "critical", "action": "restart"}, recorder=must_not_run, evidence_collector=must_not_run, evidence_verifier=must_not_run, ) assert handoff["status"] == "critical_break_glass_required" assert handoff["queued"] is False assert handoff["side_effect_performed"] is False assert calls == 0 def test_controlled_apply_uses_durable_idempotency_guard() -> None: source = inspect.getsource(run_controlled_apply_for_claim) broker_source = inspect.getsource(run_pending_check_modes_once) assert "pg_advisory_xact_lock" in source assert "apply_idempotency_key" in source assert "ansible_controlled_apply_duplicate_suppressed" in source assert "apply_duplicate_suppressed" in broker_source @pytest.mark.asyncio async def test_controlled_apply_duplicate_never_runs_subprocess( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.services import awooop_ansible_check_mode_service as service class ExistingResult: def mappings(self): return self def first(self): return { "op_id": "00000000-0000-0000-0000-000000000103", "status": "success", } class FakeDb: async def execute(self, *_args, **_kwargs): return ExistingResult() class FakeContext: async def __aenter__(self): return FakeDb() async def __aexit__(self, *_args): return False subprocess_started = False async def must_not_run(*_args, **_kwargs): nonlocal subprocess_started subprocess_started = True raise AssertionError("duplicate apply must not execute") monkeypatch.setattr( service, "get_db_context", lambda _project_id: FakeContext(), ) monkeypatch.setattr( service, "_execution_capability_timeout_seconds", lambda *_args, **_kwargs: 30, ) monkeypatch.setattr(service, "_run_ansible_command", must_not_run) claim = AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000102", source_candidate_op_id="00000000-0000-0000-0000-000000000101", incident_id="INC-IDEMPOTENT", catalog_id="ansible:188-momo-backup-user", playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml", apply_playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml", inventory_hosts=("host_188",), risk_level="low", input_payload={ "controlled_apply_allowed": True, "apply_idempotency_key": ( "ansible-apply:00000000-0000-0000-0000-000000000101:" "ansible:188-momo-backup-user" ), }, ) result = await service.run_controlled_apply_for_claim( claim, timeout_seconds=30, ) assert result is None assert subprocess_started is False assert claim.input_payload["apply_duplicate_suppressed"] is True assert claim.input_payload["existing_apply_status"] == "success" def test_ansible_subprocess_is_terminated_when_worker_task_is_cancelled() -> None: source = inspect.getsource(_run_ansible_command) assert "except asyncio.CancelledError" in source assert "process.kill()" in source assert "await process.communicate()" in source def test_controlled_apply_shutdown_marks_pending_row_before_cancelling() -> None: source = inspect.getsource(run_controlled_apply_for_claim) assert "except asyncio.CancelledError" in source assert "ansible_controlled_apply_interrupted_by_worker_shutdown" in source assert '"status": status' in source assert '"op_id": apply_op_id' in source assert "ansible_controlled_apply_interrupted" in source def test_ansible_post_apply_verifier_helpers_are_deterministic() -> None: assert _post_apply_km_path_type("03ca6836-1b76-4da2-8e3e-6d3b6df9254a") == ( "ansible_apply_receipt:03ca6836" ) successful_apply = AnsibleRunResult( returncode=0, stdout="", stderr="", duration_ms=1, ) assert _post_apply_verification_result(successful_apply) == "failed" assert _post_apply_verification_result( successful_apply, { "verification_result": "success", "all_postconditions_passed": True, "executor_returncode_trusted": False, }, ) == "success" assert _post_apply_verification_result( AnsibleRunResult(returncode=1, stdout="", stderr="", duration_ms=1) ) == "failed" assert _post_apply_verification_result( AnsibleRunResult(returncode=0, stdout="", stderr="", duration_ms=1, timed_out=True) ) == "timeout" def test_ansible_runtime_stage_receipts_are_same_run_and_public_safe() -> None: claim = AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000102", source_candidate_op_id="00000000-0000-0000-0000-000000000101", incident_id="INC-RUNTIME-STAGE", catalog_id="ansible:188-momo-backup-user", playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml", apply_playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml", inventory_hosts=("host_188",), risk_level="low", input_payload={ "automation_run_id": "00000000-0000-0000-0000-000000000101", }, ) result = AnsibleRunResult( returncode=0, stdout="changed=0", stderr="", duration_ms=42, ) pre = build_ansible_pre_apply_runtime_stage_receipts(claim) post = build_ansible_post_apply_runtime_stage_receipts( claim, result, apply_op_id="00000000-0000-0000-0000-000000000103", verifier_ready=True, ) assert {receipt["stage_id"] for receipt in pre} == { "normalized_asset_identity", "source_truth_diff", "risk_policy_decision", } assert {receipt["stage_id"] for receipt in post} == { "executor_log_projection", "retry_or_rollback", } assert all(receipt["durable_receipt"] is True for receipt in [*pre, *post]) assert all(receipt["raw_log_payload_stored"] is False for receipt in [*pre, *post]) assert all(receipt["secret_value_stored"] is False for receipt in [*pre, *post]) assert all( receipt["automation_run_id"] == "00000000-0000-0000-0000-000000000101" for receipt in [*pre, *post] ) assert inspect.iscoroutinefunction(_record_runtime_stage_receipts) receipt_writer_source = inspect.getsource(_record_runtime_stage_receipts) assert "jsonb_array_elements" in receipt_writer_source assert "WITH ORDINALITY" in receipt_writer_source assert "DISTINCT ON" in receipt_writer_source def test_ansible_context_receipts_reference_pre_decision_evidence_without_content() -> None: claim = AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000112", source_candidate_op_id="00000000-0000-0000-0000-000000000111", incident_id="INC-RUNTIME-CONTEXT", catalog_id="ansible:188-ai-web", playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml", apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml", inventory_hosts=("host_188",), risk_level="medium", input_payload={ "automation_run_id": "00000000-0000-0000-0000-000000000111", }, ) receipts = build_ansible_context_runtime_stage_receipts( claim, { "id": "evidence-111", "recent_logs": "sanitized service log summary", "mcp_health": {"prometheus": True, "kubernetes": False}, "sensors_attempted": 2, "sensors_succeeded": 1, }, ) assert {receipt["stage_id"] for receipt in receipts} == { "mcp_context", "service_log_evidence", } assert all(receipt["automation_run_id"] == claim.source_candidate_op_id for receipt in receipts) assert all(receipt["raw_log_payload_stored"] is False for receipt in receipts) service_log = next( receipt for receipt in receipts if receipt["stage_id"] == "service_log_evidence" ) assert service_log["detail"]["content_in_receipt"] is False assert "sanitized service log summary" not in str(service_log) loader_source = inspect.getsource(_load_pre_decision_context_runtime_stage_receipts) assert "evidence.collected_at <= candidate.created_at" in loader_source assert "evidence.post_execution_state IS NULL" in loader_source assert "CAST(evidence.post_execution_state AS jsonb)" in loader_source assert "= 'null'::jsonb" in loader_source def test_ansible_context_receipt_rejects_zero_success_mcp_attempts() -> None: claim = AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000132", source_candidate_op_id="00000000-0000-0000-0000-000000000131", incident_id="INC-RUNTIME-CONTEXT-FAILED", catalog_id="ansible:188-ai-web", playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml", apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml", inventory_hosts=("host_188",), risk_level="medium", input_payload={}, ) receipts = build_ansible_context_runtime_stage_receipts( claim, { "id": "evidence-failed", "recent_logs": None, "mcp_health": {"prometheus": False, "kubernetes": False}, "sensors_attempted": 2, "sensors_succeeded": 0, }, ) assert receipts == [] def test_ansible_timeline_projection_receipt_is_same_run_and_idempotent() -> None: claim = AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000122", source_candidate_op_id="00000000-0000-0000-0000-000000000121", incident_id="INC-RUNTIME-TIMELINE", catalog_id="ansible:188-ai-web", playbook_path="infra/ansible/playbooks/188-ai-web-readonly.yml", apply_playbook_path="infra/ansible/playbooks/188-ai-web.yml", inventory_hosts=("host_188",), risk_level="medium", input_payload={ "automation_run_id": "00000000-0000-0000-0000-000000000121", }, ) result = AnsibleRunResult( returncode=2, stdout="", stderr="apply failed", duration_ms=42, ) receipt = build_ansible_timeline_runtime_stage_receipt( claim, result, apply_op_id="00000000-0000-0000-0000-000000000123", timeline_event_id="timeline-123", ) assert receipt["stage_id"] == "timeline_projection" assert receipt["automation_run_id"] == claim.source_candidate_op_id assert receipt["evidence_ref"] == "timeline_events:timeline-123" assert receipt["detail"]["projection_status"] == "error" writer_source = inspect.getsource(_record_timeline_projection_receipt) assert "WITH existing AS" in writer_source assert "gen_random_uuid()::text" in writer_source assert "WHERE NOT EXISTS (SELECT 1 FROM existing)" in writer_source assert inspect.iscoroutinefunction(_record_timeline_projection_receipt) 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"] )