Sanitize a whole queue of messages in one call with POST /v1/batch/transform, draft token-only LLM replies, and restore each per audience

Batch mode runs the Detect, Tokenize, [LLM], Rehydrate pipeline over many messages in a single call. Each item gets its own sealed session (independent mode) or shares one token map across items (shared mode), so a recurring value keeps one identity across a case. Only typed tokens and safe metadata ever leave the OGuardAI boundary; the model sees tokens, never raw values.

This guide embeds one runnable example verbatim from the repository at examples/batch-python/. A developer or an agent can build the whole pipeline from the code and fixtures below without opening the repo.

Overnight support-queue triage (Python, raw HTTP)

Scenario: the fictional German green-energy utility Musterfirma GmbH runs an unattended nightly job over the day's inbound customer emails. It classifies each message and drafts a German reply with an LLM, without ever sending regulated PII (names, IBANs, addresses, phones, order numbers, or a national ID a customer pasted by mistake) to the model.

The example uses raw HTTP against three endpoints (POST /v1/batch/transform, POST /v1/transform, POST /v1/rehydrate) so it depends on no SDK beyond requests.

What it demonstrates

Overnight queue, batch/transform (independent mode). One call sanitizes six messages, each into its own sealed session. The offline drafter classifies and writes a token-only German reply, then the same reply is restored three ways from one session: the customer's gender-aware reply (full on customer_email), a redacted audit line (log_safe), and an abstract analytics row (abstract). A national-ID-shaped value is hard-redacted at transform and never reaches the model. Per-item failures are isolated, so one bad message never sinks the batch.
Safe metadata drives correct grammar, then the full restore spectrum. A single /v1/transform returns the metadata OGuardAI lets out per entity (gender, formality, language). The drafter reads the person's gender to pick the German adjective ending that agrees with the honorific the restore will insert (Sehr geehrter for Herr, Sehr geehrte for Frau), then the same reply is shown under every restore mode: full, formatted, masked, partial, abstract, none. The policy is the ceiling: the IBAN stays masked even under full, because the german-support policy caps it.
Multi-part case, batch/transform (shared mode). A customer writes three times about the same refund. Shared mode keeps ONE token per recurring value (order number, email, name) across all parts, so the case stays one identity end to end. The contrast run shows that independent mode mints a different token per part. The final reply resolves against the accumulated terminal session, including the name that only appeared in part 1.

The german-support policy governs the whole run: person restore is gender-aware (Herr/Frau), IBAN and credit cards are capped at masked, and SSN, passport, and health IDs are hard-redacted.

Prerequisites

