feat(plugins): stream an install or update's progress over server-sent events

repeater/plugins/ipc.py

@@ -131,6 +131,7 @@ class PluginIPCServer:
             "catalogue_install": self._op_catalogue_install,
             "check_update": self._op_check_update,
             "update": self._op_update,
+            "progress": self._op_progress,
             "ping": self._op_ping,
         }
 
@@ -367,6 +368,14 @@ class PluginIPCServer:
         force = bool(request.get("force_refresh") or request.get("refresh"))
         return self.manager.check_update(plugin_id, force_refresh=force)
 
+    def _op_progress(self, request: dict[str, Any]) -> dict[str, Any]:
+        plugin_id = self._require_id(request)
+        try:
+            since = int(request.get("since") or 0)
+        except (TypeError, ValueError):
+            raise PluginIPCError("since must be an integer", 400)
+        return self.manager.progress(plugin_id, since=since)
+
     def _op_update(self, request: dict[str, Any]) -> dict[str, Any]:
         plugin_id = self._require_id(request)
         version = request.get("version")
@@ -509,6 +518,9 @@ class PluginIPCClient:
             payload["version"] = version
         return self.call("catalogue_install", **payload)
 
+    def progress(self, plugin_id: str, *, since: int = 0) -> dict[str, Any]:
+        return self.call("progress", plugin_id=plugin_id, since=int(since))
+
     def check_update(self, plugin_id: str, *, force_refresh: bool = False) -> dict[str, Any]:
         return self.call("check_update", plugin_id=plugin_id, force_refresh=bool(force_refresh))
 

repeater/plugins/manager.py

@@ -31,6 +31,77 @@ from .storage import PluginStorage
 logger = logging.getLogger("PluginManager")
 
 
+PROGRESS_MAX_LINES = 500
+PROGRESS_MAX_LOGS = 32
+
+
+class OperationProgress:
+    """Bounded output log of a plugin's last install or update."""
+
+    def __init__(self, plugin_id: str, operation: str):
+        self.plugin_id = plugin_id
+        self.operation = operation
+        self.state = "running"
+        self.error: Optional[str] = None
+        self.started = time.time()
+        self.finished: Optional[float] = None
+        self._lines: list[str] = []
+        self._dropped = 0  # cursors count every line ever appended
+        self._lock = threading.Lock()
+
+    def append(self, line: str) -> None:
+        with self._lock:
+            self._lines.append(line)
+            excess = len(self._lines) - PROGRESS_MAX_LINES
+            if excess > 0:
+                del self._lines[:excess]
+                self._dropped += excess
+
+    def finish(self, error: Optional[str] = None) -> None:
+        with self._lock:
+            self.state = "error" if error else "complete"
+            self.error = error
+            self.finished = time.time()
+
+    def snapshot(self, since: int = 0) -> dict[str, Any]:
+        with self._lock:
+            start = max(0, min(int(since) - self._dropped, len(self._lines)))
+            return {
+                "id": self.plugin_id,
+                "operation": self.operation,
+                "state": self.state,
+                "error": self.error,
+                "lines": list(self._lines[start:]),
+                "next": self._dropped + len(self._lines),
+                "started": self.started,
+                "finished": self.finished,
+            }
+
+
+def _reported(operation: str):
+    """Record the operation's installer output and outcome for /plugins/progress."""
+
+    def decorate(fn):
+        @wraps(fn)
+        def wrapped(self, plugin_id, *args, **kwargs):
+            progress = self._start_progress(plugin_id, operation)
+            previous = self.runtime.on_output
+            self.runtime.on_output = progress.append
+            try:
+                result = fn(self, plugin_id, *args, **kwargs)
+            except Exception as exc:
+                progress.finish(str(exc))
+                raise
+            finally:
+                self.runtime.on_output = previous
+            progress.finish()
+            return result
+
+        return wrapped
+
+    return decorate
+
+
 def _serialized_plugin(fn):
     """Keep a complete lifecycle transaction exclusive, without queuing callers."""
 
@@ -66,6 +137,7 @@ class PluginManager:
         self.runtime = runtime or PluginRuntime(storage)
         self._lock = threading.RLock()
         self._operations: dict[str, tuple[int, int]] = {}
+        self._progress: dict[str, OperationProgress] = {}
         self.catalogue = catalogue_client or CatalogueClient(catalogue_url or DEFAULT_CATALOGUE_URL)
         self.github = github_client or GitHubReleaseClient()
 
