Hold an automatic slot unlink until the slot stays empty (issue #3186)

CHANGELOG.md

@@ -46,6 +46,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **A slot that reads empty for a moment no longer loses its spool assignment (#3186, reported by @Sawtaytoes)** — An idle X1 Carbon sent one status update that showed a whole AMS unit as empty, with no colour or material in any slot, and Bambuddy deleted all four spool assignments on it at once. The spools never moved, and the next update reported them again. Nothing brought the assignments back, and for a non-RFID spool the assignment is the only record of which spool is in the slot. The #3100 fix covered a blank slot the AMS still reported as occupied, but not one briefly reported as empty. A slot that looks empty, or that drops out of the AMS data, now keeps its assignment for two minutes and loses it only if it is still empty then. Bambuddy checks again by itself when the two minutes are up, so a spool that was really taken out does not wait for the next AMS change. A different spool the AMS can identify, by its RFID tag or its colour and material, still releases the old assignment immediately. A spool it cannot read that goes in during those two minutes releases it once they are up, rather than inheriting the old spool's assignment. Assigning a spool to the slot yourself cancels the wait. Spoolman mode's slot links follow the same rules.
 - **Number fields can be cleared and retyped (#3182, reported by @Carter3DP)** — Every number field corrected its value on each keystroke, so erasing the "1" in the print dialog's Quantity snapped straight back to 1, and getting to 6 meant typing 16 and deleting the 1. It was worse where the minimum is above 1: typing the "6" of 60 into the AMS drying temperature turned it into 45, so the field could only be set with the arrows. Fields now keep what you type while you edit, use it once it is a number in range, and settle on leaving the field: out-of-range numbers are pulled into range, and an empty field goes back to its default. This covers the 38 number fields across the print dialog, per-plate quantities, scheduling, drying, Settings, smart plugs, backups, spools, projects and SpoolBuddy. Two fields could not be set to 0 even though 0 is allowed, the smart plug's delay after drying and AMS humidity; they can now. The drying presets now also keep to their own limits, 30–65 °C on an AMS 2 Pro and 30–85 °C on an AMS HT.
 - **Installing or updating no longer runs out of memory on 2 GB machines (#3181, reported by @PhilippeP62)** — Every installer and updater builds the frontend with `npm run build`, which also ran the TypeScript type check. That check alone needs about 1 GB of Node memory, the whole default allowance on a 2 GB machine such as the standard Proxmox LXC or a 2 GB Raspberry Pi, so the build crashed with "JavaScript heap out of memory". On the Proxmox helper script that left the install without its database and nobody could sign in; Bambuddy's own update script rolled back but could never finish an update. The build now only bundles the frontend, and the type check runs on its own in development and CI (`npm run typecheck`). That also fixes CI's type-check job and the pre-commit hook, which ran `tsc --noEmit` against a config that lists no files and so had been passing without checking anything.
 - **Remember Me is back on the login page when only SSO sign-in is allowed (#2784, reported and contributed by @vuthanhtrung2010 in #3117)** — Turning off local username/password login hid the whole sign-in form, and the Remember Me checkbox went with it. An SSO sign-in could then only ever last until the browser closed, so SSO users signed in again every time they came back. The checkbox now sits above the SSO buttons when local login is off, and the "or continue with" divider, which had nothing left to divide, is hidden. Signing in with Remember Me unticked now also clears a choice left over from an abandoned attempt; before, ticking it, backing out of the sign-in provider and then signing in unticked still kept you signed in. With automatic SSO sign-in configured, Bambuddy now also remembers the page you were trying to open and returns you there after signing in, instead of to the dashboard. Automatic sign-in redirects before the page can show the checkbox, so those sign-ins still last only until the browser closes.

backend/app/api/routes/inventory.py

@@ -55,6 +55,7 @@ from backend.app.schemas.supplier import (
     SupplierStats,
     SupplierUpdate,
 )
+from backend.app.services import slot_unlink_grace
 from backend.app.services.ams_slot_presence import spool_present
 from backend.app.services.location_service import (
     DUPLICATE_LOCATION_NAME,
@@ -2195,6 +2196,7 @@ async def assign_spool(
     db.add(assignment)
     await db.commit()
     await db.refresh(assignment)
+    slot_unlink_grace.forget_slot(data.printer_id, data.ams_id, data.tray_id)
 
     # 4. Auto-configure AMS slot via MQTT.
     #

backend/app/api/routes/spoolman.py

@@ -21,6 +21,7 @@ from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spoolman_k_profile import SpoolmanKProfile
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
 from backend.app.models.user import User
+from backend.app.services import slot_unlink_grace
 from backend.app.services.printer_manager import printer_manager
 from backend.app.services.slicer_filament_resolver import resolve_slicer_filament
 from backend.app.services.slot_nozzle import resolve_slot_nozzle
@@ -896,6 +897,7 @@ async def link_spool(
                 {"printer_id": p_id, "ams_id": a_id, "tray_id": t_id, "spool_id": spool_id},
             )
             await db.commit()
+            slot_unlink_grace.forget_slot(p_id, a_id, t_id)
         except Exception as e:
             await db.rollback()
             logger.error(
@@ -1318,6 +1320,7 @@ async def create_spool_from_slot(
                 },
             )
             await db.commit()
+            slot_unlink_grace.forget_slot(req.printer_id, req.ams_id, req.tray_id)
         except Exception as exc:
             await db.rollback()
             logger.exception("Failed to persist Spoolman slot assignment")

backend/app/api/routes/spoolman_inventory.py

@@ -46,6 +46,7 @@ from backend.app.models.user import User
 from backend.app.schemas.spool import SpoolFilamentPresetBase, SpoolKProfileBase
 from backend.app.schemas.spoolman import SpoolmanFilamentPatch, SpoolmanSlotAssignmentEnriched
 from backend.app.schemas.supplier import SpoolSupplierLinkInput
+from backend.app.services import slot_unlink_grace
 from backend.app.services.location_service import (
     enrich_spool_dicts_with_location_id,
     maybe_sync_spoolman_locations,
@@ -1539,6 +1540,7 @@ async def assign_spoolman_slot(
             },
         )
         await db.commit()
+        slot_unlink_grace.forget_slot(body.printer_id, body.ams_id, body.tray_id)
     except Exception as exc:
         await db.rollback()
         logger.error("Failed to persist slot assignment: %s", exc)

backend/app/main.py

@@ -87,7 +87,7 @@ from backend.app.core.config import APP_VERSION, settings as app_settings
 from backend.app.core.database import async_session, engine, init_db
 from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
-from backend.app.services import print_dispatch_context
+from backend.app.services import print_dispatch_context, slot_unlink_grace
 from backend.app.services.archive import ArchiveService, peek_plate_index_in_3mf, swap_plate_suffix
 from backend.app.services.archive_purge import archive_purge_service
 from backend.app.services.bambu_ftp import (
@@ -2068,56 +2068,22 @@ async def on_fts_inlet_change(printer_id: int, ams_id: int, inlet: str):
         logger.warning("[Printer %s] Could not re-apply K-profiles after inlet move: %s", printer_id, e)
 
 
-async def on_ams_change(printer_id: int, ams_data: list):
-    """Handle AMS data changes - sync to Spoolman if enabled and auto mode."""
-    logger = logging.getLogger(__name__)
+async def _unlink_stale_assignments(printer_id: int, ams_data: list, printing_now: bool) -> None:
+    """Unlink built-in inventory assignments whose slot no longer holds their spool.
 
-    # Snapshot BEFORE any await: if a print is active, skip weight sync later.
-    # on_print_complete may pop _active_sessions during our awaits (#880).
-    from backend.app.services.usage_tracker import _active_sessions
-
-    _print_active = printer_id in _active_sessions
-
-    # A slot that reports empty while a print is running is a filament runout,
-    # not a spool swap: the spool is still physically in the AMS, just
-    # consumed. Dropping either inventory backend's slot link there loses the
-    # only record of which spool fed the print, so the completion path can't
-    # charge the runout segment to anything. Both cleanup passes below consult
-    # this; computed once, up front, so neither depends on the other having run.
-    _unlink_state = printer_manager.get_status(printer_id)
-    printing_now = (getattr(_unlink_state, "state", "") or "").upper() in ("RUNNING", "PAUSE")
-
-    # MQTT relay - publish AMS change
-    try:
-        printer_info = printer_manager.get_printer(printer_id)
-        if printer_info:
-            await mqtt_relay.on_ams_change(printer_id, printer_info.name, printer_info.serial_number, ams_data)
-    except Exception:
-        pass  # Don't fail AMS callback if MQTT fails
-
-    # Broadcast AMS change via WebSocket (bypasses status_key deduplication)
-    # This ensures frontend gets immediate updates when AMS slots are configured
-    try:
-        state = printer_manager.get_status(printer_id)
-        if state:
-            logger.info("[Printer %s] Broadcasting AMS change via WebSocket", printer_id)
-            await ws_manager.send_printer_status(
-                printer_id,
-                printer_state_to_dict(
-                    state,
-                    printer_id,
-                    printer_manager.get_model(printer_id),
-                    printer_manager.get_drying_targets(printer_id),
-                ),
-            )
-    except Exception as e:
-        logger.warning("Failed to broadcast AMS change for printer %s: %s", printer_id, e)
+    Runs from ``on_ams_change`` and, for removals held by ``slot_unlink_grace``,
+    from the delayed re-check -- which is why it takes the AMS data and print
+    state as arguments instead of reading them from the push (#3186).
+    """
+    logger = logging.getLogger(__name__)
 
     from backend.app.utils.color_utils import colors_similar as _colors_similar
 
-    # Auto-unlink spool assignments with stale fingerprints
+    # Auto-unlink spool assignments with stale fingerprints. Under the
+    # per-printer assignment lock since #3186: the held-removal re-check runs
+    # this outside any MQTT push, so it can now overlap one.
     try:
-        async with async_session() as db:
+        async with _get_ams_assignment_lock(printer_id), async_session() as db:
             from sqlalchemy.orm import selectinload
 
             from backend.app.api.routes.inventory import _find_tray_in_ams_data
@@ -2143,6 +2109,8 @@ async def on_ams_change(printer_id: int, ams_data: list):
             # unlinking the spool that fed the print — the next idle-time pass
             # unlinks it if the user really did take it out.
             stale = []
+            # Removals this pass is holding rather than unlinking (#3186).
+            held: set[tuple] = set()
             for assignment in assignments:
                 # External spool assignments (ams_id=255) live in vt_tray, not AMS data
                 if assignment.ams_id == 255:
@@ -2168,6 +2136,20 @@ async def on_ams_change(printer_id: int, ams_data: list):
                             assignment.tray_id,
                         )
                         continue
+                    # A whole AMS unit can drop out of one push and come back in
+                    # the next; only a slot that stays gone is a removal (#3186).
+                    hold_key = ("inventory", assignment.ams_id, assignment.tray_id, assignment.spool_id)
+                    if not slot_unlink_grace.removal_confirmed(printer_id, hold_key):
+                        held.add(hold_key)
+                        logger.info(
+                            "Auto-unlink held: spool %d AMS%d-T%d — tray not found in AMS data; "
+                            "unlinking if it is still gone in %ds",
+                            assignment.spool_id,
+                            assignment.ams_id,
+                            assignment.tray_id,
+                            int(slot_unlink_grace.GRACE_SECONDS),
+                        )
+                        continue
                     logger.info(
                         "Auto-unlink: spool %d AMS%d-T%d — tray not found in AMS data (slot empty?)",
                         assignment.spool_id,
@@ -2309,7 +2291,20 @@ async def on_ams_change(printer_id: int, ams_data: list):
                         # 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():
+                        #
+                        # Unless the slot was reported empty first: a removal
+                        # already held means a spool came out and another went
+                        # in, so the blank report keeps the hold running below
+                        # rather than cancelling it (#3186).
+                        if (
+                            spool_present(current_tray) is True
+                            and not cur_color.strip()
+                            and not cur_type.strip()
+                            and not slot_unlink_grace.is_held(
+                                printer_id,
+                                ("inventory", assignment.ams_id, assignment.tray_id, assignment.spool_id),
+                            )
+                        ):
                             logger.info(
                                 "Auto-unlink skipped: spool %d AMS%d-T%d — slot still occupied, "
                                 "tray reports no filament data yet",
@@ -2318,6 +2313,24 @@ async def on_ams_change(printer_id: int, ams_data: list):
                                 assignment.tray_id,
                             )
                             continue
+                        # A blank report the presence bit does not vouch for is
+                        # still only one push. An idle X1C cleared a whole AMS
+                        # unit's bits, colour and type for a moment and lost
+                        # four saved assignments that way (#3186); unlink only
+                        # if the slot is still blank after the grace period.
+                        if not cur_color.strip() and not cur_type.strip():
+                            hold_key = ("inventory", assignment.ams_id, assignment.tray_id, assignment.spool_id)
+                            if not slot_unlink_grace.removal_confirmed(printer_id, hold_key):
+                                held.add(hold_key)
+                                logger.info(
+                                    "Auto-unlink held: spool %d AMS%d-T%d — tray reports no filament data; "
+                                    "unlinking if it is still blank in %ds",
+                                    assignment.spool_id,
+                                    assignment.ams_id,
+                                    assignment.tray_id,
+                                    int(slot_unlink_grace.GRACE_SECONDS),
+                                )
+                                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
@@ -2390,6 +2403,7 @@ async def on_ams_change(printer_id: int, ams_data: list):
                             spool.material if spool else "?",
                         )
                         stale.append(assignment)  # Spool changed
+            slot_unlink_grace.settle(printer_id, "inventory", held)
             # Snapshot slots before delete — ORM attribute access after the
             # commit would refresh against a deleted row.
             unlinked_slots = [(a.ams_id, a.tray_id) for a in stale]
@@ -2417,6 +2431,161 @@ async def on_ams_change(printer_id: int, ams_data: list):
     except Exception as e:
         logger.warning("Spool assignment cleanup failed: %s", e, exc_info=True)
 
+
+async def _expire_spoolman_empty_slots(printer_id: int, ams_data: list, printing_now: bool) -> None:
+    """Delete Spoolman slot rows whose held removal has run its grace period.
+
+    The Spoolman half of the #3186 re-check. The full sync in ``on_ams_change``
+    talks to Spoolman for every tray; this only needs the local rows, so it
+    repeats that pass's empty-slot decision -- a tray with no type or no colour
+    (``parse_ams_tray`` returns None for exactly those), not during a print, and
+    not in a slot the presence bit calls occupied -- and nothing else.
+    """
+    logger = logging.getLogger(__name__)
+    try:
+        async with async_session() as db:
+            from backend.app.api.routes.settings import get_setting
+            from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+            from backend.app.services.ams_slot_presence import spool_present
+
+            enabled = await get_setting(db, "spoolman_enabled")
+            if not enabled or enabled.lower() != "true":
+                return
+            sync_mode = await get_setting(db, "spoolman_sync_mode")
+            if sync_mode and sync_mode != "auto":
+                return
+
+            trays: dict[tuple[int, int], dict] = {}
+            for ams_unit in ams_data or []:
+                if not isinstance(ams_unit, dict):
+                    continue
+                for tray in ams_unit.get("tray", []):
+                    if isinstance(tray, dict):
+                        trays[(int(ams_unit.get("id", 0)), int(tray.get("id", 0)))] = tray
+
+            rows = (
+                (
+                    await db.execute(
+                        select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == printer_id)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            held: set[tuple] = set()
+            expired: list[tuple[int, int]] = []
+            for row in rows:
+                tray = trays.get((row.ams_id, row.tray_id))
+                if tray is None or printing_now:
+                    continue
+                if (tray.get("tray_type") or "").strip() and (tray.get("tray_color") or "").strip():
+                    continue
+                hold_key = ("spoolman", row.ams_id, row.tray_id, row.spoolman_spool_id)
+                if spool_present(tray) is True and not slot_unlink_grace.is_held(printer_id, hold_key):
+                    continue
+                if slot_unlink_grace.removal_confirmed(printer_id, hold_key):
+                    expired.append((row.ams_id, row.tray_id))
+                else:
+                    held.add(hold_key)
+            slot_unlink_grace.settle(printer_id, "spoolman", held)
+            if not expired:
+                return
+            # A statement rather than ORM deletes, like the sync pass: the two
+            # can overlap, and a row the other already removed must not fail
+            # this commit.
+            for ams_id, tray_id in expired:
+                await db.execute(
+                    delete(SpoolmanSlotAssignment).where(
+                        SpoolmanSlotAssignment.printer_id == printer_id,
+                        SpoolmanSlotAssignment.ams_id == ams_id,
+                        SpoolmanSlotAssignment.tray_id == tray_id,
+                    )
+                )
+            await db.commit()
+            logger.info("Unlinked %d Spoolman slot(s) that stayed empty for printer %d", len(expired), printer_id)
+            for ams_id, tray_id in expired:
+                await ws_manager.broadcast(
+                    {
+                        "type": "spool_assignment_changed",
+                        "printer_id": printer_id,
+                        "ams_id": ams_id,
+                        "tray_id": tray_id,
+                    }
+                )
+    except Exception as e:
+        logger.warning("Spoolman slot re-check failed for printer %s: %s", printer_id, e, exc_info=True)
+
+
+async def _recheck_held_unlinks(printer_id: int) -> None:
+    """Re-run just the slot cleanup against the printer's current AMS state.
+
+    Scheduled by ``slot_unlink_grace`` when it holds a removal. The unlink
+    passes otherwise run only when the AMS hash changes, and a slot that went
+    empty and stayed empty may never change it again.
+    """
+    status = printer_manager.get_status(printer_id)
+    # A disconnected printer's state is its last report, not a new one: acting
+    # on it would "confirm" a removal nobody has seen for the whole grace
+    # period. Leave the holds; the first push after reconnecting decides.
+    if status is None or not status.connected:
+        return
+    ams_raw = status.raw_data.get("ams")
+    ams_data = ams_raw.get("ams", []) if isinstance(ams_raw, dict) else ams_raw if isinstance(ams_raw, list) else []
+    printing_now = (getattr(status, "state", "") or "").upper() in ("RUNNING", "PAUSE")
+    await _unlink_stale_assignments(printer_id, ams_data, printing_now)
+    await _expire_spoolman_empty_slots(printer_id, ams_data, printing_now)
+
+
+slot_unlink_grace.set_recheck(_recheck_held_unlinks)
+
+
+async def on_ams_change(printer_id: int, ams_data: list):
+    """Handle AMS data changes - sync to Spoolman if enabled and auto mode."""
+    logger = logging.getLogger(__name__)
+
+    # Snapshot BEFORE any await: if a print is active, skip weight sync later.
+    # on_print_complete may pop _active_sessions during our awaits (#880).
+    from backend.app.services.usage_tracker import _active_sessions
+
+    _print_active = printer_id in _active_sessions
+
+    # A slot that reports empty while a print is running is a filament runout,
+    # not a spool swap: the spool is still physically in the AMS, just
+    # consumed. Dropping either inventory backend's slot link there loses the
+    # only record of which spool fed the print, so the completion path can't
+    # charge the runout segment to anything. Both cleanup passes below consult
+    # this; computed once, up front, so neither depends on the other having run.
+    _unlink_state = printer_manager.get_status(printer_id)
+    printing_now = (getattr(_unlink_state, "state", "") or "").upper() in ("RUNNING", "PAUSE")
+
+    # MQTT relay - publish AMS change
+    try:
+        printer_info = printer_manager.get_printer(printer_id)
+        if printer_info:
+            await mqtt_relay.on_ams_change(printer_id, printer_info.name, printer_info.serial_number, ams_data)
+    except Exception:
+        pass  # Don't fail AMS callback if MQTT fails
+
+    # Broadcast AMS change via WebSocket (bypasses status_key deduplication)
+    # This ensures frontend gets immediate updates when AMS slots are configured
+    try:
+        state = printer_manager.get_status(printer_id)
+        if state:
+            logger.info("[Printer %s] Broadcasting AMS change via WebSocket", printer_id)
+            await ws_manager.send_printer_status(
+                printer_id,
+                printer_state_to_dict(
+                    state,
+                    printer_id,
+                    printer_manager.get_model(printer_id),
+                    printer_manager.get_drying_targets(printer_id),
+                ),
+            )
+    except Exception as e:
+        logger.warning("Failed to broadcast AMS change for printer %s: %s", printer_id, e)
+
+    await _unlink_stale_assignments(printer_id, ams_data, printing_now)
+
     # Auto-manage inventory spools from AMS tray data (skip if Spoolman manages AMS).
     # Serialised per-printer via _ams_assignment_locks: MQTT bursts can deliver
     # two AMS pushes ~30 ms apart, and without the lock both callbacks read
@@ -2426,6 +2595,8 @@ async def on_ams_change(printer_id: int, ams_data: list):
     # bug stayed latent there. See _ams_assignment_locks comment for details.
     try:
         async with _get_ams_assignment_lock(printer_id), async_session() as db:
+            from sqlalchemy.orm import selectinload
+
             from backend.app.api.routes.settings import get_setting
             from backend.app.models.spool import Spool
             from backend.app.models.spool_assignment import SpoolAssignment as SA
@@ -2770,6 +2941,7 @@ async def on_ams_change(printer_id: int, ams_data: list):
             synced = 0
             slot_changes: list[tuple[int, int, int]] = []  # (ams_id, tray_id, spoolman_spool_id) to upsert
             empty_slots: list[tuple[int, int]] = []  # (ams_id, tray_id) whose tray is now empty
+            spoolman_held: set[tuple] = set()  # removals held for the grace period (#3186)
             for ams_unit in ams_data:
                 if not isinstance(ams_unit, dict):
                     continue
@@ -2802,8 +2974,27 @@ async def on_ams_change(printer_id: int, ams_data: list):
                         # 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))
+                        #
+                        # And only once the slot has stayed empty for the grace
+                        # period -- the internal inventory's #3186 answer, for
+                        # the same one-push blank.
+                        linked_spool = spoolman_slot_map.get((ams_id, tray_id_raw))
+                        hold_key = ("spoolman", ams_id, tray_id_raw, linked_spool)
+                        if not printing_now and (
+                            spool_present(tray_data) is not True or slot_unlink_grace.is_held(printer_id, hold_key)
+                        ):
+                            if linked_spool is None or slot_unlink_grace.removal_confirmed(printer_id, hold_key):
+                                empty_slots.append((ams_id, tray_id_raw))
+                            else:
+                                spoolman_held.add(hold_key)
+                                logger.info(
+                                    "Spoolman slot unlink held: AMS%d-T%d (spool %d) reports empty; "
+                                    "unlinking if it is still empty in %ds",
+                                    ams_id,
+                                    tray_id_raw,
+                                    linked_spool,
+                                    int(slot_unlink_grace.GRACE_SECONDS),
+                                )
                         _clear_unknown_tag_dedup(printer_id, ams_id, tray_id_raw)
                         continue
 
@@ -2881,6 +3072,8 @@ async def on_ams_change(printer_id: int, ams_data: list):
                     except Exception as e:
                         logger.error("Error syncing AMS %s tray %s: %s", ams_id, tray.tray_id, e)
 
+            slot_unlink_grace.settle(printer_id, "spoolman", spoolman_held)
+
             if synced > 0:
                 logger.info("Auto-synced %s AMS trays to Spoolman for printer %s", synced, printer_id)
 
@@ -9766,6 +9959,7 @@ async def lifespan(app: FastAPI):
 
     await stop_printer_download_cleanup()
     printer_manager.disconnect_all()
+    slot_unlink_grace.reset()
     await close_spoolman_client()
 
     # Stop all virtual printer services

backend/app/services/slot_unlink_grace.py

@@ -0,0 +1,144 @@
+"""Hold an automatic slot unlink until "the spool is gone" has lasted.
+
+Both inventory backends unlink a slot's spool when the AMS says the slot is
+empty: the built-in inventory deletes the ``spool_assignment`` row, Spoolman
+mode deletes the ``spoolman_slot_assignments`` row. Until #3186 one MQTT push
+was enough. An idle X1 Carbon on X1Plus firmware sent a push that cleared an
+entire AMS unit -- presence bits off, colour and type blank -- and four saved
+assignments were deleted in the same instant. The spools never moved. When
+the next push reported them again, nothing brought the rows back, and the
+identity of a non-RFID spool exists nowhere but in that row.
+
+A spool that really was taken out stays out, so the evidence is cheap to
+confirm: note when a slot first looks empty, keep the row, and unlink only if
+it still looks empty ``GRACE_SECONDS`` later. A slot that reads normally again
+in between is forgotten. Evidence of a *different* spool -- another colour or
+type, another Bambu tag -- is not held here; callers unlink that immediately.
+
+The unlink passes run only when the AMS hash changes, so a slot that goes
+empty and stays empty might never be looked at again. Holding a removal
+therefore also schedules one re-check per printer, which re-runs just the
+cleanup passes against the printer's current AMS state.
+"""
+
+import asyncio
+import logging
+import time
+from collections.abc import Awaitable, Callable
+
+logger = logging.getLogger(__name__)
+
+GRACE_SECONDS = 120.0
+
+# A hold that nobody re-observed for this long is out of date -- the pass that
+# would have seen it did not run (Spoolman unreachable, printer offline) -- so
+# the next sighting starts the clock again instead of confirming at once.
+_STALE_AFTER = 2 * GRACE_SECONDS
+
+# (printer_id, scope, ams_id, tray_id, spool_id) -> (first_seen, last_seen)
+_held: dict[tuple, tuple[float, float]] = {}
+_recheck_tasks: dict[int, asyncio.Task] = {}
+_recheck: Callable[[int], Awaitable[None]] | None = None
+
+# Indirection so tests can move the clock without touching time.monotonic
+# itself, which the event loop reads too.
+_now = time.monotonic
+
+
+def set_recheck(callback: Callable[[int], Awaitable[None]] | None) -> None:
+    """Register the coroutine that re-runs the cleanup passes for a printer."""
+    global _recheck
+    _recheck = callback
+
+
+def removal_confirmed(printer_id: int, key: tuple) -> bool:
+    """Has this slot looked empty for the whole grace period?
+
+    ``key`` is ``(scope, ams_id, tray_id, spool_id)``, the spool id being the
+    inventory spool or the Spoolman spool linked to the slot. Returns False while the
+    removal is being held, and schedules a re-check for when it falls due.
+    """
+    full_key = (printer_id, *key)
+    now = _now()
+    first_seen, last_seen = _held.get(full_key, (now, now))
+    if now - last_seen > _STALE_AFTER:
+        first_seen = now
+    _held[full_key] = (first_seen, now)
+    if now - first_seen >= GRACE_SECONDS:
+        return True
+    _schedule_recheck(printer_id, GRACE_SECONDS - (now - first_seen))
+    return False
+
+
+def is_held(printer_id: int, key: tuple) -> bool:
+    """Is a removal already being held for this slot?
+
+    Lets a caller tell a spool swap from a spool the AMS merely cannot read.
+    A slot that reports occupied-but-blank is normally kept (#3100), but if it
+    was reported *empty* first, a spool came out and another went in -- that
+    keeps the hold running instead of cancelling it.
+    """
+    entry = _held.get((printer_id, *key))
+    return entry is not None and _now() - entry[1] <= _STALE_AFTER
+
+
+def forget_slot(printer_id: int, ams_id: int, tray_id: int) -> None:
+    """Drop any hold on a slot a spool has just been assigned or linked to.
+
+    A new assignment is fresher evidence than any empty report before it. It
+    can carry the same key as the hold -- the same spool put back and assigned
+    again -- so without this the user's own assignment could run out the old
+    clock and be deleted.
+    """
+    for full_key in [k for k in _held if k[0] == printer_id and k[2] == ams_id and k[3] == tray_id]:
+        del _held[full_key]
+
+
+def settle(printer_id: int, scope: str, held_keys: set[tuple]) -> None:
+    """Forget holds in ``scope`` that this pass did not hold again.
+
+    Called at the end of each pass with the keys it held. Anything else --
+    a slot that reads normally again, a removal just confirmed and unlinked,
+    an assignment that went away by other means -- no longer needs its clock.
+    """
+    for full_key in [k for k in _held if k[0] == printer_id and k[1] == scope and k[1:] not in held_keys]:
+        del _held[full_key]
+
+
+def _schedule_recheck(printer_id: int, delay: float) -> None:
+    if _recheck is None:
+        return
+    task = _recheck_tasks.get(printer_id)
+    if task is not None and not task.done():
+        return
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return
+    _recheck_tasks[printer_id] = loop.create_task(_run_recheck(printer_id, delay))
+
+
+async def _run_recheck(printer_id: int, delay: float) -> None:
+    await asyncio.sleep(max(delay, 0) + 1)
+    # Drop the handle first, so a hold the re-check itself renews can schedule
+    # the next one.
+    _recheck_tasks.pop(printer_id, None)
+    callback = _recheck
+    if callback is None:
+        return
+    try:
+        await callback(printer_id)
+    except Exception:
+        logger.exception("Held slot unlink re-check failed for printer %s", printer_id)
+
+
+def reset() -> None:
+    """Drop every hold and cancel pending re-checks (shutdown, tests)."""
+    for task in _recheck_tasks.values():
+        try:
+            task.cancel()
+        except RuntimeError:
+            # Its event loop is already closed; the task can never run.
+            pass
+    _recheck_tasks.clear()
+    _held.clear()

backend/app/services/spool_tag_matcher.py

@@ -9,6 +9,7 @@ from sqlalchemy.orm import selectinload
 from backend.app.models.spool import Spool
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.schemas.spool import normalize_effect_type
+from backend.app.services import slot_unlink_grace
 from backend.app.services.slot_nozzle import resolve_slot_nozzle
 from backend.app.services.spool_filament_preset import printer_safe_filament_id, resolve_spool_preset
 from backend.app.utils.tag_normalization import (
@@ -567,6 +568,7 @@ async def auto_assign_spool(
     )
     db.add(assignment)
     await db.flush()
+    slot_unlink_grace.forget_slot(printer_id, ams_id, tray_id)
 
     # Apply K-profile via MQTT (if available)
     # NOTE: Do NOT send ams_set_filament_setting here. This function is only

backend/tests/conftest.py

@@ -151,6 +151,20 @@ def reset_spoolman_location_sync_cache():
     _spoolman_location_sync_cache_clear()
 
 
+@pytest.fixture(autouse=True)
+def reset_slot_unlink_grace():
+    """Drop held slot unlinks and cancel their re-checks between tests (#3186).
+
+    The holds live in a module-level dict keyed by printer id, and every test
+    database hands out the same ids -- a hold left by one test would let the
+    next one's first blank report count as already confirmed."""
+    from backend.app.services import slot_unlink_grace
+
+    slot_unlink_grace.reset()
+    yield
+    slot_unlink_grace.reset()
+
+
 @pytest.fixture(autouse=True)
 def reset_auth_enabled_cache():
     """Drop the module-level auth-enabled cache between tests (issue #2572).

backend/tests/integration/test_inventory_assign.py

@@ -5,7 +5,7 @@ custom presets) takes priority, with slot reuse and generic fallback as
 lower-priority fallbacks.
 """
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from httpx import AsyncClient
@@ -1644,11 +1644,13 @@ class TestAutoUnlinkDuringRunout:
     async def test_cleared_tray_data_still_unlinks_when_idle(
         self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
     ):
-        """Off the print, an emptied slot really does mean the spool is gone."""
+        """Off the print, an emptied slot really does mean the spool is gone --
+        once it has stayed empty for the grace period (#3186)."""
         from unittest.mock import AsyncMock
 
         from backend.app.main import on_ams_change
         from backend.app.models.spool_assignment import SpoolAssignment
+        from backend.app.services import slot_unlink_grace
 
         printer = await printer_factory(name="H2D")
         spool = await spool_factory(material="ABS", rgba="616777FF")
@@ -1679,7 +1681,14 @@ class TestAutoUnlinkDuringRunout:
             mock_ws.send_printer_status = AsyncMock()
             mock_ws.broadcast = AsyncMock()
 
-            await on_ams_change(printer.id, ams_data)
+            clock = [1000.0]
+            with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+                await on_ams_change(printer.id, ams_data)
+                db_session.expunge_all()
+                assert await db_session.get(SpoolAssignment, assignment_id) is not None
+
+                clock[0] += slot_unlink_grace.GRACE_SECONDS
+                await on_ams_change(printer.id, ams_data)
 
         db_session.expunge_all()
         assert await db_session.get(SpoolAssignment, assignment_id) is None
@@ -1855,13 +1864,18 @@ class TestAutoUnlinkOccupiedSlot:
         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
+        from backend.app.services import slot_unlink_grace
 
         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"))
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            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
@@ -1888,6 +1902,377 @@ class TestAutoUnlinkOccupiedSlot:
         assert await db_session.get(SpoolAssignment, assignment_id) is None
 
 
+class TestAutoUnlinkHeldForGrace:
+    """#3186: one push is not proof a spool was taken out.
+
+    An idle X1 Carbon on X1Plus sent a push that cleared a whole AMS unit --
+    presence bits off, colour and type blank -- and four saved assignments were
+    deleted in the same instant (``Auto-unlinked 4 stale spool assignments``).
+    The spools never moved. A removal is now held for the grace period and
+    unlinked only if the slot is still empty then.
+    """
+
+    @staticmethod
+    async def _assign_unit(db_session, printer, spool):
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        ids = []
+        for tray_id in range(4):
+            a = SpoolAssignment(
+                spool_id=spool.id,
+                printer_id=printer.id,
+                ams_id=1,
+                tray_id=tray_id,
+                fingerprint_color="858585FF",
+                fingerprint_type="PLA",
+            )
+            db_session.add(a)
+            await db_session.commit()
+            ids.append(a.id)
+        return ids
+
+    @staticmethod
+    def _unit(blank: bool):
+        return [
+            {
+                "id": 1,
+                "tray": [
+                    {"id": t, "exists": False, "tray_type": "", "tray_color": "", "state": 9}
+                    if blank
+                    else {"id": t, "exists": True, "tray_type": "PLA", "tray_color": "858585FF", "state": 11}
+                    for t in range(4)
+                ],
+            }
+        ]
+
+    @staticmethod
+    async def _surviving(db_session, ids):
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        db_session.expunge_all()
+        return [i for i in ids if await db_session.get(SpoolAssignment, i) is not None]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_whole_unit_blanking_for_one_push_keeps_every_assignment(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        from backend.app.services import slot_unlink_grace
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="000000FF")
+        ids = await self._assign_unit(db_session, printer, spool)
+
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            blank, back = self._unit(blank=True), self._unit(blank=False)
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
+            assert await self._surviving(db_session, ids) == ids
+
+            clock[0] += 30
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, back, _make_printing_status(back, state="IDLE"))
+
+            # The recovery dropped the holds, so a blank long after the first
+            # one starts a fresh clock instead of confirming the old one.
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
+
+        assert await self._surviving(db_session, ids) == ids
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_unit_missing_from_one_push_keeps_its_assignments(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="000000FF")
+        ids = await self._assign_unit(db_session, printer, spool)
+
+        # AMS 1 dropped out of the payload entirely.
+        other_unit = [{"id": 0, "tray": [{"id": 0, "exists": True, "tray_type": "PLA", "tray_color": "FFFFFFFF"}]}]
+        await TestAutoUnlinkOccupiedSlot._run(printer.id, other_unit, _make_printing_status(other_unit, state="IDLE"))
+
+        assert await self._surviving(db_session, ids) == ids
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_slot_that_stays_empty_is_unlinked_by_the_recheck(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """No second push has to arrive: a spool taken out and left out never
+        changes the AMS hash again, so the held removal re-checks itself."""
+        from backend.app.main import _recheck_held_unlinks
+        from backend.app.services import slot_unlink_grace
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="000000FF")
+        ids = await self._assign_unit(db_session, printer, spool)
+        blank = self._unit(blank=True)
+
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
+            assert printer.id in slot_unlink_grace._recheck_tasks, "a held removal must schedule its re-check"
+            assert await self._surviving(db_session, ids) == ids
+
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            with (
+                patch("backend.app.main.printer_manager") as mock_pm_main,
+                patch("backend.app.main.ws_manager") as mock_ws,
+            ):
+                mock_pm_main.get_status.return_value = _make_printing_status(blank, state="IDLE")
+                mock_ws.broadcast = AsyncMock()
+                await _recheck_held_unlinks(printer.id)
+
+        assert await self._surviving(db_session, ids) == []
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_recheck_leaves_a_slot_that_came_back(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        from backend.app.main import _recheck_held_unlinks
+        from backend.app.services import slot_unlink_grace
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="000000FF")
+        ids = await self._assign_unit(db_session, printer, spool)
+        blank, back = self._unit(blank=True), self._unit(blank=False)
+
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            with (
+                patch("backend.app.main.printer_manager") as mock_pm_main,
+                patch("backend.app.main.ws_manager") as mock_ws,
+            ):
+                mock_pm_main.get_status.return_value = _make_printing_status(back, state="IDLE")
+                mock_ws.broadcast = AsyncMock()
+                await _recheck_held_unlinks(printer.id)
+
+        assert await self._surviving(db_session, ids) == ids
+
+
+class TestAutoUnlinkSwapDuringGrace:
+    """A non-RFID spool swapped out and another put in within the grace period.
+
+    The new spool reports occupied-but-blank, which on its own keeps an
+    assignment (#3100). After an *empty* report it means the spool changed,
+    so the held removal has to keep running rather than be cancelled -- or the
+    old spool would stay assigned to the new one indefinitely.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_swap_still_unlinks_once_the_grace_runs_out(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        from backend.app.models.spool_assignment import SpoolAssignment
+        from backend.app.services import slot_unlink_grace
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
+        assignment_id = await TestAutoUnlinkOccupiedSlot._assignment(db_session, printer, spool)
+
+        taken_out = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
+        new_spool = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, taken_out, _make_printing_status(taken_out, state="IDLE"))
+            clock[0] += 20
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, new_spool, _make_printing_status(new_spool, state="IDLE"))
+            db_session.expunge_all()
+            assert await db_session.get(SpoolAssignment, assignment_id) is not None
+
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, new_spool, _make_printing_status(new_spool, 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_new_assignment_to_the_slot_starts_clean(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """Assigning a spool to the slot after the swap -- even the same spool
+        again -- makes a new row, and the old clock does not carry over to it."""
+        from backend.app.models.spool_assignment import SpoolAssignment
+        from backend.app.services import slot_unlink_grace
+
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
+        old_id = await TestAutoUnlinkOccupiedSlot._assignment(db_session, printer, spool)
+
+        taken_out = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
+        back_in = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, taken_out, _make_printing_status(taken_out, state="IDLE"))
+
+            # The user puts the spool back and assigns it again through the
+            # endpoint, which replaces the row -- on SQLite quite possibly under
+            # the same id, so only dropping the hold keeps the old clock off it.
+            hold_key = ("inventory", 1, 1, spool.id)
+            assert slot_unlink_grace.is_held(printer.id, hold_key)
+            with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
+                mock_pm.get_client.return_value = None
+                mock_pm.get_status.return_value = _make_mock_status(ams_data=back_in)
+                response = await async_client.post(
+                    "/api/v1/inventory/assignments",
+                    json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 1, "tray_id": 1},
+                )
+            assert response.status_code == 200
+            assert not slot_unlink_grace.is_held(printer.id, hold_key), "an assignment must drop the old hold"
+            new_id = response.json()["id"]
+            assert await db_session.get(SpoolAssignment, old_id) is None or new_id == old_id
+
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await TestAutoUnlinkOccupiedSlot._run(printer.id, back_in, _make_printing_status(back_in, state="IDLE"))
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, new_id) is not None
+
+
+class TestSpoolmanSlotUnlinkHeldForGrace:
+    """Spoolman mode's half of #3186: the slot row waits out the same grace."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_recheck_deletes_a_row_only_once_the_slot_stayed_empty(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        from backend.app.main import _expire_spoolman_empty_slots
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+        from backend.app.services import slot_unlink_grace
+
+        await TestSpoolmanSlotAssignmentDuringRunout()._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
+
+        blank = [{"id": 0, "tray": [{"id": 2, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
+        clock = [1000.0]
+        with (
+            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_ws.broadcast = AsyncMock()
+            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)
+            db_session.expunge_all()
+            assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None
+
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolmanSlotAssignment, row_id) is None
+        mock_ws.broadcast.assert_awaited()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_swap_to_an_unreadable_spool_still_expires_the_row(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        from backend.app.main import _expire_spoolman_empty_slots
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+        from backend.app.services import slot_unlink_grace
+
+        await TestSpoolmanSlotAssignmentDuringRunout()._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
+
+        taken_out = [{"id": 0, "tray": [{"id": 2, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
+        new_spool = [{"id": 0, "tray": [{"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
+        clock = [1000.0]
+        with (
+            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_ws.broadcast = AsyncMock()
+            await _expire_spoolman_empty_slots(printer.id, taken_out, printing_now=False)
+            clock[0] += 20
+            await _expire_spoolman_empty_slots(printer.id, new_spool, printing_now=False)
+            db_session.expunge_all()
+            assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None
+
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await _expire_spoolman_empty_slots(printer.id, new_spool, printing_now=False)
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolmanSlotAssignment, row_id) is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_an_occupied_unreadable_slot_with_no_hold_keeps_its_row(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        """#3100 still holds for the re-check: never reported empty, never unlinked."""
+        from backend.app.main import _expire_spoolman_empty_slots
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+        from backend.app.services import slot_unlink_grace
+
+        await TestSpoolmanSlotAssignmentDuringRunout()._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
+
+        unreadable = [{"id": 0, "tray": [{"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
+        clock = [1000.0]
+        with (
+            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_ws.broadcast = AsyncMock()
+            for _ in range(3):
+                await _expire_spoolman_empty_slots(printer.id, unreadable, printing_now=False)
+                clock[0] += slot_unlink_grace.GRACE_SECONDS
+
+        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_recheck_keeps_a_row_whose_slot_reads_again(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        from backend.app.main import _expire_spoolman_empty_slots
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+        from backend.app.services import slot_unlink_grace
+
+        await TestSpoolmanSlotAssignmentDuringRunout()._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
+
+        blank = [{"id": 0, "tray": [{"id": 2, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
+        back = [{"id": 0, "tray": [{"id": 2, "exists": True, "tray_type": "PLA", "tray_color": "858585FF"}]}]
+        clock = [1000.0]
+        with (
+            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_ws.broadcast = AsyncMock()
+            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)
+            clock[0] += 30
+            await _expire_spoolman_empty_slots(printer.id, back, printing_now=False)
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolmanSlotAssignment, row_id) is not 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
@@ -1992,8 +2377,9 @@ class TestSpoolmanSlotAssignmentDuringRunout:
         self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
     ):
         """Proves the guard is what saved the row above, not an unreachable
-        code path."""
+        code path -- once the slot has stayed empty for the grace period."""
         from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+        from backend.app.services import slot_unlink_grace
 
         await self._enable_spoolman(db_session)
         printer = await printer_factory(name="H2D")
@@ -2002,7 +2388,14 @@ class TestSpoolmanSlotAssignmentDuringRunout:
         await db_session.commit()
         row_id = row.id
 
-        await self._run(printer.id, _make_printing_status(None, state="IDLE"))
+        clock = [1000.0]
+        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+            await self._run(printer.id, _make_printing_status(None, state="IDLE"))
+            db_session.expunge_all()
+            assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None
+
+            clock[0] += slot_unlink_grace.GRACE_SECONDS
+            await self._run(printer.id, _make_printing_status(None, state="IDLE"))
 
         db_session.expunge_all()
         assert await db_session.get(SpoolmanSlotAssignment, row_id) is None

backend/tests/integration/test_spoolman_ams_sync_broadcast.py

@@ -124,18 +124,25 @@ async def test_a_synced_slot_is_broadcast(async_client: AsyncClient, printer_fac
 @pytest.mark.asyncio
 @pytest.mark.integration
 async def test_an_emptied_slot_is_broadcast(async_client: AsyncClient, printer_factory, db_session: AsyncSession):
-    """The row is deleted here; a card still drawing the removed spool is the
-    same bug seen from the other side."""
+    """The row is deleted here -- once the slot has stayed empty for the grace
+    period (#3186) -- and a card still drawing the removed spool is the same
+    bug seen from the other side."""
+    from backend.app.services import slot_unlink_grace
+
     printer = await printer_factory(name="H2C")
     await _enable_spoolman(db_session)
     db_session.add(SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=1, spoolman_spool_id=7))
     await db_session.commit()
 
-    broadcast, _ = await _run_ams_change(
-        printer.id,
-        [{"id": 0, "tray": [{"id": 1}]}],
-        parsed={(0, 1): None},
-    )
+    clock = [1000.0]
+    with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
+        await _run_ams_change(printer.id, [{"id": 0, "tray": [{"id": 1}]}], parsed={(0, 1): None})
+        clock[0] += slot_unlink_grace.GRACE_SECONDS
+        broadcast, _ = await _run_ams_change(
+            printer.id,
+            [{"id": 0, "tray": [{"id": 1}]}],
+            parsed={(0, 1): None},
+        )
 
     assert (0, 1) in _slot_events(broadcast)
 

backend/tests/integration/test_spoolman_slot_assignments.py

@@ -109,6 +109,28 @@ class TestAssignSpoolmanSlot:
         assert rows[0]["ams_id"] == 0
         assert rows[0]["tray_id"] == 0
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_assign_drops_a_held_unlink_on_the_slot(
+        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
+    ):
+        """#3186: a slot that read empty has its unlink held for a grace period.
+        Linking a spool to it is fresher evidence, so the old clock must not
+        carry over and delete the user's own link."""
+        from backend.app.services import slot_unlink_grace
+
+        hold_key = ("spoolman", 0, 0, 10)
+        slot_unlink_grace.removal_confirmed(test_printer.id, hold_key)
+        assert slot_unlink_grace.is_held(test_printer.id, hold_key)
+
+        response = await async_client.post(
+            "/api/v1/spoolman/inventory/slot-assignments",
+            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
+        )
+
+        assert response.status_code == 200
+        assert not slot_unlink_grace.is_held(test_printer.id, hold_key)
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_assign_accepts_ams_ht_id(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):

backend/tests/unit/services/test_slot_unlink_grace.py

@@ -0,0 +1,133 @@
+"""slot_unlink_grace holds an automatic slot unlink until the slot has stayed
+empty for the grace period (#3186)."""
+
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from backend.app.services import slot_unlink_grace as grace
+
+KEY = ("inventory", 1, 2, 33)
+
+
+@pytest.fixture
+def clock():
+    now = [1000.0]
+    with patch.object(grace, "_now", lambda: now[0]):
+        yield now
+
+
+def test_a_first_sighting_is_held(clock):
+    assert grace.removal_confirmed(7, KEY) is False
+
+
+def test_confirmed_once_the_grace_period_has_passed(clock):
+    grace.removal_confirmed(7, KEY)
+    clock[0] += grace.GRACE_SECONDS - 1
+    assert grace.removal_confirmed(7, KEY) is False
+    clock[0] += 1
+    assert grace.removal_confirmed(7, KEY) is True
+
+
+def test_settle_forgets_a_slot_the_pass_did_not_hold_again(clock):
+    grace.removal_confirmed(7, KEY)
+    grace.settle(7, "inventory", set())
+    clock[0] += grace.GRACE_SECONDS
+    assert grace.removal_confirmed(7, KEY) is False, "the recovered slot must start a fresh clock"
+
+
+def test_settle_keeps_other_printers_and_scopes(clock):
+    grace.removal_confirmed(7, KEY)
+    grace.removal_confirmed(8, KEY)
+    grace.removal_confirmed(7, ("spoolman", 1, 2, 41))
+    grace.settle(8, "inventory", set())
+    grace.settle(7, "spoolman", set())
+    clock[0] += grace.GRACE_SECONDS
+    assert grace.removal_confirmed(7, KEY) is True
+
+
+def test_a_hold_nobody_renewed_starts_over(clock):
+    """A pass that did not run (Spoolman unreachable, printer offline) cannot
+    have watched the slot stay empty, so its old first sighting does not count."""
+    grace.removal_confirmed(7, KEY)
+    clock[0] += 3 * grace.GRACE_SECONDS
+    assert grace.removal_confirmed(7, KEY) is False
+
+
+def test_is_held_follows_the_hold(clock):
+    assert grace.is_held(7, KEY) is False
+    grace.removal_confirmed(7, KEY)
+    assert grace.is_held(7, KEY) is True
+    grace.settle(7, "inventory", set())
+    assert grace.is_held(7, KEY) is False
+
+
+def test_is_held_ignores_a_stale_hold(clock):
+    grace.removal_confirmed(7, KEY)
+    clock[0] += 3 * grace.GRACE_SECONDS
+    assert grace.is_held(7, KEY) is False
+
+
+def test_forget_slot_drops_every_hold_on_that_slot_only(clock):
+    grace.removal_confirmed(7, ("spoolman", 1, 2, 41))
+    grace.removal_confirmed(7, ("inventory", 1, 2, 99))
+    grace.removal_confirmed(7, ("spoolman", 1, 3, 42))
+    grace.removal_confirmed(8, ("spoolman", 1, 2, 41))
+
+    grace.forget_slot(7, 1, 2)
+
+    assert grace.is_held(7, ("spoolman", 1, 2, 41)) is False
+    assert grace.is_held(7, ("inventory", 1, 2, 99)) is False
+    assert grace.is_held(7, ("spoolman", 1, 3, 42)) is True
+    assert grace.is_held(8, ("spoolman", 1, 2, 41)) is True
+
+
+def test_no_recheck_without_a_registered_callback(clock):
+    with patch.object(grace, "_recheck", None):
+        grace.removal_confirmed(7, KEY)
+    assert grace._recheck_tasks == {}
+
+
+@pytest.mark.asyncio
+async def test_holds_on_one_printer_share_a_single_recheck(clock):
+    async def recheck(printer_id: int) -> None:
+        pass
+
+    with patch.object(grace, "_recheck", recheck):
+        grace.removal_confirmed(7, KEY)
+        grace.removal_confirmed(7, ("inventory", 1, 3, 34))
+        grace.removal_confirmed(8, KEY)
+
+    assert sorted(grace._recheck_tasks) == [7, 8]
+    for task in grace._recheck_tasks.values():
+        task.cancel()
+    await asyncio.gather(*grace._recheck_tasks.values(), return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test_the_recheck_runs_the_callback_and_frees_its_slot():
+    calls: list[int] = []
+
+    async def recheck(printer_id: int) -> None:
+        # Freed before the callback, so a hold the re-check renews can
+        # schedule the next one.
+        assert printer_id not in grace._recheck_tasks
+        calls.append(printer_id)
+
+    with patch.object(grace, "_recheck", recheck):
+        grace._recheck_tasks[7] = asyncio.current_task()
+        await grace._run_recheck(7, -1)
+
+    assert calls == [7]
+
+
+@pytest.mark.asyncio
+async def test_a_failing_recheck_is_logged_not_raised():
+    async def recheck(printer_id: int) -> None:
+        raise RuntimeError("boom")
+
+    with patch.object(grace, "_recheck", recheck), patch.object(grace.logger, "exception") as log:
+        await grace._run_recheck(7, -1)
+
+    log.assert_called_once()