fix(obico): distinguish missing verdicts

backend/app/api/routes/obico.py

@@ -16,7 +16,7 @@
 
 
 class TestConnectionRequest(BaseModel):
-    url: str
+    url: str | None = None
     token: str | None = None
 
 
@@ -38,19 +38,42 @@ async def get_status(
     }
 
 
+@router.get("/printer-status")
+async def get_printer_status(
+    user: User | None = RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
+):
+    """Return live card states without exposing Obico configuration."""
+    settings = await obico_detection_service._load_settings()
+    enabled_printers = settings["enabled_printers"]
+    can_see_error = user is None or user.has_permission(Permission.SETTINGS_READ.value)
+    per_printer = obico_detection_service.get_per_printer()
+    if not can_see_error:
+        per_printer = {printer_id: {**entry, "error": None} for printer_id, entry in per_printer.items()}
+    return {
+        "enabled": settings["enabled"],
+        "monitored_printers": sorted(enabled_printers) if enabled_printers is not None else None,
+        "per_printer": per_printer,
+        "last_error": obico_detection_service._last_error if can_see_error else None,
+    }
+
+
 @router.post("/test-connection")
 async def test_connection(
     req: TestConnectionRequest,
     _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
 ):
-    """Ping the Obico ML API health endpoint and check token acceptance."""
-    if not req.url:
-        return {"ok": False, "status_code": None, "body": None, "error": "URL is empty", "auth_ok": None}
+    """Ping an explicit configuration, or the saved one when values are omitted."""
+    url = req.url
     token = req.token
-    if token is None:
+    if url is None or token is None:
         settings = await obico_detection_service._load_settings()
-        token = settings.get("ml_token") or ""
-    return await obico_detection_service.test_connection(req.url, token)
+        if url is None:
+            url = settings.get("ml_url") or ""
+        if token is None:
+            token = settings.get("ml_token") or ""
+    if not url:
+        return {"ok": False, "status_code": None, "body": None, "error": "URL is empty", "auth_ok": None}
+    return await obico_detection_service.test_connection(url, token)
 
 
 @router.get("/cached-frame/{nonce}")

backend/app/services/obico_detection.py

@@ -90,6 +90,8 @@ def __init__(self):
         self._state_keys: dict[int, str] = {}
         # printer_id -> last classification ("safe"/"warning"/"failure")
         self._last_class: dict[int, str] = {}
+        # printer_id -> why the latest poll produced no verdict
+        self._errors: dict[int, str] = {}
         # printer_id -> whether an action has already been fired for the current print
         self._action_fired: dict[int, bool] = {}
         # Global detection event log (most-recent-first)
@@ -177,21 +179,29 @@ async def _loop(self):
                 self._last_error = str(e) or type(e).__name__
                 await asyncio.sleep(30)
 
+    def _clear_printer_state(self, printer_id: int) -> None:
+        """Discard cached state when a printer can no longer be checked."""
+        self._states.pop(printer_id, None)
+        self._state_keys.pop(printer_id, None)
+        self._action_fired.pop(printer_id, None)
+        self._last_class.pop(printer_id, None)
+        self._errors.pop(printer_id, None)
+
     async def _poll_once(self, settings: dict):
         # Late import to avoid cycles at module load time
         from backend.app.services.printer_manager import printer_manager
 
         statuses = printer_manager.get_all_statuses()
         for printer_id, status in list(statuses.items()):
             if settings["enabled_printers"] is not None and printer_id not in settings["enabled_printers"]:
+                self._clear_printer_state(printer_id)
                 continue
             if not printer_manager.is_connected(printer_id):
+                self._clear_printer_state(printer_id)
                 continue
             if not status or getattr(status, "state", None) != "RUNNING":
                 # Reset state when not printing so the next print starts fresh
-                self._states.pop(printer_id, None)
-                self._state_keys.pop(printer_id, None)
-                self._action_fired.pop(printer_id, None)
+                self._clear_printer_state(printer_id)
                 continue
 
             await self._check_printer(printer_id, status, settings)
