RAG Document Pipeline

From: Erika Mustermann <erika.mustermann@example.com>
To: Support Team
Subject: Account #AC-2026-8834 - Billing dispute

Hi team,

Customer Max Mustermann (max.mustermann@example.com, phone 415-555-0192)
called about a double charge of $847.50 on invoice INV-44021.
His employee contact at our company is Moritz Mustermann (ext. 4421).
Please resolve within 24 hours per our SLA.

Erika Mustermann
Customer Success Manager

curl -X POST http://localhost:3000/v1/transform \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $GUARDAI_API_KEY" \
  -d '{
    "input": "From: Erika Mustermann <erika.mustermann@example.com>\nTo: Support Team\nSubject: Account #AC-2026-8834 - Billing dispute\n\nHi team,\n\nCustomer Max Mustermann (max.mustermann@example.com, phone 415-555-0192) called about a double charge of $847.50 on invoice INV-44021. His employee contact at our company is Moritz Mustermann (ext. 4421). Please resolve within 24 hours per our SLA.\n\nErika Mustermann\nCustomer Success Manager",
    "policy": "rag-ingest"
  }'

From: `{{person:p_001:ad4f97591c16}}` <`{{email:e_001:1bcaef1a4aff}}`>
To: Support Team
Subject: Account #`{{customer_id:cid_001:d00188a0c40c}}` - Billing dispute

Hi team,

Customer `{{person:p_002:98e1f012a96f}}` (`{{email:e_002:3bd0ea511c7d}}`, phone `{{phone:ph_001:71453119be53}}`)
called about a double charge of `{{custom:money:x_001:601595ba1e2c}}` on invoice INV-44021.
His employee contact at our company is `{{person:p_003:e19cd15f15fb}}` (ext. 4421).
Please resolve within 24 hours per our SLA.

`{{person:p_001:ad4f97591c16}}`
Customer Success Manager

curl -X POST http://localhost:3000/v1/transform \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $GUARDAI_API_KEY" \
  -d '{
    "input": "What happened with the billing dispute from Max Mustermann?",
    "policy": "rag-query"
  }'

What happened with the billing dispute from `{{person:p_001:ad4f97591c16}}`?

Based on the support ticket, `{{person:p_002:98e1f012a96f}}` reported a double charge
of `{{custom:money:x_001:601595ba1e2c}}` on invoice INV-44021 for account `{{customer_id:cid_001:d00188a0c40c}}`.
The ticket was assigned to `{{person:p_003:e19cd15f15fb}}` for resolution within
24 hours per SLA. The original report was filed by `{{person:p_001:ad4f97591c16}}`
from the Customer Success team.

curl -X POST http://localhost:3000/v1/rehydrate \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $GUARDAI_API_KEY" \
  -d '{
    "output": "<LLM answer with tokens>",
    "session_state": "<encrypted-blob-from-ingest>",
    "output_channel": "internal_summary"
  }'

Based on the support ticket, Max Mustermann reported a double charge
of $847.50 on invoice INV-44021 for account AC-2026-8834. The ticket
was assigned to Moritz Mustermann for resolution within 24 hours per SLA.
The original report was filed by Erika Mustermann from the Customer Success team.

name: rag-ingest
version: "1.0"
rules:
  - entity_type: person
    protection_level: 2
    action: tokenize
    restore_mode: full
  - entity_type: email
    protection_level: 2
    action: tokenize
    restore_mode: full
  - entity_type: phone
    protection_level: 2
    action: tokenize
    restore_mode: masked
  - entity_type: customer_id
    protection_level: 2
    action: tokenize
    restore_mode: full
  - entity_type: ssn
    protection_level: 1
    action: redact

defaults:
  protection_level: 2
  action: tokenize
  restore_mode: full

metadata_policy:
  expose_gender: true
  expose_formality: true
  expose_language: true
  expose_role: true

channel_rules:
  internal_summary:
    person: full
    email: full
    phone: masked
    customer_id: full
  customer_email:
    person: partial
    email: none
    phone: none
    customer_id: masked

