"""Response integrity, span masking, journal rejection, and diagnosis contract.""" from __future__ import annotations import os import sys import tempfile from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] REPO = ROOT.parent sys.path.insert(0, str(ROOT)) os.environ["KANSHO_FAKE_PROVIDER"] = "1" os.environ["KANSHO_FAKE_DETECT"] = "1" from fastapi.testclient import TestClient from identity_store import list_mappings, remember_mapping from journal_generate import JOURNAL_NOT_ACCEPTED, JOURNAL_NOT_ACCEPTED_MESSAGE from journal_store import current_draft from main import app from privacy_gateway import ( ERROR_EGRESS_VALIDATION, ActiveReplacement, GatewayRequest, MaskingManifest, PrivacyGatewayError, complete, mask_prompt, reset_debug, validate_pre_egress, ) from providers import ChatResult def expect(ok: bool, message: str) -> None: if not ok: raise SystemExit(f"FAIL: {message}") print(f"OK {message}") def header(token: str) -> dict: return {"X-Auth-Token": token} def _run_gateway(profile_id: str, rendered: str, *, purpose: str = "dialogue_turn"): return complete( GatewayRequest( prompt_id="response-integrity", purpose=purpose, data_class="B", profile_id=profile_id, payload={"rendered": rendered, "source_text": rendered, "prompt_slug": "mvp.journal_generate"}, ) ) def _day_with_turn(client: TestClient, headers: dict, body: str, title: str = "Tag") -> tuple[str, str]: space = client.post("/api/journal/spaces", headers=headers, json={"title": title}) day = client.post( f"/api/journal/spaces/{space.json()['id']}/days", headers=headers, json={"calendar_date": "2026-08-27"}, ) conv = client.post( f"/api/journal/days/{day.json()['day']['id']}/conversations", headers=headers, json={"title": "Gespräch"}, ) turn = client.post( f"/api/journal/conversations/{conv.json()['id']}/turn", headers=headers, json={"body": body}, ) if turn.status_code != 200: raise SystemExit(f"FAIL: setup turn {turn.status_code}") print("OK setup turn 200") return day.json()["day"]["id"], conv.json()["id"] def _generate(client: TestClient, headers: dict, day_id: str, conv_id: str, reply: str, usage=None): calls = {"n": 0} def fake(_messages, _policy): calls["n"] += 1 return ChatResult( content=reply, model="fake-gpt", usage=usage or {"prompt_tokens": 7, "completion_tokens": 5, "total_tokens": 12, "cost": 0.001}, context_compression="disabled", ) with patch("privacy_gateway.complete_model", fake): response = client.post( f"/api/journal/days/{day_id}/generate", headers=headers, json={"conversation_ids": [conv_id]}, ) return response, calls["n"] def test_span_masking_and_pre_egress() -> None: text = "Ich aß Sushi. Sushi kam später." second = text.find("Sushi", text.find("Sushi") + 1) manifest = mask_prompt( text, [ { "local_label": "Sushi", "token": "PERSON:01", "entity_type": "PERSON", "start": second, "end": second + 5, "text": "Sushi", "source": "request_local", } ], "journal_generate", ) expect(manifest.masked_text == "Ich aß Sushi. [[PERSON:01]] kam später.", "only the detected span is masked") validate_pre_egress(manifest.masked_text, manifest) overlapping = mask_prompt( "Anna-Lena kam vorbei.", [ { "local_label": "Anna", "token": "PERSON:01", "entity_type": "PERSON", "start": 0, "end": 4, "text": "Anna", "source": "request_local", }, { "local_label": "Anna-Lena", "token": "PERSON:08", "entity_type": "PERSON", "start": 0, "end": 9, "text": "Anna-Lena", "source": "request_local", }, ], "journal_generate", ) expect(overlapping.masked_text.startswith("[[PERSON:08]]"), "overlapping spans pick the longest") leaky = MaskingManifest( masked_text="Anna war am Markt.", available_mapping_count=1, replacements=(ActiveReplacement("PERSON:01", "PERSON", 1, "Anna"),), ) blocked = False try: validate_pre_egress("Anna war am Markt.", leaky) except PrivacyGatewayError as exc: blocked = exc.code == ERROR_EGRESS_VALIDATION expect(blocked, "privacy stays fail-closed before the provider") def test_runtime_has_no_incident_wordlists() -> None: forbidden = ( "Kinder", "Delfine", "delphine", "Wohnzimmer", "Balkon", "Tagesreflektion", "Lebenwesen", "Baguette", ) runtime = [] for path in ROOT.glob("*.py"): if path.name.startswith("entity_detect_eval") or path.name.startswith("_"): continue text = path.read_text(encoding="utf-8") for word in forbidden: if word in text: runtime.append(f"{path.name}:{word}") expect(not runtime, f"incident words must not be hardcoded in runtime: {runtime}") def test_frontend_degraded_copy() -> None: page = (REPO / "frontend" / "src" / "pages" / "JournalDayPage.jsx").read_text(encoding="utf-8") trace = (REPO / "frontend" / "src" / "components" / "CallTrace.jsx").read_text(encoding="utf-8") log = (REPO / "frontend" / "src" / "components" / "RunLogPopup.jsx").read_text(encoding="utf-8") expect("Generierung nicht übernommen" in page, "journal page names a rejected generate") expect("Generierung nicht übernommen" in trace, "trace names a rejected generate") expect("Generierung nicht übernommen" in log, "run log names a rejected generate") expect("nur user-Zeilen" not in trace, "masking copy is not limited to user lines") expect("zwei Modellaufrufe nacheinander" not in log, "running copy does not claim two model calls") def main() -> None: reset_debug() test_span_masking_and_pre_egress() test_runtime_has_no_incident_wordlists() test_frontend_degraded_copy() with TestClient(app) as client: setup = client.post( "/api/auth/setup", json={"email": "response@example.test", "name": "Response", "password": "test-pass"}, ) headers = header(setup.json()["token"]) profile_id = setup.json()["profile_id"] calls = {"n": 0} def false_positive_cleartext(_messages, _policy): calls["n"] += 1 return ChatResult( content="Danach Sushi essen am Markt.", model="fake-gpt", usage={"prompt_tokens": 8, "completion_tokens": 6, "total_tokens": 14, "cost": 0.002}, context_compression="not_applicable", ) remember_mapping(profile_id, "Sushi", "PERSON:01") with patch("privacy_gateway.complete_model", false_positive_cleartext): result = _run_gateway(profile_id, "Ich aß Sushi. Meine Frau Sushi kam später.") expect(calls["n"] == 1, "false-positive cleartext still uses exactly one generate call") expect("Sushi" in (result.content or ""), "reconstructed common noun is kept after local normalization") expect(result.diagnostics.get("generate_calls") == 1, "gateway reports one generate call") expect( result.diagnostics.get("response_normalization") in {"active_cleartext_normalized", "none"}, "normalization is recorded without a retry", ) remember_mapping(profile_id, "Anna", "PERSON:02") day_id, conv_id = _day_with_turn(client, headers, "Heute war ich mit Anna am Markt.") gen, n = _generate( client, headers, day_id, conv_id, "Ein Markttag\n\nAnna stand den ganzen Nachmittag am Markt.", ) expect(gen.status_code == 200, f"attested person cleartext {gen.status_code}") expect(n == 1, "attested person cleartext does not retry") expect("Anna" in (gen.json().get("body") or ""), "attested person is demasked") expect((gen.json().get("trace") or {}).get("model_text_accepted") is True, "attested model text is accepted") remember_mapping(profile_id, "Clarissa", "PERSON:99") hist_day, hist_conv = _day_with_turn(client, headers, "Heute nur Markt und Kirschen.", title="Historisch") blocked, n = _generate( client, headers, hist_day, hist_conv, "Ein Markttag\n\nClarissa kaufte Kirschen am Markt.", ) expect(blocked.status_code == 409, f"style-only person {blocked.status_code}") detail = blocked.json().get("detail") or {} expect(detail.get("code") == JOURNAL_NOT_ACCEPTED, "historical-only name is rejected") expect(detail.get("message") == JOURNAL_NOT_ACCEPTED_MESSAGE, "API message is explicit") expect(n == 1, "provenance reject does not retry") expect(current_draft(profile_id, hist_day) is None, "rejected generate inserts no draft") diag = detail.get("diagnostics") or {} trace = diag.get("trace") or {} expect(trace.get("abort_reason") == "unattested_identity", "provenance abort reason remains") expect(trace.get("model_text_accepted") is False, "model text is marked not accepted") expect( trace.get("model") == "fake-gpt" or (trace.get("budget") or {}).get("model") == "fake-gpt", "error trace keeps the model", ) expect( (trace.get("generate_calls") == 1) or ((trace.get("budget") or {}).get("generate_calls") == 1) or diag.get("generate_calls") == 1, "error trace keeps generate_calls", ) expect( (trace.get("budget") or {}).get("completion_tokens") == 5 or diag.get("completion_tokens") == 5, "error trace keeps completion tokens", ) unknown_day, unknown_conv = _day_with_turn(client, headers, "Heute nur der Markt.", title="Platzhalter") unknown, n = _generate( client, headers, unknown_day, unknown_conv, "Ein Markttag\n\n[[PERSON:99]] stand am Markt.", ) expect(unknown.status_code == 409, f"unknown placeholder {unknown.status_code}") expect((unknown.json().get("detail") or {}).get("code") == JOURNAL_NOT_ACCEPTED, "unknown placeholder is rejected") expect(n == 1, "unknown placeholder does not retry") expect(current_draft(profile_id, unknown_day) is None, "unknown placeholder inserts no draft") generic_day, generic_conv = _day_with_turn(client, headers, "Heute nur der Hafen.", title="Generisch") generic, n = _generate( client, headers, generic_day, generic_conv, "Ein Hafentag\n\n[[...]] blieb am Hafen.", ) expect(generic.status_code == 409, f"generic placeholder {generic.status_code}") expect((generic.json().get("detail") or {}).get("code") == JOURNAL_NOT_ACCEPTED, "generic placeholder is rejected") expect(current_draft(profile_id, generic_day) is None, "generic placeholder inserts no draft") ok_day, ok_conv = _day_with_turn(client, headers, "Heute war ich mit Anna am Hafen.", title="Tokens") anna_token = next( ( item.get("token") for item in list_mappings(profile_id) if (item.get("canonical_label") or item.get("local_label") or "") == "Anna" ), "PERSON:02", ) ok_gen, n = _generate( client, headers, ok_day, ok_conv, f"Ein Hafentag\n\n[[{anna_token}]] stand am Hafen.", ) expect(ok_gen.status_code == 200, f"active tokens {ok_gen.status_code}") expect("Anna" in (ok_gen.json().get("body") or ""), "active tokens demask normally") expect("[[" not in (ok_gen.json().get("body") or ""), "no leftover placeholder after demask") expect(n == 1, "normal demask uses one generate call") provider_calls = {"n": 0} def boom(_messages, _policy): provider_calls["n"] += 1 raise AssertionError("provider must not be called after pre-egress failure") def leaky_mask(rendered, mappings, purpose): return MaskingManifest( masked_text=rendered, available_mapping_count=len(mappings or []), replacements=(ActiveReplacement("PERSON:02", "PERSON", 1, "Anna"),), ) with patch("privacy_gateway.mask_prompt", leaky_mask), patch("privacy_gateway.complete_model", boom): failed = False try: _run_gateway(profile_id, "Anna war am Markt.") except PrivacyGatewayError as exc: failed = exc.code == ERROR_EGRESS_VALIDATION expect(failed, "pre-egress still fail-closed") expect(provider_calls["n"] == 0, "provider is not called when confirmed identity remains") print("All privacy response integrity tests passed.") if __name__ == "__main__": main()