chemenu/tools/chemenu/commands/migrate_cmd.py

"""`wikitool migrate` - content migrations: what this instance still owes, and
whether a bulk rewrite broke anything.

Five commands around two facts. `.wikitool-kb.json` (see `chemenu/kb_state.py`)
records what shape the content is in, so the chain of outstanding migrations is
computed rather than guessed. `migrate verify` compares the corpus against a git
revision on the invariants a migration must not change (see
`chemenu/corpus_diff.py`).

**There is no `migrate run`.** An `assisted` migration is a procedure an agent
carries out page by page; the tool keeps the books and checks the result. A
`run` would claim an ability that does not exist - it arrives when mechanical
primitives do.
"""
from __future__ import annotations

import json as _json
import subprocess
from pathlib import Path
from typing import Optional

import typer

from chemenu import config, corpus_diff, kb_scan, kb_state, version as version_mod
from chemenu.commands._util import console, fail, rel_path, success, today_iso
from chemenu.frontmatter_io import read_page
from chemenu.page import Page
from chemenu.version import Version, VersionError

app = typer.Typer(help="Content migrations: outstanding chain, bookkeeping, and verification.")


# --- shared state loading --------------------------------------------------


def _versions() -> tuple[Version, Optional[Version]]:
    """(stack version, kb version).

    An unreadable VERSION or a corrupt state file exits 1 here (via `fail`,
    which raises); a *missing* kb version returns None, because that is a
    state each command explains in its own words rather than an error."""
    try:
        return version_mod.read_version(), kb_state.read_kb_version()
    except VersionError as exc:
        fail(str(exc))
        raise  # unreachable: fail() raises typer.Exit


# --- migrate list ----------------------------------------------------------


@app.command("list")
def list_command(
    json_out: bool = typer.Option(False, "--json", help="Print the migrations as JSON"),
):
    """List every migration document, oldest target first. Read-only."""
    migrations = kb_state.load_migrations()

    if json_out:
        typer.echo(
            _json.dumps(
                [
                    {
                        "name": m.name,
                        "migrates_to": str(m.target),
                        "migration_kind": m.kind,
                        "obligation": m.obligation,
                        "description": m.description,
                        "path": m.relative_path,
                    }
                    for m in migrations
                ],
                indent=2,
            )
        )
        return

    if not migrations:
        success(f"No migration documents under {rel_path(kb_state.migrations_dir())}.")
        return
    for migration in migrations:
        console.print(
            f"[bold]{migration.target}[/bold]  {migration.name}  "
            f"({migration.kind}, {migration.obligation})"
        )
        if migration.description:
            console.print(f"    {migration.description}")


# --- migrate status --------------------------------------------------------


def _report_offers(
    offered: list["kb_state.Migration"], divergent: Optional[list[str]]
) -> None:
    """Print the optional half of `status`, above the outstanding chain.

    Deliberately never affects the exit code and never says "outstanding". An
    offer is the stack proposing a better default for a file the instance owns;
    an instance that keeps its own version is in a correct state, not a late
    one. Mixing the two is how the message that actually matters - your content
    no longer fits your machinery - stops being read.
    """
    if not offered:
        return
    console.print(
        f"[cyan]{len(offered)} optional upgrade(s) available[/cyan] - none of them block:"
    )
    for migration in offered:
        console.print(f"  {migration.target}  {migration.name}  ({migration.kind})")
        if migration.description:
            console.print(f"       {migration.description}")
        console.print(f"       {migration.relative_path}")

    if divergent is None:
        console.print(
            "  [dim]This tree carries no release stamp, so which of your files still match "
            "what you were given cannot be answered here.[/dim]"
        )
        return
    if divergent:
        console.print(
            f"  [dim]{len(divergent)} file(s) differ from the release you installed - those are "
            "yours to reconcile by hand rather than overwrite:[/dim]"
        )
        for relative in divergent[:10]:
            console.print(f"    [dim]{relative}[/dim]")
        if len(divergent) > 10:
            console.print(f"    [dim]... and {len(divergent) - 10} more[/dim]")
    else:
        console.print(
            "  [dim]No file differs from the release you installed, so an offer can be taken "
            "by copying.[/dim]"
        )


