Gate LLM Judge Drift Before It Silently Changes Verdicts

id: llm-judge-consistency-gate
namespace: company.team
description: |
  Catch LLM-as-judge pipelines that quietly drift when the judge model is
  updated, before their verdicts silently change on the same gold set. An
  evaluation runner POSTs paired gold verdicts and current judge verdicts to
  a keyed webhook, Python computes the agreement rate and flags every item
  that flipped since the expected label, a Loop walks the flipped items and
  alerts on each regression, and a final If escalates and fails the run when
  agreement falls under the configured floor. Consistent judges are logged,
  and the agreement rate, flipped items and per run counters are published as
  outputs.

triggers:
  - id: on_judgments_publish
    type: io.kestra.plugin.core.trigger.Webhook
    description: The evaluation runner POSTs the gold set items with the current
      judge verdicts here so consistency is scored as soon as a judge run
      finishes.
    key: "{{ secret('LLM_JUDGE_WEBHOOK_KEY') }}"
  - id: weekly_judge_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly net on Mondays at 08:00 UTC that re-scores the last posted
      gold set run, shipped disabled until the webhook path is trusted in your
      environment.
    cron: "0 8 * * 1"
    disabled: true

inputs:
  - id: min_agreement_pct
    type: FLOAT
    defaults: 95.0
    description: Floor agreement percentage the current judge run must meet against
      the gold verdicts before the gate escalates and fails the run.
  - id: judge_model
    type: STRING
    defaults: "atlas-judge-3"
    description: Judge model identifier echoed in every alert so the on call knows
      which revision produced the verdicts under scrutiny.
  - id: gold_set_name
    type: STRING
    defaults: "support-intents-v2"
    description: Name of the gold evaluation set being scored, echoed in alerts to
      identify the contract under test.

