fix(smart-plug): don't cut power when a print restarts, honor per-plug cooldown setting (#1890)

The print-queue "auto off after this job" trigger used a second, inline
auto-off implementation (main.py, print_scheduler.py, print_queue.py)
that hardcoded wait_for_cooldown(50C, 600s) — ignoring each plug's
configured off_delay_mode / off_delay_minutes / off_temp_threshold — and
ignored the return value, powering off on the 600s timeout regardless of
print state. A print that failed and was reprinted from the touchscreen
got its power cut mid-print. The inline tasks were also uncancellable, so
a reprint couldn't abort a pending off.

Consolidate all three into SmartPlugManager.schedule_off_after_queue_job,
which schedules via the plug's configured strategy (shared with
on_print_complete through _schedule_off_per_mode) and is cancellable via
_pending_off. Add printer_manager.is_print_active() and guard the actual
power-off in _delayed_off and _temp_based_off so no path cuts power on a
loaded print. Move the on_print_start cancellation ahead of the auto_on
gate so a reprint always aborts a pending off.

backend/app/api/routes/print_queue.py

@@ -16,7 +16,6 @@ from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_owners
 from backend.app.core.config import settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
-from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_batch import PrintBatch
@@ -1272,9 +1271,7 @@ async def stop_queue_item(
     holding only _OWN saw the Stop button in the queue UI but got 403 on click.
     """
 
-    from backend.app.models.smart_plug import SmartPlug
     from backend.app.services.printer_manager import printer_manager
-    from backend.app.services.tasmota import tasmota_service
 
     user, can_modify_all = auth_result
 
@@ -1323,33 +1320,21 @@ async def stop_queue_item(
     item.error_message = "Stopped by user" if stop_sent else "Stopped by user (printer was offline)"
     await db.commit()
 
-    # Get smart plug info if auto-off is enabled
-    plug_ip = None
-    if auto_off_after:
-        result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-        plug = result.scalar_one_or_none()
-        if plug and plug.enabled:
-            plug_ip = plug.ip_address
-
     logger.info("Stopped printing queue item %s (stop command sent: %s)", item_id, stop_sent)
 
-    # Schedule background task for cooldown + power off
-    if plug_ip:
+    # Schedule power-off if the queue item opted in. Delegates to the smart-plug
+    # manager so the off honours each plug's configured strategy (time delay or
+    # temperature threshold), is cancelled if the printer starts printing again,
+    # and never cuts power on a loaded print (#1890). Previously an inline block
+    # hardcoded a 50°C / 600s cooldown wait and powered off on the timeout
+    # regardless of print state.
+    if auto_off_after:
+        from backend.app.services.smart_plug_manager import smart_plug_manager
 
-        async def cooldown_and_poweroff():
-            logger.info("Auto-off: Waiting for printer %s to cool down before power off...", printer_id)
-            await printer_manager.wait_for_cooldown(printer_id, target_temp=50.0, timeout=600)
-            # Re-fetch plug since we're in a new async context
-            from backend.app.core.database import async_session
-
-            async with async_session() as new_db:
-                result = await new_db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-                plug = result.scalar_one_or_none()
-                if plug and plug.enabled:
-                    logger.info("Auto-off: Powering off printer %s", printer_id)
-                    await tasmota_service.turn_off(plug)
-
-        spawn_background_task(cooldown_and_poweroff(), name=f"queue-cooldown-poweroff-{printer_id}")
+        try:
+            await smart_plug_manager.schedule_off_after_queue_job(printer_id, db)
+        except Exception as e:
+            logger.warning("Auto-off: Failed to schedule power-off for printer %s: %s", printer_id, e)
 
     return {"message": "Print stopped" if stop_sent else "Queue item cancelled (printer was offline)"}
 

backend/app/main.py

@@ -4333,38 +4333,19 @@ async def on_print_complete(printer_id: int, data: dict):
             except Exception:
                 pass  # Don't fail if notification fails
 
-            # Handle auto_off_after - power off printer if requested (after cooldown)
+            # Handle auto_off_after - power off printer if the queue item opted
+            # in. Delegates to the smart-plug manager so the off honours each
+            # plug's configured strategy (time delay or temperature threshold),
+            # is cancelled if the printer starts printing again, and never cuts
+            # power on a loaded print (#1890). Previously an inline block here
+            # hardcoded a 50°C / 600s cooldown wait and powered off on the
+            # timeout regardless of print state — cutting a touchscreen reprint.
             if queue_auto_off:
-                async with async_session() as db:
-                    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-                    plugs = list(result.scalars().all())
-                enabled_plugs = [p for p in plugs if p.enabled]
-                if enabled_plugs:
-                    logger.info("Auto-off requested for printer %s, waiting for cooldown...", printer_id)
-
-                    async def cooldown_and_poweroff(pid: int, plug_ids: list[int]):
-                        # Wait for nozzle to cool down
-                        await printer_manager.wait_for_cooldown(pid, target_temp=50.0, timeout=600)
-                        # Re-fetch plugs in new session and turn off each one
-                        async with async_session() as new_db:
-                            for plug_id in plug_ids:
-                                try:
-                                    result = await new_db.execute(select(SmartPlug).where(SmartPlug.id == plug_id))
-                                    p = result.scalar_one_or_none()
-                                    if p and p.enabled:
-                                        service = await smart_plug_manager.get_service_for_plug(p, new_db)
-                                        success = await service.turn_off(p)
-                                        if success:
-                                            logger.info("Powered off printer %s via smart plug '%s'", pid, p.name)
-                                        else:
-                                            logger.warning("Failed to power off plug '%s' for printer %s", p.name, pid)
-                                except Exception as e:
-                                    logger.warning("Failed to power off plug %s for printer %s: %s", plug_id, pid, e)
-
-                    spawn_background_task(
-                        cooldown_and_poweroff(printer_id, [p.id for p in enabled_plugs]),
-                        name=f"cooldown-poweroff-{printer_id}",
-                    )
+                try:
+                    async with async_session() as db:
+                        await smart_plug_manager.schedule_off_after_queue_job(printer_id, db)
+                except Exception as e:
+                    logger.warning("Failed to schedule queue auto-off for printer %s: %s", printer_id, e)
     except Exception as e:
         logging.getLogger(__name__).warning(f"Queue item update failed: {e}")
 

backend/app/services/print_scheduler.py

@@ -2329,30 +2329,20 @@ class PrintScheduler:
         return prev_item.status in ("completed", "cancelled")
 
     async def _power_off_if_needed(self, db: AsyncSession, item: PrintQueueItem):
-        """Power off printer if auto_off_after is enabled (waits for cooldown)."""
+        """Schedule power-off if the queue item enabled auto_off_after.
+
+        Delegates to the smart-plug manager so the off honours each plug's
+        configured strategy (time delay or temperature threshold), is cancelled
+        if the printer starts printing again, and never cuts power on a loaded
+        print (#1890). Previously this hardcoded a 50°C / 600s cooldown wait and
+        powered off on the timeout regardless of print state.
+        """
         if not item.auto_off_after:
             return
-
-        plugs = await self._get_smart_plugs(db, item.printer_id)
-        plug_ids = [p.id for p in plugs if p.enabled]
-        if plug_ids:
-            logger.info("Auto-off: Waiting for printer %s to cool down before power off...", item.printer_id)
-            # Wait for cooldown (up to 10 minutes)
-            await printer_manager.wait_for_cooldown(item.printer_id, target_temp=50.0, timeout=600)
-            # Re-fetch plugs in a fresh session after the long cooldown wait
-            async with async_session() as new_db:
-                for plug_id in plug_ids:
-                    try:
-                        result = await new_db.execute(select(SmartPlug).where(SmartPlug.id == plug_id))
-                        plug = result.scalar_one_or_none()
-                        if plug and plug.enabled:
-                            logger.info("Auto-off: Powering off plug '%s' for printer %s", plug.name, item.printer_id)
-                            service = await smart_plug_manager.get_service_for_plug(plug, new_db)
-                            await service.turn_off(plug)
-                    except Exception as e:
-                        logger.warning(
-                            "Auto-off: Failed to power off plug %s for printer %s: %s", plug_id, item.printer_id, e
-                        )
+        try:
+            await smart_plug_manager.schedule_off_after_queue_job(item.printer_id, db)
+        except Exception as e:
+            logger.warning("Auto-off: Failed to schedule power-off for printer %s: %s", item.printer_id, e)
 
     async def _get_job_name(self, db: AsyncSession, item: PrintQueueItem) -> str:
         """Get a human-readable name for a queue item."""

backend/app/services/printer_manager.py

@@ -602,6 +602,24 @@ class PrinterManager:
             return client.state
         return None
 
+    # Gcode states in which a job is loaded / in progress and cutting power
+    # would ruin the print. PAUSE is included on purpose — a paused print is
+    # still loaded on the bed. Used by the smart-plug auto-off guard (#1890) so
+    # a re-print started from the touchscreen isn't killed mid-print.
+    ACTIVE_PRINT_STATES = ("RUNNING", "PAUSE", "PREPARE", "SLICING")
+
+    def is_print_active(self, printer_id: int) -> bool:
+        """True when the printer currently has a print loaded / in progress.
+
+        Returns False when disconnected or in any idle/terminal state
+        (IDLE / FINISH / FAILED / unknown), so callers fail *open* only for
+        the safe "nothing is printing" case. #1890.
+        """
+        state = self.get_status(printer_id)
+        if not state or not state.connected:
+            return False
+        return state.state in self.ACTIVE_PRINT_STATES
+
     def get_model(self, printer_id: int) -> str | None:
         """Get the cached model for a printer."""
         return self._models.get(printer_id)

backend/app/services/smart_plug_manager.py

@@ -227,13 +227,16 @@ class SmartPlugManager:
                 logger.debug("Smart plug '%s' is disabled, skipping auto-on", plug.name)
                 continue
 
+            # Cancel any pending off task FIRST — a re-print must abort a
+            # scheduled auto-off regardless of the plug's auto_on setting
+            # (#1890). Previously this lived behind the auto_on gate, so a plug
+            # with auto_on disabled kept its pending off and cut power mid-print.
+            self._cancel_pending_off(plug.id)
+
             if not plug.auto_on:
                 logger.debug("Smart plug '%s' auto_on is disabled", plug.name)
                 continue
 
-            # Cancel any pending off task
-            self._cancel_pending_off(plug.id)
-
             # Turn on the plug
             logger.info("Print started on printer %s, turning on plug '%s'", printer_id, plug.name)
             try:
@@ -289,10 +292,51 @@ class SmartPlugManager:
                 plug.name,
             )
 
-            if plug.off_delay_mode == "time":
-                self._schedule_delayed_off(plug, printer_id, plug.off_delay_minutes * 60)
-            elif plug.off_delay_mode == "temperature":
-                self._schedule_temp_based_off(plug, printer_id, plug.off_temp_threshold)
+            self._schedule_off_per_mode(plug, printer_id)
+
+    def _schedule_off_per_mode(self, plug: "SmartPlug", printer_id: int):
+        """Schedule an auto-off using the plug's configured off strategy.
+
+        Honours the per-plug ``off_delay_mode`` — ``time`` waits
+        ``off_delay_minutes``; ``temperature`` waits until the nozzle drops
+        below ``off_temp_threshold`` (#1890 — the queue/scheduler auto-off
+        paths used to hardcode 50°C / 600s and ignore these settings). Both
+        branches register a cancellable task in ``_pending_off``, so a re-print
+        cancels the pending off via :meth:`on_print_start`.
+        """
+        if plug.off_delay_mode == "temperature":
+            self._schedule_temp_based_off(plug, printer_id, plug.off_temp_threshold)
+        else:
+            # Default / "time": also the safe fallback for any unexpected value.
+            self._schedule_delayed_off(plug, printer_id, plug.off_delay_minutes * 60)
+
+    async def schedule_off_after_queue_job(self, printer_id: int, db: AsyncSession):
+        """Schedule auto-off for a printer after a queue job that opted in.
+
+        The print-queue "auto off after this job" toggle (`auto_off_after`) is
+        a per-job override, independent of the plug's global ``auto_off`` flag —
+        so unlike :meth:`on_print_complete` this does NOT gate on ``plug.auto_off``.
+        It still honours ``enabled`` and skips HA-script entities (which can only
+        be triggered, not turned off), and uses each plug's configured off
+        strategy via :meth:`_schedule_off_per_mode`. Replaces the three inline
+        ``wait_for_cooldown(50°C, 600s)`` blocks that ignored plug settings,
+        fired on the cooldown *timeout* regardless of print state, and could not
+        be cancelled by a re-print (#1890).
+        """
+        plugs = await self._get_plugs_for_printer(printer_id, db)
+        for plug in plugs:
+            if not plug.enabled:
+                logger.debug("Smart plug '%s' is disabled, skipping queue auto-off", plug.name)
+                continue
+            if plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script."):
+                logger.debug("Smart plug '%s' is a HA script entity, skipping queue auto-off", plug.name)
+                continue
+            logger.info(
+                "Queue job finished on printer %s, scheduling turn-off for plug '%s'",
+                printer_id,
+                plug.name,
+            )
+            self._schedule_off_per_mode(plug, printer_id)
 
     async def on_drying_complete(self, printer_id: int, db: AsyncSession):
         """Schedule turn-off for plugs flagged ``auto_off_after_drying`` when
@@ -381,6 +425,21 @@ class SmartPlugManager:
         try:
             await asyncio.sleep(delay_seconds)
 
+            # #1890: never cut power while a print is loaded / running. The
+            # delay fires unconditionally after N minutes, so if the user
+            # re-started (or reprinted) in the meantime, the printer is active
+            # again — skip the off and clear the pending flag rather than
+            # killing the print mid-way.
+            if printer_manager.is_print_active(printer_id):
+                logger.info(
+                    "Skipping auto-off for plug %s: printer %s is printing again (state=%s)",
+                    plug_id,
+                    printer_id,
+                    getattr(printer_manager.get_status(printer_id), "state", "unknown"),
+                )
+                await self._mark_auto_off_pending(plug_id, False)
+                return
+
             # Create a minimal plug-like object for the service
             class PlugInfo:
                 def __init__(self):
@@ -489,6 +548,22 @@ class SmartPlugManager:
                         )
 
                     if max_nozzle_temp < temp_threshold:
+                        # #1890: the nozzle can dip below the threshold between
+                        # a finished print and a fresh one starting (e.g. a
+                        # touchscreen reprint during the PREPARE/heating phase).
+                        # Guard the turn-off so we never cut power on a loaded
+                        # print; keep polling until it's genuinely idle again.
+                        if printer_manager.is_print_active(printer_id):
+                            logger.info(
+                                "Deferring temp-based auto-off for plug %s: printer %s is printing again (state=%s)",
+                                plug_id,
+                                printer_id,
+                                getattr(printer_manager.get_status(printer_id), "state", "unknown"),
+                            )
+                            await asyncio.sleep(check_interval)
+                            elapsed += check_interval
+                            continue
+
                         # All nozzles are below threshold, turn off
                         class PlugInfo:
                             def __init__(self):
@@ -626,6 +701,22 @@ class SmartPlugManager:
 
                     logger.info("Resuming pending auto-off for plug '%s' (printer %s)", plug.name, plug.printer_id)
 
+                    # #1890: never resume a power-off onto a live print. If the
+                    # printer started a new print during the downtime, the stale
+                    # pending off must be dropped, not executed — same guard the
+                    # live off-executors use.
+                    if printer_manager.is_print_active(plug.printer_id):
+                        logger.info(
+                            "Not resuming auto-off for plug '%s': printer %s is printing (state=%s); clearing pending",
+                            plug.name,
+                            plug.printer_id,
+                            getattr(printer_manager.get_status(plug.printer_id), "state", "unknown"),
+                        )
+                        plug.auto_off_pending = False
+                        plug.auto_off_pending_since = None
+                        await db.commit()
+                        continue
+
                     # Resume the appropriate off mode
                     if plug.off_delay_mode == "temperature":
                         self._schedule_temp_based_off(plug, plug.printer_id, plug.off_temp_threshold)

backend/tests/unit/services/test_printer_manager.py

@@ -476,6 +476,46 @@ class TestPrinterManager:
 
         assert result is True
 
+    # ========================================================================
+    # Tests for is_print_active (#1890)
+    # ========================================================================
+
+    @pytest.mark.parametrize(
+        "state,expected",
+        [
+            ("RUNNING", True),
+            ("PAUSE", True),
+            ("PREPARE", True),
+            ("SLICING", True),
+            ("FINISH", False),
+            ("IDLE", False),
+            ("FAILED", False),
+            ("unknown", False),
+        ],
+    )
+    def test_is_print_active_state_matrix(self, manager, mock_client, state, expected):
+        """A job-loaded state is 'active'; idle/terminal states are not."""
+        mock_client.state.connected = True
+        mock_client.state.state = state
+        mock_client.check_staleness.return_value = True
+        manager._clients[1] = mock_client
+
+        assert manager.is_print_active(1) is expected
+
+    def test_is_print_active_false_when_disconnected(self, manager, mock_client):
+        """Even in RUNNING, a disconnected printer is not treated as active —
+        we fail safe (no active print) only for the 'nothing printing' cases."""
+        mock_client.state.connected = False
+        mock_client.state.state = "RUNNING"
+        mock_client.check_staleness.return_value = False
+        manager._clients[1] = mock_client
+
+        assert manager.is_print_active(1) is False
+
+    def test_is_print_active_false_for_unknown_printer(self, manager):
+        """Unknown printer id → not active (no client)."""
+        assert manager.is_print_active(999) is False
+
     # ========================================================================
     # Tests for logging methods
     # ========================================================================

backend/tests/unit/services/test_smart_plug_manager.py

@@ -4,6 +4,7 @@ These tests specifically target the auto-off behavior and toggle functionality
 that were identified as common regression points.
 """
 
+import asyncio
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -815,7 +816,7 @@ class TestPendingAutoOffPersistence:
             patch("backend.app.core.database.async_session") as mock_session_ctx,
             patch("backend.app.services.smart_plug_manager.tasmota_service") as mock_tasmota,
             patch.object(manager, "_mark_auto_off_executed", new_callable=AsyncMock) as mock_mark,
-            patch("backend.app.services.smart_plug_manager.printer_manager"),
+            patch("backend.app.services.smart_plug_manager.printer_manager") as mock_pm,
         ):
             mock_db = AsyncMock()
             mock_result = MagicMock()
@@ -826,8 +827,227 @@ class TestPendingAutoOffPersistence:
             mock_session_ctx.return_value.__aexit__ = AsyncMock()
 
             mock_tasmota.turn_off = AsyncMock(return_value=True)
+            mock_pm.is_print_active.return_value = False  # printer idle on restart
 
             await manager.resume_pending_auto_offs()
 
             mock_tasmota.turn_off.assert_called_once()
             mock_mark.assert_called_once_with(1)
+
+    @pytest.mark.asyncio
+    async def test_resume_pending_auto_off_skipped_when_printing(self, manager):
+        """#1890: on restart, a stale pending off must NOT power off a live print;
+        the pending flag is cleared instead."""
+        mock_plug = MagicMock()
+        mock_plug.id = 1
+        mock_plug.name = "Test Plug"
+        mock_plug.printer_id = 1
+        mock_plug.auto_off_pending = True
+        mock_plug.auto_off_pending_since = datetime.now(timezone.utc)
+        mock_plug.off_delay_mode = "time"
+
+        with (
+            patch("backend.app.core.database.async_session") as mock_session_ctx,
+            patch("backend.app.services.smart_plug_manager.tasmota_service") as mock_tasmota,
+            patch("backend.app.services.smart_plug_manager.printer_manager") as mock_pm,
+            patch.object(manager, "_schedule_temp_based_off") as mock_temp,
+        ):
+            mock_db = AsyncMock()
+            mock_result = MagicMock()
+            mock_result.scalars.return_value.all.return_value = [mock_plug]
+            mock_db.execute = AsyncMock(return_value=mock_result)
+            mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            mock_session_ctx.return_value.__aexit__ = AsyncMock()
+
+            mock_tasmota.turn_off = AsyncMock(return_value=True)
+            mock_pm.is_print_active.return_value = True  # printer printing again on restart
+            mock_pm.get_status.return_value = MagicMock(state="RUNNING")
+
+            await manager.resume_pending_auto_offs()
+
+            mock_tasmota.turn_off.assert_not_called()  # never cut power on the live print
+            mock_temp.assert_not_called()
+            assert mock_plug.auto_off_pending is False  # stale pending cleared
+
+
+class TestActivePrintGuard:
+    """#1890 — auto-off must never cut power while a print is loaded/running.
+
+    Covers the two off-executors (`_delayed_off`, `_temp_based_off`), the new
+    queue-override scheduler that honours per-plug settings, and the
+    on_print_start cancellation gap.
+    """
+
+    @pytest.fixture
+    def manager(self):
+        return SmartPlugManager()
+
+    @pytest.fixture
+    def mock_plug(self):
+        plug = MagicMock()
+        plug.id = 1
+        plug.name = "Test Plug"
+        plug.ip_address = "192.168.1.100"
+        plug.username = None
+        plug.password = None
+        plug.enabled = True
+        plug.auto_on = True
+        plug.auto_off = True
+        plug.off_delay_mode = "time"
+        plug.off_delay_minutes = 5
+        plug.off_temp_threshold = 70
+        plug.printer_id = 1
+        plug.plug_type = "tasmota"
+        plug.ha_entity_id = None
+        return plug
+
+    # ---- _delayed_off (time mode) ----------------------------------------
+
+    @pytest.mark.asyncio
+    async def test_delayed_off_skips_when_printer_printing_again(self, manager):
+        """Time-delay fires after N min; if a reprint is running, skip the off."""
+        with (
+            patch("backend.app.services.smart_plug_manager.printer_manager") as mock_pm,
+            patch.object(manager, "get_service_for_plug", new_callable=AsyncMock) as mock_get_svc,
+            patch.object(manager, "_mark_auto_off_pending", new_callable=AsyncMock) as mock_mark_pending,
+            patch.object(manager, "_mark_auto_off_executed", new_callable=AsyncMock) as mock_mark_exec,
+        ):
+            mock_pm.is_print_active.return_value = True
+            mock_pm.get_status.return_value = MagicMock(state="RUNNING")
+
+            await manager._delayed_off(1, "tasmota", "1.2.3.4", None, None, None, printer_id=1, delay_seconds=0)
+
+            mock_get_svc.assert_not_called()  # never even resolved a service to turn off
+            mock_mark_exec.assert_not_called()
+            mock_mark_pending.assert_awaited_with(1, False)  # pending flag cleared
+
+    @pytest.mark.asyncio
+    async def test_delayed_off_powers_off_when_idle(self, manager):
+        """When the printer is genuinely idle, the delayed off still fires."""
+        mock_service = AsyncMock()
+        mock_service.turn_off = AsyncMock(return_value=True)
+        with (
+            patch("backend.app.services.smart_plug_manager.printer_manager") as mock_pm,
+            patch.object(manager, "get_service_for_plug", new_callable=AsyncMock, return_value=mock_service),
+            patch.object(manager, "_mark_auto_off_executed", new_callable=AsyncMock),
+        ):
+            mock_pm.is_print_active.return_value = False
+
+            await manager._delayed_off(1, "tasmota", "1.2.3.4", None, None, None, printer_id=1, delay_seconds=0)
+
+            mock_service.turn_off.assert_awaited_once()
+            mock_pm.mark_printer_offline.assert_called_once_with(1)
+
+    # ---- _temp_based_off (temperature mode) ------------------------------
+
+    @pytest.mark.asyncio
+    async def test_temp_based_off_defers_while_printing_even_if_cool(self, manager):
+        """Nozzle can dip below threshold during a reprint's PREPARE/heat phase;
+        the guard must defer rather than cut power on the loaded print."""
+        with (
+            patch("backend.app.services.smart_plug_manager.printer_manager") as mock_pm,
+            patch("backend.app.services.smart_plug_manager.asyncio.sleep", new_callable=AsyncMock) as mock_sleep,
+            patch.object(manager, "get_service_for_plug", new_callable=AsyncMock) as mock_get_svc,
+        ):
+            # Cool enough to trip the threshold, but a print is active.
+            mock_pm.get_status.return_value = MagicMock(state="PREPARE", temperatures={"nozzle": 30})
+            mock_pm.is_print_active.return_value = True
+            # Break the poll loop after the first deferral so the test terminates.
+            mock_sleep.side_effect = asyncio.CancelledError()
+
+            await manager._temp_based_off(1, "tasmota", "1.2.3.4", None, None, None, printer_id=1, temp_threshold=70)
+
+            mock_get_svc.assert_not_called()  # never turned off despite temp < threshold
+
+    @pytest.mark.asyncio
+    async def test_temp_based_off_powers_off_when_cool_and_idle(self, manager):
+        """Cool nozzle + idle printer → turn off using the plug's threshold."""
+        mock_service = AsyncMock()
+        mock_service.turn_off = AsyncMock(return_value=True)
+        with (
+            patch("backend.app.services.smart_plug_manager.printer_manager") as mock_pm,
+            patch("backend.app.services.smart_plug_manager.asyncio.sleep", new_callable=AsyncMock),
+            patch.object(manager, "get_service_for_plug", new_callable=AsyncMock, return_value=mock_service),
+            patch.object(manager, "_mark_auto_off_executed", new_callable=AsyncMock),
+        ):
+            mock_pm.get_status.return_value = MagicMock(state="FINISH", temperatures={"nozzle": 40})
+            mock_pm.is_print_active.return_value = False
+
+            await manager._temp_based_off(1, "tasmota", "1.2.3.4", None, None, None, printer_id=1, temp_threshold=55)
+
+            mock_service.turn_off.assert_awaited_once()
+
+    # ---- schedule_off_after_queue_job (uses plug settings, not hardcoded 50/600)
+
+    @pytest.mark.asyncio
+    async def test_queue_off_uses_time_mode_regardless_of_global_auto_off(self, manager, mock_plug):
+        """Queue 'auto off after this job' is a per-job override — it schedules
+        even when the plug's global auto_off is disabled, and honours the plug's
+        configured time-delay mode."""
+        mock_plug.auto_off = False
+        mock_plug.off_delay_mode = "time"
+        mock_plug.off_delay_minutes = 8
+        with (
+            patch.object(manager, "_get_plugs_for_printer", new_callable=AsyncMock, return_value=[mock_plug]),
+            patch.object(manager, "_schedule_delayed_off") as mock_delayed,
+            patch.object(manager, "_schedule_temp_based_off") as mock_temp,
+        ):
+            await manager.schedule_off_after_queue_job(printer_id=1, db=AsyncMock())
+
+            mock_delayed.assert_called_once_with(mock_plug, 1, 8 * 60)  # plug's minutes, not hardcoded
+            mock_temp.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_queue_off_uses_configured_temp_threshold(self, manager, mock_plug):
+        """Temperature mode passes the plug's off_temp_threshold, not a hardcoded 50."""
+        mock_plug.off_delay_mode = "temperature"
+        mock_plug.off_temp_threshold = 65
+        with (
+            patch.object(manager, "_get_plugs_for_printer", new_callable=AsyncMock, return_value=[mock_plug]),
+            patch.object(manager, "_schedule_delayed_off") as mock_delayed,
+            patch.object(manager, "_schedule_temp_based_off") as mock_temp,
+        ):
+            await manager.schedule_off_after_queue_job(printer_id=1, db=AsyncMock())
+
+            mock_temp.assert_called_once_with(mock_plug, 1, 65)
+            mock_delayed.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_queue_off_skips_disabled_and_ha_script_plugs(self, manager, mock_plug):
+        """Disabled plugs and HA-script entities are never scheduled."""
+        disabled = MagicMock(id=2, name="disabled", enabled=False, plug_type="tasmota", ha_entity_id=None)
+        ha_script = MagicMock(
+            id=3, name="ha", enabled=True, plug_type="homeassistant", ha_entity_id="script.printer_off"
+        )
+        with (
+            patch.object(manager, "_get_plugs_for_printer", new_callable=AsyncMock, return_value=[disabled, ha_script]),
+            patch.object(manager, "_schedule_off_per_mode") as mock_sched,
+        ):
+            await manager.schedule_off_after_queue_job(printer_id=1, db=AsyncMock())
+
+            mock_sched.assert_not_called()
+
+    # ---- on_print_start cancellation gap ---------------------------------
+
+    @pytest.mark.asyncio
+    async def test_reprint_cancels_pending_off_even_when_auto_on_disabled(self, manager, mock_plug):
+        """A reprint must abort a scheduled auto-off regardless of auto_on (#1890).
+
+        Previously the cancel lived behind the auto_on gate, so a plug with
+        auto_on disabled kept its pending off and cut power mid-reprint.
+        """
+        mock_plug.auto_on = False
+        mock_task = MagicMock()
+        manager._pending_off[mock_plug.id] = mock_task
+        with (
+            patch.object(manager, "_get_plugs_for_printer", new_callable=AsyncMock, return_value=[mock_plug]),
+            patch.object(manager, "_mark_auto_off_pending", new_callable=AsyncMock),
+            patch("backend.app.services.smart_plug_manager.tasmota_service") as mock_tasmota,
+        ):
+            mock_tasmota.turn_on = AsyncMock()
+
+            await manager.on_print_start(printer_id=1, db=AsyncMock())
+
+            mock_task.cancel.assert_called_once()  # cancelled despite auto_on=False
+            assert mock_plug.id not in manager._pending_off
+            mock_tasmota.turn_on.assert_not_called()  # but not powered on