fix(queue): print a plate whose filaments are all on the external spool (issue #3087)

    The reporter's P1S heated up, sat at Heatbed preheating for ten and a half
    minutes, then paused with 07FF_8012, "Failed to get AMS mapping table".
    Resuming only reheated it. Prints that fed from the AMS were fine.

    The plate was one filament of a seven-filament MakerWorld project, mapped by
    hand to the external spool. slice_info.config numbers filaments across the
    whole project, so the mapping for that plate is [-1,-1,-1,-1,-1,-1,254]: six
    placeholders and the spool holder. The command builder decides whether a print
    needs the AMS by asking whether the mapping is entirely external, and six -1s
    answer no. So the print went out as use_ams=true carrying a flat mapping of
    nothing but -1 -- 254 is deliberately never sent raw, the firmware reads it as
    AMS tray 0 -- which is exactly the mapping table the firmware then could not
    find.

    The builder cannot fix this itself. Down there a -1 is either padding for a
    filament this plate does not print, which is BambuStudio's own convention, or a
    slot that never resolved to a tray, and sending the second one to the spool
    holder is what #2589 exists to prevent. They are the same byte.

    The scheduler knows. extract_filament_requirements drops every filament with
    used_g <= 0, so it names precisely the slots the plate prints. When all of those
    are an explicit 254/255, dispatch now sends use_ams=false and the print runs.
    When one of them resolved to nothing, the flag is left alone and the firmware
    rejects the print as it does today -- deliberately, because that is the case
    where guessing would print a filament in the wrong material without saying so.

    Single-nozzle only, mirroring the reconcile in the command builder: on a
    two-extruder printer use_ams selects which nozzle to feed rather than whether to
    use the AMS, so an H2D with a spool on each side must keep the flag it was
    given. Judged generously from the model name and from live telemetry -- a second
    nozzle reporting a diameter, an extruder map, or more than one external feed --
    because a wrong yes only preserves existing behaviour while a wrong no would
    reroute the print. H2S stays single-nozzle (#1386).

    Nothing in the command builder changed. Its own reconcile keeps the exact
    semantics #2589, #2595 and #797 gave it, and now usually agrees with a decision
    that was already made one layer up. Where the file has no parseable filament
    list the mapping is left exactly as before, the same evidence-only convention
    as #2771, and the parse itself sits behind a check for anything external at all
    so an AMS-only print never opens the file.

    Covered end to end at the dispatcher, including the reporter's seven-filament
    shape, a plate mixing the spool holder with an AMS tray, a consumed slot that
    never resolved, both external feeds on a dual-nozzle machine, and a 3MF with no
    filament list at all.

backend/app/services/print_scheduler.py

@@ -64,6 +64,7 @@ from backend.app.utils.filament_types import canonical_filament_type
 from backend.app.utils.filename import derive_remote_filename
 from backend.app.utils.local_time import utcnow_naive
 from backend.app.utils.printer_models import (
+    is_dual_nozzle_model,
     is_gcode_compatible,
     is_nozzle_rack_model,
     normalize_printer_model,
@@ -472,6 +473,77 @@ def _mapping_is_all_unresolved(mapping: list | None) -> bool:
 _EXTERNAL_TRAY_ID_MIN = 254
 
 
+def _consumed_mapping_entries(mapping: list | None, required: list[dict] | None) -> list | None:
+    """The ``mapping`` entries for the slots this plate actually prints.
+
+    ``required`` comes from ``extract_filament_requirements``, which drops any
+    filament with ``used_g <= 0`` — so a slot_id present there is one the plate
+    consumes, and one absent from it is padding. That distinction is why this
+    decision lives here and not in the MQTT command builder: a ``-1`` in the
+    mapping means either "this plate does not print filament N" or "we never
+    worked out which tray", and only the plate's own filament list separates
+    them. The builder sees both as the same byte, which is how a plate whose one
+    printed filament sat on the external spool went out as `use_ams=true` with a
+    mapping of nothing but -1 and stalled at preheat until the firmware gave up
+    with 07FF_8012 (#3087).
+
+    Returns None whenever the two cannot be lined up — no mapping, no parsed
+    requirements, or a requirement the mapping is too short to cover — so every
+    caller falls back to existing behaviour rather than acting on a guess.
+    """
+    if not isinstance(mapping, list) or not mapping or not required:
+        return None
+    entries = []
+    for filament in required:
+        slot_id = filament.get("slot_id")
+        if not isinstance(slot_id, int) or not 1 <= slot_id <= len(mapping):
+            # The mapping and the requirements disagree about how many filaments
+            # the file has. They came from different reads, so judge nothing.
+            return None
+        entries.append(mapping[slot_id - 1])
+    return entries or None
+
+
+def _is_external_tray(tray_id) -> bool:
+    """True for an explicit external-spool selection (254/255), not for an
+    unresolved slot and not for an AMS tray."""
+    if tray_id is None:
+        return False
+    try:
+        return int(tray_id) >= _EXTERNAL_TRAY_ID_MIN
+    except (TypeError, ValueError):
+        return False
+
+
+def _might_be_dual_nozzle(printer_model: str | None, status) -> bool:
+    """Whether this printer could have two extruders, judged generously.
+
+    On a dual-nozzle printer ``use_ams`` is nozzle routing rather than an
+    AMS on/off flag — H2D Pro firmware reads it as an extruder index — which is
+    why the MQTT command builder skips its own use_ams reconcile there. Anything
+    that might be dual-nozzle therefore keeps whatever ``use_ams`` it arrived
+    with, external spools or not.
+
+    Deliberately over-eager: a wrong "yes" only means this printer keeps the
+    behaviour it has always had, while a wrong "no" would rewrite a field that
+    steers which nozzle prints. The model name is the first answer (it is what
+    the command builder falls back to as well), then the same live evidence the
+    dispatcher's extruder annotation uses — a second nozzle reporting a
+    diameter, a populated ``ams_extruder_map``, or more than one external feed,
+    since a single-nozzle printer has exactly one.
+    """
+    if is_dual_nozzle_model(printer_model):
+        return True
+    nozzles = getattr(status, "nozzles", None) or []
+    if len(nozzles) > 1 and getattr(nozzles[1], "nozzle_diameter", ""):
+        return True
+    raw = getattr(status, "raw_data", None) or {}
+    if raw.get("ams_extruder_map"):
+        return True
+    vt_trays = raw.get("vt_tray") or []
+    return isinstance(vt_trays, list) and len(vt_trays) > 1
+
+
 def _int_or(value, default: int) -> int:
     """``int(value)``, or ``default`` when the field is missing or junk.
 
@@ -6808,6 +6880,54 @@ class PrintScheduler:
             if slot_extruders:
                 nozzle_slot_extruders = json.dumps(slot_extruders)
 
+        # Every filament this plate prints is on the external spool -> the print
+        # must go out with use_ams=False. The firmware answers use_ams=true plus
+        # a mapping it cannot resolve with 07FF_8012 "Failed to get AMS mapping
+        # table", which is what held the reporter's P1S at Heatbed preheating
+        # for ten minutes before it gave up (#3087). The MQTT command builder
+        # already downgrades a mapping that is *only* external ([254]), but a
+        # multi-filament project pads the slots this plate does not print with
+        # -1 — BambuStudio's own convention — and down there a -1 is
+        # indistinguishable from a slot that never resolved, which must never be
+        # sent to the spool holder (#2589). Here the plate's filament list says
+        # which is which, so the answer is exact rather than a guess.
+        #
+        # Deliberately narrow: this fires only when every consumed slot is an
+        # explicit 254/255. A consumed slot that did not resolve leaves use_ams
+        # alone and the firmware still rejects the print, exactly as today. And
+        # only for single-nozzle printers, mirroring the builder's own reconcile
+        # — on a dual-nozzle machine use_ams is which extruder to feed, not
+        # whether to use the AMS, so it is not ours to rewrite.
+        effective_use_ams = item.use_ams
+        if (
+            effective_use_ams
+            and ams_mapping
+            and file_path is not None
+            # Cheap gate before opening the file: with nothing on the spool
+            # holder anywhere in the mapping, no subset of it can be all
+            # external, so most dispatches never pay for the parse. The
+            # isinstance also keeps a malformed stored mapping (a bare number
+            # from a hand-edited row) failing where it always failed, in the
+            # command builder, rather than here.
+            and isinstance(ams_mapping, list)
+            and any(_is_external_tray(t) for t in ams_mapping)
+            and not _might_be_dual_nozzle(printer.model, pre_status)
+        ):
+            from backend.app.services.filament_requirements import extract_filament_requirements
+
+            consumed = _consumed_mapping_entries(
+                ams_mapping, extract_filament_requirements(file_path, plate_id=item.plate_id or 1)
+            )
+            if consumed and all(_is_external_tray(t) for t in consumed):
+                effective_use_ams = False
+                logger.info(
+                    "Queue item %s: every filament plate %s prints is on the external spool "
+                    "(mapping %s) — dispatching with use_ams=False (#3087)",
+                    item.id,
+                    item.plate_id or 1,
+                    ams_mapping,
+                )
+
         # Start the print with AMS mapping, plate_id and print options.
         # nozzle_mapping rides through verbatim — JSON string captured from
         # Bambu Studio's project_file on VP intake (#1780); the MQTT layer
@@ -6824,7 +6944,7 @@ class PrintScheduler:
             vibration_cali=item.vibration_cali,
             layer_inspect=item.layer_inspect,
             timelapse=effective_timelapse,
-            use_ams=item.use_ams,
+            use_ams=effective_use_ams,
             nozzle_offset_cali=item.nozzle_offset_cali,
             nozzle_mapping=item.nozzle_mapping
             or (json.dumps(resolved_nozzle_mapping) if resolved_nozzle_mapping else None),

backend/tests/integration/test_external_spool_use_ams_3087.py

@@ -0,0 +1,303 @@
+"""A plate printed entirely from the external spool must dispatch use_ams=False (#3087).
+
+The reporter's P1S sat at "Heatbed preheating" for ten and a half minutes and
+then paused with 07FF_8012, "Failed to get AMS mapping table". The plate was one
+filament, mapped by hand to the external spool, out of a seven-filament
+MakerWorld project -- so the mapping was ``[-1, -1, -1, -1, -1, -1, 254]`` and
+the command went out as ``use_ams: true`` with a flat mapping of nothing but
+-1 (254 is deliberately not sent raw: the firmware reads it as AMS tray 0).
+
+The decision belongs here rather than in the MQTT command builder. Down there a
+-1 is either padding for a filament this plate does not print -- BambuStudio's
+own convention, and what the other six entries are -- or a slot that never
+resolved, which must never be redirected to the spool holder (#2589). The two
+are the same byte. Only the plate's own filament list tells them apart, and
+``extract_filament_requirements`` already drops anything with ``used_g <= 0``,
+so it names exactly the slots that are printed.
+"""
+
+from __future__ import annotations
+
+import json
+import zipfile
+from contextlib import ExitStack
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import backend.app.models  # noqa: F401 - populate Base.metadata
+import backend.app.services.print_scheduler as scheduler_module
+from backend.app.core.database import Base
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings  # noqa: F401 - registers the table
+from backend.app.services.print_scheduler import PrintScheduler
+from backend.tests._fixtures.background_tasks import discarding_spawn_patch
+
+pytestmark = pytest.mark.integration
+
+# The reporter's plate: filament 7 of a seven-filament project, and it is the
+# only one this plate consumes. slice_info.config lists a plate's filaments by
+# their project-wide id, which is why the mapping is seven long.
+_PLATE_4_ONE_FILAMENT = '<filament id="7" used_g="12.4" type="PLA" color="#F98C36"/>'
+
+
+def _write_3mf(path: Path, plate_index: int = 4, filaments: str = _PLATE_4_ONE_FILAMENT) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr(
+            "Metadata/slice_info.config",
+            f'<config><plate><metadata key="index" value="{plate_index}"/>{filaments}</plate></config>',
+        )
+
+
+def _write_3mf_without_slice_info(path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(path, "w") as zf:
+        zf.writestr("3D/3dmodel.model", "<model/>")
+
+
+@pytest.fixture
+async def dispatch_case(tmp_path):
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+    base_dir = tmp_path / "external-spool"
+
+    async def _build(
+        mapping, *, use_ams=True, plate_id=4, filaments=_PLATE_4_ONE_FILAMENT, slice_info=True, model="P1S"
+    ):
+        archive_rel = Path("archives") / f"plate-{plate_id}-{abs(hash(str(mapping))) % 10**6}.gcode.3mf"
+        if slice_info:
+            _write_3mf(base_dir / archive_rel, plate_index=plate_id, filaments=filaments)
+        else:
+            _write_3mf_without_slice_info(base_dir / archive_rel)
+
+        async with session_maker() as db:
+            printer = Printer(
+                name="P1S",
+                serial_number=f"01P{abs(hash(str(mapping))) % 10**9}",
+                ip_address="127.0.0.1",
+                access_code="access-code",
+                model=model,
+            )
+            db.add(printer)
+            await db.flush()
+            archive = PrintArchive(
+                printer_id=printer.id,
+                filename=archive_rel.name,
+                file_path=str(archive_rel),
+                file_size=(base_dir / archive_rel).stat().st_size,
+                status="completed",
+            )
+            db.add(archive)
+            await db.flush()
+            item = PrintQueueItem(
+                printer_id=printer.id,
+                archive_id=archive.id,
+                plate_id=plate_id,
+                status="pending",
+                use_ams=use_ams,
+                ams_mapping=json.dumps(mapping) if mapping is not None else None,
+            )
+            db.add(item)
+            await db.commit()
+            return SimpleNamespace(item_id=item.id, printer_id=printer.id)
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker, base_dir=base_dir, build=_build)
+    finally:
+        await engine.dispose()
+
+
+async def _dispatch(ctx, ids, status=None):
+    scheduler = PrintScheduler()
+    start_print = MagicMock(return_value=True)
+    status = status or SimpleNamespace(state="IDLE", nozzle_rack=None, raw_data={}, nozzles=[])
+
+    with ExitStack() as stack:
+        for patcher in (
+            patch.object(scheduler_module, "async_session", ctx.session_maker),
+            patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
+            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=status)),
+            patch("backend.app.services.print_scheduler.printer_manager.start_print", start_print),
+            patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
+            patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True)),
+            patch("backend.app.services.print_scheduler.upload_file_async", AsyncMock(return_value=True)),
+            patch(
+                "backend.app.services.print_scheduler.get_ftp_retry_settings",
+                AsyncMock(return_value=(False, 3, 2.0, 30.0)),
+            ),
+            patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
+            discarding_spawn_patch(),
+            patch("backend.app.services.notification_service.notification_service.on_queue_job_started", AsyncMock()),
+            patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", AsyncMock()),
+            patch("backend.app.services.mqtt_relay.mqtt_relay.on_queue_job_started", AsyncMock()),
+            patch.object(scheduler, "_propagate_owner_to_printer_manager", AsyncMock()),
+            patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
+            patch.object(scheduler, "_preheat_and_soak", AsyncMock()),
+        ):
+            stack.enter_context(patcher)
+        await scheduler._dispatch_one(ids.item_id)
+
+    assert start_print.call_count == 1, "the print command was never sent"
+    return start_print.call_args
+
+
+class TestThePlateThatOnlyPrintsFromTheSpoolHolder:
+    async def test_the_reporters_mapping_dispatches_without_the_ams(self, dispatch_case):
+        """[-1]*6 + [254] on a plate whose only printed filament is #7."""
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254])
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is False
+        # The mapping itself still goes out untouched — the builder is what
+        # turns 254 into -1 plus ams_mapping2, and none of that changes.
+        assert call.kwargs["ams_mapping"] == [-1, -1, -1, -1, -1, -1, 254]
+
+    async def test_the_main_nozzle_sentinel_counts_too(self, dispatch_case):
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 255])
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is False
+
+    async def test_an_unpadded_single_filament_plate_is_unaffected(self, dispatch_case):
+        """[254] already worked: the MQTT command builder downgrades an
+        all-external mapping by itself. The scheduler now reaches the same
+        answer one layer earlier, so the two agree rather than one undoing the
+        other — this pins that they do."""
+        ids = await dispatch_case.build([254], filaments='<filament id="1" used_g="9.0" type="PLA"/>', plate_id=1)
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is False
+
+
+class TestWhatMustNotChange:
+    async def test_a_consumed_slot_that_never_resolved_still_goes_out_with_the_ams(self, dispatch_case):
+        """The #2589 contract, and the reason this lives in the scheduler.
+
+        Filaments 1 and 7 are both printed; 7 is on the spool holder and 1
+        resolved to nothing. Redirecting the plate to the external spool would
+        print filament 1 in the wrong material without saying so. use_ams stays
+        true and the firmware rejects the print, exactly as before.
+        """
+        ids = await dispatch_case.build(
+            [-1, -1, -1, -1, -1, -1, 254],
+            filaments='<filament id="1" used_g="8.0" type="PETG"/>' + _PLATE_4_ONE_FILAMENT,
+        )
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_a_plate_mixing_an_ams_tray_with_the_spool_holder_keeps_the_ams(self, dispatch_case):
+        ids = await dispatch_case.build(
+            [5, -1, -1, -1, -1, -1, 254],
+            filaments='<filament id="1" used_g="8.0" type="PETG"/>' + _PLATE_4_ONE_FILAMENT,
+        )
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_a_plate_printed_from_ams_trays_is_untouched(self, dispatch_case):
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 5])
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_use_ams_false_is_never_promoted_here(self, dispatch_case):
+        """Promotion is the builder's job (#2595) and stays there."""
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 5], use_ams=False)
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is False
+
+    async def test_a_3mf_with_no_filament_list_falls_back_to_the_stored_flag(self, dispatch_case):
+        """No evidence, no decision — the same convention as #2771."""
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254], slice_info=False)
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_a_plate_the_file_does_not_describe_falls_back(self, dispatch_case):
+        """The item says plate 4; the file only describes plate 1."""
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254], plate_id=4)
+        # Rewrite the archive's 3MF so its only plate is index 1.
+        async with dispatch_case.session_maker() as db:
+            archive = (await db.get(PrintQueueItem, ids.item_id)).archive_id
+            path = dispatch_case.base_dir / (await db.get(PrintArchive, archive)).file_path
+        _write_3mf(path, plate_index=1)
+
+        call = await _dispatch(dispatch_case, ids)
+        assert call.kwargs["use_ams"] is True
+
+    async def test_an_item_with_no_mapping_at_all_is_untouched(self, dispatch_case):
+        ids = await dispatch_case.build(None)
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+
+class TestDualNozzleIsNotOursToRewrite:
+    """On a two-extruder printer use_ams is which nozzle to feed, not whether to
+    use the AMS — H2D Pro firmware reads it as an extruder index. The MQTT
+    command builder skips its own reconcile for exactly that reason, and this
+    must skip it too, or a perfectly normal dual external-spool print gets its
+    routing rewritten."""
+
+    async def test_a_dual_nozzle_model_keeps_its_flag(self, dispatch_case):
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254], model="H2D")
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_both_external_feeds_on_a_dual_nozzle_are_left_alone(self, dispatch_case):
+        """254 is the deputy feed and 255 the main one — an ordinary H2D print
+        with a spool on each side, and the one this would have broken."""
+        ids = await dispatch_case.build(
+            [254, -1, -1, -1, -1, -1, 255],
+            filaments='<filament id="1" used_g="8.0" type="PLA"/>' + _PLATE_4_ONE_FILAMENT,
+            model="H2D",
+        )
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_live_telemetry_can_veto_a_single_nozzle_model_name(self, dispatch_case):
+        """A model string we do not recognise as dual is not the last word: two
+        external feeds is something only a two-extruder printer reports."""
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254], model="Something New")
+        status = SimpleNamespace(
+            state="IDLE",
+            nozzle_rack=None,
+            nozzles=[],
+            raw_data={"vt_tray": [{"id": "254"}, {"id": "255"}]},
+        )
+        call = await _dispatch(dispatch_case, ids, status=status)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_a_second_nozzle_reporting_a_diameter_vetoes_it_too(self, dispatch_case):
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254], model="Something New")
+        status = SimpleNamespace(
+            state="IDLE",
+            nozzle_rack=None,
+            nozzles=[SimpleNamespace(nozzle_diameter="0.4"), SimpleNamespace(nozzle_diameter="0.4")],
+            raw_data={},
+        )
+        call = await _dispatch(dispatch_case, ids, status=status)
+
+        assert call.kwargs["use_ams"] is True
+
+    async def test_h2s_is_single_nozzle_and_still_gets_the_fix(self, dispatch_case):
+        """H2S shares the H2 serial prefix and firmware quirks but has one
+        extruder — the #1386 distinction, which must survive here."""
+        ids = await dispatch_case.build([-1, -1, -1, -1, -1, -1, 254], model="H2S")
+        call = await _dispatch(dispatch_case, ids)
+
+        assert call.kwargs["use_ams"] is False

