bambuddy/backend/app/services/octoeverywhere_detection.py

"""OctoEverywhere AI print-failure detection service.

Each print owns a context so OctoEverywhere can combine camera snapshots over
time. Snapshots follow the configured interval, sped up to the API's
recommended interval while it suggests faster inspection, always respecting its
minimum. The API decides when to warn or pause; raw model scores are not used
for classification or actions.
"""

import asyncio
import json
import logging
import math
import time
from collections import deque
from dataclasses import dataclass
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from functools import partial
from urllib.parse import urlsplit

import httpx
from sqlalchemy import select

from backend.app.core.database import async_session
from backend.app.models.settings import Settings
from backend.app.services.octoeverywhere_images import MAX_IMAGE_BYTES, prepare_frame

logger = logging.getLogger(__name__)

CREATE_CONTEXT_URL = "https://gadget-pv1-oeapi.octoeverywhere.com/api/gadget/v1/createcontext"
DETECTION_TIMEOUT = 30.0
POLL_INTERVAL = 5
DEFAULT_POLL_INTERVAL = 20
ERROR_RETRY_INTERVAL = 60
NOTIFICATION_REPEAT_INTERVAL = 3 * 60 * 60
HISTORY_MAX = 50
# Gadget's confidence levels: 1 suggests a warning or pause soonest, with more
# false positives; 5 waits until the model is most certain.
CONFIDENCE_LEVELS = {"lowest": 1, "low": 2, "medium": 3, "high": 4, "highest": 5}
IP_RESTRICTED_ERROR = "OE_API_KEY_IP_RESTRICTED"
ACCOUNT_ERROR_CODES = {
    "OE_INVALID_API_KEY",
    "OE_API_KEY_DISABLED",
    "OE_API_KEY_BLOCKED_PAYMENT_FAILED",
    IP_RESTRICTED_ERROR,
    "OE_FREE_USAGE_LIMIT_REACHED",
}

API_ERRORS = {
    "OE_BAD_ARGS": "OctoEverywhere rejected the request arguments.",
    "OE_ARGS_PARSE_FAILED": "OctoEverywhere could not parse the request.",
    "OE_CONTEXT_RATE_LIMITED": "OctoEverywhere requested a delay before the next check. Detection will retry automatically.",
    "OE_IMAGE_DECODE_FAILED": "OctoEverywhere could not decode the camera snapshot.",
    "OE_INVALID_API_KEY": "OctoEverywhere rejected the Gadget API key. Check your API key in AI Failure Detection settings.",
    "OE_API_KEY_DISABLED": "The OctoEverywhere Gadget API key is disabled. Contact OctoEverywhere support to restore access.",
    "OE_API_KEY_BLOCKED_PAYMENT_FAILED": ("The OctoEverywhere API account is unavailable. Check the account status."),
    IP_RESTRICTED_ERROR: (
        "This IP address is already in use by another OctoEverywhere Gadget API key. "
        "Use the original key, or contact OctoEverywhere support."
    ),
    "OE_FREE_USAGE_LIMIT_REACHED": "API usage limit reached. Set up billing to continue.",
    "OE_BACKEND_THROTTLED": "OctoEverywhere is temporarily busy. Detection will retry automatically.",
    "OE_INTERNAL_ERROR": "OctoEverywhere encountered a service error. Detection will retry automatically.",
}


class _ApiError(Exception):
    """An actionable error with no remote response bodies or request secrets."""

    def __init__(
        self,
        message: str,
        status_code: int | None = None,
        retry_after: float = 0,
        *,
        context_missing: bool = False,
        error_code: str | None = None,
    ):
        super().__init__(message)
        self.status_code = status_code
        self.retry_after = retry_after
        self.context_missing = context_missing
        self.error_code = _known_error_code(error_code)


@dataclass
class _Context:
    context_id: str
    process_url: str
    fallback_url: str


@dataclass
class _PrintState:
    print_id: str | None = None
    context: _Context | None = None
    next_check_at: float = 0
    last_processed_at: float | None = None
    interval: int | None = None
    recommended_interval: int | None = None  # Active only while faster inspections are suggested.
    frame_count: int = 0
    print_quality: int | None = None
    verdict: str = "unknown"
    error: str | None = None
    error_code: str | None = None
    failures: int = 0
    warning_fired: bool = False
    action_fired: bool = False
    last_notification_at: float | None = None


