chemenu/tools/chemenu/kb_state.py

"""The KB version: which *shape* this instance's content is in.

Distinct from the two version facts that already existed, and the distinction
is the whole point:

| Fact | File | Written by | Answers |
|------|------|-----------|---------|
| Stack version | `VERSION` | `version bump` | which machinery is installed |
| Release stamp | `.wikitool-release.json` | `dist export` | where that machinery came from |
| **KB version** | `.wikitool-kb.json` | `migrate done` | what shape the content is in |

Without the third, the state *every* upgrade passes through - machinery already
replaced, content not yet migrated - cannot be represented, and `migrate status`
would have to guess from the stack version, which is wrong exactly when it
matters.

It is a separate file rather than a field in the release stamp because the two
have opposite rules: the stamp is generated and must never be hand-edited, this
one is mutable instance state. Keeping them apart keeps AGENTS.md invariant 1
stated simply.
"""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

from chemenu import config
from chemenu.version import Version, VersionError

KB_STATE_FILENAME = ".wikitool-kb.json"
KB_STATE_SCHEMA = 1

MIGRATIONS_SUBDIR = "migrations"


def kb_state_file() -> Path:
    return config.ROOT / KB_STATE_FILENAME


# Whether a migration has to run, as opposed to how it is carried out. The two
# are independent: a `mechanical` migration can be optional and an `assisted`
# one mandatory. Keeping them on one axis is what would make `migrate status`
# cry wolf - an instance nagged about an improvement it declined stops reading
# the nag that means its content no longer fits the machinery.
REQUIRED = "required"
OFFERED = "offered"
OBLIGATIONS = (REQUIRED, OFFERED)


@dataclass(frozen=True)
class Migration:
    """One migration document under `instructions/migrations/`."""

    name: str
    target: Version
    kind: str  # "mechanical" | "assisted"
    description: str
    path: Path
    obligation: str = REQUIRED

    @property
    def is_required(self) -> bool:
        return self.obligation != OFFERED

    @property
    def relative_path(self) -> str:
        try:
            return str(self.path.relative_to(config.ROOT))
        except ValueError:
            return str(self.path)


def read_kb_version() -> Optional[Version]:
    """The shape this instance's content is in, or None if it never said.

    None is a real state, not an error: an instance created before the KB
    version existed has content of unknown vintage, and guessing would be
    worse than asking (`migrate baseline`).

    Refuses a pre-release (`-beta.N`): a content *shape* has no beta channel,
    only the machinery does, so a `kb_version` naming one means something
    wrote a stack version into this field by hand or by mistake.
    """
    state = read_kb_state()
    if state is None:
        return None
    raw = state.get("kb_version")
    if not raw:
        return None
    version = Version.parse(str(raw))
    if version.is_prerelease:
        raise VersionError(
            f"{KB_STATE_FILENAME} names a pre-release kb_version ({version}) - content has no "
            "beta channel, only the stack version does"
        )
    return version


