diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py
index d6c9c03fe..c356f75ba 100644
--- a/backend/app/services/print_scheduler.py
+++ b/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),
diff --git a/backend/tests/integration/test_external_spool_use_ams_3087.py b/backend/tests/integration/test_external_spool_use_ams_3087.py
new file mode 100644
index 000000000..b89edec66
--- /dev/null
+++ b/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 = ''
+
+
+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'{filaments}',
+ )
+
+
+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", "")
+
+
+@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='', 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='' + _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='' + _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='' + _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
diff --git a/backend/tests/unit/test_external_spool_mapping_3087.py b/backend/tests/unit/test_external_spool_mapping_3087.py
new file mode 100644
index 000000000..ed91f0d93
--- /dev/null
+++ b/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