ingest  ->  chunk + tokenize each document, store the safe chunks in the vector DB
query   ->  tokenize the agent's question for vector search
context ->  bind retrieved chunks to the query identity AND enforce access control
answer  ->  restore the LLM's token-only answer, once per output channel
delete  ->  purge the sealed session when the case is closed

pip install -r requirements.txt

requests>=2.31,<3

# Full stack (server + Python NER detector)
GUARDAI_SESSION_SECRET=$(openssl rand -base64 32) \
  docker compose -f ../../deploy/docker/docker-compose.yml up -d

# Or the server alone (structured PII only, person tokens will be missing). The shipped config binds
# loopback under dev auth, so both env vars make the mapped port reachable.
docker run -p 3000:3000 -e GUARDAI_HOST=0.0.0.0 -e GUARDAI_ALLOW_DEV_NON_LOOPBACK=1 \
  ghcr.io/oronts/oguardai/server:0.1.0

export GUARDAI_BASE_URL="http://localhost:3000"
export GUARDAI_API_KEY="<key>"          # only if the server enforces auth
python main.py

[
  {
    "file": "refund-policy-de.md",
    "doc_id": "KB-REFUND-POLICY",
    "classification": "public",
    "owner": "policy-team",
    "tags": ["refund", "policy"]
  },
  {
    "file": "dispute-runbook-de.md",
    "doc_id": "KB-DISPUTE-RUNBOOK",
    "classification": "internal",
    "owner": "support-ops",
    "tags": ["refund", "runbook"]
  },
  {
    "file": "case-erika-mueller.txt",
    "doc_id": "CASE-2026-4471",
    "classification": "confidential",
    "owner": "support-agent-42",
    "tags": ["refund", "case"]
  }
]

# Rueckerstattungsrichtlinie (oeffentlich)

Rueckerstattungen werden innerhalb von fuenf Bankarbeitstagen nach Freigabe auf das
hinterlegte Konto ausgezahlt. Bei einer nachgewiesenen Fehlbuchung genuegt die
Angabe der Vorgangsnummer; ein separater Nachweis ist nicht erforderlich.

Betraege ueber 1.000 EUR erfordern eine Zweitfreigabe durch die Teamleitung. Die
Kundin oder der Kunde wird nach Auszahlung automatisch per gesichertem Posteingang
informiert.

# Runbook: Rueckerstattung bearbeiten (intern)

1. Vorgang im Kernbanksystem oeffnen und den Status pruefen.
2. Bei einer Fehlbuchung die Korrektur anstossen und die Auszahlung auf die
   hinterlegte IBAN freigeben.
3. Eskalation bei Betraegen ueber 1.000 EUR an die Teamleitung unter
   eskalation@musterbank.example, Durchwahl 089 5550 4400.
4. Kundenkontakt ausschliesslich ueber den gesicherten Posteingang. Kontodaten
   werden niemals telefonisch bestaetigt.

Vorgangsakte (vertraulich)

Kundin: Erika Mustermann
Kundennummer: KDN-4471-2026
Kontakt: erika.mustermann@example.de, Telefon 0170 5551234
Geburtsdatum: 04.03.1987

Vorgang: Rueckerstattung einer doppelten Abbuchung ueber 189,00 EUR.
Auszahlung auf IBAN DE97 9900 0000 0123 4567 89.
Status: freigegeben am 2026-07-06, Auszahlung veranlasst.

Hinweis: Die Kundin nannte zur Legitimation ungefragt eine US-Sozialversicherungs-
nummer 123-45-6789. Diese gehoert nicht in die Akte und darf kein Modell erreichen.

