from __future__ import annotations import inspect import json from contextlib import asynccontextmanager from dataclasses import replace from datetime import UTC, datetime from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from src.core import redis_client as redis_client_module from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job from src.services import awooop_ansible_check_mode_service as service from src.services import telegram_gateway as telegram_gateway_module from src.services.awooop_ansible_audit_service import ( build_typed_target_route_generation, ) from src.services.telegram_gateway import TelegramGateway from src.workers import ansible_executor_broker as broker DESTINATION_BINDING = "a" * 64 PROVIDER_DESTINATION_BINDING = "b" * 64 PROVIDER_DESTINATION_VERIFICATION_METHOD = ( "requested_chat_id_matches_provider_chat_id" ) class _MappingResult: def __init__(self, row: dict | None = None, scalar=None) -> None: self._row = row self._scalar = scalar def mappings(self) -> _MappingResult: return self def one_or_none(self) -> dict | None: return self._row def scalar_one_or_none(self): return self._scalar def scalar(self): return self._scalar class _SequenceDB: def __init__(self, *results: _MappingResult) -> None: self._results = list(results) self.statements: list[str] = [] self.parameters: list[dict] = [] async def execute(self, statement, parameters=None): self.statements.append(str(statement)) self.parameters.append(dict(parameters or {})) return self._results.pop(0) if self._results else _MappingResult() class _RowsResult: def __init__(self, rows: list[dict]) -> None: self._rows = rows def mappings(self) -> _RowsResult: return self def all(self) -> list[dict]: return self._rows def _claim() -> service.AnsibleCheckModeClaim: typed_target_route = { "schema_version": "typed_domain_target_route_v2", "route_id": "typed:host_systemd:verified-closure", "resolution_status": "resolved", "executor": "host_ansible_executor", "verifier": "host_runtime_independent_verifier", "cross_domain_fallback_allowed": False, "allowed_catalog_ids": [ "ansible:awoooi-auto-repair-canary" ], "allowed_inventory_hosts": ["host_121"], "canonical_asset_id": "service:awoooi-auto-repair-canary", "target_kind": "host_systemd", "risk_class": "medium", "controlled_apply_allowed": True, "critical_break_glass_required": False, } generation_candidates = [ { "catalog_id": "ansible:awoooi-auto-repair-canary", "catalog_revision": "", "canonical_asset_id": "service:awoooi-auto-repair-canary", "typed_domain": "host_systemd", "inventory_hosts": ["host_121"], } ] return service.AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000101", source_candidate_op_id="00000000-0000-0000-0000-000000000100", incident_id="INC-20260711-D037E5", catalog_id="ansible:awoooi-auto-repair-canary", playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml", apply_playbook_path=( "infra/ansible/playbooks/awoooi-auto-repair-canary.yml" ), inventory_hosts=("host_121",), risk_level="medium", input_payload={ "automation_run_id": "00000000-0000-0000-0000-000000000102", "approval_id": "00000000-0000-0000-0000-000000000103", "candidate_catalog_schema": "typed_domain_router_v2", "target_route_generation": build_typed_target_route_generation( typed_target_route=typed_target_route, executor_candidates=generation_candidates, ), "typed_target_route": typed_target_route, "typed_generation_candidates": generation_candidates, "cross_domain_fallback_allowed": False, "single_writer_executor": "awoooi-ansible-executor-broker", "canonical_asset_id": "service:awoooi-auto-repair-canary", "typed_domain": "host_systemd", "target_scope_validated": True, "catalog_revision": "", }, ) def _verified_result() -> service.AnsibleRunResult: return service.AnsibleRunResult( returncode=0, stdout="", stderr="", duration_ms=25, post_verifier_passed=True, ) @pytest.mark.asyncio async def test_failed_apply_replay_rejects_legacy_claim_before_ansible( monkeypatch: pytest.MonkeyPatch, ) -> None: legacy_claim = _claim() legacy_claim.input_payload.pop("candidate_catalog_schema") monkeypatch.setattr( service, "_load_open_failed_apply_retry_row", AsyncMock(return_value={"op_id": "legacy-apply-op"}), ) monkeypatch.setattr( service, "_claim_from_apply_operation_row", lambda _row: ( legacy_claim, service.AnsibleRunResult( returncode=1, stdout="", stderr="failed", duration_ms=1, ), ), ) runner = AsyncMock( side_effect=AssertionError("legacy claim must not touch Ansible") ) monkeypatch.setattr(service, "_run_ansible_command", runner) result = await service.run_failed_apply_check_mode_replay_once( timeout_seconds=30, ) assert result["scanned"] == 1 assert result["error"] == "failed_apply_replay_typed_scope_not_verified" assert result["blockers"] == ["typed_candidate_catalog_schema_required"] assert result["runtime_apply_executed"] == 0 runner.assert_not_awaited() @pytest.mark.asyncio async def test_failed_apply_replay_rejects_foreign_playbook_before_ansible( monkeypatch: pytest.MonkeyPatch, ) -> None: foreign_claim = replace( _claim(), playbook_path="infra/ansible/playbooks/foreign-readonly.yml", ) monkeypatch.setattr( service, "_load_open_failed_apply_retry_row", AsyncMock(return_value={"op_id": "foreign-playbook-apply-op"}), ) monkeypatch.setattr( service, "_claim_from_apply_operation_row", lambda _row: ( foreign_claim, service.AnsibleRunResult( returncode=1, stdout="", stderr="failed", duration_ms=1, ), ), ) runner = AsyncMock( side_effect=AssertionError("foreign playbook must not touch Ansible") ) monkeypatch.setattr(service, "_run_ansible_command", runner) result = await service.run_failed_apply_check_mode_replay_once( timeout_seconds=30, ) assert result["scanned"] == 1 assert result["error"] == "failed_apply_replay_typed_scope_not_verified" assert result["blockers"] == [ "typed_claim_check_playbook_catalog_mismatch" ] assert result["runtime_apply_executed"] == 0 runner.assert_not_awaited() @pytest.mark.asyncio async def test_terminal_apply_replaces_queued_approval_truth( monkeypatch: pytest.MonkeyPatch, ) -> None: approval = SimpleNamespace( status=service.ApprovalStatus.APPROVED, resolved_at=None, rejection_reason=None, extra_metadata={ "execution_kind": "host_ansible_check_mode_queued", "execution_state": "queued", "repair_executed": False, "repair_verified": False, "incident_closure_allowed": False, }, ) db = _SequenceDB(_MappingResult(scalar=approval)) @asynccontextmanager async def fake_get_db_context(_project_id): yield db monkeypatch.setattr(service, "get_db_context", fake_get_db_context) written = await service._finalize_controlled_approval_projection( _claim(), _verified_result(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", ) assert written is True assert approval.status == service.ApprovalStatus.EXECUTION_SUCCESS assert approval.extra_metadata["execution_kind"] == "controlled_apply" assert approval.extra_metadata["execution_state"] == "completed" assert approval.extra_metadata["repair_executed"] is True assert approval.extra_metadata["repair_verified"] is True assert approval.extra_metadata["incident_closure_allowed"] is True def _controlled_result_payload() -> dict: return { "chat_id": "sre-chat", "text": "CONTROLLED APPLY RESULT INC-20260711-D037E5", "_awoooi_canonical_telegram_route": { "product_id": "awoooi", "signal_family": "incident_lifecycle", "severity": "P1", }, "_skip_incident_thread_reply": True, "_awooop_source_envelope_extra": { "callback_reply": { "action": "controlled_apply_result", "status": "callback_reply_sent", "project_id": "awoooi", "automation_run_id": ( "00000000-0000-0000-0000-000000000102" ), "incident_id": "INC-20260711-D037E5", "apply_op_id": "00000000-0000-0000-0000-000000000104", } }, } def test_apply_result_prefers_safe_playbook_marker_over_stderr_warning() -> None: result = service.AnsibleRunResult( returncode=2, stdout=( "TASK [Require successful bounded Compose activation]\n" "fatal: [host_188]: FAILED! => {\"msg\": " "\"openclaw_compose_activation_failed:bind_source_missing\"}" ), stderr=( "[WARNING]: Platform linux on host host_188 is using the " "discovered Python interpreter." ), duration_ms=123, ) status, output, dry_run_result, error = ( service._build_apply_result_payload(result) ) assert status == "failed" assert error == ( "openclaw_compose_activation_failed:bind_source_missing" ) assert output["stderr_tail"].startswith("[WARNING]") assert dry_run_result["returncode"] == 2 def test_check_mode_result_projects_safe_playbook_failure_contract() -> None: result = service.AnsibleRunResult( returncode=2, stdout=( "TASK [Fail closed when a payload base is not hash-compatible]\n" "fatal: [host_188]: FAILED! => {\"msg\": " "\"openclaw_payload_base_hash_mismatch; " "drift_work_item=DRIFT-H188-OPENCLAW-SOURCE-001\"}" ), stderr="[WARNING]: interpreter discovery warning", duration_ms=123, ) status, output, dry_run_result, error = service._build_result_payload(result) assert status == "failed" assert error == "openclaw_payload_base_hash_mismatch" assert output["failure_class"] == "playbook_contract" assert output["failure_reason"] == "openclaw_payload_base_hash_mismatch" assert dry_run_result["failure_class"] == "playbook_contract" assert dry_run_result["failure_reason"] == "openclaw_payload_base_hash_mismatch" def test_check_mode_result_projects_fixed_transport_reason_without_raw_output() -> None: result = service.AnsibleRunResult( returncode=4, stdout=( "host_188 | UNREACHABLE! => {\"msg\": " "\"Failed to connect to the host via ssh: private detail\"}" ), stderr="", duration_ms=123, ) _status, output, dry_run_result, error = service._build_result_payload(result) assert error == "ansible_check_mode_failed_rc_4" assert output["failure_class"] == "transport" assert output["failure_reason"] == "transport_unreachable" assert dry_run_result["failure_reason"] == "transport_unreachable" assert "private detail" not in output["failure_reason"] def test_apply_result_does_not_project_untrusted_stdout_as_error() -> None: result = service.AnsibleRunResult( returncode=2, stdout="fatal: [host_188]: FAILED! => secret-shaped-untrusted-output", stderr="safe_transport_failure", duration_ms=123, ) _status, _output, _dry_run_result, error = ( service._build_apply_result_payload(result) ) assert error == "safe_transport_failure" def test_runtime_stage_ids_only_accepts_durable_receipts() -> None: stage_ids = service._runtime_stage_ids( { "runtime_stage_receipts": [ { "stage_id": "mcp_context", "durable_receipt": True, }, { "stage_id": "telegram_receipt", "durable_receipt": False, }, ] } ) assert stage_ids == {"mcp_context"} def test_incident_terminal_writer_normalizes_legacy_non_object_outcome() -> None: source = inspect.getsource(service._record_incident_terminal_disposition) assert "jsonb_typeof(CAST(outcome AS jsonb))" in source assert "jsonb_build_object(" in source assert "'legacy_outcome'" in source @pytest.mark.asyncio @pytest.mark.parametrize( ("proposal_executed", "expected_success"), [(True, False), (False, None)], ) async def test_incident_terminal_round_trips_negative_outcome_truth( monkeypatch: pytest.MonkeyPatch, proposal_executed: bool, expected_success: bool | None, ) -> None: terminal = { "schema_version": "ansible_incident_terminal_disposition_v1", "automation_run_id": "00000000-0000-0000-0000-000000000102", "apply_op_id": "00000000-0000-0000-0000-000000000104", "terminal_type": "no_write_replay_failed_waiting_repair_candidate", } db = _SequenceDB( _MappingResult({"proposal_executed": proposal_executed}), _MappingResult( { "incident_status": "MITIGATING", "updated_at": None, "resolved_at": None, "outcome": { "automation_terminal": terminal, "proposal_executed": proposal_executed, "execution_success": expected_success, }, } ), ) @asynccontextmanager async def fake_db_context(_project_id: str): yield db monkeypatch.setattr(service, "get_db_context", fake_db_context) result = await service._record_incident_terminal_disposition( _claim(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", terminal_type=terminal["terminal_type"], success_terminal=False, retry_op_id=None, telegram_receipt_acknowledged=False, ) assert result is not None assert result["proposal_executed"] is proposal_executed assert result["execution_success"] is expected_success assert db.parameters[1]["execution_success"] is expected_success @pytest.mark.asyncio async def test_closure_refuses_to_resolve_when_any_receipt_is_missing( monkeypatch: pytest.MonkeyPatch, ) -> None: atomic_writer = AsyncMock() monkeypatch.setattr( service, "_read_verified_apply_closure_prerequisites", AsyncMock( return_value={ "ready": False, "missing": ["telegram_receipt"], "receipts": {}, } ), ) monkeypatch.setattr( service, "_commit_verified_incident_closure", atomic_writer, ) result = await service._finalize_verified_apply_closure( _claim(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", ) assert result == { "status": "closure_receipts_pending", "closed": False, "missing": ["telegram_receipt"], } atomic_writer.assert_not_awaited() @pytest.mark.asyncio async def test_closure_resolves_only_after_durable_readback( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( service, "_read_verified_apply_closure_prerequisites", AsyncMock(return_value={"ready": True, "receipts": {"all": True}}), ) atomic_writer = AsyncMock( return_value={ "automation_run_id": "00000000-0000-0000-0000-000000000102", "apply_op_id": "00000000-0000-0000-0000-000000000104", "incident_resolved": True, "closure_receipt_written": True, "resolved_lifecycle_written": True, "working_memory_projection_ready": True, } ) monkeypatch.setattr( service, "_commit_verified_incident_closure", atomic_writer, ) monkeypatch.setattr( service, "_read_incident_closure_readback", AsyncMock( return_value={ "closed": True, "missing": [], "incident_status": "RESOLVED", } ), ) result = await service._finalize_verified_apply_closure( _claim(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", ) assert result["status"] == "controlled_apply_closed" assert result["closed"] is True atomic_writer.assert_awaited_once() assert atomic_writer.await_args.kwargs["closure_prerequisite_count"] == 1 @pytest.mark.asyncio async def test_closure_preserves_durable_terminal_when_projection_is_degraded( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( service, "_read_verified_apply_closure_prerequisites", AsyncMock(return_value={"ready": True, "receipts": {"all": True}}), ) monkeypatch.setattr( service, "_commit_verified_incident_closure", AsyncMock(return_value={"working_memory_projection_ready": False}), ) readback = AsyncMock( return_value={ "closed": True, "missing": [], "incident_status": "RESOLVED", } ) monkeypatch.setattr(service, "_read_incident_closure_readback", readback) result = await service._finalize_verified_apply_closure( _claim(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", ) assert result == { "status": "controlled_apply_closed_projection_degraded", "closed": True, "missing": ["working_memory_terminal_projection"], "incident_status": "RESOLVED", "working_memory_projection_ready": False, "read_model_degraded": True, } readback.assert_awaited_once() @pytest.mark.asyncio async def test_terminal_working_memory_reconciler_projects_durable_rows( monkeypatch: pytest.MonkeyPatch, ) -> None: service._WORKING_MEMORY_PROJECTION_CURSOR.clear() updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC) db = _SequenceDB( _RowsResult( [ { "incident_id": "INC-20260711-C50677", "incident_status": "RESOLVED", "updated_at": updated_at, "resolved_at": updated_at, }, { "incident_id": "INC-20260711-8A28BC", "incident_status": "RESOLVED", "updated_at": updated_at, "resolved_at": updated_at, }, ] ) ) @asynccontextmanager async def fake_db_context(_project_id: str): yield db project = AsyncMock(side_effect=["updated", "current"]) monkeypatch.setattr(service, "get_db_context", fake_db_context) monkeypatch.setattr( service, "_project_incident_terminal_to_working_memory", project, ) result = await service.reconcile_verified_incident_working_memory_once( project_id="awoooi", window_hours=24, limit=6, ) assert result == { "scanned": 2, "updated": 1, "current": 1, "missing": 0, "failed": 0, "error": None, } assert project.await_count == 2 assert db.parameters[0] == { "project_id": "awoooi", "window_hours": 24, "limit": 6, "cursor_updated_at": None, "cursor_incident_id": "", } query = db.statements[0] assert "working_memory_projection,ready" in query assert "source_updated_at_epoch_us" in query assert "cursor_updated_at" in query assert "ORDER BY incident.updated_at ASC" in query assert service._WORKING_MEMORY_PROJECTION_WINDOW_HOURS >= 7 * 24 service._WORKING_MEMORY_PROJECTION_CURSOR.clear() @pytest.mark.asyncio async def test_terminal_reconciler_advances_cursor_after_one_row_fails( monkeypatch: pytest.MonkeyPatch, ) -> None: service._WORKING_MEMORY_PROJECTION_CURSOR.clear() first_at = datetime(2026, 7, 11, 11, 40, tzinfo=UTC) second_at = datetime(2026, 7, 11, 11, 41, tzinfo=UTC) db = _SequenceDB( _RowsResult( [ { "incident_id": "INC-20260711-FAIL01", "incident_status": "RESOLVED", "updated_at": first_at, "resolved_at": first_at, }, { "incident_id": "INC-20260711-NEXT01", "incident_status": "RESOLVED", "updated_at": second_at, "resolved_at": second_at, }, ] ) ) @asynccontextmanager async def fake_db_context(_project_id: str): yield db project = AsyncMock( side_effect=[RuntimeError("redis unavailable"), "updated"] ) monkeypatch.setattr(service, "get_db_context", fake_db_context) monkeypatch.setattr( service, "_project_incident_terminal_to_working_memory", project, ) result = await service.reconcile_verified_incident_working_memory_once( project_id="awoooi", limit=2, ) assert result["scanned"] == 2 assert result["failed"] == 1 assert result["updated"] == 1 assert project.await_count == 2 assert service._WORKING_MEMORY_PROJECTION_CURSOR["awoooi"] == ( second_at, "INC-20260711-NEXT01", ) service._WORKING_MEMORY_PROJECTION_CURSOR.clear() @pytest.mark.asyncio async def test_projection_exception_does_not_erase_durable_closure( monkeypatch: pytest.MonkeyPatch, ) -> None: terminal = { "automation_run_id": "00000000-0000-0000-0000-000000000102", "apply_op_id": "00000000-0000-0000-0000-000000000104", } updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC) db = _SequenceDB( _MappingResult(), _MappingResult( { "incident_status": "RESOLVED", "updated_at": updated_at, "resolved_at": updated_at, "outcome": { "automation_terminal": terminal, "proposal_executed": True, "execution_success": True, }, "closure_receipt_written": True, "resolved_lifecycle_written": True, } ), ) transaction_completed = False @asynccontextmanager async def fake_db_context(_project_id: str): nonlocal transaction_completed yield db transaction_completed = True monkeypatch.setattr(service, "get_db_context", fake_db_context) monkeypatch.setattr( service, "_project_incident_terminal_to_working_memory", AsyncMock(side_effect=RuntimeError("redis unavailable")), ) result = await service._commit_verified_incident_closure( _claim(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", closure_prerequisite_count=12, ) assert transaction_completed is True assert result is not None assert result["incident_resolved"] is True assert result["closure_receipt_written"] is True assert result["working_memory_projection_status"] == "failed" assert result["working_memory_projection_ready"] is False @pytest.mark.asyncio async def test_broker_redis_projection_init_is_fail_soft( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( broker, "init_redis_pool", AsyncMock(side_effect=RuntimeError("redis unavailable")), ) assert await broker._init_optional_projection_redis() is False @pytest.mark.asyncio async def test_failed_redis_init_does_not_poison_projection_retry( monkeypatch: pytest.MonkeyPatch, ) -> None: failed_pool = type( "_FailedPool", (), { "ping": AsyncMock(side_effect=RuntimeError("redis unavailable")), "close": AsyncMock(), }, )() monkeypatch.setattr(redis_client_module, "_redis_pool", None) monkeypatch.setattr( redis_client_module.redis, "from_url", lambda *_args, **_kwargs: failed_pool, ) with pytest.raises(RuntimeError, match="redis unavailable"): await redis_client_module.init_redis_pool() assert redis_client_module._redis_pool is None failed_pool.close.assert_awaited_once_with() @pytest.mark.asyncio async def test_atomic_closure_rolls_back_when_bundle_readback_is_missing( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB(_MappingResult(), _MappingResult(None)) transaction_rolled_back = False @asynccontextmanager async def fake_db_context(_project_id: str): nonlocal transaction_rolled_back try: yield db except Exception: transaction_rolled_back = True raise monkeypatch.setattr(service, "get_db_context", fake_db_context) result = await service._commit_verified_incident_closure( _claim(), apply_op_id="00000000-0000-0000-0000-000000000104", project_id="awoooi", closure_prerequisite_count=12, ) assert result is None assert transaction_rolled_back is True atomic_statement = db.statements[1] assert "INSERT INTO alert_operation_log" in atomic_statement assert "UPDATE automation_operation_log apply" in atomic_statement assert "UPDATE incidents incident" in atomic_statement assert "WITH target_apply AS" in atomic_statement assert "WHEN upper(incident.status::text) = 'CLOSED'" in atomic_statement assert db.parameters[1]["apply_op_id_uuid"] == db.parameters[1]["apply_op_id"] def test_uuid_lookup_and_json_text_receipt_binds_are_type_isolated() -> None: mixed_apply_id_queries = ( service._read_verified_apply_closure_prerequisites, service._read_incident_closure_readback, service._commit_verified_incident_closure, service._verify_retry_terminal_projection_readback, ) for query_owner in mixed_apply_id_queries: source = inspect.getsource(query_owner) assert "CAST(:apply_op_id_uuid AS uuid)" in source assert "CAST(:apply_op_id AS uuid)" not in source assert '"apply_op_id_uuid": apply_op_id' in source retry_source = inspect.getsource( service._verify_retry_terminal_projection_readback ) assert "CAST(:replay_op_id_uuid AS uuid)" in retry_source assert "CAST(:replay_op_id AS uuid)" not in retry_source assert '"replay_op_id_uuid": replay_op_id' in retry_source def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None: predicate = candidate_job.verified_ansible_candidate_terminal_sql( "candidate" ) recorder_source = inspect.getsource( candidate_job.record_ansible_decision_audit ) assert "direct_terminal.status = 'dry_run'" in predicate assert "check_mode.parent_op_id = candidate.op_id" in predicate assert "apply.parent_op_id = check_mode.op_id" in predicate assert "replay.parent_op_id = apply.op_id" in predicate assert "replay.status IN ('success', 'failed')" in predicate assert ( "replay_check.parent_op_id = failed_check.parent_op_id" in predicate ) assert "replay_of_check_mode_op_id" in predicate assert "failed_check.parent_op_id = candidate.op_id" in predicate assert "failed_check.op_id = CAST(" in predicate assert predicate.count("failed_check.op_id::text") == 1 assert "verified_no_write_terminal" in predicate assert "repository_readback_verified" in predicate assert "{detail,execution_success}" in predicate assert "{detail,telegram_receipt_acknowledged}" in predicate assert "ORDER BY candidate.created_at DESC, candidate.op_id DESC" in ( recorder_source ) assert "target_route_generation" in recorder_source assert "candidate_lane_lock_key" in recorder_source @pytest.mark.asyncio async def test_projection_replay_sends_missing_receipt_without_reapplying( monkeypatch: pytest.MonkeyPatch, ) -> None: closure_coverage = { "telegram_receipt": False, "candidate": True, "check_mode": True, "controlled_apply": True, "post_apply_verifier": True, "auto_repair_execution_receipt": True, "km_playbook_writeback": True, "rag_writeback": True, "mcp_context": True, "playbook_trust": True, "timeline_projection": True, "approval_projection": True, "execution_lifecycle": True, "required_runtime_stage_receipts": True, } readback = AsyncMock( side_effect=[ {"receipts": closure_coverage}, {"receipts": {"telegram_receipt": True}}, ] ) telegram_sender = AsyncMock(return_value={"ok": True}) monkeypatch.setattr( service, "_finalize_controlled_approval_projection", AsyncMock(return_value=True), ) monkeypatch.setattr( service, "_append_alert_lifecycle_receipt", AsyncMock(return_value=True), ) monkeypatch.setattr( service, "_read_verified_apply_closure_prerequisites", readback, ) monkeypatch.setattr( service, "_send_controlled_apply_telegram_receipt", telegram_sender, ) monkeypatch.setattr( service, "_finalize_verified_apply_closure", AsyncMock(return_value={"status": "controlled_apply_closed", "closed": True}), ) result = await service._reconcile_verified_apply_closure_projections( _claim(), _verified_result(), apply_op_id="00000000-0000-0000-0000-000000000104", writeback={ "verification_passed": True, "verification_result": "success", "verification": True, "learning": True, }, project_id="awoooi", ) assert result["closed"] is True assert result["runtime_apply_executed"] is False telegram_sender.assert_awaited_once() assert telegram_sender.await_args.kwargs["closure_receipts"] == ( closure_coverage ) @pytest.mark.asyncio async def test_alert_chain_final_receipt_projects_all_learning_evidence( monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = AsyncMock() gateway.send_controlled_apply_result_receipt.return_value = { "ok": True, "_awooop_outbound_mirror_acknowledged": True, } monkeypatch.setattr( "src.services.telegram_gateway.get_telegram_gateway", lambda: gateway, ) base = _claim() claim = replace( base, catalog_id="ansible:110-alertmanager-delivery-recovery", playbook_path=( "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml" ), apply_playbook_path=( "infra/ansible/playbooks/110-alertmanager-delivery-recovery.yml" ), input_payload={ **base.input_payload, "alert_category": "alert_chain_health", }, ) acknowledged = await service._send_controlled_apply_telegram_receipt( claim, _verified_result(), apply_op_id="00000000-0000-0000-0000-000000000104", writeback={ "verification_result": "success", "verification": True, "learning": True, "rag_writeback": True, "trust_learning": True, }, closure_receipts={ "km_playbook_writeback": True, "rag_writeback": True, "mcp_context": True, "playbook_trust": True, }, project_id="awoooi", ) assert acknowledged is True call = gateway.send_controlled_apply_result_receipt.await_args.kwargs assert call["alert_category"] == "alert_chain_health" assert call["verifier_written"] is True assert call["learning_written"] is True assert call["rag_written"] is True assert call["mcp_context_ready"] is True assert call["playbook_trust_written"] is True @pytest.mark.asyncio async def test_projection_stays_open_until_durable_telegram_readback( monkeypatch: pytest.MonkeyPatch, ) -> None: lifecycle = AsyncMock(return_value=True) monkeypatch.setattr( service, "_finalize_controlled_approval_projection", AsyncMock(return_value=True), ) monkeypatch.setattr(service, "_append_alert_lifecycle_receipt", lifecycle) monkeypatch.setattr( service, "_read_verified_apply_closure_prerequisites", AsyncMock( side_effect=[ {"receipts": {"telegram_receipt": False}}, {"receipts": {"telegram_receipt": False}}, ] ), ) telegram_sender = AsyncMock(return_value=False) monkeypatch.setattr( service, "_send_controlled_apply_telegram_receipt", telegram_sender, ) monkeypatch.setattr( service, "_finalize_verified_apply_closure", AsyncMock( return_value={ "status": "closure_receipts_pending", "closed": False, "missing": ["telegram_receipt"], } ), ) result = await service._reconcile_verified_apply_closure_projections( _claim(), _verified_result(), apply_op_id="00000000-0000-0000-0000-000000000104", writeback={ "verification_passed": True, "verification_result": "success", "verification": True, "learning": True, }, project_id="awoooi", ) assert result["closed"] is False assert result["telegram_receipt_acknowledged"] is False assert result["telegram_lifecycle_written"] is False assert lifecycle.await_count == 1 telegram_sender.assert_not_awaited() @pytest.mark.asyncio async def test_controlled_result_pending_reservation_blocks_duplicate_provider_send( monkeypatch: pytest.MonkeyPatch, ) -> None: class _Response: def raise_for_status(self) -> None: return None def json(self) -> dict: return { "ok": True, "result": { "message_id": 456, "chat": {"id": "sre-chat"}, }, } class _Client: def __init__(self) -> None: self.post_count = 0 async def post(self, _url, json): self.post_count += 1 return _Response() gateway = TelegramGateway() monkeypatch.setattr( telegram_gateway_module.settings, "SRE_GROUP_CHAT_ID", "sre-chat", ) gateway._initialized = True client = _Client() gateway._http_client = client reserve = AsyncMock( side_effect=[ { "status": "reserved", "run_id": "00000000-0000-0000-0000-000000000201", "message_id": "00000000-0000-0000-0000-000000000202", }, { "status": "pending_unknown", "run_id": "00000000-0000-0000-0000-000000000201", "message_id": "00000000-0000-0000-0000-000000000202", "durable_reservation_committed": True, "provider_retry_blocked": True, }, ] ) finalize = AsyncMock(return_value=False) monkeypatch.setattr( gateway, "_reserve_controlled_apply_result_outbound", reserve, ) monkeypatch.setattr( gateway, "_finalize_controlled_apply_result_outbound", finalize, ) first = await gateway._send_request( "sendMessage", _controlled_result_payload(), ) second = await gateway._send_request( "sendMessage", _controlled_result_payload(), ) assert client.post_count == 1 assert first["_awooop_provider_send_performed"] is True assert first["_awooop_outbound_mirror_acknowledged"] is False assert second["ok"] is False assert second["_awooop_provider_send_performed"] is False assert second["_awooop_delivery_status"] == "pending_unknown" assert finalize.await_count == 1 @pytest.mark.asyncio async def test_durable_reconcile_finalizes_exact_pending_row_without_provider_send( monkeypatch: pytest.MonkeyPatch, ) -> None: pending_row = { "message_id": "00000000-0000-0000-0000-000000000202", "send_status": "pending", "provider_message_id": None, "destination_binding": DESTINATION_BINDING, "visual_requested": True, } sent_row = { **pending_row, "send_status": "sent", "provider_message_id": "456", "destination_binding": DESTINATION_BINDING, "provider_destination_binding": PROVIDER_DESTINATION_BINDING, "provider_destination_verification_method": ( PROVIDER_DESTINATION_VERIFICATION_METHOD ), "provider_destination_identity_verified": True, } db = _SequenceDB( _MappingResult(row=pending_row), _MappingResult(), _MappingResult(), _MappingResult( row={"send_status": "sent", "provider_message_id": "456"} ), _MappingResult(row=sent_row), ) @asynccontextmanager async def fake_get_db_context(_project_id): yield db monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) gateway = TelegramGateway() gateway._http_client = AsyncMock() identity = { "project_id": "awoooi", "automation_run_id": "agent99-lifecycle-exact-readback", "incident_id": "INC-20260716-AG99ACK", "apply_op_id": "recovered|info", "delivery_kind": "agent99_lifecycle", } result = await gateway._reconcile_controlled_apply_result_outbound( identity=identity, expected_provider_message_id="456", expected_destination_binding=DESTINATION_BINDING, expected_provider_destination_binding=PROVIDER_DESTINATION_BINDING, expected_provider_destination_verification_method=( PROVIDER_DESTINATION_VERIFICATION_METHOD ), ) assert result == { "status": "sent", "run_id": str( telegram_gateway_module._controlled_apply_result_delivery_run_id( identity ) ), "message_id": pending_row["message_id"], "provider_message_id": "456", "destination_binding": DESTINATION_BINDING, "provider_destination_binding": PROVIDER_DESTINATION_BINDING, "provider_destination_verification_method": ( PROVIDER_DESTINATION_VERIFICATION_METHOD ), "provider_destination_identity_verified": True, "visual_requested": True, } gateway._http_client.post.assert_not_awaited() assert len(db.statements) == 5 assert "{callback_reply,automation_run_id}" in db.statements[0] assert "provider_destination_binding" in db.statements[0] assert "provider_destination_verification_method" in db.statements[0] assert "provider_destination_identity_verified" in db.statements[0] assert db.parameters[0]["automation_run_id"] == identity["automation_run_id"] assert db.parameters[0]["incident_id"] == identity["incident_id"] assert db.parameters[0]["apply_op_id"] == identity["apply_op_id"] assert "telegram-destination-pin:" in db.parameters[2][ "destination_lock_key" ] assert "UPDATE awooop_outbound_message" in db.statements[3] @pytest.mark.asyncio async def test_controlled_result_finalize_backfills_missing_destination_binding( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB( _MappingResult(), _MappingResult(), _MappingResult(row={"send_status": "sent", "provider_message_id": "456"}), ) @asynccontextmanager async def fake_get_db_context(_project_id): yield db monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) gateway = TelegramGateway() identity = { "project_id": "awoooi", "automation_run_id": "agent99-lifecycle-missing-binding", "incident_id": "INC-20260717-WAZUHACK", "apply_op_id": "00000000-0000-0000-0000-000000000104", "delivery_kind": "controlled_apply_result", } finalized = await gateway._finalize_controlled_apply_result_outbound( identity=identity, reservation={ "message_id": "00000000-0000-0000-0000-000000000202", "run_id": str( telegram_gateway_module._controlled_apply_result_delivery_run_id( identity ) ), }, provider_message_id="456", expected_destination_binding=DESTINATION_BINDING, provider_destination_binding=PROVIDER_DESTINATION_BINDING, provider_destination_verification_method=( PROVIDER_DESTINATION_VERIFICATION_METHOD ), ) assert finalized is True update_sql = db.statements[2] assert "{canonical_route_receipt,destination_binding}" in update_sql assert "COALESCE(" in update_sql assert "IN ('', :expected_destination_binding)" in update_sql assert "CAST(" in update_sql assert ":expected_destination_binding\n" in update_sql assert "AS text" in update_sql assert ":expected_destination_binding::text" not in update_sql assert "NOT EXISTS" in update_sql assert "provider_destination_binding" in update_sql assert db.parameters[2]["expected_destination_binding"] == ( DESTINATION_BINDING ) assert db.parameters[2]["provider_destination_binding"] == ( PROVIDER_DESTINATION_BINDING ) assert db.parameters[2][ "provider_destination_verification_method" ] == PROVIDER_DESTINATION_VERIFICATION_METHOD @pytest.mark.asyncio async def test_durable_reconcile_rejects_provider_message_identity_mismatch( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB( _MappingResult( row={ "message_id": "00000000-0000-0000-0000-000000000202", "send_status": "sent", "provider_message_id": "456", "visual_requested": False, } ) ) @asynccontextmanager async def fake_get_db_context(_project_id): yield db monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) gateway = TelegramGateway() gateway._http_client = AsyncMock() identity = { "project_id": "awoooi", "automation_run_id": "agent99-lifecycle-provider-mismatch", "incident_id": "INC-20260716-AG99ACK", "apply_op_id": "recovered|info", "delivery_kind": "agent99_lifecycle", } result = await gateway._reconcile_controlled_apply_result_outbound( identity=identity, expected_provider_message_id="999", ) assert result["status"] == "provider_message_id_mismatch" assert result["provider_message_id"] == "456" assert len(db.statements) == 1 gateway._http_client.post.assert_not_awaited() @pytest.mark.asyncio async def test_controlled_result_reuses_durable_sent_reservation( monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = TelegramGateway() monkeypatch.setattr( telegram_gateway_module.settings, "SRE_GROUP_CHAT_ID", "sre-chat", ) gateway._initialized = True client = AsyncMock() gateway._http_client = client destination_binding = ( telegram_gateway_module._telegram_destination_binding("sre-chat") ) monkeypatch.setattr( gateway, "_reserve_controlled_apply_result_outbound", AsyncMock( return_value={ "status": "sent", "run_id": "00000000-0000-0000-0000-000000000201", "message_id": "00000000-0000-0000-0000-000000000202", "provider_message_id": "456", "destination_binding": destination_binding, "provider_destination_binding": destination_binding, "provider_destination_verification_method": ( PROVIDER_DESTINATION_VERIFICATION_METHOD ), "provider_destination_identity_verified": True, } ), ) result = await gateway._send_request( "sendMessage", _controlled_result_payload(), ) assert result["ok"] is True assert result["_awooop_outbound_mirror_acknowledged"] is True assert result["_awooop_provider_send_performed"] is False assert result["_awooop_delivery_context"] == { "destination_binding": destination_binding, "provider_destination_binding": destination_binding, "provider_destination_verification_method": ( PROVIDER_DESTINATION_VERIFICATION_METHOD ), "provider_destination_identity_verified": True, "destination_binding_verified": True, "durable_exact_row_readback": True, } client.post.assert_not_awaited() @pytest.mark.asyncio async def test_no_write_result_reuses_durable_suppression_without_provider_send( monkeypatch: pytest.MonkeyPatch, ) -> None: gateway = TelegramGateway() monkeypatch.setattr( telegram_gateway_module.settings, "SRE_GROUP_CHAT_ID", "sre-chat", ) gateway._initialized = True client = AsyncMock() gateway._http_client = client monkeypatch.setattr( gateway, "_reserve_controlled_apply_result_outbound", AsyncMock( return_value={ "status": "suppressed", "run_id": "00000000-0000-0000-0000-000000000201", "message_id": "00000000-0000-0000-0000-000000000202", } ), ) result = await gateway._send_request( "sendMessage", _controlled_result_payload(), ) assert result["ok"] is True assert result["_awooop_outbound_mirror_acknowledged"] is True assert result["_awooop_provider_send_performed"] is False assert result["_awooop_notification_suppressed"] is True assert result["_awooop_delivery_status"] == "suppressed" client.post.assert_not_awaited() @pytest.mark.asyncio async def test_no_write_group_reservation_persists_shadow_suppression( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB( _MappingResult(), _MappingResult(row=None), _MappingResult(scalar="00000000-0000-0000-0000-000000000299"), ) @asynccontextmanager async def fake_get_db_context(_project_id): yield db record_outbound = AsyncMock( return_value="00000000-0000-0000-0000-000000000298" ) monkeypatch.setattr( "src.db.base.get_db_context", fake_get_db_context, ) monkeypatch.setattr( "src.services.channel_hub.record_outbound_message", record_outbound, ) gateway = TelegramGateway() identity = { "project_id": "awoooi", "automation_run_id": "00000000-0000-0000-0000-000000000102", "incident_id": "INC-20260711-D037E5", "apply_op_id": "00000000-0000-0000-0000-000000000104", } source_extra = { "callback_reply": { "action": "controlled_apply_result", "execution_kind": "no_write_replay", **identity, }, "notification_policy": { "failure_fingerprint": "deadbeefdeadbeefdeadbeef", "digest_window_minutes": 30, "disposition": "send_or_suppress", }, } reservation = await gateway._reserve_controlled_apply_result_outbound( identity=identity, payload={"chat_id": "sre-chat", "text": "AI repair digest"}, source_envelope_extra=source_extra, ) assert reservation["status"] == "suppressed" assert reservation["message_id"] == ( "00000000-0000-0000-0000-000000000298" ) assert "send_status IN ('pending', 'sent')" in db.statements[2] assert db.parameters[0]["lock_key"].endswith( ":deadbeefdeadbeefdeadbeef" ) record_kwargs = record_outbound.await_args.kwargs assert record_kwargs["send_status"] == "shadow" assert record_kwargs["provider_message_id"] is None assert record_kwargs["is_shadow"] is True assert record_kwargs["source_envelope"]["notification_policy"][ "disposition" ] == "suppressed" @pytest.mark.asyncio async def test_historical_no_write_backfill_is_provider_silent( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB( _MappingResult(), _MappingResult(row=None), ) @asynccontextmanager async def fake_get_db_context(_project_id): yield db record_outbound = AsyncMock( return_value="00000000-0000-0000-0000-000000000398" ) monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) monkeypatch.setattr( "src.services.channel_hub.record_outbound_message", record_outbound, ) gateway = TelegramGateway() identity = { "project_id": "awoooi", "automation_run_id": "00000000-0000-0000-0000-000000000302", "incident_id": "INC-20260711-HISTORICAL", "apply_op_id": "00000000-0000-0000-0000-000000000304", } source_extra = { "callback_reply": { "action": "controlled_apply_result", "execution_kind": "no_write_replay", **identity, }, "notification_policy": { "failure_fingerprint": "feedfacefeedfacefeedface", "digest_window_minutes": 30, "disposition": "send_or_suppress", "provider_delivery": "shadow_only", }, } reservation = await gateway._reserve_controlled_apply_result_outbound( identity=identity, payload={"chat_id": "sre-chat", "text": "historical repair digest"}, source_envelope_extra=source_extra, ) assert reservation["status"] == "suppressed" assert len(db.statements) == 2 assert all( "send_status IN ('pending', 'sent')" not in sql for sql in db.statements ) record_kwargs = record_outbound.await_args.kwargs assert record_kwargs["send_status"] == "shadow" assert record_kwargs["provider_message_id"] is None policy = record_kwargs["source_envelope"]["notification_policy"] assert policy["disposition"] == "suppressed" assert policy["provider_send_performed"] is False assert policy["suppression_reason"] == "historical_projection_backfill" @pytest.mark.asyncio async def test_historical_verified_repair_is_provider_silent( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB( _MappingResult(), _MappingResult(row=None), ) @asynccontextmanager async def fake_get_db_context(_project_id): yield db record_outbound = AsyncMock( return_value="00000000-0000-0000-0000-000000000498" ) monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) monkeypatch.setattr( "src.services.channel_hub.record_outbound_message", record_outbound, ) gateway = TelegramGateway() identity = { "project_id": "awoooi", "automation_run_id": "00000000-0000-0000-0000-000000000402", "incident_id": "INC-20260711-HISTORICAL-SUCCESS", "apply_op_id": "00000000-0000-0000-0000-000000000404", } source_extra = { "callback_reply": { "action": "controlled_apply_result", "execution_kind": "controlled_apply", **identity, }, "notification_policy": { "failure_fingerprint": "", "disposition": "suppressed", "provider_delivery": "shadow_only", }, } reservation = await gateway._reserve_controlled_apply_result_outbound( identity=identity, payload={"chat_id": "sre-chat", "text": "historical repair closed"}, source_envelope_extra=source_extra, ) assert reservation["status"] == "suppressed" assert len(db.statements) == 2 record_kwargs = record_outbound.await_args.kwargs assert record_kwargs["send_status"] == "shadow" assert record_kwargs["provider_message_id"] is None policy = record_kwargs["source_envelope"]["notification_policy"] assert policy["suppression_reason"] == "historical_projection_backfill" assert policy["provider_send_performed"] is False @pytest.mark.asyncio async def test_new_reservation_uses_committed_context_as_durable_receipt( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB( _MappingResult(), _MappingResult(row=None), ) context_exited = False @asynccontextmanager async def fake_get_db_context(_project_id): nonlocal context_exited yield db context_exited = True record_outbound = AsyncMock( return_value="00000000-0000-0000-0000-000000000598" ) monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) monkeypatch.setattr( "src.services.channel_hub.record_outbound_message", record_outbound, ) gateway = TelegramGateway() identity = { "project_id": "awoooi", "automation_run_id": "00000000-0000-0000-0000-000000000502", "incident_id": "INC-20260715-RESERVATION", "apply_op_id": "00000000-0000-0000-0000-000000000504", } reservation = await gateway._reserve_controlled_apply_result_outbound( identity=identity, payload={"chat_id": "sre-chat", "text": "committed reservation"}, source_envelope_extra={ "callback_reply": { "action": "controlled_apply_result", **identity, } }, ) assert context_exited is True assert reservation["status"] == "reserved" assert reservation["durable_reservation_committed"] is True assert len(db.statements) == 2 record_outbound.assert_awaited_once() def test_controlled_result_uses_durable_reservation_and_single_reconcile_entry() -> None: reserve_source = inspect.getsource( TelegramGateway._reserve_controlled_apply_result_outbound ) finalize_source = inspect.getsource( TelegramGateway._finalize_controlled_apply_result_outbound ) gateway_source = inspect.getsource(TelegramGateway._send_request) live_apply_source = inspect.getsource(service.run_controlled_apply_for_claim) assert "pg_advisory_xact_lock" in reserve_source assert "record_outbound_message" in reserve_source assert 'send_status="pending"' in reserve_source assert 'send_status="shadow"' in reserve_source assert "failure_fingerprint" in reserve_source assert "digest_window_minutes" in reserve_source assert "notification_suppressed_deduplicated" in reserve_source assert "durable_reservation_committed" in reserve_source assert "provider_retry_blocked" in reserve_source assert "reused_pending_reservation" not in reserve_source assert "reservation_readback_failed" not in reserve_source assert "UPDATE awooop_outbound_message" in finalize_source assert "_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS" in finalize_source assert gateway_source.index( "_reserve_controlled_apply_result_outbound" ) < gateway_source.index("_http_client.post") assert ( "telegram_receipt_sent = await _send_controlled_apply_telegram_receipt" not in live_apply_source ) assert live_apply_source.count( "_reconcile_verified_apply_closure_projections" ) == 1 @pytest.mark.asyncio async def test_retry_telegram_readback_binds_exact_apply_operation( monkeypatch: pytest.MonkeyPatch, ) -> None: db = _SequenceDB(_MappingResult(scalar=True)) @asynccontextmanager async def fake_get_db_context(_project_id): yield db monkeypatch.setattr(service, "get_db_context", fake_get_db_context) apply_op_id = "00000000-0000-0000-0000-000000000104" acknowledged = await service._retry_telegram_receipt_acknowledged( _claim(), apply_op_id=apply_op_id, project_id="awoooi", ) assert acknowledged is True assert "{callback_reply,apply_op_id}" in db.statements[0] assert db.parameters[0]["apply_op_id"] == apply_op_id def test_all_terminal_telegram_readbacks_require_exact_apply_operation() -> None: closure_source = inspect.getsource( service._read_verified_apply_closure_prerequisites ) retry_source = inspect.getsource( service._retry_telegram_receipt_acknowledged ) retry_receipt_source = inspect.getsource( service._record_retry_runtime_stage_receipt ) projection_source = inspect.getsource( service._load_missing_retry_terminal_projection_rows ) assert closure_source.count("{callback_reply,apply_op_id}") == 1 assert retry_source.count("{callback_reply,apply_op_id}") == 1 assert retry_receipt_source.count("{callback_reply,apply_op_id}") == 1 assert projection_source.count("{callback_reply,apply_op_id}") == 2 assert projection_source.count("= apply.op_id::text") >= 2 assert "coalesce(\n outbound.source_envelope #>>\n '{callback_reply,apply_op_id}'" not in projection_source def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() -> None: proposal = candidate_job._build_backfill_proposal( { "alertname": "AwoooPAutoRepairCanaryT16", "severity": "medium", "approval_id": "00000000-0000-0000-0000-000000000103", } ) query_source = inspect.getsource( candidate_job._fetch_missing_candidate_incidents ) assert proposal["approval_id"] == "00000000-0000-0000-0000-000000000103" assert "approval_records" in query_source assert "verified_ansible_candidate_terminal_sql" in query_source assert "current_recurrence" in query_source assert "target_route_generation" in query_source assert "latest_candidate_is_terminal" in query_source assert "preflight_ansible_candidate_generation" in inspect.getsource( candidate_job.enqueue_missing_ansible_candidates_once ) assert "ORDER BY candidate.created_at DESC, candidate.op_id DESC" in query_source def test_retry_projection_only_repairs_latest_candidate_run() -> None: source = inspect.getsource( service._load_missing_retry_terminal_projection_rows ) assert "source_candidate.op_id = source_check.parent_op_id" in source assert "newer_candidate.created_at," in source assert "source_candidate.created_at," in source assert "repair_candidate_controlled_queue" in source assert "retry_receipt.receipt IS NULL" in source assert "controlled_apply_result" in source assert "LEFT JOIN LATERAL" in source assert "receipt.value ->> 'automation_run_id'" in source assert "upper(incident.status::text)" in source assert "ansible_incident_terminal_disposition_v1" in source assert "'proposal_executed'" in source assert "'execution_success'" in source assert "verification_result IN ('failed', 'timeout')" in source @pytest.mark.asyncio async def test_broker_reconciles_receipts_then_continues_claiming_fresh_work( monkeypatch: pytest.MonkeyPatch, ) -> None: retry_replayer = AsyncMock( return_value={ "scanned": 0, "replayed": 0, "check_mode_passed": 0, "check_mode_failed": 0, "runtime_stage_receipt_written": 0, "blockers": [], "error": None, } ) fresh_claimer = AsyncMock(return_value=[]) monkeypatch.setattr( service, "_expire_stale_ansible_execution_capabilities", AsyncMock(return_value=0), ) monkeypatch.setattr( service, "backfill_missing_retry_terminal_projections_once", AsyncMock( return_value={ "scanned": 0, "written": 0, "incident_receipt_written": 0, "lifecycle_written": 0, "verified": 0, "runtime_apply_executed": False, "error": None, } ), ) monkeypatch.setattr( service, "backfill_missing_auto_repair_execution_receipts_once", AsyncMock( return_value={ "scanned": 1, "written": 1, "incident_closure_written": 1, "telegram_receipt_acknowledged": 1, "error": None, } ), ) monkeypatch.setattr( service, "run_failed_apply_check_mode_replay_once", retry_replayer, ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service, "recent_ansible_transport_blockers", AsyncMock(return_value=[]), ) monkeypatch.setattr( service, "claim_stale_pending_check_modes", AsyncMock(return_value=[]), ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", AsyncMock(return_value=[]), ) monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer) result = await service.run_pending_check_modes_once(limit=1) assert result["claimed"] == 0 assert result["repair_receipt_backfill_priority_tick"] is True assert result["repair_receipt_closure_written"] == 1 retry_replayer.assert_awaited_once() fresh_claimer.assert_awaited_once() @pytest.mark.asyncio async def test_broker_errors_do_not_starve_unrelated_fresh_claims( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( service, "_expire_stale_ansible_execution_capabilities", AsyncMock(return_value=0), ) monkeypatch.setattr( service, "backfill_missing_auto_repair_execution_receipts_once", AsyncMock( return_value={ "scanned": 3, "written": 0, "incident_closure_written": 0, "error": "legacy_projection_write_failed", } ), ) monkeypatch.setattr( service, "run_failed_apply_check_mode_replay_once", AsyncMock( return_value={ "scanned": 1, "replayed": 0, "blockers": ["foreign_retry_already_claimed"], "error": "foreign_retry_reconstruction_failed", } ), ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service, "recent_ansible_transport_blockers", AsyncMock(return_value=[]), ) monkeypatch.setattr( service, "claim_stale_pending_check_modes", AsyncMock(return_value=[]), ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", AsyncMock(return_value=[]), ) fresh_claimer = AsyncMock(return_value=[]) monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer) result = await service.run_pending_check_modes_once(limit=1) assert result["repair_receipt_backfill_error"] == ( "legacy_projection_write_failed" ) assert result["failed_apply_retry_error"] == ( "foreign_retry_reconstruction_failed" ) assert result["failed_apply_retry_blockers"] == [ "foreign_retry_already_claimed" ] fresh_claimer.assert_awaited_once() @pytest.mark.asyncio async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply( monkeypatch: pytest.MonkeyPatch, ) -> None: row = { "op_id": "00000000-0000-0000-0000-000000000104", "replay_op_id": "00000000-0000-0000-0000-000000000105", "replay_status": "success", "replay_output": {"returncode": 0}, "replay_dry_run_result": {"timed_out": False}, "replay_duration_ms": 18, "retry_receipt_present": False, "terminal_type": "no_write_replay_passed_waiting_repair_candidate", "telegram_receipt_acknowledged": False, } monkeypatch.setattr( service, "_load_missing_retry_terminal_projection_rows", AsyncMock(return_value=[row]), ) monkeypatch.setattr( service, "_claim_from_apply_operation_row", lambda _row: ( _claim(), service.AnsibleRunResult( returncode=1, stdout="", stderr="", duration_ms=0, ), ), ) terminal_writer = AsyncMock( return_value={ "automation_run_id": "00000000-0000-0000-0000-000000000102", "apply_op_id": row["op_id"], "retry_op_id": row["replay_op_id"], "terminal_type": row["terminal_type"], "repository_readback_verified": True, } ) receipt_writer = AsyncMock(return_value=True) lifecycle_writer = AsyncMock(return_value=True) verifier = AsyncMock(return_value=True) telegram_sender = AsyncMock(return_value=True) telegram_readback = AsyncMock(return_value=True) monkeypatch.setattr( service, "_record_incident_terminal_disposition", terminal_writer ) monkeypatch.setattr( service, "_append_runtime_stage_receipts_to_apply", receipt_writer ) monkeypatch.setattr( service, "_append_alert_lifecycle_receipt", lifecycle_writer ) monkeypatch.setattr( service, "_verify_retry_terminal_projection_readback", verifier ) monkeypatch.setattr( service, "_send_controlled_apply_telegram_receipt", telegram_sender, ) monkeypatch.setattr( service, "_retry_telegram_receipt_acknowledged", telegram_readback, ) result = await service.backfill_missing_retry_terminal_projections_once() assert result == { "scanned": 1, "written": 1, "retry_receipt_written": 1, "telegram_receipt_acknowledged": 1, "incident_receipt_written": 1, "lifecycle_written": 1, "verified": 1, "runtime_apply_executed": False, "error": None, } terminal_writer.assert_awaited_once() assert terminal_writer.await_args.kwargs["success_terminal"] is False retry_receipt = receipt_writer.await_args_list[0].kwargs["receipts"][0] assert retry_receipt["stage_id"] == "retry_or_rollback" assert retry_receipt["derived_from_durable_chain"] is True receipt = receipt_writer.await_args_list[1].kwargs["receipts"][0] assert receipt["stage_id"] == "incident_closure" assert receipt["derived_from_durable_chain"] is True assert lifecycle_writer.await_args.args[1] == "EXECUTION_COMPLETED" assert lifecycle_writer.await_args.kwargs["success"] is False telegram_sender.assert_awaited_once() telegram_readback.assert_awaited_once() @pytest.mark.asyncio async def test_retry_projection_does_not_starve_verified_success_closure( monkeypatch: pytest.MonkeyPatch, ) -> None: class _RowsResult(_MappingResult): def __init__(self, rows: list[dict]) -> None: super().__init__() self._rows = rows def all(self) -> list[dict]: return self._rows apply_row = { "op_id": "00000000-0000-0000-0000-000000000104", "status": "success", "verifier_ready": True, } db = _SequenceDB(_MappingResult(), _RowsResult([apply_row])) @asynccontextmanager async def fake_db_context(_project_id: str): yield db monkeypatch.setattr(service, "get_db_context", fake_db_context) monkeypatch.setattr( service, "backfill_missing_retry_terminal_projections_once", AsyncMock( return_value={ "scanned": 1, "written": 0, "retry_receipt_written": 0, "telegram_receipt_acknowledged": 1, "incident_receipt_written": 0, "lifecycle_written": 0, "verified": 0, "runtime_apply_executed": False, "error": None, } ), ) monkeypatch.setattr( service, "_claim_from_apply_operation_row", lambda _row: (_claim(), _verified_result()), ) monkeypatch.setattr( service, "_record_post_apply_verifier_and_learning", AsyncMock( return_value={ "verification": True, "verification_passed": True, "verification_result": "success", "learning": True, "trust_learning": True, "rag_writeback": True, "runtime_stage_receipts": [], } ), ) monkeypatch.setattr( service, "_record_auto_repair_execution_receipt", AsyncMock(return_value=True), ) monkeypatch.setattr( service, "_record_timeline_projection_receipt", AsyncMock(return_value=None), ) monkeypatch.setattr( service, "_record_runtime_stage_receipts", AsyncMock(return_value=True), ) closure = AsyncMock( return_value={ "status": "controlled_apply_closed", "closed": True, "telegram_receipt_acknowledged": True, "runtime_apply_executed": False, } ) monkeypatch.setattr( service, "_reconcile_verified_apply_closure_projections", closure, ) result = await service.backfill_missing_auto_repair_execution_receipts_once( limit=1 ) assert result["retry_terminal_projection_scanned"] == 1 assert result["scanned"] == 2 assert result["incident_closure_written"] == 1 assert result["telegram_receipt_acknowledged"] == 1 closure.assert_awaited_once() @pytest.mark.asyncio async def test_broker_prioritizes_retry_terminal_projection_backfill( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( service, "_expire_stale_ansible_execution_capabilities", AsyncMock(return_value=0), ) receipt_backfill = AsyncMock( return_value={ "scanned": 1, "written": 0, "incident_closure_written": 1, "telegram_receipt_acknowledged": 0, "retry_terminal_projection_scanned": 1, "retry_terminal_projection_written": 1, "retry_terminal_projection_verified": 1, "retry_terminal_projection_runtime_apply_executed": False, "error": None, } ) monkeypatch.setattr( service, "backfill_missing_auto_repair_execution_receipts_once", receipt_backfill, ) monkeypatch.setattr( service, "run_failed_apply_check_mode_replay_once", AsyncMock(return_value={"scanned": 0, "replayed": 0}), ) monkeypatch.setattr(service, "_runtime_blockers", lambda: []) monkeypatch.setattr( service, "recent_ansible_transport_blockers", AsyncMock(return_value=[]), ) monkeypatch.setattr( service, "claim_stale_pending_check_modes", AsyncMock(return_value=[]), ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", AsyncMock(return_value=[]), ) fresh_claimer = AsyncMock(return_value=[]) monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claimer) result = await service.run_pending_check_modes_once(limit=1) assert result["claimed"] == 0 assert result["retry_terminal_projection_verified"] == 1 assert result["retry_terminal_projection_runtime_apply_executed"] is False assert result["repair_receipt_backfill_priority_tick"] is True receipt_backfill.assert_awaited_once() fresh_claimer.assert_awaited_once() @pytest.mark.asyncio async def test_signal_worker_retry_preflight_is_query_only( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( service, "_load_open_failed_apply_retry_row", AsyncMock(return_value={"op_id": "apply-op"}), ) result = await service.preflight_failed_apply_retry_queue_once( project_id="awoooi", window_hours=24, ) assert result["scanned"] == 1 assert result["replayed"] == 0 assert result["query_only"] is True assert result["runtime_apply_executed"] is False assert result["execution_owner"] == "awoooi-ansible-executor-broker" @pytest.mark.asyncio @pytest.mark.parametrize( ("recovery_count", "expected_replayed", "expected_terminalized"), [(0, 1, 0), (1, 0, 1)], ) async def test_stale_pending_retry_is_reclaimed_without_runtime_apply( monkeypatch: pytest.MonkeyPatch, recovery_count: int, expected_replayed: int, expected_terminalized: int, ) -> None: replay_op_id = "00000000-0000-0000-0000-000000000105" apply_op_id = "00000000-0000-0000-0000-000000000104" db = _SequenceDB( _MappingResult(), _MappingResult(scalar=replay_op_id), _MappingResult(scalar=replay_op_id), ) @asynccontextmanager async def fake_db_context(_project_id: str): yield db monkeypatch.setattr(service, "get_db_context", fake_db_context) monkeypatch.setattr( service, "_load_open_failed_apply_retry_row", AsyncMock( return_value={ "op_id": apply_op_id, "stale_replay_op_id": replay_op_id, "stale_replay_recovery_count": recovery_count, } ), ) monkeypatch.setattr( service, "_claim_from_apply_operation_row", lambda _row: ( _claim(), service.AnsibleRunResult( returncode=1, stdout="", stderr="failed apply", duration_ms=10, ), ), ) runtime_blockers = ( ["ansible_binary_missing"] if expected_terminalized else [] ) monkeypatch.setattr( service, "_runtime_blockers", lambda: runtime_blockers, ) transport_probe = AsyncMock( return_value=( ["ansible_transport_unavailable"] if expected_terminalized else [] ) ) monkeypatch.setattr( service, "recent_ansible_transport_blockers", transport_probe, ) monkeypatch.setattr( service, "build_ansible_check_mode_command", lambda **_kwargs: object(), ) replay_runner = AsyncMock( return_value=service.AnsibleRunResult( returncode=0, stdout="check ok", stderr="", duration_ms=15, ) ) monkeypatch.setattr(service, "_run_ansible_command", replay_runner) receipt_writer = AsyncMock(return_value=True) monkeypatch.setattr( service, "_record_retry_runtime_stage_receipt", receipt_writer, ) result = await service.run_failed_apply_check_mode_replay_once( timeout_seconds=30, ) assert result["replayed"] == expected_replayed if expected_terminalized: assert result["check_mode_passed"] == 0 assert result["check_mode_failed"] == 1 else: assert result["check_mode_passed"] == 1 assert result["check_mode_failed"] == 0 assert result["stale_retry_terminalized"] == expected_terminalized assert result["runtime_apply_executed"] == 0 assert "pg_advisory_xact_lock" in db.statements[0] assert "UPDATE automation_operation_log replay" in db.statements[1] assert "reclaimed_after_stale_pending" in db.statements[1] assert "input ->> 'retry_claim_id'" in db.statements[2] persisted_output = json.loads(db.parameters[2]["output"]) persisted_dry_run = json.loads(db.parameters[2]["dry_run_result"]) assert persisted_output["check_mode_executed"] is ( not bool(expected_terminalized) ) assert persisted_output["check_mode_replay_performed"] is ( not bool(expected_terminalized) ) assert persisted_output["runtime_apply_executed"] is False assert persisted_dry_run["check_mode_executed"] is ( not bool(expected_terminalized) ) assert persisted_dry_run["check_mode_replay_performed"] is ( not bool(expected_terminalized) ) assert persisted_dry_run["runtime_apply_executed"] is False if expected_terminalized: replay_runner.assert_not_awaited() transport_probe.assert_not_awaited() else: replay_runner.assert_awaited_once() transport_probe.assert_awaited_once() receipt_writer.assert_awaited_once() assert receipt_writer.await_args.kwargs[ "check_mode_replay_performed" ] is (not bool(expected_terminalized)) def test_candidate_worker_defaults_to_query_only_retry_preflight() -> None: defaults = candidate_job.enqueue_missing_ansible_candidates_once.__kwdefaults__ assert defaults is not None assert defaults["retry_replayer"] is service.preflight_failed_apply_retry_queue_once