Markdown Copy

Ask AI

DataQ — System Architecture

Keep these diagrams in sync with the code. When a new component, datasource, integration, or DB table is added, update the relevant diagram in the same PR.

☁️ These diagrams are drawn using Azure's component names (Container Apps, Key Vault, App Insights, Azure AD) because that's the concrete deployment they trace through. DataQ runs as an equally-live peer deployment on AWS (ECS Fargate, Secrets Manager, CloudWatch+X-Ray, Cognito) behind the exact same seams — see deployment parity for the side-by-side. Neither cloud is primary; read "Key Vault" / "App Insights" / "Azure AD" below as "the secret store" / "the observability backend" / "the OIDC authority" and substitute the AWS equivalent where relevant.

%%{init: {'flowchart': {'curve': 'linear'}}}%%
flowchart LR
    subgraph orch["⚙️ Orchestration — monitor + trigger only"]
        ADF["Azure Data Factory"]
        AF["Apache Airflow"]
        DBT["dbt"]
    end
    subgraph clients["🌐 Clients"]
        AICli["AI clients · Claude/Copilot/Cursor"]
        Web["Web UI · React SPA"]
    end
    subgraph platform["🟦 DataQ Platform · Azure Container Apps"]
        Frontend["Frontend · nginx + React SPA<br/>sole public ingress · runtime OIDC config"]
        API["FastAPI · internal ingress<br/>REST /api/v1 + /mcp · OIDC JWT"]
        Worker["Celery worker<br/>GX Core execution"]
        Beat["Celery beat · own process<br/>schedules only, never executes"]
        Infra["PostgreSQL · Redis · Key Vault · App Insights<br/>suites·runs·results·pipeline_runs · queue · secrets · traces"]
    end
    subgraph alerts["🔔 Alerts · ResultPublisher seam"]
        Teams["Teams · Slack"]
        Email["Email · SMTP"]
    end
    subgraph ds["📊 Datasources — GX checks execute here"]
        SF["Snowflake · DEV/QA/UAT"]
        Files["ADLS Gen2 · S3 · flat files"]
        UC["Unity Catalog · Databricks"]
        Iceberg["Apache Iceberg · native pyiceberg read<br/>object storage + catalog"]
    end

    Web -->|HTTPS| Frontend
    AICli -->|MCP · HTTP| Frontend
    orch -->|"webhook / HMAC callback<br/>+ 10-min REST poll fallback"| Frontend
    Frontend -->|"same-origin proxy<br/>/api + /mcp + /healthz"| API
    API -->|enqueue run| Worker
    API -.-> Infra
    Worker -.-> Infra
    Beat -.->|schedule tick, enqueue| Infra
    Worker -->|run outcome| alerts
    Worker -->|GX checks| ds

    classDef hub fill:#E6F1FB,stroke:#185FA5,color:#0C449C
    classDef src fill:#E1F5EE,stroke:#0F6E56,color:#085041
    classDef integ fill:#FAEEDA,stroke:#854F0B,color:#633806
    classDef notify fill:#FAECE7,stroke:#993C1D,color:#712B13
    classDef client fill:#F1EFE8,stroke:#5F5E5A,color:#444441
    class API,Worker,Beat,Infra,Frontend hub
    class SF,Files,UC,Iceberg src
    class ADF,AF,DBT integ
    class Teams,Email notify
    class Web,AICli client
    style platform fill:#F4F9FE,stroke:#185FA5
    style ds fill:#F0FBF7,stroke:#0F6E56
    style orch fill:#FDF7EC,stroke:#854F0B
    style alerts fill:#FDF3EF,stroke:#993C1D
    style clients fill:#F7F6F2,stroke:#5F5E5A
    linkStyle default stroke:#5B6B7B,stroke-width:1.5px

The flow reads left → right: inputs (Clients, Orchestration) drive the DataQ platform in the centre, which acts on its targets on the right (runs GX checks against datasources, publishes outcomes to alert channels).

Legend

Colour	Group
Grey	Clients — browser web UI + AI clients (over MCP)
Orange	Orchestration — ADF · Airflow · dbt (monitor + trigger only, never datasources)
Blue	DataQ platform — Frontend (nginx SPA) · FastAPI · Celery worker · Celery beat (own process) · PostgreSQL · Redis · Key Vault · App Insights
Green	Datasources — GX checks run against these
Red	Alert channels — Teams · Slack · Email (the ResultPublisher seam)

Data model (ER diagram)

Source of truth: backend/app/db/models.py (28 tables). Update this diagram in the same PR as any model/migration change.

