fix(aiops): require current recurrence before controlled apply

apps/api/tests/test_awooop_ansible_candidate_backfill_job.py

@@ -308,6 +308,10 @@ def test_backfill_query_reconciles_current_same_fingerprint_recurrence() -> None
     assert "source_occurrence_id" in joined
     assert "target_route_generation" in joined
     assert "source_recurrence_verified" in joined
+    assert "jsonb_array_elements" in joined
+    assert "approval_record_fingerprint" in joined
+    assert "approval.fingerprint" in joined
+    assert "OR current_recurrence.received_at" in joined
 
 
 @pytest.mark.asyncio
@@ -347,9 +351,67 @@ async def test_backfill_scans_bounded_hundred_rows_without_expanding_write_batch
     )
 
     assert observed["scan_limit"] == 100
+    assert result["scan_limit"] == 100
+    assert result["scan_window_exhausted"] is False
     assert result["queued"] == 0
 
 
+@pytest.mark.asyncio
+async def test_backfill_scan_does_not_starve_candidate_behind_25_no_catalog_rows(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.jobs import awooop_ansible_candidate_backfill_job as job
+
+    no_catalog = [
+        {
+            "incident_id": f"INC-20260715-NOCAT{index:02d}",
+            "project_id": "awoooi",
+            "status": "INVESTIGATING",
+            "severity": "warning",
+            "alertname": "UnknownAssetDrift",
+            "affected_services": [f"unknown-asset-{index}"],
+            "signals": [],
+        }
+        for index in range(25)
+    ]
+    rows = [*no_catalog, _candidate_incident()]
+
+    async def fake_fetch_missing_candidate_incidents(
+        *, project_id: str, window_hours: int, scan_limit: int
+    ) -> list[dict]:
+        assert project_id == "awoooi"
+        assert window_hours >= 1
+        return rows[:scan_limit]
+
+    monkeypatch.setattr(
+        job,
+        "_fetch_missing_candidate_incidents",
+        fake_fetch_missing_candidate_incidents,
+    )
+    monkeypatch.setattr(
+        job.settings,
+        "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER",
+        True,
+    )
+    recorder = AsyncMock(return_value=True)
+
+    result = await job.enqueue_missing_ansible_candidates_once(
+        project_id="awoooi",
+        limit=1,
+        recorder=recorder,
+        retry_replayer=AsyncMock(return_value={}),
+        evidence_collector=AsyncMock(return_value=MagicMock()),
+        evidence_verifier=AsyncMock(return_value=True),
+    )
+
+    assert result["scan_limit"] == 100
+    assert result["scanned"] == 26
+    assert result["no_catalog_candidate"] == 25
+    assert result["queued"] == 1
+    assert result["scan_window_exhausted"] is False
+    recorder.assert_awaited_once()
+
+
 def test_backfill_worker_error_backoff_is_short_and_bounded() -> None:
     from src.jobs import awooop_ansible_candidate_backfill_job as job
 

apps/api/tests/test_current_owner_controlled_apply_policy.py

@@ -213,11 +213,46 @@ async def test_legacy_medium_pending_gate_is_durably_promoted(monkeypatch) -> No
     assert len(db.events) == 1
 
 
+@pytest.mark.asyncio
+async def test_stale_critical_bounded_container_restart_is_reclassified(
+    monkeypatch,
+) -> None:
+    record = _approval_record(
+        action=(
+            "docker inspect "
+            "sentry-self-hosted-snuba-profiling-functions-consumer-1 "
+            "then docker restart the same container"
+        ),
+        risk_level=RiskLevel.CRITICAL,
+    )
+    db = _FakeDB(record)
+
+    @asynccontextmanager
+    async def fake_db_context():
+        yield db
+
+    monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context)
+
+    approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id)
+
+    assert promoted is True
+    assert approval is not None
+    assert approval.status == ApprovalStatus.APPROVED
+    assert approval.risk_level == RiskLevel.MEDIUM
+    assert approval.required_signatures == 0
+    assert approval.metadata["legacy_recorded_risk_level"] == "critical"
+    assert approval.metadata["current_policy_reclassified_risk_level"] == "medium"
+    assert approval.metadata["current_policy_reclassification_basis"] == (
+        "deterministic_action_and_blast_radius"
+    )
+    assert len(db.events) == 1
+
+
 @pytest.mark.asyncio
 async def test_legacy_critical_pending_gate_is_not_promoted(monkeypatch) -> None:
     record = _approval_record(
         action="host reboot 192.168.0.110",
-        risk_level=RiskLevel.HIGH,
+        risk_level=RiskLevel.CRITICAL,
     )
     db = _FakeDB(record)
 

apps/api/tests/test_sre_typed_domain_router.py

@@ -18,6 +18,7 @@ from src.services.awooop_ansible_audit_service import (  # noqa: E402
     _catalog_hints,
     build_ansible_decision_audit_payload,
     get_ansible_catalog_item,
+    preflight_ansible_candidate_generation,
     record_ansible_decision_audit,
 )
 from src.services.awooop_ansible_check_mode_service import (  # noqa: E402
@@ -111,6 +112,7 @@ def _sentry_runtime_proposal(occurrence_id: str) -> dict:
         "source_occurrence_id": occurrence_id,
         "source_recurrence_verified": True,
         "source_recurrence_source": "current_alert_webhook",
+        "source_recurrence_anchor": "webhook_payload",
     }
 
 
@@ -642,6 +644,9 @@ async def test_stale_broad_candidate_is_no_write_superseded_before_exact_candida
     assert candidate_input["candidate_catalog_schema"] == "typed_domain_router_v2"
     assert candidate_input["target_route_generation"].startswith("typed-route-v2:")
     assert candidate_input["source_recurrence"]["verified"] is True
+    assert candidate_input["source_recurrence"]["identity_anchor"] == (
+        "webhook_payload"
+    )
     assert candidate_input["target_selector"]["catalog_ids"] == [SENTRY_CATALOG]
     assert candidate_input["target_selector"]["inventory_hosts"] == ["host_110"]
 
@@ -734,6 +739,33 @@ async def test_same_terminal_generation_accepts_one_fresh_occurrence(
     assert sum("'decision_manager'" in sql for sql in db.statements) == 1
 
 
+@pytest.mark.asyncio
+async def test_webhook_generation_rejects_unverified_historical_recurrence(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    @asynccontextmanager
+    async def unexpected_db_context(_project_id: str = "awoooi"):
+        raise AssertionError("unverified recurrence must stop before DB evidence lookup")
+        yield
+
+    monkeypatch.setattr(
+        "src.services.awooop_ansible_audit_service.get_db_context",
+        unexpected_db_context,
+    )
+    proposal = _sentry_runtime_proposal("alert-current-1")
+    proposal.pop("source_fingerprint")
+
+    result = await preflight_ansible_candidate_generation(
+        incident=_sentry_runtime_incident(),
+        proposal_data=proposal,
+        decision_path="repair_candidate_controlled_queue",
+        not_used_reason="test",
+    )
+
+    assert result["status"] == "current_source_recurrence_not_verified"
+    assert result["should_collect_evidence"] is False
+
+
 def test_webhook_recurrence_receipt_requires_explicit_occurrence_identity() -> None:
     proposal = _sentry_runtime_proposal("")
     payload = build_ansible_decision_audit_payload(