Kansho/backend/privacy_gateway.py

"""Local privacy gateway. Personal generative egress is not allowed to skip this layer."""
from __future__ import annotations

from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from typing import Any
import contextvars
import re
import time

from entity_detect import DetectError, detect_personal_egress, reset_detect_test_hooks, user_detect_message
from identity_store import KINSHIP, is_maskable_label, is_registry_maskable_label, mapping_spellings
from journal_reconstruct import claim_texts, fake_reconstruction, is_dialogue_role_line
from prompt_budget import (
    ERROR_PROVIDER_CONTEXT_LENGTH,
    JOURNAL_PURPOSES,
    JournalBudgetError,
    USER_MESSAGES,
    assert_input_fits,
    merge_usage,
    sum_usages,
)
from pronoun_bind import bind_user_lines
from providers import ChatResult, ProviderError, complete_chat, generate_provider

MAX_EGRESS_CHARS = 24000
debug_calls = 0
last_compact: dict[str, Any] | None = None
_test_recorder: contextvars.ContextVar[list | None] = contextvars.ContextVar(
    "kansho_gateway_recorder",
    default=None,
)
COMPACT_DIAGNOSTIC_KEYS = (
    "model",
    "actual_model",
    "provider",
    "purpose",
    "prompt_slug",
    "prompt_revision",
    "generate_ms",
    "effective_context_window",
    "estimated_input_tokens",
    "prompt_tokens",
    "completion_tokens",
    "total_tokens",
    "reserved_output_tokens",
    "max_tokens",
    "safety_margin",
    "context_compression",
    "cost",
    "budget_ok",
    "abort_reason",
    "status",
    "available_mapping_count",
    "active_mapping_count",
    "masked_occurrence_count",
    "active_tokens",
    "active_entity_types",
    "leak_tokens",
    "leak_entity_types",
    "pre_egress_validation",
    "response_validation",
    "response_validation_retry",
    "response_normalization",
    "normalized_cleartext_count",
    "generate_calls",
    "model_text_accepted",
    "provenance_decision",
    "full_detection_coverage",
    "detect_provider",
    "detect_model",
    "detect_note",
    "source_chars",
    "chunk_count",
    "chunks_ok",
    "entity_counts",
    "confirmed_registry_hits",
    "confirmed_registry_applied",
    "request_local_hits",
    "semantic_identity_guaranteed",
    "detect_calls",
    "detect_prompt_tokens",
    "detect_completion_tokens",
    "detect_total_tokens",
    "detect_cost",
    "detect_ms",
    "generate_called",
    "generate_requested",
    "http_status",
    "provider_message",
    "provider_code",
    "timeout_s",
    "cost_report",
    "detect_passes",
    "detect_partial_discarded",
    "detect_attempts",
    "contract_violation",
    "invalid_entity_type",
    "omitted_non_identity_types",
    "detect_cost_unknown",
    "generation_selection",
    "style_application",
)

ERROR_EGRESS_VALIDATION = "egress_validation_failed"


class PrivacyGatewayError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 503, diagnostics: dict | None = None):
        super().__init__(message)
        self.code = code
        self.message = message
        self.status_code = status_code
        self.diagnostics = diagnostics or {}


@dataclass
class GatewayRequest:
    prompt_id: str | None
    purpose: str
    data_class: str
    payload: dict[str, Any] = field(default_factory=dict)
    profile_id: str | None = None


@dataclass
class GatewayResult:
    allowed: bool
    reason: str
    provider: str | None = None
    content: str | None = None
    diagnostics: dict[str, Any] = field(default_factory=dict)
    trace: dict[str, Any] | None = None
    local_identities: list[dict[str, Any]] = field(default_factory=list)
    checked_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


ALLOWED_CLASSES = {"A", "B", "C"}
LOCAL_ONLY_CLASS = "A"


@dataclass(frozen=True)
class ActiveReplacement:
    """One mapping that was actually replaced as identity in this request."""

    token: str
    entity_type: str
    occurrence_count: int
    local_label: str
    demask_label: str = ""
    source: str = ""
    span_based: bool = False
    labels: tuple[str, ...] = ()

    def restore_label(self) -> str:
        return self.demask_label or self.local_label

    def all_labels(self) -> tuple[str, ...]:
        if self.labels:
            return self.labels
        return (self.local_label,) if self.local_label else ()


@dataclass
class MaskingManifest:
    """Request-scoped mask result. local_label stays on this object only."""

    masked_text: str
    available_mapping_count: int
    replacements: tuple[ActiveReplacement, ...] = ()
    restore_by_token: dict[str, str] = field(default_factory=dict)

    def public_summary(self) -> dict[str, Any]:
        return {
            "available_mapping_count": self.available_mapping_count,
            "active_mapping_count": len(self.replacements),
            "masked_occurrence_count": sum(item.occurrence_count for item in self.replacements),
            "active_tokens": [item.token for item in self.replacements],
            "active_entity_types": sorted({item.entity_type for item in self.replacements}),
        }


def _snapshot_plain(value: Any, *, _stack: set[int] | None = None) -> Any:
    """Acyclic copy of compact diagnostic values. Drops cyclic back-references."""
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    if isinstance(value, (dict, list, tuple)):
        stack = _stack or set()
        ident = id(value)
        if ident in stack:
            return None
        stack = set(stack)
        stack.add(ident)
        if isinstance(value, dict):
            return {
                str(key): _snapshot_plain(item, _stack=stack)
                for key, item in value.items()
                if key not in {"trace", "diagnostics", "intern", "egress", "raw", "reply", "prompt", "messages"}
            }
        return [_snapshot_plain(item, _stack=stack) for item in value]
    return value


def compact_diagnostics(data: dict[str, Any] | None) -> dict[str, Any]:
    payload = {}
    for key in COMPACT_DIAGNOSTIC_KEYS:
        if data and key in data and data[key] is not None:
            payload[key] = _snapshot_plain(data[key])
    return payload


def _as_cost(value: Any) -> float | None:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _format_usd(value: float) -> str:
    text = f"{value:.6f}".rstrip("0").rstrip(".")
    return f"{text.replace('.', ',')} USD"


def _first_present(*values: Any) -> Any:
    for value in values:
        if value is not None:
            return value
    return None


