Route external spools to a nozzle when the printer has no AMS (#2771)

Five X2Ds with no AMS, each printing from its external spool holder,
took a job sent to a named printer and refused the same job sent to
"Any X2D": the file uploaded, the firmware answered 0700_8012 "Failed
to get AMS mapping table", and the item failed after three attempts.

A named-printer job carries a mapping the frontend resolved at queue
time, so the scheduler's matcher never runs. A model-based job has no
printer until dispatch, so the matcher does run -- and could not see an
external spool on a dual-nozzle printer. _build_loaded_filaments derived
dual-nozzle status from ams_extruder_map, which is built from AMS info
bits, so a printer with zero AMS units reported an empty map; every
external spool got extruder_id=None, and the nozzle-aware hard filter in
_match_filaments_to_slots discarded it because None equals neither 0 nor
1. The mapping came back all -1, was cleared to None, and the print
command went out as use_ams:true with no ams_mapping and no
ams_mapping2 at all.

This is the backend half of #1257, which fixed the same logic in
useFilamentMapping.ts and left this copy behind. Mirror its inference:
a populated nozzles[1].nozzle_diameter, a non-empty ams_extruder_map, or
more than one vt_tray entry. Replaying the reporter's own push-status
now yields extruder 1 for Ext-L and 0 for Ext-R, and a nozzle-1
requirement resolves to [254] -- what their working named-printer
dispatch sent. Single-nozzle printers keep extruder_id=None; nozzles
always has two entries, so its length alone must not be the signal.

Also stop dispatching a job the firmware is certain to reject. When the
matcher ran, matched nothing, and the printer has no AMS, fail the item
with the filament and nozzle it wants instead of spending an upload and
two retries on it -- that path already ended in a failed item, just an
opaque one. With an AMS attached the firmware error still stands, since
there the user can load a spool and press Resume. Fail-safe like the
nozzle-diameter guard (#1899): every branch short of a positive finding
returns None and dispatches as before.

_apply_filament_overrides is extracted from _compute_ams_mapping_for_printer
so the message names the filament the matcher looked for rather than the one
the 3MF was sliced with.

backend/app/services/print_scheduler.py

@@ -396,6 +396,43 @@ def _nozzle_mismatch_message(sliced_nozzle: float | None, installed: list[float]
     )
 
 
+def _describe_filament(entry: dict, nozzle_key: str) -> str:
+    """One-line "PETG #000000 (left nozzle)" for an error message (#2771).
+
+    Shared by the required and loaded sides, which name their extruder
+    differently: a 3MF requirement carries ``nozzle_id``, a loaded tray carries
+    ``extruder_id``. Both are MQTT extruder ids — 0 is the right/main nozzle,
+    1 the left/deputy — and both are absent on single-nozzle printers, where
+    naming a nozzle would be noise.
+    """
+    parts = [(entry.get("type") or "filament").upper()]
+    if entry.get("color"):
+        parts.append(str(entry["color"]))
+    nozzle = entry.get(nozzle_key)
+    if nozzle == 0:
+        parts.append("(right nozzle)")
+    elif nozzle == 1:
+        parts.append("(left nozzle)")
+    return " ".join(parts)
+
+
+def _unmatched_filament_message(required: list[dict], loaded: list[dict]) -> str:
+    """Explain that nothing loaded matches what the file needs (#2771).
+
+    Only ever built for a printer with no AMS, where the loaded list is short
+    enough to quote in full and there is no "load another spool and hit Resume"
+    recovery — the external spool holder is all there is, so the user needs to
+    be told which filament to put on it.
+    """
+    want = ", ".join(_describe_filament(r, "nozzle_id") for r in required)
+    have = ", ".join(_describe_filament(f, "extruder_id") for f in loaded)
+    return (
+        f"No filament loaded on this printer matches the file. It needs {want}; "
+        f"the printer has {have} and no AMS. Load the required filament on the "
+        f"external spool holder, or send this job to a printer that has it."
+    )
+
+
 class PrintScheduler:
     """Background scheduler that processes the print queue."""
 
@@ -808,7 +845,10 @@ class PrintScheduler:
                     # (all -1). A stored all-[-1] mapping is a bug artifact — a
                     # frontend status-load race can persist [-1] (#2589) — and
                     # must be recomputed from live trays rather than trusted.
-                    await self._ensure_ams_mapping(db, item.printer_id, item)
+                    unmappable = await self._ensure_ams_mapping(db, item.printer_id, item)
+                    if unmappable:
+                        await self._fail_unmappable_item(db, item, item.printer_id, unmappable)
+                        continue
 
                     # Filament-deficit pre-dispatch check (#1496). If the
                     # assigned spool can't satisfy any required slot grams,
@@ -1010,7 +1050,10 @@ class PrintScheduler:
                         # missing OR unresolved (all -1). Critical for model-based
                         # jobs where mapping wasn't computed upfront, and it also
                         # self-heals a bogus stored [-1] (#2589).
-                        await self._ensure_ams_mapping(db, printer_id, item)
+                        unmappable = await self._ensure_ams_mapping(db, printer_id, item)
+                        if unmappable:
+                            await self._fail_unmappable_item(db, item, printer_id, unmappable)
+                            continue
 
                         # Filament-deficit pre-dispatch check (#1496).
                         if await self._block_on_filament_deficit(db, item):
@@ -1625,7 +1668,7 @@ class PrintScheduler:
             # 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:
+    async def _ensure_ams_mapping(self, db: AsyncSession, printer_id: int, item: PrintQueueItem) -> str | None:
         """Ensure the queue item carries a usable AMS mapping before dispatch.
 
         Recomputes from live printer status when the stored mapping is missing OR
@@ -1641,6 +1684,14 @@ class PrintScheduler:
         external selection; the print command then keeps use_ams=True and the
         firmware surfaces a clear AMS-mapping error instead of silently printing
         to the empty external feed.
+
+        Returns an actionable message when that firmware error is the only
+        possible outcome — the matcher ran, matched nothing, and the printer has
+        no AMS to load a different spool into (#2771). The caller fails the item
+        on it instead of spending an upload on a print that cannot start.
+        Returns None everywhere else, including every case where we simply lack
+        the data to judge, so dispatch is only ever blocked on a positive
+        finding.
         """
         stored_mapping: list | None = None
         if item.ams_mapping:
@@ -1652,7 +1703,7 @@ class PrintScheduler:
         # Already resolved (present and not all-unresolved) — keep as-is so a
         # user's manual mapping is never overwritten.
         if item.ams_mapping and not _mapping_is_all_unresolved(stored_mapping):
-            return
+            return None
 
         computed_mapping = await self._compute_ams_mapping_for_printer(db, printer_id, item)
         if computed_mapping and not _mapping_is_all_unresolved(computed_mapping):
@@ -1664,7 +1715,9 @@ class PrintScheduler:
                 computed_mapping,
             )
             await db.commit()
-        elif _mapping_is_all_unresolved(stored_mapping):
+            return None
+
+        if _mapping_is_all_unresolved(stored_mapping):
             logger.warning(
                 "Queue item %s: stored ams_mapping %s is unresolved and could not be recomputed "
                 "from live status on printer %s; clearing it so dispatch does not treat it as external",
@@ -1675,6 +1728,105 @@ class PrintScheduler:
             item.ams_mapping = None
             await db.commit()
 
+        return await self._unmappable_without_ams_message(db, printer_id, item, computed_mapping)
+
+    async def _unmappable_without_ams_message(
+        self,
+        db: AsyncSession,
+        printer_id: int,
+        item: PrintQueueItem,
+        computed_mapping: list[int] | None,
+    ) -> str | None:
+        """Message for a mapping that resolved nothing on an AMS-less printer (#2771).
+
+        A print dispatched with no mapping goes out as ``use_ams: true`` with no
+        ``ams_mapping`` and no ``ams_mapping2``, which the firmware rejects with
+        0700_8012 "Failed to get AMS mapping table" — after Bambuddy has already
+        uploaded several megabytes and burned its dispatch retries. With an AMS
+        attached that error is worth reaching: the user can load the right spool
+        and press Resume, so this returns None and today's behaviour stands. With
+        no AMS there is nothing to resume into — the external spool holder is the
+        whole inventory — so the useful answer is to say which filament is
+        missing and stop.
+
+        Fail-safe by construction, mirroring the nozzle-diameter guard (#1899):
+        every branch that lacks the evidence to be sure returns None.
+        """
+        # None means the matcher never ran (no requirements parsed from the 3MF,
+        # or nothing loaded at all) rather than "ran and matched nothing". Those
+        # dispatch as they always have.
+        if not _mapping_is_all_unresolved(computed_mapping):
+            return None
+
+        status = printer_manager.get_status(printer_id)
+        if status is None:
+            return None
+
+        # "No AMS" has to be a fact the printer stated, not the absence of a
+        # statement. `raw_data["ams"]` is written only once an AMS push has been
+        # handled and is preserved across partial pushes thereafter, so a missing
+        # key means we have not heard yet — most likely a reconnect, where the
+        # trays of a fully loaded AMS would be invisible for a few seconds. An
+        # empty list is the positive report of a printer with no AMS.
+        ams_units = status.raw_data.get("ams")
+        if not isinstance(ams_units, list) or ams_units:
+            return None
+
+        required = await self._get_filament_requirements(db, item)
+        loaded = self._build_loaded_filaments(status)
+        if not required or not loaded:
+            # Both were non-empty moments ago or the matcher could not have run.
+            # If the picture changed under us, say nothing rather than fail an
+            # item on stale evidence.
+            return None
+        self._apply_filament_overrides(item, required)
+        return _unmatched_filament_message(required, loaded)
+
+    async def _fail_unmappable_item(
+        self, db: AsyncSession, item: PrintQueueItem, printer_id: int, message: str
+    ) -> None:
+        """Fail a queue item whose filament mapping cannot resolve (#2771).
+
+        This replaces a failure, not a success: without it the item is uploaded,
+        rejected by the firmware with 0700_8012, retried twice more and failed
+        anyway with "never started the print after N dispatch attempts". So this
+        applies on the model-based path too, even though it means an "Any <model>"
+        job stops at the first printer offered rather than trying its siblings —
+        deferring instead would need the check to move inside
+        ``_find_printer_for_model``'s candidate loop, since un-assigning here just
+        re-assigns the same printer on the next tick.
+        """
+        item.status = "failed"
+        item.error_message = message
+        item.completed_at = datetime.now(timezone.utc)
+        item.waiting_reason = None
+        await db.commit()
+        logger.warning(
+            "Queue item %s: no usable AMS mapping on printer %s — %s",
+            item.id,
+            printer_id,
+            message,
+        )
+
+        job_name = await self._get_job_name(db, item)
+        printer = await self._get_printer(db, printer_id)
+        await notification_service.on_queue_job_failed(
+            job_name=job_name,
+            printer_id=printer_id,
+            printer_name=printer.name if printer else "Unknown",
+            reason=message,
+            db=db,
+        )
+        try:
+            await ws_manager.send_queue_item_failed(
+                user_id=item.created_by_id,
+                queue_item_id=item.id,
+                printer_id=printer_id,
+                reason="filament_unmappable",
+            )
+        except Exception:
+            pass
+
     async def _compute_ams_mapping_for_printer(
         self, db: AsyncSession, printer_id: int, item: PrintQueueItem
     ) -> list[int] | None:
@@ -1727,37 +1879,7 @@ class PrintScheduler:
             logger.debug("No filament requirements found for queue item %s", item.id)
             return None
 
-        # Apply filament overrides if present
-        if item.filament_overrides:
-            try:
-                overrides = json.loads(item.filament_overrides)
-                override_map = {o["slot_id"]: o for o in overrides}
-                for req in filament_reqs:
-                    if req["slot_id"] in override_map:
-                        override = override_map[req["slot_id"]]
-                        req["type"] = override["type"]
-                        req["color"] = override["color"]
-                        # A manual/preference override SWAPS the slot's filament, so the
-                        # 3MF's original tray_info_idx now points at the old spool and must
-                        # be cleared — matching then falls back to type+colour. A
-                        # force_color_match override is not a swap: it carries the 3MF's
-                        # intended variant (Basic GFA00 / Matte GFA01 / Silk GFA06), so keep
-                        # it here too, letting the matcher pin the correct variant slot on a
-                        # printer holding two same-colour spools of different variants (#2650).
-                        # If that variant isn't loaded the matcher falls back to type+colour,
-                        # so an eligible printer never fails to map.
-                        req["tray_info_idx"] = (
-                            override.get("tray_info_idx", "") if override.get("force_color_match") else ""
-                        )
-                        logger.debug(
-                            "Queue item %s: Override slot %d -> %s %s",
-                            item.id,
-                            req["slot_id"],
-                            override["type"],
-                            override["color"],
-                        )
-            except (json.JSONDecodeError, KeyError, TypeError) as e:
-                logger.warning("Failed to apply filament overrides for queue item %s: %s", item.id, e)
+        self._apply_filament_overrides(item, filament_reqs)
 
         # Build loaded filaments from printer status
         loaded_filaments = self._build_loaded_filaments(status)
@@ -1791,6 +1913,47 @@ class PrintScheduler:
             filament_reqs, loaded_filaments, prefer_lowest, inventory_remain_overrides, fts_installed
         )
 
+    def _apply_filament_overrides(self, item: PrintQueueItem, filament_reqs: list[dict]) -> None:
+        """Rewrite ``filament_reqs`` in place with the item's per-slot overrides.
+
+        Extracted from ``_compute_ams_mapping_for_printer`` so the unmappable
+        diagnosis (#2771) describes the filament the matcher actually looked
+        for, not the one the 3MF was sliced with — naming the pre-override
+        filament in a user-facing error would send the user to load the wrong
+        spool.
+        """
+        if not item.filament_overrides:
+            return
+        try:
+            overrides = json.loads(item.filament_overrides)
+            override_map = {o["slot_id"]: o for o in overrides}
+            for req in filament_reqs:
+                if req["slot_id"] in override_map:
+                    override = override_map[req["slot_id"]]
+                    req["type"] = override["type"]
+                    req["color"] = override["color"]
+                    # A manual/preference override SWAPS the slot's filament, so the
+                    # 3MF's original tray_info_idx now points at the old spool and must
+                    # be cleared — matching then falls back to type+colour. A
+                    # force_color_match override is not a swap: it carries the 3MF's
+                    # intended variant (Basic GFA00 / Matte GFA01 / Silk GFA06), so keep
+                    # it here too, letting the matcher pin the correct variant slot on a
+                    # printer holding two same-colour spools of different variants (#2650).
+                    # If that variant isn't loaded the matcher falls back to type+colour,
+                    # so an eligible printer never fails to map.
+                    req["tray_info_idx"] = (
+                        override.get("tray_info_idx", "") if override.get("force_color_match") else ""
+                    )
+                    logger.debug(
+                        "Queue item %s: Override slot %d -> %s %s",
+                        item.id,
+                        req["slot_id"],
+                        override["type"],
+                        override["color"],
+                    )
+        except (json.JSONDecodeError, KeyError, TypeError) as e:
+            logger.warning("Failed to apply filament overrides for queue item %s: %s", item.id, e)
+
     def _build_override_direct_mapping(self, force_overrides: list[dict], status) -> list[int] | None:
         """Build an AMS mapping directly from force-color overrides without a 3MF.
 
@@ -1864,6 +2027,38 @@ class PrintScheduler:
         # Get ams_extruder_map for dual-nozzle printers (H2D, H2D Pro)
         ams_extruder_map = status.raw_data.get("ams_extruder_map", {})
 
+        # Dual-nozzle detection, used below to route external spools to an
+        # extruder (#2771). Mirrors `buildLoadedFilaments` in the frontend,
+        # which was corrected for #1257 while this copy kept the old signal.
+        #
+        # `ams_extruder_map` is derived from AMS info bits, so a dual-nozzle
+        # printer with zero AMS units reports an empty map — and every external
+        # spool then got `extruder_id=None`, which the nozzle-aware filter in
+        # `_match_filaments_to_slots` rejects outright because `None` equals
+        # neither 0 nor 1. On an X2D feeding from external spools only that left
+        # nothing to match, the mapping came back all -1, and the print went out
+        # with `use_ams: true` and no mapping table at all — firmware 0700_8012,
+        # "Failed to get AMS mapping table".
+        #
+        # `nozzles` is always a two-entry list (the state seeds it with two empty
+        # NozzleInfo stubs), so its length proves nothing; only a populated
+        # diameter on the second entry means real hardware. The other two signals
+        # are fallbacks for firmware revisions that surface one but not the
+        # other: a populated `ams_extruder_map` is dual-nozzle by construction,
+        # and so is more than one `vt_tray` entry, since single-nozzle printers
+        # expose exactly one external feed.
+        nozzles = getattr(status, "nozzles", None) or []
+        vt_trays = status.raw_data.get("vt_tray") or []
+        is_dual_nozzle = bool(
+            (len(nozzles) > 1 and getattr(nozzles[1], "nozzle_diameter", ""))
+            or ams_extruder_map
+            # isinstance, because a dict here would count its ~30 keys as trays.
+            # bambu_mqtt normalises vt_tray to a list before it reaches raw_data,
+            # so this is unreachable — but the loop below would raise on a dict
+            # and that is the pre-existing behaviour to keep, not to paper over.
+            or (isinstance(vt_trays, list) and len(vt_trays) > 1)
+        )
+
         # Parse AMS units from raw_data
         ams_data = status.raw_data.get("ams", [])
         for ams_unit in ams_data:
@@ -1900,7 +2095,7 @@ class PrintScheduler:
                     )
 
         # Check external spool(s) (vt_tray is a list)
-        for idx, vt in enumerate(status.raw_data.get("vt_tray") or []):
+        for idx, vt in enumerate(vt_trays):
             if vt.get("tray_type"):
                 color = self._normalize_color(vt.get("tray_color", ""))
                 tray_id = int(vt.get("id", 254))
@@ -1914,7 +2109,9 @@ class PrintScheduler:
                         "is_ht": False,
                         "is_external": True,
                         "global_tray_id": tray_id,
-                        "extruder_id": (255 - tray_id) if ams_extruder_map else None,
+                        # 254 = VIRTUAL_TRAY_DEPUTY_ID feeds extruder 1 (left),
+                        # 255 = VIRTUAL_TRAY_MAIN_ID feeds extruder 0 (right).
+                        "extruder_id": (255 - tray_id) if is_dual_nozzle else None,
                         "remain": vt.get("remain", -1),
                     }
                 )

backend/tests/unit/test_scheduler_cross_model_variants.py

@@ -299,7 +299,10 @@ async def _run_check_queue(ctx, scheduler, finder, waiting_notification=None):
         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()),
+        # None is the mapping-resolved answer; a bare AsyncMock returns a truthy
+        # sentinel, which the unmappable guard (#2771) reads as "this job can
+        # never print" and fails the item on.
+        patch.object(scheduler, "_ensure_ams_mapping", AsyncMock(return_value=None)),
         patch.object(scheduler, "_block_on_filament_deficit", AsyncMock(return_value=False)),
         patch.object(scheduler, "_launch_uploads", MagicMock()),
     ]

backend/tests/unit/test_scheduler_external_spool_nozzle_2771.py

@@ -0,0 +1,259 @@
+"""Regression tests for external-spool nozzle routing on AMS-less printers (#2771).
+
+A fleet of X2Ds with no AMS, printing from external spools, could not be sent a
+job with "Any X2D": the print uploaded, then the firmware rejected it with
+0700_8012 "Failed to get AMS mapping table" and the item failed after three
+dispatch attempts. Sending the same file to a named printer worked, because that
+path carries a mapping the *frontend* resolved and the scheduler's matcher never
+runs.
+
+Cause: ``_build_loaded_filaments`` derived dual-nozzle status from
+``ams_extruder_map``, which is built from AMS info bits — a dual-nozzle printer
+with zero AMS units reports an empty map. Every external spool then got
+``extruder_id=None``, and the nozzle-aware hard filter in
+``_match_filaments_to_slots`` rejected it because ``None`` equals neither 0 nor
+1. Nothing matched, the mapping came back all -1 and was cleared to None, and the
+print command went out as ``use_ams: true`` with no mapping table at all.
+
+This is the backend half of #1257, which fixed the identical logic in
+``useFilamentMapping.ts`` and left this copy behind; the first two tests below
+mirror its frontend regression tests.
+
+The second half covers the guard that keeps a genuinely unmappable job from
+being uploaded at all, since without an AMS there is no "load another spool and
+press Resume" recovery for the firmware error to lead to.
+"""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.print_scheduler import (
+    PrintScheduler,
+    _unmatched_filament_message,
+)
+
+# Two external feeds, as an X2D/H2D reports them: 254 is Ext-L (deputy/left,
+# extruder 1) and 255 is Ext-R (main/right, extruder 0).
+DUAL_EXTERNAL = [
+    {"id": "254", "tray_type": "PETG", "tray_color": "000000FF", "tray_info_idx": "GFG00"},
+    {"id": "255", "tray_type": "PLA", "tray_color": "FFFFFFFF", "tray_info_idx": "GFA00"},
+]
+
+REAL_NOZZLES = [
+    SimpleNamespace(nozzle_diameter="0.4"),
+    SimpleNamespace(nozzle_diameter="0.4"),
+]
+# The state seeds `nozzles` with two empty NozzleInfo stubs even on single-nozzle
+# printers, so the second entry's presence proves nothing — only a diameter does.
+STUB_NOZZLES = [
+    SimpleNamespace(nozzle_diameter="0.4"),
+    SimpleNamespace(nozzle_diameter=""),
+]
+
+
+def _status(raw_data, nozzles=None):
+    return SimpleNamespace(raw_data=raw_data, nozzles=nozzles)
+
+
+@pytest.fixture
+def scheduler():
+    return PrintScheduler()
+
+
+class TestExternalSpoolExtruderRouting:
+    """``_build_loaded_filaments`` must route external spools without an AMS."""
+
+    def test_dual_nozzle_without_ams_routes_both_external_feeds(self, scheduler):
+        """The X2D case from the report: no AMS, so ams_extruder_map is empty."""
+        loaded = scheduler._build_loaded_filaments(
+            _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
+        )
+
+        by_tray = {f["global_tray_id"]: f for f in loaded}
+        assert by_tray[254]["extruder_id"] == 1  # Ext-L -> left
+        assert by_tray[255]["extruder_id"] == 0  # Ext-R -> right
+
+    def test_single_nozzle_stub_does_not_fabricate_an_extruder(self, scheduler):
+        """A P1S/A1/X1C must keep extruder_id=None, matching pre-fix behaviour.
+
+        Sibling regression to the fix: `nozzles` always has two entries, so
+        inferring dual-nozzle from its length would hand every single-nozzle
+        printer's external spool a nozzle it does not have.
+        """
+        loaded = scheduler._build_loaded_filaments(
+            _status(
+                {"ams": [], "ams_extruder_map": {}, "vt_tray": [DUAL_EXTERNAL[0]]},
+                STUB_NOZZLES,
+            )
+        )
+
+        assert len(loaded) == 1
+        assert loaded[0]["extruder_id"] is None
+
+    def test_two_external_feeds_alone_imply_dual_nozzle(self, scheduler):
+        """Fallback signal: only dual-nozzle hardware exposes two external feeds.
+
+        Kept for firmware revisions that report the feeds but not the nozzle
+        diameters — here `nozzles` is absent entirely.
+        """
+        loaded = scheduler._build_loaded_filaments(
+            _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL})
+        )
+
+        assert {f["extruder_id"] for f in loaded} == {0, 1}
+
+    def test_populated_ams_extruder_map_still_implies_dual_nozzle(self, scheduler):
+        """The original signal keeps working when there IS an AMS."""
+        loaded = scheduler._build_loaded_filaments(
+            _status(
+                {"ams": [], "ams_extruder_map": {"0": 1}, "vt_tray": [DUAL_EXTERNAL[0]]},
+                STUB_NOZZLES,
+            )
+        )
+
+        assert loaded[0]["extruder_id"] == 1
+
+    def test_mapping_resolves_for_the_nozzle_the_spool_feeds(self, scheduler):
+        """End to end: the matcher now finds the external spool, as it did for
+        the working named-printer dispatch (which sent ams_mapping [254])."""
+        loaded = scheduler._build_loaded_filaments(
+            _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
+        )
+        req = {"slot_id": 1, "type": "PETG", "color": "#000000", "tray_info_idx": "GFG00"}
+
+        assert scheduler._match_filaments_to_slots([{**req, "nozzle_id": 1}], loaded) == [254]
+        # Nothing PETG on the right nozzle — still correctly unmatched.
+        assert scheduler._match_filaments_to_slots([{**req, "nozzle_id": 0}], loaded) == [-1]
+
+
+class TestUnmatchedFilamentMessage:
+    """The message has to name the filament and, on dual-nozzle, the nozzle."""
+
+    def test_names_type_colour_and_nozzle(self):
+        message = _unmatched_filament_message(
+            [{"slot_id": 1, "type": "PETG", "color": "#000000", "nozzle_id": 0}],
+            [{"type": "PETG", "color": "#000000", "extruder_id": 1}],
+        )
+
+        assert "PETG #000000 (right nozzle)" in message
+        assert "PETG #000000 (left nozzle)" in message
+
+    def test_omits_nozzle_on_single_nozzle_printers(self):
+        message = _unmatched_filament_message(
+            [{"slot_id": 1, "type": "ABS", "color": "#FF0000"}],
+            [{"type": "PLA", "color": "#000000"}],
+        )
+
+        assert "ABS #FF0000" in message
+        assert "nozzle" not in message
+
+
+class TestUnmappableWithoutAmsGuard:
+    """``_ensure_ams_mapping`` reports only a positive, unrecoverable finding."""
+
+    def _item(self, ams_mapping=None):
+        item = MagicMock()
+        item.id = 22
+        item.printer_id = 4
+        item.ams_mapping = ams_mapping
+        item.filament_overrides = None
+        return item
+
+    async def _ensure(self, scheduler, computed, status):
+        scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=computed)
+        scheduler._get_filament_requirements = AsyncMock(
+            return_value=[{"slot_id": 1, "type": "PETG", "color": "#000000", "nozzle_id": 0}]
+        )
+        with patch("backend.app.services.print_scheduler.printer_manager") as pm:
+            pm.get_status.return_value = status
+            return await scheduler._ensure_ams_mapping(AsyncMock(), 4, self._item())
+
+    @pytest.mark.asyncio
+    async def test_reports_when_nothing_matches_and_there_is_no_ams(self, scheduler):
+        status = _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
+
+        message = await self._ensure(scheduler, [-1], status)
+
+        assert message is not None
+        assert "no AMS" in message
+
+    @pytest.mark.asyncio
+    async def test_silent_when_an_ams_is_attached(self, scheduler):
+        """With an AMS the user can load a spool and press Resume, so the
+        firmware's own error is worth reaching — behaviour is unchanged."""
+        status = _status(
+            {
+                "ams": [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}]}],
+                "ams_extruder_map": {},
+                "vt_tray": DUAL_EXTERNAL,
+            },
+            REAL_NOZZLES,
+        )
+
+        assert await self._ensure(scheduler, [-1], status) is None
+
+    @pytest.mark.asyncio
+    async def test_silent_when_the_ams_field_has_not_arrived_yet(self, scheduler):
+        """Absence of an AMS report is not a report of no AMS.
+
+        `raw_data["ams"]` appears only once an AMS push has been handled, so a
+        missing key means a reconnect or a cold start — where a fully loaded
+        AMS is briefly invisible and everything would look unmappable.
+        """
+        status = _status({"ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
+
+        assert await self._ensure(scheduler, [-1], status) is None
+
+    @pytest.mark.asyncio
+    async def test_silent_when_the_matcher_never_ran(self, scheduler):
+        """A None mapping means no requirements parsed or nothing loaded — not
+        evidence of a mismatch. Fail-safe: dispatch as before."""
+        status = _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
+
+        assert await self._ensure(scheduler, None, status) is None
+
+    @pytest.mark.asyncio
+    async def test_silent_when_the_mapping_resolves(self, scheduler):
+        status = _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
+        item = self._item()
+        scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[254])
+
+        with patch("backend.app.services.print_scheduler.printer_manager") as pm:
+            pm.get_status.return_value = status
+            assert await scheduler._ensure_ams_mapping(AsyncMock(), 4, item) is None
+
+        assert json.loads(item.ams_mapping) == [254]
+
+    @pytest.mark.asyncio
+    async def test_silent_when_the_printer_status_is_gone(self, scheduler):
+        assert await self._ensure(scheduler, [-1], None) is None
+
+
+class TestFailUnmappableItem:
+    """The guard fails the item instead of spending an upload on it."""
+
+    @pytest.mark.asyncio
+    async def test_marks_failed_with_the_message(self, scheduler):
+        db = AsyncMock()
+        item = MagicMock()
+        item.id = 22
+        item.created_by_id = 1
+
+        with (
+            patch("backend.app.services.print_scheduler.notification_service") as notify,
+            patch("backend.app.services.print_scheduler.ws_manager"),
+        ):
+            notify.on_queue_job_failed = AsyncMock()
+            scheduler._get_job_name = AsyncMock(return_value="Fidget")
+            scheduler._get_printer = AsyncMock(return_value=SimpleNamespace(name="X2D-1"))
+
+            await scheduler._fail_unmappable_item(db, item, 4, "needs PETG")
+
+        assert item.status == "failed"
+        assert item.error_message == "needs PETG"
+        assert item.completed_at is not None
+        db.commit.assert_awaited()
+        notify.on_queue_job_failed.assert_awaited_once()