Guarded Document AI

document (raw PII, any of: .txt / .eml)
    |
    v
OGuardAI ingest    ->  /v1/transform (text)  or  /v1/transform/file (document route)
    |                   whole document tokenized: {{type:id:fingerprint}}, regulated ids redacted
    v
offline summarizer ->  reads the SAFE text, writes a token-only summary + extraction
    |                   (the model never sees a raw value)
    v
OGuardAI rehydrate ->  restore per channel: internal_summary, export, log_safe

DIENSTLEISTUNGSVERTRAG

zwischen

Musterfirma GmbH, Musterstrasse 12, 20095 Hamburg
(nachfolgend "Auftraggeber"), vertreten durch Frau Erika Mustermann,

und

Muster & Kollegen Rechtsanwaelte, Beispielstrasse 4, 80999 Muenchen
(nachfolgend "Auftragnehmer"), vertreten durch Herrn Max Mustermann.

Aktenzeichen: 3 O 271/25

Praeambel
Der Auftragnehmer beraet den Auftraggeber in einer vertragsrechtlichen
Streitigkeit mit einem Lieferanten.

Paragraph 1 Verguetung
Das Honorar betraegt 12.500,00 EUR zzgl. Umsatzsteuer. Zahlungen erfolgen
auf das Konto IBAN DE97 9900 0000 0123 4567 89 der Kanzlei.

Paragraph 2 Kontakt
Rueckfragen richten Sie an erika.mustermann@example.de oder
telefonisch an +49 40 55598877.

Paragraph 3 Laufzeit
Der Vertrag beginnt am 01.03.2026 und laeuft bis zum 28.02.2027.

Hamburg, den 15.02.2026

From: Erika Mustermann <erika.mustermann@kontakt.example>
To: support@example.de
Subject: Re: Beschwerde zu Bestellung ORD-2026-4471
Date: Mon, 06 Jul 2026 09:14:00 +0200
Message-ID: <20260706071400.4471@example.de>
Content-Type: text/plain; charset="utf-8"

Guten Tag,

vielen Dank fuer Ihre Rueckmeldung. Wie besprochen erreichen Sie mich
tagsueber unter +49 89 55512340. Meine Kundennummer lautet 5541882.

Am 04.07.2026 schrieb support@example.de:
> Sehr geehrte Frau Mustermann, wir pruefen Ihre Bestellung ORD-2026-4471
> und melden uns in Kuerze mit einem Loesungsvorschlag.

Mit freundlichen Gruessen
Erika Mustermann

RECHNUNG Nr. RE-2026-0847

Musterbank AG
Bankstrasse 1, 60311 Frankfurt am Main
USt-IdNr: DE123456789

Rechnungsempfaenger:
Erika Mustermann
Lindenweg 8, 80331 Muenchen
Kundennummer: 5541882
E-Mail: erika.mustermann@kontakt.example

Position                          Betrag
Kontofuehrung 2026                120,00 EUR
Beratung Vermoegensaufbau         380,00 EUR
Summe netto                       500,00 EUR
Umsatzsteuer 19%                   95,00 EUR
Gesamtbetrag                      595,00 EUR

Zahlbar innerhalb von 14 Tagen auf IBAN DE97 9900 0000 0123 4567 89.

ARZTBRIEF

Musterklinik, Klinik fuer Kardiologie
Musterstrasse 1, 80999 Muenchen

Patient: Herr Max Mustermann
Geburtsdatum: 14.03.1978
Versichertennummer: A123456789
Fallnummer: FN-2026-55210

Sehr geehrte Frau Kollegin, sehr geehrter Herr Kollege,

wir berichten ueber o.g. Patienten, der sich am 02.07.2026 in unserer
kardiologischen Ambulanz vorstellte. Fuer Rueckfragen erreichen Sie uns
unter +49 89 5550100 oder kardiologie@musterklinik.example.

Diagnose: Paroxysmales Vorhofflimmern (ICD-10 I48.0). Wir empfehlen eine
orale Antikoagulation und eine ambulante Verlaufskontrolle in vier Wochen.

