lore-engine-poc-v3/lore_engine_poc/migration.py

"""Slice 6.4 — Backfill helper for Setting + Material Plane + EXISTS_IN.

Per ``docs/plan/exec/06-planes.md`` sub-slice 6.4, this module
ships ``migrate_setting_id_to_exists_in()`` — the migration
helper that materialises a ``:Setting`` node, a default
``Material Plane``, and ``EXISTS_IN`` facts for a given
list of entity ids.

The helper is idempotent and side-effect-safe to invoke
multiple times — the slice 6.6 region↔plane migration calls it
once per setting, and ``01_ingest.py`` may also call it on
every ingest to keep the slice 6 layer in sync with new
entities.

Design notes (see ``docs/17-planes.md`` for the full version):

  - ``Setting`` is the world's container. The slice 6
    contract (per ``docs/plan/exec/06-planes.md``) requires
    ``id``, ``kind``, ``current_era``, ``schema_version``,
    ``created_at``. The helper supplies defaults that the
    world-builder can override via kwargs.
  - ``Plane`` is a planar subdivision. The default plane is
    a ``Material Plane`` with id ``{setting_id}.material``;
    every entity that the caller passes via ``entity_ids``
    becomes a member of this default plane until the slice
    6.5 time-bounded work assigns them to specific planes.
  - ``EXISTS_IN`` is a *timeless* type-assertion (per
    ``docs/17-planes.md``). Time-bounded planar membership is
    carried by a separate reified ``:Relation`` (slice 6.5).
    ``add_exists_in`` on the GraphBackend Protocol is the
    single write chokepoint — re-adding the same
    ``(entity_id, setting_id)`` pair is a no-op.

Why a dedicated module (not a method on ``InMemoryGraph``):

  - The migration is a one-shot operation per codex; it is
    not part of the read/write hot path. Putting it in a
    module keeps the GraphBackend Protocol narrow (the
    Protocol stays focused on read/write primitives).
  - The slice 6.6 migration script (``05_migrate_planes.py``)
    imports this helper directly, decoupled from the
    ``InMemoryGraph`` implementation.
"""

from __future__ import annotations

import datetime as _dt
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable, Optional

from .graph_backend import GraphBackend
from .setting import Plane, Setting


@dataclass(frozen=True)
class BackfillSummary:
    """The outcome of a single ``migrate_setting_id_to_exists_in``
    invocation. Returned to the caller so scripts can report
    what happened (or assert on it in tests).

    Fields:

      - ``setting_added``: True iff a new ``:Setting`` node
        was materialised. False if the Setting already
        existed (idempotency).
      - ``planes_added``: count of new ``:Plane`` nodes
        materialised. Always ``0`` or ``1`` today (the
        default Material Plane); a future slice may grow
        this to add multiple default planes (e.g. Material
        + Feywild per Setting).
      - ``exists_in_added``: count of new ``EXISTS_IN``
        facts created. Equals ``len(entity_ids)`` minus the
        number of entities that were already members of
        the setting.
    """

    setting_added: bool
    planes_added: int
    exists_in_added: int


