from __future__ import annotations import json from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from fastapi import HTTPException from src.api.v1 import webhooks from src.db.models import IncidentRecord from src.models.incident import IncidentStatus from src.services import agent99_controlled_dispatch_ledger as ledger_module from src.services import agent99_outcome_ingestion as ingestion_module from src.services.agent99_controlled_dispatch_ledger import ( PostgresAgent99DispatchLedger, build_agent99_dispatch_identity, build_agent99_dispatch_receipt_envelope, parse_agent99_dispatch_identity, record_agent99_learning_writeback, ) from src.services.agent99_sre_bridge import bridge_alertmanager_to_agent99 def _identity(): return build_agent99_dispatch_identity( project_id="awoooi", incident_id="INC-20260711-11C751", source_fingerprint="cold-start-source-fingerprint", route_id="agent99_cold_start_recovery", execution_generation="1", work_item_id="agent99-dispatch:awoooi:INC-20260711-11C751:recovery", ) def _evidence_refs() -> dict[str, str]: return { "agent99_outcome_receipt_id": "agent99-outcome:queue-1", "post_verifier_evidence_ref": "evidence:cold-start-scorecard:1", "source_event_evidence_ref": "alert:fingerprint:resolved:1", } def _outcome(identity, *, transport_ok: bool = True) -> dict: return { "identity": identity.public_dict(), "controlledApply": True, "outcome": { "identity": identity.public_dict(), "schemaVersion": "agent99_outcome_contract_v1", "state": "resolved", "transportOk": transport_ok, "verifierPassed": True, "sourceEventResolved": True, "verifiedAt": "2026-07-11T20:00:00+08:00", }, } def test_pending_inbox_never_promotes_or_authorizes() -> None: receipt = build_agent99_dispatch_receipt_envelope( identity=_identity(), dispatch_receipt={ "accepted": True, "inbox_triggered": False, "delivery_certainty": "unknown", }, controlled_apply_authorized=True, ) assert receipt["dispatch_accepted"] is True assert receipt["inbox_triggered"] is False assert receipt["dispatch_promoted"] is False assert receipt["controlled_apply_authorized"] is False assert receipt["runtime_execution_authorized"] is False assert receipt["verifier"]["status"] == "blocked_dispatch_not_promoted" assert receipt["runtime_closure_verified"] is False def test_complete_identity_is_required_and_recomputed() -> None: identity = _identity() assert parse_agent99_dispatch_identity(identity.public_dict()) == identity missing = identity.public_dict() missing["trace_id"] = "" with pytest.raises(ValueError, match="identity_fields_required:trace_id"): parse_agent99_dispatch_identity(missing) mismatched = identity.public_dict() mismatched["run_id"] = "00000000-0000-0000-0000-000000000000" with pytest.raises(ValueError, match="identity_mismatch:run_id"): parse_agent99_dispatch_identity(mismatched) def test_approval_identity_cannot_alias_run_or_idempotency() -> None: common = { "project_id": "awoooi", "incident_id": "INC-20260711-11C751", "source_fingerprint": "cold-start-source-fingerprint", "route_id": "agent99_cold_start_recovery", "execution_generation": "1", "work_item_id": "agent99-dispatch:awoooi:11c751", } first = build_agent99_dispatch_identity( **common, approval_id="00000000-0000-0000-0000-000000000751", ) second = build_agent99_dispatch_identity( **common, approval_id="00000000-0000-0000-0000-000000000752", ) assert first.run_id != second.run_id assert first.idempotency_key != second.idempotency_key assert first.public_dict()["canonical_digest"] != ( second.public_dict()["canonical_digest"] ) assert first.single_flight_key == second.single_flight_key def test_terminal_incident_status_uses_database_enum_member() -> None: status_type = IncidentRecord.__table__.c.status.type assert status_type.enum_class is IncidentStatus assert IncidentStatus.RESOLVED.name in status_type.enums assert IncidentStatus.RESOLVED.value == "resolved" class _ScalarResult: def __init__(self, value=None, row=None) -> None: self.value = value self.row = row def scalar_one_or_none(self): return self.value def one_or_none(self): return self.row class _Context: def __init__(self, db) -> None: self.db = db async def __aenter__(self): return self.db async def __aexit__(self, *_args): return False @pytest.mark.asyncio async def test_reservation_claim_tokens_are_unique_and_running_is_reconcile_only( monkeypatch, ) -> None: identity = _identity() claim_statements = [] class ReserveDB: def __init__(self) -> None: self.call = 0 async def execute(self, statement): self.call += 1 if self.call == 3: claim_statements.append(statement) return _ScalarResult(identity.run_id) if self.call == 5: return _ScalarResult( row=SimpleNamespace( state="running", error_detail=json.dumps({"identity": identity.public_dict()}), ) ) return _ScalarResult() databases = [ReserveDB(), ReserveDB()] monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(databases.pop(0)), ) ledger = PostgresAgent99DispatchLedger() payload = { "kind": "host_recovery", "suggestedMode": "Recover", "controlledApply": True, "awoooi": {"targetResource": "cold-start-gate"}, } first = await ledger.reserve(identity=identity, payload=payload) second = await ledger.reserve(identity=identity, payload=payload) assert first["claimed"] is True assert second["claimed"] is True assert first["claim_token"] != second["claim_token"] claim_sql = str(claim_statements[0]) assert "awooop_run_state.state = :state_1" in claim_sql assert "awooop_run_state.next_attempt_at IS NULL" in claim_sql assert "awooop_run_state.next_attempt_at <=" in claim_sql assert "awooop_run_state.lease_until <" not in claim_sql @pytest.mark.asyncio async def test_stale_worker_cannot_complete_newer_claim(monkeypatch) -> None: identity = _identity() class FencedDB: async def execute(self, _statement): return _ScalarResult(None) monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(FencedDB()), ) result = await PostgresAgent99DispatchLedger().complete( identity=identity, claim_token="stale-claim-token", dispatch_receipt={ "accepted": True, "inbox_triggered": True, "delivery_certainty": "delivered", }, controlled_apply_authorized=True, ) assert result["status"] == "claim_fenced_no_write" assert result["receipt_persisted"] is False assert result["runtime_closure_verified"] is False @pytest.mark.asyncio async def test_redis_lock_collision_returns_run_to_retryable_pending( monkeypatch, ) -> None: identity = _identity() initial = { "schema_version": "agent99_controlled_dispatch_receipt_v1", "identity": identity.public_dict(), "runtime_closure_verified": False, } class RetryDB: def __init__(self) -> None: self.call = 0 async def execute(self, _statement): self.call += 1 if self.call == 1: return _ScalarResult( row=SimpleNamespace( state="running", worker_id="claim-2", attempt_count=1, max_attempts=3, error_detail=json.dumps(initial), ) ) if self.call == 2: return _ScalarResult(identity.run_id) return _ScalarResult() monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(RetryDB()), ) result = await PostgresAgent99DispatchLedger().defer_claim_retryable( identity=identity, claim_token="claim-2", reason="single_flight_duplicate", retry_after_seconds=600, ) assert result["status"] == "dispatch_lock_retryable" assert result["receipt_persisted"] is True assert result["retry_policy"]["retry_same_generation"] is True assert result["controlled_apply_authorized"] is False assert result["runtime_closure_verified"] is False @pytest.mark.asyncio async def test_verifier_requires_transport_and_all_evidence(monkeypatch) -> None: identity = _identity() ledger = PostgresAgent99DispatchLedger() missing = await ledger.record_verifier( identity=identity, outcome_receipt=_outcome(identity), evidence_refs={"agent99_outcome_receipt_id": "one-ref-only"}, ) assert missing["status"] == "verifier_evidence_missing_fail_closed" assert missing["receipt_persisted"] is False promoted = build_agent99_dispatch_receipt_envelope( identity=identity, dispatch_receipt={ "accepted": True, "inbox_triggered": True, "delivery_certainty": "delivered", }, controlled_apply_authorized=True, ) class VerifierDB: def __init__(self) -> None: self.call = 0 async def execute(self, _statement): self.call += 1 if self.call == 1: return _ScalarResult( row=SimpleNamespace( state="waiting_tool", error_detail=json.dumps(promoted), ) ) if self.call == 2: return _ScalarResult(identity.run_id) if self.call == 4: return _ScalarResult(identity.incident_id) if self.call in {5, 6}: return _ScalarResult(identity.run_id) return _ScalarResult() monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(VerifierDB()), ) failed = await ledger.record_verifier( identity=identity, outcome_receipt=_outcome(identity, transport_ok=False), evidence_refs=_evidence_refs(), ) assert failed["post_verifier_passed"] is False assert failed["verifier"]["transport_ok"] is False assert failed["runtime_closure_verified"] is False monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(VerifierDB()), ) passed = await ledger.record_verifier( identity=identity, outcome_receipt=_outcome(identity, transport_ok=True), evidence_refs=_evidence_refs(), ) assert passed["post_verifier_passed"] is True assert passed["verifier"]["transport_ok"] is True assert passed["status"] == "verifier_passed_learning_writeback_pending" assert passed["runtime_closure_verified"] is False @pytest.mark.asyncio async def test_learning_writeback_is_the_only_same_run_terminal_step( monkeypatch, ) -> None: identity = _identity() ledger = PostgresAgent99DispatchLedger() verifier_passed = build_agent99_dispatch_receipt_envelope( identity=identity, dispatch_receipt={ "accepted": True, "inbox_triggered": True, "delivery_certainty": "delivered", }, controlled_apply_authorized=True, ) verifier_passed.update({ "status": "verifier_passed_learning_writeback_pending", "post_verifier_passed": True, "runtime_closure_verified": False, "closure_complete": False, "learning_writeback": { "status": "pending_writeback", "receipt_refs": {"km_writeback_ack_id": "km-only"}, }, }) class IncompleteCheckpointDB: def __init__(self) -> None: self.call = 0 async def execute(self, _statement): self.call += 1 if self.call == 1: return _ScalarResult( row=SimpleNamespace( state="waiting_tool", error_detail=json.dumps(verifier_passed), ) ) raise AssertionError("terminal tables must not be touched") incomplete_db = IncompleteCheckpointDB() monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(incomplete_db), ) incomplete = await ledger.record_learning_writeback( identity=identity, receipt_refs={ "telegram_lifecycle_receipt_id": "telegram:fake", "km_writeback_ack_id": "knowledge:fake", "playbook_trust_writeback_ack_id": "playbook:fake", }, ) assert incomplete["status"] == ( "learning_writeback_checkpoint_incomplete_fail_closed" ) assert incomplete["runtime_closure_verified"] is False assert incomplete_db.call == 1 checkpoint_refs = { "telegram_lifecycle_receipt_id": "telegram-1", "km_writeback_ack_id": "km-1", "playbook_trust_writeback_ack_id": "playbook-1", } verifier_passed["learning_writeback"]["receipt_refs"] = checkpoint_refs class LearningDB: def __init__(self) -> None: self.call = 0 async def execute(self, _statement): self.call += 1 if self.call == 1: return _ScalarResult( row=SimpleNamespace( state="waiting_tool", error_detail=json.dumps(verifier_passed), ) ) if self.call == 2: return _ScalarResult(SimpleNamespace(outcome={})) if self.call == 4: return _ScalarResult(identity.incident_id) if self.call in {5, 6}: return _ScalarResult(identity.run_id) return _ScalarResult() monkeypatch.setattr( ledger_module, "get_db_context", lambda _project_id: _Context(LearningDB()), ) closed = await ledger.record_learning_writeback( identity=identity, receipt_refs=checkpoint_refs, ) assert closed["status"] == "closed_verified_learning_written" assert closed["runtime_closure_verified"] is True assert closed["closure_complete"] is True @pytest.mark.asyncio async def test_terminal_post_commit_projection_is_project_scoped( monkeypatch, ) -> None: identity = _identity() incident = SimpleNamespace(incident_id=identity.incident_id) episodic = AsyncMock(return_value=incident) working = AsyncMock(return_value=True) monkeypatch.setattr( ledger_module._ledger, "record_learning_writeback", AsyncMock( return_value={ "status": "closed_verified_learning_written", "postgres_terminal_bundle_committed": True, "redis_projection_pending": True, "runtime_closure_verified": True, } ), ) monkeypatch.setattr( "src.services.incident_service.get_incident_service", lambda: SimpleNamespace( get_from_episodic_memory=episodic, save_to_working_memory=working, ), ) result = await record_agent99_learning_writeback( identity=identity, receipt_refs={ "telegram_lifecycle_receipt_id": "telegram:1", "km_writeback_ack_id": "knowledge:1", "playbook_trust_writeback_ack_id": "playbook:1", }, ) episodic.assert_awaited_once_with( identity.incident_id, project_id=identity.project_id, ) working.assert_awaited_once_with(incident) assert result["redis_projection_acknowledged"] is True assert result["redis_projection_pending"] is False @pytest.mark.asyncio async def test_outcome_ingestion_ignores_untrusted_refs_and_cannot_terminalize( monkeypatch, ) -> None: identity = _identity() calls: list[tuple[str, str]] = [] async def verifier(**kwargs): calls.append(("verifier", str(kwargs["identity"].run_id))) return { "status": "verifier_passed_learning_writeback_pending", "receipt_persisted": True, "post_verifier_passed": True, "runtime_closure_verified": False, } terminal = AsyncMock(side_effect=AssertionError("terminal bypass invoked")) latest = AsyncMock(side_effect=AssertionError("terminal readback invoked")) monkeypatch.setattr(ingestion_module, "record_agent99_verifier_receipt", verifier) monkeypatch.setattr( ingestion_module, "record_agent99_learning_writeback", terminal, raising=False, ) monkeypatch.setattr( ingestion_module, "read_agent99_dispatch_receipt", latest, raising=False, ) result = await ingestion_module.ingest_agent99_outcome_receipt({ "identity": identity.public_dict(), "outcome_receipt": _outcome(identity), "evidence_refs": _evidence_refs(), "learning_receipt_refs": { "telegram_lifecycle_receipt_id": "telegram:fake", "km_writeback_ack_id": "knowledge:fake", "playbook_trust_writeback_ack_id": "playbook:fake", }, }) assert calls == [("verifier", str(identity.run_id))] terminal.assert_not_awaited() latest.assert_not_awaited() assert result["status"] == "verifier_recorded_learning_pending" assert result["learning_writeback"]["status"] == ( "untrusted_external_receipts_ignored_reconciler_pending" ) assert result["untrusted_learning_receipt_refs_ignored"] is True assert result["runtime_closure_verified"] is False @pytest.mark.asyncio async def test_authenticated_backup_outcome_merges_all_nine_safe_refs( monkeypatch, ) -> None: identity = build_agent99_dispatch_identity( project_id="awoooi", incident_id="INC-BRR-20260711-001", source_fingerprint="backup-source-fingerprint", route_id="agent99:backup_health:BackupCheck", work_item_id="agent99-dispatch:awoooi:backup-readback", ) captured: dict[str, object] = {} async def verifier(**kwargs): captured.update(kwargs) return { "status": "verifier_passed_learning_writeback_pending", "receipt_persisted": True, "post_verifier_passed": True, "runtime_closure_verified": False, } monkeypatch.setattr(ingestion_module, "record_agent99_verifier_receipt", verifier) result = await ingestion_module.ingest_agent99_outcome_receipt({ "identity": identity.public_dict(), "evidence_refs": { "agent99_outcome_receipt_id": "agent99-outcome:backup-1", "post_verifier_evidence_ref": "backupcheck:verified:1", "source_event_evidence_ref": "backup-alert:resolved:1", "untrusted_extra": "must-not-persist", }, "outcome_receipt": { "identity": identity.public_dict(), "mode": "BackupCheck", "outcome": { "identity": identity.public_dict(), "schemaVersion": "agent99_outcome_contract_v1", "state": "resolved", "transportOk": True, "verifierPassed": True, "sourceEventResolved": True, "evidenceRefs": { "backup_status_evidence_ref": "backup-status:run:1", "freshness_evidence_ref": "backup-freshness:slo:1", "offsite_verify_evidence_ref": "offsite-verify:run:1", "escrow_evidence_ref": "escrow-metadata:run:1", "restore_drill_evidence_ref": "restore-drill:run:1", "source_resolution_receipt_ref": "source-resolution:run:1", "token": "must-not-persist", }, }, }, }) refs = captured["evidence_refs"] assert set(refs) == { "agent99_outcome_receipt_id", "post_verifier_evidence_ref", "source_event_evidence_ref", "backup_status_evidence_ref", "freshness_evidence_ref", "offsite_verify_evidence_ref", "escrow_evidence_ref", "restore_drill_evidence_ref", "source_resolution_receipt_ref", } assert "must-not-persist" not in str(refs) assert captured["identity"].run_id == identity.run_id assert result["status"] == "verifier_recorded_learning_pending" assert result["runtime_closure_verified"] is False def test_outcome_evidence_ref_filter_rejects_secret_shaped_values() -> None: refs = ingestion_module._public_evidence_refs( { "agent99_outcome_receipt_id": "https://unsafe.invalid/outcome", "post_verifier_evidence_ref": "verifier:public-ref", "source_event_evidence_ref": "source:public-ref", }, { "outcome": { "evidenceRefs": { "backup_status_evidence_ref": "token=must-not-persist", "freshness_evidence_ref": "freshness:public-ref", "offsite_verify_evidence_ref": "receipt?token=unsafe", "escrow_evidence_ref": "../runtime-secret", "restore_drill_evidence_ref": "Bearer must-not-persist", } } }, ) assert refs == { "post_verifier_evidence_ref": "verifier:public-ref", "source_event_evidence_ref": "source:public-ref", "freshness_evidence_ref": "freshness:public-ref", } @pytest.mark.asyncio async def test_outcome_webhook_requires_shared_token(monkeypatch) -> None: identity = _identity() request = webhooks.Agent99OutcomeWebhookPayload( identity=identity.public_dict(), outcome_receipt=_outcome(identity), evidence_refs=_evidence_refs(), ) monkeypatch.setattr( webhooks.settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "relay-token", ) with pytest.raises(HTTPException) as exc_info: await webhooks.ingest_agent99_outcome_webhook(request, None) assert exc_info.value.status_code == 401 async def accepted(_payload): return { "status": "verifier_recorded_learning_pending", "accepted": True, "runtime_closure_verified": False, } monkeypatch.setattr(webhooks, "ingest_agent99_outcome_receipt", accepted) response = await webhooks.ingest_agent99_outcome_webhook( request, "relay-token", ) assert response["accepted"] is True assert response["runtime_closure_verified"] is False @pytest.mark.asyncio async def test_safe_failed_generation_advances_bounded_and_keeps_one_flight_key( monkeypatch, ) -> None: identities = [] flight_calls: list[tuple[str, str]] = [] class GenerationLedger: async def reserve(self, *, identity, payload): identities.append(identity) if identity.execution_generation == "1": return { "status": "idempotent_failed", "claimed": False, "receipt": { "status": "dispatch_not_delivered_retry_safe", "retry_policy": { "safe_to_retry": True, "requires_generation_advance": True, "max_generation": 3, }, }, } return { "status": "reserved", "claimed": True, "claim_token": "generation-2-claim", "receipt": None, } async def complete( self, *, identity, claim_token, dispatch_receipt, **_kwargs, ): return { **build_agent99_dispatch_receipt_envelope( identity=identity, dispatch_receipt=dispatch_receipt, controlled_apply_authorized=True, ), "receipt_persisted": True, } async def mark_delivery_attempt_started(self, *, identity, claim_token): return { "status": "dispatch_delivery_unknown_reconcile_only", "identity": identity.public_dict(), "receipt_persisted": True, } async def acquire(key, owner, **_kwargs): flight_calls.append((key, owner)) return {"acquired": True, "reason": "acquired"} monkeypatch.setattr( "src.services.agent99_sre_bridge.get_agent99_dispatch_ledger", lambda: GenerationLedger(), ) monkeypatch.setattr( "src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight", acquire, ) monkeypatch.setattr( "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt", lambda _payload: { "status": "accepted_inbox_triggered", "transport": "relay", "accepted": True, "inbox_triggered": True, "delivery_certainty": "delivered", }, ) result = await bridge_alertmanager_to_agent99( alert_id="recurrence-2", alertname="ColdStartGateBlocked", severity="critical", namespace="host_recovery", target_resource="cold-start-gate", message="cold-start gate blocked", fingerprint="same-source-fingerprint", incident_id="INC-20260711-11C751", route_id="agent99_cold_start_recovery", ) assert result["status"] == "dispatched" assert [item.execution_generation for item in identities] == ["1", "2"] assert identities[0].idempotency_key != identities[1].idempotency_key assert identities[0].single_flight_key == identities[1].single_flight_key assert flight_calls == [ (identities[1].single_flight_key, "generation-2-claim") ] @pytest.mark.asyncio async def test_retry_not_before_does_not_burn_generation_or_transport( monkeypatch, ) -> None: identities = [] class CooldownLedger: async def reserve(self, *, identity, payload): # type: ignore[no-untyped-def] identities.append(identity) return { "status": "retry_not_before", "claimed": False, "next_attempt_at": "2026-07-11T20:10:00", "receipt": { "status": "dispatch_not_delivered_retry_safe", "retry_policy": { "safe_to_retry": True, "requires_generation_advance": True, "max_generation": 3, }, }, } async def unexpected_flight(*_args, **_kwargs): # type: ignore[no-untyped-def] raise AssertionError("cooldown must return before Redis or transport") monkeypatch.setattr( "src.services.agent99_sre_bridge.get_agent99_dispatch_ledger", lambda: CooldownLedger(), ) monkeypatch.setattr( "src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight", unexpected_flight, ) monkeypatch.setattr( "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt", lambda _payload: (_ for _ in ()).throw( AssertionError("cooldown must not dispatch") ), ) result = await bridge_alertmanager_to_agent99( alert_id="cooldown-recurrence", alertname="ColdStartGateBlocked", severity="critical", namespace="host_recovery", target_resource="cold-start-gate", message="cold-start gate blocked", fingerprint="same-source-fingerprint", incident_id="INC-20260711-11C751", route_id="agent99_cold_start_recovery", ) assert result["status"] == "retryable" assert result["reason"] == "durable_retry_not_before" assert result["nextAttemptAt"] == "2026-07-11T20:10:00" assert [item.execution_generation for item in identities] == ["1"] @pytest.mark.asyncio async def test_webhook_pending_inbox_is_not_queued_or_authorized(monkeypatch) -> None: append = AsyncMock() monkeypatch.setattr( "src.repositories.alert_operation_log_repository.get_alert_operation_log_repository", lambda: SimpleNamespace(append=append), ) monkeypatch.setattr( webhooks, "bridge_alertmanager_to_agent99", AsyncMock( return_value={ "status": "delivery_pending", "reason": "accepted_without_inbox_triggered", "dispatchPerformed": True, "identity": _identity().public_dict(), "dispatchReceipt": { "accepted": True, "inbox_triggered": False, "status": "accepted_inbox_pending", }, "correlatedReceipt": { "receipt_persisted": True, "dispatch_promoted": False, "controlled_apply_authorized": False, "runtime_closure_verified": False, }, } ), ) handoff = await webhooks._try_auto_repair_background( incident_id="INC-20260711-11C751", approval_id="00000000-0000-0000-0000-000000000751", alert_type="ColdStartGateBlocked", target_resource="cold-start-gate", namespace="host_recovery", risk_level="low", source_alert_id="pending-inbox", source_alertname="ColdStartGateBlocked", source_fingerprint="same-source-fingerprint", ) assert handoff["queued"] is False assert handoff["dispatch_accepted"] is True assert handoff["inbox_triggered"] is False assert handoff["dispatch_promoted"] is False assert handoff["runtime_execution_authorized"] is False assert handoff["runtime_closure_verified"] is False assert append.await_args.kwargs["success"] is False