"""Semantic entity detection for the privacy gateway.

Detect may see plaintext. Generate must not. Detector output is untrusted:
spans and types only, never tokens, never a durable registry write.
"""
from __future__ import annotations

import contextvars
import json
import re
import time
from dataclasses import dataclass, field
from typing import Any

import placeholder_mvp  # noqa: F401
from db import get_db, row_to_dict
from env_loader import allows_remote_plaintext_detect
from identity_store import (
    ENTITY_TYPES,
    confirmed_match_labels,
    is_maskable_label,
    is_registry_maskable_label,
    list_confirmed_identities,
    masking_rows_from_confirmed,
    normalize_entity_type,
    record_review_proposal,
)
from placeholders import PlaceholderError, resolve_template
from providers import ChatResult, ProviderError, complete_chat, detect_provider

DETECT_CHUNK_CHARS = 4000
DETECT_CHUNK_OVERLAP = 250
DETECT_TIMEOUT = 90.0
DETECT_MAX_TOKENS = 1024
JSON_BLOCK = re.compile(r"\{.*\}", re.DOTALL)
ALLOWED_ENTITY_FIELDS = frozenset({"start", "end", "text", "entity_type"})
ALLOWED_ROOT_FIELDS = frozenset({"entities"})
TYPE_PRIORITY = {"PERSON": 0, "PROJECT": 1, "ORG": 2, "PLACE": 3}
NON_IDENTITY_ENTITY_TYPES = frozenset(
    {
        "FOOD",
        "DISH",
        "MEAL",
        "CUISINE",
        "FOODSTUFF",
        "INGREDIENT",
    }
)

ERROR_DETECT_UNAVAILABLE = "detect_provider_unavailable"
ERROR_DETECT_INCOMPLETE = "detect_incomplete"
ERROR_DETECT_INVALID = "detect_invalid_output"
ERROR_DETECT_TRUNCATED = "detect_truncated"
ERROR_DETECT_CHUNK = "detect_chunk_failed"
DETECT_MAX_PASSES = 2
DETECT_CONTRACT_RETRY_CODES = frozenset(
    {
        ERROR_DETECT_INVALID,
        ERROR_DETECT_TRUNCATED,
        ERROR_DETECT_INCOMPLETE,
    }
)
SCHEMA_RETRY_HINT = (
    "Der vorherige Durchgang hat das verbindliche Schema nicht erfüllt. "
    "Antworte ausschließlich mit dem geforderten JSON-Objekt. "
    "Nur die Felder start, end, text und entity_type sind zulässig. "
    "Nur entity_type-Werte PERSON, PLACE, ORG, PROJECT. "
    "Lebensmittel und Gerichte weglassen, nicht als FOOD oder anderen Typ melden. "
    "Offsets sind 0-basiert und end ausschließlich, bezogen auf den Text nach «Text:». "
    "text muss ein exakter Substring dieses Texts sein. "
    "Keine zusätzlichen Felder, keine Tokens, keine Platzhalter, keine Erklärungen."
)
DETECT_DIALOGUE_FALLBACK_CODES = frozenset(
    {
        ERROR_DETECT_INVALID,
        ERROR_DETECT_TRUNCATED,
        ERROR_DETECT_INCOMPLETE,
        ERROR_DETECT_CHUNK,
    }
)
USER_DETECT_MESSAGES = {
    ERROR_DETECT_UNAVAILABLE: (
        "Persönliche Angaben können gerade nicht geschützt werden. "
        "Es wurde kein Impuls erzeugt."
    ),
    ERROR_DETECT_INCOMPLETE: (
        "Persönliche Angaben konnten nicht vollständig geprüft werden. "
        "Es wurde kein Impuls erzeugt."
    ),
    ERROR_DETECT_INVALID: (
        "Persönliche Angaben konnten nicht sicher zugeordnet werden. "
        "Es wurde kein Impuls erzeugt."
    ),
    ERROR_DETECT_TRUNCATED: (
        "Die Prüfung persönlicher Angaben wurde abgebrochen. "
        "Es wurde kein Impuls erzeugt."
    ),
    ERROR_DETECT_CHUNK: (
        "Die Prüfung persönlicher Angaben ist fehlgeschlagen. "
        "Es wurde kein Impuls erzeugt."
    ),
}


def user_detect_message(code: str, fallback: str = "") -> str:
    return USER_DETECT_MESSAGES.get(code) or fallback or "Persönliche Angaben konnten nicht geprüft werden."


_LETTER = r"A-Za-zÄÖÜäöüß"


