chemenu/tools/chemenu/version.py

"""The stack's own version: what `VERSION` holds, how a release stamps an
instance, and when two versions are compatible.

The version describes the **stack** - `tools/`, `types/`, `instructions/`,
`AGENTS.md` and the contracts - never the wiki content sitting next to it in
the same repo. That split is the whole reason it is set explicitly rather than
derived from commit messages: `publish --message "ingest: ..."` writes content
commits into this same repo, so a conventional-commit reading would turn every
ingest into a release.

**Compatibility is read off the leftmost non-zero component**, the rule Cargo's
caret ranges use: `0.1.3 -> 0.1.4` is safe, `0.1.3 -> 0.2.0` is not, and from
`1.0.0` on the same rule reads as the familiar "MAJOR means migration". Stating
it that way is what lets the 0.x era carry the migration signal at all - under
plain "MAJOR breaks" semantics every 0.x release would be indistinguishable
from every other, which is exactly the signal update detection needs. Nothing
about the mechanism changes at 1.0.0.

Paths are resolved through `config.ROOT` at call time rather than at import,
because the tests (and `dist export`'s own fixtures) relocate the root.
"""
from __future__ import annotations

import json
import os
import re
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional

from chemenu import config

VERSION_FILENAME = "VERSION"
CHANGES_FILENAME = "CHANGES.md"

# Written into an exported distribution by `dist export`, and committed with
# it: an instance has to keep knowing which stack it came from, or update
# detection has nothing to compare against. Machine-written, never hand-edited.
RELEASE_STAMP_FILENAME = ".wikitool-release.json"
STAMP_SCHEMA = 1

# Where `version check` looks when neither the environment nor a release stamp
# names something else. A fork changes this line; an instance exported from a
# fork carries the fork's URL in its stamp and never reaches this default.
DEFAULT_UPDATE_URL = (
    "https://gitea.nehmer.net/api/v1/repos/torben/chemenu/releases/latest"
)
UPDATE_URL_ENV = "WIKITOOL_UPDATE_URL"
# Optional: only needed if the origin's release feed is not readable
# anonymously. Absent is the normal case, not a misconfiguration.
UPDATE_TOKEN_ENV = "WIKITOOL_UPDATE_TOKEN"

PARTS = ("major", "minor", "patch")

# Plain `x.y.z` only: no `-rc1`, no `+build`. Pre-release channels would mean a
# second ordering rule everywhere a version is compared - the release feed, the
# migration chain, the compatibility check - to serve a workflow this stack does
# not have.
_SEMVER_RE = re.compile(r"^\s*v?(\d+)\.(\d+)\.(\d+)\s*$")

# Written into a CHANGES.md entry whose version crosses a compatibility
# boundary that needs no content migration. `docs verify` accepts it in place
# of a migration document, so the exact string is a contract between the two.
MIGRATION_NONE_MARKER = "**Migration:** none required"
# A changelog entry that names a version. Entries predating versioning start
# with a date instead and are deliberately not matched - they are history, not
# a claim about which version the tree is.
_CHANGES_ENTRY_RE = re.compile(r"^## (\d+\.\d+\.\d+)(?: - (.*))?$", re.MULTILINE)


class VersionError(ValueError):
    """A version could not be read, parsed, or fetched. Carries a message
    written to be shown to the user verbatim."""


@dataclass(frozen=True, order=True)
class Version:
    major: int
    minor: int
    patch: int

    @classmethod
    def parse(cls, text: str) -> "Version":
        match = _SEMVER_RE.match(text or "")
        if not match:
            raise VersionError(
                f"{text.strip()!r} is not a semantic version - expected MAJOR.MINOR.PATCH"
            )
        return cls(int(match.group(1)), int(match.group(2)), int(match.group(3)))

    def __str__(self) -> str:  # noqa: D105 - obvious
        return f"{self.major}.{self.minor}.{self.patch}"

    def bumped(self, part: str) -> "Version":
        if part == "major":
            return Version(self.major + 1, 0, 0)
        if part == "minor":
            return Version(self.major, self.minor + 1, 0)
        if part == "patch":
            return Version(self.major, self.minor, self.patch + 1)
        raise VersionError(f"unknown version part {part!r} - expected one of {', '.join(PARTS)}")

    @property
    def compat_key(self) -> tuple[int, ...]:
        """The prefix up to and including the leftmost non-zero component.

        Two versions are compatible exactly when this is equal. `0.1.3` and
        `0.1.9` share `(0, 1)`; `0.2.0` does not. An all-zero version has no
        non-zero component, so it compares by all three - during `0.0.x`
        every release is a breaking one, which is what that range means.
        """
        components = (self.major, self.minor, self.patch)
        for index, component in enumerate(components):
            if component:
                return components[: index + 1]
        return components


