Charge the tray the printer said it used, not the first one loaded (issue #2953)

A sliced file numbers its filaments 1..4; which AMS tray each came from is
decided when the job is sent. #2768 gave the Spoolman writer two ways to
recover that decision when the print did not come through Bambuddy: the
printer's own mapping field, and a colour match of the 3MF's slots against the
loaded trays. An A1 satisfies neither. It publishes no mapping field, and it
drops the MQTT connection when we subscribe to its request topic, so the
slicer's instruction never arrives either. That leaves the colour match, and it
compares hex strings exactly.

The reporter sliced with a generic black profile against a tray they had set to
charged slot 1 to whatever sat in the first tray -- 2.17 g onto a grey PLA+
spool, while the print was fed from tray 3. Their bundle carries the printer's
own answer: "Tray change during print: tray=3 at layer=0", recorded 90 seconds
in, and read further down the same completion pass by _print_used_tray_keys to
decide which slots the print had touched. The same pass then charged tray 0 on
a guess, and logged "AMS0-T3: remain% did not fall over the print" about the
tray that had actually done the work.

_single_slot_tray_from_state adds the third rung. For a print with exactly one
slot carrying usage, the one slot came from the one tray, so the printer's tray
reporting answers the question directly: the mid-print tray-change log, then
the tray loaded at print start, then the current one, then the last real tray
seen. That is the ladder usage_tracker.on_print_complete has consulted since it
started resolving mappings at completion -- Spoolman users were the only ones
not getting it, which is why an install running the built-in inventory has
never shown this. On this printer only the first and last rungs can fire:
tray_now_at_start is 255 because print start runs before the filament is
loaded, and the A1 parks tray_now back at 255 the moment a print ends.

Gated on exactly one slot with usage, like the internal writer: a multi-colour
print moves tray_now on every change, so one reading cannot then be attributed
to one slot. It also declines when the log holds more than one switch, because
an AMS-backup runout is split per segment (#1793) and a single-tray mapping
would land the whole print on one spool.

That gate needed the guess warning to exclude the split path too. The split
never reads slot_to_tray at all -- it charges each segment to the tray the
printer announced switching to, which is the same evidence this fallback is
built on -- so a declined mapping there is not a guess, and calling it one
suppressed the archive rewrite for exactly the prints whose attribution is best
supported. Nothing covered that combination; a test does now.

Where nothing names a tray the positional default still stands, because it is
right for an AMS loaded in slicer order. It now says so at warning level so a
support bundle carries the reason, and it no longer restamps the archive's
filament colour and material from a spool it picked by position. That restamp
is what made the fault read as data loss: the grams can be put back, whereas
overwriting what the slicer recorded leaves nothing to compare against, and the
reporter's archive had already been rewritten from #000000 to the wrong spool's
grey. A slot that consumed nothing also stops claiming a tray in the handled
set -- it was never charged, so the remain-delta path should stay free to cover
it rather than be suppressed by an estimate of zero.

The request-topic probe is the same failure reached from the other side. A
printer that refuses kills the TCP connection instead of returning a SUBACK
failure, so the only signal is "we subscribed, then got disconnected", and that
was believed the first time it happened. Every other reason a connection drops
inside the same window looks identical -- a network blip, the printer
rebooting, the container stopped mid-probe -- and the verdict was cached per
serial with no re-probe anywhere, so on a printer that supports the topic one
unlucky drop cost mapping capture for the rest of the process and every slicer
print after it was charged by tray position. It now takes two consecutive
drops, and a disconnect we asked for is not counted. A printer that genuinely
refuses answers the same way every time and pays one extra reconnect; one
already known to refuse still skips the subscription outright rather than
reopening a reconnect loop.

backend/app/services/bambu_mqtt.py

@@ -1085,6 +1085,17 @@ class BambuMQTTClient:
     # Class-level cache: serial_number -> False when request topic is known unsupported.
     # Persists across client instances so reconnects don't re-trigger failed subscriptions.
     _request_topic_cache: dict[str, bool] = {}
+    # serial_number -> consecutive disconnects seen shortly after subscribing to
+    # the request topic. A SUBACK failure is the broker answering the question;
+    # a disconnect is only circumstantial, and any drop inside the window looks
+    # identical -- a network blip, the printer rebooting, the container being
+    # stopped mid-probe. Latching on the first one costs ams_mapping capture for
+    # the rest of the process on a printer that supports it perfectly well
+    # (#2953). Require the drop to repeat before believing it; a printer that
+    # really does refuse the topic answers the same way every time and pays one
+    # extra reconnect for it.
+    _request_topic_probe_failures: dict[str, int] = {}
+    _REQUEST_TOPIC_PROBE_LIMIT: int = 2
     # Counter for generating unique MQTT client IDs across instances.
     _client_instance_counter: int = 0
 
@@ -1698,6 +1709,7 @@ class BambuMQTTClient:
                     )
                     self._request_topic_confirmed = True
                     BambuMQTTClient._request_topic_cache[self.serial_number] = True
+                    BambuMQTTClient._request_topic_probe_failures.pop(self.serial_number, None)
             self._request_topic_sub_mid = None
             self._request_topic_sub_time = 0.0
 
@@ -1754,13 +1766,30 @@ class BambuMQTTClient:
             self._request_topic_sub_time > 0
             and not self._request_topic_confirmed
             and time.time() - self._request_topic_sub_time < 10.0
+            # A disconnect we asked for says nothing about the subscription.
+            and self._disconnection_event is None
         ):