tasks:
  # --- 1. consistency score ---
  - id: score_judgments
    type: io.kestra.plugin.scripts.python.Script
    description: Read the posted gold set judgments or the bundled example run,
      compute the agreement rate between expected and current judge verdicts,
      flag every flipped item with its flip direction, and emit the consistency
      verdict through the outputs protocol.
    containerImage: python:3.12-slim
    env:
      JUDGE_RUN_JSON: "{{ trigger.body ?? '' }}"
      MIN_AGREEMENT_PCT: "{{ inputs.min_agreement_pct }}"
      JUDGE_MODEL: "{{ inputs.judge_model }}"
      GOLD_SET_NAME: "{{ inputs.gold_set_name }}"
    script: |
      import json
      import os

      DEFAULT_RUN = """
      {
        "gold_set": "support-intents-v2",
        "judge_model": "atlas-judge-3",
        "items": [
          {"item_id": "g001", "expected": "pass", "judged": "pass"},
          {"item_id": "g002", "expected": "pass", "judged": "fail"},
          {"item_id": "g003", "expected": "fail", "judged": "fail"},
          {"item_id": "g004", "expected": "pass", "judged": "pass"},
          {"item_id": "g005", "expected": "fail", "judged": "fail"},
          {"item_id": "g006", "expected": "pass", "judged": "pass"},
          {"item_id": "g007", "expected": "pass", "judged": "pass"},
          {"item_id": "g008", "expected": "fail", "judged": "pass"},
          {"item_id": "g009", "expected": "pass", "judged": "pass"},
          {"item_id": "g010", "expected": "pass", "judged": "fail"},
          {"item_id": "g011", "expected": "fail", "judged": "fail"},
          {"item_id": "g012", "expected": "pass", "judged": "pass"}
        ]
      }
      """


      def as_float(value, fallback=0.0):
          try:
              return float(value)
          except (TypeError, ValueError):
              return fallback


      raw = (os.environ.get("JUDGE_RUN_JSON") or "").strip() or DEFAULT_RUN.strip()
      min_pct = as_float(os.environ.get("MIN_AGREEMENT_PCT"), 95.0)
      judge_model = (os.environ.get("JUDGE_MODEL") or "").strip() or "atlas-judge-3"
      gold_set = (os.environ.get("GOLD_SET_NAME") or "").strip() or "support-intents-v2"

      outputs = {
          "gate_ok": False,
          "gold_set": gold_set,
          "judge_model": judge_model,
          "scored_items": 0,
          "agreement_rate": 0.0,
          "flip_count": 0,
          "flipped_items": [],
          "false_pass_count": 0,
          "false_reject_count": 0,
          "sample_sufficient": False,
          "min_agreement_pct": min_pct,
      }

      try:
          payload = json.loads(raw)
      except ValueError:
          payload = None

      if isinstance(payload, dict):
          items = payload.get("items")
          if not isinstance(items, list):
              items = []
          flips = []
          scored = 0
          false_passes = 0
          false_rejects = 0
          for item in items:
              if not isinstance(item, dict):
                  continue
              expected = str(item.get("expected") or "").strip().lower()
              judged = str(item.get("judged") or "").strip().lower()
              if expected not in ("pass", "fail") or judged not in ("pass", "fail"):
                  continue
              scored += 1
              if judged != expected:
                  direction = "false_pass" if expected == "fail" else "false_reject"
                  if direction == "false_pass":
                      false_passes += 1
                  else:
                      false_rejects += 1
                  flips.append({
                      "item_id": str(item.get("item_id") or "unknown-item"),
                      "expected": expected,
                      "judged": judged,
                      "direction": direction,
                  })
          agreement = (scored - len(flips)) * 100.0 / scored if scored > 0 else 0.0
          outputs.update({
              "gate_ok": True,
              "gold_set": str(payload.get("gold_set") or gold_set),
              "judge_model": str(payload.get("judge_model") or judge_model),
              "scored_items": scored,
              "agreement_rate": round(agreement, 2),
              "flip_count": len(flips),
              "flipped_items": flips,
              "false_pass_count": false_passes,
              "false_reject_count": false_rejects,
              "sample_sufficient": scored > 0,
          })

      print("gate_ok=%s scored=%d agreement=%.2f%% flips=%d false_pass=%d false_reject=%d" % (
          outputs["gate_ok"], outputs["scored_items"], outputs["agreement_rate"],
          outputs["flip_count"], outputs["false_pass_count"], outputs["false_reject_count"]))
      print("::" + json.dumps({"outputs": outputs}) + "::")

  # --- 2. per flip sweep ---
  - id: review_flips
    type: io.kestra.plugin.core.flow.Loop
    description: Walk the scored flip rows one at a time so every regressed gold
      item gets its own alert with its expected and current verdicts instead of
      one blended agreement message.
    concurrencyLimit: 1
    values: "{{ outputs.score_judgments.vars.flipped_items }}"
    tasks:
      - id: flag_flip
        type: io.kestra.plugin.core.flow.If
        description: Branch on the flip direction the scorer already computed, so a
          dangerous false pass is escalated loudly while a false reject is
          recorded as throughput loss.
        condition: "{{ item.value.direction == 'false_pass' }}"
        then:
          - id: alert_false_pass
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the item id, its expected and current verdicts and the judge
              model so the owning team knows the judge approved content the gold
              set marked as fail.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "LLM judge FALSE PASS on gold set {{ outputs.score_judgments.vars.gold_set }} item {{ item.value.item_id }} - gold verdict was {{ item.value.expected }} but judge {{ outputs.score_judgments.vars.judge_model }} returned {{ item.value.judged }}. The judge approved content it should have rejected. Execution {{ execution.id }}."
              }
        else:
          - id: log_false_reject
            type: io.kestra.plugin.core.log.Log
            description: Record the false reject for the run so throughput loss stays
              visible in the execution history without paging anyone.
            message: "LLM judge false reject on item {{ item.value.item_id }} - gold verdict
              was {{ item.value.expected }} but judge returned {{
              item.value.judged }}. This costs throughput but does not approve
              bad content."

  # --- 3. consistency gate ---
  - id: gate_consistency
    type: io.kestra.plugin.core.flow.If
    description: Branch on the consistency verdict so a judge run that falls under
      the agreement floor escalates and fails the run, while a consistent judge
      is logged and passes.
    condition: "{{ outputs.score_judgments.vars.gate_ok and
      (outputs.score_judgments.vars.agreement_rate <
      outputs.score_judgments.vars.min_agreement_pct or not
      outputs.score_judgments.vars.sample_sufficient) }}"
    then:
      - id: alert_drift
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the agreement rate, the floor, the flip count and the false
          pass count so the owning team knows exactly how far the judge has
          drifted before they trust its verdicts again.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "LLM judge drift ESCALATION for gold set {{ outputs.score_judgments.vars.gold_set }} - judge {{ outputs.score_judgments.vars.judge_model }} scored {{ outputs.score_judgments.vars.agreement_rate }} percent agreement against a {{ inputs.min_agreement_pct }} percent floor, {{ outputs.score_judgments.vars.flip_count }} flipped item(s) including {{ outputs.score_judgments.vars.false_pass_count }} false pass(es). Do not trust this judge revision until the drift is explained. Execution {{ execution.id }}."
          }
      - id: fail_gate
        type: io.kestra.plugin.core.execution.Fail
        description: Turn the run red so the evaluation pipeline stops instead of
          shipping verdicts from a drifted judge revision.
        errorMessage: "LLM judge drift detected - agreement {{
          outputs.score_judgments.vars.agreement_rate }} percent is under the {{
          inputs.min_agreement_pct }} percent floor, {{
          outputs.score_judgments.vars.flip_count }} item(s) flipped."
    else:
      - id: log_consistent
        type: io.kestra.plugin.core.log.Log
        description: Record the consistent verdict with the agreement rate so a passing
          judge run still leaves evidence the gate ran.
        message: "LLM judge consistent for gold set {{
          outputs.score_judgments.vars.gold_set }} - judge {{
          outputs.score_judgments.vars.judge_model }} scored {{
          outputs.score_judgments.vars.agreement_rate }} percent agreement over
          {{ outputs.score_judgments.vars.scored_items }} items, at or above the
          {{ inputs.min_agreement_pct }} percent floor."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: gate_ok
    type: BOOL
    description: True when the posted run parsed and scoring completed, false for a
      malformed or unusable payload.
    value: "{{ outputs.score_judgments.vars.gate_ok }}"
  - id: gold_set
    type: STRING
    description: Name of the gold evaluation set that was scored, taken from the
      payload or the configured default.
    value: "{{ outputs.score_judgments.vars.gold_set }}"
  - id: judge_model
    type: STRING
    description: Judge model identifier that produced the verdicts under scrutiny,
      echoed from the payload or the configured default.
    value: "{{ outputs.score_judgments.vars.judge_model }}"
  - id: scored_items
    type: INT
    description: Number of gold set items that carried both a valid expected and
      current judge verdict and were scored.
    value: "{{ outputs.score_judgments.vars.scored_items }}"
  - id: agreement_rate
    type: FLOAT
    description: Percentage of scored items where the current judge verdict matched
      the gold verdict, rounded to two decimals.
    value: "{{ outputs.score_judgments.vars.agreement_rate }}"
  - id: flip_count
    type: INT
    description: Number of scored items where the current judge verdict disagreed
      with the gold verdict.
    value: "{{ outputs.score_judgments.vars.flip_count }}"
  - id: flipped_items
    type: ARRAY
    description: Per flip rows with item id, expected verdict, judged verdict and
      flip direction, ready for gold set triage.
    value: "{{ outputs.score_judgments.vars.flipped_items }}"
  - id: false_pass_count
    type: INT
    description: Flips where the gold verdict was fail but the judge returned pass,
      the dangerous direction that approves bad content.
    value: "{{ outputs.score_judgments.vars.false_pass_count }}"
  - id: false_reject_count
    type: INT
    description: Flips where the gold verdict was pass but the judge returned fail,
      the throughput loss direction that is safe but wasteful.
    value: "{{ outputs.score_judgments.vars.false_reject_count }}"
  - id: sample_sufficient
    type: BOOL
    description: True when at least one valid gold item was scored, false when the
      sample is empty and the gate should not trust the rate.
    value: "{{ outputs.score_judgments.vars.sample_sufficient }}"