"""Vendor-neutral memory portability service.

One module, six operations:

  * `list_starter_packs`         — read `server/starter_packs/index.json`
  * `import_starter_pack`        — ingest a bundled pack into a new subject
  * `reseed_support_subject`     — rebuild `statewave-support-docs` from the
                                   bundled `statewave-support-agent` pack
  * `clone_subject`              — copy episodes/memories from one subject
                                   to another with provenance metadata
  * `export_memory_payload`      — produce a versioned JSON payload that
                                   the admin client encrypts client-side
  * `import_memory_payload`      — ingest a payload (after the admin client
                                   has decrypted it) with strict validation

These primitives back every memory action in the admin UI. There is one
generic ingest path (`_ingest_records`) reused by starter-pack import,
clone, and `.swmem` import — per the spec's "do not invent separate one-off
pipelines."

No memory content is logged anywhere in this module. Log lines carry
subject ids, counts, and pack ids only.
"""

from __future__ import annotations

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Literal

import structlog
from sqlalchemy import delete, func, select

from server.core.config import settings
from server.core.identifiers import SUBJECT_ID_CHARSET_DESC, SUBJECT_ID_RE
from server.db import engine as engine_module
from server.db.tables import EpisodeRow, MemoryRow

logger = structlog.stdlib.get_logger()

PAYLOAD_FORMAT = "statewave-memory-payload"
PAYLOAD_FORMAT_VERSION = 1
STARTER_PACK_FORMAT = "statewave-starter-pack"
STARTER_PACK_INDEX_FORMAT = "statewave-starter-pack-index"

CloneScope = Literal[
    "episodes",
    "memories",
    "episodes_and_memories",
    "episodes_memories_sources",
]
ConflictStrategy = Literal["create_copy", "merge", "cancel"]

_STARTER_PACK_KIND = Literal["support_docs", "demo_agent"]

_PACKS_ROOT = Path(__file__).resolve().parent.parent / "starter_packs"

# Subject-id contract is now defined once in server.core.identifiers and
# enforced at every write/query ingress (issue #121) — reuse it here so the
# pack path can never drift from the rest of the API. On top of the shared
# charset we additionally reject the reserved prefixes used by the marketing
# widget's per-visitor subjects (`demo_web_*`).
_SUBJECT_ID_RE = SUBJECT_ID_RE
_RESERVED_PREFIXES = ("demo_web_",)


class StarterPackError(Exception):
    """Domain error with an attached HTTP status hint for the API layer."""

    def __init__(self, message: str, *, status_code: int = 400):
        super().__init__(message)
        self.status_code = status_code


# ─── Helpers ────────────────────────────────────────────────────────────────


def _validate_subject_id(
    value: str,
    *,
    field: str = "subject_id",
    allow_reserved: bool = False,
) -> str:
    """Validate a subject id.

    `allow_reserved` exists because the reserved-prefix guard is about
    preventing operators from CREATING new subjects in the marketing
    widget's namespace (`demo_web_*`). Reading from such a subject —
    cloning it, exporting it, or otherwise inspecting it — is a legitimate
    operator action and shouldn't be blocked. Source-side call sites
    therefore set `allow_reserved=True`; target-side call sites keep the
    default and the guard fires.
    """
    if not value or not _SUBJECT_ID_RE.match(value):
        raise StarterPackError(
            f"{field!r} must be {SUBJECT_ID_CHARSET_DESC}.",
            status_code=400,
        )
    if not allow_reserved and any(value.startswith(p) for p in _RESERVED_PREFIXES):
        raise StarterPackError(
            f"{field!r} uses a reserved prefix ({', '.join(_RESERVED_PREFIXES)}).",
            status_code=400,
        )
    return value


def _suggest_target_subject_id(base: str) -> str:
    """Derive a fresh subject id from a suggested base.

    If `base` carries a reserved prefix (e.g. `demo_web_`), the prefix is
    rewritten to `imported-` so the new id can never land back in the
    marketing widget's namespace. This matters most on `.swmem` import:
    archives generated from a per-visitor subject carry the original
    `demo_web_<uuid>__support-agent` id, and we want a clean
    `imported-<uuid>__support-agent-<hex>` target instead of a collision.
    """
    base = base or "imported-subject"
    for prefix in _RESERVED_PREFIXES:
        if base.startswith(prefix):
            base = "imported-" + base[len(prefix):]
            break
    suffix = uuid.uuid4().hex[:8]
    candidate = f"{base}-{suffix}"
    return _validate_subject_id(candidate)


async def _count_subject(subject_id: str) -> tuple[int, int]:
    """Return (episode_count, memory_count) for a subject. 0/0 if missing."""
    async with engine_module.get_session_factory()() as session:
        ep = await session.scalar(
            select(func.count()).select_from(EpisodeRow).where(EpisodeRow.subject_id == subject_id)
        )
        mem = await session.scalar(
            select(func.count()).select_from(MemoryRow).where(MemoryRow.subject_id == subject_id)
        )
        return int(ep or 0), int(mem or 0)


def _load_jsonl(path: Path) -> list[dict[str, Any]]:
    if not path.exists():
        return []
    out: list[dict[str, Any]] = []
    with path.open("r", encoding="utf-8") as f:
        for raw in f:
            raw = raw.strip()
            if not raw:
                continue
            out.append(json.loads(raw))
    return out


