feat(serve): add --quiet mode and developer dashboard API

Headless serve with plain stderr logging; /api/dashboard/* (status, sessions, jobs, logs, SSE/poll events) behind a separate serve.dashboard.key.

docs/dev/rest-api/dashboard.md

@@ -0,0 +1,82 @@
+# Developer Dashboard
+
+The dashboard endpoints show the operator of a `serve` instance what the server is doing:
+version and uptime, request counts, live remote sessions, download jobs, and the recent
+log lines. They exist so you can watch a headless server (`serve --quiet`) from a web page
+instead of a terminal.
+
+The dashboard endpoints are for the person who runs the server, not for API users. They
+have their own key: a user key or a tier key never opens them.
+
+## Enable
+
+Set a dashboard key under `serve.dashboard` in `unshackle.yaml`:
+
+```yaml
+serve:
+  api_secret: ...
+  dashboard:
+    key: a-long-random-string
+```
+
+Without `dashboard.key` the server does not register the routes, and they answer `404`.
+`serve --no-key` turns off user authentication only; the dashboard key stays required.
+Once the dashboard key is set, every `/api/dashboard/` request must carry it in
+`X-Secret-Key`.
+Every other endpoint refuses the dashboard key, and the dashboard refuses user keys. The
+routes are available in `--remote-only` mode too.
+
+Run the server headless with `--quiet`: no banner, no Rich output, one plain log line per
+event on stderr, and `api.*` loggers stay at `INFO` so remote session activity reaches the
+log ring. Add `--debug` to raise the level to `DEBUG` and also write the full log to
+`directories.logs` as `unshackle_serve_<time>.log`. Nothing else the dashboard shows reaches
+the disk: stats, the log ring and the event stream live in memory and reset on restart, so a
+dashboard UI that wants history must keep its own copy.
+
+```
+unshackle serve --remote-only --quiet
+```
+
+## Endpoints
+
+All are `GET`.
+
+| Path | Returns |
+|------|---------|
+| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), loaded service count, session count, `max_sessions`, `session_ttl`, job counts by status. |
+| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the first resolved title's display name), `auth_status`, `auth_error`, `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`. `client` is whatever the remote client sent as `client` in its session create request (the current CLI sends nothing, so it is `{}`). `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. |
+| `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. |
+| `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. |
+| `/api/dashboard/events` | Server-Sent Events stream. With `?since=<seq>` it returns `{"seq", "stats", "events": [...]}` instead: every event after `seq` (the server keeps the last 20000, log records included, so a poll every few seconds is safe even under a busy access log) as one JSON burst, so a UI that polls every few seconds still sees a remote session that opened and closed in between. |
+
+`requests_rejected` counts 401 responses; those never appear in `requests_by_key`. The
+label is the `username` from the `users` entry when one is set, else the first four
+characters of the API key.
+
+## Event stream
+
+`/api/dashboard/events` sends a `stats` event (the same body as `/status`) on connect and
+then every 5 seconds, so the connection stays alive and the counters stay current. Between
+those it sends:
+
+| Event | Data |
+|-------|------|
+| `log` | One log record: `seq`, `ts`, `level`, `logger`, `msg`. |
+| `session` | `action` (`create`, `update` on every request the session makes and when titles or tracks load, or `delete`) plus the full remote session summary, so each event is a complete snapshot. A `delete` also carries `reason`: `closed` (the client ended it), `expired` (TTL sweep) or `evicted` (`max_sessions` reached). A session opened by `--skip-dl` can live only a few seconds, so a UI that wants history must keep its own record of these events. |
+| `job` | `event` (`queued`, `progress`, `status`, or the terminal status) plus the full job object. |
+
+Every frame carries an `id:` (the event seq). After a dropped connection an `EventSource`
+reconnects with `Last-Event-ID` and the server replays the events it missed. The connection
+never closes on its own.
+
+A browser `EventSource` cannot set headers, so this route also accepts the dashboard key as
+`?secret_key=`. The dashboard key then appears in proxy and access logs and in browser history, so
+use the query parameter only for `/events` and send the header everywhere else.
+
+```js
+const es = new EventSource(`${server}/api/dashboard/events?secret_key=${key}`);
+es.addEventListener("log", (e) => console.log(JSON.parse(e.data).msg));
+```
+
+CORS is open (`Access-Control-Allow-Origin: *`), so a dashboard page served from another
+origin works without a proxy.

docs/reference/configuration/services.md

@@ -119,6 +119,7 @@ is the [REST API](../../dev/rest-api/index.md) section. These are the config key
 | `users` | dict | `{}` | Per-user API keys and their allowlists (see below). |
 | `services` | list | *(unset)* | Global service allowlist. Omit to allow all. |
 | `remote_only` | bool | `false` | Expose only the remote service session endpoints (health, services, search, session) and disable the rest of the REST API. |
+| `dashboard` | dict | *(unset)* | Developer dashboard: `key` is the API key for the read-only `/api/dashboard/` endpoints (status, sessions, jobs, logs, SSE events). Unset leaves those routes unregistered. See [dashboard endpoints](../../dev/rest-api/dashboard.md). |
 | `session_ttl` | int (s) | `300` | Lifetime of an interactive auth session. |
 | `max_sessions` | int | `100` | Maximum concurrent sessions. |
 | `history_limit` | int | `100` | How many finished jobs to retain in history. |

mkdocs.yml

@@ -140,6 +140,7 @@ nav:
           - Authentication: dev/rest-api/authentication.md
           - Endpoints: dev/rest-api/endpoints.md
           - Job Lifecycle: dev/rest-api/job-lifecycle.md
+          - Dashboard: dev/rest-api/dashboard.md
           - Remote Sessions: dev/rest-api/remote-sessions.md
           - Error Responses: dev/rest-api/errors.md
   - API Reference:

tests/remote/unit/test_dashboard.py

@@ -0,0 +1,110 @@
+"""Dashboard auth split, stats middleware, log ring buffer and session events."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+import pytest
+from aiohttp import web
+
+from unshackle.core.api.events import bus
+from unshackle.core.api.handlers import dashboard_authentication
+from unshackle.core.api.routes import cors_middleware, setup_routes
+from unshackle.core.api.session_store import SessionStore
+from unshackle.core.api.stats import RingLogHandler, stats, stats_middleware
+from unshackle.core.config import config
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def dashboard_cfg(monkeypatch):
+    monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"})
+    monkeypatch.setitem(config.serve, "users", {"tier-key": {"username": "tier1"}})
+    stats.requests_total = 0
+    stats.requests_by_key.clear()
+
+
+def make_app(dashboard: bool = True) -> web.Application:
+    app = web.Application(middlewares=[cors_middleware, stats_middleware, dashboard_authentication])
+    app["config"] = {"users": {}}
+    app["debug_api"] = False
+    setup_routes(app, remote_only=True, dashboard=dashboard)
+    return app
+
+
+async def test_dashboard_key_split(aiohttp_client, dashboard_cfg) -> None:
+    client = await aiohttp_client(make_app())
+    assert (await client.get("/api/dashboard/status", headers={"X-Secret-Key": "tier-key"})).status == 401
+    assert (await client.get("/api/dashboard/status")).status == 401
+    resp = await client.get("/api/dashboard/status", headers={"X-Secret-Key": "dash-secret"})
+    assert resp.status == 200
+    body = await resp.json()
+    assert body["mode"] and "uptime_seconds" in body and body["sessions"] == 0
+    # rejected requests count only as rejected; accepted ones are keyed by masked label
+    assert stats.requests_total == 3 and stats.requests_rejected == 2 and stats.requests_by_key == {"dash…": 1}
+    await client.get("/api/health", headers={"X-Secret-Key": "tier-key"})
+    assert stats.requests_by_key["tier1"] == 1
+
+
+async def test_dashboard_routes_absent_without_key(aiohttp_client, dashboard_cfg) -> None:
+    client = await aiohttp_client(make_app(dashboard=False))
+    assert (await client.get("/api/dashboard/status", headers={"X-Secret-Key": "dash-secret"})).status == 404
+
+
+def test_ring_since_and_level() -> None:
+    ring = RingLogHandler(maxlen=3)
+    ring.setFormatter(logging.Formatter("%(message)s"))
+    logger = logging.getLogger("test.ring")
+    logger.propagate = False
+    logger.addHandler(ring)
+    logger.setLevel(logging.DEBUG)
+    for i in range(5):
+        (logger.warning if i % 2 else logger.info)(f"m{i}")
+    assert [r["msg"] for r in ring.records] == ["m2", "m3", "m4"]
+    assert ring.seq == 5
+    assert [r["msg"] for r in ring.since(3)] == ["m3", "m4"]
+    assert [r["msg"] for r in ring.since(0, "warning")] == ["m3"]
+    assert ring.since(0, logger="test") and not ring.since(0, logger="other")
+
+
+async def test_session_events() -> None:
+    queue = bus.subscribe()
+    try:
+        store = SessionStore()
+        entry = await store.create("EXAMPLE", object())
+        assert [e.session_id for e in store.list()] == [entry.session_id]
+        await store.delete(entry.session_id)
+        create, delete = queue.get_nowait(), queue.get_nowait()
+        assert create["event"] == "session" and create["data"]["action"] == "create"
+        assert delete["data"]["action"] == "delete" and delete["data"]["id"] == entry.session_id
+        assert "owner_key" not in create["data"]
+    finally:
+        bus.unsubscribe(queue)
+    await asyncio.sleep(0)
+
+
+async def test_events_burst_mode(aiohttp_client, dashboard_cfg) -> None:
+    client = await aiohttp_client(make_app())
+    mark = bus.seq
+    bus.publish("session", {"action": "create", "id": "s1"})
+    bus.publish("session", {"action": "delete", "id": "s1"})
+    resp = await client.get(f"/api/dashboard/events?since={mark}", headers={"X-Secret-Key": "dash-secret"})
+    assert resp.status == 200
+    body = await resp.json()
+    assert body["seq"] == bus.seq and "stats" in body
+    assert [e["data"]["action"] for e in body["events"] if e["event"] == "session"] == ["create", "delete"]
+
+
+async def test_session_action_log() -> None:
+    store = SessionStore()
+    entry = await store.create("EXAMPLE", object(), owner_key="k")
+    queue = bus.subscribe()
+    try:
+        store.record_action(entry.session_id, {"action": "titles", "status": 200})
+        store.record_action("missing", {"action": "titles", "status": 200})
+        assert entry.summary()["actions"] == [{"action": "titles", "status": 200}]
+        assert queue.get_nowait()["data"]["action"] == "update" and queue.empty()
+    finally:
+        bus.unsubscribe(queue)

tests/remote/unit/test_serve_cli.py

@@ -32,6 +32,7 @@ def test_serve_help_lists_documented_flags(runner: CliRunner) -> None:
         "--debug-api",
         "--debug",
         "--remote-only",
+        "--quiet",
     ):
         assert flag in out, f"missing flag in --help: {flag}"
 

unshackle/commands/serve.py

@@ -2,7 +2,9 @@ import asyncio
 import hmac
 import logging
 import subprocess
+import sys
 from contextlib import suppress
+from datetime import datetime
 
 import click
 from aiohttp import web
@@ -10,8 +12,15 @@ from aiohttp import web
 from unshackle.core import binaries
 from unshackle.core.api import cors_middleware, setup_routes, setup_swagger
 from unshackle.core.api.compression import compression_middleware
-from unshackle.core.api.handlers import request_secret_key
+from unshackle.core.api.handlers import (
+    DASHBOARD_PREFIX,
+    dashboard_authentication,
+    dashboard_key,
+    request_secret_key,
+)
+from unshackle.core.api.stats import ring, stats, stats_middleware
 from unshackle.core.config import config
+from unshackle.core.console import console
 from unshackle.core.constants import context_settings
 from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit
 
@@ -92,6 +101,13 @@ def _install_service_refresh(app: web.Application) -> None:
     default=False,
     help="Only expose remote service session endpoints (health, services, search, session).",
 )
+@click.option(
+    "-q",
+    "--quiet",
+    is_flag=True,
+    default=False,
+    help="Headless mode: no banner and no Rich output, plain log lines on stderr.",
+)
 def serve(
     host: str,
     port: int,
@@ -103,6 +119,7 @@ def serve(
     debug_api: bool,
     debug: bool,
     remote_only: bool,
+    quiet: bool,
 ) -> None:
     """
     Serve your Local Widevine and PlayReady Devices and REST API for Remote Access.
@@ -136,12 +153,35 @@ def serve(
 
     log = logging.getLogger("serve")
 
+    if quiet:
+        logging.basicConfig(
+            force=True,
+            level=logging.DEBUG if debug else logging.INFO,
+            format="%(asctime)s %(name)s %(levelname)s %(message)s",
+            stream=sys.stderr,
+        )
+        console.quiet = True
+        if not debug:
+            for handler in logging.getLogger().handlers:
+                handler.addFilter(lambda record: record.name != "aiohttp.access")
+        if debug:
+            log_path = config.directories.logs / config.filenames.log.format(
+                name="serve", time=datetime.now().strftime("%Y%m%d-%H%M%S")
+            )
+            log_path.parent.mkdir(parents=True, exist_ok=True)
+            file_handler = logging.FileHandler(log_path, encoding="utf-8")
+            file_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s"))
+            logging.getLogger().addHandler(file_handler)
+            log.info(f"Writing log to {log_path}")
+    elif debug:
+        logging.getLogger().setLevel(logging.DEBUG)
     if debug:
-        logging.basicConfig(level=logging.DEBUG, format="%(name)s - %(levelname)s - %(message)s")
         log.info("Debug logging enabled for API operations")
-    else:
+    elif not quiet:
         logging.getLogger("api").setLevel(logging.WARNING)
         logging.getLogger("api.remote").setLevel(logging.WARNING)
+    ring.setFormatter(logging.Formatter("%(message)s"))
+    logging.getLogger().addHandler(ring)
 
     if not no_key:
         api_secret = config.serve.get("api_secret")
@@ -180,7 +220,7 @@ def serve(
         @web.middleware
         async def api_key_authentication(request: web.Request, handler) -> web.StreamResponse:
             """Authenticate API requests using X-Secret-Key header."""
-            if request.path == "/api/health":
+            if request.path == "/api/health" or request.path.startswith(DASHBOARD_PREFIX):
                 return await handler(request)
             secret_key = request_secret_key(request)
             if not secret_key:
@@ -194,6 +234,13 @@ def serve(
         remote_only = remote_only or config.serve.get("remote_only", False)
         if remote_only:
             api_only = True
+        stats.host, stats.port = host, port
+        stats.mode = "remote_only" if remote_only else "api_only" if api_only else "full"
+        dashboard = bool(dashboard_key())
+        if dashboard:
+            log.info(f"Developer dashboard endpoints available at http://{host}:{port}/api/dashboard/")
+        elif quiet:
+            log.info("Developer dashboard disabled: set serve.dashboard.key in unshackle.yaml to enable it")
 
         try:
             global_speed_limit = parse_speed_limit(config.serve.get("global_speed_limit"))
@@ -220,10 +267,20 @@ def serve(
         if api_only:
             log.info("Starting REST API server (pywidevine/pyplayready CDM disabled)")
             if no_key:
-                app = web.Application(middlewares=[cors_middleware, compression_middleware])
+                app = web.Application(
+                    middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
+                )
                 app["config"] = {"users": {}}
             else:
-                app = web.Application(middlewares=[cors_middleware, api_key_authentication, compression_middleware])
+                app = web.Application(
+                    middlewares=[
+                        cors_middleware,
+                        stats_middleware,
+                        dashboard_authentication,
+                        api_key_authentication,
+                        compression_middleware,
+                    ]
+                )
                 api_users: dict = {api_secret: {"devices": [], "username": "api_user"}}
                 if isinstance(users, dict):
                     api_users.update(users)
@@ -245,9 +302,9 @@ def serve(
             app.on_cleanup.append(stop_session_cleanup)
             _install_service_refresh(app)
 
-            setup_routes(app, remote_only=remote_only)
+            setup_routes(app, remote_only=remote_only, dashboard=dashboard)
             if not remote_only:
-                setup_swagger(app)
+                setup_swagger(app, dashboard=dashboard)
                 log.info(f"REST API endpoints available at http://{host}:{port}/api/")
                 log.info(f"Swagger UI available at http://{host}:{port}/api/docs/")
             else:
@@ -295,6 +352,8 @@ def serve(
                 @web.middleware
                 async def serve_authentication(request: web.Request, handler) -> web.StreamResponse:
                     secret_key = request_secret_key(request)
+                    if request.path.startswith(DASHBOARD_PREFIX):
+                        return await handler(request)
                     if serve_playready_flag and request.path in ("/playready", "/playready/"):
                         response = await handler(request)
                     elif secret_key and not request.headers.get("X-Secret-Key"):
@@ -316,10 +375,20 @@ def serve(
                 return serve_authentication
 
             if no_key:
-                app = web.Application(middlewares=[cors_middleware, compression_middleware])
+                app = web.Application(
+                    middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
+                )
             else:
                 serve_auth = create_serve_authentication(serve_playready and bool(prd_devices))
-                app = web.Application(middlewares=[cors_middleware, serve_auth, compression_middleware])
+                app = web.Application(
+                    middlewares=[
+                        cors_middleware,
+                        dashboard_authentication,
+                        serve_auth,
+                        stats_middleware,
+                        compression_middleware,
+                    ]
+                )
 
             app["config"] = serve_config
             app["debug_api"] = debug_api
@@ -382,14 +451,14 @@ def serve(
             elif serve_playready:
                 log.info("No PlayReady devices found, skipping PlayReady CDM endpoints")
 
-            setup_routes(app, remote_only=remote_only)
+            setup_routes(app, remote_only=remote_only, dashboard=dashboard)
 
             if serve_widevine:
                 log.info(f"Widevine CDM endpoints available at http://{host}:{port}/{{device}}/open")
             if remote_only:
                 log.info(f"Remote service endpoints available at http://{host}:{port}/api/session/")
             else:
-                setup_swagger(app)
+                setup_swagger(app, dashboard=dashboard)
                 log.info(f"REST API endpoints available at http://{host}:{port}/api/")
                 log.info(f"Swagger UI available at http://{host}:{port}/api/docs/")
             log.info("(Press CTRL+C to quit)")

unshackle/core/__main__.py

@@ -1,5 +1,6 @@
 import atexit
 import logging
+import sys
 from datetime import datetime
 
 import click
@@ -52,6 +53,9 @@ def main(version: bool, debug: bool) -> None:
 
     traceback.install(console=console, width=80, suppress=[click])
 
+    if "--quiet" in sys.argv[1:] or "-q" in sys.argv[1:]:
+        return
+
     console.print(
         Padding(
             Group(

unshackle/core/api/download_manager.py

@@ -15,6 +15,7 @@ from enum import Enum
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional
 
+from unshackle.core.api.events import bus
 from unshackle.core.api.sanitize import sanitize_log
 from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text
 
@@ -675,6 +676,7 @@ class DownloadQueueManager:
         self._job_queue.put_nowait(job)
 
         log.info(f"Created download job {job_id} for {sanitize_log(service)}:{sanitize_log(title_id)}")
+        bus.publish("job", {"event": "queued", **job.to_dict(include_full_details=True)})
         return job
 
     def get_job(self, job_id: str) -> Optional[DownloadJob]:
@@ -703,6 +705,7 @@ class DownloadQueueManager:
 
     def publish(self, job: DownloadJob, event: str) -> None:
         """Fan an event carrying the job's full detail dict out to every listener."""
+        bus.publish("job", {"event": event, **job.to_dict(include_full_details=True)})
         queues = self._subscribers.get(job.job_id)
         if not queues:
             return

unshackle/core/api/events.py

@@ -0,0 +1,56 @@
+"""Server-wide event broadcaster feeding the dashboard SSE event stream."""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from collections import deque
+from contextlib import suppress
+from typing import Any, Deque, Dict, List, Optional
+
+
+class EventBus:
+    """Fan events out to every subscriber queue; drop when a subscriber falls behind."""
+
+    def __init__(self, history: int = 20000) -> None:
+        self._subs: List[asyncio.Queue] = []
+        self.seq = 0
+        self.history: Deque[Dict[str, Any]] = deque(maxlen=history)
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+        self._loop_thread: Optional[threading.Thread] = None
+
+    def subscribe(self) -> asyncio.Queue:
+        self._loop = asyncio.get_running_loop()
+        self._loop_thread = threading.current_thread()
+        queue: asyncio.Queue = asyncio.Queue(maxsize=200)
+        self._subs.append(queue)
+        return queue
+
+    def unsubscribe(self, queue: asyncio.Queue) -> None:
+        with suppress(ValueError):
+            self._subs.remove(queue)
+
+    def since(self, seq: int) -> List[Dict[str, Any]]:
+        return [item for item in self.history if item["seq"] > seq]
+
+    def publish(self, event: str, data: Dict[str, Any]) -> None:
+        self.seq += 1
+        item = {"seq": self.seq, "event": event, "data": data}
+        self.history.append(item)
+        if not self._subs:
+            return
+        on_loop = self._loop is None or threading.current_thread() is self._loop_thread
+        for queue in list(self._subs):
+            if on_loop:
+                _put_drop(queue, item)
+            else:
+                assert self._loop is not None
+                self._loop.call_soon_threadsafe(_put_drop, queue, item)
+
+
+def _put_drop(queue: asyncio.Queue, item: Dict[str, Any]) -> None:
+    with suppress(asyncio.QueueFull):
+        queue.put_nowait(item)
+
+
+bus = EventBus()

unshackle/core/api/handlers.py

@@ -15,6 +15,7 @@ from unshackle.core.api.compression import safe_inflate
 from unshackle.core.api.errors import APIError, APIErrorCode, handle_api_exception
 from unshackle.core.api.input_bridge import AuthStatus, InputBridge
 from unshackle.core.api.sanitize import safe_cache_key, sanitize_log
+from unshackle.core.api.session_store import SessionStore
 from unshackle.core.config import config
 from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP
 from unshackle.core.providers.anilist import parse_anilist_ref
@@ -325,6 +326,9 @@ def server_proxy_allowed(request: Optional[web.Request] = None) -> bool:
 
 
 JOB_EVENTS_ROUTE = "/api/download/jobs/{job_id}/events"
+DASHBOARD_PREFIX = "/api/dashboard/"
+DASHBOARD_EVENTS_ROUTE = DASHBOARD_PREFIX + "events"
+SSE_QUERY_KEY_ROUTES = {JOB_EVENTS_ROUTE, DASHBOARD_EVENTS_ROUTE}
 
 CORS_HEADERS = {
     "Access-Control-Allow-Origin": "*",
@@ -335,16 +339,42 @@ CORS_HEADERS = {
 
 
 def request_secret_key(request: web.Request) -> Optional[str]:
-    """The caller's API key: the X-Secret-Key header, or on the events route the secret_key
-    query param (EventSource cannot send headers)."""
+    """The caller's API key: the X-Secret-Key header, or on SSE routes the secret_key query param."""
     key = request.headers.get("X-Secret-Key")
     if not key:
         resource = request.match_info.route.resource
-        if resource is not None and resource.canonical == JOB_EVENTS_ROUTE:
+        if resource is not None and resource.canonical in SSE_QUERY_KEY_ROUTES:
             key = request.query.get("secret_key")
     return key
 
 
+def dashboard_key() -> Optional[str]:
+    """The developer dashboard key from serve.dashboard.key, or None when the dashboard is off."""
+    dashboard = (config.serve or {}).get("dashboard")
+    key = dashboard.get("key") if isinstance(dashboard, dict) else None
+    return str(key) if key else None
+
+
+@web.middleware
+async def dashboard_authentication(request: web.Request, handler: Any) -> web.StreamResponse:
+    """Gate /api/dashboard/ behind serve.dashboard.key only; user and tier keys never pass.
+
+    This middleware runs before the user-key middleware, so it answers the dashboard routes
+    first and no other route ever accepts the dashboard key.
+    """
+    import hmac
+
+    if not request.path.startswith(DASHBOARD_PREFIX):
+        return await handler(request)
+    expected = dashboard_key()
+    if not expected:
+        return await handler(request)  # routes are unregistered, so this 404s
+    provided = request_secret_key(request)
+    if not provided or not hmac.compare_digest(provided, expected):
+        return web.json_response({"status": 401, "message": "Dashboard key is invalid."}, status=401)
+    return await handler(request)
+
+
 def caller_key(request: Optional[web.Request] = None) -> str:
     """The authenticating API key for a request, or 'anonymous' when unauthenticated."""
     if not request:
@@ -1422,7 +1452,7 @@ async def download_job_events_handler(job_id: str, request: web.Request) -> web.
     response = web.StreamResponse(
         headers={
             "Content-Type": "text/event-stream",
-            "Cache-Control": "no-cache",
+            "Cache-Control": "no-cache, no-transform",
             # Stops nginx from buffering the stream.
             "X-Accel-Buffering": "no",
             **CORS_HEADERS,
@@ -1468,6 +1498,124 @@ async def download_job_events_handler(job_id: str, request: web.Request) -> web.
     return response
 
 
+async def dashboard_status_handler(request: web.Request) -> web.Response:
+    """Server-wide stats for the developer dashboard."""
+    from collections import Counter
+
+    from unshackle.core.api.download_manager import get_download_manager
+    from unshackle.core.api.session_store import get_session_store
+    from unshackle.core.api.stats import stats
+
+    store = get_session_store()
+    jobs = Counter(job.status.value for job in get_download_manager().list_jobs())
+    return web.json_response(
+        {
+            **stats.to_dict(),
+            "services": len(Services.get_tags()),
+            "sessions": store.session_count,
+            "max_sessions": store.max_sessions,
+            "session_ttl": store.ttl,
+            "jobs": dict(jobs),
+        }
+    )
+
+
+async def dashboard_sessions_handler(request: web.Request) -> web.Response:
+    """Every live remote session, not owner-scoped."""
+    from unshackle.core.api.session_store import get_session_store
+
+    return web.json_response([entry.summary() for entry in get_session_store().list()])
+
+
+async def dashboard_jobs_handler(request: web.Request) -> web.Response:
+    """Every download job, not owner-scoped."""
+    from unshackle.core.api.download_manager import get_download_manager
+
+    return web.json_response([job.to_dict(include_full_details=True) for job in get_download_manager().list_jobs()])
+
+
+async def dashboard_logs_handler(request: web.Request) -> web.Response:
+    """Recent log records; `since` (seq) and `level` filter the ring buffer."""
+    from unshackle.core.api.stats import ring
+
+    try:
+        since = int(request.query.get("since", 0))
+    except ValueError:
+        since = 0
+    records = ring.since(since, request.query.get("level"), request.query.get("logger"))
+    return web.json_response({"seq": ring.seq, "records": records})
+
+
+async def dashboard_events_handler(request: web.Request) -> web.StreamResponse:
+    """Server-wide SSE event stream: `log`, `session` and `job` events, plus `stats` every 5 s (also the keep-alive).
+
+    Every frame carries `id: <seq>`; a reconnecting EventSource sends Last-Event-ID, and the
+    server replays the missed events from the bus history. With `?since=<seq>` the same missed
+    events come back as one JSON burst instead of an event stream, for clients that poll.
+    """
+    from unshackle.core.api.events import bus
+
+    if "since" in request.query:
+        try:
+            since = int(request.query["since"])
+        except ValueError:
+            since = 0
+        status = await dashboard_status_handler(request)
+        return web.json_response({"seq": bus.seq, "stats": json.loads(status.text or "{}"), "events": bus.since(since)})
+
+    response = web.StreamResponse(
+        headers={
+            "Content-Type": "text/event-stream",
+            # no-transform stops a CDN from compressing or buffering the event stream.
+            "Cache-Control": "no-cache, no-transform",
+            "X-Accel-Buffering": "no",
+            **CORS_HEADERS,
+        }
+    )
+    await response.prepare(request)
+
+    async def send(event: str, data: Dict[str, Any], event_id: Optional[int] = None) -> None:
+        payload = json.dumps(data, separators=(",", ":"), default=str)
+        head = f"id: {event_id}\n" if event_id is not None else ""
+        await response.write(f"{head}event: {event}\ndata: {payload}\n\n".encode())
+
+    async def send_stats() -> None:
+        status = await dashboard_status_handler(request)
+        await send("stats", json.loads(status.text or "{}"))
+
+    queue = bus.subscribe()
+    get_task: Optional[asyncio.Task] = None
+    loop = asyncio.get_running_loop()
+    try:
+        await send_stats()
+        next_stats = loop.time() + 5
+        try:
+            last_id = int(request.headers.get("Last-Event-ID", 0))
+        except ValueError:
+            last_id = 0
+        for item in bus.since(last_id) if last_id else []:
+            await send(item["event"], item["data"], item["seq"])
+        while True:
+            if get_task is None:
+                get_task = asyncio.create_task(queue.get())
+            done, _ = await asyncio.wait({get_task}, timeout=max(0.0, next_stats - loop.time()))
+            if not done:
+                await send_stats()
+                next_stats = loop.time() + 5
+                continue
+            item = get_task.result()
+            get_task = None
+            await send(item["event"], item["data"], item["seq"])
+    except (ConnectionResetError, asyncio.CancelledError):
+        log.debug("SSE dashboard client disconnected")
+    finally:
+        if get_task is not None:
+            get_task.cancel()
+        bus.unsubscribe(queue)
+
+    return response
+
+
 async def cancel_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
     """Answer the cancel/remove download job request."""
     from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
@@ -2080,10 +2228,13 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
             normalized_service,
             service_instance,
             session_id=session_id,
+            owner_key=api_key,
+            creator_ip=request.remote if request else None,
         )
-        session.creator_ip = request.remote if request else None
-        session.owner_key = api_key
         session.cache_tag = session_cache_tag
+        # Echoed to every dashboard viewer, so cap what an arbitrary client can push into it.
+        if isinstance(data.get("client"), dict) and len(json.dumps(data["client"], default=str)) <= 4096:
+            session.client = data["client"]
         session.input_bridge = bridge
         session.auth_status = AuthStatus.AUTHENTICATING
 
@@ -2145,6 +2296,7 @@ async def session_titles_handler(session_id: str, request: Optional[web.Request]
             tid = str(t.id) if hasattr(t, "id") else str(id(t))
             session.title_map[tid] = t
             serialized_titles.append(stamp_service_flags(serialize_title(t), service_instance))
+        SessionStore.publish_update(session)
 
         return web.json_response(
             {
@@ -2202,6 +2354,7 @@ async def session_tracks_handler(
             title_tracks[str(track.id)] = track
             session.tracks[str(track.id)] = track
         session.tracks_by_title[str(title_id)] = title_tracks
+        SessionStore.publish_update(session)
 
         try:
             chapters = service_instance.get_chapters(title)

unshackle/core/api/routes.py

@@ -13,11 +13,18 @@ from unshackle.core import services as services_module
 from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception
 from unshackle.core.api.handlers import (
     CORS_HEADERS,
+    DASHBOARD_EVENTS_ROUTE,
+    DASHBOARD_PREFIX,
     JOB_EVENTS_ROUTE,
     cancel_download_job_handler,
     clear_cache_handler,
     clear_finished_download_jobs_handler,
     clear_temp_handler,
+    dashboard_events_handler,
+    dashboard_jobs_handler,
+    dashboard_logs_handler,
+    dashboard_sessions_handler,
+    dashboard_status_handler,
     delete_history_handler,
     download_handler,
     download_history_handler,
@@ -1930,6 +1937,132 @@ async def session_prompt_submit(request: web.Request) -> web.Response:
         )
 
 
+@api_handler
+async def dashboard_status(request: web.Request) -> web.Response:
+    """
+    Dashboard: server status.
+    ---
+    summary: Dashboard server status
+    description: >
+      Version, uptime, bind address, mode, request counters per API key, and counts of
+      services, remote sessions and jobs. Every /api/dashboard/ route needs
+      `serve.dashboard.key` in X-Secret-Key; a user key never passes, and the routes do not
+      exist without a dashboard key.
+    tags: [Dashboard]
+    responses:
+      '200':
+        description: Server status
+      '401':
+        description: Dashboard key missing or invalid
+    """
+    return await dashboard_status_handler(request)
+
+
+@api_handler
+async def dashboard_sessions(request: web.Request) -> web.Response:
+    """
+    Dashboard: active remote sessions.
+    ---
+    summary: Dashboard remote sessions
+    description: Every live remote session on the server, with the owner key masked.
+    tags: [Dashboard]
+    responses:
+      '200':
+        description: Session list
+      '401':
+        description: Dashboard key missing or invalid
+    """
+    return await dashboard_sessions_handler(request)
+
+
+@api_handler
+async def dashboard_jobs(request: web.Request) -> web.Response:
+    """
+    Dashboard: download jobs.
+    ---
+    summary: Dashboard download jobs
+    description: Every download job on the server regardless of owner. Empty in --remote-only mode.
+    tags: [Dashboard]
+    responses:
+      '200':
+        description: Job list
+      '401':
+        description: Dashboard key missing or invalid
+    """
+    return await dashboard_jobs_handler(request)
+
+
+@api_handler
+async def dashboard_logs(request: web.Request) -> web.Response:
+    """
+    Dashboard: recent log records.
+    ---
+    summary: Dashboard log ring buffer
+    description: The last 1000 log records. Poll with `since` set to the `seq` of the last record seen.
+    tags: [Dashboard]
+    parameters:
+      - name: since
+        in: query
+        required: false
+        schema:
+          type: integer
+        description: Return only records with seq greater than this value
+      - name: level
+        in: query
+        required: false
+        schema:
+          type: string
+        description: Minimum level (DEBUG, INFO, WARNING, ERROR)
+      - name: logger
+        in: query
+        required: false
+        schema:
+          type: string
+        description: Keep only this logger and its children, e.g. serve or aiohttp.access
+    responses:
+      '200':
+        description: Records plus the current seq
+      '401':
+        description: Dashboard key missing or invalid
+    """
+    return await dashboard_logs_handler(request)
+
+
+@api_handler
+async def dashboard_events(request: web.Request) -> web.StreamResponse:
+    """
+    Dashboard: server-wide event stream.
+    ---
+    summary: Dashboard SSE stream
+    description: >
+      Server-Sent Events: a `stats` event on connect and every 5 seconds (also the keep-alive),
+      plus `log`, `session` (action create/update/delete) and `job` events as they happen. A browser
+      EventSource can authenticate with the `secret_key` query parameter.
+    tags: [Dashboard]
+    parameters:
+      - name: secret_key
+        in: query
+        required: false
+        schema:
+          type: string
+        description: Dashboard key, for clients that cannot send headers
+      - name: since
+        in: query
+        required: false
+        schema:
+          type: integer
+        description: >
+          Poll mode. Return the events after this seq as one JSON burst
+          ({seq, stats, events}) instead of opening an event stream. Use seq 0 on first call.
+    responses:
+      '200':
+        description: text/event-stream, or JSON for a request that carries `since`
+      '401':
+        description: Dashboard key missing or invalid
+    """
+    return await dashboard_events_handler(request)
+
+
 # Single source of truth for all API routes. `remote` marks endpoints exposed in
 # --remote-only server mode; full mode registers every route. Both setup_routes and
 # setup_swagger derive from this table so the live routes and the docs cannot drift.
@@ -1967,21 +2100,30 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
     ("DELETE", "/api/session/{session_id}", session_delete, True),
 ]
 
+DASHBOARD_ROUTES: list[tuple[str, str, Handler, bool]] = [
+    ("GET", DASHBOARD_PREFIX + "status", dashboard_status, True),
+    ("GET", DASHBOARD_PREFIX + "sessions", dashboard_sessions, True),
+    ("GET", DASHBOARD_PREFIX + "jobs", dashboard_jobs, True),
+    ("GET", DASHBOARD_PREFIX + "logs", dashboard_logs, True),
+    ("GET", DASHBOARD_EVENTS_ROUTE, dashboard_events, True),
+]
 
-def setup_routes(app: web.Application, remote_only: bool = False) -> None:
-    """Setup API routes. When remote_only=True, only the remote session endpoints operate."""
+
+def setup_routes(app: web.Application, remote_only: bool = False, dashboard: bool = False) -> None:
+    """Setup API routes. When remote_only=True, only the remote session endpoints operate.
+    When dashboard=True, this also registers the /api/dashboard/ routes."""
     add: dict[str, Callable[..., Any]] = {
         "GET": app.router.add_get,
         "POST": app.router.add_post,
         "DELETE": app.router.add_delete,
     }
-    for method, path, handler, remote in ROUTES:
+    for method, path, handler, remote in ROUTES + (DASHBOARD_ROUTES if dashboard else []):
         if remote_only and not remote:
             continue
         add[method](path, handler)
 
 
-def setup_swagger(app: web.Application) -> None:
+def setup_swagger(app: web.Application, dashboard: bool = False) -> None:
     """Setup Swagger UI documentation."""
     swagger = SwaggerDocs(
         app,
@@ -1994,4 +2136,6 @@ def setup_swagger(app: web.Application) -> None:
     )
 
     route: dict[str, Callable[..., Any]] = {"GET": web.get, "POST": web.post, "DELETE": web.delete}
-    swagger.add_routes([route[method](path, handler) for method, path, handler, _ in ROUTES])
+    swagger.add_routes(
+        [route[method](path, handler) for method, path, handler, _ in ROUTES + (DASHBOARD_ROUTES if dashboard else [])]
+    )

unshackle/core/api/session_store.py

@@ -11,10 +11,12 @@ from __future__ import annotations
 import asyncio
 import logging
 import uuid
+from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
-from typing import Any, Dict, List, Optional
+from typing import Any, Deque, Dict, List, Optional
 
+from unshackle.core.api.events import bus
 from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT, AuthStatus, InputBridge
 from unshackle.core.api.sanitize import sanitize_log
 from unshackle.core.config import config
@@ -41,6 +43,8 @@ class SessionEntry:
     input_bridge: Optional[InputBridge] = None
     auth_status: AuthStatus = AuthStatus.AUTHENTICATED
     auth_error: Optional[str] = None
+    client: Dict[str, Any] = field(default_factory=dict)  # version/argv/platform the client reported
+    actions: Deque[Dict[str, Any]] = field(default_factory=lambda: deque(maxlen=500))
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     last_accessed: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
@@ -48,6 +52,39 @@ class SessionEntry:
         """Update last_accessed timestamp."""
         self.last_accessed = datetime.now(timezone.utc)
 
+    def summary(self) -> Dict[str, Any]:
+        """Dashboard view of the session: no service instance, no raw owner key."""
+        from unshackle.core.api.stats import mask_key
+
+        now = datetime.now(timezone.utc)
+        return {
+            "id": self.session_id,
+            "owner": mask_key(self.owner_key),
+            "creator_ip": self.creator_ip,
+            "service": self.service_tag,
+            "title_id": next(iter(self.title_map), None),
+            "title": str(next(iter(self.title_map.values()), "")) or None,
+            "titles": len(self.title_map),
+            "tracks": len(self.tracks),
+            "auth_status": self.auth_status.value,
+            "auth_error": self.auth_error,
+            "client": self.client,
+            "actions": list(self.actions),
+            "created_at": self.created_at.isoformat(),
+            "last_accessed": self.last_accessed.isoformat(),
+            "created_ts": self.created_at.timestamp(),
+            "last_accessed_ts": self.last_accessed.timestamp(),
+            "age_seconds": int((now - self.created_at).total_seconds()),
+            "idle_seconds": int((now - self.last_accessed).total_seconds()),
+        }
+
+
+def _publish(action: str, entry: SessionEntry, reason: Optional[str] = None) -> None:
+    data = {"action": action, **entry.summary()}
+    if reason:
+        data["reason"] = reason
+    bus.publish("session", data)
+
 
 class SessionStore:
     """Thread-safe remote session store with TTL-based expiration."""
@@ -72,21 +109,26 @@ class SessionStore:
         service_tag: str,
         service_instance: Any,
         session_id: Optional[str] = None,
+        owner_key: Optional[str] = None,
+        creator_ip: Optional[str] = None,
     ) -> SessionEntry:
         """Make a new remote session with an authenticated service instance."""
         async with self._lock:
             if len(self._sessions) >= self.max_sessions:
                 oldest_id = min(self._sessions, key=lambda k: self._sessions[k].last_accessed)
                 log.warning(f"Max sessions reached ({self.max_sessions}), evicting oldest: {oldest_id}")
-                del self._sessions[oldest_id]
+                _publish("delete", self._sessions.pop(oldest_id), "evicted")
 
             session_id = session_id or str(uuid.uuid4())
             entry = SessionEntry(
                 session_id=session_id,
                 service_tag=service_tag,
                 service_instance=service_instance,
+                owner_key=owner_key,
+                creator_ip=creator_ip,
             )
             self._sessions[session_id] = entry
+            _publish("create", entry)
             log.info(f"Created session {sanitize_log(session_id)} for service {sanitize_log(service_tag)}")
             return entry
 
@@ -101,7 +143,7 @@ class SessionStore:
                 elapsed = (datetime.now(timezone.utc) - entry.last_accessed).total_seconds()
                 if elapsed > self.ttl:
                     log.info(f"Session {sanitize_log(session_id)} expired (elapsed={elapsed:.0f}s, ttl={self.ttl}s)")
-                    del self._sessions[session_id]
+                    _publish("delete", self._sessions.pop(session_id), "expired")
                     return None
 
             entry.touch()
@@ -115,6 +157,7 @@ class SessionStore:
                 if entry.input_bridge:
                     entry.input_bridge.cancel()
                 self.cleanup_cache_dir(entry.cache_tag)
+                _publish("delete", entry, "closed")
                 log.info(f"Deleted session {sanitize_log(session_id)}")
                 return True
             return False
@@ -136,6 +179,7 @@ class SessionStore:
                 if entry.input_bridge:
                     entry.input_bridge.cancel()
                 self.cleanup_cache_dir(entry.cache_tag)
+                _publish("delete", entry, "expired")
             if expired:
                 log.info(f"Cleaned up {len(expired)} expired sessions")
             return len(expired)
@@ -194,6 +238,23 @@ class SessionStore:
             except OSError:
                 break
 
+    @staticmethod
+    def publish_update(entry: SessionEntry) -> None:
+        """Tell dashboard listeners that a session's summary changed (titles or tracks loaded, auth settled)."""
+        _publish("update", entry)
+
+    def record_action(self, session_id: str, action: Dict[str, Any]) -> None:
+        """Append one request to the session's action log and tell dashboard listeners."""
+        entry = self._sessions.get(session_id)
+        if entry is None:
+            return
+        entry.actions.append(action)
+        _publish("update", entry)
+
+    def list(self) -> List[SessionEntry]:
+        """Snapshot of every live session."""
+        return list(self._sessions.values())
+
     @property
     def session_count(self) -> int:
         """Number of active sessions."""

unshackle/core/api/stats.py

@@ -0,0 +1,120 @@
+"""Request counters and a log ring buffer for the developer dashboard."""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections import Counter, deque
+from dataclasses import dataclass, field
+from typing import Any, Deque, Dict, List, Optional
+
+from aiohttp import web
+
+from unshackle.core import __code_hash__, __version__
+from unshackle.core.api.events import bus
+from unshackle.core.config import config
+
+
+def mask_key(key: Optional[str]) -> str:
+    """A display label for an API key: its configured username, else a truncated prefix."""
+    if not key:
+        return "anonymous"
+    user = (config.serve or {}).get("users", {}).get(key)
+    if isinstance(user, dict) and user.get("username"):
+        return str(user["username"])
+    return key[:4] + "…"
+
+
+@dataclass
+class ServerStats:
+    started_at: float = field(default_factory=time.time)
+    host: str = ""
+    port: int = 0
+    mode: str = "full"
+    requests_total: int = 0
+    requests_rejected: int = 0
+    requests_by_key: Counter = field(default_factory=Counter)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "version": __version__,
+            "code_hash": __code_hash__,
+            "host": self.host,
+            "port": self.port,
+            "mode": self.mode,
+            "started_at": self.started_at,
+            "uptime_seconds": int(time.time() - self.started_at),
+            "requests_total": self.requests_total,
+            "requests_rejected": self.requests_rejected,
+            "requests_by_key": dict(self.requests_by_key),
+        }
+
+
+stats = ServerStats()
+
+
+@web.middleware
+async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResponse:
+    from unshackle.core.api.handlers import request_secret_key
+
+    stats.requests_total += 1
+    started = time.perf_counter()
+    response = await handler(request)
+    if response.status == 401:
+        stats.requests_rejected += 1
+    else:
+        stats.requests_by_key[mask_key(request_secret_key(request))] += 1
+    session_id = request.match_info.get("session_id")
+    if session_id and request.method != "OPTIONS":
+        from unshackle.core.api.session_store import get_session_store
+
+        get_session_store().record_action(
+            session_id,
+            {
+                "ts": time.time(),
+                "method": request.method,
+                "action": request.path.split(f"/{session_id}", 1)[-1].strip("/") or "info",
+                "query": dict(request.query),
+                "status": response.status,
+                "ms": round((time.perf_counter() - started) * 1000, 1),
+                "bytes_in": request.content_length or 0,
+                "bytes_out": getattr(response, "content_length", None) or 0,
+            },
+        )
+    return response
+
+
+class RingLogHandler(logging.Handler):
+    """Keep the last N log records in memory and publish each one on the event bus."""
+
+    def __init__(self, maxlen: int = 1000) -> None:
+        super().__init__()
+        self.records: Deque[Dict[str, Any]] = deque(maxlen=maxlen)
+        self.seq = 0
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.seq += 1
+        item = {
+            "seq": self.seq,
+            "ts": record.created,
+            "level": record.levelname,
+            "logger": record.name,
+            "msg": self.format(record),
+        }
+        self.records.append(item)
+        bus.publish("log", item)
+
+    def since(self, seq: int = 0, level: Optional[str] = None, logger: Optional[str] = None) -> List[Dict[str, Any]]:
+        min_level = logging.getLevelName(level.upper()) if level else 0
+        if not isinstance(min_level, int):
+            min_level = 0
+        return [
+            r
+            for r in self.records
+            if r["seq"] > seq
+            and logging.getLevelName(r["level"]) >= min_level
+            and (not logger or r["logger"] == logger or r["logger"].startswith(logger + "."))
+        ]
+
+
+ring = RingLogHandler()