Feat/actions

src/agentpeek/actions/runner.py

@@ -0,0 +1,108 @@
+"""Shell-out runner for `claude plugin …` write operations.
+
+agentpeek does not edit Claude state files directly. Every write goes
+through the `claude` CLI, which handles dependency validation, version
+tracking, and cache invariants. See docs.claude.com plugin reference.
+"""
+
+import shutil
+import subprocess
+import time
+from dataclasses import dataclass
+from typing import Literal
+
+PluginVerb = Literal["enable", "disable", "update", "uninstall", "install"]
+# `managed` is accepted by `claude plugin update` for enterprise-pushed
+# plugins; enable/disable/uninstall do not accept it. If a managed-scope
+# plugin is targeted with a non-update verb, the CLI surfaces a clear error
+# in the result modal.
+Scope = Literal["user", "project", "local", "managed"]
+
+_TIMEOUT_SECONDS = 60
+
+
+@dataclass(frozen=True, slots=True)
+class ActionResult:
+    ok: bool
+    verb: str
+    target: str
+    scope: str | None
+    returncode: int
+    stdout: str
+    stderr: str
+    elapsed_ms: int
+
+    @property
+    def message(self) -> str:
+        """One-line summary suitable for a toast notification."""
+        merged = (self.stdout + self.stderr).strip().splitlines()
+        return merged[-1] if merged else ("ok" if self.ok else "failed")
+
+
+def claude_cli_available() -> bool:
+    return shutil.which("claude") is not None
+
+
+def run_plugin(verb: PluginVerb, qualified_id: str, *, scope: Scope) -> ActionResult:
+    cmd = ["claude", "plugin", verb, qualified_id, "--scope", scope]
+    return _run(cmd, verb=verb, target=qualified_id, scope=scope)
+
+
+def run_marketplace_update(name: str | None) -> ActionResult:
+    """Refresh one marketplace (or all if `name` is None)."""
+    cmd = ["claude", "plugin", "marketplace", "update"]
+    if name is not None:
+        cmd.append(name)
+    return _run(cmd, verb="marketplace update", target=name or "(all)", scope=None)
+
+
+def _run(
+    cmd: list[str], *, verb: str, target: str, scope: str | None
+) -> ActionResult:
+    start = time.monotonic()
+    try:
+        proc = subprocess.run(  # noqa: S603 - cmd is built from a fixed allowlist
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=_TIMEOUT_SECONDS,
+            check=False,
+        )
+        elapsed_ms = int((time.monotonic() - start) * 1000)
+        return ActionResult(
+            ok=proc.returncode == 0,
+            verb=verb,
+            target=target,
+            scope=scope,
+            returncode=proc.returncode,
+            stdout=proc.stdout,
+            stderr=proc.stderr,
+            elapsed_ms=elapsed_ms,
+        )
+    except subprocess.TimeoutExpired as exc:
+        elapsed_ms = int((time.monotonic() - start) * 1000)
+        partial = exc.stdout
+        if isinstance(partial, bytes):
+            partial = partial.decode("utf-8", errors="replace")
+        return ActionResult(
+            ok=False,
+            verb=verb,
+            target=target,
+            scope=scope,
+            returncode=-1,
+            stdout=partial or "",
+            stderr=f"timed out after {_TIMEOUT_SECONDS}s",
+            elapsed_ms=elapsed_ms,
+        )
+    except FileNotFoundError:
+        elapsed_ms = int((time.monotonic() - start) * 1000)
+        return ActionResult(
+            ok=False,
+            verb=verb,
+            target=target,
+            scope=scope,
+            returncode=-1,
+            stdout="",
+            stderr="`claude` CLI not found on PATH",
+            elapsed_ms=elapsed_ms,
+        )

src/agentpeek/cli.py

@@ -40,6 +40,13 @@ def main(argv: list[str] | None = None) -> int:
         default="WARNING",
         choices=["DEBUG", "INFO", "WARNING", "ERROR"],
     )
+    parser.add_argument(
+        "--read-only",
+        action="store_true",
+        help="Disable write actions. By default the TUI exposes update / "
+        "enable / disable / uninstall / marketplace-refresh bindings that "
+        "shell out to the `claude plugin` CLI.",
+    )
     args = parser.parse_args(argv)
     configure_logging(args.log_level)
 
@@ -55,5 +62,9 @@ def main(argv: list[str] | None = None) -> int:
     # TUI into module-load time when only the CLI surface is exercised.
     from agentpeek.tui.app import AgentViewApp  # noqa: PLC0415
 
-    AgentViewApp(scan_root=args.root, source_name=args.source).run()
+    AgentViewApp(
+        scan_root=args.root,
+        source_name=args.source,
+        actions=not args.read_only,
+    ).run()
     return 0

