Преглед изворни кода

fix(ams): read the firmware presence bit, not the tray state (issue #3084)

Swapping a Bambu spool for one the AMS cannot read left Assign Spool
publishing no ams_filament_setting at all. The printer kept showing "?"
on its screen and in the slicer, and only Configure, which publishes
unconditionally, put anything there.

Four places asked the tray's `state` field whether a spool was in the
slot. It cannot answer that. An AMS-HT reports its LOADED tray as 9
rather than 11, because it does not feed into a shared buffer the way a
4-slot AMS does -- the merge has skipped its own state heuristic for HT
units since #2594 for exactly this reason. And the field is partly our
own writing: apply_tray_exist_bits stamps state=9 on every slot whose
tray_exist_bits bit is 0, and when the bit comes back it refreshes only
the `exists` annotation beside it. Either way the slot sits at
exists=True, state=9 until something configures it.

That 9 also kept the deferred-configuration replay from firing -- its
own "has a spool appeared" test was the same heuristic -- which is the
deadlock #1322 removed from the assign path, still in place one step
further along. And it is what deleted the assignments in #3100: with the
replay never firing, the row kept the empty fingerprint it was stored
with, and the first tray report naming a filament was read as a swap.

All four now read tray_exist_bits first, which is the mask firmware
answers this question with and the one the printer card has drawn its
"?" from since #2527. The bit is allowed to overrule an "empty" state
and nothing else: a bit reading empty deliberately does not start
suppressing pushes that go out today, because the cost of computing a
bit position wrong is a slot that silently stops configuring, against a
saving of one message firmware would have dropped.

A blank tray report from a slot the bit calls occupied no longer unlinks
anything, off a print as well as during one, in both inventory modes --
Spoolman's parse_ams_tray calls a tray with no type empty, so a tag-less
spool assigned through the UI had its row deleted by the first idle push
after it went in. A filament the AMS cannot identify is not a filament
that was removed.
maziggy пре 4 дана
родитељ
комит
905bda4f3f

Разлика између датотеке није приказан због своје велике величине
+ 0 - 0
CHANGELOG.md


+ 30 - 1
backend/app/api/routes/inventory.py

@@ -45,6 +45,7 @@ from backend.app.schemas.spool import (
     normalize_extra_colors,
 )
 from backend.app.schemas.spool_usage import SpoolUsageHistoryResponse
+from backend.app.services.ams_slot_presence import spool_present
 from backend.app.services.location_service import (
     DUPLICATE_LOCATION_NAME,
     assign_location_name,
@@ -1850,6 +1851,9 @@ async def assign_spool(
     fingerprint_type = None
     current_tray_info_idx = ""
     tray_state: int | None = None
+    # Firmware's tray_exist_bits answer for this slot, when the payload carries
+    # one. Outranks tray_state below — see services/ams_slot_presence.py.
+    tray_has_spool: bool | None = None
     state = printer_manager.get_status(data.printer_id)
     if state and state.raw_data:
         if data.ams_id == 255:
@@ -1864,6 +1868,7 @@ async def assign_spool(
                     raw_state = vt.get("state")
                     if isinstance(raw_state, int):
                         tray_state = raw_state
+                    tray_has_spool = spool_present(vt)
                     break
         else:
             ams_data = state.raw_data.get("ams", {})
@@ -1886,6 +1891,7 @@ async def assign_spool(
                 raw_state = tray.get("state")
                 if isinstance(raw_state, int):
                     tray_state = raw_state
+                tray_has_spool = spool_present(tray)
 
     # 3. Upsert assignment (replace if same printer+ams+tray)
     existing = await db.execute(
@@ -1945,7 +1951,30 @@ async def assign_spool(
     # a doomed MQTT push when the firmware has positively confirmed "no
     # spool" — and to keep the on_ams_change replay path as the single
     # source of truth for those slots.
-    slot_is_definitely_empty = tray_state == 9 or tray_state == 10
+    #
+    # ...except that `state` cannot carry that meaning. Two independent ways
+    # a loaded slot reads 9 here:
+    #
+    #   - an AMS-HT reports its LOADED tray as 9, not 11, because it does not
+    #     feed into a shared buffer the way a 4-slot AMS does (#2594, and the
+    #     merge above skips its own state heuristic for HT units for exactly
+    #     this reason). So this branch called every HT slot empty on sight.
+    #   - apply_tray_exist_bits stamps state=9 on any slot whose tray_exist_bits
+    #     bit is 0 and never takes it back when the bit returns, so a slot that
+    #     was briefly emptied keeps the 9 until something configures it.
+    #
+    # Either way the slot sits at exists=True, state=9, this branch took the
+    # pending path, nothing was published, and the printer kept showing "?"
+    # (#3084 — reported against an H2C's AMS-HT, where both apply). Firmware's
+    # presence bit is what actually answers "is a spool in this slot", and the
+    # printer card has read it ahead of `state` since #2527.
+    #
+    # It is allowed to overrule the 9 and nothing else. A bit reading *empty*
+    # deliberately does NOT start suppressing pushes that go out today: the
+    # cost of being wrong there is a slot that silently stops configuring, on
+    # whichever AMS variant we compute the bit position wrong for, against a
+    # saving of one MQTT message the firmware would have dropped anyway.
+    slot_is_definitely_empty = tray_has_spool is not True and (tray_state == 9 or tray_state == 10)
     configured = False
     if not slot_is_definitely_empty:
         try:

+ 41 - 2
backend/app/main.py

@@ -2022,6 +2022,7 @@ async def on_ams_change(printer_id: int, ams_data: list):
             from backend.app.api.routes.inventory import _find_tray_in_ams_data
             from backend.app.models.spool import Spool as _Spool
             from backend.app.models.spool_assignment import SpoolAssignment as SA
+            from backend.app.services.ams_slot_presence import spool_present
             from backend.app.services.inventory_mode import spoolman_owns_assignments
 
             # Built-in assignments only. Since #2812 they survive a switch to
@@ -2138,7 +2139,18 @@ async def on_ams_change(printer_id: int, ams_data: list):
                     # (#1322). The state ∉ {9,10} guard keeps the firmware's
                     # explicit "empty" signals authoritative over any stale
                     # tray_type that might survive the relay's auto-clearing.
-                    loaded = cur_state == 11 or (cur_state not in (9, 10) and cur_type.strip())
+                    #
+                    # tray_exist_bits comes first because that guard cannot tell
+                    # a firmware "empty" from Bambuddy's own: apply_tray_exist_bits
+                    # writes state=9 when the bit is 0 and leaves it there when the
+                    # bit returns. A non-RFID spool inserted into a pre-assigned
+                    # slot brings no tray_type with it, so the stale 9 made this
+                    # expression false forever and the deferred config never fired
+                    # — the deadlock #1322 removed from the assign path, still in
+                    # place here (#3084, #3100).
+                    loaded = spool_present(current_tray) is True or (
+                        cur_state == 11 or (cur_state not in (9, 10) and cur_type.strip())
+                    )
                     if not fp_type.strip() and loaded and assignment.spool:
                         try:
                             from backend.app.api.routes.inventory import (
@@ -2188,6 +2200,23 @@ async def on_ams_change(printer_id: int, ams_data: list):
                                 assignment.tray_id,
                             )
                             continue
+                        # Same reasoning off the print, on firmware's own say-so:
+                        # a blank tray report from a slot whose tray_exist_bits
+                        # bit is set describes a spool the AMS cannot identify —
+                        # a non-RFID one, or one whose slot was reset — not a
+                        # spool that was taken out. Deleting the assignment there
+                        # threw away the identity the user had supplied, which is
+                        # the only place it existed (#3100). A slot the bit calls
+                        # empty, or one that carries no bit at all, still unlinks.
+                        if spool_present(current_tray) is True and not cur_color.strip() and not cur_type.strip():
+                            logger.info(
+                                "Auto-unlink skipped: spool %d AMS%d-T%d — slot still occupied, "
+                                "tray reports no filament data yet",
+                                assignment.spool_id,
+                                assignment.ams_id,
+                                assignment.tray_id,
+                            )
+                            continue
                         # Fingerprint mismatch — but check if tray now matches the
                         # assigned spool (e.g. auto-configure changed the tray).
                         # Both sides are reduced to the type the slot can carry
@@ -2594,6 +2623,7 @@ async def on_ams_change(printer_id: int, ams_data: list):
 
             from backend.app.models.spool_assignment import SpoolAssignment
             from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+            from backend.app.services.ams_slot_presence import spool_present
             from backend.app.services.inventory_mode import spoolman_owns_assignments
 
             # Built-in remaining weight, used by sync_ams_tray only when the
@@ -2662,7 +2692,16 @@ async def on_ams_change(printer_id: int, ams_data: list):
                         # completion (#1459), so deleting the row mid-print
                         # loses the runout segment's usage — the same failure
                         # the internal inventory's auto-unlink had.
-                        if not printing_now:
+                        #
+                        # Nor when firmware's presence bit says the slot is
+                        # occupied. parse_ams_tray calls a tray with no type or
+                        # no colour empty, and a spool the AMS cannot read has
+                        # neither until something configures it — so a tag-less
+                        # spool assigned through the UI had its row deleted by
+                        # the first idle push after it was inserted. Same
+                        # deletion as the internal inventory's in #3100, same
+                        # answer, so the two modes stay in step.
+                        if not printing_now and spool_present(tray_data) is not True:
                             empty_slots.append((ams_id, tray_id_raw))
                         _clear_unknown_tag_dedup(printer_id, ams_id, tray_id_raw)
                         continue

+ 60 - 0
backend/app/services/ams_slot_presence.py

@@ -0,0 +1,60 @@
+"""Is there a spool in this AMS slot?
+
+Three backend decisions turn on the answer -- whether to push
+``ams_filament_setting`` when a spool is assigned, whether a pre-assigned slot
+has just been filled, and whether an assignment has gone stale -- and all three
+used to read it off the tray's ``state`` field. That field cannot carry it.
+
+## Why ``state`` is the wrong source
+
+``state`` is firmware-variant. The A1 Mini BMCU and the P1S Standard AMS report
+3 for a loaded slot and never emit 11; the AMS-HT's codes differ again (#2670).
+Bambuddy already works around that on the printer card, where
+``getEmptySlotKind`` (``PrintersPage.tsx``) reads the presence bit first and
+only falls back to the 9/10 heuristic when there is no bit to read.
+
+Worse, ``state`` is partly Bambuddy's own writing. ``apply_tray_exist_bits``
+sets ``state = 9`` on every slot whose presence bit is 0 -- and when the bit
+comes back it leaves the 9 exactly where it was, because the "slot occupied"
+branch only annotates ``exists`` and moves on. So a slot the firmware says is
+full can sit in the cache reading ``exists=True, state=9`` indefinitely. That
+is #3084: a non-Bambu spool is swapped in, Assign Spool reads the stale 9,
+calls the slot empty, sends no MQTT, and the printer keeps showing ``?``. The
+deferred-configuration replay could not rescue it either, because its own
+"loaded" test was the same 9/10 heuristic -- the exact deadlock #1322 removed
+elsewhere. #3100 is the same stale 9 one step further on: the replay does not
+fire, the assignment keeps the empty fingerprint it was stored with, and the
+first real tray report is read as a spool swap and deleted.
+
+## What this module answers
+
+``tray_exist_bits`` is the firmware's own "which slots have a spool" bitmask --
+the one BambuStudio draws its ``?`` from -- and ``apply_tray_exist_bits``
+records it per tray as ``exists``. That bit is authoritative where it exists,
+and absent otherwise; it is never a guess. Callers that want a decision for a
+payload carrying no bit at all keep their own fallback, because the right
+fallback differs per caller: the assign path wants to know whether the push is
+doomed, the unlink pass wants to know whether a spool was removed, and a state
+of 26 ("unloaded", mid-runout) answers those two questions differently.
+"""
+
+from collections.abc import Mapping
+from typing import Any
+
+
+def spool_present(tray: Mapping[str, Any] | None) -> bool | None:
+    """Does firmware's presence bit say a spool is in this slot?
+
+    ``True`` / ``False`` straight from ``tray_exist_bits``; ``None`` when the
+    tray carries no presence annotation, which means the caller has to decide
+    on its own terms rather than assume either way.
+
+    Only the internal AMS path annotates ``exists`` (``apply_tray_exist_bits``
+    is called with ``annotate_exists=True`` there and False for the VP bridge),
+    so the external spool's ``vt_tray`` entries answer ``None`` -- they have no
+    bit in the mask.
+    """
+    if not isinstance(tray, Mapping):
+        return None
+    exists = tray.get("exists")
+    return exists if isinstance(exists, bool) else None

+ 360 - 2
backend/tests/integration/test_inventory_assign.py

@@ -1226,6 +1226,221 @@ class TestAssignSpoolEmptyDetection:
         assert body["configured"] is True
 
 
+class TestAssignSpoolPresenceBit:
+    """#3084: the slot the firmware says is full, and the cache says is empty.
+
+    ``apply_tray_exist_bits`` stamps ``state = 9`` on every slot whose
+    ``tray_exist_bits`` bit is 0 and annotates ``exists`` on every slot it
+    looks at — but when the bit comes back it only refreshes ``exists`` and
+    leaves the 9 where it was. Swap a Bambu spool for a non-RFID one and the
+    slot sits at ``exists=True, state=9`` until something configures it.
+
+    Reported on an H2D/H2C AMS-HT: remove the Bambu spool (bits ``f``), insert
+    a third-party one 9 seconds later (bits ``1000f``), then Assign Spool 28
+    seconds after that — and no ``ams_filament_setting`` was published at all.
+    Configure worked, because it publishes unconditionally.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_bit_overrules_a_stale_empty_state(
+        self, async_client: AsyncClient, printer_factory, spool_factory
+    ):
+        """exists=True with a leftover state=9 — MQTT must fire."""
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(slicer_filament="GFL05", material="PLA")
+
+        mock_client = MagicMock()
+        mock_client.ams_set_filament_setting.return_value = True
+        mock_client.extrusion_cali_sel.return_value = True
+
+        tray_data = {"id": 0, "state": 9, "exists": True, "tray_type": "", "tray_color": "", "tray_info_idx": ""}
+        status = _make_mock_status(ams_data=[{"id": 128, "tray": [tray_data]}])
+
+        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            mock_pm.get_status.return_value = status
+
+            response = await async_client.post(
+                "/api/v1/inventory/assignments",
+                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 128, "tray_id": 0},
+            )
+
+        assert response.status_code == 200
+        mock_client.ams_set_filament_setting.assert_called_once()
+        body = response.json()
+        assert body["configured"] is True
+        assert body["pending_config"] is False
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_an_empty_bit_does_not_start_suppressing_pushes(
+        self, async_client: AsyncClient, printer_factory, spool_factory
+    ):
+        """The bit overrules the 9 and nothing else.
+
+        Reading it the other way too would be tidier — skip the push firmware
+        is going to drop — but it also means a slot that silently stops
+        configuring on whichever AMS variant we compute the bit position
+        wrong for. The saving is one MQTT message; the failure is the bug
+        this commit is fixing, inverted. So a state that does not say "empty"
+        still publishes, exactly as it did before.
+        """
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(slicer_filament="GFL05", material="PLA")
+
+        mock_client = MagicMock()
+        mock_client.ams_set_filament_setting.return_value = True
+        mock_client.extrusion_cali_sel.return_value = True
+
+        tray_data = {"id": 3, "state": 11, "exists": False, "tray_type": "", "tray_color": ""}
+        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])
+
+        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            mock_pm.get_status.return_value = status
+
+            response = await async_client.post(
+                "/api/v1/inventory/assignments",
+                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
+            )
+
+        assert response.status_code == 200
+        mock_client.ams_set_filament_setting.assert_called_once()
+        assert response.json()["pending_config"] is False
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_pre_assign_workflow_still_skips_a_genuinely_empty_slot(
+        self, async_client: AsyncClient, printer_factory, spool_factory
+    ):
+        """SpoolBuddy weighs a spool and assigns it before it goes in. Bit
+        clear and state 9 agree that the slot is empty, so the push is still
+        deferred to the replay — unchanged."""
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(slicer_filament="GFL05", material="PLA")
+
+        mock_client = MagicMock()
+        mock_client.ams_set_filament_setting.return_value = True
+
+        tray_data = {"id": 3, "state": 9, "exists": False, "tray_type": "", "tray_color": ""}
+        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])
+
+        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            mock_pm.get_status.return_value = status
+
+            response = await async_client.post(
+                "/api/v1/inventory/assignments",
+                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
+            )
+
+        assert response.status_code == 200
+        mock_client.ams_set_filament_setting.assert_not_called()
+        body = response.json()
+        assert body["configured"] is False
+        assert body["pending_config"] is True
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_an_unannotated_tray_still_reads_the_state(
+        self, async_client: AsyncClient, printer_factory, spool_factory
+    ):
+        """No presence bit in the payload → the 9/10 heuristic still decides.
+
+        The external spool's ``vt_tray`` has no bit in the mask, and neither do
+        the hand-built payloads every other test in this file uses.
+        """
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(slicer_filament="GFL05", material="PLA")
+
+        mock_client = MagicMock()
+        mock_client.ams_set_filament_setting.return_value = True
+
+        status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "state": 9, "tray_type": ""}]}])
+
+        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
+            mock_pm.get_client.return_value = mock_client
+            mock_pm.get_status.return_value = status
+
+            response = await async_client.post(
+                "/api/v1/inventory/assignments",
+                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
+            )
+
+        assert response.status_code == 200
+        mock_client.ams_set_filament_setting.assert_not_called()
+        assert response.json()["pending_config"] is True
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_deferred_config_fires_for_a_spool_the_ams_cannot_name(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """The pre-assign workflow's half of the same bug.
+
+        A non-RFID spool inserted into a pre-assigned slot brings no
+        ``tray_type`` with it, and the stale 9 kept the replay's "loaded" test
+        false, so the deferred configuration never fired for it either. The
+        presence bit is the only thing in the payload that changed.
+        """
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(slicer_filament="GFL05", material="PLA")
+
+        pre_assignment = SpoolAssignment(
+            spool_id=spool.id,
+            printer_id=printer.id,
+            ams_id=2,
+            tray_id=3,
+            fingerprint_color=None,
+            fingerprint_type=None,
+        )
+        db_session.add(pre_assignment)
+        await db_session.commit()
+
+        ams_data = [{"id": 2, "tray": [{"id": 3, "state": 9, "exists": True, "tray_type": "", "tray_color": ""}]}]
+
+        mock_client = MagicMock()
+        mock_client.ams_set_filament_setting.return_value = True
+        mock_client.extrusion_cali_sel.return_value = True
+
+        status = _make_mock_status(ams_data=ams_data)
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = status
+            mock_pm_main.get_client.return_value = mock_client
+            mock_pm_main.get_model.return_value = "H2D"
+            mock_pm_inv.get_client.return_value = mock_client
+            mock_pm_inv.get_status.return_value = status
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer.id, ams_data)
+
+        mock_client.ams_set_filament_setting.assert_called_once()
+        call_kwargs = mock_client.ams_set_filament_setting.call_args.kwargs
+        assert call_kwargs["ams_id"] == 2
+        assert call_kwargs["tray_id"] == 3
+        assert call_kwargs["tray_info_idx"] == "GFL05"
+
+        # The assignment is still there — the pass that fires the config is the
+        # same pass that deletes stale ones (#3100).
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, pre_assignment.id) is not None
+
+
 class TestAssignSpoolPfcnCloudPreset:
     """Assign path for PFCN-prefix cloud presets (#1648).
 
@@ -1561,6 +1776,118 @@ class TestAutoUnlinkDuringRunout:
         assert await db_session.get(SpoolAssignment, assignment_id) is not None
 
 
+class TestAutoUnlinkOccupiedSlot:
+    """#3100: six saved assignments deleted across three X1 Carbons.
+
+    Each one had an explicit earlier assignment and then an ``Auto-unlink ...
+    fingerprint mismatch``, and the reporter recovered the mappings from logs
+    because they were gone from the inventory API, not merely hidden. Two
+    shapes, one cause: a slot the presence bit calls occupied while the tray
+    reports nothing about what is in it.
+
+    ``cur=/ fp=BCBCBCFF/PLA spool=8A8F92FF/PLA`` — an established assignment,
+    a blank idle report, and the row deleted. The spool never went anywhere;
+    the AMS simply had nothing to say about a filament it cannot read.
+    """
+
+    @staticmethod
+    async def _run(printer_id, ams_data, status):
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="X1C", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = status
+            mock_pm_main.get_model.return_value = "X1C"
+            mock_pm_main.get_client.return_value = None
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer_id, ams_data)
+
+    @staticmethod
+    async def _assignment(db_session, printer, spool):
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        assignment = SpoolAssignment(
+            spool_id=spool.id,
+            printer_id=printer.id,
+            ams_id=1,
+            tray_id=1,
+            fingerprint_color="BCBCBCFF",
+            fingerprint_type="PLA",
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        return assignment.id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_blank_report_from_an_occupied_slot_keeps_the_assignment(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
+        assignment_id = await self._assignment(db_session, printer, spool)
+
+        ams_data = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
+        await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, assignment_id) is not None, (
+            "a spool the AMS cannot identify is not a spool that was removed"
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_blank_report_from_an_empty_slot_still_unlinks(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """The guard reads the bit, not the blankness — take the spool out and
+        the assignment still goes, which is what makes the test above a
+        distinction rather than a blanket reprieve."""
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
+        assignment_id = await self._assignment(db_session, printer, spool)
+
+        ams_data = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
+        await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, assignment_id) is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_genuinely_different_filament_in_an_occupied_slot_still_unlinks(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """The guard is for a blank report only. A slot that names a filament
+        which is not the assigned spool is a swap, bit set or not."""
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
+        assignment_id = await self._assignment(db_session, printer, spool)
+
+        ams_data = [
+            {"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "PETG", "tray_color": "00FF00FF", "state": 11}]}
+        ]
+        await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, assignment_id) is None
+
+
 class TestSpoolmanSlotAssignmentDuringRunout:
     """`spoolman_slot_assignments` is how a tag-less spool assigned through the
     Bambuddy UI is resolved at completion (#1459). Deleting the row when a slot
@@ -1578,14 +1905,14 @@ class TestSpoolmanSlotAssignmentDuringRunout:
             db_session.add(Settings(key=key, value=value))
         await db_session.commit()
 
-    async def _run(self, printer_id, state):
+    async def _run(self, printer_id, state, tray=None):
         from unittest.mock import AsyncMock
 
         from backend.app.main import on_ams_change
 
         # A tray the firmware has cleared: parse_ams_tray returns None, which
         # is what marks the slot empty for the cleanup pass.
-        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]
+        ams_data = [{"id": 0, "tray": [tray or {"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]
 
         spoolman_client = MagicMock()
         spoolman_client.health_check = AsyncMock(return_value=True)
@@ -1628,6 +1955,37 @@ class TestSpoolmanSlotAssignmentDuringRunout:
         db_session.expunge_all()
         assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_an_occupied_slot_the_ams_cannot_read_keeps_its_row(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        """Spoolman mode's half of #3100.
+
+        parse_ams_tray calls a tray with no type empty, and a tag-less spool
+        has none until something configures it — so the row assigned through
+        the UI was deleted by the first idle push after the spool went in.
+        Firmware's presence bit is the same answer the built-in inventory
+        uses, so the two modes stay in step.
+        """
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+
+        await self._enable_spoolman(db_session)
+        printer = await printer_factory(name="H2D")
+        row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41)
+        db_session.add(row)
+        await db_session.commit()
+        row_id = row.id
+
+        await self._run(
+            printer.id,
+            _make_printing_status(None, state="IDLE"),
+            tray={"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9},
+        )
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_the_slot_row_is_still_cleaned_up_when_idle(

+ 45 - 0
backend/tests/unit/services/test_ams_slot_presence.py

@@ -0,0 +1,45 @@
+"""Unit tests for the firmware presence bit helper.
+
+#3084: a slot can read ``exists=True, state=9`` — the bit says a spool is in
+it, the state says the opposite — because ``apply_tray_exist_bits`` writes the
+9 itself and never takes it back. Everything downstream has to know which of
+the two to believe, and this helper is the single place that says so.
+"""
+
+from backend.app.services.ams_slot_presence import spool_present
+
+
+class TestSpoolPresent:
+    def test_the_bit_is_reported_as_it_stands(self):
+        assert spool_present({"id": 0, "exists": True}) is True
+        assert spool_present({"id": 0, "exists": False}) is False
+
+    def test_the_bit_answers_over_a_contradicting_state(self):
+        # The #3084 slot: non-Bambu spool swapped in, so the bit is set, while
+        # the 9 Bambuddy stamped on the slot when it was briefly empty is still
+        # sitting there.
+        assert spool_present({"id": 0, "exists": True, "state": 9}) is True
+        # And the converse — a spool pulled from a slot the firmware last
+        # described as loaded.
+        assert spool_present({"id": 0, "exists": False, "state": 11}) is False
+
+    def test_a_tray_with_no_annotation_answers_nothing(self):
+        # vt_tray entries and the VP bridge's cache carry no presence bit, so
+        # callers have to fall back to their own reading rather than be handed
+        # a guess dressed up as firmware's answer.
+        assert spool_present({"id": 0, "state": 11, "tray_type": "PLA"}) is None
+        assert spool_present({}) is None
+
+    def test_a_non_bool_exists_is_not_a_presence_bit(self):
+        # Only apply_tray_exist_bits writes this key, and it writes a bool.
+        # Anything else reached the dict some other way and is not firmware's
+        # answer — 0 and "" would otherwise read as a confident "empty".
+        assert spool_present({"exists": 0}) is None
+        assert spool_present({"exists": ""}) is None
+        assert spool_present({"exists": "true"}) is None
+        assert spool_present({"exists": None}) is None
+
+    def test_a_missing_tray_answers_nothing(self):
+        assert spool_present(None) is None
+        assert spool_present([]) is None
+        assert spool_present("tray") is None

Неке датотеке нису приказане због велике количине промена