Scan AI Agent Sessions for Leaked Secrets and PII

id: ai-agent-session-leakage-scanner
namespace: company.team
description: |
  Audit what your AI agents actually said. Every day this scanner pulls the previous day's agent
  session transcripts (user messages, agent replies, and tool outputs), runs them through layered
  detectors for leaked secrets (provider key shapes, tokens, private keys, high-entropy strings)
  and PII (emails, payment cards validated by Luhn, SSNs, IBANs validated by mod-97, phone
  numbers), quarantines the offending sessions as masked copies, and produces a leakage scorecard
  with a trend against the previous run. Genuine new leakage is judged by Gemini and escalated to
  a GitHub issue and a Slack alert. Detected values are never printed, logged, or posted anywhere:
  every artifact carries only a redacted form (first and last characters) plus a SHA-256
  fingerprint, so findings can be correlated without ever exposing the secret itself.
triggers:
  - id: daily_leakage_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 05:00 UTC scan of agent session transcripts; run it on demand
      after an incident or a logging change.
    cron: 0 5 * * *
inputs:
  - id: sessions_url
    type: STRING
    defaults: ''
    description: URL of a JSONL session export (one session per line) to scan
      instead of the bundled synthetic demo set. Each session carries
      session_id, agent, started_at, messages (role and content), and tool_calls
      (tool and output).
  - id: sessions_override
    type: STRING
    defaults: ''
    description: Leave empty in production. Demo and testing seam. A JSON array of
      session objects in the same shape as the export, so the scanner can be
      exercised without a live log pipeline.
  - id: entropy_threshold
    type: FLOAT
    defaults: 4.5
    description: Minimum Shannon entropy (bits per character) for the high-entropy
      token detector. The default sits above the ceiling for hex strings (4.0),
      so commit SHAs and UUIDs in transcripts do not trip it, while random
      base64-style tokens do.
  - id: max_total_findings
    type: INT
    defaults: 10
    description: The leakage gate also fails when total findings exceed this count,
      even with no critical finding. Any critical finding fails the gate on its
      own.
  - id: previous_scorecard_url
    type: STRING
    defaults: ''
    description: URL of the previous run's scorecard.json, used for the trend
      comparison. Empty marks this run as the baseline.
  - id: github_repository
    type: STRING
    defaults: acme/platform-security
    description: Repository that receives the tracking issue when the leakage gate fails.