def migrate_setting_id_to_exists_in(
    graph: GraphBackend,
    setting_id: str,
    *,
    entity_ids: Optional[Iterable[str]] = None,
    current_era: str = "unspecified",
    schema_version: str = "1.2",
    kind: str = "campaign",
) -> BackfillSummary:
    """Materialise Setting + default Material Plane + EXISTS_IN
    facts for ``entity_ids``.

    Idempotent: re-running with the same ``setting_id`` adds
    no new Setting, no new Plane, and no new EXISTS_IN fact
    beyond the first run.

    Parameters:

      - ``graph``: a ``GraphBackend`` (typically
        ``InMemoryGraph``; Neo4j is a follow-up — see the
        ``NotImplementedError`` stubs in ``neo4j_graph.py``).
      - ``setting_id``: the stable id of the Setting to
        materialise (e.g., ``"mardonari"``). Required.
      - ``entity_ids``: explicit iterable of entity names
        that should get an ``EXISTS_IN`` fact for this
        Setting. When ``None``, the helper does **not**
        walk ``all_names()`` — the caller is responsible
        for naming the entities to migrate. This prevents
        accidental cross-setting bleed when the same
        graph holds multiple settings.
      - ``current_era``: the id of the ``:Era`` the
        world-builder considers "now". Defaults to
        ``"unspecified"`` so a fresh setting is never
        silently pinned to a real era.
      - ``schema_version``: the graph-model version this
        setting is authored under. Defaults to ``"1.2"``
        (the slice 6 version, per ``docs/12-storage-strategy.md``).
      - ``kind``: the world's genre / category
        (``"campaign"``, ``"novel"``, ``"west-marches"``,
        etc.). Defaults to ``"campaign"`` — the Mardonari
        codex's shape.

    Returns:

      A ``BackfillSummary`` describing what changed.

    Raises:

      ``TypeError`` if ``graph`` does not implement the
      ``GraphBackend`` Protocol (defensive — the Protocol
      is structural and doesn't enforce at construction).
    """
    # Materialise the Setting (idempotent: re-adding the same
    # id overwrites with the new metadata, but since dataclass
    # equality is structural, an identical Setting is a no-op
    # for any consumer that compares by value).
    existing_setting = graph.find_setting(setting_id)
    setting = Setting(
        id=setting_id,
        kind=kind,
        current_era=current_era,
        schema_version=schema_version,
        # ISO-8601 UTC timestamp. The created_at is only set
        # on first materialisation — re-runs preserve the
        # original so the audit trail is stable.
        created_at=(
            existing_setting.created_at
            if existing_setting is not None
            else _dt.datetime.now(_dt.timezone.utc).isoformat()
        ),
    )
    graph.add_setting(setting)
    setting_added = existing_setting is None

    # Materialise the default Material Plane (idempotent:
    # the GraphBackend's ``add_plane`` no-ops on re-add).
    material_plane_id = f"{setting_id}.material"
    existing_material = graph.find_plane(material_plane_id)
    material_plane = Plane(
        id=material_plane_id,
        setting_id=setting_id,
        name="Material Plane",
        kind="material",
    )
    graph.add_plane(material_plane)
    planes_added = 0 if existing_material is not None else 1

    # Walk the entity list and add EXISTS_IN facts.
    # ``add_exists_in`` is itself idempotent (re-add is a no-op),
    # so the only counting needed is to detect which entities
    # are *new* members — the test surface needs that signal.
    exists_in_added = 0
    if entity_ids is not None:
        for entity_id in entity_ids:
            already_member = entity_id in graph.setting_entities(setting_id)
            graph.add_exists_in(entity_id=entity_id, setting_id=setting_id)
            if not already_member:
                exists_in_added += 1

    return BackfillSummary(
        setting_added=setting_added,
        planes_added=planes_added,
        exists_in_added=exists_in_added,
    )


# ===========================================================================
# Slice 6.6 — Region ↔ Plane migration
# ===========================================================================
#
# Per ``docs/plan/exec/06-planes.md`` sub-slice 6.6, this module
# exposes the migration as a library so the CLI script
# (``scripts/05_migrate_planes.py``) is a thin wrapper:
#
#   - ``scan_codex_for_planes(codex_dir, setting_id)`` reads
#     the codex's markdown frontmatter and returns a list of
#     :class:`PlannedPlaneMigration` records — one per entry
#     that should be promoted from ``:Region`` to ``:Plane``.
#     The function is pure: it does not touch the graph.
#
#   - ``apply_plane_migration(graph, plans, setting_id)`` takes
#     those plans, materialises ``:Plane`` nodes on the graph,
#     and creates ``LAYER_OF`` edges between co-referenced
#     Planes. Idempotent — re-running on the same plans
#     produces the same graph state.
#
# Discriminator (what makes an entry a Plane):
#
#   1. ``plane: true`` in the frontmatter, OR
#   2. ``plane`` in the entry's ``tags`` list, OR
#   3. ``type: plane`` in the frontmatter.
#
# The exec roadmap also names "under
# ``Campaign Codex - Planes/``" as a path-based signal; that
# rule is honoured when the directory exists. Today's codex
# doesn't ship such a directory, so the rule is part of the
# contract but not exercised by the default fixtures.
#
# LAYER_OF direction follows the design doc's example in
# ``docs/17-planes.md``::
#
#     (:Plane {id: 'mardonari.voldramir'})-[:LAYER_OF]->(:Plane {id: 'mardonari.material'})
#
# So a Plane ``A`` whose markdown body references ``[[B]]``
# and ``B`` is also a Plane creates ``A LAYER_OF B`` (the
# layer points at its parent). ``LAYER_OF`` is only created
# when *both* endpoints are Planes — a Plane referencing a
# non-Plane entity (e.g. a Person) does not produce an edge.


