From 19d7db93e53e4885f54981eac5d0680129438750 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 15:31:47 +0800 Subject: [PATCH] fix(aiops): require current recurrence before controlled apply --- apps/api/src/api/v1/webhooks.py | 1 + .../awooop_ansible_candidate_backfill_job.py | 85 ++++++++++++++----- apps/api/src/services/approval_db.py | 13 ++- .../services/awooop_ansible_audit_service.py | 41 ++++++++- ...t_awooop_ansible_candidate_backfill_job.py | 62 ++++++++++++++ ...t_current_owner_controlled_apply_policy.py | 37 +++++++- .../api/tests/test_sre_typed_domain_router.py | 32 +++++++ 7 files changed, 246 insertions(+), 25 deletions(-) diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index 28b5d5eb4..a0ef04ae8 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -442,6 +442,7 @@ async def _try_auto_repair_background( resolved_source_fingerprint and source_alert_id ), "source_recurrence_source": "current_alert_webhook", + "source_recurrence_anchor": "webhook_payload", }, project_id=str(getattr(incident, "project_id", None) or "awoooi"), ) diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index e89e2d5a2..504116b34 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -49,6 +49,9 @@ _BACKFILL_REASON = ( _ERROR_BACKOFF_MIN_SECONDS = 15.0 _ERROR_BACKOFF_MAX_SECONDS = 30.0 _CURRENT_RECURRENCE_MAX_AGE_SECONDS = 15 * 60 +_BACKFILL_MIN_SCAN_LIMIT = 100 +_BACKFILL_MAX_SCAN_LIMIT = 500 +_BACKFILL_SCAN_MULTIPLIER = 20 def _error_backoff_seconds() -> float: @@ -99,17 +102,12 @@ async def _fetch_missing_candidate_incidents( AS latest_candidate_source_occurrence_generation, latest_candidate.is_terminal AS latest_candidate_is_terminal, - ( - SELECT approval.id - FROM approval_records approval - WHERE approval.incident_id = incidents.incident_id - AND lower(approval.status::text) IN ( - 'approved', - 'execution_failed' - ) - ORDER BY approval.updated_at DESC - LIMIT 1 - ) AS approval_id + current_approval.id AS approval_id, + CASE + WHEN current_approval.id IS NOT NULL + THEN 'approval_record_fingerprint' + ELSE 'incident_signal_fingerprint' + END AS source_fingerprint_anchor FROM incidents JOIN LATERAL ( SELECT @@ -134,6 +132,19 @@ async def _fetch_missing_candidate_incidents( ORDER BY recurrence.received_at DESC LIMIT 1 ) current_recurrence ON TRUE + LEFT JOIN LATERAL ( + SELECT approval.id, approval.fingerprint + FROM approval_records approval + WHERE approval.incident_id = incidents.incident_id + AND lower(approval.status::text) IN ( + 'approved', + 'execution_failed' + ) + AND NULLIF(approval.fingerprint, '') + = current_recurrence.source_fingerprint + ORDER BY approval.updated_at DESC + LIMIT 1 + ) current_approval ON TRUE LEFT JOIN LATERAL ( SELECT candidate.input ->> 'target_route_generation' @@ -157,16 +168,34 @@ async def _fetch_missing_candidate_incidents( incidents.project_id = :project_id OR incidents.project_id IS NULL ) - AND incidents.created_at >= NOW() - ( - :window_hours * INTERVAL '1 hour' + AND ( + incidents.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + OR current_recurrence.received_at >= NOW() - ( + :recurrence_max_age_seconds * INTERVAL '1 second' + ) ) AND incidents.resolved_at IS NULL AND upper(coalesce(incidents.status::text, '')) NOT IN ('RESOLVED', 'CLOSED') - AND NULLIF( - incidents.signals #>> '{0,labels,fingerprint}', - '' - ) = current_recurrence.source_fingerprint + AND ( + current_approval.id IS NOT NULL + OR EXISTS ( + SELECT 1 + FROM jsonb_array_elements( + coalesce( + incidents.signals::jsonb, + '[]'::jsonb + ) + ) incident_signal(value) + WHERE NULLIF( + incident_signal.value #>> + '{labels,fingerprint}', + '' + ) = current_recurrence.source_fingerprint + ) + ) ORDER BY current_recurrence.received_at DESC LIMIT :scan_limit """.replace( @@ -208,6 +237,9 @@ def _build_backfill_proposal(incident: dict[str, Any]) -> dict[str, Any]: "source_recurrence_source": str( incident.get("source_recurrence_source") or "" ), + "source_recurrence_anchor": str( + incident.get("source_fingerprint_anchor") or "" + ), } @@ -875,6 +907,8 @@ async def enqueue_missing_ansible_candidates_once( return { "skipped": True, "scanned": 0, + "scan_limit": 0, + "scan_window_exhausted": False, "queued": 0, "already_existing_or_write_skipped": 0, "no_catalog_candidate": 0, @@ -901,13 +935,21 @@ async def enqueue_missing_ansible_candidates_once( window_hours or settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS, ) # The batch write remains bounded by ``bounded_limit`` below, but the - # read-side scan must not let a small group of newer, non-catalog alerts - # starve an older exact typed candidate. One hundred rows is still a - # bounded query and matches the existing hard ceiling. - scan_limit = 100 + # read-side scan must not let 25 newer, non-catalog alerts starve an older + # exact typed candidate. Keep the read window finite while giving each + # write slot enough room to skip non-catalog rows. + scan_limit = min( + _BACKFILL_MAX_SCAN_LIMIT, + max( + _BACKFILL_MIN_SCAN_LIMIT, + bounded_limit * _BACKFILL_SCAN_MULTIPLIER, + ), + ) stats: dict[str, Any] = { "skipped": False, "scanned": 0, + "scan_limit": scan_limit, + "scan_window_exhausted": False, "queued": 0, "already_existing_or_write_skipped": 0, "no_catalog_candidate": 0, @@ -979,6 +1021,7 @@ async def enqueue_missing_ansible_candidates_once( scan_limit=scan_limit, ) stats["scanned"] = len(incidents) + stats["scan_window_exhausted"] = len(incidents) >= scan_limit for incident in incidents: if stats["queued"] >= bounded_limit: break diff --git a/apps/api/src/services/approval_db.py b/apps/api/src/services/approval_db.py index 5328c0ab9..3fa61956b 100644 --- a/apps/api/src/services/approval_db.py +++ b/apps/api/src/services/approval_db.py @@ -469,7 +469,11 @@ class ApprovalDBService: effective_risk = classify_risk( action=str(record.action or ""), blast_radius=current.blast_radius, - explicit_level=current.risk_level, + # This is a current-policy reconciliation of a legacy pending + # row. Reusing its stored explicit level makes a stale + # CRITICAL classification self-perpetuating and prevents the + # deterministic action/blast classifier from ever running. + explicit_level=None, ) if status != "pending" or effective_risk == RiskLevel.CRITICAL: return current, False @@ -481,6 +485,13 @@ class ApprovalDBService: "owner_review_gate": "auto_waived_for_low_medium_high", "controlled_apply_required": True, "legacy_pending_promoted": True, + "legacy_recorded_risk_level": current.risk_level.value, + "current_policy_reclassified_risk_level": ( + effective_risk.value + ), + "current_policy_reclassification_basis": ( + "deterministic_action_and_blast_radius" + ), } ) record.status = ApprovalStatus.APPROVED diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 42dc7d31b..b60288396 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -54,11 +54,14 @@ def _source_recurrence_receipt( ) -> dict[str, Any]: """Build a fail-closed receipt proving this is not a historical-only replay.""" - fingerprint = _source_fingerprint(incident, proposal) + explicit_fingerprint = str( + proposal.get("source_fingerprint") or "" + ).strip() + fingerprint = explicit_fingerprint or _source_fingerprint(incident, proposal) proposal_source = str(proposal.get("source") or "").strip() occurrence_id = str(proposal.get("source_occurrence_id") or "").strip() verified = bool( - fingerprint + explicit_fingerprint and occurrence_id and proposal.get("source_recurrence_verified") is True ) @@ -68,6 +71,10 @@ def _source_recurrence_receipt( "fingerprint": fingerprint or None, "occurrence_id": occurrence_id or None, "observed_at": str(proposal.get("source_received_at") or "") or None, + "identity_anchor": str( + proposal.get("source_recurrence_anchor") or "" + ) + or None, "source": ( "current_alert_webhook" if proposal_source == "alert_webhook_controlled_router" @@ -1539,6 +1546,21 @@ async def preflight_ansible_candidate_generation( occurrence = str( payload["input"].get("source_occurrence_generation") or "" ) + source_recurrence = payload["input"].get("source_recurrence") or {} + proposal_source = str(payload["input"].get("proposal_source") or "") + if ( + proposal_source + in { + "alert_webhook_controlled_router", + "truth_chain_candidate_backfill", + } + and source_recurrence.get("verified") is not True + ): + return { + "status": "current_source_recurrence_not_verified", + "should_collect_evidence": False, + "target_route_generation": generation, + } terminal_sql = verified_ansible_candidate_terminal_sql("candidate") try: async with get_db_context(project_id) as db: @@ -1678,6 +1700,21 @@ async def record_ansible_decision_audit( payload["input"].get("target_route_generation") or "" ) source_recurrence = payload["input"].get("source_recurrence") or {} + proposal_source = str(payload["input"].get("proposal_source") or "") + if ( + proposal_source + in { + "alert_webhook_controlled_router", + "truth_chain_candidate_backfill", + } + and source_recurrence.get("verified") is not True + ): + logger.warning( + "ansible_candidate_write_blocked", + incident_id=incident_id, + reason="current_source_recurrence_not_verified", + ) + return False try: async with get_db_context(str(project_id)) as db: await db.execute( diff --git a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py index b4c0c951c..2edc05b1f 100644 --- a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py +++ b/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 diff --git a/apps/api/tests/test_current_owner_controlled_apply_policy.py b/apps/api/tests/test_current_owner_controlled_apply_policy.py index f5bbc3e65..328e53728 100644 --- a/apps/api/tests/test_current_owner_controlled_apply_policy.py +++ b/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) diff --git a/apps/api/tests/test_sre_typed_domain_router.py b/apps/api/tests/test_sre_typed_domain_router.py index de779c94e..6a6a07013 100644 --- a/apps/api/tests/test_sre_typed_domain_router.py +++ b/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(