diff --git a/app/database.py b/app/database.py
index e94c0bd..a73ef06 100644
--- a/app/database.py
+++ b/app/database.py
@@ -89,7 +89,13 @@ def decrypt_value(value: str) -> str:
     try:
         return _fernet.decrypt(value[len(_ENC_PREFIX) :].encode()).decode()
     except InvalidToken:
-        _logger.warning("Failed to decrypt config value — key may have changed")
+        _logger.warning(
+            "Failed to decrypt config value: the Fernet ENCRYPTION KEY (CASHPILOT_SECRET_KEY / "
+            "%s) does not match the key this value was encrypted with. This is NOT a bad "
+            "credential — re-enter the affected credentials, or restore the original encryption "
+            "key, to recover.",
+            _FERNET_KEY_FILE,
+        )
         return ""
 
 
diff --git a/app/deps.py b/app/deps.py
index d6bdc64..7fb4558 100644
--- a/app/deps.py
+++ b/app/deps.py
@@ -11,13 +11,14 @@
 from __future__ import annotations
 
 import ipaddress
+import os
 from typing import Any
 
 from fastapi import HTTPException, Request
 from fastapi.responses import RedirectResponse
 from fastapi.templating import Jinja2Templates
 
-from app import auth
+from app import auth, setup_token
 
 __all__ = [
     "templates",
@@ -26,8 +27,15 @@
     "_require_writer",
     "_require_owner",
     "_require_private_network",
+    "_require_first_run_access",
+    "client_ip",
 ]
 
+# Opt-in: set CASHPILOT_TRUSTED_PROXY=1 only when the app sits behind exactly one
+# reverse proxy you control. X-Forwarded-For is attacker-controlled, so we ignore
+# it unless the operator asserts a trusted proxy is stripping/appending it.
+_TRUST_PROXY = os.getenv("CASHPILOT_TRUSTED_PROXY", "").strip().lower() in ("1", "true", "yes", "on")
+
 templates = Jinja2Templates(directory="app/templates")
 
 
@@ -57,13 +65,48 @@ def _require_owner(request: Request) -> dict[str, Any]:
     return user
 
 
+def client_ip(request: Request) -> str | None:
+    """Best-effort real client IP.
+
+    Behind a trusted reverse proxy (opt-in via ``CASHPILOT_TRUSTED_PROXY``) the
+    real peer is the right-most ``X-Forwarded-For`` entry — the value appended by
+    the trusted proxy, which a client cannot forge by prepending its own. Without
+    that opt-in we never trust the header and use the direct peer.
+    """
+    if _TRUST_PROXY:
+        xff = request.headers.get("x-forwarded-for", "")
+        parts = [p.strip() for p in xff.split(",") if p.strip()]
+        if parts:
+            return parts[-1]
+    return request.client.host if request.client else None
+
+
 def _require_private_network(request: Request) -> None:
-    """Block requests from public IPs (for first-run setup)."""
-    if not request.client or not request.client.host:
+    """Block requests whose real client IP is public (first-run defense in depth)."""
+    ip_str = client_ip(request)
+    if not ip_str:
         return
     try:
-        client_ip = ipaddress.ip_address(request.client.host)
+        ip = ipaddress.ip_address(ip_str)
     except ValueError:
         return
-    if not (client_ip.is_loopback or client_ip.is_private):
+    if not (ip.is_loopback or ip.is_private):
         raise HTTPException(status_code=403, detail="First-run setup only allowed from private networks")
+
+
+def _require_first_run_access(request: Request, setup_token_value: str | None = None) -> None:
+    """Gate first-run owner creation: private network AND the one-time setup token.
+
+    The network check alone is spoofable behind a reverse proxy (the peer is then
+    the proxy), so the setup token — printed to the server logs, readable only
+    with host access — is the real gate. The token is accepted from the explicit
+    argument (a form field), the ``setup_token`` query param, or the
+    ``X-Setup-Token`` header.
+    """
+    _require_private_network(request)
+    token = setup_token_value or request.query_params.get("setup_token") or request.headers.get("x-setup-token")
+    if not setup_token.verify(token):
+        raise HTTPException(
+            status_code=403,
+            detail="First-run setup requires the setup token printed in the server logs",
+        )
diff --git a/app/exchange_rates.py b/app/exchange_rates.py
index 8e879a6..765db3b 100644
--- a/app/exchange_rates.py
+++ b/app/exchange_rates.py
@@ -108,6 +108,7 @@ def get_all() -> dict[str, Any]:
         "fiat": dict(_fiat_rates),
         "crypto_usd": dict(_crypto_usd),
         "last_updated": _last_fetch,
+        "stale": rates_stale(),
     }
 
 
diff --git a/app/main.py b/app/main.py
index b1c361f..cdeb292 100644
--- a/app/main.py
+++ b/app/main.py
@@ -31,7 +31,7 @@
 from pydantic import BaseModel
 from starlette.middleware.base import BaseHTTPMiddleware
 
-from app import auth, catalog, compose_generator, database, exchange_rates, fleet_key, metrics
+from app import auth, catalog, compose_generator, database, exchange_rates, fleet_key, metrics, setup_token
 
 logging.basicConfig(
     level=logging.INFO,
@@ -44,6 +44,30 @@
 # In-memory store for the latest collector alerts (errors from last run)
 _collector_alerts: list[dict[str, str]] = []
 _collection_lock = asyncio.Lock()
+_collection_semaphore = asyncio.Semaphore(8)
+
+# Fire-and-forget background tasks (e.g. triggered collection runs). Keeping a
+# reference prevents the task from being garbage-collected mid-run and lets us
+# retrieve/log any exception it raised (bare `asyncio.create_task(...)` drops
+# the reference and silently swallows exceptions).
+_background_tasks: set[asyncio.Task] = set()
+
+
+def _spawn(coro) -> asyncio.Task:
+    """Fire-and-forget a coroutine while keeping a reference and logging errors."""
+    task = asyncio.create_task(coro)
+    _background_tasks.add(task)
+
+    def _on_done(t: asyncio.Task) -> None:
+        _background_tasks.discard(t)
+        if not t.cancelled():
+            exc = t.exception()
+            if exc is not None:
+                logger.error("Background task failed: %s", exc, exc_info=exc)
+
+    task.add_done_callback(_on_done)
+    return task
+
 
 # Login rate limiting
 _login_attempts: dict[str, list[float]] = defaultdict(list)
@@ -51,16 +75,16 @@
 _LOGIN_WINDOW_SECONDS = 300
 
 
-def _check_login_rate(client_ip: str) -> None:
+def _check_login_rate(ip: str) -> None:
     now = monotonic()
-    attempts = _login_attempts[client_ip]
-    _login_attempts[client_ip] = [t for t in attempts if now - t < _LOGIN_WINDOW_SECONDS]
-    if len(_login_attempts[client_ip]) >= _LOGIN_MAX_ATTEMPTS:
+    attempts = _login_attempts[ip]
+    _login_attempts[ip] = [t for t in attempts if now - t < _LOGIN_WINDOW_SECONDS]
+    if len(_login_attempts[ip]) >= _LOGIN_MAX_ATTEMPTS:
         raise HTTPException(status_code=429, detail="Too many login attempts. Try again in a few minutes.")
 
 
-def _record_failed_login(client_ip: str) -> None:
-    _login_attempts[client_ip].append(monotonic())
+def _record_failed_login(ip: str) -> None:
+    _login_attempts[ip].append(monotonic())
 
 
 def _safe_json(raw: str, fallback: Any = None) -> Any:
@@ -179,6 +203,12 @@ async def _run_health_check() -> None:
         logger.warning("Health check skipped: %s", exc)
 
 
+async def _collect_bounded(collector) -> Any:
+    """Run a single collector's `collect()` under the shared concurrency limit."""
+    async with _collection_semaphore:
+        return await collector.collect()
+
+
 async def _run_collection() -> None:
     """Collect earnings from all deployed services that have collectors."""
     global _collector_alerts
@@ -186,9 +216,10 @@ async def _run_collection() -> None:
         logger.info("Collection already in progress, skipping")
         return
     async with _collection_lock:
-        start_time = metrics.record_collection_start()
         success = True
+        start_time = 0.0
         try:
+            start_time = metrics.record_collection_start()
             deployments = await database.get_deployments()
             config = await database.get_config() or {}
             if not isinstance(config, dict):
@@ -197,7 +228,7 @@ async def _run_collection() -> None:
 
             collectors = make_collectors(deployments, config)
             await _close_stale()
-            results = await asyncio.gather(*(c.collect() for c in collectors), return_exceptions=True)
+            results = await asyncio.gather(*(_collect_bounded(c) for c in collectors), return_exceptions=True)
             alerts: list[dict[str, str]] = []
             platforms_ok = 0
             for result in results:
@@ -242,6 +273,15 @@ async def _run_data_retention() -> None:
         logger.warning("Data retention error: %s", exc)
 
 
+async def _run_vacuum() -> None:
+    """Reclaim disk left by retention deletes (SQLite does not auto-shrink)."""
+    try:
+        await database.vacuum_database()
+        logger.info("Database VACUUM complete")
+    except Exception as exc:
+        logger.warning("Database VACUUM error: %s", exc)
+
+
 async def _check_stale_workers() -> None:
     """Mark workers as offline if stale, and purge workers offline > 1 hour."""
     try:
@@ -281,6 +321,21 @@ async def lifespan(app: FastAPI):
         _changed = _u.get("password_changed_at") or 0.0
         if _changed:
             auth.set_user_pwd_epoch(_u["id"], _changed)
+    # First-run setup token: while no users exist, require a one-time token
+    # (printed below) for /register so a proxy-exposed instance cannot be seized
+    # by the first public visitor. Persisted in config so it survives restarts;
+    # cleared once the owner account is created.
+    if not await database.has_any_users():
+        _tok = await database.get_config("_setup_token")
+        if not _tok:
+            _tok = setup_token.generate()
+            await database.set_config("_setup_token", _tok)
+        setup_token.set_active(_tok)
+        logger.warning(
+            "FIRST-RUN SETUP: no account exists yet. Create the owner account at "
+            "/register?setup_token=%s — this token is required and is shown only here.",
+            _tok,
+        )
     catalog.load_services()
     catalog.register_sighup()
 
@@ -324,6 +379,15 @@ def _on_job_event(event):
         coalesce=True,
         misfire_grace_time=300,
     )
+    scheduler.add_job(
+        _run_vacuum,
+        "interval",
+        weeks=1,
+        id="db_vacuum",
+        max_instances=1,
+        coalesce=True,
+        misfire_grace_time=300,
+    )
     scheduler.add_job(
         exchange_rates.refresh,
         "interval",
@@ -335,7 +399,7 @@ def _on_job_event(event):
     )
     scheduler.start()
     await exchange_rates.refresh()
-    asyncio.create_task(_run_collection())
+    _spawn(_run_collection())
     logger.info("CashPilot UI started (container ops via workers)")
 
     yield