def _read_starter_pack_manifest(directory: Path) -> dict[str, Any]:
    manifest_path = directory / "manifest.json"
    if not manifest_path.exists():
        raise StarterPackError(
            f"Starter pack at {directory.name!r} is missing manifest.json.",
            status_code=500,
        )
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    if manifest.get("format") != STARTER_PACK_FORMAT:
        raise StarterPackError(
            f"Starter pack {directory.name!r} has unexpected format "
            f"{manifest.get('format')!r}; expected {STARTER_PACK_FORMAT!r}.",
            status_code=500,
        )
    return manifest


def _starter_pack_dir(pack_id: str) -> tuple[Path, dict[str, Any]]:
    """Resolve `pack_id` to a directory + manifest. Raises 404 on miss."""
    index = _read_index()
    entry = next((p for p in index if p["pack_id"] == pack_id), None)
    if not entry:
        raise StarterPackError(
            f"Starter pack {pack_id!r} not found.", status_code=404
        )
    directory = _PACKS_ROOT / entry["directory"]
    manifest = _read_starter_pack_manifest(directory)
    return directory, manifest


def _read_index() -> list[dict[str, Any]]:
    index_path = _PACKS_ROOT / "index.json"
    if not index_path.exists():
        return []
    raw = json.loads(index_path.read_text(encoding="utf-8"))
    if raw.get("format") != STARTER_PACK_INDEX_FORMAT:
        raise StarterPackError(
            "Starter pack index has unexpected format.", status_code=500
        )
    return list(raw.get("packs", []))


# ─── Public: list starter packs ──────────────────────────────────────────────


def list_starter_packs() -> list[dict[str, Any]]:
    """Return manifest metadata for every bundled starter pack.

    Pack content (episodes/memories) is NOT included — only metadata, so
    the admin UI can render selectable cards. Manifest read errors on a
    single pack become a `__error` field on that entry rather than failing
    the whole list, so a corrupt pack can't break the picker.

    The `__error` field carries a stable, sanitized error code rather than
    the raw exception message. The full exception detail is logged
    server-side via `logger.warning(..., exc_info=True)` so operators still
    get the diagnostic info, but the API response can't accidentally leak
    filesystem paths, JSON parser positions, or future error-message text
    that may grow more sensitive over time (CodeQL py/stack-trace-exposure,
    GH alert #1). The widened `except Exception` also catches
    `json.JSONDecodeError` and `OSError` from the manifest read — both of
    which previously bubbled past the narrower `except StarterPackError`
    and broke the per-pack soft-fail contract this function promises.
    """
    out: list[dict[str, Any]] = []
    for entry in _read_index():
        directory = _PACKS_ROOT / entry["directory"]
        try:
            manifest = _read_starter_pack_manifest(directory)
        except Exception:
            logger.warning(
                "starter_pack_manifest_unreadable",
                pack_id=entry.get("pack_id"),
                directory=entry.get("directory"),
                exc_info=True,
            )
            out.append(
                {
                    "pack_id": entry["pack_id"],
                    "kind": entry.get("kind"),
                    "__error": "manifest_unreadable",
                }
            )
            continue
        out.append(
            {
                "pack_id": manifest["pack_id"],
                "kind": entry.get("kind"),
                "display_name": manifest.get("display_name"),
                "description": manifest.get("description"),
                "version": manifest.get("version"),
                "created_at": manifest.get("created_at"),
                "subject_id_suggestion": manifest.get("subject_id_suggestion"),
                "episode_count": manifest.get("episode_count", 0),
                "memory_count": manifest.get("memory_count", 0),
                "source_count": manifest.get("source_count", 0),
                "tags": manifest.get("tags", []),
            }
        )
    return out


# ─── Internal: ingest pipeline shared by starter-pack/clone/import ──────────


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


