feat(extensions): add management CLI

nanobot/cli/commands.py

@@ -72,6 +72,7 @@ from nanobot.bus.outbound_events import (  # noqa: E402
     StreamEndEvent,
     outbound_event_from_message,
 )
+from nanobot.cli.extensions import create_extensions_app  # noqa: E402
 from nanobot.cli.gateway import create_gateway_app  # noqa: E402
 from nanobot.cli.stream import StreamRenderer, ThinkingSpinner  # noqa: E402
 from nanobot.config.paths import get_workspace_path, is_default_workspace  # noqa: E402
@@ -2535,6 +2536,8 @@ def agent(
 # ============================================================================
 
 
+app.add_typer(create_extensions_app(console=console), name="extensions")
+
 channels_app = typer.Typer(help="Manage channels")
 app.add_typer(channels_app, name="channels")
 

nanobot/cli/extensions.py

@@ -0,0 +1,213 @@
+"""Typer commands for installing and governing extensions."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from typing import Any
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from nanobot.extensions import ExtensionService
+
+ServiceFactory = Callable[[], ExtensionService]
+
+
+def create_extensions_app(
+    *,
+    console: Console,
+    service_factory: ServiceFactory = ExtensionService,
+) -> typer.Typer:
+    """Build the extension command group around the transport-neutral service."""
+    app = typer.Typer(help="Discover, install, inspect, and govern extensions.")
+
+    def service() -> ExtensionService:
+        return service_factory()
+
+    def run(awaitable: Any) -> dict[str, Any]:
+        try:
+            return asyncio.run(awaitable)
+        except (KeyError, RuntimeError, ValueError) as exc:
+            console.print(f"[red]Error: {exc}[/red]")
+            raise typer.Exit(1) from exc
+
+    @app.command("list")
+    def list_extensions() -> None:
+        """List installed extensions and their activation policy."""
+        payload = run(service().status())
+        table = Table(show_header=True, header_style="bold")
+        table.add_column("Extension")
+        table.add_column("Runtime")
+        table.add_column("State")
+        table.add_column("Trust")
+        table.add_column("Version")
+        for item in payload["extensions"]:
+            state = "active" if item["active"] else ("enabled" if item["enabled"] else "disabled")
+            table.add_row(
+                item["name"],
+                item["runtime"],
+                state,
+                "trusted" if item["trusted"] else "untrusted",
+                item["version"],
+            )
+        console.print(table)
+        if not payload["extensions"]:
+            console.print("[dim]No extensions installed.[/dim]")
+        if payload["diagnostics"]:
+            console.print(f"[yellow]{len(payload['diagnostics'])} diagnostic(s)[/yellow]")
+
+    @app.command("inspect")
+    def inspect_extension(extension_id: str = typer.Argument(..., help="Extension ID")) -> None:
+        """Show manifest, dependencies, permissions, and diagnostics."""
+        payload = run(service().status())
+        item = next(
+            (candidate for candidate in payload["extensions"] if candidate["id"] == extension_id),
+            None,
+        )
+        if item is None:
+            console.print(f"[red]Extension not found: {extension_id}[/red]")
+            raise typer.Exit(1)
+        console.print(f"[bold]{item['name']}[/bold] [dim]{item['version']}[/dim]")
+        console.print(item["description"] or "[dim]No description.[/dim]")
+        console.print(f"Runtime: {item['runtime']}  Scope: {item['scope']}")
+        console.print(
+            f"State: {'active' if item['active'] else 'inactive'}  "
+            f"Trust: {'trusted' if item['trusted'] else 'untrusted'}"
+        )
+        _print_named_rows(console, "Contributions", item["contributions"], "kind", "name")
+        _print_named_rows(console, "Dependencies", item["dependencies"], "kind", "name")
+        _print_permissions(console, item["permissions"], set(item["granted_permissions"]))
+        diagnostics = [
+            diagnostic
+            for diagnostic in payload["diagnostics"]
+            if diagnostic["extension_id"] == extension_id
+        ]
+        if diagnostics:
+            console.print("\n[bold]Diagnostics[/bold]")
+            for diagnostic in diagnostics:
+                console.print(
+                    f"  [yellow]{diagnostic['code']}[/yellow] {diagnostic['message']}"
+                )
+
+    @app.command("search")
+    def search_extensions(
+        query: str = typer.Argument("", help="Package name or keyword"),
+        ecosystem: str = typer.Option(
+            "all",
+            "--ecosystem",
+            "-e",
+            help="all, nanobot, pi, or openclaw",
+        ),
+        limit: int = typer.Option(20, "--limit", min=1, max=100),
+    ) -> None:
+        """Search compatible extension packages on npm."""
+        payload = run(service().search(query, ecosystem=ecosystem, limit=limit))
+        table = Table(show_header=True, header_style="bold")
+        table.add_column("Package")
+        table.add_column("Ecosystem")
+        table.add_column("Version")
+        table.add_column("Description")
+        for package in payload["packages"]:
+            table.add_row(
+                package["name"],
+                package["ecosystem"],
+                package["version"],
+                package["description"],
+            )
+        console.print(table)
+        if not payload["packages"]:
+            console.print("[dim]No compatible packages found.[/dim]")
+
+    @app.command("install")
+    def install_extension(
+        source: str = typer.Argument(..., help="npm spec, Git URL, or local path"),
+        kind: str = typer.Option("npm", "--kind", help="npm, git, or local"),
+        ref: str = typer.Option("", "--ref", help="Git branch, tag, or commit"),
+    ) -> None:
+        """Install an extension without granting trust or permissions."""
+        payload = run(service().install(source, kind=kind, ref=ref, trusted=False))
+        record = payload["record"]
+        console.print(
+            f"[green]Installed {record['id']} {record['version']}[/green] "
+            "[yellow](untrusted)[/yellow]"
+        )
+        console.print(
+            f"Review with [bold]nanobot extensions inspect {record['id']}[/bold], "
+            "then grant permissions and trust it explicitly."
+        )
+
+    def policy_command(name: str, value: bool, label: str, help_text: str) -> None:
+        @app.command(name, help=help_text)
+        def update(extension_id: str = typer.Argument(..., help="Extension ID")) -> None:
+            payload = run(
+                service().set_enabled(extension_id, value)
+                if name in {"enable", "disable"}
+                else service().set_trusted(extension_id, value)
+            )
+            console.print(f"[green]{label}: {payload['record']['id']}[/green]")
+
+    policy_command("enable", True, "Enabled", "Allow an installed extension to activate.")
+    policy_command("disable", False, "Disabled", "Prevent an installed extension from activating.")
+    policy_command("trust", True, "Trusted", "Trust an installed extension's executable code.")
+    policy_command("untrust", False, "Trust revoked", "Revoke trust and stop extension activation.")
+
+    @app.command("permissions")
+    def set_permissions(
+        extension_id: str = typer.Argument(..., help="Extension ID"),
+        permissions: list[str] = typer.Argument(
+            None,
+            help="Exact permissions to grant; omit all to revoke every grant",
+        ),
+    ) -> None:
+        """Replace the extension's granted host permissions."""
+        payload = run(service().set_permissions(extension_id, set(permissions or [])))
+        granted = payload["record"]["granted_permissions"]
+        console.print(
+            f"[green]Updated permissions for {extension_id}:[/green] "
+            + (", ".join(granted) if granted else "none")
+        )
+
+    @app.command("uninstall")
+    def uninstall_extension(
+        extension_id: str = typer.Argument(..., help="Extension ID"),
+        yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation"),
+    ) -> None:
+        """Remove an installed extension."""
+        if not yes and not typer.confirm(f"Uninstall extension '{extension_id}'?"):
+            raise typer.Abort()
+        run(service().uninstall(extension_id))
+        console.print(f"[green]Uninstalled {extension_id}[/green]")
+
+    return app
+
+
+def _print_named_rows(
+    console: Console,
+    title: str,
+    rows: list[dict[str, Any]],
+    category_key: str,
+    name_key: str,
+) -> None:
+    console.print(f"\n[bold]{title}[/bold]")
+    if not rows:
+        console.print("  [dim]None[/dim]")
+        return
+    for row in rows:
+        console.print(f"  {row[category_key]}: {row[name_key]}")
+
+
+def _print_permissions(
+    console: Console,
+    permissions: list[dict[str, str]],
+    granted: set[str],
+) -> None:
+    console.print("\n[bold]Permissions[/bold]")
+    if not permissions:
+        console.print("  [dim]None requested[/dim]")
+        return
+    for permission in permissions:
+        status = "[green]granted[/green]" if permission["name"] in granted else "[yellow]pending[/yellow]"
+        reason = f" — {permission['reason']}" if permission["reason"] else ""
+        console.print(f"  {permission['name']} ({status}){reason}")

tests/cli/test_extensions.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from io import StringIO
+
+from rich.console import Console
+from typer.testing import CliRunner
+
+from nanobot.cli.extensions import create_extensions_app
+
+
+class _Service:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, object]] = []
+
+    async def status(self):
+        self.calls.append(("status", None))
+        return {
+            "extensions": [
+                {
+                    "id": "sample",
+                    "name": "Sample",
+                    "version": "1.0.0",
+                    "runtime": "pi",
+                    "scope": "user",
+                    "description": "Example extension",
+                    "enabled": True,
+                    "trusted": False,
+                    "active": False,
+                    "contributions": [],
+                    "dependencies": [],
+                    "permissions": [{"name": "network", "reason": "Fetch data"}],
+                    "granted_permissions": [],
+                }
+            ],
+            "diagnostics": [],
+        }
+
+    async def search(self, query, *, ecosystem, limit):
+        self.calls.append(("search", (query, ecosystem, limit)))
+        return {"packages": []}
+
+    async def install(self, source, *, kind, ref, trusted):
+        self.calls.append(("install", (source, kind, ref, trusted)))
+        return {
+            "record": {
+                "id": "sample",
+                "version": "1.0.0",
+                "trusted": trusted,
+            }
+        }
+
+    async def set_enabled(self, extension_id, enabled):
+        self.calls.append(("enabled", (extension_id, enabled)))
+        return {"record": {"id": extension_id}}
+
+    async def set_trusted(self, extension_id, trusted):
+        self.calls.append(("trusted", (extension_id, trusted)))
+        return {"record": {"id": extension_id}}
+
+    async def set_permissions(self, extension_id, permissions):
+        self.calls.append(("permissions", (extension_id, permissions)))
+        return {
+            "record": {
+                "id": extension_id,
+                "granted_permissions": sorted(permissions),
+            }
+        }
+
+    async def uninstall(self, extension_id):
+        self.calls.append(("uninstall", extension_id))
+        return {"removed": extension_id}
+
+
+def _runner(service: _Service):
+    output = StringIO()
+    app = create_extensions_app(
+        console=Console(file=output, force_terminal=False),
+        service_factory=lambda: service,
+    )
+    return CliRunner(), app, output
+
+
+def test_extension_cli_inspects_and_searches() -> None:
+    service = _Service()
+    runner, app, output = _runner(service)
+
+    inspected = runner.invoke(app, ["inspect", "sample"])
+    searched = runner.invoke(app, ["search", "web", "--ecosystem", "pi", "--limit", "7"])
+
+    assert inspected.exit_code == 0
+    assert searched.exit_code == 0
+    assert "Sample" in output.getvalue()
+    assert service.calls == [
+        ("status", None),
+        ("search", ("web", "pi", 7)),
+    ]
+
+
+def test_extension_cli_install_and_policy_commands() -> None:
+    service = _Service()
+    runner, app, _output = _runner(service)
+
+    assert runner.invoke(app, ["install", "pi-example"]).exit_code == 0
+    assert runner.invoke(app, ["trust", "sample"]).exit_code == 0
+    assert runner.invoke(app, ["disable", "sample"]).exit_code == 0
+    assert runner.invoke(
+        app,
+        ["permissions", "sample", "network", "filesystem.read"],
+    ).exit_code == 0
+    assert runner.invoke(app, ["uninstall", "sample", "--yes"]).exit_code == 0
+
+    assert service.calls == [
+        ("install", ("pi-example", "npm", "", False)),
+        ("trusted", ("sample", True)),
+        ("enabled", ("sample", False)),
+        ("permissions", ("sample", {"network", "filesystem.read"})),
+        ("uninstall", "sample"),
+    ]