Start queued jobs in list order, whether pinned or "Any <model>" (#3200)

backend/app/api/routes/print_queue.py

@@ -54,6 +54,7 @@ from backend.app.services.print_batch import (
     refresh_batch_status_for_item,
 )
 from backend.app.services.print_cost_estimate import estimate_queue_source_cost
+from backend.app.services.queue_position import lock_queue_positions, max_queue_position
 from backend.app.utils.printer_models import (
     is_gcode_compatible,
 )
@@ -1001,47 +1002,14 @@ async def add_to_queue(
         await db.flush()  # Get batch.id before creating items
         batch_id = batch.id
 
-    # Get queue scope for this printer (or for unassigned/model-based items).
-    if data.printer_id is not None:
-        queue_scope = (
-            PrintQueueItem.printer_id == data.printer_id,
-            PrintQueueItem.status == "pending",
-        )
-    else:
-        # For unassigned/model-based items, scope across all unassigned.
-        queue_scope = (
-            PrintQueueItem.printer_id.is_(None),
-            PrintQueueItem.status == "pending",
-        )
-
-    # Serialize concurrent queue inserts to the same scope (#1625-followup).
-    # The race: two concurrent ASAP inserts both compute MAX(position) before
-    # either commits; in an empty scope, both INSERT at position 1 (duplicate).
-    # In a non-empty scope, Postgres's row-level locks on the UPDATE shift
-    # serialize naturally, but the empty-scope path has no rows to lock.
-    # A transaction-scoped advisory lock keyed on the printer_id closes that
-    # window; the lock is released automatically at commit/rollback. Different
-    # printers don't contend. SQLite serializes writes implicitly so this is a
-    # no-op there.
-    #
-    # Dialect is checked against the actual session binding, NOT the
-    # `is_sqlite()` helper, because the test fixture overrides `get_db` with a
-    # SQLite engine while `settings.database_url` still points at Postgres
-    # (the helper reads settings). Inspecting the connection directly is the
-    # right shape for any code that mutates SQL based on the live dialect.
-    from sqlalchemy import text
-
-    bind = db.get_bind()
-    if bind.dialect.name == "postgresql":
-        scope_key = data.printer_id if data.printer_id is not None else 0
-        # 1625 namespaces the lock so it can't collide with other advisory
-        # locks elsewhere in the codebase.
-        await db.execute(text("SELECT pg_advisory_xact_lock(1625, :k)"), {"k": scope_key})
+    # Positions are one sequence across every pending item (#3200), so a new
+    # item lands relative to the whole list, not to its printer's share of it.
+    queue_scope = (PrintQueueItem.status == "pending",)
+    await lock_queue_positions(db)
 
     insert_position = max(1, data.insert_position or 1)
     if data.insert_at_top or data.insert_position is not None:
-        result = await db.execute(select(func.max(PrintQueueItem.position)).where(*queue_scope))
-        max_pos = result.scalar() or 0
+        max_pos = await max_queue_position(db)
         insert_position = min(insert_position, max_pos + 1)
         await db.execute(
             update(PrintQueueItem)
@@ -1051,9 +1019,7 @@ async def add_to_queue(
         )
         start_position = insert_position
     else:
-        result = await db.execute(select(func.max(PrintQueueItem.position)).where(*queue_scope))
-        max_pos = result.scalar() or 0
-        start_position = max_pos + 1
+        start_position = await max_queue_position(db) + 1
 
     # Resolve print_time_seconds for SJF scheduling (cache on item at creation)
     cached_print_time = None

backend/app/api/routes/webhook.py

@@ -14,6 +14,7 @@ from backend.app.models.printer import Printer
 from backend.app.schemas.printer import HMSErrorResponse, hms_error_responses
 from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
 from backend.app.services.printer_manager import printer_manager
+from backend.app.services.queue_position import next_queue_position
 
 logger = logging.getLogger(__name__)
 
@@ -114,18 +115,9 @@ async def webhook_add_to_queue(
     if not printer:
         raise HTTPException(status_code=404, detail="Printer not found")
 
-    # Get next position
-    result = await db.execute(
-        select(PrintQueueItem.position)
-        .where(
-            PrintQueueItem.printer_id == data.printer_id,
-            PrintQueueItem.status == "pending",
-        )
-        .order_by(PrintQueueItem.position.desc())
-        .limit(1)
-    )
-    max_position = result.scalar()
-    next_position = (max_position or 0) + 1
+    # Append to the end of the queue: positions are one sequence across all
+    # pending items, not one per printer (#3200).
+    next_position = await next_queue_position(db)
 
     # Parse scheduled time if provided
     scheduled_time = None

backend/app/schemas/print_queue.py

@@ -80,7 +80,7 @@ class PrintQueueItemCreate(BaseModel):
     require_previous_success: bool = False
     auto_off_after: bool = False  # Power off printer after print completes
     manual_start: bool = False  # Requires manual trigger to start (staged)
-    insert_at_top: bool = False  # Insert ahead of other pending items in the same queue scope
+    insert_at_top: bool = False  # Insert ahead of other pending items (one queue order across all printers, #3200)
     insert_position: int | None = None  # 1-indexed insertion position for priority queueing
     # Persistent "Print Anyway" acknowledgement (#1698-followup). When set,
     # PrintModal already showed the deficit warning and the user confirmed,

backend/app/services/print_batch.py

@@ -21,13 +21,14 @@ import logging
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 
-from sqlalchemy import func, select, text
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from backend.app.models.print_batch import PrintBatch, PrintBatchPlate
 from backend.app.models.print_log import PrintLogEntry
 from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
+from backend.app.services.queue_position import next_queue_position
 
 logger = logging.getLogger(__name__)
 
@@ -432,34 +433,6 @@ async def refresh_batch_status_for_item(db: AsyncSession, queue_item_id: int) ->
     await refresh_batch_status(db, batch)
 
 
-async def _next_position(db: AsyncSession, printer_id: int | None) -> int:
-    """Next free queue position in the scope a clone will land in.
-
-    Positions are per-queue, not global: one sequence per printer plus one
-    shared sequence for unassigned / model-based items, matching the scope the
-    add-to-queue route uses. Taking a global MAX here would drop every clone
-    at the end of whichever printer's queue happens to be longest and scramble
-    the order the user sees.
-    """
-    # Same advisory lock the add-to-queue route takes (#1625-followup): two
-    # concurrent inserts into an empty scope would otherwise both read
-    # MAX(position) as 0 and land on position 1. SQLite serialises writes
-    # implicitly and needs no equivalent.
-    bind = db.get_bind()
-    if bind.dialect.name == "postgresql":
-        await db.execute(
-            text("SELECT pg_advisory_xact_lock(1625, :k)"), {"k": printer_id if printer_id is not None else 0}
-        )
-
-    scope = PrintQueueItem.printer_id == printer_id if printer_id is not None else PrintQueueItem.printer_id.is_(None)
-    max_pos = (
-        await db.execute(
-            select(func.max(PrintQueueItem.position)).where(scope).where(PrintQueueItem.status == "pending")
-        )
-    ).scalar() or 0
-    return max_pos + 1
-
-
 def _clone_queue_item(source: PrintQueueItem, *, position: int, created_by_id: int | None) -> PrintQueueItem:
     """Copy *source*'s print configuration into a fresh pending item.
 
@@ -556,9 +529,9 @@ async def dispatch_remaining(
         if limit is not None:
             wanted = min(wanted, limit - len(created))
 
-        # One scope per source printer; clones for this plate all land in it,
-        # appended after whatever is already queued there.
-        position = await _next_position(db, source.printer_id)
+        # Clones for this plate are appended to the end of the queue (#3200),
+        # after whatever is already pending, in plate order.
+        position = await next_queue_position(db)
 
         for _ in range(wanted):
             clone = _clone_queue_item(source, position=position, created_by_id=created_by_id)

backend/app/services/print_scheduler.py

@@ -410,6 +410,27 @@ def _filament_constraints(candidate: _ModelCandidate) -> tuple[list[str] | None,
     return effective_types, filament_overrides
 
 
+def _could_take_printer(item: PrintQueueItem, printer_id: int, printer_model: str | None) -> bool:
+    """Whether ``item`` was competing for the printer another item just took.
+
+    The SJF starvation guard marks a job as jumped when a shorter one lower in
+    the queue takes a printer it wanted. A pinned job and an "Any <model>" job
+    compete for the same printer, so the guard has to look across both lanes
+    (#3200); looking only inside the dispatched item's own lane let a stream of
+    short model-based jobs hold a longer pinned job back indefinitely.
+
+    Location filters are not checked: over-marking only lifts an item that was
+    going to wait anyway, while under-marking is the starvation this prevents.
+    """
+    if item.printer_id is not None:
+        return item.printer_id == printer_id
+    if not printer_model:
+        return False
+    wanted = (normalize_printer_model(printer_model) or printer_model).lower()
+    models = [v.target_model for v in item.variants] if item.variants else [item.target_model]
+    return any(m and (normalize_printer_model(m) or m).lower() == wanted for m in models)
+
+
 def _candidates_for(item: PrintQueueItem) -> list[_ModelCandidate]:
     """Candidate files for ``item``, best first.
 
@@ -1306,11 +1327,17 @@ class PrintScheduler:
             # Check if shortest-job-first scheduling is enabled
             sjf_enabled = await self._get_bool_setting(db, "queue_shortest_first")
 
-            # Get all pending items, ordered by printer and position (or SJF order)
+            # Get all pending items in the one order the queue page shows them
+            # in (#3200). The first eligible item takes a printer, so this order
+            # decides who wins a printer that a pinned job and an "Any <model>"
+            # job both want. It used to start with ``printer_id``, which made the
+            # lane outrank the position: SQLite sorts NULL first, so model-based
+            # jobs always won; PostgreSQL sorts it last, so pinned jobs did.
+            # Neither is what the user dragged into place.
             if sjf_enabled:
-                # SJF: group by printer (and target_model for model-based jobs),
-                # then items already jumped get top priority (starvation guard),
-                # then sort by print_time ascending. Items with no print time go last.
+                # SJF: items already jumped get top priority (starvation guard),
+                # then sort by print_time ascending. Items with no print time go
+                # last, and position breaks ties.
                 result = await db.execute(
                     select(PrintQueueItem)
                     .where(PrintQueueItem.status == "pending")
@@ -1331,11 +1358,10 @@ class PrintScheduler:
                         selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
                     )
                     .order_by(
-                        PrintQueueItem.printer_id,
-                        PrintQueueItem.target_model,
                         PrintQueueItem.been_jumped.desc(),
                         PrintQueueItem.print_time_seconds.asc().nullslast(),
                         PrintQueueItem.position,
+                        PrintQueueItem.id,
                     )
                 )
             else:
@@ -1352,7 +1378,7 @@ class PrintScheduler:
                         # raise in async.
                         selectinload(PrintQueueItem.variants).selectinload(PrintQueueVariant.library_file),
                     )
-                    .order_by(PrintQueueItem.printer_id, PrintQueueItem.position)
+                    .order_by(PrintQueueItem.position, PrintQueueItem.id)
                 )
             items = list(result.scalars().all())
 
@@ -1889,11 +1915,13 @@ class PrintScheduler:
 
                     # SJF starvation guard: mark items that were jumped
                     if sjf_enabled and item.print_time_seconds is not None:
+                        pinned_model = pinned_printers.get(item.printer_id, ("", ""))[1]
                         for other in items:
                             if (
                                 other.id != item.id
+                                and other.id not in dispatch_ids
                                 and other.status == "pending"
-                                and other.printer_id == item.printer_id
+                                and _could_take_printer(other, item.printer_id, pinned_model)
                                 and not other.been_jumped
                                 and other.position < item.position
                                 and (
@@ -2093,15 +2121,14 @@ class PrintScheduler:
                         dispatch_ids.append(item.id)
                         claim_printer(printer_id)
 
-                        # SJF starvation guard: mark model-based items that were jumped
+                        # SJF starvation guard: mark items that were jumped
                         if sjf_enabled and item.print_time_seconds is not None:
                             for other in items:
                                 if (
                                     other.id != item.id
+                                    and other.id not in dispatch_ids
                                     and other.status == "pending"
-                                    and other.printer_id is None
-                                    and other.target_model
-                                    and other.target_model.upper() == item.target_model.upper()
+                                    and _could_take_printer(other, printer_id, item.target_model)
                                     and not other.been_jumped
                                     and other.position < item.position
                                     and (

backend/app/services/queue_position.py

@@ -0,0 +1,45 @@
+"""The queue's position sequence.
+
+Positions are one sequence across every pending item, not one per printer
+(#3200). The queue page lists and reorders pending items as a single list, and
+the scheduler dispatches in that order, so a pinned job and an "Any <model>" job
+compete for a printer by position. A per-printer sequence gave each lane its
+own 1, 2, 3 and made a job added last land in the middle of the list.
+"""
+
+from sqlalchemy import func, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.print_queue import PrintQueueItem
+
+# Advisory lock key for the shared sequence, in the 1625 namespace. The key
+# used to be the printer id, back when every printer had a sequence of its own.
+QUEUE_POSITION_LOCK_KEY = 0
+
+
+async def lock_queue_positions(db: AsyncSession) -> None:
+    """Serialize inserts that read MAX(position) (#1625-followup).
+
+    Two concurrent inserts into an empty queue would otherwise both read
+    MAX(position) as 0 and land on position 1. The lock is transaction-scoped
+    and released at commit/rollback. SQLite serializes writes implicitly and
+    needs no equivalent.
+
+    The dialect is read from the session binding, not the ``is_sqlite()``
+    helper: the test fixture binds a SQLite engine while
+    ``settings.database_url`` may still point at Postgres.
+    """
+    if db.get_bind().dialect.name == "postgresql":
+        await db.execute(text("SELECT pg_advisory_xact_lock(1625, :k)"), {"k": QUEUE_POSITION_LOCK_KEY})
+
+
+async def max_queue_position(db: AsyncSession) -> int:
+    """Highest position among pending items, or 0 for an empty queue."""
+    result = await db.execute(select(func.max(PrintQueueItem.position)).where(PrintQueueItem.status == "pending"))
+    return result.scalar() or 0
+
+
+async def next_queue_position(db: AsyncSession) -> int:
+    """Position for an item appended to the end of the queue, under the lock."""
+    await lock_queue_positions(db)
+    return await max_queue_position(db) + 1

backend/app/services/virtual_printer/manager.py

@@ -1045,13 +1045,11 @@ class VirtualPrinterInstance:
                     # empty queue.
                     from sqlalchemy import func, select as _sql_select
 
+                    # One sequence across all pending items, not one per
+                    # printer (#3200): the plates go to the end of the queue.
                     queue_scope = _sql_select(func.max(PrintQueueItem.position)).where(
                         PrintQueueItem.status == "pending"
                     )
-                    if self.target_printer_id is not None:
-                        queue_scope = queue_scope.where(PrintQueueItem.printer_id == self.target_printer_id)
-                    else:
-                        queue_scope = queue_scope.where(PrintQueueItem.printer_id.is_(None))
                     try:
                         max_pos_raw = (await db.execute(queue_scope)).scalar()
                         max_pos = int(max_pos_raw) if max_pos_raw is not None else 0

backend/tests/integration/test_print_batch_orders.py

@@ -498,10 +498,15 @@ class TestBatchOrderDispatch:
             # from under the rest of the order.
             assert clone.cleanup_library_after_dispatch is False
 
-    async def test_clones_land_in_their_own_printer_queue(
+    async def test_clones_land_at_the_end_of_the_queue(
         self, async_client, printer_factory, archive_factory, db_session
     ):
-        """Positions are per-printer sequences — a global MAX would scramble them."""
+        """Positions are one sequence across every printer (#3200).
+
+        The queue page shows and reorders pending items as one list and the
+        scheduler dispatches in that order, so clones go after everything
+        already queued, and the whole queue stays free of gaps and duplicates.
+        """
         printer_a = await printer_factory()
         printer_b = await printer_factory()
         archive = await archive_factory()
@@ -510,35 +515,34 @@ class TestBatchOrderDispatch:
             archive.id,
             [{"plate_id": 1, "quantity_target": 3}, {"plate_id": 2, "quantity_target": 2}],
         )
-        # Pad printer B's queue so a global MAX would push plate 1's clones
-        # past the end of printer A's much shorter queue.
         for _ in range(5):
             await async_client.post("/api/v1/queue/", json={"printer_id": printer_b.id, "archive_id": archive.id})
         await _queue_item(async_client, printer_a.id, archive.id, order["id"], plate_id=1)
         await _queue_item(async_client, printer_b.id, archive.id, order["id"], plate_id=2)
 
-        response = await async_client.post(f"/api/v1/queue/batches/{order['id']}/dispatch", json={})
-        assert response.status_code == 200
-
         from sqlalchemy import select
 
         from backend.app.models.print_queue import PrintQueueItem
 
-        for printer in (printer_a, printer_b):
-            rows = (
-                (
-                    await db_session.execute(
-                        select(PrintQueueItem)
-                        .where(PrintQueueItem.printer_id == printer.id)
-                        .where(PrintQueueItem.status == "pending")
-                    )
-                )
+        async def pending():
+            return (
+                (await db_session.execute(select(PrintQueueItem).where(PrintQueueItem.status == "pending")))
                 .scalars()
                 .all()
             )
-            positions = sorted(r.position for r in rows)
-            assert len(positions) == len(set(positions)), f"duplicate positions on printer {printer.id}"
-            assert positions == list(range(1, len(rows) + 1)), f"gap in printer {printer.id} queue"
+
+        before_ids = {r.id for r in await pending()}
+        last_before = max(r.position for r in await pending())
+
+        response = await async_client.post(f"/api/v1/queue/batches/{order['id']}/dispatch", json={})
+        assert response.status_code == 200
+
+        rows = await pending()
+        positions = sorted(r.position for r in rows)
+        assert positions == list(range(1, len(rows) + 1)), "queue positions have a gap or a duplicate"
+        clones = [r for r in rows if r.id not in before_ids]
+        assert len(clones) == 3
+        assert all(r.position > last_before for r in clones)
 
     async def test_clone_differs_from_its_source_only_in_lifecycle_state(
         self, async_client, printer_factory, archive_factory, db_session

backend/tests/integration/test_print_queue_api.py

@@ -2761,10 +2761,16 @@ class TestAbortedStatusNormalisation:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_add_to_queue_insert_position_scopes_unassigned_items(
+    async def test_add_to_queue_positions_are_one_sequence_across_printers(
         self, async_client: AsyncClient, printer_factory, archive_factory, db_session
     ):
-        """Unassigned inserts shift only the unassigned queue scope."""
+        """Positions are shared by pinned and unassigned items (#3200).
+
+        The queue page lists and reorders pending items as one list and the
+        scheduler dispatches in that order, so a pinned item added after two
+        unassigned ones lands after them, and an insert at the top shifts
+        everything, not only its own printer's items.
+        """
         printer = await printer_factory()
         unassigned_first = await archive_factory(print_name="Unassigned First")
         unassigned_second = await archive_factory(print_name="Unassigned Second")
@@ -2778,7 +2784,7 @@ class TestAbortedStatusNormalisation:
             json={"printer_id": printer.id, "archive_id": assigned.id},
         )
         assert assigned_response.status_code == 200
-        assert assigned_response.json()["position"] == 1
+        assert assigned_response.json()["position"] == 3
 
         response = await async_client.post(
             "/api/v1/queue/",
@@ -2789,19 +2795,14 @@ class TestAbortedStatusNormalisation:
         )
         assert response.status_code == 200
 
-        unassigned_response = await async_client.get("/api/v1/queue/?printer_id=-1")
-        unassigned_items = sorted(unassigned_response.json(), key=lambda item: item["position"])
-        assert [item["archive_id"] for item in unassigned_items] == [
+        all_items = sorted((await async_client.get("/api/v1/queue/")).json(), key=lambda item: item["position"])
+        assert [item["archive_id"] for item in all_items] == [
             priority.id,
             unassigned_first.id,
             unassigned_second.id,
+            assigned.id,
         ]
-        assert [item["position"] for item in unassigned_items] == [1, 2, 3]
-
-        assigned_scope_response = await async_client.get(f"/api/v1/queue/?printer_id={printer.id}&target_model=NONE")
-        assigned_items = sorted(assigned_scope_response.json(), key=lambda item: item["position"])
-        assert [item["archive_id"] for item in assigned_items] == [assigned.id]
-        assert [item["position"] for item in assigned_items] == [1]
+        assert [item["position"] for item in all_items] == [1, 2, 3, 4]
 
     @pytest.mark.asyncio
     @pytest.mark.integration

backend/tests/unit/test_scheduler_global_queue_order_3200.py

@@ -0,0 +1,251 @@
+"""One queue order for pinned and "Any <model>" jobs (#3200).
+
+Three P2S printers. The user dragged a job pinned to printer 1 above two
+"Any P2S" jobs, printer 1 finished, and it started one of the "Any" jobs from
+further down the queue. The pinned job sat at the top saying "Busy".
+
+The scheduler read the queue ``ORDER BY printer_id, position``, which put the
+lane ahead of the position. A model-based job has ``printer_id`` NULL, which
+SQLite sorts first and PostgreSQL sorts last, so which job won a printer both
+wanted was decided by the database, never by where the user put it.
+
+These run on SQLite, where the "Any" jobs used to win.
+"""
+
+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
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+async def ctx():
+    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:
+        for pid in (1, 2, 3):
+            db.add(
+                Printer(
+                    id=pid,
+                    name=f"3D printer 0{pid}",
+                    serial_number=f"P2S000{pid}",
+                    ip_address=f"10.0.0.{pid}",
+                    access_code="x",
+                    model="P2S",
+                    is_active=True,
+                )
+            )
+        await db.commit()
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker)
+    finally:
+        await engine.dispose()
+
+
+async def _add(ctx, *, position, printer_id=None, target_model=None, print_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": "P2S"},
+        )
+        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,
+            print_time_seconds=print_time,
+            manual_start=manual_start,
+        )
+        db.add(item)
+        await db.commit()
+        return item.id
+
+
+async def _pinned(ctx, position, printer_id=1, **kw):
+    return await _add(ctx, position=position, printer_id=printer_id, **kw)
+
+
+async def _any(ctx, position, **kw):
+    return await _add(ctx, position=position, target_model="P2S", **kw)
+
+
+async def _set(ctx, key, value):
+    async with ctx.session_maker() as db:
+        db.add(Settings(key=key, value=value))
+        await db.commit()
+
+
+async def _item(ctx, item_id):
+    async with ctx.session_maker() as db:
+        return (await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one()
+
+
+async def _run(ctx, *, idle_printers):
+    """One check_queue pass; returns {item_id: printer_id} for what went out."""
+    scheduler = PrintScheduler()
+    launched = MagicMock()
+    patches = [
+        patch("backend.app.services.print_scheduler.async_session", ctx.session_maker),
+        patch("backend.app.core.database.async_session", ctx.session_maker),
+        patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)),
+        patch(
+            "backend.app.services.print_scheduler.printer_manager.is_awaiting_plate_clear",
+            MagicMock(return_value=False),
+        ),
+        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,
+            "_is_printer_idle",
+            MagicMock(side_effect=lambda pid, *_a, **_k: pid in idle_printers),
+        ),
+        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, "_get_smart_plugs", AsyncMock(return_value=[])),
+        patch.object(scheduler, "_launch_uploads", launched),
+    ]
+    with ExitStack() as stack:
+        for p in patches:
+            stack.enter_context(p)
+        await scheduler.check_queue()
+
+    if not launched.called:
+        return {}
+    dispatched = {}
+    for item_id in launched.call_args[0][0]:
+        dispatched[item_id] = (await _item(ctx, item_id)).printer_id
+    return dispatched
+
+
+class TestPositionDecidesWhoGetsThePrinter:
+    @pytest.mark.asyncio
+    async def test_the_reporters_queue(self, ctx):
+        """Pinned job dragged to the top, two "Any P2S" jobs below it, only
+        printer 1 free: the pinned job starts."""
+        shuttle_a = await _any(ctx, 2)
+        shuttle_b = await _any(ctx, 3)
+        coral = await _pinned(ctx, 1)
+
+        assert await _run(ctx, idle_printers={1}) == {coral: 1}
+        assert (await _item(ctx, shuttle_a)).status == "pending"
+        assert (await _item(ctx, shuttle_b)).status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_an_any_job_above_a_pinned_one_goes_first(self, ctx):
+        """The other direction: position still decides, so a model-based job
+        the user put higher takes the printer."""
+        pinned = await _pinned(ctx, 2)
+        any_job = await _any(ctx, 1)
+
+        assert await _run(ctx, idle_printers={1}) == {any_job: 1}
+        assert (await _item(ctx, pinned)).status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_a_pinned_job_that_cannot_start_does_not_hold_up_other_printers(self, ctx):
+        """The top job is pinned to a busy printer; the "Any" job below it
+        still takes the free one."""
+        await _pinned(ctx, 1, printer_id=2)
+        any_job = await _any(ctx, 2)
+
+        assert await _run(ctx, idle_printers={1}) == {any_job: 1}
+
+    @pytest.mark.asyncio
+    async def test_several_free_printers_each_get_the_next_job(self, ctx):
+        coral = await _pinned(ctx, 1)
+        shuttle_a = await _any(ctx, 2)
+        shuttle_b = await _any(ctx, 3)
+
+        dispatched = await _run(ctx, idle_printers={1, 2, 3})
+
+        assert dispatched[coral] == 1
+        assert {dispatched[shuttle_a], dispatched[shuttle_b]} == {2, 3}
+
+    @pytest.mark.asyncio
+    async def test_a_staged_job_at_the_top_does_not_block_the_printer(self, ctx):
+        """A manual-start job waits for the user, so the next job takes the
+        printer rather than the queue stalling behind it."""
+        await _pinned(ctx, 1, manual_start=True)
+        any_job = await _any(ctx, 2)
+
+        assert await _run(ctx, idle_printers={1}) == {any_job: 1}
+
+
+class TestShortestJobFirstAcrossLanes:
+    @pytest.mark.asyncio
+    async def test_a_shorter_any_job_beats_a_longer_pinned_one(self, ctx):
+        await _set(ctx, "queue_shortest_first", "true")
+        pinned = await _pinned(ctx, 1, print_time=7200)
+        short_any = await _any(ctx, 2, print_time=600)
+
+        assert await _run(ctx, idle_printers={1}) == {short_any: 1}
+        assert (await _item(ctx, pinned)).status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_the_pinned_job_it_jumped_is_marked_and_goes_next(self, ctx):
+        """The starvation guard has to look across lanes, or a stream of short
+        "Any" jobs holds the pinned one back forever."""
+        await _set(ctx, "queue_shortest_first", "true")
+        pinned = await _pinned(ctx, 1, print_time=7200)
+        await _any(ctx, 2, print_time=600)
+
+        await _run(ctx, idle_printers={1})
+        assert (await _item(ctx, pinned)).been_jumped is True
+
+        await _any(ctx, 3, print_time=300)
+        async with ctx.session_maker() as db:
+            for row in (await db.execute(select(PrintQueueItem).where(PrintQueueItem.status != "pending"))).scalars():
+                row.status = "completed"
+            await db.commit()
+
+        assert await _run(ctx, idle_printers={1}) == {pinned: 1}
+
+    @pytest.mark.asyncio
+    async def test_a_jumped_any_job_is_marked_when_a_pinned_job_goes_first(self, ctx):
+        await _set(ctx, "queue_shortest_first", "true")
+        long_any = await _any(ctx, 1, print_time=7200)
+        short_pinned = await _pinned(ctx, 2, print_time=600)
+
+        assert await _run(ctx, idle_printers={1}) == {short_pinned: 1}
+        assert (await _item(ctx, long_any)).been_jumped is True
+
+    @pytest.mark.asyncio
+    async def test_an_any_job_for_another_model_is_not_marked(self, ctx):
+        await _set(ctx, "queue_shortest_first", "true")
+        other_model = await _add(ctx, position=1, target_model="X1C", print_time=7200)
+        await _pinned(ctx, 2, print_time=600)
+
+        await _run(ctx, idle_printers={1})
+        assert (await _item(ctx, other_model)).been_jumped is False