Redeliver outbound FTP feeds the partner never received

id: ftp-stalled-outbound-spool-redelivery-guard
namespace: company.team
description: >-
  Reconcile an outbound FTP spool against the partner inbox listing, name every
  handoff that never landed or landed short, re-upload only those files behind
  an approval gate, prove the delivered bytes with the partner side checksum
  then archive the source so nothing is ever sent twice.

labels:
  project: ftp-stalled-outbound-spool-redelivery-guard
  team: data-platform

inputs:
  - id: spool_host
    type: STRING
    displayName: Spool FTP host
    defaults: "{{ secret('FTP_SPOOL_HOST') }}"
    description: Host of the FTP server that holds the outbound spool on the producer side.

  - id: spool_port
    type: STRING
    displayName: Spool FTP port
    defaults: "21"
    description: Control port of the spool server.

  - id: spool_username
    type: STRING
    displayName: Spool FTP user
    defaults: "{{ secret('FTP_SPOOL_USERNAME') }}"
    description: Account that can list, read then rename inside the spool directory.

  - id: spool_password
    type: STRING
    displayName: Spool FTP password
    defaults: "{{ secret('FTP_SPOOL_PASSWORD') }}"
    description: Password for the spool account.

  - id: spool_directory
    type: STRING
    displayName: Outbound spool directory
    defaults: "/outbound"
    description: >-
      Directory the handoff job writes its ready to send files into. Read on
      every run. Only files inside it are ever considered.

  - id: archive_directory
    type: STRING
    displayName: Archive directory for sent files
    defaults: "/outbound/sent"
    description: >-
      A redelivered file is renamed into this directory once the partner side
      checksum matched. That rename is what stops the next run resending it.

  - id: partner_host
    type: STRING
    displayName: Partner FTP host
    defaults: "{{ secret('FTP_PARTNER_HOST') }}"
    description: Host of the partner FTP server that receives the feed.

  - id: partner_port
    type: STRING
    displayName: Partner FTP port
    defaults: "21"
    description: Control port of the partner server.

  - id: partner_username
    type: STRING
    displayName: Partner FTP user
    defaults: "{{ secret('FTP_PARTNER_USERNAME') }}"
    description: Account the partner issued for the drop zone.

  - id: partner_password
    type: STRING
    displayName: Partner FTP password
    defaults: "{{ secret('FTP_PARTNER_PASSWORD') }}"
    description: Password for the partner account.

  - id: partner_inbox
    type: STRING
    displayName: Partner inbox directory
    defaults: "/inbox"
    description: >-
      The only directory this flow ever writes to. Everything else on the
      partner server is out of reach.

  - id: deliverable_pattern
    type: STRING
    displayName: Deliverable file pattern
    defaults: "^[A-Za-z0-9._-]+\\.(csv|edi|dat)$"
    description: >-
      Write allowlist by file name. A spool entry that does not match is
      reported as not deliverable then left alone, so a README, a checksum
      sidecar or an operator scratch file is never pushed to the partner.

  - id: in_progress_suffixes
    type: ARRAY
    itemType: STRING
    displayName: Suffixes that mark a file still being written
    defaults: [ ".part", ".lock", ".tmp", ".writing" ]
    description: >-
      A spool file is skipped while a sibling marker with one of these suffixes
      exists, because the producer has not finished writing it. The markers
      themselves are never deliverable.

  - id: sla_minutes
    type: INT
    displayName: Delivery SLA (minutes)
    defaults: 60
    description: >-
      A spool file counts as a missed handoff only once it is older than this.
      Inside the window the nightly handoff still has time to deliver it, so the
      flow stays out of the way.

  - id: max_files_per_run
    type: INT
    displayName: Per-run redelivery ceiling
    defaults: 3
    description: >-
      At most this many files are re-uploaded in one run, oldest first. A spool
      with two hundred undelivered feeds is an incident, not a retry job.

  - id: max_file_megabytes
    type: INT
    displayName: Per-file size ceiling (MB)
    defaults: 64
    description: >-
      A spool file larger than this is reported but never pushed, so one feed
      cannot occupy the worker or the partner link for an unbounded time.

  - id: scan_limit
    type: INT
    displayName: Listing ceiling
    defaults: 200
    description: Maximum entries read from each directory, which bounds every listing.

  - id: dry_run
    type: BOOL
    displayName: Dry run
    defaults: true
    description: >-
      When true the flow reports every missed handoff with its spool size and
      the partner side size then writes nothing.

  - id: auto_remediate
    type: BOOL
    displayName: Allow redelivery
    defaults: false
    description: >-
      When false the flow only reports. An upload needs this true, dry_run false
      plus an explicit approval on the pause.

  - id: slack_webhook_url
    type: STRING
    displayName: Slack webhook
    defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"
    description: Incoming webhook the findings and the failure alert are posted to.

  - id: slack_channel
    type: STRING
    displayName: Slack channel
    defaults: "#data-handoffs"
    description: Channel that receives the reconciliation report.