@@ -247,31 +257,38 @@ async def _capture_frame(self, printer_id: int) -> bytes | None:
             timeout=SNAPSHOT_CAPTURE_TIMEOUT,
         )
 
+    def _no_verdict(self, printer_id: int, reason: str) -> None:
+        """Record a failed poll without reusing an earlier successful verdict."""
+        self._errors[printer_id] = reason
+        self._last_error = reason
+        logger.warning(reason)
+
     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}"
         if self._state_keys.get(printer_id) != key:
             self._states[printer_id] = PrintState()
             self._state_keys[printer_id] = key
             self._action_fired[printer_id] = False
+            self._last_class.pop(printer_id, None)
+            self._errors.pop(printer_id, None)
 
         # Capture locally first, then hand Obico a nonce URL that returns the
         # cached bytes instantly. Obico's ML API is GET-only (/p/?img=URL) with a
         # hardcoded 5s read timeout which would otherwise race our /camera/snapshot
         # 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)
+            self._no_verdict(printer_id, f"Failed to capture snapshot for printer {printer_id}")
             return
 
         external_url = settings.get("external_url") or ""
         if not external_url:
-            self._last_error = (
-                "external_url setting is empty — Obico's ML API needs a reachable URL to fetch the snapshot from. "
-                "Set Settings → General → External URL."
+            self._no_verdict(
+                printer_id,
+                "external_url (External URL) setting is empty — Obico's ML API needs a reachable URL to fetch the snapshot from. "
+                "Set Settings → General → External URL.",
             )
-            logger.warning(self._last_error)
             return
 
         nonce = await stash_frame(frame)
@@ -286,19 +303,18 @@ async def _check_printer(self, printer_id: int, status, settings: dict):
                     headers=auth_headers(settings.get("ml_token")),
                 )
                 if resp.status_code == 401:
