Kansho/backend/tests/test_journal_style_legacy_immutable.py

"""Legacy comparison voices and immutable seed IDs after style-context control."""
from __future__ import annotations

import copy
import os
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

os.environ["KANSHO_DB_PATH"] = str(Path(tempfile.gettempdir()) / "kansho-journal-legacy-immutable-test.sqlite")
os.environ["KANSHO_FAKE_PROVIDER"] = "1"
os.environ["KANSHO_FAKE_DETECT"] = "1"
Path(os.environ["KANSHO_DB_PATH"]).unlink(missing_ok=True)

from fastapi.testclient import TestClient
from db import get_db, init_db
from journal_generation_policy import (
    PURPOSE_JOURNAL,
    archive_guideline,
    clone_guideline,
    compile_selection,
    default_selection_ids,
    get_guideline,
    list_guidelines,
    load_seed_document,
    load_selection,
    publish_guideline,
    save_selection,
    seed_generation_instructions,
    update_guideline,
)
from journal_editorial import format_style_examples
from main import app
from privacy_gateway import install_test_recorder, reset_debug


# Reconstructed from git HEAD `backend/config/generation_instructions.seed.json`
# (commit 4f37991, seed_revision 2026-08-27-generation-guidelines-v1).
HEAD_VOICE = {
    "neutral": {
        "id": "journal-generate-voice-neutral",
        "instruction": "Schreibe in einem neutralen Journalstil. Das Writing Profile höchstens als leise Tendenz.",
        "label": "neutral",
        "summary": "Neutraler Journalstil, Writing Profile nur als leise Tendenz.",
    },
    "light": {
        "id": "journal-generate-voice-light",
        "instruction": "Nimm Rhythmus und Wortwahl des Writing Profiles zurückhaltend auf.",
        "label": "dezent",
        "summary": "Rhythmus und Wortwahl des Writing Profiles zurückhaltend.",
    },
    "noticeable": {
        "id": "journal-generate-voice-noticeable",
        "instruction": "Wende das Writing Profile spürbar an, ohne es in den Vordergrund zu stellen.",
        "label": "spürbar",
        "summary": "Writing Profile spürbar, ohne in den Vordergrund zu treten.",
    },
    "clear": {
        "id": "journal-generate-voice-clear",
        "instruction": "Wende Rhythmus, Wortwahl und Reflexionsdichte des Writing Profiles deutlich an, ohne den Text künstlich zu literarisieren.",
        "label": "deutlich",
        "summary": "Rhythmus, Wortwahl und Reflexionsdichte des Writing Profiles deutlich.",
    },
}
FULL_STYLE_CONTEXT = {
    "include_core": True,
    "include_facet": True,
    "include_traits": True,
    "include_style_examples": True,
}
CURRENT_VOICE_IDS = {
    "neutral": "journal-generate-voice-neutral",
    "light": "journal-generate-voice-light",
    "noticeable": "journal-generate-voice-noticeable",
    "clear": "journal-generate-voice-clear",
    "with_examples": "journal-generate-voice-with-examples",
}
LEGACY_KEYS = ("legacy_neutral", "legacy_light", "legacy_noticeable", "legacy_clear")


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 intern_of(payload: dict) -> str:
    for stage in (payload.get("trace") or {}).get("stages") or []:
        if stage.get("purpose") == "journal_generate":
            return stage.get("intern") or ""
    return (payload.get("trace") or {}).get("intern") or ""


def seed_id(slot: str, key: str) -> str:
    seed = load_seed_document()
    for item in (seed.get("slots") or {}).get(slot, {}).get("variants") or []:
        if item.get("guideline_key") == key:
            return item["id"]
    raise SystemExit(f"FAIL: missing seed {slot}/{key}")


def voice_rows() -> list[dict]:
    return list_guidelines(PURPOSE_JOURNAL, slot="voice", include_instruction=True)


