"""Two tables: the deduplicated exception, and each occurrence of it. Why two and not one. A per-failure log already exists five times over — failed provisioning tasks, dead outbox rows, failed cron runs, blocked sync items, a failed connection test. Copying those rows into a sixth log would buy nothing. What is missing is the level above: *this one problem, seen 47 times since Tuesday, on this integration, with this fix*. That is :class:`AutomationException`, and it is the object a human triages, assigns, fixes and closes. :class:`AutomationExceptionOccurrence` is the join back down to evidence. It exists for three jobs a cluster row cannot do: * **Idempotent ingestion.** ``(source, source_ref_type, source_ref_id)`` is unique, so a collector may re-scan the same window any number of times — every cron tick does — without double counting. Every counter on the parent is derived from these rows, never incremented blindly. * **Recovery detection.** A collector reports what is *currently* failing. An occurrence whose source row no longer appears has recovered, and when the last one recovers the parent resolves itself. Nothing has to notice; the absence is the signal. * **Scoped replay.** PR3 replays by occurrence — all of an integration's, or all of one signature's, or just one subject's — and each row carries the source reference needed to do it. The parent keeps counters (``occurrence_count``, ``open_occurrence_count``, ``affected_person_count``) that are recomputed from the children on every ingest rather than incremented. Derived-and-recomputed is slower and correct; incremented is faster and drifts the first time a cron runs twice. The platform has paid for that lesson in two other ledgers. """ from __future__ import annotations import uuid from datetime import datetime from typing import Any from sqlalchemy import ( JSON, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from app.core.database import Base from app.core.sql_types import GUID as Uuid from app.domains.automation_exception.taxonomy import ( CAUSE_UNKNOWN, SEVERITY_NORMAL, STATUS_OPEN, ) class AutomationException(Base): """One problem, however many times it happened. Identity is ``(source, scope_key, signature_hash)``, enforced by a unique constraint so two concurrent collectors cannot create twins. ``scope_key`` is deliberately part of it: the same vendor sentence from two integrations is two problems with two fixes. """ __tablename__ = "autoexc_exceptions" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "source", "scope_key", "signature_hash", name="uq_autoexc_source_scope_signature", ), # The workbench's default query: active rows, worst first. Index("ix_autoexc_status_last_seen", "status", "last_seen_at"), # "everything wrong with this integration" — the credential # remedy's blast radius, and the integration detail page's panel. Index("ix_autoexc_integration_status", "integration_id", "status"), Index("ix_autoexc_source_status", "source", "status"), Index("ix_autoexc_owner_status", "owner_person_id", "status"), # Recurrence sweep: resolved rows that might need reopening, and # the weekly digest's "fixed this week" slice. Index("ix_autoexc_resolved_at", "resolved_at"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) # ---- identity ------------------------------------------------- #: One of ``taxonomy.SOURCES``. source: Mapped[str] = mapped_column(String(32), nullable=False) #: Narrowest deterministic "where": integration slug, cron handler #: key, sync-target slug. Empty string (not NULL) when the source is #: its own scope — NULL in a unique constraint does not compare #: equal to NULL on MySQL, so a nullable column here would let #: duplicate clusters accumulate silently. scope_key: Mapped[str] = mapped_column(String(160), nullable=False, default="") #: 32-char digest from ``signature.signature_hash``. signature_hash: Mapped[str] = mapped_column(String(32), nullable=False) #: Human-readable cluster label from ``signature.canonical_signature``. signature: Mapped[str] = mapped_column(String(255), nullable=False) # ---- what and how bad ----------------------------------------- status: Mapped[str] = mapped_column( String(16), nullable=False, default=STATUS_OPEN, ) severity: Mapped[str] = mapped_column( String(16), nullable=False, default=SEVERITY_NORMAL, ) #: One of ``taxonomy.ROOT_CAUSES``. Collectors set it when a #: deterministic rule fires; otherwise it stays ``unknown`` until #: triage. The remedy is always derived from this, never stored #: separately — one source of truth for "what do I do". root_cause: Mapped[str] = mapped_column( String(32), nullable=False, default=CAUSE_UNKNOWN, ) #: How the current ``root_cause`` was decided: ``rule`` (a #: deterministic collector rule), ``ai`` (PR2 triage), ``manual`` #: (a human overrode it), ``none`` (still unknown). Kept because #: "who decided this" changes how much an admin should trust it, #: and because a manual verdict must never be overwritten by a #: later automatic one. cause_source: Mapped[str] = mapped_column( String(16), nullable=False, default="none", ) # ---- where it hurts ------------------------------------------- #: The integration whose credentials / config are implicated, when #: one can be resolved. This is what makes "replace the credential #: and replay everything" a single action. integration_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("integrations.id", ondelete="SET NULL"), ) #: Who should act. Resolved deterministically at ingest (see #: ``ownership.py``); NULL means nobody could be resolved, which is #: itself worth showing rather than hiding behind a default. owner_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) #: How the owner was resolved (``system_registry`` / ``integration_admin`` #: / ``it_admin`` / ``none``) — so the workbench can say why this #: landed on someone's plate. owner_source: Mapped[str] = mapped_column( String(24), nullable=False, default="none", ) # ---- counters, all recomputed from occurrences ---------------- occurrence_count: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0", ) #: Occurrences not yet recovered. This, not ``occurrence_count``, is #: what "still broken" means. open_occurrence_count: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0", ) #: Distinct people with an unrecovered occurrence. Drives severity #: and the employee-facing回音 in PR3. affected_person_count: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0", ) #: How many times this cluster has been reopened after being #: resolved. A fix that does not hold is a different conversation #: from a first failure, and PR3 escalates on it. recurrence_count: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0", ) first_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) last_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) # ---- evidence ------------------------------------------------- #: One representative raw error, pre-scrubbed. The cluster label is #: normalised beyond recognition of specifics; this is what an admin #: actually reads. sample_error: Mapped[str | None] = mapped_column(Text) #: Structured, source-specific context: handler key, target slug, #: task action, vendor error code. Never credential material. context: Mapped[dict[str, Any] | None] = mapped_column(JSON(none_as_null=True)) # ---- triage output (written by PR2, read from PR1) ------------ #: Short factual explanation. Kept nullable so PR1's rows render #: correctly with nothing in it. advice: Mapped[str | None] = mapped_column(Text) #: The fact references the advice is grounded in, same convention as #: the intelligence domain's grounding guard. advice_refs: Mapped[list[str] | None] = mapped_column(JSON(none_as_null=True)) advice_generated_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) #: Model identifier, so an admin can tell a stale verdict from a #: fresh one after a model change. advice_model: Mapped[str | None] = mapped_column(String(64)) # ---- resolution ----------------------------------------------- acknowledged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) acknowledged_by_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) resolved_by_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) #: ``auto`` when every occurrence recovered on its own, ``replay`` #: when a disposition fixed it, ``manual`` when a human closed it. resolution_kind: Mapped[str | None] = mapped_column(String(16)) resolution_note: Mapped[str | None] = mapped_column(Text) #: Set when the employee-facing "we are on it / it is fixed"回音 #: has gone out, so PR3 cannot send it twice. subject_notified_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), ) #: Set when a recurrence escalation has been sent for the current #: open episode; cleared on resolution. escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) occurrences: Mapped[list[AutomationExceptionOccurrence]] = relationship( "AutomationExceptionOccurrence", back_populates="exception", cascade="all, delete-orphan", passive_deletes=True, ) class AutomationExceptionOccurrence(Base): """One failing source row, pointed at its cluster. ``(source, source_ref_type, source_ref_id)`` is unique — that is the idempotency key for every collector, and the reason a cron can re-scan its whole window on each tick. """ __tablename__ = "autoexc_occurrences" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "source", "source_ref_type", "source_ref_id", name="uq_autoexc_occurrence_source_ref", ), # Parent drill-down and counter recomputation. Index("ix_autoexc_occ_exception_recovered", "exception_id", "recovered_at"), # The recovery pass's own query: "every still-open occurrence of # this source". It runs on every collector tick for every source, # and none of the other indexes lead with ``source`` — without # this it is a full scan of a table that grows with every failure # the platform has ever recorded. Index("ix_autoexc_occ_source_recovered", "source", "recovered_at"), # "what is broken for this person" — the employee-facing回音 and # the person detail page's panel. Index("ix_autoexc_occ_person_recovered", "person_id", "recovered_at"), Index("ix_autoexc_occ_occurred_at", "occurred_at"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) exception_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("autoexc_exceptions.id", ondelete="CASCADE"), nullable=False, ) #: Denormalised from the parent so the unique constraint can be #: enforced without a join, and so a collector can look up "have I #: already recorded this row" with one index hit. source: Mapped[str] = mapped_column(String(32), nullable=False) #: What kind of row this points at, e.g. ``provisioning_task``, #: ``notification_outbox``, ``scheduled_job_run``, #: ``identity_sync_run_item``, ``integration``. Stored as a plain #: string rather than an FK per type: five FKs would make this table #: a five-way polymorphic join and block adding a sixth source. source_ref_type: Mapped[str] = mapped_column(String(48), nullable=False) #: The source row's primary key, as text. Sources use both CHAR(36) #: UUIDs and, for integrations, a stable slug. source_ref_id: Mapped[str] = mapped_column(String(64), nullable=False) #: The person this failure is *about* — the hire whose account did #: not get created, the leaver whose seat did not get revoked, the #: recipient whose card never arrived. NULL for failures with no #: subject (a whole-cron failure, a connection test). person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) occurred_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) #: Raw error text as the source recorded it, truncated and scrubbed. error_text: Mapped[str | None] = mapped_column(Text) #: Per-occurrence structured detail (vendor code, http status, the #: action attempted). The parent's ``context`` is the representative #: copy; this is the specific one. detail: Mapped[dict[str, Any] | None] = mapped_column(JSON(none_as_null=True)) #: Set the moment a collector stops reporting this source row, or a #: replay of it succeeds. NULL = still broken. recovered_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) #: ``source_cleared`` (the row stopped failing on its own), #: ``replay`` (PR3 re-ran it and it worked), ``manual`` (a human #: declared it handled outside Helix). recovery_kind: Mapped[str | None] = mapped_column(String(16)) #: Replay bookkeeping, written by PR3. Kept on the occurrence #: because replay is per-source-row: a bounded attempt count here is #: what stops a "fix and replay" loop from becoming an unbounded #: retry storm against a vendor. replay_attempt_count: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0", ) last_replay_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_replay_error: Mapped[str | None] = mapped_column(Text) #: True when this occurrence cannot be replayed by construction (a #: cron run is not re-runnable per-row; a connection test has no #: payload). Set by the collector that knows, so the disposition UI #: never offers an action that will do nothing. replayable: Mapped[bool] = mapped_column( Boolean, nullable=False, server_default="0", ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) exception: Mapped[AutomationException] = relationship( "AutomationException", back_populates="occurrences", )