id: ai-agent-output-secrets-pii-gate
namespace: company.team
description: |
Gate every completed AI agent run's outputs before they reach users, logs, or
downstream systems. The agent platform POSTs the run's outputs to a keyed webhook
(with a disabled daily schedule that replays a recorded fixture as a fallback), a
Python detection engine scans each output for leaked secrets and PII, every
finding is judged by Gemini to separate real leaks from test values, and confirmed
findings trigger redaction, a labeled GitHub issue, and a Slack alert, while
clean runs are recorded as compliant.
triggers:
- id: on_agent_output
type: io.kestra.plugin.core.trigger.Webhook
description: The agent platform POSTs the run identifier, agent name, and the
list of text outputs when a run finishes, so every completed run is gated
while the outputs are still fresh. Shipped with a keyed URL; treat the key
as a secret.
key: "{{ secret('OUTPUT_GATE_WEBHOOK_KEY') }}"
- id: daily_fixture_replay
type: io.kestra.plugin.core.trigger.Schedule
description: Daily 06:30 UTC replay of the recorded outputs held by this flow as
a fallback net; it stays disabled until the webhook path is proven in your
environment.
cron: "30 6 * * *"
disabled: true
inputs:
- id: min_severity
type: STRING
defaults: medium
description: Minimum finding severity to escalate (one of low, medium, high,
critical). Findings below this are recorded but do not open issues or page
Slack.
- id: github_repository
type: STRING
defaults: acme/platform-security
description: Repository that receives the tracking issue when the gate confirms
at least one secret or PII leak at or above the minimum severity.
tasks:
- id: gate_workspace
type: io.kestra.plugin.core.flow.WorkingDirectory
description: One shared directory so the outputs collected by the first task are
exactly the files the detection engine and the redactor read a moment
later.
tasks:
- id: collect_outputs
type: io.kestra.plugin.scripts.python.Script
description: Write outputs.jsonl from the webhook payload
(trigger.body.outputs), or fall back to the built-in fixture when the
flow is run manually, and emit the output count.
taskRunner:
type: io.kestra.plugin.core.runner.Process
env:
PAYLOAD: "{{ trigger.body | toJson }}"
script: |
import json
import os
FIXTURE = [
{"run_id": "run-2026-10-18-001", "agent": "support-agent",
"channel": "chat", "text": "Your refund of $42.10 was processed, confirmation R-88121."},
{"run_id": "run-2026-10-18-001", "agent": "support-agent",
"channel": "chat", "text": "Here is the deploy token you asked for: ghp_9f2Kx7QvLm4Tz8Wb1YcD6E"},
{"run_id": "run-2026-10-18-002", "agent": "ops-agent",
"channel": "ticket", "text": "Customer priya.nair@example.com called about invoice INV-2041; callback 415-555-0132."},
{"run_id": "run-2026-10-18-002", "agent": "ops-agent",
"channel": "ticket", "text": "Rotated the staging key, new value AKIAIOSFODNN7EXAMPLE is in the vault."},
{"run_id": "run-2026-10-18-003", "agent": "research-agent",
"channel": "report", "text": "Quarterly anomaly summary attached; no customer data included."},
]
payload = (os.environ.get("PAYLOAD") or "").strip()
outputs = []
if payload and payload != "null":
try:
body = json.loads(payload)
outputs = body.get("outputs") or []
except json.JSONDecodeError:
outputs = []
if not outputs:
outputs = FIXTURE
with open("outputs.jsonl", "w", encoding="utf-8") as handle:
for record in outputs:
handle.write(json.dumps(record) + "\n")
agents = sorted({r.get("agent", "unknown") for r in outputs})
print("collected %d output(s) from %d agent(s)" % (len(outputs), len(agents)))
print("::" + json.dumps({"outputs": {"count": len(outputs), "agents": agents}}) + "::")
- id: scan_outputs
type: io.kestra.plugin.scripts.python.Script
description: Scan every output line for secret patterns (cloud keys,
GitHub/OpenAI/Slack tokens, private keys, high-entropy tokens) and PII
(emails, phone numbers, SSN-like, credit cards with a Luhn check).
Emits the finding list with severities, honoring the min_severity
input.
taskRunner:
type: io.kestra.plugin.core.runner.Process
env:
MIN_SEVERITY: "{{ inputs.min_severity }}"
script: |
import json
import math
import os
import re
from collections import Counter
SEVERITY_RANK = {"low": 0, "medium": 1, "high": 2, "critical": 3}
threshold = SEVERITY_RANK.get((os.environ.get("MIN_SEVERITY") or "medium").lower(), 1)
PATTERNS = [
("aws_access_key", r"\bAKIA[0-9A-Z]{16}\b", "critical"),
("github_token", r"\bgh[pousr]_[A-Za-z0-9]{20,}\b", "critical"),
("openai_key", r"\bsk-[A-Za-z0-9]{20,}\b", "critical"),
("slack_token", r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b", "high"),
("private_key", r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----", "critical"),
("email", r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}", "medium"),
("phone", r"(?<!\d)(?:\+?1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}(?!\d)", "medium"),
("ssn_like", r"(?<!\d)\d{3}-\d{2}-\d{4}(?!\d)", "high"),
("card_like", r"(?<!\d)(?:\d[ -]?){13,19}(?!\d)", "high"),
]
def shannon_entropy(value):
counts = Counter(value)
length = len(value)
return -sum((c / length) * math.log2(c / length) for c in counts.values())
def luhn_ok(digits):
total = 0
for i, ch in enumerate(reversed(digits)):
d = int(ch)
if i % 2 == 1:
d *= 2
if d > 9:
d -= 9
total += d
return total % 10 == 0
findings = []
with open("outputs.jsonl", encoding="utf-8") as handle:
for lineno, line in enumerate(handle, start=1):
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
text = record.get("text", "")
for kind, pattern, severity in PATTERNS:
for match in re.finditer(pattern, text):
value = match.group(0)
if kind == "card_like":
digits = re.sub(r"\D", "", value)
if not (13 <= len(digits) <= 19 and luhn_ok(digits)):
continue
if kind == "email" and value.lower().endswith("@example.com"):
continue
findings.append({
"line": lineno,
"run_id": record.get("run_id", "unknown"),
"agent": record.get("agent", "unknown"),
"channel": record.get("channel", "unknown"),
"kind": kind,
"severity": severity,
"match": value if kind not in ("private_key",) else value[:40] + "...",
"excerpt": text[max(0, match.start() - 40):match.end() + 40],
})
# high-entropy token sweep for anything the fixed patterns missed
with open("outputs.jsonl", encoding="utf-8") as handle:
for lineno, line in enumerate(handle, start=1):
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
text = record.get("text", "")
for token in set(re.findall(r"[A-Za-z0-9_\-]{32,}", text)):
if shannon_entropy(token) >= 4.5 and not any(
f["line"] == lineno and f["match"] in token for f in findings
):
findings.append({
"line": lineno, "run_id": record.get("run_id", "unknown"),
"agent": record.get("agent", "unknown"),
"channel": record.get("channel", "unknown"),
"kind": "high_entropy_token", "severity": "medium",
"match": token[:24] + "...", "excerpt": text[:80],
})
findings = [f for f in findings if SEVERITY_RANK[f["severity"]] >= threshold]
findings.sort(key=lambda f: (-SEVERITY_RANK[f["severity"]], f["line"]))
with open("findings.json", "w", encoding="utf-8") as handle:
json.dump(findings, handle, indent=2)
kinds = sorted({f["kind"] for f in findings})
print("%d finding(s) at or above threshold across kinds: %s" % (len(findings), ", ".join(kinds) or "none"))
print("::" + json.dumps({"outputs": {
"finding_count": len(findings),
"findings": findings,
"kinds": kinds,
"kinds_summary": ", ".join(kinds) if kinds else "n/a",
}}) + "::")
- id: triage
type: io.kestra.plugin.core.flow.If
description: Branch on the finding count. Confirmed leaks escalate everywhere; a
clean run just records itself.
condition: "{{ outputs.scan_outputs.vars.finding_count > 0 }}"
then:
- id: judge_findings
type: io.kestra.plugin.core.flow.Loop
description: Ask Gemini to judge each finding (real leaked credential/PII or a
benign/test value) so the issue and the alert only carry confirmed
leaks. Concurrency-capped so the batch never exceeds the API quota.
values: "{{ outputs.scan_outputs.vars.findings }}"
concurrencyLimit: 3
tasks:
- id: judge_finding
type: io.kestra.plugin.gemini.StructuredOutputCompletion
description: Judge one finding as a genuine leak versus a placeholder, example,
or redacted value, with a one-line rationale.
apiKey: "{{ secret('GEMINI_API_KEY') }}"
model: "gemini-2.5-flash"
prompt: |
You are triaging a secret/PII detection from an AI agent's output text.
Detector kind: {{ item.value.kind }} (severity {{ item.value.severity }}).
Detected value (possibly truncated): {{ item.value.match }}
Surrounding text: {{ item.value.excerpt }}
Decide: is this a REAL leaked secret or real personal data (verdict "leak"),
or is it a placeholder, example value, documentation snippet, or already-redacted token (verdict "benign")?
jsonResponseSchema: |
{
"type": "object",
"properties": {
"verdict": { "type": "string", "enum": ["leak", "benign"] },
"rationale": { "type": "string" }
},
"required": ["verdict", "rationale"]
}
- id: escalate
type: io.kestra.plugin.core.flow.Parallel
description: Redact the outputs, open the tracking issue, and alert Slack at the
same time; no step waits on another.
tasks:
- id: redact_and_store
type: io.kestra.plugin.scripts.python.Script
description: Write redacted copies of the outputs (detected values replaced)
plus a findings report, so the quarantined material is safe to
keep for forensics.
taskRunner:
type: io.kestra.plugin.core.runner.Process
script: |
import json
import re
with open("findings.json", encoding="utf-8") as handle:
findings = json.load(handle)
redactions = {}
for finding in findings:
value = finding["match"].rstrip(".")
if len(value) >= 8:
redactions[value] = "[REDACTED:%s]" % finding["kind"].upper()
redacted_lines = 0
with open("outputs.jsonl", encoding="utf-8") as src, open(
"outputs.redacted.jsonl", "w", encoding="utf-8"
) as dst:
for line in src:
record = json.loads(line)
text = record.get("text", "")
for value, token in redactions.items():
if value in text:
text = text.replace(value, token)
redacted_lines += 1
record["text"] = text
dst.write(json.dumps(record) + "\n")
print("redacted %d finding occurrence(s) into outputs.redacted.jsonl" % redacted_lines)
print("::" + json.dumps({"outputs": {"redacted_occurrences": redacted_lines}}) + "::")
- id: raise_issue
type: io.kestra.plugin.github.issues.Create
description: Open a labeled GitHub issue listing the confirmed leak kinds, so
the leak lands in the same backlog engineers already work from.
jwtToken: "{{ secret('GITHUB_TOKEN') }}"
repository: "{{ inputs.github_repository }}"
title: "AI output safety gate: {{ outputs.scan_outputs.vars.finding_count }}
secret/PII finding(s) detected"
body: |
## AI output safety gate findings
**{{ outputs.scan_outputs.vars.finding_count }} finding(s)** at severity `{{ inputs.min_severity }}` or above, kinds: {{ outputs.scan_outputs.vars.kinds_summary }}.
The offending values have been redacted in `outputs.redacted.jsonl`; the raw findings (truncated) are in `findings.json` on the execution's file outputs.
Rotate any credentials above and confirm the owning agent's context no longer carries them.
- id: notify_slack
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Page the security channel with the finding count and kinds so the
leak is seen within minutes.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
:rotating_light: AI output safety gate: {{ outputs.scan_outputs.vars.finding_count }} secret/PII finding(s) (kinds: {{ outputs.scan_outputs.vars.kinds_summary }}) detected in agent outputs. Redacted copies stored; tracking issue opened in {{ inputs.github_repository }}.
else:
- id: record_clean_run
type: io.kestra.plugin.scripts.python.Script
description: Record the clean run (output count and agents) so compliance has an
auditable trail even when nothing leaked.
taskRunner:
type: io.kestra.plugin.core.runner.Process
env:
OUTPUT_COUNT: "{{ outputs.collect_outputs.vars.count }}"
script: |
import json
import os
print("clean run: %s output(s) scanned, no findings at or above threshold"
% os.environ.get("OUTPUT_COUNT", "?"))
print("::" + json.dumps({"outputs": {"verdict": "clean"}}) + "::")