def snapshot_current_voices() -> dict:
    return {
        item["id"]: {
            "instruction": item["instruction"],
            "label": item["label"],
            "summary": item["summary"],
            "style_context": dict(item["style_context"]),
            "is_default": item["is_default"],
            "revision": item["revision"],
            "status": item["status"],
        }
        for item in voice_rows()
        if item["id"] in CURRENT_VOICE_IDS.values()
    }


def test_current_variants_and_legacy_seed() -> None:
    init_db()
    before = snapshot_current_voices()
    expect(CURRENT_VOICE_IDS["clear"] in before, "current default voice id remains")
    expect(before[CURRENT_VOICE_IDS["clear"]]["is_default"] is True, "current default stays on Persönliche Stimme deutlich")
    expect(
        before[CURRENT_VOICE_IDS["neutral"]]["style_context"]["include_core"] is False,
        "current profile-free voice keeps an empty style context",
    )
    expect(
        default_selection_ids()["voice_id"] == CURRENT_VOICE_IDS["clear"],
        "catalog default still points at the current clear id",
    )

    with get_db() as conn:
        seed_generation_instructions(conn)
        seed_generation_instructions(conn)
    after = snapshot_current_voices()
    expect(before == after, "reseeding does not change current voice ids or semantics")

    legacy = [item for item in voice_rows() if item["guideline_key"] in LEGACY_KEYS]
    expect(len(legacy) == 4, "legacy comparison voices are inserted once")
    expect(len({item["id"] for item in legacy}) == 4, "legacy voices have distinct ids")
    expect(all(not item["is_default"] for item in legacy), "legacy voices are not default")
    expect(all(item["status"] == "active" for item in legacy), "legacy voices are selectable")
    expect(
        all(item["style_context"] == FULL_STYLE_CONTEXT for item in legacy),
        "legacy voices reconstruct the former full style context",
    )
    by_key = {item["guideline_key"]: item for item in legacy}
    for old_key, spec in HEAD_VOICE.items():
        row = by_key[f"legacy_{old_key}"]
        expect(row["id"] != spec["id"], f"legacy {old_key} uses a new stable id")
        expect(row["instruction"] == spec["instruction"], f"legacy {old_key} keeps the reconstructed instruction")
        expect("Legacy – vor Stilkontextsteuerung" in row["label"], f"legacy {old_key} is labeled as pre-style-context")
        expect(row["id"].startswith("journal-generate-voice-legacy-"), f"legacy {old_key} id is namespaced")


def test_stored_selection_and_clones_and_archives() -> None:
    init_db()
    profile_id = "legacy-immutable-user"
    with get_db() as conn:
        conn.execute(
            """
            INSERT OR IGNORE INTO profiles (id, email, name, password_hash, role)
            VALUES (?, ?, ?, ?, 'user')
            """,
            (profile_id, "legacy@example.test", "Legacy", "x"),
        )
    stored = default_selection_ids()
    stored["voice_id"] = CURRENT_VOICE_IDS["neutral"]
    save_selection(profile_id, stored)
    cloned = clone_guideline(CURRENT_VOICE_IDS["clear"])
    original_clone_instruction = "CLONE MUST STAY"
    update_guideline(
        cloned["id"],
        {
            "guideline_key": "clone_clear",
            "label": "Klon",
            "summary": "Admin-Klon",
            "instruction": original_clone_instruction,
            "style_context": {
                "include_core": True,
                "include_facet": False,
                "include_traits": False,
                "include_style_examples": False,
            },
        },
    )
    published_clone = publish_guideline(clone_guideline(CURRENT_VOICE_IDS["light"])["id"])
    archived = archive_guideline(published_clone["id"])
    archived_instruction = get_guideline(archived["id"], include_instruction=True)["instruction"]
    archived_status = archived["status"]

    with get_db() as conn:
        seed_generation_instructions(conn)
    expect(load_selection(profile_id)["voice_id"] == CURRENT_VOICE_IDS["neutral"], "stored selection is not rewritten")
    expect(
        get_guideline(cloned["id"], include_instruction=True)["instruction"] == original_clone_instruction,
        "admin clone is not overwritten by seed",
    )
    expect(get_guideline(archived["id"], include_instruction=True)["status"] == archived_status, "archived variant stays archived")
    expect(
        get_guideline(archived["id"], include_instruction=True)["instruction"] == archived_instruction,
        "archived variant is not rewritten",
    )
    expect(
        get_guideline(CURRENT_VOICE_IDS["clear"], include_instruction=True)["is_default"] is True,
        "current default is not reset by a legacy seed pass",
    )