async def _ingest_records_async(
    *,
    target_subject_id: str,
    target_tenant_id: str | None,
    episodes_data: Iterable[dict[str, Any]],
    memories_data: Iterable[dict[str, Any]],
    extra_provenance: dict[str, Any],
    extra_metadata: dict[str, Any],
    wait_for_embeddings: bool = False,
) -> dict[str, int]:
    eps = list(episodes_data)
    mems = list(memories_data)
    if len(eps) > settings.memory_import_max_episodes:
        raise StarterPackError(
            f"Too many episodes: {len(eps)} > limit {settings.memory_import_max_episodes}.",
            status_code=413,
        )
    if len(mems) > settings.memory_import_max_memories:
        raise StarterPackError(
            f"Too many memories: {len(mems)} > limit {settings.memory_import_max_memories}.",
            status_code=413,
        )

    # Pre-baked packs (bundled starter packs, .swmem exports, clones) ship
    # compiled memories alongside their source episodes. Without this guard
    # the first /v1/memories/compile after import re-runs the LLM compiler
    # over every imported episode — adding ~15s to a fresh visitor's first
    # chat for no net memory gain. When memories are imported in the same
    # batch we treat the episodes as already compiled by default; an
    # explicit `last_compiled_at` in the input still wins.
    pack_is_pre_compiled = bool(mems)

    id_map: dict[str, str] = {}
    async with engine_module.get_session_factory()() as session:
        for ep in eps:
            old_id = ep.get("id") or ep.get("original_episode_id")
            new_id = uuid.uuid4()
            if old_id:
                id_map[str(old_id)] = str(new_id)
            metadata = dict(ep.get("metadata") or {})
            metadata.update(extra_metadata)
            provenance = dict(ep.get("provenance") or {})
            provenance.update(extra_provenance)
            ep_created_dt = _parse_iso_or_now(ep.get("created_at"))
            # Packs created before migration 0015 don't carry occurred_at;
            # fall back to created_at so restored episodes still land in a
            # sensible chronological position.
            ep_occurred_dt = (
                _parse_iso_or_now(ep["occurred_at"]) if ep.get("occurred_at") else ep_created_dt
            )
            ep_last_compiled_dt = _parse_iso_or_none(ep.get("last_compiled_at"))
            if ep_last_compiled_dt is None and pack_is_pre_compiled:
                ep_last_compiled_dt = ep_created_dt
            row = EpisodeRow(
                id=new_id,
                subject_id=target_subject_id,
                tenant_id=target_tenant_id,
                source=ep.get("source") or "memory-pack",
                type=ep.get("type") or "imported",
                payload=ep.get("payload") or {},
                metadata_=metadata,
                provenance=provenance,
                occurred_at=ep_occurred_dt,
                created_at=ep_created_dt,
                last_compiled_at=ep_last_compiled_dt,
            )
            session.add(row)

        # Track memories that arrived without an embedding so we can
        # schedule background embedding generation after the commit.
        # Bundled starter packs and most operator-supplied .swmem archives
        # ship without embeddings; without this step the import path
        # would silently leave every imported memory with `embedding IS
        # NULL`, breaking vector retrieval until a manual backfill.
        memories_needing_embedding: list[tuple[uuid.UUID, str]] = []

        for mem in mems:
            new_id = uuid.uuid4()
            source_eps: list[uuid.UUID] = []
            for raw in mem.get("source_episode_ids") or []:
                raw_str = str(raw)
                mapped = id_map.get(raw_str, raw_str)
                try:
                    source_eps.append(uuid.UUID(mapped))
                except (ValueError, TypeError):
                    continue
            metadata = dict(mem.get("metadata") or {})
            metadata.update(extra_metadata)
            content = mem.get("content") or ""
            embedding = mem.get("embedding")
            row = MemoryRow(
                id=new_id,
                subject_id=target_subject_id,
                tenant_id=target_tenant_id,
                kind=mem.get("kind") or "fact",
                content=content,
                summary=mem.get("summary") or content,
                confidence=float(mem.get("confidence", 0.9)),
                valid_from=_parse_iso_or_now(mem.get("valid_from")),
                valid_to=_parse_iso_or_none(mem.get("valid_to")),
                source_episode_ids=source_eps,
                metadata_=dict(metadata),
                status=mem.get("status") or "active",
                embedding=embedding,
                created_at=_parse_iso_or_now(mem.get("created_at")),
                updated_at=_parse_iso_or_now(mem.get("updated_at")),
            )
            session.add(row)
            if embedding is None and content:
                memories_needing_embedding.append((new_id, content))

        await session.commit()

    if memories_needing_embedding:
        # Default: fire-and-forget — must not block large operator-supplied
        # imports. The shared scheduler is the same one the compile path
        # uses, so both write paths produce embedded memories with the same
        # guarantees.
        #
        # `wait_for_embeddings=True`: caller (e.g. the marketing demo seed)
        # blocks until the embeddings are persisted. Without this, a fast
        # visitor's first `/v1/context` lookup can race the backfill and
        # land on a corpus with NULL embeddings — the vector search returns
        # nothing, retrieval falls back to recency-only, and the answer
        # quality on the very first turn is visibly worse than every turn
        # after. The wait runs against batched embeddings (one provider
        # call per import), so it adds ~0.5–2s for a typical starter pack.
        from server.services.embeddings.backfill import (
            generate_embeddings_background,
            schedule_embedding_backfill,
        )

        ids = [m[0] for m in memories_needing_embedding]
        texts = [m[1] for m in memories_needing_embedding]
        if wait_for_embeddings:
            await generate_embeddings_background(ids, texts)
        else:
            schedule_embedding_backfill(ids, texts)

    return {"episodes": len(eps), "memories": len(mems)}


def _parse_iso_or_now(value: Any) -> datetime:
    if not value:
        return datetime.now(timezone.utc)
    if isinstance(value, datetime):
        return value
    try:
        return datetime.fromisoformat(str(value))
    except ValueError:
        return datetime.now(timezone.utc)


def _parse_iso_or_none(value: Any) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        return value
    try:
        return datetime.fromisoformat(str(value))
    except ValueError:
        return None


# ─── Public: starter pack import ────────────────────────────────────────────


