Your Agent Just Leaked One Customer's Data to Another

We shipped a support agent that could look up a customer's account to answer billing questions. It worked well for months. Then one afternoon a user asked why their last payment failed, and the agent, reaching for context to be helpful, pulled the account record and wrote back a friendly paragraph that included the last four digits of a card and an email address. The card and the email belonged to a different customer whose record had been in the retrieval results because of a fuzzy name match. The agent did nothing malicious. It summarized what it was handed. But the message went out, and you cannot unsend a message.

Nothing in our stack was technically broken. The retrieval returned rows. The model summarized them. The response was well formed and polite. And it was a data breach, because the one surface we were not inspecting was the thing the agent actually said to a human being. Every layer of defense we had built pointed inward, at what came into the agent, and none of it pointed outward, at what left.

This is the blind spot in most agent deployments. Teams spend real effort on input defense, and they should. But an agent's output is the one artifact that reaches a user, a customer, an inbox, a public channel, and it is the one artifact you cannot take back once it has been sent. If you are not inspecting what your agent says on the way out, you are trusting a probabilistic model to never once say the wrong thing to the wrong person. It will.

Input defense does not cover the output

If you have hardened your agent, you have probably worked on the ingress side. You sanitize tool results, you constrain what retrieved documents can instruct, you gate which tools the model may call. That work is necessary, and I have written about the tool-call version of it in defending tool-using agents against prompt injection. None of it inspects the generated text that goes back to the user.

The reason the output needs its own layer is that a harmful reply does not require a hostile input. It can come from an entirely benign chain. The model retrieves a record it was allowed to retrieve and then surfaces a field it should not have. It hallucinates a refund policy that sounds plausible and commits your company to it in writing. It restates a competitor's name, or a slur that appeared in a support ticket, or an internal system detail that leaked into the context. The instructions were clean. The output was not. Ingress controls, by construction, cannot catch this, because the problem is created inside the model, at generation time, after every input check has already passed.

There is a second reason a system prompt is not enough. Telling the model do not reveal other customers' data is a request, not a guarantee. It usually works, which is exactly what makes it dangerous, because the failure is rare, silent, and catastrophic when it lands. A guarantee lives in code that runs every time and does not depend on the model choosing to comply. That is what an egress layer is: a deterministic checkpoint the output must pass before it is allowed to leave.

The shape of the fix: a fail-closed egress layer

Put a single stage between the model's finished output and the send. Every response, whether it goes to a user, an email API, or a webhook, passes through it. The stage runs an ordered set of checks. Each check can pass the text, rewrite it (redact a matched span), or block it outright. The default when a check errors or is unsure is not to send. Fail-closed is the whole point: a guardrail that fails open is decoration.

Start with the contract, because the structure matters more than any single check.

from dataclasses import dataclass
from enum import Enum
from typing import Callable, Protocol

class Action(Enum):
    ALLOW = "allow"      # text is fine as-is
    REDACT = "redact"    # text was rewritten; keep going with the new text
    BLOCK = "block"      # do not send at all

@dataclass
class CheckResult:
    action: Action
    text: str                 # possibly rewritten
    reason: str = ""          # for logging, never shown to the user

class Check(Protocol):
    name: str
    def __call__(self, text: str, ctx: dict) -> CheckResult: ...

def run_egress(text: str, ctx: dict, checks: list[Check]) -> CheckResult:
    current = text
    for check in checks:
        try:
            result = check(current, ctx)
        except Exception as e:
            # A check that throws must not open the gate. Fail closed.
            return CheckResult(Action.BLOCK, current, f"{check.name} errored: {e}")
        if result.action is Action.BLOCK:
            return result                      # stop immediately, nothing ships
        current = result.text                  # carry redactions forward
    return CheckResult(Action.ALLOW, current)

Three properties make this trustworthy. Checks run in order, so a cheap deterministic filter runs before an expensive model call. Redactions accumulate, so the text handed to the next check is already cleaned by the previous one. And any exception blocks, so a bug in a check can only ever make you too safe, never too loose. The ctx dict carries who this output is going to and what the request was about, which the checks need to decide what counts as a violation.

A redaction check that rewrites, and a policy check that blocks

The most common egress failure is structured sensitive data, card numbers, government IDs, other people's contact details, leaking into a reply. This one is deterministic and belongs in code, running first, before you spend a token on anything smarter.

import re

# Patterns that are cheap and high-precision. Validate where you can:
# a bare 16-digit match is noisy, so confirm card-shaped numbers with Luhn.
CARD_RE = re.compile(r"\b(?:\d[ -]*?){13,16}\b")
EMAIL_RE = re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")

def luhn_ok(digits: str) -> bool:
    nums = [int(d) for d in digits][::-1]
    total = sum(d if i % 2 == 0 else (d * 2 - 9 if d * 2 > 9 else d * 2)
                for i, d in enumerate(nums))
    return total % 10 == 0

def redact_pii(text: str, ctx: dict) -> CheckResult:
    changed = False

    def card_sub(m: re.Match) -> str:
        nonlocal changed
        digits = re.sub(r"\D", "", m.group())
        if 13 <= len(digits) <= 16 and luhn_ok(digits):
            changed = True
            return "[redacted card]"
        return m.group()

    out = CARD_RE.sub(card_sub, text)

    # Only redact emails that are not the recipient's own address. An agent
    # repeating *your* email back to you is fine; leaking someone else's is not.
    recipient = ctx.get("recipient_email", "").lower()
    def email_sub(m: re.Match) -> str:
        nonlocal changed
        if m.group().lower() == recipient:
            return m.group()
        changed = True
        return "[redacted email]"

    out = EMAIL_RE.sub(email_sub, out)
    action = Action.REDACT if changed else Action.ALLOW
    return CheckResult(action, out, "pii redacted" if changed else "")

