Power on a printer for jobs queued to a printer class (#2786)

    Queue a print against a printer class -- "Any X1C", or a Slicer Pipeline whose
    target type is Printer class -- with every printer of that class switched off,
    and nothing happened. The job sat pending and no smart plug was touched, while
    the same file pinned to a specific printer powered that printer on within one
    queue check. The reporter's log holds both halves: thirteen minutes of the item
    being polled as (133, None, ...) and passed over, then a PATCH onto printer 2,
    then "Printer 2 offline, attempting to power on via smart plug(s)" on the very
    next tick. Same item, same plug, same Auto On setting.

    Powering a printer on had only ever been written inside `if item.printer_id:`.
    The model-based branch below it walks the same queue but its matcher classes an
    offline printer as a reason to keep waiting -- printers_offline collects the
    *name*, for the waiting reason -- and nothing on that path ever looks at plugs.

    _wake_printer_for_model adds it. The model query moves into _printers_for_model
    so the matcher and the wake step answer "which printers can this job run on"
    from one place: a job can only be woken onto a printer the matcher would also
    have considered. Candidates that failed the cross-model gate are excluded --
    switching a printer on for a file that can never legally run on it leaves the
    job just as stuck, with the printer now drawing power.

    Two things it does that the fixed-printer branch does not:

    A printer awaiting plate-clear acknowledgment is skipped. Waking it buys
    nothing; it boots into IDLE and is held by the gate. That is what the reporter's
    log shows for the eighty minutes after their manual edit -- "printer 2 not
    available -- connected=True, state=IDLE, awaiting_plate_clear=True" every thirty
    seconds to the end of the capture. The flag is Bambuddy-side and persisted, so
    it is readable while the printer is still off.

    At most one printer per pass, because each wake blocks the queue loop for the
    boot wait. Several queued jobs bring several printers up over the following
    minutes rather than a whole shelf at once.

    A failed power-on opens a 600s per-printer cool-off. Without it the walk is by
    id, the pass spends its single attempt on the same broken printer every time,
    and a healthy sibling two slots down is never reached -- one unreachable plug
    starves its whole model, and costs a 180s boot timeout out of every 30s pass.
    Entries expire on read: a printer inside its cool-off is skipped before the
    power-on is reached, so a live entry can never be overwritten by a success.

    The failed printer is deliberately NOT added to busy_printers. It is off, not
    busy; labelling it busy would misdescribe it in every later item's waiting
    reason and, because an all-busy reason is treated as needing no user action,
    suppress the notification too.

    Assignment is left to the next pass. AMS trays arrive with the first status push
    after connect, so matching filament against a printer that booted five seconds
    ago can reject the printer we just woke.

    Finally, the waiting reason separates "Offline: X1C-1" from "Offline, no Auto On
    smart plug: X1C-2". Those are different problems and only the second is one the
    user has to go and fix -- it was also the first question the reporter had to be
    asked, and the queue could not answer it.

    Tests cover the wake, the plate-clear skip in both gate states, all-candidates-
    awaiting-plate-clear waking nothing, one wake per pass, the starvation case over
    two passes, cool-off expiry, no-Auto-On-plug being left alone and named, an
    incompatible sliced model waking nothing, connected printers being left alone,
    scheduled-for-later and manual-start jobs switching nothing on, and a regression
    pin on the fixed-printer branch.

backend/app/services/print_scheduler.py

@@ -474,6 +474,23 @@ class PrintScheduler:
         self._fast_check_interval = 3  # seconds
         self._power_on_wait_time = 180  # seconds to wait for printer after power on (3 min)
         self._power_on_check_interval = 10  # seconds between connection checks
+        # Printers whose class-target power-on failed, mapped to the monotonic
+        # time their cool-off expires (#2786).
+        #
+        # Without this, one printer with an unreachable plug starves every
+        # sibling of its model forever: the wake step walks candidates in id
+        # order, spends the pass's single attempt on the same broken printer
+        # every time, and the healthy one two slots down is never reached. It
+        # also costs a full ``_power_on_wait_time`` out of every 30 s pass,
+        # which delays the whole queue, not just this job.
+        #
+        # Entries expire on read rather than being cleared on success: a printer
+        # inside its cool-off is skipped before the power-on is reached, so a
+        # live entry can never be overwritten by a success anyway. A printer
+        # that comes back by any other route stops being a wake candidate the
+        # moment it connects.
+        self._wake_failures: dict[int, float] = {}
+        self._wake_failure_cooloff = 600  # seconds
         # Track which printers are currently auto-drying (printer_id -> start timestamp)
         self._drying_in_progress: dict[int, float] = {}
         # Defensive in-memory dispatch hold (#1157): a printer that just received
@@ -739,6 +756,17 @@ class PrintScheduler:
                 logger.warning("Home Assistant interlock check failed: %s", e)
                 interlocked = {}
 
+            # Printers a smart plug can bring back, read once for the whole pass
+            # (#2786). Used by the model-based branch both to word "Offline" in
+            # the waiting reason and to decide what the wake step may switch on.
+            wakeable_printer_ids = await self._wakeable_printer_ids(db)
+
+            # At most one power-on per queue check. Each one blocks this loop
+            # for the boot wait, so a queue of ten class-targeted jobs must not
+            # switch on ten printers inside a single pass — the next pass wakes
+            # the next one (#2786).
+            power_on_attempted = False
+
             # Log skip reasons once per queue check (not per item)
             skip_reasons: dict[str, int] = {}
 
@@ -962,6 +990,11 @@ class PrintScheduler:
                     printer_id = None
                     chosen: _ModelCandidate | None = None
                     per_model_reasons: list[tuple[str | None, str]] = []
+                    # Candidates that cleared the cross-model gate below. The
+                    # smart-plug wake step may only consider these — waking a
+                    # printer for a file that can never legally run on it is
+                    # worse than not waking at all (#2786).
+                    wakeable_candidates: list[_ModelCandidate] = []
 
                     if not candidates:
                         # Every candidate file has been deleted or trashed out from
@@ -1015,6 +1048,7 @@ class PrintScheduler:
                             skip_reasons["sliced_model_mismatch"] = skip_reasons.get("sliced_model_mismatch", 0) + 1
                             continue
 
+                        wakeable_candidates.append(candidate)
                         match_id, match_reason = await self._find_idle_printer_for_model(
                             db,
                             candidate.target_model,
@@ -1026,6 +1060,7 @@ class PrintScheduler:
                             item.target_location,
                             filament_overrides=filament_overrides,
                             require_plate_clear=require_plate_clear,
+                            wakeable_ids=wakeable_printer_ids,
                         )
                         if match_id:
                             printer_id = match_id
@@ -1033,6 +1068,34 @@ class PrintScheduler:
                             break
                         per_model_reasons.append((candidate.target_model, match_reason or ""))
 
+                    # Nothing is available and nothing has been woken this pass:
+                    # switch one matching printer on. Assignment is left to the
+                    # next pass, which sees the booted printer's live state
+                    # instead of guessing at it seconds after connect (#2786).
+                    if printer_id is None and not power_on_attempted and wakeable_candidates:
+                        woken_id, attempted_id = await self._wake_printer_for_model(
+                            db,
+                            wakeable_candidates,
+                            item.target_location,
+                            busy_printers | interlocked.keys(),
+                            wakeable_printer_ids,
+                            require_plate_clear,
+                        )
+                        # An attempt spends the pass's one wake whether or not
+                        # it worked: it has already blocked the queue loop for
+                        # the boot wait. A failed printer is held out of later
+                        # passes by its own cool-off, deliberately NOT by
+                        # busy_printers — it is off, not busy, and labelling it
+                        # busy would both misdescribe it in every later item's
+                        # waiting reason and suppress the notification, since
+                        # an all-busy reason is treated as needing no action.
+                        power_on_attempted = attempted_id is not None
+                        if woken_id is not None:
+                            # Hold this item back rather than dispatching onto a
+                            # printer whose AMS has not reported yet.
+                            skip_reasons["powered_on_printer"] = skip_reasons.get("powered_on_printer", 0) + 1
+                            continue
+
                     waiting_reason = None if printer_id else _collapse_waiting_reasons(per_model_reasons)
 
                     # Fold the winning variant's file and settings onto the item
@@ -1414,6 +1477,149 @@ class PrintScheduler:
                     return
                 await asyncio.sleep(0.5 * attempt)
 
+    async def _printers_for_model(
+        self,
+        db: AsyncSession,
+        model: str,
+        target_location: str | None = None,
+    ) -> list[Printer]:
+        """Active printers of *model*, optionally narrowed to one location.
+
+        Shared by the matcher and by the smart-plug wake step (#2786) so both
+        answer "which printers can this job run on" from one query — a job can
+        only be woken onto a printer the matcher would also have considered.
+        """
+        normalized_model = normalize_printer_model(model) or model
+        query = (
+            select(Printer)
+            .where(func.lower(Printer.model) == normalized_model.lower())
+            .where(Printer.is_active == True)  # noqa: E712
+        )
+        if target_location:
+            query = query.where(Printer.location == target_location)
+        result = await db.execute(query)
+        return list(result.scalars().all())
+
+    async def _wakeable_printer_ids(self, db: AsyncSession) -> set[int]:
+        """Printer IDs that at least one enabled ``auto_on`` plug can power on.
+
+        Read once per queue check rather than per printer: it decides both
+        whether the wake step has anything to do and how an offline printer is
+        worded in the waiting reason — "Offline" and "offline with no Auto On
+        plug" are different problems, and the second is the one the user has to
+        fix themselves (#2786).
+        """
+        result = await db.execute(
+            select(SmartPlug.printer_id)
+            .where(SmartPlug.printer_id.is_not(None))
+            .where(SmartPlug.enabled == True)  # noqa: E712
+            .where(SmartPlug.auto_on == True)  # noqa: E712
+        )
+        return {pid for (pid,) in result.all() if pid is not None}
+
+    def _wake_recently_failed(self, printer_id: int) -> bool:
+        """True while this printer's failed power-on is still cooling off (#2786)."""
+        deadline = self._wake_failures.get(printer_id)
+        if deadline is None:
+            return False
+        if time.monotonic() >= deadline:
+            del self._wake_failures[printer_id]
+            return False
+        return True
+
+    async def _wake_printer_for_model(
+        self,
+        db: AsyncSession,
+        candidates: list[_ModelCandidate],
+        target_location: str | None,
+        exclude_ids: set[int],
+        wakeable_ids: set[int],
+        require_plate_clear: bool,
+    ) -> tuple[int | None, int | None]:
+        """Power on one offline printer a model-based item could run on (#2786).
+
+        The fixed-printer branch has powered a printer on since smart plugs
+        existed. The model-based branch never could: its matcher drops an
+        offline printer into the "Offline:" waiting reason and nothing looks at
+        its plugs, so a class-targeted job with every matching printer switched
+        off sat pending forever. The reporter's log is the controlled
+        experiment — the same item, same plug, same Auto On setting, dispatched
+        the moment they edited it onto a specific printer.
+
+        Returns ``(woken_id, attempted_id)``. ``attempted_id`` is set whenever a
+        power-on was actually tried, so the caller can tell "nothing here was
+        wakeable" (both None — cheap, other items may still find something)
+        from "we tried and it did not come up" (only ``attempted_id`` — the
+        boot timeout has already been spent).
+
+        Deliberately does NOT go on to match the job: AMS trays arrive with the
+        first status push after connect, so a filament check against a printer
+        that booted seconds ago can reject the printer we just woke. The next
+        queue pass matches it with live state.
+
+        At most one printer per pass. Each wake blocks the queue loop for the
+        boot wait, and a queue of ten class-targeted jobs must not switch on
+        ten printers inside one check.
+        """
+        for candidate in candidates:
+            if not candidate.target_model:
+                continue
+            printers = await self._printers_for_model(db, candidate.target_model, target_location)
+            for printer in sorted(printers, key=lambda p: p.id):
+                if printer.id in exclude_ids or printer.id not in wakeable_ids:
+                    continue
+                if printer_manager.is_connected(printer.id):
+                    continue
+                if self._wake_recently_failed(printer.id):
+                    # Its plug did not bring it back a moment ago. Move on to a
+                    # sibling instead of spending this pass — and every pass —
+                    # on the same printer.
+                    continue
+                if require_plate_clear and printer_manager.is_awaiting_plate_clear(printer.id):
+                    # Waking this one buys nothing: it would boot into IDLE and
+                    # then be held by the plate-clear gate, which is exactly
+                    # what the reporter's log shows happening for 80 minutes
+                    # after a fixed-printer wake. The flag is Bambuddy-side and
+                    # persisted, so it is readable while the printer is off.
+                    logger.info(
+                        "Not powering on printer %s for a %s job: it is awaiting plate-clear acknowledgment",
+                        printer.id,
+                        candidate.target_model,
+                    )
+                    continue
+
+                plugs = await self._get_smart_plugs(db, printer.id)
+                auto_on_plugs = [p for p in plugs if p.auto_on and p.enabled]
+                if not auto_on_plugs:
+                    # wakeable_ids said otherwise — the plug changed under us
+                    # mid-pass. Nothing to do but move on.
+                    continue
+
+                logger.info(
+                    "No %s printer available for a queued job; powering on offline printer %s via smart plug(s)",
+                    candidate.target_model,
+                    printer.id,
+                )
+                primary_plug = self._pick_power_plug(auto_on_plugs)
+                if not await self._power_on_and_wait(primary_plug, printer.id, db):
+                    logger.warning(
+                        "Could not power on printer %s via smart plug; not trying it again for %ss",
+                        printer.id,
+                        self._wake_failure_cooloff,
+                    )
+                    self._wake_failures[printer.id] = time.monotonic() + self._wake_failure_cooloff
+                    return None, printer.id
+
+                for extra_plug in [p for p in auto_on_plugs if p.id != primary_plug.id]:
+                    try:
+                        service = await smart_plug_manager.get_service_for_plug(extra_plug, db)
+                        await service.turn_on(extra_plug)
+                        logger.info("Also powered on plug '%s' for printer %s", extra_plug.name, printer.id)
+                    except Exception as e:
+                        logger.warning("Failed to power on extra plug '%s': %s", extra_plug.name, e)
+                return printer.id, printer.id
+        return None, None
+
     async def _find_idle_printer_for_model(
         self,
         db: AsyncSession,
@@ -1423,6 +1629,7 @@ class PrintScheduler:
         target_location: str | None = None,
         filament_overrides: list[dict] | None = None,
         require_plate_clear: bool = True,
+        wakeable_ids: set[int] | None = None,
     ) -> tuple[int | None, str | None]:
         """Find an idle, connected printer matching the model with compatible filaments.
 
@@ -1437,26 +1644,17 @@ class PrintScheduler:
                                  ``force_color_match: true`` to require an exact type+color match
                                  on the printer for that slot. Without the flag the existing
                                  colour-preference logic applies.
+            wakeable_ids: Printers a smart plug can power on (#2786). Only changes how an
+                          offline printer is worded: one Bambuddy will switch on reads
+                          differently from one the user has to go and switch on themselves.
 
         Returns:
             Tuple of (printer_id, waiting_reason):
             - (printer_id, None) if a matching printer was found
             - (None, reason) if no printer is available, with explanation
         """
-        # Normalize model name and use case-insensitive matching
         normalized_model = normalize_printer_model(model) or model
-        query = (
-            select(Printer)
-            .where(func.lower(Printer.model) == normalized_model.lower())
-            .where(Printer.is_active == True)  # noqa: E712
-        )
-
-        # Add location filter if specified
-        if target_location:
-            query = query.where(Printer.location == target_location)
-
-        result = await db.execute(query)
-        printers = list(result.scalars().all())
+        printers = await self._printers_for_model(db, model, target_location)
 
         location_suffix = f" in {target_location}" if target_location else ""
         if not printers:
@@ -1469,6 +1667,7 @@ class PrintScheduler:
         # Track reasons for skipping printers
         printers_busy = []
         printers_offline = []
+        printers_offline_no_plug = []
         printers_missing_filament: list[tuple[str, list[str]]] = []
         candidates: list[tuple[int, int]] = []  # (printer_id, color_match_count)
 
@@ -1490,7 +1689,10 @@ class PrintScheduler:
             is_idle = self._is_printer_idle(printer.id, require_plate_clear) if is_connected else False
 
             if not is_connected:
-                printers_offline.append(printer.name)
+                if wakeable_ids is not None and printer.id not in wakeable_ids:
+                    printers_offline_no_plug.append(printer.name)
+                else:
+                    printers_offline.append(printer.name)
                 continue
 
             if not is_idle:
@@ -1590,6 +1792,11 @@ class PrintScheduler:
             reasons.append(f"Busy: {', '.join(printers_busy)}")
         if printers_offline:
             reasons.append(f"Offline: {', '.join(printers_offline)}")
+        if printers_offline_no_plug:
+            # Named separately because it is the one entry on this list the
+            # user has to act on: no enabled Auto On plug means Bambuddy will
+            # never power this printer on for the queue (#2786).
+            reasons.append(f"Offline, no Auto On smart plug: {', '.join(printers_offline_no_plug)}")
 
         return None, " | ".join(reasons) if reasons else f"No available {model} printers{location_suffix}"
 

backend/tests/unit/test_scheduler_class_target_smart_plug_2786.py

@@ -0,0 +1,424 @@
+"""Smart-plug power-on for class-targeted queue items (#2786).
+
+Powering a printer on for a queued job has existed since smart plugs did, but
+only on the branch that handles an item pinned to one printer. An item queued
+as "Any X1C" carries no ``printer_id``, takes the model-based branch, and that
+branch's matcher drops an offline printer into a "Offline:" waiting reason
+without ever looking at its plugs. With every matching printer switched off the
+job sat pending indefinitely.
+
+The reporter's log is the controlled experiment: the same item, same plug, same
+Auto On setting, powered a printer on the moment they edited it onto a specific
+printer -- and did nothing for the thirteen minutes before that.
+"""
+
+from contextlib import ExitStack
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy import select
+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.models.smart_plug import SmartPlug
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+async def queue_db():
+    """Two X1Cs, each on its own plug, so "which one" is a real question."""
+    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_all(
+            [
+                Printer(
+                    id=1,
+                    name="X1C-1",
+                    serial_number="X1C0001",
+                    ip_address="10.0.0.1",
+                    access_code="x",
+                    model="X1C",
+                    is_active=True,
+                ),
+                Printer(
+                    id=2,
+                    name="X1C-2",
+                    serial_number="X1C0002",
+                    ip_address="10.0.0.2",
+                    access_code="x",
+                    model="X1C",
+                    is_active=True,
+                ),
+            ]
+        )
+        await db.commit()
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker)
+    finally:
+        await engine.dispose()
+
+
+async def _add_plug(ctx, printer_id, *, auto_on=True, enabled=True, name=None):
+    async with ctx.session_maker() as db:
+        plug = SmartPlug(
+            name=name or f"Plug {printer_id}",
+            plug_type="tasmota",
+            ip_address=f"10.0.1.{printer_id}",
+            printer_id=printer_id,
+            enabled=enabled,
+            auto_on=auto_on,
+        )
+        db.add(plug)
+        await db.commit()
+        return plug.id
+
+
+async def _add_item(
+    ctx, *, printer_id=None, target_model=None, sliced_for="X1C", position=1, scheduled_time=None, manual_start=False
+):
+    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": sliced_for},
+        )
+        db.add(lib)
+        await db.flush()
+        item = PrintQueueItem(
+            status="pending",
+            position=position,
+            printer_id=printer_id,
+            target_model=target_model,
+            library_file_id=lib.id,
+            scheduled_time=scheduled_time,
+            manual_start=manual_start,
+        )
+        db.add(item)
+        await db.commit()
+        return item.id
+
+
+async def _run(
+    ctx,
+    scheduler,
+    *,
+    power_on=AsyncMock,
+    connected=False,
+    awaiting_plate_clear=(),
+    require_plate_clear=True,
+    launched=None,
+):
+    """Run one queue pass with every printer offline unless told otherwise.
+
+    ``power_on`` is the patched ``_power_on_and_wait``; the tests assert on the
+    printer ids it was called with, which is the whole behaviour under test.
+    """
+    power_on_mock = power_on() if isinstance(power_on, type) else power_on
+    with ExitStack() as stack:
+        for p in [
+            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(side_effect=lambda pid: pid in connected if connected else False),
+            ),
+            patch(
+                "backend.app.services.print_scheduler.printer_manager.is_awaiting_plate_clear",
+                MagicMock(side_effect=lambda pid: pid in awaiting_plate_clear),
+            ),
+            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(
+                "backend.app.services.notification_service.notification_service.on_queue_job_assigned",
+                AsyncMock(),
+            ),
+            patch.object(scheduler, "_check_auto_drying", AsyncMock()),
+            patch.object(scheduler, "_ensure_ams_mapping", AsyncMock(return_value=None)),
+            patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=False)),
+            patch.object(scheduler, "_launch_uploads", launched or MagicMock()),
+            patch.object(scheduler, "_power_on_and_wait", power_on_mock),
+            patch.object(
+                scheduler,
+                "_get_bool_setting",
+                AsyncMock(
+                    side_effect=lambda db, key, default=False: (
+                        require_plate_clear if key == "require_plate_clear" else default
+                    )
+                ),
+            ),
+        ]:
+            stack.enter_context(p)
+        await scheduler.check_queue()
+    return power_on_mock
+
+
+async def _get_item(ctx, item_id):
+    async with ctx.session_maker() as db:
+        return (await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one()
+
+
+def _woken_printer_ids(power_on_mock):
+    """Printer ids ``_power_on_and_wait(plug, printer_id, db)`` was called for."""
+    return [call.args[1] for call in power_on_mock.await_args_list]
+
+
+class TestClassTargetWakesAPrinter:
+    @pytest.mark.asyncio
+    async def test_offline_printers_are_powered_on_for_an_any_model_job(self, queue_db):
+        """The bug: this used to do nothing at all."""
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        item_id = await _add_item(queue_db, target_model="X1C")
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        assert _woken_printer_ids(power_on) == [1]
+        # Assignment is left to the next pass, once the printer has reported.
+        item = await _get_item(queue_db, item_id)
+        assert item.status == "pending"
+        assert item.printer_id is None
+
+    @pytest.mark.asyncio
+    async def test_a_printer_awaiting_plate_clear_is_passed_over(self, queue_db):
+        """Waking it buys nothing -- the plate-clear gate would hold it anyway.
+
+        This is what the reporter's log shows after a fixed-printer wake: the
+        printer booted and then reported ``awaiting_plate_clear=True`` every 30
+        seconds for the next 80 minutes. The flag is Bambuddy-side and
+        persisted, so it is readable while the printer is still switched off.
+        """
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        await _add_item(queue_db, target_model="X1C")
+
+        power_on = await _run(
+            queue_db,
+            PrintScheduler(),
+            power_on=AsyncMock(return_value=True),
+            awaiting_plate_clear=(1,),
+        )
+
+        assert _woken_printer_ids(power_on) == [2]
+
+    @pytest.mark.asyncio
+    async def test_nothing_is_woken_when_every_candidate_awaits_plate_clear(self, queue_db):
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        await _add_item(queue_db, target_model="X1C")
+
+        power_on = await _run(
+            queue_db,
+            PrintScheduler(),
+            power_on=AsyncMock(return_value=True),
+            awaiting_plate_clear=(1, 2),
+        )
+
+        power_on.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_plate_clear_gate_off_wakes_anyway(self, queue_db):
+        """With the gate disabled the flag is not a reason to skip a printer."""
+        await _add_plug(queue_db, 1)
+        await _add_item(queue_db, target_model="X1C")
+
+        power_on = await _run(
+            queue_db,
+            PrintScheduler(),
+            power_on=AsyncMock(return_value=True),
+            awaiting_plate_clear=(1,),
+            require_plate_clear=False,
+        )
+
+        assert _woken_printer_ids(power_on) == [1]
+
+    @pytest.mark.asyncio
+    async def test_at_most_one_printer_is_woken_per_pass(self, queue_db):
+        """Each wake blocks the queue loop for the boot wait.
+
+        Ten class-targeted jobs must not switch on ten printers inside one
+        check; the next pass wakes the next one.
+        """
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        await _add_item(queue_db, target_model="X1C", position=1)
+        await _add_item(queue_db, target_model="X1C", position=2)
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        assert _woken_printer_ids(power_on) == [1]
+
+    @pytest.mark.asyncio
+    async def test_a_failed_power_on_is_not_retried_in_the_same_pass(self, queue_db):
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        item_a = await _add_item(queue_db, target_model="X1C", position=1)
+        item_b = await _add_item(queue_db, target_model="X1C", position=2)
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=False))
+
+        assert _woken_printer_ids(power_on) == [1]
+        # A printer we failed to switch on is off, not busy. Calling it busy
+        # would misdescribe it here and, because an all-busy reason is treated
+        # as needing no user action, silence the notification as well.
+        for item_id in (item_a, item_b):
+            reason = (await _get_item(queue_db, item_id)).waiting_reason or ""
+            assert "Busy" not in reason
+            assert "Offline" in reason
+
+    @pytest.mark.asyncio
+    async def test_a_dead_plug_does_not_starve_its_siblings(self, queue_db):
+        """One unreachable plug must not hold every sibling of its model hostage.
+
+        Candidates are walked in id order and a pass spends only one power-on
+        attempt, so without a cool-off the broken printer is picked again on
+        every pass and the healthy one behind it is never reached. It also
+        costs a full boot timeout out of each 30s pass, which delays the whole
+        queue rather than just this job.
+        """
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        await _add_item(queue_db, target_model="X1C")
+
+        scheduler = PrintScheduler()
+        power_on = AsyncMock(return_value=False)
+        await _run(queue_db, scheduler, power_on=power_on)
+        await _run(queue_db, scheduler, power_on=power_on)
+
+        assert _woken_printer_ids(power_on) == [1, 2]
+
+    @pytest.mark.asyncio
+    async def test_a_printer_is_tried_again_once_its_cooloff_expires(self, queue_db):
+        """The skip is a cool-off, not a blacklist — a fixed plug is picked up."""
+        await _add_plug(queue_db, 1)
+        await _add_item(queue_db, target_model="X1C")
+
+        scheduler = PrintScheduler()
+        scheduler._wake_failure_cooloff = 0
+        power_on = AsyncMock(return_value=False)
+        await _run(queue_db, scheduler, power_on=power_on)
+        await _run(queue_db, scheduler, power_on=power_on)
+
+        assert _woken_printer_ids(power_on) == [1, 1]
+
+    @pytest.mark.asyncio
+    async def test_an_expired_cooloff_is_not_left_behind(self, queue_db):
+        """The map holds one key per currently-failing printer, not per printer
+        this process has ever failed to wake."""
+        await _add_plug(queue_db, 1)
+        await _add_item(queue_db, target_model="X1C")
+
+        scheduler = PrintScheduler()
+        scheduler._wake_failure_cooloff = 0
+        await _run(queue_db, scheduler, power_on=AsyncMock(return_value=False))
+        assert 1 in scheduler._wake_failures
+
+        await _run(queue_db, scheduler, power_on=AsyncMock(return_value=True))
+        assert scheduler._wake_failures == {}
+
+
+class TestWhatIsNotWokenUp:
+    @pytest.mark.asyncio
+    async def test_a_printer_with_no_auto_on_plug_is_left_alone_and_said_so(self, queue_db):
+        """The first question asked of the reporter was whether Auto On was on.
+
+        "Offline" and "offline with no Auto On plug" are different problems and
+        only the second is one the user has to go and fix, so they must not
+        share a waiting reason.
+        """
+        await _add_plug(queue_db, 1, auto_on=False)
+        await _add_plug(queue_db, 2, enabled=False)
+        item_id = await _add_item(queue_db, target_model="X1C")
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        power_on.assert_not_awaited()
+        item = await _get_item(queue_db, item_id)
+        assert item.waiting_reason is not None
+        assert "no Auto On smart plug" in item.waiting_reason
+        assert "X1C-1" in item.waiting_reason and "X1C-2" in item.waiting_reason
+
+    @pytest.mark.asyncio
+    async def test_an_incompatible_file_never_wakes_anything(self, queue_db):
+        """The cross-model gate (#2578) runs before the wake, not after it.
+
+        Switching a printer on for a file that can never legally run on it is
+        worse than leaving it off: the job still cannot start, and now the
+        printer is drawing power.
+        """
+        await _add_plug(queue_db, 1)
+        await _add_item(queue_db, target_model="X1C", sliced_for="A1")
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        power_on.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_a_job_scheduled_for_later_does_not_switch_anything_on_now(self, queue_db):
+        """Otherwise a print set for 3am powers a printer up the moment it is queued."""
+        await _add_plug(queue_db, 1)
+        await _add_item(
+            queue_db,
+            target_model="X1C",
+            scheduled_time=datetime.now(timezone.utc) + timedelta(hours=6),
+        )
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        power_on.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_a_manual_start_job_does_not_switch_anything_on(self, queue_db):
+        """Manual start means the user presses play; nothing happens until they do."""
+        await _add_plug(queue_db, 1)
+        await _add_item(queue_db, target_model="X1C", manual_start=True)
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        power_on.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_an_already_connected_printer_is_not_powered_on(self, queue_db):
+        await _add_plug(queue_db, 1)
+        await _add_plug(queue_db, 2)
+        await _add_item(queue_db, target_model="X1C")
+
+        power_on = await _run(
+            queue_db,
+            PrintScheduler(),
+            power_on=AsyncMock(return_value=True),
+            connected=(1, 2),
+        )
+
+        power_on.assert_not_awaited()
+
+
+class TestFixedPrinterBranchStillWakes:
+    @pytest.mark.asyncio
+    async def test_an_item_pinned_to_a_printer_still_powers_it_on(self, queue_db):
+        """The branch that always worked, pinned so a refactor cannot drop it."""
+        await _add_plug(queue_db, 2)
+        await _add_item(queue_db, printer_id=2)
+
+        power_on = await _run(queue_db, PrintScheduler(), power_on=AsyncMock(return_value=True))
+
+        assert _woken_printer_ids(power_on) == [2]