erDiagram
    users {
        uuid id PK
        string aad_object_id UK "nullable — NULL for non-AAD (email-OTP) identities"
        string oidc_issuer "issuer that vouched for aad_object_id (provider-neutral OIDC)"
        string email UK "unique on lower(email) — the cross-authenticator identity key"
        string display_name
        bool display_name_override "user-set name survives IdP re-sync"
        string role "admin | member | viewer — CHECK-constrained, ADR 0033"
        timestamptz last_seen_at
    }
    workspace_members {
        uuid id PK
        string email UK "unique on lower(email) - who is admitted to the workspace"
        string initial_role "seeds the users row on first sign-in only - ADR 0043"
        string source "admin | auto_import - auto_import rows await review"
        uuid invited_by FK "SET NULL - the membership outlives the inviter"
        timestamptz created_at
    }
    api_keys {
        uuid id PK
        uuid user_id FK "CASCADE"
        string name "label, e.g. ci-smoke"
        string key_prefix "dq_live_ + 4 — identifies without revealing"
        string key_hash UK "sha256 hex; plaintext never stored"
        timestamptz expires_at "mandatory expiry"
        timestamptz revoked_at "soft revocation"
        timestamptz last_used_at "throttled telemetry"
    }
    sessions {
        uuid id PK
        uuid user_id FK "CASCADE"
        string token_hash UK "sha256 hex of the dq_sess_ cookie token"
        timestamptz expires_at "fixed horizon — no refresh pair"
        timestamptz revoked_at "logout"
    }
    otp_codes {
        uuid id PK
        string email "normalized (strip+lower); deliberately NOT FK'd to users"
        string code_hash "sha256 hex of the 6-digit code"
        timestamptz expires_at "10-minute TTL"
        timestamptz consumed_at "redeemed OR superseded by a re-request"
        int attempts "atomically incremented before each compare; capped"
    }
    connections {
        uuid id PK
        string name "unique per env"
        string type "snowflake / adls_gen2 / s3 / unity_catalog / iceberg / adf / airflow / dbt"
        string env "dev / qa / uat / prod"
        jsonb config "non-secret datasource config"
        string secret_ref "SecretStore key, never the credential"
        uuid created_by FK
        timestamptz last_polled_at "orchestration poll health: + last_poll_error, consecutive_poll_failures, health_alerted_at"
        timestamptz credential_expires_at "readable credential lifetime: + credential_expiry_checked_at — NULL = unknown, never 'does not expire'"
        timestamptz lineage_last_refresh_at "lineage pull state (ADR 0034): + lineage_watermark, lineage_last_tier, lineage_degraded_reason, lineage_last_error"
        timestamptz inventory_sync_last_attempted_at "inventory sync state (ADR 0040): + inventory_sync_last_error, _failing_since, _last_table_count, _zero_since"
    }
    connection_versions {
        uuid id PK
        uuid connection_id FK
        int version_no "per-connection sequence"
        string name
        string type
        string env
        jsonb config
        uuid changed_by FK "SET NULL"
    }
    assets {
        uuid id PK
        string namespace "OpenLineage dataset namespace (ADR 0034)"
        string name "OpenLineage dataset name — (namespace,name) unique"
        string env "metadata, not identity"
        uuid connection_id FK "provenance hint (SET NULL)"
        uuid owner_user_id FK "incident-routing hop (SET NULL — reserved for future routing use)"
        string description "workspace-Admin-set (ADR 0034 §4)"
        jsonb column_tags "warehouse governance classifications, cached per read"
        timestamptz column_tags_refreshed_at "'never looked' vs 'looked, found none'"
        timestamptz first_seen
        timestamptz last_seen
    }
    lineage_edges {
        uuid id PK
        uuid upstream_asset_id FK "CASCADE (ADR 0034)"
        uuid downstream_asset_id FK "CASCADE"
        string source "lineage source, e.g. 'dbt'"
        uuid connection_id FK "CASCADE — refreshing conn (provenance + prune scope); (up,down,source,connection_id) unique"
        jsonb columns "column-level mapping when the source provides one (none_as_null)"
        timestamptz first_seen
        timestamptz last_seen
    }
    suites {
        uuid id PK
        string name
        string description
        uuid connection_id FK
        uuid asset_id FK "resolved target asset (SET NULL, fail-soft)"
        jsonb target "table / path / UC name the checks run against"
        jsonb column_policy "failing-sample redaction policy"
        uuid created_by FK
    }
    checks {
        uuid id PK
        uuid suite_id FK
        string name
        string kind "expectation / freshness / volume / schema_drift / anomaly / comparison (ADR 0012/0015)"
        string expectation_type
        string dimension "DQ dimension (ADR 0038) — nullable; NULL = unclassified"
        uuid source_connection_id FK "comparison baseline datasource (ADR 0015; RESTRICT) — set iff kind='comparison'"
        numeric warn_threshold
        numeric fail_threshold
        numeric critical_threshold
        jsonb config "GX expectation kwargs"
        timestamptz alert_snoozed_until
    }
    monitor_baselines {
        uuid id PK
        uuid check_id FK "UNIQUE - one current baseline"
        string kind "denormalized from check"
        jsonb baseline "kind-shaped: schema snapshot / anomaly params"
        timestamptz captured_at
        uuid captured_by FK "NULL = run-captured"
    }
    check_versions {
        uuid id PK
        uuid check_id FK
        int version_no "per-check sequence"
        string name
        string kind
        string expectation_type
        string dimension "DQ dimension (ADR 0038) — nullable; NULL = unclassified"
        uuid source_connection_id "snapshot — plain UUID, deliberately no FK"
        jsonb config
        numeric warn_threshold
        numeric fail_threshold
        numeric critical_threshold
        uuid changed_by FK "SET NULL"
    }
    audit_events {
        uuid id PK
        timestamptz occurred_at
        string action_class "config (ADR 0041 phase 1) / access (phase 2)"
        string action "check.update / share.grant / connection.reauth / …"
        string entity_type
        uuid entity_id "NO FK — the audit row must outlive the entity"
        uuid actor_user_id FK "SET NULL"
        string actor_kind "user / pat / webhook — deliberately no 'system'"
        string actor_label "denormalized, survives the SET NULL"
        jsonb before
        jsonb after
        string request_id
        string prev_hash "hash chain (ADR 0041 §9)"
        string row_hash "NULL = written before the chain shipped, not backfilled"
    }
    audit_chain_state {
        int id PK "singleton — always 1"
        string head_hash "locked via FOR UPDATE at commit time"
        uuid head_event_id "NO FK — see audit_events.entity_id"
        timestamptz updated_at
    }
    audit_chain_checkpoints {
        uuid id PK
        timestamptz created_at
        timestamptz cutoff "the retention cutoff this purge ran against"
        int deleted_count
        uuid last_deleted_event_id "NO FK — about to be deleted"
        string last_deleted_row_hash
        uuid first_surviving_event_id "NO FK — nullable if the purge deleted everything"
        boolean anchored "whether the external anchor publish succeeded"
    }
    runs {
        uuid id PK
        uuid suite_id FK
        uuid asset_id FK "stamped at dispatch (SET NULL) — targets change; history shouldn't rewrite"
        string status "queued / running / succeeded / failed / cancelled"
        string triggered_by "manual / schedule / provider:pipeline:run_id"
        string celery_task_id
        string failure_reason "operational failure classification"
        timestamptz started_at
        timestamptz finished_at
    }
    results {
        uuid id PK
        uuid run_id FK
        uuid check_id FK
        string status "pass warn fail critical + skip error"
        numeric metric_value "SQL-aggregatable scalar (ADR 0012)"
        int duration_ms
        jsonb observed_value
        jsonb expected_value
        jsonb sample_failures "redacted failing rows"
        timestamptz sample_failures_purged_at
        jsonb sampling "scale-aware read strategy + counts — NULL = full read"
    }
    shares {
        uuid id PK
        uuid suite_id FK
        uuid user_id FK
        string permission "view / edit (admin schema-legal, never granted)"
    }
    pipeline_runs {
        uuid id PK
        string provider "adf / airflow / dbt"
        uuid connection_id FK
        string provider_run_id "unique per provider"
        string pipeline_or_dag_id
        string env
        string status
        timestamptz started_at
        timestamptz finished_at
        string failure_reason
        timestamptz last_updated_at "staleness guard for the polling upsert"
    }
    trigger_bindings {
        uuid id PK
        string provider
        string pipeline_or_dag_id
        string env
        uuid suite_id FK
        bool enabled
    }
    schedules {
        uuid id PK
        uuid suite_id FK
        string cron "5-field, evaluated in timezone"
        string timezone "IANA name, default UTC"
        bool enabled
        timestamptz next_run_at "precomputed next fire"
        timestamptz last_run_at
        uuid created_by FK
    }
    suite_notifications {
        uuid id PK
        uuid suite_id FK "unique - one row per suite"
        bool enabled
        string alert_on "fail / warn / always"
        string webhook_secret_ref "per-suite Teams webhook, SecretStore key"
        string slack_webhook_secret_ref "per-suite Slack webhook ref (NULL → workspace webhook)"
        string email_recipients "comma-separated addresses (not secret; NULL → workspace EMAIL_TO)"
        bool auto_resolve_incidents "auto-resolve on pass (default true, ADR 0034)"
    }
    notification_channels {
        uuid id PK
        string name "admin-chosen label"
        string type "teams / slack / email"
        string webhook_secret_ref "teams/slack - SecretStore key"
        string email_recipients "email type - comma-separated addresses, not secret"
        uuid created_by FK "SET NULL"
    }
    suite_notification_channels {
        uuid suite_id PK,FK "CASCADE"
        uuid channel_id PK,FK "RESTRICT - unlink before deleting a live channel"
    }
    workspace_health {
        string key PK "signal name, e.g. orchestration_poll_staleness"
        timestamptz alerted_at "delivered-first flag - set only after a publish succeeded"
    }
    privacy_settings {
        int id PK "always 1 - singleton zero-sample privacy switch"
        bool zero_sample_mode "default false - effective state is the env floor OR this"
        uuid updated_by FK "the admin who last changed it; SET NULL on erasure"
    }
    scoring_settings {
        int id PK "always 1 - singleton health-score weights (ADR 0005 amendment)"
        numeric warn_weight "0 <= warn <= fail <= critical, critical > 0 (table CHECK)"
        numeric fail_weight
        numeric critical_weight "also the normaliser - all-critical scores 0"
        uuid updated_by FK "the admin who last changed it; SET NULL on erasure"
    }
    llm_settings {
        int id PK "always 1 - singleton provider config (ADR 0042)"
        string provider "anthropic / openai_compatible"
        string base_url "required for openai_compatible; the credential's destination"
        string model
        string api_key_secret_ref "SecretStore ref - the key itself is never stored"
        string structured_output "native / prompt_json"
        bool enabled "default false - no row or disabled means no LLM features"
    }
    llm_invocations {
        uuid id PK
        string kind "ping / sql_generation / check_suggestion / rca_narrative"
        string status "pending / running / succeeded / failed"
        uuid requested_by_user_id FK "SET NULL - record outlives requester"
        uuid suite_id FK "SET NULL - the cost record outlives its scope"
        jsonb request "caller ask - never warehouse values"
        string context_fingerprint "sha256 of the assembled prompt"
        jsonb response
        string error
        int input_tokens "+ output_tokens, duration_ms - the cost record"
    }
    incidents {
        uuid id PK
        uuid asset_id FK "CASCADE (ADR 0034)"
        uuid check_id FK "CASCADE — (asset,check) anchor; ≤1 active per pair (partial unique index)"
        uuid suite_id FK "CASCADE — authz + routing"
        string status "open / acknowledged / resolved"
        string resolved_by "user / auto (NULL until resolved)"
        int occurrence_count "repeat failures attach, not duplicate"
        timestamptz last_seen_at
        timestamptz acknowledged_at "+ acknowledge_note — lifecycle transition record"
        uuid acknowledged_by FK "SET NULL"
        timestamptz resolved_at "+ resolution_note"
        uuid resolved_by_user_id FK "SET NULL"
        uuid prior_incident_id FK "SET NULL — reopen chain"
        jsonb evidence "deterministic layer-1 card (no sample rows — PII)"
    }

    users ||--o{ api_keys : "PATs (CASCADE — keys die with the user)"
    users |o--o{ workspace_members : "invited_by (SET NULL)"
    users ||--o{ sessions : "OTP browser sessions (CASCADE)"
    users ||--o{ connections : "created_by"
    users ||--o{ suites : "created_by"
    users ||--o{ schedules : "created_by"
    users ||--o{ shares : "grantee (CASCADE)"
    users |o--o{ connection_versions : "changed_by (SET NULL)"
    users |o--o{ check_versions : "changed_by (SET NULL)"
    users |o--o{ audit_events : "actor_user_id (SET NULL)"
    users |o--o{ llm_invocations : "requested_by (SET NULL)"
    suites |o--o{ llm_invocations : "context scope (SET NULL)"

    connections ||--o{ connection_versions : "config history (CASCADE)"
    connections ||--o{ suites : "datasource for"
    connections ||--o{ pipeline_runs : "orchestrator connection"
    connections |o--o{ assets : "provenance hint (SET NULL)"

    users |o--o{ assets : "owner_user_id (SET NULL)"
    assets |o--o{ suites : "resolved target (SET NULL)"
    assets |o--o{ runs : "stamped at dispatch (SET NULL)"
    assets ||--o{ lineage_edges : "upstream (CASCADE)"
    assets ||--o{ lineage_edges : "downstream (CASCADE)"
    connections ||--o{ lineage_edges : "refreshed by (CASCADE)"

    suites ||--o{ checks : "contains (CASCADE)"
    connections |o--o{ checks : "comparison source (ADR 0015, RESTRICT)"
    suites ||--o{ runs : "executed as (CASCADE)"
    suites ||--o{ shares : "shared via (CASCADE)"
    suites ||--o{ trigger_bindings : "triggered by (CASCADE)"
    suites ||--o{ schedules : "scheduled by (CASCADE)"
    suites ||--o| suite_notifications : "alert config (CASCADE)"
    suites ||--o{ suite_notification_channels : "linked channels (CASCADE)"
    notification_channels ||--o{ suite_notification_channels : "referenced by (RESTRICT)"
    users |o--o{ notification_channels : "created_by (SET NULL)"

    checks ||--o| monitor_baselines : "diff reference (CASCADE)"
    users |o--o{ monitor_baselines : "captured_by (SET NULL)"
    checks ||--o{ check_versions : "config history (CASCADE)"
    checks ||--o{ results : "evaluated as (CASCADE)"
    runs ||--o{ results : "produces (CASCADE)"

    assets ||--o{ incidents : "anchored to (CASCADE, ADR 0034)"
    checks ||--o{ incidents : "anchored to (CASCADE)"
    suites ||--o{ incidents : "authz + routing (CASCADE)"
    incidents |o--o| incidents : "reopen chain — prior_incident_id (SET NULL)"

    pipeline_runs ||..o{ runs : "triggered_by marker (no FK)"
    trigger_bindings }o..o{ pipeline_runs : "(provider, pipeline, env) match (no FK)"

Reading notes

• Conventions (elided from the diagram for noise): every table has a gen_random_uuid() UUID PK and created_at; user-editable entities also carry updated_at (runs/results deliberately don't — they are engine-mutated event rows whose lifecycle lives in started_at/finished_at/sample_failures_purged_at). Status/type columns are TEXT + CHECK constraints, not native PG enums (migration ergonomics).
• shares.permission grants are view/edit only. admin is legal in the DB CHECK but never granted (share_service.GRANTABLE_PERMISSIONS): workspace-admin is implicit on every suite (a stored users.role = 'admin', or the WORKSPACE_ADMIN_EMAILS bootstrap/break-glass allowlist — ADR 0033, amending ADR 0027) and owner is suites.created_by, not a share. Authorization is two axes: the stored workspace role gates whole resource classes (connection mutations are Admin-only), the per-suite grant gates individual suites; neither replaces the other.
• Cascade posture (ADR 0020): deleting a suite cascades its checks, runs, results, shares, trigger bindings, schedules, and notification config; deleting a connection cascades its version history. History is not retained past entity deletion — accepted. Version snapshots survive their author (changed_by is SET NULL), not their entity.
• pipeline_runs ≠ runs — no FK between them. Orchestrator pipeline executions correlate to the DQ suite runs they trigger only via the string marker runs.triggered_by = '<provider>:<pipeline_or_dag_id>:<provider_run_id>' (dotted lines above); trigger_bindings matches pipeline runs by (provider, pipeline_or_dag_id, env), also without an FK. A partial unique index on runs (suite_id, triggered_by) dedupes orchestration-triggered runs.
• Singleton constraints: at most one orchestrator connection per (type, env) (partial unique index over adf/airflow/dbt only — datasources may repeat); one suite_notifications row per suite; one live shares row per (suite, user).
• Secrets are never in these tables. connections.secret_ref / suite_notifications.webhook_secret_ref hold SecretStore keys; version snapshots deliberately omit credentials, so a credential rotation records no version.

Runtime flows

The diagrams above show structure (who talks to whom, what is stored); these two show ordering for the flows that cross the most components.

Suite run lifecycle

Every run — manual (POST /suites/{id}/run), scheduled (the 60s beat dispatcher), or orchestration-triggered — converges on the same path once the Run row exists:

sequenceDiagram
    autonumber
    participant Trig as Trigger<br/>(manual API · schedule beat · pipeline event)
    participant API as FastAPI / beat dispatcher
    participant PG as PostgreSQL
    participant BR as Redis broker
    participant W as Celery worker (run_suite)
    participant KV as SecretStore (Key Vault)
    participant DS as Datasource
    participant Pub as ResultPublisher (Teams · Slack · email)

    Trig->>API: trigger suite run
    API->>PG: INSERT runs (status=queued, triggered_by marker)
    API->>BR: send_task run_suite(run_id)
    Note over API,BR: broker down → run marked terminal failed,<br/>never left stuck queued
    API->>PG: store celery_task_id (enables cancel/revoke)

    BR->>W: deliver task
    W->>PG: load run — already cancelled? stop (cooperative cancel)
    W->>PG: load suite + connection + checks, resolve target
    W->>KV: get connection credential (secret_ref)
    W->>W: build CheckRunner by connection.type (registry, ADR 0011)
    Note over W,KV: any setup failure → run terminal failed
    W->>DS: materialize flat-file batch path
    Note over W,DS: batch absent → every check skip,<br/>run still succeeds
    W->>PG: UPDATE runs SET status=running
    W->>DS: execute by check.kind (ADR 0012) —<br/>expectation → GX validate · freshness/volume → monitor SQL
    DS-->>W: one CheckOutcome per check
    W->>W: severity.resolve_status — band unexpected-% against<br/>warn/fail/critical thresholds (ADR 0005/0016)
    W->>PG: INSERT results (status, metric_value, redacted sample_failures)
    W->>PG: UPDATE runs SET status=succeeded<br/>(failed = the adapter raised)
    W->>Pub: publish_run_outcome — snooze suppression → dedup →<br/>publish (severity/alert_on routing inside the publisher)
    Note over W,Pub: best-effort — a publish failure never<br/>affects the persisted run

Key sources: worker/tasks.py (run_suite), services/run_service.py (execute_run), services/run_dispatch.py, services/severity.py, alerting/dispatch.py.

Orchestration event flow

How an ADF / Airflow / dbt pipeline outcome becomes (at most) a triggered suite run. Everything goes through the OrchestrationProvider seam (ADR 0004) — no provider branching:

sequenceDiagram
    autonumber
    participant ORC as ADF / Airflow / dbt
    participant FE as Frontend nginx (public ingress)
    participant API as FastAPI (internal)
    participant KV as SecretStore (Key Vault)
    participant PG as PostgreSQL
    participant BR as Redis broker
    participant W as Celery worker

    Note over ORC: pipeline / DAG / dbt build reaches a terminal state
    ORC->>FE: POST /api/v1/orchestration/events/{provider}<br/>(ADF — Azure Monitor alert · Airflow — on_*_callback · dbt — post-build callback)
    FE->>API: same-origin proxy
    API->>KV: load receiver secret
    API->>API: authenticate — ADF constant-time token query param (ADR 0006)<br/>· Airflow HMAC-SHA256 over raw body (ADR 0007)
    API->>API: OrchestrationProvider.parse → RunUpdate or AlertPing

    alt AlertPing — run-anonymous Azure Monitor alert
        API->>BR: enqueue targeted poll-now (provider + resource)
        API-->>ORC: 200 (body status = reconciling)
    else RunUpdate
        API->>PG: upsert pipeline_runs ON (provider, provider_run_id)
        alt status = succeeded
            API->>PG: match enabled trigger_bindings (provider, pipeline_or_dag_id, env)
            API->>PG: INSERT runs, triggered_by = provider:pipeline:run_id<br/>ON CONFLICT DO NOTHING (dedup index)
            API->>BR: dispatch run_suite per triggered run
        else status = failed
            API->>API: alert the user only — failures never trigger suites
        end
        API-->>ORC: 200 (body status = recorded)
    end

    Note over W,PG: polling fallback — a 10-min beat sweeps every orchestrator connection<br/>via the provider REST API (list_recent_runs, 15-min lookback) into the SAME<br/>upsert + trigger path — a 30-min gap-recovery sweep (1-hour window) covers downtime

Key sources: api/v1/orchestration.py, services/orchestration_service.py, worker/tasks.py (poll_orchestration_runs, recover_orchestration_gaps).

Status semantics

Run lifecycle

runs.status describes execution, not data quality — a run whose checks all failed is still succeeded.

stateDiagram-v2
    [*] --> queued : created (manual · schedule · trigger)
    queued --> running : worker picks up
    queued --> cancelled : user cancel
    queued --> failed : dispatch/setup failure · reaper
    running --> succeeded : execution completed
    running --> cancelled : user cancel
    running --> failed : adapter raised · reaper
    succeeded --> [*]
    failed --> [*]
    cancelled --> [*]

• succeeded means executed — checks may still have failed; the data-quality outcome lives in results.status.
• Cancel works on any non-terminal run: the API sets cancelled and best-effort revokes the Celery task; the worker also honours the status cooperatively (start-check before executing).
• The reaper drives runs orphaned in queued/running past a threshold (task never published, or the worker died mid-run) to terminal failed.

Result status derivation

results.status has two orthogonal families: the four severity tiers (ADR 0005) and the two operational statuses. Only the tiers carry health-score weight (0.5 / 1.0 / 2.0 for warn / fail / critical); skip/error must be excluded from the health-score denominator.

flowchart TD
    O["CheckOutcome (from the runner)"] --> E{"runner could evaluate it?"}
    E -- "no — evaluation raised" --> ERR["error — operational<br/>no tier · no metric · error message in observed_value"]
    E -- yes --> S{"flat-file batch landed?<br/>(decided before execution)"}
    S -- no --> SKIP["skip — operational<br/>not evaluated at all"]
    S -- yes --> M{"thresholds set AND a<br/>bandable metric_value exists?"}
    M -- "no — binary fallback (ADR 0005)" --> BIN{"GX success?"}
    BIN -- yes --> PASS[pass]
    BIN -- no --> FAIL[fail]
    M -- yes --> BAND["band unexpected-% (0-100, higher = worse)<br/>against warn / fail / critical thresholds (ADR 0016)"]
    BAND --> TIER["pass / warn / fail / critical<br/>thresholds are policy — they OVERRIDE GX success"]

    classDef op fill:#F1EFE8,stroke:#5F5E5A,color:#444441
    classDef tier fill:#E6F1FB,stroke:#185FA5,color:#0C449C
    class ERR,SKIP op
    class PASS,FAIL,TIER tier

The single decision lives in services/severity.py (resolve_status), shared by run persistence and the check-editor dry-run so a preview can never disagree with the run it previews.

Trust boundaries & authentication

The one-diagram consolidation of ADRs 0006 / 0007 / 0008 / 0028 — what crosses each boundary and what credential it carries:

flowchart LR
    subgraph internet["🌍 Untrusted — public internet"]
        B["Browser (React SPA)"]
        AI["AI clients (MCP)"]
        WH["ADF / Airflow / dbt webhooks"]
    end
    IDP["🔑 OIDC authority<br/>(Azure AD or AWS Cognito behind the generic DATAQ_AUTH_* contract)"]
    subgraph aca["🟦 Trust boundary — Container Apps env"]
        FE["Frontend nginx — SOLE public ingress (TLS)<br/>serves the SPA + runtime auth config (non-secret)"]
        API["FastAPI — internal ingress only<br/>validates EVERY request itself (no EasyAuth)"]
        WK["Celery worker — no ingress"]
        BT["Celery beat — no ingress, own process"]
    end
    subgraph backing["🔒 Credentialed backing services"]
        KV["Key Vault (all connection + webhook secrets)"]
        PG[("PostgreSQL — dataq db,<br/>least-priv dataq_app role")]
        RD[("Redis — password auth")]
        APPI["App Insights (PII-redacted logs + traces)"]
    end
    subgraph egress["🌐 Outbound — credentials fetched from Key Vault per use"]
        DS["Datasources — Snowflake · ADLS · S3 · Unity Catalog · Iceberg"]
        AL["Teams / Slack webhooks · SMTP"]
        OAPI["ADF / Airflow REST APIs (polling)"]
        LLM["LLM provider — Anthropic / OpenAI-compat<br/>(off by default; masked profiler stats or a<br/>redacted observed_value only — never raw samples)"]
    end

    B -- "OIDC auth-code + PKCE" --> IDP
    IDP -. "same bearer token, minted by the user's<br/>web-app sign-in and copied into the client<br/>(no client-driven flow — ADR 0008)" .-> AI
    B -- "HTTPS · bearer JWT on /api" --> FE
    AI -- "bearer JWT on /mcp" --> FE
    WH -- "ADF: shared-secret token in URL (ADR 0006)<br/>Airflow / dbt: HMAC-SHA256 body signature (ADR 0007/0029)" --> FE
    FE -- "same-origin proxy /api · /mcp · /healthz (HTTP/1.1)" --> API
    API -- "JWT validated (fastapi-azure-auth / MCP JWTVerifier)<br/>+ per-suite authz + sample redaction" --> API
    API -- "enqueue tasks" --> RD
    RD -- "deliver tasks" --> WK
    BT -- "schedule tick, enqueue tasks" --> RD
    API -- "UAMI — no stored credential" --> KV
    WK -- "UAMI — no stored credential" --> KV
    API --> PG
    WK --> PG
    BT --> PG
    API --> APPI
    WK --> APPI
    BT --> APPI
    WK -- "GX checks / monitor SQL" --> DS
    WK -- "run-outcome alerts" --> AL
    WK -- "10-min poll fallback" --> OAPI
    WK -- "sql_generation / check_suggestion / rca_narrative,<br/>via the dedicated llm Celery queue" --> LLM

    classDef hub fill:#E6F1FB,stroke:#185FA5,color:#0C449C
    classDef ext fill:#F1EFE8,stroke:#5F5E5A,color:#444441
    classDef sec fill:#FAEEDA,stroke:#854F0B,color:#633806
    classDef out fill:#E1F5EE,stroke:#0F6E56,color:#085041
    class FE,API,WK,BT hub
    class B,AI,WH,IDP ext
    class KV,PG,RD,APPI sec
    class DS,AL,OAPI,LLM out
    style internet fill:#F7F6F2,stroke:#5F5E5A
    style aca fill:#F4F9FE,stroke:#185FA5
    style backing fill:#FDF7EC,stroke:#854F0B
    style egress fill:#F0FBF7,stroke:#0F6E56

Boundary notes:

• Defense in depth, not perimeter trust: the API validates every request's bearer JWT itself (fastapi-azure-auth for REST, JWTVerifier for MCP — same tenant/audience/scope) even though it is only reachable through the frontend proxy. Platform-level auth (SWA EasyAuth) is explicitly disabled.
• The only endpoints that bypass user JWT auth are the two orchestration webhook receivers (each with its own secret scheme, above) and the health probe. Webhook secrets live in Key Vault and are compared constant-time; they are never logged.
• Nothing secret is baked into images or served to the browser. The frontend's runtime DATAQ_AUTH_* config is non-secret OIDC metadata (ADR 0028); all real secrets resolve at use-time from Key Vault via user-assigned managed identity.
• MCP is fail-closed: without resolvable auth config the /mcp mount does not come up at all (ADR 0008).

Key invariants

• The frontend (Container App on Azure, ECS Fargate task on AWS) is the sole public surface (ADR 0028 §5). It's one generic nginx image whose auth is injected at runtime (DATAQ_AUTH_* → generic OIDC, validated against Azure AD or AWS Cognito — no MSAL, nothing cloud-specific baked in), and it reverse-proxies /api + /mcp + /healthz same-origin to the internal-ingress API. The API is not reachable directly from the internet; external orchestrator webhooks land on the frontend and are proxied through.
• Orchestration providers (ADF · Airflow · dbt) are not datasources. They live in pipeline_runs, not runs. Trigger bindings map (provider, pipeline_id, env) → suite_id.
• Scheduled/triggered suite runs are Celery-only. FastAPI never enqueues GX itself for a full suite run; it dispatches a task. Exception — synchronous preview paths: the check dry-run (POST /suites/{id}/checks/dryrun) and the column profiler (POST /suites/{id}/profile) run a single GX check / a profiling query against the datasource synchronously in a threadpool (persisting nothing) — interactive authoring aids, not scheduled runs.
• The worker consumes two Celery queues. run_suite and the periodic beat-dispatched tasks stay on the default celery queue; the three LLM intelligence tasks (sql_generation · check_suggestion · rca_narrative) route to a dedicated llm queue, so a backlog of long-running suite runs never starves an LLM request behind it. Both queues are consumed by the same worker process (-Q celery,llm) — this is queue separation for fair scheduling, not a second worker deployment.
• celery-beat is its own process/service, never embedded in the worker. Beat only schedules — it enqueues tasks onto the broker, it never executes one — so a worker OOM (overlapping large suites under worker_concurrency=4 and a 2 GiB hard limit) can never take the scheduler down with it. This is what stops orchestration polling, scheduled-suite dispatch, and every daily sweep from going silently dark because the process that ran them died. Exactly one beat instance may ever run (min=max=1 replica / desired_count=1) — a second would double-fire every periodic task.
• The outbound LLM intelligence layer is live, off by default. An admin must configure a provider (Anthropic or an OpenAI-compatible endpoint) and credential before any call leaves the deployment. SQL generation and check suggestions send masked aggregate profiler statistics only; RCA narratives additionally send the triggering check's own observed_value, routed through the same column-policy/warehouse-tag redaction floor every other results surface applies, plus its expected_value (a check-authored threshold, never masked). Raw sample rows are never sent on any path. See Security & data handling.
• All connection secrets via the deployment's secret store in production / staging — Key Vault on Azure, Secrets Manager on AWS. Local dev may resolve secrets via KV_SECRET_* env vars through the EnvSecretStore backend (see backend/app/core/secrets.py). No credentials are ever hardcoded.
• The /mcp endpoint exposes the same service layer to AI clients. The 48 FastMCP tools (25 read-only, 18 that change state, 5 live-probe tools gated like writes) are thin wrappers reusing the same services + per-suite authz + sample redaction as the REST API — no logic duplication. It mounts under either sign-in mode (SSO, email OTP) and stays unmounted, fail-closed, only when none is configured. Under SSO it validates the same OIDC bearer (Azure AD or Cognito — a JWTVerifier on the same tenant/audience/scope) or a PAT; under email OTP a PAT is the only accepted credential — a raw JWT and a session cookie are both rejected there, since there is no IdP-issued bearer to validate and a session is a browser-only credential. See ADR 0008 / ADR 0032.
• Interactive API docs are off in production. /docs, /redoc, and /openapi.json are disabled when ENVIRONMENT=prod (the prod-docs gate); available in dev/staging.