Add cross-model variant resolution to the queue scheduler (#671)

Adds print_queue_variants: the candidate files a queue item may run, each
with its own model, plate, AMS mapping and nozzle mapping. The scheduler
walks them in priority order and takes the first whose model has an idle
printer, then folds that candidate onto the queue row before the
selection commit — so upload, archive creation, print history and reprint
keep seeing an ordinary single-file item.

Candidates are ordered least-attempted first, so a printer that accepts
the file and never starts hands the job to the alternative on the next
lap instead of spending the item's whole retry budget on the machine
that is wedged. The item-level DISPATCH_MAX_ATTEMPTS bound is unchanged.

An item whose candidate files have all been deleted is held pending with
an actionable reason rather than failing deep in the upload, and waiting
notifications name the job and every model it is waiting on.

backend/app/models/print_queue.py

@@ -165,6 +165,82 @@ class PrintQueueItem(Base):
     project: Mapped["Project | None"] = relationship(back_populates="queue_items")
     batch: Mapped["PrintBatch | None"] = relationship(back_populates="queue_items")
     created_by: Mapped["User | None"] = relationship()
+    variants: Mapped[list["PrintQueueVariant"]] = relationship(
+        back_populates="queue_item",
+        cascade="all, delete-orphan",
+        order_by="PrintQueueVariant.position",
+    )
+
+
+class PrintQueueVariant(Base):
+    """One candidate file for a queue item that may print on several models (#671).
+
+    A user with an H2S and an H2C slices the same job twice and does not care
+    which machine runs it. Each slice becomes a variant; the scheduler walks them
+    in ``position`` order and takes the first whose model has an idle printer.
+
+    **This is a snapshot, not a pointer.** The candidate list is copied from the
+    library's variant group when the item is queued, and every per-file setting
+    the dispatcher needs is copied with it. Two reasons:
+
+    - Editing the library group afterwards must not silently change a job that is
+      already waiting in the queue.
+    - The per-file settings genuinely differ between candidates and are choices
+      the user made for *this* job, not properties of the file. An H2C slice is
+      dual-nozzle and will not have the same slot count, AMS mapping or nozzle
+      mapping as the H2S slice of the same model.
+
+    On a match the winning variant's fields are written onto the queue row before
+    the dispatch commit, so everything downstream — upload, archive creation,
+    print history, reprint — sees an ordinary single-file item and needs no
+    knowledge that variants exist.
+
+    Variants reference library files only. An archive records a print that already
+    happened, of one specific file, so it is never a candidate for "which of these
+    should we run".
+    """
+
+    __tablename__ = "print_queue_variants"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    queue_item_id: Mapped[int] = mapped_column(
+        ForeignKey("print_queue.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    # User's priority order. When two printers are idle in the same scheduler
+    # pass, the lowest position wins — so the choice is reproducible instead of
+    # depending on which match the matcher happened to find first.
+    position: Mapped[int] = mapped_column(Integer, default=0)
+
+    # CASCADE: deleting the file drops this candidate but leaves the item and its
+    # other candidates alone. Losing the *last* candidate is handled by the
+    # resolver, which holds the item pending with an explicit waiting_reason
+    # rather than letting it sit there looking dispatchable forever.
+    library_file_id: Mapped[int] = mapped_column(ForeignKey("library_files.id", ondelete="CASCADE"), nullable=False)
+    # Normalized short name ("H2S"), taken from the file's own sliced_for_model
+    # at creation, or picked by the user for a legacy file that declares none.
+    target_model: Mapped[str] = mapped_column(String(50), nullable=False)
+
+    # Per-file dispatch settings, same semantics as the identically named columns
+    # on PrintQueueItem — see there for the formats.
+    plate_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    ams_mapping: Mapped[str | None] = mapped_column(Text, nullable=True)
+    nozzle_mapping: Mapped[str | None] = mapped_column(Text, nullable=True)
+    filament_overrides: Mapped[str | None] = mapped_column(Text, nullable=True)
+    required_filament_types: Mapped[str | None] = mapped_column(Text, nullable=True)
+    print_time_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    # How many times this candidate has been dispatched and bounced back to
+    # pending by the start-watchdog. The resolver tries least-attempted first, so
+    # a printer that accepts the file and never starts (#1678) hands the job to
+    # the other machine on the next lap instead of burning the item's whole
+    # DISPATCH_MAX_ATTEMPTS budget against the same wedged printer — which is the
+    # entire reason the user queued an alternative.
+    attempt_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+
+    queue_item: Mapped["PrintQueueItem"] = relationship(back_populates="variants")
+    library_file: Mapped["LibraryFile"] = relationship()
 
 
 from backend.app.models.archive import PrintArchive  # noqa: E402

backend/app/services/print_scheduler.py

@@ -4,6 +4,7 @@ import asyncio
 import json
 import logging
 import time
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -17,7 +18,7 @@ from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
-from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
 from backend.app.models.printer import Printer
 from backend.app.models.settings import Settings
 from backend.app.models.smart_plug import SmartPlug
@@ -156,6 +157,123 @@ def _canonical_filament_type(ftype: str) -> str:
     return _FILAMENT_EQUIV_MAP.get(upper, upper)
 
 
+@dataclass(slots=True)
+class _ModelCandidate:
+    """One (file, printer model) pair the model-based matcher may try.
+
+    Model-based assignment used to have exactly one of these per item, held
+    directly in the item's own columns. Cross-model queue items (#671) have
+    several, held in ``print_queue_variants``. Both shapes are normalised into
+    this so the matching, the cross-model gate and the waiting-reason handling
+    are written once and an item without variants provably takes the same path
+    it took before variants existed.
+
+    ``variant`` is None for the item's own columns and set for a real variant
+    row, which is what :meth:`PrintScheduler._resolve_variant` writes onto the
+    item once that candidate wins.
+    """
+
+    target_model: str | None
+    sliced_for: str | None
+    required_filament_types: str | None
+    filament_overrides: str | None
+    variant: "PrintQueueVariant | None" = None
+
+
+def _sliced_for_model(archive, library_file) -> str | None:
+    """Model a 3MF declares it was sliced for, from whichever source holds it."""
+    if archive is not None:
+        return archive.sliced_for_model
+    if library_file is not None and library_file.file_metadata:
+        return library_file.file_metadata.get("sliced_for_model")
+    return None
+
+
+def _candidates_for(item: PrintQueueItem) -> list[_ModelCandidate]:
+    """Candidate files for ``item``, best first.
+
+    An item with no variant rows yields exactly one candidate built from its own
+    columns — the pre-#671 behaviour, unchanged.
+
+    Variants come back least-attempted first, ties broken by the user's
+    ``position``. On the first pass every count is zero, so this is purely the
+    user's priority order. After a start-watchdog bounce the printer that failed
+    drops behind, so the next lap tries the other machine rather than spending the
+    item's whole retry budget on the one that is wedged. Once every candidate has
+    been tried equally often they cycle again, which keeps the item-level
+    ``DISPATCH_MAX_ATTEMPTS`` bound from #2555 intact — a job with alternatives
+    still gives up, it just does not give up without trying them.
+    """
+    if not item.variants:
+        return [
+            _ModelCandidate(
+                target_model=item.target_model,
+                sliced_for=_sliced_for_model(item.archive, item.library_file),
+                required_filament_types=item.required_filament_types,
+                filament_overrides=item.filament_overrides,
+            )
+        ]
+
+    ordered = sorted(item.variants, key=lambda v: (v.attempt_count or 0, v.position, v.id))
+    return [
+        _ModelCandidate(
+            target_model=v.target_model,
+            sliced_for=_sliced_for_model(None, v.library_file),
+            required_filament_types=v.required_filament_types,
+            filament_overrides=v.filament_overrides,
+            variant=v,
+        )
+        for v in ordered
+    ]
+
+
+def _collapse_waiting_reasons(per_model: list[tuple[str | None, str]]) -> str | None:
+    """Fold one waiting reason per candidate into a single line for the item.
+
+    A cross-model item produces a reason per candidate, and pasting them
+    together unlabelled reads as gibberish ("No idle printer; PETG not loaded"
+    — on which machine?). Each reason is prefixed with its model, except in the
+    single-candidate case where the item already displays its target model and
+    the prefix would be noise.
+
+    Identical reasons collapse rather than repeat, so three idle-less models
+    read as one clause.
+
+    When *every* candidate is merely busy the parts are joined with the ``" | "``
+    separator :meth:`PrintScheduler._is_busy_only` already parses, and left
+    unprefixed. That case must keep testing busy-only: a fleet that is simply
+    printing needs no user action, and labelling the clauses would turn each pass
+    over a two-model item into a "job waiting" notification.
+    """
+    reasons = [(model, reason) for model, reason in per_model if reason]
+    if not reasons:
+        return None
+    if len(reasons) == 1:
+        return reasons[0][1]
+
+    distinct = list(dict.fromkeys(reason for _model, reason in reasons))
+    if len(distinct) == 1:
+        return distinct[0]
+
+    if all(PrintScheduler._is_busy_only(reason) for _model, reason in reasons):
+        return " | ".join(distinct)
+
+    return "; ".join(f"{model or 'unassigned'}: {reason}" for model, reason in reasons)
+
+
+def _candidate_model_label(candidates: list[_ModelCandidate]) -> str | None:
+    """Human label for the models an item is waiting on ("H2S or H2C").
+
+    Notifications take a single target model. For a cross-model item the item's
+    own ``target_model`` is whichever variant happens to be first, which reads as
+    a lie once it is the H2C that actually runs — so name all of them.
+    """
+    models = list(dict.fromkeys(c.target_model for c in candidates if c.target_model))
+    if not models:
+        return None
+    return " or ".join(models)
+
+
 def _mapping_is_all_unresolved(mapping: list | None) -> bool:
     """True if ``mapping`` is a non-empty list whose every entry is the
     unresolved sentinel (-1 / None) — i.e. no required slot ever matched a tray.
@@ -404,6 +522,10 @@ class PrintScheduler:
                     .options(
                         selectinload(PrintQueueItem.archive),
                         selectinload(PrintQueueItem.library_file),
+                        # Cross-model candidates (#671), plus each candidate's file
+                        # for the same cross-model gate. Lazy-loading either would
+                        # raise in async.
+                        selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
                     )
                     .order_by(
                         PrintQueueItem.printer_id,
@@ -422,6 +544,10 @@ class PrintScheduler:
                     .options(
                         selectinload(PrintQueueItem.archive),
                         selectinload(PrintQueueItem.library_file),
+                        # Cross-model candidates (#671), plus each candidate's file
+                        # for the same cross-model gate. Lazy-loading either would
+                        # raise in async.
+                        selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
                     )
                     .order_by(PrintQueueItem.printer_id, PrintQueueItem.position)
                 )
@@ -684,60 +810,94 @@ class PrintScheduler:
                                 other.been_jumped = True
                         await db.commit()
 
-                elif item.target_model:
-                    # Model-based assignment - find any idle printer of matching model
-                    # Parse required filament types if present
-                    required_types = None
-                    if item.required_filament_types:
-                        try:
-                            required_types = json.loads(item.required_filament_types)
-                        except json.JSONDecodeError:
-                            pass  # Ignore malformed filament types; treat as no constraint
+                elif item.target_model or item.variants:
+                    # Model-based assignment - find any idle printer of matching model.
+                    # A plain model-based item has exactly one candidate, built from
+                    # its own columns. A cross-model item (#671) has one per sliced
+                    # variant and takes the first that matches, walking them in the
+                    # user's priority order so the pick is reproducible when more
+                    # than one printer is free in the same pass.
+                    candidates = _candidates_for(item)
+                    printer_id = None
+                    chosen: _ModelCandidate | None = None
+                    per_model_reasons: list[tuple[str | None, str]] = []
+
+                    if not item.variants and not item.archive_id and not item.library_file_id:
+                        # Every candidate file was deleted out from under this item
+                        # (variant rows go with their library file). Dispatching would
+                        # fail deep in the upload with "No archive_id or library_file_id";
+                        # hold it here with something the user can act on instead.
+                        candidates = []
+                        per_model_reasons.append(
+                            (
+                                item.target_model,
+                                "Every file for this job has been deleted — add a file back or remove the item",
+                            )
+                        )
 
-                    # Parse filament overrides if present
-                    filament_overrides = None
-                    if item.filament_overrides:
-                        try:
-                            filament_overrides = json.loads(item.filament_overrides)
-                        except json.JSONDecodeError:
-                            pass
+                    for candidate in candidates:
+                        # Parse required filament types if present
+                        required_types = None
+                        if candidate.required_filament_types:
+                            try:
+                                required_types = json.loads(candidate.required_filament_types)
+                            except json.JSONDecodeError:
+                                pass  # Ignore malformed filament types; treat as no constraint
+
+                        # Parse filament overrides if present
+                        filament_overrides = None
+                        if candidate.filament_overrides:
+                            try:
+                                filament_overrides = json.loads(candidate.filament_overrides)
+                            except json.JSONDecodeError:
+                                pass
+
+                        # If overrides exist, use override types for validation instead
+                        effective_types = required_types
+                        if filament_overrides:
+                            override_types = sorted({o["type"] for o in filament_overrides if "type" in o})
+                            if override_types:
+                                # Merge: keep original types for non-overridden slots, add override types
+                                effective_types = sorted(set(required_types or []) | set(override_types))
+
+                        # Cross-model safety gate (#2578): never hand a 3MF sliced
+                        # for an incompatible model to a printer, no matter how the
+                        # row got into the DB (old rows, direct API writes). Held
+                        # as pending with an actionable waiting_reason — the user
+                        # fixes it by editing the item's target model.
+                        if not is_gcode_compatible(candidate.sliced_for, candidate.target_model):
+                            per_model_reasons.append(
+                                (
+                                    candidate.target_model,
+                                    f"File was sliced for {candidate.sliced_for}, which is not compatible with "
+                                    f"{candidate.target_model} — edit the item and fix its target model",
+                                )
+                            )
+                            skip_reasons["sliced_model_mismatch"] = skip_reasons.get("sliced_model_mismatch", 0) + 1
+                            continue
 
-                    # If overrides exist, use override types for validation instead
-                    effective_types = required_types
-                    if filament_overrides:
-                        override_types = sorted({o["type"] for o in filament_overrides if "type" in o})
-                        if override_types:
-                            # Merge: keep original types for non-overridden slots, add override types
-                            effective_types = sorted(set(required_types or []) | set(override_types))
-
-                    # Cross-model safety gate (#2578): never hand a 3MF sliced
-                    # for an incompatible model to a printer, no matter how the
-                    # row got into the DB (old rows, direct API writes). Held
-                    # as pending with an actionable waiting_reason — the user
-                    # fixes it by editing the item's target model.
-                    sliced_for = None
-                    if item.archive:
-                        sliced_for = item.archive.sliced_for_model
-                    elif item.library_file and item.library_file.file_metadata:
-                        sliced_for = item.library_file.file_metadata.get("sliced_for_model")
-
-                    if not is_gcode_compatible(sliced_for, item.target_model):
-                        printer_id = None
-                        waiting_reason = (
-                            f"File was sliced for {sliced_for}, which is not compatible with "
-                            f"{item.target_model} — edit the item and fix its target model"
-                        )
-                        skip_reasons["sliced_model_mismatch"] = skip_reasons.get("sliced_model_mismatch", 0) + 1
-                    else:
-                        printer_id, waiting_reason = await self._find_idle_printer_for_model(
+                        match_id, match_reason = await self._find_idle_printer_for_model(
                             db,
-                            item.target_model,
+                            candidate.target_model,
                             busy_printers,
                             effective_types,
                             item.target_location,
                             filament_overrides=filament_overrides,
                             require_plate_clear=require_plate_clear,
                         )
+                        if match_id:
+                            printer_id = match_id
+                            chosen = candidate
+                            break
+                        per_model_reasons.append((candidate.target_model, match_reason or ""))
+
+                    waiting_reason = None if printer_id else _collapse_waiting_reasons(per_model_reasons)
+
+                    # Fold the winning variant's file and settings onto the item
+                    # before anything else looks at them — the guards below and
+                    # every step of the dispatch read the item's own columns.
+                    if chosen is not None:
+                        self._resolve_variant(item, chosen)
 
                     # Update waiting_reason if changed and send notification when first waiting
                     if item.waiting_reason != waiting_reason:
@@ -751,7 +911,7 @@ class PrintScheduler:
                             job_name = await self._get_job_name(db, item)
                             await notification_service.on_queue_job_waiting(
                                 job_name=job_name,
-                                target_model=item.target_model,
+                                target_model=_candidate_model_label(candidates) or item.target_model,
                                 waiting_reason=waiting_reason,
                                 db=db,
                             )
@@ -1387,6 +1547,46 @@ class PrintScheduler:
                 matches += 1
         return matches
 
+    def _resolve_variant(self, item: PrintQueueItem, candidate: _ModelCandidate) -> None:
+        """Fold the winning candidate's file and settings onto the queue row (#671).
+
+        This is the whole trick that keeps cross-model items cheap: the many-to-many
+        never escapes the selection loop. By the time the pass commits, the row
+        looks exactly like an ordinary single-file model-based item, so the upload,
+        archive creation, expected-print registration, print history and reprint
+        paths need no knowledge that variants exist.
+
+        No-ops for a non-variant candidate, which is already the item's own columns.
+
+        Safe to run and re-run: the item's file columns are only ever *read* when it
+        has no variants, so an item that gets resolved and then skipped (library-row
+        conflict, previous-print gate) is simply resolved again on the next pass.
+        """
+        variant = candidate.variant
+        if variant is None:
+            return
+
+        item.library_file_id = variant.library_file_id
+        item.library_file = variant.library_file
+        # The dispatcher checks archive_id first and would print that instead of
+        # the file we just picked. Creation refuses to combine the two, so this
+        # only ever fires on a hand-written row — clear it rather than silently
+        # dispatch something the matcher never considered.
+        item.archive_id = None
+        item.archive = None
+
+        item.target_model = variant.target_model
+        item.plate_id = variant.plate_id
+        item.ams_mapping = variant.ams_mapping
+        item.nozzle_mapping = variant.nozzle_mapping
+        item.filament_overrides = variant.filament_overrides
+        item.required_filament_types = variant.required_filament_types
+        if variant.print_time_seconds is not None:
+            # The row carried the shortest candidate's estimate so SJF could order
+            # it before a printer was known; now that one is chosen, record what is
+            # actually going to run so history and the ETA agree with reality.
+            item.print_time_seconds = variant.print_time_seconds
+
     async def _ensure_ams_mapping(self, db: AsyncSession, printer_id: int, item: PrintQueueItem) -> None:
         """Ensure the queue item carries a usable AMS mapping before dispatch.
 
@@ -2975,6 +3175,22 @@ class PrintScheduler:
             library_file = result.scalar_one_or_none()
             if library_file:
                 return library_file.filename.replace(".gcode.3mf", "").replace(".3mf", "")
+        # A cross-model item (#671) holds no file of its own until a printer is
+        # picked, so name it after its first candidate — otherwise every waiting
+        # notification for one reads "Job #12". Queried rather than read off
+        # item.variants because callers outside the selection loop have not
+        # eager-loaded them, and a lazy load raises in async.
+        first_variant_name = (
+            await db.execute(
+                select(LibraryFile.filename)
+                .join(PrintQueueVariant, PrintQueueVariant.library_file_id == LibraryFile.id)
+                .where(PrintQueueVariant.queue_item_id == item.id)
+                .order_by(PrintQueueVariant.position, PrintQueueVariant.id)
+                .limit(1)
+            )
+        ).scalar_one_or_none()
+        if first_variant_name:
+            return first_variant_name.replace(".gcode.3mf", "").replace(".3mf", "")
         return f"Job #{item.id}"
 
     async def _get_printer(self, db: AsyncSession, printer_id: int) -> Printer | None:
@@ -3996,6 +4212,17 @@ class PrintScheduler:
                 return "already_moved_on"
             item.dispatch_attempts = (item.dispatch_attempts or 0) + 1
             item.started_at = None
+            # Charge the attempt to the candidate that was actually dispatched, so
+            # a cross-model item (#671) reaches for its other file next lap instead
+            # of retrying the printer that just failed to start. Matched by file
+            # because that is what the resolver copied onto the row.
+            if item.library_file_id is not None:
+                await db.execute(
+                    update(PrintQueueVariant)
+                    .where(PrintQueueVariant.queue_item_id == item.id)
+                    .where(PrintQueueVariant.library_file_id == item.library_file_id)
+                    .values(attempt_count=PrintQueueVariant.attempt_count + 1)
+                )
             if command_rejected:
                 # No retry budget for this one: the printer refused to verify the
                 # command, and re-uploading the same 3MF to the same printer will

backend/tests/unit/test_scheduler_cross_model_variants.py

@@ -0,0 +1,456 @@
+"""Cross-model queue items — one job, several sliced files (#671).
+
+The reporter has an H2S and an H2C and does not care which one runs the job.
+He slices it twice; both slices become variants of a single queue item, and the
+scheduler takes the first whose model has an idle printer.
+
+The design constraint that shapes everything here: the many-to-many must never
+escape the selection loop. Once a candidate wins, its file and settings are
+folded onto the queue row, so the upload, archive creation, print history and
+reprint paths keep seeing an ordinary single-file item. These tests assert both
+halves — that the right candidate is picked, and that the row afterwards looks
+like it was queued for that file all along.
+"""
+
+from contextlib import ExitStack
+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, PrintQueueVariant
+from backend.app.models.printer import Printer
+from backend.app.services.print_scheduler import (
+    PrintScheduler,
+    _candidate_model_label,
+    _candidates_for,
+    _collapse_waiting_reasons,
+)
+
+# ---------------------------------------------------------------------------
+# Candidate ordering — pure
+# ---------------------------------------------------------------------------
+
+
+def _fake_variant(*, vid, position, model, attempts=0):
+    return SimpleNamespace(
+        id=vid,
+        position=position,
+        target_model=model,
+        attempt_count=attempts,
+        library_file=SimpleNamespace(file_metadata={"sliced_for_model": model}),
+        required_filament_types=None,
+        filament_overrides=None,
+    )
+
+
+def _fake_item(variants):
+    return SimpleNamespace(
+        variants=variants,
+        target_model=None,
+        archive=None,
+        library_file=None,
+        required_filament_types=None,
+        filament_overrides=None,
+    )
+
+
+def test_no_variants_yields_the_items_own_columns():
+    """The pre-#671 path must be provably unchanged: one candidate, built from
+    the item itself."""
+    item = SimpleNamespace(
+        variants=[],
+        target_model="H2D",
+        archive=None,
+        library_file=SimpleNamespace(file_metadata={"sliced_for_model": "H2D"}),
+        required_filament_types='["PLA"]',
+        filament_overrides=None,
+    )
+    candidates = _candidates_for(item)
+    assert len(candidates) == 1
+    assert candidates[0].target_model == "H2D"
+    assert candidates[0].sliced_for == "H2D"
+    assert candidates[0].required_filament_types == '["PLA"]'
+    assert candidates[0].variant is None
+
+
+def test_variants_come_back_in_user_priority_order():
+    item = _fake_item(
+        [
+            _fake_variant(vid=2, position=1, model="H2C"),
+            _fake_variant(vid=1, position=0, model="H2S"),
+        ]
+    )
+    assert [c.target_model for c in _candidates_for(item)] == ["H2S", "H2C"]
+
+
+def test_least_attempted_candidate_is_tried_first():
+    """A printer that accepts the file and never starts must not eat the item's
+    whole retry budget — the alternative gets the next lap."""
+    item = _fake_item(
+        [
+            _fake_variant(vid=1, position=0, model="H2S", attempts=1),
+            _fake_variant(vid=2, position=1, model="H2C", attempts=0),
+        ]
+    )
+    assert [c.target_model for c in _candidates_for(item)] == ["H2C", "H2S"]
+
+
+def test_equal_attempts_fall_back_to_priority():
+    """Once every candidate has failed equally often they cycle in the user's
+    order, so the item still reaches its DISPATCH_MAX_ATTEMPTS ceiling."""
+    item = _fake_item(
+        [
+            _fake_variant(vid=1, position=0, model="H2S", attempts=2),
+            _fake_variant(vid=2, position=1, model="H2C", attempts=2),
+        ]
+    )
+    assert [c.target_model for c in _candidates_for(item)] == ["H2S", "H2C"]
+
+
+# ---------------------------------------------------------------------------
+# Waiting reasons — pure
+# ---------------------------------------------------------------------------
+
+
+def test_single_candidate_reason_is_unprefixed():
+    """One candidate means the card already shows the model; prefixing it would
+    just be noise."""
+    assert _collapse_waiting_reasons([("H2D", "Busy: H2D-1 (Printing)")]) == "Busy: H2D-1 (Printing)"
+
+
+def test_identical_reasons_collapse_to_one_clause():
+    collapsed = _collapse_waiting_reasons([("H2S", "Busy: shared-1 (Printing)"), ("H2C", "Busy: shared-1 (Printing)")])
+    assert collapsed == "Busy: shared-1 (Printing)"
+
+
+def test_all_busy_stays_busy_only_so_no_notification_fires():
+    """Two models busy on differently-named printers still has to read as
+    busy-only. Labelling the clauses would make every pass over a cross-model
+    item look like it needs the user, when it just needs a printer to finish."""
+    scheduler = PrintScheduler()
+    collapsed = _collapse_waiting_reasons([("H2S", "Busy: H2S-1 (Printing)"), ("H2C", "Busy: H2C-1 (Printing)")])
+    assert collapsed == "Busy: H2S-1 (Printing) | Busy: H2C-1 (Printing)"
+    assert scheduler._is_busy_only(collapsed)
+
+
+def test_differing_reasons_are_labelled_by_model():
+    collapsed = _collapse_waiting_reasons([("H2S", "No PETG loaded"), ("H2C", "Busy: H2C-1 (Printing)")])
+    assert collapsed == "H2S: No PETG loaded; H2C: Busy: H2C-1 (Printing)"
+    assert not PrintScheduler._is_busy_only(collapsed), "a real blocker must still notify"
+
+
+def test_empty_reasons_are_dropped():
+    assert _collapse_waiting_reasons([("H2S", "")]) is None
+    assert _collapse_waiting_reasons([]) is None
+
+
+def test_model_label_names_every_candidate():
+    item = _fake_item(
+        [
+            _fake_variant(vid=1, position=0, model="H2S"),
+            _fake_variant(vid=2, position=1, model="H2C"),
+        ]
+    )
+    assert _candidate_model_label(_candidates_for(item)) == "H2S or H2C"
+
+
+# ---------------------------------------------------------------------------
+# Scheduler behaviour
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def queue_db():
+    """In-memory DB with one H2S and one H2C."""
+    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="H2S-1",
+                    serial_number="H2S0001",
+                    ip_address="10.0.0.1",
+                    access_code="x",
+                    model="H2S",
+                    is_active=True,
+                ),
+                Printer(
+                    id=2,
+                    name="H2C-1",
+                    serial_number="H2C0001",
+                    ip_address="10.0.0.2",
+                    access_code="x",
+                    model="H2C",
+                    is_active=True,
+                ),
+            ]
+        )
+        await db.commit()
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker)
+    finally:
+        await engine.dispose()
+
+
+async def _add_variant_item(ctx, specs):
+    """Seed one pending queue item with a variant per (model, overrides) spec."""
+    async with ctx.session_maker() as db:
+        item = PrintQueueItem(
+            status="pending",
+            position=1,
+            target_model=specs[0]["model"],
+        )
+        db.add(item)
+        await db.flush()
+
+        for position, spec in enumerate(specs):
+            lib = LibraryFile(
+                filename=f"job_{spec['model']}.gcode.3mf",
+                file_path=f"/library/job_{spec['model']}.gcode.3mf",
+                file_size=10,
+                file_type="gcode.3mf",
+                file_metadata={"sliced_for_model": spec.get("sliced_for", spec["model"])},
+            )
+            db.add(lib)
+            await db.flush()
+            db.add(
+                PrintQueueVariant(
+                    queue_item_id=item.id,
+                    position=position,
+                    library_file_id=lib.id,
+                    target_model=spec["model"],
+                    plate_id=spec.get("plate_id"),
+                    ams_mapping=spec.get("ams_mapping"),
+                    nozzle_mapping=spec.get("nozzle_mapping"),
+                    print_time_seconds=spec.get("print_time_seconds"),
+                    attempt_count=spec.get("attempts", 0),
+                )
+            )
+        await db.commit()
+        return item.id
+
+
+async def _run_check_queue(ctx, scheduler, finder, waiting_notification=None):
+    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.notification_service.notification_service.on_queue_job_waiting",
+            waiting_notification or AsyncMock(),
+        ),
+        patch(
+            "backend.app.services.notification_service.notification_service.on_queue_job_assigned",
+            AsyncMock(),
+        ),
+        patch.object(scheduler, "_find_idle_printer_for_model", finder),
+        patch.object(scheduler, "_check_auto_drying", AsyncMock()),
+        # Selection is what's under test — keep AMS recomputation and the
+        # filament-deficit probe out of the way, and never actually dispatch.
+        patch.object(scheduler, "_ensure_ams_mapping", AsyncMock()),
+        patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=False)),
+        patch.object(scheduler, "_launch_uploads", MagicMock()),
+    ]
+    with ExitStack() as stack:
+        for p in patches:
+            stack.enter_context(p)
+        return await scheduler.check_queue()
+
+
+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 _finder_for(available: dict[str, int]):
+    """Matcher that offers a printer only for the listed models."""
+
+    async def _find(db, model, exclude_ids, *args, **kwargs):
+        if model in available:
+            return available[model], None
+        return None, f"No idle {model} printer"
+
+    return AsyncMock(side_effect=_find)
+
+
+@pytest.mark.asyncio
+async def test_first_matching_variant_wins_and_is_folded_onto_the_row(queue_db):
+    """The H2C is free, the H2S is not — the item runs the H2C slice, and every
+    downstream consumer sees a plain single-file item pointing at it."""
+    item_id = await _add_variant_item(
+        queue_db,
+        [
+            {"model": "H2S", "plate_id": 1, "ams_mapping": "[1]", "print_time_seconds": 900},
+            {
+                "model": "H2C",
+                "plate_id": 3,
+                "ams_mapping": "[4, 5]",
+                "nozzle_mapping": "[0, 1]",
+                "print_time_seconds": 1200,
+            },
+        ],
+    )
+    scheduler = PrintScheduler()
+
+    await _run_check_queue(queue_db, scheduler, _finder_for({"H2C": 2}))
+
+    item = await _get_item(queue_db, item_id)
+    assert item.printer_id == 2, "assigned to the H2C"
+    assert item.target_model == "H2C"
+    assert item.plate_id == 3
+    assert item.ams_mapping == "[4, 5]"
+    assert item.nozzle_mapping == "[0, 1]"
+    assert item.print_time_seconds == 1200, "the estimate now describes what will actually run"
+    assert item.waiting_reason is None
+    assert item.archive_id is None
+
+    async with queue_db.session_maker() as db:
+        chosen = (
+            await db.execute(select(PrintQueueVariant).where(PrintQueueVariant.target_model == "H2C"))
+        ).scalar_one()
+        assert item.library_file_id == chosen.library_file_id
+
+
+@pytest.mark.asyncio
+async def test_priority_order_decides_when_both_are_free(queue_db):
+    """Both printers idle in the same pass: the user's first choice runs, so the
+    outcome is reproducible rather than whichever match came back first."""
+    item_id = await _add_variant_item(queue_db, [{"model": "H2S"}, {"model": "H2C"}])
+    scheduler = PrintScheduler()
+
+    await _run_check_queue(queue_db, scheduler, _finder_for({"H2S": 1, "H2C": 2}))
+
+    item = await _get_item(queue_db, item_id)
+    assert item.printer_id == 1
+    assert item.target_model == "H2S"
+
+
+@pytest.mark.asyncio
+async def test_cross_model_gate_is_applied_per_candidate(queue_db):
+    """A variant whose file disagrees with its own model is skipped, and the
+    other one still runs — the gate must not condemn the whole item."""
+    item_id = await _add_variant_item(
+        queue_db,
+        [
+            {"model": "H2S", "sliced_for": "X1C"},
+            {"model": "H2C"},
+        ],
+    )
+    scheduler = PrintScheduler()
+    finder = _finder_for({"H2S": 1, "H2C": 2})
+
+    await _run_check_queue(queue_db, scheduler, finder)
+
+    assert [c.args[1] for c in finder.await_args_list] == ["H2C"], "the mismatched variant never reaches the matcher"
+    item = await _get_item(queue_db, item_id)
+    assert item.printer_id == 2
+    assert item.target_model == "H2C"
+
+
+@pytest.mark.asyncio
+async def test_no_match_reports_every_model_it_tried(queue_db):
+    """Nothing is free: the user must be able to tell which machines were
+    considered, not just that "a printer" was unavailable."""
+    item_id = await _add_variant_item(queue_db, [{"model": "H2S"}, {"model": "H2C"}])
+    scheduler = PrintScheduler()
+    waiting = AsyncMock()
+
+    await _run_check_queue(queue_db, scheduler, _finder_for({}), waiting)
+
+    item = await _get_item(queue_db, item_id)
+    assert item.printer_id is None
+    assert item.status == "pending"
+    assert "H2S: No idle H2S printer" in item.waiting_reason
+    assert "H2C: No idle H2C printer" in item.waiting_reason
+    assert waiting.await_args.kwargs["target_model"] == "H2S or H2C"
+    # The item holds no file of its own yet — the alert still has to name the job.
+    assert waiting.await_args.kwargs["job_name"] == "job_H2S"
+
+
+@pytest.mark.asyncio
+async def test_item_with_no_files_left_is_held_with_an_actionable_reason(queue_db):
+    """Deleting a library file takes its variant with it. An item stripped of
+    every candidate used to sail into dispatch and die there on "No archive_id
+    or library_file_id"; hold it where the user can see why."""
+    async with queue_db.session_maker() as db:
+        db.add(PrintQueueItem(status="pending", position=1, target_model="H2S"))
+        await db.commit()
+    scheduler = PrintScheduler()
+    finder = _finder_for({"H2S": 1})
+
+    await _run_check_queue(queue_db, scheduler, finder)
+
+    finder.assert_not_awaited()
+    async with queue_db.session_maker() as db:
+        item = (await db.execute(select(PrintQueueItem))).scalar_one()
+    assert item.status == "pending"
+    assert item.printer_id is None
+    assert "has been deleted" in item.waiting_reason
+
+
+@pytest.mark.asyncio
+async def test_plain_model_based_item_is_untouched(queue_db):
+    """Regression guard: an item with no variants takes exactly the path it took
+    before variants existed."""
+    async with queue_db.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": "H2S"},
+        )
+        db.add(lib)
+        await db.flush()
+        db.add(
+            PrintQueueItem(
+                status="pending",
+                position=1,
+                target_model="H2S",
+                library_file_id=lib.id,
+                plate_id=2,
+            )
+        )
+        await db.commit()
+    scheduler = PrintScheduler()
+
+    await _run_check_queue(queue_db, scheduler, _finder_for({"H2S": 1}))
+
+    async with queue_db.session_maker() as db:
+        item = (await db.execute(select(PrintQueueItem))).scalar_one()
+    assert item.printer_id == 1
+    assert item.target_model == "H2S"
+    assert item.plate_id == 2, "nothing overwrote the item's own settings"
+
+
+@pytest.mark.asyncio
+async def test_failed_candidate_steps_aside_for_the_alternative(queue_db):
+    """The H2S burned an attempt on the last lap. Both are free now — the H2C
+    goes first, which is the entire point of queueing an alternative."""
+    item_id = await _add_variant_item(
+        queue_db,
+        [
+            {"model": "H2S", "attempts": 1},
+            {"model": "H2C", "attempts": 0},
+        ],
+    )
+    scheduler = PrintScheduler()
+
+    await _run_check_queue(queue_db, scheduler, _finder_for({"H2S": 1, "H2C": 2}))
+
+    item = await _get_item(queue_db, item_id)
+    assert item.printer_id == 2
+    assert item.target_model == "H2C"