OpenAI

support ticket (raw PII)
    |
    v
OGuardAI /v1/transform   ->  safe text ({{type:id:fingerprint}} tokens) + sealed session
    |
    v
openai.chat.completions.create(system = safe metadata, user = safe text)
    |                             (the model sees tokens only, never a raw value)
    v
OGuardAI /v1/rehydrate   ->  real values restored, per output channel and restore mode

# Start the runtime WITH the NER detector (person detection needs it). 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,internal_summary,export,log_safe
docker compose -f ../../deploy/docker/docker-compose.yml up -d

pip install -r requirements.txt

export GUARDAI_URL=http://localhost:3000
export GUARDAI_API_KEY="$GUARDAI_API_KEY_1"   # the bundled stack runs api_key auth
export OPENAI_API_KEY="sk-..."     # optional: draft with the real model
python guard.py

draft_reply = _openai_draft if os.getenv("OPENAI_API_KEY") else _offline_draft

Betreff: Beschaedigte Lieferung zu Bestellung ORD-2026-4471

Sehr geehrtes Serviceteam,

mein Name ist Erika Mustermann. Die Lieferung zu meiner Bestellung
ORD-2026-4471 ist beschaedigt angekommen; das Regal hat einen Riss in
der Seitenwand. Ich bitte um Erstattung des vollen Kaufpreises von
249,00 EUR auf mein Konto DE97 9900 0000 0123 4567 89.

Sie erreichen mich unter erika.mustermann@mail.example oder telefonisch unter
+49 89 55512340. Meine Kundennummer lautet 5541882.

Mit freundlichen Gruessen
Erika Mustermann

requests>=2.31,<3
# Only the live path (a real OPENAI_API_KEY) needs the OpenAI SDK. The default
# offline stub uses only requests and needs no provider account.
openai>=1.0.0

from __future__ import annotations

import os
import re
from pathlib import Path

import requests

GUARDAI_URL = (os.getenv("GUARDAI_URL") or "http://localhost:3000").rstrip("/")
GUARDAI_API_KEY = os.getenv("GUARDAI_API_KEY", "")
POLICY = os.getenv("GUARDAI_POLICY", "german-support")
LANGUAGE = os.getenv("GUARDAI_LANG", "de")

HERE = Path(__file__).resolve().parent
TICKET_FILE = HERE / "support_ticket.txt"

# Raw values that must never survive into the tokenized text sent to the model. The structured values
# are caught by the built-in regex detectors, so a leak is a hard failure in every mode. The name
# needs the NER detector: a leak is a hard failure when NER is active, and a printed NOTE when the
# runtime runs without it.
NER_SENSITIVE = ["Erika Mustermann"]
STRUCTURED_SENSITIVE = [
    "erika.mustermann@mail.example",
    "+49 89 55512340",
    "DE97 9900 0000 0123 4567 89",
    "ORD-2026-4471",
]

_TOKEN_RE = re.compile(r"\{\{([^}]+)\}\}")
RULE = "=" * 82


class GuardError(RuntimeError):
    """An OGuardAI request that did not succeed. The example fails closed on this: it never falls
    back to sending raw text to the model."""

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


def _headers() -> dict[str, str]:
    headers = {"Content-Type": "application/json"}
    if GUARDAI_API_KEY:
        headers["X-API-Key"] = GUARDAI_API_KEY
    return headers


def _post(path: str, payload: dict) -> dict:
    resp = requests.post(f"{GUARDAI_URL}{path}", json=payload, headers=_headers(), timeout=30)
    if resp.status_code >= 400:
        body = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
        # Never echo the raw upstream body: it can carry the request's PII. Trust only OGuardAI's
        # own structured error string; otherwise report the status code generically.
        error = body.get("error")
        message = error if isinstance(error, str) else f"OGuardAI request failed (HTTP {resp.status_code})"
        raise GuardError(body.get("code", f"HTTP_{resp.status_code}"), message)
    return resp.json()