"""Guarded RAG support copilot for Musterbank AG (a fictional German retail bank),
built on OGuardAI's /v1/rag/* endpoints (raw HTTP).

THE PROBLEM. An internal support copilot answers agent questions from a knowledge base
that mixes three sensitivity tiers: a public refund policy, an internal dispute runbook,
and a confidential customer case file full of regulated PII (name, email, phone, IBAN,
customer number, date of birth, and a national ID a customer pasted by mistake). A naive
RAG stack embeds every document and lets vector search return whatever is on topic. The
vector store has no notion of who is asking, so an over-permissioned retrieval hands the
LLM confidential rows it was never cleared to see. Sending that raw is a GDPR incident;
stripping it leaves the copilot unable to answer a real customer question.

THE PIPELINE (Detect -> Tokenize -> [LLM] -> Rehydrate), mapped to the RAG endpoints:

    ingest   -> chunk + tokenize each document; store the safe chunks in the vector DB
    query    -> tokenize the agent's question before it hits vector search
    context  -> bind the retrieved chunks to the query identity AND enforce row-level
                access control at the trust boundary (this is the load-bearing step)
    answer   -> restore the LLM's token-only answer per output channel
    delete   -> purge the sealed session when the case is closed (data minimization)

WHY THE ACCESS CONTROL LIVES HERE. Retrieval is topic-based and clearance-blind: it
returns every refund-tagged chunk. OGuardAI's `access_level` filter at the context stage
drops any chunk whose classification exceeds the caller's clearance BEFORE it reaches the
LLM, so the confidential case never enters the prompt for an agent with only internal
clearance. Missing classifications fail closed (treated as restricted). That is defense in
depth your vector store cannot give you.

Cross-document token identity is kept stable by a shared `corpus_id`: the same customer
maps to the same token in the case file and in the query, so retrieval and generation stay
coherent and the final answer restores the right person.

Prerequisites:
    pip install -r requirements.txt
    # OGuardAI server with the NER detector (person/company/location), e.g.:
    #   docker compose -f ../../deploy/docker/docker-compose.yml up -d
    # Structured PII (email, phone, IBAN, DOB, SSN) works on the builtins alone; person
    # names and the customer identity in the answer need 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 of
                                                       # the drafter. The provider is never
                                                       # contacted by this example.
    python main.py
"""

from __future__ import annotations

import json
import os
import re
import sys
from dataclasses import dataclass
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", "")

# One vertical policy governs the whole pipeline: the `enterprise` policy tokenizes every
# direct identifier (person, email, phone, address, customer_id, order, DOB), caps the IBAN
# at masked so an account number is never echoed in full, and hard-redacts SSN, passport,
# and credit-card values so a mistakenly pasted national ID can never become a restorable
# token. It also exposes safe metadata (gender, formality, language) for correct phrasing.
POLICY = os.environ.get("GUARDAI_POLICY", "enterprise")
LANG = os.environ.get("GUARDAI_LANG", "de")

# A shared corpus id gives the same value the same token across every ingest and the query.
CORPUS = "aurelis-support-kb"

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

# Presence of a provider key only flips the printed label; the drafter is a deterministic
# offline stand-in so the example runs without network egress. See draft_answer().
LLM_LABEL = "OpenAI (token-only)" if os.environ.get("OPENAI_API_KEY") else "offline template drafter"

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

# Raw sentinels from the confidential case file and the internal runbook, split by which detection
# tier masks them. The STRUCTURED values are caught by the builtin regex detectors on every lane
# (the SSN is hard redacted), so any one crossing to the vector store or the model is a hard
# failure. The customer name needs NER; whether a raw name crossing is a failure or a noted
# degradation is decided by ner_active, not by inspecting the output. The bare case number matches
# no builtin detector, so it is not sentineled here and is never claimed protected.
STRUCTURED_MARKERS = {
    "email erika.mustermann@example.de": "erika.mustermann@example.de",
    "email eskalation@musterbank.example": "eskalation@musterbank.example",
    "phone 0170 5551234": "0170 5551234",
    "phone 089 5550 4400": "089 5550 4400",
    "dob 04.03.1987": "04.03.1987",
    "iban DE97 9900 0000 0123 4567 89": "DE97 9900 0000 0123 4567 89",
    "ssn 123-45-6789": "123-45-6789",
}
NAME_MARKERS = ("Erika Mustermann",)

class GuardError(RuntimeError):
    """An OGuardAI request that did not succeed. The pipeline fails closed: on any error we
    stop rather than fall back to embedding, retrieving, or answering on raw PII."""

    def __init__(self, code: str, message: str, path: str) -> None:
        super().__init__(f"{code} at {path}: {message}")
        self.code = code