-            logger.warning(
-                "[%s] Disconnected shortly after request topic subscription. Disabling request topic for this printer.",
-                self.serial_number,
-            )
-            self._request_topic_supported = False
-            BambuMQTTClient._request_topic_cache[self.serial_number] = False
+            failures = BambuMQTTClient._request_topic_probe_failures.get(self.serial_number, 0) + 1
+            BambuMQTTClient._request_topic_probe_failures[self.serial_number] = failures
+            if failures >= BambuMQTTClient._REQUEST_TOPIC_PROBE_LIMIT:
+                logger.warning(
+                    "[%s] Disconnected shortly after request topic subscription %d times. "
+                    "Disabling request topic for this printer — ams_mapping capture from "
+                    "slicer-initiated prints is unavailable, and their filament will be "
+                    "attributed from the printer's own tray reporting instead.",
+                    self.serial_number,
+                    failures,
+                )
+                self._request_topic_supported = False
+                BambuMQTTClient._request_topic_cache[self.serial_number] = False
+            else:
+                logger.info(
+                    "[%s] Disconnected shortly after request topic subscription (%d/%d). "
+                    "Retrying it on the next connection before giving up.",
+                    self.serial_number,
+                    failures,
+                    BambuMQTTClient._REQUEST_TOPIC_PROBE_LIMIT,
+                )
         self._request_topic_sub_mid = None
         self._request_topic_sub_time = 0.0
 

backend/app/services/spoolman_tracking.py

@@ -40,6 +40,11 @@ _ZERO_TAG_UID = "0000000000000000"
 _MAX_REAL_TRAY_ID = 254
 
 
+def _is_real_tray_id(value) -> bool:
+    """True when ``value`` names a physical slot rather than "nothing loaded"."""
+    return isinstance(value, int) and not isinstance(value, bool) and 0 <= value <= _MAX_REAL_TRAY_ID
+
+
 def _is_non_zero_identifier(value: str) -> bool:
     """Return True when identifier is non-empty and not all zeros."""
     if not value:
