Kopia Snapshot Shrink Retention Pin Guard

id: kopia-snapshot-shrink-retention-pin-guard
namespace: company.team

description: |
  Learn what a normal Kopia snapshot looks like for every backup source, detect a snapshot that suddenly
  holds far fewer files or bytes than that baseline (an emptied mount, a wiped volume, a half-failed upload),
  and pin the newest known-good snapshots in the repository before Kopia's automatic retention rotates them
  out. Pinning and releasing pins is gated by a human approval and re-verified against the repository.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: s3_endpoint
    type: STRING
    displayName: S3 endpoint
    description: host:port of the S3-compatible storage holding the Kopia repository
      (MinIO, AWS S3, Ceph, Wasabi). Must be reachable from the Docker task
      runner.
    defaults: "host.docker.internal:9000"

  - id: s3_bucket
    type: STRING
    displayName: Bucket
    defaults: "kopia-backups"

  - id: s3_prefix
    type: STRING
    displayName: Repository prefix
    description: Object key prefix of the repository inside the bucket. Leave empty
      when the repository sits at the bucket root.
    defaults: ""

  - id: s3_disable_tls
    type: BOOL
    displayName: Disable TLS
    description: Only for a local plain-HTTP MinIO. Keep false for AWS S3 and any
      TLS endpoint.
    defaults: false

  - id: max_file_drop_pct
    type: INT
    displayName: Max file-count drop (%)
    description: A snapshot with fewer files than the baseline median minus this
      percentage is anomalous.
    defaults: 30

  - id: max_size_drop_pct
    type: INT
    displayName: Max size drop (%)
    description: A snapshot with fewer total bytes than the baseline median minus
      this percentage is anomalous.
    defaults: 40

  - id: min_baseline_samples
    type: INT
    displayName: Minimum baseline samples
    description: Healthy snapshots needed per source before drops are judged.
      Earlier snapshots are accepted as the bootstrap baseline.
    defaults: 3

  - id: baseline_window
    type: INT
    displayName: Baseline window
    description: Number of most recent healthy snapshots per source that define "normal".
    defaults: 14

  - id: pin_count
    type: INT
    displayName: Snapshots to pin per alarmed source
    description: How many of the newest known-good snapshots are protected from
      retention while a source is anomalous.
    defaults: 2

  - id: release_after_healthy
    type: INT
    displayName: Healthy snapshots before releasing pins
    description: Consecutive healthy snapshots required before the guard proposes unpinning.
    defaults: 3

  - id: sustained_shift_after
    type: INT
    displayName: Sustained shift threshold
    description: Consecutive anomalous snapshots after which the shift is reported
      as possibly legitimate.
    defaults: 7

  - id: rebaseline_sources
    type: ARRAY
    itemType: STRING
    displayName: Accept new normal for sources
    description: Source keys such as root@db1:/srv/data whose current level should
      become the new baseline. Their guard pins are proposed for release.
    defaults: []

variables:
  pin_name: kestra-retention-guard
  state_key: "kopia_guard_state_{{ (inputs.s3_bucket ~ '-' ~ inputs.s3_prefix) |
    slugify }}"
  audit_key: "kopia_guard_audit_{{ (inputs.s3_bucket ~ '-' ~ inputs.s3_prefix) |
    slugify }}"

triggers:
  - id: after_backup_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Call this endpoint from the backup job right after `kopia snapshot
      create` finishes, so a bad snapshot is judged before the next retention
      pass.
    key: "{{ secret('KOPIA_GUARD_WEBHOOK_KEY') }}"

  - id: six_hourly_safety_net
    type: io.kestra.plugin.core.trigger.Schedule
    description: Catch backups that never called the webhook. Shipped disabled until
      the S3 inputs point at your repository.
    cron: "15 */6 * * *"
    disabled: true