@dataclass(frozen=True)
class PlannedPlaneMigration:
    """One entry the migration will promote to ``:Plane``.

    The dataclass is immutable so the planner (pure) and
    the applier (mutating) cannot accidentally disagree on
    what the plan was. The applier reads these fields and
    calls ``graph.add_plane(...)`` for each plan, plus
    ``graph.add_edge(...)`` for LAYER_OF edges.
    """
    entity_name: str
    plane_id: str  # ``{setting_id}.{entity-slug}``
    setting_id: str
    plane_kind: str  # ``material`` / ``demiplane`` / ``reflection`` / ...
    source_path: str
    # Markdown body — kept on the plan so the applier can
    # walk ``[[wiki-links]]`` for LAYER_OF edges without
    # re-reading the codex file.
    markdown_body: str = ""
    # File's parsed frontmatter — kept so the applier (or
    # future tools) can introspect the source of the
    # promotion signal (``plane: true`` vs ``tags``).
    frontmatter: dict = field(default_factory=dict)


@dataclass(frozen=True)
class PlaneMigrationSummary:
    """Return value of :func:`apply_plane_migration`.

    Pinned by the tests so future refactors of the apply
    step can be checked against the same shape.
    """
    planes_added: int
    layer_of_edges_added: int
    setting_added: bool


def _slugify(name: str) -> str:
    """Convert an entity name into the ``Plane.id`` slug part.

    ``"Voldramir"`` → ``"voldramir"``;
    ``"Lord Aldric"`` → ``"lord_aldric"``.

    The setting-id prefix (``mardonari.``) is added by the
    caller, not here, so the slug rule is local to the
    entity-name side.
    """
    s = name.strip().lower()
    # Replace any run of non-alphanumeric characters with
    # a single underscore; collapse leading/trailing
    # underscores.
    s = re.sub(r"[^a-z0-9]+", "_", s)
    s = s.strip("_")
    return s or "unnamed"


def _parse_simple_frontmatter(text: str) -> tuple[dict, str]:
    """A *very* small YAML-frontmatter parser for the fields
    the migration cares about (``plane: true``, ``type``,
    ``tags``).

    The codex's full YAML parsing is done by
    :func:`lore_engine_poc.parsers._yaml.load_yaml`; the
    migration is intentionally narrow — it only inspects a
    handful of fields and is OK rejecting weird files by
    treating them as "no plane signal".

    Returns ``(frontmatter_dict, markdown_body)`` where the
    body is the markdown after the closing ``---`` fence.
    """
    if not text.startswith("---"):
        return {}, text
    rest = text[3:]
    end = rest.find("\n---")
    if end < 0:
        return {}, text
    head = rest[:end]
    body = rest[end + 4 :]  # skip "\n---"
    # Strip a single leading newline if present so the body
    # starts at the first heading.
    if body.startswith("\n"):
        body = body[1:]

    fm: dict = {}
    tags_block_pending = False
    for raw_line in head.splitlines():
        line = raw_line.rstrip()
        if not line or line.lstrip().startswith("#"):
            continue

        # Indented continuation for block-list ``tags:``.
        if line.startswith("  - ") and tags_block_pending:
            fm.setdefault("tags", []).append(
                line[4:].strip().strip('"\'').strip()
            )
            continue

        if ":" not in line:
            continue
        key, _, value = line.partition(":")
        key = key.strip()
        value = value.strip()
        # Strip surrounding quotes if present.
        if (
            len(value) >= 2
            and value[0] == value[-1]
            and value[0] in {'"', "'"}
        ):
            value = value[1:-1]
        if key == "plane":
            fm["plane"] = value.lower() == "true"
        elif key == "type":
            fm["type"] = value
        elif key == "kind":
            fm["kind"] = value
        elif key == "tags":
            tags_block_pending = False
            if value.startswith("[") and value.endswith("]"):
                inner = value[1:-1]
                fm["tags"] = [
                    t.strip().strip('"\'').strip()
                    for t in inner.split(",") if t.strip()
                ]
            elif value == "":
                tags_block_pending = True
            else:
                # Single tag on one line: ``tags: plane``.
                fm["tags"] = [value]
        elif key and value:
            # Catch-all for arbitrary frontmatter fields the
            # migration doesn't introspect — kept so the
            # returned dict is round-trippable.
            fm.setdefault("_extras", {})[key] = value

    return fm, body