@@ -163,7 +168,73 @@ def _resolve_global_tray_id(slot_id: int, slot_to_tray: list | None, ams_trays:
     return slot_id - 1
 
 
-def _resolve_slot_to_tray_fallback(printer_id: int, filament_usage: list[dict]) -> tuple[list[int] | None, str]:
+def _single_slot_tray_from_state(
+    state,
+    filament_usage: list[dict],
+    tray_now_at_start: int | None = None,
+) -> tuple[int, int] | None:
+    """The tray a single-slot print actually drew from, read off the printer.
+
+    A1, A1 mini, P1S and P2S publish no ``mapping`` field and drop the MQTT
+    connection when we subscribe to their request topic, so neither of the
+    other two fallbacks can answer for them. What they do report is which tray
+    the extruder is fed from, and for a print that uses exactly one filament
+    slot that is the same question: the one slot came from the one tray.
+
+    The ladder mirrors ``usage_tracker.on_print_complete`` step 5, which has
+    consulted these same fields since it started resolving mappings at
+    completion. Spoolman users were the only ones not getting them (#2953).
+
+    Gated on exactly one slot with usage, like the internal writer: on a
+    multi-colour print every filament change moves ``tray_now``, so a single
+    tray reading says nothing about which slot it belongs to.
+
+    More than one tray-change entry means the print switched trays mid-run
+    (AMS backup on runout, #957). ``report_usage`` splits those per segment
+    and must not be handed a single-tray mapping instead, so this declines.
+
+    Returns ``(slot_id, global_tray_id)``, or None when the printer offered no
+    usable reading and the positional default stands.
+    """
+    nonzero = [u for u in filament_usage or [] if u.get("used_g", 0) > 0]
+    if len(nonzero) != 1:
+        return None
+    slot_id = nonzero[0].get("slot_id", 0)
+    if slot_id <= 0:
+        return None
+
+    changes = list(getattr(state, "tray_change_log", None) or [])
+    if len(changes) > 1:
+        return None
+    if len(changes) == 1:
+        entry = changes[0]
+        if isinstance(entry, (tuple, list)) and entry and _is_real_tray_id(entry[0]):
+            # Strongest evidence there is: the printer announced this switch
+            # while the job was running, so it describes this print and no
+            # other. On the reporter's A1 it read (3, 0) -- tray 3 at layer 0
+            # -- while the positional default was charging tray 0.
+            return slot_id, entry[0]
+
+    # No mid-print switch recorded. Fall back to the standing tray readings,
+    # newest evidence first. ``tray_now`` is 255 both at rest and while
+    # nothing is loaded, which is why _MAX_REAL_TRAY_ID excludes it; A1
+    # firmware parks there the moment a print ends, leaving last_loaded_tray
+    # as the only survivor.
+    for candidate in (
+        tray_now_at_start,
+        getattr(state, "tray_now", None),
+        getattr(state, "last_loaded_tray", None),
+    ):
+        if _is_real_tray_id(candidate):
+            return slot_id, candidate
+    return None
+
+
+def _resolve_slot_to_tray_fallback(
+    printer_id: int,
+    filament_usage: list[dict],
+    tray_now_at_start: int | None = None,
+) -> tuple[list[int] | None, str]:
     """Recover a slot-to-tray mapping at completion when print start captured none.
 
     ``store_print_data`` can only learn the mapping from two sources: the
@@ -180,9 +251,17 @@ def _resolve_slot_to_tray_fallback(printer_id: int, filament_usage: list[dict])
     The printer knows the real answer. Its ``mapping`` field carries the actual
     slot-to-tray assignment for the running job, and for the models that never
     publish it (A1, P1S, P2S) the 3MF's per-slot colours can be matched against
-    the loaded trays instead. The built-in inventory writer has consulted both
-    for as long as it has resolved mappings at completion; this gives the
-    Spoolman writer the same two fallbacks at the same moment.
+    the loaded trays instead. Failing both, a print that used a single filament
+    slot can be pinned to the tray the printer reported feeding from
+    (``_single_slot_tray_from_state``).
+
+    The built-in inventory writer has consulted all three for as long as it has
+    resolved mappings at completion. The first version of this function offered
+    only the first two, which left A1-class printers -- no ``mapping`` field, no
+    request topic -- with nothing but the colour match, and that needs the
+    slicer's filament colour to equal the tray's exactly. A generic black
+    profile against a tray set to #111111 does not match, and the print is
+    charged to whichever spool happens to sit in the first tray (#2953).
 
     Deliberately at completion rather than inside ``store_print_data``: the
     printer keeps publishing ``mapping`` long after a job ends — it is still in
@@ -193,28 +272,46 @@ def _resolve_slot_to_tray_fallback(printer_id: int, filament_usage: list[dict])
 
     Args:
         printer_id: Printer whose live state is consulted.
-        filament_usage: The 3MF's per-slot estimates, needed by the colour
-            match. Only the ``slot_id``/``color`` keys are read.
+        filament_usage: The 3MF's per-slot estimates. The colour match reads
+            ``slot_id``/``color``; the tray-state fallback reads
+            ``slot_id``/``used_g``.
+        tray_now_at_start: The tray the printer was feeding from when the print
+            began, as captured by ``store_print_data``. Only consulted by the
+            tray-state fallback.
 
     Returns:
-        ``(mapping, source)``, or ``(None, "none")`` when neither fallback
-        produced anything and the positional default stands.
+        ``(mapping, source)``, or ``(None, "none")`` when no fallback produced
+        anything and the positional default stands.
     """
     from backend.app.services.printer_manager import printer_manager
     from backend.app.services.usage_tracker import _decode_mqtt_mapping, _match_slots_by_color
 
     state = printer_manager.get_status(printer_id)
     raw_data = getattr(state, "raw_data", None) if state else None
-    if not raw_data:
-        return None, "none"
 
-    decoded = _decode_mqtt_mapping(raw_data.get("mapping"))
-    if decoded:
-        return decoded, "mqtt"
-
-    matched = _match_slots_by_color(filament_usage, raw_data.get("ams"))
-    if matched:
-        return matched, "color_match"
+    # Both of the first two fallbacks read the status payload; the third reads
+    # fields ``bambu_mqtt`` maintains on the state object itself, so an empty
+    # payload must not short-circuit past it.
+    if raw_data:
+        decoded = _decode_mqtt_mapping(raw_data.get("mapping"))
+        if decoded:
+            return decoded, "mqtt"
+
+        matched = _match_slots_by_color(filament_usage, raw_data.get("ams"))
+        if matched:
+            return matched, "color_match"
+
+    single = _single_slot_tray_from_state(state, filament_usage, tray_now_at_start)
+    if single is not None:
+        slot_id, global_tray_id = single
+        # Only the one slot is claimed. The -1 padding is the array's existing
+        # "not an AMS tray" value, and the slots carrying it consumed nothing,
+        # so no caller resolves them: ``_report_spool_usage_for_slots`` skips
+        # zero-gram slots before resolving, ``_print_used_tray_keys`` skips
+        # negatives, and report_usage's handled-set skips them too.
+        mapping = [-1] * slot_id
+        mapping[slot_id - 1] = global_tray_id
+        return mapping, "tray_state"
 
     return None, "none"
 
@@ -929,7 +1026,11 @@ async def _report_partial_usage(
     # ``_resolve_global_tray_id`` (#2768). An aborted print charges the wrong
     # spool just as readily as a finished one.
     if not slot_to_tray:
-        slot_to_tray, _partial_mapping_source = _resolve_slot_to_tray_fallback(printer_id, filament_usage)
+        slot_to_tray, _partial_mapping_source = _resolve_slot_to_tray_fallback(
+            printer_id,
+            filament_usage,
+            getattr(tracking, "tray_now_at_start", None),
+        )
         logger.info(
             "[SPOOLMAN] Partial usage: slot_to_tray=%s (source: %s)",
             slot_to_tray,
@@ -1124,13 +1225,34 @@ async def report_usage(printer_id: int, archive_id: int):
         # AMS slot directly, so there is nothing to recover for it.
         mapping_source = "stored" if slot_to_tray else "none"
         if filament_usage and not slot_to_tray:
-            slot_to_tray, mapping_source = _resolve_slot_to_tray_fallback(printer_id, filament_usage)
+            slot_to_tray, mapping_source = _resolve_slot_to_tray_fallback(printer_id, filament_usage, tray_now_at_start)
         logger.info(
             "[SPOOLMAN] Archive %s: slot_to_tray=%s (source: %s)",
             archive_id,
             slot_to_tray,
             mapping_source,
         )
+        # Nothing named a tray for this print and no fallback could recover
+        # one, so every slot is about to be resolved by position -- slicer slot
+        # 1 to the first loaded tray, and so on. That guess is right for an AMS
+        # loaded in slicer order and wrong for any other, and the caller has no
+        # way to tell which it got. Say so at a level that survives the default
+        # log filter, so a support bundle carries the reason (#2953).
+        #
+        # Excludes the tray-split path. It never reads ``slot_to_tray`` at all:
+        # it charges each segment to the tray the printer announced switching
+        # to, which is the same evidence the tray-state fallback is built on and
+        # is not a guess. Calling it one would suppress the archive rewrite for
+        # exactly the prints -- an AMS-backup runout on a Studio job (#1793 in
+        # #2768's conditions) -- whose attribution is best supported.
+        mapping_is_guess = bool(filament_usage) and mapping_source == "none" and len(tray_changes) <= 1
+        if mapping_is_guess:
+            logger.warning(
+                "[SPOOLMAN] Archive %s: no slot-to-tray mapping from any source -- "
+                "charging by tray position, which is a guess. Verify the spool weights "
+                "if the AMS is not loaded in slicer order.",
+                archive_id,
+            )
 
         slot_colors: dict[int, str] = {}
         slot_materials: dict[int, str] = {}
@@ -1199,6 +1321,13 @@ async def report_usage(printer_id: int, archive_id: int):
                 # Track which physical slots the 3MF path already covered so
                 # Path 2 doesn't double-charge them.
                 for u in filament_usage:
+                    if u.get("used_g", 0) <= 0:
+                        # ``_report_spool_usage_for_slots`` skipped this slot
+                        # before resolving a tray for it, so nothing was
+                        # charged and Path 2 is free to cover the slot from
+                        # remain% -- claiming it here would suppress a real
+                        # drop on the strength of a zero-gram estimate.
+                        continue
                     slot_id = u.get("slot_id", 0)
                     handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays))
 
@@ -1231,11 +1360,25 @@ async def report_usage(printer_id: int, archive_id: int):
         # Stamp the archive's filament colour from the matched Spoolman spools
         # so it reflects the curated inventory colour, not the slicer's 3MF
         # value (#1494) — mirrors the built-in inventory path in usage_tracker.
-        await _apply_spool_colors_to_archive(db, archive_id, filament_usage, slot_colors)
+        #
+        # Skipped when the mapping was a positional guess. Charging the wrong
+        # spool costs grams the owner can put back; rewriting the archive's
+        # colour and material on top of it overwrites what the slicer actually
+        # recorded, and the print then reads as a different filament than the
+        # one that made it, with nothing left to compare against (#2953).
+        if mapping_is_guess:
+            if slot_colors or slot_materials:
+                logger.info(
+                    "[SPOOLMAN] Archive %s: leaving filament colour/type as sliced — "
+                    "the spools were matched by position, not by a known mapping",
+                    archive_id,
+                )
+        else:
+            await _apply_spool_colors_to_archive(db, archive_id, filament_usage, slot_colors)
 
-        # Same for the material: a slot mapped to a differently-typed spool than
-        # it was sliced for otherwise records the sliced type (#2563).
-        await _apply_spool_types_to_archive(db, archive_id, filament_usage, slot_materials)
+            # Same for the material: a slot mapped to a differently-typed spool
+            # than it was sliced for otherwise records the sliced type (#2563).
+            await _apply_spool_types_to_archive(db, archive_id, filament_usage, slot_materials)
 
 
 def _print_used_tray_keys(

backend/tests/unit/services/test_bambu_mqtt.py

@@ -1805,6 +1805,7 @@ class TestRequestTopicFailSafe:
         from backend.app.services.bambu_mqtt import BambuMQTTClient
 
         BambuMQTTClient._request_topic_cache.clear()
+        BambuMQTTClient._request_topic_probe_failures.clear()
 
     @pytest.fixture
     def mqtt_client(self):
@@ -1858,7 +1859,14 @@ class TestRequestTopicFailSafe:
         assert mqtt_client._request_topic_supported is True
 
     def test_disconnect_after_subscription_disables_topic(self, mqtt_client):
-        """Disconnect within 10s of subscription attempt disables request topic."""
+        """Repeated disconnects within 10s of a subscription attempt disable the
+        request topic.
+
+        One is not enough (#2953). Any drop inside the window looks the same as
+        the broker refusing the topic, so the first is treated as circumstantial
+        and the subscription is retried on the next connection. A printer that
+        really does refuse answers the same way every time.
+        """
         import time
 
         mqtt_client._request_topic_sub_time = time.time()
@@ -1867,6 +1875,12 @@ class TestRequestTopicFailSafe:
 
         mqtt_client._on_disconnect(None, None)
 
+        assert mqtt_client._request_topic_supported is True
+        assert mqtt_client._request_topic_sub_time == 0.0
+
+        mqtt_client._request_topic_sub_time = time.time()
+        mqtt_client._on_disconnect(None, None)
+
         assert mqtt_client._request_topic_supported is False
         assert mqtt_client._request_topic_sub_time == 0.0
 
@@ -1924,11 +1938,13 @@ class TestRequestTopicFailSafe:
         )
         assert client1._request_topic_supported is True
 
-        # Simulate disconnect-after-subscribe disabling the topic
-        client1._request_topic_sub_time = __import__("time").time()
+        # Simulate disconnect-after-subscribe disabling the topic. Takes two
+        # (#2953) — the first drop is not treated as an answer.
         client1._request_topic_confirmed = False
         client1._last_message_time = 0.0
-        client1._on_disconnect(None, None)
+        for _ in range(BambuMQTTClient._REQUEST_TOPIC_PROBE_LIMIT):
+            client1._request_topic_sub_time = __import__("time").time()
+            client1._on_disconnect(None, None)
         assert client1._request_topic_supported is False
 
         # New instance for same serial should inherit the cached state

backend/tests/unit/services/test_spoolman_tray_state_fallback_2953.py

@@ -0,0 +1,388 @@
+"""Tray-state slot mapping for printers that can answer no other way (#2953).
+
+#2768 gave the Spoolman writer two ways to recover a print's slot-to-tray
+mapping when print start captured none: the printer's ``mapping`` field, and a
+colour match of the 3MF's slots against the loaded trays. An A1 can satisfy
+neither. It publishes no ``mapping`` field, and it drops the MQTT connection
+when Bambuddy subscribes to its request topic, so the slicer's own instruction
+never arrives either. That leaves the colour match, which compares hex strings
+exactly.
+
+The reporter sliced with a generic black profile (``#000000``) against a tray
+they had set to ``#111111``. No match, so every print fell through to the
+positional default and charged slot 1 to the first loaded tray -- the grey PLA+
+in tray 0 -- while the print was actually fed from tray 3. Their log carries
+the printer's own answer, ``Tray change during print: tray=3 at layer=0``,
+recorded 90 seconds into the print and already read by ``_print_used_tray_keys``
+further down the same completion pass.
+
+Values throughout are the ones from archive 12 of the reporter's support
+bundle.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.spoolman_tracking import (
+    _resolve_slot_to_tray_fallback,
+    _single_slot_tray_from_state,
+)
+
+# The reporter's AMS at completion: tray 1 is empty, and no tray is #000000.
+REPORTER_AMS = [
+    {
+        "id": 0,
+        "tray": [
+            {"id": 0, "tray_color": "888888FF", "tray_type": "PLA+"},
+            {"id": 1, "tray_color": None, "tray_type": None},
+            {"id": 2, "tray_color": "5F4036FF", "tray_type": "PLA"},
+            {"id": 3, "tray_color": "111111FF", "tray_type": "PLA+"},
+        ],
+    }
+]
+REPORTER_USAGE = [{"slot_id": 1, "used_g": 2.17, "type": "PLA+", "color": "#000000"}]
+
+
+class _AsyncCtx:
+    def __init__(self, db):
+        self._db = db
+
+    async def __aenter__(self):
+        return self._db
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+def _state(tray_change_log=None, tray_now=255, last_loaded_tray=-1, **raw):
+    """An A1 as it looks at completion: no mapping field, nothing loaded."""
+    return SimpleNamespace(
+        raw_data=raw,
+        layer_num=0,
+        total_layers=0,
+        tray_change_log=list(tray_change_log or []),
+        tray_now=tray_now,
+        last_loaded_tray=last_loaded_tray,
+    )
+
+
+def _patched_pm(state):
+    pm = MagicMock()
+    pm.get_status.return_value = state
+    return pm
+
+
+class TestSingleSlotTrayFromState:
+    def test_the_mid_print_tray_change_is_the_answer(self):
+        """``Tray change during print: tray=3 at layer=0``. The printer
+        announced the switch while the job was running, so it describes this
+        print and no other."""
+        state = _state(tray_change_log=[(3, 0)])
+        assert _single_slot_tray_from_state(state, REPORTER_USAGE, 255) == (1, 3)
+
+    def test_declines_a_multi_slot_print(self):
+        """Every colour change moves ``tray_now``, so one tray reading can't be
+        attributed to one slot. Same gate the internal writer uses."""
+        usage = [
+            {"slot_id": 1, "used_g": 10.0, "color": "#000000"},
+            {"slot_id": 2, "used_g": 5.0, "color": "#FFFFFF"},
+        ]
+        assert _single_slot_tray_from_state(_state(tray_change_log=[(3, 0)]), usage, None) is None
+
+    def test_declines_when_the_print_switched_trays(self):
+        """Two entries means AMS backup swapped a spool mid-print (#957).
+        ``report_usage`` splits those per segment; handing it a single-tray
+        mapping instead would charge the whole print to one of them."""
+        state = _state(tray_change_log=[(3, 0), (0, 120)])
+        assert _single_slot_tray_from_state(state, REPORTER_USAGE, None) is None
+
+    def test_slots_that_consumed_nothing_do_not_count_as_a_second_slot(self):
+        """A purge-only slot is in the 3MF with zero grams. It is not a second
+        filament and must not disqualify the print."""
+        usage = [
+            {"slot_id": 1, "used_g": 2.17, "color": "#000000"},
+            {"slot_id": 2, "used_g": 0.0, "color": "#FFFFFF"},
+        ]
+        assert _single_slot_tray_from_state(_state(tray_change_log=[(3, 0)]), usage, None) == (1, 3)
+
+    def test_falls_back_to_tray_now_at_start(self):
+        state = _state(tray_change_log=[])
+        assert _single_slot_tray_from_state(state, REPORTER_USAGE, 2) == (1, 2)
+
+    def test_falls_back_to_current_tray_now(self):
+        state = _state(tray_change_log=[], tray_now=2)
+        assert _single_slot_tray_from_state(state, REPORTER_USAGE, 255) == (1, 2)
+
+    def test_falls_back_to_last_loaded_tray(self):
+        """The reporter's A1 parks ``tray_now`` at 255 the moment the print
+        ends, and their ``tray_now_at_start`` was 255 too because print start
+        fires before the filament is loaded. ``last_loaded_tray`` only ever
+        latches real trays, so it is the one still holding the answer."""
+        state = _state(tray_change_log=[], tray_now=255, last_loaded_tray=3)
+        assert _single_slot_tray_from_state(state, REPORTER_USAGE, 255) == (1, 3)
+
+    def test_255_is_not_a_tray(self):
+        """255 is ``tray_now`` at rest and what an unparseable reading falls
+        back to. Reading it as a slot would charge a spool on no evidence."""
+        state = _state(tray_change_log=[], tray_now=255, last_loaded_tray=255)
+        assert _single_slot_tray_from_state(state, REPORTER_USAGE, 255) is None
+
+    def test_no_state_at_all(self):
+        assert _single_slot_tray_from_state(None, REPORTER_USAGE, None) is None
+
+
+class TestResolveSlotToTrayFallbackRung:
+    def test_the_reporters_print_resolves_to_tray_3(self):
+        """End of the chain: no mapping field, colours don't match, and the
+        tray-change log settles it."""
+        pm = _patched_pm(_state(tray_change_log=[(3, 0)], ams=REPORTER_AMS))
+
+        with patch("backend.app.services.printer_manager.printer_manager", pm):
+            mapping, source = _resolve_slot_to_tray_fallback(1, REPORTER_USAGE, 255)
+
+        assert mapping == [3]
+        assert source == "tray_state"
+
+    def test_colour_match_still_wins(self):
+        """When the slicer colour does equal a tray's, that is a direct
+        statement about this slot and outranks a tray reading."""
+        usage = [{"slot_id": 1, "used_g": 2.17, "color": "#5F4036"}]
+        pm = _patched_pm(_state(tray_change_log=[(3, 0)], ams=REPORTER_AMS))
+
+        with patch("backend.app.services.printer_manager.printer_manager", pm):
+            mapping, source = _resolve_slot_to_tray_fallback(1, usage, 255)
+
+        assert mapping == [2]
+        assert source == "color_match"
+
+    def test_mapping_field_still_wins(self):
+        pm = _patched_pm(_state(tray_change_log=[(3, 0)], mapping=[0], ams=REPORTER_AMS))
+
+        with patch("backend.app.services.printer_manager.printer_manager", pm):
+            mapping, source = _resolve_slot_to_tray_fallback(1, REPORTER_USAGE, 255)
+
+        assert mapping == [0]
+        assert source == "mqtt"
+
+    def test_an_empty_status_payload_still_reaches_the_tray_rung(self):
+        """``tray_change_log`` lives on the state object, not in the status
+        payload, so an empty payload must not short-circuit past it."""
+        pm = _patched_pm(_state(tray_change_log=[(3, 0)]))
+
+        with patch("backend.app.services.printer_manager.printer_manager", pm):
+            mapping, source = _resolve_slot_to_tray_fallback(1, REPORTER_USAGE, None)
+
+        assert mapping == [3]
+        assert source == "tray_state"
+
+    def test_only_the_used_slot_is_claimed(self):
+        """A print whose one filament is slot 3 pads the array so the index
+        lines up. The padding is -1, which no caller resolves: those slots
+        consumed nothing."""
+        usage = [
+            {"slot_id": 1, "used_g": 0.0, "color": "#AAAAAA"},
+            {"slot_id": 3, "used_g": 2.17, "color": "#000000"},
+        ]
+        pm = _patched_pm(_state(tray_change_log=[(3, 0)]))
+
+        with patch("backend.app.services.printer_manager.printer_manager", pm):
+            mapping, source = _resolve_slot_to_tray_fallback(1, usage, None)
+
+        assert mapping == [-1, -1, 3]
+        assert source == "tray_state"
+
+    def test_still_says_none_when_the_printer_offers_nothing(self):
+        pm = _patched_pm(_state(tray_change_log=[], tray_now=255, last_loaded_tray=-1, ams=REPORTER_AMS))
+
+        with patch("backend.app.services.printer_manager.printer_manager", pm):
+            mapping, source = _resolve_slot_to_tray_fallback(1, REPORTER_USAGE, 255)
+
+        assert mapping is None
+        assert source == "none"
+
+
+class TestReportUsageChargesTheRightSpool:
+    """The reporter's archive 12, end to end."""
+
+    @staticmethod
+    def _run(tracking, state, spool_by_tag, archive):
+        rows = iter([tracking])
+
+        def _next_row(*_args, **_kwargs):
+            result = MagicMock()
+            result.scalar_one_or_none.return_value = next(rows, archive)
+            return result
+
+        db = AsyncMock()
+        db.execute = AsyncMock(side_effect=_next_row)
+        db.delete = AsyncMock()
+        db.commit = AsyncMock()
+
+        client = AsyncMock()
+        client.find_spool_by_tag = AsyncMock(side_effect=lambda tag: spool_by_tag.get(tag))
+        client.use_spool = AsyncMock()
+
+        pm = _patched_pm(state)
+
+        async def _go():
+            from backend.app.services.spoolman_tracking import report_usage
+
+            with (
+                patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
+                patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+                patch(
+                    "backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
+                    AsyncMock(return_value=client),
+                ),
+                patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SER")),
+                patch(
+                    "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+                    AsyncMock(return_value=None),
+                ),
+                patch("backend.app.services.printer_manager.printer_manager", pm),
+            ):
+                await report_usage(printer_id=1, archive_id=12)
+
+        return _go, client
+
+    @staticmethod
+    def _tracking():
+        return SimpleNamespace(
+            filament_usage=list(REPORTER_USAGE),
+            ams_trays={
+                "0": {"tray_uuid": "TRAY0", "tag_uid": "", "tray_type": "PLA+"},
+                "2": {"tray_uuid": "TRAY2", "tag_uid": "", "tray_type": "PLA"},
+                "3": {"tray_uuid": "TRAY3", "tag_uid": "", "tray_type": "PLA+"},
+            },
+            slot_to_tray=None,
+            tray_remain_start=None,
+            layer_usage=None,
+            filament_properties=None,
+            tray_now_at_start=255,
+        )
+
+    SPOOLS = {
+        # Spool 41 is the grey PLA+ that was wrongly charged 2.17 g.
+        "TRAY0": {"id": 41, "filament": {"color_hex": "888888", "material": "PLA+"}},
+        "TRAY2": {"id": 20, "filament": {"color_hex": "5F4036", "material": "PLA"}},
+        "TRAY3": {"id": 46, "filament": {"color_hex": "111111", "material": "PLA+"}},
+    }
+
+    @pytest.mark.asyncio
+    async def test_the_tray_the_printer_named_is_charged(self):
+        archive = SimpleNamespace(filament_color="#000000", filament_type="PLA+")
+        state = _state(tray_change_log=[(3, 0)], tray_now=255, last_loaded_tray=3, ams=REPORTER_AMS)
+
+        run, client = self._run(self._tracking(), state, self.SPOOLS, archive)
+        await run()
+
+        client.use_spool.assert_awaited_once_with(46, 2.17)
+
+    @pytest.mark.asyncio
+    async def test_the_archive_is_not_restamped_from_a_positional_guess(self):
+        """Nothing named a tray, so slot 1 is charged by position. The grams
+        can be put back; overwriting what the slicer recorded cannot, so the
+        archive keeps the colour and material it was printed with."""
+        archive = SimpleNamespace(filament_color="#000000", filament_type="PLA+")
+        state = _state(tray_change_log=[], tray_now=255, last_loaded_tray=-1, ams=REPORTER_AMS)
+
+        run, client = self._run(self._tracking(), state, self.SPOOLS, archive)
+        await run()
+
+        client.use_spool.assert_awaited_once_with(41, 2.17)
+        assert archive.filament_color == "#000000"
+        assert archive.filament_type == "PLA+"
+
+    @pytest.mark.asyncio
+    async def test_a_resolved_mapping_still_restamps_the_archive(self):
+        """#1494 and #2563 are unchanged when the mapping is actually known."""
+        archive = SimpleNamespace(filament_color="#000000", filament_type="PLA+")
+        state = _state(tray_change_log=[(3, 0)], tray_now=255, last_loaded_tray=3, ams=REPORTER_AMS)
+
+        run, _client = self._run(self._tracking(), state, self.SPOOLS, archive)
+        await run()
+
+        assert archive.filament_color == "#111111"
+
+
+class TestTraySplitIsNotAPositionalGuess:
+    """A print that switched trays mid-run is attributed per segment from the
+    tray-change log (#1793). That path never reads ``slot_to_tray``, so the
+    absence of a mapping says nothing about it -- treating it as a positional
+    guess would suppress the archive rewrite for the prints whose attribution
+    is best supported, and log a warning naming a mechanism that did not run.
+    """
+
+    @pytest.mark.asyncio
+    async def test_a_runout_switch_without_a_mapping_still_restamps_the_archive(self):
+        from backend.app.services.spoolman_tracking import report_usage
+
+        tracking = SimpleNamespace(
+            filament_usage=[{"slot_id": 1, "used_g": 72.56, "type": "PLA+", "color": "#000000"}],
+            ams_trays={
+                "0": {"tray_uuid": "TRAY0", "tag_uid": "", "tray_type": "PLA+"},
+                "3": {"tray_uuid": "TRAY3", "tag_uid": "", "tray_type": "PLA+"},
+            },
+            # The #2768 condition: a Studio print, so print start stored nothing.
+            slot_to_tray=None,
+            tray_remain_start=None,
+            layer_usage={},
+            filament_properties={},
+            tray_now_at_start=255,
+        )
+        # The #1793 condition: AMS backup switched tray 0 -> tray 3 at layer 50.
+        state = SimpleNamespace(
+            raw_data={"ams": REPORTER_AMS},
+            tray_change_log=[(0, 0), (3, 50)],
+            total_layers=100,
+            layer_num=100,
+            tray_now=255,
+            last_loaded_tray=3,
+        )
+        spools = {
+            "TRAY0": {"id": 41, "filament": {"color_hex": "888888", "material": "PLA+"}},
+            "TRAY3": {"id": 46, "filament": {"color_hex": "111111", "material": "PLA+"}},
+        }
+        archive = SimpleNamespace(filament_color="#000000", filament_type="PLA+")
+
+        rows = iter([tracking])
+
+        def _next_row(*_args, **_kwargs):
+            result = MagicMock()
+            result.scalar_one_or_none.return_value = next(rows, archive)
+            return result
+
+        db = AsyncMock()
+        db.execute = AsyncMock(side_effect=_next_row)
+        db.delete = AsyncMock()
+        db.commit = AsyncMock()
+
+        client = AsyncMock()
+        client.find_spool_by_tag = AsyncMock(side_effect=lambda tag: spools.get(tag))
+        client.use_spool = AsyncMock()
+
+        with (
+            patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+            patch(
+                "backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
+                AsyncMock(return_value=client),
+            ),
+            patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SER")),
+            patch(
+                "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+                AsyncMock(return_value=None),
+            ),
+            patch("backend.app.services.printer_manager.printer_manager", _patched_pm(state)),
+        ):
+            await report_usage(printer_id=1, archive_id=12)
+
+        # Both segments charged, so the split path is what ran.
+        charged = {c.args[0] for c in client.use_spool.await_args_list}
+        assert charged == {41, 46}
+
+        # And the archive was rewritten from the spools the segments named,
+        # rather than being left alone as a guess would be.
+        assert archive.filament_color != "#000000"

backend/tests/unit/test_request_topic_probe_2953.py

@@ -0,0 +1,131 @@
+"""The request-topic probe must not latch on a single unexplained drop (#2953).
+
+Bambuddy subscribes to the printer's own request topic to intercept the
+``ams_mapping`` a slicer sends with a print. A1-class printers refuse: their
+broker kills the TCP connection instead of returning a SUBACK failure, so the
+only signal is "we subscribed and then got disconnected". That signal is
+circumstantial. Every other reason a connection can drop inside the same few
+seconds -- a network blip, the printer rebooting, the container being stopped
+mid-probe -- looks exactly the same.
+
+Latching on the first one costs ams_mapping capture for the rest of the
+process on a printer that supports the topic perfectly well, and every Studio
+print after that is charged to a spool picked by tray position. Requiring the
+drop to repeat costs a printer that genuinely refuses one extra reconnect.
+"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+
+@pytest.fixture(autouse=True)
+def _clear_class_state():
+    BambuMQTTClient._request_topic_cache.clear()
+    BambuMQTTClient._request_topic_probe_failures.clear()
+    yield
+    BambuMQTTClient._request_topic_cache.clear()
+    BambuMQTTClient._request_topic_probe_failures.clear()
+
+
+def _client(serial="SER2953"):
+    client = BambuMQTTClient(ip_address="10.0.0.9", serial_number=serial, access_code="12345678")
+    client._stale_reconnecting = False
+    client._last_message_time = 0.0
+    client.last_connect_error = None
+    return client
+
+
+def _mid_probe(client):
+    """Put the client where it is right after subscribing to the request topic."""
+    import time
+
+    client._request_topic_sub_mid = 7
+    client._request_topic_sub_time = time.time()
+    client._request_topic_confirmed = False
+
+
+def _drop(client):
+    client._on_disconnect(None, None, disconnect_flags=None, rc=SimpleNamespace(is_failure=True))
+
+
+def test_one_drop_keeps_the_request_topic_enabled():
+    """A blip, as far as we can tell. Try again on the next connection."""
+    client = _client()
+    _mid_probe(client)
+
+    _drop(client)
+
+    assert client._request_topic_supported is True
+    assert BambuMQTTClient._request_topic_cache.get("SER2953") is None
+    assert BambuMQTTClient._request_topic_probe_failures["SER2953"] == 1
+
+
+def test_a_second_drop_disables_it():
+    """The A1's answer: same response every time. Stop asking, and stop
+    causing a reconnect on every connection."""
+    client = _client()
+    _mid_probe(client)
+    _drop(client)
+    _mid_probe(client)
+
+    _drop(client)
+
+    assert client._request_topic_supported is False
+    assert BambuMQTTClient._request_topic_cache["SER2953"] is False
+
+
+def test_a_new_client_for_a_disabled_printer_does_not_re_probe():
+    """Unchanged: once disabled, later instances skip the subscription
+    entirely rather than reopening the reconnect loop."""
+    client = _client()
+    _mid_probe(client)
+    _drop(client)
+    _mid_probe(client)
+    _drop(client)
+
+    assert _client()._request_topic_supported is False
+
+
+def test_a_successful_suback_clears_the_count():
+    """A printer that answered once has answered. A later isolated drop must
+    start from zero, not from a half-spent budget."""
+    client = _client()
+    _mid_probe(client)
+    _drop(client)
+    assert BambuMQTTClient._request_topic_probe_failures["SER2953"] == 1
+
+    client._request_topic_sub_mid = 7
+    client._on_subscribe(None, None, 7, [SimpleNamespace(is_failure=False, value=0, getName=lambda: "ok")])
+
+    assert BambuMQTTClient._request_topic_cache["SER2953"] is True
+    assert "SER2953" not in BambuMQTTClient._request_topic_probe_failures
+
+
+def test_a_suback_rejection_still_disables_immediately():
+    """A SUBACK failure is the broker answering the question, not evidence
+    about it. One is enough."""
+    client = _client()
+    client._request_topic_sub_mid = 7
+
+    client._on_subscribe(None, None, 7, [SimpleNamespace(is_failure=True, value=135, getName=lambda: "Not authorized")])
+
+    assert client._request_topic_supported is False
+    assert BambuMQTTClient._request_topic_cache["SER2953"] is False
+
+
+def test_a_disconnect_we_asked_for_is_not_evidence():
+    """Shutting the container down mid-probe used to count against the
+    printer. ``disconnect()`` sets the event before closing the socket."""
+    import threading
+
+    client = _client()
+    _mid_probe(client)
+    client._disconnection_event = threading.Event()
+
+    _drop(client)
+
+    assert client._request_topic_supported is True
+    assert "SER2953" not in BambuMQTTClient._request_topic_probe_failures