Kansho/backend/tests/test_detect_contract_retry.py

"""Bounded detect-contract retry, acyclic diagnostics, persist must not hide EngineError.

Run from backend/: python tests/test_detect_contract_retry.py
Fake detect/provider only. No live calls.
"""
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))

os.environ["KANSHO_FAKE_PROVIDER"] = "1"
os.environ["KANSHO_FAKE_DETECT"] = "1"

from fastapi.testclient import TestClient
from debug_store import persist_engine_error, sanitize
from engine import EngineError
from entity_detect import (
    DetectError,
    detect_personal_egress,
    install_test_detect_failure,
    install_test_detect_script,
    reset_detect_test_hooks,
    split_detect_chunks,
    validate_detected_entity,
)
from journal_generate import _compose_journal_trace, _stage_trace
from main import app
from privacy_gateway import GatewayRequest, PrivacyGatewayError, complete, reset_debug
from providers import ChatResult


USAGE_A = {"prompt_tokens": 10, "completion_tokens": 3, "total_tokens": 13, "cost": 0.001}
USAGE_B = {"prompt_tokens": 8, "completion_tokens": 2, "total_tokens": 10, "cost": 0.0005}


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 _ids(obj, stack=None) -> bool:
    """True if obj is acyclic."""
    if not isinstance(obj, (dict, list, tuple)):
        return True
    stack = stack or set()
    ident = id(obj)
    if ident in stack:
        return False
    stack.add(ident)
    try:
        items = obj.values() if isinstance(obj, dict) else obj
        return all(_ids(item, stack) for item in items)
    finally:
        stack.discard(ident)


def _budget_is_not_diagnostics(payload: dict) -> bool:
    diag = payload.get("diagnostics") if isinstance(payload.get("diagnostics"), dict) else payload
    trace = diag.get("trace") if isinstance(diag.get("trace"), dict) else {}
    budget = trace.get("budget")
    if budget is diag:
        return False
    for stage in trace.get("stages") or []:
        if isinstance(stage, dict) and stage.get("budget") is diag:
            return False
    return True


def _run(profile_id: str, rendered: str, purpose: str = "dialogue_turn"):
    return complete(
        GatewayRequest(
            prompt_id="detect-retry-test",
            purpose=purpose,
            data_class="B",
            profile_id=profile_id,
            payload={"rendered": rendered, "source_text": rendered},
        )
    )


def _chat(entities: list[dict], usage: dict) -> ChatResult:
    return ChatResult(
        content=json.dumps({"entities": entities}),
        model="fake-detect",
        usage=usage,
        finish_reason="stop",
    )


def _invalid_type():
    return [{"start": 0, "end": 1, "text": "x", "entity_type": "HUMAN"}]


def _extra_field():
    return [{"start": 0, "end": 1, "text": "x", "entity_type": "PERSON", "score": 0.9}]


def _missing_field():
    return [{"start": 0, "text": "x", "entity_type": "PERSON"}]


def _bad_offset(chunk_text: str):
    return [{"start": 0, "end": 5, "text": "ZZZXQ", "entity_type": "PERSON"}]


def open_space_day(client, headers, title="Alltag", date="2026-08-29"):
    space = client.post("/api/journal/spaces", headers=headers, json={"title": title})
    expect(space.status_code == 200, f"create space {space.status_code}")
    day = client.post(
        f"/api/journal/spaces/{space.json()['id']}/days",
        headers=headers,
        json={"calendar_date": date},
    )
    expect(day.status_code == 200, f"open day {day.status_code}")
    return space.json(), day.json()