backend/tests/unit/test_external_spool_mapping_3087.py

@@ -0,0 +1,156 @@
+"""Which mapping entries a plate actually prints, and what the builder does
+with the answer (#3087).
+
+The dispatch-level behaviour is covered in
+``backend/tests/integration/test_external_spool_use_ams_3087.py``. This is the
+pure part: separating a padding ``-1`` from a slot that never resolved, which
+is the distinction the MQTT command builder cannot make and the reason the
+decision was put in the scheduler.
+"""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from backend.app.services.bambu_mqtt import BambuMQTTClient
+from backend.app.services.print_scheduler import (
+    _consumed_mapping_entries,
+    _is_external_tray,
+    _might_be_dual_nozzle,
+)
+
+
+def _required(*slot_ids):
+    """What `extract_filament_requirements` returns: one entry per filament the
+    plate consumes, keyed by its project-wide slot_id. Anything with
+    `used_g <= 0` is already dropped there, so everything here is printed."""
+    return [{"slot_id": s, "type": "PLA", "used_grams": 10.0} for s in slot_ids]
+
+
+class TestConsumedMappingEntries:
+    def test_it_picks_out_the_slot_the_plate_prints(self):
+        # The reporter's plate: seven project filaments, only #7 printed.
+        assert _consumed_mapping_entries([-1, -1, -1, -1, -1, -1, 254], _required(7)) == [254]
+
+    def test_padding_is_not_reported_as_unresolved(self):
+        assert _consumed_mapping_entries([-1, 5, -1], _required(2)) == [5]
+
+    def test_a_slot_the_plate_prints_reports_its_own_unresolved_entry(self):
+        assert _consumed_mapping_entries([-1, -1, -1, -1, -1, -1, 254], _required(1, 7)) == [-1, 254]
+
+    def test_several_printed_slots_come_back_in_slot_order(self):
+        assert _consumed_mapping_entries([4, -1, 254], _required(1, 3)) == [4, 254]
+
+    @pytest.mark.parametrize(
+        "mapping,required",
+        [
+            (None, _required(1)),
+            ([], _required(1)),
+            ([254], None),
+            ([254], []),
+        ],
+    )
+    def test_it_declines_to_answer_without_both_halves(self, mapping, required):
+        # No evidence, no decision — the caller then leaves use_ams alone.
+        assert _consumed_mapping_entries(mapping, required) is None
+
+    def test_a_requirement_the_mapping_is_too_short_for_declines(self):
+        # The two were read at different times and disagree about the file.
+        assert _consumed_mapping_entries([254], _required(7)) is None
+
+    def test_a_junk_slot_id_declines(self):
+        assert _consumed_mapping_entries([254], [{"slot_id": "7"}]) is None
+        assert _consumed_mapping_entries([254], [{"slot_id": 0}]) is None
+        assert _consumed_mapping_entries([254], [{}]) is None
+
+
+class TestIsExternalTray:
+    @pytest.mark.parametrize("tray_id", [254, 255, "254"])
+    def test_the_spool_holder(self, tray_id):
+        assert _is_external_tray(tray_id) is True
+
+    @pytest.mark.parametrize("tray_id", [None, -1, 0, 5, 253, 128, "", "x", 1.5])
+    def test_everything_else(self, tray_id):
+        # 128-253 are AMS-HT units, -1/None unresolved, and junk is not a
+        # licence to redirect a print to the spool holder.
+        assert _is_external_tray(tray_id) is False
+
+
+class TestMightBeDualNozzle:
+    """Over-eager on purpose: a wrong yes only leaves a printer with the
+    behaviour it already had, a wrong no rewrites which nozzle prints."""
+
+    def _status(self, *, nozzles=(), **raw):
+        return SimpleNamespace(nozzles=list(nozzles), raw_data=dict(raw))
+
+    @pytest.mark.parametrize("model", ["H2D", "H2C", "X2D", "H2D Pro"])
+    def test_the_model_name_is_enough(self, model):
+        assert _might_be_dual_nozzle(model, self._status()) is True
+
+    @pytest.mark.parametrize("model", ["P1S", "X1C", "A1", "P2S", "H2S", None, ""])
+    def test_single_nozzle_models_pass(self, model):
+        # H2S is the #1386 case: H2 family, one extruder.
+        assert _might_be_dual_nozzle(model, self._status()) is False
+
+    def test_two_external_feeds_give_it_away(self):
+        # Only a two-extruder printer reports more than one vt_tray.
+        assert _might_be_dual_nozzle("P1S", self._status(vt_tray=[{"id": "254"}, {"id": "255"}])) is True
+
+    def test_one_external_feed_does_not(self):
+        assert _might_be_dual_nozzle("P1S", self._status(vt_tray=[{"id": "254"}])) is False
+
+    def test_a_second_nozzle_with_a_diameter_gives_it_away(self):
+        nozzles = [SimpleNamespace(nozzle_diameter="0.4"), SimpleNamespace(nozzle_diameter="0.4")]
+        assert _might_be_dual_nozzle("P1S", self._status(nozzles=nozzles)) is True
+
+    def test_a_stub_second_nozzle_does_not(self):
+        # The status model can carry placeholder NozzleInfo entries; only a
+        # populated diameter means real hardware.
+        nozzles = [SimpleNamespace(nozzle_diameter="0.4"), SimpleNamespace(nozzle_diameter="")]
+        assert _might_be_dual_nozzle("P1S", self._status(nozzles=nozzles)) is False
+
+    def test_an_extruder_map_gives_it_away(self):
+        assert _might_be_dual_nozzle("P1S", self._status(ams_extruder_map={"0": 1})) is True
+
+    def test_no_status_at_all_is_not_evidence_of_two(self):
+        assert _might_be_dual_nozzle("P1S", None) is False
+
+    def test_a_vt_tray_dict_is_not_counted_as_many_trays(self):
+        # bambu_mqtt normalises vt_tray to a list, but a dict here would
+        # otherwise count its keys and read as dual-nozzle.
+        assert _might_be_dual_nozzle("P1S", self._status(vt_tray={"id": "254", "tray_type": "PLA"})) is False
+
+
+class TestTheBuilderHonoursTheDecision:
+    """The scheduler's answer has to survive the command builder, which has its
+    own use_ams reconcile (#2589/#2595). It must not promote the flag back."""
+
+    @pytest.fixture
+    def mqtt_client(self):
+        client = BambuMQTTClient(ip_address="192.168.1.100", serial_number="01P00A452600691", access_code="x")
+        client.model = "P1S"
+        client._client = MagicMock()
+        client.state.connected = True
+        return client
+
+    def _sent(self, mqtt_client):
+        return json.loads(mqtt_client._client.publish.call_args.args[1])["print"]
+
+    def test_the_reporters_command_now_goes_out_printable(self, mqtt_client):
+        mqtt_client.start_print("plate_4.3mf", ams_mapping=[-1] * 6 + [254], use_ams=False)
+        cmd = self._sent(mqtt_client)
+
+        assert cmd["use_ams"] is False
+        # 254 is still never sent raw in the flat array — the firmware reads it
+        # as AMS tray 0 — and ams_mapping2 still carries the spool holder.
+        assert cmd["ams_mapping"] == [-1] * 7
+        assert cmd["ams_mapping2"][6] == {"ams_id": 255, "slot_id": 0}
+        assert cmd["ams_mapping2"][0] == {"ams_id": 255, "slot_id": 255}
+
+    def test_the_builder_still_rejects_an_unresolved_mapping_as_external(self, mqtt_client):
+        """The #2589 contract, unchanged: nothing here treats -1 as the spool."""
+        mqtt_client.start_print("plate_4.3mf", ams_mapping=[-1] * 6 + [254], use_ams=True)
+
+        assert self._sent(mqtt_client)["use_ams"] is True