374 lines
17 KiB
Python
374 lines
17 KiB
Python
"""Style-application variants control which writing-profile parts reach generate."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
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_editorial import (
|
|
compile_journal_style_application,
|
|
style_application_trace,
|
|
)
|
|
from journal_generation_policy import (
|
|
clone_guideline,
|
|
compile_selection,
|
|
default_selection_ids,
|
|
get_guideline,
|
|
save_selection,
|
|
seed_generation_instructions,
|
|
)
|
|
from journal_generate import pack_narration_context
|
|
from main import app
|
|
from privacy_gateway import install_test_recorder, reset_debug
|
|
from prompt_budget import JournalBudget
|
|
from writing_profile_store import (
|
|
compile_task_brief,
|
|
import_text,
|
|
set_lifecycle,
|
|
update_facet,
|
|
upsert_trait,
|
|
)
|
|
|
|
|
|
CORE_MARK = "ZXCORE_MARKER kurze trockene Saetze"
|
|
FACET_MARK = "ZXFACET_MARKER ruhige Abendreflexion"
|
|
TRAIT_MARK = "ZXTRAIT_MARKER keine rhythmischen Aufzaehlungen"
|
|
EXAMPLE_MARK = "ZXEXAMPLE_MARKER Der Hafen blieb hinter der Faehre"
|
|
TODAY_MARK = "ZXTODAY_MARKER am heutigen Markt"
|
|
|
|
|
|
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 style_region(intern: str) -> str:
|
|
return (intern or "").split("\nCURRENT_DAY_SOURCES\n")[0]
|
|
|
|
|
|
def has_heading(intern: str, title: str) -> bool:
|
|
body = intern or ""
|
|
return f"\n{title}\n" in body or body.startswith(f"{title}\n")
|
|
|
|
|
|
def seed_id(slot: str, key: str) -> str:
|
|
from journal_generation_policy import load_seed_document
|
|
|
|
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 selection_of(voice_key: str) -> dict[str, str]:
|
|
ids = default_selection_ids()
|
|
ids["voice_id"] = seed_id("voice", voice_key)
|
|
return ids
|
|
|
|
|
|
def ensure_local_profile(profile_id: str = "style-context-local") -> str:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT OR IGNORE INTO profiles (id, email, name, password_hash, role)
|
|
VALUES (?, ?, ?, ?, 'user')
|
|
""",
|
|
(profile_id, f"{profile_id}@example.test", "Style", "x"),
|
|
)
|
|
return profile_id
|
|
|
|
|
|
def install_profile(profile_id: str) -> None:
|
|
update_facet(profile_id, "core", value=CORE_MARK)
|
|
update_facet(profile_id, "autobiographical_journal", value=FACET_MARK)
|
|
upsert_trait(
|
|
profile_id,
|
|
slug="journal-rhythm",
|
|
facet_key="autobiographical_journal",
|
|
label="Rhythmus",
|
|
statement=TRAIT_MARK,
|
|
origin="manual",
|
|
)
|
|
set_lifecycle(profile_id, "confirmed")
|
|
import_text(
|
|
profile_id,
|
|
EXAMPLE_MARK + " in knappen ruhigen Saetzen ohne Pathoswolken.",
|
|
occurred_at="2026-07-01",
|
|
)
|
|
|
|
|
|
def test_compile_flags() -> None:
|
|
init_db()
|
|
profile_id = ensure_local_profile()
|
|
install_profile(profile_id)
|
|
full = compile_task_brief(profile_id)
|
|
expect(CORE_MARK in full and FACET_MARK in full and TRAIT_MARK in full, "full brief keeps selected parts")
|
|
core_only = compile_task_brief(
|
|
profile_id,
|
|
include_core=True,
|
|
include_facet=False,
|
|
include_traits=False,
|
|
)
|
|
expect(CORE_MARK in core_only, "core-only brief keeps core")
|
|
expect(FACET_MARK not in core_only, "core-only brief omits facet")
|
|
expect(TRAIT_MARK not in core_only, "core-only brief omits traits")
|
|
none = compile_task_brief(
|
|
profile_id,
|
|
include_core=False,
|
|
include_facet=False,
|
|
include_traits=False,
|
|
)
|
|
expect(none == "", "no selected profile parts yield an empty brief")
|
|
bundle = compile_journal_style_application(profile_id, compile_selection(selection_of("neutral")).style_context)
|
|
expect(bundle["profile_block"] == "", "profile-free omits the wrapped profile block")
|
|
expect(bundle["examples_block"] == "", "profile-free omits the wrapped examples block")
|
|
|
|
|
|
def test_budget_omission_in_effective_trace() -> None:
|
|
init_db()
|
|
prompt = {
|
|
"id": "style-pack",
|
|
"slug": "mvp.journal_generate",
|
|
"prompt_type": "base",
|
|
"template": "{{writing_profile}}{{style_examples}}DAY\n{{reconstruction}}\nOLD\n{{existing_text}}\n",
|
|
}
|
|
assembled = {
|
|
"writing_profile": "WRITING_PROFILE\nCore: kurze Sätze.\n",
|
|
"reconstruction": "Heute Markt.",
|
|
"style_examples": "",
|
|
"existing_text": "",
|
|
}
|
|
huge = "Stil " + ("Beispielwort " * 400)
|
|
from engine import preview_prompt
|
|
from prompt_budget import estimate_tokens
|
|
|
|
rendered = preview_prompt(prompt, assembled)["rendered"]
|
|
budget = JournalBudget(
|
|
model="fake",
|
|
purpose="journal_generate",
|
|
effective_context_window=32_768,
|
|
reserved_output_tokens=256,
|
|
safety_margin=0.15,
|
|
available_input_tokens=estimate_tokens(rendered) + 80,
|
|
chars_per_token=2.0,
|
|
)
|
|
packed, dropped = pack_narration_context(
|
|
prompt,
|
|
budget,
|
|
assembled,
|
|
style_examples=huge,
|
|
existing_text="Bestehende Fassung",
|
|
include_existing=True,
|
|
)
|
|
expect("style_examples" in dropped, "budget drops style examples first")
|
|
expect("Beispielwort" not in (packed.get("style_examples") or ""), "dropped examples leave the packed prompt")
|
|
compiled = compile_selection(selection_of("with_examples"))
|
|
bundle = {
|
|
"requested": dict(compiled.style_context),
|
|
"brief": "Core: kurze Sätze.",
|
|
"profile_block": assembled["writing_profile"],
|
|
"examples_block": huge,
|
|
"example_rows": [{"kind": "imported_text", "excerpt": huge}],
|
|
"omitted": [],
|
|
"effective": {"neutral_fallback": False, "trait_count": 0},
|
|
}
|
|
trace = style_application_trace(compiled, bundle, dropped=dropped)
|
|
expect(trace["effective"]["style_example_count"] == 0, "effective trace has no examples after budget drop")
|
|
expect(
|
|
any(item.get("part") == "style_examples" and item.get("reason") == "budget" for item in trace["omitted"]),
|
|
"budget drop is recorded on the effective trace",
|
|
)
|
|
expect(trace["id"] == seed_id("voice", "with_examples"), "trace keeps the requested voice id")
|
|
expect(trace["revision"] == 1, "trace keeps the requested revision")
|
|
|
|
|
|
def main() -> None:
|
|
test_compile_flags()
|
|
test_budget_omission_in_effective_trace()
|
|
|
|
from tests.harness import reset_postgres_data
|
|
|
|
reset_postgres_data()
|
|
reset_debug()
|
|
with TestClient(app) as client:
|
|
setup = client.post(
|
|
"/api/auth/setup",
|
|
json={"email": "voice@example.test", "name": "Ada", "password": "test-pass"},
|
|
)
|
|
headers = header(setup.json()["token"])
|
|
profile_id = setup.json()["profile_id"]
|
|
install_profile(profile_id)
|
|
|
|
space = client.post("/api/journal/spaces", headers=headers, json={"title": "Stil"})
|
|
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": f"Ich war am Markt. {TODAY_MARK}. 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"]
|
|
|
|
def generate(voice_key: str) -> dict:
|
|
reset_debug()
|
|
recorder = install_test_recorder()
|
|
result = client.post(
|
|
f"/api/journal/days/{day_id}/generate",
|
|
headers=headers,
|
|
json={
|
|
"conversation_ids": [conv_id],
|
|
"generation_selection": selection_of(voice_key),
|
|
"remember_generation_selection": False,
|
|
},
|
|
)
|
|
expect(result.status_code == 200, f"{voice_key} generate {result.status_code}")
|
|
payload = result.json()
|
|
intern = intern_of(payload)
|
|
style = style_region(intern)
|
|
app_trace = (payload.get("trace") or {}).get("style_application") or {}
|
|
calls = [item.get("purpose") for item in recorder]
|
|
expect(calls == ["journal_generate"], f"{voice_key} still uses exactly one narration call")
|
|
expect(app_trace.get("id") == seed_id("voice", voice_key), f"{voice_key} id reaches the trace")
|
|
expect(app_trace.get("key") == voice_key, f"{voice_key} key reaches the trace")
|
|
expect(app_trace.get("revision") == 1, f"{voice_key} revision reaches the trace")
|
|
expect(
|
|
(payload.get("trace") or {}).get("generation_selection", {}).get("ids", {}).get("voice")
|
|
== seed_id("voice", voice_key),
|
|
f"{voice_key} selection id matches prompt/trace",
|
|
)
|
|
expect(TODAY_MARK not in style, f"{voice_key} does not use today as a style example")
|
|
expect(TODAY_MARK in intern.split("\nCURRENT_DAY_SOURCES\n")[-1], f"{voice_key} keeps today as content")
|
|
expect("keine neuen Tatsachen" in intern, f"{voice_key} keeps fact fidelity")
|
|
expect("[[PERSON:01]]" in intern, f"{voice_key} keeps privacy placeholders")
|
|
return {"payload": payload, "intern": intern, "style": style, "trace": app_trace}
|
|
|
|
free = generate("neutral")
|
|
expect(not has_heading(free["intern"], "WRITING_PROFILE"), "profile-free omits the profile heading")
|
|
expect(not has_heading(free["intern"], "STYLE_EXAMPLES"), "profile-free omits the examples heading")
|
|
expect("STYLE_EXAMPLES" not in free["intern"], "profile-free has no orphaned STYLE_EXAMPLES reference")
|
|
expect(CORE_MARK not in free["intern"], "profile-free sends no core")
|
|
expect(FACET_MARK not in free["intern"], "profile-free sends no facet")
|
|
expect(TRAIT_MARK not in free["intern"], "profile-free sends no traits")
|
|
expect(EXAMPLE_MARK not in free["intern"], "profile-free sends no style examples")
|
|
expect(free["trace"]["effective"]["include_core"] is False, "profile-free effective core is no")
|
|
expect(free["trace"]["effective"]["style_example_count"] == 0, "profile-free effective examples are zero")
|
|
|
|
light = generate("light")
|
|
expect(CORE_MARK in light["style"], "light sends core")
|
|
expect(FACET_MARK not in light["intern"], "light does not send facet even indirectly")
|
|
expect(TRAIT_MARK not in light["intern"], "light does not send traits even indirectly")
|
|
expect(EXAMPLE_MARK not in light["intern"], "light does not send examples even indirectly")
|
|
expect(has_heading(light["intern"], "WRITING_PROFILE"), "light keeps a profile heading")
|
|
expect(not has_heading(light["intern"], "STYLE_EXAMPLES"), "light omits the examples heading")
|
|
expect("STYLE_EXAMPLES" not in light["intern"], "light has no orphaned STYLE_EXAMPLES reference")
|
|
expect(light["trace"]["effective"]["include_core"] is True, "light effective core is yes")
|
|
expect(light["trace"]["effective"]["include_facet"] is False, "light effective facet is no")
|
|
expect(light["trace"]["effective"]["trait_count"] == 0, "light effective traits are zero")
|
|
|
|
clear = generate("clear")
|
|
expect(CORE_MARK in clear["style"], "clear sends core")
|
|
expect(FACET_MARK in clear["style"], "clear sends the relevant facet")
|
|
expect(TRAIT_MARK in clear["style"], "clear sends traits")
|
|
expect(EXAMPLE_MARK not in clear["intern"], "clear does not send examples")
|
|
expect(not has_heading(clear["intern"], "STYLE_EXAMPLES"), "clear omits the examples heading")
|
|
expect("STYLE_EXAMPLES" not in clear["intern"], "clear has no orphaned STYLE_EXAMPLES reference")
|
|
expect(clear["trace"]["effective"]["include_core"] is True, "clear effective core is yes")
|
|
expect(clear["trace"]["effective"]["include_facet"] is True, "clear effective facet is yes")
|
|
expect(clear["trace"]["effective"]["trait_count"] >= 1, "clear effective traits are counted")
|
|
expect(clear["trace"]["effective"]["style_example_count"] == 0, "clear effective examples are zero")
|
|
|
|
examples = generate("with_examples")
|
|
expect(CORE_MARK in examples["style"], "with-examples sends core")
|
|
expect(FACET_MARK in examples["style"], "with-examples sends facet")
|
|
expect(TRAIT_MARK in examples["style"], "with-examples sends traits")
|
|
expect(EXAMPLE_MARK in examples["style"], "with-examples sends style examples")
|
|
expect(has_heading(examples["intern"], "STYLE_EXAMPLES"), "with-examples labels the examples block")
|
|
expect("ausschließlich als Stilreferenz" in examples["intern"] or "sprachliche" in examples["style"], "examples stay style-only")
|
|
expect("keine Tatsachen des heutigen" in examples["style"] or "nicht übernommen" in examples["style"], "example facts are forbidden")
|
|
expect(examples["trace"]["effective"]["style_example_count"] >= 1, "with-examples effective count is used")
|
|
expect(EXAMPLE_MARK not in (examples["payload"].get("body") or ""), "example facts are not copied into today's draft")
|
|
|
|
settings = client.get("/api/journal/generation-settings", headers=headers).json()
|
|
voice_options = settings["options"]["voice"]
|
|
free_opt = next(item for item in voice_options if item["id"] == seed_id("voice", "neutral"))
|
|
expect(free_opt["style_context"]["include_core"] is False, "generation-settings expose style context without prompt text")
|
|
expect("instruction" not in free_opt, "generation-settings still hide prompt wording")
|
|
|
|
save_selection(profile_id, selection_of("neutral"))
|
|
remembered = generate("noticeable")
|
|
expect(CORE_MARK in remembered["style"], "legacy noticeable remains selectable")
|
|
profile_run = client.post(
|
|
f"/api/journal/days/{day_id}/generate",
|
|
headers=headers,
|
|
json={"conversation_ids": [conv_id]},
|
|
)
|
|
expect(profile_run.status_code == 200, f"stored selection generate {profile_run.status_code}")
|
|
stored_intern = intern_of(profile_run.json())
|
|
expect(not has_heading(stored_intern, "WRITING_PROFILE"), "stored profile-free selection stays profile-free")
|
|
expect(CORE_MARK not in stored_intern, "stored profile-free selection still omits core")
|
|
|
|
cloned = clone_guideline(seed_id("voice", "clear"))
|
|
original_instruction = get_guideline(seed_id("voice", "clear"), include_instruction=True)["instruction"]
|
|
with get_db() as conn:
|
|
seed_generation_instructions(conn)
|
|
expect(
|
|
get_guideline(seed_id("voice", "clear"), include_instruction=True)["instruction"] == original_instruction,
|
|
"seed refresh does not replace the published original with a clone",
|
|
)
|
|
expect(
|
|
get_guideline(cloned["id"], include_instruction=True)["style_context"]["include_core"] is True,
|
|
"cloned voice keeps the structured style context",
|
|
)
|
|
|
|
def compile_one(key: str) -> str:
|
|
bundle = compile_journal_style_application(profile_id, compile_selection(selection_of(key)).style_context)
|
|
return (bundle.get("brief") or "") + "\n" + (bundle.get("style_examples") or "")
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
first = pool.submit(compile_one, "neutral")
|
|
second = pool.submit(compile_one, "with_examples")
|
|
free_text = first.result()
|
|
example_text = second.result()
|
|
expect(CORE_MARK not in free_text, "parallel profile-free compile stays empty")
|
|
expect(CORE_MARK in example_text and EXAMPLE_MARK in example_text, "parallel with-examples compile keeps all parts")
|
|
expect(EXAMPLE_MARK not in free_text, "parallel requests do not mix style configurations")
|
|
|
|
print("journal style context tests passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|