Org-Wide Dependency Vulnerability Triage with Approval and Expiring Waivers

id: dependency-vulnerability-triage-with-expiring-waivers
namespace: company.team

description: |
  Org-wide dependency vulnerability triage against OSV.dev. One file plays three roles (`mode`): orchestrate fans out by calling itself in scan mode, scan checks one lockfile, replay retries the dead-letter list.
  HIGH and CRITICAL findings block until a human approves a waiver, stored with a KV TTL so it expires on its own.

inputs:
  - id: mode
    type: SELECT
    displayName: Mode
    values: [ orchestrate, scan, replay ]
    defaults: orchestrate
    description: orchestrate = triage every lockfile in `repos`. replay = triage
      only the lockfiles in the dead-letter list. scan = one lockfile, findings
      only (the flow calls itself in this mode).

  - id: repos
    type: JSON
    displayName: Lockfiles to triage (orchestrate)
    description: JSON list of {"repo", "lockfile_url"}. `repo` must be unique and is
      used as the dead-letter key, so use owner/name:path when one repository
      has several lockfiles. Supported files are npm package-lock.json (v1, v2,
      v3) and pinned pip requirements files. The defaults are public
      repositories pinned at fixed commits, so the results are reproducible.
    defaults: |
      [
        {"repo": "anxolerd/dvpwa", "lockfile_url": "https://raw.githubusercontent.com/anxolerd/dvpwa/5ee2dc23502bcedcbc9ca1f6fce21f9137d4fe8c/requirements.txt"},
        {"repo": "adeyosemanputra/pygoat", "lockfile_url": "https://raw.githubusercontent.com/adeyosemanputra/pygoat/aaf8b545aba83c12b43a2f2c47ba33934d135deb/requirements.txt"},
        {"repo": "django/djangoproject.com", "lockfile_url": "https://raw.githubusercontent.com/django/djangoproject.com/4b2a648aa6ed2f0681a4d777d0de7d812bdf8f26/requirements/prod.txt"},
        {"repo": "OWASP/NodeGoat", "lockfile_url": "https://raw.githubusercontent.com/OWASP/NodeGoat/9336e34a7c0a440a24c20807469523bd0c1ee8aa/package-lock.json"},
        {"repo": "expressjs/expressjs.com", "lockfile_url": "https://raw.githubusercontent.com/expressjs/expressjs.com/0eff3bf2856b980b4428fd9b268604f68af1f38e/package-lock.json"}
      ]

  - id: repo
    type: STRING
    displayName: Repository (scan)
    required: false
    description: Set by the orchestrator in scan mode.

  - id: lockfile_url
    type: STRING
    displayName: Lockfile URL (scan)
    required: false
    description: Set by the orchestrator in scan mode. Must be a raw https URL, not
      a GitHub /blob/ page.

  - id: block_at
    type: SELECT
    displayName: Block at severity
    values: [ CRITICAL, HIGH, MEDIUM ]
    defaults: HIGH
    description: Findings at or above this severity block unless a live waiver
      covers them. UNKNOWN severity is reported but never blocks.

  - id: include_dev
    type: BOOL
    displayName: Include npm devDependencies
    defaults: false
    description: npm lockfiles flag build-only packages as dev. They are skipped by
      default because they do not ship to production.

  - id: approval_timeout
    type: STRING
    displayName: Approval timeout
    defaults: PT4H
    validator: ^(P[0-9]+D|PT[0-9]+[HMS])$
    description: How long the approval step waits. When it expires without a
      decision the run is treated as REJECT. Keep it shorter than the SLA below.