def public_cost_report(diagnostics: dict[str, Any] | None) -> dict[str, Any]:
    """User-facing billed/unbilled summary. No prompts, no mapping labels."""
    data = diagnostics or {}
    nested = data.get("trace") if isinstance(data.get("trace"), dict) else {}
    budget = nested.get("budget") if isinstance(nested.get("budget"), dict) else {}
    generate_called = bool(
        data.get("generate_called")
        or data.get("generate_calls")
        or nested.get("generate_called")
        or nested.get("generate_calls")
    )
    generate_requested = bool(
        data.get("generate_requested")
        or nested.get("generate_requested")
        or generate_called
    )
    detect_calls = _first_present(data.get("detect_calls"), nested.get("detect_calls"))
    try:
        detect_calls_n = int(detect_calls) if detect_calls is not None else 0
    except (TypeError, ValueError):
        detect_calls_n = 0
    detect_called = detect_calls_n > 0
    generate_cost = _as_cost(_first_present(data.get("cost"), budget.get("cost")))
    detect_cost = _as_cost(_first_present(data.get("detect_cost"), nested.get("detect_cost")))
    known = [item for item in (detect_cost, generate_cost) if item is not None]
    total = sum(known) if known else None
    if (detect_cost or 0) > 0 or (generate_cost or 0) > 0:
        billed = "yes"
    elif generate_requested and generate_cost is None:
        billed = "unknown"
    elif detect_called and detect_cost is None and not generate_called:
        billed = "unknown"
    else:
        billed = "no"
    parts: list[str] = []
    if detect_called or detect_cost is not None:
        if detect_cost is not None:
            parts.append(f"Detect {_format_usd(detect_cost)}")
        elif detect_called:
            parts.append("Detect aufgerufen, Kosten unbekannt")
    if generate_called and generate_cost is not None:
        parts.append(f"Generate {_format_usd(generate_cost)}")
    elif generate_called:
        parts.append("Generate abgeschlossen, Kosten unbekannt")
    elif generate_requested:
        parts.append("Generate-Aufruf gesendet, Kosten unbekannt")
    else:
        parts.append("Generate nicht aufgerufen")
    if billed == "yes":
        note = "Beim Anbieter sind Kosten entstanden"
        if parts:
            note += ": " + "; ".join(parts) + "."
        else:
            note += "."
    elif billed == "unknown":
        note = "Ob beim Anbieter Kosten entstanden sind, ist unbekannt"
        if parts:
            note += " (" + "; ".join(parts) + ")."
        else:
            note += "."
    else:
        if detect_called or generate_requested or generate_called:
            note = "Beim Anbieter sind für diesen Lauf keine Kosten ausgewiesen"
            if parts:
                note += " (" + "; ".join(parts) + ")."
            else:
                note += "."
        else:
            note = "Es wurde kein Provider aufgerufen. Es sind keine Kosten entstanden."
    return {
        "generate_called": generate_called,
        "generate_requested": generate_requested,
        "detect_called": detect_called,
        "detect_calls": detect_calls_n,
        "cost": generate_cost,
        "detect_cost": detect_cost,
        "total_cost": total,
        "billed": billed,
        "note": note,
    }


def public_error_detail(exc) -> dict[str, Any]:
    diagnostics = dict(getattr(exc, "diagnostics", None) or {})
    report = public_cost_report(diagnostics)
    if report and diagnostics.get("cost_report") is None:
        diagnostics["cost_report"] = report
    detail: dict[str, Any] = {
        "code": getattr(exc, "code", None) or "error",
        "message": getattr(exc, "message", None) or str(exc),
        "cost_report": report,
    }
    http_status = diagnostics.get("http_status")
    if http_status is not None:
        detail["http_status"] = http_status
    provider_message = diagnostics.get("provider_message")
    if provider_message:
        detail["provider_message"] = provider_message
    if diagnostics:
        detail["diagnostics"] = diagnostics
    return detail


_LOG_FORBIDDEN_KEYS = frozenset(
    {
        "local_label",
        "label",
        "masked_text",
        "rendered",
        "intern",
        "egress",
        "raw",
        "reply",
        "content",
        "messages",
        "prompt",
        "mapping_table",
        "replacements",
    }
)


def _log_event(events: list[dict[str, Any]], started: float, kind: str, **fields: Any) -> None:
    """Compact run log. No prompts, no answers, no mapping labels."""
    row: dict[str, Any] = {
        "kind": kind,
        "t_ms": int((time.perf_counter() - started) * 1000),
    }
    for key, value in fields.items():
        if value is None or value == "" or key in _LOG_FORBIDDEN_KEYS:
            continue
        row[key] = value
    events.append(row)


def _usage_bits(usage: dict[str, Any] | None) -> dict[str, Any]:
    data = usage or {}
    bits: dict[str, Any] = {}
    for key in ("prompt_tokens", "completion_tokens", "total_tokens", "cost"):
        if data.get(key) is not None:
            bits[key] = data[key]
    return bits


def _raise_with_log(
    exc: PrivacyGatewayError,
    events: list[dict[str, Any]],
    *,
    purpose: str | None = None,
    extra: dict[str, Any] | None = None,
) -> None:
    diagnostics = dict(exc.diagnostics or {})
    diagnostics["log"] = events
    if purpose:
        diagnostics["purpose"] = purpose
    if extra:
        diagnostics.update({key: value for key, value in extra.items() if value is not None})
    diagnostics["cost_report"] = public_cost_report(diagnostics)
    raise PrivacyGatewayError(exc.code, exc.message, exc.status_code, diagnostics) from exc


def install_test_recorder() -> list[dict[str, Any]]:
    """Test-local observer. Production never installs this."""
    records: list[dict[str, Any]] = []
    _test_recorder.set(records)
    return records


def _record_test(event: dict[str, Any]) -> None:
    records = _test_recorder.get()
    if records is not None:
        records.append(event)


def reset_debug() -> None:
    global last_compact, debug_calls
    last_compact = None
    debug_calls = 0
    _test_recorder.set(None)
    reset_detect_test_hooks()


def inspect(request: GatewayRequest) -> GatewayResult:
    """Fail closed unless a policy-complete generative provider is configured."""
    data_class = (request.data_class or "").upper()
    if data_class not in ALLOWED_CLASSES:
        return GatewayResult(allowed=False, reason="unknown_data_class")
    if data_class == LOCAL_ONLY_CLASS:
        return GatewayResult(allowed=False, reason="class_a_never_leaves_local_zone")
    config = generate_provider()
    if not config:
        return GatewayResult(allowed=False, reason="no_egress_provider_configured")
    return GatewayResult(allowed=True, reason="policy_ok", provider=config.name)


