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.

CHANGELOG.md

@@ -19,6 +19,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Error and warning toasts now stay up twice as long** — Every pop-up notification disappeared after three seconds regardless of what it said. That is about right for "Settings saved", which confirms something you just did and is skimmed rather than read, but errors and warnings are a different kind of message: they carry a reason, often one relayed from the printer or the backend, and they run to a couple of lines. Three seconds was not long enough to finish reading one, and a missed error message is gone for good — there is no notification history to go back to. Errors and warnings now hold for six seconds. Success and informational toasts keep the three-second default, so the common case of clicking something and seeing it confirmed is unchanged, and the close button and the manual dismiss work exactly as before on all of them. The background print-dispatch toast is unaffected: it stays up while it has work in progress and clears itself shortly after the last job settles. Covered by frontend tests.
 
 ### Fixed
+- **A job queued to "Any {model}" now switches a printer on, like a job queued to one printer always has (#2786, reported by @TheUltimateC0der)** — 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 at the wall, and nothing happened. The job sat pending, no smart plug was touched, and the only way out was to edit the item onto a specific printer, at which point Bambuddy powered that printer on immediately. The reporter's log holds that comparison exactly: thirteen minutes of the job being polled and passed over, then the edit, then a power-on on the very next check -- same job, same plug, same Auto Power On setting. Powering a printer on had only ever been written into the branch that handles a job pinned to one printer; the branch that picks a printer by model listed an offline one as a reason to keep waiting and never looked at its plugs. It does now. It also picks with a little more care than the older branch: a printer waiting for a plate-clear acknowledgment is passed over, because switching it on only leaves it idling behind that gate -- which is what the reporter's own log shows happening for the eighty minutes after their manual edit -- and a printer whose class the file cannot legally run on is never switched on at all. One printer comes up per queue check rather than a whole shelf at once, so several queued jobs wake several printers over the following minutes. Finally, a printer that is off and has no enabled Auto Power On plug now says that in the job's waiting reason instead of hiding behind the same "Offline" as the printers Bambuddy can bring back itself -- that distinction was the first question the reporter had to be asked.
 - **A printer whose file service stops answering is named as such instead of quietly emptying your archives (#2780, reported by @Utility9298 and @AntonPalmqvist)** — Two printers went on printing normally while every archive they produced arrived holding nothing but a filename: no filament totals, no layer count, no cover image, no timelapse. The Connection Diagnostic reported the file-transfer port as reachable, because it was — the printer accepted the connection and then answered it with something that was not TLS at all, and the sliced file could never be read back. Bambuddy said nothing about that on screen; it retried. Because each candidate location opened its own connection, one reporter's log carried 1813 identical handshake failures and another's 3511, against a printer that could not have answered any of them. This is not a model or a firmware problem — the same printers worked for days before and after the fault, and other installs run the same models untouched. It is the printer's own file service getting stuck, and a power-cycle clears it. Bambuddy now stops after the first failed handshake and leaves that printer alone for five minutes, so the log carries a handful of entries that say what went wrong and what to do about it rather than thousands that say neither. The Connection Diagnostic now completes a real handshake instead of merely opening the port, so a printer in this state reads as a warning that names a restart — not as a green tick. Scanning for a timelapse on such a printer reports the file service, where it used to return one error message that covered both "the printer is unreachable" and "this printer has no timelapse folder", and asking for a cover image says the same thing instead of the "no cover for this print" it used to claim. Printing is unaffected throughout: the control connection is a separate service, which is exactly why the fault was invisible.
 - **Prints queued from a Slicer Pipeline or the Library are checked for enough filament again (#2779, reported by @wylyn3d)** — A job needing 20.5 g was dispatched onto a spool holding 9 g, and the printer started. The same file, printed from the Print dialog, was correctly refused. The check that stands between the queue and the printer reads the sliced file to learn how much each slot needs, and it looked for that file in the wrong place: a file in the Library records where it lives relative to Bambuddy's data directory, and this one check read that as a path from wherever the process happened to be running. It found nothing, and a source it cannot find has always meant "nothing to verify" rather than "stop" — so the job passed a check that never actually ran. Every path that queues a Library file was affected: Slicer Pipeline jobs, which are always Library-backed, and anything added through the Library's **Add to queue**. Both the automatic dispatcher and the Play button on the queue were equally blind, so the deficit could not be caught by starting the job by hand either. Prints queued from print history were never affected, and neither was the Print dialog, which finds the file its own way. Two things changed: the check now resolves a Library file the same way the eleven other places that read one already did, and a source file that is configured but missing now writes a warning to the log naming the item and the path it looked at. That case still dispatches — the upload needs the same file moments later and fails there, where blocking would strand a queue on a file the user may have moved — but it no longer passes in silence, which is what let this go unnoticed. Covered by backend tests, including the reporter's exact 20.5 g against 9 g.
 - **A Forgejo token limited to a single repository can now be used for backups (#2775, reported by @AnthonyGrondin)** — Forgejo v15 lets you mint an access token that only reaches one repository, which is the safest token you can give a backup: leak it and the damage stops at the repository it was made for. Bambuddy refused it. **Test connection** asked Forgejo who the token belonged to before it asked whether the token could reach the repository, and a repository-scoped token is not allowed to answer that question — it may only carry read and write on issues and repositories — so the check failed on a token that would have backed up perfectly well. The identity question is now asked but no longer decides: only an outright rejection of the token is conclusive, and everything else falls through to the repository check, which is the one that matters. Nothing else in a backup ever needed the wider permission — the push writes through the repository's own contents endpoint and a restore reads its commits, trees and blobs — so ordinary tokens are unaffected. The message shown when the repository cannot be reached now names the scope to look for and the possibility that the token is scoped to a different repository, rather than only explaining Forgejo's habit of reporting a private repository as missing. The hint under the token field is also per provider now: it read "fine-grained token with Contents read/write" for all four, advice that only ever applied to GitHub, and now names GitHub's, GitLab's, Gitea's and Forgejo's own scopes in every language Bambuddy speaks. Covered by backend and frontend tests.

backend/app/services/print_scheduler.py

@@ -466,6 +466,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
@@ -726,6 +743,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] = {}
 
@@ -949,6 +977,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
@@ -1002,6 +1035,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,
@@ -1013,6 +1047,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
@@ -1020,6 +1055,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
@@ -1364,6 +1427,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,
@@ -1373,6 +1579,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.
 
@@ -1387,26 +1594,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:
@@ -1419,6 +1617,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)
 
@@ -1440,7 +1639,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:
@@ -1540,6 +1742,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]