Keep the external spool in a virtual printer's saved AMS mapping (#3237)

The slicer writes an external spool as -1 in ams_mapping, the same as an
unused filament, and names it only in ams_mapping2. Save AMS mapping read
the flat list alone, so dispatch sent the filament as unmapped and an H2C
stopped with 0700-8012 before the first layer. Resolve it through
ams_mapping2 to the global tray (254/255, by the target's nozzle count),
as the usage capture does since #3166.

backend/app/services/virtual_printer/manager.py

@@ -20,6 +20,7 @@ from backend.app.models.virtual_printer import (
     VP_MODE_QUEUE,
     normalize_vp_mode,
 )
+from backend.app.services.bambu_mqtt import resolve_external_spools_in_mapping
 from backend.app.services.virtual_printer.bind_server import BindServer
 from backend.app.services.virtual_printer.certificate import CertificateService
 from backend.app.services.virtual_printer.ftp_server import VirtualPrinterFTPServer, compute_passive_port_slice
@@ -155,7 +156,7 @@ def _tristate_from_slicer(data: dict, bool_field: str, int_field: str) -> str |
     return None
 
 
-def _extract_slicer_ams_mapping_json(data: dict, log_prefix: str) -> str | None:
+def _extract_slicer_ams_mapping_json(data: dict, log_prefix: str, *, is_dual_nozzle: bool = False) -> str | None:
     """Pull the slicer's own AMS-slot pick out of a captured project_file payload.
 
     BambuStudio/OrcaSlicer resolves the physical AMS tray for each filament
@@ -184,6 +185,14 @@ def _extract_slicer_ams_mapping_json(data: dict, log_prefix: str) -> str | None:
     per-slot force-color overrides all live inside that function. Callers are
     responsible for the gating — this parser only says what the slicer sent.
 
+    An external spool is ``-1`` in the flat list, the same as a filament with
+    no tray, and is named only in ``ams_mapping2``. It is resolved from there
+    to its global tray (254/255), as the usage capture does (#3166); dispatch
+    turns that back into the external-spool entry. Kept as ``-1`` it dispatched
+    as unmapped and the printer stopped with 0700-8012 (#3237).
+    ``is_dual_nozzle`` is the target printer's, since that decides which
+    global tray an external spool is.
+
     Returns ``None`` when the field is absent, unparsable, or the classic
     "all -1" unresolved-race sentinel (#2589) — never worth trusting over a
     fresh live computation.
@@ -202,6 +211,13 @@ def _extract_slicer_ams_mapping_json(data: dict, log_prefix: str) -> str | None:
     # valid mapping.
     if not isinstance(raw, list) or not raw or not all(isinstance(v, int) and not isinstance(v, bool) for v in raw):
         return None
+    detail = data.get("ams_mapping2")
+    if isinstance(detail, str):
+        try:
+            detail = json.loads(detail)
+        except json.JSONDecodeError:
+            detail = None
+    raw = resolve_external_spools_in_mapping(raw, detail, is_dual_nozzle)
     if all(v < 0 for v in raw):
         # #2589 sentinel — every slot unresolved. Let the scheduler compute a
         # fresh mapping from live AMS state instead of trusting this.
@@ -395,6 +411,26 @@ class VirtualPrinterInstance:
             # IDLE reset was wrong — staying at FINISH is the designed
             # behaviour. The next upload's PREPARE→FINISH cycle starts fresh.
 
+    def _target_is_dual_nozzle(self) -> bool:
+        """Whether the target printer has two nozzles, which decides the global
+        tray of an external spool in the slicer's mapping (#3237).
+
+        Same signal dispatch uses: the live client's detection, else its model.
+        Without a client, the VP's own model is what the slicer sliced for.
+        """
+        from backend.app.utils.printer_models import is_dual_nozzle_model
+
+        client = (
+            self._printer_manager.get_client(self.target_printer_id)
+            if self._printer_manager is not None and self.target_printer_id is not None
+            else None
+        )
+        if client is not None:
+            return bool(getattr(client, "_is_dual_nozzle", False)) or is_dual_nozzle_model(
+                getattr(client, "model", None)
+            )
+        return is_dual_nozzle_model(self.model)
+
     async def on_print_command(self, filename: str, data: dict) -> None:
         """Handle print command from MQTT.
 
@@ -534,7 +570,9 @@ class VirtualPrinterInstance:
         # `_compute_ams_mapping_for_printer`, and with it prefer-lowest and the
         # #1766 backup gate).
         ams_mapping_json = (
-            _extract_slicer_ams_mapping_json(data, f"[VP {self.name}] Late MQTT")
+            _extract_slicer_ams_mapping_json(
+                data, f"[VP {self.name}] Late MQTT", is_dual_nozzle=self._target_is_dual_nozzle()
+            )
             if self.target_printer_id is not None and self.save_ams_mapping
             else None
         )
@@ -979,7 +1017,9 @@ class VirtualPrinterInstance:
                 # printer actually gets the job.
                 ams_mapping_json: str | None = None
                 if slicer_opts is not None and self.target_printer_id is not None and self.save_ams_mapping:
-                    ams_mapping_json = _extract_slicer_ams_mapping_json(slicer_opts, f"[VP {self.name}]")
+                    ams_mapping_json = _extract_slicer_ams_mapping_json(
+                        slicer_opts, f"[VP {self.name}]", is_dual_nozzle=self._target_is_dual_nozzle()
+                    )
 
                 # `Force color match` is the user asking Bambuddy to do the
                 # matching strictly, against the printer's live trays. Its only

backend/tests/unit/services/test_vp_external_spool_mapping_3237.py

@@ -0,0 +1,145 @@
+"""A VP with "Save AMS mapping" keeps the slicer's external-spool pick (#3237).
+
+The slicer writes an external spool as -1 in the flat ``ams_mapping``, the same
+as a filament with no tray, and names it only in ``ams_mapping2``. Saving the
+flat list alone made dispatch send the external filament as unmapped, and the
+printer stopped with 0700-8012 before the first layer.
+"""
+
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.virtual_printer.manager import (
+    VirtualPrinterInstance,
+    _extract_slicer_ams_mapping_json,
+)
+
+EXT_LEFT = {"ams_id": 254, "slot_id": 0}
+EXT_RIGHT = {"ams_id": 255, "slot_id": 0}
+UNMAPPED = {"ams_id": 255, "slot_id": 255}
+
+# The reporter's H2C job: left external spool plus AMS 0 slots 3 and 2.
+H2C_PAYLOAD = {
+    "command": "project_file",
+    "ams_mapping": [-1, 3, 2],
+    "ams_mapping2": [EXT_LEFT, {"ams_id": 0, "slot_id": 3}, {"ams_id": 0, "slot_id": 2}],
+}
+
+
+def _extract(data, is_dual_nozzle):
+    raw = _extract_slicer_ams_mapping_json(data, "[test]", is_dual_nozzle=is_dual_nozzle)
+    return None if raw is None else json.loads(raw)
+
+
+class TestExtract:
+    def test_left_external_spool_on_dual_nozzle(self):
+        assert _extract(H2C_PAYLOAD, is_dual_nozzle=True) == [254, 3, 2]
+
+    def test_right_external_spool_on_dual_nozzle(self):
+        data = {"ams_mapping": [-1, 3], "ams_mapping2": [EXT_RIGHT, {"ams_id": 0, "slot_id": 3}]}
+        assert _extract(data, is_dual_nozzle=True) == [255, 3]
+
+    def test_single_nozzle_external_spool_is_254(self):
+        # Single-nozzle printers have one external spool; the wire calls it
+        # 255, Bambuddy's global tray is 254.
+        data = {"ams_mapping": [0, -1], "ams_mapping2": [{"ams_id": 0, "slot_id": 0}, EXT_RIGHT]}
+        assert _extract(data, is_dual_nozzle=False) == [0, 254]
+
+    def test_unused_filament_stays_unmapped(self):
+        data = {"ams_mapping": [-1, 3], "ams_mapping2": [UNMAPPED, {"ams_id": 0, "slot_id": 3}]}
+        assert _extract(data, is_dual_nozzle=True) == [-1, 3]
+
+    def test_external_only_job_is_kept(self):
+        # Every flat entry is -1, but this is a real pick, not the #2589
+        # unresolved sentinel.
+        assert _extract({"ams_mapping": [-1], "ams_mapping2": [EXT_LEFT]}, is_dual_nozzle=True) == [254]
+
+    def test_unresolved_sentinel_still_dropped(self):
+        data = {"ams_mapping": [-1, -1], "ams_mapping2": [UNMAPPED, UNMAPPED]}
+        assert _extract(data, is_dual_nozzle=True) is None
+
+    def test_stringified_mapping2_is_parsed(self):
+        data = {"ams_mapping": "[-1, 3, 2]", "ams_mapping2": json.dumps(H2C_PAYLOAD["ams_mapping2"])}
+        assert _extract(data, is_dual_nozzle=True) == [254, 3, 2]
+
+    def test_missing_or_unusable_mapping2_keeps_flat_list(self):
+        assert _extract({"ams_mapping": [-1, 3, 2]}, is_dual_nozzle=True) == [-1, 3, 2]
+        assert _extract({"ams_mapping": [-1, 3, 2], "ams_mapping2": "not json"}, is_dual_nozzle=True) == [-1, 3, 2]
+        # Length mismatch: positions can't be paired, leave it alone.
+        assert _extract({"ams_mapping": [-1, 3, 2], "ams_mapping2": [EXT_LEFT]}, is_dual_nozzle=True) == [-1, 3, 2]
+
+
+def _instance(tmp_path, *, model="O1C", printer_manager=None, session_factory=None):
+    return VirtualPrinterInstance(
+        vp_id=3237,
+        name="ExtSpool",
+        mode="queue",
+        model=model,
+        access_code="12345678",
+        serial_suffix="391803237",
+        base_dir=tmp_path,
+        session_factory=session_factory,
+        save_ams_mapping=True,
+        target_printer_id=7,
+        printer_manager=printer_manager,
+    )
+
+
+def _manager_with(client):
+    manager = MagicMock()
+    manager.get_client = MagicMock(return_value=client)
+    return manager
+
+
+class TestTargetIsDualNozzle:
+    def test_live_detection_wins(self, tmp_path):
+        client = MagicMock(_is_dual_nozzle=True, model="Some future model")
+        assert _instance(tmp_path, model="BL-P001", printer_manager=_manager_with(client))._target_is_dual_nozzle()
+
+    def test_client_model(self, tmp_path):
+        client = MagicMock(_is_dual_nozzle=False, model="H2C")
+        assert _instance(tmp_path, model="BL-P001", printer_manager=_manager_with(client))._target_is_dual_nozzle()
+
+    def test_single_nozzle_client(self, tmp_path):
+        client = MagicMock(_is_dual_nozzle=False, model="X1C")
+        assert not _instance(tmp_path, printer_manager=_manager_with(client))._target_is_dual_nozzle()
+
+    def test_falls_back_to_vp_model_without_client(self, tmp_path):
+        assert _instance(tmp_path, model="O1C", printer_manager=_manager_with(None))._target_is_dual_nozzle()
+        assert not _instance(tmp_path, model="BL-P001")._target_is_dual_nozzle()
+
+
+@pytest.mark.asyncio
+async def test_queue_item_and_archive_keep_left_external_spool(tmp_path):
+    added_items = []
+    mock_db = AsyncMock()
+    mock_db.add = MagicMock(side_effect=added_items.append)
+    mock_db.commit = AsyncMock()
+    session_ctx = AsyncMock()
+    session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
+    session_ctx.__aexit__ = AsyncMock(return_value=False)
+    session_factory = MagicMock(return_value=session_ctx)
+
+    client = MagicMock(_is_dual_nozzle=True, model="H2C")
+    inst = _instance(tmp_path, printer_manager=_manager_with(client), session_factory=session_factory)
+
+    file_path = tmp_path / "test.3mf"
+    file_path.write_bytes(b"fake3mf")
+    await inst.on_print_command(file_path.name, dict(H2C_PAYLOAD))
+
+    mock_archive = MagicMock(id=1, print_name="test")
+    with (
+        patch("backend.app.api.routes.settings.get_setting", new_callable=AsyncMock, return_value=None),
+        patch(
+            "backend.app.services.archive.ArchiveService.archive_print",
+            new_callable=AsyncMock,
+            return_value=mock_archive,
+        ) as archive_print,
+    ):
+        await inst._add_to_print_queue(file_path, "192.168.1.100")
+
+    assert len(added_items) == 1
+    assert json.loads(added_items[0].ams_mapping) == [254, 3, 2]
+    assert archive_print.await_args.kwargs["slicer_ams_mapping"] == [254, 3, 2]