P928: prepare the Neovim artifact the editor launches

.github/workflows/ci.yml

@@ -60,14 +60,6 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: '3.14'
-      - name: Install Neovim
-        run: |
-          curl --fail --location --silent --show-error \
-            https://github.com/neovim/neovim/releases/download/v0.12.4/nvim-linux-x86_64.tar.gz \
-            --output "$RUNNER_TEMP/neovim.tar.gz"
-          tar --extract --gzip --file "$RUNNER_TEMP/neovim.tar.gz" --directory "$RUNNER_TEMP"
-          echo "$RUNNER_TEMP/nvim-linux-x86_64/bin" >> "$GITHUB_PATH"
-          "$RUNNER_TEMP/nvim-linux-x86_64/bin/nvim" --version
       - uses: ./.github/actions/core-from-main
         with:
           path: .artifacts/core
@@ -88,6 +80,8 @@ jobs:
           with open(os.environ['GITHUB_ENV'], 'a') as env:
               env.write(f'AC_NATIVE_COPIED_PACKAGE={package.resolve()}\n')
           PYCORE
+      - name: Prepare the declared Neovim requirement
+        run: .venv/bin/python -m toad.neovim_runtime
       # xdist reports a collection error as a worker crash; collect alone first so it reads as one.
       - run: .venv/bin/pytest --collect-only -q
       # A test stuck past 120 s dumps every thread, so the red names where it hung.

pyproject.toml