def _minimize(text: str, purpose: str = "") -> str:
    cleaned = (text or "").strip()
    if purpose in JOURNAL_PURPOSES:
        return cleaned
    if len(cleaned) <= MAX_EGRESS_CHARS:
        return cleaned
    return cleaned[:MAX_EGRESS_CHARS]


_LETTER = r"A-Za-zÄÖÜäöüß"
_WORD = re.compile(rf"[{_LETTER}]+")
THING_GOVERNORS = {
    "esse", "essen", "isst", "aß", "aßest", "gegessen",
    "koche", "kochen", "kochte", "gekocht",
    "trinke", "trinken", "trank", "getrunken",
    "kaufe", "kaufen", "kaufte", "gekauft",
    "bestelle", "bestellen", "bestellte",
    "hole", "holen", "holte",
}
THING_OBJECT_AFTER = {
    "essen", "kochen", "trinken", "kaufen", "bestellen", "holen",
    "gegessen", "gekocht", "getrunken", "gekauft",
}


def _words_before(text: str, index: int, n: int = 4) -> list[str]:
    return [word.lower() for word in _WORD.findall(text[:index])[-n:]]


def _words_after(text: str, index: int, n: int = 3) -> list[str]:
    return [word.lower() for word in _WORD.findall(text[index:])[:n]]


def _is_person_token(token: str) -> bool:
    raw = (token or "").strip().upper()
    raw = raw[2:-2] if raw.startswith("[[") and raw.endswith("]]") else raw
    return raw.startswith("PERSON:")


def _mask_person_hit(text: str, start: int, end: int) -> bool:
    prev = _words_before(text, start)
    nxt = _words_after(text, end)
    if prev and prev[-1] in KINSHIP:
        return True
    if any(word in THING_GOVERNORS for word in prev):
        return False
    if nxt and nxt[0] in THING_OBJECT_AFTER:
        return False
    return True


def canonical_token(token: str) -> str:
    """Public, intent-neutral: placeholder token without surrounding [[ ]]."""
    raw = (token or "").strip()
    if raw.startswith("[[") and raw.endswith("]]"):
        return raw[2:-2]
    return raw


def identity_label_pattern(label: str) -> re.Pattern[str]:
    """Public, intent-neutral: word-boundary pattern for a local identity label."""
    return re.compile(
        rf"(?<![{_LETTER}]){re.escape(label)}(?![{_LETTER}])",
        re.IGNORECASE,
    )


def is_identity_mention(text: str, start: int, end: int, token: str) -> bool:
    """Public, intent-neutral: same rule as masking. Food/thing homonyms are not identity."""
    if not _is_person_token(token):
        return True
    return _mask_person_hit(text, start, end)


def identity_occurrence_count(text: str, label: str, token: str) -> int:
    """Public, intent-neutral count of identity mentions for one mapping."""
    if not label or not is_registry_maskable_label(label):
        return 0
    return sum(
        1
        for match in identity_label_pattern(label).finditer(text or "")
        if is_identity_mention(text, match.start(), match.end(), token)
    )


def _is_identity_mention(text: str, start: int, end: int, token: str) -> bool:
    return is_identity_mention(text, start, end, token)


def _label_pattern(label: str) -> re.Pattern[str]:
    return identity_label_pattern(label)


def _canonical_token(token: str) -> str:
    return canonical_token(token)


def _entity_type_of(item: dict) -> str:
    explicit = (item.get("entity_type") or "").strip().upper()
    if explicit:
        return explicit
    token = _canonical_token(item.get("token") or "").upper()
    if ":" in token:
        return token.split(":", 1)[0]
    return token or "UNKNOWN"


def _identity_occurrence_count(text: str, label: str, token: str) -> int:
    return identity_occurrence_count(text, label, token)


def _mapping_label_allowed(item: dict) -> bool:
    label = (item.get("local_label") or "").strip()
    token = (item.get("token") or "").strip()
    if not label or not token:
        return False
    if (item.get("source") or "") == "confirmed_registry":
        return is_registry_maskable_label(label)
    return is_maskable_label(label)


def _active_replacements(text: str, mappings: list[dict]) -> tuple[ActiveReplacement, ...]:
    found: list[ActiveReplacement] = []
    for item in mappings or []:
        label = (item.get("local_label") or "").strip()
        token = (item.get("token") or "").strip()
        if not _mapping_label_allowed(item):
            continue
        count = _identity_occurrence_count(text, label, token)
        if count <= 0:
            continue
        found.append(
            ActiveReplacement(
                token=_canonical_token(token),
                entity_type=_entity_type_of(item),
                occurrence_count=count,
                local_label=label,
                demask_label=(item.get("demask_label") or item.get("canonical_label") or label),
            )
        )
    return tuple(found)


_SPAN_TYPE_PRIORITY = {"PERSON": 0, "PROJECT": 1, "ORG": 2, "PLACE": 3}


def _placeholder_of(token: str) -> str:
    raw = (token or "").strip()
    if raw.startswith("[[") and raw.endswith("]]"):
        return f"[[{_canonical_token(raw)}]]"
    return f"[[{_canonical_token(raw)}]]"


def _ground_mapping_span(text: str, item: dict) -> tuple[int, int] | None:
    label = (item.get("local_label") or "").strip()
    if not label or "start" not in item or "end" not in item:
        return None
    try:
        start = int(item.get("start"))
        end = int(item.get("end"))
    except (TypeError, ValueError):
        return None
    if start < 0 or end > len(text or "") or start >= end:
        return None
    slice_text = (text or "")[start:end]
    if slice_text == label or slice_text.casefold() == label.casefold():
        return start, end
    return None


def _resolve_span_mappings(items: list[dict]) -> list[dict]:
    ordered = sorted(
        items,
        key=lambda row: (
            -(int(row["end"]) - int(row["start"])),
            int(row["start"]),
            _SPAN_TYPE_PRIORITY.get((row.get("entity_type") or "").upper(), 9),
        ),
    )
    kept: list[dict] = []
    occupied: list[tuple[int, int]] = []
    for item in ordered:
        start, end = int(item["start"]), int(item["end"])
        if any(start < right and end > left for left, right in occupied):
            continue
        kept.append(item)
        occupied.append((start, end))
    return sorted(kept, key=lambda row: int(row["start"]))