def _entry_is_plane(frontmatter: dict) -> bool:
    """True if the frontmatter signals ``:Plane`` promotion.

    Pins: ``plane: true`` *or* ``tags contains "plane"``
    *or* ``type: plane``. The tags convention is what the
    existing codex uses; ``plane: true`` is the exec
    roadmap's preferred form; ``type: plane`` is honoured
    so the migration accepts both legacy and new codexes.
    """
    if frontmatter.get("plane") is True:
        return True
    tags = frontmatter.get("tags") or []
    if any(t.lower() == "plane" for t in tags):
        return True
    if (frontmatter.get("type") or "").lower() == "plane":
        return True
    return False


def _infer_plane_kind(frontmatter: dict) -> str:
    """Decide the ``Plane.kind`` value from the entry's
    frontmatter.

    Rules:

      - ``plane: true`` (without any other kind hint) → ``material``.
        This is the "baseline plane of a setting" convention.
      - ``tags: [plane, demiplane]`` → ``demiplane`` (explicit).
      - ``tags: [plane, ...]`` (just ``plane``, no specific kind) →
        ``demiplane`` is the conservative default.
      - ``kind: <value>`` in the frontmatter → that value.

    The function is intentionally simple. Authoring a plane
    of a more specific kind (reflection, transit, ethereal,
    outer, inner, transcendent) requires the author to
    either tag it explicitly or set ``kind:``.
    """
    if frontmatter.get("kind"):
        return str(frontmatter["kind"]).lower()
    # ``plane: true`` (the bare flag, no other hint) is the
    # convention for "this setting's Material Plane" — the
    # baseline.
    if frontmatter.get("plane") is True and "tags" not in frontmatter:
        return "material"
    tags = frontmatter.get("tags") or []
    tag_set = {t.lower() for t in tags}
    for known in (
        "material", "demiplane", "reflection",
        "transit", "ethereal", "outer", "inner", "transcendent",
    ):
        if known in tag_set:
            return known
    # Default — ``plane`` is tagged but no specific kind.
    return "demiplane"


def _iter_codex_markdown(codex_root: Path) -> Iterable[tuple[Path, str]]:
    """Yield ``(path, contents)`` for every ``.md`` file
    under the codex's ``Campaign Codex - Regions/`` and
    ``Campaign Codex - Planes/`` directories.

    The migration reads from both folders: regions that
    are tagged-as-planes live in ``- Regions/`` (today's
    codex convention), and the roadmap allows future
    Plane entries to live in ``- Planes/`` directly.

    The function is local — the migration does not need
    the codex-wide walker that ``parsers.iter_codex``
    provides, and keeping it local makes the migration's
    surface explicit.
    """
    for sub in ("Campaign Codex - Regions", "Campaign Codex - Planes"):
        d = codex_root / sub
        if not d.is_dir():
            continue
        for path in sorted(d.glob("*.md")):
            yield path, path.read_text(encoding="utf-8")


def scan_codex_for_planes(
    codex_root: Path,
    setting_id: str,
) -> list[PlannedPlaneMigration]:
    """Scan the codex and return the list of entries to
    promote to ``:Plane``.

    The function is pure: it does not write to any graph.
    Re-running it produces the same plans. The plans are
    what :func:`apply_plane_migration` consumes.
    """
    out: list[PlannedPlaneMigration] = []
    for path, contents in _iter_codex_markdown(codex_root):
        fm, body = _parse_simple_frontmatter(contents)
        if not _entry_is_plane(fm):
            continue
        # Extract the entity name from the first H1 in the
        # markdown body — the same convention the rest of
        # the codex uses for entity identity.
        name = _first_h1(body) or path.stem
        out.append(PlannedPlaneMigration(
            entity_name=name,
            plane_id=f"{setting_id}.{_slugify(name)}",
            setting_id=setting_id,
            plane_kind=_infer_plane_kind(fm),
            source_path=str(path),
            markdown_body=body,
            frontmatter=fm,
        ))
    return out


def _first_h1(body: str) -> str | None:
    """Return the text of the first ``# H1`` heading in
    ``body``, or ``None`` if no H1 is present.
    """
    for line in body.splitlines():
        line = line.strip()
        if line.startswith("# "):
            return line[2:].strip()
    return None


_WIKILINK = re.compile(r"\[\[([^\]]+)\]\]")