@@ -52,6 +52,9 @@ textual-diff-view = { git = "https://github.com/OpenHCSDev/textual-diff-view.git
 [project.scripts]
 toad = "toad.cli:main"
 
+[project.entry-points."agent_comms.runtime_preparation"]
+neovim = "toad.neovim_runtime:Neovim"
+
 [project.entry-points."agent_comms.writer_owned_stores"]
 toad = "toad.db"
 preferences = "toad.preferences"

src/toad/neovim_editing.py

@@ -27,6 +27,7 @@
 from textual.keys import key_to_character
 from textual.message import Message
 from textual.widgets import TextArea
+from toad.neovim_runtime import Neovim
 
 
 class NeovimKey(DeclaredFamily, affix="Key"):
@@ -873,7 +874,7 @@ async def _run(self) -> None:
         try:
             async with AsyncExitStack() as cleanup:
                 async def acquire_child() -> AttachedChild:
-                    child = await AttachedChild.start(("nvim", "--embed", "--clean", "-n"))
+                    child = await AttachedChild.start(Neovim().command())
                     cleanup.push_async_callback(child.stop)
                     return child
 

src/toad/neovim_runtime.py

@@ -0,0 +1,219 @@
+"""The pinned Neovim artifact, prepared before publication rather than on input."""
+
+from __future__ import annotations
+
+import errno
+import hashlib
+import os
+import platform
+import stat
+import sys
+import tarfile
+from dataclasses import dataclass, field
+from functools import cached_property
+from pathlib import Path, PurePosixPath
+from tempfile import TemporaryDirectory
+from typing import ClassVar
+from urllib.request import urlopen
+
+from agent_comms.declared_family import DeclaredFamily
+from agent_comms.native_build_inputs import Executable
+from agent_comms.runtime_preparation import RuntimePreparation
+from toad.platform import LinuxPlatform, MacPlatform, Platform, host
+
+
+class NeovimUnavailable(OSError):
+    """The declared artifact is missing, unsupported or fails verification."""
+
+
+class Architecture(DeclaredFamily, affix="Architecture"):
+    """Decode external machine spellings once into release architecture identity."""
+
+    tag: ClassVar[str]
+    aliases: ClassVar[tuple[str, ...]] = ()
+
+    @classmethod
+    def decode(cls, machine: str) -> type[Architecture]:
+        spelling = machine.casefold()
+        for member in cls.members_with(Architecture):
+            if spelling in (member.tag, *member.aliases):
+                return member
+        raise NeovimUnavailable(f"No declared Neovim architecture for {machine!r}")
+
+
+class X64Architecture(Architecture):
+    tag = "x86_64"
+    aliases = ("amd64",)
+
+
+class Arm64Architecture(Architecture):
+    tag = "arm64"
+    aliases = ("aarch64",)
+
+
+@dataclass(frozen=True)
+class ReleaseArchive:
+    """Trusted release metadata, not a checksum fetched beside an untrusted download."""
+
+    platform: type[Platform]
+    architecture: type[Architecture]
+    size: int
+    sha256: bytes
+
+    @property
+    def name(self) -> str:
+        return f"{Neovim.program}-{self.platform.key}-{self.architecture.tag}.tar.gz"
+
+    @property
+    def directory(self) -> str:
+        return self.name.removesuffix(".tar.gz")
+
+    def verify(self, path: Path) -> None:
+        if path.stat().st_size != self.size:
+            raise NeovimUnavailable(f"Neovim archive size mismatch: {path}")
+        with path.open("rb") as stream:
+            actual = hashlib.file_digest(stream, "sha256").digest()
+        if actual != self.sha256:
+            raise NeovimUnavailable(f"Neovim archive SHA-256 mismatch: {path}")
+
+    def download(self, url: str, destination: Path) -> None:
+        remaining = self.size
+        with urlopen(url, timeout=30) as response, destination.open("xb") as output:
+            while block := response.read(min(1024 * 1024, remaining + 1)):
+                remaining -= len(block)
+                if remaining < 0:
+                    raise NeovimUnavailable(f"Neovim archive exceeds declared size: {url}")
+                output.write(block)
+        self.verify(destination)
+
+
+@dataclass(frozen=True)
+class Neovim(RuntimePreparation, Executable):
+    """One release declaration supplies preparation, runtime launch and CI."""
+
+    program: ClassVar[str] = "nvim"
+    version: ClassVar[str] = "0.12.4"
+    executable_mode: ClassVar[int] = 0o755
+    # Official neovim/neovim v0.12.4 release metadata, retained by P928.
+    # These are external artifact records, not a roster of editing-mode members.
+    archives: ClassVar[tuple[ReleaseArchive, ...]] = (
+        ReleaseArchive(LinuxPlatform, Arm64Architecture, 11353219, bytes.fromhex(
+            "ceb7e88c6b681f0515d135dcdfad54f5eb4373b25ce6172197cd9a69c758063f")),
+        ReleaseArchive(LinuxPlatform, X64Architecture, 11404178, bytes.fromhex(
+            "012bf3fcac5ade43914df3f174668bf64d05e049a4f032a388c027b1ebd78628")),
+        ReleaseArchive(MacPlatform, Arm64Architecture, 9630250, bytes.fromhex(
+            "51ab83afa66d663627c2ab1be43209b0f4e81360d4598b53efaa4d8195f24c89")),
+        ReleaseArchive(MacPlatform, X64Architecture, 9896697, bytes.fromhex(
+            "03fe16f8dd9f1e9eaf52d5e294913a39917b9e2faea30d7fb0fb385fbd36fe59")),
+    )
+
+    prefix: Path = Path(sys.prefix)
+    system: type[Platform] = host
+    architecture: type[Architecture] = field(
+        default_factory=lambda: Architecture.decode(platform.machine())
+    )
+
+    @cached_property
+    def archive(self) -> ReleaseArchive:
+        for archive in self.archives:
+            if (archive.platform, archive.architecture) == (self.system, self.architecture):
+                return archive
+        raise NeovimUnavailable(
+            f"Neovim {self.version} has no declared asset for "
+            f"{self.system.system}/{self.architecture.tag}"
+        )
+
+    @property
+    def root(self) -> Path:
+        return self.prefix / "share" / "toad" / "neovim" / self.version / self.archive.directory
+
+    @property
+    def payload(self) -> Path:
+        return self.root / self.archive.name
+
+    @property
+    def executable(self) -> Path:
+        path = self.root / self.archive.directory / "bin" / self.program
+        try:
+            info = path.lstat()
+        except FileNotFoundError as error:
+            raise NeovimUnavailable(
+                f"Neovim {self.version} is not prepared at {path}; "
+                "run target-runtime preparation before selecting the native editor"
+            ) from error
+        if (not stat.S_ISREG(info.st_mode)
+                or stat.S_IMODE(info.st_mode) != self.executable_mode):
+            raise NeovimUnavailable(f"Neovim executable has unsafe type or permissions: {path}")
+        return path
+
+    def command(self) -> tuple[str, ...]:
+        """No PATH discovery, download or alternate editor on the runtime path."""
+        return str(self.executable), "--embed", "--clean", "-n"
+
+    def verify(self) -> None:
+        """The trusted archive supplies every extracted file and link's expected answer."""
+        self.archive.verify(self.payload)
+        expected_paths = {self.payload}
+        with tarfile.open(self.payload, "r:gz") as archive:
+            for member in archive:
+                parts = PurePosixPath(member.name).parts
+                if not parts or parts[0] != self.archive.directory:
+                    raise NeovimUnavailable(f"Unexpected Neovim archive root: {member.name}")
+                tarfile.data_filter(member, str(self.root))
+                relative = Path(member.name)
+                path = self.root / relative
+                expected_paths.add(path)
+                expected_paths.update(
+                    self.root / parent for parent in relative.parents if parent != Path(".")
+                )
+                if member.issym():
+                    if not path.is_symlink() or os.readlink(path) != member.linkname:
+                        raise NeovimUnavailable(f"Neovim link differs from archive: {path}")
+                elif member.isfile() or member.islnk():
+                    if path.is_symlink() or not path.is_file():
+                        raise NeovimUnavailable(f"Neovim file missing or replaced: {path}")
+                    expected = archive.extractfile(member)
+                    if expected is None:
+                        raise NeovimUnavailable(f"Neovim archive file has no payload: {member.name}")
+                    with expected, path.open("rb") as actual:
+                        if (hashlib.file_digest(expected, "sha256").digest()
+                                != hashlib.file_digest(actual, "sha256").digest()):
+                            raise NeovimUnavailable(f"Neovim file differs from archive: {path}")
+                elif member.isdir():
+                    if path.is_symlink() or not path.is_dir():
+                        raise NeovimUnavailable(f"Neovim directory missing or replaced: {path}")
+                else:
+                    raise NeovimUnavailable(f"Unsupported Neovim archive member: {member.name}")
+        if set(self.root.rglob("*")) != expected_paths:
+            raise NeovimUnavailable(f"Neovim tree contains undeclared or missing paths: {self.root}")
+        self.executable
+
+    def prepare(self) -> None:
+        """Filesystem admission is separate from use; only a completed tree is published."""
+        if self.root.exists():
+            self.verify()
+            return
+        self.root.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        with TemporaryDirectory(prefix=".neovim-", dir=self.root.parent) as temporary:
+            staged = Neovim(Path(temporary), self.system, self.architecture)
+            staged.root.mkdir(mode=0o700, parents=True)
+            url = (f"https://github.com/neovim/neovim/releases/download/"
+                   f"v{self.version}/{self.archive.name}")
+            self.archive.download(url, staged.payload)
+            with tarfile.open(staged.payload, "r:gz") as archive:
+                archive.extractall(staged.root, filter="data")
+            staged.executable.chmod(self.executable_mode)
+            staged.verify()
+            try:
+                os.rename(staged.root, self.root)
+            except OSError as error:
+                if error.errno not in (errno.EEXIST, errno.ENOTEMPTY):
+                    raise
+                # A concurrent preparer may publish the same complete artifact first.
+                self.verify()
+
+
+if __name__ == "__main__":
+    requirement = Neovim()
+    requirement.prepare()
+    print(requirement.executable)

tests/model/test_neovim_preparation.py

@@ -0,0 +1,94 @@
+"""Real release controls; only the integration owner executes this selection."""
+
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from toad.neovim_runtime import (
+    Neovim, NeovimUnavailable, ReleaseArchive, X64Architecture,
+)
+from toad.platform import WindowsPlatform
+
+
+@pytest.mark.parametrize("archive", Neovim.archives)
+def test_release_selection_uses_the_declared_pair(archive: ReleaseArchive, tmp_path: Path):
+    requirement = Neovim(tmp_path, archive.platform, archive.architecture)
+    assert requirement.archive is archive
+    assert not requirement.root.exists()
+
+
+def test_unsupported_pair_refuses_without_preparation(tmp_path: Path):
+    requirement = Neovim(tmp_path, WindowsPlatform, X64Architecture)
+    with pytest.raises(NeovimUnavailable):
+        requirement.command()
+    assert not (tmp_path / "share").exists()
+
+
+@pytest.fixture(scope="module")
+def prepared_neovim(tmp_path_factory: pytest.TempPathFactory) -> Neovim:
+    requirement = Neovim(tmp_path_factory.mktemp("neovim-release"))
+    assert not requirement.root.exists()
+    requirement.prepare()
+    return requirement
+
+
+def test_first_preparation_runs_the_declared_release(prepared_neovim: Neovim):
+    requirement = prepared_neovim
+    output = subprocess.run(
+        [str(requirement.executable), "--version"],
+        check=True, capture_output=True, text=True, timeout=30,
+    ).stdout
+    assert output.splitlines()[0] == f"NVIM v{requirement.version}"
+    assert requirement.command() == (
+        str(requirement.executable), "--embed", "--clean", "-n"
+    )
+    assert requirement.executable.is_relative_to(requirement.prefix)
+
+
+def test_reusable_artifact_is_verified_without_republication(prepared_neovim: Neovim):
+    paths = (prepared_neovim.payload, prepared_neovim.executable)
+    before = tuple((path.stat().st_ino, path.stat().st_mtime_ns) for path in paths)
+    prepared_neovim.prepare()
+    assert tuple((path.stat().st_ino, path.stat().st_mtime_ns) for path in paths) == before
+
+
+def test_corrupt_archive_is_refused_without_replacement(tmp_path: Path):
+    requirement = Neovim(tmp_path)
+    requirement.root.mkdir(parents=True)
+    requirement.payload.write_bytes(b"unfinished or corrupted release")
+    with pytest.raises(NeovimUnavailable):
+        requirement.prepare()
+    assert requirement.payload.read_bytes() == b"unfinished or corrupted release"
+
+
+def test_corrupt_extracted_executable_is_refused(
+    prepared_neovim: Neovim, tmp_path: Path,
+):
+    requirement = Neovim(tmp_path)
+    shutil.copytree(prepared_neovim.root, requirement.root, symlinks=True)
+    requirement.executable.write_bytes(b"not the declared Neovim executable")
+    with pytest.raises(NeovimUnavailable):
+        requirement.prepare()
+    assert requirement.executable.read_bytes() == b"not the declared Neovim executable"
+
+
+def test_same_size_archive_corruption_is_refused(
+    prepared_neovim: Neovim, tmp_path: Path,
+):
+    requirement = Neovim(tmp_path)
+    requirement.root.mkdir(parents=True)
+    shutil.copyfile(prepared_neovim.payload, requirement.payload)
+    with requirement.payload.open("r+b") as stream:
+        original = stream.read(1)
+        stream.seek(0)
+        stream.write(bytes((original[0] ^ 1,)))
+    with pytest.raises(NeovimUnavailable):
+        requirement.prepare()
+
+
+def test_runtime_does_not_prepare_or_search_path(tmp_path: Path):
+    requirement = Neovim(tmp_path)
+    with pytest.raises(NeovimUnavailable):
+        requirement.command()
+    assert not requirement.root.exists()

tests/textual/test_prompt_enter_insert.py

@@ -99,6 +99,10 @@ async def test_standard_chooser_retires_neovim_and_saved_selection_reopens(works
         await ui.until(lambda: ui.conversation.prompt.prompt_text_area.key_mode is Standard,
                        why="the actual saved Standard preference after reopen")
         area = ui.conversation.prompt.prompt_text_area
+        assert type(area.editing_mode) is Standard
         area.focus()
         await ui.until(lambda: str(ui.conversation.prompt.editing_mode_label.render()) == Standard.label,
                        why="the saved Standard label, without starting the engine")
+        await ui.type("B o stays native")
+        assert "B o stays native" in area.text
+        assert type(area.editing_mode) is Standard