def _mask_body(text: str, mappings: list[dict]) -> str:
    masked = text
    for item in sorted(mappings, key=lambda row: len(row.get("local_label") or ""), reverse=True):
        label = (item.get("local_label") or "").strip()
        token = (item.get("token") or "").strip()
        if not _mapping_label_allowed(item):
            continue
        placeholder = _placeholder_of(token)
        pattern = _label_pattern(label)

        def repl(match: re.Match, *, _token=token, _ph=placeholder) -> str:
            if not _is_identity_mention(masked, match.start(), match.end(), _token):
                return match.group(0)
            return _ph

        masked = pattern.sub(repl, masked)
    return masked


def _mask(text: str, mappings: list[dict], *, personal_lines_only: bool = False) -> str:
    if not personal_lines_only:
        return _mask_body(text, mappings)
    parts: list[str] = []
    for line in (text or "").splitlines(keepends=True):
        raw = line[:-1] if line.endswith("\n") else line
        newline = "\n" if line.endswith("\n") else ""
        if is_dialogue_role_line(raw):
            parts.append(_mask_body(raw, mappings) + newline)
        else:
            parts.append(line)
    return "".join(parts)


PLACEHOLDER_RE = re.compile(r"\[\[\s*([^\[\]]+?)\s*\]\]")
GENERIC_PLACEHOLDER_INNER = frozenset({"…", "...", "..", "...."})


def _restore_table(manifest: MaskingManifest | None) -> dict[str, str]:
    table: dict[str, str] = {}
    if not manifest:
        return table
    for token, label in (manifest.restore_by_token or {}).items():
        key = canonical_token(str(token)).upper().replace(" ", "")
        if key and label:
            table[key] = label
    for item in manifest.replacements:
        key = canonical_token(item.token).upper().replace(" ", "")
        label = item.restore_label()
        if key and label:
            table.setdefault(key, label)
    return table


def _lookup_restore(table: dict[str, str], inner: str) -> str | None:
    raw = canonical_token(inner or "").upper().replace(" ", "")
    if not raw or raw in GENERIC_PLACEHOLDER_INNER:
        return None
    if raw in table:
        return table[raw]
    if ":" in raw:
        prefix, suffix = raw.split(":", 1)
        if suffix.isdigit():
            padded = f"{prefix}:{int(suffix):02d}"
            if padded in table:
                return table[padded]
            unpadded = f"{prefix}:{int(suffix)}"
            if unpadded in table:
                return table[unpadded]
    return None


def _restore_map_from_mappings(mappings: list[dict] | None) -> dict[str, str]:
    table: dict[str, str] = {}
    for item in mappings or []:
        token = canonical_token(item.get("token") or "").upper().replace(" ", "")
        label = (item.get("demask_label") or item.get("canonical_label") or item.get("local_label") or "").strip()
        if token and label:
            table[token] = label
    return table


def mask_prompt(rendered: str, mappings: list[dict], purpose: str) -> MaskingManifest:
    """Request-scoped mask result. Detected spans replace only those offsets.

    Request-local detections never globally replace a label. Confirmed-registry
    rows without spans remain a separate identity-mention safety net.
    """
    text = rendered or ""
    replacements: dict[str, ActiveReplacement] = {}

    def _merge_labels(existing: tuple[str, ...], item: dict) -> tuple[str, ...]:
        labels = list(existing)
        seen = {label.casefold() for label in labels}
        for label in mapping_spellings(item):
            key = label.casefold()
            if key in seen:
                continue
            seen.add(key)
            labels.append(label)
        return tuple(labels)

    def remember(item: dict, count: int, *, span_based: bool) -> None:
        if count <= 0:
            return
        token = _canonical_token(item.get("token") or "")
        if not token:
            return
        existing = replacements.get(token)
        replacements[token] = ActiveReplacement(
            token=token,
            entity_type=_entity_type_of(item),
            occurrence_count=(existing.occurrence_count if existing else 0) + count,
            local_label=(item.get("local_label") or (existing.local_label if existing else "")).strip(),
            demask_label=(
                item.get("demask_label")
                or item.get("canonical_label")
                or (existing.demask_label if existing else "")
                or item.get("local_label")
                or ""
            ),
            source=(item.get("source") or (existing.source if existing else "")),
            span_based=bool((existing.span_based if existing else False) or span_based),
            labels=_merge_labels(existing.labels if existing else (), item),
        )

    span_rows: list[dict] = []
    label_rows: list[dict] = []
    for item in mappings or []:
        label = (item.get("local_label") or "").strip()
        token = (item.get("token") or "").strip()
        if not _mapping_label_allowed(item):
            continue
        has_span_fields = item.get("start") is not None and item.get("end") is not None
        grounded = _ground_mapping_span(text, item) if has_span_fields else None
        if grounded is not None:
            span_rows.append({**item, "start": grounded[0], "end": grounded[1]})
        elif not has_span_fields:
            label_rows.append(item)

    applied = _resolve_span_mappings(span_rows)
    masked = text
    for item in sorted(applied, key=lambda row: int(row["start"]), reverse=True):
        start, end = int(item["start"]), int(item["end"])
        masked = masked[:start] + _placeholder_of(item.get("token") or "") + masked[end:]
        remember(item, 1, span_based=True)

    for item in sorted(label_rows, key=lambda row: len(row.get("local_label") or ""), reverse=True):
        label = (item.get("local_label") or "").strip()
        token = (item.get("token") or "").strip()
        placeholder = _placeholder_of(token)
        pattern = _label_pattern(label)
        current = masked
        count = 0

        def repl(match: re.Match, *, _token=token, _ph=placeholder, _text=current) -> str:
            nonlocal count
            if not _is_identity_mention(_text, match.start(), match.end(), _token):
                return match.group(0)
            count += 1
            return _ph

        masked = pattern.sub(repl, current)
        remember(item, count, span_based=False)

    if purpose == "dialogue_turn":
        masked = bind_user_lines(masked)
    for token, item in list(replacements.items()):
        labels = item.labels
        for row in mappings or []:
            if _canonical_token(row.get("token") or "") != token:
                continue
            labels = _merge_labels(labels, row)
        if labels != item.labels:
            replacements[token] = replace(item, labels=labels)
    return MaskingManifest(
        masked_text=masked,
        available_mapping_count=len(mappings or []),
        replacements=tuple(replacements.values()),
        restore_by_token=_restore_map_from_mappings(mappings),
    )