@app.command("status")
def status_command(
    json_out: bool = typer.Option(False, "--json", help="Print the chain as JSON"),
):
    """Show the migrations this instance still owes, in the order they run.

    Read-only. Exit 1 only when the KB version is undeclared - that is a
    question the tool refuses to answer by guessing."""
    stack, kb_version = _versions()
    migrations = kb_state.load_migrations()

    if kb_version is None:
        if json_out:
            typer.echo(
                _json.dumps(
                    {"stack_version": str(stack), "kb_version": None, "pending": None}, indent=2
                )
            )
        fail(
            f"{kb_state.KB_STATE_FILENAME} is missing - this instance has never declared what "
            f"shape its content is in, and guessing would be wrong exactly when it matters.\n"
            f"Declare it once: `wikitool migrate baseline <version>` (use {stack} if this "
            f"instance's content has never been migrated behind its machinery)."
        )
        return

    pending = kb_state.chain(migrations, kb_version, stack.base)
    offered = kb_state.offers(migrations, kb_state.applied_names(kb_state.read_kb_state()))
    divergent = kb_state.divergent_files()

    if json_out:
        typer.echo(
            _json.dumps(
                {
                    "stack_version": str(stack),
                    "kb_version": str(kb_version),
                    "pending": [
                        {"name": m.name, "migrates_to": str(m.target), "migration_kind": m.kind}
                        for m in pending
                    ],
                    "offered": [
                        {"name": m.name, "migrates_to": str(m.target), "migration_kind": m.kind}
                        for m in offered
                    ],
                    "divergent_files": divergent,
                },
                indent=2,
            )
        )
        return

    console.print(f"stack {stack}, content {kb_version}")
    _report_offers(offered, divergent)
    if not pending:
        if kb_version < stack:
            console.print(
                f"[green]Nothing outstanding[/green] - no migration targets the range "
                f"({kb_version}, {stack}]."
            )
        else:
            console.print("[green]Nothing outstanding[/green] - content matches the machinery.")
        return

    console.print(f"[cyan]{len(pending)} migration(s) outstanding, in this order:[/cyan]")
    for position, migration in enumerate(pending, start=1):
        console.print(f"  {position}. {migration.target}  {migration.name}  ({migration.kind})")
        if migration.description:
            console.print(f"       {migration.description}")
        console.print(f"       {migration.relative_path}")
    console.print(
        f"\nRun the first one, then record it: `wikitool migrate done {pending[0].target}`.\n"
        "The procedure is instructions/migrate-corpus.md."
    )


# --- migrate done / baseline ----------------------------------------------


