Fix spool usage lost when spool runs empty mid-print (#459)

When a spool ran empty during a print and the AMS auto-switched to a
backup, on_ams_change eagerly deleted the SpoolAssignment (fingerprint
mismatch). on_print_complete then queried the now-deleted assignment
and silently dropped usage.

Snapshot all spool assignments into PrintSession at print start. At
completion, prefer the snapshot over live DB queries in both the 3MF
primary path and the AMS remain% delta fallback. Fall back to live
query for backward compatibility with in-flight sessions started
before upgrade.

CHANGELOG.md

@@ -5,6 +5,7 @@ All notable changes to Bambuddy will be documented in this file.
 ## [0.2.1b2] - Unreleased
 
 ### Fixed
+- **Spool Usage Lost When Spool Runs Empty Mid-Print** ([#459](https://github.com/maziggy/bambuddy/issues/459)) — When a spool ran empty during a print and the AMS auto-switched to a backup spool, the `on_ams_change` handler eagerly deleted the empty spool's `SpoolAssignment` record (fingerprint mismatch). When `on_print_complete` later ran, it queried `SpoolAssignment` live from the database, found nothing, and silently dropped usage. Now snapshots all spool assignments at print start into the `PrintSession`, so usage is correctly attributed at completion regardless of mid-print AMS changes.
 - **K-Profile Response Race Condition Crash** ([#462](https://github.com/maziggy/bambuddy/issues/462)) — An unsolicited or late K-profile MQTT response could crash the MQTT handler with `AttributeError: 'NoneType' object has no attribute 'set'`. The MQTT callback thread checked `self._pending_kprofile_response` (not None) at line 2698, but between that check and the `.set()` call, the asyncio thread's `finally` block in `get_kprofiles()` could clear the attribute to `None` after a timeout — a classic TOCTOU race. Fixed by capturing the event reference in a local variable before the check.
 - **Queue Stuck on "Busy" for "Any Model" Jobs** ([#435](https://github.com/maziggy/bambuddy/issues/435)) — When a print was queued with "Any [Model]" (e.g., "Any P1S"), it was created with `printer_id=NULL` and `target_model="P1S"`. After the assigned printer finished, the queue widget queried only for items matching `printer_id=X`, missing the next pending model-based item (`printer_id IS NULL`). With no next item found, the "Clear Plate & Start Next" button never appeared, leaving the scheduler stuck reporting "Busy". The queue API now accepts an optional `target_model` parameter; when combined with `printer_id`, it uses OR logic to also return unassigned items whose `target_model` matches the printer's model. The frontend passes the printer's model through to this query. Additionally, the backend now resolves the printer's model server-side from the database when the frontend doesn't provide `target_model` (e.g., when the printer was added without selecting a model), ensuring the OR logic works regardless of whether the client knows the printer's model.
 
@@ -13,6 +14,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Include Beta Updates Setting** — New toggle in Settings → Updates to opt in to beta/prerelease update notifications. Default: off (stable only). The update checker now fetches `/releases` instead of `/releases/latest` and filters by `parse_version()` prerelease detection (not GitHub's `prerelease` flag, which may not be set correctly). Users on the Docker `latest` tag will no longer see notifications for beta releases they can't install.
 
 ### Improved
+- **Spool Assignment Snapshot Test Coverage** — Added 7 backend unit tests covering spool assignment snapshotting at print start, snapshot-preferred spool lookup in both 3MF and AMS delta paths, fallback to live query for pre-upgrade sessions, and the core mid-print unlink scenario from #459.
 - **Background Dispatch Test Coverage** — Added 5 backend unit tests for dispatch cancel races (single-lock TOCTOU fix), batch counter reset re-check, and job lifecycle. Added 2 FTP regression tests for voidresp error handling (upload-loop prevention) and A1 model voidresp skip. Added 1 frontend test for reprint toast suppression.
 - **Frontend Pre-Commit Hooks** ([#458](https://github.com/maziggy/bambuddy/issues/458)) — Added `frontend-typecheck` (`tsc --noEmit`) and `frontend-lint` (`eslint .`) hooks to the pre-commit config. Both hooks only trigger when `frontend/src/**/*.{ts,tsx}` files are staged.
 

backend/app/main.py

@@ -1055,10 +1055,10 @@ async def on_print_start(printer_id: int, data: dict):
             from backend.app.api.routes.settings import get_setting
 
             _spoolman_on = await get_setting(db, "spoolman_enabled")
-        if not _spoolman_on or _spoolman_on.lower() != "true":
-            from backend.app.services.usage_tracker import on_print_start as usage_on_print_start
+            if not _spoolman_on or _spoolman_on.lower() != "true":
+                from backend.app.services.usage_tracker import on_print_start as usage_on_print_start
 
-            await usage_on_print_start(printer_id, data, printer_manager)
+                await usage_on_print_start(printer_id, data, printer_manager, db=db)
     except Exception as e:
         logger.warning("Usage tracker on_print_start failed: %s", e)
 

backend/app/services/usage_tracker.py

@@ -155,14 +155,17 @@ class PrintSession:
     tray_remain_start: dict[tuple[int, int], int] = field(default_factory=dict)
     # tray_now at print start (correct value, unlike at completion where it's 255)
     tray_now_at_start: int = -1
+    # Snapshot of spool assignments at print start: {(ams_id, tray_id): spool_id}
+    # Prevents usage loss when on_ams_change unlinks a spool mid-print
+    spool_assignments: dict[tuple[int, int], int] = field(default_factory=dict)
 
 
 # Module-level storage, keyed by printer_id
 _active_sessions: dict[int, PrintSession] = {}
 
 
-async def on_print_start(printer_id: int, data: dict, printer_manager) -> None:
-    """Capture AMS tray remain% at print start."""
+async def on_print_start(printer_id: int, data: dict, printer_manager, db: AsyncSession | None = None) -> None:
+    """Capture AMS tray remain% and spool assignments at print start."""
     state = printer_manager.get_status(printer_id)
     if not state or not state.raw_data:
         logger.debug("[UsageTracker] No state for printer %d, skipping", printer_id)
@@ -215,6 +218,20 @@ async def on_print_start(printer_id: int, data: dict, printer_manager) -> None:
             )
         logger.info("[UsageTracker] PRINT START printer %d AMS %d: %s", printer_id, ams_id, ", ".join(tray_summary))
 
+    # Snapshot spool assignments so usage isn't lost if on_ams_change unlinks mid-print
+    spool_assignments: dict[tuple[int, int], int] = {}
+    if db:
+        assign_result = await db.execute(select(SpoolAssignment).where(SpoolAssignment.printer_id == printer_id))
+        for assignment in assign_result.scalars().all():
+            spool_assignments[(assignment.ams_id, assignment.tray_id)] = assignment.spool_id
+        if spool_assignments:
+            logger.info(
+                "[UsageTracker] Snapshotted %d spool assignments for printer %d: %s",
+                len(spool_assignments),
+                printer_id,
+                {f"{k[0]}-{k[1]}": v for k, v in spool_assignments.items()},
+            )
+
     # Always create session (even without valid remain data) so print_name
     # is available at completion for 3MF-based tracking
     session = PrintSession(
@@ -223,6 +240,7 @@ async def on_print_start(printer_id: int, data: dict, printer_manager) -> None:
         started_at=datetime.now(timezone.utc),
         tray_remain_start=tray_remain_start,
         tray_now_at_start=tray_now_at_start,
+        spool_assignments=spool_assignments,
     )
     _active_sessions[printer_id] = session
 
@@ -294,6 +312,7 @@ async def on_print_complete(
             tray_now_at_start=session.tray_now_at_start if session else -1,
             last_progress=data.get("last_progress", 0.0),
             last_layer_num=data.get("last_layer_num", 0),
+            spool_assignments=session.spool_assignments if session else None,
         )
         results.extend(threemf_results)
 
@@ -328,20 +347,23 @@ async def on_print_complete(
                     if delta_pct <= 0:
                         continue  # No consumption or tray was refilled
 
-                    # Look up SpoolAssignment for this slot
-                    result = await db.execute(
-                        select(SpoolAssignment).where(
-                            SpoolAssignment.printer_id == printer_id,
-                            SpoolAssignment.ams_id == ams_id,
-                            SpoolAssignment.tray_id == tray_id,
+                    # Look up spool: prefer snapshot (survives mid-print unlink), fall back to live query
+                    spool_id = session.spool_assignments.get(key) if session.spool_assignments else None
+                    if spool_id is None:
+                        result = await db.execute(
+                            select(SpoolAssignment).where(
+                                SpoolAssignment.printer_id == printer_id,
+                                SpoolAssignment.ams_id == ams_id,
+                                SpoolAssignment.tray_id == tray_id,
+                            )
                         )
-                    )
-                    assignment = result.scalar_one_or_none()
-                    if not assignment:
-                        continue
+                        assignment = result.scalar_one_or_none()
+                        if not assignment:
+                            continue
+                        spool_id = assignment.spool_id
 
                     # Load spool
-                    spool_result = await db.execute(select(Spool).where(Spool.id == assignment.spool_id))
+                    spool_result = await db.execute(select(Spool).where(Spool.id == spool_id))
                     spool = spool_result.scalar_one_or_none()
                     if not spool:
                         continue
@@ -405,6 +427,7 @@ async def _track_from_3mf(
     tray_now_at_start: int = -1,
     last_progress: float = 0.0,
     last_layer_num: int = 0,
+    spool_assignments: dict[tuple[int, int], int] | None = None,
 ) -> list[dict]:
     """Track usage from 3MF per-filament slicer data (primary path).
 
@@ -613,21 +636,26 @@ async def _track_from_3mf(
         if key in handled_trays:
             continue
 
-        # Find spool assignment for this tray
-        assign_result = await db.execute(
-            select(SpoolAssignment).where(
-                SpoolAssignment.printer_id == printer_id,
-                SpoolAssignment.ams_id == ams_id,
-                SpoolAssignment.tray_id == tray_id,
+        # Find spool: prefer snapshot (survives mid-print unlink), fall back to live query
+        spool_id = spool_assignments.get(key) if spool_assignments else None
+        if spool_id is None:
+            assign_result = await db.execute(
+                select(SpoolAssignment).where(
+                    SpoolAssignment.printer_id == printer_id,
+                    SpoolAssignment.ams_id == ams_id,
+                    SpoolAssignment.tray_id == tray_id,
+                )
             )
-        )
-        assignment = assign_result.scalar_one_or_none()
-        if not assignment:
-            logger.info("[UsageTracker] 3MF: no spool assignment at printer %d AMS%d-T%d", printer_id, ams_id, tray_id)
-            continue
+            assignment = assign_result.scalar_one_or_none()
+            if not assignment:
+                logger.info(
+                    "[UsageTracker] 3MF: no spool assignment at printer %d AMS%d-T%d", printer_id, ams_id, tray_id
+                )
+                continue
+            spool_id = assignment.spool_id
 
         # Load spool
-        spool_result = await db.execute(select(Spool).where(Spool.id == assignment.spool_id))
+        spool_result = await db.execute(select(Spool).where(Spool.id == spool_id))
         spool = spool_result.scalar_one_or_none()
         if not spool:
             continue

backend/tests/unit/services/test_usage_tracker.py

@@ -399,3 +399,274 @@ class TestTrackFrom3MF:
         assert len(results) == 1
         assert results[0]["ams_id"] == 1
         assert results[0]["tray_id"] == 0
+
+
+class TestSpoolAssignmentSnapshot:
+    """Tests for spool assignment snapshotting at print start (#459).
+
+    When a spool runs empty mid-print, on_ams_change deletes the SpoolAssignment.
+    The snapshot captured at print start ensures usage is still attributed correctly.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _clear_sessions(self):
+        _active_sessions.clear()
+        yield
+        _active_sessions.clear()
+
+    @pytest.mark.asyncio
+    async def test_on_print_start_snapshots_assignments_with_db(self):
+        """on_print_start captures spool assignments when db is provided."""
+        ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}, {"id": 1, "remain": 60}]}]
+        pm = _make_printer_manager(_make_printer_state(ams_data, tray_now=0))
+
+        assignment_0 = _make_assignment(spool_id=10, printer_id=1, ams_id=0, tray_id=0)
+        assignment_1 = _make_assignment(spool_id=20, printer_id=1, ams_id=0, tray_id=1)
+
+        db = AsyncMock()
+        scalars_mock = MagicMock()
+        scalars_mock.all.return_value = [assignment_0, assignment_1]
+        result_mock = MagicMock()
+        result_mock.scalars.return_value = scalars_mock
+        db.execute = AsyncMock(return_value=result_mock)
+
+        await on_print_start(1, {"subtask_name": "Benchy"}, pm, db=db)
+
+        session = _active_sessions[1]
+        assert session.spool_assignments == {(0, 0): 10, (0, 1): 20}
+
+    @pytest.mark.asyncio
+    async def test_on_print_start_empty_snapshot_without_db(self):
+        """on_print_start creates empty snapshot when no db provided."""
+        ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]
+        pm = _make_printer_manager(_make_printer_state(ams_data, tray_now=0))
+
+        await on_print_start(1, {"subtask_name": "Benchy"}, pm)
+
+        session = _active_sessions[1]
+        assert session.spool_assignments == {}
+
+    @pytest.mark.asyncio
+    async def test_3mf_uses_snapshot_instead_of_live_query(self):
+        """_track_from_3mf uses snapshot spool_id without querying SpoolAssignment."""
+        spool = _make_spool(id=42, label_weight=1000)
+        archive = MagicMock()
+        archive.file_path = "archives/test.3mf"
+
+        # db: archive, queue_item(None), spool — NO assignment query needed
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        pm = _make_printer_manager(_make_printer_state([], tray_now=0))
+        filament_usage = [{"slot_id": 1, "used_g": 15.0, "type": "PLA", "color": "#FF0000"}]
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
+        ):
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=10,
+                status="completed",
+                print_name="Test",
+                handled_trays=set(),
+                printer_manager=pm,
+                db=db,
+                spool_assignments={(0, 0): 42},
+            )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 42
+        assert results[0]["weight_used"] == 15.0
+
+    @pytest.mark.asyncio
+    async def test_3mf_falls_back_to_live_query_without_snapshot(self):
+        """_track_from_3mf queries SpoolAssignment when no snapshot exists."""
+        spool = _make_spool(id=5, label_weight=1000)
+        assignment = _make_assignment(spool_id=5)
+        archive = MagicMock()
+        archive.file_path = "archives/test.3mf"
+
+        # db: archive, queue_item(None), assignment, spool
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        pm = _make_printer_manager(_make_printer_state([], tray_now=0))
+        filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": "#FF0000"}]
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
+        ):
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=10,
+                status="completed",
+                print_name="Test",
+                handled_trays=set(),
+                printer_manager=pm,
+                db=db,
+                spool_assignments=None,
+            )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 5
+
+    @pytest.mark.asyncio
+    async def test_ams_delta_uses_snapshot_over_live_query(self):
+        """AMS remain% fallback uses snapshot spool_id instead of live query."""
+        spool = _make_spool(id=77, label_weight=1000)
+
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="Benchy",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 80},
+            spool_assignments={(0, 0): 77},
+        )
+
+        # Current remain = 70% → 10% delta → 100g
+        ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
+        pm = _make_printer_manager(_make_printer_state(ams_data))
+
+        # db only returns spool (NO assignment query)
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        results = await on_print_complete(
+            printer_id=1,
+            data={"status": "completed"},
+            printer_manager=pm,
+            db=db,
+            archive_id=None,
+        )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 77
+        assert results[0]["weight_used"] == 100.0
+
+    @pytest.mark.asyncio
+    async def test_ams_delta_falls_back_to_live_query_without_snapshot(self):
+        """AMS remain% fallback queries SpoolAssignment when snapshot is empty."""
+        spool = _make_spool(id=33, label_weight=1000)
+        assignment = _make_assignment(spool_id=33)
+
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="Benchy",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 80},
+            spool_assignments={},  # Empty snapshot (pre-upgrade session)
+        )
+
+        ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]
+        pm = _make_printer_manager(_make_printer_state(ams_data))
+
+        # db returns assignment then spool
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        results = await on_print_complete(
+            printer_id=1,
+            data={"status": "completed"},
+            printer_manager=pm,
+            db=db,
+            archive_id=None,
+        )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 33
+
+    @pytest.mark.asyncio
+    async def test_snapshot_survives_mid_print_unlink(self):
+        """Core bug scenario: snapshot provides spool_id after mid-print unlink.
+
+        Simulates the #459 scenario: spool runs empty mid-print, on_ams_change
+        deletes the SpoolAssignment, but the snapshot from print start still
+        has the spool_id so usage is correctly attributed at print completion.
+        """
+        spool = _make_spool(id=8, label_weight=1000, weight_used=50)
+        archive = MagicMock()
+        archive.file_path = "archives/big_print.3mf"
+
+        # Session was created at print start WITH snapshot
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="Big Print",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 90},
+            spool_assignments={(0, 0): 8},  # Snapshot from print start
+        )
+
+        pm = _make_printer_manager(
+            _make_printer_state(
+                [{"id": 0, "tray": [{"id": 0, "remain": 75}]}],
+                tray_now=0,
+            )
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 14.2, "type": "PLA", "color": "#FF0000"}]
+
+        # db: archive, queue_item(None), spool
+        # NOTE: No assignment in db — it was deleted by on_ams_change mid-print!
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=filament_usage),
+        ):
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await on_print_complete(
+                printer_id=1,
+                data={"status": "completed"},
+                printer_manager=pm,
+                db=db,
+                archive_id=100,
+            )
+
+        # Usage should be tracked despite assignment being deleted mid-print
+        assert len(results) >= 1
+        assert results[0]["spool_id"] == 8
+        assert results[0]["weight_used"] == 14.2
+        # Spool weight should be updated: 50 + 14.2 = 64.2
+        assert spool.weight_used == 64.2