"""Turn collector output into ledger rows, idempotently. Called once per source per cron tick. Everything about it is written so that running it twice in a row changes nothing the second time: * Occurrences are keyed on ``(source, source_ref_type, source_ref_id)`` and upserted, never appended. * Parent counters are **recomputed from the children** after each source's pass, never incremented. An incremented counter is wrong the first time a tick overlaps itself, and this platform runs overlapping ticks (a deploy mid-run, a manual trigger next to a scheduled one). * Status transitions are computed from the counters, so the lifecycle cannot disagree with the evidence. Three rules that are easy to get wrong and expensive to get wrong: **A manual verdict outranks an automatic one.** Once a human sets the root cause, or closes something as ``wont_fix``, no later collector pass may overwrite it. ``wont_fix`` in particular must stay closed even as occurrences keep arriving — that is what "we have decided to live with this" means, and reopening it nightly is how a ledger becomes noise nobody reads. **A partial scan may not close anything.** If a collector failed halfway, its candidate list is not evidence of absence. The engine ingests what arrived and skips absence-based recovery entirely. **Recovery is per occurrence; resolution is per cluster.** The parent resolves only when its last open occurrence recovers, and it resolves itself — no notification, no human step. The parent reopens when a new occurrence arrives against a ``resolved`` row, and *that* is worth telling someone, which is why ``recurrence_count`` is tracked here and acted on in PR3. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass, field from datetime import UTC, datetime from sqlalchemy import case, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.automation_exception.collectors.base import ( Candidate, CollectorResult, ) from app.domains.automation_exception.models import ( AutomationException, AutomationExceptionOccurrence, ) from app.domains.automation_exception.ownership import resolve_owner from app.domains.automation_exception.rules import classify_cause, compute_severity from app.domains.automation_exception.signature import ( canonical_signature, signature_hash, ) from app.domains.automation_exception.taxonomy import ( CAUSE_UNKNOWN, CLOSED_STATUSES, REOPENABLE_STATUSES, SEVERITY_NORMAL, SEVERITY_ORDER, STATUS_ACKNOWLEDGED, STATUS_OPEN, STATUS_RESOLVED, STATUS_WONT_FIX, validate_cause, validate_severity, validate_source, ) from app.domains.automation_exception.timeutil import aware as _aware from app.integrations.models import Integration logger = logging.getLogger(__name__) #: How much raw vendor text we keep per occurrence. Enough to read the #: sentence and the code; not enough for a multi-megabyte traceback to #: dominate the table. _MAX_ERROR_TEXT = 4000 #: ``cause_source`` values. CAUSE_SOURCE_NONE = "none" CAUSE_SOURCE_RULE = "rule" CAUSE_SOURCE_AI = "ai" CAUSE_SOURCE_MANUAL = "manual" #: Verdicts an automatic pass may never overwrite. ``ai`` is here #: alongside ``manual`` because PR2's triage costs a model call and #: reasons over the whole cluster; letting a one-line regex silently #: replace it on the next tick would throw that away and make the #: displayed cause flap between two answers. _PROTECTED_CAUSE_SOURCES = frozenset({CAUSE_SOURCE_MANUAL, CAUSE_SOURCE_AI}) #: Recovery kinds on the occurrence. RECOVERY_SOURCE_CLEARED = "source_cleared" RECOVERY_REPLAY = "replay" RECOVERY_MANUAL = "manual" #: Resolution kinds on the parent. RESOLUTION_AUTO = "auto" RESOLUTION_REPLAY = "replay" RESOLUTION_MANUAL = "manual" @dataclass(slots=True) class IngestStats: """What one source's pass did — persisted as the cron's summary.""" source: str candidates: int = 0 exceptions_created: int = 0 exceptions_reopened: int = 0 exceptions_resolved: int = 0 occurrences_created: int = 0 occurrences_recovered: int = 0 absence_recovery_applied: bool = False partial: bool = False warnings: list[str] = field(default_factory=list) def as_dict(self) -> dict[str, object]: return { "source": self.source, "candidates": self.candidates, "exceptions_created": self.exceptions_created, "exceptions_reopened": self.exceptions_reopened, "exceptions_resolved": self.exceptions_resolved, "occurrences_created": self.occurrences_created, "occurrences_recovered": self.occurrences_recovered, "absence_recovery_applied": self.absence_recovery_applied, "partial": self.partial, "warnings": self.warnings[:20], } def _clip(text: str | None) -> str | None: if text is None: return None stripped = text.strip() if not stripped: return None return stripped[:_MAX_ERROR_TEXT] #: Keyword arguments ``rules.compute_severity`` accepts from a #: collector's ``severity_facts``. Filtered rather than splatted blindly: #: a collector that adds a fact this build does not know must degrade to #: "that fact is not considered", never to a ``TypeError`` that takes the #: whole ingest pass down. _SEVERITY_FACT_KEYS = frozenset({ "action", "subject_is_pending_hire", "subject_is_leaver", "notification_bucket", "slo_tier", "integration_enabled", }) def _worse(left: str | None, right: str) -> str: """The louder of two severities. Lower ``SEVERITY_ORDER`` sorts first.""" if left is None: return right return left if SEVERITY_ORDER.get(left, 9) < SEVERITY_ORDER.get(right, 9) else right def _cluster_severity( *, source: str, root_cause: str, candidates: list[Candidate], ) -> str: """Worst severity any candidate in this cluster justifies. 🔴 The collectors' ``severity_facts`` are the ONLY place the facts that make a failure critical exist — a revoke that did not happen for a leaver, a pending hire with no account, a P0 cron, an undelivered actionable notification. They are per-occurrence and the parent stores none of them, so if they are not folded in here they are folded in nowhere: every cluster would be rated from the four columns ``_severity_for`` can see, and the ledger's "worst first" first screen would never show the rows that are actually worst. Worst-wins across the cluster because the cluster is one problem: a Zoom 429 affecting forty seat grants, one of which is a leaver's revoke, is a critical problem with thirty-nine ordinary symptoms. """ worst: str | None = None for cand in candidates: facts = { key: value for key, value in (cand.severity_facts or {}).items() if key in _SEVERITY_FACT_KEYS } worst = _worse( worst, compute_severity(source=source, root_cause=root_cause, **facts), ) return worst or SEVERITY_NORMAL async def _load_integrations( db: AsyncSession, ids: set[uuid.UUID], ) -> dict[uuid.UUID, Integration]: """One query for every integration the batch references.""" if not ids: return {} rows = await db.execute( select(Integration).where(Integration.id.in_(sorted(ids, key=str))), ) return {row.id: row for row in rows.scalars().all()} async def ingest_source( db: AsyncSession, *, source: str, result: CollectorResult, now: datetime | None = None, ) -> IngestStats: """Fold one collector's answer into the ledger. Flushes but does not commit — the caller owns the transaction, so a cron tick that ingests five sources either records all five or none and can be retried without half-applied state. """ validate_source(source) now = now or datetime.now(UTC) stats = IngestStats( source=source, candidates=len(result.candidates), partial=result.partial, warnings=list(result.warnings), ) integrations = await _load_integrations( db, {c.integration_id for c in result.candidates if c.integration_id}, ) # ---- 1. group candidates into clusters ------------------------- #: (scope_key, signature_hash) -> (signature text, candidates) clusters: dict[tuple[str, str], tuple[str, list[Candidate]]] = {} for cand in result.candidates: signature = canonical_signature(cand.error_text) digest = signature_hash( source=source, scope_key=cand.scope_key or None, signature=signature, ) key = (cand.scope_key or "", digest) if key not in clusters: clusters[key] = (signature, []) clusters[key][1].append(cand) seen_refs: set[str] = {c.source_ref_id for c in result.candidates} #: Clusters an occurrence LEFT this pass. Step 4 discovers touched #: parents by asking "who owns the refs we saw", which by then #: answers with the NEW parent — the old one would keep an #: ``open_occurrence_count`` for a row it no longer has and sit in #: the workbench as a live problem with no evidence under it until #: the daily reconciler swept it. vacated: set[uuid.UUID] = set() # ---- 2. upsert each cluster and its occurrences ---------------- for (scope_key, digest), (signature, cands) in clusters.items(): exc = ( await db.execute( select(AutomationException).where( AutomationException.source == source, AutomationException.scope_key == scope_key, AutomationException.signature_hash == digest, ).limit(1), ) ).scalar_one_or_none() representative = max(cands, key=lambda c: c.occurred_at) integration = ( integrations.get(representative.integration_id) if representative.integration_id else None ) earliest = min(c.occurred_at for c in cands) latest = representative.occurred_at if exc is None: owner = await resolve_owner( db, integration=integration, software_id=representative.software_id, ) rule_cause = classify_cause(representative.error_text) new_cause = validate_cause(rule_cause or CAUSE_UNKNOWN) exc = AutomationException( source=source, scope_key=scope_key, signature_hash=digest, signature=signature[:255], status=STATUS_OPEN, root_cause=new_cause, severity=validate_severity( _cluster_severity( source=source, root_cause=new_cause, candidates=cands, ), ), cause_source=( CAUSE_SOURCE_RULE if rule_cause else CAUSE_SOURCE_NONE ), integration_id=representative.integration_id, owner_person_id=owner.person_id, owner_source=owner.source, first_seen_at=earliest, last_seen_at=latest, sample_error=_clip(representative.error_text), context=representative.detail, ) db.add(exc) await db.flush() stats.exceptions_created += 1 else: # A cluster that was resolved and is failing again. Reopen — # unless a human closed it as wont_fix, which stays closed. if exc.status in REOPENABLE_STATUSES: exc.status = STATUS_OPEN exc.recurrence_count = (exc.recurrence_count or 0) + 1 exc.resolved_at = None exc.resolved_by_id = None exc.resolution_kind = None exc.escalated_at = None # Cleared for the same reason as ``escalated_at``: both # mark "we already said something about THIS episode", # and a reopen starts a new episode. Leaving it set made # the recovery echo once-per-lifetime — the person whose # account broke a second time would never be told it was # fixed a second time, which is the exact silence this # epic exists to remove. exc.subject_notified_at = None stats.exceptions_reopened += 1 stored_first = _aware(exc.first_seen_at) stored_last = _aware(exc.last_seen_at) if stored_first is None or earliest < stored_first: exc.first_seen_at = earliest if stored_last is None or latest > stored_last: exc.last_seen_at = latest exc.sample_error = _clip(representative.error_text) exc.context = representative.detail # Re-run the rule only where nothing better exists: never # over a protected verdict, and otherwise only to fill an # unknown — a rule that already decided keeps its answer so # the cause does not flap as sample errors rotate. if exc.cause_source not in _PROTECTED_CAUSE_SOURCES and ( exc.cause_source == CAUSE_SOURCE_NONE or exc.root_cause == CAUSE_UNKNOWN ): rule_cause = classify_cause(representative.error_text) if rule_cause: exc.root_cause = validate_cause(rule_cause) exc.cause_source = CAUSE_SOURCE_RULE # Ownership can change under us (a system owner leaves, a # backup takes over). Re-resolve only while nobody is # assigned, so a resolved owner stays stable for the humans # coordinating around it. if exc.owner_person_id is None: owner = await resolve_owner( db, integration=integration, software_id=representative.software_id, ) exc.owner_person_id = owner.person_id exc.owner_source = owner.source if exc.integration_id is None and representative.integration_id: exc.integration_id = representative.integration_id # This tick's candidates carry facts the parent does not # store, so re-fold them in. Worst-wins, matching # ``_severity_for`` — the ledger never quietly gets quieter. if exc.status not in CLOSED_STATUSES: exc.severity = validate_severity( _worse( exc.severity, _cluster_severity( source=source, root_cause=exc.root_cause, candidates=cands, ), ), ) # ---- occurrences ------------------------------------------- existing_rows = ( await db.execute( select(AutomationExceptionOccurrence).where( AutomationExceptionOccurrence.source == source, AutomationExceptionOccurrence.source_ref_id.in_( [c.source_ref_id for c in cands], ), ), ) ).scalars().all() existing = { (r.source_ref_type, r.source_ref_id): r for r in existing_rows } for cand in cands: row = existing.get((cand.source_ref_type, cand.source_ref_id)) if row is None: db.add( AutomationExceptionOccurrence( exception_id=exc.id, source=source, source_ref_type=cand.source_ref_type, source_ref_id=cand.source_ref_id, person_id=cand.person_id, occurred_at=cand.occurred_at, error_text=_clip(cand.error_text), detail=cand.detail, replayable=cand.replayable, ), ) stats.occurrences_created += 1 continue # Already recorded. Re-point it if the same source row now # fails with a different signature (a credential problem # that became a quota problem) — the cluster it belongs to # is whatever it is failing with *now*. if row.exception_id != exc.id: vacated.add(row.exception_id) row.exception_id = exc.id row.person_id = cand.person_id or row.person_id stored_occurred = _aware(row.occurred_at) row.occurred_at = ( cand.occurred_at if stored_occurred is None or cand.occurred_at > stored_occurred else stored_occurred ) row.error_text = _clip(cand.error_text) row.detail = cand.detail row.replayable = cand.replayable # It is failing again, so it is not recovered. if row.recovered_at is not None: row.recovered_at = None row.recovery_kind = None await db.flush() # ---- 3. recovery ---------------------------------------------- recovered_ids = await _apply_recovery( db, source=source, result=result, seen_refs=seen_refs, now=now, ) stats.occurrences_recovered = len(recovered_ids) stats.absence_recovery_applied = ( result.absence_means_recovery and not result.partial ) # ---- 4. recompute parents from their children ------------------ # Everything this pass could have moved: clusters whose occurrence # just recovered, plus clusters that own a ref we saw failing. touched: set[uuid.UUID] = set(recovered_ids) | vacated if seen_refs: owning = await db.execute( select(AutomationExceptionOccurrence.exception_id).where( AutomationExceptionOccurrence.source == source, AutomationExceptionOccurrence.source_ref_id.in_(sorted(seen_refs)), ), ) touched.update(row[0] for row in owning.all()) stats.exceptions_resolved = await recompute_exceptions( db, exception_ids=touched, now=now, ) await db.flush() return stats async def _apply_recovery( db: AsyncSession, *, source: str, result: CollectorResult, seen_refs: set[str], now: datetime, ) -> set[uuid.UUID]: """Mark occurrences recovered. Returns the parent ids touched.""" conditions_refs: set[str] = set(result.recovered_refs) # Coerced once, outside the loop: the window bound is compared # against every open occurrence, and both sides must be aware. window_start = _aware(result.absence_window_start) stmt = select(AutomationExceptionOccurrence).where( AutomationExceptionOccurrence.source == source, AutomationExceptionOccurrence.recovered_at.is_(None), ) open_rows = (await db.execute(stmt)).scalars().all() touched: set[uuid.UUID] = set() for row in open_rows: recovered = row.source_ref_id in conditions_refs if ( not recovered and result.absence_means_recovery and not result.partial and row.source_ref_id not in seen_refs ): # Absence within the scanned window only. Anything older # than the window was never in scope for this pass and its # absence proves nothing. occurred = _aware(row.occurred_at) if ( window_start is None or occurred is None or occurred >= window_start ): recovered = True if recovered: row.recovered_at = now row.recovery_kind = RECOVERY_SOURCE_CLEARED touched.add(row.exception_id) if touched: await db.flush() return touched async def recompute_exceptions( db: AsyncSession, *, exception_ids: set[uuid.UUID] | None, now: datetime | None = None, ) -> int: """Recompute counters, severity and status from the occurrences. Returns how many clusters transitioned into ``resolved``. Pass ``exception_ids=None`` to sweep every non-closed cluster (used by the daily reconciler); pass a set to touch only what a source's pass moved. """ now = now or datetime.now(UTC) if exception_ids is not None and not exception_ids: return 0 stmt = select(AutomationException) if exception_ids is not None: stmt = stmt.where(AutomationException.id.in_(sorted(exception_ids, key=str))) else: stmt = stmt.where( AutomationException.status.notin_(sorted(CLOSED_STATUSES)), ) parents = (await db.execute(stmt)).scalars().all() if not parents: return 0 counts = ( await db.execute( select( AutomationExceptionOccurrence.exception_id, func.count(AutomationExceptionOccurrence.id), func.sum( case( (AutomationExceptionOccurrence.recovered_at.is_(None), 1), else_=0, ), ), # COUNT(DISTINCT ...) skips NULLs, so the CASE's ``else`` # branch is what excludes both recovered occurrences and # subject-less ones from the affected-people figure. func.count( func.distinct( case( ( AutomationExceptionOccurrence.recovered_at.is_(None), AutomationExceptionOccurrence.person_id, ), else_=None, ), ), ), ) .where( AutomationExceptionOccurrence.exception_id.in_( [p.id for p in parents], ), ) .group_by(AutomationExceptionOccurrence.exception_id), ) ).all() by_id = {row[0]: (int(row[1]), int(row[2] or 0), int(row[3] or 0)) for row in counts} # Batched once for the whole set: severity depends on whether the # integration is still live, and a per-parent query here would be a # round trip per row on the daily full sweep. integration_ids = {p.integration_id for p in parents if p.integration_id} enabled_by_integration: dict[uuid.UUID, bool] = {} if integration_ids: rows = await db.execute( select(Integration.id, Integration.enabled).where( Integration.id.in_(sorted(integration_ids, key=str)), ), ) enabled_by_integration = {row[0]: bool(row[1]) for row in rows.all()} resolved_now = 0 for parent in parents: total, still_open, people = by_id.get(parent.id, (0, 0, 0)) parent.occurrence_count = total parent.open_occurrence_count = still_open parent.affected_person_count = people if still_open == 0 and parent.status not in CLOSED_STATUSES: parent.status = STATUS_RESOLVED parent.resolved_at = now parent.resolution_kind = RESOLUTION_AUTO parent.escalated_at = None resolved_now += 1 elif still_open > 0 and parent.status == STATUS_RESOLVED: # Defensive: a parent can only be resolved with zero open # occurrences. If the two ever disagree the counters are # the evidence and the status is the derived value. parent.status = ( STATUS_ACKNOWLEDGED if parent.acknowledged_at else STATUS_OPEN ) parent.resolved_at = None parent.resolution_kind = None if parent.status not in CLOSED_STATUSES: parent.severity = validate_severity( _severity_for( parent, affected_person_count=people, # Unknown integration ⇒ treated as live. That is the # safe direction: a missing row should not quietly # downgrade a real problem. # ``None`` (no integration at all) and "not found" # both mean "no reason to downgrade", so both take # the True default rather than being conflated with # a genuinely disabled connector. integration_enabled=( enabled_by_integration.get(parent.integration_id, True) if parent.integration_id is not None else True ), ), ) # Flush before returning. Without it these are in-memory-only edits, # and the next statement a caller issues — a refresh, a re-read, the # detail payload the transition endpoints return — can re-populate # the objects from the unchanged rows and silently discard the # recomputation. await db.flush() return resolved_now def _severity_for( parent: AutomationException, *, affected_person_count: int, integration_enabled: bool = True, ) -> str: """Severity from the parent's stored facts. The per-candidate facts a collector supplies drive severity on every ingest pass (``_cluster_severity``); on recompute only what is persisted is available, so this recomputes from the cluster-level facts (source, cause, how many people) and — crucially — never downgrades. That "never downgrades" is what carries the ingest-time verdict through the daily reconcile: without it, the reconciler would rate every cluster from four columns and silently undo the only place the critical-making facts are ever considered. A cluster that was once critical because a leaver kept access does not become normal just because the affected-count query rounded down. """ from app.domains.automation_exception.taxonomy import ( SEVERITY_CRITICAL, SEVERITY_ORDER, ) computed = compute_severity( source=parent.source, root_cause=parent.root_cause, affected_person_count=affected_person_count, # The REAL flag, not an assumption. Hardcoding ``True`` meant a # retired integration's exceptions were recomputed as if it were # live and — because the rule below keeps the worse of the two — # promoted from ``low`` to ``high`` on the next reconcile. A # deliberately disabled connector would have grown louder on the # workbench the longer it stayed disabled, which is backwards. integration_enabled=integration_enabled, ) if parent.severity == SEVERITY_CRITICAL: return SEVERITY_CRITICAL # Keep the worse of the two so a collector's richer facts win. if SEVERITY_ORDER.get(parent.severity, 9) < SEVERITY_ORDER.get(computed, 9): return parent.severity return computed __all__ = [ "CAUSE_SOURCE_AI", "CAUSE_SOURCE_MANUAL", "CAUSE_SOURCE_NONE", "CAUSE_SOURCE_RULE", "RECOVERY_MANUAL", "RECOVERY_REPLAY", "RECOVERY_SOURCE_CLEARED", "RESOLUTION_AUTO", "RESOLUTION_MANUAL", "RESOLUTION_REPLAY", "STATUS_WONT_FIX", "IngestStats", "ingest_source", "recompute_exceptions", ]