"""Cost-conscious profile review: local drift, bundled evidence, optional AI/paste.

Does not rebuild the writing profile after every new source. Dialogue turns never
trigger an LLM review. Interaction preferences are never inferred from silence.
Semantic traits come from review, not from word counts. Continuous learning starts
only after a confirmed Initial Profile Build.
"""
from __future__ import annotations

import json
import re
import uuid
from datetime import datetime, timedelta, timezone

from db import get_db, row_to_dict
from debug_store import persist_engine_error, persist_step
from dialogue_store import StoreError
from journal_body import plain_text
from journal_store import list_versions
from writing_profile_infer import SIGNAL_KEYS, infer_features
from writing_profile_schema import (
    ACTION_ALIASES,
    APPLY_ACTIONS,
    EXISTING_BEFORE_NEW,
    LEGACY_STYLE_KEYS,
    SEED_FACETS,
    TRAIT_ACTIONS,
    coerce_slug,
    facet_label,
    hint_for_context,
    is_external_only,
    is_meta_style_text,
    normalize_evidence_basis,
    normalize_facet_key,
    normalize_mode,
    seed_catalog,
    valid_slug,
)
from writing_profile_store import (
    GOVERNANCE,
    INITIAL_BUILD_SOURCES,
    _assemble_brief,
    _queue_suggestion,
    _upsert_facet,
    ensure_profile,
    get_profile,
    has_confirmed_profile,
    import_text,
    list_corpus,
    refresh_profile,
    replace_trait_refs,
    retire_trait,
    snapshot_version,
    upsert_trait,
)
from profile_analysis import (
    KIND_PACKAGE,
    KIND_PACKAGES,
    KIND_RESULT,
    KIND_RESULT_LEGACY,
    KIND_RESULTS,
    FORMAT_VERSION as ANALYSIS_FORMAT,
    expected_result_schema,
    filter_style_evidence,
    package_note,
    render_paste_prompt as render_analysis_prompt,
    select_corpus_items,
    semantic_task,
)

KIND_PACKAGE_LEGACY = "kansho.profile_review_package"
FORMAT_VERSION = 1
JSON_BLOCK = re.compile(r"\{.*\}", re.DOTALL)
TRIGGERS = {
    "user_edit_of_draft",
    "heuristic_drift",
    "dialogue_sample",
    "periodic_batch",
    "explicit_review",
}
CORE_KEY = "core"
JOURNAL_FACET = "autobiographical_journal"
BATCH_MIN = 3
MAX_BUNDLE = 8
EXCERPT_CHARS = 480
EXAMPLE_CHARS = 220
CORPUS_EXCERPT = 360
DIALOGUE_MIN_WORDS = 60
DIALOGUE_COOLDOWN_HOURS = 24
DRIFT_MIN = 2
PERIODIC_DAYS = 7
LAYERS = {"core", "facet", "trait"}

INSTRUCTION_SHARED = (
    "Keine Diagnose, kein Persönlichkeitsmodell, keine erfundenen Stilmerkmale. "
    "Lokale Heuristiken (Wortzählungen, Signalwörter) sind Messhilfe, keine Traits. "
    "Übernimm Heuristik-Keys wie chronology, humor, detail, rhythm nicht als festes Raster. "
    "Jeder semantische Trait braucht echte Source-Evidence und darf repräsentative Originalbeispiele nennen. "
    "Urlaubstagebücher und autobiografische Journale belegen primär die Facet autobiographical_journal. "
    "Einen globalen Core nur so weit verallgemeinern, wie unterschiedliche Quellenarten das tragen. "
    "Neuere Texte wiegen stärker für die aktuelle Ausprägung; ältere Texte belegen stabile Langzeitmerkmale. "
    "Existing-before-New: " + "; ".join(EXISTING_BEFORE_NEW) + "."
)
INSTRUCTION_INCREMENTAL = (
    "Vergleiche die neuen Evidenzen mit dem bestehenden Profil. "
    "Erzeuge kein komplett neues Profil. "
    "Unterscheide Core, optionale context/output-Facets und dynamische semantische Traits darin. "
    "Schlage nur Änderungen vor, die die Evidenz trägt. "
    + INSTRUCTION_SHARED
)
INSTRUCTION_INITIAL = (
    "Bilde ein erstes Writing Profile aus dem historischen Korpus, nicht nur aus zwei aktuellen Einträgen. "
    "Die Hülle ist Core plus optionale Facets plus dynamische Traits, Evidence References und Representative Exemplars. "
    "Seed-Keys sind nur Ordnungshilfe. "
    + INSTRUCTION_SHARED
)


def _parse_stamp(raw: str | None) -> datetime | None:
    text = (raw or "").strip()
    if not text:
        return None
    try:
        stamp = datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None
    if stamp.tzinfo is None:
        stamp = stamp.replace(tzinfo=timezone.utc)
    return stamp


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _parse_json(raw: str | None, fallback):
    if not raw:
        return fallback
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return fallback
    return data if data is not None else fallback


def facet_delta(current: dict[str, str], incoming: dict[str, str]) -> list[str]:
    """Local-signal delta. Not a semantic trait diff."""
    changed: list[str] = []
    for key, value in incoming.items():
        old = (current.get(key) or "").strip()
        new = (value or "").strip()
        if old and new and old != new:
            changed.append(key)
        elif not old and new:
            changed.append(key)
    return changed


def is_strong_edit(previous: str, current: str) -> bool:
    old = plain_text(previous or "").strip()
    new = plain_text(current or "").strip()
    if not new:
        return False
    if not old:
        return len(new.split()) >= 40
    old_words = set(old.lower().split())
    new_words = set(new.lower().split())
    union = max(len(old_words | new_words), 1)
    overlap = len(old_words & new_words) / union
    length = abs(len(new) - len(old)) / max(len(old), 1)
    return overlap < 0.55 or length > 0.3


def _heuristic_baseline(profile_id: str, *, exclude_entry_id: str | None = None) -> dict[str, str]:
    texts = [
        item.get("body") or ""
        for item in list_corpus(profile_id, limit=12)
        if (item.get("entry_id") or "") != (exclude_entry_id or "")
    ]
    return infer_features(texts)


