Gate a classification model on subgroup error rates before it ships

id: model-subgroup-fairness-gate
namespace: company.team

variables:
  # SQL over the source columns. They live in the flow, not in the inputs, so only people who can edit
  # the flow can change them; whoever executes it only picks the data and the attributes.
  actual_sql: two_year_recid = 1
  predicted_sql: decile_score >= 5
  row_filter_sql: days_b_screening_arrest BETWEEN -30 AND 30 AND is_recid <> -1
    AND c_charge_degree <> 'O' AND score_text <> 'N/A'
  decision_key: "model_{{ inputs.model_name | replace({'-': '_'}) }}_fairness"

inputs:
  - id: model_name
    type: STRING
    displayName: Model name
    description: Lower case letters, digits and dashes. Names the report and the
      decision key.
    defaults: compas-recidivism
    validator: ^[a-z0-9][a-z0-9-]{0,62}$

  - id: model_ref
    type: STRING
    displayName: Model reference
    description: Opaque reference to the exact model being audited (an MLflow model
      URI, an image digest, a git SHA). Recorded with the decision and returned
      in the outputs, so the next step deploys exactly what was audited.
    required: false

  - id: predictions_file
    type: FILE
    displayName: Scored data (file from an upstream task)
    description: Use this when the flow runs as a subflow and an upstream task
      produced the scored data, for example by passing that task's `uri` output.
      When set, `predictions_url` is ignored.
    required: false

  - id: predictions_url
    type: STRING
    displayName: Scored data (HTTPS URL)
    validator: ^https://\S+$
    description: >
      HTTPS URL of the scored data when no file is given. The worker fetches it,
      so whoever can execute the flow can make the worker call any HTTPS address
      it can reach. The default is ProPublica's COMPAS data: the risk scores a
      commercial model gave about 7,000 defendants in Broward County, Florida,
      and whether each one was charged again within two years.
    defaults: https://raw.githubusercontent.com/propublica/compas-analysis/master/compas-scores-two-years.csv

  - id: data_format
    type: SELECT
    displayName: Data format
    values: [ CSV, PARQUET, JSON ]
    defaults: CSV

  - id: attributes
    type: ARRAY
    itemType: STRING
    displayName: Attributes to audit
    description: Columns that define the groups, each audited separately. Join
      columns with `+` (for example `race+sex`) to audit their intersections.
    defaults:
      - race
      - sex
      - age_cat
      - race+sex

  - id: fairness_metric
    type: SELECT
    displayName: Fairness criterion
    description: >
      EQUALIZED_ODDS compares true and false positive rates, EQUAL_OPPORTUNITY
      only true positive rates, PREDICTIVE_PARITY the precision of positive
      predictions. Pick the one your policy or regulator uses.
    values: [ EQUALIZED_ODDS, EQUAL_OPPORTUNITY, PREDICTIVE_PARITY ]
    defaults: EQUALIZED_ODDS

  - id: max_rate_gap
    type: FLOAT
    displayName: Max gap between groups
    description: Block when a compared rate differs between two groups by more than
      this (0.10 is 10 points) and the difference is significant.
    defaults: 0.10

  - id: min_impact_ratio
    type: FLOAT
    displayName: Min impact ratio
    description: Warn when a group's selection rate is below this fraction of the
      highest group's. 0.80 is the "four-fifths" rule; NYC Local Law 144 audits
      report the same ratio.
    defaults: 0.80

  - id: human_review
    type: BOOL
    displayName: Ask a reviewer on WARN or FAIL
    description: Pause and let a reviewer approve an exception (with a written
      justification) or reject. When false, the statistical decision is final.
    defaults: true

  - id: review_timeout
    type: STRING
    displayName: Review timeout (ISO 8601)
    description: Without an answer in this time, the review fails and the model
      stays unapproved. For example P7D or PT12H.
    defaults: P7D
    validator: ^P(\d+D)?(T(\d+H)?(\d+M)?(\d+S)?)?$

  - id: fail_on_block
    type: BOOL
    displayName: Fail the execution when not approved
    description: Set to false when a parent flow calls this one and branches on the
      `approved` output itself.
    defaults: true

  - id: fairlearn_cross_check
    type: BOOL
    displayName: Cross-check with Fairlearn
    description: Recompute every group's rates from the rows with Fairlearn and stop
      if they differ from the audit. Needs PyPI access from the worker.
    defaults: false

  - id: alert_webhook_url
    type: STRING
    displayName: Alert webhook (Slack compatible)
    description: Receives a short message whenever the model is not approved.
    required: false
    validator: ^https?://\S+$

  - id: mlflow_tracking_uri
    type: STRING
    displayName: MLflow tracking URI
    description: Without a trailing slash. When set with the model name and version
      below, the decision is written as the `fairness_decision` tag of that
      MLflow model version.
    required: false
    validator: ^https?://\S+$

  - id: mlflow_model_name
    type: STRING
    displayName: MLflow registered model name
    required: false

  - id: mlflow_model_version
    type: STRING
    displayName: MLflow model version
    required: false

  - id: keep_source_data
    type: BOOL
    displayName: Keep the source data
    description: By default the scored data and the DuckDB file are deleted from
      internal storage when the execution ends, whatever the outcome. Set to
      true to keep them with the execution.
    defaults: false

