fix(obico): capture snapshots locally and serve via nonce URL (#172)

  Obico's ML API has a hardcoded 5s read timeout on the URL it fetches, which
  our /camera/snapshot regularly exceeds on cold calls (TLS proxy + ffmpeg +
  RTSP keyframe wait). The detection loop now captures the JPEG locally with
  a 20s timeout we control, stashes the bytes under a single-use 32-byte
  nonce, and hands Obico a new /api/v1/obico/cached-frame/{nonce} URL that
  returns the cached bytes instantly. The 5s ceiling is no longer a factor.

  The nonce is the credential (URL-safe, 256 bits of entropy, single-use,
  30s TTL) so the endpoint can be unauthenticated without widening the
  camera access surface. Replaces the previous camera-stream-token snapshot
  URL approach, which remained vulnerable to the upstream 5s timeout even
  when auth was disabled.

  Thanks to @fblix for the detailed reproducer with timeout numbers.

CHANGELOG.md

@@ -30,6 +30,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Print Speed Icon Not Updating Live When Changed on Printer** ([#993](https://github.com/maziggy/bambuddy/issues/993)) — Changing the print speed mode from the printer's own panel (instead of from Bambuddy) did not update the speed icon on the Printers page card; the new value only appeared after a full page reload. The MQTT parser was already tracking `spd_lvl` and updating `state.speed_level` correctly, but the WebSocket serializer (`printer_state_to_dict`) was missing the field — so live status pushes never carried `speed_level`, and the frontend's merge-over-old-cache update left the icon stuck on its previous value. The REST `/status` endpoint used on initial page load already included it, which is why reloads worked. Added `speed_level` to the WebSocket payload. Thanks to @chesterakl for reporting.
 - **Camera Popup Shows "Valid camera stream token required" With Auth Enabled** ([#979](https://github.com/maziggy/bambuddy/issues/979)) — When Camera View Mode was set to "Window" and authentication was enabled, clicking the camera button opened a popup that immediately failed with `"Valid camera stream token required"`, while the embedded overlay kept working. Two root causes: (1) `window.open(...)` passed `noopener` in the popup features, which severed the opener link and prevented the browser from copying sessionStorage (where the auth token lives) into the popup — so the new window booted unauthenticated and the `POST /printers/camera/stream-token` fetch returned 401, leaving the `<img>` src without the required `?token=` query param; (2) even once the token arrived, `CameraPage` computed its URL from the module-level stream-token cache on render and never re-rendered when the cache was updated in a `useEffect`, so the first paint locked in a tokenless URL that the backend kept rejecting. Fixed by dropping `noopener` from the camera popup features (same-origin, trusted window) so sessionStorage is inherited, subscribing `CameraPage` to the `camera-stream-token` React Query so it re-renders the moment the token resolves, and appending the token directly from the reactive query value instead of the effect-synced module cache — the `<img>` src stays empty until the token is ready, so no tokenless request ever leaves the popup. Embedded-overlay mode was unaffected. Thanks to @VREmma for the reproducer.
 - **Obico ML API Got 401 When Fetching Snapshot with Auth Enabled** ([#172](https://github.com/maziggy/bambuddy/issues/172)) — The Obico failure-detection service handed the ML API container a snapshot URL (`/api/v1/printers/{id}/camera/snapshot`) for it to `GET` directly, but when Bambuddy authentication was enabled the endpoint returned 401 and the ML API surfaced "Failed to get image" (visible as a 400 from the ML API back to Bambuddy). The detection service now appends a short-lived camera-stream token to the snapshot URL — the same token scheme already used by `<img>`-based camera consumers, which the snapshot endpoint already accepts. The token is cached on the service and refreshed before its 60-minute expiry, so no extra per-call DB churn. When auth is disabled the token is simply ignored. Thanks to @fblix for reporting.
+- **Obico Detection Timed Out on Slow Snapshot** ([#172](https://github.com/maziggy/bambuddy/issues/172)) — Second wave of #172 — once the auth/401 fix landed, `fblix` reported a new failure mode: `Read timed out. (read timeout=5)`. Bambuddy's `/camera/snapshot` endpoint takes 5–10 s on cold calls (it spins up a TLS proxy + ffmpeg + waits for the next RTSP keyframe), but Obico's ML API `server.py` has a hardcoded `timeout = (0.1, 5)` on the URL it fetches — any snapshot that doesn't return within 5 s gets the entire detection cycle reported as "Failed to get image". Raising the timeout in the user's container was only a workaround; every Obico container would have the same ceiling. Fixed by flipping the flow around: the detection loop now captures the JPEG *locally* (with a long 20 s timeout that we control), stashes the bytes under a random 32-byte single-use nonce, and hands Obico's ML API a new `/api/v1/obico/cached-frame/{nonce}` URL that returns the pre-captured bytes instantly. Obico's 5 s timeout no longer races the capture pipeline — its fetch is a pure in-memory lookup. The nonce is the credential (URL-safe, 256 bits of entropy), single-use (popped on read), and expires in 30 s, so the endpoint can be unauthenticated without widening the camera access surface. External cameras (MJPEG/RTSP/HTTP snapshot) are captured via the same `capture_frame` helper used by the snapshot endpoint. Thanks to @fblix for the detailed reproducer with the exact timeout numbers.
 - **Direct Print from Library Not Attributed to User** — Clicking the Print button on a library file dispatched the job with no `created_by_id`, so the resulting archive had no owner and the print didn't show up in per-user statistics. The Queue and Reprint paths already forwarded the authenticated user; the library `POST /files/{file_id}/print` endpoint now does the same, reading the user from the JWT and passing it through to the dispatcher so direct prints are attributed like queued and reprinted ones.
 - **Add/Edit Printer Modal Clipped on Short Viewports** ([#964](https://github.com/maziggy/bambuddy/issues/964)) — On short or zoomed-in browser windows, the Add Printer and Edit Printer dialogs exceeded the viewport height with no scroll, hiding the lower fields (Access Code, Model, Location) and the Save button. Users had to zoom the browser out to complete the form. The modal overlay now scrolls and the card caps at `calc(100vh - 2rem)` with internal overflow so every field stays reachable regardless of viewport height. Thanks to @MartinNYHC for reporting.
 - **AMS Drying Silently Does Nothing** ([#971](https://github.com/maziggy/bambuddy/issues/971)) — Clicking Start Drying on a supported printer (e.g. P1S with AMS 2 Pro) could publish the MQTT command successfully but leave the AMS idle with no UI feedback. Two issues: (1) the firmware rejects the command when `dry_sf_reason` reports a blocking state (most commonly code 8 — AMS 2 Pro external power adapter not plugged in — but also "AMS busy", "already drying", etc.), and Bambuddy parsed that array but never surfaced it to the user; (2) the payload sent `filament: ""`, which some firmwares treat as an invalid-field refusal. The `/drying/start` endpoint now inspects the live `dry_sf_reason` for the target AMS unit and returns a descriptive 409 (e.g. "Plug in the external AMS power adapter to start drying") instead of silently publishing, and backfills an empty `filament` from the first loaded tray's type (defaulting to `PLA`) so the printer never rejects the command for a missing field. Thanks to @MartinNYHC for reporting.

backend/app/api/routes/obico.py

@@ -2,13 +2,13 @@
 
 import logging
 
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException, Response
 from pydantic import BaseModel
 
 from backend.app.core.auth import RequirePermissionIfAuthEnabled
 from backend.app.core.permissions import Permission
 from backend.app.models.user import User
-from backend.app.services.obico_detection import obico_detection_service
+from backend.app.services.obico_detection import obico_detection_service, pop_frame
 
 logger = logging.getLogger(__name__)
 
@@ -46,3 +46,24 @@ async def test_connection(
     if not req.url:
         return {"ok": False, "status_code": None, "body": None, "error": "URL is empty"}
     return await obico_detection_service.test_connection(req.url)
+
+
+@router.get("/cached-frame/{nonce}")
+async def cached_frame(nonce: str):
+    """Serve a pre-captured JPEG to the Obico ML API.
+
+    The detection loop captures a snapshot locally (where we control the timeout),
+    stashes the bytes under a one-shot random nonce, then hands this URL to Obico's
+    ML API. Obico's hardcoded 5s read timeout never races our snapshot pipeline.
+
+    Unauthenticated: the unguessable 32-byte nonce is single-use and expires in
+    seconds, so exposing this path doesn't widen the camera access surface.
+    """
+    data = await pop_frame(nonce)
+    if data is None:
+        raise HTTPException(status_code=404, detail="Frame not found or expired")
+    return Response(
+        content=data,
+        media_type="image/jpeg",
+        headers={"Cache-Control": "no-store"},
+    )

backend/app/main.py

@@ -4272,6 +4272,9 @@ PUBLIC_API_PATTERNS = [
     # Camera (streams loaded via <img> tag)
     "/camera/stream",  # /printers/{id}/camera/stream
     "/camera/snapshot",  # /printers/{id}/camera/snapshot
+    # Obico ML API fetches JPEG frames by one-shot nonce (issue #172 follow-up).
+    # The nonce itself is the credential: 32-byte random, single-use, ~30s TTL.
+    "/obico/cached-frame/",  # /obico/cached-frame/{nonce}
     # Slicer token-authenticated downloads — protocol handlers (bambustudioopen://,
     # orcaslicer://) cannot send auth headers. These endpoints validate a short-lived
     # download token in the URL path instead.

backend/app/services/obico_detection.py

@@ -10,18 +10,16 @@ See `obico_smoothing.py` for the per-print EWM + rolling-mean math.
 import asyncio
 import json
 import logging
+import secrets
+import time
 from collections import deque
-from datetime import datetime, timedelta, timezone
-from urllib.parse import urlencode
+from datetime import datetime, timezone
 
 import httpx
 from sqlalchemy import select
 
-from backend.app.core.auth import (
-    CAMERA_STREAM_TOKEN_EXPIRE_MINUTES,
-    create_camera_stream_token,
-)
 from backend.app.core.database import async_session
+from backend.app.models.printer import Printer
 from backend.app.models.settings import Settings
 from backend.app.services.obico_smoothing import (
     PrintState,
@@ -35,6 +33,45 @@ logger = logging.getLogger(__name__)
 HISTORY_MAX = 50
 HEALTH_TIMEOUT = 5.0
 DETECTION_TIMEOUT = 30.0
+SNAPSHOT_CAPTURE_TIMEOUT = 20  # seconds — we control this, not Obico
+FRAME_CACHE_TTL = 30.0  # seconds — Obico usually fetches within 1s of receiving the URL
+
+# Module-level one-shot frame cache. Obico's ML API has a hardcoded 5s read timeout
+# on the URL it fetches, which our /camera/snapshot endpoint can exceed (RTSP keyframe
+# wait + ffmpeg startup on cold calls). We capture locally first, stash the JPEG under
+# a random nonce, and hand Obico a URL that serves the cached bytes instantly.
+_frame_cache: dict[str, tuple[bytes, float]] = {}
+_frame_cache_lock = asyncio.Lock()
+
+
+def _prune_frame_cache() -> None:
+    """Drop entries older than FRAME_CACHE_TTL. Called under the cache lock."""
+    now = time.monotonic()
+    expired = [k for k, (_b, ts) in _frame_cache.items() if now - ts > FRAME_CACHE_TTL]
+    for k in expired:
+        _frame_cache.pop(k, None)
+
+
+async def stash_frame(data: bytes) -> str:
+    """Store JPEG bytes and return a URL-safe nonce that serves them once."""
+    nonce = secrets.token_urlsafe(32)
+    async with _frame_cache_lock:
+        _prune_frame_cache()
+        _frame_cache[nonce] = (data, time.monotonic())
+    return nonce
+
+
+async def pop_frame(nonce: str) -> bytes | None:
+    """Return and remove a cached frame by nonce; None if missing or expired."""
+    async with _frame_cache_lock:
+        _prune_frame_cache()
+        entry = _frame_cache.pop(nonce, None)
+    if entry is None:
+        return None
+    data, ts = entry
+    if time.monotonic() - ts > FRAME_CACHE_TTL:
+        return None
+    return data
 
 
 class ObicoDetectionService:
@@ -53,10 +90,6 @@ class ObicoDetectionService:
         # Global detection event log (most-recent-first)
         self._history: deque = deque(maxlen=HISTORY_MAX)
         self._last_error: str | None = None
-        # Cached camera-stream token so the ML API can fetch snapshots when
-        # auth is enabled. Refreshed before expiry; harmless when auth is off.
-        self._snapshot_token: str | None = None
-        self._snapshot_token_expires_at: datetime | None = None
 
     # ---- lifecycle ----
 
@@ -72,26 +105,6 @@ class ObicoDetectionService:
             self._task = None
             logger.info("Stopped Obico detection service")
 
-    # ---- snapshot auth ----
-
-    async def _get_snapshot_token(self) -> str:
-        """Return a valid camera-stream token, refreshing it before expiry.
-
-        The ML API fetches the snapshot URL directly, so when Bambuddy's auth
-        is enabled the URL must carry a token (same scheme used by <img>-based
-        camera consumers). When auth is disabled the token is simply ignored.
-        """
-        now = datetime.now(timezone.utc)
-        refresh_before = timedelta(minutes=5)
-        if (
-            self._snapshot_token is None
-            or self._snapshot_token_expires_at is None
-            or self._snapshot_token_expires_at - now <= refresh_before
-        ):
-            self._snapshot_token = await create_camera_stream_token()
-            self._snapshot_token_expires_at = now + timedelta(minutes=CAMERA_STREAM_TOKEN_EXPIRE_MINUTES)
-        return self._snapshot_token
-
     # ---- settings ----
 
     async def _load_settings(self) -> dict:
@@ -172,6 +185,35 @@ class ObicoDetectionService:
 
             await self._check_printer(printer_id, status, settings)
 
+    async def _capture_frame(self, printer_id: int) -> bytes | None:
+        """Capture one JPEG frame from the printer camera. Returns None on failure.
+
+        Uses a long local timeout because we control it — Obico's ML API never
+        waits on the slow path (it fetches from the nonce-cached URL).
+        """
+        # Late import to avoid cycles at module load time
+        from backend.app.services.camera import capture_camera_frame_bytes
+        from backend.app.services.external_camera import capture_frame as capture_external_frame
+
+        async with async_session() as db:
+            printer = await db.get(Printer, printer_id)
+        if printer is None:
+            self._last_error = f"Printer {printer_id} not found"
+            return None
+
+        if printer.external_camera_enabled and printer.external_camera_url:
+            return await capture_external_frame(
+                printer.external_camera_url,
+                printer.external_camera_type,
+                timeout=SNAPSHOT_CAPTURE_TIMEOUT,
+            )
+        return await capture_camera_frame_bytes(
+            ip_address=printer.ip_address,
+            access_code=printer.access_code,
+            model=printer.model,
+            timeout=SNAPSHOT_CAPTURE_TIMEOUT,
+        )
+
     async def _check_printer(self, printer_id: int, status, settings: dict):
         task_name = getattr(status, "task_name", None) or getattr(status, "subtask_name", "") or ""
         key = f"{task_name}"
@@ -180,10 +222,18 @@ class ObicoDetectionService:
             self._state_keys[printer_id] = key
             self._action_fired[printer_id] = False
 
-        token = await self._get_snapshot_token()
-        snapshot_url = (
-            f"{settings['external_url']}/api/v1/printers/{printer_id}/camera/snapshot?{urlencode({'token': token})}"
-        )
+        # Capture locally first, then hand Obico a nonce URL that returns the
+        # cached bytes instantly. Obico's ML API has a hardcoded 5s read timeout
+        # which would otherwise race our /camera/snapshot endpoint's keyframe wait.
+        frame = await self._capture_frame(printer_id)
+        if not frame:
+            self._last_error = f"Failed to capture snapshot for printer {printer_id}"
+            logger.warning(self._last_error)
+            return
+
+        # secrets.token_urlsafe() already produces a URL-safe path segment.
+        nonce = await stash_frame(frame)
+        snapshot_url = f"{settings['external_url']}/api/v1/obico/cached-frame/{nonce}"
         ml_url = f"{settings['ml_url']}/p/"
 
         try:

backend/tests/integration/test_obico_api.py

@@ -0,0 +1,71 @@
+"""Integration tests for Obico API endpoints (#172 follow-up).
+
+Verifies the /obico/cached-frame/{nonce} endpoint used by Obico's ML API to fetch
+pre-captured JPEG frames. This endpoint lets the detection loop sidestep Obico's
+hardcoded 5s read timeout by pre-populating a cache before issuing the ML call.
+"""
+
+import pytest
+from httpx import AsyncClient
+
+from backend.app.services.obico_detection import _frame_cache, stash_frame
+
+FAKE_JPEG = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xd9"
+
+
+@pytest.fixture(autouse=True)
+def clear_cache():
+    _frame_cache.clear()
+    yield
+    _frame_cache.clear()
+
+
+class TestObicoCachedFrame:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_valid_nonce_returns_jpeg(self, async_client: AsyncClient):
+        """A stashed nonce returns the stored JPEG bytes with image/jpeg."""
+        nonce = await stash_frame(FAKE_JPEG)
+        response = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
+        assert response.status_code == 200
+        assert response.headers["content-type"] == "image/jpeg"
+        assert response.content == FAKE_JPEG
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_unknown_nonce_is_404(self, async_client: AsyncClient):
+        """An unguessable URL must not leak that the endpoint exists — return 404."""
+        response = await async_client.get("/api/v1/obico/cached-frame/definitely-not-a-real-nonce")
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_nonce_is_single_use(self, async_client: AsyncClient):
+        """A second fetch with the same nonce returns 404 — prevents replay."""
+        nonce = await stash_frame(FAKE_JPEG)
+        first = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
+        assert first.status_code == 200
+        second = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
+        assert second.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_endpoint_is_public(self, async_client: AsyncClient):
+        """Obico's ML API can't send auth headers, so the nonce IS the credential.
+        The path must be in PUBLIC_API_PATTERNS (no auth wall)."""
+        nonce = await stash_frame(FAKE_JPEG)
+        # Intentionally omit any auth headers even if the fixture would normally inject them
+        response = await async_client.get(
+            f"/api/v1/obico/cached-frame/{nonce}",
+            headers={},  # no Authorization header
+        )
+        assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_response_is_not_cached(self, async_client: AsyncClient):
+        """Browsers/proxies must not hold onto the image after Obico consumes it."""
+        nonce = await stash_frame(FAKE_JPEG)
+        response = await async_client.get(f"/api/v1/obico/cached-frame/{nonce}")
+        assert response.status_code == 200
+        assert "no-store" in response.headers.get("cache-control", "")

backend/tests/unit/test_obico_detection.py

@@ -5,9 +5,17 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from backend.app.schemas.settings import AppSettingsUpdate
-from backend.app.services.obico_detection import ObicoDetectionService
+from backend.app.services.obico_detection import (
+    FRAME_CACHE_TTL,
+    ObicoDetectionService,
+    _frame_cache,
+    pop_frame,
+    stash_frame,
+)
 from backend.app.services.obico_smoothing import WARMUP_FRAMES
 
+FAKE_JPEG = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00\xff\xd9"
+
 
 class TestSettingsSchemaValidators:
     """Guard rails on the new obico_* AppSettings fields."""
@@ -145,10 +153,7 @@ class TestPollOneStateLifecycle:
 
         with (
             patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
-            patch(
-                "backend.app.services.obico_detection.create_camera_stream_token",
-                new=AsyncMock(return_value="tok"),
-            ),
+            patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
         ):
             await svc._check_printer(1, status, settings)
 
@@ -178,10 +183,7 @@ class TestPollOneStateLifecycle:
 
         with (
             patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
-            patch(
-                "backend.app.services.obico_detection.create_camera_stream_token",
-                new=AsyncMock(return_value="tok"),
-            ),
+            patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
         ):
             await svc._check_printer(1, status, settings)
 
@@ -224,13 +226,140 @@ class TestPollOneStateLifecycle:
         with (
             patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
             patch("backend.app.services.obico_actions.execute_action", new=AsyncMock()) as mock_action,
-            patch(
-                "backend.app.services.obico_detection.create_camera_stream_token",
-                new=AsyncMock(return_value="tok"),
-            ),
+            patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
         ):
             await svc._check_printer(1, status, settings)
             assert mock_action.call_count == 1
             await svc._check_printer(1, status, settings)
             # Second call must not dispatch again
             assert mock_action.call_count == 1
+
+
+class TestFrameCache:
+    """One-shot JPEG cache that lets us sidestep Obico's 5s read timeout.
+
+    Obico's ML API fetches snapshots via `GET /p/?img=URL` with `timeout=(0.1, 5)`.
+    Our /camera/snapshot can exceed that on cold calls (RTSP keyframe wait). So the
+    detection loop captures locally, stashes the JPEG bytes under a nonce, then hands
+    Obico a URL that returns those bytes instantly. The cache is single-use + TTLed
+    so a leaked nonce can't be replayed.
+    """
+
+    def setup_method(self):
+        _frame_cache.clear()
+
+    @pytest.mark.asyncio
+    async def test_stash_and_pop_roundtrip(self):
+        nonce = await stash_frame(FAKE_JPEG)
+        assert nonce  # non-empty URL-safe token
+        data = await pop_frame(nonce)
+        assert data == FAKE_JPEG
+
+    @pytest.mark.asyncio
+    async def test_nonce_is_single_use(self):
+        nonce = await stash_frame(FAKE_JPEG)
+        assert await pop_frame(nonce) == FAKE_JPEG
+        # Second pop returns None — caches replay protection
+        assert await pop_frame(nonce) is None
+
+    @pytest.mark.asyncio
+    async def test_unknown_nonce_returns_none(self):
+        assert await pop_frame("not-a-real-nonce") is None
+
+    @pytest.mark.asyncio
+    async def test_stash_produces_unique_nonces(self):
+        nonces = {await stash_frame(FAKE_JPEG) for _ in range(10)}
+        assert len(nonces) == 10
+
+    @pytest.mark.asyncio
+    async def test_expired_entries_are_pruned_on_stash(self):
+        """New entries trigger pruning of TTL-expired ones — prevents unbounded growth."""
+        # Manually seed an entry with a stale timestamp
+        import time as time_module
+
+        _frame_cache["stale-nonce"] = (FAKE_JPEG, time_module.monotonic() - FRAME_CACHE_TTL - 1)
+        await stash_frame(FAKE_JPEG)
+        # Stale entry was pruned
+        assert "stale-nonce" not in _frame_cache
+
+    @pytest.mark.asyncio
+    async def test_pop_rejects_expired_nonce(self):
+        """Even if the entry is still in the dict, an expired TTL returns None."""
+        import time as time_module
+
+        _frame_cache["aging-nonce"] = (FAKE_JPEG, time_module.monotonic() - FRAME_CACHE_TTL - 1)
+        assert await pop_frame("aging-nonce") is None
+
+
+class TestCheckPrinterUsesCachedFrameUrl:
+    """The URL sent to Obico must point at our nonce endpoint, not /camera/snapshot."""
+
+    def setup_method(self):
+        _frame_cache.clear()
+
+    @pytest.mark.asyncio
+    async def test_ml_api_called_with_cached_frame_url(self):
+        svc = ObicoDetectionService()
+        settings = {
+            "enabled": True,
+            "ml_url": "http://obico:3333",
+            "sensitivity": "medium",
+            "action": "notify",
+            "poll_interval": 10,
+            "enabled_printers": None,
+            "external_url": "http://bambuddy:8000",
+        }
+        status = MagicMock(state="RUNNING", task_name="job", subtask_name="")
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"detections": []}
+        mock_response.raise_for_status = MagicMock()
+        mock_client = MagicMock()
+        mock_client.get = AsyncMock(return_value=mock_response)
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
+            patch.object(svc, "_capture_frame", new=AsyncMock(return_value=FAKE_JPEG)),
+        ):
+            await svc._check_printer(1, status, settings)
+
+        # Extract the img URL handed to the ML API
+        _args, kwargs = mock_client.get.call_args
+        img_url = kwargs["params"]["img"]
+        assert img_url.startswith("http://bambuddy:8000/api/v1/obico/cached-frame/")
+        # The path segment after /cached-frame/ is the nonce itself — that nonce must
+        # resolve back to our stashed frame (single-use guarantees freshness).
+        nonce = img_url.rsplit("/", 1)[-1]
+        assert await pop_frame(nonce) == FAKE_JPEG
+
+    @pytest.mark.asyncio
+    async def test_capture_failure_skips_ml_call(self):
+        """If we can't capture a frame, don't bother the ML API."""
+        svc = ObicoDetectionService()
+        settings = {
+            "enabled": True,
+            "ml_url": "http://obico:3333",
+            "sensitivity": "medium",
+            "action": "notify",
+            "poll_interval": 10,
+            "enabled_printers": None,
+            "external_url": "http://bambuddy:8000",
+        }
+        status = MagicMock(state="RUNNING", task_name="job", subtask_name="")
+
+        mock_client = MagicMock()
+        mock_client.get = AsyncMock()
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("backend.app.services.obico_detection.httpx.AsyncClient", return_value=mock_client),
+            patch.object(svc, "_capture_frame", new=AsyncMock(return_value=None)),
+        ):
+            await svc._check_printer(1, status, settings)
+
+        mock_client.get.assert_not_called()
+        assert svc._last_error is not None
+        assert "Failed to capture snapshot" in svc._last_error