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 cli_contract, 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 ----------------------------------------------------------


@cli_contract.record(cli_contract.CommandRecord(
    path="migrate list",
    summary="List every migration document under `instructions/migrations/`.",
    synopsis=(cli_contract.Variant(usage="migrate list [--json]"),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.READ,
        idempotent=cli_contract.Idempotent.YES,
        atomic="Read-only",
        budget=cli_contract.Budget.EXEMPT,
    ),
    notes=(
        "Lists every migration document under `instructions/migrations/`, oldest target first, "
        "with its kind and obligation.",
        "Never fails. Read-only and **exempt from the Iteration Budget Gate**.",
    ),
    failures=(),
    examples=(
        "tools/wikitool migrate list",
    ),
    see_also=(
        "`wikitool migrate status` - which of them this instance still owes",
        "`instructions/migrate-corpus.md` - how a migration is run",
    ),
))
@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]"
        )


@cli_contract.record(cli_contract.CommandRecord(
    path="migrate status",
    summary="Show the migrations this instance still owes, in the order they must run.",
    synopsis=(cli_contract.Variant(usage="migrate status [--json]"),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.READ,
        idempotent=cli_contract.Idempotent.YES,
        atomic="Read-only",
        budget=cli_contract.Budget.EXEMPT,
    ),
    notes=(
        "Shows every **required** migration whose `migrates_to` lies in "
        "`(kb_version, VERSION]`, in the order it must run.",
        "`offered` migrations are listed separately above the chain: they never block, never "
        "count as owed, and are bounded by the applied ledger rather than by `kb_version` - "
        "taking one does not move the version.",
        "With a release stamp present, also reports which shipped files this instance has "
        "since edited (from the per-file sha256 in `.wikitool-release.json`) - which says "
        "whether an offer may be copied over or has to be reconciled by hand. Without a stamp "
        "that question is reported as unanswerable rather than answered.",
        "Exits 1 only when the content version is undeclared (`.wikitool-kb.json` missing) or "
        "`VERSION` is unreadable; it never guesses the content's shape.",
        "Read-only, safe to retry freely, and exempt from the Iteration Budget Gate.",
    ),
    failures=(
        cli_contract.Failure(
            cause="`.wikitool-kb.json` is missing - the content version is undeclared",
            reaction="Run `migrate baseline <version>` once, then retry",
        ),
        cli_contract.Failure(
            cause="`VERSION` is unreadable",
            reaction="Fix `VERSION`, then retry",
        ),
    ),
    examples=(
        "tools/wikitool migrate status",
        "tools/wikitool migrate status --json",
    ),
    see_also=(
        "`wikitool migrate done` - records one as applied",
        "`instructions/migrate-corpus.md` - how a migration is run",
        "`instructions/upgrade-instance.md` - where an upgrade checks this",
    ),
))
@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 ----------------------------------------------


@cli_contract.record(cli_contract.CommandRecord(
    path="migrate done",
    summary="Record one migration as applied, advancing `kb_version` in `.wikitool-kb.json`.",
    synopsis=(cli_contract.Variant(usage="migrate done <version> [--pages N] [--dry-run]"),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.WRITE,
        idempotent=cli_contract.Idempotent.NO,
        atomic="Yes - single file write",
        budget=cli_contract.Budget.COUNTED,
    ),
    notes=(
        "Records one migration as applied, advancing `kb_version` in `.wikitool-kb.json` to its "
        "target.",
        "**Refuses any required version that is not the next link in the chain.**",
        "An `offered` migration is recorded in the applied ledger *without* moving "
        "`kb_version` and with no ordering rule applied. Re-recording one already in the ledger "
        "is a no-op, not an error - idempotent and safe to repeat.",
        "Not idempotent for a required migration: it advances the chain.",
        "`--dry-run` reports without writing.",
    ),
    failures=(
        cli_contract.Failure(
            cause="Unknown version, no `.wikitool-kb.json`, or nothing outstanding",
            reaction="Check `migrate status`, fix the argument, then retry once",
        ),
        cli_contract.Failure(
            cause="A *required* version that is not the next link in the chain",
            reaction="Run `migrate status` and apply the migrations in the order it prints",
        ),
    ),
    examples=(
        "tools/wikitool migrate done 7.0.0 --pages 42 --dry-run",
        "tools/wikitool migrate done 7.0.0 --pages 42",
    ),
    never=(
        "Never force the order of required migrations.",
        "Never hand-edit `.wikitool-kb.json` to advance the version.",
    ),
    see_also=(
        "`wikitool migrate status` - the order to apply them in",
        "`instructions/migrate-corpus.md` - the migration procedure",
    ),
))
@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, so there is
    nothing to skip, and requiring the chain first would make an unrelated
    file upgrade wait on it. 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."
        )
    )