def mask_for_egress(rendered: str, mappings: list[dict], purpose: str) -> str:
    """String wrapper for callers that only need the masked egress text."""
    return mask_prompt(rendered, mappings, purpose).masked_text


def validate_pre_egress(masked_text: str, manifest: MaskingManifest) -> None:
    """Fail closed before the provider if a confirmed identity remains as plaintext.

    Request-local span replacements may leave the same wording unmasked where it
    was not detected as identity. That remaining wording is not an egress leak.
    """
    leaked_tokens: list[str] = []
    leaked_types: list[str] = []
    body = masked_text or ""
    for item in manifest.replacements:
        if item.span_based and item.source != "confirmed_registry":
            placeholder = f"[[{canonical_token(item.token)}]]"
            if placeholder not in body:
                leaked_tokens.append(item.token)
                leaked_types.append(item.entity_type)
            continue
        leaked = False
        for label in item.all_labels():
            for match in _label_pattern(label).finditer(body):
                if _is_identity_mention(body, match.start(), match.end(), item.token):
                    leaked_tokens.append(item.token)
                    leaked_types.append(item.entity_type)
                    leaked = True
                    break
            if leaked:
                break
    if leaked_tokens:
        raise PrivacyGatewayError(
            ERROR_EGRESS_VALIDATION,
            "Aktive Identität war nach der Maskierung noch als Klartext im Egress.",
            diagnostics={
                "pre_egress_validation": "failed",
                "leak_tokens": leaked_tokens,
                "leak_entity_types": leaked_types,
                **manifest.public_summary(),
            },
        )


def _demask(text: str, manifest: MaskingManifest | None) -> str:
    """Rehydrate placeholders that belong to this request. Unknown tokens stay placeholders."""
    result = text or ""
    table = _restore_table(manifest)
    if not table:
        return result

    def repl(match: re.Match) -> str:
        inner = match.group(1) or ""
        restored = _lookup_restore(table, inner)
        return restored if restored else match.group(0)

    return PLACEHOLDER_RE.sub(repl, result)


def _visible_response(content: str, manifest: MaskingManifest | None) -> str:
    """Stored output: original model text with only emitted placeholders rehydrated."""
    return _demask((content or "").strip(), manifest)


def _normalize_active_cleartext(text: str, manifest: MaskingManifest | None) -> tuple[str, int]:
    """Map active identity cleartext back to the request token. Check copy only; never stored."""
    if not manifest or not manifest.replacements:
        return text or "", 0
    hits: list[tuple[int, int, str]] = []
    occupied: list[tuple[int, int]] = []
    labeled: list[tuple[ActiveReplacement, str]] = []
    for item in manifest.replacements:
        for label in item.all_labels():
            if label:
                labeled.append((item, label))
    items = sorted(labeled, key=lambda row: len(row[1]), reverse=True)
    for item, label in items:
        placeholder = _placeholder_of(item.token)
        for match in _label_pattern(label).finditer(text or ""):
            if not _is_identity_mention(text, match.start(), match.end(), item.token):
                continue
            if any(match.start() < right and match.end() > left for left, right in occupied):
                continue
            occupied.append((match.start(), match.end()))
            hits.append((match.start(), match.end(), placeholder))
    result = text or ""
    for start, end, placeholder in sorted(hits, key=lambda row: row[0], reverse=True):
        result = result[:start] + placeholder + result[end:]
    return result, len(hits)


def _process_response(content: str, manifest: MaskingManifest | None = None) -> tuple[str, dict[str, Any]]:
    """Integrity check after the model reply. Tokenized copy is not the visible output."""
    text = (content or "").strip()
    if not text:
        raise PrivacyGatewayError("empty_provider_response", "Der Provider lieferte keine Antwort.")
    normalized, count = _normalize_active_cleartext(text, manifest)
    meta = {
        "response_validation": "ok",
        "response_normalization": "active_cleartext_normalized" if count else "none",
        "normalized_cleartext_count": count,
    }
    return normalized, meta


def _validate_response(content: str, manifest: MaskingManifest | None = None) -> str:
    """Compatibility wrapper: tokenized check copy. Not the stored user-facing text."""
    text, _meta = _process_response(content, manifest)
    return text


def _fake_journal_from_sources(rendered: str) -> str:
    """Deterministic fake body from labeled CURRENT_DAY_SOURCES. Not a quality claim."""
    text = rendered or ""
    marker = "\nCURRENT_DAY_SOURCES\n"
    if marker not in text:
        marker = "CURRENT_DAY_SOURCES"
        if marker not in text:
            return ""
    after = text.split(marker, 1)[1]
    for stop in ("WRITING_PROFILE", "STYLE_EXAMPLES", "EXISTING_TEXT", "EDITORIAL_MODE", "AUSGABE", "DATENSCHUTZ"):
        if f"\n{stop}" in after:
            after = after.split(f"\n{stop}", 1)[0]
    parts: list[str] = []
    current: list[str] = []
    for line in after.splitlines():
        stripped = line.strip()
        if stripped.startswith("[u") and stripped.endswith("]"):
            if current:
                parts.append(" ".join(current).strip())
                current = []
            continue
        if stripped and not stripped.startswith("(") and "einzige Tatsachen" not in stripped:
            current.append(stripped)
    if current:
        parts.append(" ".join(current).strip())
    return " ".join(part for part in parts if part).strip()
    parts: list[str] = []
    current: list[str] = []
    for line in after.splitlines():
        stripped = line.strip()
        if stripped.startswith("[u") and stripped.endswith("]"):
            if current:
                parts.append(" ".join(current).strip())
                current = []
            continue
        if stripped and not stripped.startswith("(") and "einzige Tatsachen" not in stripped:
            current.append(stripped)
    if current:
        parts.append(" ".join(current).strip())
    return " ".join(part for part in parts if part).strip()