tasks:
  - id: list_snapshots
    type: io.kestra.plugin.scripts.shell.Commands
    description: Connect read-only and dump every snapshot manifest of every user,
      host and source as JSON (--all, because by default Kopia only lists the
      current user and host).
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: kopia/kopia:0.23.1
    env:
      KOPIA_PASSWORD: "{{ secret('KOPIA_PASSWORD') }}"
      S3_ACCESS_KEY: "{{ secret('KOPIA_S3_ACCESS_KEY') }}"
      S3_SECRET_KEY: "{{ secret('KOPIA_S3_SECRET_KEY') }}"
      S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
      S3_BUCKET: "{{ inputs.s3_bucket }}"
      S3_PREFIX: "{{ inputs.s3_prefix }}"
      S3_DISABLE_TLS: "{{ inputs.s3_disable_tls }}"
      KOPIA_CONNECT_MODE: "--readonly"
    outputFiles:
      - snapshots.json
    commands:
      - |
        set -eu
        tls_flag=""
        if [ "$S3_DISABLE_TLS" = "true" ]; then tls_flag="--disable-tls"; fi
        kopia repository connect s3 $KOPIA_CONNECT_MODE --bucket="$S3_BUCKET" --endpoint="$S3_ENDPOINT" --prefix="$S3_PREFIX" $tls_flag --access-key="$S3_ACCESS_KEY" --secret-access-key="$S3_SECRET_KEY"
        kopia snapshot list --all --show-identical --json > snapshots.json
        test -s snapshots.json

  - id: analyze_snapshots
    type: io.kestra.plugin.scripts.node.Script
    description: Compare each new snapshot with the per-source baseline stored in
      Kestra KV, classify it, and plan which known-good snapshots to pin or
      release.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: node:20-slim
    inputFiles:
      snapshots.json: "{{ outputs.list_snapshots.outputFiles['snapshots.json'] }}"
      state.json: "{{ kv(key=vars.state_key, errorOnMissing=false) | default('{}') }}"
    env:
      MAX_FILE_DROP_PCT: "{{ inputs.max_file_drop_pct }}"
      MAX_SIZE_DROP_PCT: "{{ inputs.max_size_drop_pct }}"
      MIN_BASELINE_SAMPLES: "{{ inputs.min_baseline_samples }}"
      BASELINE_WINDOW: "{{ inputs.baseline_window }}"
      PIN_COUNT: "{{ inputs.pin_count }}"
      RELEASE_AFTER_HEALTHY: "{{ inputs.release_after_healthy }}"
      SUSTAINED_SHIFT_AFTER: "{{ inputs.sustained_shift_after }}"
      REBASELINE_SOURCES: "{{ inputs.rebaseline_sources | toJson }}"
      GUARD_PIN_NAME: "{{ vars.pin_name }}"
    outputFiles:
      - state.json
      - ops.json
      - findings.json
    script: |
      const fs = require("fs");

      const env = (key, fallback) => (process.env[key] !== undefined && process.env[key] !== "" ? process.env[key] : fallback);
      const num = (key, fallback) => {
        const value = Number(env(key, fallback));
        if (!Number.isFinite(value) || value < 0) throw new Error(`Invalid numeric setting ${key}`);
        return value;
      };

      const cfg = {
        fileDrop: num("MAX_FILE_DROP_PCT", "30"),
        sizeDrop: num("MAX_SIZE_DROP_PCT", "40"),
        minSamples: num("MIN_BASELINE_SAMPLES", "3"),
        window: num("BASELINE_WINDOW", "14"),
        pinCount: num("PIN_COUNT", "2"),
        releaseAfter: num("RELEASE_AFTER_HEALTHY", "3"),
        shiftAfter: num("SUSTAINED_SHIFT_AFTER", "7"),
        pinName: env("GUARD_PIN_NAME", "kestra-retention-guard"),
        rebaseline: JSON.parse(env("REBASELINE_SOURCES", "[]")),
      };

      const readJson = (file, fallback) => {
        const text = fs.readFileSync(file, "utf8").trim();
        return text === "" || text === "null" ? fallback : JSON.parse(text);
      };

      const snapshots = readJson("snapshots.json", null);
      if (!Array.isArray(snapshots)) throw new Error("snapshots.json is not a JSON array: kopia snapshot list --json did not return manifests");
      if (snapshots.length === 0) throw new Error("The repository lists no snapshots at all, so nothing can be judged. Check the bucket, prefix and credentials.");
      const prev = readJson("state.json", {});
      const prevSources = (prev && prev.sources) || {};

      const median = (values) => {
        const sorted = [...values].sort((a, b) => a - b);
        const mid = Math.floor(sorted.length / 2);
        return sorted.length % 2 ? sorted[mid] : (sorted[mid - 1] + sorted[mid]) / 2;
      };

      const bySource = {};
      for (const s of snapshots) {
        if (!s.id || !s.source || !s.stats || typeof s.stats.fileCount !== "number" || typeof s.stats.totalSize !== "number") {
          throw new Error(`Snapshot manifest ${s.id || "?"} lacks id/source/stats.fileCount/stats.totalSize`);
        }
        const ts = Date.parse(s.endTime);
        if (Number.isNaN(ts)) throw new Error(`Snapshot ${s.id} has an unparseable endTime: ${s.endTime}`);
        const key = `${s.source.userName}@${s.source.host}:${s.source.path}`;
        (bySource[key] = bySource[key] || []).push({ ...s, ts });
      }

      const newState = { version: 1, sources: {} };
      const findings = [];
      const ops = [];
      const fmt = (n) => (n >= 1073741824 ? `${(n / 1073741824).toFixed(1)} GiB` : n >= 1048576 ? `${(n / 1048576).toFixed(1)} MiB` : `${n} B`);

      for (const key of Object.keys(bySource).sort()) {
        const all = bySource[key].sort((a, b) => a.ts - b.ts || a.id.localeCompare(b.id));
        const complete = all.filter((s) => !s.incomplete);
        const st = prevSources[key] || { baseline: [], recent: [], last_end_ms: -1, alerted_id: "" };
        let baseline = [...st.baseline];
        let recent = [...st.recent];
        let lastEnd = st.last_end_ms;
        const known = new Set(recent.map((e) => e.id));

        const entryOf = (s, status) => ({ id: s.id, end: s.endTime, ts: s.ts, files: s.stats.fileCount, size: s.stats.totalSize, status });
        const rebaseline = cfg.rebaseline.includes(key);

        if (rebaseline) {
          const tail = complete.slice(-Math.max(cfg.window, cfg.minSamples));
          baseline = tail.map((s) => entryOf(s, "OK"));
          recent = tail.map((s) => entryOf(s, "OK"));
          if (complete.length) lastEnd = complete[complete.length - 1].ts;
        } else {
          for (const s of complete.filter((c) => c.ts > lastEnd && !known.has(c.id))) {
            let status = "OK";
            if (s.stats.errorCount > 0) {
              status = "UPLOAD_ERRORS";
            } else if (baseline.length >= cfg.minSamples) {
              const medFiles = median(baseline.map((b) => b.files));
              const medSize = median(baseline.map((b) => b.size));
              if (medFiles > 0 && s.stats.fileCount < medFiles * (1 - cfg.fileDrop / 100)) status = "FILE_COUNT_DROP";
              else if (medSize > 0 && s.stats.totalSize < medSize * (1 - cfg.sizeDrop / 100)) status = "SIZE_DROP";
            }
            const entry = entryOf(s, status);
            recent.push(entry);
            if (status === "OK") baseline.push(entry);
            lastEnd = Math.max(lastEnd, s.ts);
          }
        }
        baseline = baseline.slice(-cfg.window);
        recent = recent.slice(-50);

        const present = new Set(all.map((s) => s.id));
        const guardPinned = all.filter((s) => (s.pins || []).includes(cfg.pinName)).map((s) => s.id);
        const latest = recent[recent.length - 1];
        const alarm = Boolean(latest) && latest.status !== "OK";
        let trailingBad = 0;
        for (let i = recent.length - 1; i >= 0 && recent[i].status !== "OK"; i--) trailingBad++;
        let trailingOk = 0;
        for (let i = recent.length - 1; i >= 0 && recent[i].status === "OK"; i--) trailingOk++;

        const medFiles = baseline.length ? median(baseline.map((b) => b.files)) : 0;
        const medSize = baseline.length ? median(baseline.map((b) => b.size)) : 0;
        const base = { source: key, latest_id: latest ? latest.id : "", latest_end: latest ? latest.end : "", files: latest ? latest.files : 0, size: latest ? latest.size : 0, baseline_files: medFiles, baseline_size: medSize, trailing_bad: trailingBad };
        const isNew = !latest || latest.id !== st.alerted_id;

        if (rebaseline) {
          if (guardPinned.length) ops.push({ op: "unpin", source: key, ids: guardPinned, ids_str: guardPinned.join(" ") });
          findings.push({ ...base, kind: "REBASELINED", severity: "info", is_new: true, detail: `Baseline reset to the last ${baseline.length} complete snapshot(s) at the operator's request.` });
        } else if (alarm) {
          const candidates = baseline.filter((b) => present.has(b.id)).slice(-cfg.pinCount).map((b) => b.id);
          const toPin = candidates.filter((id) => !guardPinned.includes(id));
          if (!candidates.length) {
            findings.push({ ...base, kind: "NO_GOOD_SNAPSHOT_AVAILABLE", severity: "critical", is_new: isNew, detail: `Latest snapshot ${latest.id} is ${latest.status} and no known-good snapshot remains in the repository to pin.` });
          } else {
            const detail = latest.status === "UPLOAD_ERRORS"
              ? `Latest snapshot ${latest.id} finished with fatal upload errors.`
              : `Latest snapshot ${latest.id} has ${latest.files} files / ${fmt(latest.size)} against a baseline median of ${medFiles} files / ${fmt(medSize)} (${latest.status}).`;
            findings.push({ ...base, kind: latest.status, severity: "high", is_new: isNew, detail: `${detail} Known-good snapshots to protect: ${candidates.join(", ")}.` });
            if (toPin.length) ops.push({ op: "pin", source: key, ids: toPin, ids_str: toPin.join(" ") });
          }
          if (trailingBad >= cfg.shiftAfter) {
            findings.push({ ...base, kind: "SUSTAINED_SHIFT", severity: "warning", is_new: isNew, detail: `${trailingBad} consecutive snapshots are below baseline. If this is a legitimate change, rerun with rebaseline_sources containing "${key}" to accept the new normal and release the pins.` });
          }
        } else if (guardPinned.length && trailingOk >= cfg.releaseAfter) {
          ops.push({ op: "unpin", source: key, ids: guardPinned, ids_str: guardPinned.join(" ") });
          findings.push({ ...base, kind: "RECOVERED", severity: "info", is_new: true, detail: `${trailingOk} consecutive healthy snapshots since the anomaly. Guard pins on ${guardPinned.length} snapshot(s) can be released so normal retention resumes.` });
        }

        newState.sources[key] = { baseline, recent, last_end_ms: lastEnd, alerted_id: alarm && latest ? latest.id : "" };
      }

      const attention = findings.filter((f) => f.is_new);
      const hasActions = ops.length > 0;
      const lines = attention.map((f) => `• *${f.source}* [${f.kind}] ${f.detail}`);
      const opLines = ops.map((o) => `• ${o.op.toUpperCase()} on ${o.source}: ${o.ids.join(", ")}`);
      const summary = [...lines, ...(opLines.length ? ["", "*Planned repository changes (need approval):*", ...opLines] : [])].join("\n");

      fs.writeFileSync("state.json", JSON.stringify(newState));
      fs.writeFileSync("ops.json", JSON.stringify(ops));
      fs.writeFileSync("findings.json", JSON.stringify(findings));

      console.log("::" + JSON.stringify({ outputs: {
        sources_audited: Object.keys(bySource).length,
        findings_count: attention.length,
        ops_count: ops.length,
        has_attention: attention.length > 0 || hasActions,
        has_actions: hasActions,
        ops,
        summary,
      } }) + "::");

  - id: gate_attention
    type: io.kestra.plugin.core.flow.If
    description: Only alert and plan changes when a new anomaly, a recovery or a
      rebaseline needs attention.
    condition: "{{ outputs.analyze_snapshots.vars.has_attention }}"
    then:
      - id: notify_findings
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Report the anomalous sources and the repository changes that are
          waiting for approval.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: >-
          {{ {
            "text": ":warning: *Kopia snapshot guard* on `" ~ inputs.s3_bucket ~ "/" ~ inputs.s3_prefix ~ "` (execution `" ~ execution.id ~ "`)",
            "attachments": [ { "color": "#E2B93B", "text": outputs.analyze_snapshots.vars.summary } ]
          } | toJson }}

      - id: gate_actions
        type: io.kestra.plugin.core.flow.If
        description: A pin or release changes the repository, so it must go through the
          approval gate.
        condition: "{{ outputs.analyze_snapshots.vars.has_actions }}"
        then:
          - id: pause_for_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Hold the plan until an operator resumes it. Unapproved plans fail
              after 12 hours instead of running on their own.
            pauseDuration: PT12H
            behavior: FAIL
            onResume:
              - id: release_pins
                type: BOOL
                defaults: false
                description: Also remove guard pins from recovered sources. Pinned snapshots
                  stop being protected, so tick this only when the source is
                  confirmed healthy.
              - id: approver_note
                type: STRING
                defaults: ""
                description: Ticket or reason recorded in the audit entry.

          - id: apply_plan
            type: io.kestra.plugin.core.flow.Loop
            description: Apply one planned operation per backup source, one at a time.
            values: "{{ outputs.analyze_snapshots.vars.ops }}"
            concurrencyLimit: 1
            tasks:
              - id: route_op
                type: io.kestra.plugin.core.flow.If
                description: Pin known-good snapshots, or release guard pins when the operator
                  approved it.
                condition: "{{ item.value.op == 'pin' }}"
                then:
                  - id: pin_snapshots
                    type: io.kestra.plugin.scripts.shell.Commands
                    description: Add the guard pin so Kopia retention can no longer expire these
                      snapshots.
                    taskRunner:
                      type: io.kestra.plugin.scripts.runner.docker.Docker
                    containerImage: kopia/kopia:0.23.1
                    env:
                      KOPIA_PASSWORD: "{{ secret('KOPIA_PASSWORD') }}"
                      S3_ACCESS_KEY: "{{ secret('KOPIA_S3_ACCESS_KEY') }}"
                      S3_SECRET_KEY: "{{ secret('KOPIA_S3_SECRET_KEY') }}"
                      S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
                      S3_BUCKET: "{{ inputs.s3_bucket }}"
                      S3_PREFIX: "{{ inputs.s3_prefix }}"
                      S3_DISABLE_TLS: "{{ inputs.s3_disable_tls }}"
                      PIN_NAME: "{{ vars.pin_name }}"
                      SNAPSHOT_IDS: "{{ item.value.ids_str }}"
                    commands:
                      - |
                        set -eu
                        tls_flag=""
                        if [ "$S3_DISABLE_TLS" = "true" ]; then tls_flag="--disable-tls"; fi
                        kopia repository connect s3 --bucket="$S3_BUCKET" --endpoint="$S3_ENDPOINT" --prefix="$S3_PREFIX" $tls_flag --access-key="$S3_ACCESS_KEY" --secret-access-key="$S3_SECRET_KEY"
                        kopia snapshot pin --add "$PIN_NAME" $SNAPSHOT_IDS
                else:
                  - id: release_gate
                    type: io.kestra.plugin.core.flow.If
                    description: Release pins only when the operator ticked release_pins at the
                      approval step.
                    condition: "{{ outputs.pause_for_approval.onResume.release_pins }}"
                    then:
                      - id: unpin_snapshots
                        type: io.kestra.plugin.scripts.shell.Commands
                        description: Remove the guard pin so normal retention applies to the recovered
                          source again.
                        taskRunner:
                          type: io.kestra.plugin.scripts.runner.docker.Docker
                        containerImage: kopia/kopia:0.23.1
                        env:
                          KOPIA_PASSWORD: "{{ secret('KOPIA_PASSWORD') }}"
                          S3_ACCESS_KEY: "{{ secret('KOPIA_S3_ACCESS_KEY') }}"
                          S3_SECRET_KEY: "{{ secret('KOPIA_S3_SECRET_KEY') }}"
                          S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
                          S3_BUCKET: "{{ inputs.s3_bucket }}"
                          S3_PREFIX: "{{ inputs.s3_prefix }}"
                          S3_DISABLE_TLS: "{{ inputs.s3_disable_tls }}"
                          PIN_NAME: "{{ vars.pin_name }}"
                          SNAPSHOT_IDS: "{{ item.value.ids_str }}"
                        commands:
                          - |
                            set -eu
                            tls_flag=""
                            if [ "$S3_DISABLE_TLS" = "true" ]; then tls_flag="--disable-tls"; fi
                            kopia repository connect s3 --bucket="$S3_BUCKET" --endpoint="$S3_ENDPOINT" --prefix="$S3_PREFIX" $tls_flag --access-key="$S3_ACCESS_KEY" --secret-access-key="$S3_SECRET_KEY"
                            kopia snapshot pin --remove "$PIN_NAME" $SNAPSHOT_IDS
                    else:
                      - id: log_release_skipped
                        type: io.kestra.plugin.core.log.Log
                        description: Record that the release was left for a later run.
                        message: "Release of guard pins on {{ item.value.source }} skipped: release_pins
                          was not ticked at approval."

          - id: relist_snapshots
            type: io.kestra.plugin.scripts.shell.Commands
            description: Re-read the repository after the changes, read-only, to prove what
              actually landed.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
            containerImage: kopia/kopia:0.23.1
            env:
              KOPIA_PASSWORD: "{{ secret('KOPIA_PASSWORD') }}"
              S3_ACCESS_KEY: "{{ secret('KOPIA_S3_ACCESS_KEY') }}"
              S3_SECRET_KEY: "{{ secret('KOPIA_S3_SECRET_KEY') }}"
              S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
              S3_BUCKET: "{{ inputs.s3_bucket }}"
              S3_PREFIX: "{{ inputs.s3_prefix }}"
              S3_DISABLE_TLS: "{{ inputs.s3_disable_tls }}"
              KOPIA_CONNECT_MODE: "--readonly"
            outputFiles:
              - snapshots_after.json
            commands:
              - |
                set -eu
                tls_flag=""
                if [ "$S3_DISABLE_TLS" = "true" ]; then tls_flag="--disable-tls"; fi
                kopia repository connect s3 $KOPIA_CONNECT_MODE --bucket="$S3_BUCKET" --endpoint="$S3_ENDPOINT" --prefix="$S3_PREFIX" $tls_flag --access-key="$S3_ACCESS_KEY" --secret-access-key="$S3_SECRET_KEY"
                kopia snapshot list --all --show-identical --json > snapshots_after.json
                test -s snapshots_after.json

          - id: verify_plan
            type: io.kestra.plugin.scripts.node.Script
            description: Check that every planned pin is present, every approved release is
              gone, and no pinned snapshot disappeared.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
            containerImage: node:20-slim
            inputFiles:
              snapshots_after.json: "{{
                outputs.relist_snapshots.outputFiles['snapshots_after.json'] }}"
              ops.json: "{{ outputs.analyze_snapshots.outputFiles['ops.json'] }}"
            env:
              GUARD_PIN_NAME: "{{ vars.pin_name }}"
              RELEASE_APPROVED: "{{ outputs.pause_for_approval.onResume.release_pins }}"
              APPROVER_NOTE: "{{ outputs.pause_for_approval.onResume.approver_note }}"
              EXECUTION_ID: "{{ execution.id }}"
            outputFiles:
              - audit.json
            script: |
              const fs = require("fs");

              const pinName = process.env.GUARD_PIN_NAME || "kestra-retention-guard";
              const releaseApproved = process.env.RELEASE_APPROVED === "true";

              const after = JSON.parse(fs.readFileSync("snapshots_after.json", "utf8"));
              if (!Array.isArray(after) || after.length === 0) throw new Error("snapshots_after.json is not a non-empty JSON array");
              const ops = JSON.parse(fs.readFileSync("ops.json", "utf8"));
              const byId = new Map(after.map((s) => [s.id, s]));
              const hasPin = (id) => Boolean(byId.get(id) && (byId.get(id).pins || []).includes(pinName));

              const mismatches = [];
              let pinned = 0;
              let released = 0;
              let skipped = 0;
              for (const op of ops) {
                for (const id of op.ids) {
                  if (op.op === "pin") {
                    if (hasPin(id)) pinned++;
                    else mismatches.push(`${op.source}: snapshot ${id} is not pinned after the pin step`);
                  } else if (releaseApproved) {
                    if (!byId.has(id) || !hasPin(id)) released++;
                    else mismatches.push(`${op.source}: snapshot ${id} still carries the guard pin after the release step`);
                  } else {
                    skipped++;
                  }
                }
              }

              const verified = mismatches.length === 0;
              const audit = {
                executed_at: new Date().toISOString(),
                execution_id: process.env.EXECUTION_ID || "",
                verified,
                pinned,
                released,
                release_skipped: skipped,
                release_approved: releaseApproved,
                approver_note: process.env.APPROVER_NOTE || "",
                mismatches,
                ops,
              };
              fs.writeFileSync("audit.json", JSON.stringify(audit));

              console.log("::" + JSON.stringify({ outputs: { verified, pinned, released, release_skipped: skipped, mismatch_summary: mismatches.join("; ") } }) + "::");

          - id: check_verification
            type: io.kestra.plugin.core.flow.If
            description: Fail loudly if the repository does not match the plan.
            condition: "{{ not outputs.verify_plan.vars.verified }}"
            then:
              - id: fail_verification
                type: io.kestra.plugin.core.execution.Fail
                description: Stop before the baseline is advanced so the next run replans the
                  same changes.
                errorMessage: "Kopia repository does not match the approved plan: {{
                  outputs.verify_plan.vars.mismatch_summary }}"

          - id: record_audit
            type: io.kestra.plugin.core.kv.Set
            description: Keep the last verified change set, approver note and counts for
              compliance review.
            key: "{{ vars.audit_key }}"
            kvType: JSON
            value: "{{ read(outputs.verify_plan.outputFiles['audit.json']) }}"

          - id: notify_result
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Confirm the verified result to the channel.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: >-
              {{ {
                "text": ":white_check_mark: *Kopia snapshot guard* verified: " ~ outputs.verify_plan.vars.pinned ~ " snapshot(s) pinned, " ~ outputs.verify_plan.vars.released ~ " released, " ~ outputs.verify_plan.vars.release_skipped ~ " release(s) skipped (execution `" ~ execution.id ~ "`)."
              } | toJson }}
        else:
          - id: log_alert_only
            type: io.kestra.plugin.core.log.Log
            description: Findings that need no repository change, such as a sustained shift,
              stay alert-only.
            message: "{{ outputs.analyze_snapshots.vars.findings_count }} finding(s)
              reported, no pin or release needed."
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        description: Every source is within its baseline.
        message: "Audited {{ outputs.analyze_snapshots.vars.sources_audited }} backup
          source(s): all snapshots are within baseline."

  - id: persist_baseline
    type: io.kestra.plugin.core.kv.Set
    description: Advance the per-source baseline only after the run finished
      cleanly, so a rejected or failed run replans the same snapshots.
    key: "{{ vars.state_key }}"
    kvType: STRING
    value: "{{ read(outputs.analyze_snapshots.outputFiles['state.json']) }}"

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Page the channel when the guard fails, the approval times out, or
      verification finds a mismatch.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {
        "text": ":rotating_light: *Kopia snapshot guard failed* for `" ~ inputs.s3_bucket ~ "/" ~ inputs.s3_prefix ~ "` (execution `" ~ execution.id ~ "`). The baseline was not advanced; the next run replans the same snapshots."
      } | toJson }}