Validate REPL imports with the selected Lean toolchain

servers/README.md

@@ -13,6 +13,10 @@ runtime for the current AutoformBot installation, Unix user, and compute node.
 That runtime owns one REPL admission pool and LSP session per active Lean
 project. LSP sessions stay warm; each public REPL call gets a fresh child and
 no process-owned environment or proof-state handle survives the response.
+Before that child starts, the selected Lake toolchain's `lean --deps-json`
+parser validates the submitted header and its complete import set; a rejected
+or unrecognized parser response fails closed. Execution names the
+package-qualified `@repl/repl` target so a project target cannot shadow it.
 Closing the session that started it does not stop it; after a crash, the next
 tool call starts it again. Runtime sockets include a code fingerprint, so an
 in-place upgrade gracefully replaces the older build.

servers/lean_runtime.py

@@ -172,7 +172,7 @@ def from_environment(cls) -> "LeanRuntimeConfig":
                 "AUTOFORM_REPL_TOTAL_WORKERS"
             )
         repl_project_limit = min(max_projects, total_workers // workers_per_project)
-        repl_command = tuple(shlex.split(os.environ.get("LEAN_REPL_CMD", "lake exe repl")))
+        repl_command = tuple(shlex.split(os.environ.get("LEAN_REPL_CMD", "lake exe @repl/repl")))
         lsp_command = tuple(shlex.split(os.environ.get("LEAN_LSP_CMD", "lake serve")))
         if not repl_command:
             raise ValueError("LEAN_REPL_CMD must not be empty")

servers/repl/core.py

@@ -13,6 +13,7 @@
 import select
 import signal
 import subprocess
+import sys
 import threading
 import time
 from dataclasses import dataclass, field
@@ -34,6 +35,11 @@
 _STDERR_TAIL_BYTES = 200
 _PUBLIC_DIAGNOSTIC_FIELDS = frozenset({"severity", "data", "pos", "endPos"})
 _PUBLIC_SORRY_FIELDS = frozenset({"goal", "pos", "endPos"})
+_LEAN_HEADER_LAUNCHER = (
+    "import os; "
+    "lean = os.path.join(os.environ['LEAN_SYSROOT'], 'bin', 'lean'); "
+    "os.execv(lean, [lean, '--deps-json', '/dev/stdin'])"
+)
 
 
 # ---------------------------------------------------------------------------
@@ -290,6 +296,191 @@ def _validate_command_response(
     return environment, messages
 
 
+def _communicate_bounded(
+    process: subprocess.Popen[bytes],
+    input_bytes: bytes,
+    *,
+    deadline: float,
+    max_output_bytes: int,
+) -> tuple[bytes, bytes]:
+    """Exchange bytes with a child without unbounded output or time."""
+    stdin_fd = process.stdin.fileno()
+    stdout_fd = process.stdout.fileno()
+    stderr_fd = process.stderr.fileno()
+    for fd in (stdin_fd, stdout_fd, stderr_fd):
+        os.set_blocking(fd, False)
+
+    pending = memoryview(input_bytes)
+    offset = 0
+    stdin_open = True
+    stdout_open = True
+    stderr_open = True
+    stdout = bytearray()
+    stderr = bytearray()
+
+    def close_stdin() -> None:
+        nonlocal stdin_open
+        if not stdin_open:
+            return
+        stdin_open = False
+        try:
+            process.stdin.close()
+        except OSError:
+            pass
+
+    while stdin_open or stdout_open or stderr_open:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError("timed out checking the Lean header")
+        readable_fds: list[int] = []
+        if stdout_open:
+            readable_fds.append(stdout_fd)
+        if stderr_open:
+            readable_fds.append(stderr_fd)
+        writable_fds = [stdin_fd] if stdin_open else []
+        try:
+            readable, writable, _ = select.select(
+                readable_fds,
+                writable_fds,
+                [],
+                remaining,
+            )
+        except InterruptedError:
+            continue
+        if not readable and not writable:
+            raise TimeoutError("timed out checking the Lean header")
+
+        for fd, target in ((stdout_fd, stdout), (stderr_fd, stderr)):
+            if fd not in readable:
+                continue
+            try:
+                chunk = os.read(fd, 65536)
+            except BlockingIOError:
+                continue
+            if not chunk:
+                if fd == stdout_fd:
+                    stdout_open = False
+                else:
+                    stderr_open = False
+                continue
+            if len(stdout) + len(stderr) + len(chunk) > max_output_bytes:
+                raise ValueError(
+                    f"Lean header parser output exceeded {max_output_bytes} bytes"
+                )
+            target.extend(chunk)
+
+        if stdin_fd in writable:
+            try:
+                written = os.write(stdin_fd, pending[offset : offset + 65536])
+            except BlockingIOError:
+                continue
+            except BrokenPipeError:
+                close_stdin()
+            else:
+                if written <= 0:
+                    close_stdin()
+                else:
+                    offset += written
+                    if offset == len(pending):
+                        close_stdin()
+
+    close_stdin()
+    if process.poll() is None:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError("timed out checking the Lean header")
+        try:
+            process.wait(timeout=remaining)
+        except subprocess.TimeoutExpired:
+            raise TimeoutError("timed out checking the Lean header") from None
+    return bytes(stdout), bytes(stderr)
+
+
+@dataclass(frozen=True)
+class _LeanHeaderAnalysis:
+    """Lean-owned facts needed to validate and compose one submitted header."""
+
+    modules: tuple[str, ...]
+    accepts_leading_imports: bool
+
+
+def _decode_header_analysis(stdout: bytes) -> _LeanHeaderAnalysis:
+    """Decode the strict schemas emitted by Lean's fast import parser."""
+    try:
+        payload = _decode_repl_json(stdout)
+        if not isinstance(payload, dict) or set(payload) != {"imports"}:
+            raise ValueError
+        entries = payload["imports"]
+        if not isinstance(entries, list) or len(entries) != 1:
+            raise ValueError
+        entry = entries[0]
+        if not isinstance(entry, dict):
+            raise ValueError
+        errors = entry.get("errors")
+        if not isinstance(errors, list) or not all(
+            isinstance(error, str) and error for error in errors
+        ):
+            raise ValueError
+        if errors:
+            raise ValueError(errors[0])
+
+        has_result = "result" in entry
+        has_imports = "imports" in entry
+        if has_result == has_imports:
+            raise ValueError
+        is_module: bool | None = None
+        if has_result:
+            result = entry["result"]
+            if not isinstance(result, dict):
+                raise ValueError
+            imports = result.get("imports")
+            is_module = result.get("isModule")
+            if is_module is not None and type(is_module) is not bool:
+                raise ValueError
+        else:
+            imports = entry["imports"]
+        if not isinstance(imports, list):
+            raise ValueError
+
+        modules: list[str] = []
+        has_implicit_init = False
+        has_composition_metadata = has_result and is_module is not None
+        for item in imports:
+            if not isinstance(item, dict):
+                raise ValueError
+            module = item.get("module")
+            if not isinstance(module, str) or not module:
+                raise ValueError
+            if has_result:
+                is_meta = item.get("isMeta")
+                if is_meta is not None and type(is_meta) is not bool:
+                    raise ValueError
+                if is_meta is None:
+                    has_composition_metadata = False
+                if module == "Init" and is_meta:
+                    has_implicit_init = True
+            if module != "Init":
+                modules.append(module)
+        # Only the current schema exposes enough parser state to distinguish
+        # Lean's implicit Init imports from an explicit ``prelude``/``module``
+        # header. The legacy schema remains valid for allowlist checks, but
+        # fails closed on source rewriting and sends the validated frame exact.
+        return _LeanHeaderAnalysis(
+            modules=tuple(modules),
+            accepts_leading_imports=(
+                has_composition_metadata
+                and has_implicit_init
+                and is_module is False
+            ),
+        )
+    except (ReplProtocolError, TypeError, KeyError, json.JSONDecodeError):
+        raise ValueError("unrecognized output from lean --deps-json") from None
+    except ValueError as error:
+        if str(error):
+            raise
+        raise ValueError("unrecognized output from lean --deps-json") from None
+
+
 def _split_imports_and_body(code: str) -> tuple[list[str], str, int]:
     """Split Lean code into import statements and body.
 
@@ -336,7 +527,10 @@ class LeanReplConfig:
     allowed_imports: frozenset[str] = ALLOWED_IMPORTS
     warmup_imports: frozenset[str] = WARMUP_IMPORTS
 
-    repl_command: list[str] = field(default_factory=lambda: ["lake", "exe", "repl"])
+    repl_command: list[str] = field(default_factory=lambda: ["lake", "exe", "@repl/repl"])
+    header_deps_command: list[str] = field(
+        default_factory=lambda: ["lake", "env", sys.executable, "-c", _LEAN_HEADER_LAUNCHER]
+    )
 
     # stdout is capped per response. stderr has no protocol framing, so its
     # ceiling applies to the entire process generation and resets on restart.
@@ -726,6 +920,37 @@ def is_clean(self) -> bool:
         """Return whether this wrapper owns no process generation."""
         return self.process is None and self._process_group_id is None
 
+    def _check_header(self, code: str, deadline: float) -> _LeanHeaderAnalysis:
+        """Analyze the submitted header with the selected toolchain's Lean parser.
+
+        The parser is this wrapper's process generation until close() verifies
+        that it exited, so a failed cleanup is retried before the slot is reused.
+        """
+        env = _inherit_clean_env()
+        env.update(self.config.env)
+        self.process = subprocess.Popen(
+            self.config.header_deps_command,
+            cwd=self.cwd,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=env,
+            start_new_session=True,
+        )
+        self._process_group_id = self.process.pid
+        stdout, stderr = _communicate_bounded(
+            self.process,
+            code.encode(),
+            deadline=deadline,
+            max_output_bytes=self.config.max_buffer_bytes,
+        )
+        returncode = self.process.returncode
+        self.close(deadline=deadline)
+        if returncode != 0:
+            detail = stderr.decode(errors="replace").strip().splitlines()
+            raise ValueError(detail[0] if detail else f"exit status {returncode}")
+        return _decode_header_analysis(stdout)
+
     def run_disposable(
         self,
         code: str,
@@ -752,12 +977,26 @@ def remaining() -> float:
                 # generation can be created from this slot.
                 self.close(deadline=deadline)
                 imports, _, _ = _split_imports_and_body(code)
-                submitted_roots = {statement.split(".")[0] for statement in imports}
+                accepts_leading_imports = True
                 if (
                     self.config.validate_imports
                     and self._allowed_import_roots is not None
                 ):
-                    disallowed = submitted_roots - self._allowed_import_roots
+                    try:
+                        header = self._check_header(code, deadline)
+                    except ValueError as error:
+                        result = {"repl_error": f"Rejected Lean header: {error}"}
+                    else:
+                        imports = list(header.modules)
+                        accepts_leading_imports = header.accepts_leading_imports
+                submitted_roots = {statement.split(".")[0] for statement in imports}
+                if (
+                    result is None
+                    and self.config.validate_imports
+                    and self._allowed_import_roots is not None
+                ):
+                    warmup_roots = {root.split(".")[0] for root in self.config.warmup_imports}
+                    disallowed = (submitted_roots | warmup_roots) - self._allowed_import_roots
                     if disallowed:
                         result = {
                             "repl_error": (
@@ -770,7 +1009,7 @@ def remaining() -> float:
                     added_imports = tuple(
                         root
                         for root in sorted(self.config.warmup_imports)
-                        if root not in submitted_roots
+                        if accepts_leading_imports and root not in submitted_roots
                     )
                     prefix = "\n".join(
                         f"import {root}" for root in added_imports
@@ -813,7 +1052,7 @@ def remaining() -> float:
                 result = {"repl_error": str(error)}
             except (ReplProtocolError, ReplOutcomeUnknown) as error:
                 result = {"repl_error": str(error), "outcome_unknown": True}
-            except (ReplProcessExited, TimeoutError, RuntimeError) as error:
+            except (ReplProcessExited, TimeoutError, RuntimeError, OSError) as error:
                 result = {"repl_error": str(error)}
             except BaseException as error:
                 request_error = error

skills/setup/SKILL.md

@@ -82,6 +82,11 @@ For an incomplete existing repository, preserve its authored configuration and
 run `autoform init` through the same `uv run --project "$AUTOFORM_PLUGIN_ROOT"`
 prefix only for the Autoform vault/site repair overlay.
 
+Neither `project new` nor `init` adds the Lean REPL that the REPL tool runs.
+Declare `leanprover-community/repl` at an immutable revision tested against the
+project's exact Lean toolchain, such as the bundled example's pin for v4.32.2,
+then verify `lake build @repl/repl`; never guess a compatible revision.
+
 `autoform init` is the whole vault: `blueprint/` with its landing page,
 `roadmap/README.md`, `coverage/`, and `sources/`, plus `mkdocs.yml`, the theme
 override, both workflows, and ignore rules. Do not hand-build any of it and do
@@ -138,6 +143,7 @@ then run the publication sequence:
 lake update          # only when the project has no lake-manifest.json
 lake exe cache get   # skip only when the project has no Mathlib dependency
 lake build
+lake build @repl/repl
 ```
 
 Then validate, visualize, render, and strict-build the site, keeping

skills/setup/assets/cabannes-thesis-project/README.md

@@ -9,7 +9,7 @@ slice as a handoff example.
 Developed with [AutoformBot](https://github.com/facebookresearch/autoform-bot).
 
 - `lean-toolchain`, `lakefile.toml`, and `CabannesThesis/` pin matching stable
-  Lean and Mathlib `v4.32.2` releases.
+  Lean and Mathlib `v4.32.2` releases plus the compatible upstream REPL commit.
 - `blueprint/` is an Obsidian-compatible Markdown vault with roadmap, coverage,
   sources, and a seven-node theorem DAG spanning two formalization chapters.
 - `mkdocs.yml` builds the `autoform render` output as a leanblueprint-styled