def _fake_complete(purpose: str, rendered: str) -> str:
    if purpose == "profile_review":
        return (
            '{"kind":"kansho.profile_analysis_result","format_version":1,"target":"writing",'
            '"evidence_basis":["kansho_sources"],'
            '"changes":[{"layer":"trait","key":"dry_humor","slug":"dry_humor","facet_key":"autobiographical_journal",'
            '"action":"add","label":"Trockener Humor",'
            '"proposed_value":"gelegentlich trocken, nie aufgesetzt",'
            '"evidence_basis":["kansho_sources"],'
            '"rationale":"Fake-Review: Journal-Evidenz trägt Humor in autobiographical_journal, nicht als globalen Core.",'
            '"evidence_ids":[],"exemplars":[{"excerpt":"haha das war irgendwie lustig","role":"exemplar","evidence_basis":["kansho_sources"]}]}]}'
        )
    if purpose == "journal_reconstruct":
        return fake_reconstruction(rendered)
    if purpose == "journal_generate":
        body = _fake_journal_from_sources(rendered)
        if body:
            return f"Ein Tag\n\n{body}"
        match = re.search(r"\{.*\}", rendered, re.DOTALL)
        if match:
            try:
                import json

                data = json.loads(match.group(0))
                parts = claim_texts(data)
                body = " ".join(str(part) for part in parts if part)
                if body:
                    return f"Ein Tag\n\n{body}"
            except (ValueError, TypeError):
                pass
        lines = []
        for line in rendered.splitlines():
            if line.startswith("user:"):
                lines.append(line[5:].strip())
        body = " ".join(part for part in lines if part) or "Ein stiller Tag."
        return f"Ein Tag\n\n{body}"
    if "Keine Interviewfrage" in rendered:
        return '{"operation":"fortfuehren","impulse":"Der letzte Faden bleibt offen."}'
    impulse = (
        "Was davon möchtest du vertiefen, ohne Unerwähntes als nicht geschehen zu behandeln?"
        if ("nicht geschehen" in rendered or "nicht erwähnt" in rendered)
        else "Was davon möchtest du festhalten?"
    )
    return '{"operation":"erleben_vertiefen","impulse":"' + impulse + '"}'


def _generate_timeout(purpose: str) -> float:
    if purpose in JOURNAL_PURPOSES:
        return 90
    if purpose == "profile_review":
        return 120
    return 60


def complete_model(messages: list[dict], policy: dict[str, Any]) -> ChatResult:
    config = generate_provider()
    if not config:
        raise PrivacyGatewayError("no_egress_provider_configured", "Es ist kein Egress-Provider konfiguriert.")
    if config.mode == "fake":
        rendered = "\n".join(item.get("content") or "" for item in messages)
        return ChatResult(
            content=_fake_complete(policy.get("purpose") or "", rendered),
            model=config.model,
            usage={},
            context_compression="disabled" if policy.get("disable_context_compression") else "not_applicable",
        )
    try:
        return complete_chat(
            config,
            messages,
            timeout=_generate_timeout(policy.get("purpose") or ""),
            max_tokens=policy.get("max_tokens"),
            disable_context_compression=bool(policy.get("disable_context_compression")),
        )
    except ProviderError as exc:
        if exc.code == "provider_context_length_rejected":
            raise PrivacyGatewayError(
                ERROR_PROVIDER_CONTEXT_LENGTH,
                USER_MESSAGES[ERROR_PROVIDER_CONTEXT_LENGTH],
                exc.status_code,
                diagnostics=exc.diagnostics,
            ) from exc
        raise PrivacyGatewayError(exc.code, exc.message, exc.status_code, getattr(exc, "diagnostics", None)) from exc