def classify_journal_trigger(profile_id: str, entry_id: str, body: str, origin: str | None) -> str | None:
    text = plain_text(body or "").strip()
    if not text:
        return None
    origin = (origin or "").strip()
    if origin == "user_edit" and entry_id:
        versions = list_versions(profile_id, entry_id)
        if len(versions) >= 2:
            previous = versions[-2]
            if (previous.get("origin") or "") == "accepted_draft" and is_strong_edit(
                previous.get("body") or "", text
            ):
                return "user_edit_of_draft"
    incoming = infer_features([text])
    baseline = _heuristic_baseline(profile_id, exclude_entry_id=entry_id)
    if baseline and len(facet_delta(baseline, incoming)) >= DRIFT_MIN:
        return "heuristic_drift"
    return None


def enqueue_evidence(
    profile_id: str,
    *,
    trigger: str,
    excerpt: str,
    source_kind: str = "",
    source_id: str | None = None,
) -> dict | None:
    if trigger not in TRIGGERS:
        return None
    text = plain_text(excerpt or "").strip()
    if not text:
        return None
    row = ensure_profile(profile_id)
    if (row.get("governance") or "learning") == "frozen" and trigger != "explicit_review":
        return None
    signals = infer_features([text])
    example_for = [key for key in signals if key in SIGNAL_KEYS][:6]
    excerpt = text[:EXCERPT_CHARS]
    with get_db() as conn:
        existing = row_to_dict(
            conn.execute(
                """
                SELECT id FROM writing_profile_evidence
                WHERE profile_id = ? AND trigger = ? AND excerpt = ? AND status = 'pending'
                LIMIT 1
                """,
                (profile_id, trigger, excerpt),
            ).fetchone()
        )
        if existing:
            return existing
        item = {
            "id": str(uuid.uuid4()),
            "profile_id": profile_id,
            "trigger": trigger,
            "source_kind": source_kind,
            "source_id": source_id,
            "excerpt": excerpt,
            "local_signals_json": json.dumps(signals, ensure_ascii=False),
            "example_for_json": json.dumps(example_for, ensure_ascii=False),
            "status": "pending",
        }
        conn.execute(
            """
            INSERT INTO writing_profile_evidence
              (id, profile_id, trigger, source_kind, source_id, excerpt, local_signals_json, example_for_json, status)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                item["id"],
                profile_id,
                trigger,
                source_kind,
                source_id,
                excerpt,
                item["local_signals_json"],
                item["example_for_json"],
                item["status"],
            ),
        )
    _refresh_ready(profile_id)
    return item


def _pending_evidence(profile_id: str, limit: int = MAX_BUNDLE) -> list[dict]:
    with get_db() as conn:
        rows = [
            row_to_dict(item)
            for item in conn.execute(
                """
                SELECT * FROM writing_profile_evidence
                WHERE profile_id = ? AND status = 'pending'
                ORDER BY created
                LIMIT ?
                """,
                (profile_id, limit),
            ).fetchall()
        ]
    for item in rows:
        item["local_signals"] = _parse_json(item.get("local_signals_json"), {})
        item["example_for"] = _parse_json(item.get("example_for_json"), [])
    return rows


def _refresh_ready(profile_id: str) -> None:
    with get_db() as conn:
        pending = conn.execute(
            "SELECT COUNT(*) AS c FROM writing_profile_evidence WHERE profile_id = ? AND status = 'pending'",
            (profile_id,),
        ).fetchone()["c"]
        row = row_to_dict(
            conn.execute(
                "SELECT last_reviewed, lifecycle FROM writing_profiles WHERE profile_id = ?",
                (profile_id,),
            ).fetchone()
        )
    if (row or {}).get("lifecycle") != "confirmed":
        with get_db() as conn:
            conn.execute("UPDATE writing_profiles SET review_ready = 0 WHERE profile_id = ?", (profile_id,))
        return
    last = (row or {}).get("last_reviewed") or ""
    stamp = _parse_stamp(last)
    stale = False
    if stamp:
        stale = _now() - stamp >= timedelta(days=PERIODIC_DAYS) and pending >= 1
    elif last:
        stale = pending >= BATCH_MIN
    else:
        stale = pending >= 1
    ready = 1 if pending >= BATCH_MIN or stale else 0
    with get_db() as conn:
        conn.execute(
            "UPDATE writing_profiles SET review_ready = ? WHERE profile_id = ?",
            (ready, profile_id),
        )


def ingest_journal(profile_id: str, entry_id: str, body: str, origin: str | None = None) -> dict:
    """After a saved entry: assemble sources. Queue evidence only once the profile is confirmed."""
    refresh_profile(profile_id)
    if not has_confirmed_profile(profile_id):
        return get_profile(profile_id)
    trigger = classify_journal_trigger(profile_id, entry_id, body, origin)
    if trigger:
        enqueue_evidence(
            profile_id,
            trigger=trigger,
            excerpt=body,
            source_kind="journal_entry",
            source_id=entry_id,
        )
    elif origin == "imported_text":
        enqueue_evidence(
            profile_id,
            trigger="heuristic_drift",
            excerpt=body,
            source_kind="imported_text",
            source_id=None,
        )
    return get_profile(profile_id)


def ingest_import(
    profile_id: str,
    body: str,
    *,
    occurred_at: str | None = None,
    context_hint: str | None = None,
) -> dict:
    import_text(profile_id, body, occurred_at=occurred_at, context_hint=context_hint)
    if not has_confirmed_profile(profile_id):
        return get_profile(profile_id)
    enqueue_evidence(
        profile_id,
        trigger="heuristic_drift",
        excerpt=body,
        source_kind="imported_text",
    )
    return get_profile(profile_id)


def consider_dialogue(profile_id: str, last_user: str) -> None:
    """Sample only. Never an LLM call, never an interaction preference, never before confirm."""
    if not has_confirmed_profile(profile_id):
        return
    text = plain_text(last_user or "").strip()
    if len(text.split()) < DIALOGUE_MIN_WORDS:
        return
    row = ensure_profile(profile_id)
    if (row.get("governance") or "learning") == "frozen":
        return
    cutoff = (_now() - timedelta(hours=DIALOGUE_COOLDOWN_HOURS)).strftime("%Y-%m-%d %H:%M:%S")
    with get_db() as conn:
        recent = conn.execute(
            """
            SELECT id FROM writing_profile_evidence
            WHERE profile_id = ? AND trigger = 'dialogue_sample' AND created >= ?
            LIMIT 1
            """,
            (profile_id, cutoff),
        ).fetchone()
    if recent:
        return
    enqueue_evidence(
        profile_id,
        trigger="dialogue_sample",
        excerpt=text,
        source_kind="dialogue_style",
    )


def review_status(profile_id: str) -> dict:
    ensure_profile(profile_id)
    with get_db() as conn:
        pending = [
            row_to_dict(item)
            for item in conn.execute(
                """
                SELECT id, trigger, source_kind, created, status
                FROM writing_profile_evidence
                WHERE profile_id = ? AND status = 'pending'
                ORDER BY created
                """,
                (profile_id,),
            ).fetchall()
        ]
        open_review = row_to_dict(
            conn.execute(
                """
                SELECT id, channel, status, created FROM writing_profile_reviews
                WHERE profile_id = ? AND status IN ('open', 'proposed')
                ORDER BY created DESC LIMIT 1
                """,
                (profile_id,),
            ).fetchone()
        )
        versions = [
            row_to_dict(item)
            for item in conn.execute(
                """
                SELECT seq, cause, created FROM writing_profile_versions
                WHERE profile_id = ? ORDER BY seq DESC LIMIT 8
                """,
                (profile_id,),
            ).fetchall()
        ]
        row = row_to_dict(
            conn.execute(
                """
                SELECT version, review_ready, last_reviewed, governance, lifecycle
                FROM writing_profiles WHERE profile_id = ?
                """,
                (profile_id,),
            ).fetchone()
        )
    lifecycle = (row or {}).get("lifecycle") or "uninitialized"
    return {
        "version": (row or {}).get("version") or 0,
        "review_ready": bool((row or {}).get("review_ready")),
        "last_reviewed": (row or {}).get("last_reviewed"),
        "governance": (row or {}).get("governance") or "learning",
        "lifecycle": lifecycle,
        "confirmed": lifecycle == "confirmed",
        "pending_evidence": pending,
        "open_review": open_review,
        "versions": versions,
        "existing_before_new": list(EXISTING_BEFORE_NEW),
        "note": (
            "Erst nach einem bestätigten Initial Profile Build beginnt kontinuierliches Learning. "
            if lifecycle != "confirmed"
            else "Lokale Heuristik entscheidet, ob eine Review nötig ist. Kein Call nach jedem Dialog."
        ),
    }


def _review_mode(profile_id: str, requested: str | None = None) -> str:
    requested = normalize_mode(requested) or requested
    if requested in {"initial_build", "review"}:
        return requested
    return "review" if has_confirmed_profile(profile_id) else "initial_build"


def _corpus_supports_core(corpus: list[dict]) -> bool:
    hints = set()
    for item in corpus:
        hint = hint_for_context(item.get("facet_hint") or item.get("context_hint"))
        if not hint and item.get("kind") == "journal_entry":
            hint = JOURNAL_FACET
        if hint:
            hints.add(hint)
    return len(hints) >= 2


def _public_trait(item: dict) -> dict:
    return {
        "id": item.get("id"),
        "slug": item.get("slug"),
        "label": item.get("label") or item.get("slug"),
        "facet_key": item.get("facet_key") or CORE_KEY,
        "statement": item.get("statement") or "",
        "origin": item.get("origin") or "manual",
        "locked": bool(item.get("locked")),
        "status": item.get("status") or "active",
        "observed_from": item.get("observed_from"),
        "observed_to": item.get("observed_to"),
        "evidence_refs": [
            {
                "excerpt": ref.get("excerpt") or "",
                "occurred_at": ref.get("occurred_at"),
                "source_id": ref.get("source_id"),
            }
            for ref in item.get("evidence_refs") or []
            if ref.get("excerpt")
        ],
        "exemplars": [
            {
                "excerpt": ref.get("excerpt") or "",
                "occurred_at": ref.get("occurred_at"),
                "source_id": ref.get("source_id"),
            }
            for ref in item.get("exemplars") or []
            if ref.get("excerpt")
        ],
    }


def _public_facet(item: dict | None) -> dict:
    if not item:
        return {}
    key = item.get("facet_key")
    return {
        "key": key,
        "label": item.get("label") or facet_label(key or ""),
        "layer": item.get("layer") or "context",
        "value": item.get("value") or "",
        "origin": item.get("origin") or "manual",
        "locked": bool(item.get("locked")),
        "traits": [_public_trait(trait) for trait in item.get("traits") or []],
    }


def _public_evidence(item: dict) -> dict:
    return {
        "id": item.get("id"),
        "trigger": item.get("trigger"),
        "source_kind": item.get("source_kind") or "",
        "source_id": item.get("source_id"),
        "excerpt": item.get("excerpt") or "",
        "local_signals": item.get("local_signals") or {},
        "example_for": item.get("example_for") or [],
    }


def _public_corpus(item: dict) -> dict:
    excerpt = plain_text(item.get("body") or "")[:CORPUS_EXCERPT]
    return {
        "id": item.get("id"),
        "kind": item.get("kind"),
        "occurred_at": item.get("occurred_at"),
        "recency_weight": item.get("recency_weight"),
        "recency_role": item.get("recency_role"),
        "context_hint": item.get("context_hint") or "",
        "facet_hint": item.get("facet_hint") or "",
        "excerpt": excerpt,
    }


def build_package(
    profile_id: str,
    *,
    trigger: str = "explicit_review",
    target: str = "writing",
    mode: str | None = None,
) -> dict:
    if target != "writing":
        raise StoreError("unsupported_target", "Nur Writing-Profile-Review ist gebündelt; Interaction bleibt explizit.")
    ensure_profile(profile_id)
    mode = _review_mode(profile_id, mode)
    if mode == "initial_build":
        with get_db() as conn:
            conn.execute(
                "UPDATE writing_profiles SET lifecycle = 'initial_pending', updated = datetime('now') WHERE profile_id = ?",
                (profile_id,),
            )
    evidences = filter_style_evidence(_pending_evidence(profile_id))
    raw_corpus = list_corpus(profile_id, limit=INITIAL_BUILD_SOURCES)
    corpus = select_corpus_items(raw_corpus, mode=mode)
    profile = get_profile(profile_id)
    facets = profile.get("facets") or []
    core = next((item for item in facets if item.get("facet_key") == CORE_KEY or item.get("layer") == "core"), None)
    others = [item for item in facets if item is not core]
    empty = not corpus and not evidences
    package = {
        "kind": KIND_PACKAGE,
        "format_version": FORMAT_VERSION,
        "target": "writing",
        "mode": mode,
        "instruction": semantic_task(mode),
        "existing_before_new": list(EXISTING_BEFORE_NEW),
        "seed_catalog": seed_catalog(),
        "note": package_note(mode),
        "kansho_evidence_empty": empty,
        "selection": {
            "budget_chars": int(__import__("os").environ.get("KANSHO_PROFILE_PACKAGE_CHARS") or "12000"),
            "role": "representative current, older baseline, distinct contexts — not a full dump",
        },
        "profile": {
            "version": profile.get("version") or 0,
            "governance": profile.get("governance") or "learning",
            "lifecycle": profile.get("lifecycle") or "uninitialized",
            "core": _public_facet(core),
            "facets": [_public_facet(item) for item in others],
            "traits": [_public_trait(item) for item in profile.get("traits") or []],
        },
        "corpus": [_public_corpus(item) for item in corpus],
        "evidences": [_public_evidence(item) for item in evidences],
        "expected_result": expected_result_schema(mode),
        "compiled_brief_excluded": True,
    }
    return package


def render_paste_prompt(package: dict) -> str:
    return render_analysis_prompt(package)


def parse_result(raw) -> dict:
    if isinstance(raw, dict):
        data = raw
    else:
        text = (raw or "").strip()
        if text.startswith("```"):
            text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text, flags=re.I | re.S)
        match = JSON_BLOCK.search(text)
        if not match:
            raise StoreError("invalid_review_result", "Kein JSON-Review-Ergebnis.")
        try:
            data = json.loads(match.group(0))
        except json.JSONDecodeError as exc:
            raise StoreError("invalid_review_result", "Review-Ergebnis ist kein gültiges JSON.") from exc
    if not isinstance(data, dict):
        raise StoreError("invalid_review_result", "Review-Ergebnis muss ein Objekt sein.")
    if data.get("compiled_brief") and not data.get("changes") and not data.get("profile"):
        raise StoreError("invalid_review_result", "Der aktuelle Brief ist kein Profile-Importformat.")
    kind = (data.get("kind") or "").strip()
    if kind not in KIND_RESULTS:
        raise StoreError("invalid_review_result", "kind muss kansho.profile_analysis_result sein.")
    version = data.get("format_version", FORMAT_VERSION)
    try:
        version = int(version)
    except (TypeError, ValueError) as exc:
        raise StoreError("unsupported_format", "format_version ungültig.") from exc
    if version != FORMAT_VERSION:
        raise StoreError("unsupported_format", f"format_version {version} wird nicht unterstützt.")
    mode = normalize_mode(data.get("mode")) or (data.get("mode") or "")
    raw_changes = data.get("changes")
    if raw_changes is None:
        raw_changes = []
    if not isinstance(raw_changes, list):
        raise StoreError("invalid_review_result", "changes muss eine Liste sein.")
    raw_changes = list(raw_changes) + _changes_from_proposal(_proposal_from_result(data))
    cleaned = []
    for item in raw_changes:
        parsed_change = _parse_change(item)
        if parsed_change:
            cleaned.append(parsed_change)
    basis = normalize_evidence_basis(data.get("evidence_basis"))
    if not basis:
        for change in cleaned:
            basis.extend(item for item in change.get("evidence_basis") or [] if item not in basis)
    return {
        "kind": KIND_RESULT,
        "format_version": FORMAT_VERSION,
        "target": "writing",
        "mode": mode,
        "evidence_basis": basis,
        "uncertainties": [
            str(item) for item in (data.get("uncertainties") or []) if item
        ],
        "changes": cleaned,
    }


def _proposal_from_result(data: dict) -> dict:
    nested = data.get("profile") if isinstance(data.get("profile"), dict) else {}
    profile = dict(nested)
    for key in ("core", "facets", "traits"):
        if not profile.get(key) and data.get(key):
            profile[key] = data[key]
    return profile


def _parse_basis(raw) -> list[str]:
    return normalize_evidence_basis(raw)


def _changes_from_proposal(profile: dict) -> list[dict]:
    if not isinstance(profile, dict) or not profile:
        return []
    changes = []
    core = profile.get("core")
    if isinstance(core, dict):
        value = core.get("value") or core.get("statement") or core.get("summary") or ""
        if value:
            changes.append(
                {
                    "layer": "core",
                    "key": CORE_KEY,
                    "action": "update",
                    "proposed_value": value,
                    "rationale": core.get("rationale") or core.get("generalizability") or "Profile Proposal Core",
                    "evidence_basis": core.get("evidence_basis") or [],
                    "exemplars": core.get("exemplars") or [],
                }
            )
    for item in profile.get("facets") or []:
        if not isinstance(item, dict):
            continue
        value = item.get("value") or item.get("statement") or item.get("summary") or ""
        if not value:
            continue
        changes.append(
            {
                "layer": "facet",
                "key": item.get("key") or item.get("facet_key") or JOURNAL_FACET,
                "action": "update",
                "proposed_value": value,
                "rationale": item.get("rationale") or "Profile Proposal Facet",
                "evidence_basis": item.get("evidence_basis") or [],
            }
        )
    for item in profile.get("traits") or []:
        if not isinstance(item, dict):
            continue
        scope = (item.get("scope") or "").strip()
        facet_key = item.get("facet_key") or (CORE_KEY if scope == "core" else "") or JOURNAL_FACET
        changes.append(
            {
                "layer": "trait",
                "slug": item.get("slug") or item.get("key"),
                "facet_key": facet_key,
                "action": item.get("action") or "add",
                "label": item.get("label") or "",
                "proposed_value": item.get("statement") or item.get("proposed_value") or item.get("summary") or "",
                "rationale": item.get("rationale") or "Profile Proposal Trait",
                "evidence_basis": item.get("evidence_basis") or [],
                "exemplars": item.get("exemplars") or [],
                "evidence_ids": item.get("evidence_ids") or [],
            }
        )
    return changes


def _parse_change(item: dict | None) -> dict | None:
    if not isinstance(item, dict):
        return None
    layer = (item.get("layer") or "trait").strip()
    if layer not in LAYERS:
        return None
    action = ACTION_ALIASES.get((item.get("action") or "keep").strip(), (item.get("action") or "keep").strip())
    if action not in TRAIT_ACTIONS:
        return None
    action = ACTION_ALIASES.get(action, action)
    key = (item.get("key") or item.get("slug") or "").strip()
    slug = coerce_slug(item.get("slug") or key or item.get("label") or "trait")
    if layer == "core":
        key = CORE_KEY
        slug = CORE_KEY
    elif layer == "facet":
        key = normalize_facet_key(key) or JOURNAL_FACET
        slug_hint = item.get("slug") or ""
        as_trait = bool(slug_hint) or action == "add" or key in LEGACY_STYLE_KEYS or (
            valid_slug(key) and key not in SEED_FACETS and key != CORE_KEY
        )
        if as_trait:
            layer = "trait"
            slug = coerce_slug(slug_hint or key)
        else:
            slug = coerce_slug(key)
    elif not valid_slug(slug):
        return None
    facet_key = normalize_facet_key(item.get("facet_key") or (key if layer == "facet" else "") or CORE_KEY) or CORE_KEY
    if layer == "core":
        facet_key = CORE_KEY
    exemplars = []
    for ex in item.get("exemplars") or []:
        parsed_ex = _parse_exemplar(ex)
        if parsed_ex:
            exemplars.append(parsed_ex)
    split_into = []
    for part in item.get("split_into") or []:
        if not isinstance(part, dict):
            continue
        part_slug = coerce_slug(part.get("slug") or part.get("label") or "")
        if not part_slug:
            continue
        split_into.append(
            {
                "slug": part_slug,
                "label": part.get("label") or part_slug,
                "statement": part.get("statement") or part.get("proposed_value") or "",
                "facet_key": normalize_facet_key(part.get("facet_key") or facet_key) or facet_key,
            }
        )
    basis = _parse_basis(item.get("evidence_basis"))
    if not basis and (item.get("evidence_ids") or any(ex.get("source_id") for ex in exemplars)):
        basis = ["kansho_sources"]
    if is_external_only(basis):
        for ex in exemplars:
            ex.pop("source_id", None)
    return {
        "layer": layer,
        "key": key,
        "slug": slug,
        "facet_key": facet_key,
        "action": action,
        "proposed_value": item.get("proposed_value") or item.get("statement") or "",
        "label": item.get("label") or "",
        "rationale": item.get("rationale") or "",
        "evidence_ids": [str(eid) for eid in (item.get("evidence_ids") or []) if eid],
        "evidence_basis": basis,
        "exemplars": exemplars,
        "merge_slugs": [
            coerce_slug(value)
            for value in (item.get("merge_slugs") or item.get("merge_ids") or [])
            if value
        ],
        "split_into": split_into,
        "trait_id": item.get("trait_id") or "",
    }


def _parse_exemplar(ex) -> dict | None:
    if isinstance(ex, str) and ex.strip() and not is_meta_style_text(ex):
        return {"excerpt": ex.strip(), "role": "exemplar", "evidence_basis": []}
    if not isinstance(ex, dict):
        return None
    excerpt = (ex.get("excerpt") or "").strip()
    if not excerpt or is_meta_style_text(excerpt):
        return None
    basis = _parse_basis(ex.get("evidence_basis"))
    source_id = ex.get("source_id")
    if is_external_only(basis):
        source_id = None
    return {
        "excerpt": excerpt,
        "role": ex.get("role") or "exemplar",
        "source_id": source_id,
        "occurred_at": ex.get("occurred_at"),
        "evidence_basis": basis or (["kansho_sources"] if source_id else []),
    }


def _store_review(profile_id: str, package: dict, channel: str) -> dict:
    review_id = str(uuid.uuid4())
    with get_db() as conn:
        conn.execute(
            """
            INSERT INTO writing_profile_reviews
              (id, profile_id, target, channel, status, package_json)
            VALUES (?, ?, 'writing', ?, 'open', ?)
            """,
            (review_id, profile_id, channel, json.dumps(package, ensure_ascii=False)),
        )
        ids = [item.get("id") for item in package.get("evidences") or [] if item.get("id")]
        if ids:
            placeholders = ",".join("?" * len(ids))
            conn.execute(
                f"UPDATE writing_profile_evidence SET status = 'bundled' WHERE id IN ({placeholders})",
                ids,
            )
    return {"id": review_id, **review_status(profile_id), "package": package}


def open_paste_review(profile_id: str, *, mode: str | None = None) -> dict:
    package = build_package(profile_id, trigger="explicit_review", mode=mode)
    stored = _store_review(profile_id, package, "paste")
    stored["prompt"] = render_paste_prompt(package)
    return stored


def open_api_review(profile_id: str, *, mode: str | None = None) -> dict:
    package = build_package(profile_id, trigger="explicit_review", mode=mode)
    return _store_review(profile_id, package, "api")


def run_api_review(profile_id: str, review_id: str | None = None, *, mode: str | None = None) -> dict:
    from engine import EngineError, execute_prompt, load_active_prompt

    if review_id:
        with get_db() as conn:
            row = row_to_dict(
                conn.execute(
                    "SELECT * FROM writing_profile_reviews WHERE id = ? AND profile_id = ?",
                    (review_id, profile_id),
                ).fetchone()
            )
        if not row:
            raise StoreError("not_found", "Review nicht gefunden", 404)
        package = _parse_json(row.get("package_json"), {})
        if (row.get("status") or "") not in {"open", "proposed"}:
            raise StoreError("review_closed", "Diese Review ist bereits abgeschlossen.")
    else:
        opened = open_api_review(profile_id, mode=mode)
        review_id = opened["id"]
        package = opened["package"]
    prompt = load_active_prompt("mvp.profile_review")
    try:
        result = execute_prompt(
            prompt,
            profile_id,
            purpose="profile_review",
            data_class="B",
            context={
                "review_package": json.dumps(package, ensure_ascii=False, indent=2),
                "source_text": "\n".join(
                    [item.get("excerpt") or "" for item in package.get("evidences") or []]
                    + [item.get("excerpt") or "" for item in package.get("corpus") or []]
                ),
            },
        )
    except EngineError as exc:
        persist_engine_error(
            profile_id,
            purpose="profile_review",
            exc=exc,
            subject_type="writing_profile_review",
            subject_id=review_id or "",
        )
        raise
    try:
        parsed = parse_result(result.get("content") or "")
    except StoreError as exc:
        diag = dict(result.get("diagnostics") or {})
        if result.get("trace") is not None:
            diag.setdefault("trace", result.get("trace"))
        diag["generate_called"] = True if diag.get("generate_called") is None else diag.get("generate_called")
        diag["generate_requested"] = True
        diag["abort_reason"] = exc.code
        wrapped = EngineError(exc.code, exc.message, getattr(exc, "status_code", 400), diag)
        persist_engine_error(
            profile_id,
            purpose="profile_review",
            exc=wrapped,
            subject_type="writing_profile_review",
            subject_id=review_id or "",
        )
        raise wrapped from exc
    if not parsed.get("mode"):
        parsed["mode"] = package.get("mode") or ""
    staged = _stage_proposal(profile_id, parsed, review_id=review_id, package=package)
    staged["trace"] = result.get("trace")
    staged["review_id"] = review_id
    persist_step(
        profile_id,
        purpose="profile_review",
        status="ok",
        subject_type="writing_profile_review",
        subject_id=review_id or "",
        trace=result.get("trace"),
        extra={"review_id": review_id},
    )
    return staged


def import_result(profile_id: str, raw, review_id: str | None = None) -> dict:
    parsed = parse_result(raw)
    package = {}
    if not review_id:
        with get_db() as conn:
            open_row = row_to_dict(
                conn.execute(
                    """
                    SELECT id, package_json FROM writing_profile_reviews
                    WHERE profile_id = ? AND status IN ('open', 'proposed')
                    ORDER BY created DESC LIMIT 1
                    """,
                    (profile_id,),
                ).fetchone()
            )
        review_id = (open_row or {}).get("id")
        package = _parse_json((open_row or {}).get("package_json"), {})
        if not review_id:
            stored = _store_review(
                profile_id,
                build_package(profile_id, trigger="explicit_review"),
                "paste",
            )
            review_id = stored["id"]
            package = stored.get("package") or {}
    else:
        with get_db() as conn:
            row = row_to_dict(
                conn.execute(
                    "SELECT package_json FROM writing_profile_reviews WHERE id = ? AND profile_id = ?",
                    (review_id, profile_id),
                ).fetchone()
            )
        package = _parse_json((row or {}).get("package_json"), {})
    if not parsed.get("mode"):
        parsed["mode"] = package.get("mode") or normalize_mode(parsed.get("mode")) or _review_mode(profile_id)
    if not parsed.get("changes"):
        raise StoreError(
            "empty_review_result",
            "Keine übernehmbaren Änderungen gefunden. Für initial_build werden core, facets und traits oder eine changes-Liste benötigt.",
        )
    return _stage_proposal(profile_id, parsed, review_id=review_id, package=package)


def _known_ids(package: dict) -> set[str]:
    ids = set()
    for item in (package.get("evidences") or []) + (package.get("corpus") or []):
        if item.get("id"):
            ids.add(str(item["id"]))
        if item.get("source_id"):
            ids.add(str(item["source_id"]))
    return ids


def _unknown_refs(parsed: dict, package: dict) -> list[str]:
    known = _known_ids(package)
    unknown = []
    for change in parsed.get("changes") or []:
        if is_external_only(change.get("evidence_basis")):
            continue
        for eid in change.get("evidence_ids") or []:
            if str(eid) not in known:
                unknown.append(str(eid))
        for ex in change.get("exemplars") or []:
            source_id = ex.get("source_id")
            if source_id and str(source_id) not in known:
                unknown.append(str(source_id))
    return sorted(set(unknown))


def _stage_proposal(profile_id: str, parsed: dict, *, review_id: str, package: dict) -> dict:
    unknown = _unknown_refs(parsed, package)
    current = get_profile(profile_id)
    for index, change in enumerate(parsed.get("changes") or []):
        payload = {**change, "index": index, "unknown_refs": [item for item in unknown if item in (change.get("evidence_ids") or [])]}
        _queue_suggestion(
            profile_id,
            change.get("facet_key") or change.get("key") or "",
            change.get("proposed_value") or "",
            change.get("rationale") or "",
            trait_slug=change.get("slug") or "",
            action=change.get("action") or "update",
            payload=payload,
        )
    with get_db() as conn:
        conn.execute(
            """
            UPDATE writing_profile_reviews
            SET result_json = ?, status = 'proposed', channel = COALESCE(channel, 'paste')
            WHERE id = ? AND profile_id = ?
            """,
            (json.dumps(parsed, ensure_ascii=False), review_id, profile_id),
        )
    profile = get_profile(profile_id)
    suggested = len(parsed.get("changes") or [])
    status = review_status(profile_id)
    return {
        **status,
        "applied": 0,
        "suggested": suggested,
        "skipped": 0,
        "accepted": False,
        "review_id": review_id,
        "mode": parsed.get("mode") or package.get("mode"),
        "unknown_refs": unknown,
        "proposal": parsed,
        "result": parsed,
        "profile": profile,
        "note": f"{suggested} Vorschläge geprüft. Nichts wurde am Current Profile geändert.",
        "current_version": current.get("version") or 0,
    }


def accept_proposal(
    profile_id: str,
    review_id: str | None = None,
    *,
    indexes: list[int] | None = None,
    baseline: bool = False,
) -> dict:
    with get_db() as conn:
        if review_id:
            row = row_to_dict(
                conn.execute(
                    "SELECT * FROM writing_profile_reviews WHERE id = ? AND profile_id = ?",
                    (review_id, profile_id),
                ).fetchone()
            )
        else:
            row = row_to_dict(
                conn.execute(
                    """
                    SELECT * FROM writing_profile_reviews
                    WHERE profile_id = ? AND status = 'proposed'
                    ORDER BY created DESC LIMIT 1
                    """,
                    (profile_id,),
                ).fetchone()
            )
    if not row:
        raise StoreError("not_found", "Kein offenes Profile-Proposal.", 404)
    parsed = parse_result(_parse_json(row.get("result_json"), {}))
    package = _parse_json(row.get("package_json"), {})
    changes = parsed.get("changes") or []
    remaining = []
    if indexes is not None:
        chosen = set(indexes)
        selected = [changes[i] for i in indexes if 0 <= i < len(changes)]
        remaining = [item for i, item in enumerate(changes) if i not in chosen]
        parsed = {**parsed, "changes": selected}
    return apply_result(
        profile_id,
        parsed,
        review_id=row["id"],
        package=package,
        baseline=baseline or ((parsed.get("mode") or package.get("mode")) == "initial_build" and indexes is None),
        user_accepted=True,
        remaining=remaining,
    )


def reject_proposal(profile_id: str, review_id: str | None = None, *, indexes: list[int] | None = None) -> dict:
    with get_db() as conn:
        if review_id:
            row = row_to_dict(
                conn.execute(
                    "SELECT * FROM writing_profile_reviews WHERE id = ? AND profile_id = ?",
                    (review_id, profile_id),
                ).fetchone()
            )
        else:
            row = row_to_dict(
                conn.execute(
                    """
                    SELECT * FROM writing_profile_reviews
                    WHERE profile_id = ? AND status IN ('open', 'proposed')
                    ORDER BY created DESC LIMIT 1
                    """,
                    (profile_id,),
                ).fetchone()
            )
        if not row:
            raise StoreError("not_found", "Kein offenes Profile-Proposal.", 404)
        parsed = parse_result(_parse_json(row.get("result_json"), {})) if row.get("result_json") else {}
        changes = parsed.get("changes") or []
        remaining = []
        if indexes is not None:
            chosen = set(indexes)
            remaining = [item for i, item in enumerate(changes) if i not in chosen]
        if remaining:
            kept = {**parsed, "changes": remaining}
            conn.execute(
                """
                UPDATE writing_profile_reviews
                SET result_json = ?, status = 'proposed'
                WHERE id = ? AND profile_id = ?
                """,
                (json.dumps(kept, ensure_ascii=False), row["id"], profile_id),
            )
            dropped = {item.get("slug") or item.get("key") for i, item in enumerate(changes) if i in set(indexes)}
            for slug in dropped:
                if not slug:
                    continue
                conn.execute(
                    """
                    UPDATE writing_profile_suggestions
                    SET status = 'rejected', resolved = datetime('now')
                    WHERE profile_id = ? AND status = 'pending' AND (trait_slug = ? OR facet_key = ?)
                    """,
                    (profile_id, slug, slug),
                )
            return {
                "accepted": False,
                "applied": 0,
                "review_id": row["id"],
                "proposal": kept,
                "result": kept,
                **review_status(profile_id),
                "profile": get_profile(profile_id),
            }
        conn.execute(
            """
            UPDATE writing_profile_reviews
            SET status = 'dismissed', resolved = datetime('now')
            WHERE id = ? AND profile_id = ?
            """,
            (row["id"], profile_id),
        )
        conn.execute(
            """
            UPDATE writing_profile_suggestions
            SET status = 'rejected', resolved = datetime('now')
            WHERE profile_id = ? AND status = 'pending'
            """,
            (profile_id,),
        )
    return {"accepted": False, "applied": 0, **review_status(profile_id), "profile": get_profile(profile_id)}


def _refs_from_change(change: dict, evidences: list[dict], corpus: list[dict]) -> list[dict]:
    refs = []
    basis = change.get("evidence_basis") or []
    external_only = is_external_only(basis)
    by_id = {item.get("id"): item for item in evidences if item.get("id")}
    by_source = {item.get("id"): item for item in corpus if item.get("id")}
    if not external_only:
        for eid in change.get("evidence_ids") or []:
            item = by_id.get(eid) or by_source.get(eid)
            if not item:
                continue
            excerpt = plain_text(item.get("excerpt") or item.get("body") or "")
            if not excerpt or is_meta_style_text(excerpt):
                continue
            refs.append(
                {
                    "role": "evidence",
                    "excerpt": excerpt,
                    "source_id": item.get("source_id") or item.get("id"),
                    "occurred_at": item.get("occurred_at"),
                }
            )
    for item in change.get("exemplars") or []:
        excerpt = plain_text(item.get("excerpt") or "")
        if not excerpt or is_meta_style_text(excerpt):
            continue
        source_id = None if external_only else item.get("source_id")
        refs.append(
            {
                "role": item.get("role") or "exemplar",
                "excerpt": excerpt,
                "source_id": source_id,
                "occurred_at": item.get("occurred_at"),
            }
        )
    return refs


def _apply_change(
    profile_id: str,
    change: dict,
    *,
    origin: str,
    force: bool,
    allow_core: bool,
    evidences: list[dict],
    corpus: list[dict],
) -> str:
    action = APPLY_ACTIONS.get(change.get("action") or "keep", change.get("action") or "confirm")
    layer = change.get("layer") or "trait"
    value = (change.get("proposed_value") or "").strip()
    rationale = (change.get("rationale") or "AI Review").strip()
    refs = _refs_from_change(change, evidences, corpus)
    if layer == "core":
        if not allow_core:
            return "skipped"
        if action in {"confirm", "keep"} and not value:
            return "applied"
        if not value:
            return "skipped"
        return _upsert_facet(
            profile_id,
            CORE_KEY,
            value,
            evidence=rationale,
            origin=origin,
            force=force,
        )
    if layer == "facet" and action not in {"create", "split", "merge"}:
        key = normalize_facet_key(change.get("key") or change.get("facet_key") or JOURNAL_FACET)
        if not value and action in {"confirm", "keep"}:
            return "applied"
        if not value:
            return "skipped"
        return _upsert_facet(profile_id, key, value, evidence=rationale, origin=origin, force=force)

    slug = coerce_slug(change.get("slug") or change.get("key") or "trait")
    facet_key = normalize_facet_key(change.get("facet_key") or JOURNAL_FACET) or JOURNAL_FACET
    if action == "remove":
        retire_trait(profile_id, slug, "retired")
        return "applied"
    if action == "create" and not value:
        return "skipped"
    if action == "rescope":
        upsert_trait(
            profile_id,
            slug=slug,
            facet_key=facet_key,
            label=change.get("label") or slug,
            statement=value or None,
            origin=origin,
            force=force,
        )
        if refs:
            replace_trait_refs(profile_id, slug, refs)
        return "applied"
    if action == "merge":
        upsert_trait(
            profile_id,
            slug=slug,
            facet_key=facet_key,
            label=change.get("label") or slug,
            statement=value or None,
            origin=origin,
            force=force,
        )
        if refs:
            replace_trait_refs(profile_id, slug, refs)
        for other in change.get("merge_slugs") or []:
            if other and other != slug:
                retire_trait(profile_id, other, "merged")
        return "applied"
    if action == "split":
        created = 0
        for part in change.get("split_into") or []:
            upsert_trait(
                profile_id,
                slug=part["slug"],
                facet_key=part.get("facet_key") or facet_key,
                label=part.get("label") or part["slug"],
                statement=part.get("statement") or "",
                origin=origin,
                force=force,
            )
            created += 1
        if created:
            retire_trait(profile_id, slug, "split")
            return "applied"
        return "skipped"
    if action in {"confirm", "precisify", "create", "keep", "update"}:
        if action == "confirm" and not value:
            if refs:
                replace_trait_refs(profile_id, slug, refs)
            return "applied"
        outcome = upsert_trait(
            profile_id,
            slug=slug,
            facet_key=facet_key,
            label=change.get("label") or slug,
            statement=value,
            origin=origin,
            force=force,
        )
        if refs:
            replace_trait_refs(profile_id, slug, refs)
        return outcome
    return "skipped"


def apply_result(
    profile_id: str,
    result: dict,
    review_id: str | None = None,
    package: dict | None = None,
    *,
    baseline: bool = False,
    user_accepted: bool = False,
    remaining: list[dict] | None = None,
) -> dict:
    if not user_accepted:
        raise StoreError("proposal_not_accepted", "Profile Proposal wird erst nach Nutzerbestätigung übernommen.")
    parsed = parse_result(result)
    row = ensure_profile(profile_id)
    governance = row.get("governance") or "learning"
    if governance not in GOVERNANCE:
        governance = "learning"
    mode = normalize_mode(parsed.get("mode") or (package or {}).get("mode")) or _review_mode(profile_id)
    origin = "initial_build" if mode == "initial_build" or baseline else "accepted_suggestion"
    corpus = list_corpus(profile_id, limit=INITIAL_BUILD_SOURCES)
    evidences = list((package or {}).get("evidences") or []) + _pending_evidence(profile_id, limit=40)
    allow_core = _corpus_supports_core(corpus)
    leftover = list(remaining or [])
    close_review = not leftover
    snapshot_version(profile_id, cause="before_review")
    applied = 0
    skipped = 0
    for change in parsed.get("changes") or []:
        if (row.get("governance") or "") == "frozen" and not baseline:
            skipped += 1
            continue
        outcome = _apply_change(
            profile_id,
            change,
            origin=origin,
            force=True,
            allow_core=allow_core or baseline,
            evidences=evidences,
            corpus=corpus,
        )
        if outcome == "applied":
            applied += 1
        else:
            skipped += 1
    _assemble_brief(profile_id)
    if applied:
        snapshot_version(profile_id, cause="accepted_review")
    if close_review and (mode == "initial_build" or baseline) and applied:
        with get_db() as conn:
            conn.execute(
                """
                UPDATE writing_profiles
                SET lifecycle = 'confirmed', updated = datetime('now')
                WHERE profile_id = ?
                """,
                (profile_id,),
            )
        _assemble_brief(profile_id)
    evidence_ids = [
        eid
        for change in parsed.get("changes") or []
        for eid in change.get("evidence_ids") or []
    ]
    applied_slugs = [change.get("slug") or change.get("key") for change in parsed.get("changes") or []]
    leftover_doc = {**parsed, "changes": leftover} if leftover else None
    with get_db() as conn:
        if close_review:
            conn.execute(
                "UPDATE writing_profiles SET last_reviewed = datetime('now'), review_ready = 0 WHERE profile_id = ?",
                (profile_id,),
            )
        if evidence_ids:
            placeholders = ",".join("?" * len(evidence_ids))
            conn.execute(
                f"UPDATE writing_profile_evidence SET status = 'consumed' WHERE profile_id = ? AND id IN ({placeholders})",
                (profile_id, *evidence_ids),
            )
        elif close_review:
            conn.execute(
                "UPDATE writing_profile_evidence SET status = 'consumed' WHERE profile_id = ? AND status IN ('bundled', 'pending')",
                (profile_id,),
            )
        if review_id and leftover_doc:
            conn.execute(
                """
                UPDATE writing_profile_reviews
                SET result_json = ?, status = 'proposed'
                WHERE id = ? AND profile_id = ?
                """,
                (json.dumps(leftover_doc, ensure_ascii=False), review_id, profile_id),
            )
        elif review_id:
            conn.execute(
                """
                UPDATE writing_profile_reviews
                SET status = 'applied', resolved = datetime('now')
                WHERE id = ? AND profile_id = ?
                """,
                (review_id, profile_id),
            )
        if close_review:
            conn.execute(
                """
                UPDATE writing_profile_suggestions
                SET status = 'accepted', resolved = datetime('now')
                WHERE profile_id = ? AND status = 'pending'
                """,
                (profile_id,),
            )
        else:
            for slug in applied_slugs:
                if not slug:
                    continue
                conn.execute(
                    """
                    UPDATE writing_profile_suggestions
                    SET status = 'accepted', resolved = datetime('now')
                    WHERE profile_id = ? AND status = 'pending' AND (trait_slug = ? OR facet_key = ?)
                    """,
                    (profile_id, slug, slug),
                )
    _refresh_ready(profile_id)
    profile = get_profile(profile_id)
    return {
        "governance": governance,
        "mode": mode,
        "applied": applied,
        "suggested": len(leftover),
        "skipped": skipped,
        "accepted": True,
        "baseline": baseline,
        "review_id": review_id,
        "proposal": leftover_doc,
        "result": leftover_doc or parsed,
        "profile": profile,
        **review_status(profile_id),
    }