Auto-bootstrap BSIPA when re-enabling plugins

Run bootstrap automatically on enable paths when health is bad, skip bsipa
in disable-all, and add tests for the shared bootstrap gate.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
pleb
2026-07-01 22:55:47 -07:00
parent f160e4b349
commit c86e51b85a
8 changed files with 425 additions and 37 deletions
+3 -2
View File
@@ -127,8 +127,9 @@ from an interactive terminal.
The menu reads `plugin-helper.local.toml` when present, shows each discovered
Beat Saber install with its resolved state directory, and lets you toggle
managed plugins with arrow keys and Space. In the plugin table, use `d` to
disable all currently enabled managed plugins and `e` to enable all currently
disabled managed plugins.
disable all currently enabled managed plugins (except `bsipa`) and `e` to enable all
currently disabled managed plugins. Re-enabling a plugin auto-bootstraps BSIPA when
needed.
The individual subcommands are mostly for automation and debugging. If you use
them, pass `--state-dir` directly only when you intentionally want to override
+44 -1
View File
@@ -8,7 +8,7 @@ from pathlib import Path
from typing import Any, Callable
from urllib.request import Request, urlopen
from .bsipa import BSIPA_PLUGIN_ID, check_bsipa_health
from .bsipa import BSIPA_PLUGIN_ID, bootstrap_health_error, check_bsipa_health, planning_requires_bootstrap
from .config import is_windows
from .fsutil import sha256_file
from .installer import apply_plan
@@ -296,3 +296,46 @@ def run_bootstrap(
if completed["returncode"] != 0:
raise RuntimeError(f"IPA.exe -n failed with exit code {completed['returncode']}; state written to {state['statePath']}")
return state
def ensure_healthy_bootstrap(
*,
instance: str,
instance_path: Path,
beat_saber_version: str,
registry: Registry,
lockfile: Lockfile,
state_root: Path,
repo_root: Path,
selected_ids: set[str],
proton: Path | None = None,
native: bool | None = None,
progress: Callable[[str], None] | None = None,
ipa_timeout_seconds: int = 120,
) -> None:
if not planning_requires_bootstrap(lockfile.plugins, selected_ids):
return
health = check_bsipa_health(instance_path, state_root, instance)
if health["ok"]:
return
tell = progress or (lambda _message: None)
tell("BSIPA bootstrap is unhealthy; running bootstrap")
run_bootstrap(
instance=instance,
instance_path=instance_path,
beat_saber_version=beat_saber_version,
registry=registry,
lockfile=lockfile,
state_root=state_root,
repo_root=repo_root,
proton=proton,
native=native,
progress=progress,
ipa_timeout_seconds=ipa_timeout_seconds,
)
health = check_bsipa_health(instance_path, state_root, instance)
if not health["ok"]:
raise ValueError(bootstrap_health_error(health))
+11
View File
@@ -10,6 +10,17 @@ from .state import bootstrap_state_path, load_bootstrap_state
BSIPA_PLUGIN_ID = "bsipa"
def planning_requires_bootstrap(lockfile_plugins: tuple[Any, ...] | list[Any], selected_ids: set[str]) -> bool:
has_locked_bsipa = any(plugin.id == BSIPA_PLUGIN_ID for plugin in lockfile_plugins)
planning_ordinary_plugins = any(plugin_id != BSIPA_PLUGIN_ID for plugin_id in selected_ids)
return has_locked_bsipa and planning_ordinary_plugins
def bootstrap_health_error(health: dict[str, Any]) -> str:
joined = "; ".join(health["messages"])
return f"BSIPA bootstrap is not healthy; run bootstrap first: {joined}"
def latest_log_path(instance_path: Path) -> Path:
return instance_path / "Logs" / "_latest.log"
+2
View File
@@ -504,6 +504,7 @@ def run(argv: list[str] | None = None) -> int:
if args.command == "enable":
instance = get_instance(inst_roots, args.instance)
progress = (lambda message: print(f" {message}", flush=True))
result = enable_disabled_plugin(
instance=args.instance,
instance_path=instance.path,
@@ -511,6 +512,7 @@ def run(argv: list[str] | None = None) -> int:
plugin_id=args.plugin,
registry=args.registry,
lockfile=args.lockfile,
progress=progress,
)
print(f"Enabled: {args.plugin}")
print(f"Plan: {result['planPath']}")
+106 -8
View File
@@ -1,15 +1,34 @@
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
from typing import Any
from .bootstrap import ensure_healthy_bootstrap
from .config import repo_root
from .installer import apply_plan
from .models import load_lockfile, load_registry
from .models import Lockfile, Registry, load_lockfile, load_registry
from .planner import create_plan
from .state import load_installed_state
def _resolve_paths(
*,
instance: str,
registry: str,
lockfile: str | None,
repo: Path | None,
) -> tuple[Path, Path, Path, Lockfile, Registry]:
root = repo or repo_root()
registry_path = (root / registry).resolve() if not Path(registry).is_absolute() else Path(registry)
lock_path = Path(lockfile) if lockfile else root / "locks" / f"{instance}.lock.toml"
if not lock_path.is_absolute():
lock_path = (root / lock_path).resolve()
loaded_lockfile = load_lockfile(lock_path)
loaded_registry = load_registry(registry_path)
return root, registry_path, lock_path, loaded_lockfile, loaded_registry
def enable_disabled_plugin(
*,
instance: str,
@@ -19,25 +38,38 @@ def enable_disabled_plugin(
registry: str = "registry/plugins.toml",
lockfile: str | None = None,
repo: Path | None = None,
progress: Callable[[str], None] | None = None,
) -> dict[str, Any]:
installed_state = load_installed_state(state_root, instance)
if plugin_id not in installed_state.get("disabledPlugins", {}):
raise KeyError(f"plugin is not recorded as disabled: {plugin_id}")
root = repo or repo_root()
registry_path = (root / registry).resolve() if not Path(registry).is_absolute() else Path(registry)
lock_path = Path(lockfile) if lockfile else root / "locks" / f"{instance}.lock.toml"
if not lock_path.is_absolute():
lock_path = (root / lock_path).resolve()
loaded_lockfile = load_lockfile(lock_path)
root, registry_path, _lock_path, loaded_lockfile, loaded_registry = _resolve_paths(
instance=instance,
registry=registry,
lockfile=lockfile,
repo=repo,
)
if not any(plugin.id == plugin_id for plugin in loaded_lockfile.plugins):
raise KeyError(f"plugin is disabled but not locked for this instance: {plugin_id}")
ensure_healthy_bootstrap(
instance=instance,
instance_path=instance_path,
beat_saber_version=loaded_lockfile.beat_saber_version,
registry=loaded_registry,
lockfile=loaded_lockfile,
state_root=state_root,
repo_root=root,
selected_ids={plugin_id},
progress=progress,
)
plan, path = create_plan(
instance=instance,
instance_path=instance_path,
beat_saber_version=loaded_lockfile.beat_saber_version,
registry=load_registry(registry_path),
registry=loaded_registry,
lockfile=loaded_lockfile,
state_root=state_root,
repo_root=root,
@@ -45,3 +77,69 @@ def enable_disabled_plugin(
)
result = apply_plan(plan, state_root)
return {"planPath": str(path), **result}
def enable_disabled_plugins(
*,
instance: str,
instance_path: Path,
state_root: Path,
plugin_ids: list[str],
registry: str = "registry/plugins.toml",
lockfile: str | None = None,
repo: Path | None = None,
progress: Callable[[str], None] | None = None,
) -> dict[str, Any]:
if not plugin_ids:
return {"enabled": [], "errors": []}
root, _registry_path, _lock_path, loaded_lockfile, loaded_registry = _resolve_paths(
instance=instance,
registry=registry,
lockfile=lockfile,
repo=repo,
)
installed_state = load_installed_state(state_root, instance)
disabled_plugins = installed_state.get("disabledPlugins", {})
selected_ids = set(plugin_ids)
ensure_healthy_bootstrap(
instance=instance,
instance_path=instance_path,
beat_saber_version=loaded_lockfile.beat_saber_version,
registry=loaded_registry,
lockfile=loaded_lockfile,
state_root=state_root,
repo_root=root,
selected_ids=selected_ids,
progress=progress,
)
enabled: list[dict[str, Any]] = []
errors: list[dict[str, str]] = []
for plugin_id in plugin_ids:
if plugin_id not in disabled_plugins:
errors.append({"plugin": plugin_id, "error": f"plugin is not recorded as disabled: {plugin_id}"})
continue
if not any(plugin.id == plugin_id for plugin in loaded_lockfile.plugins):
errors.append(
{"plugin": plugin_id, "error": f"plugin is disabled but not locked for this instance: {plugin_id}"}
)
continue
try:
plan, path = create_plan(
instance=instance,
instance_path=instance_path,
beat_saber_version=loaded_lockfile.beat_saber_version,
registry=loaded_registry,
lockfile=loaded_lockfile,
state_root=state_root,
repo_root=root,
selected={plugin_id},
)
result = apply_plan(plan, state_root)
enabled.append({"plugin": plugin_id, "planPath": str(path), "applied": len(result["applied"])})
except Exception as exc:
errors.append({"plugin": plugin_id, "error": str(exc)})
return {"enabled": enabled, "errors": errors}
+3 -6
View File
@@ -9,7 +9,7 @@ from zipfile import ZipFile
from .fsutil import ensure_relative, sha256_bytes, sha256_file
from .models import Lockfile, Registry, VALID_STRATEGIES
from .bsipa import BSIPA_PLUGIN_ID, check_bsipa_health
from .bsipa import bootstrap_health_error, check_bsipa_health, planning_requires_bootstrap
from .state import downloads_dir, plans_dir, plugin_downloads_dir
@@ -81,13 +81,10 @@ def create_plan(
changes: list[dict[str, Any]] = []
warnings: list[str] = []
has_locked_bsipa = any(plugin.id == BSIPA_PLUGIN_ID for plugin in lockfile.plugins)
planning_ordinary_plugins = any(plugin_id != BSIPA_PLUGIN_ID for plugin_id in selected_ids)
if require_bootstrap and has_locked_bsipa and planning_ordinary_plugins:
if require_bootstrap and planning_requires_bootstrap(lockfile.plugins, selected_ids):
health = check_bsipa_health(instance_path, state_root, instance)
if not health["ok"]:
joined = "; ".join(health["messages"])
raise ValueError(f"BSIPA bootstrap is not healthy; run bootstrap first: {joined}")
raise ValueError(bootstrap_health_error(health))
for locked in lockfile.plugins:
if locked.id not in selected_ids:
+33 -18
View File
@@ -9,9 +9,10 @@ from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.widgets import DataTable, Footer, Header, Static
from .bsipa import BSIPA_PLUGIN_ID
from .installer import disable_plugin
from .models import load_lockfile, load_registry
from .operations import enable_disabled_plugin
from .operations import enable_disabled_plugin, enable_disabled_plugins
from .reports import installed_plugins_report
from .state import load_installed_state
@@ -124,6 +125,7 @@ class PluginHelperTui(App[int]):
state_root=target.state_root,
plugin_id=plugin_id,
repo=self.repo_root,
progress=lambda message: self._set_status(f"Bootstrapping: {message}"),
)
self._set_status(f"Enabled {plugin_id}; applied {len(result['applied'])} files.")
else:
@@ -138,7 +140,14 @@ class PluginHelperTui(App[int]):
if self.mode != "plugins" or self.selected_installation is None:
return
target = self.selected_installation
enabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "enabled"]
enabled = [
plugin
for plugin in self.plugin_rows
if plugin["status"] == "enabled" and plugin["id"] != BSIPA_PLUGIN_ID
]
skipped_bsipa = any(
plugin["status"] == "enabled" and plugin["id"] == BSIPA_PLUGIN_ID for plugin in self.plugin_rows
)
changed = 0
errors: list[str] = []
for plugin in enabled:
@@ -157,7 +166,10 @@ class PluginHelperTui(App[int]):
errors.append(f"{plugin_id}: {self._format_skipped(result['skipped'])}")
except Exception as exc:
errors.append(f"{plugin_id}: {exc}")
self._set_bulk_status("Disabled", changed, errors)
if skipped_bsipa and not errors:
self._set_status(f"Disabled {changed} plugins (bsipa kept enabled).")
else:
self._set_bulk_status("Disabled", changed, errors)
self._show_plugins(preserve_status=True)
def action_enable_all_plugins(self) -> None:
@@ -165,21 +177,24 @@ class PluginHelperTui(App[int]):
return
target = self.selected_installation
disabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "disabled"]
changed = 0
errors: list[str] = []
for plugin in disabled:
plugin_id = plugin["id"]
try:
enable_disabled_plugin(
instance=target.instance_name,
instance_path=target.instance_path,
state_root=target.state_root,
plugin_id=plugin_id,
repo=self.repo_root,
)
changed += 1
except Exception as exc:
errors.append(f"{plugin_id}: {exc}")
if not disabled:
self._set_status("No disabled plugins to enable.")
return
plugin_ids = [plugin["id"] for plugin in disabled]
try:
result = enable_disabled_plugins(
instance=target.instance_name,
instance_path=target.instance_path,
state_root=target.state_root,
plugin_ids=plugin_ids,
repo=self.repo_root,
progress=lambda message: self._set_status(f"Bootstrapping: {message}"),
)
except Exception as exc:
self._set_status(f"Could not enable plugins: {exc}")
return
changed = len(result["enabled"])
errors = [f"{item['plugin']}: {item['error']}" for item in result["errors"]]
self._set_bulk_status("Enabled", changed, errors)
self._show_plugins(preserve_status=True)
+223 -2
View File
@@ -13,8 +13,8 @@ from rich.text import Text
from textual.coordinate import Coordinate
from textual.widgets import DataTable
from plugin_helper.bootstrap import _run_ipa, build_bootstrap_command
from plugin_helper.bsipa import check_bsipa_health
from plugin_helper.bootstrap import _run_ipa, build_bootstrap_command, ensure_healthy_bootstrap
from plugin_helper.bsipa import check_bsipa_health, planning_requires_bootstrap
from plugin_helper.beatmods import by_version_id, normalize_mods
from plugin_helper.checker import check_lock
from plugin_helper.cli import installed_plugins_report, run
@@ -23,6 +23,7 @@ from plugin_helper.fsutil import sha256_file
from plugin_helper.installer import apply_plan, disable_plugin, uninstall_plugin
from plugin_helper.instances import get_instance, list_instances
from plugin_helper.models import Lockfile, LockedPlugin, Registry, RegistryPlugin
from plugin_helper.operations import enable_disabled_plugin
from plugin_helper.planner import create_plan
from plugin_helper.scanner import scan_bootstrap_files, scan_instance
from plugin_helper.state import downloads_dir, load_installed_state, plugin_downloads_dir, save_bootstrap_state, save_installed_state
@@ -380,6 +381,226 @@ state_dir = ".state"
self.assertFalse(result["ok"])
self.assertIn("missing Logs/_latest.log", result["messages"])
def test_planning_requires_bootstrap(self) -> None:
lockfile_plugins = (
LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None),
LockedPlugin(id="example", repo=None, tag="v1.0.0", asset="Example.dll", sha256=None),
)
self.assertTrue(planning_requires_bootstrap(lockfile_plugins, {"example"}))
self.assertFalse(planning_requires_bootstrap(lockfile_plugins, {"bsipa"}))
self.assertFalse(planning_requires_bootstrap((), {"example"}))
def test_ensure_healthy_bootstrap_noop_when_healthy(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
instance = work / "1.44.1"
state = work / "state"
instance.mkdir(parents=True)
(instance / "IPA").mkdir()
(instance / "Libs").mkdir()
(instance / "IPA.exe").write_bytes(b"ipa")
(instance / "winhttp.dll").write_bytes(b"proxy")
save_bootstrap_state(
state,
"1.44.1",
{
"bootstrapMode": "native",
"ipaExitCode": 0,
"ipaTimedOut": False,
"files": scan_bootstrap_files(instance),
},
)
lockfile = Lockfile(
beat_saber_version="1.44.1",
instance="1.44.1",
plugins=(
LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None),
LockedPlugin(id="example", repo=None, tag="v1.0.0", asset="Example.dll", sha256=None),
),
)
registry = Registry(
{
"bsipa": RegistryPlugin(id="bsipa", name="BSIPA", repo=None, install_strategy="root-zip"),
"example": RegistryPlugin(
id="example",
name="Example",
repo=None,
install_strategy="dll-to-plugins",
),
}
)
with patch("plugin_helper.bootstrap.run_bootstrap") as run_bootstrap:
ensure_healthy_bootstrap(
instance="1.44.1",
instance_path=instance,
beat_saber_version="1.44.1",
registry=registry,
lockfile=lockfile,
state_root=state,
repo_root=work,
selected_ids={"example"},
)
run_bootstrap.assert_not_called()
def test_ensure_healthy_bootstrap_runs_when_unhealthy(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
instance = work / "1.44.1"
state = work / "state"
instance.mkdir(parents=True)
lockfile = Lockfile(
beat_saber_version="1.44.1",
instance="1.44.1",
plugins=(
LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None),
LockedPlugin(id="example", repo=None, tag="v1.0.0", asset="Example.dll", sha256=None),
),
)
registry = Registry(
{
"bsipa": RegistryPlugin(id="bsipa", name="BSIPA", repo=None, install_strategy="root-zip"),
"example": RegistryPlugin(
id="example",
name="Example",
repo=None,
install_strategy="dll-to-plugins",
),
}
)
healthy = {
"ok": True,
"messages": [],
"statePath": str(state / "instances" / "1.44.1" / "bootstrap.json"),
"logPath": str(instance / "Logs" / "_latest.log"),
}
unhealthy = {**healthy, "ok": False, "messages": ["missing IPA.exe"]}
with patch("plugin_helper.bootstrap.check_bsipa_health", side_effect=[unhealthy, healthy]) as check_health:
with patch("plugin_helper.bootstrap.run_bootstrap") as run_bootstrap:
ensure_healthy_bootstrap(
instance="1.44.1",
instance_path=instance,
beat_saber_version="1.44.1",
registry=registry,
lockfile=lockfile,
state_root=state,
repo_root=work,
selected_ids={"example"},
)
self.assertEqual(check_health.call_count, 2)
run_bootstrap.assert_called_once()
def test_ensure_healthy_bootstrap_skips_for_bsipa_only(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
instance = work / "1.44.1"
state = work / "state"
instance.mkdir(parents=True)
lockfile = Lockfile(
beat_saber_version="1.44.1",
instance="1.44.1",
plugins=(LockedPlugin(id="bsipa", repo=None, tag="4.3.7", asset="BSIPA.zip", sha256=None),),
)
registry = Registry(
{"bsipa": RegistryPlugin(id="bsipa", name="BSIPA", repo=None, install_strategy="root-zip")}
)
with patch("plugin_helper.bootstrap.check_bsipa_health") as check_health:
with patch("plugin_helper.bootstrap.run_bootstrap") as run_bootstrap:
ensure_healthy_bootstrap(
instance="1.44.1",
instance_path=instance,
beat_saber_version="1.44.1",
registry=registry,
lockfile=lockfile,
state_root=state,
repo_root=work,
selected_ids={"bsipa"},
)
check_health.assert_not_called()
run_bootstrap.assert_not_called()
def test_enable_disabled_plugin_bootstraps_before_plan(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)
instance = work / "instances" / "1.44.1"
state = work / "state"
instance.mkdir(parents=True)
(instance / "Beat Saber_Data").mkdir()
(instance / "Plugins").mkdir()
asset = plugin_downloads_dir(state, "1.44.1", "example") / "Example.dll"
asset.write_bytes(b"managed dll")
save_installed_state(
state,
"1.44.1",
{
"disabledPlugins": {
"example": {
"disabledAt": "2026-01-01T00:00:00Z",
"files": [{"path": "Plugins/Example.dll", "sha256": sha256_file(asset)}],
}
}
},
)
(work / "locks").mkdir()
(work / "registry").mkdir()
lock_path = work / "locks" / "1.44.1.lock.toml"
lock_path.write_text(
"""
beat_saber_version = "1.44.1"
instance = "1.44.1"
[[plugins]]
id = "bsipa"
tag = "4.3.7"
asset = "BSIPA.zip"
[[plugins]]
id = "example"
tag = "v1.0.0"
asset = "Example.dll"
sha256 = "%s"
""" % sha256_file(asset),
encoding="utf-8",
)
registry_path = work / "registry" / "plugins.toml"
registry_path.write_text(
"""
[bsipa]
name = "BSIPA"
install_strategy = "root-zip"
[example]
name = "Example"
install_strategy = "dll-to-plugins"
""",
encoding="utf-8",
)
call_order: list[str] = []
def _ensure(**_kwargs: object) -> None:
call_order.append("ensure")
def _create_plan(**_kwargs: object) -> tuple[dict[str, object], Path]:
call_order.append("plan")
return {"changes": [], "instance": "1.44.1", "instancePath": str(instance)}, work / "plan.json"
def _apply_plan(_plan: dict[str, object], _state_root: Path) -> dict[str, object]:
call_order.append("apply")
return {"applied": [], "statePath": str(state / "instances" / "1.44.1" / "installed.json")}
with patch("plugin_helper.operations.ensure_healthy_bootstrap", side_effect=_ensure):
with patch("plugin_helper.operations.create_plan", side_effect=_create_plan):
with patch("plugin_helper.operations.apply_plan", side_effect=_apply_plan):
enable_disabled_plugin(
instance="1.44.1",
instance_path=instance,
state_root=state,
plugin_id="example",
registry=str(registry_path),
lockfile=str(lock_path),
repo=work,
)
self.assertEqual(call_order, ["ensure", "plan", "apply"])
def test_plan_apply_and_uninstall_dll(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
work = Path(tmp)