feat: read tapes from any Bub tape store in plugin mode

README.md

@@ -13,6 +13,7 @@ uvx bub-walkman ~/.bub/tapes/<tape>.jsonl
 When installed into Bub's environment (for example `bub install bub-walkman`), `bub-walkman` also works as a Bub plugin via `bub walkman <tape>`.
 
 - **Tape directory**: `bub.home / "tapes"` when Bub is installed, otherwise `$BUB_HOME/tapes` (default `~/.bub/tapes`).
+- **Other tape stores**: as `bub walkman`, tapes come from the tape store Bub is configured with (the `provide_tape_store` hook), so tapes kept by a store plugin (a database, for example) work too. Such a store has no files: the list shows entry counts instead of sizes, a followed tape is re-read every second, and deleting a tape resets it and its sidecars in the store. Bub's default file store is still read from the tape directory.
 - **Server**: starts a local server at `http://localhost:7357` by default (or a free port if `7357` is taken) and opens the page in the browser.
 - **Options**: `--host`, `--port`, `--no-open`, `--version`.
 

src/bub_walkman/cli.py

@@ -58,41 +58,50 @@ def _version(value: bool) -> None:
 
 
 def walkman(
+    ctx: typer.Context,
     tape: Annotated[
         Path | None,
         typer.Argument(
-            help="Tape .jsonl file, or the name of a tape in Bub's tape directory. Without one, the page lists the tapes "
+            help="Tape .jsonl file, or the name of a tape in Bub's tape store. Without one, the page lists the tapes "
             "(running ones first) to pick from.",
             show_default=False,
         ),
     ] = None,
     host: Annotated[str, typer.Option(help="Bind address.")] = "127.0.0.1",
     port: Annotated[int, typer.Option(help="Port; falls back to a free one when taken.")] = 7357,
     open_browser: Annotated[bool, typer.Option("--open/--no-open", help="Open the page in a browser.")] = True,
-    version: Annotated[
-        bool | None, typer.Option("--version", callback=_version, is_eager=True, help="Show the version and exit.")
-    ] = None,
+    version: Annotated[bool | None, typer.Option("--version", callback=_version, is_eager=True, help="Show the version and exit.")] = None,
 ) -> None:
     """Render a Bub tape as an agent timeline in the browser. Live following and replay are controlled from the web page."""
     import uvicorn
 
     from .server import create_app
+    from .sources import BubSource, FileSource, TapeSource
 
+    # As `bub walkman`, Bub passes its framework: read tapes from whatever tape store Bub is configured with.
+    framework = ctx.obj if hasattr(ctx.obj, "get_tape_store") and hasattr(ctx.obj, "running") else None
     directory = tapes_dir()
     path = None
+    name = None
     if tape is not None:
         path = resolve_tape(tape)
-        if path is None:
+        if path is not None:
+            name = path.stem
+        elif framework is not None and len(tape.parts) == 1:
+            name = tape.name.removesuffix(".jsonl")  # a tape in the store; the page says so if there is none
+        else:
             typer.echo(f"bub-walkman: tape not found: {tape} (also looked in {directory})", err=True)
             raise typer.Exit(2)
 
+    source: TapeSource = BubSource(framework, directory, path) if framework is not None else FileSource(directory, path)
     sock = _bind(host, port)
     bound = sock.getsockname()[1]
     shown = "localhost" if host in ("127.0.0.1", "0.0.0.0", "::") else host
-    url = f"http://{shown}:{bound}/" + (f"?tape={quote(path.stem)}" if path else "")
-    typer.echo(f"bub-walkman  {path.name if path else directory}\n  {url}\n  Ctrl+C to stop")
+    url = f"http://{shown}:{bound}/" + (f"?tape={quote(name)}" if name else "")
+    where = path.name if path else name or ("Bub's tape store" if framework is not None else directory)
+    typer.echo(f"bub-walkman  {where}\n  {url}\n  Ctrl+C to stop")
 
-    app = create_app(directory, path)
+    app = create_app(source)
 
     class Server(uvicorn.Server):
         @override