@app.command("done")
def done_command(
    version: str = typer.Argument(..., help="The migration's target version, e.g. 1.4.0"),
    pages: Optional[int] = typer.Option(None, "--pages", help="How many pages it touched"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Report without writing"),
):
    """Record one migration as applied, advancing the KB version to its target.

    Refuses any version that is not the *next* link in the chain: skipping a
    migration is how a corpus ends up in a shape no version describes, and an
    interrupted multi-step upgrade has to be resumable rather than guessable.

    An `offered` migration is recorded but does not move the version, and no
    ordering rule applies to it - it is not a link in the chain. The record is
    the only thing that distinguishes an offer someone took from one they
    ignored, precisely because the version stays put."""
    stack, kb_version = _versions()
    if kb_version is None:
        fail(
            f"{kb_state.KB_STATE_FILENAME} is missing - run `wikitool migrate baseline <version>` "
            "before recording a migration."
        )
        return

    try:
        target = Version.parse(version)
    except VersionError as exc:
        fail(str(exc))
        return

    migrations = kb_state.load_migrations()
    state = kb_state.read_kb_state() or {}

    # An offer is recorded but does not advance the version: it is not a link in
    # the chain, so there is no ordering rule to check and nothing to skip. The
    # ledger is what makes it stop being offered - without that record there
    # would be no way to tell a taken offer from an ignored one, because
    # `kb_version` deliberately does not move.
    offered = {m.name: m for m in migrations if not m.is_required}
    taken = next((m for m in offered.values() if str(m.target) == version), None)
    if taken is not None:
        if taken.name in kb_state.applied_names(state):
            success(f"{taken.name} is already recorded as taken. Nothing to do.")
            return
        if dry_run:
            success(f"Dry run: would record the optional {taken.name}. Nothing written.")
            return
        applied = list(state.get("applied") or [])
        entry = {"migration": taken.name, "at": today_iso(), "obligation": kb_state.OFFERED}
        if pages is not None:
            entry["pages"] = pages
        applied.append(entry)
        kb_state.write_kb_state(kb_version, applied)
        success(
            f"Recorded the optional {taken.name}. Content stays at {kb_version} - an offer "
            "changes a file you own, not the shape of your content."
        )
        return

    expected = kb_state.next_link(migrations, kb_version, stack.base)
    if expected is None:
        fail(
            f"Nothing is outstanding: content is at {kb_version}, machinery at {stack}, and no "
            f"migration targets the range in between."
        )
        return
    if expected.target != target:
        fail(
            f"{target} is not the next migration. The chain from {kb_version} continues with "
            f"{expected.target} ({expected.name}) - applying them out of order leaves the corpus "
            f"in a shape no version describes.\nRun `wikitool migrate status` to see the order."
        )
        return

    applied = list(state.get("applied") or [])
    entry = {"migration": expected.name, "at": today_iso()}
    if pages is not None:
        entry["pages"] = pages
    applied.append(entry)

    if dry_run:
        success(f"Dry run: content {kb_version} -> {target} ({expected.name}). Nothing written.")
        return

    kb_state.write_kb_state(target, applied)
    remaining = kb_state.chain(migrations, target, stack.base)
    success(
        f"Content is now {target} ({expected.name}). "
        + (
            f"{len(remaining)} migration(s) still outstanding - next is {remaining[0].target}."
            if remaining
            else "Nothing outstanding."
        )
    )


@app.command("baseline")
def baseline_command(
    version: str = typer.Argument(..., help="The shape this instance's content is already in"),
    force: bool = typer.Option(
        False, "--force", help="Overwrite an existing declaration (not a substitute for `done`)"
    ),
):
    """Declare the KB version once, for an instance that never had one.

    Only for a tree predating `.wikitool-kb.json`. Advancing the version after
    a migration is `migrate done`, which checks the chain; this command does
    not, which is why it refuses to overwrite silently."""
    try:
        target = Version.parse(version)
    except VersionError as exc:
        fail(str(exc))
        return

    try:
        existing = kb_state.read_kb_version()
    except VersionError as exc:
        fail(str(exc))
        return

    if existing is not None and not force:
        fail(
            f"This instance already declares content version {existing}. Use "
            f"`wikitool migrate done <version>` to advance it after a migration, or --force "
            f"if the declaration itself is wrong."
        )
        return

    state = kb_state.read_kb_state() or {}
    kb_state.write_kb_state(target, list(state.get("applied") or []))
    success(f"Content version declared as {target}.")


# --- migrate verify --------------------------------------------------------


def _git_show(rev: str, relative: str) -> Optional[str]:
    result = subprocess.run(
        ["git", "show", f"{rev}:{relative}"],
        cwd=config.ROOT,
        capture_output=True,
        text=True,
    )
    return result.stdout if result.returncode == 0 else None


def _paths_at(rev: str) -> Optional[list[str]]:
    result = subprocess.run(
        ["git", "ls-tree", "-r", "--name-only", "-z", rev, "--", "kb"],
        cwd=config.ROOT,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return None
    # The same page/not-a-page rule the working tree is read with. Answering it
    # differently on the two sides reported every COLLECTION.md and INDEX.md as
    # a page that had since disappeared.
    return [
        path
        for path in result.stdout.split("\0")
        if path.startswith("kb/") and kb_scan.is_page_path(path[len("kb/"):])
    ]


def _shapes_at_revision(rev: str, wanted: set[str]) -> dict[str, corpus_diff.PageShape]:
    """Page shapes as of `rev`, keyed by repo-relative path."""
    import tempfile

    shapes: dict[str, corpus_diff.PageShape] = {}
    paths = _paths_at(rev)
    if paths is None:
        fail(f"`git show {rev}` failed - is {rev} a revision in this repository?")
        return shapes

    with tempfile.TemporaryDirectory() as tmp:
        for relative in paths:
            if wanted and not any(relative.startswith(prefix) for prefix in wanted):
                continue
            text = _git_show(rev, relative)
            if text is None:
                continue
            # read_page owns frontmatter parsing (and its error contract), so the
            # historical blob is materialised under its real filename - the stem
            # is the page title, which PageShape compares.
            scratch = Path(tmp) / Path(relative).name
            scratch.write_text(text, encoding="utf-8")
            try:
                frontmatter, body = read_page(scratch)
            except Exception:  # noqa: BLE001 - an unparseable historical page is not this tool's error
                continue
            shapes[relative] = corpus_diff.PageShape.of(Page(Path(relative), frontmatter, body))
    return shapes


def _shapes_now(wanted: set[str]) -> dict[str, corpus_diff.PageShape]:
    shapes: dict[str, corpus_diff.PageShape] = {}
    for path in kb_scan.iter_kb_pages(config.KB_DIR):
        relative = path.relative_to(config.ROOT).as_posix()
        if wanted and not any(relative.startswith(prefix) for prefix in wanted):
            continue
        try:
            frontmatter, body = read_page(path)
        except Exception:  # noqa: BLE001 - lint reports unreadable frontmatter
            continue
        shapes[relative] = corpus_diff.PageShape.of(Page(path, frontmatter, body))
    return shapes


@app.command("verify")
def verify_command(
    from_rev: str = typer.Option(..., "--from", help="Git revision to compare against, e.g. HEAD"),
    path: Optional[list[str]] = typer.Option(
        None, "--path", help="Limit to a subtree, repeatable (e.g. kb/concepts)"
    ),
    expect_body_change: bool = typer.Option(
        False, "--expect-body-change", help="Also report pages whose body did not change at all"
    ),
    json_out: bool = typer.Option(False, "--json", help="Print the diff as JSON"),
    fail_on_error: bool = typer.Option(
        False, "--fail-on-error", help="Exit 1 if any invariant changed"
    ),
):
    """Compare kb/ against a git revision on the invariants a content migration
    must not change: wikilink and citation *counts*, footnote definitions, H1,
    and structural frontmatter.

    Not migration-specific - worth running after any bulk rewrite. `lint` cannot
    answer this: it reads one revision, so a reference that went missing is
    invisible to it."""
    wanted = {p.rstrip("/") for p in (path or [])}
    before = _shapes_at_revision(from_rev, wanted)
    after = _shapes_now(wanted)
    diff = corpus_diff.compare(before, after, expect_body_change=expect_body_change)

    if json_out:
        typer.echo(
            _json.dumps(
                {
                    "from": from_rev,
                    "compared": diff.compared,
                    "added": diff.added,
                    "removed": diff.removed,
                    "findings": [
                        {"path": f.path, "kind": f.kind, "detail": f.detail} for f in diff.findings
                    ],
                },
                indent=2,
            )
        )
    else:
        typer.echo(corpus_diff.render_report(diff, from_rev))

    if diff.findings and fail_on_error:
        raise typer.Exit(code=1)