"""The triage queue must not starve. Found in the PR5 review. The cron filtered in Python (``needs_triage``) while paginating in SQL. A row that ``needs_triage`` always rejects is never stamped, so it stays in the result set forever — and because the ordering is "worst first", a handful of already-classified high-volume rows fill every page and the rows that actually need a cause are never reached. Triage would run hourly and silently do nothing. The fix put the skip conditions in the WHERE clause. ``needs_triage`` remains the single definition; this file is what stops the two from drifting apart. """ from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from app.domains.automation_exception.ingest import ( CAUSE_SOURCE_AI, CAUSE_SOURCE_MANUAL, CAUSE_SOURCE_NONE, CAUSE_SOURCE_RULE, ) from app.domains.automation_exception.models import AutomationException from app.domains.automation_exception.scheduler_handlers import ( run_automation_exception_triage, ) from app.domains.automation_exception.taxonomy import CAUSE_UNKNOWN from app.domains.automation_exception.triage import needs_triage pytestmark = pytest.mark.asyncio def _exc(**kw) -> AutomationException: now = datetime.now(UTC) return AutomationException( id=uuid.uuid4(), source="provisioning", scope_key="", signature_hash=uuid.uuid4().hex[:32], signature="vendor said no", status="open", severity="normal", root_cause=kw.get("root_cause", CAUSE_UNKNOWN), cause_source=kw.get("cause_source", CAUSE_SOURCE_NONE), occurrence_count=1, open_occurrence_count=kw.get("open_occurrence_count", 1), affected_person_count=0, recurrence_count=0, first_seen_at=now, last_seen_at=now, sample_error="boom", advice_generated_at=kw.get("advice_generated_at"), ) class TestTheQueryAgreesWithNeedsTriage: """If these two disagree, the queue either starves or wastes calls.""" @pytest.mark.parametrize( ("cause_source", "root_cause", "expected"), [ (CAUSE_SOURCE_NONE, CAUSE_UNKNOWN, True), (CAUSE_SOURCE_RULE, CAUSE_UNKNOWN, True), (CAUSE_SOURCE_RULE, "credential", False), (CAUSE_SOURCE_AI, "credential", False), (CAUSE_SOURCE_MANUAL, CAUSE_UNKNOWN, False), (CAUSE_SOURCE_MANUAL, "credential", False), ], ) def test_needs_triage_matches_the_where_clause( self, cause_source, root_cause, expected, ): row = _exc(cause_source=cause_source, root_cause=root_cause) assert needs_triage(row) is expected # The WHERE clause, restated. Kept literal rather than imported # so a change to the query has to be mirrored here deliberately. in_query = ( cause_source != CAUSE_SOURCE_MANUAL and (cause_source in (CAUSE_SOURCE_NONE, "") or root_cause == CAUSE_UNKNOWN) ) assert in_query is expected, ( "the SQL filter and needs_triage must agree, or the queue " "starves (rows that never stamp fill every page forever)" ) class TestTheQueueDoesNotStarve: async def test_classified_rows_do_not_crowd_out_the_ones_that_need_help( self, db_session, monkeypatch, ): """The exact starvation scenario, at small scale.""" seen: list[uuid.UUID] = [] async def _fake_triage(db, exc): seen.append(exc.id) exc.advice_generated_at = datetime.now(UTC) return exc monkeypatch.setattr( "app.domains.automation_exception.triage.triage_exception", _fake_triage, ) # Many already-classified rows with the HIGHEST open counts, so # they sort to the front of every page. for _ in range(10): db_session.add( _exc( cause_source=CAUSE_SOURCE_RULE, root_cause="credential", open_occurrence_count=999, ), ) # One row that actually needs a cause, with the lowest count. needy = _exc(open_occurrence_count=1) db_session.add(needy) await db_session.flush() result = await run_automation_exception_triage( db_session, {"max_per_run": 2}, ) assert needy.id in seen, ( "the row that needs a cause must be reachable even when " "classified high-volume rows sort ahead of it" ) assert result["triaged"] >= 1 class TestARetiredIntegrationGetsQuieterNotLouder: """Regression from the PR5 review. ``_severity_for`` was called with a hardcoded ``integration_enabled=True``, so an exception belonging to a deliberately disabled connector was recomputed as if it were live. Because the rule keeps the WORSE of stored and computed, it was promoted from ``low`` to ``high`` on the next reconcile — a retired integration grew louder the longer it stayed retired, which is backwards. """ async def test_a_disabled_integration_does_not_escalate_on_recompute( self, db_session, ): from app.domains.automation_exception.ingest import recompute_exceptions from app.domains.automation_exception.models import ( AutomationExceptionOccurrence, ) from app.domains.automation_exception.taxonomy import ( SEVERITY_LOW, SOURCE_INTEGRATION, ) from app.integrations.models import Integration retired = Integration( id=uuid.uuid4(), kind="provisioning", provider="zoom", name=f"retired-{uuid.uuid4().hex[:8]}", slug=f"retired-{uuid.uuid4().hex[:8]}", enabled=False, ) db_session.add(retired) await db_session.flush() exc = _exc() exc.source = SOURCE_INTEGRATION exc.severity = SEVERITY_LOW exc.integration_id = retired.id db_session.add(exc) await db_session.flush() # One still-open occurrence, so the parent stays open and the # severity branch is actually reached. db_session.add( AutomationExceptionOccurrence( id=uuid.uuid4(), exception_id=exc.id, source=SOURCE_INTEGRATION, source_ref_type="integration", source_ref_id=str(retired.id), replayable=False, occurred_at=datetime.now(UTC), error_text="creds expired", ), ) await db_session.flush() # Precondition: it really is low before the sweep. assert exc.severity == SEVERITY_LOW await recompute_exceptions(db_session, exception_ids={exc.id}) assert exc.severity == SEVERITY_LOW, ( "a deliberately disabled connector must not grow louder on " "the workbench the longer it stays disabled" ) class TestAPermanentlyFailingRowDoesNotStarveTheQueue: """The second door into the starvation the WHERE clause closed. ``FAILURE_MODEL`` and ``FAILURE_UNGROUNDED`` record the attempt on ``advice_model`` but deliberately do NOT stamp ``advice_generated_at`` — a transient model error deserves another try. So a row the model keeps answering ungroundedly stays in the result set forever, and ordered by volume alone it sits at the top of every page and consumes the whole hourly budget. Same outcome as the Python-side filtering the WHERE clause was written to fix, reached by a different route. Never-attempted rows (``advice_model IS NULL``) therefore sort first, which makes progress monotone without giving up worst-first inside each group. """ async def test_a_row_that_always_fails_does_not_block_a_fresh_one( self, db_session, monkeypatch, ): from app.domains.automation_exception.triage import FAILURE_UNGROUNDED seen: list[uuid.UUID] = [] async def _fake_triage(db, exc): seen.append(exc.id) # Always ungrounded: records the attempt, never stamps. exc.advice_model = FAILURE_UNGROUNDED return exc monkeypatch.setattr( "app.domains.automation_exception.triage.triage_exception", _fake_triage, ) # Rows the model has already refused, with the highest volume so # they win the worst-first tiebreak. for _ in range(4): stuck = _exc(open_occurrence_count=999) stuck.advice_model = FAILURE_UNGROUNDED db_session.add(stuck) fresh = _exc(open_occurrence_count=1) db_session.add(fresh) await db_session.flush() await run_automation_exception_triage(db_session, {"max_per_run": 2}) assert fresh.id in seen, ( "a row nobody has ever asked about must be reached before " "rows the model has already refused four times" )