fix(plugins): recover interrupted Python environment rebuilds

docs/plugins.md

@@ -153,6 +153,11 @@ that deliberately detach into new sessions or write other files.
 
 Failed Python-version rebuilds restore the previous venv and remain retryable;
 the new marker is accepted only after the new environment and entrypoint exist.
+A rebuild-intent record is reconciled before version checks after a manager
+restart, so interruption during replacement cannot silently accept a partial
+venv. This provides process-interruption recovery, not guaranteed filesystem
+durability across power loss; interruption during cleanup can leave an unused
+backup.
 
 Archive extraction limits expanded member size and total size and rejects
 symlink-backed destinations. Catalogue JSON and wheel response bodies have size

repeater/plugins/runtime.py

@@ -153,6 +153,24 @@ class PluginRuntime:
         version = str(state.get("version") or manifest.version)
         paths = self.storage.paths_for(plugin_id)
         venv_path = paths.venv_dir(version)
+        # Reconcile before either missing-config or matching-version early return:
+        # a newly created pyvenv.cfg does not mean pip/entrypoint validation finished.
+        rebuild_marker = venv_path.with_name(".venv-rebuild-in-progress")
+        if rebuild_marker.exists():
+            backup_name = rebuild_marker.read_text(encoding="ascii")
+            suffix = backup_name.removeprefix(".venv-backup-")
+            if backup_name != f".venv-backup-{uuid.UUID(hex=suffix).hex}":
+                raise ValueError("invalid venv rebuild backup marker")
+            interrupted_backup = venv_path.with_name(backup_name)
+            if interrupted_backup.exists():
+                if venv_path.exists():
+                    shutil.rmtree(venv_path)
+                interrupted_backup.rename(venv_path)
+            # No backup means interruption before rename or after rollback.
+            if not venv_path.exists():
+                raise RuntimeError("interrupted venv rebuild has no recoverable environment")
+            rebuild_marker.unlink()
+
         config_path = venv_path / "pyvenv.cfg"
         if not config_path.is_file():
             # Legacy/test venvs have no reliable version marker. Let normal
@@ -194,6 +212,9 @@ class PluginRuntime:
         # Build at the final absolute path: moving a freshly built venv would
         # invalidate pip-generated shebangs. Keep the old environment for rollback.
         backup = venv_path.with_name(f".venv-backup-{uuid.uuid4().hex}")
+        # Publish intent atomically before moving the old environment. This is
+        # process-interruption recovery, not a filesystem power-loss guarantee.
+        write_release_asset(venv_path.parent, rebuild_marker.name, backup.name.encode("ascii"))
         venv_path.rename(backup)
         try:
             self._create_venv(venv_path)
@@ -203,8 +224,12 @@ class PluginRuntime:
             if venv_path.exists():
                 shutil.rmtree(venv_path)
             backup.rename(venv_path)
+            rebuild_marker.unlink()
             raise
         else:
+            # Commit only after validation; an interrupted cleanup may leave an
+            # unused backup, but must never restore a partly deleted backup.
+            rebuild_marker.unlink()
             shutil.rmtree(backup)
 
     def _run_install_command(self, cmd, *, timeout, **kwargs):

tests/test_plugin_runtime_hardening.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import shutil
 import signal
 import subprocess
 import sys
@@ -169,6 +170,54 @@ def test_failed_rebuild_restores_old_environment_and_retries(tmp_path):
     assert len([cmd for cmd in calls if "pip" in cmd]) == 2
 
 
+@pytest.mark.parametrize("stage", ["old_renamed", "new_config"])
+def test_fresh_runtime_recovers_interrupted_rebuild_snapshot(tmp_path, stage):
+    snapshot = tmp_path / "snapshot"
+    calls = []
+    capturing = True
+
+    def run(cmd, **kwargs):
+        calls.append(cmd)
+        if "venv" in cmd:
+            target = Path(cmd[-1])
+            if capturing and stage == "old_renamed":
+                shutil.copytree(tmp_path / "plugins", snapshot, symlinks=True)
+                raise RuntimeError("snapshot captured")
+            (target / "bin").mkdir(parents=True)
+            (target / "bin/python").symlink_to(sys.executable)
+            (target / "pyvenv.cfg").write_text(
+                f"version = {sys.version_info.major}.{sys.version_info.minor}.0\n"
+            )
+            if capturing:
+                shutil.copytree(tmp_path / "plugins", snapshot, symlinks=True)
+                raise RuntimeError("snapshot captured")
+        else:
+            entry = Path(cmd[0]).parent / "demo-cli"
+            entry.write_text(f"#!{cmd[0]}\nprint('recovered')\n")
+            entry.chmod(0o755)
+        return subprocess.CompletedProcess(cmd, 0, "", "")
+
+    runtime, paths = prepared_runtime(tmp_path, run_factory=run)
+    venv = paths.venv_dir("1.0.0")
+    (venv / "bin").mkdir(parents=True)
+    (venv / "bin/python").write_text("old")
+    (venv / "pyvenv.cfg").write_text("version = 2.7.0\n")
+    (paths.release_dir("1.0.0") / "demo-1.0.0-py3-none-any.whl").touch()
+    with pytest.raises(RuntimeError, match="snapshot captured"):
+        runtime.ensure_venv_compatible("openhop.demo")
+
+    capturing = False
+    calls.clear()
+    fresh = PluginRuntime(PluginStorage(snapshot), run_factory=run)
+    fresh.ensure_venv_compatible("openhop.demo")
+    assert len([cmd for cmd in calls if "pip" in cmd]) == 1
+    entry = fresh.resolve_entrypoint(fresh.storage.paths_for("openhop.demo"), "1.0.0", "demo-cli")
+    assert subprocess.check_output([str(entry)], text=True, timeout=5).strip() == "recovered"
+    calls.clear()
+    fresh.ensure_venv_compatible("openhop.demo")
+    assert calls == []
+
+
 def test_install_subprocesses_have_deadlines(tmp_path):
     calls = []