@cli_contract.record(cli_contract.CommandRecord(
    path="migrate baseline",
    summary="Declare `kb_version` once, for an instance predating `.wikitool-kb.json`.",
    synopsis=(cli_contract.Variant(usage="migrate baseline <version> [--force]"),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.WRITE,
        idempotent=cli_contract.Idempotent.YES,
        atomic="Yes - single file write",
        budget=cli_contract.Budget.COUNTED,
    ),
    notes=(
        "Declares `kb_version` once, for an instance predating `.wikitool-kb.json`.",
        "Refuses to overwrite an existing declaration without `--force`. Advancing the version "
        "after a migration is `migrate done`, which checks the chain; this command does not.",
        "Safe to re-run with the same version.",
    ),
    failures=(
        cli_contract.Failure(
            cause="Unparseable version",
            reaction="Fix the version and retry",
        ),
        cli_contract.Failure(
            cause="A declaration already exists and `--force` was not passed",
            reaction="It is almost always `migrate done` that was wanted",
        ),
    ),
    examples=(
        "tools/wikitool migrate baseline 6.2.0",
    ),
    never=(
        "Never use `--force` to advance the version past a migration - that is `migrate done`.",
    ),
    see_also=(
        "`wikitool migrate done` - advances the version after a migration",
        "`wikitool migrate status` - what is owed from the declared version",
    ),
))
@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 page title (the wiki's only identity
    for a page), not by path - a page that only moved directory between `rev`
    and now must compare as itself, not as one revision's delete plus the
    other's add. `PageShape.path` still carries the path, for `moved`.

    Sorted path order, matching `kb_scan.load_kb_pages`: if two paths share a
    stem (a naming collision `lint` already reports as `duplicate_titles`),
    the later one wins here too, rather than raising mid-comparison.
    """
    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 sorted(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
            shape = corpus_diff.PageShape.of(Page(Path(relative), frontmatter, body))
            shapes[shape.title] = shape
    return shapes


def _shapes_now(wanted: set[str]) -> dict[str, corpus_diff.PageShape]:
    shapes: dict[str, corpus_diff.PageShape] = {}
    for path in sorted(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
        # The relative path, not `path` itself, so `PageShape.path` is
        # comparable to the historical side's - both repo-relative - and a
        # `moved` entry names a path rather than this machine's tmp_path.
        shape = corpus_diff.PageShape.of(Page(Path(relative), frontmatter, body))
        shapes[shape.title] = shape
    return shapes


@cli_contract.record(cli_contract.CommandRecord(
    path="migrate verify",
    summary="Compare `kb/` against a git revision on the invariants a content migration must "
    "not change.",
    synopsis=(cli_contract.Variant(
        usage="migrate verify --from <rev> [--path P ...] [--expect-body-change] [--json] "
        "[--fail-on-error]",
    ),),
    properties=cli_contract.Properties(
        effect=cli_contract.Effect.READ,
        idempotent=cli_contract.Idempotent.YES,
        atomic="Read-only",
        budget=cli_contract.Budget.EXEMPT,
    ),
    notes=(
        "Compares `kb/` against the revision `--from` on the invariants a content migration "
        "must not change: wikilink and citation **counts** (not sets), footnote definitions, "
        "H1, structural frontmatter, and the **count of generated-region marker pairs**.",
        "Pages are matched by **title**, not path, so a page `move` (or `move --reconcile`) "
        "relocated compares as itself - reported separately as `moved` - rather than as a "
        "removed-and-added pair.",
        "Reports added and removed pages without failing on them.",
        "`--expect-body-change` additionally flags a page whose body did not change at all.",
        "Not migration-specific: worth running after any bulk rewrite.",
        "Exits 0 whatever it finds unless `--fail-on-error` is passed.",
        "Read-only and exempt from the Iteration Budget Gate.",
    ),
    failures=(
        cli_contract.Failure(
            cause="Only with `--fail-on-error`: an invariant changed",
            reaction="Act on the findings - exit 1 here means \"act on the findings\", not \"the "
            "tool is broken\". A finding names a page and what changed on it; it is never fixed "
            "by re-running",
        ),
        cli_contract.Failure(
            cause="`--from` is not a revision in this repository",
            reaction="Fix the revision and retry",
        ),
    ),
    examples=(
        "tools/wikitool migrate verify --from HEAD",
        "tools/wikitool migrate verify --from HEAD --path kb/concepts --expect-body-change",
    ),
    never=(
        "Never re-run to make a finding go away - fix the page it names.",
    ),
    see_also=(
        "`instructions/migrate-corpus.md` - where a migration runs this",
        "`wikitool lint` - the single-revision checks",
    ),
))
@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.

    Marker pairs are compared by *count*, not by the set of region names: a
    page that went from one links region to two has the same set and a
    different count, and a lost marker turns a generated region into prose
    the next write appends a second one beside. This is the one question
    `lint` cannot answer - it reads a single revision, so it cannot see that
    something went missing.

    Pages are matched by title, not path, so a page that only moved directory
    (see `wikitool move`) compares as itself rather than as a
    removed-and-added pair - its path change is reported separately, as
    `moved`, and never counted as a finding on its own.

    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,
                    "moved": [
                        {"page": title, "from": old, "to": new} for title, old, new in diff.moved
                    ],
                    "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)