tasks:
  - id: mark_pending
    type: io.kestra.plugin.core.kv.Set
    description: Replace any earlier decision first, so an audit that fails half way
      or is killed never leaves an old approval behind.
    key: "{{ render(vars.decision_key) }}"
    kvType: JSON
    value: |
      {"decision": "PENDING", "model_ref": {{ inputs.model_ref | toJson }}, "execution_id": "{{ execution.id }}", "flow_revision": {{ flow.revision }}, "started_at": {{ now() | toJson }}}

  - id: fetch
    type: io.kestra.plugin.core.flow.If
    description: Download the data only when no upstream file was passed in.
    condition: "{{ inputs.predictions_file is null }}"
    then:
      - id: download
        type: io.kestra.plugin.core.http.Download
        uri: "{{ inputs.predictions_url }}"

  - id: load_predictions
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: >
      Keep only what the audit needs: the attribute columns as text, the true
      outcome and the prediction. Other columns of the source, such as names,
      are not copied.
    inputFiles:
      source.data: "{{ inputs.predictions_file ?? outputs.download.uri }}"
    outputDbFile: true
    fetchType: NONE
    sql: |
      CREATE TABLE predictions AS
      SELECT COLUMNS(c -> list_contains([{% for attribute in inputs.attributes %}{% for column in attribute | split('\+') %}'{{ column | trim | replace({"'": "''"}) }}', {% endfor %}{% endfor %}], c))::VARCHAR,
             ({{ vars.actual_sql }}) AS actual,
             ({{ vars.predicted_sql }}) AS predicted
      FROM {% if inputs.data_format == 'PARQUET' %}read_parquet{% elseif inputs.data_format == 'JSON' %}read_json_auto{% else %}read_csv{% endif %}('{{ workingDir }}/source.data')
      WHERE {{ vars.row_filter_sql }};

  - id: per_attribute
    type: io.kestra.plugin.core.flow.Loop
    description: Count true and false positives and negatives per group, one query
      per attribute or intersection.
    values: "{{ inputs.attributes }}"
    outputs:
      - id: counts
        type: STRING
        value: "{{ {'attribute': item.value, 'groups': outputs.group_counts.rows} |
          toJson }}"
    tasks:
      - id: group_counts
        type: io.kestra.plugin.jdbc.duckdb.Query
        description: Column names are quoted as identifiers, so DuckDB rejects any name
          that is not a column of the data.
        databaseUri: "{{ outputs.load_predictions.databaseUri }}"
        fetchType: FETCH
        sql: |
          SELECT concat_ws(' / ', {% for column in item.value | split('\+') %}coalesce("{{ column | trim | replace({'"': '""'}) }}", '(missing)'){% if not loop.last %}, {% endif %}{% endfor %}) AS grp,
                 count(*) FILTER (WHERE actual AND predicted) AS tp,
                 count(*) FILTER (WHERE actual AND NOT predicted) AS fn,
                 count(*) FILTER (WHERE NOT actual AND predicted) AS fp,
                 count(*) FILTER (WHERE NOT actual AND NOT predicted) AS tn
          FROM predictions
          GROUP BY 1
          ORDER BY 1

  - id: audit
    type: io.kestra.plugin.scripts.python.Script
    description: >
      Compare every pair of groups on the chosen rates with Fisher's exact test,
      correct the p-values for the number of tests (Holm), judge each attribute
      PASS, WARN, FAIL or INCONCLUSIVE, compare with the previous audits of this
      model and write the report. Python standard library only.
    containerImage: python:3.12-slim
    env:
      COUNTS: "{{ loopOutputs(outputs.per_attribute.outputs, 'counts') | toJson }}"
      HISTORY: "{{ kv(render(vars.decision_key) ~ '_history', errorOnMissing=false) |
        toJson }}"
      METRIC: "{{ inputs.fairness_metric }}"
      MAX_GAP: "{{ inputs.max_rate_gap }}"
      MIN_RATIO: "{{ inputs.min_impact_ratio }}"
      MODEL: "{{ inputs.model_name }}"
      EXECUTION_ID: "{{ execution.id }}"
    outputFiles:
      - fairness_report.md
    script: |
      import itertools, json, math, os
      from datetime import datetime, timezone

      ALPHA = 0.05          # family-wise error rate over every test of the run (Holm)
      MIN_DENOMINATOR = 50  # a rate is only judged when at least this many rows are behind it
      HISTORY_LENGTH = 52   # audits kept per model for the trend
      max_gap, min_ratio = float(os.environ["MAX_GAP"]), float(os.environ["MIN_RATIO"])
      # rate -> (numerator, denominator columns, what the denominator counts)
      ALL_RATES = {"tpr": ("tp", ("tp", "fn"), "actual positives"),
                   "fpr": ("fp", ("fp", "tn"), "actual negatives"),
                   "ppv": ("tp", ("tp", "fp"), "positive predictions")}
      RATES = {r: ALL_RATES[r] for r in {"EQUALIZED_ODDS": ("tpr", "fpr"), "EQUAL_OPPORTUNITY": ("tpr",),
                                         "PREDICTIVE_PARITY": ("ppv",)}[os.environ["METRIC"]]}

      counts = [json.loads(c) if isinstance(c, str) else c for c in json.loads(os.environ.get("COUNTS") or "[]")]
      counts.sort(key=lambda c: c["attribute"])   # Loop iterations can finish in any order
      history = json.loads(os.environ.get("HISTORY") or "null") or []

      def log_comb(n, k):
          return math.lgamma(n + 1) - math.lgamma(k + 1) - math.lgamma(n - k + 1)

      def fisher_p(x1, n1, x2, n2):
          # two-sided Fisher exact test that x1/n1 and x2/n2 are equal; exact, so fine for sparse counts
          k, n = x1 + x2, n1 + n2
          logp = {i: log_comb(n1, i) + log_comb(n2, k - i) - log_comb(n, k) for i in range(max(0, k - n2), min(k, n1) + 1)}
          return min(1.0, sum(math.exp(v) for v in logp.values() if v <= logp[x1] + 1e-7))

      def pct(x):
          return "n/a" if x is None else "%.1f%%" % (100 * x)

      audits, tests = {}, []
      for entry in counts:
          attribute, stats, reasons, unjudged = entry["attribute"], {}, [], {}
          for g in entry["groups"]:
              c = {k: int(g[k]) for k in ("tp", "fn", "fp", "tn")}
              n = sum(c.values())
              s = dict(c, n=n, base_rate=(c["tp"] + c["fn"]) / n if n else None,
                       selection_rate=(c["tp"] + c["fp"]) / n if n else None)
              for rate, (num, den, _) in ALL_RATES.items():
                  d = sum(c[x] for x in den)
                  s[rate], s[rate + "_denominator"] = (c[num] / d if d else None), d
              stats[str(g["grp"])] = s
          for rate, (num, den, what) in RATES.items():
              judged = sorted(g for g, s in stats.items() if s[rate + "_denominator"] >= MIN_DENOMINATOR)
              # a group with too few rows behind a rate is left out of it, never counted as 0
              unjudged[rate] = sorted(g for g in stats if g not in judged)
              if len(judged) < 2:
                  reasons.append("%s: fewer than 2 groups with %d or more %s" % (rate.upper(), MIN_DENOMINATOR, what))
              for a, b in itertools.combinations(judged, 2):
                  sa, sb = stats[a], stats[b]
                  tests.append({"attribute": attribute, "rate": rate, "groups": [a, b],
                                "gap": abs(sa[rate] - sb[rate]),
                                "p_value": fisher_p(sa[num], sa[rate + "_denominator"], sb[num], sb[rate + "_denominator"])})
          judged = [g for g, s in stats.items() if s["n"] >= MIN_DENOMINATOR]
          selection = None
          if len(judged) >= 2:
              hi = max(judged, key=lambda g: stats[g]["selection_rate"])
              lo = min(judged, key=lambda g: stats[g]["selection_rate"])
              top = stats[hi]["selection_rate"]
              for g in judged:
                  # nobody predicted positive in any group is equal treatment, not a division by zero
                  stats[g]["impact_ratio"] = stats[g]["selection_rate"] / top if top > 0 else 1.0
              selection = {"ratio": round(stats[lo]["impact_ratio"], 3), "lowest": lo, "highest": hi}
          audits[attribute] = {"groups": stats, "reasons": reasons, "selection": selection, "unjudged": unjudged}

      # Holm step-down: the i-th smallest of m p-values is multiplied by (m - i), kept monotonic
      running = 0.0
      for i, t in enumerate(sorted(tests, key=lambda t: t["p_value"])):
          running = max(running, min(1.0, (len(tests) - i) * t["p_value"]))
          t["p_adjusted"] = running

      previous = history[-1] if history else None
      results, lines = {}, []
      for attribute, a in audits.items():
          own = [t for t in tests if t["attribute"] == attribute]
          flagged = [t for t in own if t["gap"] > max_gap and t["p_adjusted"] < ALPHA]
          if flagged:
              verdict = "FAIL"
          elif a["reasons"]:
              verdict = "INCONCLUSIVE"
          elif a["selection"]["ratio"] < min_ratio:
              verdict = "WARN"
          else:
              verdict = "PASS"
          largest = {}
          for rate in RATES:
              of_rate = [t for t in own if t["rate"] == rate]
              largest[rate] = max(of_rate, key=lambda t: t["gap"]) if of_rate else None
          results[attribute] = {
              "verdict": verdict, "reasons": a["reasons"], "selection": a["selection"],
              "largest_gaps": largest, "flagged_pairs": flagged, "unjudged_groups": a["unjudged"], "groups": a["groups"],
          }

          lines += ["## " + attribute + ": " + verdict, "",
                    "| group | rows | actually positive | selection rate | impact ratio | " + " | ".join(r.upper() + " (rows behind it)" for r in RATES) + " |",
                    "|---|---|---|---|---|" + "---|" * len(RATES)]
          for g, s in sorted(a["groups"].items()):
              cells = ["%s (%d%s)" % (pct(s[r]), s[r + "_denominator"], ", not judged" if g in a["unjudged"][r] else "") for r in RATES]
              ratio = "%.2f" % s["impact_ratio"] if "impact_ratio" in s else "not judged"
              lines.append("| %s | %d | %s | %s | %s | %s |" % (g, s["n"], pct(s["base_rate"]), pct(s["selection_rate"]), ratio, " | ".join(cells)))
          lines.append("")
          for rate, t in largest.items():
              if t:
                  lines.append("- Largest %s gap: %s vs %s, %.1f points, p = %.2g, Holm-adjusted p = %.2g." % (
                      rate.upper(), t["groups"][0], t["groups"][1], 100 * t["gap"], t["p_value"], t["p_adjusted"]))
          for t in flagged:
              lines.append("- Blocking: %s %s vs %s differ by %.1f points (adjusted p = %.2g)." % (
                  t["rate"].upper(), t["groups"][0], t["groups"][1], 100 * t["gap"], t["p_adjusted"]))
          if a["selection"]:
              lines.append("- Lowest impact ratio %.2f (%s vs %s)." % (a["selection"]["ratio"], a["selection"]["lowest"], a["selection"]["highest"]))
          for rate in RATES:
              if a["unjudged"][rate]:
                  lines.append("- Left out of %s (fewer than %d %s): %s." % (rate.upper(), MIN_DENOMINATOR, RATES[rate][2], ", ".join(a["unjudged"][rate])))
          lines += ["- Not judged: " + r for r in a["reasons"]]
          if previous and attribute in previous.get("attributes", {}):
              before = previous["attributes"][attribute]
              lines.append("- Previous audit (%s): %s, largest gap %s points." % (
                  previous["audited_at"][:10], before["verdict"], before["largest_gap_points"]))
          lines.append("")

      order = ["PASS", "WARN", "INCONCLUSIVE", "FAIL"]
      overall = max((r["verdict"] for r in results.values()), key=order.index) if results else "INCONCLUSIVE"
      decision = {"PASS": "APPROVED", "WARN": "NEEDS_REVIEW", "INCONCLUSIVE": "INCONCLUSIVE", "FAIL": "BLOCKED"}[overall]
      method = [
          "## Method", "",
          "- Criterion: %s. TPR is the share of actual positives predicted positive, FPR the share of actual negatives predicted positive, PPV the share of positive predictions that are right." % os.environ["METRIC"],
          "- Every pair of groups is compared on each rate with a two-sided Fisher exact test. The %d tests of this run (all attributes and rates together) are corrected with Holm's method at a family-wise level of %.2f." % (len(tests), ALPHA),
          "- A gap blocks the model only when it is larger than %.0f points and its adjusted p-value is below %.2f. A group is compared on a rate only with at least %d rows behind that rate; smaller groups are shown but left out, never counted as 0. An attribute with fewer than two comparable groups is INCONCLUSIVE." % (100 * max_gap, ALPHA, MIN_DENOMINATOR),
          "- The impact ratio (a group's selection rate over the highest one) only warns, below %.2f: groups can have different base rates." % min_ratio,
          "- Assumes rows are independent cases and the recorded outcome is correct. Measured outcomes can be biased themselves (for example re-arrests rather than offences).",
          "",
      ]
      if not results:
          lines = ["No attribute was audited.", ""]
      report = "# Subgroup fairness report: %s (%s, %s)\n\n" % (os.environ["MODEL"], overall, decision) + "\n".join(lines + method)
      with open("fairness_report.md", "w") as f:
          f.write(report)
      print(report)

      now = datetime.now(timezone.utc).isoformat()
      history = (history + [{
          "execution_id": os.environ["EXECUTION_ID"], "audited_at": now, "decision": decision,
          "attributes": {a: {"verdict": r["verdict"],
                             "largest_gap_points": round(100 * max((t["gap"] for t in r["largest_gaps"].values() if t), default=0), 1)}
                         for a, r in results.items()},
      }])[-HISTORY_LENGTH:]
      summary = "%s: %s (%s). %s" % (os.environ["MODEL"], decision, overall, "; ".join(
          "%s %s" % (a, r["verdict"]) for a, r in results.items()) or "no attribute audited")
      print("::" + json.dumps({"outputs": {
          "verdict": overall, "decision": decision, "results": results, "summary": summary,
          "report_markdown": report, "history": history,
          "failed_attributes": ", ".join(a for a, r in results.items() if r["verdict"] == "FAIL"),
          "inconclusive_attributes": ", ".join(a for a, r in results.items() if r["verdict"] == "INCONCLUSIVE"),
      }}) + "::")

  - id: cross_check
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.fairlearn_cross_check }}"
    then:
      - id: fairlearn
        type: io.kestra.plugin.scripts.python.Script
        description: Recompute every group's rates from the rows with Fairlearn's
          MetricFrame, independently of the SQL counts.
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install --quiet --disable-pip-version-check fairlearn==0.13.0
            duckdb pandas
        inputFiles:
          predictions.duckdb: "{{ outputs.load_predictions.databaseUri }}"
        env:
          RESULTS: "{{ outputs.audit.vars.results | toJson }}"
        script: |
          import json, os
          import duckdb
          from fairlearn.metrics import MetricFrame, false_positive_rate, selection_rate, true_positive_rate
          from sklearn.metrics import precision_score

          print("duckdb", duckdb.__version__)
          with duckdb.connect("predictions.duckdb", read_only=True) as con:
              rows = con.sql("SELECT * FROM predictions").df()
          y_true, y_pred = rows["actual"].astype(int), rows["predicted"].astype(int)
          metrics = {"tpr": true_positive_rate, "fpr": false_positive_rate, "selection_rate": selection_rate,
                     "ppv": lambda t, p: precision_score(t, p, zero_division=0)}
          worst = 0.0
          for attribute, result in json.loads(os.environ["RESULTS"]).items():
              columns = [c.strip() for c in attribute.split("+")]
              groups = rows[columns].fillna("(missing)").astype(str).agg(" / ".join, axis=1)
              frame = MetricFrame(metrics=metrics, y_true=y_true, y_pred=y_pred, sensitive_features=groups).by_group
              for group, ours in result["groups"].items():
                  for metric in metrics:
                      if ours.get(metric) is not None:
                          worst = max(worst, abs(float(frame.loc[group, metric]) - ours[metric]))
          print("largest difference with Fairlearn: %.3g" % worst)
          print("::" + json.dumps({"outputs": {"max_difference": worst}}) + "::")

      - id: stop_on_mismatch
        type: io.kestra.plugin.core.execution.Fail
        description: The audit and Fairlearn disagree, so the audit cannot be trusted;
          the decision stays PENDING.
        condition: "{{ outputs.fairlearn.vars.max_difference > 0.000001 }}"
        errorMessage: "The audit and Fairlearn disagree by {{
          outputs.fairlearn.vars.max_difference }}. The decision stays PENDING."

  - id: publish_report
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "fairness/{{ inputs.model_name }}/fairness_report.md": "{{ outputs.audit.outputFiles['fairness_report.md'] }}"

  - id: record_history
    type: io.kestra.plugin.core.kv.Set
    key: "{{ render(vars.decision_key) }}_history"
    kvType: JSON
    value: "{{ outputs.audit.vars.history | toJson }}"

  - id: review
    type: io.kestra.plugin.core.flow.If
    description: A person can approve an exception or reject, and their name and
      reason are recorded with the decision.
    condition: "{{ inputs.human_review and (outputs.audit.vars.decision ==
      'NEEDS_REVIEW' or outputs.audit.vars.decision == 'BLOCKED') }}"
    then:
      - id: record_awaiting_review
        type: io.kestra.plugin.core.kv.Set
        key: "{{ render(vars.decision_key) }}"
        kvType: JSON
        value: |
          {"decision": "AWAITING_REVIEW", "audit_decision": {{ outputs.audit.vars.decision | toJson }}, "model_ref": {{ inputs.model_ref | toJson }}, "execution_id": "{{ execution.id }}", "flow_revision": {{ flow.revision }}}

      - id: await_review
        type: io.kestra.plugin.core.flow.Pause
        pauseDuration: "{{ inputs.review_timeout }}"
        behavior: FAIL
        onResume:
          - id: approve_exception
            type: BOOL
            displayName: Approve an exception
            defaults: false
          - id: reviewer
            type: STRING
            displayName: Your name
            validator: ^\S.{1,99}$
          - id: justification
            type: STRING
            displayName: Justification
            description: Why the model may ship despite the result, or why it is rejected
              (at least 20 characters).
            validator: ^[\s\S]{20,2000}$

  - id: finalize
    type: io.kestra.plugin.core.output.OutputValues
    values:
      decision: "{% if outputs.await_review is defined %}{{
        outputs.await_review.onResume.approve_exception ?
        'APPROVED_WITH_EXCEPTION' : 'REJECTED' }}{% else %}{{
        outputs.audit.vars.decision }}{% endif %}"
      reviewer: "{% if outputs.await_review is defined %}{{
        outputs.await_review.onResume.reviewer }}{% endif %}"
      justification: "{% if outputs.await_review is defined %}{{
        outputs.await_review.onResume.justification }}{% endif %}"

  - id: record_decision
    type: io.kestra.plugin.core.kv.Set
    description: Written on every outcome, so this key always holds the latest audit
      of the model.
    key: "{{ render(vars.decision_key) }}"
    kvType: JSON
    value: |
      {
        "decision": {{ outputs.finalize.values.decision | toJson }},
        "audit_decision": {{ outputs.audit.vars.decision | toJson }},
        "verdict": {{ outputs.audit.vars.verdict | toJson }},
        "model_ref": {{ inputs.model_ref | toJson }},
        "reviewer": {{ (outputs.finalize.values.reviewer ?? null) | toJson }},
        "justification": {{ (outputs.finalize.values.justification ?? null) | toJson }},
        "execution_id": "{{ execution.id }}",
        "flow_revision": {{ flow.revision }},
        "audited_at": {{ now() | toJson }},
        "data": {{ (inputs.predictions_file is null ? inputs.predictions_url : 'upstream file') | toJson }},
        "attributes": {{ inputs.attributes | toJson }},
        "fairness_metric": {{ inputs.fairness_metric | toJson }},
        "max_rate_gap": {{ inputs.max_rate_gap }},
        "min_impact_ratio": {{ inputs.min_impact_ratio }},
        "report": "fairness/{{ inputs.model_name }}/fairness_report.md"
      }

  - id: tag_mlflow
    type: io.kestra.plugin.core.flow.If
    description: Write the decision where a model registry deploy reads it.
    condition: "{{ inputs.mlflow_tracking_uri is not null and
      inputs.mlflow_model_name is not null and inputs.mlflow_model_version is
      not null }}"
    then:
      - id: mlflow_decision_tag
        type: io.kestra.plugin.core.http.Request
        method: POST
        uri: "{{ inputs.mlflow_tracking_uri }}/api/2.0/mlflow/model-versions/set-tag"
        contentType: application/json
        body: "{{ {'name': inputs.mlflow_model_name, 'version':
          inputs.mlflow_model_version, 'key': 'fairness_decision', 'value':
          outputs.finalize.values.decision} | toJson }}"
      - id: mlflow_execution_tag
        type: io.kestra.plugin.core.http.Request
        method: POST
        uri: "{{ inputs.mlflow_tracking_uri }}/api/2.0/mlflow/model-versions/set-tag"
        contentType: application/json
        body: "{{ {'name': inputs.mlflow_model_name, 'version':
          inputs.mlflow_model_version, 'key': 'fairness_audit_execution',
          'value': execution.id} | toJson }}"

  - id: alert
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.alert_webhook_url is not null and
      outputs.finalize.values.decision != 'APPROVED' }}"
    then:
      - id: send_alert
        type: io.kestra.plugin.core.http.Request
        method: POST
        uri: "{{ inputs.alert_webhook_url }}"
        contentType: application/json
        body: "{{ {'text': 'Fairness gate ' ~ outputs.finalize.values.decision ~ ' for '
          ~ inputs.model_name ~ ': ' ~ outputs.audit.vars.summary ~ ' Report:
          fairness/' ~ inputs.model_name ~ '/fairness_report.md, execution ' ~
          execution.id} | toJson }}"

  - id: decision
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ outputs.finalize.values.decision }}"
    cases:
      APPROVED:
        - id: approved
          type: io.kestra.plugin.core.log.Log
          message: "{{ outputs.audit.vars.summary }}"
      APPROVED_WITH_EXCEPTION:
        - id: approved_with_exception
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: "Approved by {{ outputs.finalize.values.reviewer }} despite {{
            outputs.audit.vars.decision }}: {{
            outputs.finalize.values.justification }}"
      NEEDS_REVIEW:
        - id: needs_review
          type: io.kestra.plugin.core.log.Log
          level: WARN
          message: "Error rates are within {{ inputs.max_rate_gap }} for every group, but
            some impact ratios are below {{ inputs.min_impact_ratio }}. Check
            the 'actually positive' column of the report before shipping."
    defaults:
      - id: not_approved
        type: io.kestra.plugin.core.execution.Fail
        description: BLOCKED, REJECTED or INCONCLUSIVE. Skipped when a parent flow
          handles the outcome (`fail_on_block` false).
        condition: "{{ inputs.fail_on_block }}"
        errorMessage: "{{ outputs.finalize.values.decision }}: {{
          outputs.audit.vars.summary }} See fairness/{{ inputs.model_name
          }}/fairness_report.md."