src/agentpeek/tui/app.py

@@ -18,10 +18,17 @@ class AgentViewApp(App[None]):
     ]
     TITLE = "agentpeek"
 
-    def __init__(self, *, scan_root: Path | None, source_name: str | None) -> None:
+    def __init__(
+        self,
+        *,
+        scan_root: Path | None,
+        source_name: str | None,
+        actions: bool = True,
+    ) -> None:
         super().__init__()
         self._scan_root = scan_root
         self._source_name = source_name
+        self._actions = actions
 
     def rescan(self) -> ScanReport:
         """Re-run the scanner with the cached args and update the subtitle.
@@ -38,4 +45,10 @@ def rescan(self) -> ScanReport:
 
     def on_mount(self) -> None:
         report = self.rescan()
-        self.push_screen(MainScreen(report, explicit_root=self._scan_root is not None))
+        self.push_screen(
+            MainScreen(
+                report,
+                explicit_root=self._scan_root is not None,
+                actions=self._actions,
+            )
+        )

src/agentpeek/tui/screens/action_result.py

@@ -0,0 +1,62 @@
+from typing import ClassVar
+
+from textual.app import ComposeResult
+from textual.binding import Binding, BindingType
+from textual.containers import Container, VerticalScroll
+from textual.content import Content
+from textual.screen import ModalScreen
+from textual.widgets import Static
+
+from agentpeek.actions.runner import ActionResult
+
+
+class ActionResultModal(ModalScreen[None]):
+    """Surface the raw stdout/stderr of a `claude plugin ...` call.
+
+    Used for failures (and for successes when output is non-trivial).
+    Successful one-line outcomes use `self.notify(...)` instead.
+    """
+
+    BINDINGS: ClassVar[list[BindingType]] = [
+        Binding("q,escape,enter", "dismiss_modal", "Close"),
+    ]
+
+    def __init__(self, result: ActionResult) -> None:
+        super().__init__()
+        self._result = result
+
+    def compose(self) -> ComposeResult:
+        r = self._result
+        header_color = "$success" if r.ok else "$error"
+        verdict = "OK" if r.ok else f"FAILED (exit {r.returncode})"
+        scope_part = f" --scope {r.scope}" if r.scope else ""
+        with Container(id="action-result-card"):
+            yield Static(
+                Content.assemble(
+                    (f"{verdict}  ", f"bold {header_color}"),
+                    (f"{r.verb} {r.target}{scope_part}", "$text-muted"),
+                ),
+                id="action-result-title",
+            )
+            yield Static(
+                Content.assemble(
+                    ("elapsed ", "$text-muted"),
+                    (f"{r.elapsed_ms} ms", "bold"),
+                ),
+                id="action-result-meta",
+            )
+            with VerticalScroll(id="action-result-body"):
+                if r.stdout.strip():
+                    yield Static("stdout", classes="action-result-section-title")
+                    yield Static(r.stdout.rstrip(), classes="action-result-stream")
+                if r.stderr.strip():
+                    yield Static("stderr", classes="action-result-section-title")
+                    yield Static(r.stderr.rstrip(), classes="action-result-stream")
+                if not r.stdout.strip() and not r.stderr.strip():
+                    yield Static(
+                        "(no output)",
+                        classes="action-result-stream",
+                    )
+
+    def action_dismiss_modal(self) -> None:
+        self.dismiss()

src/agentpeek/tui/screens/confirm.py

@@ -0,0 +1,38 @@
+from typing import ClassVar
+
+from textual.app import ComposeResult
+from textual.binding import Binding, BindingType
+from textual.containers import Container, Horizontal
+from textual.screen import ModalScreen
+from textual.widgets import Button, Static
+
+
+class ConfirmModal(ModalScreen[bool]):
+    """Yes/no modal. `push_screen_wait()` returns True on confirm, False on cancel."""
+
+    BINDINGS: ClassVar[list[BindingType]] = [
+        Binding("y", "confirm", "Yes", show=False),
+        Binding("n,escape,q", "cancel", "No", show=False),
+    ]
+
+    def __init__(self, message: str, *, title: str = "Confirm") -> None:
+        super().__init__()
+        self._message = message
+        self._title = title
+
+    def compose(self) -> ComposeResult:
+        with Container(id="confirm-card"):
+            yield Static(self._title, id="confirm-title")
+            yield Static(self._message, id="confirm-body")
+            with Horizontal(id="confirm-buttons"):
+                yield Button("Yes (y)", id="confirm-yes", variant="warning")
+                yield Button("No (n)", id="confirm-no")
+
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        self.dismiss(event.button.id == "confirm-yes")
+
+    def action_confirm(self) -> None:
+        self.dismiss(True)
+
+    def action_cancel(self) -> None:
+        self.dismiss(False)