def post(path: str, body: dict) -> dict:
    resp = requests.post(f"{BASE}{path}", json=body, headers=HEADERS, timeout=60)
    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 fall back to the raw upstream body: it can echo request PII. Keep OGuardAI's own
        # structured error string when present, else a generic status-only message.
        error = detail.get("error")
        message = error if isinstance(error, str) else f"OGuardAI request failed (HTTP {resp.status_code})"
        raise GuardError(code, message, path)
    return resp.json()


# --- Token helpers --------------------------------------------------------------------

def token_type(inner: str) -> str:
    """Type key from a token body: 'person:p_001:hash' -> 'person'."""
    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 to the EXACT token string (hash included) the model must echo.
    Short forms like {{person:p_001}} do not resolve; the hash is load-bearing."""
    out: dict[str, str] = {}
    for inner in TOKEN_RE.findall(safe_text):
        out.setdefault(token_type(inner), "{{" + inner + "}}")
    return out

# --- Knowledge base + ingest ----------------------------------------------------------

@dataclass
class StoredChunk:
    """One tokenized chunk as it would sit in a vector store: the safe text (embedded and
    searched), its source classification, and the sealed per-chunk session that lets the
    context stage restore this chunk's tokens later."""

    doc_id: str
    classification: str
    safe_text: str
    chunk_session_state: str


def load_manifest() -> list[dict]:
    return json.loads((KB / "manifest.json").read_text(encoding="utf-8"))


def ingest_all() -> list[StoredChunk]:
    """Stage 1: chunk and tokenize every catalogued document, carrying its classification
    into OGuardAI as document metadata. Only tokenized chunks leave the boundary; the raw
    files never reach the vector store."""
    stored: list[StoredChunk] = []
    for entry in load_manifest():
        text = (KB / entry["file"]).read_text(encoding="utf-8")
        res = post(
            "/v1/rag/ingest",
            {
                "text": text,
                "corpus_id": CORPUS,
                "policy": POLICY,
                "language": LANG,
                "chunking_strategy": "paragraph",
                "document_metadata": {
                    "document_id": entry["doc_id"],
                    "classification": entry["classification"],
                    "owner": entry["owner"],
                    "tags": entry["tags"],
                },
            },
        )
        for chunk in res["chunks"]:
            stored.append(
                StoredChunk(
                    doc_id=entry["doc_id"],
                    classification=entry["classification"],
                    safe_text=chunk["safe_text"],
                    chunk_session_state=chunk["chunk_session_state"],
                )
            )
    return stored


# --- Retrieval (a stand-in for the vector store) --------------------------------------

_TOPIC_TERMS = ("rueckerstattung", "erstattung", "fehlbuchung", "refund", "rueckzahlung")


def retrieve(stored: list[StoredChunk], safe_query: str) -> list[StoredChunk]:
    """Simulated vector search. It is clearance-blind on purpose: it returns every chunk
    that is on topic, exactly like a real ANN index that has no row-level security. The
    access decision is deferred to the context stage, where OGuardAI can enforce it at the
    trust boundary. Matching is on the shared topic terms and on any token the query and a
    chunk have in common (cross-document identity via the shared corpus)."""
    query_tokens = set(TOKEN_RE.findall(safe_query))
    hits: list[StoredChunk] = []
    for chunk in stored:
        hay = chunk.safe_text.lower()
        on_topic = any(term in hay for term in _TOPIC_TERMS)
        shares_identity = bool(query_tokens & set(TOKEN_RE.findall(chunk.safe_text)))
        if on_topic or shares_identity:
            hits.append(chunk)
    return hits

# --- The LLM stand-in -----------------------------------------------------------------
# In production this is ONE call to your chat model. It reads OPENAI_API_KEY exactly like a
# real client would and is handed only the safe query and the safe context, so it sees
# tokens and safe metadata, never raw PII. To keep the example deterministic and offline it
# composes a token-only German answer from the retrieved chunks; the provider is never
# contacted here.