class DetectError(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(frozen=True)
class DetectedSpan:
    start: int
    end: int
    text: str
    entity_type: str
    chunk_index: int


@dataclass
class DetectionStats:
    detect_provider: str | None = None
    detect_model: str | None = None
    detect_note: str | None = None
    source_chars: int = 0
    chunk_count: int = 0
    chunks_ok: int = 0
    full_detection_coverage: bool = False
    entity_counts: dict[str, int] = field(default_factory=dict)
    confirmed_registry_hits: int = 0
    request_local_hits: int = 0
    confirmed_registry_applied: bool = False
    semantic_identity_guaranteed: bool = False
    detect_calls: int = 0
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0
    cost: float = 0.0
    cost_known: bool = False
    detect_ms: int = 0
    abort_reason: str | None = None
    detect_passes: int = 0
    detect_partial_discarded: bool = False
    detect_attempts: list[dict[str, Any]] = field(default_factory=list)
    contract_violation: str | None = None
    invalid_entity_type: str | None = None
    omitted_non_identity_types: list[str] = field(default_factory=list)

    def public(self) -> dict[str, Any]:
        payload = {
            "detect_provider": self.detect_provider,
            "detect_model": self.detect_model,
            "detect_note": self.detect_note,
            "source_chars": self.source_chars,
            "chunk_count": self.chunk_count,
            "chunks_ok": self.chunks_ok,
            "full_detection_coverage": self.full_detection_coverage,
            "entity_counts": dict(self.entity_counts),
            "confirmed_registry_hits": self.confirmed_registry_hits,
            "confirmed_registry_applied": self.confirmed_registry_applied,
            "request_local_hits": self.request_local_hits,
            "semantic_identity_guaranteed": False,
            "detect_calls": self.detect_calls,
            "detect_prompt_tokens": self.prompt_tokens,
            "detect_completion_tokens": self.completion_tokens,
            "detect_total_tokens": self.total_tokens,
            "detect_ms": self.detect_ms,
            "detect_passes": self.detect_passes,
            "detect_partial_discarded": self.detect_partial_discarded,
            "detect_attempts": [dict(item) for item in self.detect_attempts],
            "generate_called": False,
        }
        if self.cost_known:
            payload["detect_cost"] = self.cost
            payload["detect_cost_unknown"] = False
        else:
            payload["detect_cost_unknown"] = True
        if self.abort_reason:
            payload["abort_reason"] = self.abort_reason
        if self.contract_violation:
            payload["contract_violation"] = self.contract_violation
        if self.invalid_entity_type:
            payload["invalid_entity_type"] = self.invalid_entity_type
        if self.omitted_non_identity_types:
            payload["omitted_non_identity_types"] = list(self.omitted_non_identity_types)
        return payload


@dataclass
class DetectionOutcome:
    mappings: list[dict]
    stats: DetectionStats
    local_identities: list[dict]


_injected_spans: contextvars.ContextVar[list[dict] | None] = contextvars.ContextVar(
    "kansho_detect_spans",
    default=None,
)
_injected_fail: contextvars.ContextVar[DetectError | None] = contextvars.ContextVar(
    "kansho_detect_fail",
    default=None,
)
_injected_truncated: contextvars.ContextVar[bool] = contextvars.ContextVar(
    "kansho_detect_truncated",
    default=False,
)
_injected_script: contextvars.ContextVar[Any] = contextvars.ContextVar(
    "kansho_detect_script",
    default=None,
)


def install_test_spans(spans: list[dict] | None) -> None:
    """Test-only span injection. Production never calls this."""
    _injected_spans.set(None if spans is None else [dict(item) for item in spans])


def install_test_detect_failure(error: DetectError | None) -> None:
    _injected_fail.set(error)


def install_test_truncated(flag: bool = True) -> None:
    _injected_truncated.set(bool(flag))


def install_test_detect_script(handler=None) -> None:
    """Test-only per-attempt chunk script. Production never calls this.

    handler(attempt, chunk_index, chunk_text, offset, schema_retry=...) may return
    entity lists, raw JSON, ChatResult, or DetectError. Attempt is 1-based.
    """
    _injected_script.set(handler)


def reset_detect_test_hooks() -> None:
    _injected_spans.set(None)
    _injected_fail.set(None)
    _injected_truncated.set(False)
    _injected_script.set(None)


def uses_llm_detect(config) -> bool:
    """Local detect stays a provider role. Remote plaintext detect needs an explicit allow."""
    if not config or config.mode != "http":
        return False
    if config.local:
        return True
    return allows_remote_plaintext_detect()


def _detect_prompt() -> dict:
    with get_db() as conn:
        row = row_to_dict(
            conn.execute(
                "SELECT * FROM ai_prompts WHERE slug = ? AND active = 1",
                ("mvp.entity_detect",),
            ).fetchone()
        )
    if not row or not (row.get("template") or "").strip():
        raise DetectError(
            "detect_prompt_missing",
            "Prompt mvp.entity_detect fehlt in der Konfiguration.",
        )
    return row


def split_detect_chunks(text: str, *, chunk_chars: int | None = None, overlap: int | None = None) -> list[tuple[int, str]]:
    source = text or ""
    if not source:
        return [(0, "")]
    size = max(32, int(DETECT_CHUNK_CHARS if chunk_chars is None else chunk_chars))
    overlap_n = DETECT_CHUNK_OVERLAP if overlap is None else overlap
    overlap_n = max(0, min(int(overlap_n), size // 2))
    if len(source) <= size:
        return [(0, source)]
    chunks: list[tuple[int, str]] = []
    start = 0
    n = len(source)
    while start < n:
        end = min(n, start + size)
        if end < n:
            window = source[start:end]
            cut = max(window.rfind("\n"), window.rfind(" "))
            if cut >= size // 3:
                end = start + cut + 1
        chunks.append((start, source[start:end]))
        if end >= n:
            break
        nxt = end - overlap_n
        if nxt <= start:
            nxt = end
        start = nxt
    return chunks


def _span_dict(text: str, label: str, entity_type: str, *, from_index: int = 0) -> dict | None:
    start = text.find(label, from_index)
    if start < 0:
        return None
    return {
        "start": start,
        "end": start + len(label),
        "text": label,
        "entity_type": entity_type,
    }


def _contract_fake_spans(text: str) -> list[dict]:
    """Deterministic contract fixture. Not a semantic quality claim."""
    source = text or ""
    found: list[dict] = []
    seen: set[tuple[int, int, str]] = set()

    def add(label: str, entity_type: str, *, require: str | None = None) -> None:
        if require and require not in source:
            return
        start = 0
        while True:
            item = _span_dict(source, label, entity_type, from_index=start)
            if not item:
                return
            key = (item["start"], item["end"], entity_type)
            if key not in seen:
                seen.add(key)
                found.append(item)
            start = item["end"]

    if re.search(r"(?i)Frau\s+Sushi", source):
        add("Sushi", "PERSON")
    else:
        for match in re.finditer(r"Sushi", source):
            if source[match.end() : match.end() + 4] == " kam":
                key = (match.start(), match.end(), "PERSON")
                if key not in seen:
                    seen.add(key)
                    found.append(
                        {
                            "start": match.start(),
                            "end": match.end(),
                            "text": "Sushi",
                            "entity_type": "PERSON",
                        }
                    )
    if re.search(r"(?i)Projekt\s+Aurora", source):
        add("Aurora", "PROJECT")
    if re.search(r"(?i)(Organisation|Firma|bei)\s+Nordwerk", source):
        add("Nordwerk", "ORG")
    if re.search(r"(?i)(in|nach|aus)\s+Hamburg", source):
        add("Hamburg", "PLACE")
    for name in ("Anna", "Clarissa", "Hanna", "Maren"):
        if re.search(rf"(?<![{_LETTER}]){name}(?![{_LETTER}])", source):
            add(name, "PERSON")
    return found


def _contract_error(
    message: str,
    *,
    violation: str,
    code: str = ERROR_DETECT_INVALID,
    invalid_entity_type: str | None = None,
    status_code: int = 503,
) -> DetectError:
    diagnostics: dict[str, Any] = {"contract_violation": violation}
    if invalid_entity_type:
        delivered = re.sub(r"[^A-Z0-9_:-]", "", str(invalid_entity_type).strip().upper())[:40]
        if delivered:
            diagnostics["invalid_entity_type"] = delivered
    return DetectError(code, message, status_code, diagnostics)


def _parse_detect_json(raw: str) -> dict:
    text = (raw or "").strip()
    if text.startswith("```"):
        text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.IGNORECASE | re.DOTALL).strip()
    match = JSON_BLOCK.search(text)
    if not match:
        raise _contract_error("Detect-Antwort war kein gültiges JSON.", violation="invalid_json")
    try:
        data = json.loads(match.group(0))
    except json.JSONDecodeError as exc:
        raise _contract_error("Detect-Antwort war kein gültiges JSON.", violation="invalid_json") from exc
    if not isinstance(data, dict):
        raise _contract_error("Detect-Antwort muss ein Objekt mit entities sein.", violation="invalid_json")
    extra = set(data.keys()) - ALLOWED_ROOT_FIELDS
    if extra:
        raise _contract_error("Detect-Antwort enthält unerwartete Felder.", violation="extra_fields")
    items = data.get("entities")
    if not isinstance(items, list):
        raise _contract_error("Detect-entities muss eine Liste sein.", violation="invalid_json")
    return data


def _label_occurrences(chunk_text: str, label: str) -> list[tuple[int, int]]:
    if not label:
        return []
    exact = list(
        re.finditer(rf"(?<![{_LETTER}]){re.escape(label)}(?![{_LETTER}])", chunk_text or "")
    )
    if exact:
        return [(match.start(), match.end()) for match in exact]
    found = list(
        re.finditer(
            rf"(?<![{_LETTER}]){re.escape(label)}(?![{_LETTER}])",
            chunk_text or "",
            re.IGNORECASE,
        )
    )
    return [(match.start(), match.end()) for match in found]


def _ground_span(chunk_text: str, start: int, end: int, text: str) -> tuple[int, int, str] | None:
    """Trust the reported word only if it exists in this chunk. Offsets are a hint."""
    source = chunk_text or ""
    label = (text or "").strip()
    if not label:
        return None
    n = len(source)
    windows: list[tuple[int, int]] = []
    if 0 <= start < end <= n:
        windows.append((start, end))
    if 0 <= start < end + 1 <= n and (start, end + 1) not in windows:
        windows.append((start, end + 1))
    for left, right in windows:
        sliced = source[left:right]
        if sliced == text:
            return left, right, sliced
        if sliced.strip() == label:
            inner = left + (len(sliced) - len(sliced.lstrip()))
            if source[inner : inner + len(label)] == label:
                return inner, inner + len(label), label
    hits = _label_occurrences(source, label)
    if not hits:
        return None
    best = min(hits, key=lambda item: (abs(item[0] - start), item[0]))
    return best[0], best[1], source[best[0] : best[1]]


def _delivered_entity_type(raw: Any) -> str:
    if not isinstance(raw, str):
        return ""
    return re.sub(r"[^A-Z0-9_:-]", "", raw.strip().upper())[:40]


def _identity_mention(chunk_text: str, start: int, end: int) -> bool:
    from privacy_gateway import is_identity_mention

    return is_identity_mention(chunk_text, start, end, "PERSON:00")


def _span_from_non_identity(item: dict, chunk_text: str, chunk_index: int) -> DetectedSpan | None:
    """Food/dish types are not identities. Keep only a person-shaped mention of the same word."""
    text = item.get("text")
    if not isinstance(text, str) or not text.strip():
        return None
    try:
        start = int(item.get("start") or 0)
    except (TypeError, ValueError):
        start = 0
    try:
        end = int(item.get("end") or 0)
    except (TypeError, ValueError):
        end = 0
    grounded = _ground_span(chunk_text, start, end, text)
    if not grounded:
        return None
    left, right, local_text = grounded
    if not _identity_mention(chunk_text, left, right):
        return None
    return DetectedSpan(
        start=left,
        end=right,
        text=local_text,
        entity_type="PERSON",
        chunk_index=chunk_index,
    )


def validate_detected_entity(item: Any, chunk_text: str, chunk_index: int) -> DetectedSpan | None:
    if not isinstance(item, dict):
        raise _contract_error("Detect-Entity muss ein Objekt sein.", violation="invalid_entity")
    if "token" in item or "placeholder" in item:
        raise _contract_error("Detect darf keine Tokens festlegen.", violation="extra_fields")
    raw_type = item.get("entity_type")
    delivered = _delivered_entity_type(raw_type)
    kind = normalize_entity_type(raw_type if isinstance(raw_type, str) else "", default="")
    if not kind and delivered in NON_IDENTITY_ENTITY_TYPES:
        return _span_from_non_identity(item, chunk_text, chunk_index)
    extra = set(item.keys()) - ALLOWED_ENTITY_FIELDS
    if extra:
        raise _contract_error("Detect-Entity enthält unerwartete Felder.", violation="extra_fields")
    missing = ALLOWED_ENTITY_FIELDS - set(item.keys())
    if missing:
        raise _contract_error("Detect-Entity ist unvollständig.", violation="missing_fields")
    try:
        start = int(item["start"])
    except (TypeError, ValueError):
        start = 0
    try:
        end = int(item["end"])
    except (TypeError, ValueError):
        end = 0
    text = item.get("text")
    if not isinstance(text, str):
        raise _contract_error("Detect-text fehlt.", violation="missing_fields")
    if not text.strip():
        raise _contract_error("Detect-text ist leer.", violation="invalid_text")
    if not kind:
        delivered_raw = raw_type if isinstance(raw_type, str) else ""
        raise _contract_error(
            "Detect-entity_type ist nicht erlaubt.",
            violation="unknown_entity_type",
            invalid_entity_type=delivered_raw,
        )
    grounded = _ground_span(chunk_text, start, end, text)
    if not grounded:
        raise _contract_error(
            "Detect-Offsets sind nicht im Chunk verankert.",
            violation="unusable_offsets",
        )
    left, right, local_text = grounded
    return DetectedSpan(start=left, end=right, text=local_text, entity_type=kind, chunk_index=chunk_index)


def resolve_overlaps(spans: list[DetectedSpan]) -> list[DetectedSpan]:
    """Longest span wins, then leftmost, then type PERSON > PROJECT > ORG > PLACE."""
    ordered = sorted(
        spans,
        key=lambda item: (
            -(item.end - item.start),
            item.start,
            TYPE_PRIORITY.get(item.entity_type, 9),
            item.chunk_index,
        ),
    )
    kept: list[DetectedSpan] = []
    occupied: list[tuple[int, int]] = []
    for span in ordered:
        if any(span.start < end and span.end > start for start, end in occupied):
            continue
        kept.append(span)
        occupied.append((span.start, span.end))
    return sorted(kept, key=lambda item: (item.start, item.end))


def _add_usage(stats: DetectionStats, usage: dict | None) -> None:
    data = usage or {}
    if data.get("prompt_tokens") is not None:
        stats.prompt_tokens += int(data.get("prompt_tokens") or 0)
    if data.get("completion_tokens") is not None:
        stats.completion_tokens += int(data.get("completion_tokens") or 0)
    if data.get("total_tokens") is not None:
        stats.total_tokens += int(data.get("total_tokens") or 0)
    extra = data.get("cost")
    if extra is None:
        extra = data.get("total_cost")
    if extra is not None:
        try:
            stats.cost += float(extra)
            stats.cost_known = True
        except (TypeError, ValueError):
            pass


def _llm_chunk(config, excerpt: str, *, schema_retry: bool = False) -> ChatResult:
    prompt = resolve_template(
        _detect_prompt()["template"],
        {"source_text": excerpt, "known_labels": ""},
    )
    if schema_retry:
        prompt = f"{prompt.rstrip()}\n\n{SCHEMA_RETRY_HINT}"
    return complete_chat(
        config,
        [{"role": "user", "content": prompt}],
        timeout=DETECT_TIMEOUT,
        max_tokens=DETECT_MAX_TOKENS,
        disable_context_compression=True,
    )


def _note_omitted_type(stats: DetectionStats | None, delivered: str) -> None:
    if not stats or not delivered or delivered not in NON_IDENTITY_ENTITY_TYPES:
        return
    if delivered not in stats.omitted_non_identity_types:
        stats.omitted_non_identity_types.append(delivered)


def _entities_from_result(
    result: ChatResult,
    chunk_text: str,
    chunk_index: int,
    stats: DetectionStats | None = None,
) -> list[DetectedSpan]:
    if (result.finish_reason or "").lower() in {"length", "max_tokens"}:
        raise DetectError(
            ERROR_DETECT_TRUNCATED,
            "Detect-Ausgabe wurde abgeschnitten. Generate wird nicht freigegeben.",
            diagnostics={"contract_violation": "truncated"},
        )
    data = _parse_detect_json(result.content or "")
    spans: list[DetectedSpan] = []
    for item in data.get("entities") or []:
        span = validate_detected_entity(item, chunk_text, chunk_index)
        if span is None:
            delivered = _delivered_entity_type(item.get("entity_type") if isinstance(item, dict) else "")
            _note_omitted_type(stats, delivered)
            continue
        spans.append(span)
    return spans


def _to_global(span: DetectedSpan, offset: int) -> DetectedSpan:
    return DetectedSpan(
        start=span.start + offset,
        end=span.end + offset,
        text=span.text,
        entity_type=span.entity_type,
        chunk_index=span.chunk_index,
    )


def _dedupe_spans(spans: list[DetectedSpan]) -> list[DetectedSpan]:
    unique: dict[tuple[int, int, str, str], DetectedSpan] = {}
    for span in spans:
        key = (span.start, span.end, span.text, span.entity_type)
        unique[key] = span
    return resolve_overlaps(list(unique.values()))


def _next_token(used: set[str], entity_type: str) -> str:
    prefix = f"{entity_type}:"
    n = 1
    while True:
        token = f"{prefix}{n:02d}"
        if token not in used:
            used.add(token)
            return token
        n += 1


def _assign_request_tokens(
    spans: list[DetectedSpan],
    confirmed: list[dict],
) -> tuple[list[dict], int, int]:
    used_tokens = {(item.get("token") or "").upper() for item in confirmed if item.get("token")}
    by_label: dict[str, dict] = {}
    request_local = 0
    confirmed_hits = 0
    mappings: list[dict] = []

    def confirmed_for(label: str) -> dict | None:
        needle = (label or "").strip().casefold()
        for item in confirmed:
            for candidate in confirmed_match_labels(item):
                if candidate.casefold() == needle:
                    return item
        return None

    for span in spans:
        label = span.text
        if not is_maskable_label(label):
            continue
        found = confirmed_for(label)
        if found:
            token = found["token"]
            canonical = found["canonical_label"]
            source = "confirmed_registry"
            confirmed_hits += 1
            entity_type = found["entity_type"]
            demask = canonical
            aliases = list(found.get("aliases") or [])
            labels = confirmed_match_labels(found)
        else:
            key = f"{span.entity_type}:{label.casefold()}"
            if key not in by_label:
                by_label[key] = {
                    "token": _next_token(used_tokens, span.entity_type),
                    "entity_type": span.entity_type,
                }
                request_local += 1
            token = by_label[key]["token"]
            entity_type = span.entity_type
            source = "request_local"
            demask = label
            aliases = []
            labels = [label]
        mappings.append(
            {
                "token": token,
                "local_label": label,
                "canonical_label": demask,
                "demask_label": demask,
                "entity_type": entity_type,
                "source": source,
                "start": span.start,
                "end": span.end,
                "aliases": aliases,
                "labels": labels,
            }
        )
    return mappings, request_local, confirmed_hits


def _merge_confirmed_safety_net(text: str, mappings: list[dict], profile_id: str | None) -> tuple[list[dict], int]:
    if not profile_id:
        return mappings, 0
    already_labels = {
        (item.get("local_label") or "").casefold()
        for item in mappings
        if item.get("start") is None and item.get("source") == "confirmed_registry"
    }
    extra: list[dict] = []
    hits = 0
    for row in masking_rows_from_confirmed(profile_id):
        label = row.get("local_label") or ""
        if not label or not is_registry_maskable_label(label) or label.casefold() in already_labels:
            continue
        if not re.search(rf"(?<![{_LETTER}]){re.escape(label)}(?![{_LETTER}])", text or "", re.IGNORECASE):
            continue
        extra.append(row)
        already_labels.add(label.casefold())
        hits += 1
    return mappings + extra, hits


def _merge_pass_stats(overall: DetectionStats, pass_stats: DetectionStats) -> None:
    overall.detect_calls += pass_stats.detect_calls
    overall.prompt_tokens += pass_stats.prompt_tokens
    overall.completion_tokens += pass_stats.completion_tokens
    overall.total_tokens += pass_stats.total_tokens
    overall.chunk_count = pass_stats.chunk_count
    overall.chunks_ok = pass_stats.chunks_ok
    overall.detect_note = pass_stats.detect_note or overall.detect_note
    overall.detect_model = pass_stats.detect_model or overall.detect_model
    if pass_stats.cost_known:
        overall.cost_known = True
        overall.cost += pass_stats.cost
    for kind in pass_stats.omitted_non_identity_types:
        if kind not in overall.omitted_non_identity_types:
            overall.omitted_non_identity_types.append(kind)


def _pass_snapshot(
    pass_stats: DetectionStats,
    *,
    attempt: int,
    coverage: bool,
    abort: str | None = None,
) -> dict[str, Any]:
    row: dict[str, Any] = {
        "attempt": attempt,
        "schema_retry": attempt > 1,
        "chunk_count": pass_stats.chunk_count,
        "chunks_ok": pass_stats.chunks_ok,
        "detect_calls": pass_stats.detect_calls,
        "full_detection_coverage": coverage,
        "detect_prompt_tokens": pass_stats.prompt_tokens,
        "detect_completion_tokens": pass_stats.completion_tokens,
        "detect_total_tokens": pass_stats.total_tokens,
        "detect_ms": pass_stats.detect_ms,
    }
    if pass_stats.cost_known:
        row["detect_cost"] = pass_stats.cost
        row["detect_cost_unknown"] = False
    else:
        row["detect_cost_unknown"] = True
    if abort:
        row["abort_reason"] = abort
    if pass_stats.contract_violation:
        row["contract_violation"] = pass_stats.contract_violation
    if pass_stats.invalid_entity_type:
        row["invalid_entity_type"] = pass_stats.invalid_entity_type
    if pass_stats.omitted_non_identity_types:
        row["omitted_non_identity_types"] = list(pass_stats.omitted_non_identity_types)
    return row


def _script_result(raw: Any, *, model: str | None) -> ChatResult:
    if isinstance(raw, DetectError):
        raise raw
    if isinstance(raw, ChatResult):
        return raw
    if isinstance(raw, str):
        return ChatResult(content=raw, model=model, usage={}, finish_reason="stop")
    if isinstance(raw, dict) and "entities" in raw:
        return ChatResult(content=json.dumps(raw), model=model, usage={}, finish_reason="stop")
    if isinstance(raw, list):
        return ChatResult(
            content=json.dumps({"entities": raw}),
            model=model,
            usage={},
            finish_reason="stop",
        )
    raise _contract_error("Detect-Testskript lieferte keine verwertbare Antwort.", violation="invalid_json")


def _chunk_chat_result(
    config,
    chunk_text: str,
    *,
    offset: int,
    index: int,
    attempt: int,
    schema_retry: bool,
    stats: DetectionStats,
) -> ChatResult:
    if _injected_truncated.get():
        raise DetectError(
            ERROR_DETECT_TRUNCATED,
            "Detect-Ausgabe wurde abgeschnitten. Generate wird nicht freigegeben.",
            diagnostics={"contract_violation": "truncated"},
        )
    script = _injected_script.get()
    if script is not None:
        stats.detect_note = "injected"
        try:
            raw = script(attempt, index, chunk_text, offset, schema_retry=schema_retry)
        except TypeError:
            raw = script(attempt, index, chunk_text, offset)
        return _script_result(raw, model=config.model)
    spans = _injected_spans.get()
    if spans is not None:
        raw_items = [
            item
            for item in spans
            if int(item.get("start") or 0) >= offset
            and int(item.get("end") or 0) <= offset + len(chunk_text)
        ]
        local_items = []
        for item in raw_items:
            local = dict(item)
            local["start"] = int(item["start"]) - offset
            local["end"] = int(item["end"]) - offset
            local_items.append(local)
        stats.detect_note = "injected"
        return ChatResult(
            content=json.dumps({"entities": local_items}),
            model=config.model,
            usage={},
            finish_reason="stop",
        )
    if config.mode == "fake":
        stats.detect_note = "fake"
        return ChatResult(
            content=json.dumps({"entities": _contract_fake_spans(chunk_text)}),
            model=config.model,
            usage={},
            finish_reason="stop",
        )
    if uses_llm_detect(config):
        try:
            result = _llm_chunk(config, chunk_text, schema_retry=schema_retry)
        except PlaceholderError as exc:
            raise DetectError(exc.code, exc.message) from exc
        except ProviderError as exc:
            raise DetectError(
                ERROR_DETECT_CHUNK,
                "Ein Detect-Chunk ist fehlgeschlagen. Generate wird nicht freigegeben.",
                exc.status_code,
            ) from exc
        stats.detect_note = "local_llm" if config.local else "remote_llm"
        stats.detect_model = result.model or stats.detect_model
        return result
    raise DetectError(
        ERROR_DETECT_UNAVAILABLE,
        "Semantische Detection ist nicht verfügbar. Generate wird nicht freigegeben.",
    )


def _run_detection_pass(
    config,
    chunks: list[tuple[int, str]],
    *,
    attempt: int,
    schema_retry: bool,
    stats: DetectionStats,
) -> list[DetectedSpan]:
    collected: list[DetectedSpan] = []
    for index, (offset, chunk_text) in enumerate(chunks):
        result = _chunk_chat_result(
            config,
            chunk_text,
            offset=offset,
            index=index,
            attempt=attempt,
            schema_retry=schema_retry,
            stats=stats,
        )
        stats.detect_calls += 1
        _add_usage(stats, result.usage)
        spans = _entities_from_result(result, chunk_text, index, stats)
        collected.extend(_to_global(span, offset) for span in spans)
        stats.chunks_ok += 1
    if stats.chunks_ok != stats.chunk_count:
        raise DetectError(
            ERROR_DETECT_INCOMPLETE,
            "Detection hat nicht alle Chunks geprüft. Generate wird nicht freigegeben.",
            diagnostics={"contract_violation": "incomplete"},
        )
    return collected


def detect_personal_egress(profile_id: str | None, source_text: str) -> DetectionOutcome:
    """Full semantic detection of the personal generate-egress. Fail closed if incomplete."""
    started = time.perf_counter()
    stats = DetectionStats(source_chars=len(source_text or ""))
    injected_fail = _injected_fail.get()
    if injected_fail is not None:
        stats.abort_reason = injected_fail.code
        stats.detect_ms = int((time.perf_counter() - started) * 1000)
        stats.detect_passes = 0
        raise DetectError(
            injected_fail.code,
            injected_fail.message,
            injected_fail.status_code,
            {**(injected_fail.diagnostics or {}), **stats.public()},
        )
    config = detect_provider()
    if not config:
        stats.abort_reason = ERROR_DETECT_UNAVAILABLE
        stats.detect_ms = int((time.perf_counter() - started) * 1000)
        raise DetectError(
            ERROR_DETECT_UNAVAILABLE,
            "Semantische Detection ist nicht konfiguriert. Generate wird nicht freigegeben.",
            diagnostics=stats.public(),
        )
    if config.mode == "http" and not config.local and not allows_remote_plaintext_detect():
        stats.abort_reason = "remote_detect_blocked_production"
        stats.detect_ms = int((time.perf_counter() - started) * 1000)
        raise DetectError(
            ERROR_DETECT_UNAVAILABLE,
            "Externes Klartext-Detect ist im Produktivmodus nicht zulässig, und ein lokales Detect-Modell fehlt.",
            diagnostics=stats.public(),
        )
    stats.detect_provider = config.name
    stats.detect_model = config.model
    chunks = split_detect_chunks(source_text or "")
    stats.chunk_count = len(chunks)
    collected: list[DetectedSpan] = []
    last_error: DetectError | None = None
    for attempt in range(1, DETECT_MAX_PASSES + 1):
        pass_stats = DetectionStats(
            source_chars=stats.source_chars,
            detect_provider=stats.detect_provider,
            detect_model=stats.detect_model,
            chunk_count=len(chunks),
        )
        pass_started = time.perf_counter()
        try:
            collected = _run_detection_pass(
                config,
                chunks,
                attempt=attempt,
                schema_retry=attempt > 1,
                stats=pass_stats,
            )
            pass_stats.full_detection_coverage = True
            pass_stats.detect_ms = int((time.perf_counter() - pass_started) * 1000)
            _merge_pass_stats(stats, pass_stats)
            stats.detect_attempts.append(
                _pass_snapshot(pass_stats, attempt=attempt, coverage=True)
            )
            last_error = None
            break
        except DetectError as exc:
            inner = dict(exc.diagnostics or {})
            pass_stats.abort_reason = exc.code
            pass_stats.contract_violation = inner.get("contract_violation")
            pass_stats.invalid_entity_type = inner.get("invalid_entity_type")
            pass_stats.full_detection_coverage = False
            pass_stats.detect_ms = int((time.perf_counter() - pass_started) * 1000)
            _merge_pass_stats(stats, pass_stats)
            stats.detect_attempts.append(
                _pass_snapshot(pass_stats, attempt=attempt, coverage=False, abort=exc.code)
            )
            collected = []
            last_error = DetectError(exc.code, exc.message, exc.status_code, inner)
            retryable = exc.code in DETECT_CONTRACT_RETRY_CODES
            if attempt < DETECT_MAX_PASSES and retryable:
                stats.detect_partial_discarded = True
                continue
            stats.abort_reason = exc.code
            stats.contract_violation = pass_stats.contract_violation
            stats.invalid_entity_type = pass_stats.invalid_entity_type
            stats.full_detection_coverage = False
            stats.detect_passes = attempt
            stats.detect_ms = int((time.perf_counter() - started) * 1000)
            raise DetectError(
                exc.code,
                exc.message,
                exc.status_code,
                {**inner, **stats.public()},
            ) from exc
    if last_error is not None:
        stats.detect_passes = len(stats.detect_attempts)
        stats.detect_ms = int((time.perf_counter() - started) * 1000)
        raise DetectError(
            last_error.code,
            last_error.message,
            last_error.status_code,
            {**(last_error.diagnostics or {}), **stats.public()},
        )
    stats.detect_passes = len(stats.detect_attempts)
    merged = _dedupe_spans(collected)
    confirmed = list_confirmed_identities(profile_id) if profile_id else []
    mappings, request_local, confirmed_from_spans = _assign_request_tokens(merged, confirmed)
    mappings, extra_confirmed = _merge_confirmed_safety_net(source_text or "", mappings, profile_id)
    stats.request_local_hits = request_local
    stats.confirmed_registry_hits = confirmed_from_spans + extra_confirmed
    stats.confirmed_registry_applied = bool(profile_id)
    stats.semantic_identity_guaranteed = False
    counts: dict[str, int] = {kind: 0 for kind in ENTITY_TYPES}
    for item in mappings:
        kind = (item.get("entity_type") or "PERSON").upper()
        counts[kind] = counts.get(kind, 0) + 1
    stats.entity_counts = {key: value for key, value in counts.items() if value}
    stats.full_detection_coverage = True
    stats.detect_ms = int((time.perf_counter() - started) * 1000)
    if profile_id:
        seen_proposals: set[tuple[str, str]] = set()
        for item in mappings:
            if item.get("source") != "request_local":
                continue
            key = ((item.get("local_label") or ""), (item.get("entity_type") or ""))
            if key in seen_proposals:
                continue
            seen_proposals.add(key)
            record_review_proposal(profile_id, item.get("local_label") or "", item.get("entity_type") or "")
    local_identities = [
        {
            "local_label": item.get("local_label"),
            "token": item.get("token"),
            "entity_type": item.get("entity_type"),
            "demask_label": item.get("demask_label") or item.get("local_label"),
            "canonical_label": item.get("canonical_label") or item.get("demask_label") or item.get("local_label"),
            "aliases": list(item.get("aliases") or []),
            "labels": list(item.get("labels") or []),
            "source": item.get("source"),
        }
        for item in mappings
    ]
    return DetectionOutcome(mappings=mappings, stats=stats, local_identities=local_identities)