"""Named, versioned journal generation guidelines. Not model temperature. Four independent dimensions are selected by ID. Instruction text lives in generation_guidelines, seeded from JSON. This module selects, validates and composes. It does not own prompt wording and has no numeric 0–100 path. """ from __future__ import annotations import json import re import uuid from dataclasses import dataclass from pathlib import Path from db import get_db, row_to_dict from journal_policy import PolicyError from placeholders import CONTEXT_PATTERN PURPOSE_JOURNAL = "journal_generate" USER_SLOTS = ("transformation", "detail", "voice", "narrative") ALLOWED_SLOTS = USER_SLOTS LEGACY_SOURCE_MODE_SLOT = "source_mode" SELECTION_KEYS = ( "transformation_id", "detail_id", "voice_id", "narrative_id", ) SLOT_TO_ID_KEY = { "transformation": "transformation_id", "detail": "detail_id", "voice": "voice_id", "narrative": "narrative_id", } ID_KEY_TO_SLOT = {value: key for key, value in SLOT_TO_ID_KEY.items()} SLOT_INSTRUCTION_KEYS = { "transformation": "transformation_instructions", "detail": "detail_instructions", "voice": "voice_instructions", "narrative": "narrative_instructions", } REQUIRED_JOURNAL_PLACEHOLDERS = ( "transformation_instructions", "detail_instructions", "voice_instructions", "narrative_instructions", "writing_profile", "style_examples", "reconstruction", "existing_text", ) RETIRED_JOURNAL_PLACEHOLDERS = ( "source_mode_instructions", "editorial_mode", "editorial_instructions", ) STATUS_DRAFT = "draft" STATUS_ACTIVE = "active" STATUS_ARCHIVED = "archived" STATUSES = (STATUS_DRAFT, STATUS_ACTIVE, STATUS_ARCHIVED) MAX_INSTRUCTION_CHARS = 1000 MAX_LABEL_CHARS = 80 MAX_SUMMARY_CHARS = 160 KEY_RE = re.compile(r"^[a-z][a-z0-9_]{0,40}$") SEED_PATH = Path(__file__).resolve().parent / "config" / "generation_instructions.seed.json" STYLE_CONTEXT_KEYS = ( "include_core", "include_facet", "include_traits", "include_style_examples", ) EMPTY_STYLE_CONTEXT = {key: False for key in STYLE_CONTEXT_KEYS} FULL_STYLE_CONTEXT = {key: True for key in STYLE_CONTEXT_KEYS} LEGACY_VOICE_STYLE_CONTEXT = { "neutral": dict(EMPTY_STYLE_CONTEXT), "light": {**EMPTY_STYLE_CONTEXT, "include_core": True}, "noticeable": { "include_core": True, "include_facet": True, "include_traits": True, "include_style_examples": False, }, "clear": { "include_core": True, "include_facet": True, "include_traits": True, "include_style_examples": False, }, "with_examples": dict(FULL_STYLE_CONTEXT), "legacy_neutral": dict(FULL_STYLE_CONTEXT), "legacy_light": dict(FULL_STYLE_CONTEXT), "legacy_noticeable": dict(FULL_STYLE_CONTEXT), "legacy_clear": dict(FULL_STYLE_CONTEXT), } LEGACY_COMPARISON_KEYS = frozenset( { "legacy_neutral", "legacy_light", "legacy_noticeable", "legacy_clear", } ) STYLE_EXAMPLES_STANDING_SENTENCE = ( "STYLE_EXAMPLES dienen ausschließlich als Stilreferenz; ihre Inhalte gehören nicht zum heutigen Tag. " ) PUBLIC_FIELDS = ( "id", "purpose", "slot", "guideline_key", "label", "summary", "sort_order", "status", "revision", "cloned_from", "is_default", "is_system_seed", "seed_revision", "used_at", "created", "updated", ) class GenerationPolicyError(PolicyError): def __init__( self, message: str, *, code: str = "invalid_generation_selection", status_code: int = 400, ): super().__init__(code, message, status_code) class CatalogError(GenerationPolicyError): def __init__(self, message: str, *, code: str = "generation_policy_invalid", status_code: int = 409): super().__init__(message, code=code, status_code=status_code) @dataclass(frozen=True) class CompiledPolicy: ids: dict[str, str] keys: dict[str, str] labels: dict[str, str] summaries: dict[str, str] revisions: dict[str, int] instructions: dict[str, str] seed_revision: str style_context: dict[str, bool] cloned_from: dict[str, str] def load_seed_document() -> dict: return json.loads(SEED_PATH.read_text(encoding="utf-8")) def _as_bool(raw) -> bool: return bool(int(raw)) if not isinstance(raw, bool) else raw def default_style_context_for_key(guideline_key: str, *, slot: str) -> dict[str, bool]: if slot != "voice": return {} mapped = LEGACY_VOICE_STYLE_CONTEXT.get((guideline_key or "").strip()) if mapped: return dict(mapped) return dict(FULL_STYLE_CONTEXT) def normalize_style_context(raw, *, slot: str, required: bool = False) -> dict[str, bool]: if slot != "voice": if raw in (None, "", {}, []): return {} raise CatalogError( "Stilkontext gehört nur zur Dimension Persönliche Stimme.", code="invalid_generation_guideline", status_code=400, ) if raw in (None, "", {}): if required: raise CatalogError( "Die Stilanwendung braucht eine Stilkontext-Konfiguration.", code="invalid_generation_guideline", status_code=400, ) return dict(EMPTY_STYLE_CONTEXT) if not isinstance(raw, dict): raise CatalogError( "style_context muss ein Objekt mit den vier Stilquellen sein.", code="invalid_generation_guideline", status_code=400, ) unknown = sorted(set(raw) - set(STYLE_CONTEXT_KEYS)) if unknown: raise CatalogError( "Unbekannte Stilquellen: " + ", ".join(unknown), code="invalid_generation_guideline", status_code=400, ) missing = [key for key in STYLE_CONTEXT_KEYS if key not in raw] if missing and required: raise CatalogError( "style_context ist unvollständig: " + ", ".join(missing), code="invalid_generation_guideline", status_code=400, ) parsed: dict[str, bool] = dict(EMPTY_STYLE_CONTEXT) for key in STYLE_CONTEXT_KEYS: if key not in raw: continue value = raw[key] if isinstance(value, bool): parsed[key] = value elif value in (0, 1): parsed[key] = bool(value) else: raise CatalogError( f"{key} muss wahr oder falsch sein, keine Gewichtung.", code="invalid_generation_guideline", status_code=400, ) return parsed def parse_style_context_json(raw, *, slot: str, guideline_key: str = "") -> dict[str, bool]: if slot != "voice": return {} text = (raw or "").strip() if not text or text in ("{}", "null"): return default_style_context_for_key(guideline_key, slot=slot) try: data = json.loads(text) except json.JSONDecodeError: return default_style_context_for_key(guideline_key, slot=slot) if not isinstance(data, dict) or not any(key in data for key in STYLE_CONTEXT_KEYS): return default_style_context_for_key(guideline_key, slot=slot) try: return normalize_style_context(data, slot=slot, required=False) except CatalogError: return default_style_context_for_key(guideline_key, slot=slot) def style_context_json(context: dict[str, bool] | None, *, slot: str) -> str: if slot != "voice": return "{}" payload = normalize_style_context(context or EMPTY_STYLE_CONTEXT, slot=slot, required=False) return json.dumps(payload, ensure_ascii=False, sort_keys=True) def _limits() -> tuple[int, int, int]: seed = load_seed_document() return ( int(seed.get("max_instruction_chars") or MAX_INSTRUCTION_CHARS), int(seed.get("max_label_chars") or MAX_LABEL_CHARS), int(seed.get("max_summary_chars") or MAX_SUMMARY_CHARS), ) def _clean_text(raw, field: str, *, max_chars: int, allow_empty: bool = False) -> str: text = (raw or "").strip() if not text and not allow_empty: raise CatalogError(f"{field} darf nicht leer sein.", code="invalid_generation_guideline", status_code=400) if len(text) > max_chars: raise CatalogError(f"{field} darf höchstens {max_chars} Zeichen haben.", code="invalid_generation_guideline", status_code=400) return text def _clean_key(raw) -> str: key = (raw or "").strip() if not KEY_RE.match(key): raise CatalogError( "guideline_key muss aus Kleinbuchstaben, Ziffern und Unterstrich bestehen.", code="invalid_generation_guideline", status_code=400, ) return key def _row(item: dict | None, *, include_instruction: bool = False) -> dict | None: if not item: return None payload = {field: item.get(field) for field in PUBLIC_FIELDS} payload["is_default"] = _as_bool(item.get("is_default", 0)) payload["is_system_seed"] = _as_bool(item.get("is_system_seed", 0)) payload["revision"] = int(item.get("revision") or 1) payload["sort_order"] = int(item.get("sort_order") or 0) payload["guideline_key"] = item.get("guideline_key") or "" payload["style_context"] = parse_style_context_json( item.get("style_context_json"), slot=item.get("slot") or "", guideline_key=payload["guideline_key"], ) if include_instruction: payload["instruction"] = item.get("instruction") or "" payload["seed_id"] = item.get("seed_id") or "" return payload def _fetch(guideline_id: str) -> dict | None: with get_db() as conn: return row_to_dict( conn.execute("SELECT * FROM generation_guidelines WHERE id = ?", (guideline_id,)).fetchone() ) def get_guideline(guideline_id: str, *, include_instruction: bool = True) -> dict: row = _fetch(guideline_id) if not row: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) return _row(row, include_instruction=include_instruction) def list_guidelines( purpose: str = PURPOSE_JOURNAL, *, slot: str | None = None, statuses: tuple[str, ...] | None = None, include_instruction: bool = False, ) -> list[dict]: query = "SELECT * FROM generation_guidelines WHERE purpose = ?" params: list = [purpose] if slot: query += " AND slot = ?" params.append(slot) if statuses: query += " AND status IN (" + ",".join("?" for _ in statuses) + ")" params.extend(statuses) query += " ORDER BY slot, sort_order, revision, label" with get_db() as conn: rows = [row_to_dict(row) for row in conn.execute(query, params).fetchall()] return [_row(row, include_instruction=include_instruction) for row in rows] def overview_payload(purpose: str = PURPOSE_JOURNAL) -> dict: seed = load_seed_document() items = list_guidelines(purpose) slots = {slot: [item for item in items if item["slot"] == slot] for slot in ALLOWED_SLOTS} return { "purpose": purpose, "seed_revision": seed.get("seed_revision") or "", "max_instruction_chars": int(seed.get("max_instruction_chars") or MAX_INSTRUCTION_CHARS), "max_label_chars": int(seed.get("max_label_chars") or MAX_LABEL_CHARS), "max_summary_chars": int(seed.get("max_summary_chars") or MAX_SUMMARY_CHARS), "slots": slots, } def active_user_options(purpose: str = PURPOSE_JOURNAL) -> dict[str, list[dict]]: items = list_guidelines(purpose, statuses=(STATUS_ACTIVE,)) return { slot: [ { "id": item["id"], "key": item["guideline_key"], "label": item["label"], "summary": item["summary"], "revision": item["revision"], "is_default": item["is_default"], "style_context": item.get("style_context") or {}, } for item in items if item["slot"] == slot ] for slot in USER_SLOTS } def default_selection_ids(purpose: str = PURPOSE_JOURNAL) -> dict[str, str]: items = list_guidelines(purpose, statuses=(STATUS_ACTIVE,)) chosen = {} for slot in USER_SLOTS: slot_items = [item for item in items if item["slot"] == slot] if not slot_items: raise CatalogError(f"{slot} hat keine aktive Ausprägung.") preferred = next((item for item in slot_items if item["is_default"]), slot_items[0]) chosen[SLOT_TO_ID_KEY[slot]] = preferred["id"] return chosen def _require_active_for_run(row: dict, slot: str) -> dict: if not row: raise GenerationPolicyError(f"{slot} ist unbekannt.") if row.get("slot") != slot: raise GenerationPolicyError(f"{slot} verweist auf die falsche Dimension.") if row.get("status") != STATUS_ACTIVE: raise GenerationPolicyError( f"{slot} ist nicht für neue Läufe verfügbar.", code="invalid_generation_selection", ) if not (row.get("instruction") or "").strip(): raise CatalogError(f"{slot} hat keine Anweisung.") return row def validate_selection(raw, *, purpose: str = PURPOSE_JOURNAL) -> dict[str, dict]: if not isinstance(raw, dict): raise GenerationPolicyError("generation_selection muss die vier Ausprägungs-IDs enthalten.") unknown = sorted(set(raw) - set(SELECTION_KEYS)) if unknown: raise GenerationPolicyError("Unbekannte Auswahl: " + ", ".join(unknown)) missing = [key for key in SELECTION_KEYS if not (raw.get(key) or "").strip()] if missing: raise GenerationPolicyError("generation_selection ist unvollständig: " + ", ".join(missing)) chosen = {} for key in SELECTION_KEYS: slot = ID_KEY_TO_SLOT[key] row = _fetch(str(raw[key]).strip()) chosen[slot] = _require_active_for_run(row, slot) if (row.get("purpose") or purpose) != purpose: raise GenerationPolicyError(f"{slot} gehört nicht zu diesem Zweck.") return chosen def compile_selection( selection: dict[str, str], *, purpose: str = PURPOSE_JOURNAL, ) -> CompiledPolicy: chosen = validate_selection(selection, purpose=purpose) instructions = { SLOT_INSTRUCTION_KEYS[slot]: chosen[slot]["instruction"] for slot in USER_SLOTS } revision = "" for row in chosen.values(): revision = row.get("seed_revision") or revision return CompiledPolicy( ids={slot: chosen[slot]["id"] for slot in USER_SLOTS}, keys={slot: chosen[slot]["guideline_key"] for slot in USER_SLOTS}, labels={slot: chosen[slot]["label"] for slot in USER_SLOTS}, summaries={slot: chosen[slot].get("summary") or "" for slot in USER_SLOTS}, revisions={slot: int(chosen[slot].get("revision") or 1) for slot in USER_SLOTS}, instructions=instructions, seed_revision=revision, style_context=parse_style_context_json( chosen["voice"].get("style_context_json"), slot="voice", guideline_key=chosen["voice"].get("guideline_key") or "", ), cloned_from={slot: chosen[slot].get("cloned_from") or "" for slot in USER_SLOTS}, ) def policy_trace( compiled: CompiledPolicy, *, source: str, remembered: bool, ) -> dict: return { "source": source, "remembered": remembered, "ids": dict(compiled.ids), "keys": dict(compiled.keys), "labels": dict(compiled.labels), "revisions": dict(compiled.revisions), "seed_revision": compiled.seed_revision, "style_context": dict(compiled.style_context), "cloned_from": dict(compiled.cloned_from), } def draft_snapshot(compiled: CompiledPolicy, *, prompt: dict | None = None, model: str = "") -> dict: """Persistable run snapshot. No prompt bodies or instruction texts.""" return { "transformation": { "id": compiled.ids["transformation"], "key": compiled.keys["transformation"], "label": compiled.labels["transformation"], "revision": compiled.revisions["transformation"], }, "detail": { "id": compiled.ids["detail"], "key": compiled.keys["detail"], "label": compiled.labels["detail"], "revision": compiled.revisions["detail"], }, "voice": { "id": compiled.ids["voice"], "key": compiled.keys["voice"], "label": compiled.labels["voice"], "revision": compiled.revisions["voice"], "cloned_from": compiled.cloned_from.get("voice") or "", "style_context": dict(compiled.style_context), }, "narrative": { "id": compiled.ids["narrative"], "key": compiled.keys["narrative"], "label": compiled.labels["narrative"], "revision": compiled.revisions["narrative"], }, "prompt_slug": (prompt or {}).get("slug") or "", "prompt_revision": (prompt or {}).get("seed_revision") or "", "model": model or "", } def snapshot_summary(snapshot: dict | None) -> str: data = snapshot or {} labels = [ (data.get("transformation") or {}).get("label") or "", (data.get("detail") or {}).get("label") or "", (data.get("voice") or {}).get("label") or "", (data.get("narrative") or {}).get("label") or "", ] labels = [item for item in labels if item] if not labels: return "" line = " · ".join(item[:1].upper() + item[1:] if item else item for item in labels) return f"Erzeugt mit:\n{line}" def assert_journal_prompt_contract(template: str) -> None: keys = CONTEXT_PATTERN.findall(template or "") retired = [key for key in RETIRED_JOURNAL_PLACEHOLDERS if key in keys] if retired: shown = ", ".join(f"{{{{{key}}}}}" for key in retired) raise CatalogError( "Dieser Journal-Prompt verwendet den veralteten Platzhalter " + shown + ". Die Unterscheidung zwischen Fließtext- und Stichpunktmodus ist entfallen. " "Bitte den Prompt unter Admin → Prompts aktualisieren.", code="prompt_contract_incompatible", ) missing = [key for key in REQUIRED_JOURNAL_PLACEHOLDERS if key not in keys] if missing: shown = ", ".join(f"{{{{{key}}}}}" for key in missing) raise CatalogError( "Dieser Journal-Prompt erfüllt den aktuellen Vertrag nicht. Es fehlen: " + shown + ". Bitte den Prompt unter Admin → Prompts aktualisieren.", code="prompt_contract_incompatible", ) def assert_template_resolved(rendered: str) -> None: leftover = CONTEXT_PATTERN.findall(rendered or "") if leftover: raise CatalogError( "Promptplatzhalter konnten nicht aufgelöst werden: " + ", ".join(f"{{{{{key}}}}}" for key in leftover), code="unresolved_placeholder", ) def omit_orphaned_style_example_rule(template: str) -> str: """Drop the standing STYLE_EXAMPLES sentence when no example block is sent.""" text = template or "" return text.replace(STYLE_EXAMPLES_STANDING_SENTENCE, "").replace( "STYLE_EXAMPLES dienen ausschließlich als Stilreferenz; ihre Inhalte gehören nicht zum heutigen Tag.", "", ) def prompt_without_orphaned_style_example_rule(prompt: dict, *, include_style_examples: bool) -> dict: if include_style_examples: return prompt template = prompt.get("template") or "" cleaned = omit_orphaned_style_example_rule(template) if cleaned == template: return prompt return {**prompt, "template": cleaned} def load_selection(profile_id: str) -> dict[str, str] | None: with get_db() as conn: row = row_to_dict( conn.execute( """ SELECT transformation_id, detail_id, voice_id, narrative_id, updated FROM journal_generation_selection WHERE profile_id = ? """, (profile_id,), ).fetchone() ) if not row: return None values = {key: row[key] for key in SELECTION_KEYS} values["updated"] = row.get("updated") or "" return values def save_selection(profile_id: str, selection: dict[str, str]) -> dict[str, str]: chosen = validate_selection(selection) checked = {SLOT_TO_ID_KEY[slot]: chosen[slot]["id"] for slot in USER_SLOTS} with get_db() as conn: conn.execute( """ INSERT INTO journal_generation_selection ( profile_id, transformation_id, detail_id, voice_id, narrative_id, updated ) VALUES (?, ?, ?, ?, ?, datetime('now')) ON CONFLICT(profile_id) DO UPDATE SET transformation_id = excluded.transformation_id, detail_id = excluded.detail_id, voice_id = excluded.voice_id, narrative_id = excluded.narrative_id, updated = datetime('now') """, ( profile_id, checked["transformation_id"], checked["detail_id"], checked["voice_id"], checked["narrative_id"], ), ) stored = load_selection(profile_id) or {**checked, "updated": ""} return stored def get_or_create_selection(profile_id: str) -> dict[str, str]: existing = load_selection(profile_id) if existing is not None: return existing return save_selection(profile_id, default_selection_ids()) def settings_payload(profile_id: str) -> dict: stored = get_or_create_selection(profile_id) selection = {key: stored[key] for key in SELECTION_KEYS} return { "selection": selection, "updated": stored.get("updated") or "", "options": active_user_options(), "defaults": default_selection_ids(), } def resolve_run_selection( profile_id: str, snapshot: dict | None, remember: bool, ) -> tuple[dict[str, str], dict]: if snapshot is None: stored = get_or_create_selection(profile_id) values = {key: stored[key] for key in SELECTION_KEYS} validate_selection(values) return values, {"source": "profile", "remembered": False} values = {key: str(snapshot.get(key) or "").strip() for key in SELECTION_KEYS} validate_selection(values) remembered = False if remember: save_selection(profile_id, values) remembered = True return values, {"source": "request", "remembered": remembered} def mark_guidelines_used(ids: list[str]) -> None: clean = [item for item in ids if item] if not clean: return with get_db() as conn: conn.executemany( """ UPDATE generation_guidelines SET used_at = COALESCE(used_at, datetime('now')) WHERE id = ? """, [(item,) for item in clean], ) def _write_fields(raw: dict, *, slot: str) -> dict: max_instruction, max_label, max_summary = _limits() style_context = normalize_style_context( raw.get("style_context"), slot=slot, required=False, ) return { "guideline_key": _clean_key(raw.get("guideline_key") or raw.get("key")), "label": _clean_text(raw.get("label"), "label", max_chars=max_label), "summary": _clean_text(raw.get("summary"), "summary", max_chars=max_summary, allow_empty=True), "instruction": _clean_text(raw.get("instruction"), "instruction", max_chars=max_instruction), "sort_order": int(raw.get("sort_order") or 0), "style_context": style_context, "style_context_json": style_context_json(style_context, slot=slot), } def create_guideline(purpose: str, slot: str, body: dict) -> dict: if slot not in ALLOWED_SLOTS: raise CatalogError("Unbekannter Slot.", code="invalid_generation_guideline", status_code=400) fields = _write_fields(body, slot=slot) guideline_id = str(uuid.uuid4()) with get_db() as conn: count = conn.execute( "SELECT COALESCE(MAX(sort_order), -1) AS n FROM generation_guidelines WHERE purpose = ? AND slot = ?", (purpose, slot), ).fetchone()["n"] conn.execute( """ INSERT INTO generation_guidelines ( id, purpose, slot, guideline_key, label, summary, instruction, style_context_json, sort_order, status, revision, cloned_from, is_default, is_system_seed, seed_id, seed_revision, created, updated ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, NULL, 0, 0, '', '', datetime('now'), datetime('now')) """, ( guideline_id, purpose, slot, fields["guideline_key"], fields["label"], fields["summary"], fields["instruction"], fields["style_context_json"], fields["sort_order"] if body.get("sort_order") is not None else int(count) + 1, STATUS_DRAFT, ), ) return get_guideline(guideline_id) def clone_guideline(guideline_id: str) -> dict: current = _fetch(guideline_id) if not current: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) if current.get("slot") == LEGACY_SOURCE_MODE_SLOT: raise CatalogError( "Der Quellenmodus ist kein aktiver Slot mehr.", code="invalid_generation_guideline", status_code=400, ) new_id = str(uuid.uuid4()) with get_db() as conn: conn.execute( """ INSERT INTO generation_guidelines ( id, purpose, slot, guideline_key, label, summary, instruction, style_context_json, sort_order, status, revision, cloned_from, is_default, is_system_seed, seed_id, seed_revision, created, updated ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, '', ?, datetime('now'), datetime('now')) """, ( new_id, current["purpose"], current["slot"], current["guideline_key"], current["label"], current["summary"], current["instruction"], current.get("style_context_json") or style_context_json( parse_style_context_json( current.get("style_context_json"), slot=current.get("slot") or "", guideline_key=current.get("guideline_key") or "", ), slot=current.get("slot") or "", ), int(current.get("sort_order") or 0), STATUS_DRAFT, int(current.get("revision") or 1) + 1, current["id"], current.get("seed_revision") or "", ), ) return get_guideline(new_id) def update_guideline(guideline_id: str, body: dict) -> dict: current = _fetch(guideline_id) if not current: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) if current.get("status") != STATUS_DRAFT: raise CatalogError( "Aktive oder archivierte Ausprägungen sind unveränderlich. Bitte klonen.", code="guideline_immutable", status_code=409, ) merged = {**current, **body, "guideline_key": body.get("guideline_key") or current["guideline_key"]} if "style_context" not in body: merged["style_context"] = parse_style_context_json( current.get("style_context_json"), slot=current.get("slot") or "", guideline_key=current.get("guideline_key") or "", ) fields = _write_fields(merged, slot=current.get("slot") or "") with get_db() as conn: conn.execute( """ UPDATE generation_guidelines SET guideline_key = ?, label = ?, summary = ?, instruction = ?, style_context_json = ?, sort_order = ?, updated = datetime('now') WHERE id = ? """, ( fields["guideline_key"], fields["label"], fields["summary"], fields["instruction"], fields["style_context_json"], fields["sort_order"], guideline_id, ), ) return get_guideline(guideline_id) def publish_guideline(guideline_id: str) -> dict: current = _fetch(guideline_id) if not current: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) if current.get("status") == STATUS_ARCHIVED: raise CatalogError("Archivierte Ausprägungen können nicht veröffentlicht werden.", code="guideline_archived", status_code=409) max_instruction, max_label, max_summary = _limits() _clean_text(current.get("label"), "label", max_chars=max_label) _clean_text(current.get("summary"), "summary", max_chars=max_summary, allow_empty=True) _clean_text(current.get("instruction"), "instruction", max_chars=max_instruction) if current.get("slot") == "voice": normalize_style_context( parse_style_context_json( current.get("style_context_json"), slot="voice", guideline_key=current.get("guideline_key") or "", ), slot="voice", required=True, ) with get_db() as conn: conn.execute( "UPDATE generation_guidelines SET status = ?, updated = datetime('now') WHERE id = ?", (STATUS_ACTIVE, guideline_id), ) return get_guideline(guideline_id, include_instruction=False) def archive_guideline(guideline_id: str) -> dict: current = _fetch(guideline_id) if not current: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) if current.get("status") == STATUS_DRAFT: raise CatalogError("Drafts werden gelöscht, nicht archiviert.", code="invalid_generation_guideline", status_code=400) with get_db() as conn: remaining = conn.execute( """ SELECT COUNT(*) AS n FROM generation_guidelines WHERE purpose = ? AND slot = ? AND status = ? AND id != ? """, (current["purpose"], current["slot"], STATUS_ACTIVE, guideline_id), ).fetchone()["n"] if int(remaining or 0) < 1: raise CatalogError( "Die letzte aktive Ausprägung dieser Dimension kann nicht archiviert werden.", code="guideline_last_active", status_code=409, ) with get_db() as conn: conn.execute( """ UPDATE generation_guidelines SET status = ?, is_default = 0, updated = datetime('now') WHERE id = ? """, (STATUS_ARCHIVED, guideline_id), ) return get_guideline(guideline_id, include_instruction=False) def set_default_guideline(guideline_id: str) -> dict: current = _fetch(guideline_id) if not current: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) if current.get("status") != STATUS_ACTIVE: raise CatalogError("Nur aktive Ausprägungen können Standard sein.", code="invalid_generation_guideline", status_code=400) if current["slot"] not in USER_SLOTS: raise CatalogError("Nur die vier Gestaltungsdimensionen haben einen Nutzerstandard.", code="invalid_generation_guideline", status_code=400) with get_db() as conn: conn.execute( """ UPDATE generation_guidelines SET is_default = CASE WHEN id = ? THEN 1 ELSE 0 END, updated = datetime('now') WHERE purpose = ? AND slot = ? """, (guideline_id, current["purpose"], current["slot"]), ) return get_guideline(guideline_id, include_instruction=False) def delete_guideline(guideline_id: str) -> None: current = _fetch(guideline_id) if not current: raise CatalogError("Ausprägung nicht gefunden.", code="guideline_missing", status_code=404) if current.get("status") != STATUS_DRAFT: raise CatalogError("Nur unverwendete Drafts können gelöscht werden.", code="guideline_immutable", status_code=409) if current.get("used_at"): raise CatalogError("Verwendete Drafts können nicht gelöscht werden.", code="guideline_in_use", status_code=409) with get_db() as conn: conn.execute("DELETE FROM generation_guidelines WHERE id = ?", (guideline_id,)) def reset_seed_drafts(purpose: str = PURPOSE_JOURNAL) -> dict: """Create new drafts from the current seed. Never overwrite published variants.""" seed = load_seed_document() created = [] slots = seed.get("slots") or {} for slot in ALLOWED_SLOTS: for index, item in enumerate((slots.get(slot) or {}).get("variants") or []): created.append( create_guideline( purpose, slot, { "guideline_key": item.get("guideline_key") or item.get("variant_key"), "label": item.get("label") or "", "summary": item.get("summary") or "", "instruction": item.get("instruction") or "", "sort_order": index, "style_context": item.get("style_context") or {}, }, ) ) overview = overview_payload(purpose) overview["reset_drafts"] = created return overview def preview_selection(selection: dict[str, str], *, purpose: str = PURPOSE_JOURNAL) -> dict: from engine import load_active_prompt, preview_prompt compiled = compile_selection(selection, purpose=purpose) prompt = load_active_prompt("mvp.journal_generate") assert_journal_prompt_contract(prompt.get("template") or "") include_examples = bool(compiled.style_context.get("include_style_examples")) prompt = prompt_without_orphaned_style_example_rule( prompt, include_style_examples=include_examples, ) rendered = preview_prompt( prompt, { **compiled.instructions, "writing_profile": "(nicht enthalten)", "style_examples": "(Stilbeispiele)" if include_examples else "", "reconstruction": "(nicht enthalten)", "existing_text": "", }, ) assert_template_resolved(rendered.get("rendered") or "") return { **compiled.instructions, "rendered": rendered.get("rendered") or "", "prompt_slug": prompt.get("slug") or "", "prompt_revision": prompt.get("seed_revision") or "", "selection": { "ids": dict(compiled.ids), "keys": dict(compiled.keys), "labels": dict(compiled.labels), "revisions": dict(compiled.revisions), "cloned_from": dict(compiled.cloned_from), "style_context": dict(compiled.style_context), }, "seed_revision": compiled.seed_revision, } def _seed_rows(seed: dict, purpose: str) -> list[tuple]: revision = seed.get("seed_revision") or "" rows = [] slots = seed.get("slots") or {} for slot in ALLOWED_SLOTS: variants = (slots.get(slot) or {}).get("variants") or [] for index, item in enumerate(variants): context = normalize_style_context(item.get("style_context") or {}, slot=slot, required=False) rows.append( ( item.get("id") or str(uuid.uuid4()), purpose, slot, item.get("guideline_key") or item.get("variant_key"), item.get("label") or "", item.get("summary") or "", item.get("instruction") or "", style_context_json(context, slot=slot), index, STATUS_ACTIVE, 1, 1 if item.get("is_default") else 0, 1, item.get("id") or "", revision, ) ) return rows def _semantic_signature(item: dict | None = None, *, seed_row: tuple | None = None) -> tuple: if seed_row is not None: slot = seed_row[2] or "" key = seed_row[3] or "" return ( key, seed_row[4] or "", seed_row[5] or "", seed_row[6] or "", style_context_json( parse_style_context_json(seed_row[7], slot=slot, guideline_key=key), slot=slot, ), ) slot = (item or {}).get("slot") or "" key = (item or {}).get("guideline_key") or "" return ( key, (item or {}).get("label") or "", (item or {}).get("summary") or "", (item or {}).get("instruction") or "", style_context_json( parse_style_context_json((item or {}).get("style_context_json"), slot=slot, guideline_key=key), slot=slot, ), ) def _successor_id(base_id: str, revision: int) -> str: root = re.sub(r"-r\d+$", "", base_id or "") return f"{root}-r{int(revision)}" def _catalog_maps(conn, purpose: str) -> tuple[dict, dict]: rows = [row_to_dict(row) for row in conn.execute( "SELECT * FROM generation_guidelines WHERE purpose = ?", (purpose,), ).fetchall()] by_id = {row["id"]: row for row in rows} by_seed_id = {row["seed_id"]: row for row in rows if row.get("seed_id")} return by_id, by_seed_id def _find_existing_seed_row(by_id: dict, by_seed_id: dict, seed_row: tuple, all_rows: list[dict]) -> dict | None: current = by_id.get(seed_row[0]) or by_seed_id.get(seed_row[13]) if current: return current key = seed_row[3] or "" slot = seed_row[2] or "" if key not in LEGACY_COMPARISON_KEYS: return None for item in all_rows: if item.get("slot") == slot and item.get("guideline_key") == key: return item return None def _find_matching_successor(all_rows: list[dict], predecessor_id: str, seed_row: tuple) -> dict | None: wanted = _semantic_signature(seed_row=seed_row) for item in all_rows: if (item.get("cloned_from") or "") != predecessor_id: continue if _semantic_signature(item) == wanted: return item return None def _insert_seed_guideline( conn, seed_row: tuple, *, guideline_id: str | None = None, cloned_from: str | None = None, revision: int | None = None, is_default: int | None = None, ) -> None: row_id = guideline_id or seed_row[0] seed_id = row_id if guideline_id else seed_row[13] conn.execute( """ INSERT INTO generation_guidelines ( id, purpose, slot, guideline_key, label, summary, instruction, style_context_json, sort_order, status, revision, cloned_from, is_default, is_system_seed, seed_id, seed_revision, created, updated ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now')) """, ( row_id, seed_row[1], seed_row[2], seed_row[3], seed_row[4], seed_row[5], seed_row[6], seed_row[7], seed_row[8], STATUS_ACTIVE, int(revision if revision is not None else seed_row[10] or 1), cloned_from, int(is_default if is_default is not None else seed_row[11] or 0), 1, seed_id, seed_row[14], ), ) def _assign_slot_default(conn, purpose: str, slot: str, guideline_id: str) -> None: conn.execute( """ UPDATE generation_guidelines SET is_default = CASE WHEN id = ? THEN 1 ELSE 0 END, updated = datetime('now') WHERE purpose = ? AND slot = ? AND status = ? """, (guideline_id, purpose, slot, STATUS_ACTIVE), ) def seed_generation_instructions(conn, seed: dict | None = None) -> None: """Insert missing seed guidelines. Never rewrite existing IDs, clones, or archives. A semantic change in the seed creates a new successor ID. New defaults may point at that successor. Stored selections keep their previous IDs. """ seed = seed if seed is not None else load_seed_document() purpose = seed.get("purpose") or PURPOSE_JOURNAL conn.execute( """ UPDATE generation_guidelines SET status = ?, is_default = 0, updated = datetime('now') WHERE purpose = ? AND slot = ? AND status != ? """, (STATUS_ARCHIVED, purpose, LEGACY_SOURCE_MODE_SLOT, STATUS_ARCHIVED), ) for row in _seed_rows(seed, purpose): by_id, by_seed_id = _catalog_maps(conn, purpose) all_rows = list(by_id.values()) current = _find_existing_seed_row(by_id, by_seed_id, row, all_rows) if current: if not _as_bool(current.get("is_system_seed", 0)): continue if current.get("status") != STATUS_ACTIVE: continue if _semantic_signature(current) == _semantic_signature(seed_row=row): continue successor = _find_matching_successor(all_rows, current["id"], row) if successor: if row[11] and successor.get("status") == STATUS_ACTIVE: _assign_slot_default(conn, purpose, row[2], successor["id"]) continue next_revision = int(current.get("revision") or 1) + 1 taken = set(by_id) successor_id = _successor_id(current["id"], next_revision) while successor_id in taken: next_revision += 1 successor_id = _successor_id(current["id"], next_revision) _insert_seed_guideline( conn, row, guideline_id=successor_id, cloned_from=current["id"], revision=next_revision, is_default=int(row[11] or 0), ) if row[11]: _assign_slot_default(conn, purpose, row[2], successor_id) continue _insert_seed_guideline(conn, row) if row[11]: _assign_slot_default(conn, purpose, row[2], row[0]) def backfill_missing_settings(conn) -> None: seed_generation_instructions(conn) defaults = {} for slot, key in SLOT_TO_ID_KEY.items(): row = conn.execute( """ SELECT id FROM generation_guidelines WHERE purpose = ? AND slot = ? AND status = ? AND is_default = 1 ORDER BY sort_order LIMIT 1 """, (PURPOSE_JOURNAL, slot, STATUS_ACTIVE), ).fetchone() if not row: row = conn.execute( """ SELECT id FROM generation_guidelines WHERE purpose = ? AND slot = ? AND status = ? ORDER BY sort_order LIMIT 1 """, (PURPOSE_JOURNAL, slot, STATUS_ACTIVE), ).fetchone() if row: defaults[key] = row["id"] if len(defaults) != 4: return conn.execute( """ INSERT OR IGNORE INTO journal_generation_selection ( profile_id, transformation_id, detail_id, voice_id, narrative_id, updated ) SELECT id, ?, ?, ?, ?, datetime('now') FROM profiles """, ( defaults["transformation_id"], defaults["detail_id"], defaults["voice_id"], defaults["narrative_id"], ), )