feat: self-management commands

src/bub/builtin/cli.py

@@ -4,6 +4,11 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import subprocess
+import sys
+from functools import lru_cache
+from pathlib import Path
 
 import typer
 
@@ -81,3 +86,116 @@ def chat(
         raise typer.Exit(1)
     channel.set_metadata(chat_id=chat_id, session_id=session_id)  # type: ignore[attr-defined]
     asyncio.run(manager.listen_and_run())
+
+
+@lru_cache(maxsize=1)
+def _find_uv() -> str:
+    import shutil
+    import sysconfig
+
+    bin_path = sysconfig.get_path("scripts")
+    uv_path = shutil.which("uv", path=os.pathsep.join([bin_path, os.getenv("PATH", "")]))
+    if uv_path is None:
+        raise FileNotFoundError("uv executable not found in PATH or scripts directory.")
+    return uv_path
+
+
+@lru_cache(maxsize=1)
+def _default_project() -> Path:
+    from .settings import load_settings
+
+    settings = load_settings()
+    project = settings.home / "bub-project"
+    project.mkdir(exist_ok=True, parents=True)
+    return project
+
+
+def _is_in_venv() -> bool:
+    return sys.prefix != getattr(sys, "base_prefix", sys.prefix)
+
+
+project_opt = typer.Option(
+    default_factory=_default_project,
+    help="Path to the project directory (default: ~/.bub/bub-project)",
+    envvar="BUB_PROJECT",
+)
+
+
+def _uv(*args: str, cwd: Path) -> subprocess.CompletedProcess:
+    uv_executable = _find_uv()
+    if not _is_in_venv():
+        typer.secho("Please install Bub in a virtual environment to use this command.", err=True, fg="red")
+        raise typer.Exit(1)
+    env = {**os.environ, "VIRTUAL_ENV": sys.prefix}
+    try:
+        return subprocess.run([uv_executable, *args], env=env, check=True, cwd=cwd)
+    except subprocess.CalledProcessError as e:
+        typer.secho(f"Command 'uv {' '.join(args)}' failed with exit code {e.returncode}.", err=True, fg="red")
+        raise typer.Exit(e.returncode) from e
+
+
+BUB_CONTRIB_REPO = "https://github.com/bubbuild/bub-contrib.git"
+
+
+def _build_requirement(spec: str) -> str:
+    if spec.startswith(("git@", "https://")):
+        # Git URL
+        return f"git+{spec}"
+    elif "/" in spec:
+        # owner/repo format
+        repo, *rest = spec.partition("@")
+        ref = "".join(rest)
+        return f"git+https://github.com/{repo}.git{ref}"
+    else:
+        # Assume it's a package name in bub-contrib
+        name, *rest = spec.partition("@")
+        ref = "".join(rest)
+        return f"git+{BUB_CONTRIB_REPO}{ref}#subdirectory=packages/{name}"
+
+
+def _ensure_project(project: Path) -> None:
+    if (project / "pyproject.toml").is_file():
+        return
+    _uv("init", "--bare", "--name", "bub-project", "--app", cwd=project)
+
+
+def install(
+    specs: list[str] = typer.Argument(
+        default_factory=list,
+        help="Package specification to install, can be a git URL, owner/repo, or package name in bub-contrib.",
+    ),
+    project: Path = project_opt,
+) -> None:
+    """Install a plugin into Bub's environment, or sync the environment if no specifications are provided."""
+    _ensure_project(project)
+    if not specs:
+        _uv("sync", "--active", "--inexact", cwd=project)
+    else:
+        _uv("add", "--active", *map(_build_requirement, specs), cwd=project)
+
+
+def uninstall(
+    packages: list[str] = typer.Argument(..., help="Package name to uninstall (must match the name in pyproject.toml)"),
+    project: Path = project_opt,
+) -> None:
+    """Uninstall a plugin from Bub's environment."""
+    _ensure_project(project)
+    _uv("remove", "--active", "--no-sync", *packages, cwd=project)
+    _uv("sync", "--active", "--frozen", "--inexact", cwd=project)
+
+
+def update(
+    packages: list[str] = typer.Argument(
+        default_factory=list, help="Optional package name to update (must match the name in pyproject.toml)"
+    ),
+    project: Path = project_opt,
+) -> None:
+    """Update selected package or all packages in Bub's environment."""
+    _ensure_project(project)
+    if not packages:
+        _uv("sync", "--active", "--upgrade", "--inexact", cwd=project)
+    else:
+        package_args: list[str] = []
+        for pkg in packages:
+            package_args.extend(["--upgrade-package", pkg])
+        _uv("sync", "--active", "--inexact", *package_args, cwd=project)

src/bub/builtin/hook_impl.py

@@ -118,6 +118,10 @@ def register_cli_commands(self, app: typer.Typer) -> None:
         app.add_typer(cli.login_app)
         app.command("hooks", hidden=True)(cli.list_hooks)
         app.command("gateway")(cli.gateway)
+        app.command("install")(cli.install)
+        # TODO: uninstall command can't work properly
+        # app.command("uninstall")(cli.uninstall)
+        app.command("update")(cli.update)
 
     def _read_agents_file(self, state: State) -> str:
         workspace = state.get("_runtime_workspace", str(Path.cwd()))