@@ -87,6 +159,34 @@ class PluginManager:
                 else:
                     self._operations[plugin_id] = (owner, depth - 1)
 
+    def _start_progress(self, plugin_id: str, operation: str) -> OperationProgress:
+        progress = OperationProgress(plugin_id, operation)
+        with self._lock:
+            self._progress[plugin_id] = progress
+            finished = sorted(
+                (p for p in self._progress.values() if p.state != "running"),
+                key=lambda p: p.finished or 0,
+            )
+            for old in finished[: max(0, len(self._progress) - PROGRESS_MAX_LOGS)]:
+                del self._progress[old.plugin_id]
+        return progress
+
+    def progress(self, plugin_id: str, since: int = 0) -> dict[str, Any]:
+        with self._lock:
+            progress = self._progress.get(plugin_id)
+        if progress is None:
+            return {
+                "id": plugin_id,
+                "operation": None,
+                "state": "idle",
+                "error": None,
+                "lines": [],
+                "next": 0,
+                "started": None,
+                "finished": None,
+            }
+        return progress.snapshot(since)
+
     @contextmanager
     def stage_upload(self, wheel_path: str):
         """Own a bounded regular-file copy before IPC acknowledges processing.
@@ -430,6 +530,7 @@ class PluginManager:
         return {"schema": catalogue.schema, "plugins": plugins_out}
 
     @_serialized_plugin
+    @_reported("install")
     def install_from_catalogue(
         self,
         plugin_id: str,
@@ -568,6 +669,7 @@ class PluginManager:
         }
 
     @_serialized_plugin
+    @_reported("update")
     def update_plugin(
         self,
         plugin_id: str,

repeater/plugins/runtime.py

@@ -72,6 +72,7 @@ class PluginRuntime:
         self.crash_max_exits = crash_max_exits
         self._popen = popen_factory or subprocess.Popen
         self._run = run_factory or self._run_install_command
+        self.on_output: Optional[Callable[[str], None]] = None
         self._lock = threading.RLock()
         self._handles: dict[str, ProcessHandle] = {}
         self._runtime_state: dict[str, PluginState] = {}
@@ -232,6 +233,13 @@ class PluginRuntime:
             rebuild_marker.unlink()
             shutil.rmtree(backup)
 
+    def _announce(self, line: str) -> None:
+        if self.on_output is not None:
+            try:
+                self.on_output(line)
+            except Exception:
+                logger.debug("install output listener failed", exc_info=True)
+
     def _run_install_command(self, cmd, *, timeout, **kwargs):
         """Drain output into bounded memory and kill the entire installer group."""
         # Callers construct Python/venv/pip argv; installing trusted wheels is intentional.
@@ -245,12 +253,25 @@ class PluginRuntime:
             "installer", proc, None, process_group=proc.pid if os.name == "posix" else None
         )
         output = bytearray()
+        pending = bytearray()
+
+        def emit(line: bytes) -> None:
+            self._announce(line.decode("utf-8", errors="replace").rstrip("\r"))
 
         def drain():
             with proc.stdout as source:
                 while chunk := source.read1(64 * 1024):
                     output.extend(chunk)
                     del output[:-INSTALL_OUTPUT_MAX_BYTES]
+                    if self.on_output is not None:
+                        pending.extend(chunk)
+                        *lines, rest = pending.split(b"\n")
+                        pending[:] = rest
+                        del pending[:-INSTALL_OUTPUT_MAX_BYTES]
+                        for line in lines:
+                            emit(line)
+            if pending:
+                emit(bytes(pending))
 
         reader = threading.Thread(target=drain, name="plugin-install-output", daemon=True)
         reader.start()
@@ -271,6 +292,7 @@ class PluginRuntime:
             if py.is_file():
                 return
         venv_path.parent.mkdir(parents=True, exist_ok=True)
+        self._announce(f"Creating environment: {venv_path.name}")
         result = self._run(
             [self.python_executable, "-m", "venv", str(venv_path)],
             timeout=VENV_TIMEOUT_SECONDS,
@@ -288,6 +310,7 @@ class PluginRuntime:
         if not py.is_file():
             raise RuntimeError(f"venv python missing: {py}")
         install_path = self._ensure_pip_wheel_filename(wheel_path)
+        self._announce(f"Installing {wheel_path.name} with pip")
         try:
             result = self._run(
                 [str(py), "-m", "pip", "install", "--upgrade", str(install_path)],

repeater/web/openapi.yaml

@@ -4701,6 +4701,50 @@ paths:
           description: Updated plugin status
         '400':
           description: Update unavailable or invalid
+  /plugins/progress:
+    get:
+      tags: [Plugins]
+      summary: Stream an install or update's progress
+      description: |
+        Server-sent events for a plugin's last catalogue install or update, as
+        /update/progress streams the repeater's own updater. JSON events:
+        `connected`, `line`, `status` (idle, running, complete, error),
+        `keepalive`, and one `done` that ends the stream. Idle streams end after
+        a minute. Uploaded wheels are not reported.
+      security:
+        - BearerAuth: []
+        - ApiKeyAuth: []
+      parameters:
+        - name: id
+          in: query
+          required: true
+          schema:
+            type: string
+        - name: since
+          in: query
+          required: false
+          description: Resume from this line index
+          schema:
+            type: integer
+            default: 0
+        - name: fresh
+          in: query
+          required: false
+          description: Ignore a log already finished when the stream opens; wait for the next operation.
+          schema:
+            type: boolean
+            default: false
+      responses:
+        '504':
+          $ref: '#/components/responses/PluginOutcomeUnknown'
+        '200':
+          description: Event stream
+          content:
+            text/event-stream:
+              schema:
+                type: string
+        '503':
+          description: Plugin manager unavailable
   /plugins/settings:
     get:
       tags: [Plugins]

repeater/web/plugin_endpoints.py

@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import re
 import shutil
 import tempfile
+import time
 from pathlib import Path
 from typing import Any, Optional
 
@@ -22,6 +24,10 @@ from repeater.plugins.storage import resolve_plugin_socket_path, resolve_plugins
 
 logger = logging.getLogger("HTTPServer")
 
+PROGRESS_POLL_SECONDS = 0.75
+PROGRESS_MAX_SECONDS = 900.0  # the IPC completion budget
+PROGRESS_IDLE_SECONDS = 60.0
+
 
 class PluginAPIEndpoints:
     """CherryPy nested app for /api/plugins/*."""
@@ -306,6 +312,102 @@ class PluginAPIEndpoints:
             raise cherrypy.HTTPError(400, "tail must be an integer")
         return self._handle_ipc(lambda: self._client_or_raise().logs(plugin_id, tail=tail))
 
+    @cherrypy.expose
+    def progress(self, **kwargs):
+        """GET /api/plugins/progress?id=&since=&fresh= — SSE of an install or update."""
+        if cherrypy.request.method == "OPTIONS":
+            return ""
+        if cherrypy.request.method not in ("GET", "HEAD"):
+            raise cherrypy.HTTPError(405)
+        plugin_id = self._plugin_id_from(kwargs)
+        try:
+            since = int(kwargs.get("since", 0))
+        except (TypeError, ValueError):
+            raise cherrypy.HTTPError(400, "since must be an integer")
+        fresh = str(kwargs.get("fresh", "")).strip().lower() in {"1", "true", "yes", "on"}
+        cherrypy.response.headers["Content-Type"] = "text/event-stream"
+        if cherrypy.request.method == "HEAD":
+            return ""
+        cherrypy.response.headers["Cache-Control"] = "no-cache"
+        cherrypy.response.headers["X-Accel-Buffering"] = "no"
+        cherrypy.response.headers["Connection"] = "keep-alive"
+        return self._progress_events(plugin_id, since, fresh=fresh)
+
+    progress._cp_config = {"response.stream": True}
+
+    def _progress_events(
+        self,
+        plugin_id: str,
+        since: int,
+        *,
+        fresh: bool = False,
+        sleep=time.sleep,
+        clock=time.monotonic,
+    ):
+        def event(payload: dict) -> str:
+            return f"data: {json.dumps(payload)}\n\n"
+
+        def generate():
+            yield event({"type": "connected", "id": plugin_id})
+            cursor, last_state, quiet, stale = since, None, 0, None
+            opened = clock()
+            while True:
+                try:
+                    snap = self._client_or_raise().progress(plugin_id, since=cursor)
+                except GeneratorExit:
+                    return
+                except Exception as exc:
+                    yield event({"type": "done", "state": "error", "error": str(exc)})
+                    return
+
+                state, started = snap.get("state"), snap.get("started")
+                # fresh: a log already finished when the stream opened is not this operation's
+                if fresh and last_state is None and state in ("complete", "error"):
+                    stale = started
+                if stale is not None and started == stale:
+                    state, snap = "idle", {"next": cursor, "operation": None, "started": None}
+
+                changed = False
+                for line in snap.get("lines") or []:
+                    yield event({"type": "line", "line": line})
+                    changed = True
+                cursor = int(snap.get("next") or cursor)
+                if state != last_state:
+                    yield event(
+                        {
+                            "type": "status",
+                            "state": state,
+                            "operation": snap.get("operation"),
+                            "started": snap.get("started"),
+                        }
+                    )
+                    last_state, changed = state, True
+                if state in ("complete", "error"):
+                    yield event(
+                        {
+                            "type": "done",
+                            "state": state,
+                            "error": snap.get("error"),
+                            "started": snap.get("started"),
+                        }
+                    )
+                    return
+                now = clock()
+                if now >= opened + PROGRESS_MAX_SECONDS:
+                    yield event(
+                        {"type": "done", "state": "timeout", "error": "progress stream timed out"}
+                    )
+                    return
+                if state == "idle" and now >= opened + PROGRESS_IDLE_SECONDS:
+                    yield event({"type": "done", "state": "idle", "error": None})
+                    return
+                quiet = 0 if changed else quiet + 1
+                if quiet and quiet % 4 == 0:
+                    yield event({"type": "keepalive"})
+                sleep(PROGRESS_POLL_SECONDS)
+
+        return generate()
+
     @cherrypy.expose
     @cherrypy.tools.json_out()
     @cherrypy.tools.json_in(force=False)

tests/test_plugin_progress.py

@@ -0,0 +1,166 @@
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from repeater.plugins.ipc import PluginIPCClient, PluginIPCServer
+from repeater.plugins.manager import (
+    PROGRESS_MAX_LINES,
+    PROGRESS_MAX_LOGS,
+    OperationProgress,
+    PluginManager,
+    PluginManagerError,
+)
+from repeater.plugins.runtime import PluginRuntime
+from repeater.plugins.storage import PluginStorage
+from repeater.web.plugin_endpoints import PluginAPIEndpoints
+
+
+def test_runtime_hands_each_output_line_to_the_listener(tmp_path: Path):
+    runtime = PluginRuntime(PluginStorage(tmp_path / "plugins"))
+    heard: list[str] = []
+    runtime.on_output = heard.append
+    script = "import sys; print('Collecting demo'); print('Installing', end=''); sys.stdout.flush()"
+    result = runtime._run_install_command([sys.executable, "-c", script], timeout=30)
+    assert result.returncode == 0
+    assert heard == ["Collecting demo", "Installing"]
+    assert "Collecting demo" in result.stdout
+
+    def broken(_line: str) -> None:
+        raise RuntimeError("listener bug")
+
+    runtime.on_output = broken
+    assert (
+        runtime._run_install_command([sys.executable, "-c", "print('ok')"], timeout=30).returncode
+        == 0
+    )
+
+
+def test_progress_log_is_bounded_and_pages_by_cursor():
+    log = OperationProgress("openhop.demo", "update")
+    for i in range(PROGRESS_MAX_LINES + 20):
+        log.append(f"line {i}")
+    snap = log.snapshot()
+    assert len(snap["lines"]) == PROGRESS_MAX_LINES
+    assert snap["lines"][0] == "line 20"
+    assert snap["next"] == PROGRESS_MAX_LINES + 20
+    log.append("tail")
+    assert log.snapshot(since=snap["next"])["lines"] == ["tail"]
+    log.finish("boom")
+    assert (log.snapshot()["state"], log.snapshot()["error"]) == ("error", "boom")
+
+
+def test_manager_records_the_outcome_and_restores_the_listener(tmp_path: Path):
+    mgr = PluginManager(PluginStorage(tmp_path / "plugins"))
+    assert mgr.progress("missing.plugin")["state"] == "idle"
+    with pytest.raises(PluginManagerError):
+        mgr.update_plugin("missing.plugin")
+    snap = mgr.progress("missing.plugin")
+    assert (snap["operation"], snap["state"]) == ("update", "error")
+    assert "plugin not found" in snap["error"]
+    assert mgr.runtime.on_output is None
+
+
+def test_manager_keeps_a_bounded_number_of_logs(tmp_path: Path):
+    mgr = PluginManager(PluginStorage(tmp_path / "plugins"))
+    for i in range(PROGRESS_MAX_LOGS + 10):
+        with pytest.raises(PluginManagerError):
+            mgr.update_plugin(f"nobody.{i:03d}")
+    assert len(mgr._progress) == PROGRESS_MAX_LOGS
+    assert mgr.progress("nobody.000")["state"] == "idle"
+
+
+def test_ipc_round_trips_progress(tmp_path: Path):
+    storage = PluginStorage(tmp_path / "plugins")
+    manager = PluginManager(storage, PluginRuntime(storage, popen_factory=lambda *a, **k: None))
+    sock_path = Path(f"/tmp/oh-pm-progress-{tmp_path.name}.sock")
+    server = PluginIPCServer(sock_path, manager)
+    server.start()
+    try:
+        client = PluginIPCClient(sock_path)
+        assert client.progress("openhop.demo")["state"] == "idle"
+        with pytest.raises(PluginManagerError):
+            manager.update_plugin("openhop.demo")
+        assert client.progress("openhop.demo", since=0)["state"] == "error"
+    finally:
+        server.stop()
+
+
+def _snap(state, lines=(), nxt=0, started=None, error=None):
+    return {
+        "state": state,
+        "operation": "update",
+        "lines": list(lines),
+        "next": nxt,
+        "error": error,
+        "started": started,
+    }
+
+
+def _events(tmp_path, snapshots, *, ticks=None, fresh=False):
+    api = PluginAPIEndpoints({"storage": {"storage_dir": str(tmp_path)}})
+    answers = iter(snapshots)
+    client = SimpleNamespace(progress=lambda _id, since=0: next(answers, snapshots[-1]))
+    clock = iter(ticks or [0.0] * 1000)
+    with patch.object(api, "_client_or_raise", return_value=client):
+        chunks = api._progress_events(
+            "openhop.demo", 0, fresh=fresh, sleep=lambda _s: None, clock=lambda: next(clock)
+        )
+        return [json.loads(chunk[len("data: ") : -2]) for chunk in chunks]
+
+
+def test_stream_plays_lines_status_and_one_done(tmp_path):
+    events = _events(
+        tmp_path,
+        [_snap("running", ["a"], 1), _snap("running", ["b", "c"], 3), _snap("complete", nxt=3)],
+    )
+    assert [e["type"] for e in events] == [
+        "connected",
+        "line",
+        "status",
+        "line",
+        "line",
+        "status",
+        "done",
+    ]
+    assert [e["line"] for e in events if e["type"] == "line"] == ["a", "b", "c"]
+    assert events[-1] == {"type": "done", "state": "complete", "error": None, "started": None}
+
+    events = _events(tmp_path, [_snap("error", ["pip failed"], 1, error="pip")])
+    assert events[-1]["state"] == "error" and events[-1]["error"] == "pip"
+
+
+def test_idle_stream_keeps_alive_then_ends(tmp_path):
+    events = _events(tmp_path, [_snap("idle")], ticks=[0.0] + [1.0] * 12 + [100.0])
+    types = [e["type"] for e in events]
+    assert types[:2] == ["connected", "status"]
+    assert types.count("keepalive") == 2
+    assert events[-1] == {"type": "done", "state": "idle", "error": None}
+
+    events = _events(tmp_path, [_snap("running", started=5.0)], ticks=[0.0, 100.0, 500.0, 901.0])
+    assert events[-1]["state"] == "timeout"
+
+
+def test_fresh_stream_ignores_a_previous_operations_log(tmp_path):
+    previous = _snap("complete", ["old"], 1, started=100.0)
+    current = _snap("running", ["new"], 1, started=200.0)
+    events = _events(
+        tmp_path, [previous, previous, current, _snap("complete", nxt=1, started=200.0)], fresh=True
+    )
+    assert [e["type"] for e in events] == [
+        "connected",
+        "status",
+        "line",
+        "status",
+        "status",
+        "done",
+    ]
+    assert events[1]["state"] == "idle"
+    assert events[2]["line"] == "new"
+    assert events[-1]["started"] == 200.0
+
+    events = _events(tmp_path, [previous])
+    assert [e["type"] for e in events] == ["connected", "line", "status", "done"]