Monitor ClinVar genes for variant classification changes

id: clinvar-variant-reclassification-monitor
namespace: company.team

concurrency:
  limit: 1

inputs:
  - id: genes
    type: ARRAY
    itemType: STRING
    displayName: Gene symbols to monitor
    description: HGNC gene symbols to look up in ClinVar. Each gene is processed one
      at a time and keeps its own snapshot in the KV store. Symbols are
      case-insensitive, so BRCA1 and brca1 are the same gene, and a gene listed
      more than once is processed once. An entry that is not a single symbol of
      letters, digits, '.', '_' or '-' stops the run before any request.
    defaults:
      - BRCA1
      - BRCA2
      - TP53
      - EGFR

  - id: overlap_days
    type: INT
    displayName: Overlap in days
    description: Incremental runs fetch records modified since the last checkpoint
      minus this many days. The overlap covers late ClinVar updates and
      day-level date granularity. Duplicates are harmless because every record
      is compared with the snapshot.
    defaults: 7
    min: 1
    max: 60

  - id: reconcile_days
    type: INT
    displayName: Full scan interval in days
    description: A full scan of the gene runs instead of the incremental scan when
      the last one is at least this old. It also logs how many of its changes
      fall outside the date window an incremental run would have used. The first
      run for a gene is always a full scan.
    defaults: 28
    min: 1
    max: 365

  - id: extra_query
    type: STRING
    displayName: Optional extra ClinVar search clause
    description: 'Optional ClinVar search syntax ANDed to every gene query, for
      example a review status or condition filter. ESearch uses the same query
      language as the ClinVar website, so test the clause there first. Changing
      it later requires a new state_key_prefix. Leave empty to monitor every
      record of the gene.'
    defaults: ""

  - id: classifications_to_monitor
    type: ARRAY
    itemType: STRING
    displayName: Classifications to monitor
    description: 'Optional filter on germline classification names as ClinVar writes
      them, for example Pathogenic or Likely pathogenic. A change or new variant
      is reported only when the old or new classification is in this list. Leave
      empty to report every change.'
    defaults: []

  - id: alert_on_new_variants
    type: BOOL
    displayName: Alert on new variants
    description: Also report variants that were not in the stored snapshot. They are
      always listed separately from reclassifications. The first run only builds
      the baseline and never alerts.
    defaults: true

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post findings to Slack when something changed. Needs the
      SLACK_WEBHOOK_URL secret. Findings are always written to the execution
      logs.
    defaults: true

  - id: contact_email
    type: STRING
    displayName: Contact email for NCBI
    description: NCBI asks API users to identify themselves. When set, it is sent as
      the email parameter so NCBI can contact you instead of blocking the
      traffic.
    defaults: ""
    validator: '^$|^[^@\s&=?#]+@[^@\s&=?#]+$'

  - id: state_key_prefix
    type: STRING
    displayName: KV key prefix for snapshots
    description: Snapshots are stored under <prefix>_<gene> in the KV store of the
      flow namespace. Use a different prefix for every copy of this flow in the
      same namespace, because copies with different queries must not share
      snapshots.
    defaults: clinvar_snapshot
    validator: '^[A-Za-z0-9][A-Za-z0-9._-]*$'

  - id: fail_if_snapshot_missing
    type: BOOL
    displayName: Fail when a snapshot is missing
    description: Leave off for the first run, which builds the baseline. Turn it on
      afterwards so a deleted or expired snapshot stops the run instead of
      silently starting a new baseline and hiding the changes in between.
    defaults: false

  - id: use_ncbi_api_key
    type: BOOL
    displayName: Use an NCBI API key
    description: Send the NCBI_API_KEY secret with every request, which raises the
      E-utilities limit from 3 to 10 requests per second. Leave off to run
      without a key.
    defaults: false

triggers:
  - id: weekly_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Runs every Tuesday at 06:00 UTC. ClinVar updates its web site on
      Mondays, so Tuesday avoids racing the release. The overlap window also
      covers a late update.
    cron: "0 6 * * 2"
    timezone: UTC

