Fix spool weight tracking when auto-archive disabled for queue/reprint prints (#839)

  When auto_archive was disabled but the print was dispatched by BamBuddy
  (queue/reprint), on_print_start discarded the expected print entry and
  returned early. The archive was never promoted to _active_prints, so at
  completion archive_id and ams_mapping were both None — all tracking paths
  failed silently. Now detects expected prints before the auto-archive
  early-return and falls through to the normal promotion path. Also injects
  the stored ams_mapping into the usage tracker session for printers where
  MQTT request topic subscription fails (P1S, A1).

CHANGELOG.md

@@ -25,7 +25,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Camera Snapshot Temp Files World-Readable** — Camera snapshot and plate detection endpoints created temporary JPEG files in `/tmp` with default 0644 permissions, making them readable by any local user. Switched from `NamedTemporaryFile(delete=False)` to `mkstemp` with explicit 0600 permissions so only the application user can read them. Cleanup was already handled via `finally` blocks. Reported responsibly by Sacha Vaudey via security@bambuddy.cool.
 
 ### Fixed
-- **Spool Weight Not Updated After Print** ([#839](https://github.com/maziggy/bambuddy/issues/839)) — Filament usage tracking failed silently in several scenarios: (1) when FTP download failed and a fallback archive was created without a 3MF file, the primary tracking path was skipped entirely — now falls back to matching the 3MF from the library or a previous archive of the same file; (2) external/VT tray spools were never tracked by the AMS remain% fallback because it only iterated AMS unit trays — now captures and tracks VT tray remain% deltas; (3) notifications showed "Unknown" for time and filament on fallback archives — now enriches notifications with usage tracker results and captures estimated print time from MQTT at archive creation; (4) when auto-archive was disabled, `archive_id` was None at print completion so the entire 3MF tracking path was skipped — now searches library files and previous archives by filename to find the 3MF even without an archive, and captures the AMS slot-to-tray mapping at print start so it's available at completion regardless of archive state.
+- **Spool Weight Not Updated After Print** ([#839](https://github.com/maziggy/bambuddy/issues/839)) — Filament usage tracking failed silently in several scenarios: (1) when FTP download failed and a fallback archive was created without a 3MF file, the primary tracking path was skipped entirely — now falls back to matching the 3MF from the library or a previous archive of the same file; (2) external/VT tray spools were never tracked by the AMS remain% fallback because it only iterated AMS unit trays — now captures and tracks VT tray remain% deltas; (3) notifications showed "Unknown" for time and filament on fallback archives — now enriches notifications with usage tracker results and captures estimated print time from MQTT at archive creation; (4) when auto-archive was disabled, `archive_id` was None at print completion so the entire 3MF tracking path was skipped — now searches library files and previous archives by filename to find the 3MF even without an archive, and captures the AMS slot-to-tray mapping at print start so it's available at completion regardless of archive state; (5) when auto-archive was disabled but the print was dispatched by BamBuddy (queue/reprint), the on_print_start callback discarded the expected print entry and returned early — the archive was never promoted to `_active_prints`, so at completion `archive_id` and `ams_mapping` were both None, making all tracking paths fail. Now detects expected prints before the auto-archive early-return and falls through to the normal promotion path, also injecting the stored `ams_mapping` into the usage tracker session.
 - **File Manager Stale UI After Deleting Folders/Files** — Deleting a folder, file, or bulk-deleting items in the file manager appeared to succeed (toast shown) but the UI didn't update until a page reload. The delete endpoints (`delete_folder`, `delete_file`, `bulk_delete`) relied on FastAPI's dependency cleanup auto-commit which runs after the response is sent — the frontend received the success response, refetched the folder/file list, but the delete hadn't been committed yet. Added explicit `db.commit()` before returning in all three endpoints.
 - **Spool Manager Deducts Double the Filament Used** ([#880](https://github.com/maziggy/bambuddy/issues/880)) — After a print completed, the built-in spool manager subtracted twice the actual filament consumption. The printer's MQTT status message contains both updated AMS remain percentages and the `FINISH` state, which triggered two independent deduction paths in the same event loop cycle: the AMS weight sync (absolute SET from remain%) and the usage tracker (additive delta from 3MF data). The AMS weight sync now skips updates while a print session is active, letting the usage tracker handle deductions precisely via 3MF slicer data.
 - **Thumbnails Broken After Backend Restart** — Archive and library thumbnails returned 401 Unauthorized after a backend restart because stream tokens are stored in memory and lost on restart. The frontend now detects failed token-protected image loads and automatically refreshes the stream token, so thumbnails recover without a page reload.

backend/app/main.py

@@ -1383,44 +1383,54 @@ async def on_print_start(printer_id: int, data: dict):
                 # Don't block print on plate detection errors
                 logger.warning("[PLATE CHECK] Plate detection failed for printer %s: %s", printer_id, plate_err)
 
-        if not printer or not printer.auto_archive:
-            # Send notification without archive data (auto-archive disabled)
-            logger.info(
-                f"[CALLBACK] Skipping archive - printer: {printer is not None}, auto_archive: {printer.auto_archive if printer else 'N/A'}"
-            )
+        if not printer:
+            logger.info("[CALLBACK] Skipping archive - printer not found in database")
             if not notification_sent:
-                # Even with auto-archive disabled, try to recover created_by_id from
-                # a registered expected print (e.g. a library-file queue item) so the
-                # user start email can still be sent.
-                _fn = data.get("filename", "")
-                _sn = data.get("subtask_name", "")
-                _no_archive_creator_keys: list[tuple[int, str]] = []
-                if _sn:
-                    _no_archive_creator_keys += [
-                        (printer_id, _sn),
-                        (printer_id, f"{_sn}.3mf"),
-                        (printer_id, f"{_sn}.gcode.3mf"),
-                    ]
-                if _fn:
-                    _base_fn = _fn.split("/")[-1] if "/" in _fn else _fn
-                    _no_archive_creator_keys.append((printer_id, _base_fn))
-                    _no_archive_base = _base_fn.replace(".gcode", "").replace(".3mf", "")
-                    _no_archive_creator_keys += [
-                        (printer_id, _no_archive_base),
-                        (printer_id, f"{_no_archive_base}.3mf"),
-                    ]
-                _no_archive_creator: int | None = None
-                for _key in _no_archive_creator_keys:
-                    # Clean up all dicts for every key to avoid memory leaks
-                    _expected_prints.pop(_key, None)
-                    _expected_print_registered_at.pop(_key, None)
-                    popped_creator = _expected_print_creators.pop(_key, None)
-                    if _no_archive_creator is None:
-                        _no_archive_creator = popped_creator
-                _creator_data = {"created_by_id": _no_archive_creator} if _no_archive_creator else None
-                await _send_print_start_notification(printer_id, data, _creator_data, logger)
+                await _send_print_start_notification(printer_id, data, logger=logger)
             return
 
+        if not printer.auto_archive:
+            # auto-archive disabled — check if there's an expected print (dispatched
+            # by BamBuddy via queue/reprint) that already has an archive to promote.
+            # If so, fall through to the expected-print handling below so the archive
+            # is tracked in _active_prints and usage tracking works at completion.
+            _fn = data.get("filename", "")
+            _sn = data.get("subtask_name", "")
+            _check_keys: list[tuple[int, str]] = []
+            if _sn:
+                _check_keys += [
+                    (printer_id, _sn),
+                    (printer_id, f"{_sn}.3mf"),
+                    (printer_id, f"{_sn}.gcode.3mf"),
+                ]
+            if _fn:
+                _base_fn = _fn.split("/")[-1] if "/" in _fn else _fn
+                _check_keys.append((printer_id, _base_fn))
+                _no_archive_base = _base_fn.replace(".gcode", "").replace(".3mf", "")
+                _check_keys += [
+                    (printer_id, _no_archive_base),
+                    (printer_id, f"{_no_archive_base}.3mf"),
+                ]
+
+            _has_expected = any(k in _expected_prints for k in _check_keys)
+
+            if not _has_expected:
+                # No expected print — truly external print (started from slicer/touchscreen)
+                logger.info("[CALLBACK] Skipping archive - auto_archive: False, no expected print")
+                if not notification_sent:
+                    _no_archive_creator: int | None = None
+                    for _key in _check_keys:
+                        _expected_prints.pop(_key, None)
+                        _expected_print_registered_at.pop(_key, None)
+                        popped_creator = _expected_print_creators.pop(_key, None)
+                        if _no_archive_creator is None:
+                            _no_archive_creator = popped_creator
+                    _creator_data = {"created_by_id": _no_archive_creator} if _no_archive_creator else None
+                    await _send_print_start_notification(printer_id, data, _creator_data, logger)
+                return
+            else:
+                logger.info("[CALLBACK] auto_archive disabled but expected print found — promoting archive")
+
         # Get the filename and subtask_name
         filename = data.get("filename", "")
         subtask_name = data.get("subtask_name", "")
@@ -1487,6 +1497,21 @@ async def on_print_start(printer_id: int, data: dict):
                 if subtask_name:
                     _active_prints[(printer_id, f"{subtask_name}.3mf")] = archive.id
 
+                # Inject ams_mapping into usage tracker session — the session was created
+                # before expected-print promotion, so it may have ams_mapping=None when
+                # the MQTT request topic subscription failed (common on P1S/A1).
+                _stored_map = _print_ams_mappings.get(expected_archive_id)
+                if _stored_map:
+                    try:
+                        from backend.app.services.usage_tracker import _active_sessions
+
+                        _ut_session = _active_sessions.get(printer_id)
+                        if _ut_session and not _ut_session.ams_mapping:
+                            _ut_session.ams_mapping = _stored_map
+                            logger.info("[CALLBACK] Injected ams_mapping into usage tracker session: %s", _stored_map)
+                    except Exception:
+                        pass
+
                 # Set up energy tracking
                 try:
                     plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))

backend/tests/unit/test_print_start_expected_promotion.py

@@ -0,0 +1,341 @@
+"""Tests for expected print promotion when auto_archive is disabled (#839).
+
+When auto_archive=False but a print was dispatched by BamBuddy (queue/reprint),
+the on_print_start callback must still promote the expected print to _active_prints
+so that at print completion the archive_id and ams_mapping are available for
+filament usage tracking.
+
+These are pure unit tests that verify the module-level dict manipulation logic
+directly, NOT by calling the full on_print_start callback.
+"""
+
+import time
+
+import pytest
+
+from backend.app.main import (
+    _active_prints,
+    _expected_print_creators,
+    _expected_print_registered_at,
+    _expected_prints,
+    _print_ams_mappings,
+    register_expected_print,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_dicts():
+    """Clear module-level tracking dicts before and after each test."""
+    _expected_prints.clear()
+    _expected_print_registered_at.clear()
+    _expected_print_creators.clear()
+    _print_ams_mappings.clear()
+    _active_prints.clear()
+    yield
+    _expected_prints.clear()
+    _expected_print_registered_at.clear()
+    _expected_print_creators.clear()
+    _print_ams_mappings.clear()
+    _active_prints.clear()
+
+
+class TestRegisterExpectedPrint:
+    """Verify register_expected_print populates all tracking dicts."""
+
+    def test_registers_filename_and_variants(self):
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        assert _expected_prints[(1, "Box.3mf")] == 54
+        assert _expected_prints[(1, "Box")] == 54
+        assert _expected_prints[(1, "Box.gcode")] == 54
+
+    def test_stores_ams_mapping(self):
+        register_expected_print(1, "test.3mf", archive_id=10, ams_mapping=[2, -1, 3])
+        assert _print_ams_mappings[10] == [2, -1, 3]
+
+    def test_no_ams_mapping_when_none(self):
+        register_expected_print(1, "test.3mf", archive_id=10, ams_mapping=None)
+        assert 10 not in _print_ams_mappings
+
+    def test_stores_creator(self):
+        register_expected_print(1, "test.3mf", archive_id=10, created_by_id=5)
+        assert _expected_print_creators[(1, "test.3mf")] == 5
+
+    def test_stores_registered_at(self):
+        before = time.monotonic()
+        register_expected_print(1, "test.3mf", archive_id=10)
+        after = time.monotonic()
+
+        ts = _expected_print_registered_at[(1, "test.3mf")]
+        assert before <= ts <= after
+
+
+class TestExpectedPrintDetection:
+    """Verify the expected-print detection logic used in on_print_start.
+
+    Reproduces the key-building and lookup logic from the auto_archive=False
+    block in on_print_start to verify that expected prints are correctly
+    detected across all filename variations.
+    """
+
+    @staticmethod
+    def _build_check_keys(printer_id: int, filename: str, subtask_name: str):
+        """Reproduce the key-building logic from on_print_start."""
+        check_keys = []
+        if subtask_name:
+            check_keys += [
+                (printer_id, subtask_name),
+                (printer_id, f"{subtask_name}.3mf"),
+                (printer_id, f"{subtask_name}.gcode.3mf"),
+            ]
+        if filename:
+            base_fn = filename.split("/")[-1] if "/" in filename else filename
+            check_keys.append((printer_id, base_fn))
+            no_archive_base = base_fn.replace(".gcode", "").replace(".3mf", "")
+            check_keys += [
+                (printer_id, no_archive_base),
+                (printer_id, f"{no_archive_base}.3mf"),
+            ]
+        return check_keys
+
+    def test_detects_expected_print_by_subtask(self):
+        """Expected print is found when subtask_name matches."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+        keys = self._build_check_keys(1, filename="", subtask_name="Box")
+        assert any(k in _expected_prints for k in keys)
+
+    def test_detects_expected_print_by_filename(self):
+        """Expected print is found when filename matches."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+        keys = self._build_check_keys(1, filename="Box.3mf", subtask_name="")
+        assert any(k in _expected_prints for k in keys)
+
+    def test_detects_expected_print_by_gcode_filename(self):
+        """Expected print is found when MQTT reports .gcode filename."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+        # MQTT sometimes reports gcode filename
+        keys = self._build_check_keys(1, filename="Box.gcode", subtask_name="Box")
+        assert any(k in _expected_prints for k in keys)
+
+    def test_no_false_positive_for_different_file(self):
+        """Expected print NOT found for a different filename."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+        keys = self._build_check_keys(1, filename="Benchy.3mf", subtask_name="Benchy")
+        assert not any(k in _expected_prints for k in keys)
+
+    def test_no_false_positive_for_different_printer(self):
+        """Expected print NOT found when printer_id doesn't match."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+        keys = self._build_check_keys(2, filename="Box.3mf", subtask_name="Box")
+        assert not any(k in _expected_prints for k in keys)
+
+    def test_empty_expected_prints_returns_false(self):
+        """No detection when _expected_prints is empty."""
+        keys = self._build_check_keys(1, filename="test.3mf", subtask_name="test")
+        assert not any(k in _expected_prints for k in keys)
+
+    def test_filename_with_spaces_and_parens(self):
+        """Handles filenames with spaces and parentheses (e.g. 'Box3.0_(2)_plate_5.3mf')."""
+        register_expected_print(1, "Box3.0_(2)_plate_5.3mf", archive_id=54, ams_mapping=[1])
+        keys = self._build_check_keys(
+            1,
+            filename="Box3.0_(2)_plate_5.gcode",
+            subtask_name="Box3.0_(2)_plate_5",
+        )
+        assert any(k in _expected_prints for k in keys)
+
+
+class TestExpectedPrintPromotion:
+    """Verify that expected prints are correctly promoted to _active_prints.
+
+    Reproduces the expected-print pop + promotion logic from on_print_start
+    (lines 1468-1496) to verify that _active_prints is populated and
+    _expected_prints is cleaned up.
+    """
+
+    @staticmethod
+    def _simulate_expected_print_promotion(printer_id: int, subtask_name: str, filename: str, archive_filename: str):
+        """Simulate the expected-print lookup and promotion from on_print_start."""
+        expected_keys = []
+        if subtask_name:
+            expected_keys.append((printer_id, subtask_name))
+            expected_keys.append((printer_id, f"{subtask_name}.3mf"))
+            expected_keys.append((printer_id, f"{subtask_name}.gcode.3mf"))
+        if filename:
+            fname = filename.split("/")[-1] if "/" in filename else filename
+            expected_keys.append((printer_id, fname))
+            base = fname.replace(".gcode", "").replace(".3mf", "")
+            expected_keys.append((printer_id, base))
+            expected_keys.append((printer_id, f"{base}.3mf"))
+
+        expected_archive_id = None
+        for key in expected_keys:
+            expected_archive_id = _expected_prints.pop(key, None)
+            _expected_print_registered_at.pop(key, None)
+            if expected_archive_id:
+                for other_key in expected_keys:
+                    _expected_prints.pop(other_key, None)
+                    _expected_print_registered_at.pop(other_key, None)
+                break
+
+        if expected_archive_id:
+            _active_prints[(printer_id, archive_filename)] = expected_archive_id
+            if subtask_name:
+                _active_prints[(printer_id, f"{subtask_name}.3mf")] = expected_archive_id
+
+        return expected_archive_id
+
+    def test_promotion_populates_active_prints(self):
+        """After promotion, archive is in _active_prints."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        archive_id = self._simulate_expected_print_promotion(
+            printer_id=1,
+            subtask_name="Box",
+            filename="Box.gcode",
+            archive_filename="Box.3mf",
+        )
+
+        assert archive_id == 54
+        assert _active_prints[(1, "Box.3mf")] == 54
+
+    def test_promotion_cleans_up_expected_prints(self):
+        """After promotion, _expected_prints is empty for this print."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        self._simulate_expected_print_promotion(
+            printer_id=1,
+            subtask_name="Box",
+            filename="Box.gcode",
+            archive_filename="Box.3mf",
+        )
+
+        # All variants should be cleaned up
+        assert (1, "Box.3mf") not in _expected_prints
+        assert (1, "Box") not in _expected_prints
+        assert (1, "Box.gcode") not in _expected_prints
+
+    def test_ams_mapping_survives_promotion(self):
+        """_print_ams_mappings is NOT consumed during promotion — it's needed at completion."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        self._simulate_expected_print_promotion(
+            printer_id=1,
+            subtask_name="Box",
+            filename="Box.gcode",
+            archive_filename="Box.3mf",
+        )
+
+        # ams_mapping should still be available for on_print_complete
+        assert _print_ams_mappings[54] == [1]
+
+    def test_completion_lookup_finds_promoted_archive(self):
+        """Simulate on_print_complete finding the archive in _active_prints."""
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        self._simulate_expected_print_promotion(
+            printer_id=1,
+            subtask_name="Box",
+            filename="Box.gcode",
+            archive_filename="Box.3mf",
+        )
+
+        # Simulate on_print_complete key building
+        completion_keys = [
+            (1, "Box.3mf"),
+            (1, "Box.gcode.3mf"),
+            (1, "Box"),
+        ]
+        found_id = None
+        for key in completion_keys:
+            found_id = _active_prints.pop(key, None)
+            if found_id:
+                break
+
+        assert found_id == 54
+        # And ams_mapping is retrievable
+        assert _print_ams_mappings.pop(54, None) == [1]
+
+    def test_no_promotion_for_external_print(self):
+        """When no expected print exists, nothing is promoted."""
+        archive_id = self._simulate_expected_print_promotion(
+            printer_id=1,
+            subtask_name="Benchy",
+            filename="Benchy.gcode",
+            archive_filename="Benchy.3mf",
+        )
+
+        assert archive_id is None
+        assert len(_active_prints) == 0
+
+
+class TestAMSMappingInjection:
+    """Verify ams_mapping injection into usage tracker session."""
+
+    def test_injection_into_session(self):
+        """ams_mapping from _print_ams_mappings is injectable into a session."""
+        from datetime import datetime, timezone
+
+        from backend.app.services.usage_tracker import PrintSession, _active_sessions
+
+        _active_sessions.clear()
+
+        # Create a session without ams_mapping (simulates MQTT not providing it)
+        session = PrintSession(
+            printer_id=1,
+            print_name="Box",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={},
+            tray_now_at_start=-1,
+            spool_assignments={},
+            ams_mapping=None,
+        )
+        _active_sessions[1] = session
+
+        # Register expected print with ams_mapping
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        # Simulate the injection logic from on_print_start
+        _stored_map = _print_ams_mappings.get(54)
+        assert _stored_map == [1]
+
+        ut_session = _active_sessions.get(1)
+        assert ut_session is not None
+        assert ut_session.ams_mapping is None  # before injection
+
+        ut_session.ams_mapping = _stored_map  # injection
+        assert ut_session.ams_mapping == [1]
+
+        _active_sessions.clear()
+
+    def test_no_injection_when_session_already_has_mapping(self):
+        """Don't overwrite existing ams_mapping in session."""
+        from datetime import datetime, timezone
+
+        from backend.app.services.usage_tracker import PrintSession, _active_sessions
+
+        _active_sessions.clear()
+
+        session = PrintSession(
+            printer_id=1,
+            print_name="Box",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={},
+            tray_now_at_start=-1,
+            spool_assignments={},
+            ams_mapping=[5, 6],  # already has mapping from MQTT
+        )
+        _active_sessions[1] = session
+
+        register_expected_print(1, "Box.3mf", archive_id=54, ams_mapping=[1])
+
+        _stored_map = _print_ams_mappings.get(54)
+        ut_session = _active_sessions.get(1)
+
+        # Guard: don't overwrite if session already has a mapping
+        if ut_session and not ut_session.ams_mapping:
+            ut_session.ams_mapping = _stored_map
+
+        assert ut_session.ams_mapping == [5, 6]  # unchanged
+
+        _active_sessions.clear()