Python with requests: pip install -r requirements.txt.
A running OGuardAI server (default base URL http://localhost:3000).
Needs-NER: person names (and the gender-aware German salutation) require the NER detector sidecar, and the batch's german-support policy declares person detection as required. The batch endpoint reports a detector outage per item, not with a top-level status: each affected item comes back in the HTTP 200 batch response with its error field set (the message detection unavailable) instead of an un-tokenized name, while items whose detection succeeded still return their safe text. The built-in regex still covers structural PII (email, phone, IBAN, order, address, IDs), but the batch needs the full stack with the detector active:

# The bundled stack requires these four secrets, and its client key ships least-privilege (channel
# user_output only), so widen it to the channels this example restores on.
export GUARDAI_SESSION_SECRET=$(openssl rand -base64 32)
export GUARDAI_API_KEY_1=$(openssl rand -hex 32)
export GUARDAI_DETECTOR_API_KEY=$(openssl rand -hex 32)
export GUARDAI_REDIS_PASSWORD=$(openssl rand -hex 32)
export GUARDAI_API_KEY_1_ALLOWED_OUTPUT_CHANNELS=user_output,customer_email,log_safe
docker compose -f ../../deploy/docker/docker-compose.yml up -d

The docker compose path above is relative to examples/batch-python/ (deploy/docker/docker-compose.yml from the repository root).

Run

export GUARDAI_BASE_URL="http://localhost:3000"   # or GUARDAI_URL / GUARDAI_BASE
export GUARDAI_API_KEY="$GUARDAI_API_KEY_1"        # the bundled stack runs api_key auth
export OPENAI_API_KEY="<key>"                      # optional; only flips a label. The provider
                                                   # is never contacted; drafting stays offline.
python main.py

To see session_id instead of session_state, run the server with a server-side backend (set session.backend: memory in the server config, or session.backend: redis with GUARDAI_REDIS_URL, replay_backend: redis, and revocation_backend: redis, since a redis session backend requires a redis replay backend and advertises a multi-replica topology that startup gates on the shared redis revocation authority).

The code

The full example is examples/batch-python/main.py. It is embedded below in contiguous blocks that together reconstruct the file.

Module docstring, imports, and the run configuration (base URL, policy, headers). The base URL accepts any of the repo's env conventions. From main.py:

"""Overnight support-queue triage for Musterfirma GmbH (a fictional German
green-energy utility), built on OGuardAI's batch endpoint.

THE PROBLEM. Every night the care platform holds a backlog of the day's inbound
customer emails. Before an agent opens them in the morning, an unattended job wants
an LLM to classify each message and draft a German reply. The messages are full of
regulated PII (names, IBANs, addresses, phones, order numbers, the odd national ID
a customer pasted by mistake). Sending that to a model raw is a GDPR incident;
stripping it leaves the model unable to write a correct, personal reply.

THE PIPELINE (Detect -> Tokenize -> [LLM] -> Rehydrate), run over a whole queue:

    1. batch/transform (independent) -> one call sanitizes every message; only typed
       tokens ever leave the boundary. The model classifies and drafts on tokens.
    2. transform (single, with metadata) -> the safe metadata that OGuardAI lets out
       (gender, formality, language) drives a grammatically correct formal salutation,
       then the SAME reply is restored under every restore mode to show the spectrum.
    3. batch/transform (shared) -> a customer's multi-part case keeps ONE token per
       recurring value (order number, email, name) across all parts, so the case
       stays coherent and the final reply resolves against the accumulated session.

Fail-closed by policy: a national-ID-shaped value is hard-redacted at step 1 and never
reaches the model. Per-item failures are isolated: one bad message never sinks the batch.

Prerequisites:
    pip install requests
    # OGuardAI server with the NER detector (person/company/location):
    #   docker compose -f ../../deploy/docker/docker-compose.yml up -d
    # The batch sanitization is structural PII; the gender-aware person restore needs NER.

Usage:
    export GUARDAI_BASE_URL="http://localhost:3000"   # or GUARDAI_URL / GUARDAI_BASE
    export GUARDAI_API_KEY="<key>"                     # only if the server enforces auth
    export OPENAI_API_KEY="<key>"                      # optional; only flips the label that
                                                       # a real model would draft the replies.
                                                       # The provider is never contacted here.
    python main.py
"""

from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

import requests

# Accept whichever base-URL var is set so the example drops into any of the repo's
# env conventions without edits.
BASE = (
    os.environ.get("GUARDAI_BASE_URL")
    or os.environ.get("GUARDAI_URL")
    or os.environ.get("GUARDAI_BASE")
    or "http://localhost:3000"
).rstrip("/")
API_KEY = os.environ.get("GUARDAI_API_KEY", "")

# A single vertical policy governs the whole run: German formal restore (Herr/Frau),
# IBAN capped at masked, national IDs redacted. Per-item language pins honorifics to German.
POLICY = "german-support"
LANG = "de"

HEADERS = {"Content-Type": "application/json"}
if API_KEY:
    HEADERS["X-API-Key"] = API_KEY

HERE = Path(__file__).resolve().parent
TOKEN_RE = re.compile(r"\{\{([^}]+)\}\}")

HTTP and token helpers. post fails closed on any error and never surfaces the raw upstream body, which could echo request PII. token_map maps each entity type in one safe text to the exact token string (hash included) the LLM must reuse. From main.py:

def post(path: str, body: dict) -> dict:
    resp = requests.post(f"{BASE}{path}", json=body, headers=HEADERS, timeout=60)
    # Fail closed: on any transform/rehydrate error we stop rather than fall back to
    # processing raw text. Surface the server's typed error code when it sent one.
    if resp.status_code >= 400:
        detail = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
        code = detail.get("code", f"HTTP_{resp.status_code}")
        # Never surface the raw upstream body: it can echo request PII. Use OGuardAI's typed
        # error string when present, else a generic status-only message.
        error = detail.get("error")
        message = error if isinstance(error, str) else f"HTTP {resp.status_code}"
        raise SystemExit(f"OGuardAI request to {path} failed: {code} - {message}")
    return resp.json()


def load_jsonl(name: str) -> list[dict]:
    with (HERE / name).open(encoding="utf-8") as fh:
        return [json.loads(line) for line in fh if line.strip()]


def token_type(inner: str) -> str:
    """Type key from a token body: 'person:p_001:hash' -> 'person',
    'custom:policy_number:x_001:hash' -> 'custom:policy_number'."""
    return inner.rsplit(":", 2)[0] if inner.count(":") >= 2 else inner.split(":")[0]


def token_map(safe_text: str) -> dict[str, str]:
    """Map each entity type in one safe text to the exact token string (hash included)
    the LLM must reuse. A well-behaved model echoes these verbatim; the rehydrator
    resolves the same string back. Short forms like {{person:p_001}} do NOT resolve."""
    out: dict[str, str] = {}
    for inner in TOKEN_RE.findall(safe_text):
        out.setdefault(token_type(inner), "{{" + inner + "}}")
    return out


def rehydrate(draft: str, session_state: str, *, restore_mode: str | None = None, channel: str | None = None) -> str:
    body: dict = {"output": draft, "session_state": session_state}
    if restore_mode:
        body["restore_mode"] = restore_mode
    if channel:
        body["output_channel"] = channel
    return post("/v1/rehydrate", body)["restored_text"]

The model stand-in. In production this is one LLM call over the whole safe batch; here it classifies on safe keywords and fills a template with the message's own tokens, so the example stays deterministic and offline. It reads OPENAI_API_KEY exactly like a real client would and only ever sees tokenized text. From main.py:

# --- The "model" stand-in -------------------------------------------------------------
# In production this is ONE LLM call over the whole safe batch. It reads OPENAI_API_KEY
# exactly like a real client would and sees only tokenized text, never raw PII. To keep
# the example deterministic and offline it classifies on safe keywords and fills a
# template with the message's own tokens; the real provider is never contacted here.
LLM_ENABLED = bool(os.environ.get("OPENAI_API_KEY"))

INTENTS = [
    ("refund", ("fehlbuchung", "erstattung", "gutschrift", "rueckerstattung")),
    ("meter_reading", ("zaehlerstand", "zaehler", "ablesung")),
    ("cancellation", ("kuendigung", "kuendigen", "vertrag beenden")),
    ("address_change", ("adressaenderung", "anschrift", "adresse aendern", "umzug")),
    ("billing", ("abrechnung", "rechnung", "abschlag")),
]


def classify(safe_text: str, subject: str) -> str:
    hay = f"{subject} {safe_text}".lower()
    for intent, keys in INTENTS:
        if any(k in hay for k in keys):
            return intent
    return "general"


def draft_reply(intent: str, tok: dict[str, str]) -> str:
    """Token-only German reply. Every {{...}} is a placeholder the rehydrator fills per
    channel; the drafter never handles a raw value. The batch result carries no
    per-message gender, so the salutation uses "Guten Tag", which is correct German for
    any gender; the honorific (Frau/Herr) is still inserted correctly by the restore."""
    person = tok.get("person", "")
    salutation = f"Guten Tag {person}" if person else "Sehr geehrte Damen und Herren"
    order = tok.get("order")
    iban = tok.get("iban")
    ref = f" zu Bestellung {order}" if order else ""
    if intent == "refund":
        pay = f" Die Erstattung erfolgt auf {iban}." if iban else ""
        return f"{salutation}, wir haben Ihre Fehlbuchung{ref} geprueft und veranlassen die Korrektur.{pay}"
    if intent == "meter_reading":
        return f"{salutation}, vielen Dank fuer Ihre Zaehlerstandsmeldung. Wir haben den Wert uebernommen."
    if intent == "cancellation":
        return f"{salutation}, wir haben Ihre Kuendigung{ref} erhalten und bestaetigen den Eingang."
    if intent == "address_change":
        return f"{salutation}, Ihre neue Anschrift ist hinterlegt. Die Aenderung ist ab sofort aktiv."
    if intent == "billing":
        return f"{salutation}, wir pruefen Ihre Rueckfrage zur Abrechnung und melden uns zeitnah."
    return f"{salutation}, vielen Dank fuer Ihre Nachricht. Ein Mitarbeiter meldet sich in Kuerze."


def entities_of(safe_text: str) -> int:
    return len(TOKEN_RE.findall(safe_text))

The leak-boundary contract. Every structural identifier crossing to the model is a hard failure on every lane. The batch's german-support policy declares person detection as required, so on a detector outage the batch returns HTTP 200 with the affected items carrying a per-item error (the message detection unavailable) rather than let an un-tokenized name cross; the script reads /v1/capabilities.ner_active to report that state. The sentinels are the exact values planted in the fixtures, checked against every string that crosses to the model. From main.py:

# --- Leak boundary contract -----------------------------------------------------------
# The one leak-exemption signal is /v1/capabilities.ner_active: the runtime's configured
# detector intent. It is the ONLY thing that turns a person name reaching the model from a
# failure into a documented degradation. Everything structural stays a failure on every lane.
def ner_active() -> bool:
    """True unless capabilities reports ner_active explicitly false. A fetch miss, a
    malformed body, or a non-boolean field defaults to available, so the example never
    grants the no-NER name exemption it cannot justify."""
    try:
        data = requests.get(f"{BASE}/v1/capabilities", headers=HEADERS, timeout=10).json()
    except requests.RequestException:
        return True
    val = data.get("ner_active", True) if isinstance(data, dict) else True
    return val is not False


# Sentinels planted in inbound_queue.jsonl / case_thread.jsonl and the section-2 message.
# STRUCTURED ids are builtin-regex-detectable, so they must tokenize on every lane and their
# raw form must never survive into a string the model reads. NAMES need the NER detector;
# without it they cross raw, which is a note on this fixture, not a structural leak.
STRUCTURED_SENTINELS = (
    "erika.mustermann@example.de",
    "erika.mustermann@example.de",
    "erika.mustermann@example.de",
    "ORD-2026-4471",
    "ORD-2026-5510",
    "ORD-2026-6021",
    "DE97 9900 0000 0123 4567 89",
    "DE97 9900 0000 0123 4567 89",
    "+49 89 5512347",
    "+49 40 6633120",
    "123-45-6789",
)
NAME_SENTINELS = (
    "Erika Mustermann",
    "Max Mustermann",
    "Erika Mustermann",
    "Max Mustermann",
    "Erika Mustermann",
    "Max Mustermann",
)


def scan_crossing(text: str, leaks: dict[str, set[str]]) -> None:
    """Record every raw sentinel that survived into a string that crosses to the model
    (the safe text OGuardAI vouched for, and the token-only draft built from it)."""
    for value in STRUCTURED_SENTINELS:
        if value in text:
            leaks["structured"].add(value)
    for name in NAME_SENTINELS:
        if name in text:
            leaks["names"].add(name)

Section 1, the independent-mode triage. One POST /v1/batch/transform sanitizes the whole queue, then each draft is restored into three audiences from its own sealed session. From main.py:

# --- Pipeline sections ----------------------------------------------------------------

def run_independent_triage(leaks: dict[str, set[str]]) -> None:
    """Section 1: sanitize the whole overnight queue in one call, draft a reply per
    message, and restore each draft into three audiences from its own sealed session."""
    queue = load_jsonl("inbound_queue.jsonl")
    items = [{"text": m["body"], "language": LANG} for m in queue]

    print("=" * 78)
    print("1. OVERNIGHT QUEUE -> batch/transform (independent: one session per message)")
    print("=" * 78)
    res = post("/v1/batch/transform", {"items": items, "session_mode": "independent", "policy": POLICY})
    model = "OpenAI (token-only batch)" if LLM_ENABLED else "offline template drafter"
    print(f"   sanitized {res['stats']['total_items']} messages, "
          f"{res['stats']['total_entities']} entities, in {res['stats']['total_time_ms']:.0f} ms")
    print(f"   drafting with: {model}\n")

    for msg, result in zip(queue, res["results"]):
        # Isolate per-item failures: a bad message reports an error and is skipped, the
        # rest of the batch is unaffected.
        if result.get("error"):
            print(f"   [{msg['message_id']}] SKIPPED: {result['error']}")
            continue

        safe = result["safe_text"]
        state = result["session_state"]
        redacted = "[REDACTED:" in safe  # a hard identifier was dropped, never tokenized
        scan_crossing(safe, leaks)

        print(f"   [{msg['message_id']}] {msg['subject']}  ({entities_of(safe)} tokens"
              + (", 1 hard-redacted" if redacted else "") + ")")
        print(f"     safe (left the boundary): {safe}")

        intent = classify(safe, msg["subject"])
        draft = draft_reply(intent, token_map(safe))
        scan_crossing(draft, leaks)
        print(f"     intent={intent}  draft (token-only): {draft}")

        # One draft, three audiences, from the SAME sealed session:
        agent = rehydrate(draft, state, restore_mode="full", channel="customer_email")
        audit = rehydrate(draft, state, channel="log_safe")
        stats = rehydrate(draft, state, restore_mode="abstract")
        print(f"     -> customer reply : {agent}")
        print(f"     -> audit log line : {audit}")
        print(f"     -> analytics row  : {stats}")
        if redacted:
            print("     note: a national-ID-shaped value was hard-redacted at transform; "
                  "the model never saw it.")
        print()

Section 2, the safe-metadata salutation and the full restore spectrum. A single /v1/transform returns entity_context with gender, formality, and language; the drafter uses gender to pick the correct adjective ending, then the same draft is restored under every mode. The trailing checks decide the exit code, so they raise rather than assert. From main.py:

# The person restore already inserts the honorific (Herr/Frau). The gender metadata only
# has to pick the German adjective ending that AGREES with it: "geehrter" for Herr,
# "geehrte" for Frau. That agreement is exactly what a gender-blind drafter gets wrong.
SALUTATION = {"male": "Sehr geehrter", "female": "Sehr geehrte"}


def run_metadata_and_restore_spectrum(leaks: dict[str, set[str]]) -> None:
    """Section 2: a single transform that returns the safe metadata (gender/formality),
    used to build a grammatically correct formal salutation, then the SAME reply under
    every restore mode. Core rule: a mode only tightens what the policy allows, it never
    widens it (the IBAN stays masked even under `full`, because the policy caps it)."""
    msg = ("Kundin Erika Mustermann meldet eine Fehlbuchung zu Bestellung ORD-2026-4471. "
           "Erstattung bitte auf DE97 9900 0000 0123 4567 89.")
    res = post("/v1/transform", {"input": msg, "language": LANG, "policy": POLICY})
    safe, state = res["safe_text"], res["session_state"]
    scan_crossing(safe, leaks)
    tok = token_map(safe)

    print("=" * 78)
    print("2. SAFE METADATA -> correct salutation, then every RESTORE MODE (one session)")
    print("=" * 78)
    print(f"   safe (left the boundary): {safe}")
    print("   metadata that left the boundary (drives grammar, never the raw name):")
    for ent in res.get("entity_context", []):
        print(f"     {ent['token'][:26]:28} gender={ent.get('gender'):8} "
              f"formality={ent.get('formality'):8} lang={ent.get('language')}")

    # The drafter is gender-blind on the token, but the metadata is safe to read: pick the
    # adjective ending that agrees with the honorific, so the German is correct rather than
    # the naive "Sehr geehrte Herr ...". Unknown gender falls back to gender-neutral "Guten Tag".
    # Without NER no person token appears, so the salutation degrades to the neutral opener
    # while the restore spectrum still runs on the structured tokens.
    person = tok.get("person", "")
    person_gender = next((e.get("gender") for e in res.get("entity_context", [])
                          if e["token"] == person), "unknown") if person else "unknown"
    opener = SALUTATION.get(person_gender, "Guten Tag")
    greeting = f"{opener} {person}" if person else opener
    draft = (f"{greeting}, Bestellung {tok.get('order', '')} wird auf "
             f"{tok.get('iban', '')} erstattet.")
    scan_crossing(draft, leaks)
    print(f"\n   metadata-driven salutation (gender={person_gender}): {opener} ...")
    print(f"   token-only draft: {draft}\n")

    restored = {}
    for mode in ("full", "formatted", "masked", "partial", "abstract", "none"):
        restored[mode] = rehydrate(draft, state, restore_mode=mode)
        print(f"   [{mode:9}] {restored[mode]}")
    print("   (IBAN never widens past masked under any mode: the policy caps it.)\n")

    # exit 0 is not proof the restore worked. When a person actually tokenized this run the
    # name must come back in the clear under `full`; a revoked value restores as [DELETED]
    # (revocation is value-digest-keyed and crosses sessions, so a fixture shared with a
    # revoke demo would poison this), and a token the drafter left unresolvable survives as a
    # literal {{...}}. The honorific prefix a detector may prepend ("[PERSON]Frau ...") is
    # tolerated; only the surname is required. These decide the exit code, so they raise
    # rather than assert, which -O would strip.
    full = restored["full"]
    if "{{person" in safe and "Mustermann" not in full:
        raise SystemExit(f"expected the customer name restored under full mode, got: {full}")
    if "[DELETED]" in full:
        raise SystemExit(f"[DELETED] under full restore, a revoked value leaked: {full}")
    if "{{" in full:
        raise SystemExit(f"full restore still carries an unresolved token: {full}")

Section 3, the shared-mode multi-part case. Shared mode reuses one token per recurring value across all parts; the contrast run under independent mode mints a fresh token per part. The terminal session accumulates part 1's name token so the final reply resolves it even though later parts omit the name. From main.py:

def run_shared_thread(leaks: dict[str, set[str]]) -> None:
    """Section 3: a customer's multi-part case. Shared mode keeps ONE token per recurring
    value across all parts, so the order number, email, and name are the same identity end
    to end; independent mode would mint a fresh token per part. The final reply resolves
    against the accumulated terminal session even for a name that only appeared in part 1."""
    thread = load_jsonl("case_thread.jsonl")
    items = [{"text": p["body"], "language": LANG} for p in thread]

    print("=" * 78)
    print("3. MULTI-PART CASE -> batch/transform (shared: token map carries forward)")
    print("=" * 78)
    shared = post("/v1/batch/transform", {"items": items, "session_mode": "shared", "policy": POLICY})

    # Accumulate the token per type across all parts: shared mode reuses ids, so part 1's
    # name token is valid against the terminal session even though later parts omit the name.
    acc: dict[str, str] = {}
    terminal_state = None
    order_tokens, email_tokens = [], []
    for part, result in zip(thread, shared["results"]):
        safe = result["safe_text"]
        scan_crossing(safe, leaks)
        part_tok = token_map(safe)
        for etype, token in part_tok.items():
            acc.setdefault(etype, token)
        if part_tok.get("order"):
            order_tokens.append(part_tok["order"])
        if part_tok.get("email"):
            email_tokens.append(part_tok["email"])
        if result.get("session_state"):
            terminal_state = result["session_state"]
        print(f"   part {part['part']}: {safe}")

    same_order = len(set(order_tokens)) == 1
    same_email = len(set(email_tokens)) == 1
    print(f"\n   order token identical across all parts: {same_order}  ({order_tokens[0] if order_tokens else 'n/a'})")
    print(f"   email token identical across all parts: {same_email}")

    # Contrast: the identical bodies under independent mode get DIFFERENT tokens per part.
    indep = post("/v1/batch/transform", {"items": items, "session_mode": "independent", "policy": POLICY})
    for r in indep["results"]:
        scan_crossing(r["safe_text"], leaks)
    indep_orders = [token_map(r["safe_text"]).get("order") for r in indep["results"]]
    print(f"   under independent mode the same order gets {len(set(indep_orders))} different tokens (no carry-forward)")

    if terminal_state and acc.get("person"):
        draft = (f"Guten Tag {acc['person']}, zu Bestellung {acc.get('order', '')} ist die "
                 f"Erstattung nun angewiesen. Wir melden uns an {acc.get('email', '')}.")
        scan_crossing(draft, leaks)
        reply = rehydrate(draft, terminal_state, restore_mode="full", channel="customer_email")
        print(f"\n   thread reply (from the terminal session): {reply}")
    print()

The entry point. It health-checks the server, warns when NER is inactive, runs the three sections, then converts the accumulated leak record into the exit code: a structured id crossing raw is always a failure; a name crossing raw is a failure only when the runtime claims NER, else a note. From main.py:

def main() -> None:
    try:
        health = requests.get(f"{BASE}/v1/health", headers=HEADERS, timeout=10).json()
    except requests.RequestException as exc:
        raise SystemExit(f"Cannot reach OGuardAI at {BASE}: {exc}")

    ner_available = ner_active()
    if not ner_available:
        detector = health.get("components", {}).get("detector", {}).get("message", "") \
            if isinstance(health, dict) else ""
        print("WARNING: NER detector not active; person names will not tokenize and reach "
              "the model raw. Start the full stack via docker compose for the German "
              f"honorific restore. [detector: {detector}]\n", file=sys.stderr)

    leaks: dict[str, set[str]] = {"structured": set(), "names": set()}
    run_independent_triage(leaks)
    run_metadata_and_restore_spectrum(leaks)
    run_shared_thread(leaks)

    # One leak record drives both the exit code and the closing line. A structured id crossing
    # raw is always a failure; a name crossing raw is a failure only when the runtime claims NER,
    # else a note that never prints the blanket assurance.
    structured = sorted(leaks["structured"])
    names = sorted(leaks["names"])
    if structured:
        raise SystemExit("FAIL: structured PII crossed to the model raw: " + ", ".join(structured))
    if names and ner_available:
        raise SystemExit(
            "FAIL: a person name crossed to the model raw while NER is active: " + ", ".join(names)
        )
    if names:
        listed = ", ".join(names)
        print(f"NOTE: {listed} reached the model because this runtime runs without NER. "
              "Start the full stack via docker compose to tokenize person names.")
        print(f"Done. Structured PII stayed inside the runtime; {listed} reached the model "
              "because this runtime runs without NER.")
        return
    print("Done. Raw PII stayed inside the trusted zone; no untokenized PII reached the model.")


if __name__ == "__main__":
    main()

Fixtures and dependencies

requirements.txt:

requests>=2.31,<3

inbound_queue.jsonl holds the six fictional overnight messages consumed in independent mode. All PII is invented. Each line is one JSON object with message_id, received_at, channel, subject, and body:

{"message_id": "NL-INB-88213", "received_at": "2026-07-07T18:42:11+02:00", "channel": "email", "subject": "Fehlbuchung Abschlag", "body": "Kundin Erika Mustermann meldet eine Fehlbuchung zu Bestellung ORD-2026-4471. Erstattung bitte auf DE97 9900 0000 0123 4567 89. Kontakt: erika.mustermann@example.de."}
{"message_id": "NL-INB-88231", "received_at": "2026-07-07T19:05:44+02:00", "channel": "webform", "subject": "Zaehlerstand melden", "body": "Mein Name ist Max Mustermann. Bitte notieren Sie meinen Zaehlerstand fuer die Anschrift Lindenstrasse 12, 80331 Muenchen. Telefon +49 89 5512347."}
{"message_id": "NL-INB-88240", "received_at": "2026-07-07T20:11:02+02:00", "channel": "email", "subject": "Vertrag kuendigen", "body": "Kundin Erika Mustermann bittet um Kuendigung des Vertrags zu Bestellung ORD-2026-5510. Rueckfragen an erika.mustermann@example.de oder +49 40 6633120."}
{"message_id": "NL-INB-88255", "received_at": "2026-07-07T21:36:29+02:00", "channel": "email", "subject": "Frage zur Abrechnung", "body": "Der Kunde Max Mustermann hat eine Frage zur Jahresabrechnung und hat versehentlich seine Sozialversicherungsnummer 123-45-6789 in die Nachricht kopiert."}
{"message_id": "NL-INB-88268", "received_at": "2026-07-07T22:04:57+02:00", "channel": "webform", "subject": "Adressaenderung", "body": "Hier schreibt Erika Mustermann. Bitte aendern Sie meine Anschrift auf Gartenweg 4, 20095 Hamburg. Neue E-Mail: erika.mustermann@example.de."}
{"message_id": "NL-INB-88274", "received_at": "2026-07-07T22:48:13+02:00", "channel": "email", "subject": "Beschwerde Gutschrift", "body": "Der Kunde Max Mustermann beschwert sich ueber Bestellung ORD-2026-6021 und erwartet eine Gutschrift auf DE97 9900 0000 0123 4567 89."}

case_thread.jsonl holds one fictional customer's three-part refund case consumed in shared mode. The order number and email recur across all three parts:

{"message_id": "NL-INB-88213", "received_at": "2026-07-07T18:42:11+02:00", "part": 1, "body": "Kundin Erika Mustermann meldet eine Fehlbuchung zu Bestellung ORD-2026-4471, Kontakt erika.mustermann@example.de."}
{"message_id": "NL-INB-88301", "received_at": "2026-07-08T09:13:50+02:00", "part": 2, "body": "Nachtrag zu Bestellung ORD-2026-4471: die Erstattung fehlt weiterhin, bitte an erika.mustermann@example.de antworten."}
{"message_id": "NL-INB-88355", "received_at": "2026-07-08T11:27:04+02:00", "part": 3, "body": "Letzte Erinnerung zu Bestellung ORD-2026-4471, der Kontakt bleibt erika.mustermann@example.de."}

Session handle by backend

Each result carries the handle its backend uses; rehydrate with whichever one is present.

Backend	Per-result handle
Sealed (default)	`session_state` (a client-held encrypted blob)
Memory / Redis	`session_id` (the token map lives server-side)

Expected output (sealed backend, abridged)

1. OVERNIGHT QUEUE -> batch/transform (independent: one session per message)
   sanitized 6 messages, 21 entities, in 680 ms
   drafting with: offline template drafter

   [NL-INB-88213] Fehlbuchung Abschlag  (4 tokens)
     safe (left the boundary): Kundin {{person:p_001:...}} meldet eine Fehlbuchung zu Bestellung {{order:o_001:...}}. Erstattung bitte auf {{iban:ib_001:...}}. Kontakt: {{email:e_001:...}}.
     intent=refund  draft (token-only): Guten Tag {{person:p_001:...}}, wir haben Ihre Fehlbuchung ...
     -> customer reply : Guten Tag Frau Erika Mustermann, wir haben Ihre Fehlbuchung zu Bestellung ORD-2026-4471 ... Die Erstattung erfolgt auf D*** **** **** **** **** *9.
     -> audit log line : Guten Tag [REDACTED], wir haben Ihre Fehlbuchung zu Bestellung [REDACTED] ...
     -> analytics row  : Guten Tag (Kundin), wir haben Ihre Fehlbuchung zu Bestellung (order on file) ...

   [NL-INB-88255] Frage zur Abrechnung  (1 tokens, 1 hard-redacted)
     safe (left the boundary): Der Kunde {{person:p_001:...}} ... seine Sozialversicherungsnummer [REDACTED:ssn] ...
     note: a national-ID-shaped value was hard-redacted at transform; the model never saw it.

2. SAFE METADATA -> correct salutation, then every RESTORE MODE (one session)
   metadata that left the boundary (drives grammar, never the raw name):
     {{person:p_001:...}}          gender=female   formality=formal   lang=de
   metadata-driven salutation (gender=female): Sehr geehrte ...
   [full     ] Sehr geehrte Frau Erika Mustermann, Bestellung ORD-2026-4471 wird auf D*** **** **** **** **** *9 erstattet.
   [formatted] Sehr geehrte Frau Erika Mustermann, Bestellung O***********1 wird auf D*** **** **** **** **** *9 erstattet.
   [masked   ] Sehr geehrte E**** *********n, Bestellung O***********1 wird auf D*** **** **** **** **** *9 erstattet.
   [partial  ] Sehr geehrte E. Mustermann, Bestellung ************* wird auf D*** **** **** **** **** *9 erstattet.
   [abstract ] Sehr geehrte (Kundin), Bestellung (order on file) wird auf (iban on file) erstattet.
   [none     ] Sehr geehrte [REDACTED], Bestellung [REDACTED] wird auf [REDACTED] erstattet.
   (IBAN never widens past masked under any mode: the policy caps it.)

3. MULTI-PART CASE -> batch/transform (shared: token map carries forward)
   order token identical across all parts: True
   email token identical across all parts: True
   under independent mode the same order gets 3 different tokens (no carry-forward)
   thread reply (from the terminal session): Guten Tag Frau Erika Mustermann, zu Bestellung ORD-2026-4471 ist die Erstattung nun angewiesen. Wir melden uns an erika.mustermann@example.de.

Done. Raw PII stayed inside the trusted zone; the model and every downstream sink saw tokens only.

Batch Processing