async def import_starter_pack(
    *,
    pack_id: str,
    target_subject_id: str | None,
    target_display_name: str | None,
    target_tenant_id: str | None,
    conflict_strategy: ConflictStrategy = "create_copy",
    allow_reserved_target: bool = False,
) -> dict[str, Any]:
    """Import a starter pack into a target subject.

    `allow_reserved_target` opts the caller into landing the pack on a
    subject inside a reserved namespace (currently `demo_web_*`). The
    default `False` matches the operator-facing admin UI's policy: an
    admin who clicks "Import" should never accidentally fork into the
    marketing widget's per-visitor namespace. The marketing edge
    function — which is the legitimate caller for `demo_web_*` targets,
    seeding a visitor's subject from the bundled showcase pack — sets
    this to `True` to skip the guard.
    """
    directory, manifest = _starter_pack_dir(pack_id)
    suggested = manifest.get("subject_id_suggestion") or pack_id

    if target_subject_id:
        target_subject_id = _validate_subject_id(
            target_subject_id,
            field="target_subject_id",
            allow_reserved=allow_reserved_target,
        )

    final_subject_id = await _resolve_target_subject(
        explicit=target_subject_id,
        suggested=suggested,
        conflict_strategy=conflict_strategy,
    )

    episodes_data = _load_jsonl(directory / "episodes.jsonl")
    memories_data = _load_jsonl(directory / "memories.jsonl")

    extra_metadata = {
        "starter_pack_id": pack_id,
        "starter_pack_version": manifest.get("version"),
        "imported_at": _now_iso(),
    }
    if target_display_name:
        extra_metadata["display_name"] = target_display_name

    counts = await _ingest_records_async(
        target_subject_id=final_subject_id,
        target_tenant_id=target_tenant_id,
        episodes_data=episodes_data,
        memories_data=memories_data,
        extra_provenance={"starter_pack_id": pack_id},
        extra_metadata=extra_metadata,
        # Block on embedding backfill so the visitor's first /v1/context
        # call hits a fully-embedded corpus. Starter packs are bounded
        # (~60 memories) so the added wait is small relative to the rest
        # of the seed step (which the client already shows a spinner for).
        wait_for_embeddings=True,
    )

    logger.info(
        "starter_pack_imported",
        pack_id=pack_id,
        target_subject_id=final_subject_id,
        episodes=counts["episodes"],
        memories=counts["memories"],
        conflict_strategy=conflict_strategy,
    )

    return {
        "pack_id": pack_id,
        "target_subject_id": final_subject_id,
        "imported_episodes": counts["episodes"],
        "imported_memories": counts["memories"],
        "imported_sources": 0,
        "conflict_strategy": conflict_strategy,
        "imported_at": _now_iso(),
    }


async def _resolve_target_subject(
    *,
    explicit: str | None,
    suggested: str,
    conflict_strategy: ConflictStrategy,
) -> str:
    """Pick the final subject id given the conflict strategy."""
    if explicit:
        ep_count, mem_count = await _count_subject(explicit)
        already_populated = ep_count > 0 or mem_count > 0
        if not already_populated:
            return explicit
        if conflict_strategy == "merge":
            return explicit
        if conflict_strategy == "cancel":
            raise StarterPackError(
                f"Subject {explicit!r} already has data ({ep_count} episodes, "
                f"{mem_count} memories). Pick a different target id, choose "
                "'merge', or delete the existing subject first.",
                status_code=409,
            )
        # create_copy: pick a new id derived from the explicit one
        return _suggest_target_subject_id(explicit)

    # No explicit id: derive from the suggested base.
    #
    # If `suggested` lives in a reserved namespace (e.g. an archive
    # captured from a `demo_web_*` visitor subject), we MUST go through
    # `_suggest_target_subject_id` so the prefix gets rewritten — using
    # the suggested id verbatim would land the new subject back in the
    # marketing widget's namespace.
    if any(suggested.startswith(p) for p in _RESERVED_PREFIXES):
        return _suggest_target_subject_id(suggested)

    ep_count, mem_count = await _count_subject(suggested)
    if ep_count == 0 and mem_count == 0:
        return _validate_subject_id(suggested)
    return _suggest_target_subject_id(suggested)


# ─── Public: support reseed (vendor-neutral) ────────────────────────────────