Mit kollegialen Gruessen
Dr. med. Erika Mustermann

pip install -r requirements.txt

# Server with the NER detector (person/company names need it)
GUARDAI_SESSION_SECRET=$(openssl rand -base64 32) \
  docker compose -f ../../deploy/docker/docker-compose.yml up -d

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

oguardai-sdk>=0.1.0
# Optional: only the OCR/image sub-demo renders a sample card image to feed the
# image route. The example self-skips that step when Pillow or the server's
# Tesseract OCR is unavailable, so the core document flow needs only the SDK.
Pillow>=10,<12
# Optional: only the live summarizer (a real OPENAI_API_KEY) needs the OpenAI SDK.
# openai>=1.0.0

from __future__ import annotations

import email
import io
import os
import re
from dataclasses import dataclass, field
from email import policy as email_policy
from pathlib import Path

from guardai_sdk import OGuardAIClient
from guardai_sdk.errors import OGuardAIError

GUARDAI_URL = os.getenv("GUARDAI_URL", "http://localhost:3000")
GUARDAI_API_KEY = os.getenv("GUARDAI_API_KEY", "") or None
LANGUAGE = os.getenv("GUARDAI_LANG", "de")

HERE = Path(__file__).resolve().parent
DOCS = HERE / "docs"

_TOKEN_RE = re.compile(r"\{\{([^}]+)\}\}")
_REDACTION_RE = re.compile(r"\[REDACTED:([a-z_]+)\]")
RULE = "=" * 86

@dataclass(frozen=True)
class Document:
    """One document in the pipeline, plus the policy that governs it and how it is ingested."""

    label: str
    filename: str
    policy: str
    ingest: str  # "text", "eml", or "file"
    caller_role: str | None = None
    # Structured raw values the builtin detectors always tokenize or redact, checked absent from
    # every string that crosses to the summarizer as an environment-independent leak gate.
    sensitive: list[str] = field(default_factory=list)
    # Person names, tokenized only when the NER detector is active. A raw name crossing is a hard
    # failure when NER is active and a recorded note otherwise, so a no-NER run never over-claims.
    names: list[str] = field(default_factory=list)
    # Identifiers with no detectable prefix (a bare customer number the builtin CustomerId pattern
    # cannot match). Never asserted as protected; recorded so the closing states honestly that they
    # reach the summarizer in the clear, outside the builtin detection floor.
    undetected: list[str] = field(default_factory=list)


DOCUMENTS = [
    Document(
        "German service contract", "contract.txt", "legal", "file",
        sensitive=["erika.mustermann@example.de", "+49 40 55598877", "DE97 9900 0000 0123 4567 89"],
        names=["Erika Mustermann", "Max Mustermann"],
    ),
    Document(
        "Customer email thread", "email_thread.eml", "german-support", "eml",
        sensitive=[
            "erika.mustermann@kontakt.example", "support@example.de",
            "+49 89 55512340", "ORD-2026-4471",
        ],
        names=["Erika Mustermann"],
        undetected=["5541882"],
    ),
    Document(
        "Bank invoice", "invoice.txt", "financial", "text",
        sensitive=[
            "erika.mustermann@kontakt.example", "DE123456789", "DE97 9900 0000 0123 4567 89",
        ],
        names=["Erika Mustermann"],
        undetected=["5541882"],
    ),
    Document(
        "Hospital discharge letter", "clinical_letter.txt", "healthcare-clinical", "text",
        caller_role="clinician",
        sensitive=[
            "kardiologie@musterklinik.example", "+49 89 30680000", "A123456789",
        ],
        names=["Max Mustermann", "Erika Mustermann"],
    ),
]

