643 lines
31 KiB
Python
643 lines
31 KiB
Python
"""Semantic detection contracts. Fake provider proves schema and fail-closed, not live quality.
|
|
|
|
Run from backend/: python tests/test_privacy_detect.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from tests.harness import configure_test_engine
|
|
|
|
configure_test_engine()
|
|
|
|
os.environ["KANSHO_FAKE_PROVIDER"] = "1"
|
|
os.environ["KANSHO_FAKE_DETECT"] = "1"
|
|
|
|
from fastapi.testclient import TestClient
|
|
from identity_store import (
|
|
STATUS_LEGACY,
|
|
confirm_identity,
|
|
is_maskable_label,
|
|
is_registry_maskable_label,
|
|
list_confirmed_identities,
|
|
list_mappings,
|
|
list_registry,
|
|
list_review_proposals,
|
|
remember_mapping,
|
|
)
|
|
from main import app
|
|
from privacy_gateway import (
|
|
GatewayRequest,
|
|
PrivacyGatewayError,
|
|
_demask,
|
|
_validate_response,
|
|
_visible_response,
|
|
complete,
|
|
mask_prompt,
|
|
reset_debug,
|
|
)
|
|
from entity_detect import (
|
|
DETECT_CHUNK_CHARS,
|
|
DetectError,
|
|
detect_personal_egress,
|
|
install_test_detect_failure,
|
|
install_test_spans,
|
|
install_test_truncated,
|
|
reset_detect_test_hooks,
|
|
split_detect_chunks,
|
|
uses_llm_detect,
|
|
validate_detected_entity,
|
|
)
|
|
from providers import ProviderConfig
|
|
|
|
|
|
def expect(ok: bool, message: str) -> None:
|
|
if not ok:
|
|
raise SystemExit(f"FAIL: {message}")
|
|
print(f"OK {message}")
|
|
|
|
|
|
def _run(profile_id: str, rendered: str, purpose: str = "dialogue_turn"):
|
|
return complete(
|
|
GatewayRequest(
|
|
prompt_id="detect-test",
|
|
purpose=purpose,
|
|
data_class="B",
|
|
profile_id=profile_id,
|
|
payload={"rendered": rendered, "source_text": rendered},
|
|
)
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
reset_debug()
|
|
with TestClient(app) as client:
|
|
setup = client.post(
|
|
"/api/auth/setup",
|
|
json={"email": "lars@example.test", "name": "Lars", "password": "test-pass"},
|
|
)
|
|
headers = {"X-Auth-Token": setup.json()["token"]}
|
|
profile_id = setup.json()["profile_id"]
|
|
space = client.post("/api/journal/spaces", headers=headers, json={"title": "Alltag"})
|
|
day = client.post(
|
|
f"/api/journal/spaces/{space.json()['id']}/days",
|
|
headers=headers,
|
|
json={"calendar_date": "2026-08-22"},
|
|
)
|
|
conv = client.post(
|
|
f"/api/journal/days/{day.json()['day']['id']}/conversations",
|
|
headers=headers,
|
|
json={"title": "Gespräch"},
|
|
)
|
|
|
|
balcony = _run(profile_id, "Ich ging auf den Balkon und setzte mich.")
|
|
expect(balcony.allowed, "balcony sentence is allowed")
|
|
expect("Balkon" in (balcony.trace.get("egress") or ""), "capitalized common noun stays unmasked")
|
|
expect("[[PERSON:" not in (balcony.trace.get("egress") or ""), "balcony is not a person placeholder")
|
|
|
|
typo = _run(profile_id, "Ich ging auf den Bakon und setzte mich.")
|
|
expect("Bakon" in (typo.trace.get("egress") or ""), "misspelled common noun is not stored as a person")
|
|
expect(
|
|
not any((item.get("canonical_label") or "") == "Bakon" for item in list_confirmed_identities(profile_id)),
|
|
"misspelled noun is not a confirmed identity",
|
|
)
|
|
|
|
food = _run(profile_id, "Ich aß Sushi auf dem Teller.")
|
|
expect("Sushi" in (food.trace.get("egress") or ""), "food context keeps the word")
|
|
person_sushi = _run(profile_id, "Sushi kam ins Wohnzimmer.")
|
|
expect("[[PERSON:" in (person_sushi.trace.get("egress") or ""), "same word can be a person in another sentence")
|
|
expect("Sushi" not in (person_sushi.trace.get("egress") or "").replace("[[PERSON:", ""), "person Sushi is masked")
|
|
|
|
mixed = _run(
|
|
profile_id,
|
|
"Ich traf Anna. Ich arbeitete am privaten Projekt Aurora. "
|
|
"Die Organisation Nordwerk schickte Unterlagen. Ich war in Hamburg.",
|
|
)
|
|
egress = mixed.trace.get("egress") or ""
|
|
expect("[[PERSON:" in egress, "person is typed as PERSON")
|
|
expect("[[PROJECT:" in egress, "private project is typed as PROJECT")
|
|
expect("[[ORG:" in egress, "organisation is typed as ORG")
|
|
expect("[[PLACE:" in egress, "named city is typed as PLACE")
|
|
|
|
sample = "Ich traf Anna."
|
|
start = sample.index("Anna")
|
|
span = validate_detected_entity(
|
|
{"start": start, "end": start + 4, "text": "Anna", "entity_type": "PERSON"},
|
|
sample,
|
|
0,
|
|
)
|
|
expect(span.text == "Anna" and span.start == start, "valid span matches the local substring")
|
|
expect(
|
|
validate_detected_entity(
|
|
{"start": 0, "end": 4, "text": "Anna", "entity_type": "PERSON"},
|
|
sample,
|
|
0,
|
|
).start
|
|
== start,
|
|
"wrong offsets are grounded to the local word",
|
|
)
|
|
expect(
|
|
validate_detected_entity(
|
|
{"start": -1, "end": 4, "text": "Anna", "entity_type": "PERSON"},
|
|
sample,
|
|
0,
|
|
).start
|
|
== start,
|
|
"negative offsets still ground when the word exists",
|
|
)
|
|
try:
|
|
validate_detected_entity(
|
|
{"start": 0, "end": 4, "text": "Nein", "entity_type": "PERSON"},
|
|
sample,
|
|
0,
|
|
)
|
|
missing_word = False
|
|
except DetectError as exc:
|
|
missing_word = exc.code == "detect_invalid_output"
|
|
expect(
|
|
(exc.diagnostics or {}).get("contract_violation") == "unusable_offsets",
|
|
"a word that is not in the chunk is a contract violation",
|
|
)
|
|
expect(missing_word, "a word that is not in the chunk fails closed")
|
|
expect(
|
|
validate_detected_entity(
|
|
{"start": "x", "end": 4, "text": "Anna", "entity_type": "PERSON"},
|
|
sample,
|
|
0,
|
|
).start
|
|
== start,
|
|
"non-numeric offsets still ground when the word exists",
|
|
)
|
|
rejected = []
|
|
for payload, label in (
|
|
({"start": start, "end": start + 4, "text": "Anna", "entity_type": "PET"}, "unknown type"),
|
|
({"start": start, "end": start + 4, "text": "Anna", "entity_type": "PERSON", "token": "PERSON:01"}, "detector token"),
|
|
({"start": start, "end": start + 4, "text": "Anna", "entity_type": "PERSON", "score": 1}, "extra field"),
|
|
):
|
|
try:
|
|
validate_detected_entity(payload, sample, 0)
|
|
except DetectError:
|
|
rejected.append(label)
|
|
expect(rejected == ["unknown type", "detector token", "extra field"], "invalid detector output is rejected")
|
|
|
|
first = _run(profile_id, "Ich war mit Anna in Hamburg und danach allein.")
|
|
expect(first.trace.get("detect_provider") == "fake-detect", "detect provider is separate")
|
|
expect("Anna" not in (first.trace.get("egress") or ""), "Anna masked before generate")
|
|
expect("Hamburg" not in (first.trace.get("egress") or ""), "Hamburg masked before generate")
|
|
expect(first.trace.get("full_detection_coverage") is True, "successful detect records technical chunk coverage")
|
|
expect(first.trace.get("semantic_identity_guaranteed") is False, "chunk coverage is not semantic certainty")
|
|
expect(first.diagnostics.get("semantic_identity_guaranteed") is False, "compact trace does not claim semantic completeness")
|
|
expect(first.trace.get("confirmed_registry_applied") is True, "confirmed registry is applied as a safety net")
|
|
expect(first.trace.get("generate_called") is True, "generate runs after complete detection")
|
|
expect(first.diagnostics.get("detect_calls") == 1, "normal short text uses one detect call")
|
|
labels_after = {item["canonical_label"] for item in list_confirmed_identities(profile_id)}
|
|
expect("Anna" not in labels_after, "detect hit is not auto-confirmed")
|
|
expect("Hamburg" not in labels_after, "place detect hit is not auto-confirmed")
|
|
proposals = [item["observed_label"] for item in list_review_proposals(profile_id)]
|
|
expect("Anna" in proposals, "unconfirmed proposal may be stored for review")
|
|
|
|
second = _run(profile_id, "Anna war ruhig.")
|
|
expect("Anna" not in (second.trace.get("egress") or ""), "second request still runs full detection")
|
|
expect(second.trace.get("request_local_hits", 0) >= 1, "known labels are not excluded from the new detection")
|
|
|
|
poisoned = remember_mapping(profile_id, "Wohnzimmer", "PERSON:77")
|
|
expect(poisoned["status"] == "confirmed", "remember_mapping is explicit confirmation")
|
|
update = client.patch(
|
|
f"/api/admin/identities/{poisoned['id']}",
|
|
headers=headers,
|
|
json={"status": "inactive"},
|
|
)
|
|
expect(update.status_code == 200, "admin can deactivate a false mapping")
|
|
expect(update.json()["status"] == "inactive", "deactivated mapping is inactive")
|
|
later = _run(profile_id, "Ich saß im Wohnzimmer.")
|
|
expect("Wohnzimmer" in (later.trace.get("egress") or ""), "inactive mapping does not keep masking a room")
|
|
expect(
|
|
not any(item.get("canonical_label") == "Wohnzimmer" and item.get("status") == "confirmed" for item in list_registry(profile_id)),
|
|
"later request is not bound to the deactivated mapping",
|
|
)
|
|
|
|
confirmed_project = confirm_identity(profile_id, "Kanshō", entity_type="PROJECT", aliases=["Kansho"])
|
|
canonical = _run(profile_id, "Ich arbeitete an Kansho weiter.")
|
|
expect("[[PROJECT:" in (canonical.trace.get("egress") or ""), "confirmed alias is masked")
|
|
expect("Kansho" not in (canonical.trace.get("egress") or ""), "confirmed alias does not leak")
|
|
project_manifest = mask_prompt(
|
|
"Ich arbeitete an Kansho weiter.",
|
|
[
|
|
{
|
|
"local_label": "Kansho",
|
|
"token": confirmed_project["token"],
|
|
"entity_type": "PROJECT",
|
|
"demask_label": "Kanshō",
|
|
}
|
|
],
|
|
"journal_generate",
|
|
)
|
|
expect(
|
|
_demask(f"[[{confirmed_project['token']}]] weiter", project_manifest) == "Kanshō weiter",
|
|
"confirmed alias demasks to the canonical spelling",
|
|
)
|
|
|
|
observed_manifest = mask_prompt(
|
|
"Ich traf Anna am Abend.",
|
|
[{"local_label": "Anna", "token": "PERSON:01", "demask_label": "Anna"}],
|
|
"journal_generate",
|
|
)
|
|
expect(_demask("[[PERSON:01]] am Abend.", observed_manifest) == "Anna am Abend.", "request-local hit demasks to the observed spelling")
|
|
|
|
leftover = _demask("[[PERSON:01]] und [[PERSON:99]]", mask_prompt("Anna war da.", [{"local_label": "Anna", "token": "PERSON:01", "demask_label": "Anna"}], "journal_generate"))
|
|
expect("[[PERSON:99]]" in leftover, "unknown placeholder is not materialized")
|
|
|
|
long_text = ("Anfang. " + ("x" * 1200) + " Anna in der Mitte. " + ("y" * 1200) + " Ende.")
|
|
with patch("entity_detect.DETECT_CHUNK_CHARS", 400), patch("entity_detect.DETECT_CHUNK_OVERLAP", 80):
|
|
chunks = split_detect_chunks(long_text, chunk_chars=400, overlap=80)
|
|
expect(len(chunks) >= 3, "long egress is chunked")
|
|
covered = 0
|
|
prev_end = 0
|
|
for offset, chunk in chunks:
|
|
expect(offset <= prev_end, "chunks overlap or abut")
|
|
covered = max(covered, offset + len(chunk))
|
|
prev_end = offset + len(chunk)
|
|
expect(covered == len(long_text), "chunk union covers the full text")
|
|
boundary = _run(profile_id, long_text)
|
|
expect(boundary.trace.get("chunk_count") >= 2, "gateway chunks a long prompt")
|
|
expect(boundary.trace.get("chunks_ok") == boundary.trace.get("chunk_count"), "every chunk is marked ok")
|
|
expect("Anna" not in (boundary.trace.get("egress") or ""), "entity at a chunk boundary is still masked")
|
|
expect(boundary.trace.get("full_detection_coverage") is True, "coverage is true only after every chunk")
|
|
|
|
generate_calls = {"n": 0}
|
|
|
|
def boom(_messages, _policy):
|
|
generate_calls["n"] += 1
|
|
raise AssertionError("generate must not run")
|
|
|
|
install_test_detect_failure(DetectError("detect_chunk_failed", "chunk failed"))
|
|
blocked = False
|
|
with patch("privacy_gateway.complete_model", boom):
|
|
try:
|
|
_run(profile_id, "Ich traf Anna.")
|
|
except PrivacyGatewayError as exc:
|
|
blocked = exc.code == "detect_chunk_failed"
|
|
expect(exc.diagnostics.get("generate_called") is False, "failed chunk does not call generate")
|
|
expect(exc.diagnostics.get("full_detection_coverage") is False, "failed chunk is not full coverage")
|
|
expect("Substring" not in (exc.message or ""), "detect errors omit the internal substring wording")
|
|
expect("Angaben" in (exc.message or ""), "detect errors use a user-facing message")
|
|
reset_detect_test_hooks()
|
|
expect(blocked, "failed chunk fails closed")
|
|
expect(generate_calls["n"] == 0, "generate is not called when a chunk fails")
|
|
|
|
install_test_truncated(True)
|
|
truncated = False
|
|
generate_calls["n"] = 0
|
|
with patch("privacy_gateway.complete_model", boom):
|
|
try:
|
|
_run(profile_id, "Ich traf Anna.")
|
|
except PrivacyGatewayError as exc:
|
|
truncated = exc.code == "detect_truncated"
|
|
expect(exc.diagnostics.get("generate_called") is False, "truncated detect does not call generate")
|
|
reset_detect_test_hooks()
|
|
expect(truncated, "truncated detect output fails closed")
|
|
|
|
start = "Ich traf Anna.".index("Anna")
|
|
install_test_spans([{"start": start, "end": start + 4, "text": "Anna", "entity_type": "PERSON", "token": "PERSON:09"}])
|
|
extra = False
|
|
with patch("privacy_gateway.complete_model", boom):
|
|
try:
|
|
_run(profile_id, "Ich traf Anna.")
|
|
except PrivacyGatewayError as exc:
|
|
extra = exc.code == "detect_invalid_output"
|
|
reset_detect_test_hooks()
|
|
expect(extra, "detector-assigned tokens fail closed")
|
|
|
|
install_test_spans([{"start": 0, "end": 4, "text": "Anna", "entity_type": "PERSON"}])
|
|
try:
|
|
grounded_run = _run(profile_id, "Ich traf Anna am Abend.")
|
|
expect("Anna" not in (grounded_run.trace.get("egress") or ""), "grounded name is still masked")
|
|
expect(grounded_run.trace.get("generate_called") is True, "wrong offsets do not block generate")
|
|
finally:
|
|
reset_detect_test_hooks()
|
|
|
|
results = []
|
|
|
|
def worker(label: str):
|
|
reset_debug()
|
|
out = _run(profile_id, f"Ich traf {label}.")
|
|
results.append(
|
|
(
|
|
label,
|
|
tuple(out.trace.get("active_tokens") or []),
|
|
out.trace.get("egress") or "",
|
|
out.trace.get("intern") or "",
|
|
)
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
one = pool.submit(worker, "Anna")
|
|
two = pool.submit(worker, "Maren")
|
|
one.result()
|
|
two.result()
|
|
labels = {item[0] for item in results}
|
|
expect(labels == {"Anna", "Maren"}, "parallel requests keep their own manifests")
|
|
by_label = {item[0]: item for item in results}
|
|
expect("Maren" not in by_label["Anna"][2] and "Maren" not in by_label["Anna"][3], "Anna request does not carry Maren")
|
|
expect("Anna" not in by_label["Maren"][2] and "Anna" not in by_label["Maren"][3], "Maren request does not carry Anna")
|
|
expect(by_label["Anna"][1] and by_label["Maren"][1], "each parallel request has its own tokens")
|
|
|
|
from db import get_db
|
|
import uuid
|
|
|
|
legacy_id = str(uuid.uuid4())
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO identity_mappings
|
|
(id, profile_id, token, local_label, canonical_label, entity_type, status, origin)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(legacy_id, profile_id, "PERSON:55", "Wohnblock", "Wohnblock", "PERSON", STATUS_LEGACY, "legacy_auto"),
|
|
)
|
|
expect(
|
|
not any(item.get("id") == legacy_id for item in list_mappings(profile_id)),
|
|
"legacy mappings are not treated as confirmed",
|
|
)
|
|
legacy_run = _run(profile_id, "Der Wohnblock war ruhig.")
|
|
expect("Wohnblock" in (legacy_run.trace.get("egress") or ""), "legacy auto-mapping does not skip detection or force masking")
|
|
|
|
listed = client.get("/api/admin/identities", headers=headers)
|
|
expect(listed.status_code == 200, "admin identities list is authenticated")
|
|
expect("registry" in listed.json() and "proposals" in listed.json(), "admin list returns registry and proposals")
|
|
expect("local_label" not in json.dumps(first.diagnostics), "compact diagnostics omit labels")
|
|
expect("Anna" not in json.dumps(first.diagnostics), "compact diagnostics omit clear names")
|
|
expect("Anna" not in json.dumps(first.trace.get("log") or []), "run log omits labels")
|
|
|
|
provider_seen = {"n": 0}
|
|
|
|
def mark_provider(messages, policy):
|
|
provider_seen["n"] += 1
|
|
from privacy_gateway import ChatResult as _unused # noqa: F401
|
|
from providers import ChatResult
|
|
|
|
return ChatResult(content="[[PERSON:01]] war ruhig.", model="fake", usage={})
|
|
|
|
with patch("privacy_gateway.complete_model", mark_provider):
|
|
_run(profile_id, "Ich traf Anna.")
|
|
expect(provider_seen["n"] == 1, "successful detection still uses exactly one generate call")
|
|
|
|
from identity_store import coalesce_mappings, normalize_token
|
|
from privacy_gateway import _mask
|
|
from pronoun_bind import bind_user_lines, bind_pronouns
|
|
from privacy_gateway import mask_for_egress
|
|
|
|
expect(normalize_token("PERSON:KURZ") is None, "template token KURZ rejected")
|
|
first_map = remember_mapping(profile_id, "meine Frau Sushi", "PERSON:08")
|
|
second_map = remember_mapping(profile_id, "Sushi", "PERSON:KURZ")
|
|
expect(first_map["token"] == second_map["token"], "alias reuses token")
|
|
expect(second_map["canonical_label"] == "Sushi", "canonical label is the name")
|
|
extra_id = str(uuid.uuid4())
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO identity_mappings (id, profile_id, token, local_label, canonical_label, entity_type, status, origin)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(extra_id, profile_id, "PERSON:KURZ", "meine Frau Sushi", "meine Frau Sushi", "PERSON", "confirmed", "local_authoritative"),
|
|
)
|
|
coalesce_mappings(profile_id)
|
|
sushi_rows = [
|
|
item
|
|
for item in list_registry(profile_id)
|
|
if (item.get("canonical_label") or "").lower() == "sushi"
|
|
]
|
|
expect(len(sushi_rows) == 1, "coalesce keeps one Sushi row")
|
|
expect(sushi_rows[0]["token"].upper() != "PERSON:KURZ", "stable token wins over KURZ")
|
|
joined = _mask("meine Frau Sushi kam. Sushi stellte fest.", [{"local_label": "Sushi", "token": sushi_rows[0]["token"]}])
|
|
expect(joined.count(f"[[{sushi_rows[0]['token']}]]") == 2, "both mentions share one placeholder")
|
|
expect("meine Frau " in joined, "kinship stays visible")
|
|
homonym = _mask(
|
|
"Ich esse gerne Sushi mit meiner Frau Sushi.",
|
|
[{"local_label": "Sushi", "token": sushi_rows[0]["token"]}],
|
|
)
|
|
expect("esse gerne Sushi" in homonym, "food homonym stays")
|
|
expect(homonym.count("[[") == 1, "only the person mention is masked")
|
|
eating = _mask("Heute nur von Sushi essen erzählt.", [{"local_label": "Sushi", "token": sushi_rows[0]["token"]}])
|
|
expect("Sushi essen" in eating, "noun before eat stays food")
|
|
|
|
anna = [{"local_label": "Anna", "token": "PERSON:01"}]
|
|
bound = bind_user_lines(
|
|
_mask(
|
|
"user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.",
|
|
anna,
|
|
)
|
|
)
|
|
expect(bound.count("[[PERSON:01]]") == 3, "name and both pronouns share one token")
|
|
journal_masked = mask_for_egress(
|
|
"user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.",
|
|
anna,
|
|
"journal_generate",
|
|
)
|
|
expect(journal_masked.count("[[PERSON:01]]") == 1, "journal masks the name only")
|
|
dialogue_masked = mask_for_egress(
|
|
"user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.",
|
|
anna,
|
|
"dialogue_turn",
|
|
)
|
|
expect(dialogue_masked.count("[[PERSON:01]]") == 3, "dialogue still binds pronouns")
|
|
herr = [{"local_label": "Berg", "token": "PERSON:02"}]
|
|
male, _ = bind_pronouns(_mask("Herr Berg kam. Er setzte sich.", herr))
|
|
expect(male.count("[[PERSON:02]]") == 2, "masculine pronoun follows last man")
|
|
address, _ = bind_pronouns(_mask("Anna kam. Seid ihr schon da?", anna))
|
|
expect("ihr" in address, "address ihr is not a person")
|
|
plural, _ = bind_pronouns(_mask("Anna kam. Sie waren zu dritt.", anna))
|
|
expect("Sie waren" in plural or "sie waren" in plural.lower(), "plural sie stays")
|
|
|
|
expect(not is_maskable_label("ich"), "pronoun ich is not a mapping")
|
|
expect(not is_maskable_label("user"), "role user is not a mapping")
|
|
expect(not is_maskable_label("6:00 Uhr"), "clock time is not identity")
|
|
expect(is_maskable_label("Sushi"), "name-shaped label stays structurally maskable")
|
|
expect(is_maskable_label("Himmel"), "weather word is not banned by a stopword list")
|
|
expect(not is_maskable_label("Lia"), "detect gate still skips short labels")
|
|
expect(is_registry_maskable_label("Lia"), "confirmed registry may mask short names")
|
|
expect(not is_registry_maskable_label("ich"), "registry gate still skips pronouns")
|
|
|
|
from db import get_db as _get_db
|
|
|
|
with _get_db() as conn:
|
|
row = conn.execute(
|
|
"SELECT seed_revision, template, default_template FROM ai_prompts WHERE slug = ?",
|
|
("mvp.entity_detect",),
|
|
).fetchone()
|
|
expect(row["seed_revision"] == "2026-08-27-detect-ground-v1", "detect prompt revision is stored")
|
|
expect("{{known_labels}}" not in (row["template"] or ""), "new detect prompt has no known_labels skip list")
|
|
expect("Zwiebeln" not in (row["template"] or ""), "detect prompt has no food word list")
|
|
|
|
with _get_db() as conn:
|
|
conn.execute(
|
|
"UPDATE ai_prompts SET template = ? WHERE slug = ?",
|
|
("CUSTOM DETECT {{source_text}}", "mvp.entity_detect"),
|
|
)
|
|
from db import init_db
|
|
|
|
init_db()
|
|
with _get_db() as conn:
|
|
custom = conn.execute(
|
|
"SELECT template, seed_revision FROM ai_prompts WHERE slug = ?",
|
|
("mvp.entity_detect",),
|
|
).fetchone()
|
|
expect(custom["template"] == "CUSTOM DETECT {{source_text}}", "user-edited detect prompt is not overwritten")
|
|
expect(custom["seed_revision"] == "2026-08-27-detect-ground-v1", "revision still updates the default")
|
|
|
|
remote = ProviderConfig(
|
|
role="detect",
|
|
name="openrouter",
|
|
mode="http",
|
|
url="https://openrouter.ai/api/v1/chat/completions",
|
|
model="x",
|
|
key="sk-test",
|
|
local=False,
|
|
zdr=True,
|
|
no_train=True,
|
|
)
|
|
expect(uses_llm_detect(remote), "development allows remote detect")
|
|
os.environ["KANSHO_ENV"] = "production"
|
|
expect(not uses_llm_detect(remote), "production blocks remote detect")
|
|
local = ProviderConfig(
|
|
role="detect",
|
|
name="ollama",
|
|
mode="http",
|
|
url="http://127.0.0.1:11434/v1/chat/completions",
|
|
model="x",
|
|
key="",
|
|
local=True,
|
|
zdr=True,
|
|
no_train=True,
|
|
)
|
|
expect(uses_llm_detect(local), "production still allows local detect")
|
|
os.environ.pop("KANSHO_ENV", None)
|
|
|
|
mapped = [{"local_label": "Anna", "token": "PERSON:01"}]
|
|
anna_manifest = mask_prompt("Anna war am Markt.", mapped, "journal_generate")
|
|
kept = _validate_response("[[PERSON:01]] war ruhig.", anna_manifest)
|
|
expect(_demask(kept, anna_manifest) == "Anna war ruhig.", "demask after validation")
|
|
expect(
|
|
_demask("[[ person:01 ]] war ruhig.", anna_manifest) == "Anna war ruhig.",
|
|
"demask is case- and space-insensitive",
|
|
)
|
|
tokenized = _validate_response("Anna war ruhig.", anna_manifest)
|
|
expect(tokenized.startswith("[[PERSON:01]]"), "active cleartext is normalized, not discarded")
|
|
expect(_demask(tokenized, anna_manifest) == "Anna war ruhig.", "normalized plaintext demasks")
|
|
expect(
|
|
_visible_response("Anna war ruhig.", anna_manifest) == "Anna war ruhig.",
|
|
"visible output keeps model plaintext instead of the check roundtrip",
|
|
)
|
|
|
|
no_user = client.get("/api/admin/identities")
|
|
expect(no_user.status_code in {401, 403}, "identity registry requires auth")
|
|
admin_calls = {"n": 0}
|
|
|
|
def no_egress(*_args, **_kwargs):
|
|
admin_calls["n"] += 1
|
|
raise AssertionError("admin identity actions must not call detect")
|
|
|
|
with patch("entity_detect.complete_chat", no_egress), patch("privacy_gateway.complete_model", no_egress):
|
|
created = client.post(
|
|
"/api/admin/identities",
|
|
headers=headers,
|
|
json={"canonical_label": "Nordlicht", "entity_type": "PROJECT", "aliases": ["Nordlicht-Projekt"]},
|
|
)
|
|
expect(created.status_code == 200, f"admin create stays local {created.text}")
|
|
expect(admin_calls["n"] == 0, "admin registry actions create no external egress")
|
|
expect(created.json()["status"] == "confirmed", "admin create confirms locally")
|
|
|
|
before_confirm = {item.get("canonical_label") for item in list_confirmed_identities(profile_id)}
|
|
confirm_identity(profile_id, "Lia", aliases=["Mara"])
|
|
expect(is_registry_maskable_label("Lia"), "short confirmed name remains maskable")
|
|
egress_body = (
|
|
"WRITING_PROFILE\nLia schreibt knapp.\n"
|
|
"STYLE_EXAMPLES\nMara lachte am Markt.\n"
|
|
"CURRENT_DAY_SOURCES\nLia und Mara gingen später.\n"
|
|
"EXISTING_TEXT\nLia blieb."
|
|
)
|
|
covered = _run(profile_id, egress_body)
|
|
masked_all = covered.trace.get("egress") or ""
|
|
intern = covered.trace.get("intern") or ""
|
|
expect("Lia" in intern and "Mara" in intern, "clear names remain in the local template")
|
|
expect("Lia" not in masked_all, "confirmed short name is masked in every egress section")
|
|
expect("Mara" not in masked_all, "confirmed alias is masked in every egress section")
|
|
expect(
|
|
masked_all.count("[[") >= intern.count("Lia") + intern.count("Mara"),
|
|
"canonical and alias mentions are replaced",
|
|
)
|
|
expect("WRITING_PROFILE" in masked_all and "STYLE_EXAMPLES" in masked_all, "writing profile and style examples stay in egress")
|
|
expect("CURRENT_DAY_SOURCES" in masked_all and "EXISTING_TEXT" in masked_all, "sources and existing text stay in egress")
|
|
expect(covered.trace.get("confirmed_registry_applied") is True, "registry safety net is marked applied")
|
|
expect(covered.trace.get("semantic_identity_guaranteed") is False, "registry coverage is not semantic certainty")
|
|
expect(covered.trace.get("pre_egress_validation") == "ok", "confirmed plaintext is not left in egress")
|
|
|
|
group_text = "Die Kinder verstanden kein Wort. Herkunft: Nordland."
|
|
kinder_start = group_text.index("Kinder")
|
|
nord_start = group_text.index("Nordland")
|
|
install_test_spans(
|
|
[
|
|
{
|
|
"start": kinder_start,
|
|
"end": kinder_start + len("Kinder"),
|
|
"text": "Kinder",
|
|
"entity_type": "PERSON",
|
|
},
|
|
{
|
|
"start": nord_start,
|
|
"end": nord_start + len("Nordland"),
|
|
"text": "Nordland",
|
|
"entity_type": "PLACE",
|
|
},
|
|
]
|
|
)
|
|
try:
|
|
overmasked = _run(profile_id, group_text)
|
|
proposals = list_review_proposals(profile_id)
|
|
labels = {(item.get("observed_label") or item.get("canonical_label") or item.get("local_label") or "").casefold() for item in proposals}
|
|
expect("kinder" in labels or "nordland" in labels, "semantic overmask stays request-local as a proposal")
|
|
confirmed_now = {item.get("canonical_label") for item in list_confirmed_identities(profile_id)}
|
|
expect("Kinder" not in confirmed_now, "generic group is not persisted as an identity")
|
|
expect("Nordland" not in confirmed_now, "origin designation is not persisted as an identity")
|
|
expect(confirmed_now == before_confirm | {"Lia"}, "only the confirmed registry row is durable")
|
|
expect(overmasked.trace.get("full_detection_coverage") is True, "injected spans still mark technical coverage")
|
|
expect(overmasked.trace.get("semantic_identity_guaranteed") is False, "overmasking is not claimed as certainty")
|
|
finally:
|
|
reset_detect_test_hooks()
|
|
|
|
turn = client.post(
|
|
f"/api/journal/conversations/{conv.json()['id']}/turn",
|
|
headers=headers,
|
|
json={"body": "Ich war mit Anna in Hamburg und danach allein."},
|
|
)
|
|
expect(turn.status_code == 200, f"turn {turn.status_code}")
|
|
expect((turn.json().get("trace") or {}).get("generate_called") is True, "dialogue generate still runs once after detect")
|
|
|
|
install_test_detect_failure(DetectError("detect_invalid_output", "Detect-text entspricht nicht dem lokalen Substring."))
|
|
try:
|
|
held = client.post(
|
|
f"/api/journal/conversations/{conv.json()['id']}/turn",
|
|
headers=headers,
|
|
json={"body": "Heute nur der Markt."},
|
|
)
|
|
expect(held.status_code == 200, f"detect fallback turn {held.status_code} {held.text}")
|
|
expect((held.json().get("assistant") or {}).get("body"), "detect failure still leaves a local impulse")
|
|
expect((held.json().get("decision") or {}).get("guard") == "detect_blocked", "detect fallback is marked locally")
|
|
finally:
|
|
reset_detect_test_hooks()
|
|
|
|
print("All privacy detect tests passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|