fix(spoolman): per-print weight tracker falls back to local slot-assignment table for tag-less spools (#1459)

  Reporter on Postgres + Spoolman saw weight never decremented after
  prints. Traced to _report_spool_usage_for_slots calling only
  client.find_spool_by_tag() — which returns None when extra.tag is empty.
  Non-RFID spools assigned via the Bambuddy UI intentionally leave
  extra.tag empty (per #1457 — we don't want fallback tags polluting
  Spoolman), so tag-less spools never got matched and weight tracking
  silently no-op'd. The tracker never consulted the local
  spoolman_slot_assignments table that has the binding.

  Adds _resolve_spool_id_via_slot_assignment() as stage 2 of the
  resolution chain. Stage 1 (existing tag-lookup) wins when present so
  RFID auto-sync remains unchanged. (ams_id, tray_id) derived from
  global_tray_id via the existing _global_tray_id_to_ams_slot helper,
  so external slots and AMS-HT slots resolve correctly. Threaded
  printer_id through the three callers (partial G-code, partial linear,
  final-usage report). Resolution path is logged ("via tag" vs "via
  slot-assignment") so support bundles confirm the fix is live.

  extra.tag is deliberately NOT auto-populated — that would re-introduce
  the exact pollution #1457 cleaned up. Slot-assignment table is the
  source of truth for non-RFID; extra.tag is reserved for hardware RFID.

backend/app/services/spoolman_tracking.py

@@ -355,6 +355,28 @@ async def _get_spoolman_client_with_fallback():
     return client
 
 
+async def _resolve_spool_id_via_slot_assignment(printer_id: int, ams_id: int, tray_id: int) -> int | None:
+    """Look up the Spoolman spool ID locally bound to (printer, ams, tray).
+
+    Fallback path for #1459: when a tag-less spool was assigned via the
+    Bambuddy UI, the user's deterministic fallback tag is intentionally NOT
+    written to Spoolman's extra.tag (kept clean per #1457), so
+    find_spool_by_tag misses. The local spoolman_slot_assignments table is
+    the authoritative binding for those spools.
+    """
+    from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+
+    async with async_session() as db:
+        result = await db.execute(
+            select(SpoolmanSlotAssignment.spoolman_spool_id).where(
+                SpoolmanSlotAssignment.printer_id == printer_id,
+                SpoolmanSlotAssignment.ams_id == ams_id,
+                SpoolmanSlotAssignment.tray_id == tray_id,
+            )
+        )
+        return result.scalar_one_or_none()
+
+
 async def _report_spool_usage_for_slots(
     client,
     filament_usage_items: list[tuple[int, float]],
@@ -362,9 +384,17 @@ async def _report_spool_usage_for_slots(
     slot_to_tray: list | None,
     method_label: str,
     printer_serial: str = "",
+    printer_id: int | None = None,
 ) -> int:
     """Report usage to Spoolman for a list of (slot_id, grams) pairs.
 
+    Resolution order per slot: (1) Spoolman extra.tag match against the
+    tray's RFID or deterministic fallback tag, (2) #1459 fallback —
+    local spoolman_slot_assignments table keyed by (printer_id, ams_id,
+    tray_id). Without (2), tag-less spools assigned via the Bambuddy UI
+    never get their weight decremented because their extra.tag is empty
+    on the Spoolman side.
+
     Returns number of spools successfully updated.
     """
     spools_updated = 0
@@ -388,22 +418,43 @@ async def _report_spool_usage_for_slots(
             is_external,
         )
 
-        spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id)
-        if not spool_tag:
-            logger.debug("[SPOOLMAN] Slot %s: no identifier for tray %s", slot_id, global_tray_id)
-            continue
+        spool_id_to_use: int | None = None
+        resolution_path = ""
 
-        spool = await client.find_spool_by_tag(spool_tag)
-        if not spool:
-            logger.debug("[SPOOLMAN] Slot %s: no spool for tag %s...", slot_id, spool_tag[:16])
+        spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id)
+        if spool_tag:
+            spool = await client.find_spool_by_tag(spool_tag)
+            if spool:
+                spool_id_to_use = spool["id"]
+                resolution_path = "tag"
+
+        if spool_id_to_use is None and printer_id is not None:
+            ams_id, tray_id = _global_tray_id_to_ams_slot(global_tray_id)
+            spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id)
+            if spool_id_to_use is not None:
+                resolution_path = "slot-assignment"
+
+        if spool_id_to_use is None:
+            logger.debug(
+                "[SPOOLMAN] Slot %s: no spool resolved (tag=%s, no slot-assignment)",
+                slot_id,
+                spool_tag[:16] if spool_tag else "none",
+            )
             continue
 
         try:
-            await client.use_spool(spool["id"], grams_used)
-            logger.info("[SPOOLMAN] %s: slot %s: %sg -> spool %s", method_label, slot_id, grams_used, spool["id"])
+            await client.use_spool(spool_id_to_use, grams_used)
+            logger.info(
+                "[SPOOLMAN] %s: slot %s: %sg -> spool %s (via %s)",
+                method_label,
+                slot_id,
+                grams_used,
+                spool_id_to_use,
+                resolution_path,
+            )
             spools_updated += 1
         except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