def _process_interval(server_interval: int | None, poll_interval: int, recommended_interval: int | None = None) -> int:
    """Use the configured cadence, sped up by an active timing hint, respecting the minimum."""
    # A faster-inspection hint may only shorten the interval the user chose.
    interval = poll_interval if recommended_interval is None else min(recommended_interval, poll_interval)
    return max(server_interval or interval, interval)


def _known_error_code(value) -> str | None:
    """Only expose documented, recognized codes, never arbitrary remote text."""
    return value if isinstance(value, str) and value in API_ERRORS else None


def _process_url(value) -> str:
    """Only send the API key to HTTPS endpoints owned by OctoEverywhere."""
    if not isinstance(value, str):
        raise _ApiError("OctoEverywhere returned an invalid processing URL.")
    try:
        parsed = urlsplit(value)
        valid = (
            parsed.scheme == "https"
            and parsed.hostname is not None
            and parsed.hostname.endswith(".octoeverywhere.com")
            and parsed.port in (None, 443)
            and parsed.username is None
            and parsed.password is None
            and not parsed.fragment
            and not any(character.isspace() for character in value)
        )
    except ValueError:
        valid = False
    if not valid:
        raise _ApiError("OctoEverywhere returned an invalid processing URL.")
    return value


def _retry_after(response: httpx.Response) -> float:
    value = response.headers.get("Retry-After", "")
    try:
        seconds = float(value)
    except ValueError:
        try:
            seconds = (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            return 0
    return max(0, seconds) if math.isfinite(seconds) else 0


def _response_payload(response: httpx.Response) -> dict:
    try:
        payload = response.json()
    except ValueError:
        payload = None
    if not 200 <= response.status_code < 300:
        error_type = payload.get("ErrorType") if isinstance(payload, dict) else None
        # The v1 server uses this specific 500 response for a missing context,
        # including one removed by its 14-day TTL. Other internal errors must
        # retain their context so a transient outage does not reset the model.
        context_missing = (
            response.status_code == 500
            and isinstance(payload, dict)
            and error_type == "OE_INTERNAL_ERROR"
            and payload.get("ErrorDetails") == "Failed to get context object."
        )
        message = API_ERRORS.get(error_type) if isinstance(error_type, str) else None
        if context_missing:
            message = "OctoEverywhere print context is no longer available. Detection will reconnect automatically."
        if not message:
            if response.status_code in (401, 403):
                message = API_ERRORS["OE_INVALID_API_KEY"]
                error_type = "OE_INVALID_API_KEY"
            elif response.status_code == 429:
                message = API_ERRORS["OE_CONTEXT_RATE_LIMITED"]
            else:
                message = f"OctoEverywhere API request failed (HTTP {response.status_code})."
        raise _ApiError(
            message,
            response.status_code,
            _retry_after(response),
            context_missing=context_missing,
            error_code=error_type,
        )
    if not isinstance(payload, dict):
        raise _ApiError("OctoEverywhere returned an invalid response.", response.status_code)
    return payload


class OctoEverywhereDetectionService:
    """Singleton service that uploads snapshots from monitored, running prints."""

    def __init__(self):
        self._task: asyncio.Task | None = None
        self._running = False
        self._checks: dict[int, asyncio.Task] = {}
        self._actions: set[asyncio.Task] = set()
        self._states: dict[int, _PrintState] = {}
        self._configuration: tuple[str, str] | None = None
        self._settings: dict | None = None
        self._settings_refresh_at = 0.0
        self._generation = 0
        self._blocked_api_key: str | None = None
        self._blocked_error_code: str | None = None
        self._history: deque = deque(maxlen=HISTORY_MAX)
        self._last_error: str | None = None
        self._last_error_code: str | None = None

    def _set_last_error(self, reason: str | None, error_code: str | None = None):
        if self._blocked_error_code:
            reason = API_ERRORS[self._blocked_error_code]
            error_code = self._blocked_error_code
        self._last_error = reason
        self._last_error_code = _known_error_code(error_code) if reason else None

    def _refresh_last_error(self):
        state = next((state for state in self._states.values() if state.error), None)
        self._set_last_error(state.error if state else None, state.error_code if state else None)

    def _block_account(self, api_key: str, error_code: str):
        """Account failures stop every printer until the user explicitly retries."""
        self._blocked_api_key = api_key
        self._blocked_error_code = error_code
        self._generation += 1
        for task in self._checks.values():
            if task is not asyncio.current_task():
                task.cancel()
        self._refresh_last_error()
        logger.warning("OctoEverywhere inspections stopped: %s", API_ERRORS[error_code])

    def _clear_account_error(self):
        # Keep contexts, action tracking, and reserved server deadlines intact.
        self._blocked_api_key = None
        self._blocked_error_code = None
        self._refresh_last_error()

    async def start(self):
        self._running = True
        if self._task is not None and not self._task.done():
            return
        logger.info("Starting OctoEverywhere detection service")
        self._task = asyncio.create_task(self._loop())

    def stop(self):
        self._running = False
        if self._task:
            self._task.cancel()
            self._task = None
            logger.info("Stopped OctoEverywhere detection service")
        for task in self._checks.values():
            task.cancel()
        self._checks.clear()
        for task in self._actions:
            task.cancel()
        self._actions.clear()
        self._states.clear()
        self._generation += 1
        self._configuration = None
        self._settings = None
        self._clear_account_error()
        self._set_last_error(None)

    def reset_printer(self, printer_id: int):
        """Discard a completed print, including any inference still in flight."""
        task = self._checks.pop(printer_id, None)
        if task is not None and task is not asyncio.current_task():
            task.cancel()
        self._states.pop(printer_id, None)
        self._refresh_last_error()

    def forget_printer(self, printer_id: int) -> None:
        """Forget a deleted printer and reload its committed monitoring limits."""
        self.reset_printer(printer_id)
        self._history = deque(
            (entry for entry in self._history if entry["printer_id"] != printer_id), maxlen=HISTORY_MAX
        )
        self._generation += 1
        self._settings = None
        self._settings_refresh_at = 0

    # ---- settings ----

    async def refresh_settings(self):
        """Invalidate in-flight decisions immediately after settings are saved."""
        self._generation += 1
        self._settings = None
        generation = self._generation
        task = self._task
        self._task = None
        try:
            if task is not None:
                task.cancel()
                await asyncio.gather(task, return_exceptions=True)
            # Cancel stale inspections, but let already committed actions finish.
            checks = [task for task in self._checks.values() if task is not asyncio.current_task()]
            for task in checks:
                task.cancel()
            if checks:
                await asyncio.gather(*checks, return_exceptions=True)
            settings = await self._load_settings()
            if self._generation == generation:
                self._cache_settings(settings)
                self._apply_settings(settings)
        finally:
            # A failed/cancelled refresh must not permanently stop detection.
            # Lifecycle intent is separate from settings generations so stop()
            # during this refresh cannot accidentally resurrect the scheduler.
            if self._running and self._task is None:
                await self.start()

    def _cache_settings(self, settings: dict):
        self._settings = settings
        self._settings_refresh_at = time.monotonic() + settings["poll_interval"]

    def _apply_settings(self, settings: dict):
        configuration = (settings["api_key"], settings["confidence"])
        if self._configuration != configuration:
            self._generation += 1
            for task in self._checks.values():
                if task is not asyncio.current_task():
                    task.cancel()
            if settings["api_key"] != self._blocked_api_key:
                self._clear_account_error()
            # A settings edit is still the same print. Replace its remote
            # context, but keep server pacing and actions already performed.
            for state in self._states.values():
                state.context = None
                state.recommended_interval = None
                state.frame_count = 0
                state.print_quality = None
                state.verdict = "unknown"
                state.error = None
                state.error_code = None
                state.failures = 0
            self._configuration = configuration
            self._set_last_error(None)
        if not settings["enabled"] or not settings["api_key"]:
            for printer_id in self._states.keys() | self._checks.keys():
                self.reset_printer(printer_id)
            self._set_last_error(None)
        if settings["enabled"] and not settings["api_key"]:
            self._set_last_error(
                "An OctoEverywhere Gadget API key is required. Add a key in Failure Detection settings."
            )
        if settings["enabled_printers"] is not None:
            for printer_id in self._states.keys() | self._checks.keys():
                if printer_id not in settings["enabled_printers"]:
                    self.reset_printer(printer_id)

    async def _load_settings(self) -> dict:
        keys = [
            "obico_enabled",
            "octoeverywhere_enabled",
            "octoeverywhere_api_key",
            "octoeverywhere_confidence",
            "octoeverywhere_action",
            "octoeverywhere_enabled_printers",
            "octoeverywhere_poll_interval",
        ]
        async with async_session() as db:
            result = await db.execute(select(Settings).where(Settings.key.in_(keys)))
            rows = {r.key: r.value for r in result.scalars().all()}

        enabled_printers_raw = rows.get("octoeverywhere_enabled_printers", "")
        enabled_printers = None
        if enabled_printers_raw:
            try:
                printer_ids = json.loads(enabled_printers_raw)
                enabled_printers = (
                    set(printer_ids)
                    if isinstance(printer_ids, list) and all(type(pid) is int for pid in printer_ids)
                    else set()
                )
            except (TypeError, ValueError):
                enabled_printers = set()

        try:
            poll_interval = int(rows.get("octoeverywhere_poll_interval", DEFAULT_POLL_INTERVAL))
            if not 5 <= poll_interval <= 30:
                poll_interval = DEFAULT_POLL_INTERVAL
        except (TypeError, ValueError):
            poll_interval = DEFAULT_POLL_INTERVAL

        return {
            # Restores can bypass PUT /settings and leave both flags enabled.
            # Refuse to act while the other provider is enabled as well.
            "enabled": rows.get("octoeverywhere_enabled", "false").lower() == "true"
            and rows.get("obico_enabled", "false").lower() != "true",
            "api_key": (rows.get("octoeverywhere_api_key") or "").strip(),
            "confidence": rows.get("octoeverywhere_confidence", "medium"),
            "action": rows.get("octoeverywhere_action", "notify"),
            "enabled_printers": enabled_printers,
            "poll_interval": poll_interval,
        }

    # ---- main loop ----

    async def _loop(self):
        while True:
            try:
                # Printer deadlines need one-second resolution, but settings
                # only need the configured polling cadence or an explicit refresh.
                if self._settings is None or time.monotonic() >= self._settings_refresh_at:
                    generation = self._generation
                    settings = await self._load_settings()
                    if generation != self._generation:
                        continue
                    self._cache_settings(settings)
                else:
                    settings = self._settings
                await self._poll_once(settings)
                # Enabled printers need whole-second scheduling for values such
                # as 21 seconds; disabled detection can check settings less often.
                await asyncio.sleep(1 if settings["enabled"] else POLL_INTERVAL)
            except asyncio.CancelledError:
                break
            except Exception as exc:
                # Never include exception text: transport errors can contain
                # request URLs and user-supplied credentials.
                self._set_last_error("OctoEverywhere detection is temporarily unavailable. Retrying automatically.")
                logger.error("OctoEverywhere detection loop error: %s", type(exc).__name__)
                await asyncio.sleep(ERROR_RETRY_INTERVAL)

    async def _poll_once(self, settings: dict):
        from backend.app.services.printer_manager import printer_manager

        self._apply_settings(settings)
        if not settings["enabled"] or not settings["api_key"]:
            return

        statuses = printer_manager.get_all_statuses()
        for printer_id in self._states.keys() | self._checks.keys():
            if printer_id not in statuses:
                self.reset_printer(printer_id)

        for printer_id, status in list(statuses.items()):
            if settings["enabled_printers"] is not None and printer_id not in settings["enabled_printers"]:
                self.reset_printer(printer_id)
                continue
            if not printer_manager.is_connected(printer_id):
                task = self._checks.get(printer_id)
                if task:
                    task.cancel()
                state = self._states.get(printer_id)
                if state:
                    self._no_verdict(printer_id, state, "Printer is disconnected. Detection will resume on reconnect.")
                continue
            print_state = getattr(status, "state", None)
            if print_state in ("FINISH", "FAILED", "IDLE"):
                self.reset_printer(printer_id)
                continue
            # Unknown or transitional status is not evidence that a print ended.
            # Keep its context, but only upload while the printer is running.
            if print_state != "RUNNING":
                task = self._checks.get(printer_id)
                if task:
                    task.cancel()
                continue
            state = self._states.get(printer_id)
            if state and not self._same_print(state, status):
                self.reset_printer(printer_id)
            task = self._checks.get(printer_id)
            if task is not None and not task.done():
                continue
            # One check per printer; a slow camera or API request must not
            # delay another printer's next inspection or lifecycle updates.
            task = asyncio.create_task(self._check_printer(printer_id, status, settings))
            self._checks[printer_id] = task
            task.add_done_callback(partial(self._check_done, printer_id))

    def _check_done(self, printer_id: int, task: asyncio.Task):
        # A canceled check may finish after its printer has started a new one.
        if self._checks.get(printer_id) is task:
            self._checks.pop(printer_id, None)
        if not task.cancelled():
            error = task.exception()
            if error is not None:
                logger.error("OctoEverywhere printer check failed: %s", type(error).__name__)

    @staticmethod
    def _print_id(status) -> str | None:
        subtask_id = getattr(status, "subtask_id", None)
        if subtask_id and str(subtask_id) != "0":
            return str(subtask_id)
        return None

    def _same_print(self, state: _PrintState, status) -> bool:
        """Learn late job IDs without discarding the context or forgetting a known ID."""
        print_id = self._print_id(status)
        if state.print_id and print_id and state.print_id != print_id:
            return False
        # Names and filenames may arrive late or change independently. Without
        # two distinct job IDs, rely on the print start/end callbacks instead.
        state.print_id = print_id or state.print_id
        return True

    def _is_current(self, printer_id: int, state: _PrintState, status, generation: int) -> bool:
        return (
            self._states.get(printer_id) is state
            and self._generation == generation
            and getattr(status, "state", None) == "RUNNING"
            and self._same_print(state, status)
        )

    async def _capture_frame(self, printer_id: int) -> bytes | None:
        from backend.app.services.failure_detection_camera import capture_detection_frame

        return await capture_detection_frame(printer_id)

    async def _create_context(self, client: httpx.AsyncClient, api_key: str, confidence: str) -> tuple[_Context, int]:
        level = CONFIDENCE_LEVELS.get(confidence, 3)
        response = await client.post(
            CREATE_CONTEXT_URL,
            headers={"X-API-Key": api_key},
            json={"WarningConfidenceLevel": level, "PauseConfidenceLevel": level},
        )
        payload = _response_payload(response)
        context_id = payload.get("ContextId")
        if not isinstance(context_id, str) or not context_id.strip():
            raise _ApiError("OctoEverywhere returned an invalid context.", response.status_code)
        return (
            _Context(
                context_id=context_id,
                process_url=_process_url(payload.get("ProcessRequestUrl")),
                fallback_url=_process_url(payload.get("FallbackProcessRequestUrl")),
            ),
            response.status_code,
        )

    def _no_verdict(self, printer_id: int, state: _PrintState, reason: str, error_code: str | None = None):
        if self._states.get(printer_id) is not state:
            return
        if state.error != reason:
            logger.warning("OctoEverywhere printer %s: %s", printer_id, reason)
        state.error = reason
        state.error_code = _known_error_code(error_code)
        self._set_last_error(reason, state.error_code)

    async def _check_printer(self, printer_id: int, status, settings: dict):
        generation = self._generation
        state = self._states.get(printer_id)
        if state is None or not self._same_print(state, status):
            state = _PrintState(print_id=self._print_id(status))
            self._states[printer_id] = state
            self._refresh_last_error()
        if self._blocked_error_code:
            return
        if state.last_processed_at is not None and not state.error:
            # Apply interval edits to an existing context on the next poll.
            # Failed/canceled requests clear last_processed_at, preserving their
            # reserved retry deadline regardless of the configured interval.
            state.next_check_at = state.last_processed_at + _process_interval(
                state.interval, settings["poll_interval"], state.recommended_interval
            )
        if time.monotonic() < state.next_check_at:
            return

        # Reserve this poll before awaiting camera/network I/O. Even a failed
        # capture must not create a hot loop or compete with a live camera.
        state.next_check_at = time.monotonic() + _process_interval(
            state.interval, settings["poll_interval"], state.recommended_interval
        )
        logger.debug(
            "OctoEverywhere inspection starting for printer %s (frame=%s, context_reused=%s)",
            printer_id,
            state.frame_count + 1,
            state.context is not None,
        )
        processing = False
        try:
            frame = await self._capture_frame(printer_id)
            if not frame:
                raise _ApiError("Could not capture a camera snapshot. Detection will retry automatically.")
            if len(frame) > MAX_IMAGE_BYTES:
                try:
                    frame = await asyncio.to_thread(prepare_frame, frame)
                except ValueError:
                    raise _ApiError("Could not reduce the camera snapshot to OctoEverywhere's 6 MiB image limit.")
            if not self._is_current(printer_id, state, status, generation):
                logger.debug(
                    "OctoEverywhere inspection discarded for printer %s: print status or settings changed", printer_id
                )
                return
            async with httpx.AsyncClient(timeout=DETECTION_TIMEOUT, follow_redirects=False) as client:
                if state.context is None:
                    context, _ = await self._create_context(client, settings["api_key"], settings["confidence"])
                    if not self._is_current(printer_id, state, status, generation):
                        logger.debug(
                            "OctoEverywhere inspection discarded for printer %s: print status or settings changed",
                            printer_id,
                        )
                        return
                    state.context = context
                    logger.debug("OctoEverywhere context created for printer %s", printer_id)
                if not self._is_current(printer_id, state, status, generation):
                    logger.debug(
                        "OctoEverywhere inspection discarded for printer %s: print status or settings changed",
                        printer_id,
                    )
                    return
                processing = True
                response = await client.post(
                    state.context.process_url,
                    headers={"X-API-Key": settings["api_key"]},
                    files={"image": ("snapshot.jpg", frame, "image/jpeg")},
                )
                payload = _response_payload(response)
                interval = payload.get("NextProcessIntervalSec")
                if not isinstance(interval, dict):
                    raise _ApiError("OctoEverywhere returned an invalid processing interval.")
                minimum_interval = interval.get("Minimum")
                if type(minimum_interval) is int and minimum_interval > 0:
                    # Preserve the latest minimum even if the assessment is
                    # invalid or canceled before its timing hint is accepted.
                    state.interval = minimum_interval
                    state.next_check_at = time.monotonic() + _process_interval(
                        minimum_interval, settings["poll_interval"], state.recommended_interval
                    )
                else:
                    raise _ApiError("OctoEverywhere returned an invalid processing interval.")
                faster_inspection_suggested = payload.get("FasterInspectionSuggested") is True
                recommended_interval = interval.get("Recommended") if faster_inspection_suggested else None
                # This hint is optional. Older responses and malformed hints
                # keep fixed timing without discarding a valid assessment.
                if type(recommended_interval) is not int or recommended_interval <= 0:
                    recommended_interval = None
                print_quality = payload.get("PrintQuality")
                warning = payload.get("WarningSuggested")
                pause = payload.get("PauseSuggested")
                if (
                    type(print_quality) is not int
                    or not 1 <= print_quality <= 10
                    or type(warning) is not bool
                    or type(pause) is not bool
                ):
                    raise _ApiError("OctoEverywhere returned an invalid print assessment.")
        except asyncio.CancelledError:
            # A request canceled during a settings edit may already have been
            # processed remotely. Preserve the previous minimum before retrying.
            state.next_check_at = max(
                state.next_check_at,
                time.monotonic()
                + _process_interval(state.interval, settings["poll_interval"], state.recommended_interval),
            )
            state.last_processed_at = None
            logger.debug("OctoEverywhere inspection canceled for printer %s", printer_id)
            raise
        except Exception as exc:
            state.last_processed_at = None
            # Account failures still apply if this print ended during the
            # request, but an old key or generation must not block new work.
            if (
                isinstance(exc, _ApiError)
                and exc.error_code in ACCOUNT_ERROR_CODES
                and generation == self._generation
                and (self._configuration is None or settings["api_key"] == self._configuration[0])
            ):
                self._block_account(settings["api_key"], exc.error_code)
                logger.debug(
                    "OctoEverywhere inspection failed for printer %s (error_code=%s); inspections stopped",
                    printer_id,
                    exc.error_code,
                )
                return
            if not self._is_current(printer_id, state, status, generation):
                state.next_check_at = max(
                    state.next_check_at,
                    time.monotonic()
                    + _process_interval(state.interval, settings["poll_interval"], state.recommended_interval),
                )
                logger.debug(
                    "OctoEverywhere inspection discarded for printer %s: print status or settings changed", printer_id
                )
                return
            if processing and state.context:
                # Keep the fallback for the rest of this context. Retrying on
                # the next scheduled poll also respects the API's rate limit.
                if isinstance(exc, _ApiError) and exc.context_missing:
                    state.context = None  # Recreate after backoff; preserve actions already performed.
                    state.recommended_interval = None
                elif isinstance(exc, httpx.RequestError) or (
                    isinstance(exc, _ApiError)
                    and exc.status_code is not None
                    and exc.status_code >= 500
                    and exc.error_code in (None, "OE_INTERNAL_ERROR", "OE_BACKEND_THROTTLED")
                ):
                    state.context.process_url = state.context.fallback_url
            state.failures += 1
            retry_after = exc.retry_after if isinstance(exc, _ApiError) else 0
            delay = max(
                _process_interval(state.interval, settings["poll_interval"], state.recommended_interval),
                ERROR_RETRY_INTERVAL * 2 ** min(state.failures - 1, 4),
                retry_after,
            )
            state.next_check_at = time.monotonic() + delay
            if isinstance(exc, _ApiError):
                reason = str(exc)
            elif isinstance(exc, httpx.TimeoutException):
                reason = "OctoEverywhere request timed out. Detection will retry automatically."
            elif isinstance(exc, httpx.RequestError):
                reason = "Could not reach OctoEverywhere. Check the internet connection."
            else:
                reason = "OctoEverywhere detection failed. Detection will retry automatically."
            self._no_verdict(printer_id, state, reason, exc.error_code if isinstance(exc, _ApiError) else None)
            logger.debug(
                "OctoEverywhere inspection failed for printer %s (error_code=%s, retry_in=%.1fs): %s",
                printer_id,
                state.error_code,
                delay,
                reason,
            )
            return

        if not self._is_current(printer_id, state, status, generation):
            logger.debug(
                "OctoEverywhere inspection discarded for printer %s: print status or settings changed", printer_id
            )
            return
        state.recommended_interval = recommended_interval
        state.last_processed_at = time.monotonic()
        state.next_check_at = state.last_processed_at + _process_interval(
            state.interval, settings["poll_interval"], state.recommended_interval
        )
        state.frame_count += 1
        state.print_quality = print_quality
        state.verdict = "failure" if pause else "warning" if warning else "safe"
        state.error = None
        state.error_code = None
        state.failures = 0
        self._refresh_last_error()
        logger.debug(
            "OctoEverywhere inspection result for printer %s: frame=%s, verdict=%s, print_quality=%s, "
            "warning_suggested=%s, pause_suggested=%s, faster_inspection_suggested=%s, "
            "minimum_interval=%ss, next_check_in=%ss",
            printer_id,
            state.frame_count,
            state.verdict,
            print_quality,
            warning,
            pause,
            faster_inspection_suggested,
            state.interval,
            _process_interval(state.interval, settings["poll_interval"], state.recommended_interval),
        )
        task_name = getattr(status, "task_name", None) or getattr(status, "subtask_name", "") or ""
        if state.verdict != "safe":
            self._history.appendleft(
                {
                    "printer_id": printer_id,
                    "task_name": task_name,
                    "timestamp": datetime.now(timezone.utc).isoformat(),
                    "class": state.verdict,
                    "print_quality": print_quality,
                }
            )

        notification_due = not state.warning_fired or (
            state.last_notification_at is not None
            and time.monotonic() - state.last_notification_at >= NOTIFICATION_REPEAT_INTERVAL
        )
        if pause and not state.action_fired:
            action = settings["action"]
            # A pause or power cut always notifies, even after a warning: the
            # user needs to know the printer stopped. Notify-only follows the
            # same reminder cadence as warnings.
            if action == "notify" and not notification_due:
                state.action_fired = True
                return
        elif (warning or pause) and notification_due:
            # Only a fresh positive assessment can trigger a reminder. Never
            # repeat hardware control, even if Gadget still suggests pausing.
            action = "notify"
        else:
            return

        # Preparation is still cancellable: a print/settings change before any
        # side effect must prevent stale hardware control.
        printer_name = await self._prepare_action(printer_id)
        if not self._is_current(printer_id, state, status, generation):
            return
        # Once committed, a pause/disconnect or autosave must neither interrupt
        # the remaining plug-off/notification nor make this print act twice.
        state.action_fired = state.action_fired or pause
        state.warning_fired = True
        # Reserve the reminder deadline before yielding, including when a
        # settings save cancels this check while the notification is in flight.
        state.last_notification_at = time.monotonic()
        action_task = asyncio.create_task(
            self._dispatch_action(printer_id, action, task_name, print_quality, frame, printer_name=printer_name)
        )
        self._actions.add(action_task)
        action_task.add_done_callback(self._action_done)
        await asyncio.shield(action_task)

    async def _prepare_action(self, printer_id: int) -> str:
        from backend.app.services.obico_actions import _get_printer_name

        return await _get_printer_name(printer_id)

    def _action_done(self, task: asyncio.Task):
        self._actions.discard(task)
        if not task.cancelled():
            error = task.exception()
            if error is not None:
                logger.error("OctoEverywhere action task failed: %s", type(error).__name__)

    async def _dispatch_action(
        self,
        printer_id: int,
        action: str,
        task_name: str,
        print_quality: int,
        frame: bytes | None = None,
        *,
        printer_name: str | None = None,
    ):
        from backend.app.services.octoeverywhere_actions import execute_action

        logger.warning(
            "OctoEverywhere: print issue on printer %s (quality=%s) — action=%s", printer_id, print_quality, action
        )
        try:
            await execute_action(printer_id, action, task_name, print_quality, frame, printer_name=printer_name)
        except Exception as exc:
            self._set_last_error("OctoEverywhere could not execute the configured failure action.")
            logger.error("OctoEverywhere action dispatch failed: %s", type(exc).__name__)

    # ---- queries ----

    def get_per_printer(self) -> dict:
        return {
            printer_id: {
                "class": "error" if self._blocked_error_code or state.error else state.verdict,
                "print_quality": state.print_quality,
                "frame_count": state.frame_count,
                "error": API_ERRORS[self._blocked_error_code] if self._blocked_error_code else state.error,
                "error_code": self._blocked_error_code or state.error_code,
            }
            for printer_id, state in self._states.items()
        }

    def get_status(self) -> dict:
        return {
            "is_running": self._task is not None and not self._task.done(),
            "last_error": self._last_error,
            "last_error_code": self._last_error_code,
            "per_printer": self.get_per_printer(),
            "history": list(self._history),
        }

    async def test_connection(self, api_key: str, confidence: str = "medium") -> dict:
        """Validate a key with a context request; never upload a camera frame."""
        generation = self._generation
        api_key = api_key.strip()
        if not api_key:
            return {
                "ok": False,
                "status_code": None,
                "error": "An OctoEverywhere Gadget API key is required.",
                "error_code": None,
            }
        try:
            async with httpx.AsyncClient(timeout=DETECTION_TIMEOUT, follow_redirects=False) as client:
                _, status_code = await self._create_context(client, api_key, confidence)
        except _ApiError as exc:
            if (
                generation == self._generation
                and self._configuration
                and api_key == self._configuration[0]
                and exc.error_code in ACCOUNT_ERROR_CODES
            ):
                self._block_account(api_key, exc.error_code)
            return {"ok": False, "status_code": exc.status_code, "error": str(exc), "error_code": exc.error_code}
        except httpx.TimeoutException:
            return {"ok": False, "status_code": None, "error": "OctoEverywhere request timed out.", "error_code": None}
        except Exception:
            return {
                "ok": False,
                "status_code": None,
                "error": "Could not connect to OctoEverywhere.",
                "error_code": None,
            }
        if generation == self._generation and self._configuration and api_key == self._configuration[0]:
            self._clear_account_error()
        return {"ok": True, "status_code": status_code, "error": None, "error_code": None}


octoeverywhere_detection_service = OctoEverywhereDetectionService()