def complete(request: GatewayRequest) -> GatewayResult:
    global last_compact, debug_calls
    started = time.perf_counter()
    events: list[dict[str, Any]] = []
    result = inspect(request)
    if not result.allowed:
        _log_event(events, started, "blocked", purpose=request.purpose, code=result.reason)
        raise PrivacyGatewayError(
            result.reason,
            "Persönlicher KI-Aufruf wurde vom Privacy Gateway blockiert. "
            "Es ist kein Egress-Provider konfiguriert."
            if result.reason == "no_egress_provider_configured"
            else "Persönlicher KI-Aufruf wurde vom Privacy Gateway blockiert.",
            diagnostics={"log": events, "purpose": request.purpose},
        )
    rendered = _minimize(str((request.payload or {}).get("rendered") or ""), request.purpose)
    detect_name = None
    detect_note = None
    detect_stats: dict[str, Any] = {}
    local_identities: list[dict[str, Any]] = []
    precomputed = request.payload.get("precomputed_mappings") if request.payload else None
    try:
        if precomputed is not None:
            mappings = list(precomputed)
            detect_stats = {
                "detect_note": "precomputed_learning_review",
                "full_detection_coverage": True,
                "semantic_identity_guaranteed": False,
            }
            local_identities = [
                {
                    "local_label": item.get("local_label"),
                    "token": item.get("token"),
                    "entity_type": item.get("entity_type"),
                    "source": item.get("source"),
                }
                for item in mappings
            ]
            detect_name = "learning_review"
        else:
            outcome = detect_personal_egress(request.profile_id, rendered)
            mappings = outcome.mappings
            detect_stats = outcome.stats.public()
            detect_name = outcome.stats.detect_provider
            detect_note = outcome.stats.detect_note
            local_identities = outcome.local_identities
    except DetectError as exc:
        detect_stats = dict(exc.diagnostics or {})
        _log_event(
            events,
            started,
            "detect",
            status="error",
            code=exc.code,
            full_detection_coverage=False,
            generate_called=False,
            chunk_count=detect_stats.get("chunk_count"),
            chunks_ok=detect_stats.get("chunks_ok"),
            detect_calls=detect_stats.get("detect_calls"),
        )
        raise PrivacyGatewayError(
            exc.code,
            user_detect_message(exc.code, exc.message),
            exc.status_code,
            {
                "log": events,
                "purpose": request.purpose,
                "generate_called": False,
                "generate_requested": False,
                **detect_stats,
            },
        ) from exc
    _log_event(
        events,
        started,
        "detect",
        status="ok",
        detect_provider=detect_name,
        detect_note=detect_note,
        detect_model=detect_stats.get("detect_model"),
        full_detection_coverage=detect_stats.get("full_detection_coverage"),
        chunk_count=detect_stats.get("chunk_count"),
        chunks_ok=detect_stats.get("chunks_ok"),
        detect_calls=detect_stats.get("detect_calls"),
        request_local_hits=detect_stats.get("request_local_hits"),
        confirmed_registry_hits=detect_stats.get("confirmed_registry_hits"),
        confirmed_registry_applied=detect_stats.get("confirmed_registry_applied"),
        semantic_identity_guaranteed=False,
        entity_counts=detect_stats.get("entity_counts"),
        detect_ms=detect_stats.get("detect_ms"),
        detect_prompt_tokens=detect_stats.get("detect_prompt_tokens"),
        detect_completion_tokens=detect_stats.get("detect_completion_tokens"),
        detect_total_tokens=detect_stats.get("detect_total_tokens"),
        detect_cost=detect_stats.get("detect_cost"),
        mapping_count=len(mappings),
    )
    manifest = mask_prompt(rendered, mappings, request.purpose)
    masked = manifest.masked_text
    summary = manifest.public_summary()
    budget = (request.payload or {}).get("budget")
    diagnostics = dict((request.payload or {}).get("diagnostics") or {})
    diagnostics.update(summary)
    diagnostics.update({key: value for key, value in detect_stats.items() if value is not None})
    diagnostics["generate_called"] = False
    diagnostics["generate_requested"] = False
    if request.purpose in JOURNAL_PURPOSES:
        if budget is None:
            _log_event(events, started, "budget", status="error", code="model_metadata_unknown")
            raise PrivacyGatewayError(
                "model_metadata_unknown",
                USER_MESSAGES["model_metadata_unknown"],
                503,
                diagnostics={"reason": "journal_budget_missing", "log": events},
            )
        try:
            assert_input_fits(budget, masked)
        except JournalBudgetError as exc:
            _log_event(events, started, "budget", status="error", code=exc.code)
            raise PrivacyGatewayError(
                exc.code,
                exc.message,
                exc.status_code,
                {**(exc.diagnostics or {}), "log": events},
            ) from exc
        diagnostics.update(budget.as_diagnostics())
        diagnostics["estimated_input_tokens"] = budget.estimated_input_tokens
        _log_event(events, started, "budget", status="ok")
    debug_calls += 1
    config = generate_provider()
    layer = {
        "journal_generate": "journalentwurf",
        "journal_reconstruct": "journalrekonstruktion",
        "profile_review": "profilreview",
    }.get(request.purpose, "dialogzug")
    max_tokens = (request.payload or {}).get("max_tokens")
    if max_tokens is not None:
        diagnostics["max_tokens"] = max_tokens
    diagnostics["provider"] = result.provider
    diagnostics["purpose"] = request.purpose
    diagnostics["prompt_slug"] = (request.payload or {}).get("prompt_slug")
    diagnostics["prompt_revision"] = (request.payload or {}).get("prompt_revision")
    diagnostics["model"] = config.model if config else None
    request_trace = {
        "purpose": request.purpose,
        "layer": layer,
        "data_class": request.data_class,
        "prompt_slug": (request.payload or {}).get("prompt_slug"),
        "prompt_revision": (request.payload or {}).get("prompt_revision"),
        "rendered": rendered,
        "masked": masked,
        "mask_input": rendered,
        "provider": result.provider,
        "model": config.model if config else None,
        "detect_provider": detect_name,
        "detect_note": detect_note,
        "detect_model": detect_stats.get("detect_model"),
        "full_detection_coverage": detect_stats.get("full_detection_coverage"),
        "chunk_count": detect_stats.get("chunk_count"),
        "chunks_ok": detect_stats.get("chunks_ok"),
        "entity_counts": detect_stats.get("entity_counts"),
        "confirmed_registry_hits": detect_stats.get("confirmed_registry_hits"),
        "confirmed_registry_applied": detect_stats.get("confirmed_registry_applied"),
        "request_local_hits": detect_stats.get("request_local_hits"),
        "semantic_identity_guaranteed": False,
        "detect_calls": detect_stats.get("detect_calls"),
        "detect_ms": detect_stats.get("detect_ms"),
        "generate_called": False,
        "mapping_count": len(mappings),
        "available_mapping_count": summary["available_mapping_count"],
        "active_mapping_count": summary["active_mapping_count"],
        "masked_occurrence_count": summary["masked_occurrence_count"],
        "active_tokens": summary["active_tokens"],
        "active_entity_types": summary["active_entity_types"],
        "budget": compact_diagnostics(diagnostics),
        "log": events,
    }
    model_policy = {
        "purpose": request.purpose,
        "zdr": True,
        "no_train": True,
        "max_tokens": max_tokens,
        "disable_context_compression": bool(
            (request.payload or {}).get("disable_context_compression")
        ),
    }
    try:
        validate_pre_egress(masked, manifest)
        diagnostics["pre_egress_validation"] = "ok"
        request_trace["pre_egress_validation"] = "ok"
        _log_event(
            events,
            started,
            "pre_egress_validation",
            status="ok",
            active_mapping_count=summary["active_mapping_count"],
            masked_occurrence_count=summary["masked_occurrence_count"],
        )
    except PrivacyGatewayError as exc:
        leak = dict(exc.diagnostics or {})
        diagnostics["pre_egress_validation"] = "failed"
        request_trace["pre_egress_validation"] = "failed"
        _log_event(
            events,
            started,
            "pre_egress_validation",
            status="failed",
            code=exc.code,
            leak_tokens=leak.get("leak_tokens"),
            leak_entity_types=leak.get("leak_entity_types"),
        )
        failed = merge_usage(
            {**diagnostics, "budget_ok": False, "abort_reason": exc.code, "status": "error", "generate_calls": 0},
            None,
            config.model if config else None,
        )
        request_trace["abort_reason"] = exc.code
        request_trace["generate_calls"] = 0
        request_trace["model_text_accepted"] = False
        request_trace["budget"] = compact_diagnostics(failed)
        last_compact = compact_diagnostics(failed)
        _record_test({"purpose": request.purpose, "ok": False, "code": exc.code})
        _raise_with_log(
            exc,
            events,
            purpose=request.purpose,
            extra={**failed, "trace": public_trace(request_trace)},
        )
    usages: list[dict] = []
    raw = ""
    chat = None
    try:
        _log_event(
            events,
            started,
            "model_call",
            attempt=1,
            purpose=request.purpose,
            model=config.model if config else None,
        )
        diagnostics["generate_requested"] = True
        request_trace["generate_requested"] = True
        chat = complete_model([{"role": "user", "content": masked}], model_policy)
        usages.append(chat.usage or {})
        diagnostics["generate_called"] = True
        diagnostics["generate_calls"] = 1
        request_trace["generate_called"] = True
        request_trace["generate_calls"] = 1
        _log_event(events, started, "model_result", attempt=1, **_usage_bits(chat.usage))
        raw = chat.content
        _tokenized, process_meta = _process_response(raw, manifest)
        diagnostics["response_validation"] = process_meta["response_validation"]
        diagnostics["response_normalization"] = process_meta["response_normalization"]
        diagnostics["normalized_cleartext_count"] = process_meta["normalized_cleartext_count"]
        request_trace["response_validation"] = process_meta["response_validation"]
        request_trace["response_normalization"] = process_meta["response_normalization"]
        request_trace["normalized_cleartext_count"] = process_meta["normalized_cleartext_count"]
        diagnostics["model_text_accepted"] = True
        _log_event(
            events,
            started,
            "response_validation",
            attempt=1,
            status="ok",
            response_normalization=process_meta["response_normalization"],
        )
    except PrivacyGatewayError as exc:
        failed = merge_usage(
            {
                **diagnostics,
                "budget_ok": False,
                "abort_reason": exc.code,
                "status": "error",
                "generate_calls": len(usages),
                "generate_ms": int((time.perf_counter() - started) * 1000),
            },
            sum_usages(usages) or None,
            (chat.model if chat else None) or (config.model if config else None),
        )
        request_trace["raw"] = raw
        request_trace["abort_reason"] = exc.code
        request_trace["generate_calls"] = len(usages)
        request_trace["model_text_accepted"] = False
        request_trace["budget"] = compact_diagnostics(failed)
        request_trace["log"] = events
        last_compact = compact_diagnostics(failed)
        _record_test({"purpose": request.purpose, "ok": False, "code": exc.code})
        _raise_with_log(
            exc,
            events,
            purpose=request.purpose,
            extra={**failed, "trace": public_trace(request_trace)},
        )
    diagnostics = merge_usage(
        {
            **diagnostics,
            "context_compression": chat.context_compression,
            "budget_ok": True,
            "status": "ok",
            "response_validation": "ok",
            "generate_ms": int((time.perf_counter() - started) * 1000),
            "generate_calls": 1,
            "model_text_accepted": True,
        },
        sum_usages(usages) or chat.usage,
        chat.model or (config.model if config else None),
    )
    result.content = _visible_response(raw, manifest)
    result.allowed = True
    result.local_identities = local_identities
    result.diagnostics = compact_diagnostics(diagnostics)
    request_trace["raw"] = raw
    request_trace["reply"] = result.content
    request_trace["model"] = chat.model or request_trace.get("model")
    request_trace["response_validation"] = "ok"
    request_trace["model_text_accepted"] = True
    request_trace["budget"] = compact_diagnostics(diagnostics)
    request_trace["log"] = events
    result.trace = public_trace(request_trace)
    last_compact = compact_diagnostics(diagnostics)
    _record_test({"purpose": request.purpose, "ok": True, "prompt_slug": request_trace.get("prompt_slug")})
    return result