def read_kb_state() -> Optional[dict]:
    path = kb_state_file()
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError) as exc:
        raise VersionError(f"{KB_STATE_FILENAME} is not readable JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise VersionError(f"{KB_STATE_FILENAME} does not contain a JSON object")
    return data


def render_kb_state(version: Version, applied: list[dict]) -> str:
    return (
        json.dumps(
            {"schema": KB_STATE_SCHEMA, "kb_version": str(version), "applied": applied},
            indent=2,
        )
        + "\n"
    )


def write_kb_state(version: Version, applied: list[dict]) -> None:
    kb_state_file().write_text(render_kb_state(version, applied), encoding="utf-8")


def migrations_dir() -> Path:
    return config.INSTRUCTIONS_DIR / MIGRATIONS_SUBDIR


def load_migrations(directory: Optional[Path] = None) -> list[Migration]:
    """Every migration document under `directory`, sorted by target version.

    `directory` defaults to this instance's own `instructions/migrations/`.
    `dist upgrade` (Gitea #7) passes the *new* tree's migrations directory
    instead: the migrations owed after an upgrade are documented in the
    machinery being installed, not in the one still on disk - an old instance
    cannot know a new version's migration chain by reading its own tree.

    A malformed document is skipped rather than fatal here - `instructions
    verify` is what reports it, and `migrate status` staying usable while one
    document is broken is worth more than a second error path.
    """
    from chemenu.frontmatter_io import read_page

    base = directory if directory is not None else migrations_dir()
    if not base.is_dir():
        return []

    migrations: list[Migration] = []
    for path in sorted(base.glob("*.md")):
        try:
            frontmatter, _ = read_page(path)
        except Exception:  # noqa: BLE001 - a broken document is verify's finding, not ours
            continue
        raw_target = frontmatter.get("migrates_to")
        if not raw_target:
            continue
        try:
            target = Version.parse(str(raw_target))
        except VersionError:
            continue
        obligation = str(frontmatter.get("obligation") or REQUIRED)
        migrations.append(
            Migration(
                name=str(frontmatter.get("name") or path.stem),
                target=target,
                kind=str(frontmatter.get("migration_kind") or "assisted"),
                description=str(frontmatter.get("description") or ""),
                path=path,
                obligation=obligation if obligation in OBLIGATIONS else REQUIRED,
            )
        )
    return sorted(migrations, key=lambda m: m.target)


def chain(
    migrations: list[Migration], kb_version: Version, stack_version: Version
) -> list[Migration]:
    """The migrations still owed, in the order they must run.

    Every **required** migration whose target lies in
    `(kb_version, stack_version]`, oldest first. An instance at 1.3.1 upgrading
    to 2.0.0 gets 1.4.0, 1.7.0, 2.0.0 - and the absence of any migration
    targeting 1.3.x is not a special case, it simply is not in the interval.
    Targets above the installed machinery are excluded: the instance has no code
    for them yet.

    `stack_version` must be release-shaped (no `-beta.N`) - pass `.base` when
    the installed machinery is a running candidate. A migration document
    targets a release (`migrates_to: 4.4.0`), and a candidate's own version
    sorts *before* that release (`4.4.0-beta.1 < 4.4.0`), so comparing against
    the raw candidate would drop its own target out of the interval right
    when the machinery that owes it is installed.

    `offered` migrations are deliberately absent. They are not links in the
    version chain: declining one leaves the content in a shape the machinery
    still accepts, so counting it as owed would make `kb_version` unreachable
    for an instance that simply kept its own file.
    """
    return [
        m for m in migrations if m.is_required and kb_version < m.target <= stack_version
    ]


def applied_names(state: Optional[dict]) -> set[str]:
    """Every migration this instance has recorded as carried out."""
    entries = (state or {}).get("applied") or []
    return {
        str(entry.get("migration"))
        for entry in entries
        if isinstance(entry, dict) and entry.get("migration")
    }


def offers(migrations: list[Migration], applied: set[str]) -> list[Migration]:
    """Optional upgrades this instance has not taken, oldest target first.

    Bounded by the **applied ledger**, not by `kb_version`, and that is not a
    detail: taking an offer deliberately does not move `kb_version`, so the
    version says nothing about whether an offer was taken. Filtering by it
    would hide every offer the moment some unrelated required migration ran.

    Not bounded above by the stack version either. An offer is about a file the
    instance owns rather than about the shape of its content, so it stays on the
    table until it is recorded - or until the operator deletes the document.
    """
    return [m for m in migrations if not m.is_required and m.name not in applied]


def next_link(
    migrations: list[Migration], kb_version: Version, stack_version: Version
) -> Optional[Migration]:
    pending = chain(migrations, kb_version, stack_version)
    return pending[0] if pending else None


# --- what this instance changed about what it was given --------------------

# What `compare_against_stamp` answers for one path: present and matching its
# recorded digest, present but not matching, or gone entirely. `dist upgrade`
# needs the three-way answer to tell a locally deleted file from a locally
# edited one; `divergent_files` (below) only ever needed the yes/no of
# "does this count as diverged", which both `MODIFIED` and `DELETED` answer
# the same way.
UNCHANGED = "unchanged"
MODIFIED = "modified"
DELETED = "deleted"


def compare_against_stamp(stamp_files: dict, root: Optional[Path] = None) -> dict[str, str]:
    """Classify every path in `stamp_files` (relative -> recorded sha256, the
    shape of a release stamp's own `files` block) against what is actually on
    disk under `root` - `UNCHANGED`, `MODIFIED`, or `DELETED`.

    `root` defaults to `config.ROOT`. The general form `divergent_files` is
    built on: that function only ever asks the question against this
    instance's own tree, but `dist upgrade` (Gitea #7) asks it against an
    already-installed tree while planning what to write, and a *second* time
    against the tree it just wrote, before recording the new stamp - two trees
    neither of which is necessarily `config.ROOT`.
    """
    import hashlib

    base = root if root is not None else config.ROOT
    result: dict[str, str] = {}
    for relative, digest in sorted(stamp_files.items()):
        path = base / relative
        if not path.is_file():
            result[relative] = DELETED
            continue
        current = "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest()
        result[relative] = UNCHANGED if current == digest else MODIFIED
    return result


def divergent_files() -> Optional[list[str]]:
    """Files whose content no longer matches the release this instance installed.

    Reads the per-file sha256 in `.wikitool-release.json`, which `dist export`
    has been writing since the stamp existed. Its own docstring says why it is
    there: it is the only way a later upgrade can tell a file the instance
    *edited* from one it merely *received* - `dist upgrade` (Gitea #7) is that
    later upgrade, built on the general `compare_against_stamp` above.

    That distinction is also what makes an `offered` migration actionable. The
    stack proposing a better `entity` template needs to know whether it may be
    copied over or whether the instance has its own version that a person has
    to reconcile - and only the recorded hash can answer that.

    Returns None when the question is unanswerable (a development tree, which
    carries no stamp), which is different from `[]` (nothing diverged).
    """
    from chemenu import version as version_mod

    try:
        stamp = version_mod.read_stamp()
    except VersionError:
        return None
    if not stamp:
        return None
    recorded = stamp.get("files")
    if not isinstance(recorded, dict):
        return None

    statuses = compare_against_stamp(recorded)
    return [relative for relative, status in statuses.items() if status != UNCHANGED]