async def get_support_subject_state() -> dict[str, Any]:
    """Snapshot the support subject's reseed metadata for the admin drawer.

    Returns the bundled manifest version and what the live subject reports
    about its last reseed (version, timestamp, reason, episode + memory
    counts). Distinguishes pack-owned rows from operator-added rows so the
    UI can warn before any destructive action.

    Owned-row detection: rows carry `metadata_.starter_pack_id == pack_id`
    when imported via this code path. Older bootstrap_docs_pack runs (pre-
    versioning) wrote `metadata.pack == "statewave-support-docs"` instead;
    we recognise both so a long-lived deployment doesn't think its own
    docs pack is operator data.
    """
    from sqlalchemy import func, or_, select

    target = settings.support_subject_id
    pack_id = settings.support_starter_pack_id

    _, manifest = _starter_pack_dir(pack_id)
    bundled_version = manifest.get("version")

    owned_predicate_ep = or_(
        EpisodeRow.metadata_["starter_pack_id"].astext == pack_id,
        EpisodeRow.metadata_["pack"].astext == "statewave-support-docs",
    )
    owned_predicate_mem = or_(
        MemoryRow.metadata_["starter_pack_id"].astext == pack_id,
        MemoryRow.metadata_["pack"].astext == "statewave-support-docs",
    )

    async with engine_module.get_session_factory()() as session:
        # Total rows on the subject (includes operator-added).
        total_eps = await session.scalar(
            select(func.count(EpisodeRow.id)).where(EpisodeRow.subject_id == target)
        )
        total_mems = await session.scalar(
            select(func.count(MemoryRow.id)).where(MemoryRow.subject_id == target)
        )
        # Rows the auto-update path is allowed to touch.
        owned_eps = await session.scalar(
            select(func.count(EpisodeRow.id)).where(
                EpisodeRow.subject_id == target, owned_predicate_ep
            )
        )
        owned_mems = await session.scalar(
            select(func.count(MemoryRow.id)).where(
                MemoryRow.subject_id == target, owned_predicate_mem
            )
        )
        # Most recent reseed identity. Memories tend to outnumber episodes,
        # but they all carry the same metadata — query whichever has rows.
        last_row = await session.execute(
            select(
                EpisodeRow.metadata_["starter_pack_version"].astext,
                EpisodeRow.metadata_["imported_at"].astext,
                EpisodeRow.metadata_["reseed_reason"].astext,
            )
            .where(EpisodeRow.subject_id == target, owned_predicate_ep)
            .order_by(EpisodeRow.created_at.desc())
            .limit(1)
        )
        last = last_row.first()

    installed_version = last[0] if last else None
    last_imported_at = last[1] if last else None
    last_reseed_reason = last[2] if last else None

    return {
        "subject_id": target,
        "pack_id": pack_id,
        "bundled_version": bundled_version,
        "installed_version": installed_version,
        "is_up_to_date": (
            installed_version is not None and installed_version == bundled_version
        ),
        "total_episode_count": total_eps or 0,
        "total_memory_count": total_mems or 0,
        "owned_episode_count": owned_eps or 0,
        "owned_memory_count": owned_mems or 0,
        "operator_episode_count": (total_eps or 0) - (owned_eps or 0),
        "operator_memory_count": (total_mems or 0) - (owned_mems or 0),
        "last_reseed": {
            "imported_at": last_imported_at,
            "reason": last_reseed_reason,
            "version": installed_version,
        },
    }


async def reseed_support_subject(
    *,
    reason: str | None = None,
    force: bool = False,
) -> dict[str, Any]:
    """Rebuild the shared `statewave-support-docs` subject from the bundled pack.

    Version-aware: skips when the live subject is already on the bundled
    pack's version (no-op fast path for container-restart auto-update).
    Pass `force=True` to reseed regardless of the version check — that's
    what the drawer's "Restore" button does when an operator wants to
    reset to the bundled state.

    Selective purge: only rows whose metadata identifies them as belonging
    to this pack are deleted before reimport. Rows added by an operator
    alongside ours (no `starter_pack_id` metadata) survive untouched. Any
    legacy rows from older `bootstrap_docs_pack` runs that wrote
    `metadata.pack == "statewave-support-docs"` are also recognised as
    owned, so a long-lived deployment can upgrade without losing its docs
    pack to the operator-data exclusion.

    Per-visitor `demo_web_*__statewave-support` subjects are never
    touched — the strict subject-id match scopes everything to the
    configured shared subject (`settings.support_subject_id`).

    Always logs the structured outcome including the reason, so an
    operator searching support-side telemetry can find why a particular
    reseed happened.
    """
    from sqlalchemy import or_

    target = settings.support_subject_id
    pack_id = settings.support_starter_pack_id

    directory, manifest = _starter_pack_dir(pack_id)
    bundled_version = manifest.get("version")

    state = await get_support_subject_state()
    installed_version = state["installed_version"]

    # Version-aware short-circuit: identical version + no force → no-op.
    # We still write a log line so operator-triggered no-ops are visible
    # in support telemetry.
    if not force and state["is_up_to_date"]:
        logger.info(
            "support_reseed_skipped",
            target_subject_id=target,
            pack_id=pack_id,
            installed_version=installed_version,
            bundled_version=bundled_version,
            reason=reason,
            outcome="already_current",
        )
        return {
            "subject_id": target,
            "pack_id": pack_id,
            "pack_version": bundled_version,
            "installed_version": installed_version,
            "imported_episodes": 0,
            "imported_memories": 0,
            "reseeded_at": None,
            "reason": reason,
            "updated": False,
            "outcome": "already_current",
        }

    episodes_data = _load_jsonl(directory / "episodes.jsonl")
    memories_data = _load_jsonl(directory / "memories.jsonl")

    # Selective purge: only delete rows our pack owns. Operator-added
    # episodes and memories on the same subject (rare but possible — e.g.
    # an internal note appended for the support team) are preserved. The
    # owned predicate matches both the modern `starter_pack_id` provenance
    # and the legacy `pack="statewave-support-docs"` metadata so an upgrade
    # from a pre-versioning install doesn't strand its own docs pack.
    owned_predicate_ep = or_(
        EpisodeRow.metadata_["starter_pack_id"].astext == pack_id,
        EpisodeRow.metadata_["pack"].astext == "statewave-support-docs",
    )
    owned_predicate_mem = or_(
        MemoryRow.metadata_["starter_pack_id"].astext == pack_id,
        MemoryRow.metadata_["pack"].astext == "statewave-support-docs",
    )

    async with engine_module.get_session_factory()() as session:
        await session.execute(
            delete(EpisodeRow).where(
                EpisodeRow.subject_id == target, owned_predicate_ep
            )
        )
        await session.execute(
            delete(MemoryRow).where(
                MemoryRow.subject_id == target, owned_predicate_mem
            )
        )
        await session.commit()

    counts = await _ingest_records_async(
        target_subject_id=target,
        target_tenant_id=None,
        episodes_data=episodes_data,
        memories_data=memories_data,
        extra_provenance={"starter_pack_id": pack_id, "support_reseed": True},
        extra_metadata={
            "starter_pack_id": pack_id,
            "starter_pack_version": bundled_version,
            "imported_at": _now_iso(),
            "reseed_reason": reason,
        },
    )

    outcome = (
        "force_reseeded"
        if force
        else ("seeded" if installed_version is None else "auto_updated")
    )
    logger.info(
        "support_reseed_completed",
        target_subject_id=target,
        pack_id=pack_id,
        installed_version_before=installed_version,
        installed_version_after=bundled_version,
        episodes=counts["episodes"],
        memories=counts["memories"],
        operator_episodes_preserved=state["operator_episode_count"],
        operator_memories_preserved=state["operator_memory_count"],
        reason=reason,
        outcome=outcome,
    )

    return {
        "subject_id": target,
        "pack_id": pack_id,
        "pack_version": bundled_version,
        "installed_version": bundled_version,
        "previous_version": installed_version,
        "imported_episodes": counts["episodes"],
        "imported_memories": counts["memories"],
        "operator_episodes_preserved": state["operator_episode_count"],
        "operator_memories_preserved": state["operator_memory_count"],
        "reseeded_at": _now_iso(),
        "reason": reason,
        "updated": True,
        "outcome": outcome,
    }