def last_trace() -> dict | None:
    """Compact diagnostics only. Full prompts live on the current GatewayResult."""
    return last_compact


def public_trace(trace: dict | None) -> dict | None:
    """Admin test view for the current response. No mapping table, no secrets."""
    if not trace:
        return None
    budget = dict(trace.get("budget") or {})
    for key in ("rendered", "masked", "intern", "raw", "reply", "prompt", "messages"):
        budget.pop(key, None)
    return {
        "purpose": trace.get("purpose"),
        "layer": trace.get("layer"),
        "data_class": trace.get("data_class"),
        "prompt_slug": trace.get("prompt_slug"),
        "prompt_revision": trace.get("prompt_revision"),
        "provider": trace.get("provider"),
        "model": trace.get("model"),
        "detect_provider": trace.get("detect_provider"),
        "detect_note": trace.get("detect_note"),
        "detect_model": trace.get("detect_model"),
        "full_detection_coverage": trace.get("full_detection_coverage"),
        "chunk_count": trace.get("chunk_count"),
        "chunks_ok": trace.get("chunks_ok"),
        "entity_counts": trace.get("entity_counts"),
        "confirmed_registry_hits": trace.get("confirmed_registry_hits"),
        "confirmed_registry_applied": trace.get("confirmed_registry_applied"),
        "request_local_hits": trace.get("request_local_hits"),
        "semantic_identity_guaranteed": False,
        "detect_calls": trace.get("detect_calls"),
        "stored_title": trace.get("stored_title"),
        "stored_body": trace.get("stored_body"),
        "detect_ms": trace.get("detect_ms"),
        "generate_ms": (trace.get("budget") or {}).get("generate_ms") if isinstance(trace.get("budget"), dict) else trace.get("generate_ms"),
        "generate_called": trace.get("generate_called"),
        "generate_requested": trace.get("generate_requested"),
        "mapping_count": trace.get("mapping_count"),
        "available_mapping_count": trace.get("available_mapping_count"),
        "active_mapping_count": trace.get("active_mapping_count"),
        "masked_occurrence_count": trace.get("masked_occurrence_count"),
        "active_tokens": trace.get("active_tokens"),
        "active_entity_types": trace.get("active_entity_types"),
        "pre_egress_validation": trace.get("pre_egress_validation"),
        "response_validation": trace.get("response_validation"),
        "response_normalization": trace.get("response_normalization"),
        "normalized_cleartext_count": trace.get("normalized_cleartext_count"),
        "generate_calls": trace.get("generate_calls"),
        "model_text_accepted": trace.get("model_text_accepted"),
        "provenance_decision": trace.get("provenance_decision"),
        "abort_reason": trace.get("abort_reason"),
        "leak_tokens": trace.get("leak_tokens"),
        "leak_entity_types": trace.get("leak_entity_types"),
        "intern": trace.get("rendered") or trace.get("intern"),
        "egress": trace.get("masked") or trace.get("egress"),
        "mask_input": trace.get("mask_input"),
        "raw": trace.get("raw"),
        "reply": trace.get("reply"),
        "budget": budget or None,
        "log": trace.get("log") or None,
        "response_validation_retry": trace.get("response_validation_retry"),
        "stage1": trace.get("stage1"),
        "reason": trace.get("reason"),
        "guard": trace.get("guard"),
        "stages": trace.get("stages"),
        "status": trace.get("status"),
        "coverage": trace.get("coverage"),
        "source_count": trace.get("source_count"),
    }