src/bub_walkman/server.py

@@ -1,4 +1,4 @@
-"""Starlette app serving the tapes of a directory: static UI, tape list, JSON snapshots and SSE tail streams."""
+"""Starlette app serving tapes: static UI, tape list, JSON snapshots and SSE tail streams."""
 
 import asyncio
 import json
@@ -17,55 +17,32 @@
 )
 from starlette.routing import Mount, Route
 from starlette.staticfiles import StaticFiles
-from watchfiles import awatch
 
-from .tapes import TapeDirectory, TapeReader, read_all, read_spill, sidecar_path
+from .sources import FileSource, TapeSource
 
 STATIC_DIR = Path(__file__).parent / "static"
 HEARTBEAT_INTERVAL = 15.0
-FALLBACK_POLL = 2.0
-
-
-def tape_info(path: Path) -> dict[str, Any]:
-    try:
-        st = path.stat()
-        size, mtime = st.st_size, st.st_mtime
-    except OSError:
-        size, mtime = 0, 0.0
-    return {"name": path.stem, "path": str(path), "size": size, "mtime": mtime}
 
 
 def sse(event: str, data: Any) -> str:
     return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
 
 
-async def _changes(path: Path, stop: asyncio.Event) -> AsyncIterator[None]:
-    """Yield whenever ``path`` may have changed: on filesystem events, or every few seconds."""
-    watcher = awatch(
-        path.parent,
-        stop_event=stop,
-        debounce=50,
-        step=50,
-        rust_timeout=int(FALLBACK_POLL * 1000),
-        yield_on_timeout=True,
-        watch_filter=lambda _, p: Path(p).name == path.name,
-    )
-    async for _ in watcher:
-        yield None
-
+def create_app(source: TapeSource | Path, tape: Path | None = None) -> Starlette:
+    """Serve the tapes of ``source``: a tape store, or a tape directory (plus ``tape``, which may live elsewhere).
 
-def create_app(tapes_dir: Path, tape: Path | None = None) -> Starlette:
-    """Serve the tapes in ``tapes_dir`` (plus ``tape``, which may live elsewhere). API calls pick one with ``?tape=<name>``."""
-    directory = TapeDirectory(tapes_dir, tape)
-    home = str(Path.home())
-    shown_dir = "~" + str(tapes_dir)[len(home) :] if str(tapes_dir).startswith(home + "/") else str(tapes_dir)
+    API calls pick one tape with ``?tape=<name>``.
+    """
+    if isinstance(source, Path):
+        source = FileSource(source, tape)
+    tapes_source = source
 
-    def pick(request: Request) -> Path | Response:
+    async def pick(request: Request) -> tuple[str, dict[str, Any]] | Response:
         name = request.query_params.get("tape")
         if not name:
             return JSONResponse({"error": "missing ?tape=<name>"}, 400)
-        path = directory.resolve(name)
-        return path if path is not None else JSONResponse({"error": f"tape not found: {name}"}, 404)
+        info = await tapes_source.info(name)
+        return (name, info) if info is not None else JSONResponse({"error": f"tape not found: {name}"}, 404)
 
     # Stop events of open SSE streams; set them all on shutdown so streams end and the server exits promptly.
     streams: set[asyncio.Event] = set()
@@ -83,8 +60,8 @@ async def favicon(request: Request) -> Response:
         return FileResponse(icon, media_type="image/svg+xml", headers={"Cache-Control": "max-age=86400"})
 
     async def tapes(request: Request) -> Response:
-        items = await asyncio.to_thread(directory.summaries)
-        return JSONResponse({"directory": shown_dir, "tapes": items}, headers={"Cache-Control": "no-store"})
+        items = await tapes_source.summaries()
+        return JSONResponse({"directory": tapes_source.label, "tapes": items}, headers={"Cache-Control": "no-store"})
 
     async def delete(request: Request) -> Response:
         # Deleting files is only for the person at this machine: a loopback client, our own page (not a page from