afterExecution:
  - id: retention
    type: io.kestra.plugin.core.flow.If
    description: Runs once the execution has ended, whatever the outcome, including
      a failed gate or a killed run.
    condition: "{{ not inputs.keep_source_data }}"
    then:
      - id: purge_source_data
        type: io.kestra.plugin.core.storage.PurgeCurrentExecutionFiles
        description: Delete the scored data and the DuckDB file from internal storage.
          The report stays in namespace files and in the outputs.

outputs:
  # Every value falls back to a safe default, so a run that stops early still reports "not approved".
  - id: decision
    type: STRING
    description: APPROVED, APPROVED_WITH_EXCEPTION, NEEDS_REVIEW, REJECTED, BLOCKED,
      INCONCLUSIVE, or PENDING when the audit did not finish.
    value: "{{ outputs.finalize.values.decision ?? 'PENDING' }}"
  - id: approved
    type: BOOL
    description: True only for APPROVED and APPROVED_WITH_EXCEPTION. Deploy on this.
    value: "{{ (outputs.finalize.values.decision ?? 'PENDING') == 'APPROVED' or
      (outputs.finalize.values.decision ?? 'PENDING') ==
      'APPROVED_WITH_EXCEPTION' }}"
  - id: verdict
    type: STRING
    value: "{{ outputs.audit.vars.verdict ?? 'NOT_RUN' }}"
  - id: model_ref
    type: STRING
    description: The model reference that was audited, passed through for the next step.
    value: "{{ inputs.model_ref ?? '' }}"
  - id: summary
    type: STRING
    value: "{{ outputs.audit.vars.summary ?? 'The audit did not finish.' }}"
  - id: report_markdown
    type: STRING
    value: "{{ outputs.audit.vars.report_markdown ?? '' }}"
  - id: results
    type: JSON
    value: "{{ (outputs.audit.vars.results ?? {}) | toJson }}"

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Re-audit every Monday once the inputs and variables point at your
      own scored data.
    cron: "0 9 * * 1"
    disabled: true