def _build_layer_of_edges(
    plans: list[PlannedPlaneMigration],
    setting_id: str,
) -> list[tuple[str, str]]:
    """Walk the markdown body of each plan and collect
    ``(subject, object)`` tuples for ``LAYER_OF`` edges.

    The subject is the *referencing* plane (``A`` whose
    markdown says ``[[B]]``). The object is the
    *referenced* plane (``B``). If ``B`` is not in the
    plan set, the edge is dropped — ``LAYER_OF`` is a
    plane-to-plane relation only.

    The returned list is unsorted; the applier adds them
    in whatever order.
    """
    # Index: which planes were promoted? (slug-based
    # lookup so Voldramir's ``[[Underdark]]`` finds the
    # ``Underdark`` plan even if the H1 casing differs.)
    promoted: dict[str, str] = {
        _slugify(p.entity_name): p.plane_id for p in plans
    }
    edges: list[tuple[str, str]] = []
    for p in plans:
        for match in _WIKILINK.finditer(p.markdown_body):
            target = match.group(1).strip()
            target_slug = _slugify(target)
            target_plane_id = promoted.get(target_slug)
            if target_plane_id is None:
                continue
            # LAYER_OF direction: A (this plane) → B (target).
            edges.append((p.plane_id, target_plane_id))
    # Deduplicate — the same pair may be referenced from
    # multiple places in the body.
    return sorted(set(edges))


def apply_plane_migration(
    graph: GraphBackend,
    plans: list[PlannedPlaneMigration],
    setting_id: str,
) -> PlaneMigrationSummary:
    """Materialise the planned planes on the graph.

    Steps:

      1. Ensure the setting exists (the migration is the
         first writer for some settings).
      2. For each plan, call ``graph.add_plane(...)``.
      3. Walk each plane's markdown for ``[[Other]]``
         references; create ``LAYER_OF`` edges to other
         promoted planes.

    Idempotent: re-running with the same plans produces
    the same graph state (the ``add_plane`` chokepoint
    overwrites an existing plane with the same id, and the
    edge path de-duplicates).
    """
    planes_added = 0
    setting_added = False

    if plans and graph.find_setting(setting_id) is None:
        graph.add_setting(Setting(
            id=setting_id,
            kind="campaign",
            current_era="unspecified",
            schema_version="1.2",
            created_at=_dt.datetime.now(_dt.timezone.utc).isoformat(),
        ))
        setting_added = True

    # Plane layer. ``add_plane`` is the chokepoint; the
    # backend overwrites if the id already exists.
    #
    # Idempotency at the plane layer: we count "newly added"
    # planes, not "plans processed", so the summary is
    # accurate on re-runs. The overwrite still happens (the
    # field values may have changed in the source codex),
    # but the count reflects the graph delta.
    for p in plans:
        existing_plane = graph.find_plane(p.plane_id)
        graph.add_plane(Plane(
            id=p.plane_id,
            setting_id=p.setting_id,
            name=p.entity_name,
            kind=p.plane_kind,
        ))
        if existing_plane is None:
            planes_added += 1

    # LAYER_OF edges (plane-to-plane only). We use the
    # ``Edge`` dataclass via the backend's ``add`` chokepoint
    # so the in-memory and Neo4j backends both handle this
    # path uniformly (slice 5 established ``add(Edge)`` as
    # the canonical write entrypoint).
    #
    # Idempotency: the migration is re-runnable, so we
    # check for an existing ``(subject, relation, object)``
    # edge before adding. This pins the LAYER_OF dedup
    # behaviour at the migration layer (the in-memory
    # backend's ``add`` does not itself dedupe by
    # ``(subject, relation, object)`` — it indexes by
    # ``edge_id``).
    from .tools import Edge
    edges = _build_layer_of_edges(plans, setting_id)
    layer_of_added = 0
    for subject, obj in edges:
        existing = [
            e for e in graph.edges_for_subject(subject, relation="LAYER_OF")
            if e.object == obj
        ]
        if existing:
            # Already wired up — skip. Re-running the
            # migration is a no-op for this edge.
            continue
        graph.add(Edge(
            subject=subject,
            relation="LAYER_OF",
            object=obj,
        ))
        layer_of_added += 1

    return PlaneMigrationSummary(
        planes_added=planes_added,
        layer_of_edges_added=layer_of_added,
        setting_added=setting_added,
    )