from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from src.api.v1 import webhooks from src.services.awooop_ansible_audit_service import ( _catalog_hints, build_ansible_decision_audit_payload, build_ansible_truth, ) from src.services.controlled_alert_target_router import ( build_controlled_recovery_handoff, build_controlled_recovery_promotion_contract, resolve_controlled_alert_target, ) def _cold_start_incident() -> dict: return { "incident_id": "INC-20260711-11C751", "project_id": "awoooi", "alertname": "ColdStartGateBlocked", "alert_category": "general", "affected_services": ["cold-start-gate"], "signals": [ { "alert_name": "ColdStartGateBlocked", "labels": { "component": "cold-start-gate", "namespace": "default", "host": "110", }, "annotations": { "summary": ( "full-stack cold-start recovery blocked; backup and " "Docker evidence must be checked" ) }, } ], } def test_normalizer_separates_source_namespace_from_execution_domain() -> None: route = resolve_controlled_alert_target( alertname="ColdStartGateBlocked", target_resource="cold-start-gate", namespace="default", ) assert route is not None assert route["source_namespace"] == "default" assert route["normalized_execution_namespace"] == "host_recovery" assert route["kubernetes_namespace_applicable"] is False assert route["executor"] == "Agent99" assert route["canonical_asset_id"] == "control-plane:cold-start-gate" assert route["allowed_inventory_hosts"] == [] assert route["controlled_apply_allowed"] is False assert route["no_write_terminal_required"] is True assert route["runtime_execution_authorized"] is False assert resolve_controlled_alert_target( alertname="SentryRpsZero", target_resource="sentry", namespace="awoooi-prod", ) is None def test_cold_start_contract_requires_same_run_receipts_before_closure() -> None: contract = build_controlled_recovery_promotion_contract( alertname="ColdStartGateBlocked", target_resource="cold-start-gate", namespace="default", incident_id="INC-20260711-11C751", ) assert contract is not None fields = {row["field"]: row for row in contract["fields"]} assert fields["target_selector"]["status"] == "ready" assert fields["controlled_apply_route"]["status"] == "blocked" for required in ( "source_sensor_receipt", "check_mode_receipt", "dispatch_receipt", "no_write_terminal_receipt", "post_apply_verifier", "incident_closure_receipt", "telegram_receipt", "km_writeback", "playbook_trust_writeback", ): assert fields[required]["status"] == "pending" assert contract["runtime_execution_authorized"] is False assert contract["owner_review_required"] is False assert contract["needs_human"] is False assert contract["controlled_playbook_queue"] is False assert contract["completion_status"] == ( "no_write_terminal_pending_durable_receipt" ) assert contract["completion_blocker"] == "exact_inventory_scope_missing" def test_cold_start_is_excluded_from_generic_ansible_keyword_catalog() -> None: hints = _catalog_hints(_cold_start_incident(), None) assert hints["match_mode"] == "typed_domain_router_v2" assert hints["decision_effect"] == "delegate_to_domain_executor" assert hints["candidates"] == [] assert hints["controlled_executor"]["executor"] == "Agent99" assert "ansible:110-devops" in hints["unmatched_catalog_ids"] assert "ansible:188-momo-backup-user" in hints["unmatched_catalog_ids"] truth = build_ansible_truth([], incident=_cold_start_incident(), drift=None) assert truth["not_used_reason"] == "typed_non_ansible_domain_route" assert truth["candidate_catalog"]["controlled_executor"]["route_id"] == ( "agent99_recover_after_owner_review" ) def test_cold_start_does_not_emit_ansible_candidate_audit() -> None: incident = SimpleNamespace( incident_id="INC-20260711-11C751", project_id="awoooi", alertname="ColdStartGateBlocked", alert_category="general", notification_type="TYPE-3", severity=SimpleNamespace(value="P3"), affected_services=["cold-start-gate"], signals=[ SimpleNamespace( alert_name="ColdStartGateBlocked", labels={"component": "cold-start-gate", "namespace": "default"}, annotations={"summary": "full-stack cold-start blocked"}, ) ], ) payload = build_ansible_decision_audit_payload( incident=incident, proposal_data={"source": "test", "risk_level": "low"}, decision_path="alert_webhook_controlled_router", not_used_reason="test", ) assert payload is None def test_handoff_requires_no_write_terminal_until_inventory_scope_is_exact() -> None: handoff = build_controlled_recovery_handoff( alertname="custom", target_resource="cold-start-gate", namespace="default", incident_id="INC-20260711-11C751", ) assert handoff is not None assert handoff["single_writer_executor"] == "Agent99" assert handoff["queued"] is False assert handoff["side_effect_performed"] is False assert handoff["runtime_execution_authorized"] is False assert handoff["owner_review_required"] is False assert handoff["needs_human"] is False assert handoff["status"] == "cold_start_no_write_terminal_pending_receipt" assert handoff["active_blockers"] == ["exact_inventory_scope_missing"] assert handoff["safe_next_action"] == ( "persist_no_write_terminal_without_claim_or_transport" ) @pytest.mark.asyncio @pytest.mark.parametrize( "operation_receipt_id", ["no-write-receipt-11c751", "", None], ) async def test_webhook_router_persists_named_cold_start_no_write_terminal( monkeypatch: pytest.MonkeyPatch, operation_receipt_id: str | None, ) -> None: queue = AsyncMock(side_effect=AssertionError("generic Ansible must not be queued")) append = AsyncMock( return_value=( SimpleNamespace(id=operation_receipt_id) if operation_receipt_id is not None else None ) ) incident_lookup = AsyncMock(side_effect=AssertionError("incident lookup not needed")) bridge = AsyncMock( return_value={ "status": "no_write_terminal", "dispatchPerformed": False, "runtimeWritePerformed": False, "controlledApplyAuthorized": False, "runtimeClosureVerified": False, "noWriteTerminal": { "schema_version": "agent99_cold_start_no_write_terminal_v1", "status": "no_write_terminal", "reason": "exact_inventory_scope_missing", "incident_id": "INC-20260711-11C751", "route_id": "agent99:host_recovery:Recover", "canonical_asset_id": "control-plane:cold-start-gate", "source_namespace": "default", "normalized_execution_namespace": "host_recovery", "kubernetes_namespace_applicable": False, "allowed_inventory_hosts": [], "runtime_write_performed": False, "agent99_dispatch_performed": False, "production_executor_invoked": False, "incident_resolution_allowed": False, "transport": { "claim_performed": False, "transport_performed": False, }, "next_safe_action": ( "bind_exact_inventory_scope_before_creating_a_new_recover_candidate" ), "verifier_contract": { "name": "cold_start_no_write_terminal_contract", "status": "ready_for_independent_source_validation", "scope": "source_no_write_only", "runtime_verifier_applicable": False, }, }, } ) monkeypatch.setattr( "src.jobs.awooop_ansible_candidate_backfill_job.enqueue_ai_decision_ansible_candidate", queue, ) monkeypatch.setattr( "src.repositories.alert_operation_log_repository.get_alert_operation_log_repository", lambda: SimpleNamespace(append=append), ) monkeypatch.setattr( webhooks, "get_incident_service", lambda: SimpleNamespace(get_from_working_memory=incident_lookup), ) monkeypatch.setattr(webhooks, "bridge_alertmanager_to_agent99", bridge) handoff = await webhooks._try_auto_repair_background( incident_id="INC-20260711-11C751", approval_id="00000000-0000-0000-0000-000000000751", alert_type="custom", target_resource="cold-start-gate", namespace="default", risk_level="low", source_alert_id="alert-11c751", source_alertname="ColdStartGateBlocked", source_fingerprint="source-fingerprint", ) queue.assert_not_awaited() incident_lookup.assert_not_awaited() bridge.assert_awaited_once() assert handoff["single_writer_executor"] == "Agent99" assert handoff["policy_route_id"] == "agent99_recover_after_owner_review" assert handoff["route_id"] == "agent99:host_recovery:Recover" assert bridge.await_args.kwargs["route_id"] == "agent99:host_recovery:Recover" assert bridge.await_args.kwargs["work_item_id"] == ( "agent99-dispatch:awoooi:INC-20260711-11C751:" "agent99:host_recovery:Recover" ) assert handoff["execution_priority"] == 30 assert handoff["queued"] is False assert handoff["side_effect_performed"] is False if operation_receipt_id: assert handoff["status"] == "cold_start_no_write_terminal" assert handoff["receipt_persisted"] is True assert handoff["no_write_terminal_receipt_id"] == ( "no-write-receipt-11c751" ) else: assert handoff["status"] == ( "cold_start_no_write_terminal_receipt_unavailable" ) assert handoff["receipt_persisted"] is False assert handoff["active_blockers"] == [ "durable_no_write_receipt_unavailable" ] assert handoff["runtime_execution_authorized"] is False assert handoff["runtime_closure_verified"] is False append.assert_awaited_once() receipt = append.await_args.kwargs assert receipt["action_detail"] == "controlled_no_write_terminal" assert receipt["success"] is None assert receipt["context"]["safe_next_action"] == ( "bind_exact_inventory_scope_before_creating_a_new_recover_candidate" ) assert receipt["context"]["no_write_terminal"]["reason"] == ( "exact_inventory_scope_missing" ) assert receipt["context"]["no_write_verifier_receipt"] == { "schema_version": "cold_start_no_write_source_verifier_receipt_v1", "verifier": "alert_webhook_no_write_terminal_validator", "status": "passed_source_contract", "producer_independent": True, "runtime_verifier": False, "runtime_closure_verified": False, } assert receipt["context"]["runtime_closure_verified"] is False @pytest.mark.asyncio async def test_webhook_router_queues_host188_callback_ansible_before_agent99( monkeypatch: pytest.MonkeyPatch, ) -> None: append = AsyncMock() incident = SimpleNamespace( incident_id="INC-H188-CALLBACK", project_id="awoooi", ) queue = AsyncMock( return_value={ "schema_version": "ai_decision_ansible_candidate_handoff_v1", "status": "ansible_candidate_queued", "queued": True, "side_effect_performed": False, "single_writer_executor": "awoooi-ansible-executor-broker", "automation_run_id": "run-h188-callback", "active_blockers": [], } ) bridge = AsyncMock( side_effect=AssertionError( "Docker Compose mutation must use Host188 Ansible, not Agent99" ) ) monkeypatch.setattr( "src.jobs.awooop_ansible_candidate_backfill_job.enqueue_ai_decision_ansible_candidate", queue, ) monkeypatch.setattr( "src.repositories.alert_operation_log_repository.get_alert_operation_log_repository", lambda: SimpleNamespace(append=append), ) monkeypatch.setattr( webhooks, "get_incident_service", lambda: SimpleNamespace( get_from_working_memory=AsyncMock(return_value=incident) ), ) monkeypatch.setattr(webhooks, "bridge_alertmanager_to_agent99", bridge) handoff = await webhooks._try_auto_repair_background( incident_id="INC-H188-CALLBACK", approval_id="00000000-0000-0000-0000-000000000188", alert_type="custom", target_resource="openclaw", namespace="default", risk_level="medium", source_alert_id="alert-h188-callback", source_alertname="TelegramCallbackForwarderDrift", source_severity="warning", source_message="callback ingress receipt missing", source_labels={ "alertname": "TelegramCallbackForwarderDrift", "component": "openclaw", "host": "188", "instance": "192.168.0.188", }, source_fingerprint="callback-fingerprint", alert_category="ai_agent", notification_type="TYPE-3", ) bridge.assert_not_awaited() queue.assert_awaited_once() proposal = queue.await_args.kwargs["proposal_data"] assert proposal["source"] == "alert_webhook_controlled_router" assert proposal["source_occurrence_id"] == "alert-h188-callback" assert handoff["queued"] is True assert handoff["single_writer_executor"] == "awoooi-ansible-executor-broker" assert handoff["execution_priority"] == 20 append.assert_awaited_once() assert append.await_args.kwargs["success"] is True