"""Read surfaces and manual transitions for the exception ledger. The read side answers three questions and is shaped around them: * :func:`list_exceptions` — *what is broken, worst first.* * :func:`get_exception` — *this one problem: its evidence, its people, its history, and what to do about it.* * :func:`get_overview` — *is it getting better or worse.* The write side in PR1 is deliberately small: acknowledge, set the root cause, close as won't-fix, reopen. Every one of those is a human recording a judgement, and each is audited. The interesting writes — apply a remedy, replay, notify the affected people — are PR3's, and they build on these states rather than replacing them. One invariant runs through the whole module: **the ledger never says more than it knows.** A cluster with no resolved owner reports no owner. A cause the rules declined reports ``unknown`` with the remedy "investigate". Nothing is defaulted into looking handled. """ from __future__ import annotations import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import Select, case, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.automation_exception.collectors import SOURCE_REF_TYPES from app.domains.automation_exception.ingest import ( CAUSE_SOURCE_MANUAL, RESOLUTION_MANUAL, recompute_exceptions, ) from app.domains.automation_exception.models import ( AutomationException, AutomationExceptionOccurrence, ) from app.domains.automation_exception.taxonomy import ( ACTIVE_STATUSES, CLOSED_STATUSES, SEVERITIES, SEVERITY_ORDER, SOURCES, STATUS_ACKNOWLEDGED, STATUS_OPEN, STATUS_RESOLVED, STATUS_WONT_FIX, remedy_for, validate_cause, ) from app.identity.person_models import IdentityPerson from app.integrations.models import Integration #: Page size ceiling. Matches the platform's standard admin-list cap so #: the frontend's page-size selector cannot ask for more than the #: backend will return — a mismatch that has silently truncated lists #: elsewhere in this codebase. MAX_PAGE_SIZE = 200 #: How many occurrences a detail view returns. The full set can be #: thousands for a throttling cluster; the newest are what an admin #: reads, and the counters already carry the totals. DETAIL_OCCURRENCE_LIMIT = 100 @dataclass(frozen=True, slots=True) class PersonCard: person_id: uuid.UUID display_name: str work_email: str | None employment_status: str def as_dict(self) -> dict[str, Any]: return { "person_id": str(self.person_id), "display_name": self.display_name, "work_email": self.work_email, "employment_status": self.employment_status, } async def _person_cards( db: AsyncSession, ids: set[uuid.UUID], ) -> dict[uuid.UUID, PersonCard]: if not ids: return {} rows = await db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.employment_status, ).where(IdentityPerson.id.in_(sorted(ids, key=str))), ) return { row[0]: PersonCard( person_id=row[0], display_name=row[1], work_email=row[2], employment_status=row[3], ) for row in rows.all() } async def _integration_cards( db: AsyncSession, ids: set[uuid.UUID], ) -> dict[uuid.UUID, dict[str, Any]]: if not ids: return {} rows = await db.execute( select(Integration).where(Integration.id.in_(sorted(ids, key=str))), ) return { row.id: { "integration_id": str(row.id), "slug": row.slug, "name": row.name, "kind": row.kind, "provider": row.provider, "enabled": bool(row.enabled), "last_test_status": row.last_test_status, "last_tested_at": ( row.last_tested_at.isoformat() if row.last_tested_at else None ), } for row in rows.scalars().all() } def _severity_sort_expression(): """ORDER BY that puts critical first without relying on alphabetics. ``critical`` < ``high`` < ``low`` < ``normal`` alphabetically, which would put ``low`` above ``normal``. The explicit CASE is the only way the list's first screen is reliably the worst screen. """ return case( *[(AutomationException.severity == sev, SEVERITY_ORDER[sev]) for sev in SEVERITIES], else_=len(SEVERITIES), ) def _base_query() -> Select[Any]: return select(AutomationException) def _apply_filters( stmt: Select[Any], *, status: str | None, statuses: list[str] | None, source: str | None, severity: str | None, root_cause: str | None, integration_id: uuid.UUID | None, owner_person_id: uuid.UUID | None, unowned_only: bool, search: str | None, ) -> Select[Any]: if statuses: stmt = stmt.where(AutomationException.status.in_(sorted(set(statuses)))) elif status == "active": # The default view. "Active" is one definition, imported from # the taxonomy, so the list, the counters and the digest can # never disagree about what still needs attention. stmt = stmt.where(AutomationException.status.in_(sorted(ACTIVE_STATUSES))) elif status == "closed": stmt = stmt.where(AutomationException.status.in_(sorted(CLOSED_STATUSES))) elif status and status != "all": stmt = stmt.where(AutomationException.status == status) if source: stmt = stmt.where(AutomationException.source == source) if severity: stmt = stmt.where(AutomationException.severity == severity) if root_cause: stmt = stmt.where(AutomationException.root_cause == root_cause) if integration_id: stmt = stmt.where(AutomationException.integration_id == integration_id) if owner_person_id: stmt = stmt.where(AutomationException.owner_person_id == owner_person_id) if unowned_only: stmt = stmt.where(AutomationException.owner_person_id.is_(None)) if search: needle = f"%{search.strip()}%" stmt = stmt.where( or_( AutomationException.signature.ilike(needle), AutomationException.scope_key.ilike(needle), AutomationException.sample_error.ilike(needle), ), ) return stmt def _exception_dict( row: AutomationException, *, owner: PersonCard | None, integration: dict[str, Any] | None, ) -> dict[str, Any]: return { "id": str(row.id), "source": row.source, "source_ref_type": SOURCE_REF_TYPES.get(row.source), "scope_key": row.scope_key or None, "signature": row.signature, "signature_hash": row.signature_hash, "status": row.status, "severity": row.severity, "root_cause": row.root_cause, "cause_source": row.cause_source, # Derived, never stored — one source of truth for "what do I do". "remedy": remedy_for(row.root_cause), "occurrence_count": row.occurrence_count, "open_occurrence_count": row.open_occurrence_count, "affected_person_count": row.affected_person_count, "recurrence_count": row.recurrence_count, "first_seen_at": row.first_seen_at.isoformat() if row.first_seen_at else None, "last_seen_at": row.last_seen_at.isoformat() if row.last_seen_at else None, "sample_error": row.sample_error, "context": row.context, "advice": row.advice, "advice_refs": row.advice_refs, "advice_generated_at": ( row.advice_generated_at.isoformat() if row.advice_generated_at else None ), "advice_model": row.advice_model, "owner": owner.as_dict() if owner else None, "owner_source": row.owner_source, "integration": integration, "acknowledged_at": ( row.acknowledged_at.isoformat() if row.acknowledged_at else None ), "resolved_at": row.resolved_at.isoformat() if row.resolved_at else None, "resolution_kind": row.resolution_kind, "resolution_note": row.resolution_note, "escalated_at": row.escalated_at.isoformat() if row.escalated_at else None, } async def list_exceptions( db: AsyncSession, *, status: str | None = "active", statuses: list[str] | None = None, source: str | None = None, severity: str | None = None, root_cause: str | None = None, integration_id: uuid.UUID | None = None, owner_person_id: uuid.UUID | None = None, unowned_only: bool = False, search: str | None = None, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: """One page of exceptions, worst first.""" page = max(1, page) page_size = max(1, min(page_size, MAX_PAGE_SIZE)) stmt = _apply_filters( _base_query(), status=status, statuses=statuses, source=source, severity=severity, root_cause=root_cause, integration_id=integration_id, owner_person_id=owner_person_id, unowned_only=unowned_only, search=search, ) total = ( await db.execute( select(func.count()).select_from(stmt.subquery()), ) ).scalar_one() rows = ( await db.execute( stmt.order_by( _severity_sort_expression(), AutomationException.last_seen_at.desc(), ) .offset((page - 1) * page_size) .limit(page_size), ) ).scalars().all() owners = await _person_cards( db, {r.owner_person_id for r in rows if r.owner_person_id}, ) integrations = await _integration_cards( db, {r.integration_id for r in rows if r.integration_id}, ) return { "items": [ _exception_dict( row, owner=owners.get(row.owner_person_id) if row.owner_person_id else None, integration=( integrations.get(row.integration_id) if row.integration_id else None ), ) for row in rows ], "total": int(total), "page": page, "page_size": page_size, } async def get_exception( db: AsyncSession, exception_id: uuid.UUID, ) -> dict[str, Any]: """One exception with its recent occurrences and affected people.""" row = ( await db.execute( select(AutomationException) .where(AutomationException.id == exception_id) .options(selectinload(AutomationException.occurrences)) .limit(1), ) ).scalar_one_or_none() if row is None: raise NotFoundError("AutomationException", str(exception_id)) occurrences = sorted( row.occurrences, key=lambda o: o.occurred_at, reverse=True, )[:DETAIL_OCCURRENCE_LIMIT] # 🔴 Every STILL-AFFECTED person, from the full occurrence set — not # from the newest-100 slice above. ``affected_person_count`` on the # header is a full COUNT(DISTINCT) over the same set, so deriving the # NAMES from the truncated slice made a busy cluster say "23 people # affected" and then list nine of them, with nothing on the page # explaining the gap. still_affected = { o.person_id for o in row.occurrences if o.person_id and o.recovered_at is None } people = await _person_cards( db, still_affected | {o.person_id for o in occurrences if o.person_id} | ({row.owner_person_id} if row.owner_person_id else set()), ) integrations = await _integration_cards( db, {row.integration_id} if row.integration_id else set(), ) payload = _exception_dict( row, owner=people.get(row.owner_person_id) if row.owner_person_id else None, integration=( integrations.get(row.integration_id) if row.integration_id else None ), ) payload["occurrences"] = [ { "id": str(o.id), "source": o.source, "source_ref_type": o.source_ref_type, "source_ref_id": o.source_ref_id, "occurred_at": o.occurred_at.isoformat() if o.occurred_at else None, "error_text": o.error_text, "detail": o.detail, "recovered_at": o.recovered_at.isoformat() if o.recovered_at else None, "recovery_kind": o.recovery_kind, "replayable": bool(o.replayable), "replay_attempt_count": o.replay_attempt_count, "last_replay_at": ( o.last_replay_at.isoformat() if o.last_replay_at else None ), "last_replay_error": o.last_replay_error, "person": ( people[o.person_id].as_dict() if o.person_id and o.person_id in people else None ), } for o in occurrences ] # How many rows the replay button would actually act on. Computed # from the same authority the engine uses (the row's own # ``replayable`` flag, set by the collector that produced it), so # the button can never offer an action that provably does nothing — # which is the dead end this whole surface exists to remove. payload["replayable_occurrence_count"] = sum( 1 for o in occurrences if o.replayable and o.recovered_at is None ) payload["occurrences_truncated"] = ( len(row.occurrences) > DETAIL_OCCURRENCE_LIMIT ) # Distinct still-affected people, so the detail page can name them # rather than only counting them. payload["affected_people"] = [ people[pid].as_dict() for pid in sorted(still_affected, key=str) if pid in people ] return payload async def get_overview( db: AsyncSession, *, window_days: int = 7, ) -> dict[str, Any]: """Counters and trend for the workbench header.""" window_days = max(1, min(window_days, 180)) now = datetime.now(UTC) since = now - timedelta(days=window_days) by_status_rows = await db.execute( select(AutomationException.status, func.count()) .group_by(AutomationException.status), ) by_status = {row[0]: int(row[1]) for row in by_status_rows.all()} active_filter = AutomationException.status.in_(sorted(ACTIVE_STATUSES)) by_severity_rows = await db.execute( select(AutomationException.severity, func.count()) .where(active_filter) .group_by(AutomationException.severity), ) by_source_rows = await db.execute( select(AutomationException.source, func.count()) .where(active_filter) .group_by(AutomationException.source), ) by_cause_rows = await db.execute( select(AutomationException.root_cause, func.count()) .where(active_filter) .group_by(AutomationException.root_cause), ) unowned = ( await db.execute( select(func.count()).select_from(AutomationException).where( active_filter, AutomationException.owner_person_id.is_(None), ), ) ).scalar_one() affected_people = ( await db.execute( select( func.count(func.distinct(AutomationExceptionOccurrence.person_id)), ).where( AutomationExceptionOccurrence.recovered_at.is_(None), AutomationExceptionOccurrence.person_id.is_not(None), ), ) ).scalar_one() # Self-healing rate in the window: of the clusters that were FIXED, # how many fixed themselves. This is the number that says whether the # platform is getting better at recovering. # # 🔴 ``status == resolved``, not ``resolved_at IS NOT NULL``. # ``close_wont_fix`` also stamps ``resolved_at`` — that column means # "closed", not "fixed" — and it writes ``resolution_kind='manual'``, # the same value a human-confirmed fix writes, so nothing downstream # could separate them. Counting won't-fix rows here inflated the # denominator the workbench divides ``auto`` by, i.e. every problem an # admin consciously decided to live with pushed the displayed # self-healing rate DOWN. Same confusion as the recovery echo's # (see ``echo.send_recovery_echoes``), one surface over. The won't-fix # count is not lost: ``by_status`` above carries it. resolved_rows = await db.execute( select(AutomationException.resolution_kind, func.count()) .where( AutomationException.status == STATUS_RESOLVED, AutomationException.resolved_at.is_not(None), AutomationException.resolved_at >= since, ) .group_by(AutomationException.resolution_kind), ) resolved_by_kind = { (row[0] or "unknown"): int(row[1]) for row in resolved_rows.all() } resolved_total = sum(resolved_by_kind.values()) new_in_window = ( await db.execute( select(func.count()).select_from(AutomationException).where( AutomationException.first_seen_at >= since, ), ) ).scalar_one() recurring = ( await db.execute( select(func.count()).select_from(AutomationException).where( active_filter, AutomationException.recurrence_count > 0, ), ) ).scalar_one() return { "window_days": window_days, "generated_at": now.isoformat(), "active_total": sum( count for status, count in by_status.items() if status in ACTIVE_STATUSES ), "by_status": by_status, "by_severity": {row[0]: int(row[1]) for row in by_severity_rows.all()}, "by_source": {row[0]: int(row[1]) for row in by_source_rows.all()}, "by_root_cause": {row[0]: int(row[1]) for row in by_cause_rows.all()}, "unowned_active": int(unowned), "affected_people_now": int(affected_people), "resolved_in_window": resolved_total, "resolved_by_kind": resolved_by_kind, "new_in_window": int(new_in_window), "recurring_active": int(recurring), } # -------------------------------------------------------------------- # Manual transitions # -------------------------------------------------------------------- async def _load(db: AsyncSession, exception_id: uuid.UUID) -> AutomationException: row = ( await db.execute( select(AutomationException) .where(AutomationException.id == exception_id) .limit(1), ) ).scalar_one_or_none() if row is None: raise NotFoundError("AutomationException", str(exception_id)) return row async def acknowledge( db: AsyncSession, exception_id: uuid.UUID, *, actor_user_id: uuid.UUID, trace_id: str, note: str | None = None, ) -> dict[str, Any]: """Record that a human has seen it and accepts it is real.""" row = await _load(db, exception_id) if row.status in CLOSED_STATUSES: raise AppException( code="AUTOEXC_ALREADY_CLOSED", message="This exception is already closed.", status_code=409, ) row.status = STATUS_ACKNOWLEDGED row.acknowledged_at = datetime.now(UTC) row.acknowledged_by_id = actor_user_id if note: row.resolution_note = note[:2000] await db.flush() await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.acknowledge", status="success", resource_type="automation_exception", resource_id=str(row.id), detail={"signature": row.signature[:200], "source": row.source}, ) return await get_exception(db, exception_id) async def set_root_cause( db: AsyncSession, exception_id: uuid.UUID, *, root_cause: str, actor_user_id: uuid.UUID, trace_id: str, note: str | None = None, ) -> dict[str, Any]: """A human's verdict on the cause. Outranks every automatic one.""" row = await _load(db, exception_id) validate_cause(root_cause) previous = row.root_cause row.root_cause = root_cause row.cause_source = CAUSE_SOURCE_MANUAL if note: row.resolution_note = note[:2000] await db.flush() await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.set_root_cause", status="success", resource_type="automation_exception", resource_id=str(row.id), detail={ "from": previous, "to": root_cause, "remedy": remedy_for(root_cause), }, ) return await get_exception(db, exception_id) async def close_wont_fix( db: AsyncSession, exception_id: uuid.UUID, *, reason: str, actor_user_id: uuid.UUID, trace_id: str, ) -> dict[str, Any]: """Close without a fix. Requires a reason, and stays closed. The reason is mandatory on purpose: this is the one transition that silences a real, still-failing problem, and six months later the only defence against "why is this off" is what somebody wrote here. Collector passes will not reopen it — see the ingest engine. """ if not reason or not reason.strip(): raise AppException( code="AUTOEXC_REASON_REQUIRED", message="A reason is required to close an exception without a fix.", status_code=422, ) row = await _load(db, exception_id) row.status = STATUS_WONT_FIX row.resolved_at = datetime.now(UTC) row.resolved_by_id = actor_user_id row.resolution_kind = RESOLUTION_MANUAL row.resolution_note = reason.strip()[:2000] row.escalated_at = None await db.flush() await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.close_wont_fix", status="success", resource_type="automation_exception", resource_id=str(row.id), detail={ "signature": row.signature[:200], "reason": reason.strip()[:500], "open_occurrences_at_close": row.open_occurrence_count, }, ) return await get_exception(db, exception_id) async def reopen( db: AsyncSession, exception_id: uuid.UUID, *, actor_user_id: uuid.UUID, trace_id: str, note: str | None = None, ) -> dict[str, Any]: """Undo a close. The counters decide what it reopens *to*.""" row = await _load(db, exception_id) if row.status not in CLOSED_STATUSES: raise AppException( code="AUTOEXC_NOT_CLOSED", message="Only a closed exception can be reopened.", status_code=409, ) previous = row.status row.status = STATUS_ACKNOWLEDGED if row.acknowledged_at else STATUS_OPEN row.resolved_at = None row.resolved_by_id = None row.resolution_kind = None # 🔴 Both episode stamps, exactly as the automatic reopen in # ``ingest.ingest_source`` clears them. A reopen starts a NEW episode, # and these two mark "we already said something about the LAST one". # # Leaving them set made the manual door a copy of the bug the automatic # door had already fixed: an admin reopens a cluster that was closed # wrongly, it genuinely resolves later, and the people who were waiting # are never told — because a stamp from the previous episode still says # they were. Same for the recurrence escalation: the owner would never # hear that the problem came back a second time. row.subject_notified_at = None row.escalated_at = None if note: row.resolution_note = note[:2000] await db.flush() # Recompute immediately: if every occurrence is in fact recovered, # the cluster resolves itself again on the spot rather than sitting # open and misleading somebody into looking for a live problem. await recompute_exceptions(db, exception_ids={row.id}) await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.reopen", status="success", resource_type="automation_exception", resource_id=str(row.id), detail={"from": previous, "signature": row.signature[:200]}, ) return await get_exception(db, exception_id) async def assign_owner( db: AsyncSession, exception_id: uuid.UUID, *, owner_person_id: uuid.UUID | None, actor_user_id: uuid.UUID, trace_id: str, ) -> dict[str, Any]: """Put it on somebody's plate, or take it off. ``None`` clears the assignment back to unowned, which the ingest engine may then re-resolve automatically. A manual assignment is never overwritten while it stands. """ row = await _load(db, exception_id) if owner_person_id is not None: person = ( await db.execute( select(IdentityPerson).where( IdentityPerson.id == owner_person_id, IdentityPerson.deleted_at.is_(None), ).limit(1), ) ).scalar_one_or_none() if person is None: raise NotFoundError("IdentityPerson", str(owner_person_id)) previous = row.owner_person_id row.owner_person_id = owner_person_id row.owner_source = "manual" if owner_person_id is not None else "none" await db.flush() await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.assign_owner", status="success", resource_type="automation_exception", resource_id=str(row.id), detail={ "from": str(previous) if previous else None, "to": str(owner_person_id) if owner_person_id else None, }, ) return await get_exception(db, exception_id) __all__ = [ "DETAIL_OCCURRENCE_LIMIT", "MAX_PAGE_SIZE", "SOURCES", "STATUS_RESOLVED", "acknowledge", "assign_owner", "close_wont_fix", "get_exception", "get_overview", "list_exceptions", "reopen", "set_root_cause", ] async def replay_exception( db: AsyncSession, exception_id: uuid.UUID, *, actor_user_id: uuid.UUID, trace_id: str, ) -> dict[str, Any]: """Re-drive this exception's open occurrences through their own domains' retry paths. Deliberately does NOT close anything on success. A replay that returned 200 is not proof the underlying work succeeded — the collectors decide recovery from their own evidence, and inventing that proof here would put a lie in the one table built to hold the truth about failures. """ from app.domains.automation_exception.replay import ( # noqa: PLC0415 replay_for_exception, ) row = await _load(db, exception_id) if row.status in CLOSED_STATUSES: raise AppException( code="AUTOEXC_ALREADY_CLOSED", message="This exception is already closed.", status_code=409, ) outcome = await replay_for_exception( db, exception_id, actor_id=actor_user_id, trace_id=trace_id, ) await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.replay", status="success" if outcome.failed == 0 else "partial", resource_type="automation_exception", resource_id=str(row.id), detail={"signature": row.signature[:200], **outcome.as_dict()}, ) payload = await get_exception(db, exception_id) payload["replay"] = outcome.as_dict() return payload async def replay_integration( db: AsyncSession, integration_id: uuid.UUID, *, actor_user_id: uuid.UUID, trace_id: str, ) -> dict[str, Any]: """"The credential is fixed" — re-drive everything under this integration. This is the one-click path, and the reason it takes an integration rather than a list of rows: the admin has already made the only decision available. Handing them back the 47 affected rows to click through would be asking them to do the work twice. """ from app.domains.automation_exception.replay import ( # noqa: PLC0415 replay_for_integration, ) outcome = await replay_for_integration( db, integration_id, actor_id=actor_user_id, trace_id=trace_id, ) await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="automation_exception.replay_integration", status="success" if outcome.failed == 0 else "partial", resource_type="integration", resource_id=str(integration_id), detail=outcome.as_dict(), ) return outcome.as_dict()