def draft_answer(safe_chunks: list[str]) -> str:
    """Build a token-only answer that echoes the exact tokens the rehydrator will resolve.
    When a confidential case chunk is present its person and IBAN tokens are cited; when the
    caller lacks clearance those chunks were already filtered out, so the answer stays on the
    general policy and carries no customer token."""
    joined = " ".join(safe_chunks)
    tok = token_map(joined)
    person = tok.get("person")
    iban = tok.get("iban")

    lead = f"Guten Tag {person}," if person else "Guten Tag,"
    policy_line = ("Rueckerstattungen werden innerhalb von fuenf Bankarbeitstagen nach "
                   "Freigabe ausgezahlt.")
    if person and iban:
        case_line = (f" Die Rueckerstattung fuer {person} ist freigegeben und wird auf "
                     f"{iban} ausgezahlt.")
    elif person:
        case_line = f" Der Vorgang fuer {person} ist freigegeben."
    else:
        case_line = (" Fuer kundenbezogene Details ist eine hoehere Freigabestufe "
                     "erforderlich.")
    return f"{lead} {policy_line}{case_line}"


# --- Pipeline stages 2-5 --------------------------------------------------------------

def run_query(question: str) -> dict:
    """Stage 2: tokenize the agent's question before vector search."""
    return post("/v1/rag/query", {"query": question, "corpus_id": CORPUS, "policy": POLICY, "language": LANG})


def run_context(retrieved: list[StoredChunk], query_state: str, access_level: str) -> dict:
    """Stage 3: bind the retrieved chunks to the query identity and enforce access control.
    `access_level` is the caller's clearance; any chunk classified above it is filtered out
    before the LLM sees it. `document_sessions` threads each chunk's ingest session so the
    surviving chunks can be restored in the final answer."""
    return post(
        "/v1/rag/context",
        {
            "chunks": [c.safe_text for c in retrieved],
            "chunk_classifications": [c.classification for c in retrieved],
            "document_sessions": [c.chunk_session_state for c in retrieved],
            "access_level": access_level,
            "session_state": query_state,
            "corpus_id": CORPUS,
            "policy": POLICY,
            "language": LANG,
        },
    )


def answer_per_channel(answer: str, context_state: str) -> dict[str, dict]:
    """Stage 4: restore the token-only answer once per output channel, from ONE sealed
    session. The channel selects the restore rule; the policy is the ceiling, so a channel
    can only tighten what the base rule allows, never widen it."""
    views: dict[str, dict] = {}
    for channel in ("user_output", "export", "log_safe"):
        res = post("/v1/rag/answer", {"answer": answer, "session_state": context_state, "output_channel": channel})
        views[channel] = res
    return views


def teardown(context_state: str) -> dict:
    """Stage 5: purge the sealed session. Closing the case invalidates the token map so no
    residual mapping can be replayed later. This is the data-minimization / right-to-erasure
    handle for a per-request RAG session."""
    return post("/v1/rag/delete", {"session_state": context_state})

# --- Boundary guard -------------------------------------------------------------------

def fetch_ner_active() -> bool:
    """The one leak-exemption signal: whether this runtime is configured to run NER, read from
    /v1/capabilities. If it cannot be read, treat NER as available so a raw customer name is a
    hard failure rather than a silently excused leak."""
    try:
        resp = requests.get(f"{BASE}/v1/capabilities", headers=HEADERS, timeout=10)
        if resp.status_code >= 400:
            return True
        val = resp.json().get("ner_active", True)
        return val is not False
    except (requests.RequestException, ValueError):
        return True


def check_boundary(where: str, text: str, ner_available: bool) -> list[str]:
    """Guard one string that crosses to the vector store or the model. Any structured marker
    present is a leak that fails on every lane. A raw name fails only when the runtime runs NER;
    otherwise it is a printed note, never silent. Returns the name notes recorded, if any."""
    leaked = sorted(label for label, raw in STRUCTURED_MARKERS.items() if raw in text)
    if leaked:
        raise SystemExit(f"FAIL: structured PII crossed to {where}: {', '.join(leaked)}")
    notes: list[str] = []
    for name in NAME_MARKERS:
        if name in text:
            if ner_available:
                raise SystemExit(
                    f"FAIL: the name {name} crossed to {where}; "
                    "this runtime runs NER, so names must be tokenized"
                )
            print(f"   NOTE: the name {name} crossed to {where}; this runtime runs without NER, "
                  "so names are not tokenized. Start the full stack for name-level protection.")
            notes.append(name)
    return notes