tasks:
  - id: validate_genes
    type: io.kestra.plugin.core.execution.Fail
    description: Stop before any request is made when a gene entry is blank or
      contains unusual characters, so a typo cannot create an empty or
      meaningless snapshot. The characters are exactly those Kestra allows in a
      KV key after the prefix, because the key is <prefix>_<gene in lower case>.
      Two checks, either of which stops the run. First, each valid symbol
      collapses to a single g, so only a list of valid symbols reduces to
      exactly g. Second, the number of g left must equal the number of list
      entries, so one entry that itself contains a comma (BRCA1,TP53) cannot
      pass as two symbols.
    condition: "{{ ((((inputs.genes | join(',')) | replace({'[A-Za-z0-9._-]+': 'g'},
      regexp=true)) | replace({'g,': ''})) != 'g') or ((((((inputs.genes |
      join(',')) | replace({'[A-Za-z0-9._-]+': 'g'}, regexp=true)) |
      replace({',': ''})) | length) ~ '') != ((inputs.genes | length) ~ '')) }}"
    errorMessage: "Invalid gene list. Every entry must be a non-empty gene symbol
      made of letters, digits, '.', '_' or '-', without spaces, so that the KV
      key built from it is valid."

  - id: watch_genes
    type: io.kestra.plugin.core.flow.Loop
    description: One gene at a time. Serial iteration keeps the run under the NCBI
      limit of 3 requests per second without an API key.
    values: "{{ inputs.genes | toJson | lower }}"
    concurrencyLimit: 1
    outputs:
      - id: gene_summary
        type: STRING
        value: "{{ outputs.compare_snapshots.vars.summary }}"
    tasks:
      - id: read_snapshot
        type: io.kestra.plugin.core.kv.Get
        description: The snapshot stored by the previous run for this gene. A missing
          key means this is the first run, so the baseline is built instead of
          alerting.
        key: "{{ inputs.state_key_prefix }}_{{ item.value | trim | lower }}"
        errorOnMissing: false

      - id: compare_snapshots
        type: io.kestra.plugin.scripts.python.Script
        description: Queries ClinVar through NCBI E-utilities (paged ESearch, ESummary
          batches of 200, retries with backoff, rate limiting), compares the
          records with the stored snapshot, and writes the next snapshot. It
          runs a full scan for the baseline and when a full scan is due,
          otherwise only records modified since the checkpoint minus the
          overlap. Any API surprise fails the run instead of being read as no
          changes.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        timeout: PT45M
        warningOnStdErr: false
        env:
          NCBI_API_KEY: "{{ inputs.use_ncbi_api_key ? secret('NCBI_API_KEY') : '' }}"
        inputFiles:
          snapshot.json: "{{ outputs.read_snapshot.value ?? '' }}"
          config.json: |
            {
              "gene": "{{ item.value | trim }}",
              "classifications": {{ (inputs.classifications_to_monitor ?? []) | toJson }},
              "alert_on_new_variants": {{ inputs.alert_on_new_variants }},
              "extra_query": {{ ((inputs.extra_query ?? '') | trim) | toJson }},
              "contact_email": {{ ((inputs.contact_email ?? '') | trim) | toJson }},
              "fail_if_snapshot_missing": {{ inputs.fail_if_snapshot_missing }},
              "overlap_days": {{ inputs.overlap_days }},
              "reconcile_days": {{ inputs.reconcile_days }},
              "max_lines": 15
            }
        outputFiles:
          - snapshot_out.json
          - report.json
        script: |
          import json
          import os
          import time
          import urllib.error
          import urllib.parse
          import urllib.request
          from datetime import date, datetime, timedelta, timezone
          from pathlib import Path


          # Only NCBI_API_KEY is set by the flow. The other environment variables exist so the unit tests can
          # point the script at a local mock server; they are unset in production.
          EUTILS = os.environ.get("EUTILS_BASE_URL", "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/")
          API_KEY = os.environ.get("NCBI_API_KEY", "").strip()
          TIMEOUT = float(os.environ.get("EUTILS_TIMEOUT", "60"))
          ATTEMPTS = int(os.environ.get("EUTILS_ATTEMPTS", "4"))
          BACKOFF = float(os.environ.get("EUTILS_BACKOFF", "2"))
          INTERVAL = float(os.environ.get("EUTILS_MIN_INTERVAL", "0.12" if API_KEY else "0.4"))
          LINK = "https://www.ncbi.nlm.nih.gov/clinvar/variation/%s/"
          SCHEMA = 2
          SEARCH_PAGE = 400
          SUMMARY_BATCH = 200
          MAX_SNAPSHOT_BYTES = int(os.environ.get("SNAPSHOT_MAX_BYTES", "8000000"))
          MAX_SLACK_CHARS = 3900
          OMITTED_RESERVE = 120
          MAX_TITLE_CHARS = 160
          DISCLAIMER = "_Research monitoring only. Aggregate germline classifications can change for many reasons and are not clinical advice._"


          def fail(message):
              raise SystemExit(message)


          cfg = json.loads(Path("config.json").read_text())
          gene = cfg["gene"].strip().upper()
          monitored = {c.strip().lower() for c in cfg.get("classifications") or [] if c and c.strip()}
          alert_on_new = bool(cfg.get("alert_on_new_variants", True))
          query = (cfg.get("extra_query") or "").strip()
          email = (cfg.get("contact_email") or "").strip()
          require_snapshot = bool(cfg.get("fail_if_snapshot_missing", False))
          overlap_days = int(cfg.get("overlap_days", 7))
          reconcile_days = int(cfg.get("reconcile_days", 28))
          max_lines = max(1, int(cfg.get("max_lines", 15)))
          if overlap_days < 1 or reconcile_days < 1:
              fail("overlap_days and reconcile_days must be at least 1")
          now_dt = datetime.now(timezone.utc)
          now_iso = now_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
          now_txt = now_dt.strftime("%Y-%m-%d %H:%M UTC")
          today = date.fromisoformat(os.environ["CLINVAR_TODAY"]) if os.environ.get("CLINVAR_TODAY") else now_dt.date()


          # ---- 1. E-utilities client: URL-encoded parameters, rate limit, bounded retries with backoff.
          # ---- A response is returned only when it is a JSON object; anything else is retried and then fails.
          last_request = [0.0]
          requests_made = [0]


          def rate_limited(data):
              # NCBI can answer HTTP 200 with a rate-limit error in the JSON body instead of HTTP 429.
              messages = [data.get("error")]
              result = data.get("esearchresult")
              if isinstance(result, dict):
                  messages.append(result.get("ERROR"))
              return any(isinstance(m, str) and ("rate limit" in m.lower() or "too many requests" in m.lower()) for m in messages)


          def capped_delay(delay, retry_after):
              # Retry-After in seconds only (the HTTP-date form is ignored), never longer than 60 seconds.
              if retry_after and str(retry_after).strip().isdigit():
                  return max(delay, min(int(str(retry_after).strip()), 60))
              return delay


          def call(endpoint, params):
              query_params = {"db": "clinvar", "retmode": "json", "tool": "kestra-clinvar-monitor"}
              if email:
                  query_params["email"] = email
              if API_KEY:
                  query_params["api_key"] = API_KEY
              query_params.update(params)
              url = EUTILS + endpoint + "?" + urllib.parse.urlencode(query_params)
              # The label never contains the API key or the long id list.
              label = "%s %s" % (endpoint, ", ".join("%s=%s" % (k, str(v)[:80]) for k, v in sorted(params.items()) if k != "id"))
              problem = "unknown error"
              for attempt in range(1, ATTEMPTS + 1):
                  wait = INTERVAL - (time.monotonic() - last_request[0])
                  if wait > 0:
                      time.sleep(wait)
                  last_request[0] = time.monotonic()
                  requests_made[0] += 1
                  delay = BACKOFF * (2 ** (attempt - 1))
                  try:
                      request = urllib.request.Request(url, headers={"User-Agent": "kestra-clinvar-monitor"})
                      with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
                          retry_after = response.headers.get("Retry-After")
                          data = json.loads(response.read().decode("utf-8"))
                      if not isinstance(data, dict):
                          raise ValueError("not a JSON object")
                      if not rate_limited(data):
                          return data
                      problem = "rate limit reported in the response body"  # the body itself is never printed
                      delay = capped_delay(delay, retry_after)
                  except urllib.error.HTTPError as exc:
                      problem = "HTTP %d" % exc.code
                      if exc.code not in (429, 500, 502, 503, 504):
                          fail("%s failed: %s" % (label, problem))
                      delay = capped_delay(delay, exc.headers.get("Retry-After") if exc.headers else None)
                  except (urllib.error.URLError, OSError) as exc:
                      problem = "network error (%s)" % type(exc).__name__
                  except ValueError:
                      problem = "response is not valid JSON"
                  if attempt < ATTEMPTS:
                      print("  retry %d of %d for %s after %s" % (attempt, ATTEMPTS - 1, label, problem))
                      time.sleep(delay)
              fail("%s failed after %d attempts: %s" % (label, ATTEMPTS, problem))


          def search_page(term, start):
              data = call("esearch.fcgi", {"term": term, "retmax": SEARCH_PAGE, "retstart": start})
              result = data.get("esearchresult")
              if not isinstance(result, dict):
                  fail("ESearch response has no 'esearchresult' object; the API format may have changed")
              if result.get("ERROR"):
                  fail("ESearch returned an error: %s" % result["ERROR"])
              errors = result.get("errorlist") or {}
              if errors.get("phrasesnotfound") or errors.get("fieldsnotfound"):
                  fail("ClinVar does not recognise part of the query for %s: %s. Check the gene symbol and the extra query."
                       % (gene, errors.get("phrasesnotfound") or errors.get("fieldsnotfound")))
              warnings = result.get("warninglist") or {}
              if warnings.get("phrasesignored") or warnings.get("quotedphrasesnotfound"):
                  fail("ClinVar ignored part of the query for %s: %s"
                       % (gene, warnings.get("phrasesignored") or warnings.get("quotedphrasesnotfound")))
              try:
                  count = int(result.get("count"))
              except (TypeError, ValueError):
                  fail("ESearch returned no valid count; the API format may have changed")
              return count, [str(i) for i in result.get("idlist") or []]


          def search_ids(term):
              ids, seen, start, first_count, pages = [], set(), 0, None, 0
              while True:
                  count, page = search_page(term, start)
                  if first_count is None:
                      first_count = count
                  pages += 1
                  for uid in page:
                      if uid not in seen:
                          seen.add(uid)
                          ids.append(uid)
                  start += SEARCH_PAGE
                  if start >= count:
                      break
                  if not page:
                      fail("ESearch returned an empty page at retstart=%d although %d records match" % (start - SEARCH_PAGE, count))
                  if pages > count // SEARCH_PAGE + 10:
                      fail("ESearch paging did not finish after %d pages" % pages)
              if len(ids) < min(first_count, count):
                  fail("ESearch paging was incomplete: %d unique ids for %d matching records. Run again; "
                       "ClinVar may have changed during the scan" % (len(ids), min(first_count, count)))
              return ids, count


          def normalize(uid, rec):
              germline = rec.get("germline_classification") or {}
              return {
                  "variation_id": uid,
                  "accession": rec.get("accession_version") or rec.get("accession"),
                  "title": rec.get("title"),
                  "classification": (germline.get("description") or "").strip() or None,
                  "review_status": (germline.get("review_status") or "").strip() or None,
              }


          def fetch_records(ids):
              # A record that ESummary cannot summarise fails the whole run (the snapshot and checkpoint stay as they were),
              # so it can never silently drop out of future monitoring. A record that stays unsummarisable blocks the gene.
              records, unusable = {}, []
              for n in range(0, len(ids), SUMMARY_BATCH):
                  batch = ids[n:n + SUMMARY_BATCH]
                  data = call("esummary.fcgi", {"id": ",".join(batch)})
                  if data.get("error"):
                      fail("ESummary returned an error: %s" % data["error"])
                  result = data.get("result")
                  if not isinstance(result, dict):
                      fail("ESummary response has no 'result' object; the API format may have changed")
                  returned = {str(u) for u in result.get("uids") or []}
                  missing = [uid for uid in batch if uid not in returned]
                  if missing:
                      fail("ESummary did not return %d of %d requested records (first missing: %s)" % (len(missing), len(batch), missing[0]))
                  for uid in batch:
                      rec = result.get(uid)
                      if not isinstance(rec, dict) or rec.get("error"):
                          unusable.append(uid)
                      else:
                          records[uid] = normalize(uid, rec)
              if unusable:
                  fail("ESummary could not summarise %d record(s) (first: %s). The run fails so the snapshot and checkpoint stay "
                       "unchanged and the next run covers the same period again" % (len(unusable), ", ".join(unusable[:5])))
              return records


          # ---- 2. Snapshot schema 2: one compact entry per variant, [classification index, review index].
          # ----   {"schema": 2, "gene", "query", "checkpoint", "reconciled", "updated", "classes", "reviews", "variants"}
          def parse_day(value, label):
              try:
                  return date.fromisoformat(value)
              except (TypeError, ValueError):
                  fail("Stored snapshot is corrupted: %s is not a date" % label)


          def decode(snap):
              try:
                  classes, reviews, packed = snap["classes"], snap["reviews"], snap["variants"]
                  if not isinstance(classes, list) or not isinstance(reviews, list) or not isinstance(packed, dict):
                      raise ValueError("bad shape")
                  out = {}
                  for uid, pair in packed.items():
                      ci, ri = pair
                      out[str(uid)] = (classes[ci] if ci >= 0 else None, reviews[ri] if ri >= 0 else None)
                  return out
              except (KeyError, ValueError, TypeError, IndexError):
                  fail("Stored snapshot is corrupted; delete the KV key to re-baseline")


          def encode(variants, checkpoint, reconciled):
              classes, reviews, cidx, ridx, packed = [], [], {}, {}, {}

              def index(value, table, lookup):
                  if value is None:
                      return -1
                  if value not in lookup:
                      lookup[value] = len(table)
                      table.append(value)
                  return lookup[value]

              for uid, (cls, review) in variants.items():
                  packed[uid] = [index(cls, classes, cidx), index(review, reviews, ridx)]
              return {"schema": SCHEMA, "gene": gene, "query": query, "checkpoint": checkpoint.isoformat(),
                      "reconciled": reconciled.isoformat(), "updated": now_iso,
                      "classes": classes, "reviews": reviews, "variants": packed}


          # ---- 3. Decide between baseline, incremental scan and full reconciliation.
          # An empty snapshot.json means the KV key does not exist. Anything else must be a valid, non-empty snapshot
          # object: a stored [], 0, false, null or {} is corruption, never a reason to start a new baseline.
          stored_text = Path("snapshot.json").read_text().strip()
          snapshot = None
          if stored_text:
              try:
                  snapshot = json.loads(stored_text)
              except ValueError as exc:
                  fail("Stored snapshot is not valid JSON: %s" % exc)
              if not isinstance(snapshot, dict) or not snapshot:
                  fail("Stored snapshot has an unexpected structure (found %s); delete the KV key to re-baseline" % stored_text[:40])
          previous, baseline, mode, reason = {}, False, "FULL", ""
          last_reconciled, checkpoint = today, None
          if snapshot is None:
              if require_snapshot:
                  fail("No stored snapshot for %s and fail_if_snapshot_missing is on; "
                       "turn it off for one run to build a new baseline" % gene)
              baseline, reason = True, "no stored snapshot"
          else:
              if not isinstance(snapshot, dict) or snapshot.get("gene") != gene or not isinstance(snapshot.get("variants"), dict):
                  fail("Stored snapshot is corrupted or belongs to another gene; delete the KV key to re-baseline")
              if (snapshot.get("query") or "") != query:
                  fail("Stored snapshot was built with a different extra query (%r, now %r); "
                       "use another state_key_prefix or delete the KV key to re-baseline" % (snapshot.get("query") or "", query))
              schema = snapshot.get("schema")
              if schema == 1:
                  # Schema 1 keyed variants as "GENE:uid" with full records; keep the classifications.
                  try:
                      previous = {k.split(":", 1)[1]: (v.get("classification"), v.get("review_status"))
                                  for k, v in snapshot["variants"].items()}
                  except (IndexError, AttributeError):
                      fail("Stored snapshot is corrupted; delete the KV key to re-baseline")
                  reason = "migrating snapshot schema 1 to 2"
              elif schema == SCHEMA:
                  previous = decode(snapshot)
                  checkpoint = parse_day(snapshot.get("checkpoint"), "checkpoint")
                  last_reconciled = parse_day(snapshot.get("reconciled"), "reconciled")
                  if checkpoint > today or last_reconciled > today:
                      fail("Stored snapshot is corrupted: its checkpoint is in the future")
                  age = (today - last_reconciled).days
                  if age >= reconcile_days:
                      reason = "full scan due (%d days since the last one)" % age
                  else:
                      mode = "INCREMENTAL"
              else:
                  fail("Stored snapshot has unsupported schema %r; delete the KV key to re-baseline" % (schema,))

          base_term = "%s[gene]" % gene + (" AND (%s)" % query if query else "")

          def date_window():
              # The inclusive [mdat] range an incremental run uses. The end is tomorrow so a record touched today is never cut off.
              return "%s:%s" % ((checkpoint - timedelta(days=overlap_days)).strftime("%Y/%m/%d"),
                                (today + timedelta(days=1)).strftime("%Y/%m/%d"))


          window = date_window() if mode == "INCREMENTAL" else ""
          term = "%s AND %s[mdat]" % (base_term, window) if window else base_term
          print("%s: %s scan (%s)%s" % (gene, mode, reason or "last checkpoint %s" % checkpoint, " window " + window if window else ""))

          # ---- 4. Fetch. An API failure, a bad page or a missing record stops the run before anything is stored.
          ids, total_matches = search_ids(term)
          if mode == "FULL" and not ids:
              if previous:
                  fail("The full scan returned no records for %s although the snapshot holds %d. "
                       "Refusing to replace it; check the gene symbol, the extra query and the NCBI status" % (gene, len(previous)))
              print("  NOTE: ClinVar returned no records for %s; an empty baseline is stored" % gene)
          records = fetch_records(ids) if ids else {}

          # ---- 5. Compare with the snapshot. A missing classification is never treated as a change.
          def in_scope(*values):
              return not monitored or any(v and v.lower() in monitored for v in values)


          new_variants, reclassified, newly_classified, review_only, lost = [], [], [], [], []
          if not baseline:
              for uid, cur in records.items():
                  if uid not in previous:
                      if in_scope(cur["classification"]):
                          new_variants.append(cur)
                      continue
                  old_cls, old_review = previous[uid]
                  new_cls = cur["classification"]
                  if new_cls is None and old_cls is not None:
                      lost.append(cur)
                  elif old_cls is None and new_cls is not None:
                      if in_scope(new_cls):
                          newly_classified.append(cur)
                  elif old_cls != new_cls:
                      if in_scope(old_cls, new_cls):
                          reclassified.append(dict(cur, previous_classification=old_cls, previous_review_status=old_review))
                  elif old_review != cur["review_status"]:
                      review_only.append(cur)

          variants = dict(previous)
          for uid, cur in records.items():
              cls, review = cur["classification"], cur["review_status"]
              if cls is None and uid in previous and previous[uid][0] is not None:
                  cls, review = previous[uid]
              variants[uid] = (cls, review)
          seen_ids = set(ids)
          not_returned = len([uid for uid in previous if uid not in seen_ids]) if mode == "FULL" and not baseline else 0

          # ---- 5b. On a scheduled full scan (a snapshot with a checkpoint exists), ask which of the detected changes the
          # ---- incremental query would have returned. Only changes OUTSIDE that ID set count. Baselines and schema
          # ---- migrations have no comparable window, and a failed comparison is reported as UNAVAILABLE, never as a number.
          changed_ids = {r["variation_id"] for r in reclassified + newly_classified + new_variants}
          window_check, outside_window = "NOT APPLICABLE", None
          if mode == "FULL" and checkpoint is not None:
              try:
                  in_window = set(search_ids("%s AND %s[mdat]" % (base_term, date_window()))[0])
                  window_check, outside_window = "COMPARED", len(changed_ids - in_window)
              except SystemExit as exc:
                  window_check = "UNAVAILABLE"
                  print("  WARNING the date-window comparison failed (%s); changes_outside_date_window is not reported" % exc)

          next_snapshot = encode(variants, today, today if mode == "FULL" else last_reconciled)
          snapshot_text = json.dumps(next_snapshot, separators=(",", ":"))
          if len(snapshot_text.encode("utf-8")) > MAX_SNAPSHOT_BYTES:
              fail("The snapshot for %s would be %d bytes, above the %d byte safety limit. Kestra documents no KV size limit, "
                   "but the value is also passed through task outputs; use a narrower extra query or another storage strategy"
                   % (gene, len(snapshot_text.encode("utf-8")), MAX_SNAPSHOT_BYTES))


          # ---- 6. Report. Untrusted text is escaped for Slack mrkdwn.
          def esc(value, limit=None):
              # Clip first, then escape, so a cut never splits an entity such as &amp;.
              text = str("" if value is None else value)
              if limit and len(text) > limit:
                  text = text[:limit] + "..."
              return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


          def conflict(*values):
              return any(v and "conflicting" in v.lower() for v in values)


          def line(rec, text):
              flag = " :warning: conflicting submissions" if conflict(rec["classification"], rec.get("previous_classification")) else ""
              review = " [%s]" % esc(rec["review_status"], 120) if rec["review_status"] else ""
              return "- <%s|%s> %s: %s%s%s" % (LINK % urllib.parse.quote(rec["variation_id"]), esc(rec["accession"] or rec["variation_id"], 40),
                                                esc(rec["title"], MAX_TITLE_CHARS), text, review, flag)


          alertable = (not baseline) and (bool(reclassified) or (alert_on_new and bool(new_variants or newly_classified)))
          groups = [("*Classification changes*", reclassified,
                     lambda r: line(r, "%s -&gt; %s" % (esc(r["previous_classification"], 120), esc(r["classification"], 120))))]
          if alert_on_new:
              groups.append(("*New variants*", new_variants,
                             lambda r: line(r, esc(r["classification"], 120) if r["classification"] else "no classification yet")))
              groups.append(("*Newly classified*", newly_classified, lambda r: line(r, esc(r["classification"], 120))))
          reportable = sum(len(items) for _, items, _ in groups)


          def build_slack_text():
              # Totals first, reclassifications before everything else, and an explicit count of whatever did not fit.
              head = ":dna: *ClinVar changes for %s* (detected %s, %s scan)" % (esc(gene), now_txt, mode.lower())
              totals = "Totals: " + ", ".join("%d %s" % (len(items), title.strip("*").lower()) for title, items, _ in groups)
              budget = MAX_SLACK_CHARS - len(head) - len(totals) - len(DISCLAIMER) - OMITTED_RESERVE
              body, used, shown = [], 0, 0
              for title, items, render in groups:
                  section = []
                  for rec in items[:max_lines]:
                      row = render(rec)
                      cost = len(row) + 1 + (0 if section else len(title) + 1)
                      if used + cost > budget:
                          break
                      if not section:
                          section.append(title)
                      section.append(row)
                      used += cost
                      shown += 1
                  body += section
              rows = [head, totals] + body
              if shown < reportable:
                  rows.append("%d additional change(s) omitted; see the execution logs." % (reportable - shown))
              rows.append(DISCLAIMER)
              return "\n".join(rows)


          slack_text = build_slack_text() if alertable else ""

          phase = "BASELINE" if baseline else ("CHANGES" if alertable else "NO CHANGES")
          text = ("%s: %s (%s). scanned=%d of %d matching records, new=%d, reclassified=%d, newly_classified=%d, "
                  "review_status_only=%d, classification_missing=%d, snapshot_size=%d, requests=%d"
                  % (gene, phase, mode.lower(), len(records), total_matches, len(new_variants), len(reclassified),
                     len(newly_classified), len(review_only), len(lost), len(variants), requests_made[0]))
          print(text)
          if baseline:
              print("  Baseline stored for %s: %d records. No alert is sent on the first run." % (gene, len(records)))
          if window_check == "COMPARED":
              print("  RECONCILIATION: %d of %d change(s) found by this full scan were outside the date window an incremental run "
                    "would have used" % (outside_window, len(changed_ids)))
          if not_returned:
              print("  NOTE: %d snapshot records were not returned by the full scan (withdrawn or merged); they are kept" % not_returned)
          # The execution log keeps the detail that a shortened Slack message leaves out.
          for label_, items_ in (("reclassified", reclassified), ("new variant", new_variants), ("newly classified", newly_classified)):
              for r in items_[:200]:
                  was = "%s -> " % r["previous_classification"] if label_ == "reclassified" else ""
                  print("  %s %s: %s%s (%s)" % (label_, r["variation_id"], was, r["classification"] or "no classification yet", r["title"]))
              if len(items_) > 200:
                  print("  ...and %d more %s record(s) are listed in report.json" % (len(items_) - 200, label_))
          for r in lost:
              print("  WARNING %s has no classification in this response; previous value kept" % r["variation_id"])

          Path("snapshot_out.json").write_text(snapshot_text)
          Path("report.json").write_text(json.dumps({
              "gene": gene, "mode": mode, "detected_at": now_iso, "baseline": baseline, "window": window,
              "date_window_comparison": window_check, "changes_outside_date_window": outside_window,
              "new_variants": new_variants, "reclassified": reclassified,
              "newly_classified": newly_classified, "review_status_only": review_only,
              "classification_missing": lost,
          }, indent=2))

          # Kestra reads task outputs from this stdout marker (the same one the kestra pip package prints), so no package is needed.
          print("::" + json.dumps({"outputs": {
              "gene": gene,
              "phase": phase,
              "mode": mode,
              "baseline": baseline,
              "alertable": alertable,
              "scanned": len(records),
              "total_matches": total_matches,
              "new_count": len(new_variants),
              "reclassified_count": len(reclassified),
              "newly_classified_count": len(newly_classified),
              "review_only_count": len(review_only),
              "missing_count": len(lost),
              "date_window_comparison": window_check,
              "changes_outside_date_window": outside_window,
              "summary": text,
              "slack_text": slack_text,
          }}) + "::")

      - id: report_gene
        type: io.kestra.plugin.core.log.Log
        description: Every run logs the verdict for the gene, including baseline runs
          and runs without changes.
        message: "{{ outputs.compare_snapshots.vars.summary }}"

      - id: notify_changes
        type: io.kestra.plugin.core.flow.If
        description: Send Slack only when something changed and notifications are on.
          The baseline run and unchanged genes stay quiet.
        condition: "{{ inputs.notify_slack and outputs.compare_snapshots.vars.alertable
          }}"
        then:
          - id: slack_alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: One message per gene with reclassifications listed apart from new
              variants, each with a ClinVar link.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: "{{ outputs.compare_snapshots.vars.slack_text }}"

      - id: save_snapshot
        type: io.kestra.plugin.core.kv.Set
        description: Persist the new snapshot and checkpoint only after the comparison
          and the alert succeeded. If Slack fails, the old snapshot and
          checkpoint stay and the same changes are found and reported again on
          the next run.
        key: "{{ inputs.state_key_prefix }}_{{ item.value | trim | lower }}"
        kvType: STRING
        value: "{{ read(outputs.compare_snapshots.outputFiles['snapshot_out.json']) }}"

  - id: run_summary
    type: io.kestra.plugin.core.log.Log
    description: One line per gene so the whole run can be read in a single place.
    message: |
      ClinVar monitor finished.
      {% for line in loopOutputs(outputs.watch_genes.outputs, 'gene_summary') %}- {{ line }}
      {% endfor %}

errors:
  - id: log_failure
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    description: A failed run is never reported as "no changes". The stored
      snapshots and checkpoints were not advanced past the failing gene.
    message: "ClinVar monitor FAILED in flow {{ flow.id }}, execution {{
      execution.id }}. Likely causes: NCBI rate limit or outage, an unexpected
      API response, an unknown gene symbol, or a corrupted stored snapshot. No
      success notification was sent."

  - id: alert_failure
    type: io.kestra.plugin.core.flow.If
    description: Tell Slack the check could not run, so silence is not mistaken for
      a clean result.
    condition: "{{ inputs.notify_slack }}"
    then:
      - id: slack_failure
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: ":x: ClinVar variant monitor FAILED (execution {{ execution.id }}).
          Results for this run are incomplete, so do not read the absence of
          alerts as no changes."