# ─── Public: clone subject ──────────────────────────────────────────────────


async def clone_subject(
    *,
    source_subject_id: str,
    target_subject_id: str | None,
    target_display_name: str | None,
    target_tenant_id: str | None,
    clone_scope: CloneScope = "episodes_memories_sources",
    cloned_by: str | None = None,
) -> dict[str, Any]:
    """Clone a subject's records into a new subject.

    Reads the source through the existing `export_subject` primitive and
    routes the result through the same `_ingest_records_async` ingest
    pipeline used by starter-pack import — so clone, starter-pack import,
    and `.swmem` import share one write path.

    Source-side validation accepts reserved-prefix ids so operators can
    fork a visitor memory subject (`demo_web_<uuid>__support-agent`) for
    inspection. Target-side validation keeps the guard so the new id
    never lands back in the marketing widget's namespace.

    Provenance metadata is written onto every copied record:
      * `cloned_from_subject_id` — source subject id
      * `cloned_at` — ISO timestamp of the clone operation
      * `cloned_by` — operator email when supplied by the admin proxy
      * `original_episode_id` / `original_memory_id` — the source record
        id, so a future export of the cloned subject can still reference
        the original observation.

    Sources / citations are NOT yet first-class records in the storage
    schema — `source_count` always returns 0 even when the scope is
    `episodes_memories_sources`. This is documented as a known limitation;
    the rest of the scopes are honoured.
    """
    # Source side: allow reserved-prefix ids so operators can clone visitor
    # memory subjects (e.g. demo_web_<uuid>__support-agent) for inspection.
    # Target side keeps the guard so the clone can't *create* a new id in
    # that namespace.
    _validate_subject_id(source_subject_id, field="source_subject_id", allow_reserved=True)

    if target_subject_id:
        _validate_subject_id(target_subject_id, field="target_subject_id")

    src_eps, src_mems = await _count_subject(source_subject_id)
    if src_eps == 0 and src_mems == 0:
        raise StarterPackError(
            f"Source subject {source_subject_id!r} not found "
            "(no episodes or memories under that id).",
            status_code=404,
        )

    final_target = await _resolve_target_subject(
        explicit=target_subject_id,
        suggested=f"{source_subject_id}-clone",
        conflict_strategy="cancel",  # clone never overwrites without explicit retry
    )

    # Read source records via the existing backup primitive — same shape the
    # export endpoint produces, so the ingest path stays unified.
    from server.services.backup import export_subject

    source_doc = await export_subject(source_subject_id, tenant_id=None)

    include_episodes = clone_scope in (
        "episodes",
        "episodes_and_memories",
        "episodes_memories_sources",
    )
    include_memories = clone_scope in (
        "memories",
        "episodes_and_memories",
        "episodes_memories_sources",
    )

    eps = source_doc.get("episodes", []) if include_episodes else []
    mems = source_doc.get("memories", []) if include_memories else []

    cloned_at = _now_iso()
    extra_provenance: dict[str, Any] = {
        "cloned_from_subject_id": source_subject_id,
        "cloned_at": cloned_at,
    }
    if cloned_by:
        extra_provenance["cloned_by"] = cloned_by
    extra_metadata = dict(extra_provenance)
    if target_display_name:
        extra_metadata["display_name"] = target_display_name

    # Stamp each record with its original id BEFORE the ingest pipeline
    # rewrites the row id. The pipeline already preserves any keys we
    # pass on `metadata` / `provenance`; we just need to lift the source
    # id from the top-level shape it arrives in.
    eps_with_provenance = [
        {
            **ep,
            "metadata": {
                **(ep.get("metadata") or {}),
                "original_episode_id": ep.get("id"),
            },
            "provenance": {
                **(ep.get("provenance") or {}),
                "original_episode_id": ep.get("id"),
            },
        }
        for ep in eps
    ]
    mems_with_provenance = [
        {
            **mem,
            "metadata": {
                **(mem.get("metadata") or {}),
                "original_memory_id": mem.get("id"),
            },
        }
        for mem in mems
    ]

    counts = await _ingest_records_async(
        target_subject_id=final_target,
        target_tenant_id=target_tenant_id,
        episodes_data=eps_with_provenance,
        memories_data=mems_with_provenance,
        extra_provenance=extra_provenance,
        extra_metadata=extra_metadata,
    )

    logger.info(
        "subject_cloned",
        source_subject_id=source_subject_id,
        target_subject_id=final_target,
        clone_scope=clone_scope,
        episode_count=counts["episodes"],
        memory_count=counts["memories"],
        cloned_by=cloned_by,
    )

    return {
        "status": "cloned",
        "source_subject_id": source_subject_id,
        "target_subject_id": final_target,
        "target_display_name": target_display_name,
        "clone_scope": clone_scope,
        "episode_count": counts["episodes"],
        "memory_count": counts["memories"],
        # Sources/citations are not yet first-class cloneable records —
        # see the docstring above. Returning the literal 0 rather than
        # omitting the key keeps the response shape stable.
        "source_count": 0,
        "cloned_at": cloned_at,
    }