def fetch_ner_active() -> bool:
    """Read ner_active from GET /v1/capabilities: the CONFIGURED detector intent, not sidecar
    liveness. It is the one signal that decides whether a person name is expected to tokenize. When it
    cannot be read, treat NER as available so an untokenized name fails rather than passing silently."""
    try:
        resp = requests.get(f"{GUARDAI_URL}/v1/capabilities", headers=_headers(), timeout=10)
        resp.raise_for_status()
        value = resp.json().get("ner_active")
    except (requests.RequestException, ValueError):
        return True
    return value if isinstance(value, bool) else True


def transform(text: str) -> dict:
    """Replace PII with semantic tokens under the configured policy."""
    return _post("/v1/transform", {"input": text, "input_type": "text", "policy": POLICY, "language": LANGUAGE})


def rehydrate(output: str, session_state: str, *, channel: str, mode: str = "full") -> dict:
    """Restore tokens to real values for one output channel and restore mode."""
    return _post(
        "/v1/rehydrate",
        {"output": output, "session_state": session_state, "output_channel": channel, "restore_mode": mode},
    )


def tokens_by_type(safe_text: str) -> dict[str, str]:
    """Map ``person``/``order``/``iban``/... to the EXACT token string (fingerprint included) found
    in ``safe_text``. The fingerprint is load-bearing: a short ``{{person:p_001}}`` does not resolve,
    so the drafter must echo the full token verbatim."""
    out: dict[str, str] = {}
    for inner in _TOKEN_RE.findall(safe_text):
        out.setdefault(inner.split(":")[0], "{{" + inner + "}}")
    return out


def check_crossing(text: str, where: str, *, ner_available: bool, name_notes: list[str]) -> None:
    """Fail closed on anything sensitive that crossed to the model. Structured PII is caught by the
    built-in regex detectors, so a leak is a hard failure on every lane. A person name is a hard
    failure when NER is active; when the runtime runs without NER it cannot tokenize the name, so the
    leak is reported as a NOTE and collected for the closing summary."""
    structured = [value for value in STRUCTURED_SENSITIVE if value in text]
    if structured:
        raise SystemExit(f"FAIL: structured PII reached {where}: {structured}")
    names = [value for value in NER_SENSITIVE if value in text]
    if not names:
        print(f"  leak check on {where}: PASS, no tracked raw value present")
        return
    if ner_available:
        raise SystemExit(f"FAIL: a name reached {where} while NER is active: {names}")
    for name in names:
        print(f"  NOTE: {name} reached {where}; without NER the runtime cannot tokenize the name")
        name_notes.append(name)

# --- The LLM step --------------------------------------------------------------------------------
# In production this is one call to your chat model. It is handed only the tokenized text and the
# safe metadata, so it never sees a raw value. The offline stub keeps the example deterministic and
# free of network egress; the OpenAI path is the single-line switch below.


def _offline_draft(safe_text: str, system: str) -> str:
    """Deterministic stand-in for the model: a formal German reply that echoes the exact tokens it
    saw, so every token resolves on restore. It invents no name, order, account, or contact detail.
    The salutation is intentionally gender-neutral ("Guten Tag {token}"); the model does not know the
    customer's gender, so OGuardAI injects the correct "Frau"/"Herr" honorific during restore."""
    _ = system  # the offline stub does not need the system prompt; the live path does
    t = tokens_by_type(safe_text)
    return (
        f"Guten Tag {t.get('person', '')},\n\n"
        f"vielen Dank fuer Ihre Nachricht. Wir haben Ihre Bestellung {t.get('order', '')} geprueft "
        f"und veranlassen die Erstattung des vollen Kaufpreises auf Ihr Konto {t.get('iban', '')}.\n\n"
        f"Eine Bestaetigung senden wir an {t.get('email', '')}; fuer Rueckfragen erreichen wir Sie "
        f"unter {t.get('phone', '')}.\n\n"
        "Mit freundlichen Gruessen\nIhr Serviceteam der Musterfirma"
    )