@@ -393,9 +457,11 @@ async def dispatch(self, request, call_next):
 from app.deps import (  # noqa: E402
     _login_redirect,  # noqa: F401  (re-exported for app.main.* test/router surface)
     _require_auth_api,
+    _require_first_run_access,  # noqa: F401  (re-exported for app.main.* router surface)
     _require_owner,
     _require_private_network,  # noqa: F401  (re-exported for app.main.* router surface)
     _require_writer,
+    client_ip,  # noqa: F401  (re-exported for app.main.* router surface)
     templates,  # noqa: F401  (re-exported for app.main.* router/test surface)
 )
 
@@ -528,6 +594,10 @@ async def api_services_deployed(request: Request) -> list[dict[str, Any]]:
             "health_score": health.get("score"),
             "uptime_pct": health.get("uptime_pct"),
             "restarts_7d": health.get("restarts", 0),
+            "crashes_7d": health.get("crashes", 0),
+            # "unstable" flags a service that has crashed repeatedly in the health window
+            # so the dashboard can surface it at a glance (not just via the score number).
+            "unstable": health.get("crashes", 0) >= 3,
             "instances": len(agg["instances"]),
             "instance_details": instance_details,
             "collector_disconnected": slug in alert_slugs,
@@ -567,6 +637,8 @@ async def api_services_deployed(request: Request) -> list[dict[str, Any]]:
             "health_score": None,
             "uptime_pct": None,
             "restarts_7d": 0,
+            "crashes_7d": 0,
+            "unstable": False,
             "instances": 0,
             "instance_details": [],
             "collector_disconnected": slug in alert_slugs,