Notice the check uses context to avoid over-redacting: repeating the recipient's own email back to them is not a leak, so the recipient address in ctx keeps the filter from mangling a legitimate reply. That distinction between the recipient and everyone else is exactly the thing a blanket system-prompt instruction cannot encode.

Redaction handles data that can be surgically removed. Some outputs cannot be fixed by cutting a span; the whole message is the problem. A reply that invents a refund policy or promises a delivery date you never authorized has to be stopped, not trimmed. That is a judgment call, which is where a model based check earns its cost. Run it last, only on text that survived the cheap checks.

UNAUTHORIZED_COMMITMENTS = (
    "Flag replies that make a binding commitment the company has not "
    "authorized: refunds, discounts, delivery dates, legal or medical "
    "assurances, or policy the agent invented. Reply with JSON "
    '{"violation": true/false, "kind": "..."}. Judge only the text given.'
)

def check_commitments(text: str, ctx: dict) -> CheckResult:
    verdict = judge_model.classify(   # a small, cheap model or a moderation endpoint
        system=UNAUTHORIZED_COMMITMENTS,
        user=text,
    )
    if verdict.get("violation"):
        # Block and hand back a safe fallback the caller can send instead.
        return CheckResult(
            Action.BLOCK, text,
            f"unauthorized commitment: {verdict.get('kind')}",
        )
    return CheckResult(Action.ALLOW, text)

When this blocks, the caller does not ship the model's text. It sends a safe fallback, something like let me connect you with a person who can confirm that, and it logs the blocked draft for review. A blocked output is not an error to hide from the user. It is the system working, catching a promise you would have had to honor or publicly walk back.

Wiring the layer into the request path is the easy part, and it is the same regardless of which checks you run.

CHECKS = [redact_pii, check_links, check_commitments]  # cheap to expensive

def respond(model_output: str, ctx: dict) -> str:
    result = run_egress(model_output, ctx, CHECKS)
    if result.action is Action.BLOCK:
        log.warning("egress blocked", reason=result.reason, draft=model_output)
        return SAFE_FALLBACK
    return result.text   # ALLOW or REDACT: send the cleaned text

The streaming trap, and other pitfalls

Streaming defeats the whole guarantee if you do it naively. The appeal of streaming is that tokens reach the user as they are generated. That is also the problem: if you stream raw tokens, you have already sent the first eight digits of a card number before any check could see the ninth. There is no such thing as un-streaming. The fix is to buffer to a natural boundary, a sentence or a small block, run the deterministic checks on that unit, and release it only once it clears. The user sees output arrive in chunks rather than a smooth character crawl, which is a small perceptual cost for the guarantee that nothing unredacted ever leaves. Model based checks that need the whole message run once, at the end, before the final send.

def stream_guarded(token_iter, ctx: dict):
    buffer = ""
    for token in token_iter:
        buffer += token
        # Release on sentence boundaries so a checked, clean unit goes out.
        if buffer.endswith((". ", "! ", "? ", "\n")):
            checked = run_egress(buffer, ctx, DETERMINISTIC_CHECKS)
            if checked.action is Action.BLOCK:
                yield SAFE_FALLBACK
                return
            yield checked.text
            buffer = ""
    if buffer:  # flush the tail
        checked = run_egress(buffer, ctx, DETERMINISTIC_CHECKS)
        yield SAFE_FALLBACK if checked.action is Action.BLOCK else checked.text

Do not log the thing you just redacted. It is easy to redact a card number in the reply and then write the raw draft straight into your logs for debugging, which recreates the leak in a system that is often less locked down than the one you were protecting. Log the reason and a hash or a masked preview, not the sensitive span itself. The guardrail's own telemetry is a place PII loves to reappear.

Over-redaction erodes trust in the layer. A filter that mangles every number into [redacted] will eat order totals, dates, and quantities, and the team will start routing around it. Precision matters as much as recall here. Validate where you can, the Luhn check on card-shaped numbers is the pattern, use context to spare legitimate values like the recipient's own details, and tune against real transcripts rather than shipping a greedy regex and hoping.

A guardrail is not a license to skip the input work. Egress filtering catches what leaks out; it does not fix why the model reached for the wrong record in the first place. If retrieval is handing the agent other customers' rows, tighten retrieval and access control too. The output layer is your last line, not your only one, and treating it as a substitute for correct data scoping means you are relying on a regex to cover a permissions bug.

Test it like the security control it is. The egress layer is exactly the kind of code that quietly regresses in a refactor, and its failures are invisible until one reaches a customer. Pin the behavior with cases: feed it a reply containing a valid test card and assert it redacts, feed it an invented refund promise and assert it blocks, feed it the recipient's own email and assert it passes. This is the same discipline as validating structured output at the boundary, which I covered in making LLM JSON output reliable, pointed at safety invariants instead of schema shape.

The takeaway

Your agent's output is the one thing it produces that you cannot recall. Input defenses, however good, all point the wrong way to catch a harmful reply, because a harmful reply can come from a perfectly clean input the moment the model summarizes the wrong row or invents a policy. The fix is a fail-closed egress layer that every output passes through: deterministic redaction first, model based judgment last, block on anything that cannot be safely trimmed, and never fail open. It is a few hundred lines, it runs on every response, and it turns your agent from something that will eventually say the wrong thing to the wrong person into something that physically cannot.

If your agents are talking to customers, sending email, or posting anywhere a mistake becomes public, this egress layer is worth building before the incident that forces it. Book a consultation call and we can map out what your agent should never be allowed to say, and put a gate in front of it.