-            logger.warning("[SPOOLMAN] Failed to record usage for spool %s: %s", spool["id"], exc)
+            logger.warning("[SPOOLMAN] Failed to record usage for spool %s: %s", spool_id_to_use, exc)
 
     return spools_updated
 
@@ -526,7 +577,13 @@ async def _report_partial_usage(
                 usage_items.append((slot_id, grams_used))
 
             spools_updated = await _report_spool_usage_for_slots(
-                client, usage_items, ams_trays, slot_to_tray, "Partial (G-code)", printer_serial
+                client,
+                usage_items,
+                ams_trays,
+                slot_to_tray,
+                "Partial (G-code)",
+                printer_serial,
+                printer_id=printer_id,
             )
             if spools_updated > 0:
                 logger.info("[SPOOLMAN] Reported partial usage to %s spool(s) using G-code data", spools_updated)
@@ -558,7 +615,13 @@ async def _report_partial_usage(
             usage_items.append((slot_id, partial_used_g))
 
     spools_updated = await _report_spool_usage_for_slots(
-        client, usage_items, ams_trays, slot_to_tray, "Partial (linear)", printer_serial
+        client,
+        usage_items,
+        ams_trays,
+        slot_to_tray,
+        "Partial (linear)",
+        printer_serial,
+        printer_id=printer_id,
     )
     if spools_updated > 0:
         logger.info("[SPOOLMAN] Reported partial usage to %s spool(s) using linear interpolation", spools_updated)
@@ -613,7 +676,13 @@ async def report_usage(printer_id: int, archive_id: int):
 
         usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage]
         spools_updated = await _report_spool_usage_for_slots(
-            client, usage_items, ams_trays, slot_to_tray, f"Archive {archive_id}", printer_serial
+            client,
+            usage_items,
+            ams_trays,
+            slot_to_tray,
+            f"Archive {archive_id}",
+            printer_serial,
+            printer_id=printer_id,
         )
 
         if spools_updated == 0:

backend/tests/integration/test_spoolman_tracking_slot_fallback.py

@@ -0,0 +1,182 @@
+"""Integration tests for #1459 — per-print weight tracker falls back to the
+local spoolman_slot_assignments table when Spoolman's extra.tag is empty.
+
+Without this, tag-less spools assigned via the Bambuddy UI never get their
+weight decremented because the Assign route intentionally leaves extra.tag
+unset (per #1457 — fallback tags must not pollute Spoolman).
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+from backend.app.services.spoolman_tracking import _report_spool_usage_for_slots
+
+
+@pytest.fixture
+def mock_spoolman_client():
+    client = MagicMock()
+    # Default: every tag-lookup returns None (the bug case — no extra.tag on Spoolman side).
+    client.find_spool_by_tag = AsyncMock(return_value=None)
+    client.use_spool = AsyncMock(return_value={"id": 0})
+    return client
+
+
+@pytest.fixture
+def patch_async_session(test_engine):
+    """Route the tracker's async_session() to the test engine so the slot-assignment
+    fallback lookup sees rows committed via db_session in the same test."""
+    test_async_session = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    with patch("backend.app.services.spoolman_tracking.async_session", test_async_session):
+        yield
+
+
+@pytest.fixture
+async def test_printer(db_session):
+    from backend.app.models.printer import Printer
+
+    printer = Printer(
+        name="Tracking Test",
+        serial_number="TRACKTEST123456",
+        ip_address="192.168.0.99",
+        access_code="12345678",
+        model="P1S",
+        is_active=True,
+        auto_archive=True,
+    )
+    db_session.add(printer)
+    await db_session.commit()
+    await db_session.refresh(printer)
+    return printer
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.usefixtures("patch_async_session")
+class TestSlotAssignmentFallback:
+    async def test_falls_back_to_slot_assignment_when_tag_missing(self, test_printer, mock_spoolman_client, db_session):
+        """Tag-less spool assigned via Bambuddy UI: extra.tag is empty (find_spool_by_tag
+        returns None) but the local spoolman_slot_assignments row says spool 42 lives in
+        AMS 0 tray 2 — the tracker must still report usage to spool 42.
+
+        slot_id is 1-based; ams_trays is keyed by global_tray_id. For AMS 0 tray 2,
+        global_tray_id = 2, so we hand the tracker slot_id=3 (since slot_id-1=global=2).
+        """
+        db_session.add(SpoolmanSlotAssignment(printer_id=test_printer.id, ams_id=0, tray_id=2, spoolman_spool_id=42))
+        await db_session.commit()
+
+        ams_trays = {2: {"tray_uuid": "", "tag_uid": "", "tray_type": "PLA"}}
+        usage_items = [(3, 15.5)]
+
+        spools_updated = await _report_spool_usage_for_slots(
+            mock_spoolman_client,
+            usage_items,
+            ams_trays,
+            slot_to_tray=None,
+            method_label="Test",
+            printer_serial=test_printer.serial_number,
+            printer_id=test_printer.id,
+        )
+
+        assert spools_updated == 1
+        mock_spoolman_client.use_spool.assert_awaited_once_with(42, 15.5)
+
+    async def test_tag_match_wins_over_slot_assignment(self, test_printer, mock_spoolman_client, db_session):
+        """When both paths could resolve a spool, the tag-match wins — RFID is the
+        authoritative binding when present. Order matters so RFID auto-sync continues
+        to bind to the spool whose extra.tag literally holds that RFID, even if the
+        slot-assignment table happens to point at a different spool."""
+        db_session.add(SpoolmanSlotAssignment(printer_id=test_printer.id, ams_id=0, tray_id=0, spoolman_spool_id=999))
+        await db_session.commit()
+
+        mock_spoolman_client.find_spool_by_tag = AsyncMock(return_value={"id": 7})
+
+        ams_trays = {0: {"tray_uuid": "A" * 32, "tag_uid": "", "tray_type": "PLA"}}
+        # slot_id=1 → global_tray_id=0 (AMS 0 tray 0).
+        usage_items = [(1, 10.0)]
+
+        spools_updated = await _report_spool_usage_for_slots(
+            mock_spoolman_client,
+            usage_items,
+            ams_trays,
+            slot_to_tray=None,
+            method_label="Test",
+            printer_serial=test_printer.serial_number,
+            printer_id=test_printer.id,
+        )
+
+        assert spools_updated == 1
+        mock_spoolman_client.use_spool.assert_awaited_once_with(7, 10.0)
+
+    async def test_skips_when_neither_path_resolves(self, test_printer, mock_spoolman_client, db_session):
+        """No tag in Spoolman AND no slot-assignment row → tracker skips the slot
+        rather than crashing or reporting against the wrong spool."""
+        ams_trays = {0: {"tray_uuid": "", "tag_uid": "", "tray_type": "PLA"}}
+        # slot_id=1 → global_tray_id=0 (AMS 0 tray 0); no assignment row exists.
+        usage_items = [(1, 5.0)]
+
+        spools_updated = await _report_spool_usage_for_slots(
+            mock_spoolman_client,
+            usage_items,
+            ams_trays,
+            slot_to_tray=None,
+            method_label="Test",
+            printer_serial=test_printer.serial_number,
+            printer_id=test_printer.id,
+        )
+
+        assert spools_updated == 0
+        mock_spoolman_client.use_spool.assert_not_called()
+
+    async def test_skips_when_printer_id_not_supplied(self, test_printer, mock_spoolman_client, db_session):
+        """Slot-assignment fallback requires printer_id to look up the binding —
+        when callers don't supply it (legacy call shape) the lookup is skipped
+        and the slot is reported as unresolved, matching pre-#1459 behaviour for
+        those callers."""
+        db_session.add(SpoolmanSlotAssignment(printer_id=test_printer.id, ams_id=0, tray_id=0, spoolman_spool_id=42))
+        await db_session.commit()
+
+        ams_trays = {0: {"tray_uuid": "", "tag_uid": "", "tray_type": "PLA"}}
+        usage_items = [(1, 5.0)]
+
+        spools_updated = await _report_spool_usage_for_slots(
+            mock_spoolman_client,
+            usage_items,
+            ams_trays,
+            slot_to_tray=None,
+            method_label="Test",
+            printer_serial=test_printer.serial_number,
+            # printer_id omitted on purpose
+        )
+
+        assert spools_updated == 0
+        mock_spoolman_client.use_spool.assert_not_called()
+
+    async def test_external_slot_falls_back_via_correct_ams_tray_pair(
+        self, test_printer, mock_spoolman_client, db_session
+    ):
+        """External spool slots use global_tray_id 254/255 which map to ams_id=255,
+        tray_id=0/1. The slot-assignment lookup must use that translated pair, not the
+        raw global id, otherwise the row is never found."""
+        db_session.add(SpoolmanSlotAssignment(printer_id=test_printer.id, ams_id=255, tray_id=0, spoolman_spool_id=88))
+        await db_session.commit()
+
+        # Position-based default with ams_trays={254: ...}: sorted_tray_ids=[254],
+        # slot_id=1 → sorted_tray_ids[0] = 254 (global) → ams_id=255 tray_id=0.
+        ams_trays = {254: {"tray_uuid": "", "tag_uid": "", "tray_type": "PLA"}}
+        usage_items = [(1, 25.0)]
+
+        spools_updated = await _report_spool_usage_for_slots(
+            mock_spoolman_client,
+            usage_items,
+            ams_trays,
+            slot_to_tray=None,
+            method_label="Test",
+            printer_serial=test_printer.serial_number,
+            printer_id=test_printer.id,
+        )
+
+        assert spools_updated == 1
+        mock_spoolman_client.use_spool.assert_awaited_once_with(88, 25.0)