fix(queue): scope force-colour overrides to the plate the item prints (#2551)

Queueing several plates of one 3MF built a single filament-override list from
every selected plate and posted that same list with each plate's item. A
force_color_match entry blocks dispatch until the printer has that exact colour
loaded, so a single-colour plate waited on the whole batch's palette. The same
shared list also widened required_filament_types, making a PLA plate refuse
every printer that lacked a sibling plate's PETG.

Narrow the overrides to the slots the plate actually consumes, on create and on
update -- in the backend, where the 3MF is, so it holds for every writer of the
queue. Dispatch already re-parsed requirements per plate and keyed overrides by
slot, so the dropped entries were inert there. When the plate's slots cannot be
read the overrides are kept whole: an item waiting on a colour it does not need
is visible, one that silently lost a forced colour prints in the wrong filament.

Items queued before this would stay stuck with a waiting reason that explains
nothing, so a startup migration re-scopes the pending ones. Printing and
finished items keep their overrides -- that is a record of what they dispatched
with, not an instruction.

backend/app/api/routes/print_queue.py

@@ -35,6 +35,7 @@ from backend.app.schemas.print_queue import (
     PrintQueueReorder,
 )
 from backend.app.services.filament_deficit import compute_deficit_for_queue_item
+from backend.app.services.filament_requirements import overrides_for_plate
 from backend.app.services.notification_service import notification_service
 from backend.app.utils.printer_models import normalize_printer_model, normalize_printer_model_id
 from backend.app.utils.threemf_tools import (
@@ -115,6 +116,22 @@ def _extract_filament_types_from_3mf(file_path: Path, plate_id: int | None = Non
 _extract_print_time_from_3mf = extract_print_time_from_3mf
 
 
+async def _resolve_source_path(db: AsyncSession, item: PrintQueueItem) -> Path | None:
+    """Resolve an existing queue item's source 3MF on disk, or None."""
+    if item.archive_id:
+        result = await db.execute(select(PrintArchive).where(PrintArchive.id == item.archive_id))
+        archive = result.scalar_one_or_none()
+        if archive:
+            return settings.base_dir / archive.file_path
+    elif item.library_file_id:
+        result = await db.execute(LibraryFile.active().where(LibraryFile.id == item.library_file_id))
+        library_file = result.scalar_one_or_none()
+        if library_file:
+            lib_path = Path(library_file.file_path)
+            return lib_path if lib_path.is_absolute() else settings.base_dir / library_file.file_path
+    return None
+
+
 def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse:
     """Add nested archive/printer/library_file info to response."""
     # Parse ams_mapping from JSON string BEFORE model_validate
@@ -451,9 +468,9 @@ async def add_to_queue(
 
     # Extract filament types for model-based assignment (used by scheduler for validation)
     required_filament_types = None
+    file_path = None
     if target_model_norm:
         # Get file path from archive or library file
-        file_path = None
         if archive:
             file_path = settings.base_dir / archive.file_path
         elif library_file:
@@ -469,15 +486,17 @@ async def add_to_queue(
     # If filament overrides are provided, update required_filament_types to match override types
     filament_overrides_json = None
     if data.filament_overrides and target_model_norm:
-        filament_overrides_json = json.dumps(data.filament_overrides)
-        # Update required_filament_types from overrides so scheduler validates against overridden types
-        override_types = sorted({o["type"] for o in data.filament_overrides if "type" in o})
-        if override_types:
-            # Merge with existing types (overrides may only cover some slots)
-            existing_types = set(json.loads(required_filament_types)) if required_filament_types else set()
-            # Replace types for overridden slots, keep others
-            all_types = existing_types | set(override_types)
-            required_filament_types = json.dumps(sorted(all_types))
+        plate_overrides = overrides_for_plate(data.filament_overrides, file_path, data.plate_id)
+        if plate_overrides:
+            filament_overrides_json = json.dumps(plate_overrides)
+            # Update required_filament_types from overrides so scheduler validates against overridden types
+            override_types = sorted({o["type"] for o in plate_overrides if "type" in o})
+            if override_types:
+                # Merge with existing types (overrides may only cover some slots)
+                existing_types = set(json.loads(required_filament_types)) if required_filament_types else set()
+                # Replace types for overridden slots, keep others
+                all_types = existing_types | set(override_types)
+                required_filament_types = json.dumps(sorted(all_types))
 
     # Validate quantity
     quantity = max(1, data.quantity)
@@ -1084,11 +1103,18 @@ async def update_queue_item(
     if "ams_mapping" in update_data:
         update_data["ams_mapping"] = json.dumps(update_data["ams_mapping"]) if update_data["ams_mapping"] else None
 
-    # Serialize filament_overrides to JSON for TEXT column storage
+    # Serialize filament_overrides to JSON for TEXT column storage, keeping only
+    # the slots this item's plate actually prints (#2551 — same shared-override
+    # list the create path narrows).
     if "filament_overrides" in update_data:
-        update_data["filament_overrides"] = (
-            json.dumps(update_data["filament_overrides"]) if update_data["filament_overrides"] else None
-        )
+        overrides = update_data["filament_overrides"]
+        if overrides:
+            overrides = overrides_for_plate(
+                overrides,
+                await _resolve_source_path(db, item),
+                update_data.get("plate_id", item.plate_id),
+            )
+        update_data["filament_overrides"] = json.dumps(overrides) if overrides else None
 
     # Serialize H2C rack-swap nozzle pick (#1780) to JSON for TEXT column
     # storage; same Text-as-opaque-blob convention as ams_mapping above.

backend/app/core/database.py

@@ -439,6 +439,77 @@ async def _migrate_normalize_printer_ids(conn) -> None:
             await conn.execute(text("UPDATE api_keys SET printer_ids = NULL WHERE printer_ids::text = '[]'"))
 
 
+async def _migrate_scope_force_color_overrides_to_plate(conn) -> None:
+    """Re-scope queue items that carry another plate's filament overrides (#2551).
+
+    Queueing several plates of one 3MF used to store the union of every selected
+    plate's overrides on each item, so a ``force_color_match`` plate printing one
+    colour sat at Waiting until a printer had the whole batch's palette loaded.
+    The write paths now narrow to the plate, but items queued before the fix would
+    stay stuck until the user deleted and re-added them by hand — with a waiting
+    reason that gives no hint as to why. Repair them here instead.
+
+    Only pending items are touched: a printing or finished item's overrides are a
+    record of what it dispatched with, not an instruction. An item whose plate we
+    cannot read keeps every override, per ``overrides_for_plate``. Idempotent —
+    an already-scoped item narrows to itself and is not rewritten.
+    """
+    import json
+    from pathlib import Path
+
+    from sqlalchemy import text
+
+    from backend.app.services.filament_requirements import overrides_for_plate
+
+    rows = (
+        await conn.execute(
+            text(
+                "SELECT q.id, q.plate_id, q.filament_overrides, "
+                "a.file_path AS archive_path, l.file_path AS library_path "
+                "FROM print_queue q "
+                "LEFT JOIN print_archives a ON a.id = q.archive_id "
+                "LEFT JOIN library_files l ON l.id = q.library_file_id "
+                "WHERE q.status = 'pending' "
+                "AND q.plate_id IS NOT NULL "
+                "AND q.filament_overrides IS NOT NULL"
+            )
+        )
+    ).fetchall()
+
+    repaired = 0
+    for row in rows:
+        try:
+            overrides = json.loads(row.filament_overrides)
+        except (json.JSONDecodeError, TypeError):
+            continue
+        if not isinstance(overrides, list) or not overrides:
+            continue
+
+        stored_path = row.archive_path or row.library_path
+        if not stored_path:
+            continue
+        path = Path(stored_path)
+        if not path.is_absolute():
+            path = settings.base_dir / stored_path
+
+        scoped = overrides_for_plate(overrides, path, row.plate_id)
+        if len(scoped) == len(overrides):
+            continue
+
+        async with conn.begin_nested():
+            await conn.execute(
+                text("UPDATE print_queue SET filament_overrides = :overrides WHERE id = :id"),
+                {"overrides": json.dumps(scoped) if scoped else None, "id": row.id},
+            )
+        repaired += 1
+
+    if repaired:
+        logger.info(
+            "Re-scoped the filament overrides of %d queued item(s) to the plate they print (#2551)",
+            repaired,
+        )
+
+
 async def _migrate_drop_library_print_name(conn) -> None:
     """Strip the embedded 3MF Title (``print_name``) from library file metadata (#1489).
 
@@ -3179,6 +3250,10 @@ async def run_migrations(conn):
     # file metadata so the FileManager displays the filename, not the title (#1489).
     await _migrate_drop_library_print_name(conn)
 
+    # Data migration: queue items written before #2551 carry every selected plate's
+    # filament overrides, so a force-colour plate waits on colours it never prints.
+    await _migrate_scope_force_color_overrides_to_plate(conn)
+
     # Backfill NULL print_archives.created_at — older rows (and rows imported
     # via the SQLite ↔ Postgres cross-DB restore path) can land with NULL
     # because the column was originally created without a DEFAULT clause and

backend/app/services/filament_requirements.py

@@ -102,6 +102,57 @@ def extract_filament_requirements(file_path: Path, plate_id: int | None = None)
     return filaments
 
 
+def overrides_for_plate(
+    overrides: list[dict],
+    file_path: Path | None,
+    plate_id: int | None,
+) -> list[dict]:
+    """Drop the filament overrides whose slots this plate never prints.
+
+    Queueing several plates of one 3MF builds a single override list out of every
+    selected plate's filaments and hands that same list to each plate's item. A
+    ``force_color_match`` entry blocks dispatch until the printer has that exact
+    colour loaded, so a single-colour plate ended up waiting on every colour in
+    the batch (#2551). Each item may only demand what its own plate consumes.
+
+    Overrides are kept as-is when the plate's slots cannot be established (whole
+    file selected, source gone, unreadable 3MF, malformed entry): an item that
+    waits on a colour it does not need is visible and fixable, whereas one that
+    silently loses a forced colour can dispatch the print in the wrong filament.
+    """
+    if not overrides or plate_id is None or file_path is None or not file_path.exists():
+        return overrides
+
+    plate_slots = {f["slot_id"] for f in extract_filament_requirements(file_path, plate_id)}
+    if not plate_slots:
+        logger.warning(
+            "Cannot read the filaments of plate %s in %s; keeping all %d filament override(s)",
+            plate_id,
+            file_path.name,
+            len(overrides),
+        )
+        return overrides
+
+    narrowed = []
+    for override in overrides:
+        try:
+            slot_id = int(override["slot_id"])
+        except (KeyError, TypeError, ValueError):
+            narrowed.append(override)
+            continue
+        if slot_id in plate_slots:
+            narrowed.append(override)
+
+    if len(narrowed) != len(overrides):
+        logger.info(
+            "Plate %s: kept %d of %d filament override(s) — the rest belong to other plates",
+            plate_id,
+            len(narrowed),
+            len(overrides),
+        )
+    return narrowed
+
+
 def _collect_filaments(parent: ET.Element, into: list[dict]) -> None:
     """Walk every `./filament` child under `parent` and append normalised
     entries to `into`. Skips filaments with `used_g <= 0` (slot present in

backend/tests/integration/test_print_queue_api.py

@@ -2807,3 +2807,260 @@ class TestReorderEndpoint:
         await db_session.refresh(item2)
         assert item1.position == 2
         assert item2.position == 1
+
+
+class TestForceColorOverridesAreScopedToThePlate:
+    """Queueing several plates of one 3MF must not make each plate wait on the
+    colours of its siblings (#2551).
+
+    The print dialog builds one override list from every selected plate and posts
+    that same list with each plate's item, so the API is what has to keep only the
+    slots the plate prints -- a ``force_color_match`` entry blocks dispatch until
+    the printer has that exact colour loaded.
+    """
+
+    THREE_PLATES = """<?xml version="1.0" encoding="UTF-8"?>
+    <config>
+        <plate>
+            <metadata key="index" value="1"/>
+            <filament id="1" used_g="50.0" type="PLA" color="#0B2C7A"/>
+        </plate>
+        <plate>
+            <metadata key="index" value="2"/>
+            <filament id="2" used_g="40.0" type="PLA" color="#9B9EA0"/>
+        </plate>
+        <plate>
+            <metadata key="index" value="3"/>
+            <filament id="3" used_g="30.0" type="PLA" color="#F4EE2A"/>
+        </plate>
+    </config>
+    """
+
+    # What the dialog posts for every plate: the union of all three plates'
+    # filaments, each one force-matched.
+    ALL_THREE_COLORS = [
+        {"slot_id": 1, "type": "PLA", "color": "#0B2C7A", "color_name": "Army Blue", "force_color_match": True},
+        {"slot_id": 2, "type": "PLA", "color": "#9B9EA0", "color_name": "Ash Grey", "force_color_match": True},
+        {"slot_id": 3, "type": "PLA", "color": "#F4EE2A", "color_name": "Sunshine Yellow", "force_color_match": True},
+    ]
+
+    @pytest.fixture
+    async def multi_plate_archive(self, db_session, tmp_path):
+        """An archive whose 3MF really exists on disk, one colour per plate."""
+        import zipfile
+
+        from backend.app.models.archive import PrintArchive
+
+        file_path = tmp_path / "three_plates.gcode.3mf"
+        with zipfile.ZipFile(file_path, "w") as zf:
+            zf.writestr("Metadata/slice_info.config", self.THREE_PLATES)
+
+        archive = PrintArchive(
+            filename="three_plates.gcode.3mf",
+            print_name="Three Plates",
+            file_path=str(file_path),
+            file_size=file_path.stat().st_size,
+            content_hash="platehash0001",
+            status="completed",
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+        return archive
+
+    @pytest.fixture
+    async def x1c(self, db_session):
+        from backend.app.models.printer import Printer
+
+        printer = Printer(
+            name="Force Color X1C",
+            ip_address="192.168.1.210",
+            serial_number="FORCECOLOR01",
+            access_code="12345678",
+            model="X1C",
+        )
+        db_session.add(printer)
+        await db_session.commit()
+        await db_session.refresh(printer)
+        return printer
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_each_plate_keeps_only_the_colour_it_prints(
+        self, async_client: AsyncClient, multi_plate_archive, x1c
+    ):
+        """The bug: plate 1 prints Army Blue only, but was stored demanding all three."""
+        stored = {}
+        for plate_id in (1, 2, 3):
+            response = await async_client.post(
+                "/api/v1/queue/",
+                json={
+                    "target_model": "X1C",
+                    "archive_id": multi_plate_archive.id,
+                    "plate_id": plate_id,
+                    "filament_overrides": self.ALL_THREE_COLORS,
+                },
+            )
+            assert response.status_code == 200
+            stored[plate_id] = response.json()["filament_overrides"]
+
+        assert [o["color_name"] for o in stored[1]] == ["Army Blue"]
+        assert [o["color_name"] for o in stored[2]] == ["Ash Grey"]
+        assert [o["color_name"] for o in stored[3]] == ["Sunshine Yellow"]
+        # The slot each entry maps to has to survive narrowing untouched, or the
+        # dispatch-time AMS mapping would key the override onto the wrong slot.
+        assert [o["slot_id"] for o in stored[2]] == [2]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_whole_file_queue_keeps_every_colour(self, async_client: AsyncClient, multi_plate_archive, x1c):
+        """No plate_id means the job prints the whole file, so every colour is needed."""
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "target_model": "X1C",
+                "archive_id": multi_plate_archive.id,
+                "filament_overrides": self.ALL_THREE_COLORS,
+            },
+        )
+        assert response.status_code == 200
+        assert len(response.json()["filament_overrides"]) == 3
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_unreadable_3mf_keeps_every_colour(self, async_client: AsyncClient, db_session, tmp_path, x1c):
+        """When the plate's slots can't be read, keep the overrides rather than drop them.
+
+        An item waiting on a colour it doesn't need is visible and fixable; one that
+        silently lost its forced colour would dispatch in the wrong filament.
+        """
+        from backend.app.models.archive import PrintArchive
+
+        file_path = tmp_path / "not_a_zip.gcode.3mf"
+        file_path.write_text("this is not a 3mf")
+        archive = PrintArchive(
+            filename="not_a_zip.gcode.3mf",
+            print_name="Corrupt",
+            file_path=str(file_path),
+            file_size=file_path.stat().st_size,
+            content_hash="platehash0002",
+            status="completed",
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "target_model": "X1C",
+                "archive_id": archive.id,
+                "plate_id": 1,
+                "filament_overrides": self.ALL_THREE_COLORS,
+            },
+        )
+        assert response.status_code == 200
+        assert len(response.json()["filament_overrides"]) == 3
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_required_types_stay_scoped_to_the_plate(self, async_client: AsyncClient, db_session, tmp_path, x1c):
+        """Override types are merged into required_filament_types, so a shared list
+        also widened the type gate -- a PLA-only plate demanded PETG as well."""
+        import zipfile
+
+        from backend.app.models.archive import PrintArchive
+
+        xml = """<?xml version="1.0" encoding="UTF-8"?>
+        <config>
+            <plate>
+                <metadata key="index" value="1"/>
+                <filament id="1" used_g="50.0" type="PLA" color="#0B2C7A"/>
+            </plate>
+            <plate>
+                <metadata key="index" value="2"/>
+                <filament id="2" used_g="40.0" type="PETG" color="#9B9EA0"/>
+            </plate>
+        </config>
+        """
+        file_path = tmp_path / "mixed_types.gcode.3mf"
+        with zipfile.ZipFile(file_path, "w") as zf:
+            zf.writestr("Metadata/slice_info.config", xml)
+        archive = PrintArchive(
+            filename="mixed_types.gcode.3mf",
+            print_name="Mixed",
+            file_path=str(file_path),
+            file_size=file_path.stat().st_size,
+            content_hash="platehash0003",
+            status="completed",
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        await db_session.refresh(archive)
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "target_model": "X1C",
+                "archive_id": archive.id,
+                "plate_id": 1,
+                "filament_overrides": [
+                    {"slot_id": 1, "type": "PLA", "color": "#0B2C7A", "force_color_match": True},
+                    {"slot_id": 2, "type": "PETG", "color": "#9B9EA0", "force_color_match": True},
+                ],
+            },
+        )
+        assert response.status_code == 200
+        assert response.json()["required_filament_types"] == ["PLA"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_editing_an_item_narrows_the_overrides_too(
+        self, async_client: AsyncClient, db_session, multi_plate_archive, x1c
+    ):
+        """The edit dialog posts the same shared list, so PATCH narrows it as well."""
+        from backend.app.models.print_queue import PrintQueueItem
+
+        item = PrintQueueItem(
+            target_model="X1C",
+            archive_id=multi_plate_archive.id,
+            plate_id=2,
+            status="pending",
+            position=1,
+        )
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+
+        response = await async_client.patch(
+            f"/api/v1/queue/{item.id}",
+            json={"filament_overrides": self.ALL_THREE_COLORS},
+        )
+        assert response.status_code == 200
+        assert [o["color_name"] for o in response.json()["filament_overrides"]] == ["Ash Grey"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_editing_the_plate_renarrows_against_the_new_plate(
+        self, async_client: AsyncClient, db_session, multi_plate_archive, x1c
+    ):
+        """Moving an item to another plate must re-scope its colours to that plate."""
+        from backend.app.models.print_queue import PrintQueueItem
+
+        item = PrintQueueItem(
+            target_model="X1C",
+            archive_id=multi_plate_archive.id,
+            plate_id=1,
+            status="pending",
+            position=1,
+        )
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+
+        response = await async_client.patch(
+            f"/api/v1/queue/{item.id}",
+            json={"plate_id": 3, "filament_overrides": self.ALL_THREE_COLORS},
+        )
+        assert response.status_code == 200
+        assert [o["color_name"] for o in response.json()["filament_overrides"]] == ["Sunshine Yellow"]

backend/tests/unit/test_force_color_plate_scope_migration.py

@@ -0,0 +1,230 @@
+"""Regression test for the force-colour override re-scoping migration (#2551).
+
+Queueing several plates of one 3MF used to store the union of every selected
+plate's filament overrides on each item, so a `force_color_match` plate printing
+a single colour sat at Waiting until a printer had the whole batch's palette
+loaded. The write paths now scope overrides to the plate; this migration repairs
+the items queued before the fix, which would otherwise stay stuck forever with a
+waiting reason that explains nothing.
+"""
+
+from __future__ import annotations
+
+import json
+import zipfile
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from backend.app.core.database import run_migrations
+
+THREE_PLATES = """<?xml version="1.0" encoding="UTF-8"?>
+<config>
+    <plate>
+        <metadata key="index" value="1"/>
+        <filament id="1" used_g="50.0" type="PLA" color="#0B2C7A"/>
+    </plate>
+    <plate>
+        <metadata key="index" value="2"/>
+        <filament id="2" used_g="40.0" type="PLA" color="#9B9EA0"/>
+    </plate>
+    <plate>
+        <metadata key="index" value="3"/>
+        <filament id="3" used_g="30.0" type="PLA" color="#F4EE2A"/>
+    </plate>
+</config>
+"""
+
+ALL_THREE_COLORS = [
+    {"slot_id": 1, "type": "PLA", "color": "#0B2C7A", "color_name": "Army Blue", "force_color_match": True},
+    {"slot_id": 2, "type": "PLA", "color": "#9B9EA0", "color_name": "Ash Grey", "force_color_match": True},
+    {"slot_id": 3, "type": "PLA", "color": "#F4EE2A", "color_name": "Sunshine Yellow", "force_color_match": True},
+]
+
+
+@pytest.fixture(autouse=True)
+def force_sqlite_dialect(monkeypatch):
+    """Force the SQLite branch regardless of test env settings."""
+    from backend.app.core import database as database_module, db_dialect
+
+    monkeypatch.setattr(db_dialect, "is_sqlite", lambda: True)
+    monkeypatch.setattr(db_dialect, "is_postgres", lambda: False)
+    monkeypatch.setattr(database_module, "is_sqlite", lambda: True)
+
+
+def _register_all_models():
+    from backend.app.models import (  # noqa: F401
+        ams_history,
+        ams_label,
+        api_key,
+        archive,
+        color_catalog,
+        external_link,
+        filament,
+        group,
+        kprofile_note,
+        library,
+        maintenance,
+        notification,
+        notification_template,
+        print_log,
+        print_queue,
+        printer,
+        project,
+        project_bom,
+        settings,
+        slot_preset,
+        smart_plug,
+        smart_plug_energy_snapshot,
+        spool,
+        spool_assignment,
+        spool_catalog,
+        spool_k_profile,
+        spool_usage_history,
+        spoolbuddy_device,
+        user,
+        user_email_pref,
+        virtual_printer,
+    )
+
+
+@pytest.fixture
+async def engine():
+    from backend.app.core.database import Base
+
+    _register_all_models()
+
+    eng = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield eng
+    await eng.dispose()
+
+
+@pytest.fixture
+def three_plate_3mf(tmp_path):
+    path = tmp_path / "three_plates.gcode.3mf"
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr("Metadata/slice_info.config", THREE_PLATES)
+    return path
+
+
+async def _seed(engine, *, file_path, items) -> None:
+    """Insert one archive plus the queue items pointing at it.
+
+    Goes through the models rather than raw INSERTs so the columns this test
+    doesn't care about get their defaults.
+    """
+    from sqlalchemy.ext.asyncio import async_sessionmaker
+
+    from backend.app.models.archive import PrintArchive
+    from backend.app.models.print_queue import PrintQueueItem
+
+    async with async_sessionmaker(engine, expire_on_commit=False)() as session:
+        session.add(
+            PrintArchive(
+                id=1,
+                filename="three_plates.gcode.3mf",
+                file_path=str(file_path),
+                file_size=0,
+                status="completed",
+            )
+        )
+        for item_id, plate_id, overrides, status in items:
+            session.add(
+                PrintQueueItem(
+                    id=item_id,
+                    archive_id=1,
+                    target_model="X1C",
+                    plate_id=plate_id,
+                    filament_overrides=json.dumps(overrides),
+                    status=status,
+                    position=item_id,
+                )
+            )
+        await session.commit()
+
+
+async def _overrides(conn, item_id: int):
+    raw = (
+        await conn.execute(text("SELECT filament_overrides FROM print_queue WHERE id = :id"), {"id": item_id})
+    ).scalar_one()
+    return json.loads(raw) if raw else None
+
+
+@pytest.mark.asyncio
+async def test_each_stuck_item_is_rescoped_to_its_own_plate(engine, three_plate_3mf):
+    """The reporter's queue: three plates, each item demanding all three colours."""
+    await _seed(
+        engine,
+        file_path=three_plate_3mf,
+        items=[
+            (1, 1, ALL_THREE_COLORS, "pending"),
+            (2, 2, ALL_THREE_COLORS, "pending"),
+            (3, 3, ALL_THREE_COLORS, "pending"),
+        ],
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    async with engine.connect() as conn:
+        assert [o["color_name"] for o in await _overrides(conn, 1)] == ["Army Blue"]
+        assert [o["color_name"] for o in await _overrides(conn, 2)] == ["Ash Grey"]
+        assert [o["color_name"] for o in await _overrides(conn, 3)] == ["Sunshine Yellow"]
+
+
+@pytest.mark.asyncio
+async def test_it_is_idempotent(engine, three_plate_3mf):
+    """Every boot re-runs the migration set; an already-scoped item narrows to
+    itself and must not be rewritten or emptied."""
+    await _seed(engine, file_path=three_plate_3mf, items=[(1, 2, ALL_THREE_COLORS, "pending")])
+
+    for _ in range(2):
+        async with engine.begin() as conn:
+            await run_migrations(conn)
+
+    async with engine.connect() as conn:
+        assert [o["slot_id"] for o in await _overrides(conn, 1)] == [2]
+
+
+@pytest.mark.asyncio
+async def test_a_dispatched_item_is_left_alone(engine, three_plate_3mf):
+    """A printing item's overrides record what it dispatched with — that is
+    history, not an instruction, and rewriting it would falsify the record."""
+    await _seed(engine, file_path=three_plate_3mf, items=[(1, 1, ALL_THREE_COLORS, "printing")])
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    async with engine.connect() as conn:
+        assert len(await _overrides(conn, 1)) == 3
+
+
+@pytest.mark.asyncio
+async def test_a_whole_file_item_keeps_every_colour(engine, three_plate_3mf):
+    """No plate_id means the item prints the whole file, so it really does need
+    all three colours."""
+    await _seed(engine, file_path=three_plate_3mf, items=[(1, None, ALL_THREE_COLORS, "pending")])
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    async with engine.connect() as conn:
+        assert len(await _overrides(conn, 1)) == 3
+
+
+@pytest.mark.asyncio
+async def test_a_missing_source_file_does_not_strip_the_overrides(engine, tmp_path):
+    """The archive's 3MF is gone, so the plate's slots cannot be read. Keep the
+    overrides: an item waiting on a colour it does not need is visible and
+    fixable, one that silently lost a forced colour prints in the wrong filament.
+    """
+    await _seed(engine, file_path=tmp_path / "deleted.gcode.3mf", items=[(1, 1, ALL_THREE_COLORS, "pending")])
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    async with engine.connect() as conn:
+        assert len(await _overrides(conn, 1)) == 3