@dataclass(frozen=True)
class UpdateStatus:
    """The answer `version check` reports. `state` is the actionable part:
    `migration` is not a louder `update`, it is a different instruction."""

    local: Version
    latest: Version
    state: str  # "current" | "update" | "migration" | "ahead"
    release_url: Optional[str] = None
    published_at: Optional[str] = None

    @property
    def headline(self) -> str:
        if self.state == "current":
            return f"Up to date: {self.local} is the latest release."
        if self.state == "ahead":
            return (
                f"Local stack {self.local} is ahead of the latest release {self.latest} "
                "- an unreleased tree."
            )
        if self.state == "migration":
            return (
                f"Update available: {self.local} -> {self.latest}. This crosses a "
                "compatibility boundary - the release notes name the migration required."
            )
        return f"Update available: {self.local} -> {self.latest} (compatible)."


def compare(local: Version, latest: Version) -> str:
    if latest == local:
        return "current"
    if latest < local:
        return "ahead"
    return "update" if latest.compat_key == local.compat_key else "migration"


def version_file() -> Path:
    return config.ROOT / VERSION_FILENAME


def stamp_file() -> Path:
    return config.ROOT / RELEASE_STAMP_FILENAME


def changes_file() -> Path:
    return config.ROOT / CHANGES_FILENAME


def read_version() -> Version:
    """This tree's stack version. Raises rather than guessing: a stack with no
    declared version cannot answer "is there an update", and a placeholder
    would answer it wrongly."""
    path = version_file()
    if not path.is_file():
        raise VersionError(
            f"{VERSION_FILENAME} is missing - this tree declares no stack version"
        )
    return Version.parse(path.read_text(encoding="utf-8"))


def write_version(version: Version) -> None:
    version_file().write_text(f"{version}\n", encoding="utf-8")


def read_stamp() -> Optional[dict]:
    """The release stamp, if this tree came from one. `None` is a normal
    answer - a dev checkout has no stamp - so a malformed one is the only
    case worth failing over."""
    path = stamp_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"{RELEASE_STAMP_FILENAME} is not readable JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise VersionError(f"{RELEASE_STAMP_FILENAME} does not contain a JSON object")
    return data


def update_url(stamp: Optional[dict] = None) -> str:
    """Where to ask for the latest release: the environment overrides, then
    the stamp this instance was exported with, then the compiled-in default."""
    override = os.environ.get(UPDATE_URL_ENV, "").strip()
    if override:
        return override
    if stamp:
        from_stamp = str(stamp.get("update_url") or "").strip()
        if from_stamp:
            return from_stamp
    return DEFAULT_UPDATE_URL


Fetcher = Callable[[str, Optional[str], float], bytes]


def _urlopen_fetch(url: str, token: Optional[str], timeout: float) -> bytes:
    request = urllib.request.Request(url, headers={"Accept": "application/json"})
    if token:
        request.add_header("Authorization", f"token {token}")
    with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310 - explicit https URL
        return response.read()