def main() -> None:
    reset_debug()
    reset_detect_test_hooks()

    sample = "Ich traf Anna."
    try:
        validate_detected_entity(
            {"start": 0, "end": 4, "text": "Anna", "entity_type": "HUMAN"},
            sample,
            0,
        )
        typed = False
    except DetectError as exc:
        typed = exc.code == "detect_invalid_output"
        expect((exc.diagnostics or {}).get("invalid_entity_type") == "HUMAN", "unknown type is recorded without the span text")
        expect((exc.diagnostics or {}).get("contract_violation") == "unknown_entity_type", "unknown type has a violation category")
        expect("Anna" not in json.dumps(exc.diagnostics or {}), "span plaintext is not stored on the type violation")
    expect(typed, "unknown entity_type is a contract error")

    cyclic = {}
    nested = {"budget": cyclic, "stages": [cyclic]}
    cyclic["trace"] = nested
    cyclic["local_label"] = "secret-name"
    cleaned = sanitize(cyclic)
    expect(cleaned is not None, "cyclic sanitize returns a value")
    expect(_ids(cleaned), "sanitized cyclic input is acyclic")
    expect("local_label" not in json.dumps(cleaned), "sanitize still drops mapping labels")
    expect(json.dumps(cleaned, default=str), "sanitized cyclic input is JSON-serializable")

    with TestClient(app) as client:
        setup = client.post(
            "/api/auth/setup",
            json={"email": "lars@example.test", "name": "Lars", "password": "test-pass"},
        )
        expect(setup.status_code == 200, "setup")
        token = setup.json()["token"]
        headers = header(token)
        profile_id = setup.json()["profile_id"]

        calls = []

        def retry_then_ok(attempt, chunk_index, chunk_text, offset, schema_retry=False):
            usage = USAGE_A if attempt == 1 else USAGE_B
            calls.append((attempt, chunk_index, schema_retry))
            if attempt == 1:
                return _chat(_invalid_type(), usage)
            return _chat([], usage)

        long_body = (
            "WRITING_PROFILE\nkurze trockene Saetze. " * 12
            + "STYLE_EXAMPLES\nDer Abend blieb ruhig. " * 12
            + "CURRENT_DAY_SOURCES\nHeute der Markt, danach der Hafen. "
        )
        with patch("entity_detect.DETECT_CHUNK_CHARS", 90), patch("entity_detect.DETECT_CHUNK_OVERLAP", 12):
            chunks = split_detect_chunks(long_body, chunk_chars=90, overlap=12)
            expect(len(chunks) >= 2, "long profile+examples text is chunked")
            install_test_detect_script(retry_then_ok)
            try:
                outcome = detect_personal_egress(profile_id, long_body)
            finally:
                reset_detect_test_hooks()
        attempts = {item[0] for item in calls}
        expect(attempts == {1, 2}, "invalid first pass starts a second full pass")
        first_chunks = [item[1] for item in calls if item[0] == 1]
        second_chunks = [item[1] for item in calls if item[0] == 2]
        expect(second_chunks == list(range(len(chunks))), "second pass re-checks every chunk")
        expect(first_chunks == [0], "failed first pass stops at the invalid chunk and discards partials")
        expect(any(item[2] for item in calls if item[0] == 2), "second pass is marked as schema retry")
        expect(not any(item[2] for item in calls if item[0] == 1), "first pass has no schema retry hint")
        expect(outcome.stats.detect_partial_discarded is True, "partial first-pass spans are discarded")
        expect(outcome.stats.detect_passes == 2, "successful retry records two passes")
        expect(outcome.stats.detect_calls == len(calls), "detect calls sum both passes")
        expect(outcome.stats.cost_known is True, "injected usage keeps known detect cost")
        expected_cost = USAGE_A["cost"] * len(first_chunks) + USAGE_B["cost"] * len(second_chunks)
        expect(abs(outcome.stats.cost - expected_cost) < 1e-9, "detect costs of both passes are aggregated")
        expect(
            abs(
                outcome.stats.prompt_tokens
                - (USAGE_A["prompt_tokens"] * len(first_chunks) + USAGE_B["prompt_tokens"] * len(second_chunks))
            )
            < 1e-9,
            "detect tokens of both passes are aggregated",
        )
        expect(outcome.stats.full_detection_coverage is True, "valid second pass has full coverage")
        expect(outcome.stats.detect_ms >= 0, "retry records duration")

        generate_n = {"n": 0}

        def count_generate(messages, policy):
            generate_n["n"] += 1
            from privacy_gateway import _fake_complete

            return ChatResult(content=_fake_complete(policy.get("purpose") or "", messages[0].get("content") or ""), model="fake")

        install_test_detect_script(retry_then_ok)
        generate_n["n"] = 0
        with patch("privacy_gateway.complete_model", count_generate):
            try:
                ran = _run(profile_id, "Heute der Markt und danach der Hafen.")
            finally:
                reset_detect_test_hooks()
        expect(ran.trace.get("generate_called") is True, "valid retry allows exactly one narration call")
        expect(generate_n["n"] == 1, "narration runs once after a recovered detect pass")
        expect(_ids(ran.diagnostics or {}), "successful diagnostics are acyclic")
        expect(_ids(ran.trace or {}), "successful trace is acyclic")

        def always_invalid(attempt, chunk_index, chunk_text, offset, schema_retry=False):
            return _chat(_invalid_type(), USAGE_A)

        generate_n["n"] = 0
        install_test_detect_script(always_invalid)
        blocked = None
        with patch("privacy_gateway.complete_model", count_generate):
            try:
                _run(profile_id, "Heute der Markt und danach der Hafen.")
            except PrivacyGatewayError as exc:
                blocked = exc
            finally:
                reset_detect_test_hooks()
        expect(blocked is not None and blocked.code == "detect_invalid_output", "two invalid passes fail closed")
        expect(generate_n["n"] == 0, "invalid detect never calls narration")
        expect(blocked.diagnostics.get("generate_called") is False, "final abort keeps generate_called false")
        expect(blocked.diagnostics.get("detect_passes") == 2, "both invalid passes are counted")
        expect(blocked.diagnostics.get("contract_violation") == "unknown_entity_type", "final abort keeps the violation category")
        expect(blocked.diagnostics.get("invalid_entity_type") == "HUMAN", "delivered type is stored without span text")
        expect(blocked.diagnostics.get("detect_partial_discarded") is True, "failed retry discarded the first pass")
        expect(_ids(blocked.diagnostics), "failed diagnostics are acyclic")
        expect("HUMAN" in json.dumps(blocked.diagnostics), "invalid type remains visible")
        expect("secret" not in json.dumps(blocked.diagnostics).lower(), "no extra identity payload")
        expect(blocked.status_code != 500, "detect abort is not a generic 500")

        def extra_then_fail(attempt, chunk_index, chunk_text, offset, schema_retry=False):
            return _chat(_extra_field(), USAGE_A)

        install_test_detect_script(extra_then_fail)
        extra_err = None
        try:
            detect_personal_egress(profile_id, "Heute der Markt.")
        except DetectError as exc:
            extra_err = exc
        finally:
            reset_detect_test_hooks()
        expect(extra_err is not None and extra_err.diagnostics.get("contract_violation") == "extra_fields", "extra fields fail closed after retry")

        def missing_then_fail(attempt, chunk_index, chunk_text, offset, schema_retry=False):
            return _chat(_missing_field(), USAGE_A)

        install_test_detect_script(missing_then_fail)
        missing_err = None
        try:
            detect_personal_egress(profile_id, "Heute der Markt.")
        except DetectError as exc:
            missing_err = exc
        finally:
            reset_detect_test_hooks()
        expect(missing_err is not None and missing_err.diagnostics.get("contract_violation") == "missing_fields", "missing fields fail closed after retry")

        def offset_then_fail(attempt, chunk_index, chunk_text, offset, schema_retry=False):
            return _chat(_bad_offset(chunk_text), USAGE_A)

        install_test_detect_script(offset_then_fail)
        offset_err = None
        try:
            detect_personal_egress(profile_id, "Heute der Markt.")
        except DetectError as exc:
            offset_err = exc
        finally:
            reset_detect_test_hooks()
        expect(offset_err is not None and offset_err.diagnostics.get("contract_violation") == "unusable_offsets", "unusable offsets fail closed after retry")

        net_calls = []

        def network_fail(attempt, chunk_index, chunk_text, offset, schema_retry=False):
            net_calls.append(attempt)
            raise DetectError("detect_chunk_failed", "chunk failed")

        install_test_detect_script(network_fail)
        net_err = None
        try:
            detect_personal_egress(profile_id, "Heute der Markt.")
        except DetectError as exc:
            net_err = exc
        finally:
            reset_detect_test_hooks()
        expect(net_err is not None and net_err.code == "detect_chunk_failed", "provider/chunk failure stays distinct")
        expect(net_calls == [1], "network-like chunk failure is not doubled by the contract retry")

        install_test_detect_failure(DetectError("detect_chunk_failed", "chunk failed"))
        immediate = None
        try:
            detect_personal_egress(profile_id, "Heute der Markt.")
        except DetectError as exc:
            immediate = exc
        finally:
            reset_detect_test_hooks()
        expect(immediate is not None and immediate.diagnostics.get("detect_passes") == 0, "injected provider failure does not start a contract retry")

        mixed = []

        def worker(kind: str):
            reset_debug()
            seen = []

            def script(attempt, chunk_index, chunk_text, offset, schema_retry=False):
                seen.append((kind, attempt))
                if attempt == 1:
                    entity_type = "HUMAN" if kind == "alpha" else "ANIMAL"
                    return _chat([{"start": 0, "end": 1, "text": "x", "entity_type": entity_type}], USAGE_A)
                return _chat([], USAGE_B)

            install_test_detect_script(script)
            try:
                out = detect_personal_egress(profile_id, f"Heute der Markt {kind}.")
                mixed.append((kind, tuple(seen), out.stats.invalid_entity_type, out.stats.detect_calls, id(out.stats)))
            finally:
                reset_detect_test_hooks()

        with ThreadPoolExecutor(max_workers=2) as pool:
            one = pool.submit(worker, "alpha")
            two = pool.submit(worker, "beta")
            one.result()
            two.result()
        kinds = {item[0] for item in mixed}
        expect(kinds == {"alpha", "beta"}, "parallel detect retries both complete")
        by_kind = {item[0]: item for item in mixed}
        expect(
            all(kind == row[0] for row in by_kind.values() for kind, _attempt in row[1]),
            "parallel scripts stay on their request",
        )
        expect(by_kind["alpha"][2] in (None, "HUMAN"), "alpha does not keep beta's type")
        expect(by_kind["beta"][2] in (None, "ANIMAL"), "beta does not keep alpha's type")
        expect(by_kind["alpha"][4] != by_kind["beta"][4], "parallel stats objects are distinct")

        failed = {
            "diagnostics": {"generate_called": False, "detect_cost": 0.001, "cost": None, "purpose": "journal_generate"},
            "trace": {"purpose": "journal_generate", "generate_called": False},
            "content": "",
        }
        staged = _stage_trace(failed, "journal_generate")
        expect(staged.get("budget") is not failed["diagnostics"], "stage budget is a snapshot, not the live diagnostics object")
        failed["diagnostics"]["trace"] = {"budget": staged.get("budget")}
        expect(_ids(failed), "attaching the snapshot does not create a diagnostics cycle")
        composed = _compose_journal_trace(
            {"trace": {"purpose": "local_source_artifact"}, "diagnostics": {"prompt_tokens": 0}},
            failed,
            run_log=[{"kind": "detect", "status": "error"}],
            narrate_prompt={"slug": "mvp.journal_generate", "seed_revision": "test"},
            profile_meta={"present": False},
            style_meta={"count": 0},
            dropped=[],
            extra={"generation_selection": {"voice_id": "journal-generate-voice-clear"}, "style_application": {"include_core": False}},
        )
        failed["diagnostics"]["trace"] = composed
        expect(_ids(failed["diagnostics"]), "composed journal diagnostics stay acyclic")
        expect(_budget_is_not_diagnostics(failed), "composed budget is not a back-reference")
        expect(json.dumps(sanitize(failed), default=str), "composed failure is serializable")

        space, day = open_space_day(client, headers)
        day_id = day["day"]["id"]
        conv = client.post(
            f"/api/journal/days/{day_id}/conversations",
            headers=headers,
            json={"title": "Gespräch"},
        )
        turn = client.post(
            f"/api/journal/conversations/{conv.json()['id']}/turn",
            headers=headers,
            json={"body": "Heute war der Markt voll und danach der Hafen ruhig."},
        )
        expect(turn.status_code == 200, f"turn {turn.status_code}")

        enabled = client.put("/api/admin/debug", headers=headers, json={"persist_enabled": True})
        expect(enabled.status_code == 200 and enabled.json()["persist_enabled"] is True, "persist enabled")
        before = client.get("/api/admin/debug/runs", headers=headers).json()["runs"]

        install_test_detect_script(always_invalid)
        try:
            gen = client.post(
                f"/api/journal/days/{day_id}/generate",
                headers=headers,
                json={"conversation_ids": [conv.json()["id"]]},
            )
        finally:
            reset_detect_test_hooks()
        expect(gen.status_code == 503, f"journal detect abort status {gen.status_code}")
        detail = gen.json()["detail"]
        expect(detail["code"] == "detect_invalid_output", "API keeps the detect contract code")
        expect(detail["code"] != "error" and gen.status_code != 500, "detect abort is not HTTP 500")
        expect("Angaben" in (detail.get("message") or ""), "API uses the user-facing detect message")
        expect(detail.get("cost_report"), "API includes a cost report")
        expect(detail["cost_report"].get("generate_called") is False, "cost report says generate was not called")
        expect(
            (detail.get("diagnostics") or {}).get("generate_called") is False,
            "diagnostics keep generate_called false",
        )
        expect(_ids(detail), "API error payload is acyclic")
        expect(_budget_is_not_diagnostics(detail), "API error budget is not a diagnostics alias")
        after = client.get("/api/admin/debug/runs", headers=headers).json()["runs"]
        error_runs = [
            item
            for item in after
            if item.get("purpose") == "journal_generate" and item.get("status") == "error"
        ]
        expect(len(error_runs) == 1, "failed journal request is stored exactly once")
        run = error_runs[0]
        expect(run.get("journal_day_id") == day_id, "error run is attached to the journal day")
        detail_run = client.get(f"/api/admin/debug/runs/{run['id']}", headers=headers)
        expect(detail_run.status_code == 200, f"error run detail {detail_run.status_code}")
        payload = detail_run.json().get("payload") or {}
        expect(payload.get("code") == "detect_invalid_output", "stored run keeps the detect code")
        expect(payload.get("generate_called") is False, "stored run records generate_called false")
        expect(payload.get("detect_passes") == 2, "stored run records both detect passes")
        expect(payload.get("detect_partial_discarded") is True, "stored run records discarded partials")
        expect(payload.get("contract_violation") == "unknown_entity_type", "stored run records the violation")
        expect(payload.get("invalid_entity_type") == "HUMAN", "stored run records the delivered type")
        expect(payload.get("generation_selection") or (payload.get("trace") or {}).get("generation_selection"), "stored run keeps generation selection")
        expect("local_label" not in json.dumps(payload), "stored run has no mapping labels")
        exported = client.get(
            f"/api/admin/debug/export?format=json&journal_day_id={day_id}&purpose=journal_generate",
            headers=headers,
        )
        expect(exported.status_code == 200, f"error run export {exported.status_code}")
        document = exported.json()
        encoded = json.dumps(document)
        expect(encoded, "exported error run is JSON-serializable")
        expect("detect_invalid_output" in encoded, "export contains the detect code")

        hidden = EngineError("detect_invalid_output", "Persönliche Angaben konnten nicht sicher zugeordnet werden.", 503)
        with patch("debug_store.persist_step", side_effect=RuntimeError("persist boom")):
            persist_engine_error(profile_id, purpose="journal_generate", exc=hidden, journal_day_id=day_id)
        install_test_detect_script(always_invalid)
        try:
            with patch("debug_store.persist_step", side_effect=RuntimeError("persist boom")):
                gen_hidden = client.post(
                    f"/api/journal/days/{day_id}/generate",
                    headers=headers,
                    json={"conversation_ids": [conv.json()["id"]]},
                )
        finally:
            reset_detect_test_hooks()
        expect(gen_hidden.status_code == 503, f"persist failure keeps detect status {gen_hidden.status_code}")
        expect(gen_hidden.json()["detail"]["code"] == "detect_invalid_output", "persist failure does not replace EngineError")
        expect(gen_hidden.status_code != 500, "persist failure is not HTTP 500")

        reset_detect_test_hooks()
        control_space, control_day = open_space_day(client, headers, "Kontrolle", "2026-08-28")
        control_conv = client.post(
            f"/api/journal/days/{control_day['day']['id']}/conversations",
            headers=headers,
            json={"title": "Gespräch"},
        )
        control_turn = client.post(
            f"/api/journal/conversations/{control_conv.json()['id']}/turn",
            headers=headers,
            json={"body": "Heute Tee auf dem Balkon, danach nur der Wind."},
        )
        expect(control_turn.status_code == 200, "control turn")
        control = client.post(
            f"/api/journal/days/{control_day['day']['id']}/generate",
            headers=headers,
            json={"conversation_ids": [control_conv.json()["id"]]},
        )
        expect(control.status_code == 200, f"control generate without profile {control.status_code}")
        expect((control.json().get("trace") or {}).get("generate_called") is True, "control run still narrates once")
        expect(_ids(control.json().get("trace") or {}), "control trace is acyclic")

        client.put("/api/admin/debug", headers=headers, json={"persist_enabled": False})

    print("ALL TESTS PASSED")


if __name__ == "__main__":
    main()