def tokens_by_type(safe_text: str) -> dict[str, str]:
    """Map each entity type to the EXACT token (fingerprint included) found in ``safe_text``. The
    fingerprint is load-bearing: a short ``{{person:p_001}}`` does not resolve, so the summarizer
    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 redaction_types(safe_text: str) -> set[str]:
    """Entity types that were redacted at ingest (``[REDACTED:iban]`` -> ``iban``). A redacted value
    never became a token, so it can never be restored."""
    return set(_REDACTION_RE.findall(safe_text))


def leaked(text: str, sensitive: list[str]) -> list[str]:
    return [value for value in sensitive if value in text]

@dataclass
class LeakLedger:
    """The single record every crossing string is folded into. Both the exit code and the closing
    summary read it, so neither can claim protection the run did not actually deliver."""

    structured: list[str] = field(default_factory=list)
    names: list[str] = field(default_factory=list)
    undetected: list[str] = field(default_factory=list)


def check_crossing(
    where: str, text: str, doc: Document, ner_available: bool, ledger: LeakLedger
) -> None:
    """Check one string that crosses to the summarizer against the document's sentinels and fold
    the outcome into the shared ledger. A structured value present is always a leak. A raw name is a
    leak only when NER is active; without NER it is the documented degradation, kept as a note. An
    undetectable bare identifier is recorded for the closing, never gated, so the run never claims to
    protect a value the builtin detectors cannot match."""
    structured = leaked(text, doc.sensitive)
    if structured:
        ledger.structured.extend(structured)
        print(f"  FAIL: structured PII reached the {where}: {structured}")
        raise SystemExit(f"structured PII reached the {where}: {structured}")
    for name in leaked(text, doc.names):
        ledger.names.append(name)
        if ner_available:
            print(f"  FAIL: a name reached the {where} while NER is active: {name!r}")
            raise SystemExit(f"a name reached the {where} while NER is active: {name!r}")
        print(f"  NOTE: {name!r} reached the {where}; names do not tokenize without the NER detector")
    for value in leaked(text, doc.undetected):
        if value not in ledger.undetected:
            ledger.undetected.append(value)

# --- The summarizer step -------------------------------------------------------------------------
# In production this is one call to your chat/summarization model, handed only the tokenized text.
# The offline stub reads the safe text and echoes back the exact tokens it saw, so the summary and
# the extraction restore cleanly and reference no invented value.


def _offline_summarize(safe_text: str) -> str:
    """Deterministic token-only summary: names the entities the document is about using only the
    tokens present in the safe text, and states which identifiers were redacted at ingest."""
    tok = tokens_by_type(safe_text)
    redacted = redaction_types(safe_text)
    parts: list[str] = []
    parties = [tok[t] for t in ("person", "company") if tok.get(t)]
    if parties:
        parts.append("Beteiligte: " + ", ".join(parties) + ".")
    reference = tok.get("order") or tok.get("case_ref") or tok.get("customer_id")
    if reference:
        parts.append(f"Referenz: {reference}.")
    if tok.get("date_of_birth"):
        parts.append(f"Geburtsdatum: {tok['date_of_birth']}.")
    if tok.get("iban"):
        parts.append(f"Bankverbindung: {tok['iban']}.")
    elif "iban" in redacted:
        parts.append("Bankverbindung: bei Erfassung unwiderruflich geschwaerzt.")
    if tok.get("email"):
        parts.append(f"Kontakt: {tok['email']}.")
    return " ".join(parts) if parts else "Keine wiederherstellbaren Entitaeten im Dokument."


def _openai_summarize(safe_text: str) -> str:
    """Summarize with the real model. It receives only the tokenized text; no raw value is sent."""
    from openai import OpenAI  # lazy import: only needed when OPENAI_API_KEY is set

    system = (
        "Du fasst deutsche Dokumente zusammen. Der Text enthaelt Platzhalter der Form "
        "{{typ:id:fingerprint}}. Gib jeden Platzhalter ZEICHENGENAU wieder und erfinde keine neuen "
        "Platzhalter oder Werte. Fasse Beteiligte, Referenzen und Kontaktdaten in wenigen Saetzen "
        "zusammen."
    )
    resp = 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 resp.choices[0].message.content or ""


summarize = _openai_summarize if os.getenv("OPENAI_API_KEY") else _offline_summarize
LLM_LABEL = "OpenAI (live)" if os.getenv("OPENAI_API_KEY") else "offline summarizer stub"

summarize = _openai_summarize if os.getenv("OPENAI_API_KEY") else _offline_summarize

def load_email_thread(path: Path) -> str:
    """Parse an .eml with the stdlib and flatten it to the text the pipeline ingests: the routing
    headers plus the plain-text body (including the quoted prior message)."""
    msg = email.message_from_bytes(path.read_bytes(), policy=email_policy.default)
    if msg.is_multipart():
        body_part = msg.get_body(preferencelist=("plain",))
        body = body_part.get_content() if body_part is not None else ""
    else:
        body = msg.get_content()
    header = f"Von: {msg['From']}\nAn: {msg['To']}\nBetreff: {msg['Subject']}"
    return f"{header}\n\n{body}".strip()

def ingest(client: OGuardAIClient, doc: Document) -> tuple[str, str, list, str | None]:
    """Detect and tokenize the whole document. Returns (safe_text, session_state, entities, note).

    A .txt contract is sent through the /v1/transform/file document route to show the multipart
    ingest path; the rest go through the text route. The clinical letter passes a caller_role so the
    role-gated healthcare policy resolves definitively instead of clamping."""
    path = DOCS / doc.filename
    context = {"caller_role": doc.caller_role} if doc.caller_role else None

    if doc.ingest == "file":
        res = client.transform_file(path.read_bytes(), doc.filename, policy=doc.policy, language=LANGUAGE)
        note = f"via /v1/transform/file  ({res.original_format}, {res.original_size} bytes)"
        return res.safe_text, res.session_state or "", res.entities, note

    text = load_email_thread(path) if doc.ingest == "eml" else path.read_text(encoding="utf-8")
    res = client.transform(text, policy=doc.policy, language=LANGUAGE, context=context)
    return res.safe_text, res.session_state or "", res.entities, "via /v1/transform (text)"

def entity_type_counts(entities: list) -> dict[str, int]:
    counts: dict[str, int] = {}
    for ent in entities:
        counts[ent.type] = counts.get(ent.type, 0) + 1
    return counts


def process(client: OGuardAIClient, doc: Document, ner_available: bool, ledger: LeakLedger) -> None:
    print("\n" + RULE)
    print(f"{doc.label}   policy={doc.policy}")
    print(RULE)

    safe_text, session_state, entities, note = ingest(client, doc)
    print(f"  ingest: {note}")
    counts = entity_type_counts(entities)
    print(f"  detected entity types: {counts or '(none; is the NER detector running?)'}")
    redacted = redaction_types(safe_text)
    if redacted:
        print(f"  redacted at ingest (never restorable): {sorted(redacted)}")

    check_crossing("safe text", safe_text, doc, ner_available, ledger)
    print(f"  leak check (structured PII): PASS, none of {len(doc.sensitive)} raw values in the safe text")

    if not session_state:
        print("  no session_state returned; skipping summarize/restore")
        return

    summary = summarize(safe_text)
    print(f"\n  token-only summary ({LLM_LABEL}):")
    print("    " + summary.replace("\n", "\n    "))
    check_crossing("summary", summary, doc, ner_available, ledger)
    print("  summary leak check: PASS, summary carries tokens only")

    print("\n  restored per output channel (one sealed session):")
    for channel, who in (
        ("internal_summary", "case worker"),
        ("export", "analytics export"),
        ("log_safe", "audit log"),
    ):
        restored = client.rehydrate(
            summary, session_state=session_state, output_channel=channel, restore_mode="full"
        )
        first = restored.restored_text.splitlines()[0] if restored.restored_text else ""
        print(f"    [{channel:16}] {who:16}: {first}")
        if channel == "internal_summary":
            print(f"      tokens_resolved={restored.tokens_resolved} unresolved={restored.tokens_unresolved}")

def run_ocr_capability_demo(base_url: str, api_key: str | None) -> None:
    """Capability-gated OCR path: render a card image, OCR it, and tokenize the extracted text. The
    image route returns 503 GUARDAI_OCR_UNAVAILABLE when the server has no Tesseract, and Pillow may
    not be installed, so this step self-skips on either instead of failing the run."""
    print("\n" + RULE)
    print("OCR / image route (capability-gated, self-skips when unavailable)")
    print(RULE)
    try:
        from PIL import Image, ImageDraw
    except ImportError:
        print("  skipped: Pillow is not installed (pip install pillow) to render the sample image.")
        return

    img = Image.new("RGB", (620, 180), "white")
    draw = ImageDraw.Draw(img)
    for offset, line in enumerate(
        ["Name: Erika Mustermann", "E-Mail: erika.mustermann@kontakt.example", "Tel: +49 89 55512340"]
    ):
        draw.text((20, 20 + offset * 45), line, fill="black")
    buffer = io.BytesIO()
    img.save(buffer, format="PNG")

    # retry=False so an OCR-unavailable 503 skips immediately instead of retrying the transient status.
    probe = OGuardAIClient(base_url=base_url, api_key=api_key, retry=False)
    try:
        result = probe.transform_image(buffer.getvalue(), "insurance_card.png", language=LANGUAGE)
        print(f"  OCR ran. safe text: {result.safe_text!r}")
        print(f"  entities located in the image: {len(result.entities)}  bounding boxes: {len(result.bounding_boxes)}")
        print("  (the same tokens can now be restored per channel exactly like the text documents)")
    except OGuardAIError as exc:
        if exc.status_code == 503 or exc.code == "GUARDAI_OCR_UNAVAILABLE":
            print("  skipped: the server has no Tesseract OCR (install with: apt install tesseract-ocr).")
        else:
            raise
    finally:
        probe.close()

def run_fail_closed_proof(client: OGuardAIClient) -> None:
    print("\n" + RULE)
    print("Fail closed: a tampered session is rejected, never half-restored")
    print(RULE)
    try:
        client.rehydrate(
            "Beteiligte: {{person:p_001:deadbeef0000}}.",
            session_state="tampered-session-blob",
            output_channel="internal_summary",
        )
        raise SystemExit("fail-closed check failed: a tampered session was accepted")
    except OGuardAIError as exc:
        print(f"  rejected with [{exc.code}]; no raw value was returned")

def main() -> int:
    print(f"OGuardAI Document AI   runtime={GUARDAI_URL}  summarizer={LLM_LABEL}")
    client = OGuardAIClient(base_url=GUARDAI_URL, api_key=GUARDAI_API_KEY)
    try:
        client.health()
    except OGuardAIError:
        print(f"\ncannot reach the OGuardAI server at {GUARDAI_URL}; start it first (see the README).")
        client.close()
        return 1
    try:
        ner_available = client.capabilities().ner_active is not False
    except OGuardAIError:
        ner_available = True
    if not ner_available:
        print("  NOTE: the NER detector is not active; person names will not tokenize and will reach")
        print("  the summarizer in the clear. Structured PII (email, phone, IBAN, order id) still")
        print("  tokenizes or redacts, so those leak checks hold on every lane.")

    ledger = LeakLedger()
    try:
        for doc in DOCUMENTS:
            process(client, doc, ner_available, ledger)
        run_ocr_capability_demo(GUARDAI_URL, GUARDAI_API_KEY)
        run_fail_closed_proof(client)
    except OGuardAIError as exc:
        print(f"\nOGuardAIError [{exc.code}]: {exc}")
        return 1
    finally:
        client.close()

    print("\n" + RULE)
    print("Every builtin-detectable identifier the documents carried (email, phone, IBAN, order id,")
    print("VAT id, passport, insurance id) was tokenized or redacted before the summarizer saw it,")
    print("then restored only for the reader each channel is cleared for.")
    if ledger.undetected:
        bare = ", ".join(sorted(set(ledger.undetected)))
        print(f"  NOTE: a bare customer number ({bare}) carries no detectable prefix, so the builtin")
        print("  detectors cannot match it; it reaches the summarizer in the clear, outside the")
        print("  detection floor. Only the structured identifiers above stayed inside the trusted zone.")
    if ledger.names:
        crossed = ", ".join(sorted(set(ledger.names)))
        print(f"  NOTE: names reached the model in the clear ({crossed}) because this runtime runs")
        print("  without the NER detector; start it to tokenize names as well.")
    print(RULE)
    return 0


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