diff --git a/backend/app/services/bambu_mqtt.py b/backend/app/services/bambu_mqtt.py
index 51707ae39..5577221cc 100644
--- a/backend/app/services/bambu_mqtt.py
+++ b/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
 
diff --git a/backend/app/services/spoolman_tracking.py b/backend/app/services/spoolman_tracking.py
index fe87f0097..b752095e8 100644
--- a/backend/app/services/spoolman_tracking.py
+++ b/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"
+    # 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"
+        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(
diff --git a/backend/tests/unit/services/test_bambu_mqtt.py b/backend/tests/unit/services/test_bambu_mqtt.py
index 5619ac3e8..5bee3f04f 100644
--- a/backend/tests/unit/services/test_bambu_mqtt.py
+++ b/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
diff --git a/backend/tests/unit/services/test_spoolman_tray_state_fallback_2953.py b/backend/tests/unit/services/test_spoolman_tray_state_fallback_2953.py
new file mode 100644
index 000000000..523364f30
--- /dev/null
+++ b/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"
diff --git a/backend/tests/unit/test_request_topic_probe_2953.py b/backend/tests/unit/test_request_topic_probe_2953.py
new file mode 100644
index 000000000..b3527d5ec
--- /dev/null
+++ b/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