Gate Apache Superset Dataset Sync and Cache Warming on dbt Builds and DuckDB Reconciliation

id: dbt-superset-exposure-cache-warm-gate
namespace: company.analytics

description: |
  Build dbt marts and exposures on DuckDB, enforce parameterized row-volume and
  metric reconciliation gates against a persisted KV baseline, and refresh Apache
  Superset dataset schemas and dashboard caches via native HTTP tasks when contracts
  pass, or quarantine BI caches and run AI root-cause triage when data drifts.

concurrency:
  limit: 1
  behavior: QUEUE

triggers:
  - id: on_upstream_ingestion_completed
    type: io.kestra.plugin.core.trigger.Flow
    description: Fires automatically when the upstream raw ingestion pipeline
      finishes successfully.
    dependsOn:
      - namespace: company.data
        flowId: extract-load-postgres
        states:
          - SUCCESS

  - id: inbound_elt_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP webhook triggered by Fivetran, Airbyte, dlt, or
      Git CI post-merge hooks.
    key: "{{ secret('WEBHOOK_KEY') }}"

  - id: morning_bi_refresh_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekday morning build and BI cache warm-up at 06:00 UTC before
      business hours. Shipped disabled by default.
    cron: "0 6 * * 1-5"
    timezone: UTC
    disabled: true

inputs:
  - id: dbt_repository_url
    type: URI
    displayName: dbt Project Repository URL
    description: Git repository containing the dbt models, seeds, tests, and
      exposure definitions.
    defaults: https://github.com/kestra-io/dbt-demo

  - id: dbt_branch
    type: STRING
    displayName: Git Branch
    description: Target branch to clone and build.
    defaults: main

  - id: mart_table
    type: STRING
    displayName: Target Mart Table
    description: Schema-qualified dbt mart table audited for row volume and metric
      reconciliation.
    defaults: main.orders

  - id: primary_key_column
    type: STRING
    displayName: Primary Key Column
    description: Natural or surrogate key column on the target mart table checked
      for nulls and duplicates.
    defaults: order_id

  - id: metric_column
    type: STRING
    displayName: Reconciled Metric Column
    description: Numeric or financial column on the target mart table checked for
      negative values and total drift.
    defaults: amount

  - id: max_volume_drop_pct
    type: FLOAT
    displayName: Max Allowed Row Volume Drop (%)
    description: Maximum percentage drop in mart row count allowed versus the
      previous certified KV baseline (1 to 100).
    defaults: 20.0
    min: 1.0
    max: 100.0

  - id: superset_base_url
    type: URI
    displayName: Apache Superset Base URL
    description: Base URL of the Apache Superset instance hosting the downstream
      executive dashboards.
    defaults: https://superset.internal.example.com

  - id: superset_auth_provider
    type: SELECT
    displayName: Superset Auth Provider
    description: Authentication provider passed to Apache Superset POST
      /api/v1/security/login.
    values:
      - db
      - ldap
    defaults: db

  - id: superset_dry_run
    type: BOOL
    displayName: Superset API Dry-Run Mode
    description: When true, logs the exact Apache Superset dataset refresh and cache
      warm-up requests without calling a live cluster or requiring Superset
      secrets.
    defaults: true

  - id: enable_ai_triage
    type: BOOL
    displayName: Enable Gemini AI Breach Triage
    description: When true, invokes Google Gemini (requires GEMINI_API_KEY) to
      synthesize a root-cause report when the quality gate blocks a refresh.
    defaults: false

  - id: notify_slack
    type: BOOL
    displayName: Send Slack Notifications
    description: When true, dispatches certification scorecards and quarantine
      alerts to Slack (requires SLACK_WEBHOOK_URL).
    defaults: false

  - id: slack_channel
    type: STRING
    displayName: Slack Alert Channel
    description: Destination Slack channel for BI certification scorecards and
      quarantine incident alerts.
    defaults: "#analytics-bi-ops"

  - id: fault_injection_mode
    type: SELECT
    displayName: Fault Injection Mode (Drill)
    description: Optional fault drill mode to test that the quality gate quarantines
      corrupt batches before touching Apache Superset caches.
    values:
      - NONE
      - DROP_90_PCT_VOLUME
      - INJECT_NEGATIVE_REVENUE
      - BREAK_PRIMARY_KEY_UNIQUENESS
    defaults: NONE