def _openai_draft(safe_text: str, system: str) -> str:
    """Draft with the real model. It receives only the tokenized text and the safe-metadata system
    prompt; no raw value ever reaches OpenAI."""
    from openai import OpenAI  # lazy import: only needed when OPENAI_API_KEY is set

    completion = OpenAI().chat.completions.create(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        temperature=0.2,
        messages=[{"role": "system", "content": system}, {"role": "user", "content": safe_text}],
    )
    return completion.choices[0].message.content or ""


# The one-line switch: real model when a key is present, deterministic stub otherwise.
draft_reply = _openai_draft if os.getenv("OPENAI_API_KEY") else _offline_draft
LLM_LABEL = "OpenAI (live, token-only)" if os.getenv("OPENAI_API_KEY") else "offline deterministic stub"


def build_system_prompt(entity_context: list[dict]) -> str:
    """Build the system prompt from the SAFE metadata only. entity_context carries the token, its
    type, and safe hints (gender/formality/language); it never carries a raw value, so the whole
    prompt is safe to send to an external provider."""
    lines = [
        "Du bist ein deutscher Kundenservice-Agent der Musterfirma GmbH.",
        "Antworte formell und freundlich auf Deutsch in der Sie-Form.",
        "Der Eingabetext enthaelt Platzhalter der Form {{typ:id:fingerprint}}.",
        "Gib jeden Platzhalter ZEICHENGENAU und unveraendert wieder.",
        "Erfinde niemals neue Platzhalter, Namen, Kontonummern oder Kontaktdaten.",
    ]
    if entity_context:
        lines.append("\nKontext zu den Platzhaltern (nur sichere Metadaten, keine Rohwerte):")
        for ctx in entity_context:
            hint = ", ".join(
                f"{key}={ctx[key]}" for key in ("gender", "formality", "language") if ctx.get(key)
            )
            lines.append(f"  {ctx['token']} = {ctx['type']}" + (f" ({hint})" if hint else ""))
    return "\n".join(lines)

def section(title: str) -> None:
    print("\n" + RULE)
    print(title)
    print(RULE)


def run(ner_available: bool) -> list[str]:
    ticket = TICKET_FILE.read_text(encoding="utf-8").strip()
    name_notes: list[str] = []

    section("1) transform: PII leaves the boundary as tokens, never as raw values")
    result = transform(ticket)
    safe_text = result["safe_text"]
    session_state = result.get("session_state") or ""
    print("  safe text (this is all the model receives):\n")
    print("    " + safe_text.replace("\n", "\n    "))
    check_crossing(safe_text, "the safe text sent to the model", ner_available=ner_available, name_notes=name_notes)
    if not session_state:
        raise GuardError("GUARDAI_NO_SESSION", "transform returned no session_state; refusing to draft")

    section("2) draft with the model (tokens in, tokens out)")
    system = build_system_prompt(result.get("entity_context") or [])
    print(f"  drafting with: {LLM_LABEL}")
    draft = draft_reply(safe_text, system)
    print("  model draft (still tokenized):\n")
    print("    " + draft.replace("\n", "\n    "))
    check_crossing(draft, "the model draft", ner_available=ner_available, name_notes=name_notes)

    section("3) rehydrate per output channel (same draft, different reader)")
    # The customer sees their own order and contact details in full; german-support leaves
    # customer_email untightened. Every other channel tightens contact details (and names) back down.
    customer = rehydrate(draft, session_state, channel="customer_email", mode="full")
    print("  [customer_email] the reply the customer receives (own order + contact, IBAN masked):")
    print("    " + customer["restored_text"].replace("\n", "\n    "))
    print(f"    tokens_resolved={customer.get('tokens_resolved')} unresolved={customer.get('tokens_unresolved')}")

    export = rehydrate(draft, session_state, channel="export", mode="full")
    print("\n  [export] the analytics copy (name/order/contact tightened to masks by the channel):")
    print("    " + export["restored_text"].split("\n")[0])

    audit = rehydrate(draft, session_state, channel="log_safe", mode="full")
    print("\n  [log_safe] the audit-log copy (everything redacted):")
    print("    " + audit["restored_text"].split("\n")[0])

    section("4) restore-mode ladder for the salutation (internal_summary channel)")
    # The restore mode is an independent knob from the channel. Walk it for the salutation line so
    # the modes are concrete side by side; a gender-honorific name shows full/formatted, then narrows.
    for mode in ("full", "formatted", "partial", "masked"):
        line = rehydrate(draft, session_state, channel="internal_summary", mode=mode)["restored_text"].split("\n")[0]
        print(f"  {mode:9} -> {line}")

    section("5) fail closed: a tampered session is rejected, not half-restored")
    try:
        rehydrate("Guten Tag {{person:p_001:deadbeef0000}}.", session_state[:-8] + "deadbeef", channel="user_output")
        raise SystemExit("fail-closed check failed: a tampered session was accepted")
    except GuardError as exc:
        print(f"  rejected with [{exc.code}]; no raw value was returned")

    return name_notes