triggers:
  - id: nightly_reconcile
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    description: >-
      Enable this to reconcile after the nightly handoff window in report only
      mode. It forces dry_run true plus auto_remediate false, so turning the
      schedule on can never upload a file on its own.
    cron: "30 5 * * *"
    timezone: UTC
    inputs:
      dry_run: true
      auto_remediate: false

  - id: spool_arrival
    type: io.kestra.plugin.fs.ftp.Trigger
    disabled: true
    description: >-
      Enable this to reconcile whenever the spool changes instead of waiting for
      the nightly schedule. action NONE leaves every file in place and relies on
      the plugin's own path, size plus last modified state, so the trigger only
      reads. It carries no inputs, which means dry_run stays true plus
      auto_remediate stays false on the default values. The plugin matches
      regExp against the whole remote path, not the bare file name, so this
      pattern names the directory as well. Change both halves together if you
      point from somewhere else.
    host: "{{ secret('FTP_SPOOL_HOST') }}"
    port: "21"
    username: "{{ secret('FTP_SPOOL_USERNAME') }}"
    password: "{{ secret('FTP_SPOOL_PASSWORD') }}"
    from: "/outbound/"
    regExp: "^/outbound/[A-Za-z0-9._-]+\\.(csv|edi|dat)$"
    action: NONE
    maxFiles: 25
    interval: PT5M

