Stop tearing down AMS drying for a print that cannot start (#2801)

    A printer in FINISH with an unacknowledged plate and something pending in
    its queue stopped and restarted drying once per scheduler tick, for as
    long as the plate stayed unacknowledged. The reporter's Home Assistant
    history recorded about 2000 state changes over ten days. No cycle ever
    ran long enough to remove moisture, and cycles the user had started by
    hand on other AMS units of the same printer were torn down with it.

    Two concerns had become tangled. Plate-clear answers "is the bed ready
    for the next job" and says nothing about whether the AMS may heat. The
    gap between a finished print and the acknowledgment is when drying is
    most useful -- the printer is free and nobody is waiting on it -- and
    leaving the plate unacknowledged is also how people hold the queue by
    hand, so the hold was costing them the drying it should have enabled.

    Four faults, all in print_scheduler.

    The "print takes priority" stop sat inside the not-idle branch. Drying is
    not one of the things _is_printer_idle looks at, so stopping a cycle can
    never turn a non-idle printer into an idle one: the stop was futile every
    time it fired, and never fired on the dispatches where it was supposed to
    mean something. It now runs when the printer is actually dispatchable,
    and only where the model cannot dry through a print -- #2758 settled that
    capable hardware should keep its cycle.

    mid_print was inferred from busy_printers, which means "the queue could
    not dispatch here this pass", not "is printing". A plate-held printer was
    therefore treated as printing: the mid-print spool-protection cap
    silently lowered its drying temperature, the cycle was logged as
    (mid-print) in FINISH, and it bypassed the very gate meant to hold it.
    busy_printers keeps its dispatch role; auto-drying now gets a narrow set
    snapshotted before the item loop -- running, held post-dispatch, or
    mid-upload -- and mid_print comes from the printer's own state. The
    interlock comment at the seed already documented this hazard and worked
    around it by staying out of the set; this generalises that instead of
    adding a third special case. The other call site was already passing the
    narrow set, so the wide one was the inconsistency.

    _stop_drying sent a stop to every AMS reporting dry_time > 0. One
    auto-dried unit was enough to kill a manual cycle on a different unit of
    the same printer, contradicting the contract _sync_drying_state already
    documents: the entry gate only knows about cycles Bambuddy began, so the
    action must not reach past them. Consequence worth stating -- after a
    restart Bambuddy cannot prove a running cycle is its own, so it leaves it
    alone rather than risk stopping somebody's manual dry.

    Fourth, and the reason #2770's guard did not catch this: a reading at or
    below the threshold popped the unit's whole entry, ended_at included, so
    the 30-minute re-arm cooldown went with it. An AMS reads higher warm than
    cool, which is #2770's own finding, so a unit a point or two above the
    threshold dipped below it as it cooled, wiped its history, and re-armed
    immediately. Lifting a suspension now clears the judgement and keeps the
    clock.

    queue_drying_block changes behaviour as a result. It previously had no
    effect on dispatch at all -- both branches skipped anyway, and it only
    decided whether drying was needlessly killed. With the stop on the
    dispatch path it now does what it says: a queued print waits for a
    running cycle. Off by default.

    Reported by @superflyer11, who traced both defects to the line and
    brought ten days of external sensor history to date the cadence.

backend/app/services/print_scheduler.py

@@ -719,7 +719,7 @@ class PrintScheduler:
                 # printing. Report the pass as productive while uploads run so the
                 # loop stays on the fast interval.
                 inflight_printers = {pid for (_task, pid) in self._inflight.values() if pid is not None}
-                await self._check_auto_drying(db, [], inflight_printers, require_plate_clear=require_plate_clear)
+                await self._check_auto_drying(db, [], inflight_printers)
                 return bool(self._inflight)
 
             logger.info(
@@ -764,6 +764,23 @@ class PrintScheduler:
                 if inflight_pid is not None:
                     busy_printers.add(inflight_pid)
 
+            # Snapshot taken here, before the item loop adds anything (#2801).
+            #
+            # The three sources above all mean the same thing: a print on this
+            # printer is running or imminent. Everything the loop adds below
+            # means only "the queue could not dispatch to it this pass", which
+            # is a different statement -- a printer waiting on a plate-clear
+            # acknowledgment, an offline printer, one with no matching file.
+            #
+            # Auto-drying must only see the first kind. Reading the whole set
+            # as "is currently printing" is what put a plate-held printer down
+            # the mid-print path: it capped the drying temperature, logged the
+            # cycle as (mid-print), and skipped the very gate that was supposed
+            # to hold it. The interlock block below already documents the same
+            # hazard and works around it by staying out of busy_printers; this
+            # generalises that workaround instead of repeating it per case.
+            dispatching_printers: set[int] = set(busy_printers)
+
             # Printers held by a Home Assistant sensor interlock (#1148) — an
             # enclosure door left open, say. The fixed-printer branch turns
             # this into a waiting_reason the user can act on; the model-based
@@ -923,24 +940,17 @@ class PrintScheduler:
 
                     # Check if printer is idle (busy with another print)
                     if not printer_idle:
-                        # If printer is drying (not truly busy), handle based on queue_drying_block
-                        if self._drying_in_progress.get(item.printer_id):
-                            block_for_drying = await self._get_bool_setting(db, "queue_drying_block")
-                            if block_for_drying:
-                                # Drying blocks queue — skip this printer
-                                busy_printers.add(item.printer_id)
-                                continue
-                            else:
-                                # Print takes priority — stop drying
-                                await self._stop_drying(item.printer_id)
-                                # Re-check idle after stopping drying
-                                printer_idle = self._is_printer_idle(item.printer_id, require_plate_clear)
-                                if not printer_idle:
-                                    busy_printers.add(item.printer_id)
-                                    continue
-                        else:
-                            busy_printers.add(item.printer_id)
-                            continue
+                        busy_printers.add(item.printer_id)
+                        continue
+
+                    # Drying blocks the queue, if the user asked it to. A hold
+                    # is a skip like any other, so it belongs here with the
+                    # rest of the availability checks.
+                    if self._drying_in_progress.get(item.printer_id) and await self._get_bool_setting(
+                        db, "queue_drying_block"
+                    ):
+                        busy_printers.add(item.printer_id)
+                        continue
 
                     # Check condition (previous print success)
                     if item.require_previous_success:
@@ -988,6 +998,28 @@ class PrintScheduler:
                         busy_printers.add(item.printer_id)
                         continue
 
+                    # Print takes priority: stop a cycle Bambuddy armed, now
+                    # that this item is definitely going out.
+                    #
+                    # Placement is the whole point (#2801). This used to sit up
+                    # with the availability checks, inside the not-idle branch
+                    # -- so it fired only on the passes where the print was NOT
+                    # going to start, and never on the ones where it was.
+                    # Drying is not one of the things `_is_printer_idle` looks
+                    # at, so a stop could never have unblocked that printer
+                    # anyway; the cycle was spent for nothing, auto-drying
+                    # re-armed on the next tick, and a plate left
+                    # unacknowledged turned that into a loop on the scheduler
+                    # interval. Every skip between there and here -- a failed
+                    # previous print, an unmappable item, a filament deficit, a
+                    # contested library row -- is another way to lose a cycle
+                    # for a print that never happens, which is why this waits
+                    # until the decision is actually made.
+                    if self._drying_in_progress.get(
+                        item.printer_id
+                    ) and not await self._drying_may_continue_through_print(db, item.printer_id):
+                        await self._stop_drying(item.printer_id)
+
                     # Queue the dispatch instead of running it here — see
                     # _dispatch_selected(). busy_printers still gets the printer
                     # immediately, so nothing else in this pass can target it.
@@ -1284,7 +1316,7 @@ class PrintScheduler:
                 self._launch_uploads(dispatch_ids, item_printers, upload_limit)
 
             # Auto-drying: start drying on idle printers that have no pending queue items
-            await self._check_auto_drying(db, items, busy_printers, require_plate_clear=require_plate_clear)
+            await self._check_auto_drying(db, items, dispatching_printers)
 
             # Keep the loop on the fast interval while any upload is in flight so
             # a slot freed mid-tick refills within seconds rather than after the
@@ -3050,9 +3082,7 @@ class PrintScheduler:
         self,
         db: AsyncSession,
         queue_items: list[PrintQueueItem],
-        busy_printers: set[int],
-        *,
-        require_plate_clear: bool = True,
+        dispatching_printers: set[int],
     ):
         """Start drying on idle printers based on humidity.
 
@@ -3125,12 +3155,20 @@ class PrintScheduler:
             model = printer_manager.get_model(pid)
             firmware = state.firmware_version
 
-            mid_print = (
-                pid in busy_printers and print_drying_enabled and supports_drying_while_printing(model, firmware)
-            )
-
-            if pid in busy_printers and not mid_print:
-                logger.debug("Auto-drying: printer %d skipped — busy", pid)
+            # "Mid-print" has to mean the printer is actually printing (#2801).
+            # It used to be inferred from the dispatch set, which also holds
+            # printers that merely could not be dispatched to -- so a printer
+            # sitting in FINISH behind an unacknowledged plate was treated as
+            # printing, had its drying temperature capped by the mid-print
+            # spool protection, and was logged as (mid-print) while idle.
+            is_printing = state.state in _ACTIVE_PRINT_STATES
+            mid_print = is_printing and print_drying_enabled and supports_drying_while_printing(model, firmware)
+
+            # A printer whose print is running or imminent is left alone unless
+            # it can dry through it. `dispatching_printers` is deliberately the
+            # narrow set: running, held post-dispatch, or mid-upload.
+            if (is_printing or pid in dispatching_printers) and not mid_print:
+                logger.debug("Auto-drying: printer %d skipped — printing or about to", pid)
                 continue
 
             if not mid_print:
@@ -3149,7 +3187,14 @@ class PrintScheduler:
             if not printer_manager.is_connected(pid):
                 logger.debug("Auto-drying: printer %d skipped — not connected", pid)
                 continue
-            if not mid_print and not self._is_printer_idle(pid, require_plate_clear):
+            # Plate-clear is deliberately ignored here (#2801). It answers
+            # "is the bed ready for the next job", which says nothing about
+            # whether the AMS may heat -- and the gap between a finished print
+            # and the acknowledgment is exactly when drying is most useful,
+            # because the printer is free and nobody is waiting on it. Leaving
+            # the plate unacknowledged is also how people hold the queue by
+            # hand, and that hold should not cost them their drying.
+            if not mid_print and not self._is_printer_idle(pid, require_plate_clear=False):
                 logger.debug("Auto-drying: printer %d skipped — not idle", pid)
                 continue
 
@@ -3279,7 +3324,18 @@ class PrintScheduler:
                             humidity,
                             humidity_threshold,
                         )
-                    self._auto_dry_units.pop(unit_key, None)
+                    # Clear the judgement, keep the clock (#2801). Dropping the
+                    # whole entry also dropped `ended_at`, and with it the
+                    # 30-minute cooldown -- so a reading that dips to the
+                    # threshold as the AMS cools and comes back above it once
+                    # warm wiped its own history and re-armed immediately. That
+                    # oscillation is the very thing #2770's cooldown exists to
+                    # ride out, and it is worst at exactly the margin that makes
+                    # a unit dry repeatedly: a point or two above the threshold.
+                    if unit_state is not None:
+                        unit_state.pop("suspended", None)
+                        unit_state.pop("unproductive", None)
+                        unit_state.pop("best_end_humidity", None)
                     logger.debug(
                         "Auto-drying: printer %d AMS %d skipped — humidity %s <= threshold %d",
                         pid,
@@ -3472,8 +3528,36 @@ class PrintScheduler:
         for key in [k for k in self._auto_dry_units if printer_manager.get_status(k[0]) is None]:
             self._auto_dry_units.pop(key, None)
 
+    async def _drying_may_continue_through_print(self, db: AsyncSession, printer_id: int) -> bool:
+        """True when a running cycle can be left alone while the next print runs.
+
+        Some hardware dries happily through a print and #2758 settled that we
+        should not tear those cycles down: the X2D there was refusing to
+        *start* a job, which is a different problem, and stopping drying before
+        every dispatch would throw away cycles the printer was content to run.
+        Where the model cannot do it, or the user has not enabled it, the print
+        takes priority and the cycle stops -- which is what the queue_drying_block
+        setting has always promised in its off position.
+        """
+        if not await self._get_bool_setting(db, "print_drying_enabled"):
+            return False
+        status = printer_manager.get_status(printer_id)
+        return supports_drying_while_printing(
+            printer_manager.get_model(printer_id),
+            status.firmware_version if status else None,
+        )
+
     async def _stop_drying(self, printer_id: int):
-        """Stop all active drying on a printer (print takes priority)."""
+        """Stop drying cycles Bambuddy armed on a printer (print takes priority).
+
+        Scoped to units in ``_auto_dry_units``. It used to send a stop to every
+        AMS reporting ``dry_time > 0``, which meant one auto-dried unit was
+        enough to kill a cycle the user had started by hand on a *different*
+        unit of the same printer (#2801). That contradicted the contract
+        ``_sync_drying_state`` already documents -- the entry gate deliberately
+        only knows about cycles Bambuddy began, so the action must not reach
+        past them either.
+        """
         state = printer_manager.get_status(printer_id)
         if not state:
             self._drying_in_progress.pop(printer_id, None)
@@ -3484,6 +3568,13 @@ class PrintScheduler:
             dry_time = int(ams_data.get("dry_time") or 0)
             if dry_time > 0:
                 ams_id = int(ams_data.get("id", 0))
+                if (printer_id, ams_id) not in self._auto_dry_units:
+                    logger.debug(
+                        "Auto-drying: leaving printer %d AMS %d alone — not a cycle Bambuddy started",
+                        printer_id,
+                        ams_id,
+                    )
+                    continue
                 logger.info(
                     "Auto-drying: stopping drying on printer %d AMS %d — print takes priority",
                     printer_id,

backend/tests/unit/test_scheduler_auto_drying.py

@@ -199,6 +199,105 @@ class TestSyncDryingState:
         assert 1 not in scheduler._drying_in_progress
 
 
+class TestPlateHoldDoesNotGateDrying:
+    """#2801 — an unacknowledged plate must not stop the AMS heating.
+
+    Plate-clear answers "is the bed ready for the next job". It says nothing
+    about whether filament may be dried, and the gap between a finished print
+    and the acknowledgment is exactly when drying is most useful: the printer
+    is free and nobody is waiting on it. Leaving the plate unacknowledged is
+    also how people hold the queue by hand.
+
+    Before this, such a printer landed in the dispatch set, was read as
+    "currently printing", took the mid-print path -- capped temperature,
+    (mid-print) in the log -- and bypassed the very gate that was meant to
+    hold it, while the queue loop tore the cycle down once a tick.
+    """
+
+    @pytest.fixture
+    def scheduler(self):
+        return PrintScheduler()
+
+    @staticmethod
+    def _finished_printer_state():
+        state = MagicMock()
+        state.state = "FINISH"
+        state.firmware_version = "01.03.00.00"
+        state.raw_data = {
+            "ams": [
+                {
+                    "id": 0,
+                    "module_type": "n3f",
+                    "dry_time": 0,
+                    "humidity_raw": "75",
+                    "dry_sf_reason": [],
+                    "tray": [{"tray_type": "PLA"}],
+                }
+            ]
+        }
+        return state
+
+    def _db(self):
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=TestAmbientDrying._make_db_side_effect(
+                {
+                    "queue_drying_enabled": TestAmbientDrying._make_setting("false"),
+                    "ambient_drying_enabled": TestAmbientDrying._make_setting("true"),
+                    "print_drying_enabled": TestAmbientDrying._make_setting("true"),
+                    "ams_humidity_fair": TestAmbientDrying._make_setting("60"),
+                    "queue_drying_block": TestAmbientDrying._make_setting("false"),
+                    "drying_presets": None,
+                }
+            )
+        )
+        return db
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    @patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
+    async def test_finished_printer_with_dirty_plate_dries_at_full_temperature(self, mock_sd, mock_pm, scheduler):
+        mock_pm.get_status.return_value = self._finished_printer_state()
+        mock_pm.is_connected.return_value = True
+        mock_pm.get_model.return_value = "P2S"
+        mock_pm.send_drying_command.return_value = True
+        scheduler._is_printer_idle = MagicMock(return_value=True)
+
+        await scheduler._check_auto_drying(self._db(), [], set())
+
+        # 45 degC is the uncapped PLA preset: mid-print would have sent 40.
+        mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    @patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
+    async def test_idleness_is_judged_without_the_plate_gate(self, mock_sd, mock_pm, scheduler):
+        mock_pm.get_status.return_value = self._finished_printer_state()
+        mock_pm.is_connected.return_value = True
+        mock_pm.get_model.return_value = "P2S"
+        mock_pm.send_drying_command.return_value = True
+        scheduler._is_printer_idle = MagicMock(return_value=True)
+
+        await scheduler._check_auto_drying(self._db(), [], set())
+
+        scheduler._is_printer_idle.assert_called_with(1, require_plate_clear=False)
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    @patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
+    async def test_a_printer_about_to_print_is_still_left_alone(self, mock_sd, mock_pm, scheduler):
+        """The narrow set keeps its job: an imminent print must not be dried into."""
+        mock_pm.get_status.return_value = self._finished_printer_state()
+        mock_pm.is_connected.return_value = True
+        mock_pm.get_model.return_value = "P2S"
+        mock_pm.send_drying_command.return_value = True
+        scheduler._is_printer_idle = MagicMock(return_value=True)
+
+        await scheduler._check_auto_drying(self._db(), [], {1})
+
+        assert not mock_pm.send_drying_command.called
+
+
 class TestStopDrying:
     """Test _stop_drying — sends stop commands and clears tracking."""
 
@@ -209,8 +308,10 @@ class TestStopDrying:
     @pytest.mark.asyncio
     @patch("backend.app.services.print_scheduler.printer_manager")
     async def test_stops_all_ams_units(self, mock_pm, scheduler):
-        """Sends stop command to each AMS unit that is drying."""
+        """Sends stop command to each auto-armed AMS unit that is drying."""
         scheduler._drying_in_progress = {1: time.monotonic()}
+        scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
+        scheduler._auto_dry_units[(1, 128)] = {"ended_at": None}
         state = MagicMock()
         state.raw_data = {
             "ams": [
@@ -230,6 +331,46 @@ class TestStopDrying:
         assert calls[1].args == (1, 128, 0, 0)
         assert 1 not in scheduler._drying_in_progress
 
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_leaves_cycles_bambuddy_did_not_start(self, mock_pm, scheduler):
+        """A hand-started dry on another unit survives (#2801).
+
+        One auto-dried unit used to be enough to stop every AMS on the
+        printer reporting dry_time > 0, which took the user's own cycle with
+        it. The entry gate only ever knew about cycles Bambuddy began; the
+        action now matches.
+        """
+        scheduler._drying_in_progress = {1: time.monotonic()}
+        scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
+        state = MagicMock()
+        state.raw_data = {"ams": [{"id": 0, "dry_time": 120}, {"id": 1, "dry_time": 600}]}
+        mock_pm.get_status.return_value = state
+
+        await scheduler._stop_drying(1)
+
+        calls = mock_pm.send_drying_command.call_args_list
+        assert [c.args[1] for c in calls] == [0]
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_stops_nothing_it_cannot_prove_it_started(self, mock_pm, scheduler):
+        """After a restart Bambuddy cannot tell its own cycle from a manual one.
+
+        _sync_drying_state prunes but never adopts, for exactly this reason, so
+        a cycle armed before the restart is left running rather than risking a
+        stop on somebody's manual dry. Tracking is still cleared.
+        """
+        scheduler._drying_in_progress = {1: time.monotonic()}
+        state = MagicMock()
+        state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
+        mock_pm.get_status.return_value = state
+
+        await scheduler._stop_drying(1)
+
+        assert not mock_pm.send_drying_command.called
+        assert 1 not in scheduler._drying_in_progress
+
     @pytest.mark.asyncio
     @patch("backend.app.services.print_scheduler.printer_manager")
     async def test_clears_tracking_when_no_state(self, mock_pm, scheduler):
@@ -417,6 +558,8 @@ class TestAutoStopOnFeatureDisabled:
     async def test_stops_drying_when_disabled(self, mock_pm, scheduler):
         """Disabling auto-drying should send stop commands to all drying printers."""
         scheduler._drying_in_progress = {1: time.monotonic(), 2: time.monotonic()}
+        scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
+        scheduler._auto_dry_units[(2, 0)] = {"ended_at": None}
 
         # Printer 1: drying, Printer 2: drying
         def get_status(pid):
@@ -481,6 +624,7 @@ class TestAutoStopOnNoScheduledItems:
     async def test_stops_when_no_scheduled_items(self, mock_pm, scheduler):
         """Auto-drying stops when queue has no scheduled items (queue mode only)."""
         scheduler._drying_in_progress = {1: time.monotonic()}
+        scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
 
         state = MagicMock()
         state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
@@ -510,6 +654,7 @@ class TestAutoStopOnNoScheduledItems:
     async def test_stops_when_empty_queue(self, mock_pm, scheduler):
         """Auto-drying stops when queue is completely empty (queue mode only)."""
         scheduler._drying_in_progress = {1: time.monotonic()}
+        scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
 
         state = MagicMock()
         state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
@@ -692,6 +837,7 @@ class TestAmbientDrying(_DryingTestBase):
     async def test_ambient_off_stops_drying_without_queue(self, mock_pm, scheduler):
         """Disabling ambient drying stops drying on printers without queue items."""
         scheduler._drying_in_progress = {1: time.monotonic()}
+        scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
 
         state = MagicMock()
         state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
@@ -1059,7 +1205,9 @@ class TestMidPrintDrying(_DryingTestBase):
     @patch("backend.app.services.print_scheduler.printer_manager")
     async def test_running_printer_dries_when_enabled_and_capable(self, mock_pm, scheduler):
         """Toggle ON + capable hardware: running printer dries at capped temp."""
-        mock_pm.get_status.return_value = self._state("01.03.00.00")
+        state = self._state("01.03.00.00")
+        state.state = "RUNNING"
+        mock_pm.get_status.return_value = state
         mock_pm.is_connected.return_value = True
         mock_pm.get_model.return_value = "H2D"
         mock_pm.send_drying_command.return_value = True
@@ -1076,7 +1224,7 @@ class TestMidPrintDrying(_DryingTestBase):
         }
         db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
 
-        # Printer 1 is in busy_printers — would normally be skipped
+        # Actually printing (RUNNING), so the mid-print path applies
         await scheduler._check_auto_drying(db, [], {1})
 
         # PLA preset is 45 degC for n3f; mid-print cap is max(40, 45-5) = 40
@@ -1101,6 +1249,7 @@ class TestMidPrintDrying(_DryingTestBase):
             ]
         }
         state.firmware_version = "01.03.00.00"
+        state.state = "RUNNING"
         mock_pm.get_status.return_value = state
         mock_pm.is_connected.return_value = True
         mock_pm.get_model.return_value = "H2D"
@@ -1347,7 +1496,9 @@ class TestAutoDryRearmGuards(_DryingTestBase):
         }
 
         await self._pass(scheduler, mock_pm, db, 0, self.BELOW)
-        assert (1, 0) not in scheduler._auto_dry_units
+        # The judgement is cleared, but the re-arm clock is kept (#2801).
+        assert not scheduler._auto_dry_units[(1, 0)].get("suspended")
+        assert not scheduler._auto_dry_units[(1, 0)].get("unproductive")
 
         await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
         mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")

backend/tests/unit/test_scheduler_drying_plate_hold_2801.py

@@ -0,0 +1,213 @@
+"""Auto-drying versus the plate-clear hold, from check_queue (#2801).
+
+A printer in FINISH whose plate has not been acknowledged, with something
+pending in its queue, stopped and restarted AMS drying once per scheduler tick
+for as long as the plate stayed unacknowledged -- roughly 2000 state changes
+over ten days on the reporter's P2S. Drying never ran long enough to do
+anything, and manual cycles on other AMS units of the same printer were killed
+with it.
+
+Two ideas were tangled together. Plate-clear answers "is the bed ready for the
+next job"; it is not a statement about whether the AMS may heat. And the
+"print takes priority" stop was reached only when the print was NOT going to
+start, so it spent the drying cycle for nothing.
+
+These tests drive the real check_queue so the wiring is covered end to end,
+not just the predicates.
+"""
+
+from contextlib import ExitStack
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import backend.app.models  # noqa: F401 - populate Base.metadata
+from backend.app.core.database import Base
+from backend.app.models.library import LibraryFile
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+async def queue_db():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    async with session_maker() as db:
+        db.add(
+            Printer(
+                id=1,
+                name="P2S-1",
+                serial_number="P2S0001",
+                ip_address="10.0.0.1",
+                access_code="x",
+                model="P2S",
+                is_active=True,
+            )
+        )
+        await db.commit()
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker)
+    finally:
+        await engine.dispose()
+
+
+async def _add_item(ctx):
+    async with ctx.session_maker() as db:
+        lib = LibraryFile(
+            filename="job.gcode.3mf",
+            file_path="/library/job.gcode.3mf",
+            file_size=10,
+            file_type="gcode.3mf",
+            file_metadata={"sliced_for_model": "P2S"},
+        )
+        db.add(lib)
+        await db.flush()
+        db.add(
+            PrintQueueItem(
+                status="pending",
+                position=1,
+                printer_id=1,
+                library_file_id=lib.id,
+            )
+        )
+        await db.commit()
+
+
+async def _run(ctx, scheduler, *, idle, stop_drying, drying=None, deficit=False, launched=None):
+    """One check_queue pass with the plate hold expressed through _is_printer_idle."""
+    patches = [
+        patch("backend.app.services.print_scheduler.async_session", ctx.session_maker),
+        patch("backend.app.core.database.async_session", ctx.session_maker),
+        patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)),
+        patch(
+            "backend.app.services.print_scheduler.ha_sensor_manager.blocked_printers",
+            AsyncMock(return_value={}),
+        ),
+        patch(
+            "backend.app.services.notification_service.notification_service.on_queue_job_waiting",
+            AsyncMock(),
+        ),
+        patch.object(scheduler, "_is_printer_idle", MagicMock(return_value=idle)),
+        patch.object(scheduler, "_check_auto_drying", drying or AsyncMock()),
+        patch.object(scheduler, "_ensure_ams_mapping", AsyncMock(return_value=None)),
+        patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=deficit)),
+        patch.object(scheduler, "_launch_uploads", launched or MagicMock()),
+        patch.object(scheduler, "_stop_drying", stop_drying),
+    ]
+    with ExitStack() as stack:
+        for p in patches:
+            stack.enter_context(p)
+        return await scheduler.check_queue()
+
+
+@pytest.mark.asyncio
+async def test_drying_is_not_stopped_for_a_print_that_cannot_start(queue_db):
+    """The reported loop. Plate unacknowledged, so nothing dispatches -- and
+    stopping the cycle could not have changed that, because drying is not one
+    of the things _is_printer_idle looks at."""
+    await _add_item(queue_db)
+    scheduler = PrintScheduler()
+    scheduler._drying_in_progress[1] = 1.0
+    stop = AsyncMock()
+
+    await _run(queue_db, scheduler, idle=False, stop_drying=stop)
+
+    stop.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_a_plate_held_printer_is_not_offered_to_drying_as_printing(queue_db):
+    """It lands in busy_printers so the queue leaves it alone, but auto-drying
+    is handed the narrow set and must not see it there -- otherwise it takes
+    the mid-print path, which caps the temperature and skips the idle gate."""
+    await _add_item(queue_db)
+    scheduler = PrintScheduler()
+    drying = AsyncMock()
+
+    await _run(queue_db, scheduler, idle=False, stop_drying=AsyncMock(), drying=drying)
+
+    dispatching = drying.await_args[0][2]
+    assert 1 not in dispatching
+
+
+@pytest.mark.asyncio
+async def test_print_takes_priority_still_stops_drying_when_it_can_dispatch(queue_db):
+    """The setting keeps its meaning: on hardware that cannot dry through a
+    print, a dispatch that is actually going to happen stops the cycle."""
+    await _add_item(queue_db)
+    scheduler = PrintScheduler()
+    scheduler._drying_in_progress[1] = 1.0
+    stop = AsyncMock()
+
+    with patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=False)):
+        await _run(queue_db, scheduler, idle=True, stop_drying=stop)
+
+    stop.assert_awaited_once_with(1)
+
+
+@pytest.mark.asyncio
+async def test_block_mode_holds_the_print_and_keeps_the_cycle(queue_db):
+    """queue_drying_block on: the print waits, and the cycle is never touched.
+
+    The setting previously had no observable effect on dispatch -- both
+    branches skipped the item anyway, and all it really decided was whether
+    drying got needlessly killed. Now it does what it says.
+    """
+    await _add_item(queue_db)
+    scheduler = PrintScheduler()
+    scheduler._drying_in_progress[1] = 1.0
+    stop = AsyncMock()
+    launched = MagicMock()
+
+    with patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=True)):
+        await _run(queue_db, scheduler, idle=True, stop_drying=stop, launched=launched)
+
+    stop.assert_not_awaited()
+    assert not launched.called
+
+
+@pytest.mark.asyncio
+async def test_drying_survives_an_item_that_is_skipped_after_the_idle_check(queue_db):
+    """The idle check is not the last thing that can stop a dispatch.
+
+    A failed previous print, an unmappable item, a filament deficit or a
+    contested library row all skip the item further down the loop. Deciding on
+    drying before those is the same defect in a smaller costume: the cycle goes
+    and the print still does not happen.
+    """
+    await _add_item(queue_db)
+    scheduler = PrintScheduler()
+    scheduler._drying_in_progress[1] = 1.0
+    stop = AsyncMock()
+
+    with patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=False)):
+        # Deficit gate holds the item back, after the printer passed as idle.
+        await _run(queue_db, scheduler, idle=True, stop_drying=stop, deficit=True)
+
+    stop.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_capable_hardware_keeps_drying_through_the_print(queue_db):
+    """#2758's finding stands: where the printer dries happily while printing
+    and the user has allowed it, the cycle is left running."""
+    await _add_item(queue_db)
+    scheduler = PrintScheduler()
+    scheduler._drying_in_progress[1] = 1.0
+    stop = AsyncMock()
+
+    with (
+        patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=True)),
+        patch.object(scheduler, "_drying_may_continue_through_print", AsyncMock(return_value=True)),
+    ):
+        await _run(queue_db, scheduler, idle=True, stop_drying=stop)
+
+    stop.assert_not_awaited()