def fetch_latest(
    url: str,
    token: Optional[str] = None,
    timeout: float = 10.0,
    fetcher: Optional[Fetcher] = None,
) -> Version:
    """The version the release feed reports as latest.

    The network call sits behind `fetcher` so every caller above this line -
    and every test - can run without a network. This is the one place in
    `wikitool` that talks to a remote host, and it is reached only from
    `version check`, never implicitly from another command.
    """
    fetch = fetcher or _urlopen_fetch
    try:
        payload = fetch(url, token, timeout)
    except urllib.error.HTTPError as exc:
        hint = ""
        if exc.code in (401, 403):
            hint = f" - the feed needs authentication; set ${UPDATE_TOKEN_ENV}"
        elif exc.code == 404:
            hint = " - no release published yet, or the URL names the wrong repository"
        raise VersionError(f"{url} answered HTTP {exc.code}{hint}") from exc
    except (urllib.error.URLError, OSError, TimeoutError) as exc:
        raise VersionError(f"Could not reach {url}: {exc}") from exc

    try:
        data = json.loads(payload)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise VersionError(f"{url} did not answer with JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise VersionError(f"{url} answered with JSON that is not an object")

    tag = str(data.get("tag_name") or "").strip()
    if not tag:
        raise VersionError(f"{url} answered without a `tag_name` - not a release feed")
    return Version.parse(tag)


def fetch_latest_release(
    url: str,
    token: Optional[str] = None,
    timeout: float = 10.0,
    fetcher: Optional[Fetcher] = None,
) -> tuple[Version, Optional[str], Optional[str]]:
    """`fetch_latest` plus the two display fields a report wants: the release's
    own page and its publication date."""
    fetch = fetcher or _urlopen_fetch
    captured: dict = {}

    def capturing(u: str, t: Optional[str], to: float) -> bytes:
        payload = fetch(u, t, to)
        try:
            parsed = json.loads(payload)
            if isinstance(parsed, dict):
                captured.update(parsed)
        except (json.JSONDecodeError, UnicodeDecodeError):
            pass
        return payload

    version = fetch_latest(url, token, timeout, capturing)
    html_url = captured.get("html_url") or captured.get("url")
    published = captured.get("published_at") or captured.get("created_at")
    return version, (str(html_url) if html_url else None), (str(published) if published else None)


# --- CHANGES.md ------------------------------------------------------------
#
# The changelog is prose and stays the author's job. What is mechanical is the
# heading - which version, which date - and checking that the topmost one
# agrees with VERSION. Same split as `new`: the tool writes structure, the LLM
# writes the entry.


def top_changes_version(text: str) -> Optional[Version]:
    """The version named by the topmost versioned entry, or `None` when the
    changelog has none. `None` is valid: a fresh distribution ships a changelog
    with no entries at all, and this repo's own history predates versioning."""
    match = _CHANGES_ENTRY_RE.search(text)
    if not match:
        return None
    return Version.parse(match.group(1))


def changes_section(text: str, version: Version) -> Optional[str]:
    """The body of one version's entry, heading included, ready to become
    release notes.

    The entry ends at the next `##` heading *of any kind*, not the next
    versioned one: entries below `0.1.0` predate versioning and are headed by
    a date, so terminating on a versioned heading would run the newest entry
    all the way to the end of the file - which is exactly what it did.
    """
    for match in _CHANGES_ENTRY_RE.finditer(text):
        if Version.parse(match.group(1)) != version:
            continue
        rest = text[match.start():]
        following = re.search(r"^## ", rest[1:], re.MULTILINE)
        section = rest[: following.start() + 1] if following else rest
        return section.rstrip().removesuffix("---").rstrip() + "\n"
    return None


def insert_changes_entry(
    text: str,
    version: Version,
    date: str,
    title: str,
    author: str,
    no_migration_reason: Optional[str] = None,
) -> str:
    """Add a heading for `version` above the newest existing entry.

    Only the skeleton: heading, date, author, and - when a compatibility
    boundary is crossed without a migration - the line that says so. The
    entry's actual content is written afterwards by whoever made the change,
    which is also why `bump` refuses to invent a title.
    """
    lines = [f"## {version} - {date} - {title}", "", f"**Author:** {author}", ""]
    if no_migration_reason:
        lines += [f"{MIGRATION_NONE_MARKER} - {no_migration_reason}", ""]
    entry = "\n".join(lines) + "\n---\n\n"
    anchor = re.search(r"^## ", text, re.MULTILINE)
    if anchor:
        return text[: anchor.start()] + entry + text[anchor.start():]
    return text.rstrip() + "\n\n---\n\n" + entry