-                    self._last_error = (
+                    reason = (
                         "Obico ML API rejected the token (401). Set Settings > Failure Detection > "
                         "ML API Token to the ML_API_TOKEN the server runs with, or clear ML_API_TOKEN "
                         "on the server."
                     )
-                    logger.warning("%s (printer %s)", self._last_error, printer_id)
+                    self._no_verdict(printer_id, reason)
                     return
                 resp.raise_for_status()
                 payload = resp.json()
         except Exception as e:
             detail = str(e) or type(e).__name__
-            self._last_error = f"ML API call failed for printer {printer_id}: {detail}"
-            logger.warning(self._last_error)
+            self._no_verdict(printer_id, f"ML API call failed for printer {printer_id}: {detail}")
             return
 
         detections = payload.get("detections", []) if isinstance(payload, dict) else []
@@ -310,6 +326,7 @@ async def _check_printer(self, printer_id: int, status, settings: dict):
         # A successful capture + ML call clears any transient error from previous
         # polls (typical case: cold-start RTSP timeout on first frame after startup,
         # followed by healthy polls that otherwise leave the banner stuck in the UI).
+        self._errors.pop(printer_id, None)
         self._last_error = None
 
         # Log every non-safe sample — safe samples would flood history
@@ -348,6 +365,19 @@ async def _dispatch_action(self, printer_id: int, action: str, task_name: str, s
 
     # ---- queries ----
 
+    def get_per_printer(self) -> dict:
+        """Return honest live states for printers with an active monitored print."""
+        result = {}
+        for printer_id, state in self._states.items():
+            error = self._errors.get(printer_id)
+            result[printer_id] = {
+                "class": "error" if error else self._last_class.get(printer_id, "unknown"),
+                "frame_count": state.frame_count,
+                "score": round(state.ewm_mean, 4),
+                "error": error,
+            }
+        return result
+
     def get_status(self, sensitivity: str = "medium") -> dict:
         # Report the thresholds for the configured sensitivity, not a hardcoded
         # "medium" — otherwise the Status panel always shows the medium row
@@ -357,14 +387,7 @@ def get_status(self, sensitivity: str = "medium") -> dict:
         return {
             "is_running": self._task is not None and not self._task.done(),
             "last_error": self._last_error,
-            "per_printer": {
-                pid: {
-                    "class": self._last_class.get(pid, "safe"),
-                    "frame_count": state.frame_count,
-                    "score": round(state.ewm_mean, 4),
-                }
-                for pid, state in self._states.items()
-            },
+            "per_printer": self.get_per_printer(),
             "thresholds": {"low": low, "high": high},
             "history": list(self._history),
         }

backend/tests/integration/test_issue_141_obico_api.py

@@ -0,0 +1,95 @@
+"""API regressions for Obico printer verdicts and permission redaction (#141)."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from httpx import AsyncClient
+
+from backend.app.api.routes.obico import (
+    TestConnectionRequest as ObicoTestConnectionRequest,
+    get_printer_status,
+    test_connection as obico_test_connection,
+)
+from backend.app.services.obico_detection import obico_detection_service
+from backend.app.services.obico_smoothing import PrintState
+
+
+@pytest.fixture(autouse=True)
+def clear_detection_state():
+    obico_detection_service._states.clear()
+    obico_detection_service._last_class.clear()
+    obico_detection_service._errors.clear()
+    obico_detection_service._last_error = None
+    yield
+    obico_detection_service._states.clear()
+    obico_detection_service._last_class.clear()
+    obico_detection_service._errors.clear()
+    obico_detection_service._last_error = None
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_printer_status_exposes_error_class_and_reason_without_reporting_safe(
+    async_client: AsyncClient,
+):
+    obico_detection_service._states[1] = PrintState()
+    obico_detection_service._errors[1] = "Obico ML API rejected the token (401)."
+    loaded = {"enabled": True, "enabled_printers": None}
+
+    with patch.object(
+        obico_detection_service,
+        "_load_settings",
+        new=AsyncMock(return_value=loaded),
+    ):
+        response = await async_client.get("/api/v1/obico/printer-status")
+
+    assert response.status_code == 200
+    entry = response.json()["per_printer"]["1"]
+    assert entry["class"] == "error"
+    assert entry["error"] == "Obico ML API rejected the token (401)."
+
+
+@pytest.mark.asyncio
+async def test_printer_reader_without_settings_permission_gets_class_but_not_internal_reason():
+    obico_detection_service._states[1] = PrintState()
+    obico_detection_service._errors[1] = "ML API http://192.168.8.9:3333 refused"
+    obico_detection_service._last_error = obico_detection_service._errors[1]
+    user = MagicMock()
+    user.has_permission.return_value = False
+    loaded = {"enabled": True, "enabled_printers": None}
+
+    with patch.object(
+        obico_detection_service,
+        "_load_settings",
+        new=AsyncMock(return_value=loaded),
+    ):
+        data = await get_printer_status(user=user)
+
+    assert data["per_printer"][1]["class"] == "error"
+    assert data["per_printer"][1]["error"] is None
+    assert data["last_error"] is None
+    assert "192.168.8.9" not in str(data)
+
+
+@pytest.mark.asyncio
+async def test_connection_without_payload_values_checks_saved_configuration():
+    loaded = {
+        "ml_url": "http://saved-obico:3333",
+        "ml_token": "saved-token",
+    }
+    with (
+        patch.object(
+            obico_detection_service,
+            "_load_settings",
+            new=AsyncMock(return_value=loaded),
+        ),
+        patch.object(
+            obico_detection_service,
+            "test_connection",
+            new=AsyncMock(return_value={"ok": True}),
+        ) as probe,
+    ):
+        result = await obico_test_connection(ObicoTestConnectionRequest())
+
+    assert result == {"ok": True}
+    probe.assert_awaited_once_with("http://saved-obico:3333", "saved-token")