fix(usage): scope 3MF filament tracking to dispatched plate (#1697)

  When a print targets a single plate from a multi-plate 3MF, both the
  internal Filament Inventory tracker and the Spoolman-mode tracker parsed
  the 3MF without a plate filter and summed every plate's filament — so a
  single lid print debited the spool the entire file's grey + black totals.

  The 3MF parser already supports plate_id (queue pre-flight uses it at
  print_queue.py:254/:286). Plumbed it through both dispatch paths:

  Queue path:
  - PrintSession gains a plate_id field; on_print_start queries the
    printer's currently-printing queue row and records queue_item.plate_id
    onto the session.
  - _track_from_3mf accepts plate_id and passes it to the extractor.
  - store_print_data moves its existing queue-item lookup above the
    extract and uses queue_item.plate_id as the plate filter.

  Direct-Print path (reprintArchive / printLibraryFile — never goes
  through the queue):
  - _print_plate_ids dict added in main.py, parallel to _print_ams_mappings.
  - register_expected_print accepts plate_id and stores it; the 2 sites in
    background_dispatch.py and the 1 site in print_scheduler.py now pass
    it (resolve was already happening, just needed reordering before the
    register call so the value is available).
  - Expected-print promotion in main.py injects _print_plate_ids[archive_id]
    into the session, guarded so a queue capture wins over the dict.
  - _get_start_plate_id helper feeds plate_id into all 3
    _store_spoolman_print_data call sites; spoolman_tracking.store_print_data
    takes the caller value first, falls back to queue_item.plate_id.

  PrintArchive.filament_used_grams stays file-level summed by design
  (#1593's contract — the archive describes the file, not the run); only
  the per-run usage attribution becomes plate-aware. Single-plate direct
  prints resolve to plate_id=1 → plate 1 = whole file, identical to the
  prior no-filter behaviour.

backend/app/main.py

@@ -349,6 +349,13 @@ _expected_prints: dict[tuple[int, str], int] = {}
 # Used by usage tracker to map 3MF slots to physical AMS trays
 _print_ams_mappings: dict[int, list[int]] = {}
 
+# Track plate_id for prints from multi-plate 3MFs: {archive_id: plate_id}
+# Used by usage tracker to scope 3MF parsing to the dispatched plate (#1697).
+# Populated by direct-Print and queue dispatch paths; queue prints also have a
+# redundant queue-item lookup in on_print_start so this dict isn't load-bearing
+# for the queue path. Cleared on print completion or TTL eviction.
+_print_plate_ids: dict[int, int] = {}
+
 # Track progress milestones for notifications: {printer_id: last_milestone_notified}
 # Milestones are 25, 50, 75. Value of 0 means no milestone notified yet for current print.
 _last_progress_milestone: dict[int, int] = {}
@@ -567,6 +574,7 @@ def register_expected_print(
     archive_id: int,
     ams_mapping: list[int] | None = None,
     created_by_id: int | None = None,
+    plate_id: int | None = None,
 ):
     """Register an expected print from reprint/scheduled so we don't create duplicate archives."""
     # Store with multiple filename variations to catch different naming patterns
@@ -579,6 +587,11 @@ def register_expected_print(
     # Store AMS mapping for usage tracking at print completion
     if ams_mapping is not None:
         _print_ams_mappings[archive_id] = ams_mapping
+    # Store plate_id for usage tracking when this is a single-plate dispatch from
+    # a multi-plate 3MF — without this, the direct-Print path attributes the whole
+    # file's filament total to the spool instead of just the printed plate (#1697).
+    if plate_id is not None:
+        _print_plate_ids[archive_id] = plate_id
     # Store created_by_id so the user start email can be sent even when the archive
     # itself has no created_by_id (e.g. library-file-based queue prints)
     if created_by_id is not None:
@@ -595,7 +608,7 @@ def register_expected_print(
         _expected_print_registered_at[(printer_id, base)] = _registered_at
         _expected_print_registered_at[(printer_id, f"{base}.gcode")] = _registered_at
     logging.getLogger(__name__).info(
-        f"Registered expected print: printer={printer_id}, file={filename}, archive={archive_id}, ams_mapping={ams_mapping}"
+        f"Registered expected print: printer={printer_id}, file={filename}, archive={archive_id}, ams_mapping={ams_mapping}, plate_id={plate_id}"
     )
 
 
@@ -639,6 +652,19 @@ def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | No
     return stored_ams_mapping
 
 
+def _get_start_plate_id(archive_id: int | None) -> int | None:
+    """Resolve plate_id for print start without consuming stored direct-Print state.
+
+    Direct-Print of a single plate from a multi-plate 3MF registers plate_id in
+    ``_print_plate_ids`` at dispatch time; this lets the spoolman / usage tracker
+    read it back at print-start without popping (the entry is popped on print
+    completion or TTL eviction, mirroring ``_print_ams_mappings``).
+    """
+    if archive_id is None:
+        return None
+    return _print_plate_ids.get(archive_id)
+
+
 def _extract_filament_data_from_mqtt(data: dict, ams_mapping: list[int] | None = None) -> dict[str, str]:
     """Best-effort filament metadata from the MQTT print-start snapshot.
 
@@ -2215,14 +2241,21 @@ async def on_print_start(printer_id: int, data: dict):
                 # 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:
+                _stored_plate_id = _print_plate_ids.get(expected_archive_id)
+                if _stored_map or _stored_plate_id is not None:
                     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:
+                        if _ut_session and _stored_map 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)
+                        # plate_id injection covers direct-Print of plate N of a multi-plate
+                        # 3MF — queue prints already capture it via the on_print_start queue
+                        # lookup, but direct-Print never goes through the queue (#1697).
+                        if _ut_session and _stored_plate_id is not None and _ut_session.plate_id is None:
+                            _ut_session.plate_id = _stored_plate_id
+                            logger.info("[CALLBACK] Injected plate_id into usage tracker session: %s", _stored_plate_id)
                     except Exception:
                         pass
 
@@ -2265,6 +2298,7 @@ async def on_print_start(printer_id: int, data: dict):
                         db,
                         printer_manager,
                         ams_mapping=_get_start_ams_mapping(data, archive.id),
+                        plate_id=_get_start_plate_id(archive.id),
                     )
                 except Exception as e:
                     logger.warning("[SPOOLMAN] Failed to store tracking data: %s", e)
@@ -2798,6 +2832,7 @@ async def on_print_start(printer_id: int, data: dict):
                         db,
                         printer_manager,
                         ams_mapping=_get_start_ams_mapping(data, fallback_archive.id),
+                        plate_id=_get_start_plate_id(fallback_archive.id),
                     )
                 except Exception as e:
                     logger.debug("[SPOOLMAN] Could not store tracking for fallback archive: %s", e)
@@ -2899,6 +2934,7 @@ async def on_print_start(printer_id: int, data: dict):
                         db,
                         printer_manager,
                         ams_mapping=_get_start_ams_mapping(data, archive.id),
+                        plate_id=_get_start_plate_id(archive.id),
                     )
                 except Exception as e:
                     logger.warning("[SPOOLMAN] Failed to store tracking data: %s", e)
@@ -3923,6 +3959,12 @@ async def on_print_complete(printer_id: int, data: dict):
     if not stored_ams_mapping and archive_id:
         stored_ams_mapping = _print_ams_mappings.pop(archive_id, None)
 
+    # Always drain the plate_id register on completion — the session already
+    # consumed it at print-start injection; leaving it would leak into the next
+    # print on the same archive_id (rare but possible with reprints) (#1697).
+    if archive_id:
+        _print_plate_ids.pop(archive_id, None)
+
     # Internal inventory: track AMS remain% deltas (skip if Spoolman handles usage)
     try:
         async with async_session() as db:
@@ -5126,12 +5168,14 @@ def _evict_stale_expected_prints() -> None:
         _expected_print_creators.pop(key, None)
         _expected_print_registered_at.pop(key, None)
 
-    # Also clean up _print_ams_mappings for archive_ids that have no remaining
-    # live keys in _expected_prints (i.e. all variants were just evicted).
+    # Also clean up _print_ams_mappings and _print_plate_ids for archive_ids
+    # that have no remaining live keys in _expected_prints (all variants
+    # were just evicted).
     live_archive_ids = set(_expected_prints.values())
     for archive_id in evicted_archive_ids:
         if archive_id not in live_archive_ids:
             _print_ams_mappings.pop(archive_id, None)
+            _print_plate_ids.pop(archive_id, None)
 
     logging.getLogger(__name__).info(
         "Evicted %d stale expected-print entries (TTL=%ds)", len(stale_keys), _EXPECTED_PRINT_TTL_SECONDS

backend/app/services/background_dispatch.py

@@ -670,15 +670,19 @@ class BackgroundDispatchService:
                         "Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT)."
                     )
 
+                # Resolve plate_id before register so usage tracking can scope the
+                # 3MF parse to the dispatched plate at print-start (#1697). Pure
+                # transform of file_path + options, safe to reorder.
+                plate_id = self._resolve_plate_id(file_path, job.options.get("plate_id"))
+
                 register_expected_print(
                     job.printer_id,
                     remote_filename,
                     job.source_id,
                     ams_mapping=job.options.get("ams_mapping"),
+                    plate_id=plate_id,
                 )
 
-                plate_id = self._resolve_plate_id(file_path, job.options.get("plate_id"))
-
                 self._raise_if_cancel_requested(job)
 
                 effective_timelapse = await self._resolve_effective_timelapse(db, archive, job)
@@ -874,15 +878,18 @@ class BackgroundDispatchService:
                         "Failed to upload file to printer. Check if SD card is inserted and properly formatted (FAT32/exFAT)."
                     )
 
+                # Resolve plate_id before register so usage tracking can scope the
+                # 3MF parse to the dispatched plate at print-start (#1697).
+                plate_id = self._resolve_plate_id(file_path, job.options.get("plate_id"))
+
                 register_expected_print(
                     job.printer_id,
                     remote_filename,
                     archive.id,
                     ams_mapping=job.options.get("ams_mapping"),
+                    plate_id=plate_id,
                 )
 
-                plate_id = self._resolve_plate_id(file_path, job.options.get("plate_id"))
-
                 self._raise_if_cancel_requested(job)
 
                 effective_timelapse = await self._resolve_effective_timelapse(db, archive, job)

backend/app/services/print_scheduler.py

@@ -2143,6 +2143,7 @@ class PrintScheduler:
                 archive.id,
                 ams_mapping=ams_mapping,
                 created_by_id=item.created_by_id,
+                plate_id=item.plate_id,
             )
 
         # Propagate the queue item's owner into printer_manager so the

backend/app/services/spoolman_tracking.py

@@ -189,6 +189,7 @@ async def store_print_data(
     db,
     printer_manager,
     ams_mapping: list[int] | None = None,
+    plate_id: int | None = None,
 ):
     """Store Spoolman tracking data at print start (persisted to database).
 
@@ -196,6 +197,13 @@ async def store_print_data(
     how the internal Filament Inventory works. The legacy AMS-remain%-based sync
     is no longer used as a weight writer (#1119), so this runs whenever Spoolman
     is enabled regardless of the deprecated `spoolman_disable_weight_sync` flag.
+
+    ``plate_id``, when set, scopes the 3MF filament extract to a single plate so
+    queue / direct-Print dispatch of plate N of a multi-plate file doesn't
+    attribute every plate's filament to the printed spool (#1697). When unset,
+    the queue item's plate_id (if any) is used; otherwise the whole-file sum is
+    extracted, which is correct for direct prints that target the first/only
+    plate of a single-plate file.
     """
     from backend.app.api.routes.settings import get_setting
     from backend.app.models.active_print_spoolman import ActivePrintSpoolman
@@ -219,8 +227,20 @@ async def store_print_data(
         logger.debug("[SPOOLMAN] 3MF file not found: %s", full_path)
         return
 
-    # Extract per-filament usage from 3MF (total usage per slot)
-    filament_usage = extract_filament_usage_from_3mf(full_path)
+    # Resolve the queue item once — used both for the plate-scoped 3MF parsing
+    # fallback (#1697: multi-plate file dispatched for one plate must only count
+    # that plate's filament) and for the ams_mapping fallback below.
+    queue_result = await db.execute(
+        select(PrintQueueItem).where(PrintQueueItem.archive_id == archive_id).where(PrintQueueItem.status == "printing")
+    )
+    queue_item = queue_result.scalar_one_or_none()
+    # Caller-supplied plate_id wins (direct-Print path); fall back to the queue
+    # item's plate_id (queue dispatch path).
+    effective_plate_id = plate_id if plate_id is not None else (queue_item.plate_id if queue_item is not None else None)
+
+    # Extract per-filament usage from 3MF (total usage for the dispatched plate,
+    # or the whole file for direct/library prints with no plate_id).
+    filament_usage = extract_filament_usage_from_3mf(full_path, effective_plate_id)
     if not filament_usage:
         logger.debug("[SPOOLMAN] No filament usage data in 3MF for archive %s", archive_id)
         return
@@ -234,18 +254,11 @@ async def store_print_data(
     # Prefer the explicit mapping captured from the print command, then fall back
     # to any queue mapping stored for scheduled/reprint jobs.
     slot_to_tray = ams_mapping if ams_mapping is not None else None
-    if not slot_to_tray:
-        queue_result = await db.execute(
-            select(PrintQueueItem)
-            .where(PrintQueueItem.archive_id == archive_id)
-            .where(PrintQueueItem.status == "printing")
-        )
-        queue_item = queue_result.scalar_one_or_none()
-        if queue_item and queue_item.ams_mapping:
-            try:
-                slot_to_tray = json.loads(queue_item.ams_mapping)
-            except json.JSONDecodeError:
-                pass  # Ignore malformed AMS mapping; fall back to default slot assignment
+    if not slot_to_tray and queue_item and queue_item.ams_mapping:
+        try:
+            slot_to_tray = json.loads(queue_item.ams_mapping)
+        except json.JSONDecodeError:
+            pass  # Ignore malformed AMS mapping; fall back to default slot assignment
 
     # Parse G-code for per-layer filament usage (for accurate partial usage tracking)
     layer_usage = extract_layer_filament_usage_from_3mf(full_path)

backend/app/services/usage_tracker.py

@@ -214,6 +214,10 @@ class PrintSession:
     spool_assignments: dict[tuple[int, int], int] = field(default_factory=dict)
     # AMS mapping from print command (captured at start, needed when auto-archive is off)
     ams_mapping: list[int] | None = None
+    # Queue item's plate_id when this print is a multi-plate 3MF dispatched for a
+    # single plate (#1697). None for non-queue prints — the file's first/only plate
+    # is the default and the 3MF parser already returns the full file in that case.
+    plate_id: int | None = None
 
 
 # Module-level storage, keyed by printer_id
@@ -380,6 +384,21 @@ async def on_print_start(printer_id: int, data: dict, printer_manager, db: Async
                 {f"{k[0]}-{k[1]}": v for k, v in spool_assignments.items()},
             )
 
+    # Capture the queue item's plate_id so 3MF parsing at completion is scoped to
+    # the plate that actually ran, not the whole multi-plate file (#1697).
+    plate_id: int | None = None
+    if db:
+        from backend.app.models.print_queue import PrintQueueItem
+
+        queue_result = await db.execute(
+            select(PrintQueueItem)
+            .where(PrintQueueItem.printer_id == printer_id)
+            .where(PrintQueueItem.status == "printing")
+        )
+        queue_item = queue_result.scalars().first()
+        if queue_item is not None:
+            plate_id = queue_item.plate_id
+
     # Always create session (even without valid remain data) so print_name
     # is available at completion for 3MF-based tracking
     session = PrintSession(
@@ -390,6 +409,7 @@ async def on_print_start(printer_id: int, data: dict, printer_manager, db: Async
         tray_now_at_start=tray_now_at_start,
         spool_assignments=spool_assignments,
         ams_mapping=data.get("ams_mapping"),
+        plate_id=plate_id,
     )
     _active_sessions[printer_id] = session
 
@@ -490,6 +510,7 @@ async def on_print_complete(
             spool_assignments=session.spool_assignments if session else None,
             print_started_at=session.started_at if session else None,
             threemf_path=threemf_path,
+            plate_id=session.plate_id if session else None,
         )
         results.extend(threemf_results)
 
@@ -855,6 +876,7 @@ async def _track_from_3mf(
     spool_assignments: dict[tuple[int, int], int] | None = None,
     print_started_at: datetime | None = None,
     threemf_path=None,
+    plate_id: int | None = None,
 ) -> list[dict]:
     """Track usage from 3MF per-filament slicer data (primary path).
 
@@ -865,6 +887,11 @@ async def _track_from_3mf(
     When archive_id is None (auto-archive disabled), a pre-resolved threemf_path
     can be provided to still track filament usage from slicer data.
 
+    When ``plate_id`` is set (queue prints of a single plate from a multi-plate
+    3MF), only that plate's filaments contribute. Without it the 3MF parser sums
+    every plate, which is correct for direct/library Print flows that always
+    target the first or only plate (#1697).
+
     Slot-to-tray mapping priority:
     1. Stored ams_mapping from print command (reprints/direct prints)
     2. MQTT mapping field from printer state (universal, all print sources)
@@ -904,12 +931,12 @@ async def _track_from_3mf(
         logger.info("[UsageTracker] 3MF: no file available for archive %s, skipping", archive_id)
         return []
 
-    filament_usage = extract_filament_usage_from_3mf(file_path)
+    filament_usage = extract_filament_usage_from_3mf(file_path, plate_id)
     if not filament_usage:
         logger.info("[UsageTracker] 3MF: no filament usage data in %s", file_path)
         return []
 
-    logger.info("[UsageTracker] 3MF: archive %s, filament_usage=%s", archive_id, filament_usage)
+    logger.info("[UsageTracker] 3MF: archive %s, plate_id=%s, filament_usage=%s", archive_id, plate_id, filament_usage)
 
     # --- Resolve slot-to-tray mapping ---
     mapping_source = None

backend/tests/unit/services/test_spoolman_tracking.py

@@ -1,5 +1,6 @@
 """Unit tests for Spoolman tracking service helpers."""
 
+import json
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -213,8 +214,14 @@ class TestStorePrintData:
     @pytest.mark.asyncio
     async def test_prefers_explicit_ams_mapping_over_queue_mapping(self):
         db = AsyncMock()
+        # store_print_data now queries the queue item unconditionally (to pick up
+        # plate_id for multi-plate 3MFs, #1697), then deletes any stale spoolman
+        # row before inserting the new one. Two execute calls in that order.
+        queue_item = SimpleNamespace(ams_mapping=json.dumps([2, -1, -1, -1]), plate_id=None)
+        queue_result = MagicMock()
+        queue_result.scalar_one_or_none.return_value = queue_item
         delete_result = MagicMock()
-        db.execute = AsyncMock(side_effect=[delete_result])
+        db.execute = AsyncMock(side_effect=[queue_result, delete_result])
         db.add = MagicMock()
         db.commit = AsyncMock()
 
@@ -250,7 +257,7 @@ class TestStorePrintData:
         db.add.assert_called_once()
         tracking = db.add.call_args.args[0]
         assert tracking.slot_to_tray == [1, -1, -1, -1]
-        db.execute.assert_called_once()
+        assert db.execute.await_count == 2
 
     @pytest.mark.asyncio
     async def test_stores_tracking_when_disable_weight_sync_is_false(self):

backend/tests/unit/test_print_start_expected_promotion.py

@@ -18,7 +18,9 @@ from backend.app.main import (
     _expected_print_creators,
     _expected_print_registered_at,
     _expected_prints,
+    _get_start_plate_id,
     _print_ams_mappings,
+    _print_plate_ids,
     register_expected_print,
 )
 
@@ -30,12 +32,14 @@ def _clear_dicts():
     _expected_print_registered_at.clear()
     _expected_print_creators.clear()
     _print_ams_mappings.clear()
+    _print_plate_ids.clear()
     _active_prints.clear()
     yield
     _expected_prints.clear()
     _expected_print_registered_at.clear()
     _expected_print_creators.clear()
     _print_ams_mappings.clear()
+    _print_plate_ids.clear()
     _active_prints.clear()
 
 
@@ -69,6 +73,24 @@ class TestRegisterExpectedPrint:
         ts = _expected_print_registered_at[(1, "test.3mf")]
         assert before <= ts <= after
 
+    def test_stores_plate_id(self):
+        """plate_id is registered so usage tracking can scope multi-plate 3MFs (#1697)."""
+        register_expected_print(1, "test.3mf", archive_id=10, plate_id=2)
+        assert _print_plate_ids[10] == 2
+
+    def test_no_plate_id_when_none(self):
+        """Direct-Print of a single-plate file passes plate_id=None; nothing stored."""
+        register_expected_print(1, "test.3mf", archive_id=10, plate_id=None)
+        assert 10 not in _print_plate_ids
+
+    def test_get_start_plate_id_reads_back(self):
+        register_expected_print(1, "test.3mf", archive_id=10, plate_id=3)
+        assert _get_start_plate_id(10) == 3
+
+    def test_get_start_plate_id_returns_none_for_unregistered(self):
+        assert _get_start_plate_id(10) is None
+        assert _get_start_plate_id(None) is None
+
 
 class TestExpectedPrintDetection:
     """Verify the expected-print detection logic used in on_print_start.
@@ -339,3 +361,79 @@ class TestAMSMappingInjection:
         assert ut_session.ams_mapping == [5, 6]  # unchanged
 
         _active_sessions.clear()
+
+
+class TestPlateIdInjection:
+    """Verify plate_id injection into usage tracker session for direct-Print of
+    a non-first plate from a multi-plate 3MF (#1697)."""
+
+    def test_injection_into_session(self):
+        """plate_id from _print_plate_ids gets injected when session has none."""
+        from datetime import datetime, timezone
+
+        from backend.app.services.usage_tracker import PrintSession, _active_sessions
+
+        _active_sessions.clear()
+
+        # Session created by on_print_start before expected-print promotion;
+        # plate_id is None because no queue item was found (direct-Print path).
+        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,
+            plate_id=None,
+        )
+        _active_sessions[1] = session
+
+        register_expected_print(1, "Box.3mf", archive_id=54, plate_id=2)
+
+        # Mirror the injection branch from main.py.
+        _stored_plate_id = _print_plate_ids.get(54)
+        assert _stored_plate_id == 2
+
+        ut_session = _active_sessions.get(1)
+        assert ut_session is not None
+        assert ut_session.plate_id is None  # before injection
+
+        ut_session.plate_id = _stored_plate_id  # injection
+        assert ut_session.plate_id == 2
+
+        _active_sessions.clear()
+
+    def test_no_injection_when_session_already_has_plate_id(self):
+        """Queue path: on_print_start already captured plate_id from queue_item;
+        don't overwrite with the dict value."""
+        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=None,
+            plate_id=3,  # captured from queue_item by on_print_start
+        )
+        _active_sessions[1] = session
+
+        register_expected_print(1, "Box.3mf", archive_id=54, plate_id=2)
+
+        _stored_plate_id = _print_plate_ids.get(54)
+        ut_session = _active_sessions.get(1)
+
+        # Guard: don't overwrite if session already has a plate_id
+        if ut_session and ut_session.plate_id is None:
+            ut_session.plate_id = _stored_plate_id
+
+        assert ut_session.plate_id == 3  # queue value preserved
+
+        _active_sessions.clear()

backend/tests/unit/test_spoolman_tracking.py

@@ -1,5 +1,6 @@
 """Unit tests for Spoolman tracking service helpers."""
 
+import json
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -183,8 +184,14 @@ class TestStorePrintData:
     @pytest.mark.asyncio
     async def test_prefers_explicit_ams_mapping_over_queue_mapping(self):
         db = AsyncMock()
+        # store_print_data now queries the queue item unconditionally (to pick up
+        # plate_id for multi-plate 3MFs, #1697), then deletes any stale spoolman
+        # row before inserting the new one. Two execute calls in that order.
+        queue_item = SimpleNamespace(ams_mapping=json.dumps([2, -1, -1, -1]), plate_id=None)
+        queue_result = MagicMock()
+        queue_result.scalar_one_or_none.return_value = queue_item
         delete_result = MagicMock()
-        db.execute = AsyncMock(side_effect=[delete_result])
+        db.execute = AsyncMock(side_effect=[queue_result, delete_result])
         db.add = MagicMock()
         db.commit = AsyncMock()
 
@@ -220,4 +227,48 @@ class TestStorePrintData:
         db.add.assert_called_once()
         tracking = db.add.call_args.args[0]
         assert tracking.slot_to_tray == [1, -1, -1, -1]
-        db.execute.assert_called_once()
+        assert db.execute.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_passes_queue_plate_id_to_3mf_extract(self):
+        """Multi-plate 3MFs queued for one plate must only count that plate's filament (#1697)."""
+        db = AsyncMock()
+        queue_item = SimpleNamespace(ams_mapping=None, plate_id=2)
+        queue_result = MagicMock()
+        queue_result.scalar_one_or_none.return_value = queue_item
+        delete_result = MagicMock()
+        db.execute = AsyncMock(side_effect=[queue_result, delete_result])
+        db.add = MagicMock()
+        db.commit = AsyncMock()
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]}
+        )
+
+        mock_settings = MagicMock()
+        mock_path = MagicMock()
+        mock_path.exists.return_value = True
+        mock_settings.base_dir.__truediv__.return_value = mock_path
+
+        extract_mock = MagicMock(return_value=[{"slot_id": 1, "used_g": 190.0, "type": "PETG", "color": "#888888"}])
+
+        with (
+            patch("backend.app.services.spoolman_tracking.app_settings", mock_settings),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(side_effect=["true", "true"])),
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", extract_mock),
+            patch("backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf", return_value=None),
+            patch("backend.app.utils.threemf_tools.extract_filament_properties_from_3mf", return_value={}),
+        ):
+            await store_print_data(
+                printer_id=1,
+                archive_id=15,
+                file_path="archives/test.3mf",
+                db=db,
+                printer_manager=printer_manager,
+                ams_mapping=[1, -1, -1, -1],
+            )
+
+        # plate_id=2 must be passed as the second positional arg
+        assert extract_mock.call_count == 1
+        assert extract_mock.call_args.args[1] == 2

backend/tests/unit/test_usage_tracker.py

@@ -56,11 +56,12 @@ def _make_archive(archive_id=1, file_path="archives/1/test.3mf", extra_data=None
     return archive
 
 
-def _make_queue_item(ams_mapping=None, status="printing"):
+def _make_queue_item(ams_mapping=None, status="printing", plate_id=None):
     """Create a mock PrintQueueItem object."""
     item = MagicMock()
     item.ams_mapping = ams_mapping
     item.status = status
+    item.plate_id = plate_id
     return item
 
 
@@ -2008,6 +2009,48 @@ class TestOnPrintStartAmsMapping:
 
         assert _active_sessions[1].ams_mapping is None
 
+    @pytest.mark.asyncio
+    async def test_captures_queue_plate_id(self):
+        """on_print_start records the queue item's plate_id onto the session (#1697)."""
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
+            tray_now=0,
+        )
+
+        queue_item = _make_queue_item(plate_id=2)
+        # on_print_start now executes: SpoolAssignment lookup, then PrintQueueItem lookup.
+        db = AsyncMock()
+        assignment_result = MagicMock()
+        assignment_result.scalars.return_value.all.return_value = []
+        queue_result = MagicMock()
+        queue_result.scalars.return_value.first.return_value = queue_item
+        db.execute = AsyncMock(side_effect=[assignment_result, queue_result])
+
+        await on_print_start(1, {"subtask_name": "Test"}, printer_manager, db=db)
+
+        assert _active_sessions[1].plate_id == 2
+
+    @pytest.mark.asyncio
+    async def test_plate_id_none_when_no_queue_item(self):
+        """Direct/library prints with no queue item leave session.plate_id = None."""
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
+            tray_now=0,
+        )
+
+        db = AsyncMock()
+        assignment_result = MagicMock()
+        assignment_result.scalars.return_value.all.return_value = []
+        queue_result = MagicMock()
+        queue_result.scalars.return_value.first.return_value = None
+        db.execute = AsyncMock(side_effect=[assignment_result, queue_result])
+
+        await on_print_start(1, {"subtask_name": "Test"}, printer_manager, db=db)
+
+        assert _active_sessions[1].plate_id is None
+
 
 class TestFindThreemfByFilename:
     """Tests for _find_3mf_by_filename() — library/archive search without archive_id."""
@@ -2205,3 +2248,91 @@ class TestTrackFrom3mfWithPreresolvedPath:
 
         assert len(results) == 1
         assert results[0]["weight_used"] == 2.0
+
+
+class TestTrackFrom3mfPlateId:
+    """plate_id must propagate from PrintSession through _track_from_3mf to the
+    3MF parser, so multi-plate files dispatched for one plate only count that
+    plate's filament (#1697)."""
+
+    @pytest.mark.asyncio
+    async def test_passes_plate_id_to_3mf_extract(self):
+        spool = _make_spool(spool_id=1, label_weight=1000)
+        assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
+
+        db = _mock_db_sequential([assignment, spool])
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": []}]},
+            tray_now=0,
+            last_loaded_tray=0,
+            tray_change_log=[],
+        )
+
+        extract_mock = MagicMock(return_value=[{"slot_id": 1, "used_g": 190.0, "type": "PETG", "color": "#888888"}])
+
+        with (
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", extract_mock),
+            patch("backend.app.core.config.settings") as mock_settings,
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+
+            await _track_from_3mf(
+                printer_id=1,
+                archive_id=None,
+                status="completed",
+                print_name="GridfinityLid",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                tray_now_at_start=0,
+                threemf_path=mock_path,
+                plate_id=2,
+            )
+
+        # plate_id=2 passed positionally as second arg
+        assert extract_mock.call_count == 1
+        assert extract_mock.call_args.args[1] == 2
+
+    @pytest.mark.asyncio
+    async def test_plate_id_none_for_non_queue_print(self):
+        spool = _make_spool(spool_id=1, label_weight=1000)
+        assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
+
+        db = _mock_db_sequential([assignment, spool])
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": []}]},
+            tray_now=0,
+            last_loaded_tray=0,
+            tray_change_log=[],
+        )
+
+        extract_mock = MagicMock(return_value=[{"slot_id": 1, "used_g": 5.0, "type": "PLA", "color": "#FF0000"}])
+
+        with (
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", extract_mock),
+            patch("backend.app.core.config.settings") as mock_settings,
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+
+            # No plate_id kwarg — direct/library Print flow.
+            await _track_from_3mf(
+                printer_id=1,
+                archive_id=None,
+                status="completed",
+                print_name="DirectPrint",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                tray_now_at_start=0,
+                threemf_path=mock_path,
+            )
+
+        assert extract_mock.call_args.args[1] is None