variables:
  repos_problem: >-
    {{ inputs.repos | jq('if type != "array" or length == 0 or length > 200 then
    "repos must be a JSON list of 1 to 200 objects" elif any(.[]; (.repo | type)
    != "string" or (.repo | test("^[A-Za-z0-9][A-Za-z0-9._/@:-]*$") | not)) then
    "every entry needs a repo made of letters, digits and . _ / @ : -" elif
    any(.[]; (.lockfile_url | type) != "string" or (.lockfile_url |
    test("^https://[^ ]+$") | not)) then "every lockfile_url must be an https
    URL" elif ([.[].repo] | unique | length) != length then "each repo must
    appear once, because it is the dead-letter key (use owner/name:path for
    several lockfiles)" else "" end') | first }}

  # errorLogs() also returns the ERROR lines of allowFailure Slack tasks, so they are skipped when naming the failed task.
  failed_task: >-
    {{ errorLogs() | jq('[.[] | select(.taskId != null and (.taskId |
    test("^(request_approval|slack_)") | not))] | (last // {}) | .taskId //
    "unknown"') | first }}
  failure_message: >-
    {{ errorLogs() | jq('[.[] | select(.taskId != null and (.taskId |
    test("^(request_approval|slack_)") | not))] as $e | ($e | last | .taskId) as
    $t | [$e[] | select(.taskId == $t) | .message][0] // "see the logs"') |
    first | abbreviate(300) }}

checks:
  - when: "{{ inputs.mode != 'orchestrate' or render(vars.repos_problem) is empty }}"
    message: "Invalid repos input. Send a JSON list of 1 to 200 objects, each with a
      unique `repo` and an https `lockfile_url`."
    style: ERROR
    behavior: BLOCK_EXECUTION
  - when: "{{ inputs.mode != 'scan' or ({'r': inputs.repo ?? '', 'u':
      inputs.lockfile_url ?? ''} | jq('(.r | length > 0) and (.u |
      test(\"^https://[^ ]+$\"))') | first) }}"
    message: scan mode needs `repo` and an https `lockfile_url`.
    style: ERROR
    behavior: BLOCK_EXECUTION

concurrency:
  limit: 6
  behavior: QUEUE

sla:
  - id: triage_deadline
    type: MAX_DURATION
    duration: PT8H
    behavior: FAIL
    labels:
      sla: breached

triggers:
  - id: lockfile_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('VULN_TRIAGE_WEBHOOK_KEY') }}"
    description: 'POST {"repos": [{"repo": ..., "lockfile_url": ...}]} or the bare
      list, from CI or a repository inventory job. A bad body fails in
      validate_input within a second, because webhook executions skip flow-level
      checks.'
    inputs:
      mode: orchestrate
      repos: "{{ trigger.body.repos ?? trigger.body }}"

  - id: weekly_org_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: Every Monday 06:00 UTC, re-triage the default list. Lockfiles do
      not change but OSV.dev does, and waivers that expired during the week
      block again.
    cron: "0 6 * * 1"
    timezone: UTC

tasks:
  - id: route_by_mode
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ inputs.mode }}"
    cases:
      scan:
        - id: scan_lockfile
          type: io.kestra.plugin.core.flow.Sequential
          description: One lockfile. The local `errors` branch writes a dead-letter entry
            and lets the execution fail, so the orchestrator sees FAILED and
            moves on.
          tasks:
            - id: download_lockfile
              type: io.kestra.plugin.core.http.Download
              description: Hard fail after retries. A lockfile that cannot be fetched cannot
                be judged, and calling it clean would be wrong.
              uri: "{{ inputs.lockfile_url }}"
              headers:
                User-Agent: kestra-blueprint-dependency-vulnerability-triage/1.0
              timeout: PT1M
              retry:
                type: exponential
                interval: PT3S
                maxInterval: PT30S
                maxAttempts: 3

            - id: scan_osv
              type: io.kestra.plugin.scripts.python.Script
              description: |
                Standard library only. Parses the lockfile, sends one OSV querybatch per 1,000 packages and follows next_page_token, then fetches each distinct advisory once for severity, aliases and fixed versions.
                No task-level retry on purpose: a parse error is permanent and goes straight to the dead-letter list. Each OSV call retries 429 and 5xx with exponential backoff inside the script.
              timeout: PT15M
              taskRunner:
                type: io.kestra.plugin.scripts.runner.docker.Docker
              containerImage: python:3.13-slim
              inputFiles:
                lockfile: "{{ outputs.download_lockfile.uri }}"
              outputFiles:
                - findings.jsonl
              env:
                REPO: "{{ inputs.repo }}"
                INCLUDE_DEV: "{{ inputs.include_dev }}"
              script: |
                import json, os, re, sys, time, urllib.request, urllib.error
                from concurrent.futures import ThreadPoolExecutor

                OSV = "https://api.osv.dev/v1"
                HEADERS = {"Content-Type": "application/json", "User-Agent": "kestra-blueprint-dependency-vulnerability-triage/1.0"}
                repo, include_dev = os.environ["REPO"], os.environ["INCLUDE_DEV"].lower() == "true"
                text = open("lockfile", encoding="utf-8", errors="replace").read()

                def parse_error(msg):
                    print("PARSE_ERROR " + msg, file=sys.stderr)
                    sys.exit(2)

                def call(url, body=None, attempts=4):
                    for n in range(attempts):
                        try:
                            req = urllib.request.Request(url, data=json.dumps(body).encode() if body else None, headers=HEADERS)
                            with urllib.request.urlopen(req, timeout=30) as r:
                                return json.loads(r.read())
                        except urllib.error.HTTPError as e:
                            if e.code not in (429, 500, 502, 503, 504) or n == attempts - 1:
                                raise
                        except (urllib.error.URLError, TimeoutError):
                            if n == attempts - 1:
                                raise
                        time.sleep(2 ** n)

                pkgs, skipped = set(), {"dev": 0, "unpinned": 0, "url_or_path": 0, "include": 0, "linked": 0}
                head = text.lstrip()[:1]
                if head == "<":
                    parse_error("the URL returned HTML, not a lockfile. Use the raw file URL (raw.githubusercontent.com), not the /blob/ page.")
                if head == "{":
                    ecosystem = "npm"
                    try:
                        lock = json.loads(text)
                    except ValueError as e:
                        parse_error("invalid package-lock.json: %s" % e)
                    if "packages" in lock:  # lockfileVersion 2 and 3, flat map keyed by node_modules path
                        for path, meta in lock["packages"].items():
                            if not path:
                                continue
                            if meta.get("link"):
                                skipped["linked"] += 1
                            elif meta.get("dev") and not include_dev:
                                skipped["dev"] += 1
                            elif re.match(r"^\d", meta.get("version", "")):
                                pkgs.add((meta.get("name") or path.rsplit("node_modules/", 1)[-1], meta["version"]))
                    elif "dependencies" in lock:  # lockfileVersion 1, nested tree
                        stack = list(lock["dependencies"].items())
                        while stack:
                            name, meta = stack.pop()
                            stack.extend(meta.get("dependencies", {}).items())
                            if meta.get("dev") and not include_dev:
                                skipped["dev"] += 1
                            elif re.match(r"^\d", meta.get("version", "")):
                                pkgs.add((name, meta["version"]))
                    else:
                        parse_error("JSON file without `packages` or `dependencies`; not an npm package-lock.json.")
                else:
                    ecosystem = "PyPI"
                    for raw in re.sub(r"\\\r?\n", " ", text).splitlines():
                        line = re.split(r"(?:^|\s)#", raw, maxsplit=1)[0].strip()
                        if not line:
                            continue
                        if line.startswith("-"):
                            skipped["include"] += line.startswith(("-r", "-c", "--requirement", "--constraint"))
                            continue
                        spec = line.split(";")[0].split(" --hash")[0].strip()
                        if "@" in spec or spec.startswith((".", "/", "git+", "http")):
                            skipped["url_or_path"] += 1
                            continue
                        m = re.match(r"^([A-Za-z0-9][A-Za-z0-9._-]*)\s*(?:\[[^\]]*\])?\s*===?\s*([0-9][A-Za-z0-9.+!_-]*)$", spec)
                        if m:
                            pkgs.add((re.sub(r"[-_.]+", "-", m.group(1)).lower(), m.group(2)))  # PEP 503 name
                        else:
                            skipped["unpinned"] += 1
                    if not pkgs:
                        parse_error("no pinned name==version lines (%d unpinned). Point at a compiled lockfile such as pip-compile or uv export output." % skipped["unpinned"])

                keys, hits, osv_requests = sorted(pkgs), {}, 0
                pending = [(k, None) for k in keys]
                while pending:
                    batch, pending = pending[:1000], pending[1000:]
                    queries = [dict({"package": {"name": k[0], "ecosystem": ecosystem}, "version": k[1]}, **({"page_token": t} if t else {})) for k, t in batch]
                    results = call(OSV + "/querybatch", {"queries": queries})["results"]
                    osv_requests += 1
                    for (k, _), res in zip(batch, results):
                        hits.setdefault(k, set()).update(v["id"] for v in res.get("vulns", []))
                        if res.get("next_page_token"):
                            pending.append((k, res["next_page_token"]))

                ids = sorted({i for s in hits.values() for i in s})
                with ThreadPoolExecutor(max_workers=8) as pool:
                    records = dict(zip(ids, pool.map(lambda i: call(OSV + "/vulns/" + i), ids)))

                findings = 0
                with open("findings.jsonl", "w") as out:
                    for (name, version), vids in sorted(hits.items()):
                        for vid in sorted(vids):
                            rec = records[vid]
                            if rec.get("withdrawn"):
                                continue
                            sev = ((rec.get("database_specific") or {}).get("severity") or "").upper()
                            sev = {"MODERATE": "MEDIUM"}.get(sev, sev) or ("CRITICAL" if vid.startswith("MAL-") else "UNKNOWN")
                            fixed = sorted({e["fixed"] for a in rec.get("affected", [])
                                            if (a.get("package") or {}).get("name", "").lower() == name.lower()
                                            for r in a.get("ranges", []) for e in r.get("events", []) if "fixed" in e})
                            out.write(json.dumps({"repo": repo, "ecosystem": ecosystem, "package": name, "version": version,
                                                  "id": vid, "aliases": rec.get("aliases", []), "severity": sev,
                                                  "fixed": ", ".join(fixed[:3]),
                                                  "summary": (rec.get("summary") or rec.get("details") or "")[:160]}) + "\n")
                            findings += 1

                stats = {"repo": repo, "ecosystem": ecosystem, "packages": len(keys), "skipped": skipped, "osv_requests": osv_requests,
                         "advisories": len(ids), "withdrawn": sum(1 for r in records.values() if r.get("withdrawn")), "findings": findings}
                print("::" + json.dumps({"outputs": {"stats": stats}}) + "::")

            - id: clear_dead_letter
              type: io.kestra.plugin.core.kv.Delete
              description: A successful scan removes any earlier dead-letter entry for this
                repo, so replays and fixed URLs heal the list.
              key: "vulntriage_dlq_{{ inputs.repo | lower | replace({'/': '__', ':': '_', '@':
                '_'}) }}"
              errorOnMissing: false
          errors:
            - id: record_dead_letter
              type: io.kestra.plugin.core.kv.Set
              description: One key per repo, so parallel scans never overwrite each other. The
                attempt counter survives replays.
              key: "vulntriage_dlq_{{ inputs.repo | lower | replace({'/': '__', ':': '_', '@':
                '_'}) }}"
              kvType: JSON
              value: >-
                {{ {
                  'repo': inputs.repo,
                  'lockfile_url': inputs.lockfile_url,
                  'reason': (render(vars.failed_task) ~ '') == 'download_lockfile' ? 'UNREACHABLE' : ((errorLogs() | jq('any(.[]; .message | contains("PARSE_ERROR"))') | first) ? 'PARSE_ERROR' : 'OSV_ERROR'),
                  'failed_task': render(vars.failed_task),
                  'error': render(vars.failure_message),
                  'attempts': ((kv('vulntriage_dlq_' ~ (inputs.repo | lower | replace({'/': '__', ':': '_', '@': '_'})), errorOnMissing=false) ?? {}).attempts ?? 0) + 1,
                  'last_failed_at': now() | date("yyyy-MM-dd'T'HH:mm:ss'Z'", timeZone='UTC'),
                  'execution_id': execution.id
                } | toJson }}

      orchestrate:
        - id: validate_input
          type: io.kestra.plugin.core.execution.Fail
          description: Same rule as the flow check. It is repeated here because executions
            created by the Webhook trigger do not run flow-level checks on
            Kestra 2.0.5.
          condition: "{{ render(vars.repos_problem) is not empty }}"
          errorMessage: "Invalid repos input: {{ render(vars.repos_problem) }}."

        - id: build_targets
          type: io.kestra.plugin.core.debug.Return
          description: Copies include_dev into every item, because Loop iterations do not
            reliably see the parent's inputs.
          format: >-
            [{% for r in inputs.repos %}{{ {'repo': r.repo, 'lockfile_url':
            r.lockfile_url, 'include_dev': inputs.include_dev} | toJson }}{% if
            not loop.last %},{% endif %}{% endfor %}]

        - id: fan_out
          type: io.kestra.plugin.core.flow.Loop
          description: One child execution per lockfile, three at a time to stay polite to
            OSV.dev and GitHub. transmitFailed false on the Subflow isolates
            failures; each iteration reports its state instead.
          values: "{{ outputs.build_targets.value }}"
          concurrencyLimit: 3
          outputs:
            - id: scan
              type: JSON
              value: "{{ {'repo': fromJson(item.value).repo, 'state': outputs.scan_repo.state
                ?? 'UNKNOWN', 'execution_id': outputs.scan_repo.executionId ??
                '', 'report': outputs.scan_repo.outputs.report ?? '', 'stats':
                outputs.scan_repo.outputs.summary ?? {}} | toJson }}"
          tasks:
            - id: scan_repo
              type: io.kestra.plugin.core.flow.Subflow
              namespace: "{{ flow.namespace }}"
              flowId: "{{ flow.id }}"
              wait: true
              transmitFailed: false
              inputs:
                mode: scan
                repo: "{{ fromJson(item.value).repo }}"
                lockfile_url: "{{ fromJson(item.value).lockfile_url }}"
                include_dev: "{{ fromJson(item.value).include_dev }}"

        - id: dead_letters
          type: io.kestra.plugin.core.kv.GetKeys
          prefix: vulntriage_dlq_

        - id: require_one_scan
          type: io.kestra.plugin.core.execution.Fail
          description: Partial results are fine, no results are not. If every lockfile
            failed, stop here rather than report a clean organisation.
          condition: "{{ (outputs.fan_out.outputs | jq('[.[].outputs.scan | select(.state
            == \"SUCCESS\")] | length') | first) == 0 }}"
          errorMessage: "None of the {{ inputs.repos | length }} lockfiles could be
            scanned. They are in the dead-letter list ({{
            outputs.dead_letters.keys | join(', ') }}). Fix the URLs or OSV.dev
            access, then run mode=replay."

        - id: merge_findings
          type: io.kestra.plugin.core.storage.Concat
          description: Joins the JSONL files that the successful child executions returned
            as outputs.
          files: "{{ outputs.fan_out.outputs | jq('[.[].outputs.scan | select(.state ==
            \"SUCCESS\" and .report != \"\") | .report]') | first | toJson }}"
          extension: .jsonl

        - id: live_waivers
          type: io.kestra.plugin.core.kv.GetKeys
          description: Expired waivers are already gone from this list. That is the whole
            expiry mechanism.
          prefix: vulntriage_waiver_

        - id: triage
          type: io.kestra.plugin.jdbc.duckdb.Queries
          description: Dedupes the same advisory across alias ids (CVE, GHSA, PYSEC) and
            across repos, applies the severity policy and live waivers, compares
            with the previous run and writes the triage CSV.
          timeout: PT5M
          inputFiles:
            findings.jsonl: "{{ outputs.merge_findings.uri }}"
            waivers.json: "{{ outputs.live_waivers.keys | toJson }}"
            previous.json: "{{ (kv('vulntriage_open_findings', errorOnMissing=false) ?? [])
              | toJson }}"
            scanned.json: "{{ outputs.fan_out.outputs | jq('[.[].outputs.scan |
              select(.state == \"SUCCESS\") | .repo]') | first | toJson }}"
          outputFiles:
            - triage.csv
          fetchType: FETCH
          sql: |
            SET TimeZone = 'UTC';
            CREATE TABLE raw AS SELECT *, list_distinct(list_prepend(id, coalesce(aliases, []))) AS all_ids
            FROM read_json('{{ workingDir }}/findings.jsonl', format = 'newline_delimited',
              columns = {repo: 'VARCHAR', ecosystem: 'VARCHAR', package: 'VARCHAR', version: 'VARCHAR', id: 'VARCHAR',
                         aliases: 'VARCHAR[]', severity: 'VARCHAR', fixed: 'VARCHAR', summary: 'VARCHAR'});
            -- One advisory id per alias group: the lowest CVE, else the lowest GHSA, else the OSV id.
            -- The finding key (advisory + ecosystem + package) is also the waiver key.
            CREATE TABLE canon AS
            SELECT *, 'vulntriage_waiver_' || lower(regexp_replace(advisory || '_' || ecosystem || '_' || package, '[^A-Za-z0-9]+', '_', 'g')) AS waiver_key
            FROM (SELECT *,
                    coalesce(list_sort(list_filter(all_ids, x -> x LIKE 'CVE-%'))[1], list_sort(list_filter(all_ids, x -> x LIKE 'GHSA-%'))[1], id) AS advisory,
                    CASE severity WHEN 'CRITICAL' THEN 4 WHEN 'HIGH' THEN 3 WHEN 'MEDIUM' THEN 2 WHEN 'LOW' THEN 1 ELSE 0 END AS rank
                  FROM raw);
            CREATE TABLE waivers AS SELECT unnest(json_extract_string(content, '$[*]')) AS waiver_key FROM read_text('{{ workingDir }}/waivers.json');
            CREATE TABLE scanned AS SELECT unnest(json_extract_string(content, '$[*]')) AS repo FROM read_text('{{ workingDir }}/scanned.json');
            CREATE TABLE previous AS SELECT json_extract_string(f, '$.repo') AS repo, json_extract_string(f, '$.key') AS waiver_key
              FROM (SELECT unnest(json_extract(content, '$[*]')) AS f FROM read_text('{{ workingDir }}/previous.json'));
            CREATE TABLE grouped AS
            SELECT advisory, ecosystem, package, waiver_key,
                   CASE max(rank) WHEN 4 THEN 'CRITICAL' WHEN 3 THEN 'HIGH' WHEN 2 THEN 'MEDIUM' WHEN 1 THEN 'LOW' ELSE 'UNKNOWN' END AS severity,
                   max(rank) AS rank,
                   string_agg(DISTINCT repo, ' ' ORDER BY repo) AS repos,
                   count(DISTINCT repo) AS repo_count,
                   string_agg(DISTINCT version, ' ' ORDER BY version) AS versions,
                   array_to_string(list_sort(list_distinct(flatten(list(all_ids)))), ' ') AS osv_ids,
                   max(fixed) AS fixed,
                   max_by(summary, CASE WHEN id LIKE 'GHSA-%' THEN 1 ELSE 0 END) AS summary
            FROM canon GROUP BY advisory, ecosystem, package, waiver_key;
            CREATE TABLE report AS
            SELECT g.*, g.rank >= {{ inputs.block_at == 'CRITICAL' ? 4 : (inputs.block_at == 'HIGH' ? 3 : 2) }} AS blocking,
                   g.waiver_key IN (SELECT waiver_key FROM waivers) AS waived,
                   CASE WHEN g.waiver_key IN (SELECT waiver_key FROM previous) THEN 'KNOWN' ELSE 'NEW' END AS status
            FROM grouped g;
            COPY (SELECT * EXCLUDE (rank) FROM report ORDER BY rank DESC, repo_count DESC, advisory) TO '{{ outputFiles["triage.csv"] }}' (HEADER);
            SELECT (SELECT count(*) FROM scanned) AS repos_scanned,
                   (SELECT count(*) FROM raw) AS raw_findings,
                   count(*) AS advisories,
                   count(*) FILTER (WHERE repo_count > 1) AS shared_across_repos,
                   count(*) FILTER (WHERE severity = 'CRITICAL') AS critical,
                   count(*) FILTER (WHERE severity = 'HIGH') AS high,
                   count(*) FILTER (WHERE blocking) AS blocking,
                   count(*) FILTER (WHERE blocking AND waived) AS waived,
                   count(*) FILTER (WHERE blocking AND NOT waived) AS unwaived_blocking,
                   count(*) FILTER (WHERE status = 'NEW') AS new_since_last_run,
                   (SELECT count(DISTINCT p.waiver_key) FROM previous p WHERE p.repo IN (SELECT repo FROM scanned)
                      AND NOT EXISTS (SELECT 1 FROM canon c WHERE c.repo = p.repo AND c.waiver_key = p.waiver_key)) AS resolved_since_last_run
            FROM report;
            SELECT advisory, ecosystem, package, severity, repos, versions, fixed, waiver_key, summary
            FROM report WHERE blocking AND NOT waived ORDER BY rank DESC, repo_count DESC, advisory;
            SELECT CAST(coalesce(json_group_array(json_object('repo', repo, 'key', waiver_key)), '[]') AS VARCHAR) AS snapshot FROM (
              SELECT DISTINCT repo, waiver_key FROM canon
              UNION SELECT repo, waiver_key FROM previous WHERE repo NOT IN (SELECT repo FROM scanned));

        - id: save_snapshot
          type: io.kestra.plugin.core.kv.Set
          description: Open findings per repo for the NEW and resolved counts next time.
            Repos that failed this run keep their previous rows, so a
            dead-lettered repo never looks fixed.
          key: vulntriage_open_findings
          kvType: JSON
          value: "{{ outputs.triage.outputs[2].rows[0].snapshot }}"

        - id: log_triage
          type: io.kestra.plugin.core.log.Log
          message: |
            Triage over {{ outputs.triage.outputs[0].rows[0].repos_scanned }} of {{ inputs.repos | length }} lockfiles (block at {{ inputs.block_at }}): {{ outputs.triage.outputs[0].rows[0] | toJson }}
            Dead-letter list: {{ outputs.dead_letters.keys | length }} {{ outputs.dead_letters.keys | toJson }}
            Report: {{ outputs.triage.outputFiles['triage.csv'] }}

        - id: approval_gate
          type: io.kestra.plugin.core.flow.If
          description: Ask a human only for blocking findings that no live waiver covers.
          condition: "{{ outputs.triage.outputs[0].rows[0].unwaived_blocking > 0 }}"
          then:
            - id: request_approval
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              allowFailure: true
              description: allowFailure keeps the approval step reachable when Slack is not
                configured. The Pause shows up in the UI either way.
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              messageText: |
                Dependency triage needs a decision: {{ outputs.triage.outputs[0].rows[0].unwaived_blocking }} {{ inputs.block_at }}+ advisories have no live waiver.
                {{ outputs.triage.outputs[1].rows | jq('[.[:10][] | "- " + .severity + " " + .advisory + " in " + .package + " (" + .repos + ")"] | join("\n")') | first }}
                Decide within {{ inputs.approval_timeout }} (no answer = REJECT): {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

            - id: wait_for_decision
              type: io.kestra.plugin.core.flow.Pause
              description: behavior RESUME on timeout continues without onResume values, and
                apply_decision reads that as REJECT.
              pauseDuration: "{{ inputs.approval_timeout }}"
              behavior: RESUME
              onResume:
                - id: decision
                  type: SELECT
                  values: [ REJECT, WAIVE ]
                  defaults: REJECT
                  description: WAIVE stores a time-boxed exception for the listed advisories.
                    REJECT keeps them blocking.
                - id: waive_only
                  type: STRING
                  required: false
                  description: Optional comma-separated advisory ids (as shown in the report) to
                    waive. Empty = all blocking advisories in this run.
                - id: waiver_ttl
                  type: STRING
                  defaults: P30D
                  validator: ^(P[0-9]+D|PT[0-9]+[HMS])$
                  description: How long the waiver lives, for example P30D or P7D. It is deleted
                    automatically when it expires.
                - id: approver
                  type: STRING
                  description: Who decided. Stored with every waiver.
                - id: justification
                  type: STRING
                  description: Why. Stored with every waiver and shown in the failure message on
                    REJECT.

            - id: apply_decision
              type: io.kestra.plugin.graalvm.js.Eval
              description: Turns the decision into waiver records. Runs in the worker JVM, so
                it needs no container.
              outputs: [ decision, waived, remaining, waivers_json, remaining_text, unknown_ids ]
              script: |
                (function () {
                  const resume = {{ (outputs.wait_for_decision.onResume ?? {}) | toJson }};
                  const blocking = {{ outputs.triage.outputs[1].rows | toJson }};
                  const decision = resume.decision ? resume.decision : 'REJECT_TIMEOUT';
                  const only = (resume.waive_only || '').split(',').map((s) => s.trim().toUpperCase()).filter((s) => s);
                  const picked = decision === 'WAIVE' ? blocking.filter((r) => only.length === 0 || only.indexOf(r.advisory.toUpperCase()) >= 0) : [];
                  const waivers = picked.map((r) => ({
                    key: r.waiver_key,
                    ttl: resume.waiver_ttl || 'P30D',
                    value: JSON.stringify({ advisory: r.advisory, package: r.package, ecosystem: r.ecosystem, severity: r.severity,
                      repos: r.repos, approver: resume.approver, justification: resume.justification, ttl: resume.waiver_ttl,
                      approved_at: new Date().toISOString(), execution_id: {{ execution.id | toJson }} })
                  }));
                  const remaining = blocking.filter((r) => picked.indexOf(r) < 0);
                  return {
                    decision: decision,
                    waived: waivers.length,
                    remaining: remaining.length,
                    waivers_json: JSON.stringify(waivers),
                    remaining_text: remaining.slice(0, 15).map((r) => r.severity + ' ' + r.advisory + ' in ' + r.package).join(', '),
                    unknown_ids: only.filter((id) => !blocking.some((r) => r.advisory.toUpperCase() === id)).join(', ')
                  };
                })()

            - id: log_decision
              type: io.kestra.plugin.core.log.Log
              level: "{{ outputs.apply_decision.outputs.unknown_ids == '' ? 'INFO' : 'WARN'
                }}"
              message: "Decision {{ outputs.apply_decision.outputs.decision }}: {{
                outputs.apply_decision.outputs.waived }} waived{{
                outputs.apply_decision.outputs.waived > 0 ? ' for ' ~
                outputs.wait_for_decision.onResume.waiver_ttl : '' }}, {{
                outputs.apply_decision.outputs.remaining }} still blocking.{{
                outputs.apply_decision.outputs.unknown_ids == '' ? '' : '
                Ignored ids that are not blocking in this run: ' ~
                outputs.apply_decision.outputs.unknown_ids }}"

            - id: write_waivers
              type: io.kestra.plugin.core.flow.Loop
              description: One KV key per advisory and package, each with its own TTL.
                Re-approving the same advisory refreshes the key, so this is
                safe to repeat.
              runIf: "{{ outputs.apply_decision.outputs.waived > 0 }}"
              values: "{{ outputs.apply_decision.outputs.waivers_json }}"
              concurrencyLimit: 5
              tasks:
                - id: store_waiver
                  type: io.kestra.plugin.core.kv.Set
                  key: "{{ fromJson(item.value).key }}"
                  value: "{{ fromJson(item.value).value }}"
                  kvType: JSON
                  ttl: "{{ fromJson(item.value).ttl }}"

            - id: enforce_policy
              type: io.kestra.plugin.core.execution.Fail
              description: The gate. A deliberate failure that CI can read, not an error, so
                the errors branch skips its alert for it.
              condition: "{{ outputs.apply_decision.outputs.remaining > 0 }}"
              errorMessage: "Blocked: {{ outputs.apply_decision.outputs.remaining }}
                advisories at or above {{ inputs.block_at }} have no waiver
                (decision {{ outputs.apply_decision.outputs.decision }}{{
                outputs.wait_for_decision.onResume.approver is defined ? ' by '
                ~ outputs.wait_for_decision.onResume.approver ~ ': ' ~
                outputs.wait_for_decision.onResume.justification : '' }}). {{
                outputs.apply_decision.outputs.remaining_text }}"
          else:
            - id: all_covered
              type: io.kestra.plugin.core.log.Log
              message: "No decision needed: {{ outputs.triage.outputs[0].rows[0].blocking }}
                blocking advisories, all covered by live waivers."

      replay:
        - id: dead_letter_entries
          type: io.kestra.plugin.core.kv.GetKeys
          prefix: vulntriage_dlq_

        - id: replay_gate
          type: io.kestra.plugin.core.flow.If
          condition: "{{ outputs.dead_letter_entries.keys | length > 0 }}"
          then:
            - id: replay_orchestration
              type: io.kestra.plugin.core.flow.Subflow
              description: Replay is orchestrate over the dead-letter list, so it gets the
                same fan-out, policy and approval. Entries that scan cleanly
                delete themselves.
              namespace: "{{ flow.namespace }}"
              flowId: "{{ flow.id }}"
              wait: true
              transmitFailed: true
              inputs:
                mode: orchestrate
                block_at: "{{ inputs.block_at }}"
                include_dev: "{{ inputs.include_dev }}"
                approval_timeout: "{{ inputs.approval_timeout }}"
                repos: "[{% for k in outputs.dead_letter_entries.keys %}{{ {'repo': kv(k).repo,
                  'lockfile_url': kv(k).lockfile_url} | toJson }}{% if not
                  loop.last %},{% endif %}{% endfor %}]"
          else:
            - id: nothing_to_replay
              type: io.kestra.plugin.core.log.Log
              message: The dead-letter list is empty. Nothing to replay.

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.core.flow.If
    description: Scan children are already captured by the dead-letter list, and a
      gate rejection (remaining > 0) is the policy working as designed, so
      neither pages anyone. errorLogs() is only read inside log_failure and
      slack_failure, because flowable conditions are evaluated outside the
      worker, where errorLogs() is not available.
    condition: "{{ inputs.mode == 'orchestrate' and
      (outputs.apply_decision.outputs.remaining ?? 0) == 0 }}"
    then:
      - id: log_failure
        type: io.kestra.plugin.core.log.Log
        level: ERROR
        message: "Dependency triage failed in task {{ render(vars.failed_task) }}: {{
          render(vars.failure_message) }}"
      - id: slack_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        allowFailure: true
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Dependency triage ({{ inputs.mode }}) failed in task {{ render(vars.failed_task) }}: {{ render(vars.failure_message) }}
          {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

finally:
  - id: run_summary
    type: io.kestra.plugin.core.log.Log
    description: Runs on success, gate failure and errors (not after an SLA kill;
      afterExecution covers that).
    message: >-
      {% if inputs.mode == 'scan' %}scan {{ inputs.repo }}: {{
      (outputs.scan_osv.vars.stats ?? 'failed, see the dead-letter entry') |
      toJson }} {%- elseif inputs.mode == 'replay' %}replay of {{
      outputs.dead_letter_entries.keys | length }} dead-letter entries {%- else
      %}orchestrate: {{ (outputs.fan_out.outputs ?? []) | jq('[.[].outputs.scan
      | .repo + "=" + .state + "(" + ((.stats.findings // "-") | tostring) +
      ")"] | join(", ")') | first }}; decision {{
      outputs.apply_decision.outputs.decision ?? 'NOT_REQUIRED' }}, waivers
      written {{ outputs.apply_decision.outputs.waived ?? 0 }}{% endif %}

afterExecution:
  - id: notify_outcome
    type: io.kestra.plugin.core.flow.If
    description: Runs once the final state is known, including after an SLA breach,
      when finally and errors do not run.
    condition: "{{ inputs.mode == 'orchestrate' }}"
    then:
      - id: slack_outcome
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        allowFailure: true
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          Dependency triage finished {{ execution.state }}{{ (labels.sla ?? '') == 'breached' ? ' (stopped by the PT8H SLA)' : '' }}.
          {{ (outputs.triage.outputs[0].rows[0] ?? {}) | toJson }}
          {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

outputs:
  - id: report
    type: STRING
    description: Internal-storage URI. scan = raw findings JSONL for one lockfile;
      orchestrate = deduplicated triage CSV.
    value: "{{ outputs.triage.outputFiles['triage.csv'] ??
      outputs.scan_osv.outputFiles['findings.jsonl'] ?? '' }}"
  - id: summary
    type: JSON
    value: "{{ outputs.triage.outputs[0].rows[0] ?? outputs.scan_osv.vars.stats ??
      {} }}"
  - id: decision
    type: STRING
    value: "{{ outputs.apply_decision.outputs.decision ?? 'NOT_REQUIRED' }}"