tasks:
  - id: load_previous_mart_baseline
    type: io.kestra.plugin.core.kv.Get
    description: Load the last certified mart row count and metric sum from Kestra
      KV Store so silent volume drops are caught across runs.
    key: "dbt_superset_baseline_{{ inputs.mart_table | slugify }}"
    errorOnMissing: false

  - id: dbt_exposure_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Shared workspace to clone the dbt repository, materialize DuckDB
      marts, and export exposure metadata and the mart Parquet snapshot.
    tasks:
      - id: clone_dbt_project
        type: io.kestra.plugin.git.Clone
        description: Clone the dbt project repository at the target branch.
        url: "{{ inputs.dbt_repository_url }}"
        branch: "{{ inputs.dbt_branch }}"
        depth: 1

      - id: build_exposed_marts
        type: io.kestra.plugin.dbt.cli.DbtCLI
        description: Seed data, attach Apache Superset exposure metadata if absent, and
          run dbt build across all models upstream of BI exposures.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: ghcr.io/kestra-io/dbt-duckdb:latest
        env:
          FAULT_MODE: "{{ inputs.fault_injection_mode }}"
          SUPERSET_URL: "{{ inputs.superset_base_url }}"
        profiles: |
          jaffle_shop:
            target: prod
            outputs:
              prod:
                type: duckdb
                path: warehouse.duckdb
                schema: main
                threads: 4
        beforeCommands:
          - |
            if [ ! -f models/bi_exposures.yml ]; then
              cat > models/bi_exposures.yml <<EOF
            version: 2
            exposures:
              - name: executive_revenue_dashboard
                label: "Executive Revenue & Order Velocity"
                type: dashboard
                maturity: high
                url: "${SUPERSET_URL}/superset/dashboard/14/"
                description: "C-suite daily revenue, net order volume, and payment method breakdown in Apache Superset."
                depends_on:
                  - ref('orders')
                  - ref('customers')
                owner:
                  name: "Analytics Engineering"
                  email: "bi-owners@example.com"
                meta:
                  superset_dataset_id: 7
                  superset_dashboard_id: 14
                  superset_table_name: "main.orders"
              - name: customer_cohort_retention
                label: "Customer Cohort LTV & Repeat Rate"
                type: dashboard
                maturity: high
                url: "${SUPERSET_URL}/superset/dashboard/19/"
                description: "Customer lifetime value and cohort repeat order retention dashboard in Apache Superset."
                depends_on:
                  - ref('customers')
                owner:
                  name: "Growth Analytics"
                  email: "growth-bi@example.com"
                meta:
                  superset_dataset_id: 8
                  superset_dashboard_id: 19
                  superset_table_name: "main.customers"
            EOF
            fi
          - |
            case "$FAULT_MODE" in
              DROP_90_PCT_VOLUME)
                sed -i 's/select \* from final/select * from final limit 10/' models/orders.sql
                ;;
              INJECT_NEGATIVE_REVENUE)
                sed -i 's/order_payments\.total_amount as amount/(-1 * order_payments.total_amount) as amount/' models/orders.sql
                ;;
              BREAK_PRIMARY_KEY_UNIQUENESS)
                sed -i 's/select \* from renamed/select * from renamed union all select * from renamed where order_id = 1/' models/staging/stg_orders.sql
                ;;
            esac
        commands:
          - dbt build --select +exposure:* || true
          - cp target/manifest.json dbt_manifest.json && cp
            target/run_results.json dbt_run_results.json

      - id: parse_manifest_exposures
        type: io.kestra.plugin.scripts.python.Script
        description: Parse dbt manifest and run_results to extract Apache Superset
          dataset/dashboard IDs and export the target mart table to Parquet for
          version-safe DuckDB auditing.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: ghcr.io/kestra-io/dbt-duckdb:latest
        env:
          MART_TABLE: "{{ inputs.mart_table }}"
          PK_COL: "{{ inputs.primary_key_column }}"
          METRIC_COL: "{{ inputs.metric_column }}"
        outputFiles:
          - mart_snapshot.parquet
        script: |
          import json
          import os
          import duckdb

          with open("dbt_manifest.json", "r", encoding="utf-8") as mf:
              manifest = json.load(mf)
          with open("dbt_run_results.json", "r", encoding="utf-8") as rf:
              run_results = json.load(rf)

          failed_nodes = []
          passed_count = 0
          for res in run_results.get("results", []):
              status = str(res.get("status", "")).lower()
              if status in ("fail", "error"):
                  failed_nodes.append({
                      "node": res.get("unique_id", "unknown"),
                      "status": status,
                      "message": (res.get("message") or "Check failed").replace("\n", " ")[:140]
                  })
              elif status in ("pass", "success"):
                  passed_count += 1

          exposures = []
          for uid, node in sorted(manifest.get("exposures", {}).items()):
              meta = node.get("meta") or {}
              exposures.append({
                  "unique_id": uid,
                  "name": node.get("name"),
                  "owner_email": (node.get("owner") or {}).get("email", "unknown"),
                  "dataset_id": int(meta.get("superset_dataset_id", 1)),
                  "dashboard_id": int(meta.get("superset_dashboard_id", 1)),
                  "table_name": str(meta.get("superset_table_name", "main.orders"))
              })

          mart_table = os.environ["MART_TABLE"]
          pk_col = os.environ["PK_COL"]
          metric_col = os.environ["METRIC_COL"]

          con = duckdb.connect("warehouse.duckdb")
          try:
              con.execute(f"COPY (SELECT * FROM {mart_table}) TO 'mart_snapshot.parquet' (FORMAT PARQUET)")
          except Exception as exc:
              print(f"Mart table {mart_table} unavailable ({exc}); writing empty Parquet stub.")
              con.execute(f"COPY (SELECT CAST(NULL AS INT) AS {pk_col}, CAST(0.0 AS DOUBLE) AS {metric_col} WHERE 1=0) TO 'mart_snapshot.parquet' (FORMAT PARQUET)")
          finally:
              con.close()

          outputs = {
              "failed_count": len(failed_nodes),
              "passed_count": passed_count,
              "failed_summary": "; ".join(f"{n['node']} ({n['message']})" for n in failed_nodes) if failed_nodes else "none",
              "exposure_count": len(exposures),
              "exposures": exposures,
              "exposure_names": ", ".join(e["name"] for e in exposures) if exposures else "none"
          }
          print("::" + json.dumps({"outputs": outputs}) + "::")

  - id: audit_mart_volume_and_reconciliation
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Query the exported mart Parquet snapshot in DuckDB to verify key
      uniqueness, non-negative metrics, and row-volume stability against the KV
      baseline.
    url: "jdbc:duckdb:"
    inputFiles:
      mart_snapshot.parquet: "{{
        outputs.parse_manifest_exposures.outputFiles['mart_snapshot.parquet']
        }}"
    fetchType: FETCH_ONE
    sql: |
      WITH current_stats AS (
        SELECT
          COUNT(*) AS row_count,
          COUNT(DISTINCT {{ inputs.primary_key_column }}) AS distinct_keys,
          COUNT(*) FILTER (WHERE {{ inputs.primary_key_column }} IS NULL) AS null_key_count,
          COUNT(*) FILTER (WHERE {{ inputs.metric_column }} < 0) AS negative_metric_count,
          ROUND(COALESCE(SUM(CAST({{ inputs.metric_column }} AS DOUBLE)), 0.0), 2) AS metric_sum
        FROM read_parquet('{{ workingDir }}/mart_snapshot.parquet')
      ),
      prev_baseline AS (
        SELECT
          CAST({{ (outputs.load_previous_mart_baseline.value ?? {'row_count': 0, 'metric_sum': 0.0}).row_count }} AS DOUBLE) AS prev_rows,
          CAST({{ (outputs.load_previous_mart_baseline.value ?? {'row_count': 0, 'metric_sum': 0.0}).metric_sum }} AS DOUBLE) AS prev_metric_sum
      ),
      evaluated AS (
        SELECT
          c.row_count,
          c.metric_sum,
          c.null_key_count,
          c.negative_metric_count,
          (c.row_count - c.distinct_keys) AS duplicate_key_count,
          p.prev_rows,
          CASE
            WHEN p.prev_rows > 0 THEN ROUND(((c.row_count - p.prev_rows) / p.prev_rows) * 100.0, 2)
            ELSE 0.0
          END AS volume_delta_pct
        FROM current_stats c
        CROSS JOIN prev_baseline p
      )
      SELECT
        row_count,
        metric_sum,
        null_key_count,
        negative_metric_count,
        duplicate_key_count,
        prev_rows,
        volume_delta_pct,
        CASE
          WHEN row_count = 0 THEN false
          WHEN null_key_count > 0 OR duplicate_key_count > 0 THEN false
          WHEN negative_metric_count > 0 OR metric_sum <= 0 THEN false
          WHEN prev_rows > 0 AND volume_delta_pct < (-1.0 * {{ inputs.max_volume_drop_pct }}) THEN false
          ELSE true
        END AS contract_passed,
        CASE
          WHEN row_count = 0 THEN 'EMPTY_MART_TABLE'
          WHEN null_key_count > 0 OR duplicate_key_count > 0 THEN 'PRIMARY_KEY_INTEGRITY_BREACH'
          WHEN negative_metric_count > 0 OR metric_sum <= 0 THEN 'NEGATIVE_METRIC_RECONCILIATION_FAILURE'
          WHEN prev_rows > 0 AND volume_delta_pct < (-1.0 * {{ inputs.max_volume_drop_pct }}) THEN 'VOLUME_DROP_EXCEEDED_THRESHOLD'
          ELSE 'CERTIFIED'
        END AS contract_verdict
      FROM evaluated;

  - id: publish_execution_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Emit structured mart reconciliation telemetry before downstream
      HTTP or Slack calls so metrics are preserved even if notifications fail.
    values:
      contract_verdict: "{{ outputs.audit_mart_volume_and_reconciliation.row.contract_verdict }}"
      contract_passed: "{{
        outputs.audit_mart_volume_and_reconciliation.row.contract_passed and
        outputs.parse_manifest_exposures.vars.failed_count == 0 and
        outputs.parse_manifest_exposures.vars.passed_count > 0 and
        outputs.parse_manifest_exposures.vars.exposure_count > 0 }}"
      mart_row_count: "{{ outputs.audit_mart_volume_and_reconciliation.row.row_count }}"
      mart_metric_sum: "{{ outputs.audit_mart_volume_and_reconciliation.row.metric_sum }}"
      volume_delta_pct: "{{ outputs.audit_mart_volume_and_reconciliation.row.volume_delta_pct }}"
      exposed_dashboards_count: "{{ outputs.parse_manifest_exposures.vars.exposure_count }}"

  - id: bi_exposure_quality_gate
    type: io.kestra.plugin.core.flow.If
    description: Gate Apache Superset dataset schema sync and cache warming on
      passing dbt builds, non-empty exposures, and passing DuckDB
      reconciliation.
    condition: "{{ outputs.parse_manifest_exposures.vars.failed_count == 0 and
      outputs.parse_manifest_exposures.vars.passed_count > 0 and
      outputs.parse_manifest_exposures.vars.exposure_count > 0 and
      outputs.audit_mart_volume_and_reconciliation.row.contract_passed == true
      }}"
    then:
      - id: authenticate_superset_api
        type: io.kestra.plugin.core.http.Request
        description: Obtain a short-lived JWT access token from Apache Superset POST
          /api/v1/security/login when superset_dry_run is false.
        runIf: "{{ not inputs.superset_dry_run }}"
        uri: "{{ inputs.superset_base_url }}/api/v1/security/login"
        method: POST
        contentType: application/json
        body: |
          {
            "username": "{{ secret('SUPERSET_USERNAME') }}",
            "password": "{{ secret('SUPERSET_PASSWORD') }}",
            "provider": "{{ inputs.superset_auth_provider }}",
            "refresh": true
          }
        retry:
          type: exponential
          interval: PT2S
          maxInterval: PT10S
          maxAttempts: 3

      - id: sync_and_warm_superset_exposures
        type: io.kestra.plugin.core.flow.Loop
        description: Iterate over each dbt exposure with bounded concurrency to refresh
          its Superset dataset schema and warm its chart cache.
        values: "{{ outputs.parse_manifest_exposures.vars.exposures }}"
        concurrencyLimit: 2
        tasks:
          - id: route_superset_execution_mode
            type: io.kestra.plugin.core.flow.If
            description: Execute live Apache Superset REST API calls in parallel when
              superset_dry_run is false, or log the dry-run payload when true.
            condition: "{{ not inputs.superset_dry_run }}"
            then:
              - id: parallel_superset_exposure_sync
                type: io.kestra.plugin.core.flow.Parallel
                description: Concurrently refresh dataset physical columns and warm dashboard
                  chart caches for the current exposure.
                tasks:
                  - id: refresh_superset_dataset_columns
                    type: io.kestra.plugin.core.http.Request
                    description: Sync physical dataset columns in Apache Superset (PUT
                      /api/v1/dataset/{id}/refresh) so schema additions in dbt
                      propagate immediately.
                    uri: "{{ inputs.superset_base_url }}/api/v1/dataset/{{
                      fromJson(item.value).dataset_id }}/refresh"
                    method: PUT
                    headers:
                      Authorization: "Bearer {{
                        fromJson(outputs.authenticate_superset_api.body).access\
                        _token }}"
                    retry:
                      type: exponential
                      interval: PT2S
                      maxInterval: PT10S
                      maxAttempts: 3

                  - id: warm_superset_dashboard_cache
                    type: io.kestra.plugin.core.http.Request
                    description: Trigger Apache Superset PUT /api/v1/chart/warm_up_cache for the
                      exposure's backing table and dashboard ID.
                    uri: "{{ inputs.superset_base_url }}/api/v1/chart/warm_up_cache"
                    method: PUT
                    contentType: application/json
                    headers:
                      Authorization: "Bearer {{
                        fromJson(outputs.authenticate_superset_api.body).access\
                        _token }}"
                    body: |
                      {
                        "table_name": {{ fromJson(item.value).table_name | toJson }},
                        "dashboard_id": {{ fromJson(item.value).dashboard_id }}
                      }
                    retry:
                      type: exponential
                      interval: PT2S
                      maxInterval: PT10S
                      maxAttempts: 3
            else:
              - id: log_superset_dry_run_sync
                type: io.kestra.plugin.core.log.Log
                description: Log the exact Apache Superset REST endpoints and payloads that
                  would be invoked in live mode.
                message: >-
                  [DRY-RUN] Certified exposure '{{ fromJson(item.value).name }}'
                  (owner: {{ fromJson(item.value).owner_email }}) -> 1) PUT {{
                  inputs.superset_base_url }}/api/v1/dataset/{{
                  fromJson(item.value).dataset_id }}/refresh 2) PUT {{
                  inputs.superset_base_url }}/api/v1/chart/warm_up_cache with
                  {"table_name": "{{ fromJson(item.value).table_name }}",
                  "dashboard_id": {{ fromJson(item.value).dashboard_id }}}

      - id: save_certified_baseline
        type: io.kestra.plugin.core.kv.Set
        description: Persist the certified mart row count and metric sum in Kestra KV
          Store only after Superset exposure sync succeeds.
        key: "dbt_superset_baseline_{{ inputs.mart_table | slugify }}"
        kvType: JSON
        overwrite: true
        value: |
          {
            "mart_table": {{ inputs.mart_table | toJson }},
            "row_count": {{ outputs.audit_mart_volume_and_reconciliation.row.row_count }},
            "metric_sum": {{ outputs.audit_mart_volume_and_reconciliation.row.metric_sum }},
            "certified_execution_id": {{ execution.id | toJson }}
          }

      - id: log_certification_scorecard
        type: io.kestra.plugin.core.log.Log
        description: Record the certification summary in the execution log.
        message: >-
          CERTIFIED {{ inputs.mart_table }}: {{
          outputs.audit_mart_volume_and_reconciliation.row.row_count }} rows
          (volume delta {{
          outputs.audit_mart_volume_and_reconciliation.row.volume_delta_pct
          }}%), {{ inputs.metric_column }} sum={{
          outputs.audit_mart_volume_and_reconciliation.row.metric_sum }}. Synced
          and warmed {{ outputs.parse_manifest_exposures.vars.exposure_count }}
          Apache Superset exposures: {{
          outputs.parse_manifest_exposures.vars.exposure_names }}.

      - id: notify_slack_bi_certified
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post a BI certification scorecard to Slack when notify_slack is
          enabled.
        runIf: "{{ inputs.notify_slack }}"
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        retry:
          type: exponential
          interval: PT2S
          maxInterval: PT10S
          maxAttempts: 3
        messageText: >-
          dbt Marts Certified & Apache Superset Caches Warmed ({{
          inputs.slack_channel }}): mart={{ inputs.mart_table }}, rows={{
          outputs.audit_mart_volume_and_reconciliation.row.row_count }} ({{
          outputs.audit_mart_volume_and_reconciliation.row.volume_delta_pct }}%
          volume delta), {{ inputs.metric_column }} sum={{
          outputs.audit_mart_volume_and_reconciliation.row.metric_sum }},
          exposures warmed ({{
          outputs.parse_manifest_exposures.vars.exposure_count }}): {{
          outputs.parse_manifest_exposures.vars.exposure_names }} (execution {{
          execution.id }}).
    else:
      - id: diagnose_exposure_breach
        type: io.kestra.plugin.ai.agent.AIAgent
        description: Analyze the dbt test or DuckDB reconciliation breach with Google
          Gemini when enable_ai_triage is true.
        runIf: "{{ inputs.enable_ai_triage }}"
        provider:
          type: io.kestra.plugin.ai.provider.GoogleGemini
          apiKey: "{{ secret('GEMINI_API_KEY') }}"
          modelName: gemini-2.5-flash
        configuration:
          temperature: 0.1
          responseFormat:
            type: JSON
            jsonSchema:
              type: object
              required:
                - severity
                - root_cause_analysis
                - quarantined_dashboards
                - remediation_steps
              properties:
                severity:
                  type: string
                  enum:
                    - CRITICAL
                    - HIGH
                    - MEDIUM
                root_cause_analysis:
                  type: string
                quarantined_dashboards:
                  type: array
                  items:
                    type: string
                remediation_steps:
                  type: string
        systemMessage: |
          You are a Principal Analytics Reliability Engineer.
          A dbt + DuckDB quality gate blocked an Apache Superset cache warm-up so corrupt data would not overwrite cached executive dashboards.
          Use only the supplied telemetry to explain the failure, list the protected Superset exposures, and provide concrete dbt/SQL remediation steps.
        prompt: |
          Mart Table: {{ inputs.mart_table }}
          Contract Verdict: {{ outputs.audit_mart_volume_and_reconciliation.row.contract_verdict }}
          Failed dbt Nodes ({{ outputs.parse_manifest_exposures.vars.failed_count }}): {{ outputs.parse_manifest_exposures.vars.failed_summary }}
          Current Row Count: {{ outputs.audit_mart_volume_and_reconciliation.row.row_count }} (Baseline: {{ outputs.audit_mart_volume_and_reconciliation.row.prev_rows }}, Delta: {{ outputs.audit_mart_volume_and_reconciliation.row.volume_delta_pct }}%)
          Reconciled {{ inputs.metric_column }} Sum: {{ outputs.audit_mart_volume_and_reconciliation.row.metric_sum }} (Negative Rows: {{ outputs.audit_mart_volume_and_reconciliation.row.negative_metric_count }})
          Null / Duplicate Keys ({{ inputs.primary_key_column }}): {{ outputs.audit_mart_volume_and_reconciliation.row.null_key_count }} / {{ outputs.audit_mart_volume_and_reconciliation.row.duplicate_key_count }}
          Downstream Apache Superset Exposures: {{ outputs.parse_manifest_exposures.vars.exposures | toJson }}

      - id: log_quarantine_alert
        type: io.kestra.plugin.core.log.Log
        level: WARN
        description: Log the BI cache quarantine reason and affected Apache Superset
          exposures.
        message: >-
          BI CACHE QUARANTINE on {{ inputs.mart_table }}: verdict={{
          outputs.audit_mart_volume_and_reconciliation.row.contract_verdict }},
          rows={{ outputs.audit_mart_volume_and_reconciliation.row.row_count }}
          (volume delta {{
          outputs.audit_mart_volume_and_reconciliation.row.volume_delta_pct
          }}%), {{ inputs.metric_column }} sum={{
          outputs.audit_mart_volume_and_reconciliation.row.metric_sum }}, failed
          dbt nodes={{ outputs.parse_manifest_exposures.vars.failed_count }} ({{
          outputs.parse_manifest_exposures.vars.failed_summary }}). Apache
          Superset caches for [{{
          outputs.parse_manifest_exposures.vars.exposure_names }}] were NOT
          refreshed and KV baseline was NOT overwritten.

      - id: notify_slack_bi_quarantine
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alert analytics engineers on Slack when notify_slack is enabled
          that Apache Superset caches were frozen due to a contract breach.
        runIf: "{{ inputs.notify_slack }}"
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        retry:
          type: exponential
          interval: PT2S
          maxInterval: PT10S
          maxAttempts: 3
        messageText: >-
          BI Cache Quarantine ({{ inputs.slack_channel }}) on {{
          inputs.mart_table }}: verdict={{
          outputs.audit_mart_volume_and_reconciliation.row.contract_verdict }},
          rows={{ outputs.audit_mart_volume_and_reconciliation.row.row_count }}
          ({{ outputs.audit_mart_volume_and_reconciliation.row.volume_delta_pct
          }}% volume delta), {{ inputs.metric_column }} sum={{
          outputs.audit_mart_volume_and_reconciliation.row.metric_sum }}, failed
          dbt nodes={{ outputs.parse_manifest_exposures.vars.failed_count }} ({{
          outputs.parse_manifest_exposures.vars.failed_summary }}). Protected
          Superset exposures: {{
          outputs.parse_manifest_exposures.vars.exposure_names }}. {{
          inputs.enable_ai_triage ? ('AI Triage (' ~
          (fromJson(outputs.diagnose_exposure_breach.textOutput ??
          '{}').severity ?? 'HIGH') ~ '): ' ~
          (fromJson(outputs.diagnose_exposure_breach.textOutput ??
          '{}').root_cause_analysis ?? '') ~ ' | Remediation: ' ~
          (fromJson(outputs.diagnose_exposure_breach.textOutput ??
          '{}').remediation_steps ?? '')) : 'Enable enable_ai_triage for
          automated Gemini root-cause remediation.' }} (execution {{
          execution.id }}).

      - id: halt_on_quarantine
        type: io.kestra.plugin.core.execution.Fail
        description: Mark the execution as FAILED after quarantine logging and alerting
          so downstream dependencies do not treat a quarantined run as clean.
        errorMessage: "BI Exposure Gate blocked Apache Superset refresh for {{
          inputs.mart_table }}: {{
          outputs.audit_mart_volume_and_reconciliation.row.contract_verdict }}
          (failed dbt nodes: {{
          outputs.parse_manifest_exposures.vars.failed_count }})."