@@ -104,64 +81,50 @@ async def delete(request: Request) -> Response:
             names = [str(n) for n in body["names"]]
         except (ValueError, KeyError, TypeError):
             return JSONResponse({"error": 'expected {"names": [...], "force": false}'}, 400)
-        deleted, errors = await asyncio.to_thread(directory.delete, names, bool(body.get("force")))
+        deleted, errors = await tapes_source.delete(names, bool(body.get("force")))
         return JSONResponse({"deleted": deleted, "errors": errors}, 200 if not errors else 409)
 
     async def snapshot(request: Request) -> Response:
-        tape = pick(request)
-        if isinstance(tape, Response):
-            return tape
-        entries, offset = await asyncio.to_thread(read_all, tape)
-        return JSONResponse({"tape": tape_info(tape), "entries": entries, "offset": offset})
+        picked = await pick(request)
+        if isinstance(picked, Response):
+            return picked
+        name, info = picked
+        entries, offset = await tapes_source.snapshot(name)
+        return JSONResponse({"tape": info, "entries": entries, "offset": offset})
 
     async def spill(request: Request) -> Response:
-        tape = pick(request)
-        if isinstance(tape, Response):
-            return tape
+        picked = await pick(request)
+        if isinstance(picked, Response):
+            return picked
         handle = request.path_params["handle"]
         if not handle.isalnum():
             return JSONResponse({"error": "bad handle"}, 400)
-        path = sidecar_path(tape, "spill")
-        if not path.is_file():
-            return JSONResponse({"error": "spill sidecar not found"}, 404)
-        data = await asyncio.to_thread(read_spill, path, handle)
+        data = await tapes_source.spill(picked[0], handle)
+        if isinstance(data, str):
+            return JSONResponse({"error": data}, 404)
         if not data["chunks"]:
             return JSONResponse({"error": "spill handle not found (sidecar may have been reset)"}, 404)
         return JSONResponse(data)
 
     async def stream(request: Request) -> Response:
-        picked = pick(request)
+        picked = await pick(request)
         if isinstance(picked, Response):
             return picked
-        tape = picked
+        name = picked[0]
         offset = int(request.query_params.get("offset") or 0)
 
         async def events() -> AsyncIterator[str]:
-            reader = TapeReader(tape)
-            try:
-                st = tape.stat()
-            except OSError:
-                st = None
-            if st is not None and 0 < offset <= st.st_size:
-                reader.resume(offset, st)  # continue right after the client's snapshot
-            elif offset:
-                yield sse("reset", {"offset": 0})
-            yield sse("hello", {"tape": tape_info(tape), "offset": reader.offset})
-
             stop = asyncio.Event()
             streams.add(stop)
             loop = asyncio.get_running_loop()
             last_beat = loop.time()
             try:
-                async for _ in _changes(tape, stop):
-                    if await request.is_disconnected():
-                        break
-                    reset, entries = await asyncio.to_thread(reader.poll)
-                    if reset:
-                        yield sse("reset", {"offset": 0})
-                    if entries:
-                        yield sse("entries", {"entries": entries, "offset": reader.offset})
+                async for event in tapes_source.follow(name, offset, stop):
+                    if event is not None:
+                        yield sse(*event)
                         last_beat = loop.time()
+                    elif await request.is_disconnected():
+                        break
                     elif loop.time() - last_beat > HEARTBEAT_INTERVAL:
                         yield ": beat\n\n"
                         last_beat = loop.time()
@@ -176,6 +139,7 @@ async def events() -> AsyncIterator[str]:
         )
 
     app = Starlette(
+        lifespan=lambda app: tapes_source.lifespan(),
         routes=[
             Route("/", index),
             Route("/favicon.svg", favicon),
@@ -185,8 +149,9 @@ async def events() -> AsyncIterator[str]:
             Route("/api/stream", stream),
             Route("/api/spill/{handle}", spill),
             Mount("/assets", StaticFiles(directory=STATIC_DIR / "assets", check_dir=False), name="assets"),
-        ]
+        ],
     )
+
     def close_streams() -> None:
         for stop in list(streams):
             stop.set()