# --- Scenario driver ------------------------------------------------------------------

def rule(title: str) -> None:
    print("=" * 82)
    print(title)
    print("=" * 82)


def run_scenario(stored: list[StoredChunk], question: str, *, clearance: str, label: str,
                 ner_available: bool) -> list[str]:
    """One agent, one clearance, end to end. Prints what left the boundary at each stage so
    the trust boundary is auditable: the safe context the LLM saw, the filtered chunk count,
    the token-only draft, and the answer restored per channel. Returns the name notes recorded."""
    rule(f"AGENT CLEARANCE = {clearance.upper()}  ({label})")
    notes: list[str] = []

    q = run_query(question)
    print(f"   safe query (search): {q['safe_query']}")
    notes += check_boundary("the vector search", q["safe_query"], ner_available)

    retrieved = retrieve(stored, q["safe_query"])
    print(f"   retrieved (topic match, clearance-blind): {len(retrieved)} chunk(s) "
          f"from {sorted({c.doc_id for c in retrieved})}")

    ctx = run_context(retrieved, q["session_state"], clearance)
    kept = ctx["safe_chunks"]
    print(f"   access filter      : {ctx.get('chunks_filtered', 0)} chunk(s) above '{clearance}' "
          f"dropped at the boundary, {len(kept)} passed to the LLM")
    for c in kept:
        print(f"     safe context     : {c}")
        notes += check_boundary("the retrieved context", c, ner_available)

    draft = draft_answer(kept)
    print(f"\n   drafting with      : {LLM_LABEL}")
    print(f"   token-only answer  : {draft}")
    notes += check_boundary("the token-only answer", draft, ner_available)

    views = answer_per_channel(draft, ctx["session_state"])
    print("\n   restored per output channel (one sealed session):")
    print(f"     agent (user_output): {views['user_output']['restored_answer']}")
    print(f"     analytics (export) : {views['export']['restored_answer']}")
    print(f"     audit (log_safe)   : {views['log_safe']['restored_answer']}")
    resolved = views["user_output"].get("tokens_resolved")
    unresolved = views["user_output"].get("tokens_unresolved") or []
    print(f"     tokens_resolved={resolved}  tokens_unresolved={len(unresolved)}")

    purge = teardown(ctx["session_state"])
    print(f"\n   session teardown   : deleted={purge.get('deleted')} "
          f"entities_purged={purge.get('entities_purged')}\n")
    return notes


