356 lines
15 KiB
Python
356 lines
15 KiB
Python
"""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_FAKE_PROVIDER"] = "1"
|
||
os.environ["KANSHO_FAKE_DETECT"] = "1"
|
||
|
||
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:
|
||
from tests.harness import reset_postgres_data
|
||
|
||
reset_postgres_data()
|
||
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()
|