def test_later_semantic_seed_creates_successor() -> None:
    init_db()
    profile_id = "legacy-successor-user"
    with get_db() as conn:
        conn.execute(
            """
            INSERT OR IGNORE INTO profiles (id, email, name, password_hash, role)
            VALUES (?, ?, ?, ?, 'user')
            """,
            (profile_id, "succ@example.test", "Succ", "x"),
        )
    stored = default_selection_ids()
    save_selection(profile_id, stored)
    original = get_guideline(CURRENT_VOICE_IDS["clear"], include_instruction=True)
    mutated = copy.deepcopy(load_seed_document())
    for item in mutated["slots"]["voice"]["variants"]:
        if item.get("id") == CURRENT_VOICE_IDS["clear"]:
            item["instruction"] = "CHANGED FUTURE SEMANTICS FOR CLEAR"
            item["label"] = "Persönliche Stimme deutlich (neu)"
            break
    mutated["seed_revision"] = "2026-08-30-future-voice-v1"
    with get_db() as conn:
        seed_generation_instructions(conn, mutated)
        seed_generation_instructions(conn, mutated)
    kept = get_guideline(CURRENT_VOICE_IDS["clear"], include_instruction=True)
    expect(kept["instruction"] == original["instruction"], "future seed does not overwrite the existing variant id")
    expect(kept["label"] == original["label"], "future seed does not change the existing label")
    successors = [
        item
        for item in voice_rows()
        if item.get("cloned_from") == CURRENT_VOICE_IDS["clear"]
        and item["instruction"] == "CHANGED FUTURE SEMANTICS FOR CLEAR"
    ]
    expect(len(successors) == 1, "changed semantics are inserted once as a successor")
    successor = successors[0]
    expect(successor["id"] != CURRENT_VOICE_IDS["clear"], "successor uses a new id")
    expect(successor["revision"] == int(original["revision"]) + 1, "successor raises the revision")
    expect(successor["cloned_from"] == CURRENT_VOICE_IDS["clear"], "successor records the predecessor")
    expect(successor["is_default"] is True, "new default may point at the successor")
    expect(kept["is_default"] is False, "previous default is released when the seed names a new default")
    expect(
        load_selection(profile_id)["voice_id"] == CURRENT_VOICE_IDS["clear"],
        "stored selection keeps the previous id after a semantic successor is added",
    )
    compiled = compile_selection({**default_selection_ids(), "voice_id": CURRENT_VOICE_IDS["clear"]})
    expect(compiled.ids["voice"] == CURRENT_VOICE_IDS["clear"], "previous id remains selectable with its old meaning")
    expect(compiled.cloned_from["voice"] == "", "original current variant has no predecessor")
    compiled_new = compile_selection({**default_selection_ids(), "voice_id": successor["id"]})
    expect(compiled_new.cloned_from["voice"] == CURRENT_VOICE_IDS["clear"], "compiled policy exposes the predecessor")