def run_fail_closed_proof() -> None:
    """A corrupted session must not leak. rag/answer on a tampered blob returns a typed error
    and the pipeline stops rather than emitting an unresolved or guessed answer."""
    rule("FAIL-CLOSED PROOF  (a tampered session must not resolve)")
    try:
        post("/v1/rag/answer", {"answer": "Guten Tag {{person:p_001:deadbeef}}.", "session_state": "tampered-session-blob"})
        raise SystemExit("fail-closed check failed: a tampered session resolved an answer")
    except GuardError as exc:
        print(f"   rejected with {exc.code}; no answer was produced.\n")


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

    ner_available = fetch_ner_active()
    if not ner_available:
        print("WARNING: NER is not active on this runtime; person tokens and the restored "
              "customer identity in the answer will be missing. Structured PII (email, phone, "
              "IBAN, DOB, SSN) still works. Start the full stack via docker compose.\n",
              file=sys.stderr)

    print(f"Server: {BASE}  policy={POLICY}  corpus={CORPUS}  drafter={LLM_LABEL}\n")

    name_notes: list[str] = []
    try:
        rule("1. INGEST  (chunk + tokenize the catalogued knowledge base)")
        stored = ingest_all()
        for c in stored:
            print(f"   [{c.classification:12}] {c.doc_id}: {c.safe_text.strip()[:96]}"
                  f"{'...' if len(c.safe_text.strip()) > 96 else ''}")
        print()
        for c in stored:
            name_notes += check_boundary(f"the vector store ({c.doc_id})", c.safe_text, ner_available)

        question = "Wie ist der Stand der Rueckerstattung fuer Erika Mustermann?"
        # Same question, two clearances. Internal cannot see the confidential case; the
        # customer's PII never enters the prompt. Confidential can, and the answer restores it.
        name_notes += run_scenario(stored, question, clearance="internal",
                                   label="support agent, standard desk", ner_available=ner_available)
        name_notes += run_scenario(stored, question, clearance="confidential",
                                   label="case handler on this file", ner_available=ner_available)
        run_fail_closed_proof()
    except GuardError as exc:
        raise SystemExit(f"OGuardAI request failed: {exc}")

    if name_notes:
        unique = ", ".join(sorted(set(name_notes)))
        print("Done. The detectable structured PII (email, phone, IBAN, DOB, SSN) stayed inside "
              "the runtime and confidential rows never crossed a clearance they were not entitled "
              f"to; the customer name(s) {unique} reached the vector store and the model because "
              "this runtime runs without NER. Start the full stack for name-level protection.")
    else:
        print("Done. The detectable structured PII (email, phone, IBAN, DOB, SSN) stayed inside "
              "the trusted zone as tokens; the vector store, the model, and every downstream sink saw "
              "them as tokens only, and confidential rows never crossed a clearance they were not "
              "entitled to.")
    print("Note: the case file's bare customer number carries no prefix the builtin detector "
          "matches, so it is below the detection floor and is not tokenized or claimed protected.")


if __name__ == "__main__":
    main()

1. INGEST  (chunk + tokenize the catalogued knowledge base)
   [public      ] KB-REFUND-POLICY: Rueckerstattungen werden innerhalb von fuenf Bankarbeitstagen ...
   [internal    ] KB-DISPUTE-RUNBOOK: ... Eskalation ... an {{email:e_001:...}}, Durchwahl {{phone:ph_001:...}}.
   [confidential] CASE-2026-4471: Kundin {{person:p_001:...}} ... IBAN {{iban:ib_001:...}} ... [REDACTED:ssn] ...

==================================================================================
AGENT CLEARANCE = INTERNAL  (support agent, standard desk)
==================================================================================
   safe query (search): Wie ist der Stand der Rueckerstattung fuer {{person:p_001:...}}?
   retrieved (topic match, clearance-blind): chunk(s) from ['CASE-2026-4471', 'KB-DISPUTE-RUNBOOK', 'KB-REFUND-POLICY']
   access filter      : the confidential CASE-2026-4471 chunk(s) dropped at the boundary, public + internal passed to the LLM
   token-only answer  : Guten Tag, Rueckerstattungen werden ... Fuer kundenbezogene Details ist eine hoehere Freigabestufe erforderlich.
   restored per output channel (one sealed session):
     agent (user_output): Guten Tag, Rueckerstattungen werden innerhalb von fuenf Bankarbeitstagen ...
   session teardown   : deleted=True entities_purged=...

==================================================================================
AGENT CLEARANCE = CONFIDENTIAL  (case handler on this file)
==================================================================================
   access filter      : nothing above 'confidential' dropped, every retrieved chunk passed to the LLM
   token-only answer  : Guten Tag {{person:p_001:...}}, ... wird auf {{iban:ib_001:...}} ausgezahlt.
   restored per output channel (one sealed session):
     agent (user_output): Guten Tag Erika Mustermann, ... wird auf D*** **** **** **** **** *0 ausgezahlt.
     analytics (export) : Guten Tag (person on file), ... auf (iban on file) ...
     audit (log_safe)   : Guten Tag [REDACTED], ... auf [REDACTED] ...

FAIL-CLOSED PROOF  (a tampered session must not resolve)
   rejected with GUARDAI_...; no answer was produced.