errors:
  - id: log_pipeline_error
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    description: Always log the failure context even when Slack alerting is disabled.
    message: "dbt + Apache Superset Exposure Gate aborted in {{ flow.namespace }}.{{
      flow.id }} (execution {{ execution.id }}). Check task logs for container,
      SQL, or gate failures."

  - id: alert_pipeline_crash
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Notify the analytics operations channel when notify_slack is true
      and an unhandled error or gate failure occurs.
    runIf: "{{ inputs.notify_slack }}"
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    retry:
      type: exponential
      interval: PT2S
      maxInterval: PT10S
      maxAttempts: 3
    messageText: "dbt + Apache Superset Exposure Gate failed in {{ flow.namespace
      }}.{{ flow.id }} (execution {{ execution.id }}, channel {{
      inputs.slack_channel }}). Inspect task logs and quarantine telemetry
      before refreshing BI caches."

outputs:
  - id: contract_verdict
    type: STRING
    value: "{{ outputs.publish_execution_metrics.values.contract_verdict }}"
  - id: contract_passed
    type: BOOL
    value: "{{ outputs.publish_execution_metrics.values.contract_passed }}"
  - id: mart_row_count
    type: INT
    value: "{{ outputs.publish_execution_metrics.values.mart_row_count }}"
  - id: mart_metric_sum
    type: FLOAT
    value: "{{ outputs.publish_execution_metrics.values.mart_metric_sum }}"
  - id: exposed_dashboards_count
    type: INT
    value: "{{ outputs.publish_execution_metrics.values.exposed_dashboards_count }}"