def test_legacy_generate_trace_and_prompt() -> None:
    Path(os.environ["KANSHO_DB_PATH"]).unlink(missing_ok=True)
    init_db()
    client = TestClient(app)
    setup = client.post(
        "/api/auth/setup",
        json={"email": "legacy-gen@example.test", "password": "pass-pass", "name": "Legacy"},
    )
    expect(setup.status_code == 200, f"setup {setup.status_code}")
    token = setup.json()["token"]
    headers = header(token)
    space = client.post("/api/journal/spaces", headers=headers, json={"title": "Urlaub"})
    expect(space.status_code == 200, f"space {space.status_code}")
    day = client.post(
        f"/api/journal/spaces/{space.json()['id']}/days",
        headers=headers,
        json={"calendar_date": "2026-08-29"},
    )
    conv = client.post(
        f"/api/journal/days/{day.json()['day']['id']}/conversations",
        headers=headers,
        json={"title": "Tag"},
    )
    turn = client.post(
        f"/api/journal/conversations/{conv.json()['id']}/turn",
        headers=headers,
        json={"body": "Ich war am Markt. Vielleicht bleibe ich kürzer."},
    )
    expect(turn.status_code == 200, f"turn {turn.status_code}")
    day_id = day.json()["day"]["id"]
    conv_id = conv.json()["id"]
    selection = default_selection_ids()

    reset_debug()
    recorder = install_test_recorder()
    current = client.post(
        f"/api/journal/days/{day_id}/generate",
        headers=headers,
        json={"conversation_ids": [conv_id], "generation_selection": selection, "remember_generation_selection": False},
    )
    expect(current.status_code == 200, f"current generate {current.status_code}")
    current_payload = current.json()
    current_intern = intern_of(current_payload)
    current_trace = (current_payload.get("trace") or {}).get("style_application") or {}
    expect([item.get("purpose") for item in recorder] == ["journal_generate"], "current path still uses one narration call")
    expect(current_trace.get("id") == CURRENT_VOICE_IDS["clear"], "trace keeps the current voice id")
    expect(current_trace.get("revision") == 1, "trace keeps the current revision")
    expect("STYLE_EXAMPLES" not in current_intern, "current default prompt has no STYLE_EXAMPLES reference")
    expect("\nSTYLE_EXAMPLES\n" not in current_intern, "current default prompt has no example block")

    formatted = format_style_examples(
        [{"kind": "journal_entry", "occurred_at": "2026-08-01", "excerpt": "Nur ein Stilbeispiel."}]
    )
    expect("keine Tatsachen des heutigen" in formatted or "nicht übernommen" in formatted, "examples stay style-only when present")

    reset_debug()
    recorder = install_test_recorder()
    selection["voice_id"] = seed_id("voice", "legacy_neutral")
    legacy = client.post(
        f"/api/journal/days/{day_id}/generate",
        headers=headers,
        json={"conversation_ids": [conv_id], "generation_selection": selection, "remember_generation_selection": False},
    )
    expect(legacy.status_code == 200, f"legacy generate {legacy.status_code}")
    payload = legacy.json()
    intern = intern_of(payload)
    app_trace = (payload.get("trace") or {}).get("style_application") or {}
    expect([item.get("purpose") for item in recorder] == ["journal_generate"], "legacy path still uses one narration call")
    expect(app_trace.get("id") == seed_id("voice", "legacy_neutral"), "legacy trace uses the reconstructed id")
    expect(app_trace.get("revision") == 1, "legacy trace uses the reconstructed revision")
    expect(app_trace.get("key") == "legacy_neutral", "legacy trace names the reconstructed key")
    expect("höchstens als leise Tendenz" in intern, "legacy instruction reaches the provider prompt")
    expect(app_trace.get("requested", {}).get("include_style_examples") is True, "legacy requests the former full style context")


def main() -> None:
    test_current_variants_and_legacy_seed()
    test_stored_selection_and_clones_and_archives()
    test_later_semantic_seed_creates_successor()
    test_legacy_generate_trace_and_prompt()
    print("journal style legacy immutable tests passed.")


if __name__ == "__main__":
    main()