fix(notifications): missing-spool-assignment check now unions both assignment tables (#1473)

notify_missing_spool_assignments_on_print_start queried only the legacy
  SpoolAssignment table. In Spoolman mode that table is empty -- bindings
  live in spoolman_slot_assignments -- so assigned_global_trays came back
  empty and every used tray was flagged missing, firing a false-positive
  notification on every print.

  Union SpoolAssignment + SpoolmanSlotAssignment rows for the printer
  before computing the missing set. Both tables expose printer_id /
  ams_id / tray_id identically, so _global_tray_from_assignment is
  unchanged. Union-only, so legacy-mode behavior cannot regress.

backend/app/services/spool_assignment_notifications.py

@@ -4,6 +4,7 @@ from backend.app.core.database import async_session
 from backend.app.core.websocket import ws_manager
 from backend.app.models.printer import Printer
 from backend.app.models.spool_assignment import SpoolAssignment
+from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
 from backend.app.services.bambu_mqtt import PrinterState
 from backend.app.services.notification_service import notification_service
 from backend.app.services.printer_manager import printer_manager
@@ -127,12 +128,23 @@ async def notify_missing_spool_assignments_on_print_start(
             printer = await db.get(Printer, printer_id)
             printer_name = printer.name if printer else f"Printer {printer_id}"
 
-            assignments_result = await db.execute(
-                SpoolAssignment.__table__.select().where(SpoolAssignment.printer_id == printer_id)
-            )
-            assignments = assignments_result.fetchall()
+            # A tray is "assigned" if it has a row in EITHER table: the legacy
+            # spool_assignment table (internal-inventory mode) or
+            # spoolman_slot_assignments (Spoolman mode — the binding
+            # source-of-truth since #1119). Querying only the legacy table
+            # flagged every used tray as missing on every Spoolman-mode print
+            # (#1473). Both tables expose printer_id / ams_id / tray_id in the
+            # same shape, so _global_tray_from_assignment works on either.
+            legacy_rows = (
+                await db.execute(SpoolAssignment.__table__.select().where(SpoolAssignment.printer_id == printer_id))
+            ).fetchall()
+            spoolman_rows = (
+                await db.execute(
+                    SpoolmanSlotAssignment.__table__.select().where(SpoolmanSlotAssignment.printer_id == printer_id)
+                )
+            ).fetchall()
             assigned_global_trays = {
-                _global_tray_from_assignment(assignment.ams_id, assignment.tray_id) for assignment in assignments
+                _global_tray_from_assignment(row.ams_id, row.tray_id) for row in (*legacy_rows, *spoolman_rows)
             }
 
             missing_global = sorted(used_global_trays - assigned_global_trays)

backend/tests/unit/services/test_spool_assignment_notifications.py

@@ -18,9 +18,18 @@ class _FakeAssignmentsResult:
 
 
 class _FakeSession:
-    def __init__(self, printer_name: str, assignments: list[SimpleNamespace]):
+    """Fake DB session that returns legacy vs. Spoolman assignment rows based
+    on which table the SELECT targets, so tests can exercise either mode."""
+
+    def __init__(
+        self,
+        printer_name: str,
+        legacy: list[SimpleNamespace] | None = None,
+        spoolman: list[SimpleNamespace] | None = None,
+    ):
         self._printer = SimpleNamespace(name=printer_name)
-        self._assignments = assignments
+        self._legacy = legacy or []
+        self._spoolman = spoolman or []
 
     async def __aenter__(self):
         return self
@@ -32,7 +41,10 @@ class _FakeSession:
         return self._printer
 
     async def execute(self, statement):
-        return _FakeAssignmentsResult(self._assignments)
+        table = statement.get_final_froms()[0].name
+        if table == "spoolman_slot_assignments":
+            return _FakeAssignmentsResult(self._spoolman)
+        return _FakeAssignmentsResult(self._legacy)
 
 
 @pytest.mark.asyncio
@@ -75,3 +87,85 @@ async def test_missing_assignment_broadcasts_websocket_event_and_push_notificati
     assert notify_kwargs["printer_id"] == 1
     assert notify_kwargs["printer_name"] == "Printer A"
     assert notify_kwargs["missing_slots"] == [{"slot": "A2", "profile": "Unknown", "color": "Unknown"}]
+
+
+def _patches(session):
+    """Common patch set: the fake session + stubbed printer state / emitters."""
+    return (
+        patch(
+            "backend.app.services.spool_assignment_notifications.async_session",
+            return_value=session,
+        ),
+        patch("backend.app.services.spool_assignment_notifications.printer_manager.get_status", return_value=None),
+        patch(
+            "backend.app.services.spool_assignment_notifications.ws_manager.send_missing_spool_assignment",
+            new_callable=AsyncMock,
+        ),
+        patch(
+            "backend.app.services.spool_assignment_notifications.notification_service.on_print_missing_spool_assignment",
+            new_callable=AsyncMock,
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_spoolman_only_assignment_suppresses_notification():
+    """#1473 — trays bound only via spoolman_slot_assignments must NOT be
+    flagged missing (the legacy spool_assignment table is empty in Spoolman
+    mode, so checking it alone fired a false positive on every print)."""
+    logger = logging.getLogger(__name__)
+    data = {"ams_mapping": [0, 1], "raw_data": {}}  # print uses A1 + A2
+
+    # Both used trays bound via Spoolman; legacy table empty.
+    session = _FakeSession(
+        "Printer A",
+        legacy=[],
+        spoolman=[SimpleNamespace(ams_id=0, tray_id=0), SimpleNamespace(ams_id=0, tray_id=1)],
+    )
+    p_session, p_status, p_ws, p_notify = _patches(session)
+    with p_session, p_status, p_ws as mock_ws, p_notify as mock_notify:
+        await notify_missing_spool_assignments_on_print_start(1, data, logger)
+
+    mock_ws.assert_not_awaited()
+    mock_notify.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_spoolman_partial_coverage_flags_only_uncovered_tray():
+    """A Spoolman assignment for A1 only, with a print using A1 + A2, flags
+    A2 alone."""
+    logger = logging.getLogger(__name__)
+    data = {"ams_mapping": [0, 1], "raw_data": {}}
+
+    session = _FakeSession(
+        "Printer A",
+        legacy=[],
+        spoolman=[SimpleNamespace(ams_id=0, tray_id=0)],  # A1 only
+    )
+    p_session, p_status, p_ws, p_notify = _patches(session)
+    with p_session, p_status, p_ws as mock_ws, p_notify as mock_notify:
+        await notify_missing_spool_assignments_on_print_start(1, data, logger)
+
+    mock_ws.assert_awaited_once()
+    assert mock_ws.await_args.kwargs["missing_slots"] == [{"slot": "A2", "profile": "Unknown", "color": "Unknown"}]
+    mock_notify.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_mixed_mode_union_covers_all_used_trays():
+    """A1 bound in the legacy table, A2 bound in spoolman_slot_assignments —
+    the union covers both used trays, so no notification fires."""
+    logger = logging.getLogger(__name__)
+    data = {"ams_mapping": [0, 1], "raw_data": {}}
+
+    session = _FakeSession(
+        "Printer A",
+        legacy=[SimpleNamespace(ams_id=0, tray_id=0)],  # A1
+        spoolman=[SimpleNamespace(ams_id=0, tray_id=1)],  # A2
+    )
+    p_session, p_status, p_ws, p_notify = _patches(session)
+    with p_session, p_status, p_ws as mock_ws, p_notify as mock_notify:
+        await notify_missing_spool_assignments_on_print_start(1, data, logger)
+
+    mock_ws.assert_not_awaited()
+    mock_notify.assert_not_awaited()