tasks:
  - id: scan_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the sessions, the findings, the quarantine
      copies, and the scorecard are the same files every stage reads. Raw
      transcripts never leave this workspace; only redacted artifacts flow
      downstream.
    tasks:
      - id: ingest_sessions
        type: io.kestra.plugin.scripts.python.Script
        description: Load session transcripts from sessions_url or sessions_override, or
          write the bundled synthetic demo set (four sessions with planted leaks
          and two clean controls). Every session is validated for the fields the
          scanner needs and written to sessions.jsonl.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          SESSIONS_URL: '{{ inputs.sessions_url }}'
          SESSIONS_OVERRIDE: '{{ inputs.sessions_override }}'
        script: |
          import json, os, urllib.request
          from kestra import Kestra

          DEMO = [
              {"session_id": "clean-support-1", "agent": "support-agent", "started_at": "2026-10-21T09:14:00Z",
               "messages": [
                   {"role": "user", "content": "Hi, where is my order 8841? It was due on Tuesday."},
                   {"role": "agent", "content": "Your order 8841 shipped on Monday and arrives on Thursday. The old card on file ends in 1112; the full number 4111 1111 1111 1112 was declined, so please update it in your account page."}],
               "tool_calls": [{"tool": "order_lookup", "output": "{\"status\": \"shipped\", \"carrier\": \"DHL\"}"}]},
              {"session_id": "clean-dev-1", "agent": "dev-agent", "started_at": "2026-10-21T10:02:00Z",
               "messages": [
                   {"role": "user", "content": "The build failed at commit 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b, can you check? The staging reference DE89370400440532013001 did not resolve either."},
                   {"role": "agent", "content": "That commit is on main. The release id is 3f6c9e2a-7b1d-4e5a-9c8f-2d6b1a0e5f77 and version 2.4.1 passed all checks."}],
               "tool_calls": []},
              {"session_id": "leak-aws-1", "agent": "dev-agent", "started_at": "2026-10-21T11:40:00Z",
               "messages": [
                   {"role": "user", "content": "My script fails with this key: AKIAIOSFODNN7EXAMPLE, why?"},
                   {"role": "agent", "content": "That key id looks disabled in IAM. Rotate it and update your local credentials file."}],
               "tool_calls": []},
              {"session_id": "leak-github-1", "agent": "dev-agent", "started_at": "2026-10-21T12:05:00Z",
               "messages": [
                   {"role": "user", "content": "Read the repo config and summarise it."},
                   {"role": "agent", "content": "I emailed the summary to jane.doe@example.com."}],
               "tool_calls": [{"tool": "file_read", "output": "ci:\n  token: ghp_aBcDeFgHiJkLmNoPqRsTuVwXyZ0123456789ab\n  retries: 2"}]},
              {"session_id": "leak-payment-1", "agent": "checkout-agent", "started_at": "2026-10-21T13:22:00Z",
               "messages": [
                   {"role": "user", "content": "Save my card 4111 1111 1111 1111 for checkout, my SSN is 123-45-6789, and call me on +1 415 555 0132 if anything fails."},
                   {"role": "agent", "content": "Saved. Your IBAN DE89 3704 0044 0532 0130 00 is also on file for refunds."}],
               "tool_calls": []},
              {"session_id": "leak-config-1", "agent": "dev-agent", "started_at": "2026-10-21T14:47:00Z",
               "messages": [
                   {"role": "user", "content": "Dump the environment of the worker and the session cache key."},
                   {"role": "agent", "content": "Here is the worker environment and cache key you asked for."}],
               "tool_calls": [{"tool": "env_dump", "output": "OPENAI_API_KEY=sk-proj-AbCdEfGhIjKlMnOpQrStUvWxYz0123456789abcd\n-----BEGIN OPENSSH PRIVATE KEY-----\ncache_key=kX7vQ2mZ9pL4wR8tN3yH6sD1fG5jK0xB"}]},
          ]

          url = os.environ.get("SESSIONS_URL", "").strip()
          override_raw = os.environ.get("SESSIONS_OVERRIDE", "").strip()
          if override_raw:
              sessions = json.loads(override_raw)
              source = "override"
          elif url:
              with urllib.request.urlopen(url, timeout=30) as resp:
                  sessions = [json.loads(line) for line in resp.read().decode().splitlines() if line.strip()]
              source = url
          else:
              sessions = DEMO
              source = "bundled demo set"

          required = {"session_id", "agent", "messages"}
          for session in sessions:
              missing = required - set(session)
              if missing:
                  raise ValueError(f"session {session.get('session_id', '?')} is missing fields: {sorted(missing)}")
              session.setdefault("started_at", "")
              session.setdefault("tool_calls", [])

          with open("sessions.jsonl", "w") as fh:
              for session in sessions:
                  fh.write(json.dumps(session) + "\n")

          Kestra.outputs({"session_count": len(sessions), "source": source})
          print(f"sessions loaded: {len(sessions)} (source: {source})")

      - id: scan_sessions
        type: io.kestra.plugin.scripts.python.Script
        description: Scan every message and tool output with layered detectors. Known
          secret shapes and credential assignments are matched by pattern,
          payment cards are confirmed with the Luhn checksum and IBANs with
          mod-97 (so lookalike numbers stay clean), and a Shannon-entropy pass
          catches unknown token shapes. Each span is claimed once, so a value is
          never double-counted. Findings record only the detector, a redacted
          form, and a SHA-256 fingerprint; the raw value is never written to any
          file or log.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          ENTROPY_THRESHOLD: '{{ inputs.entropy_threshold }}'
        script: |
          import hashlib, json, math, os, re
          from collections import Counter
          from kestra import Kestra

          def redact(value):
              v = value.strip()
              fingerprint = hashlib.sha256(v.encode()).hexdigest()[:12]
              if len(v) <= 6:
                  return "***", fingerprint
              return v[:2] + "***" + v[-2:], fingerprint

          def shannon_entropy(text):
              counts = Counter(text)
              n = len(text)
              return -sum((c / n) * math.log2(c / n) for c in counts.values())

          def luhn_ok(digits):
              total, alt = 0, False
              for ch in reversed(digits):
                  d = int(ch)
                  if alt:
                      d = d * 2 - 9 if d * 2 > 9 else d * 2
                  total += d
                  alt = not alt
              return total % 10 == 0

          def iban_ok(candidate):
              compact = re.sub(r"\s+", "", candidate).upper()
              if not re.fullmatch(r"[A-Z]{2}\d{2}[A-Z0-9]{11,30}", compact):
                  return False
              rearranged = compact[4:] + compact[:4]
              numeric = "".join(str(int(ch, 36)) if ch.isalpha() else ch for ch in rearranged)
              remainder = 0
              for i in range(0, len(numeric), 7):
                  remainder = int(str(remainder) + numeric[i:i + 7]) % 97
              return remainder == 1

          PATTERNS = [
              ("private_key_block", "secrets", "critical",
               re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH |PGP )?PRIVATE KEY-----"), None),
              ("aws_access_key_id", "secrets", "critical",
               re.compile(r"\bAKIA[0-9A-Z]{16}\b"), None),
              ("github_token", "secrets", "critical",
               re.compile(r"\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{36,}\b|\bgithub_pat_[A-Za-z0-9_]{22,}\b"), None),
              ("slack_token", "secrets", "critical",
               re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b"), None),
              ("anthropic_api_key", "secrets", "critical",
               re.compile(r"\bsk-ant-[A-Za-z0-9_-]{20,}\b"), None),
              ("openai_api_key", "secrets", "critical",
               re.compile(r"\bsk-(?:proj-)?[A-Za-z0-9_-]{20,}\b"), None),
              ("jwt_token", "secrets", "high",
               re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b"), None),
              ("generic_credential_assignment", "secrets", "high",
               re.compile(r"(?i)\b(?:api[_-]?key|client[_-]?secret|auth[_-]?token|access[_-]?token|password|passwd)\b\s*[:=]\s*[\"']?([^\s\"',;]{8,})"), 1),
              ("email_address", "pii", "medium",
               re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"), None),
              ("us_ssn", "pii", "high",
               re.compile(r"\b(?!000|666|9\d\d)\d{3}-(?!00)\d{2}-(?!0000)\d{4}\b"), None),
              ("phone_number", "pii", "low",
               re.compile(r"\+\d{1,3}[\s.-]\(?\d{3}\)?[\s.-]\d{3}[\s.-]\d{4}\b"), None),
          ]
          CARD_CANDIDATE = re.compile(r"\b(?:\d[ -]?){13,19}\b")
          IBAN_CANDIDATE = re.compile(r"\b[A-Z]{2}\d{2}(?: ?[A-Z0-9]{4}){2,7}(?: ?[A-Z0-9]{1,3})?\b")
          TOKEN_CANDIDATE = re.compile(r"[A-Za-z0-9+/=_-]{20,}")

          threshold = float(os.environ.get("ENTROPY_THRESHOLD", "4.5"))

          def scan_text(text):
              findings = []
              claimed = []

              def overlaps(start, end):
                  return any(start < c_end and c_start < end for c_start, c_end in claimed)

              def add(detector, category, severity, start, end, value):
                  masked, fingerprint = redact(value)
                  findings.append({
                      "detector": detector, "category": category, "severity": severity,
                      "start": start, "end": end, "redacted": masked, "fingerprint": fingerprint,
                  })
                  claimed.append((start, end))

              for name, category, severity, pattern, group in PATTERNS:
                  for match in pattern.finditer(text):
                      span = match.span(group) if group else match.span()
                      if overlaps(*span):
                          continue
                      add(name, category, severity, span[0], span[1], match.group(group) if group else match.group(0))
              for match in CARD_CANDIDATE.finditer(text):
                  digits = re.sub(r"[ -]", "", match.group(0))
                  if 13 <= len(digits) <= 19 and luhn_ok(digits) and not overlaps(*match.span()):
                      add("credit_card", "pii", "high", match.start(), match.end(), digits)
              for match in IBAN_CANDIDATE.finditer(text):
                  if iban_ok(match.group(0)) and not overlaps(*match.span()):
                      add("iban", "pii", "high", match.start(), match.end(), re.sub(r"\s+", "", match.group(0)))
              for match in TOKEN_CANDIDATE.finditer(text):
                  token = match.group(0)
                  if overlaps(*match.span()):
                      continue
                  if re.fullmatch(r"[0-9a-fA-F-]+", token):
                      continue
                  if len(set(token)) < 10:
                      continue
                  if shannon_entropy(token) >= threshold:
                      add("high_entropy_token", "secrets", "medium", match.start(), match.end(), token)
              return findings

          def mask_text(text, findings):
              masked = text
              for finding in sorted(findings, key=lambda f: f["start"], reverse=True):
                  masked = masked[:finding["start"]] + f"[REDACTED:{finding['detector']}]" + masked[finding["end"]:]
              return masked

          sessions = [json.loads(line) for line in open("sessions.jsonl") if line.strip()]
          all_findings = []
          with open("masked_sessions.jsonl", "w") as masked_fh:
              for session in sessions:
                  masked_session = dict(session)
                  masked_messages = []
                  for idx, message in enumerate(session["messages"]):
                      text = message.get("content", "")
                      field_findings = scan_text(text)
                      masked_text = mask_text(text, field_findings)
                      for finding in field_findings:
                          all_findings.append({
                              "session_id": session["session_id"],
                              "agent": session["agent"],
                              "location": f"messages[{idx}] ({message.get('role', '?')})",
                              "detector": finding["detector"],
                              "category": finding["category"],
                              "severity": finding["severity"],
                              "redacted": finding["redacted"],
                              "fingerprint": finding["fingerprint"],
                              "context_masked": masked_text[:240],
                          })
                      masked_message = dict(message)
                      masked_message["content"] = masked_text
                      masked_messages.append(masked_message)
                  masked_session["messages"] = masked_messages
                  masked_calls = []
                  for idx, call in enumerate(session["tool_calls"]):
                      text = call.get("output", "")
                      field_findings = scan_text(text)
                      masked_text = mask_text(text, field_findings)
                      for finding in field_findings:
                          all_findings.append({
                              "session_id": session["session_id"],
                              "agent": session["agent"],
                              "location": f"tool_calls[{idx}] ({call.get('tool', '?')})",
                              "detector": finding["detector"],
                              "category": finding["category"],
                              "severity": finding["severity"],
                              "redacted": finding["redacted"],
                              "fingerprint": finding["fingerprint"],
                              "context_masked": masked_text[:240],
                          })
                      masked_call = dict(call)
                      masked_call["output"] = masked_text
                      masked_calls.append(masked_call)
                  masked_session["tool_calls"] = masked_calls
                  masked_fh.write(json.dumps(masked_session) + "\n")

          with open("findings.jsonl", "w") as fh:
              for finding in all_findings:
                  fh.write(json.dumps(finding) + "\n")

          by_detector = Counter(f["detector"] for f in all_findings)
          Kestra.outputs({
              "findings_total": len(all_findings),
              "detectors_triggered": sorted(by_detector),
          })
          print(f"scan complete: {len(all_findings)} findings across {len(by_detector)} detectors; "
                f"values recorded redacted only")

      - id: quarantine_sessions
        type: io.kestra.plugin.scripts.python.Script
        description: Quarantine every session with a high or critical finding. The
          quarantine copy has every detected value replaced by a redaction
          marker, so reviewers can inspect the session safely, and the manifest
          records why each session was held with counts per detector. Sessions
          with only low or medium findings are reported but not held.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        script: |
          import json
          from collections import Counter
          from kestra import Kestra

          findings = [json.loads(line) for line in open("findings.jsonl") if line.strip()]
          masked_sessions = {s["session_id"]: s for s in
                             (json.loads(l) for l in open("masked_sessions.jsonl") if l.strip())}

          held_reasons = {}
          for finding in findings:
              if finding["severity"] in ("high", "critical"):
                  held_reasons.setdefault(finding["session_id"], Counter())[finding["detector"]] += 1

          quarantined = []
          with open("quarantine.jsonl", "w") as fh:
              for session_id, reasons in sorted(held_reasons.items()):
                  masked = masked_sessions[session_id]
                  entry = {
                      "session_id": session_id,
                      "agent": masked["agent"],
                      "started_at": masked.get("started_at", ""),
                      "reasons": dict(reasons),
                      "finding_count": sum(1 for f in findings if f["session_id"] == session_id),
                      "handling": "held for security review; the copy below is masked, the raw "
                                  "transcript stays in the execution workspace",
                      "masked_messages": masked["messages"],
                      "masked_tool_calls": masked["tool_calls"],
                  }
                  fh.write(json.dumps(entry) + "\n")
                  quarantined.append(entry)

          Kestra.outputs({
              "quarantined_count": len(quarantined),
              "quarantined_sessions": [q["session_id"] for q in quarantined],
          })
          print(f"quarantined {len(quarantined)} sessions: "
                + ", ".join(q["session_id"] for q in quarantined))

      - id: build_scorecard
        type: io.kestra.plugin.scripts.python.Script
        description: Build the leakage scorecard (JSON and Markdown). Totals are broken
          down by detector, by category (secrets vs PII), and by severity, with
          the leakage rate across sessions, a trend against the previous
          scorecard, and the gate verdict. The scorecard carries redacted
          examples only (detector, session, first and last characters,
          fingerprint), never a raw value.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          MAX_TOTAL_FINDINGS: '{{ inputs.max_total_findings }}'
          PREVIOUS_SCORECARD_URL: '{{ inputs.previous_scorecard_url }}'
        script: |
          import json, os, urllib.request
          from collections import Counter
          from kestra import Kestra

          max_total = int(os.environ.get("MAX_TOTAL_FINDINGS", "10"))
          previous_url = os.environ.get("PREVIOUS_SCORECARD_URL", "").strip()

          findings = [json.loads(line) for line in open("findings.jsonl") if line.strip()]
          sessions = [json.loads(line) for line in open("sessions.jsonl") if line.strip()]
          quarantined = [json.loads(line) for line in open("quarantine.jsonl") if line.strip()]

          by_detector = dict(sorted(Counter(f["detector"] for f in findings).items()))
          by_category = dict(sorted(Counter(f["category"] for f in findings).items()))
          by_severity = dict(sorted(Counter(f["severity"] for f in findings).items()))
          flagged_sessions = sorted({f["session_id"] for f in findings})
          total = len(findings)
          critical = by_severity.get("critical", 0)
          leakage_rate = round(100.0 * len(flagged_sessions) / len(sessions), 1) if sessions else 0.0
          passed = critical == 0 and total <= max_total

          trend = {"previous_findings": None, "delta": None, "note": "baseline run, no previous scorecard"}
          if previous_url:
              try:
                  with urllib.request.urlopen(previous_url, timeout=30) as resp:
                      previous = json.loads(resp.read().decode())
                  prev_total = int(previous["findings_total"])
                  delta = total - prev_total
                  note = "more leakage than previous run" if delta > 0 else (
                      "less leakage than previous run" if delta < 0 else "same leakage as previous run")
                  trend = {"previous_findings": prev_total, "delta": delta, "note": note}
              except Exception as exc:
                  trend = {"previous_findings": None, "delta": None,
                           "note": f"previous scorecard unreadable: {exc}"}

          examples = []
          seen = set()
          for finding in findings:
              if finding["detector"] in seen:
                  continue
              seen.add(finding["detector"])
              examples.append(f"{finding['detector']} in {finding['session_id']}: "
                              f"{finding['redacted']} (sha256 {finding['fingerprint']})")
              if len(examples) >= 8:
                  break

          judge_detectors = sorted({f["detector"] for f in findings
                                    if f["severity"] in ("critical", "high")})
          scorecard = {
              "sessions_scanned": len(sessions),
              "sessions_flagged": len(flagged_sessions),
              "sessions_quarantined": len(quarantined),
              "leakage_rate": leakage_rate,
              "findings_total": total,
              "by_detector": by_detector,
              "by_category": by_category,
              "by_severity": by_severity,
              "max_total_findings": max_total,
              "passed": passed,
              "trend": trend,
              "redacted_examples": examples,
          }
          with open("scorecard.json", "w") as fh:
              json.dump(scorecard, fh, indent=2)

          lines = ["# Agent session leakage scorecard", "",
                   f"Scanned **{len(sessions)} sessions**: {len(flagged_sessions)} flagged "
                   f"({leakage_rate}% leakage rate), {len(quarantined)} quarantined, "
                   f"**{total} findings** (gate: {'PASS' if passed else 'FAIL'}). "
                   f"Trend: {trend['note']}"
                   + (f" ({trend['delta']:+d} vs {trend['previous_findings']})" if trend["delta"] is not None else "") + ".",
                   "", "| Detector | Findings |", "| --- | --- |"]
          for name, count in by_detector.items():
              lines.append(f"| {name} | {count} |")
          lines += ["", "| Severity | Findings |", "| --- | --- |"]
          for name, count in by_severity.items():
              lines.append(f"| {name} | {count} |")
          if examples:
              lines += ["", "## Redacted examples (raw values are never recorded)", ""]
              for example in examples:
                  lines.append(f"- {example}")
          with open("scorecard.md", "w") as fh:
              fh.write("\n".join(lines) + "\n")

          detector_summary = ", ".join(f"{k}: {v}" for k, v in by_detector.items()) or "none"
          Kestra.outputs(scorecard)
          Kestra.vars({
              "passed": passed,
              "findings_total": total,
              "critical_count": critical,
              "leakage_rate": leakage_rate,
              "detector_summary": detector_summary,
              "redacted_examples": "\n".join(examples),
              "judge_detectors": judge_detectors,
              "trend_note": trend["note"],
          })
          print(f"scorecard: {total} findings, {critical} critical, leakage rate {leakage_rate}%, "
                f"gate {'PASS' if passed else 'FAIL'}")

      - id: write_trend_record
        type: io.kestra.plugin.scripts.python.Script
        description: Append this run to trend.jsonl (date, findings, critical count,
          leakage rate, per-detector counts, trend note), the series the next
          run's scorecard is compared against when its URL is published as
          previous_scorecard_url.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          RUN_DATE: '{{ execution.startDate }}'
        script: |
          import json, os
          from kestra import Kestra

          with open("scorecard.json") as fh:
              scorecard = json.load(fh)
          entry = {
              "date": os.environ.get("RUN_DATE", ""),
              "findings_total": scorecard["findings_total"],
              "critical": scorecard["by_severity"].get("critical", 0),
              "leakage_rate": scorecard["leakage_rate"],
              "passed": scorecard["passed"],
              "by_detector": scorecard["by_detector"],
              "trend": scorecard["trend"]["note"],
          }
          with open("trend.jsonl", "a") as fh:
              fh.write(json.dumps(entry) + "\n")
          Kestra.outputs({"trend_record": entry})
          print("trend record appended:", json.dumps(entry))
  - id: triage
    type: io.kestra.plugin.core.flow.If
    description: Branch on the leakage gate. A passing run just records itself; a
      failing run gets each flagged detector judged and then escalated.
    condition: '{{ outputs.build_scorecard.vars.passed == false }}'
    then:
      - id: judge_flagged_detectors
        type: io.kestra.plugin.core.flow.Loop
        description: Ask Gemini to judge each detector with a high or critical finding,
          meaning a genuine leak the team must rotate and purge, or likely test
          data and fixtures that belong on an allowlist. The judge only ever
          sees the detector name and redacted examples, never a raw value.
        values: '{{ outputs.build_scorecard.vars.judge_detectors }}'
        concurrencyLimit: 3
        tasks:
          - id: judge_detector
            type: io.kestra.plugin.gemini.StructuredOutputCompletion
            description: Judge whether the findings from one detector are a genuine leak or
              test and fixture data, with a one-line rationale.
            apiKey: '{{ secret(''GEMINI_API_KEY'') }}'
            model: gemini-2.5-flash
            prompt: 'You are reviewing AI agent session leakage findings. Detector: {{
              item.value }}.

              The scanner found high or critical severity matches for this
              detector in agent session transcripts. The evidence you may
              consider is limited to redacted forms (first and last characters)
              and SHA-256 fingerprints; raw values are never shared.

              Decide: is this a GENUINE leak that must be rotated and purged
              (verdict "genuine_leak"), or is it likely test data, documentation
              examples, or fixtures that should be allowlisted (verdict
              "test_or_fixture")?'
            jsonResponseSchema: "{

              \  \"type\": \"object\",

              \  \"properties\": {

              \    \"verdict\": { \"type\": \"string\", \"enum\":
              [\"genuine_leak\", \"test_or_fixture\"] },

              \    \"rationale\": { \"type\": \"string\" }

              \  },

              \  \"required\": [\"verdict\", \"rationale\"]

              }\n"
      - id: escalate
        type: io.kestra.plugin.core.flow.Parallel
        description: Open the tracking issue and alert Slack at the same time; both
          carry counts and redacted examples only.
        tasks:
          - id: raise_issue
            type: io.kestra.plugin.github.issues.Create
            description: Open a labeled GitHub issue with the leakage totals, the detector
              breakdown, and the redacted examples, so the leak lands in the
              security backlog.
            jwtToken: '{{ secret(''GITHUB_TOKEN'') }}'
            repository: '{{ inputs.github_repository }}'
            title: 'AI agent session leakage: {{ outputs.build_scorecard.vars.findings_total
              }} findings, {{ outputs.build_scorecard.vars.critical_count }}
              critical'
            body: '## AI agent session leakage results


              **{{ outputs.build_scorecard.vars.findings_total }} findings** ({{
              outputs.build_scorecard.vars.critical_count }} critical) across
              agent session transcripts. Leakage rate: {{
              outputs.build_scorecard.vars.leakage_rate }}% of sessions. Trend:
              {{ outputs.build_scorecard.vars.trend_note }}.


              Detectors: {{ outputs.build_scorecard.vars.detector_summary }}.


              Flagged sessions were quarantined as masked copies for review.
              Evidence below is redacted by design (first and last characters
              plus a SHA-256 fingerprint); raw values never leave the scan
              execution, so rotate by fingerprint correlation with your secret
              stores.


              {{ outputs.build_scorecard.vars.redacted_examples }}


              The full scorecard is in `scorecard.json` and `scorecard.md` on
              the execution''s file outputs. If Gemini judged a detector as test
              or fixture data, allowlist it in the scanner configuration instead
              of rotating.

              '
          - id: notify_slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Page the security channel with the leakage totals so a genuine leak
              is seen within minutes.
            url: '{{ secret(''SLACK_WEBHOOK_URL'') }}'
            messageText: ':warning: AI agent session leakage: {{
              outputs.build_scorecard.vars.findings_total }} findings, {{
              outputs.build_scorecard.vars.critical_count }} critical, leakage
              rate {{ outputs.build_scorecard.vars.leakage_rate }}%. Detectors:
              {{ outputs.build_scorecard.vars.detector_summary }}. Tracking
              issue opened in {{ inputs.github_repository }}. Evidence is
              redacted by design.

              '
    else:
      - id: record_clean_run
        type: io.kestra.plugin.scripts.python.Script
        description: Record the passing run (sessions scanned, findings within
          tolerance) so the team has an auditable trail of leakage posture over
          time.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          TOTAL: '{{ outputs.build_scorecard.vars.findings_total }}'
        script: "import json

          import os

          from kestra import Kestra


          print(\"passing run: %s findings within tolerance, no critical
          leakage\" % os.environ.get(\"TOTAL\", \"?\"))

          Kestra.outputs({\"verdict\": \"passing\"})\n"