Restore an Argo CD sync policy switched off during an incident

id: argocd-sync-policy-reenable-guard
namespace: company.team
description: >-
  Find Argo CD Applications whose automated sync, prune or selfHeal was switched
  off during an incident and never switched back, report the Git commits that
  are stranded because of it, restore the declared policy behind an approval
  gate, then prove the stranded commit actually reached the cluster before the
  run is allowed to pass.

labels:
  project: argocd-sync-policy-reenable-guard
  team: platform

inputs:
  - id: argocd_server
    type: STRING
    displayName: Argo CD API server
    defaults: "{{ secret('ARGOCD_SERVER') }}"
    description: >-
      host:port of the Argo CD API server, reachable from the Kestra worker. The
      CLI talks to this address for every read plus every write.

  - id: argocd_token
    type: STRING
    displayName: Argo CD API token
    defaults: "{{ secret('ARGOCD_TOKEN') }}"
    description: >-
      Token for an Argo CD account that can read every audited application plus
      update the applications inside the restorable projects. Nothing else is
      needed: no cluster admin, no repository write access.

  - id: argocd_cli_version
    type: STRING
    displayName: Argo CD CLI version
    defaults: "2.13.3"
    description: >-
      CLI version the plugin downloads. Pin it to the version of your Argo CD
      server so the patch payload is parsed the same way the server expects.

  - id: cli_docker_network
    type: STRING
    displayName: Docker network for the Argo CD CLI container
    defaults: bridge
    description: >-
      The Argo CD plugin runs the CLI in a short lived container. Point this at
      the docker network from which your Argo CD server address resolves. Leave
      it on bridge when that address is routable from the worker host.

  - id: k8s_master_url
    type: STRING
    displayName: Kubernetes API server URL
    defaults: "{{ secret('K8S_MASTER_URL') }}"
    description: >-
      API server that holds the Application objects. It is used for one read:
      the inventory of Application names plus their annotations, because the
      Argo CD plugin can only fetch one named application at a time.

  - id: k8s_oauth_token
    type: STRING
    displayName: Kubernetes bearer token
    defaults: "{{ secret('K8S_OAUTH_TOKEN') }}"
    description: >-
      Token for a ServiceAccount that can get plus list applications.argoproj.io
      in the Argo CD namespace. No write verb is used on this credential.

  - id: k8s_ca_cert_data
    type: STRING
    displayName: Kubernetes CA bundle
    defaults: "{{ secret('K8S_CA_CERT_DATA') }}"
    description: Base64 encoded PEM CA bundle for the Kubernetes API server.

  - id: argocd_namespace
    type: STRING
    displayName: Argo CD namespace
    defaults: argocd
    description: Namespace the Application objects live in.

  - id: policy_baseline
    type: JSON
    displayName: Declared policy baseline
    defaults: |
      {"automated": true, "prune": true, "selfHeal": true}
    description: >-
      The sync policy every audited application is expected to carry. A missing
      automated block means Git changes never reach the cluster, selfHeal false
      means hand edits in the cluster are never corrected, prune false means
      resources deleted from Git stay behind forever.

  - id: baseline_overrides
    type: JSON
    displayName: Per application baseline overrides
    defaults: |
      {}
    description: >-
      Declared exceptions, keyed by application name, for example
      {"batch-loader": {"prune": false}}. An application that matches its own
      override is healthy, so a deliberately different policy is never rewritten
      back to the fleet default.

  - id: restorable_projects
    type: ARRAY
    itemType: STRING
    displayName: Projects where a policy may be restored
    defaults: [ "production" ]
    description: >-
      Write allowlist, matched on spec.project. An application outside these
      projects is read, classified plus reported in full, never patched. Keep
      the Argo CD token scoped to the same projects so the allowlist is enforced
      twice.

  - id: exempt_annotation
    type: STRING
    displayName: Opt out annotation
    defaults: policy-guard.kestra.io/exempt
    description: >-
      An Application carrying this annotation is skipped before its policy is
      even read. This is how a genuinely manual application stays manual.

  - id: max_apps_per_run
    type: INT
    displayName: Per run ceiling
    defaults: 3
    description: >-
      At most this many applications are patched in one run, worst drift first.
      Twenty applications with automated sync off is an incident, not a cleanup
      job. The rest are listed as held back plus picked up by the next run.

  - id: dry_run
    type: BOOL
    displayName: Dry run
    defaults: true
    description: >-
      When true the flow reports every policy gap, names the stranded Git commit
      plus the exact patch it would send, then writes nothing.

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

  - id: slack_channel
    type: STRING
    displayName: Slack channel
    defaults: "#platform-alerts"
    description: Channel the findings plus the failure alert are posted to.