tasks:
  - id: list_spool
    type: io.kestra.plugin.fs.ftp.List
    description: >-
      Read the outbound spool through the FTP plugin. Everything is listed, not
      just the deliverable pattern, because the in progress markers have to be
      seen before a sibling file can be called late.
    host: "{{ inputs.spool_host }}"
    port: "{{ inputs.spool_port }}"
    username: "{{ inputs.spool_username }}"
    password: "{{ inputs.spool_password }}"
    from: "{{ inputs.spool_directory }}/"
    regExp: ".*"
    maxFiles: "{{ inputs.scan_limit }}"
    sort: LAST_MODIFIED_ASC

  - id: list_archive
    type: io.kestra.plugin.fs.ftp.List
    description: >-
      Read the archive of files this flow already handed over. A name that is
      here is not redelivered, which is the first half of the no duplicate rule.
    host: "{{ inputs.spool_host }}"
    port: "{{ inputs.spool_port }}"
    username: "{{ inputs.spool_username }}"
    password: "{{ inputs.spool_password }}"
    from: "{{ inputs.archive_directory }}/"
    regExp: ".*"
    maxFiles: "{{ inputs.scan_limit }}"
    sort: LAST_MODIFIED_ASC

  - id: list_partner_inbox
    type: io.kestra.plugin.fs.ftp.List
    description: >-
      Read the partner inbox on the partner's own server. This listing is the
      only authority on what the partner actually holds. A handoff job that
      reported success is not evidence.
    host: "{{ inputs.partner_host }}"
    port: "{{ inputs.partner_port }}"
    username: "{{ inputs.partner_username }}"
    password: "{{ inputs.partner_password }}"
    from: "{{ inputs.partner_inbox }}/"
    regExp: ".*"
    maxFiles: "{{ inputs.scan_limit }}"
    sort: LAST_MODIFIED_ASC

  - id: reconcile
    type: io.kestra.plugin.scripts.python.Script
    description: >-
      Join the two listings by file name. A spool file past the SLA whose
      partner copy is absent is a missed handoff, one whose partner copy is
      smaller is a truncated transfer. Equal sizes mean delivered, a larger
      partner copy means the two sides diverged. Zero byte sources, oversized
      sources, names outside the allowlist, files with a writer marker, names
      already archived plus everything over the ceiling are each refused with
      their own reason.
    containerImage: python:3.13-slim
    inputFiles:
      spool.json: "{{ outputs.list_spool | toJson }}"
      archive.json: "{{ outputs.list_archive | toJson }}"
      partner.json: "{{ outputs.list_partner_inbox | toJson }}"
    env:
      SPOOL_DIR: "{{ inputs.spool_directory }}"
      ARCHIVE_DIR: "{{ inputs.archive_directory }}"
      PARTNER_DIR: "{{ inputs.partner_inbox }}"
      DELIVERABLE_PATTERN: "{{ inputs.deliverable_pattern }}"
      IN_PROGRESS_SUFFIXES: "{{ inputs.in_progress_suffixes | toJson }}"
      SLA_MINUTES: "{{ inputs.sla_minutes }}"
      MAX_FILES: "{{ inputs.max_files_per_run }}"
      MAX_MEGABYTES: "{{ inputs.max_file_megabytes }}"
    script: |
      import json
      import os
      import re
      from datetime import datetime, timezone

      def files(path):
          with open(path) as handle:
              payload = json.load(handle)
          rows = []
          for entry in payload.get("files") or []:
              if entry.get("fileType") != "FILE":
                  continue
              rows.append({
                  "name": entry.get("name"),
                  "path": entry.get("path"),
                  "size": int(entry.get("size") or 0),
                  "updatedDate": entry.get("updatedDate"),
              })
          return rows

      spool_dir = os.environ["SPOOL_DIR"].rstrip("/")
      archive_dir = os.environ["ARCHIVE_DIR"].rstrip("/")
      partner_dir = os.environ["PARTNER_DIR"].rstrip("/")
      pattern = re.compile(os.environ["DELIVERABLE_PATTERN"])
      suffixes = tuple(json.loads(os.environ["IN_PROGRESS_SUFFIXES"]))
      sla_minutes = int(os.environ["SLA_MINUTES"])
      max_files = int(os.environ["MAX_FILES"])
      max_bytes = int(os.environ["MAX_MEGABYTES"]) * 1024 * 1024
      now = datetime.now(timezone.utc)

      spool = files("spool.json")
      archive = {row["name"]: row for row in files("archive.json")}
      partner = {row["name"]: row for row in files("partner.json")}
      spool_names = {row["name"] for row in spool}

      def age_minutes(stamp):
          if not stamp:
              return 0.0
          moment = datetime.fromisoformat(str(stamp).replace("Z", "+00:00"))
          return (now - moment).total_seconds() / 60.0

      findings = []
      for row in spool:
          name = row["name"]
          partner_row = partner.get(name)
          partner_size = int(partner_row["size"]) if partner_row else None
          entry = {
              "name": name,
              "spoolPath": "%s/%s" % (spool_dir, name),
              "archivePath": "%s/%s" % (archive_dir, name),
              "partnerPath": "%s/%s" % (partner_dir, name),
              "size": row["size"],
              "partnerSize": partner_size,
              "updatedDate": row["updatedDate"],
              "ageMinutes": round(age_minutes(row["updatedDate"]), 1),
          }
          marker = next((s for s in suffixes if "%s%s" % (name, s) in spool_names), None)
          if name.endswith(suffixes):
              entry["verdict"] = "writer_marker"
              entry["note"] = ("A producer marker, not a feed file. It is never delivered "
                               "plus never reported as a gap.")
          elif not pattern.match(name or ""):
              entry["verdict"] = "not_deliverable"
              entry["note"] = ("Outside the deliverable allowlist, so it is reported "
                               "then left in the spool.")
          elif marker:
              entry["verdict"] = "writer_active"
              entry["note"] = ("Spared. %s%s exists, so the producer is still writing this "
                               "file. Delivering it now would hand over a partial feed."
                               % (name, marker))
          elif row["size"] == 0:
              entry["verdict"] = "empty_source"
              entry["note"] = ("Refused. The spool copy is zero bytes, so the producer "
                               "aborted. Uploading it would look like an empty feed day.")
          elif row["size"] > max_bytes:
              entry["verdict"] = "too_large"
              entry["note"] = ("Refused. %d bytes is over the %d byte per-file ceiling."
                               % (row["size"], max_bytes))
          elif entry["ageMinutes"] < sla_minutes:
              entry["verdict"] = "within_sla"
              entry["note"] = ("Spared. Ready %s minutes ago, inside the %s minute delivery "
                               "window, so the handoff job still owns it."
                               % (entry["ageMinutes"], sla_minutes))
          elif name in archive:
              entry["verdict"] = "already_archived"
              entry["note"] = ("Refused. %s/%s already exists at %d bytes, so this name was "
                               "handed over once. Re-dropping the same name needs a person: "
                               "the flow will not overwrite a delivered feed."
                               % (archive_dir, name, archive[name]["size"]))
          elif partner_size is None:
              entry["verdict"] = "missing"
              entry["note"] = ("Missed handoff. %d bytes sat in the spool for %s minutes plus "
                               "the partner inbox has no copy at all."
                               % (row["size"], entry["ageMinutes"]))
          elif partner_size < row["size"]:
              entry["verdict"] = "short_delivery"
              entry["note"] = ("Truncated transfer. The partner holds %d of %d bytes, so the "
                               "feed is short by %d bytes."
                               % (partner_size, row["size"], row["size"] - partner_size))
          elif partner_size > row["size"]:
              entry["verdict"] = "diverged"
              entry["note"] = ("Refused. The partner copy is %d bytes against %d in the "
                               "spool, so the two sides disagree about the content. "
                               "Overwriting could destroy data the partner already has."
                               % (partner_size, row["size"]))
          else:
              entry["verdict"] = "delivered"
              entry["note"] = ("Already delivered at %d bytes on both sides, so it is never "
                               "sent again." % row["size"])
          findings.append(entry)

      findings.sort(key=lambda item: item["ageMinutes"], reverse=True)
      repairable = [row for row in findings if row["verdict"] in ("missing", "short_delivery")]
      targets = repairable[:max_files]
      for row in repairable[max_files:]:
          row["verdict"] = "over_ceiling"
          row["note"] = ("Held back by the %d file per-run ceiling. The next run picks it up."
                         % max_files)

      counts = {}
      for row in findings:
          counts[row["verdict"]] = counts.get(row["verdict"], 0) + 1

      gap_verdicts = ("missing", "short_delivery", "diverged", "empty_source",
                      "too_large", "already_archived", "over_ceiling")
      gap_count = sum(counts.get(key, 0) for key in gap_verdicts)
      target_bytes = sum(row["size"] - (row["partnerSize"] or 0) for row in targets)

      lines = []
      for row in findings:
          partner_text = "absent" if row["partnerSize"] is None else "%d bytes" % row["partnerSize"]
          lines.append("%s: %s spool=%d bytes partner=%s age=%sm. %s"
                       % (row["verdict"], row["name"], row["size"], partner_text,
                          row["ageMinutes"], row["note"]))
      for line in lines:
          print(line)
      print("spool=%d partner=%d gaps=%d targets=%d buckets=%s"
            % (len(findings), len(partner), gap_count, len(targets),
               json.dumps(counts, sort_keys=True)))

      print("::" + json.dumps({"outputs": {
          "findings": findings,
          "targets": targets,
          "targetCount": len(targets),
          "targetNames": [row["name"] for row in targets],
          "gapCount": gap_count,
          "spoolCount": len(findings),
          "partnerCount": len(partner),
          "deliveredCount": counts.get("delivered", 0),
          "missingCount": counts.get("missing", 0),
          "shortCount": counts.get("short_delivery", 0),
          "divergedCount": counts.get("diverged", 0),
          "sparedCount": (counts.get("within_sla", 0) + counts.get("writer_active", 0)
                          + counts.get("not_deliverable", 0) + counts.get("writer_marker", 0)),
          "missingBytes": target_bytes,
          "buckets": counts,
          "report": "\n".join(lines) if lines else "The outbound spool is empty.",
      }}) + "::")

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: One line per spool entry, naming the partner side size next to the
      spool size.
    message: |
      Reconciled {{ inputs.spool_directory }} on {{ inputs.spool_host }} against {{ inputs.partner_inbox }} on {{ inputs.partner_host }}.
      spool={{ outputs.reconcile.vars.spoolCount }} partner={{ outputs.reconcile.vars.partnerCount }} gaps={{ outputs.reconcile.vars.gapCount }} targets={{ outputs.reconcile.vars.targetCount }} missing={{ outputs.reconcile.vars.missingCount }} short={{ outputs.reconcile.vars.shortCount }} diverged={{ outputs.reconcile.vars.divergedCount }} delivered={{ outputs.reconcile.vars.deliveredCount }} spared={{ outputs.reconcile.vars.sparedCount }}
      {{ outputs.reconcile.vars.report }}

  - id: handoff_gap
    type: io.kestra.plugin.core.flow.If
    description: Everything below runs only when the reconciliation found a gap.
    condition: "{{ outputs.reconcile.vars.gapCount > 0 }}"
    then:
      - id: notify_gap
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: >-
          One message per reconciliation listing every file the partner is
          missing, how short each one is plus why the flow will or will not
          redeliver it.
        url: "{{ inputs.slack_webhook_url }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": ":outbox_tray: {{ outputs.reconcile.vars.gapCount }} of {{ outputs.reconcile.vars.spoolCount }} file(s) in `{{ inputs.spool_host }}{{ inputs.spool_directory }}` need attention against `{{ inputs.partner_host }}{{ inputs.partner_inbox }}`. {{ outputs.reconcile.vars.targetCount }} can be redelivered now, {{ outputs.reconcile.vars.missingBytes }} byte(s) outstanding. Full reconciliation, spared entries included:{% for row in outputs.reconcile.vars.findings %}\n• `{{ row.name }}` spool {{ row.size }}B, partner {{ row.partnerSize ?? 'absent' }}, age {{ row.ageMinutes }}m, verdict *{{ row.verdict }}*. {{ row.note }}{% endfor %}"
          }

  - id: route_redelivery
    type: io.kestra.plugin.core.flow.If
    description: >-
      An upload needs three things true at once. auto_remediate allows
      redelivery to be considered, dry_run false allows it to be written plus
      there has to be a file that cleared the allowlist, the writer markers, the
      SLA window, the size bounds, the archive, the divergence check and the
      per-run ceiling. Miss one and the run stays a report.
    condition: "{{ inputs.auto_remediate == true and inputs.dry_run == false and
      outputs.reconcile.vars.targetCount > 0 }}"
    then:
      - id: approval
        type: io.kestra.plugin.core.flow.Pause
        description: >-
          Re-uploading a feed hands data to a third party a second time. Read
          the findings first: a partner that already processed a short file may
          need the overwrite announced, plus a name the partner renamed on
          arrival will not show up in the inbox listing at all.
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: >-
              Set true to re-upload the listed files. Anything else leaves both
              servers untouched.

      - id: approved_redelivery
        type: io.kestra.plugin.core.flow.If
        description: An unanswered pause reads as a refusal, never as approval.
        condition: "{{ (outputs.approval.onResume.approved ?? false) == true }}"
        then:
          - id: recheck_spool
            type: io.kestra.plugin.fs.ftp.List
            description: >-
              The approval is a snapshot of the spool as it was before a person
              read it. Read the spool again.
            host: "{{ inputs.spool_host }}"
            port: "{{ inputs.spool_port }}"
            username: "{{ inputs.spool_username }}"
            password: "{{ inputs.spool_password }}"
            from: "{{ inputs.spool_directory }}/"
            regExp: ".*"
            maxFiles: "{{ inputs.scan_limit }}"
            sort: LAST_MODIFIED_ASC

          - id: recheck_archive
            type: io.kestra.plugin.fs.ftp.List
            description: Read the archive again so a name handed over during the pause is
              seen.
            host: "{{ inputs.spool_host }}"
            port: "{{ inputs.spool_port }}"
            username: "{{ inputs.spool_username }}"
            password: "{{ inputs.spool_password }}"
            from: "{{ inputs.archive_directory }}/"
            regExp: ".*"
            maxFiles: "{{ inputs.scan_limit }}"
            sort: LAST_MODIFIED_ASC

          - id: recheck_partner
            type: io.kestra.plugin.fs.ftp.List
            description: >-
              Read the partner inbox again. A retry that succeeded while the
              approval waited must not be repeated.
            host: "{{ inputs.partner_host }}"
            port: "{{ inputs.partner_port }}"
            username: "{{ inputs.partner_username }}"
            password: "{{ inputs.partner_password }}"
            from: "{{ inputs.partner_inbox }}/"
            regExp: ".*"
            maxFiles: "{{ inputs.scan_limit }}"
            sort: LAST_MODIFIED_ASC

          - id: revalidate_plan
            type: io.kestra.plugin.scripts.python.Script
            description: >-
              Compare every approved target against the second pair of listings.
              The spool file has to still be there at the same size with the
              same last modified stamp, no writer marker may have appeared, the
              name must still be absent from the archive plus the partner side
              must still be absent or still short by the same amount. Anything
              else makes the plan stale.
            containerImage: python:3.13-slim
            inputFiles:
              spool.json: "{{ outputs.recheck_spool | toJson }}"
              archive.json: "{{ outputs.recheck_archive | toJson }}"
              partner.json: "{{ outputs.recheck_partner | toJson }}"
              targets.json: "{{ outputs.reconcile.vars.targets | toJson }}"
            env:
              IN_PROGRESS_SUFFIXES: "{{ inputs.in_progress_suffixes | toJson }}"
            script: |
              import json
              import os

              def index(path):
                  with open(path) as handle:
                      payload = json.load(handle)
                  out = {}
                  for entry in payload.get("files") or []:
                      if entry.get("fileType") != "FILE":
                          continue
                      out[entry.get("name")] = {
                          "size": int(entry.get("size") or 0),
                          "updatedDate": entry.get("updatedDate"),
                      }
                  return out

              with open("targets.json") as handle:
                  targets = json.load(handle)

              spool = index("spool.json")
              archive = index("archive.json")
              partner = index("partner.json")
              suffixes = tuple(json.loads(os.environ["IN_PROGRESS_SUFFIXES"]))

              confirmed = []
              stale = []
              for row in targets:
                  name = row["name"]
                  source = spool.get(name)
                  partner_row = partner.get(name)
                  partner_size = partner_row["size"] if partner_row else None
                  reason = None
                  if source is None:
                      reason = "the spool copy disappeared during the approval"
                  elif source["size"] != row["size"]:
                      reason = ("the spool copy changed from %d to %d bytes during the approval"
                                % (row["size"], source["size"]))
                  elif source["updatedDate"] != row.get("updatedDate"):
                      reason = ("the spool copy was rewritten during the approval, last "
                                "modified moved from %s to %s"
                                % (row.get("updatedDate"), source["updatedDate"]))
                  elif any("%s%s" % (name, s) in spool for s in suffixes):
                      reason = "a producer marker appeared during the approval"
                  elif name in archive:
                      reason = ("the name was archived during the approval at %d bytes, so "
                                "another pass already handed it over" % archive[name]["size"])
                  elif partner_size == row["size"]:
                      reason = ("the partner received the full %d bytes during the approval, "
                                "so re-uploading would duplicate a delivered file" % row["size"])
                  elif partner_size is not None and partner_size > row["size"]:
                      reason = ("the partner copy grew to %d bytes against %d in the spool "
                                "during the approval" % (partner_size, row["size"]))
                  elif partner_size != row.get("partnerSize"):
                      reason = ("the partner side size moved from %s to %s during the approval"
                                % (row.get("partnerSize"), partner_size))
                  if reason:
                      stale.append({"name": name, "reason": reason})
                      print("stale: %s. %s" % (name, reason))
                  else:
                      confirmed.append(row)
                      print("confirmed: %s spool=%d bytes partner=%s. Eligibility unchanged "
                            "since the plan was built."
                            % (name, row["size"],
                               "absent" if partner_size is None else "%d bytes" % partner_size))

              print("confirmed=%d stale=%d" % (len(confirmed), len(stale)))
              print("::" + json.dumps({"outputs": {
                  "confirmedTargets": confirmed,
                  "confirmedCount": len(confirmed),
                  "confirmedNames": [row["name"] for row in confirmed],
                  "staleCount": len(stale),
                  "staleReport": "; ".join("%s: %s" % (row["name"], row["reason"])
                                           for row in stale),
              }}) + "::")

          - id: plan_unchanged
            type: io.kestra.plugin.core.flow.If
            description: >-
              A single stale target stops the whole batch before any byte is
              written, because the approval was given for the plan as a whole.
            condition: "{{ outputs.revalidate_plan.vars.staleCount == 0 }}"
            then:
              - id: redeliver
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  One file at a time so a refused upload or a checksum mismatch
                  never hides behind a sibling iteration. Each iteration reads
                  the spool copy, uploads it to the partner inbox, downloads the
                  partner copy back with the plugin's own checksum comparison
                  then only renames the source into the archive once those bytes
                  matched.
                values: "{{ outputs.revalidate_plan.vars.confirmedTargets }}"
                concurrencyLimit: 1
                tasks:
                  - id: fetch_spool_file
                    type: io.kestra.plugin.fs.ftp.Download
                    description: >-
                      Read the spool copy into Kestra's internal storage and
                      record its SHA-256. This hash is the expected value the
                      partner side has to reproduce.
                    host: "{{ inputs.spool_host }}"
                    port: "{{ inputs.spool_port }}"
                    username: "{{ inputs.spool_username }}"
                    password: "{{ inputs.spool_password }}"
                    from: "{{ fromJson(item.value).spoolPath }}"
                    checksumAlgorithm: SHA_256

                  - id: push_to_partner
                    type: io.kestra.plugin.fs.ftp.Upload
                    description: >-
                      The only write that touches the partner. It replaces a
                      truncated copy of the same name or creates the missing
                      one, inside the approved inbox directory.
                    host: "{{ inputs.partner_host }}"
                    port: "{{ inputs.partner_port }}"
                    username: "{{ inputs.partner_username }}"
                    password: "{{ inputs.partner_password }}"
                    from: "{{ outputs.fetch_spool_file.to }}"
                    to: "{{ fromJson(item.value).partnerPath }}"
                    overwrite: true

                  - id: confirm_partner_bytes
                    type: io.kestra.plugin.fs.ftp.Download
                    description: >-
                      Read the file back off the partner server and compare its
                      SHA-256 with the spool copy. A mismatch or a short read
                      fails the iteration, so the archive rename below is never
                      reached on a bad delivery.
                    host: "{{ inputs.partner_host }}"
                    port: "{{ inputs.partner_port }}"
                    username: "{{ inputs.partner_username }}"
                    password: "{{ inputs.partner_password }}"
                    from: "{{ fromJson(item.value).partnerPath }}"
                    checksumAlgorithm: SHA_256
                    validateChecksum: true
                    checksumExpected: "{{ outputs.fetch_spool_file.checksum }}"

                  - id: archive_spool_file
                    type: io.kestra.plugin.fs.ftp.Move
                    description: >-
                      Rename the source out of the spool now that the partner
                      holds the same bytes. This is what makes the next run a
                      no-op instead of a second delivery.
                    host: "{{ inputs.spool_host }}"
                    port: "{{ inputs.spool_port }}"
                    username: "{{ inputs.spool_username }}"
                    password: "{{ inputs.spool_password }}"
                    from: "{{ fromJson(item.value).spoolPath }}"
                    to: "{{ fromJson(item.value).archivePath }}"

              - id: reread_partner
                type: io.kestra.plugin.fs.ftp.List
                description: >-
                  Read the partner inbox a third time. An upload that returned
                  success proves nothing: this listing decides.
                host: "{{ inputs.partner_host }}"
                port: "{{ inputs.partner_port }}"
                username: "{{ inputs.partner_username }}"
                password: "{{ inputs.partner_password }}"
                from: "{{ inputs.partner_inbox }}/"
                regExp: ".*"
                maxFiles: "{{ inputs.scan_limit }}"
                sort: LAST_MODIFIED_ASC

              - id: reread_spool
                type: io.kestra.plugin.fs.ftp.List
                description: Read the spool again to confirm every delivered source left it.
                host: "{{ inputs.spool_host }}"
                port: "{{ inputs.spool_port }}"
                username: "{{ inputs.spool_username }}"
                password: "{{ inputs.spool_password }}"
                from: "{{ inputs.spool_directory }}/"
                regExp: ".*"
                maxFiles: "{{ inputs.scan_limit }}"
                sort: LAST_MODIFIED_ASC

              - id: reread_archive
                type: io.kestra.plugin.fs.ftp.List
                description: Read the archive again to confirm every delivered source landed in
                  it.
                host: "{{ inputs.spool_host }}"
                port: "{{ inputs.spool_port }}"
                username: "{{ inputs.spool_username }}"
                password: "{{ inputs.spool_password }}"
                from: "{{ inputs.archive_directory }}/"
                regExp: ".*"
                maxFiles: "{{ inputs.scan_limit }}"
                sort: LAST_MODIFIED_ASC

              - id: verify_delivery
                type: io.kestra.plugin.scripts.python.Script
                description: >-
                  Exit non-zero unless every approved file now sits in the
                  partner inbox at the full spool size, has left the spool, has
                  landed in the archive at the same size plus every partner file
                  the flow did not target still has the size it had before. One
                  failed check fails the run.
                containerImage: python:3.13-slim
                inputFiles:
                  partner_before.json: "{{ outputs.recheck_partner | toJson }}"
                  partner_after.json: "{{ outputs.reread_partner | toJson }}"
                  spool_after.json: "{{ outputs.reread_spool | toJson }}"
                  archive_after.json: "{{ outputs.reread_archive | toJson }}"
                  targets.json: "{{ outputs.revalidate_plan.vars.confirmedTargets | toJson }}"
                script: |
                  import json
                  import sys

                  def index(path):
                      with open(path) as handle:
                          payload = json.load(handle)
                      out = {}
                      for entry in payload.get("files") or []:
                          if entry.get("fileType") != "FILE":
                              continue
                          out[entry.get("name")] = int(entry.get("size") or 0)
                      return out

                  with open("targets.json") as handle:
                      targets = json.load(handle)

                  before = index("partner_before.json")
                  after = index("partner_after.json")
                  spool = index("spool_after.json")
                  archive = index("archive_after.json")
                  names = {row["name"] for row in targets}

                  failures = []
                  delivered = []
                  bytes_delivered = 0
                  for row in targets:
                      name = row["name"]
                      expected = row["size"]
                      landed = after.get(name)
                      if landed is None:
                          failures.append("%s is still absent from the partner inbox" % name)
                          continue
                      if landed != expected:
                          failures.append("%s is %d bytes on the partner against %d in the "
                                          "spool" % (name, landed, expected))
                          continue
                      if name in spool:
                          failures.append("%s is still in the outbound spool, so the next run "
                                          "would send it again" % name)
                          continue
                      if archive.get(name) != expected:
                          failures.append("%s is not archived at %d bytes, archive holds %s"
                                          % (name, expected, archive.get(name)))
                          continue
                      delivered.append(name)
                      bytes_delivered += expected - (row.get("partnerSize") or 0)
                      print("%s: delivered %d bytes, partner went from %s to %d, source "
                            "archived" % (name, expected,
                                          "absent" if row.get("partnerSize") is None
                                          else "%d" % row["partnerSize"], landed))

                  for name, size in sorted(before.items()):
                      if name in names:
                          continue
                      if after.get(name) != size:
                          failures.append("untargeted partner file %s changed from %d to %s "
                                          "bytes" % (name, size, after.get(name)))
                      else:
                          print("%s: untouched at %d bytes, not a target" % (name, size))

                  print("delivered=%d of %d, bytes=%d, failures=%d"
                        % (len(delivered), len(targets), bytes_delivered, len(failures)))
                  print("::" + json.dumps({"outputs": {
                      "delivered": delivered,
                      "deliveredCount": len(delivered),
                      "bytesDelivered": bytes_delivered,
                      "failureCount": len(failures),
                      "failures": failures,
                  }}) + "::")

                  if failures:
                      print("re-verify failed: " + "; ".join(failures), file=sys.stderr)
                      sys.exit(1)

              - id: audit_delivered_bytes
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  After every write, read the archived source then the partner
                  copy once more and let the plugin compare their SHA-256. This
                  proves the two sides hold identical bytes at the end of the
                  run, not only at the moment of the upload.
                values: "{{ outputs.revalidate_plan.vars.confirmedTargets }}"
                concurrencyLimit: 1
                tasks:
                  - id: archived_checksum
                    type: io.kestra.plugin.fs.ftp.Download
                    description: SHA-256 of the archived source copy.
                    host: "{{ inputs.spool_host }}"
                    port: "{{ inputs.spool_port }}"
                    username: "{{ inputs.spool_username }}"
                    password: "{{ inputs.spool_password }}"
                    from: "{{ fromJson(item.value).archivePath }}"
                    checksumAlgorithm: SHA_256

                  - id: partner_checksum_matches
                    type: io.kestra.plugin.fs.ftp.Download
                    description: >-
                      Fails the run when the partner bytes no longer match the
                      archived source.
                    host: "{{ inputs.partner_host }}"
                    port: "{{ inputs.partner_port }}"
                    username: "{{ inputs.partner_username }}"
                    password: "{{ inputs.partner_password }}"
                    from: "{{ fromJson(item.value).partnerPath }}"
                    checksumAlgorithm: SHA_256
                    validateChecksum: true
                    checksumExpected: "{{ outputs.archived_checksum.checksum }}"

              - id: log_redelivered
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Redelivered {{ outputs.verify_delivery.vars.deliveredCount }}
                  file(s) carrying {{
                  outputs.verify_delivery.vars.bytesDelivered }} outstanding
                  byte(s) to {{ inputs.partner_host }}{{ inputs.partner_inbox
                  }}: {{ outputs.verify_delivery.vars.delivered | join(' ') }}.
                  Each one matched on SHA-256 then left the spool for {{
                  inputs.archive_directory }}.
            else:
              - id: stale_plan
                type: io.kestra.plugin.core.execution.Fail
                description: >-
                  The approved plan no longer matches the two servers, so the
                  run stops without writing anything.
                errorMessage: >-
                  Refused before any upload. {{
                  outputs.revalidate_plan.vars.staleCount }} of {{
                  outputs.reconcile.vars.targetCount }} approved file(s) changed
                  while the approval waited: {{
                  outputs.revalidate_plan.vars.staleReport }}. Run the flow
                  again to build a fresh plan.
        else:
          - id: log_declined
            type: io.kestra.plugin.core.log.Log
            message: >-
              Approval declined or left unanswered, so nothing was uploaded. {{
              outputs.reconcile.vars.targetCount }} file(s) are still waiting on
              {{ inputs.partner_host }}{{ inputs.partner_inbox }}.
    else:
      - id: log_report_only
        type: io.kestra.plugin.core.log.Log
        message: >-
          Report only, nothing was written. gaps={{
          outputs.reconcile.vars.gapCount }} targets={{
          outputs.reconcile.vars.targetCount }} auto_remediate={{
          inputs.auto_remediate }} dry_run={{ inputs.dry_run }}. Set
          auto_remediate true with dry_run false to be offered the approval
          gate.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: >-
      Fires on any failure, including a checksum mismatch on the partner side
      plus a plan that went stale while the approval waited.
    url: "{{ inputs.slack_webhook_url }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": ":rotating_light: ftp-stalled-outbound-spool-redelivery-guard failed between `{{ inputs.spool_host }}{{ inputs.spool_directory }}` and `{{ inputs.partner_host }}{{ inputs.partner_inbox }}`. Execution {{ execution.id }}. If the verification step failed, at least one file was uploaded but is not confirmed on the partner side, so check the inbox listing by hand before the next run."
      }