# ─── Public: export memory payload ──────────────────────────────────────────


async def export_memory_payload(
    *,
    subject_ids: list[str],
    tenant_id: str | None,
    export_scope: CloneScope = "episodes_memories_sources",
) -> dict[str, Any]:
    if not subject_ids:
        raise StarterPackError("subject_ids is required.", status_code=400)
    if len(subject_ids) > settings.memory_import_max_subjects:
        raise StarterPackError(
            f"Too many subjects in one export: {len(subject_ids)} > "
            f"limit {settings.memory_import_max_subjects}.",
            status_code=413,
        )
    for sid in subject_ids:
        # Same source-side carve-out as clone: exporting a visitor subject
        # for inspection / backup is legitimate.
        _validate_subject_id(sid, field="subject_ids", allow_reserved=True)

    include_episodes = export_scope in (
        "episodes",
        "episodes_and_memories",
        "episodes_memories_sources",
    )
    include_memories = export_scope in (
        "memories",
        "episodes_and_memories",
        "episodes_memories_sources",
    )

    from server.services.backup import export_subject

    subjects: list[dict[str, Any]] = []
    episodes: list[dict[str, Any]] = []
    memories: list[dict[str, Any]] = []

    for sid in subject_ids:
        doc = await export_subject(sid, tenant_id=tenant_id)
        subjects.append(
            {
                "original_subject_id": sid,
                "tenant_id": doc.get("tenant_id"),
                "display_name": None,
                "tags": [],
                "metadata": {},
                "exported_at": doc.get("exported_at"),
            }
        )
        if include_episodes:
            for ep in doc.get("episodes", []):
                episodes.append(
                    {
                        "original_episode_id": ep.get("id"),
                        "subject_id": ep.get("subject_id") or sid,
                        "source": ep.get("source"),
                        "type": ep.get("type"),
                        "payload": ep.get("payload"),
                        "metadata": ep.get("metadata") or {},
                        "provenance": ep.get("provenance") or {},
                        # `occurred_at` is forward-compatible: packs created
                        # before migration 0015 won't have it, in which case
                        # the import path below falls back to `created_at`.
                        "occurred_at": ep.get("occurred_at"),
                        "created_at": ep.get("created_at"),
                        "last_compiled_at": ep.get("last_compiled_at"),
                    }
                )
        if include_memories:
            for mem in doc.get("memories", []):
                memories.append(
                    {
                        "original_memory_id": mem.get("id"),
                        "subject_id": mem.get("subject_id") or sid,
                        "kind": mem.get("kind"),
                        "content": mem.get("content"),
                        "summary": mem.get("summary"),
                        "confidence": mem.get("confidence"),
                        "valid_from": mem.get("valid_from"),
                        "valid_to": mem.get("valid_to"),
                        "source_episode_ids": mem.get("source_episode_ids") or [],
                        "status": mem.get("status"),
                        "metadata": mem.get("metadata") or {},
                        "created_at": mem.get("created_at"),
                        "updated_at": mem.get("updated_at"),
                    }
                )

    payload = {
        "format": PAYLOAD_FORMAT,
        "format_version": PAYLOAD_FORMAT_VERSION,
        "export_id": uuid.uuid4().hex,
        "exported_at": _now_iso(),
        "export_scope": export_scope,
        "subjects": subjects,
        "episodes": episodes,
        "memories": memories,
        "sources": [],
        "metadata": {
            "statewave_version": "0.1",
            "subject_count": len(subjects),
            "episode_count": len(episodes),
            "memory_count": len(memories),
            "source_count": 0,
        },
    }
    logger.info(
        "memory_export_built",
        subject_count=len(subjects),
        episode_count=len(episodes),
        memory_count=len(memories),
        export_scope=export_scope,
    )
    return payload


# ─── Public: import memory payload ──────────────────────────────────────────