@@ -750,7 +822,7 @@ async def api_deploy(request: Request, slug: str, body: DeployRequest, worker_id
     await database.save_deployment(slug=slug, container_id=container_id)
     await database.record_health_event(slug, "start", f"deployed to worker {worker_id}")
     metrics.record_container_lifecycle("deploy", slug)
-    asyncio.create_task(_run_collection())
+    _spawn(_run_collection())
     return {"status": "deployed", "container_id": container_id}
 
 
@@ -759,6 +831,7 @@ async def api_stop(request: Request, slug: str, worker_id: int | None = None) ->
     _require_writer(request)
     worker_id = await _resolve_worker_id(worker_id)
     result = await _proxy_worker_command(worker_id, "stop", slug)
+    await database.record_health_event(slug, "stop")
     metrics.record_container_lifecycle("stop", slug)
     return result
 
@@ -768,6 +841,7 @@ async def api_restart(request: Request, slug: str, worker_id: int | None = None)
     _require_writer(request)
     worker_id = await _resolve_worker_id(worker_id)
     result = await _proxy_worker_command(worker_id, "restart", slug)
+    await database.record_health_event(slug, "restart")
     metrics.record_container_lifecycle("restart", slug)
     return result
 
@@ -781,6 +855,7 @@ async def api_remove(
     params = {"delete_volumes": "true"} if delete_volumes else None
     result = await _proxy_worker_command(worker_id, "remove", slug, params=params)
     await database.remove_deployment(slug)
+    await database.record_health_event(slug, "remove")
     metrics.record_container_lifecycle("remove", slug)
     return result
 
@@ -872,6 +947,9 @@ def _validate_worker_url(raw_url: str) -> str:
 
     Resolves hostnames and validates the resolved IP(s) so a DNS name that
     points at a metadata/loopback address is rejected (DNS-rebinding guard).
+    Synchronous — its only blocking op is DNS resolution; event-loop callers
+    MUST invoke it via ``asyncio.to_thread`` (see _get_verified_worker_url) so a
+    slow/hanging resolver never blocks the whole UI.
     """
     parsed = urlparse(raw_url)
     if parsed.scheme not in _ALLOWED_WORKER_SCHEMES:
@@ -918,7 +996,7 @@ def _validate_worker_url(raw_url: str) -> str:
     return raw_url.rstrip("/")
 
 
-def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str]]:
+async def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str]]:
     """Validate a worker record and return (url, headers)."""
     if not worker:
         raise HTTPException(status_code=404, detail="Worker not found")
@@ -926,7 +1004,7 @@ def _get_verified_worker_url(worker: dict[str, Any]) -> tuple[str, dict[str, str
         raise HTTPException(status_code=503, detail="Worker is offline")
     if not worker["url"]:
         raise HTTPException(status_code=503, detail="Worker URL not known")
-    url = _validate_worker_url(worker["url"])
+    url = await asyncio.to_thread(_validate_worker_url, worker["url"])
     headers: dict[str, str] = {}
     if FLEET_API_KEY:
         headers["Authorization"] = f"Bearer {FLEET_API_KEY}"
@@ -938,7 +1016,7 @@ async def _proxy_worker_command(
 ) -> dict[str, Any]:
     """Forward a container command (restart/stop/start/remove) to a worker."""
     worker = await database.get_worker(worker_id)
-    url, headers = _get_verified_worker_url(worker)
+    url, headers = await _get_verified_worker_url(worker)
 
     try:
         async with httpx.AsyncClient(timeout=30) as client:
@@ -958,7 +1036,7 @@ async def _proxy_worker_command(
 async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any]) -> dict[str, Any]:
     """Forward a deploy command with full spec to a worker."""
     worker = await database.get_worker(worker_id)
-    url, headers = _get_verified_worker_url(worker)
+    url, headers = await _get_verified_worker_url(worker)
 
     try:
         async with httpx.AsyncClient(timeout=60) as client:
@@ -975,7 +1053,7 @@ async def _proxy_worker_deploy(worker_id: int, slug: str, spec: dict[str, Any])
 async def _proxy_worker_logs(worker_id: int, slug: str, lines: int = 50) -> dict[str, str]:
     """Forward a logs request to a worker."""
     worker = await database.get_worker(worker_id)
-    url, headers = _get_verified_worker_url(worker)
+    url, headers = await _get_verified_worker_url(worker)
 
     try:
         async with httpx.AsyncClient(timeout=30) as client:
@@ -1004,6 +1082,7 @@ async def api_service_restart(request: Request, slug: str, worker_id: int | None
     worker_id = await _resolve_worker_id(worker_id)
     result = await _proxy_worker_command(worker_id, "restart", slug)
     await database.record_health_event(slug, "restart")
+    metrics.record_container_lifecycle("restart", slug)
     return result
 
 
@@ -1013,6 +1092,7 @@ async def api_service_stop(request: Request, slug: str, worker_id: int | None =
     worker_id = await _resolve_worker_id(worker_id)
     result = await _proxy_worker_command(worker_id, "stop", slug)
     await database.record_health_event(slug, "stop")
+    metrics.record_container_lifecycle("stop", slug)
     return result
 
 
@@ -1022,6 +1102,7 @@ async def api_service_start(request: Request, slug: str, worker_id: int | None =
     worker_id = await _resolve_worker_id(worker_id)
     result = await _proxy_worker_command(worker_id, "start", slug)
     await database.record_health_event(slug, "start")
+    metrics.record_container_lifecycle("start", slug)
     return result
 
 
@@ -1043,6 +1124,8 @@ async def api_service_remove(
     params = {"delete_volumes": "true"} if delete_volumes else None
     result = await _proxy_worker_command(worker_id, "remove", slug, params=params)
     await database.remove_deployment(slug)
+    await database.record_health_event(slug, "remove")
+    metrics.record_container_lifecycle("remove", slug)
     return result
 
 
@@ -1235,7 +1318,7 @@ async def api_health_scores(request: Request, days: int = 7) -> list[dict[str, A
 @app.post("/api/collect")
 async def api_collect(request: Request) -> dict[str, str]:
     _require_writer(request)
-    asyncio.create_task(_run_collection())
+    _spawn(_run_collection())
     return {"status": "collection_started"}
 
 
@@ -1325,7 +1408,7 @@ async def api_set_preferences(request: Request, body: PreferencesUpdate) -> dict
     )
     # If setup is completed, trigger an immediate collection
     if body.setup_completed:
-        asyncio.create_task(_run_collection())
+        _spawn(_run_collection())
     return {"status": "saved"}
 
 
@@ -1403,17 +1486,11 @@ async def api_collectors_meta(request: Request) -> list[dict[str, Any]]:
     _require_owner(request)
     from app.collectors import _COLLECTOR_ARGS, COLLECTOR_MAP
 
-    secret_args = {
-        "password",
-        "token",
-        "auth_token",
-        "access_token",
-        "api_key",
-        "session_cookie",
-        "auth_cookie",
-        "oauth_token",
-        "brd_sess_id",
-    }
+    # Single-sourced from database.SECRET_CONFIG_KEYS so this endpoint can never
+    # disagree with the encryption-at-rest / masking logic about which config
+    # keys are secret (a hand-maintained duplicate here previously missed
+    # `remember_web` and `xsrf_token`, unmasking them).
+    secret_args = database.SECRET_CONFIG_KEYS
     # Per-service hints on how to obtain the credentials
     hints: dict[str, str] = {
         "bitping": "Use your Bitping account email and password (same as <a href='https://nodes.bitping.com' target='_blank'>nodes.bitping.com</a>).",
@@ -1705,10 +1782,15 @@ class WorkerCommand(BaseModel):
 @app.post("/api/workers/{worker_id}/command")
 async def api_worker_command(request: Request, worker_id: int, body: WorkerCommand) -> dict[str, Any]:
     """Send a command to a worker by proxying to its REST API."""
-    _require_writer(request)
+    # Deploy is owner-gated everywhere else (see /api/deploy/{slug}); a writer
+    # must not be able to bypass that gate by sending command="deploy" here.
+    if body.command == "deploy":
+        _require_owner(request)
+    else:
+        _require_writer(request)
 
     worker = await database.get_worker(worker_id)
-    url, headers = _get_verified_worker_url(worker)
+    url, headers = await _get_verified_worker_url(worker)
 
     try:
         async with httpx.AsyncClient(timeout=30) as client:
diff --git a/app/metrics.py b/app/metrics.py
index 9046bc1..cbe52d4 100644
--- a/app/metrics.py
+++ b/app/metrics.py
@@ -17,12 +17,15 @@
 from __future__ import annotations
 
 import contextlib
+import logging
 import os
 import re
 import time
 
 from fastapi import FastAPI
 
+logger = logging.getLogger(__name__)
+
 METRICS_ENABLED = os.getenv("CASHPILOT_METRICS_ENABLED", "").lower() in ("1", "true", "yes")
 
 _registry = None
@@ -231,6 +234,15 @@ def setup(app: FastAPI) -> None:
 
     _init_metrics()
 
+    # /metrics is served UNAUTHENTICATED (Prometheus convention). It exposes earnings and
+    # health totals, so warn on startup: keep it behind a reverse proxy / auth and never
+    # expose the app's port directly to an untrusted network.
+    logger.warning(
+        "Prometheus /metrics is ENABLED and served UNAUTHENTICATED — it exposes your "
+        "earnings and health data to anyone who can reach this port. Keep it behind a "
+        "reverse proxy with auth; do not expose the port directly to an untrusted network."
+    )
+
     from fastapi import Response
     from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
     from starlette.middleware.base import BaseHTTPMiddleware
diff --git a/app/routers/auth.py b/app/routers/auth.py
index aa8ff62..fe53613 100644
--- a/app/routers/auth.py
+++ b/app/routers/auth.py
@@ -45,7 +45,7 @@ async def do_login(
     username: str = Form(...),
     password: str = Form(...),
 ):
-    client_ip = request.client.host if request.client else "unknown"
+    client_ip = main.client_ip(request) or "unknown"
     try:
         main._check_login_rate(client_ip)
     except HTTPException:
@@ -86,7 +86,7 @@ async def page_register(request: Request, error: str = ""):
         if not user or user.get("r") != "owner":
             return RedirectResponse("/login", status_code=303)
     if is_first:
-        main._require_private_network(request)
+        main._require_first_run_access(request)
 
     return main.templates.TemplateResponse(
         request,
@@ -99,6 +99,8 @@ async def page_register(request: Request, error: str = ""):
             "button_text": "Create Account",
             "error": error,
             "is_first": is_first,
+            # Carry the setup token through the form so the POST passes the same gate.
+            "setup_token": request.query_params.get("setup_token", "") if is_first else "",
         },
     )
 
@@ -109,6 +111,7 @@ async def do_register(
     username: str = Form(...),
     password: str = Form(...),
     password_confirm: str = Form(...),
+    setup_token: str = Form(""),
 ):
     is_first = not await main.database.has_any_users()
 
@@ -119,7 +122,7 @@ async def do_register(
             raise HTTPException(status_code=403, detail="Only owners can add users")
 
     if is_first:
-        main._require_private_network(request)
+        main._require_first_run_access(request, setup_token)
 
     if not re.match(r"^[a-zA-Z0-9_-]{3,32}$", username):
         return main.templates.TemplateResponse(
@@ -133,6 +136,7 @@ async def do_register(
                 "button_text": "Create Account",
                 "error": "Username must be 3-32 alphanumeric characters (a-z, 0-9, _ -)",
                 "is_first": is_first,
+                "setup_token": setup_token if is_first else "",
             },
             status_code=400,
         )
@@ -149,6 +153,7 @@ async def do_register(
                 "button_text": "Create Account",
                 "error": "Passwords do not match",
                 "is_first": is_first,
+                "setup_token": setup_token if is_first else "",
             },
             status_code=400,
         )
@@ -165,6 +170,7 @@ async def do_register(
                 "button_text": "Create Account",
                 "error": "Password must be at least 10 characters",
                 "is_first": is_first,
+                "setup_token": setup_token if is_first else "",
             },
             status_code=400,
         )
@@ -182,6 +188,7 @@ async def do_register(
                 "button_text": "Create Account",
                 "error": "Username already taken",
                 "is_first": is_first,
+                "setup_token": setup_token if is_first else "",
             },
             status_code=400,
         )
@@ -191,6 +198,11 @@ async def do_register(
     hashed = main.auth.hash_password(password)
     user_id = await main.database.create_user(username, hashed, role)
 
+    if is_first:
+        # Owner now exists — retire the one-time setup token permanently.
+        await main.database.delete_config_keys(["_setup_token"])
+        main.setup_token.clear()
+
     token = main.auth.create_session_token(user_id, username, role)
     dest = "/setup" if is_first else "/"
     response = RedirectResponse(dest, status_code=303)
diff --git a/app/routers/users.py b/app/routers/users.py
index 2b2bed9..26f21f2 100644
--- a/app/routers/users.py
+++ b/app/routers/users.py
@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import time
 from typing import Any
 
 from fastapi import APIRouter, HTTPException, Request
@@ -43,6 +44,10 @@ async def api_update_user_role(request: Request, user_id: int, body: UserRoleUpd
         if owner_count <= 1:
             raise HTTPException(status_code=400, detail="Cannot remove the last owner")
     await main.database.update_user_role(user_id, body.role)
+    # Invalidate any outstanding session tokens for this user — they carry the
+    # old role and must not be trusted until re-issued (same mechanism used
+    # for password changes).
+    main.auth.set_user_pwd_epoch(user_id, time.time())
     return {"status": "updated"}
 
 
@@ -55,4 +60,8 @@ async def api_delete_user(request: Request, user_id: int) -> dict[str, str]:
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
     await main.database.delete_user(user_id)
+    # Invalidate any outstanding session tokens for the now-deleted account.
+    # The epoch cache is in-memory keyed by uid, so it survives the row being
+    # gone from the DB — decode_session_token rejects the token on iat alone.
+    main.auth.set_user_pwd_epoch(user_id, time.time())
     return {"status": "deleted"}
diff --git a/app/setup_token.py b/app/setup_token.py
new file mode 100644
index 0000000..cf95235
--- /dev/null
+++ b/app/setup_token.py
@@ -0,0 +1,62 @@
+"""First-run setup-token gate.
+
+The very first account created on a fresh install becomes the ``owner``. The
+private-network check in :mod:`app.deps` is not enough on its own: behind a
+reverse proxy ``request.client`` is the *proxy* (a loopback/private address), so
+the network check always passes and the first person to reach ``/register`` from
+the public internet could seize the owner account.
+
+This module adds a proxy-independent second factor: a one-time token generated at
+startup while no users exist, printed to the container logs. Only someone who can
+read the server logs (i.e. already has host access) can complete first-run setup.
+Once the owner account is created the token is cleared and never required again
+(further users are added by an authenticated owner).
+
+The active token is held in a module global so the synchronous request guard can
+check it without a DB round-trip; it is also persisted in the ``config`` table so
+it survives restarts until consumed.
+"""
+
+from __future__ import annotations
+
+import hmac
+import secrets
+
+# The token currently required for first-run registration, or ``None`` when no
+# first-run gate is active (owner already exists, or not yet initialised).
+_active: str | None = None
+
+
+def generate() -> str:
+    """Return a fresh, URL-safe setup token."""
+    return secrets.token_urlsafe(24)
+
+
+def set_active(token: str | None) -> None:
+    """Install (or clear, with ``None``) the token required for first-run setup."""
+    global _active
+    _active = token or None
+
+
+def clear() -> None:
+    """Drop the first-run gate — called once the owner account exists."""
+    set_active(None)
+
+
+def active() -> str | None:
+    """Return the currently required setup token, or ``None`` if none is active."""
+    return _active
+
+
+def verify(provided: str | None) -> bool:
+    """Check a caller-supplied token against the active one.
+
+    Returns ``True`` when no token is active (nothing to enforce) or when the
+    supplied value matches in constant time; ``False`` otherwise.
+    """
+    current = _active
+    if current is None:
+        return True
+    if not provided:
+        return False
+    return hmac.compare_digest(current, provided)
diff --git a/app/static/js/app.js b/app/static/js/app.js
index 4458645..fdfe957 100644
--- a/app/static/js/app.js
+++ b/app/static/js/app.js
@@ -75,13 +75,19 @@ const CP = (() => {
   function sanitizeHint(html) {
     const el = document.createElement('div');
     el.innerHTML = html;
+    const allowedSchemes = ['http:', 'https:', 'mailto:'];
     el.querySelectorAll('*').forEach(node => {
       if (!['A', 'B', 'CODE'].includes(node.tagName)) {
         node.replaceWith(document.createTextNode(node.textContent));
         return;
       }
       for (const attr of [...node.attributes]) {
-        if (node.tagName === 'A' && (attr.name === 'href' || attr.name === 'target')) continue;
+        if (node.tagName === 'A' && attr.name === 'href') {
+          const scheme = (attr.value || '').trim().toLowerCase();
+          if (!allowedSchemes.some(s => scheme.startsWith(s))) node.removeAttribute('href');
+          continue;
+        }
+        if (node.tagName === 'A' && attr.name === 'target') continue;
         node.removeAttribute(attr.name);
       }
     });
@@ -178,7 +184,9 @@ const CP = (() => {
   async function loadExchangeRates() {
     try {
       _exchangeRates = await api('/api/exchange-rates');
-    } catch { /* keep defaults */ }
+    } catch (err) {
+      console.warn('Could not refresh exchange rates, keeping previous values:', err);
+    }
   }
 
   async function loadTopbarEarnings() {
@@ -186,8 +194,9 @@ const CP = (() => {
       await loadExchangeRates();
       const data = await api('/api/earnings/summary');
       setTextContent('topbar-total', formatCurrency(data.total || 0));
-    } catch {
-      setTextContent('topbar-total', formatCurrency(0));
+    } catch (err) {
+      // Keep whatever was already shown rather than fabricating $0.
+      console.warn('Could not refresh topbar earnings, keeping previous value:', err);
     }
   }
 
@@ -239,12 +248,9 @@ const CP = (() => {
         setChangeIndicator('month-change', data.month_change);
       }
     } catch (err) {
-      // API not yet implemented — fill with placeholder
-      setTextContent('total-earnings', formatCurrency(0));
-      setTextContent('today-earnings', formatCurrency(0));
-      setTextContent('month-earnings', formatCurrency(0));
-      setTextContent('active-services', '0');
-      setTextContent('topbar-total', formatCurrency(0));
+      // Keep whatever was already displayed — a transient fetch failure
+      // must not look like the dashboard's earnings dropped to zero.
+      toast(err.message || 'Could not refresh dashboard', 'error');
     }
   }
 
@@ -401,9 +407,15 @@ const CP = (() => {
         }
       });
     } catch (err) {
-      // Keep existing table if we had one; only show spinner on truly empty state
+      // Keep the existing table on a refresh failure. Only replace the
+      // perpetual "Loading..." spinner with a clear error + retry affordance
+      // when there was nothing on screen yet.
       if (!container.querySelector('.breakdown-table')) {
-        container.innerHTML = `<div style="display:flex; align-items:center; justify-content:center; gap:8px; padding:24px 0; color:var(--text-muted);"><div class="spinner"></div> Loading services...</div>`;
+        container.innerHTML = `
+          <div style="display:flex; flex-direction:column; align-items:center; justify-content:center; gap:8px; padding:24px 0; color:var(--text-muted); text-align:center;">
+            <span>Couldn't load services${err && err.message ? `: ${escapeHtml(err.message)}` : ''}.</span>
+            <button class="btn btn-ghost btn-sm" onclick="CP.loadServicesTable()">Retry</button>
+          </div>`;
       }
     }
   }
@@ -434,8 +446,16 @@ const CP = (() => {
     let healthBadge = '<span style="color:var(--text-muted);">--</span>';
     if (!isExternal && svc.health_score !== null && svc.health_score !== undefined) {
       const score = svc.health_score;
-      const hClass = score >= 80 ? 'badge-running' : score >= 50 ? 'badge-error' : 'badge-stopped';
-      healthBadge = `<span class="badge ${hClass}" title="Health ${score}/100">${score}</span>`;
+      const crashes = svc.crashes_7d || 0;
+      const restarts = svc.restarts_7d || 0;
+      const unstable = svc.unstable === true;
+      // Unstable (repeated crashes in the 7-day window) takes the worst tone + an explicit
+      // label so a crash-looping service is legible at a glance, not just via a low number.
+      const hClass = (unstable || score < 50) ? 'badge-stopped' : score >= 80 ? 'badge-running' : 'badge-error';
+      const uptime = (svc.uptime_pct !== null && svc.uptime_pct !== undefined) ? `${svc.uptime_pct}% uptime · ` : '';
+      const title = `Health ${score}/100 · ${uptime}${restarts} restarts · ${crashes} crashes (7d)`;
+      const label = unstable ? `unstable · ${crashes} crashes` : String(score);
+      healthBadge = `<span class="badge ${hClass}" title="${title}">${label}</span>`;
     }
 
     // Balance + delta from breakdown
@@ -501,7 +521,7 @@ const CP = (() => {
       ? (eligible ? 'Cash out earnings' : 'View payout details')
       : 'No payout info available';
     const claimDisabled = !co.dashboard_url;
-    const claimBtn = `<button class="btn btn-icon ${eligible ? 'btn-success' : ''}" onclick="${claimDisabled ? '' : `CP.openClaimModal('${svc.slug}')`}" title="${claimTitle}"${claimDisabled ? ' disabled' : ''}>
+    const claimBtn = `<button class="btn btn-icon ${eligible ? 'btn-success' : ''}" onclick="${claimDisabled ? '' : `CP.openClaimModal('${escapeHtml(svc.slug)}')`}" title="${claimTitle}"${claimDisabled ? ' disabled' : ''}>
            <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><line x1="12" y1="1" x2="12" y2="23"/><path d="M17 5H9.5a3.5 3.5 0 000 7h5a3.5 3.5 0 010 7H6"/></svg>
          </button>`;
 
@@ -520,7 +540,7 @@ const CP = (() => {
     // For single instance: show action buttons directly
     let actionBtns;
     if (isMulti) {
-      const chevron = `<button class="btn btn-icon expand-toggle" onclick="event.stopPropagation(); CP.toggleInstances('${svc.slug}')" title="Expand instances">
+      const chevron = `<button class="btn btn-icon expand-toggle" onclick="event.stopPropagation(); CP.toggleInstances('${escapeHtml(svc.slug)}')" title="Expand instances">
         <svg class="expand-chevron" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><polyline points="6 9 12 15 18 9"/></svg>
       </button>`;
       actionBtns = `<div class="action-btns">${claimBtn}${settingsBtn}${chevron}</div>`;
@@ -536,13 +556,13 @@ const CP = (() => {
           ${claimBtn}
           ${settingsBtn}
           ${_canWrite ? `
-          <button class="btn btn-icon" onclick="CP.restartService('${svc.slug}${wParam})" title="Restart"${disabledAttr}>
+          <button class="btn btn-icon" onclick="CP.restartService('${escapeHtml(svc.slug)}${wParam})" title="Restart"${disabledAttr}>
             <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><polyline points="23 4 23 10 17 10"/><path d="M20.49 15a9 9 0 11-2.12-9.36L23 10"/></svg>
           </button>
-          <button class="btn btn-icon" onclick="CP.stopService('${svc.slug}${wParam})" title="Stop"${disabledAttr}>
+          <button class="btn btn-icon" onclick="CP.stopService('${escapeHtml(svc.slug)}${wParam})" title="Stop"${disabledAttr}>
             <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><rect x="6" y="6" width="12" height="12" rx="1"/></svg>
           </button>
-          <button class="btn btn-icon" onclick="CP.viewLogs('${svc.slug}${wParam})" title="Logs"${disabledAttr}>
+          <button class="btn btn-icon" onclick="CP.viewLogs('${escapeHtml(svc.slug)}${wParam})" title="Logs"${disabledAttr}>
             <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><path d="M14 2H6a2 2 0 00-2 2v16a2 2 0 002 2h12a2 2 0 002-2V8z"/><polyline points="14 2 14 8 20 8"/><line x1="16" y1="13" x2="8" y2="13"/><line x1="16" y1="17" x2="8" y2="17"/></svg>
           </button>` : ''}
         </div>`;
@@ -590,13 +610,13 @@ const CP = (() => {
           <td style="text-align:center; white-space:nowrap;">
             <div class="action-btns">
               ${_canWrite ? `
-              <button class="btn btn-icon" onclick="CP.restartService('${svc.slug}${wParam})" title="Restart on ${nodeLabel}"${disabledAttr}>
+              <button class="btn btn-icon" onclick="CP.restartService('${escapeHtml(svc.slug)}${wParam})" title="Restart on ${nodeLabel}"${disabledAttr}>
                 <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><polyline points="23 4 23 10 17 10"/><path d="M20.49 15a9 9 0 11-2.12-9.36L23 10"/></svg>
               </button>
-              <button class="btn btn-icon" onclick="CP.stopService('${svc.slug}${wParam})" title="Stop on ${nodeLabel}"${disabledAttr}>
+              <button class="btn btn-icon" onclick="CP.stopService('${escapeHtml(svc.slug)}${wParam})" title="Stop on ${nodeLabel}"${disabledAttr}>
                 <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><rect x="6" y="6" width="12" height="12" rx="1"/></svg>
               </button>
-              <button class="btn btn-icon" onclick="CP.viewLogs('${svc.slug}${wParam})" title="Logs on ${nodeLabel}"${disabledAttr}>
+              <button class="btn btn-icon" onclick="CP.viewLogs('${escapeHtml(svc.slug)}${wParam})" title="Logs on ${nodeLabel}"${disabledAttr}>
                 <svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"><path d="M14 2H6a2 2 0 00-2 2v16a2 2 0 002 2h12a2 2 0 002-2V8z"/><polyline points="14 2 14 8 20 8"/><line x1="16" y1="13" x2="8" y2="13"/><line x1="16" y1="17" x2="8" y2="17"/></svg>
               </button>` : ''}
             </div>
@@ -818,7 +838,7 @@ const CP = (() => {
     inputs.forEach(input => {
       const key = input.dataset.config;
       const val = input.value.trim();
-      if (val && val !== '********') data[key] = val;
+      if (val) data[key] = val;
     });
     if (!Object.keys(data).length) {
       toast('Enter at least one credential', 'warning');
@@ -918,8 +938,6 @@ const CP = (() => {
   // -----------------------------------------------------------
   // Claim Modal
   // -----------------------------------------------------------
-  let _breakdownCache = [];
-
   async function openClaimModal(platform) {
     openModal('claim-modal');
     const title = document.getElementById('claim-modal-title');
@@ -1758,7 +1776,7 @@ const CP = (() => {
         <div style="font-size:0.8rem; color:var(--text-muted); background:var(--bg-subtle, rgba(255,255,255,0.03)); border:1px solid var(--border); border-radius:6px; padding:8px 10px; margin:10px 0;">
           The credentials above run the service. To also see its <strong>balance</strong> on the dashboard,
           add earnings-tracking credentials under
-          <a href="#" onclick="event.preventDefault(); CP.openCredentialModal('${svc.slug}')" style="color:var(--accent, #3b82f6);">Settings → Collectors</a>
+          <a href="#" onclick="event.preventDefault(); CP.openCredentialModal('${escapeHtml(svc.slug)}')" style="color:var(--accent, #3b82f6);">Settings → Collectors</a>
           after deploying. This is optional — the service earns either way.
         </div>`;
       }
@@ -1974,7 +1992,7 @@ const CP = (() => {
     const data = {};
     inputs.forEach(input => {
       const val = input.value.trim();
-      if (val && val !== '********') {
+      if (val) {
         data[input.dataset.envKey] = val;
       }
     });
@@ -2056,7 +2074,7 @@ const CP = (() => {
     inputs.forEach(input => {
       const key = input.dataset.config;
       const val = input.value.trim();
-      if (val && val !== '********') {
+      if (val) {
         data[key] = val;
       }
     });
@@ -2194,6 +2212,10 @@ const CP = (() => {
 
     try {
       const alerts = await api('/api/collector-alerts');
+      // Clear any muted "alerts unavailable" styling left over from a prior
+      // failed poll now that the fetch succeeded.
+      badge.style.background = '';
+      badge.title = '';
       if (!alerts || alerts.length === 0) {
         badge.style.display = 'none';
         list.innerHTML = '<div class="notify-empty">All collectors healthy</div>';
@@ -2215,7 +2237,14 @@ const CP = (() => {
         </div>
       `).join('');
     } catch {
-      badge.style.display = 'none';
+      // The bell's own fetch failed — this is unknown, not healthy. Show a
+      // neutral/muted indicator rather than hiding the badge (which would
+      // read as "all collectors healthy").
+      badge.style.display = '';
+      badge.style.background = 'var(--text-muted)';
+      badge.textContent = '?';
+      badge.title = 'Alerts unavailable';
+      list.innerHTML = '<div class="notify-empty">Alerts unavailable — could not reach the server</div>';
     }
   }
 
@@ -2425,6 +2454,7 @@ const CP = (() => {
     filterCatalog,
     refreshServices,
     openClaimModal,
+    loadServicesTable,
     toggleInstances,
     deployServiceToWorkers,
     workerAction,
diff --git a/app/templates/auth.html b/app/templates/auth.html
index c3b5da8..e457129 100644
--- a/app/templates/auth.html
+++ b/app/templates/auth.html
@@ -151,6 +151,9 @@
     {% endif %}
 
     <form method="POST" action="{{ action }}">
+      {% if mode == "register" and is_first %}
+      <input type="hidden" name="setup_token" value="{{ setup_token }}">
+      {% endif %}
       <div class="auth-field">
         <label for="username">Username</label>
         <input type="text" id="username" name="username" required autocomplete="username" autofocus>
diff --git a/app/worker_api.py b/app/worker_api.py
index 5777057..5979c6d 100644
--- a/app/worker_api.py
+++ b/app/worker_api.py
@@ -94,7 +94,7 @@ async def _send_heartbeat() -> None:
             "os": f"{platform.system()} {platform.release()}",
             "arch": platform.machine(),
             "hostname": socket.gethostname(),
-            "docker_available": orchestrator.docker_available(),
+            "docker_available": await asyncio.to_thread(orchestrator.docker_available),
         },
     }
 
@@ -148,7 +148,7 @@ async def lifespan(app: FastAPI):
     global _heartbeat_task
 
     logger.info("CashPilot Worker '%s' starting", WORKER_NAME)
-    docker_mode = "direct" if orchestrator.docker_available() else "monitor-only"
+    docker_mode = "direct" if await asyncio.to_thread(orchestrator.docker_available) else "monitor-only"
     logger.info("Docker: %s", docker_mode)
 
     if UI_URL:
@@ -177,8 +177,9 @@ async def lifespan(app: FastAPI):
 
 
 @app.get("/", response_class=HTMLResponse)
-async def worker_status_page():
+async def worker_status_page(request: Request):
     """Self-contained HTML status page for the worker."""
+    _verify_api_key(request)
     containers = []
     try:
         containers = await asyncio.to_thread(orchestrator.get_status_cached)
@@ -237,7 +238,7 @@ async def worker_status_page():
             <div><label>Name</label><span>{_esc(WORKER_NAME)}</span></div>
             <div><label>Host</label><span>{_esc(socket.gethostname())}</span></div>
             <div><label>Platform</label><span>{_esc(platform.system())} {_esc(platform.machine())}</span></div>
-            <div><label>Docker</label><span>{"Available" if orchestrator.docker_available() else "Not available"}</span></div>
+            <div><label>Docker</label><span>{"Available" if await asyncio.to_thread(orchestrator.docker_available) else "Not available"}</span></div>
             <div><label>UI Connection</label><span>{ui_status}</span></div>
             <div><label>Last Heartbeat</label><span>{_last_heartbeat}</span></div>
         </div>
@@ -285,8 +286,29 @@ class DeploySpec(BaseModel):
     resources: ResourceSpec | None = None
 
 
-_BLOCKED_VOLUME_SOURCES = {"/", "/etc", "/var/run/docker.sock", "/root", "/proc", "/sys"}
+_BLOCKED_VOLUME_ROOTS = {
+    "/",
+    "/etc",
+    "/root",
+    "/proc",
+    "/sys",
+    "/boot",
+    "/dev",
+    "/var",
+    "/usr",
+    "/home",
+    "/lib",
+    "/lib64",
+    "/bin",
+    "/sbin",
+    "/var/run",
+    "/run",
+    "/var/lib/docker",
+}
 _BLOCKED_CAPS = {"ALL", "SYS_ADMIN", "SYS_PTRACE"}
+# Named volumes (bridge/none/unset) and mysterium's legitimate host networking are allowed;
+# `container:<id>` (namespace join) and any other value are rejected.
+_ALLOWED_NETWORK_MODES = {None, "", "bridge", "none", "host"}
 # Docker memory size syntax: a positive integer with an optional b/k/m/g unit.
 _MEM_LIMIT_RE = re.compile(r"^\d+[bkmgBKMG]?$")
 
@@ -298,12 +320,14 @@ def _validate_deploy_spec(spec: DeploySpec) -> None:
         blocked = _BLOCKED_CAPS & {c.upper() for c in spec.cap_add}
         if blocked:
             raise HTTPException(status_code=403, detail=f"Blocked capabilities: {', '.join(blocked)}")
+    if spec.network_mode not in _ALLOWED_NETWORK_MODES:
+        raise HTTPException(status_code=403, detail=f"Network mode '{spec.network_mode}' is not allowed")
     for source in spec.volumes:
-        normalized = "/" + source.strip("/")
-        if normalized == "/":
-            raise HTTPException(status_code=403, detail=f"Volume mount '{source}' is blocked")
-        for blocked in _BLOCKED_VOLUME_SOURCES:
-            if normalized == blocked or normalized.startswith(blocked + "/"):
+        if not source.startswith("/"):
+            continue  # named volume (e.g. "mysterium-data") — always allowed
+        real = os.path.realpath(source)
+        for blocked in _BLOCKED_VOLUME_ROOTS:
+            if real == blocked or real.startswith(blocked + "/"):
                 raise HTTPException(status_code=403, detail=f"Volume mount '{source}' is blocked")
     _validate_resources(spec.resources)
 
@@ -335,7 +359,7 @@ async def api_worker_status(request: Request) -> dict[str, Any]:
         logger.warning("Failed to get container status: %s", exc)
     return {
         "name": WORKER_NAME,
-        "docker_available": orchestrator.docker_available(),
+        "docker_available": await asyncio.to_thread(orchestrator.docker_available),
         "ui_connected": _ui_connected,
         "container_count": len(containers),
         "running_count": sum(1 for c in containers if c.get("status") == "running"),
diff --git a/tests/conftest.py b/tests/conftest.py
index 3418065..1ee50b4 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -44,3 +44,36 @@ async def _drain():
     # No usable loop (e.g. one is already running) — best-effort cleanup.
     with contextlib.suppress(RuntimeError):
         asyncio.run(_drain())
+
+
+@pytest.fixture(autouse=True)
+def _reset_login_attempts():
+    """Clear the in-process login rate-limit bucket before every test.
+
+    ``app.main._login_attempts`` is a module-level dict keyed by client host that
+    persists for the whole test process, and TestClient's host is a constant
+    ("testclient"), so failed-login attempts from one test would otherwise leak
+    into later tests that hit /login — an order-dependent landmine (and the reason
+    a real rate-limit test couldn't be added safely before). Start each test with
+    an empty bucket.
+    """
+    with contextlib.suppress(Exception):
+        from app import main
+
+        main._login_attempts.clear()
+    yield
+
+
+@pytest.fixture(autouse=True)
+def _reset_setup_token():
+    """Clear the first-run setup-token module global before every test.
+
+    ``app.setup_token._active`` persists for the whole process; a test that runs
+    lifespan on a fresh DB (or exercises the token directly) would otherwise leak
+    an active token into later tests, making unrelated /register tests 403.
+    """
+    with contextlib.suppress(Exception):
+        from app import setup_token
+
+        setup_token.clear()
+    yield
diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py
index 3725f1c..19f2faf 100644
--- a/tests/test_coverage_gaps.py
+++ b/tests/test_coverage_gaps.py
@@ -705,7 +705,7 @@ def test_worker_command_deploy_error(self, client):
         mock_client.post.return_value = error_resp
 
         with (
-            _auth_writer(),
+            _auth_owner(),  # deploy via command is owner-gated
             patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
diff --git a/tests/test_main_deploy_routes.py b/tests/test_main_deploy_routes.py
index e0493ed..0938d36 100644
--- a/tests/test_main_deploy_routes.py
+++ b/tests/test_main_deploy_routes.py
@@ -343,6 +343,7 @@ def test_remove_service(self, client):
             patch("app.main.database.remove_deployment", new_callable=AsyncMock),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
+            patch("app.main.database.record_health_event", new_callable=AsyncMock),
         ):
             resp = client.delete("/api/services/honeygain")
             assert resp.status_code == 200
@@ -374,6 +375,7 @@ def test_old_stop_route(self, client):
             patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
+            patch("app.main.database.record_health_event", new_callable=AsyncMock),
         ):
             resp = client.post("/api/stop/honeygain")
             assert resp.status_code == 200
@@ -386,6 +388,7 @@ def test_old_restart_route(self, client):
             patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
+            patch("app.main.database.record_health_event", new_callable=AsyncMock),
         ):
             resp = client.post("/api/restart/honeygain")
             assert resp.status_code == 200
@@ -399,6 +402,7 @@ def test_old_remove_route(self, client):
             patch("app.main.database.remove_deployment", new_callable=AsyncMock),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
+            patch("app.main.database.record_health_event", new_callable=AsyncMock),
         ):
             resp = client.delete("/api/remove/honeygain")
             assert resp.status_code == 200
@@ -412,6 +416,7 @@ def test_remove_with_delete_volumes(self, client):
             patch("app.main.database.remove_deployment", new_callable=AsyncMock),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
+            patch("app.main.database.record_health_event", new_callable=AsyncMock),
         ):
             resp = client.delete("/api/services/honeygain?delete_volumes=true")
             assert resp.status_code == 200
@@ -507,9 +512,11 @@ def _setup(self):
         return worker, mock_client
 
     def test_command_deploy(self, client):
+        # Deploy via the command route is OWNER-gated (matching /api/deploy/{slug}),
+        # so it must not be reachable with a mere writer role — use owner here.
         worker, mock_client = self._setup()
         with (
-            _auth_writer(),
+            _auth_owner(),
             patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker),
             patch("app.main.httpx.AsyncClient", return_value=mock_client),
             patch("app.main.FLEET_API_KEY", "test-key"),
@@ -524,6 +531,23 @@ def test_command_deploy(self, client):
             )
             assert resp.status_code == 200
 
+    def test_command_deploy_writer_denied(self, client):
+        """A writer must NOT be able to deploy via /api/workers/{id}/command — deploy is
+        owner-only, so this route must not become an owner-gate bypass (writer stays
+        allowed for stop/restart/remove, tested separately)."""
+        worker, mock_client = self._setup()
+        with (
+            _auth_writer(),
+            patch("app.main.database.get_worker", new_callable=AsyncMock, return_value=worker),
+            patch("app.main.httpx.AsyncClient", return_value=mock_client),
+            patch("app.main.FLEET_API_KEY", "test-key"),
+        ):
+            resp = client.post(
+                "/api/workers/1/command",
+                json={"command": "deploy", "slug": "honeygain", "spec": {"image": "test"}},
+            )
+            assert resp.status_code == 403
+
     def test_command_stop(self, client):
         worker, mock_client = self._setup()
         with (
diff --git a/tests/test_main_routes.py b/tests/test_main_routes.py
index 8b0ff6e..4e082bb 100644
--- a/tests/test_main_routes.py
+++ b/tests/test_main_routes.py
@@ -264,6 +264,53 @@ def test_register_first_user_success(self, client):
             patch("app.main.database.get_user_by_username", new_callable=AsyncMock, return_value=None),
             patch("app.main.auth.hash_password", return_value="hashed"),
             patch("app.main.database.create_user", new_callable=AsyncMock, return_value=1),
+            patch("app.main.database.delete_config_keys", new_callable=AsyncMock),
+            patch("app.main.auth.create_session_token", return_value="tok"),
+            patch("app.main.auth.set_session_cookie", side_effect=lambda r, t: r),
+        ):
+            resp = client.post(
+                "/register",
+                data={
+                    "username": "admin",
+                    "password": "password123",
+                    "password_confirm": "password123",
+                },
+                follow_redirects=False,
+            )
+            assert resp.status_code == 303
+
+    def test_register_first_user_requires_setup_token(self, client):
+        # When a first-run setup token is active, registering without it is refused
+        # (the proxy-independent gate against a public visitor seizing the owner).
+        from app import setup_token
+
+        setup_token.set_active("the-token")
+        with (
+            _no_auth(),
+            patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=False),
+        ):
+            resp = client.post(
+                "/register",
+                data={
+                    "username": "admin",
+                    "password": "password123",
+                    "password_confirm": "password123",
+                },
+                follow_redirects=False,
+            )
+            assert resp.status_code == 403
+
+    def test_register_first_user_with_setup_token_clears_it(self, client):
+        from app import setup_token
+
+        setup_token.set_active("the-token")
+        with (
+            _no_auth(),
+            patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=False),
+            patch("app.main.database.get_user_by_username", new_callable=AsyncMock, return_value=None),
+            patch("app.main.auth.hash_password", return_value="hashed"),
+            patch("app.main.database.create_user", new_callable=AsyncMock, return_value=1),
+            patch("app.main.database.delete_config_keys", new_callable=AsyncMock) as del_keys,
             patch("app.main.auth.create_session_token", return_value="tok"),
             patch("app.main.auth.set_session_cookie", side_effect=lambda r, t: r),
         ):
@@ -273,10 +320,13 @@ def test_register_first_user_success(self, client):
                     "username": "admin",
                     "password": "password123",
                     "password_confirm": "password123",
+                    "setup_token": "the-token",
                 },
                 follow_redirects=False,
             )
             assert resp.status_code == 303
+            del_keys.assert_awaited_once_with(["_setup_token"])
+            assert setup_token.active() is None
 
     def test_register_password_mismatch(self, client):
         with (
@@ -639,6 +689,62 @@ def test_deployed_collector_not_needs_setup_when_creds_present(self, client):
             row = next(r for r in resp.json() if r["slug"] == "repocket")
             assert row["collector_needs_setup"] is False
 
+    def test_deployed_row_flags_unstable_on_repeated_crashes(self, client):
+        """A service with >=3 crashes in the health window is flagged unstable so the
+        dashboard can surface a crash-looping earner at a glance."""
+        workers = [
+            {
+                "id": 1,
+                "name": "w1",
+                "status": "online",
+                "system_info": json.dumps({"docker_available": True}),
+                "containers": json.dumps([{"slug": "repocket", "name": "rp", "status": "running"}]),
+                "apps": "[]",
+            }
+        ]
+        scores = [{"slug": "repocket", "score": 30, "uptime_pct": 40, "restarts": 5, "crashes": 4}]
+        with (
+            _auth_owner(),
+            patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
+            patch("app.main.database.get_earnings_summary", new_callable=AsyncMock, return_value=[]),
+            patch("app.main.database.get_health_scores", new_callable=AsyncMock, return_value=scores),
+            patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=[]),
+            patch("app.main.database.get_config", new_callable=AsyncMock, return_value={}),
+            patch("app.main.catalog.get_service", return_value={"name": "Repocket", "category": "bandwidth"}),
+        ):
+            resp = client.get("/api/services/deployed")
+            assert resp.status_code == 200
+            row = next(r for r in resp.json() if r["slug"] == "repocket")
+            assert row["unstable"] is True
+            assert row["crashes_7d"] == 4
+
+    def test_deployed_row_not_unstable_below_threshold(self, client):
+        """Fewer than 3 crashes is not flagged unstable (crashes_7d is still surfaced)."""
+        workers = [
+            {
+                "id": 1,
+                "name": "w1",
+                "status": "online",
+                "system_info": json.dumps({"docker_available": True}),
+                "containers": json.dumps([{"slug": "repocket", "name": "rp", "status": "running"}]),
+                "apps": "[]",
+            }
+        ]
+        scores = [{"slug": "repocket", "score": 85, "uptime_pct": 98, "restarts": 1, "crashes": 2}]
+        with (
+            _auth_owner(),
+            patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
+            patch("app.main.database.get_earnings_summary", new_callable=AsyncMock, return_value=[]),
+            patch("app.main.database.get_health_scores", new_callable=AsyncMock, return_value=scores),
+            patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=[]),
+            patch("app.main.database.get_config", new_callable=AsyncMock, return_value={}),
+            patch("app.main.catalog.get_service", return_value={"name": "Repocket", "category": "bandwidth"}),
+        ):
+            resp = client.get("/api/services/deployed")
+            row = next(r for r in resp.json() if r["slug"] == "repocket")
+            assert row["unstable"] is False
+            assert row["crashes_7d"] == 2
+
     def test_deployed_disconnected_takes_precedence_over_needs_setup(self, client):
         # When the collector has actually errored, show the error state, not "needs setup".
         workers = [
@@ -1735,6 +1841,7 @@ async def _run():
                 patch("app.main.database.connect_shared", new_callable=AsyncMock),
                 patch("app.main.database.close_shared", new_callable=AsyncMock),
                 patch("app.main.database.list_users_with_pwd_epoch", new_callable=AsyncMock, return_value=[]),
+                patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=True),
                 patch("app.main.catalog.load_services"),
                 patch("app.main.catalog.register_sighup"),
                 patch("app.main.exchange_rates.refresh", new_callable=AsyncMock),
@@ -1744,12 +1851,13 @@ async def _run():
                 async with main_mod.lifespan(main_mod.app):
                     sched = main_mod.scheduler
                     jobs = {j.id: j for j in sched.get_jobs()}
-                    # All five interval jobs registered.
+                    # All interval jobs registered.
                     assert set(jobs) == {
                         "collect",
                         "health_check",
                         "stale_workers",
                         "data_retention",
+                        "db_vacuum",
                         "exchange_rates",
                     }
                     # Every job carries the hardening kwargs (audit fix).
@@ -1774,6 +1882,7 @@ async def _run():
                 patch("app.main.database.connect_shared", new_callable=AsyncMock),
                 patch("app.main.database.close_shared", new_callable=AsyncMock),
                 patch("app.main.database.list_users_with_pwd_epoch", new_callable=AsyncMock, return_value=[]),
+                patch("app.main.database.has_any_users", new_callable=AsyncMock, return_value=True),
                 patch("app.main.catalog.load_services"),
                 patch("app.main.catalog.register_sighup"),
                 patch("app.main.exchange_rates.refresh", new_callable=AsyncMock),
diff --git a/tests/test_orchestrator_coverage.py b/tests/test_orchestrator_coverage.py
new file mode 100644
index 0000000..8a9f23b
--- /dev/null
+++ b/tests/test_orchestrator_coverage.py
@@ -0,0 +1,301 @@
+"""Tests targeting uncovered lines in app/orchestrator.py.
+
+Covers get_status / get_status_light (running / exited / missing-container /
+docker-unavailable branches and the image-matched external-container path),
+_collect_stats CPU/memory parsing (including zero-delta and error edge
+cases), and _find_container's label-based fallback lookup.
+
+Mocks the Docker SDK entirely — no real Docker socket is used.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key")
+
+import pytest  # noqa: E402
+
+docker = pytest.importorskip("docker")  # noqa: E402
+
+from docker.errors import APIError, NotFound  # noqa: E402
+
+from app import orchestrator  # noqa: E402
+from app.constants import LABEL_MANAGED, LABEL_SERVICE  # noqa: E402
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _mock_container(*, name, status, slug, deployed_by="worker", category="bandwidth", container_id="short123"):
+    c = MagicMock()
+    c.id = f"id-{name}"
+    c.name = name
+    c.status = status
+    c.short_id = container_id
+    c.labels = {
+        LABEL_SERVICE: slug,
+        LABEL_MANAGED: "true",
+        "cashpilot.deployed-by": deployed_by,
+        "cashpilot.category": category,
+    }
+    c.image.tags = [f"{slug}:latest"]
+    c.image.short_id = "sha256:abcdef"
+    c.attrs = {"Created": "2026-01-01T00:00:00Z"}
+    return c
+
+
+def _zero_stats():
+    return {
+        "cpu_stats": {"cpu_usage": {"total_usage": 1, "percpu_usage": [1]}, "system_cpu_usage": 10},
+        "precpu_stats": {"cpu_usage": {"total_usage": 0}, "system_cpu_usage": 5},
+        "memory_stats": {"usage": 0},
+    }
+
+
+# ---------------------------------------------------------------------------
+# _collect_stats
+# ---------------------------------------------------------------------------
+
+
+class TestCollectStats:
+    def test_parses_cpu_and_memory(self):
+        c = MagicMock()
+        c.stats.return_value = {
+            "cpu_stats": {
+                "cpu_usage": {"total_usage": 2_000_000_000, "percpu_usage": [1, 1]},
+                "system_cpu_usage": 100_000_000_000,
+                "online_cpus": 2,
+            },
+            "precpu_stats": {
+                "cpu_usage": {"total_usage": 1_000_000_000},
+                "system_cpu_usage": 90_000_000_000,
+            },
+            "memory_stats": {"usage": 209_715_200},  # 200 MB
+        }
+        cpu_pct, mem_mb = orchestrator._collect_stats(c)
+        assert cpu_pct == 20.0
+        assert mem_mb == 200.0
+
+    def test_zero_system_delta_returns_zero_cpu(self):
+        c = MagicMock()
+        c.stats.return_value = {
+            "cpu_stats": {"cpu_usage": {"total_usage": 500, "percpu_usage": [1]}, "system_cpu_usage": 500},
+            "precpu_stats": {"cpu_usage": {"total_usage": 500}, "system_cpu_usage": 500},
+            "memory_stats": {"usage": 0},
+        }
+        cpu_pct, mem_mb = orchestrator._collect_stats(c)
+        assert cpu_pct == 0.0
+        assert mem_mb == 0.0
+
+    def test_missing_memory_usage_defaults_to_zero(self):
+        c = MagicMock()
+        c.stats.return_value = {
+            "cpu_stats": {"cpu_usage": {"total_usage": 100, "percpu_usage": [1]}, "system_cpu_usage": 500},
+            "precpu_stats": {"cpu_usage": {"total_usage": 50}, "system_cpu_usage": 400},
+            "memory_stats": {},  # no "usage" key
+        }
+        _, mem_mb = orchestrator._collect_stats(c)
+        assert mem_mb == 0.0
+
+    def test_missing_key_returns_zero_tuple(self):
+        c = MagicMock()
+        c.stats.return_value = {"cpu_stats": {}}  # precpu_stats missing -> KeyError
+        assert orchestrator._collect_stats(c) == (0.0, 0.0)
+
+    def test_stats_api_error_returns_zero_tuple(self):
+        c = MagicMock()
+        c.stats.side_effect = APIError("stats unavailable")
+        assert orchestrator._collect_stats(c) == (0.0, 0.0)
+
+
+# ---------------------------------------------------------------------------
+# _find_container
+# ---------------------------------------------------------------------------
+
+
+class TestFindContainer:
+    def test_finds_by_name(self):
+        container = MagicMock()
+        client = MagicMock()
+        client.containers.get.return_value = container
+        with patch.object(orchestrator, "_get_client", return_value=client):
+            result = orchestrator._find_container("honeygain")
+        assert result is container
+        client.containers.get.assert_called_once_with("cashpilot-honeygain")
+        client.containers.list.assert_not_called()
+
+    def test_falls_back_to_label_lookup_when_renamed(self):
+        container = MagicMock()
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("nope")
+        client.containers.list.return_value = [container]
+        with patch.object(orchestrator, "_get_client", return_value=client):
+            result = orchestrator._find_container("honeygain")
+        assert result is container
+        _, kwargs = client.containers.list.call_args
+        assert kwargs["filters"]["label"] == [
+            f"{LABEL_SERVICE}=honeygain",
+            f"{LABEL_MANAGED}=true",
+        ]
+
+    def test_raises_value_error_when_not_found_anywhere(self):
+        client = MagicMock()
+        client.containers.get.side_effect = NotFound("nope")
+        client.containers.list.return_value = []
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            pytest.raises(ValueError, match="honeygain"),
+        ):
+            orchestrator._find_container("honeygain")
+
+
+# ---------------------------------------------------------------------------
+# get_status (slow path, includes CPU/mem stats)
+# ---------------------------------------------------------------------------
+
+
+class TestGetStatus:
+    def test_docker_unavailable_returns_empty(self):
+        with patch.object(orchestrator, "_get_client", side_effect=RuntimeError("no socket")):
+            assert orchestrator.get_status() == []
+
+    def test_running_container_reports_stats(self):
+        container = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain")
+        container.stats.return_value = {
+            "cpu_stats": {"cpu_usage": {"total_usage": 100, "percpu_usage": [1]}, "system_cpu_usage": 500},
+            "precpu_stats": {"cpu_usage": {"total_usage": 50}, "system_cpu_usage": 400},
+            "memory_stats": {"usage": 1_048_576},
+        }
+        client = MagicMock()
+        client.containers.list.return_value = [container]
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value={}),
+        ):
+            results = orchestrator.get_status()
+        assert len(results) == 1
+        row = results[0]
+        assert row["slug"] == "honeygain"
+        assert row["status"] == "running"
+        assert row["memory_mb"] == 1.0
+
+    def test_exited_container_missing_stats_handled_gracefully(self):
+        container = _mock_container(name="cashpilot-earnapp", status="exited", slug="earnapp")
+        container.stats.side_effect = APIError("exited, no stats")
+        client = MagicMock()
+        client.containers.list.return_value = [container]
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value={}),
+        ):
+            results = orchestrator.get_status()
+        assert results[0]["status"] == "exited"
+        assert results[0]["cpu_percent"] == 0.0
+        assert results[0]["memory_mb"] == 0.0
+
+    def test_corrupted_container_is_skipped_not_crashed(self):
+        good = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain")
+        good.stats.return_value = _zero_stats()
+        bad = MagicMock()
+        bad.id = "bad-id"
+        bad.short_id = "bad12"
+        bad.labels.get.side_effect = RuntimeError("corrupted container labels")
+        client = MagicMock()
+        client.containers.list.return_value = [bad, good]
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value={}),
+        ):
+            results = orchestrator.get_status()
+        assert len(results) == 1
+        assert results[0]["slug"] == "honeygain"
+
+    def test_image_matched_external_container_included(self):
+        external = MagicMock()
+        external.id = "ext-1"
+        external.name = "manually-run-storj"
+        external.status = "running"
+        external.image.tags = ["storjlabs/storagenode:latest"]
+        external.short_id = "ext1"
+        external.attrs = {"Created": "2026-01-01T00:00:00Z"}
+        external.stats.return_value = _zero_stats()
+        client = MagicMock()
+        client.containers.list.side_effect = [[], [external]]
+        image_map = {"storjlabs/storagenode:latest": "storj"}
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value=image_map),
+        ):
+            results = orchestrator.get_status()
+        assert len(results) == 1
+        assert results[0]["slug"] == "storj"
+        assert results[0]["deployed_by"] == "external"
+
+    def test_all_containers_listing_failure_still_returns_labeled(self):
+        labeled = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain")
+        labeled.stats.return_value = _zero_stats()
+        client = MagicMock()
+        client.containers.list.side_effect = [[labeled], Exception("docker daemon hiccup")]
+        image_map = {"some/image:latest": "svc"}
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value=image_map),
+        ):
+            results = orchestrator.get_status()
+        assert len(results) == 1
+        assert results[0]["slug"] == "honeygain"
+
+
+# ---------------------------------------------------------------------------
+# get_status_light (fast path, no CPU/mem stats)
+# ---------------------------------------------------------------------------
+
+
+class TestGetStatusLight:
+    def test_docker_unavailable_returns_empty(self):
+        with patch.object(orchestrator, "_get_client", side_effect=RuntimeError("no socket")):
+            assert orchestrator.get_status_light() == []
+
+    def test_no_stats_call_even_when_running(self):
+        container = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain")
+        client = MagicMock()
+        client.containers.list.return_value = [container]
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value={}),
+        ):
+            results = orchestrator.get_status_light()
+        assert results[0]["cpu_percent"] == 0.0
+        assert results[0]["memory_mb"] == 0.0
+        container.stats.assert_not_called()
+
+    def test_image_matched_container_skipped_when_slug_already_seen(self):
+        labeled = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain")
+        dup = MagicMock()
+        dup.id = "dup-id"
+        dup.image.tags = ["honeygain/desktop:latest"]
+        client = MagicMock()
+        client.containers.list.side_effect = [[labeled], [dup]]
+        image_map = {"honeygain/desktop:latest": "honeygain"}
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value=image_map),
+        ):
+            results = orchestrator.get_status_light()
+        # The duplicate slug from the image-matched scan must not be added again.
+        assert len(results) == 1
+
+    def test_all_containers_listing_failure_handled(self):
+        labeled = _mock_container(name="cashpilot-honeygain", status="running", slug="honeygain")
+        client = MagicMock()
+        client.containers.list.side_effect = [[labeled], Exception("boom")]
+        image_map = {"x": "y"}
+        with (
+            patch.object(orchestrator, "_get_client", return_value=client),
+            patch.object(orchestrator, "_build_image_slug_map", return_value=image_map),
+        ):
+            results = orchestrator.get_status_light()
+        assert len(results) == 1
diff --git a/tests/test_setup_token.py b/tests/test_setup_token.py
new file mode 100644
index 0000000..59621e5
--- /dev/null
+++ b/tests/test_setup_token.py
@@ -0,0 +1,105 @@
+"""Tests for the first-run setup-token gate (app/setup_token.py + app/deps)."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from app import deps, setup_token
+
+
+def _req(host="127.0.0.1", xff=None, query=None, headers=None):
+    """A minimal fake Request: real dicts for headers/query_params so .get works."""
+    req = MagicMock()
+    if host is None:
+        req.client = None
+    else:
+        req.client.host = host
+    h = {}
+    if xff is not None:
+        h["x-forwarded-for"] = xff
+    if headers:
+        h.update(headers)
+    req.headers = h
+    req.query_params = query or {}
+    return req
+
+
+class TestSetupTokenModule:
+    def test_verify_no_active_allows_anything(self):
+        setup_token.clear()
+        assert setup_token.verify(None) is True
+        assert setup_token.verify("whatever") is True
+
+    def test_verify_active_requires_exact_match(self):
+        setup_token.set_active("s3cret")
+        assert setup_token.verify("s3cret") is True
+        assert setup_token.verify("wrong") is False
+        assert setup_token.verify(None) is False
+        assert setup_token.verify("") is False
+
+    def test_generate_is_unique_and_long(self):
+        a, b = setup_token.generate(), setup_token.generate()
+        assert a != b
+        assert len(a) >= 20
+
+    def test_set_active_empty_string_deactivates(self):
+        setup_token.set_active("")
+        assert setup_token.active() is None
+
+
+class TestClientIp:
+    def test_ignores_xff_without_trusted_proxy(self):
+        with patch.object(deps, "_TRUST_PROXY", False):
+            assert deps.client_ip(_req("10.0.0.1", xff="1.2.3.4")) == "10.0.0.1"
+
+    def test_uses_rightmost_xff_with_trusted_proxy(self):
+        # The trusted proxy appends the real peer on the right; a client can only
+        # forge entries on the left, so the right-most value is authoritative.
+        with patch.object(deps, "_TRUST_PROXY", True):
+            req = _req("10.0.0.1", xff="9.9.9.9, 8.8.8.8, 203.0.113.5")
+            assert deps.client_ip(req) == "203.0.113.5"
+
+    def test_trusted_proxy_falls_back_to_peer_without_xff(self):
+        with patch.object(deps, "_TRUST_PROXY", True):
+            assert deps.client_ip(_req("10.0.0.1")) == "10.0.0.1"
+
+    def test_no_client_returns_none(self):
+        with patch.object(deps, "_TRUST_PROXY", False):
+            assert deps.client_ip(_req(host=None)) is None
+
+
+class TestRequireFirstRunAccess:
+    def test_blocks_when_token_active_and_missing(self):
+        setup_token.set_active("tok")
+        with patch.object(deps, "_TRUST_PROXY", False), pytest.raises(HTTPException) as ei:
+            deps._require_first_run_access(_req("127.0.0.1"))
+        assert ei.value.status_code == 403
+
+    def test_allows_with_correct_token_arg(self):
+        setup_token.set_active("tok")
+        with patch.object(deps, "_TRUST_PROXY", False):
+            assert deps._require_first_run_access(_req("127.0.0.1"), "tok") is None
+
+    def test_token_from_query_param(self):
+        setup_token.set_active("tok")
+        with patch.object(deps, "_TRUST_PROXY", False):
+            assert deps._require_first_run_access(_req("127.0.0.1", query={"setup_token": "tok"})) is None
+
+    def test_token_from_header(self):
+        setup_token.set_active("tok")
+        with patch.object(deps, "_TRUST_PROXY", False):
+            assert deps._require_first_run_access(_req("127.0.0.1", headers={"x-setup-token": "tok"})) is None
+
+    def test_public_ip_blocked_even_with_correct_token(self):
+        # The network check runs first: a public client is refused regardless of token.
+        setup_token.set_active("tok")
+        with patch.object(deps, "_TRUST_PROXY", False), pytest.raises(HTTPException) as ei:
+            deps._require_first_run_access(_req("8.8.8.8"), "tok")
+        assert ei.value.status_code == 403
+
+    def test_no_active_token_falls_back_to_network_only(self):
+        # Once the token is retired, first-run access is network-gated only.
+        setup_token.clear()
+        with patch.object(deps, "_TRUST_PROXY", False):
+            assert deps._require_first_run_access(_req("192.168.1.5")) is None
diff --git a/tests/test_worker_api_coverage.py b/tests/test_worker_api_coverage.py
new file mode 100644
index 0000000..7246e24
--- /dev/null
+++ b/tests/test_worker_api_coverage.py
@@ -0,0 +1,333 @@
+"""Tests targeting uncovered lines in app/worker_api.py.
+
+Covers _verify_api_key negative paths (missing header, wrong key, no key
+configured), _validate_deploy_spec rejections not already exercised by
+test_worker_resources.py (privileged, blocked capabilities, disallowed
+network_mode, blocked volume roots, named-volume allow-list), and the
+container command endpoints (status/list/deploy/restart/stop/start/remove/
+logs) success + docker-error paths.
+
+Mocks app.orchestrator entirely — no real Docker socket is used.
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import asynccontextmanager
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key")
+
+import pytest  # noqa: E402
+
+try:
+    from fastapi import HTTPException  # noqa: E402
+    from fastapi.testclient import TestClient  # noqa: E402
+
+    from app import worker_api  # noqa: E402
+    from app.worker_api import DeploySpec, _validate_deploy_spec, _verify_api_key  # noqa: E402
+except ImportError:
+    pytest.skip(
+        "Requires full app dependencies (fastapi, docker, etc.) — runs in CI",
+        allow_module_level=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Helpers — match the TestClient harness used in test_worker_resources.py
+# ---------------------------------------------------------------------------
+
+
+@asynccontextmanager
+async def _noop_lifespan(a):
+    yield
+
+
+def _client():
+    # Disable the heartbeat lifespan so the TestClient stays isolated.
+    worker_api.app.router.lifespan_context = _noop_lifespan
+    return TestClient(worker_api.app, raise_server_exceptions=False)
+
+
+def _auth():
+    return {"Authorization": f"Bearer {worker_api.API_KEY}"}
+
+
+# ---------------------------------------------------------------------------
+# _verify_api_key — negative paths
+# ---------------------------------------------------------------------------
+
+
+class TestVerifyApiKeyUnit:
+    def test_missing_header_rejected(self):
+        req = MagicMock()
+        req.headers = {}
+        with pytest.raises(HTTPException) as ei:
+            _verify_api_key(req)
+        assert ei.value.status_code == 401
+
+    def test_wrong_key_rejected(self):
+        req = MagicMock()
+        req.headers = {"Authorization": "Bearer wrong-key"}
+        with pytest.raises(HTTPException) as ei:
+            _verify_api_key(req)
+        assert ei.value.status_code == 401
+
+    def test_no_key_configured_returns_503(self):
+        req = MagicMock()
+        req.headers = {"Authorization": "Bearer anything"}
+        with patch.object(worker_api, "API_KEY", ""), pytest.raises(HTTPException) as ei:
+            _verify_api_key(req)
+        assert ei.value.status_code == 503
+
+    def test_correct_key_passes(self):
+        req = MagicMock()
+        req.headers = {"Authorization": f"Bearer {worker_api.API_KEY}"}
+        _verify_api_key(req)  # must not raise
+
+
+class TestEndpointAuthRejection:
+    def test_endpoint_rejects_missing_header(self):
+        resp = _client().get("/api/status")
+        assert resp.status_code == 401
+
+    def test_endpoint_rejects_wrong_key(self):
+        resp = _client().get("/api/status", headers={"Authorization": "Bearer nope"})
+        assert resp.status_code == 401
+
+    def test_health_endpoint_requires_no_auth(self):
+        resp = _client().get("/api/health")
+        assert resp.status_code == 200
+        assert resp.json() == {"status": "ok", "worker": worker_api.WORKER_NAME}
+
+
+# ---------------------------------------------------------------------------
+# _validate_deploy_spec — rejections not covered by test_worker_resources.py
+# ---------------------------------------------------------------------------
+
+
+class TestValidateDeploySpecRejections:
+    def test_privileged_rejected(self):
+        spec = DeploySpec(image="x", privileged=True)
+        with pytest.raises(HTTPException) as ei:
+            _validate_deploy_spec(spec)
+        assert ei.value.status_code == 403
+
+    def test_blocked_capability_rejected(self):
+        spec = DeploySpec(image="x", cap_add=["SYS_ADMIN"])
+        with pytest.raises(HTTPException) as ei:
+            _validate_deploy_spec(spec)
+        assert ei.value.status_code == 403
+
+    def test_blocked_capability_rejected_case_insensitive(self):
+        spec = DeploySpec(image="x", cap_add=["sys_ptrace"])
+        with pytest.raises(HTTPException) as ei:
+            _validate_deploy_spec(spec)
+        assert ei.value.status_code == 403
+
+    def test_allowed_capability_passes(self):
+        _validate_deploy_spec(DeploySpec(image="x", cap_add=["NET_ADMIN"]))  # must not raise
+
+    def test_disallowed_network_mode_rejected(self):
+        spec = DeploySpec(image="x", network_mode="container:abc123")
+        with pytest.raises(HTTPException) as ei:
+            _validate_deploy_spec(spec)
+        assert ei.value.status_code == 403
+
+    def test_host_network_mode_allowed(self):
+        # mysterium legitimately needs host networking.
+        _validate_deploy_spec(DeploySpec(image="x", network_mode="host"))  # must not raise
+
+    def test_bridge_network_mode_allowed(self):
+        _validate_deploy_spec(DeploySpec(image="x", network_mode="bridge"))  # must not raise
+
+    @pytest.mark.parametrize("root", ["/etc", "/root", "/var/run", "/proc"])
+    def test_blocked_volume_root_rejected(self, root):
+        # Patch realpath to identity: on macOS dev machines /etc and /var/run
+        # are themselves symlinks (-> /private/etc, /private/var/run), which
+        # would dodge the block by resolving to a path outside the blocklist.
+        # In the worker's actual Linux container this resolution is a no-op.
+        spec = DeploySpec(image="x", volumes={root: {"bind": "/data", "mode": "rw"}})
+        with (
+            patch("app.worker_api.os.path.realpath", side_effect=lambda p: p),
+            pytest.raises(HTTPException) as ei,
+        ):
+            _validate_deploy_spec(spec)
+        assert ei.value.status_code == 403
+
+    def test_blocked_volume_subpath_rejected(self):
+        spec = DeploySpec(image="x", volumes={"/etc/passwd": {"bind": "/x", "mode": "ro"}})
+        with (
+            patch("app.worker_api.os.path.realpath", side_effect=lambda p: p),
+            pytest.raises(HTTPException) as ei,
+        ):
+            _validate_deploy_spec(spec)
+        assert ei.value.status_code == 403
+
+    def test_named_volume_always_allowed(self):
+        # Not an absolute path -> named volume (e.g. mysterium-data), always allowed.
+        spec = DeploySpec(image="x", volumes={"mysterium-data": {"bind": "/data", "mode": "rw"}})
+        _validate_deploy_spec(spec)  # must not raise
+
+    def test_allowed_bind_mount_passes(self):
+        spec = DeploySpec(image="x", volumes={"/data/honeygain": {"bind": "/data", "mode": "rw"}})
+        _validate_deploy_spec(spec)  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# Container command endpoints — success + docker-error paths
+# ---------------------------------------------------------------------------
+
+
+class TestStatusAndListEndpoints:
+    def test_status_reports_docker_and_counts(self):
+        containers = [{"status": "running"}, {"status": "exited"}]
+        with (
+            patch("app.worker_api.orchestrator.get_status_cached", return_value=containers),
+            patch("app.worker_api.orchestrator.docker_available", return_value=True),
+        ):
+            resp = _client().get("/api/status", headers=_auth())
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["docker_available"] is True
+        assert body["container_count"] == 2
+        assert body["running_count"] == 1
+
+    def test_status_handles_status_fetch_exception(self):
+        with (
+            patch("app.worker_api.orchestrator.get_status_cached", side_effect=Exception("boom")),
+            patch("app.worker_api.orchestrator.docker_available", return_value=False),
+        ):
+            resp = _client().get("/api/status", headers=_auth())
+        assert resp.status_code == 200
+        assert resp.json()["container_count"] == 0
+
+    def test_list_containers_success(self):
+        containers = [{"slug": "honeygain"}]
+        with patch("app.worker_api.orchestrator.get_status_cached", return_value=containers):
+            resp = _client().get("/api/containers", headers=_auth())
+        assert resp.status_code == 200
+        assert resp.json() == containers
+
+    def test_list_containers_docker_unavailable_returns_503(self):
+        with patch("app.worker_api.orchestrator.get_status_cached", side_effect=RuntimeError("no docker")):
+            resp = _client().get("/api/containers", headers=_auth())
+        assert resp.status_code == 503
+
+
+class TestDeployEndpointErrors:
+    def test_generic_exception_returns_500(self):
+        with patch("app.worker_api.orchestrator.deploy_raw", side_effect=Exception("boom")):
+            resp = _client().post(
+                "/api/containers/honeygain/deploy",
+                json={"image": "img"},
+                headers=_auth(),
+            )
+        assert resp.status_code == 500
+
+
+class TestStopRestartStartEndpoints:
+    def test_stop_success(self):
+        with patch("app.worker_api.orchestrator.stop_service") as mock_stop:
+            resp = _client().post("/api/containers/honeygain/stop", headers=_auth())
+        assert resp.status_code == 200
+        assert resp.json() == {"status": "stopped"}
+        mock_stop.assert_called_once_with("honeygain")
+
+    def test_stop_not_found_returns_404(self):
+        with patch("app.worker_api.orchestrator.stop_service", side_effect=ValueError("not found")):
+            resp = _client().post("/api/containers/missing/stop", headers=_auth())
+        assert resp.status_code == 404
+
+    def test_stop_docker_unavailable_returns_503(self):
+        with patch("app.worker_api.orchestrator.stop_service", side_effect=RuntimeError("no docker")):
+            resp = _client().post("/api/containers/honeygain/stop", headers=_auth())
+        assert resp.status_code == 503
+
+    def test_restart_success(self):
+        with patch("app.worker_api.orchestrator.restart_service") as mock_restart:
+            resp = _client().post("/api/containers/honeygain/restart", headers=_auth())
+        assert resp.status_code == 200
+        assert resp.json() == {"status": "restarted"}
+        mock_restart.assert_called_once_with("honeygain")
+
+    def test_restart_not_found_returns_404(self):
+        with patch("app.worker_api.orchestrator.restart_service", side_effect=ValueError("gone")):
+            resp = _client().post("/api/containers/x/restart", headers=_auth())
+        assert resp.status_code == 404
+
+    def test_restart_docker_unavailable_returns_503(self):
+        with patch("app.worker_api.orchestrator.restart_service", side_effect=RuntimeError("no docker")):
+            resp = _client().post("/api/containers/honeygain/restart", headers=_auth())
+        assert resp.status_code == 503
+
+    def test_start_success(self):
+        with patch("app.worker_api.orchestrator.start_service") as mock_start:
+            resp = _client().post("/api/containers/honeygain/start", headers=_auth())
+        assert resp.status_code == 200
+        assert resp.json() == {"status": "started"}
+        mock_start.assert_called_once_with("honeygain")
+
+    def test_start_not_found_returns_404(self):
+        with patch("app.worker_api.orchestrator.start_service", side_effect=ValueError("gone")):
+            resp = _client().post("/api/containers/x/start", headers=_auth())
+        assert resp.status_code == 404
+
+    def test_start_docker_unavailable_returns_503(self):
+        with patch("app.worker_api.orchestrator.start_service", side_effect=RuntimeError("no docker")):
+            resp = _client().post("/api/containers/honeygain/start", headers=_auth())
+        assert resp.status_code == 503
+
+
+class TestRemoveEndpoint:
+    def test_remove_success(self):
+        removal = {"container": "cashpilot-honeygain", "deleted_volumes": [], "failed_volumes": []}
+        with patch("app.worker_api.orchestrator.remove_service", return_value=removal) as mock_remove:
+            resp = _client().delete("/api/containers/honeygain", headers=_auth())
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "removed"
+        assert body["container"] == "cashpilot-honeygain"
+        mock_remove.assert_called_once_with("honeygain", delete_volumes=False)
+
+    def test_remove_with_delete_volumes_query_param(self):
+        removal = {"container": "cashpilot-honeygain", "deleted_volumes": ["v1"], "failed_volumes": []}
+        with patch("app.worker_api.orchestrator.remove_service", return_value=removal) as mock_remove:
+            resp = _client().delete("/api/containers/honeygain?delete_volumes=true", headers=_auth())
+        assert resp.status_code == 200
+        mock_remove.assert_called_once_with("honeygain", delete_volumes=True)
+
+    def test_remove_not_found_returns_404(self):
+        with patch("app.worker_api.orchestrator.remove_service", side_effect=ValueError("gone")):
+            resp = _client().delete("/api/containers/x", headers=_auth())
+        assert resp.status_code == 404
+
+    def test_remove_docker_unavailable_returns_503(self):
+        with patch("app.worker_api.orchestrator.remove_service", side_effect=RuntimeError("no docker")):
+            resp = _client().delete("/api/containers/honeygain", headers=_auth())
+        assert resp.status_code == 503
+
+
+class TestLogsEndpoint:
+    def test_logs_success(self):
+        with patch("app.worker_api.orchestrator.get_service_logs", return_value="line1\nline2") as mock_logs:
+            resp = _client().get("/api/containers/honeygain/logs", headers=_auth())
+        assert resp.status_code == 200
+        assert resp.json() == {"logs": "line1\nline2"}
+        mock_logs.assert_called_once_with("honeygain", lines=50)
+
+    def test_logs_lines_capped_at_1000(self):
+        with patch("app.worker_api.orchestrator.get_service_logs", return_value="") as mock_logs:
+            resp = _client().get("/api/containers/honeygain/logs?lines=5000", headers=_auth())
+        assert resp.status_code == 200
+        mock_logs.assert_called_once_with("honeygain", lines=1000)
+
+    def test_logs_not_found_returns_404(self):
+        with patch("app.worker_api.orchestrator.get_service_logs", side_effect=ValueError("gone")):
+            resp = _client().get("/api/containers/x/logs", headers=_auth())
+        assert resp.status_code == 404
+
+    def test_logs_docker_unavailable_returns_503(self):
+        with patch("app.worker_api.orchestrator.get_service_logs", side_effect=RuntimeError("no docker")):
+            resp = _client().get("/api/containers/honeygain/logs", headers=_auth())
+        assert resp.status_code == 503