def main() -> int:
    print(f"OGuardAI + OpenAI direct   runtime={GUARDAI_URL}  policy={POLICY}  drafter={LLM_LABEL}")
    try:
        requests.get(f"{GUARDAI_URL}/v1/health", headers=_headers(), timeout=10)
    except requests.RequestException as exc:
        print(f"\ncannot reach the OGuardAI server at {GUARDAI_URL}: {exc}")
        print("start it first (see the README).")
        return 1

    ner_available = fetch_ner_active()
    if not ner_available:
        print("  NOTE: NER is not active; the person token and its gendered salutation will be missing.")
        print("  Structured PII (email, phone, IBAN) still tokenizes.")

    try:
        name_notes = run(ner_available)
    except GuardError as exc:
        print(f"\nOGuardAI request failed: {exc}")
        return 1

    print("\n" + RULE)
    if name_notes:
        reached = ", ".join(dict.fromkeys(name_notes))
        print("The detectable identifiers (email, phone, IBAN, order) stayed inside the runtime;")
        print(f"the name {reached} reached the model because this runtime runs without NER.")
        print("Enable the NER detector to tokenize person names too.")
    else:
        print("The detectable identifiers (email, phone, IBAN, order) crossed only as tokens;")
        print("OpenAI saw tokens and safe metadata only; the reply was restored for the customer,")
        print("tightened for analytics, and redacted for the audit log.")
    print("Note: the ticket's bare customer number is below the builtin CustomerId detection")
    print("floor (no matching prefix), so it is not tokenized and is not claimed protected.")
    print(RULE)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

OGuardAI + OpenAI direct   runtime=http://localhost:3000  policy=german-support  drafter=offline deterministic stub

1) transform: PII leaves the boundary as tokens, never as raw values
  safe text (this is all the model receives):

    ...
  leak check on the safe text sent to the model: PASS, no tracked raw value present

2) draft with the model (tokens in, tokens out)
  drafting with: offline deterministic stub
  model draft (still tokenized):

    ...
  leak check on the model draft: PASS, no tracked raw value present

3) rehydrate per output channel (same draft, different reader)
  [customer_email] the reply the customer receives (own order + contact, IBAN masked):
    ...
    tokens_resolved=... unresolved=...

  [export] the analytics copy (name/order/contact tightened to masks by the channel):
    ...

  [log_safe] the audit-log copy (everything redacted):
    ...

4) restore-mode ladder for the salutation (internal_summary channel)
  full      -> ...
  formatted -> ...
  partial   -> ...
  masked    -> ...

5) fail closed: a tampered session is rejected, not half-restored
  rejected with [...]; no raw value was returned

The detectable identifiers (email, phone, IBAN, order) crossed only as tokens;
OpenAI saw tokens and safe metadata only; the reply was restored for the customer,
tightened for analytics, and redacted for the audit log.
Note: the ticket's bare customer number is below the builtin CustomerId detection
floor (no matching prefix), so it is not tokenized and is not claimed protected.