_ALLOWED_PAYLOAD_KEYS = {
    "format",
    "format_version",
    "export_id",
    "exported_at",
    "export_scope",
    "subjects",
    "episodes",
    "memories",
    "sources",
    "metadata",
}


async def import_memory_payload(
    *,
    payload: dict[str, Any],
    target_tenant_id: str | None,
    conflict_strategy: ConflictStrategy = "create_copy",
) -> dict[str, Any]:
    if not isinstance(payload, dict):
        raise StarterPackError("Payload must be a JSON object.", status_code=400)
    if payload.get("format") != PAYLOAD_FORMAT:
        raise StarterPackError(
            f"Unexpected payload format {payload.get('format')!r}; "
            f"expected {PAYLOAD_FORMAT!r}.",
            status_code=400,
        )
    if int(payload.get("format_version", 0)) != PAYLOAD_FORMAT_VERSION:
        raise StarterPackError(
            f"Unsupported format_version {payload.get('format_version')}; "
            f"expected {PAYLOAD_FORMAT_VERSION}.",
            status_code=400,
        )

    extra = set(payload.keys()) - _ALLOWED_PAYLOAD_KEYS
    if extra:
        raise StarterPackError(
            f"Payload contains unknown fields: {sorted(extra)}",
            status_code=400,
        )

    # Hard limit by serialized size — same number the request body limit guard
    # uses, kept here in case a future ingress strips it.
    serialized = json.dumps(payload)
    if len(serialized.encode("utf-8")) > settings.memory_import_max_bytes:
        raise StarterPackError(
            f"Payload exceeds size limit ({settings.memory_import_max_bytes} bytes).",
            status_code=413,
        )

    subjects = payload.get("subjects") or []
    episodes = payload.get("episodes") or []
    memories = payload.get("memories") or []
    if len(subjects) > settings.memory_import_max_subjects:
        raise StarterPackError(
            f"Too many subjects: {len(subjects)} > limit {settings.memory_import_max_subjects}.",
            status_code=413,
        )

    # Map each `original_subject_id` to a final target subject id. With
    # `create_copy` (the default) every imported subject gets a fresh id;
    # `merge` reuses the original; `cancel` aborts on any pre-existing
    # populated subject.
    id_map: dict[str, str] = {}
    for s in subjects:
        original = s.get("original_subject_id")
        if not original:
            continue
        # Source-side: the original_subject_id is metadata from the archive
        # so we accept reserved-prefix ids (e.g. `demo_web_*` archives
        # captured from the marketing widget). The resulting target id is
        # rewritten to `imported-*` by `_suggest_target_subject_id` so the
        # new subject doesn't land in the marketing widget's namespace.
        # `merge` and `cancel` strategies still pass the original through
        # to `_resolve_target_subject` as `explicit`, which validates it
        # WITHOUT the source-side carve-out — that's intentional, an admin
        # cannot merge into the visitor namespace.
        _validate_subject_id(
            original,
            field="subjects[].original_subject_id",
            allow_reserved=True,
        )
        final = await _resolve_target_subject(
            explicit=original if conflict_strategy != "create_copy" else None,
            suggested=original,
            conflict_strategy=conflict_strategy,
        )
        id_map[original] = final

    # Group episodes / memories by their source subject and ingest each group
    # under the mapped target id.
    eps_by_subject: dict[str, list[dict[str, Any]]] = {}
    for ep in episodes:
        original = ep.get("subject_id") or ""
        if not original:
            continue
        eps_by_subject.setdefault(original, []).append(ep)

    mems_by_subject: dict[str, list[dict[str, Any]]] = {}
    for mem in memories:
        original = mem.get("subject_id") or ""
        if not original:
            continue
        mems_by_subject.setdefault(original, []).append(mem)

    total_episodes = 0
    total_memories = 0
    target_subject_ids: list[str] = []

    for original_id in {*eps_by_subject.keys(), *mems_by_subject.keys()}:
        if original_id not in id_map:
            # Subject metadata wasn't included; create a fresh target id so
            # the records aren't dropped silently.
            id_map[original_id] = _suggest_target_subject_id(original_id)
        target_id = id_map[original_id]
        target_subject_ids.append(target_id)
        counts = await _ingest_records_async(
            target_subject_id=target_id,
            target_tenant_id=target_tenant_id,
            episodes_data=eps_by_subject.get(original_id, []),
            memories_data=mems_by_subject.get(original_id, []),
            extra_provenance={
                "imported_from_export_id": payload.get("export_id"),
                "original_subject_id": original_id,
                "imported_at": _now_iso(),
            },
            extra_metadata={
                "imported_from_export_id": payload.get("export_id"),
                "original_subject_id": original_id,
                "imported_at": _now_iso(),
            },
        )
        total_episodes += counts["episodes"]
        total_memories += counts["memories"]

    logger.info(
        "memory_payload_imported",
        export_id=payload.get("export_id"),
        target_subjects=target_subject_ids,
        episode_count=total_episodes,
        memory_count=total_memories,
        conflict_strategy=conflict_strategy,
    )

    return {
        "imported_at": _now_iso(),
        "export_id": payload.get("export_id"),
        "conflict_strategy": conflict_strategy,
        "subject_id_map": id_map,
        "imported_subjects": list(id_map.values()),
        "imported_episodes": total_episodes,
        "imported_memories": total_memories,
        "imported_sources": 0,
    }