triggers:
  - id: quarter_hourly_audit
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    description: >-
      Enable this to audit every 15 minutes in report only mode. It forces
      dry_run true plus auto_remediate false, so turning the schedule on can
      never patch an application on its own.
    cron: "*/15 * * * *"
    timezone: UTC
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: list_applications
    type: io.kestra.plugin.kubernetes.kubectl.Get
    description: >-
      Inventory of Application objects with their annotations. The Argo CD
      plugin reads one named application per call, so the names have to come
      from somewhere: this read is that somewhere plus it is also where the opt
      out annotation is found.
    connection:
      masterUrl: "{{ inputs.k8s_master_url }}"
      oauthToken: "{{ inputs.k8s_oauth_token }}"
      caCertData: "{{ inputs.k8s_ca_cert_data }}"
    namespace: "{{ inputs.argocd_namespace }}"
    resourceType: application
    apiGroup: argoproj.io
    apiVersion: v1alpha1
    fetchType: FETCH

  - id: plan_scope
    type: io.kestra.plugin.scripts.python.Script
    description: >-
      Split the inventory into applications whose policy will be read plus
      applications excluded by the opt out annotation. Exclusion happens here,
      before any policy is fetched, so an exempt application is never part of a
      plan in the first place.
    containerImage: python:3.13-slim
    inputFiles:
      inventory.json: "{{ outputs.list_applications | toJson }}"
    env:
      EXEMPT_ANNOTATION: "{{ inputs.exempt_annotation }}"
    script: |
      import json
      import os

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

      exempt_key = os.environ["EXEMPT_ANNOTATION"].strip()
      scan, exempt = [], []
      for meta in payload.get("metadataItems") or []:
          name = meta.get("name")
          annotations = meta.get("annotations") or {}
          if exempt_key and exempt_key in annotations:
              exempt.append({"application": name, "reason": annotations[exempt_key] or "annotated exempt"})
              continue
          scan.append(name)

      scan.sort()
      for row in exempt:
          print(f"exempt: {row['application']} skipped before any read, annotation says: {row['reason']}")
      print(f"inventory={payload.get('size')} scanned={len(scan)} exempt={len(exempt)}")

      print("::" + json.dumps({"outputs": {
          "scan": scan,
          "scanCount": len(scan),
          "exempt": exempt,
          "exemptCount": len(exempt),
          "inventoryCount": payload.get("size") or 0,
      }}) + "::")

  - id: read_policies
    type: io.kestra.plugin.core.flow.Loop
    description: >-
      One Argo CD read per application, one at a time, so a single unreachable
      application cannot be hidden by a sibling iteration. The declared outputs
      block is what carries each finding back out of the loop.
    values: "{{ outputs.plan_scope.vars.scan }}"
    concurrencyLimit: 1
    fetchType: FETCH
    outputs:
      - id: finding
        type: STRING
        value: "{{ outputs.classify_policy.vars.finding }}"
    tasks:
      - id: read_policy
        type: io.kestra.plugin.argocd.apps.Status
        description: >-
          argocd app get with a forced refresh. The refresh is what makes the
          sync state current, which is how a stranded Git commit becomes visible
          instead of looking like a quiet healthy application.
        server: "{{ inputs.argocd_server }}"
        token: "{{ inputs.argocd_token }}"
        application: "{{ item.value }}"
        argoCDVersion: "{{ inputs.argocd_cli_version }}"
        plaintext: true
        refresh: true
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          networkMode: "{{ inputs.cli_docker_network }}"

      - id: classify_policy
        type: io.kestra.plugin.scripts.python.Script
        description: >-
          Compare the live policy against the baseline this application
          declares, which is the fleet default overridden by its own entry when
          it has one. Only a policy that is weaker than the declared one counts
          as drift: an application with more automation than declared is
          reported, never rewritten, because taking automation away is not this
          flow's job.
        containerImage: python:3.13-slim
        inputFiles:
          app.json: "{{ outputs.read_policy.rawOutput }}"
        env:
          BASELINE: "{{ inputs.policy_baseline | toJson }}"
          OVERRIDES: "{{ inputs.baseline_overrides | toJson }}"
          RESTORABLE_PROJECTS: "{{ inputs.restorable_projects | toJson }}"
        script: |
          import json
          import os

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

          baseline = json.loads(os.environ["BASELINE"])
          overrides = json.loads(os.environ["OVERRIDES"])
          restorable = set(json.loads(os.environ["RESTORABLE_PROJECTS"]))

          meta = app.get("metadata") or {}
          spec = app.get("spec") or {}
          status = app.get("status") or {}
          source = spec.get("source") or {}
          sync = status.get("sync") or {}
          name = meta.get("name")

          declared = dict(baseline)
          declared.update(overrides.get(name) or {})
          automated = (spec.get("syncPolicy") or {}).get("automated")
          observed = {
              "automated": automated is not None,
              "prune": bool((automated or {}).get("prune")),
              "selfHeal": bool((automated or {}).get("selfHeal")),
          }
          want = {key: bool(declared.get(key)) for key in ("automated", "prune", "selfHeal")}
          weaker = sorted(key for key in want if want[key] and not observed[key])
          stronger = sorted(key for key in want if observed[key] and not want[key])

          sync_status = sync.get("status")
          finding = {
              "application": name,
              "project": spec.get("project"),
              "repoURL": source.get("repoURL"),
              "path": source.get("path"),
              "targetRevision": source.get("targetRevision"),
              "observed": observed,
              "declared": want,
              "weakerFields": weaker,
              "strongerFields": stronger,
              "syncStatus": sync_status,
              "healthStatus": (status.get("health") or {}).get("status"),
              "syncedRevision": sync.get("revision"),
              "driftPending": sync_status == "OutOfSync",
              "rollback": (
                  "argocd app patch " + str(name) + " --type merge --patch '"
                  + json.dumps({"spec": {"syncPolicy": {"automated": automated}}})
                  + "'"
              ),
          }

          if not weaker and not stronger:
              finding["verdict"] = "matches_baseline"
              finding["note"] = "Policy equals the declared baseline."
          elif not weaker and stronger:
              finding["verdict"] = "stronger_than_baseline"
              finding["note"] = ("Reported only. " + ", ".join(stronger) + " is enabled while the "
                                 "declared baseline turns it off. This flow never removes automation.")
          elif finding["project"] not in restorable:
              finding["verdict"] = "project_not_allowed"
              finding["note"] = ("Reported only. Project " + str(finding["project"]) + " is not on the "
                                 "write allowlist, so " + ", ".join(weaker) + " stays off until a human "
                                 "widens the allowlist.")
          else:
              finding["verdict"] = "restorable"
              finding["note"] = (", ".join(weaker) + " is off while the baseline requires it. "
                                 + ("Git plus the cluster already disagree: Argo CD reports OutOfSync "
                                    "at " + str(finding["syncedRevision"])[:7] + " on "
                                    + str(finding["targetRevision"]) + " with nothing reconciling it."
                                    if finding["driftPending"]
                                    else "Nothing has drifted yet, so only the configuration is at risk."))

          print(f"{finding['verdict']}: {name} project={finding['project']} "
                f"observed={json.dumps(observed, sort_keys=True)} "
                f"declared={json.dumps(want, sort_keys=True)} sync={sync_status} "
                f"revision={str(finding['syncedRevision'])[:7]}. {finding['note']}")
          print("::" + json.dumps({"outputs": {"finding": json.dumps(finding)}}) + "::")

  - id: build_plan
    type: io.kestra.plugin.scripts.python.Script
    description: >-
      Collect every finding, order the restorable ones worst first, apply the
      per run ceiling, then record the approved snapshot for each target. The
      snapshot is what the re-read after the pause is compared against.
    containerImage: python:3.13-slim
    inputFiles:
      loop.json: "{{ outputs.read_policies.outputs | toJson }}"
      exempt.json: "{{ outputs.plan_scope.vars.exempt | toJson }}"
    env:
      MAX_APPS: "{{ inputs.max_apps_per_run }}"
    script: |
      import json
      import os

      with open("loop.json") as handle:
          iterations = json.load(handle)
      with open("exempt.json") as handle:
          exempt = json.load(handle)

      max_apps = int(os.environ["MAX_APPS"])
      findings = [json.loads(row["outputs"]["finding"]) for row in iterations]
      findings.sort(key=lambda row: (not row["driftPending"], -len(row["weakerFields"]), row["application"]))

      restorable = [row for row in findings if row["verdict"] == "restorable"]
      targets = restorable[:max_apps]
      for row in restorable[max_apps:]:
          row["verdict"] = "over_ceiling"
          row["note"] = (f"Held back by the {max_apps} application per run ceiling. "
                         "The next run picks it up.")

      snapshots = []
      for row in targets:
          snapshots.append({
              "application": row["application"],
              "project": row["project"],
              "repoURL": row["repoURL"],
              "path": row["path"],
              "targetRevision": row["targetRevision"],
              "observed": row["observed"],
              "declared": row["declared"],
              "weakerFields": row["weakerFields"],
              "driftPending": row["driftPending"],
              "syncedRevision": row["syncedRevision"],
              "rollback": row["rollback"],
              "patch": json.dumps({"spec": {"syncPolicy": {"automated": {
                  "prune": row["declared"]["prune"],
                  "selfHeal": row["declared"]["selfHeal"],
              }}}}),
          })

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

      lines = [f"{row['verdict']}: {row['application']} project={row['project']} "
               f"off={','.join(row['weakerFields']) or 'none'} sync={row['syncStatus']} "
               f"cluster_at={str(row['syncedRevision'])[:7]}. {row['note']}" for row in findings]
      lines += [f"exempt_annotation: {row['application']} skipped, {row['reason']}" for row in exempt]
      for line in lines:
          print(line)
      print(f"audited={len(findings)} targets={len(targets)} "
            f"buckets={json.dumps(counts, sort_keys=True)}")

      print("::" + json.dumps({"outputs": {
          "findings": findings,
          "targets": snapshots,
          "targetCount": len(snapshots),
          "targetNames": [row["application"] for row in snapshots],
          "auditedCount": len(findings),
          "matchesBaseline": counts.get("matches_baseline", 0),
          "strongerThanBaseline": counts.get("stronger_than_baseline", 0),
          "projectNotAllowed": counts.get("project_not_allowed", 0),
          "overCeiling": counts.get("over_ceiling", 0),
          "exemptCount": len(exempt),
          "driftPendingCount": len([row for row in snapshots if row["driftPending"]]),
          "report": "\n".join(lines) if lines else "No Application is in scope.",
      }}) + "::")

  - id: log_findings
    type: io.kestra.plugin.core.log.Log
    description: One line per audited application plus the reason it is or is not a target.
    message: |
      Audited {{ outputs.build_plan.vars.auditedCount }} Application(s) in namespace '{{ inputs.argocd_namespace }}'.
      targets={{ outputs.build_plan.vars.targetCount }} out_of_sync_targets={{ outputs.build_plan.vars.driftPendingCount }} matches_baseline={{ outputs.build_plan.vars.matchesBaseline }} stronger_than_baseline={{ outputs.build_plan.vars.strongerThanBaseline }} project_not_allowed={{ outputs.build_plan.vars.projectNotAllowed }} over_ceiling={{ outputs.build_plan.vars.overCeiling }} exempt_annotation={{ outputs.build_plan.vars.exemptCount }}
      {{ outputs.build_plan.vars.report }}

  - id: announce_findings
    type: io.kestra.plugin.core.flow.If
    description: Everything below runs only when at least one application has a
      weaker policy than declared.
    condition: "{{ outputs.build_plan.vars.targetCount > 0 or
      outputs.build_plan.vars.projectNotAllowed > 0 }}"
    then:
      - id: notify_findings
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: >-
          One message per audit naming each application, which policy field is
          off, the commit the cluster is stuck on plus whether this flow is
          allowed to touch it.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": ":arrows_counterclockwise: {{ outputs.build_plan.vars.targetCount }} Argo CD Application(s) have a sync policy weaker than declared.{% for row in outputs.build_plan.vars.findings %}\n• `{{ row.application }}` project `{{ row.project }}` off=`{{ row.weakerFields | join(' ') }}` sync `{{ row.syncStatus }}` verdict *{{ row.verdict }}*. {{ row.note }}{% endfor %}"
          }

  - id: route_remediation
    type: io.kestra.plugin.core.flow.If
    description: >-
      A patch needs three things true at once. auto_remediate allows remediation
      to be considered, dry_run false allows it to be written plus there has to
      be an application that cleared the opt out annotation, the project
      allowlist, the stronger than baseline 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.build_plan.vars.targetCount > 0 }}"
    then:
      - id: approve_restore
        type: io.kestra.plugin.core.flow.Pause
        description: >-
          Re-enabling automated sync hands the cluster back to Git immediately.
          Read the findings first: if the cluster was changed by hand during the
          incident, the first reconciliation will overwrite those changes, plus
          prune will delete anything that is no longer in Git. Confirm the
          branch really holds the state you want before approving.
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: >-
              Set true to restore the listed policies. Anything else leaves
              every application untouched.

      - id: apply_if_approved
        type: io.kestra.plugin.core.flow.If
        description: An unanswered pause reads as a refusal, never as approval.
        condition: "{{ (outputs.approve_restore.onResume.approved ?? false) == true }}"
        then:
          - id: restore_each
            type: io.kestra.plugin.core.flow.Loop
            description: >-
              One application at a time. Argo CD starts reconciling the moment a
              policy is restored, so a parallel loop would mix one application's
              recovery window with the next one's patch.
            values: "{{ outputs.build_plan.vars.targets }}"
            concurrencyLimit: 1
            fetchType: FETCH
            outputs:
              - id: result
                type: STRING
                value: "{{ outputs.assert_recovery.vars.result ??
                  outputs.guard_stale_plan.vars.result }}"
            tasks:
              - id: recheck_target
                type: io.kestra.plugin.argocd.apps.Status
                description: >-
                  Read the application again, immediately before its write. The
                  approval was given against a snapshot that is already old by
                  the time a human answers.
                server: "{{ inputs.argocd_server }}"
                token: "{{ inputs.argocd_token }}"
                application: "{{ fromJson(item.value).application }}"
                argoCDVersion: "{{ inputs.argocd_cli_version }}"
                plaintext: true
                refresh: true
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: "{{ inputs.cli_docker_network }}"

              - id: guard_stale_plan
                type: io.kestra.plugin.scripts.python.Script
                description: >-
                  Compare the fresh read against the approved snapshot. Already
                  fixed by someone else is a skip, not a write. Anything else
                  that moved, the project, the repository, the path or the
                  branch, fails the run before a single patch is sent.
                containerImage: python:3.13-slim
                inputFiles:
                  app.json: "{{ outputs.recheck_target.rawOutput }}"
                env:
                  SNAPSHOT: "{{ item.value }}"
                script: |
                  import json
                  import os
                  import sys

                  with open("app.json") as handle:
                      app = json.load(handle)
                  snapshot = json.loads(os.environ["SNAPSHOT"])

                  spec = app.get("spec") or {}
                  source = spec.get("source") or {}
                  automated = (spec.get("syncPolicy") or {}).get("automated")
                  observed = {
                      "automated": automated is not None,
                      "prune": bool((automated or {}).get("prune")),
                      "selfHeal": bool((automated or {}).get("selfHeal")),
                  }
                  identity = {
                      "project": spec.get("project"),
                      "repoURL": source.get("repoURL"),
                      "path": source.get("path"),
                      "targetRevision": source.get("targetRevision"),
                  }
                  approved_identity = {key: snapshot.get(key) for key in identity}
                  moved = sorted(key for key in identity if identity[key] != approved_identity[key])
                  want = snapshot["declared"]
                  still_weaker = sorted(key for key in want if want[key] and not observed[key])

                  result = {
                      "application": snapshot["application"],
                      "observedBeforeWrite": observed,
                      "identityNow": identity,
                      "identityApproved": approved_identity,
                  }

                  if moved:
                      result["verdict"] = "changed_refused"
                      print("::" + json.dumps({"outputs": {"result": json.dumps(result), "verdict": "changed_refused"}}) + "::")
                      print(f"refused: {snapshot['application']} changed during the approval window. "
                            f"{', '.join(moved)} now {json.dumps({k: identity[k] for k in moved})} "
                            f"instead of {json.dumps({k: approved_identity[k] for k in moved})}. "
                            "Nothing was patched. Re-run the audit to plan against the current state.",
                            file=sys.stderr)
                      sys.exit(1)

                  if not still_weaker:
                      result["verdict"] = "already_compliant"
                      result["landed"] = True
                      print(f"skip: {snapshot['application']} already carries the declared policy "
                            f"{json.dumps(observed, sort_keys=True)}. No patch sent.")
                  else:
                      result["verdict"] = "proceed"
                      result["weakerFields"] = still_weaker
                      print(f"proceed: {snapshot['application']} still has {', '.join(still_weaker)} off. "
                            f"Patch {snapshot['patch']}. Rollback if needed: {snapshot['rollback']}")

                  print("::" + json.dumps({"outputs": {
                      "result": json.dumps(result),
                      "verdict": result["verdict"],
                  }}) + "::")

              - id: write_if_eligible
                type: io.kestra.plugin.core.flow.If
                description: Only a target that is still weaker than its baseline is patched.
                condition: "{{ outputs.guard_stale_plan.vars.verdict == 'proceed' }}"
                then:
                  - id: restore_policy
                    type: io.kestra.plugin.argocd.apps.Patch
                    description: >-
                      argocd app patch with a merge patch that sets
                      spec.syncPolicy.automated to the declared values. A merge
                      patch leaves the rest of the policy alone, so syncOptions
                      plus any allowEmpty setting survive.
                    server: "{{ inputs.argocd_server }}"
                    token: "{{ inputs.argocd_token }}"
                    application: "{{ fromJson(item.value).application }}"
                    argoCDVersion: "{{ inputs.argocd_cli_version }}"
                    plaintext: true
                    patchType: MERGE
                    patch: "{{ fromJson(item.value).patch }}"
                    taskRunner:
                      type: io.kestra.plugin.scripts.runner.docker.Docker
                      networkMode: "{{ inputs.cli_docker_network }}"

                  - id: await_recovery
                    type: io.kestra.plugin.core.flow.LoopUntil
                    description: >-
                      Watch the restored policy do the work. The application is
                      read again every 15 seconds, for up to ten reads inside a
                      three minute window, until the policy is on the object
                      plus Argo CD reports the application reconciled. Only
                      reads happen in here, never the patch, so nothing can be
                      written twice. No sync is issued either: if the commit
                      arrives it arrived because automated sync is running
                      again. When the window closes without recovery,
                      failOnMaxReached ends the iteration as failed, which is
                      the case where the patch was accepted yet changed nothing.
                    condition: "{{ outputs.measure_recovery.vars.recovered == 'yes' }}"
                    checkFrequency:
                      interval: PT15S
                      maxIterations: 10
                      maxDuration: PT3M
                    failOnMaxReached: true
                    tasks:
                      - id: poll_status
                        type: io.kestra.plugin.argocd.apps.Status
                        server: "{{ inputs.argocd_server }}"
                        token: "{{ inputs.argocd_token }}"
                        application: "{{ fromJson(item.value).application }}"
                        argoCDVersion: "{{ inputs.argocd_cli_version }}"
                        plaintext: true
                        refresh: true
                        taskRunner:
                          type: io.kestra.plugin.scripts.runner.docker.Docker
                          networkMode: "{{ inputs.cli_docker_network }}"

                      - id: measure_recovery
                        type: io.kestra.plugin.scripts.python.Script
                        description: >-
                          Decide whether this read already proves recovery. The
                          policy has to be on the object plus an application
                          that was out of sync has to be reconciled. A policy
                          that reads back correctly yet changes nothing in the
                          cluster does not count, which is the whole difference
                          between a setting and a fix. This task only measures,
                          the loop condition plus assert_recovery decide the
                          outcome.
                        containerImage: python:3.13-slim
                        inputFiles:
                          app.json: "{{ outputs.poll_status.rawOutput }}"
                        env:
                          SNAPSHOT: "{{ item.value }}"
                        script: |
                          import json
                          import os

                          with open("app.json") as handle:
                              app = json.load(handle)
                          snapshot = json.loads(os.environ["SNAPSHOT"])

                          spec = app.get("spec") or {}
                          status = app.get("status") or {}
                          sync = status.get("sync") or {}
                          automated = (spec.get("syncPolicy") or {}).get("automated")
                          observed = {
                              "automated": automated is not None,
                              "prune": bool((automated or {}).get("prune")),
                              "selfHeal": bool((automated or {}).get("selfHeal")),
                          }
                          want = snapshot["declared"]
                          missing = sorted(key for key in want if want[key] and not observed[key])

                          revision = sync.get("revision")
                          sync_status = sync.get("status")
                          reconciled = sync_status == "Synced"
                          advanced = revision != snapshot["syncedRevision"]
                          needed_recovery = bool(snapshot["driftPending"])
                          recovered = not missing and (reconciled or not needed_recovery)

                          result = {
                              "application": snapshot["application"],
                              "observedAfterWrite": observed,
                              "declared": want,
                              "restoredFields": snapshot["weakerFields"],
                              "syncStatus": sync_status,
                              "healthStatus": (status.get("health") or {}).get("status"),
                              "revisionBefore": snapshot["syncedRevision"],
                              "revisionAfter": revision,
                              "driftPending": needed_recovery,
                              "revisionAdvanced": bool(needed_recovery and advanced),
                              "reconciledWithoutManualSync": bool(needed_recovery and reconciled),
                              "landed": not missing,
                              "recovered": recovered,
                              "missingFields": missing,
                              "rollback": snapshot["rollback"],
                              "claim": (("policy restored plus the stranded commit "
                                         + str(revision)[:7] + " reached the cluster with no manual sync")
                                        if needed_recovery and reconciled and advanced else
                                        ("policy restored plus the drifted resources were reconciled "
                                         "back to Git at " + str(revision)[:7] + " with no manual sync")
                                        if needed_recovery and reconciled else
                                        "policy restored only, nothing had drifted at plan time"),
                          }

                          if missing:
                              print(f"not recovered yet: {snapshot['application']} still has "
                                    f"{', '.join(missing)} off after the patch reported success. "
                                    f"Observed {json.dumps(observed, sort_keys=True)}. If this is the "
                                    "last read, something is overwriting this Application, an "
                                    "ApplicationSet or a parent app of apps owns it. Roll back with: "
                                    + snapshot["rollback"])
                          elif needed_recovery and not reconciled:
                              print(f"not recovered yet: {snapshot['application']} carries the policy "
                                    f"again yet Argo CD reports {sync_status} at {str(revision)[:7]}. "
                                    "Waiting for the restored policy to reconcile.")
                          else:
                              print(f"recovered: {snapshot['application']} policy "
                                    f"{json.dumps(observed, sort_keys=True)}, {sync_status} at "
                                    f"{str(revision)[:7]} from {str(snapshot['syncedRevision'])[:7]}. "
                                    f"{result['claim']}.")

                          print("::" + json.dumps({"outputs": {
                              "result": json.dumps(result),
                              "recovered": "yes" if recovered else "no",
                          }}) + "::")

                  - id: assert_recovery
                    type: io.kestra.plugin.scripts.python.Script
                    description: >-
                      Restate what the last read measured plus fail this
                      application if it is not compliant, so an application is
                      never reported as restored on the strength of the patch
                      exit code alone.
                    containerImage: python:3.13-slim
                    inputFiles:
                      measured.json: "{{ outputs.measure_recovery.vars.result }}"
                    script: |
                      import json
                      import sys

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

                      print("::" + json.dumps({"outputs": {"result": json.dumps(result)}}) + "::")

                      if not result.get("recovered"):
                          print(f"recovery not proven: {result['application']} ended with policy "
                                f"{json.dumps(result['observedAfterWrite'], sort_keys=True)} plus sync "
                                f"status {result['syncStatus']}. Missing fields "
                                f"{result['missingFields']}. Roll back with: {result['rollback']}",
                                file=sys.stderr)
                          sys.exit(1)
                      print(f"verified: {result['application']} policy "
                            f"{json.dumps(result['observedAfterWrite'], sort_keys=True)}, "
                            f"{result['syncStatus']} at {str(result['revisionAfter'])[:7]}. "
                            f"{result['claim']}.")

                else:
                  - id: log_skipped_target
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      {{ fromJson(item.value).application }} was left alone: {{
                      outputs.guard_stale_plan.vars.verdict }}. The re-read
                      before the write found the declared policy already in
                      place.

          - id: summarize_restores
            type: io.kestra.plugin.scripts.python.Script
            description: >-
              Aggregate what every iteration actually observed after its write
              plus fail the run when any approved target did not end up
              compliant. The per application checks already fail hard, this is
              the fleet level statement the operator reads.
            containerImage: python:3.13-slim
            inputFiles:
              loop.json: "{{ outputs.restore_each.outputs | toJson }}"
            env:
              TARGET_NAMES: "{{ outputs.build_plan.vars.targetNames | toJson }}"
            script: |
              import json
              import os
              import sys

              with open("loop.json") as handle:
                  iterations = json.load(handle)
              planned = json.loads(os.environ["TARGET_NAMES"])

              results = [json.loads(row["outputs"]["result"]) for row in iterations
                         if (row.get("outputs") or {}).get("result")]
              by_app = {row["application"]: row for row in results}
              landed = [name for name in planned if by_app.get(name, {}).get("landed")]
              reconciled = [name for name in planned
                            if by_app.get(name, {}).get("reconciledWithoutManualSync")]
              skipped = [name for name in planned
                         if by_app.get(name, {}).get("verdict") == "already_compliant"]
              missing = [name for name in planned if name not in by_app]
              failed = [name for name in planned if name in by_app and not by_app[name].get("landed")]

              for name in planned:
                  row = by_app.get(name)
                  if not row:
                      print(f"{name}: NO RESULT, the iteration never reported")
                      continue
                  print(f"{name}: landed={row.get('landed')} policy={json.dumps(row.get('observedAfterWrite') or row.get('observedBeforeWrite'), sort_keys=True)} "
                        f"sync={row.get('syncStatus')} revision={str(row.get('revisionBefore'))[:7]}->{str(row.get('revisionAfter'))[:7]} "
                        f"{row.get('claim', 'skipped, already compliant')}")
              print(f"planned={len(planned)} landed={len(landed)} reconciled={len(reconciled)} "
                    f"skipped_already_compliant={len(skipped)} no_result={len(missing)}")

              print("::" + json.dumps({"outputs": {
                  "results": results,
                  "landed": landed,
                  "landedCount": len(landed),
                  "reconciled": reconciled,
                  "reconciledCount": len(reconciled),
                  "skipped": skipped,
                  "noResult": missing,
                  "failed": failed,
              }}) + "::")

              if missing or failed:
                  print("fleet re-read failed: "
                        f"{len(failed)} approved target(s) are still not compliant {failed} plus "
                        f"{len(missing)} never reported {missing}.", file=sys.stderr)
                  sys.exit(1)

          - id: log_restored
            type: io.kestra.plugin.core.log.Log
            message: >-
              Restored the declared sync policy on {{
              outputs.summarize_restores.vars.landedCount }} of {{
              outputs.build_plan.vars.targetCount }} approved Application(s): {{
              outputs.summarize_restores.vars.landed | join(' ') }}. {{
              outputs.summarize_restores.vars.reconciledCount }} of them were
              reconciled back to Git by Argo CD itself, with no sync issued by
              this flow: {{ outputs.summarize_restores.vars.reconciled | join('
              ') }}.
        else:
          - id: log_declined
            type: io.kestra.plugin.core.log.Log
            message: >-
              Approval declined or left unanswered, so nothing was patched. {{
              outputs.build_plan.vars.targetCount }} Application(s) still have a
              policy weaker than declared plus {{
              outputs.build_plan.vars.driftPendingCount }} of them are already
              out of sync with Git.
    else:
      - id: log_report_only
        type: io.kestra.plugin.core.log.Log
        message: >-
          Report only, nothing was written. audited={{
          outputs.build_plan.vars.auditedCount }} targets={{
          outputs.build_plan.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 re-read that found the policy missing
      after a patch reported success plus a target that changed during the
      approval window.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": ":rotating_light: argocd-sync-policy-reenable-guard failed on namespace `{{ inputs.argocd_namespace }}`. Execution {{ execution.id }}. If the failure came from the re-read, the patch was accepted yet the policy is not on the Application, so check for an ApplicationSet or a parent app of apps that owns it before the next run."
      }