Attribute filament correctly when AMS backup swaps spools mid-print

    Everything the completion path needs to split a print's filament across
    the trays it fed from lived only in memory: the dispatched plate and
    slot-to-tray mapping, the spool-assignment snapshot, and the tray-change
    log. A print that outlived a restart lost all of it and fell back to
    what the printer reports at completion -- which, with AMS Filament
    Backup on, is the substitute tray. The whole print was charged to the
    spool that only finished it while the spool that ran dry was charged
    nothing.

    Persist that context in a new active_print_sessions row, append tray
    changes as they happen, and restore both the session and the printer's
    tray-change log at restart recovery. Seed the log from the current tray
    when there is nothing to restore, since last_loaded_tray advances even
    when no change is logged.

    Rank the queue item's stored ams_mapping above the printer's live
    mapping field, which is what backup rewrites. Recover plate_id from the
    archive or queue item, and give extract_layer_filament_usage_from_3mf a
    plate_id instead of taking the first .gcode member -- a Bambu Studio
    export stores plate 2 first, so per-layer figures were measured against
    the wrong plate for both inventory backends.

    Stop auto-unlinking a spool assignment when its slot reports empty
    during a running print. At a runout the spool is still in the AMS, and
    dropping the link leaves the completion path nothing to charge.

    Capture the print-start context for both inventory backends. Spoolman's
    own durable row (#1820) carries its plate-scoped figures and dispatched
    mapping but not the tray-change log, and its slot assignments -- the
    way. Registration in _active_sessions stays gated, since on_ams_change
    reads it to decide whether to skip the remain%-based weight sync (#880).

backend/app/core/database.py

@@ -250,6 +250,7 @@ async def get_db() -> AsyncSession:
 async def init_db():
     # Import models to register them with SQLAlchemy
     from backend.app.models import (  # noqa: F401
+        active_print_session,
         active_print_spoolman,
         ams_history,
         ams_label,

backend/app/main.py

@@ -1789,6 +1789,15 @@ async def on_ams_change(printer_id: int, ams_data: list):
 
     _print_active = printer_id in _active_sessions
 
+    # A slot that reports empty while a print is running is a filament runout,
+    # not a spool swap: the spool is still physically in the AMS, just
+    # consumed. Dropping either inventory backend's slot link there loses the
+    # only record of which spool fed the print, so the completion path can't
+    # charge the runout segment to anything. Both cleanup passes below consult
+    # this; computed once, up front, so neither depends on the other having run.
+    _unlink_state = printer_manager.get_status(printer_id)
+    printing_now = (getattr(_unlink_state, "state", "") or "").upper() in ("RUNNING", "PAUSE")
+
     # MQTT relay - publish AMS change
     try:
         printer_info = printer_manager.get_printer(printer_id)
@@ -1831,6 +1840,9 @@ async def on_ams_change(printer_id: int, ams_data: list):
                 .where(SA.printer_id == printer_id)
                 .options(selectinload(SA.spool).selectinload(_Spool.k_profiles))
             )
+            # ``printing_now`` (top of this function) keeps a runout from
+            # unlinking the spool that fed the print — the next idle-time pass
+            # unlinks it if the user really did take it out.
             stale = []
             for assignment in result.scalars().all():
                 # External spool assignments (ams_id=255) live in vt_tray, not AMS data
@@ -1849,6 +1861,14 @@ async def on_ams_change(printer_id: int, ams_data: list):
                 else:
                     current_tray = _find_tray_in_ams_data(ams_data, assignment.ams_id, assignment.tray_id)
                 if not current_tray:
+                    if printing_now:
+                        logger.info(
+                            "Auto-unlink skipped: spool %d AMS%d-T%d — slot empty during a running print (runout?)",
+                            assignment.spool_id,
+                            assignment.ams_id,
+                            assignment.tray_id,
+                        )
+                        continue
                     logger.info(
                         "Auto-unlink: spool %d AMS%d-T%d — tray not found in AMS data (slot empty?)",
                         assignment.spool_id,
@@ -1958,6 +1978,19 @@ async def on_ams_change(printer_id: int, ams_data: list):
                         continue
 
                     if not _colors_similar(cur_color, fp_color) or cur_type.upper() != fp_type.upper():
+                        # Blank tray data mid-print is a runout, not a swap: the
+                        # firmware clears colour and type when it unloads a spool
+                        # it just emptied. Unlinking here would erase the record
+                        # of which spool fed the print so far.
+                        if printing_now and not cur_color.strip() and not cur_type.strip():
+                            logger.info(
+                                "Auto-unlink skipped: spool %d AMS%d-T%d — tray data cleared during a running print "
+                                "(runout?)",
+                                assignment.spool_id,
+                                assignment.ams_id,
+                                assignment.tray_id,
+                            )
+                            continue
                         # Fingerprint mismatch — but check if tray now matches the
                         # assigned spool (e.g. auto-configure changed the tray).
                         spool = assignment.spool
@@ -2372,7 +2405,16 @@ async def on_ams_change(printer_id: int, ams_data: list):
                         # Empty tray slot — record for local assignment cleanup
                         # and drop any cached unknown-tag broadcast so a
                         # reinserted spool re-prompts.
-                        empty_slots.append((ams_id, tray_id_raw))
+                        #
+                        # Not during a running print: a slot that empties there
+                        # is a filament runout, and the spool is still in the
+                        # AMS. `spoolman_slot_assignments` is how a tag-less
+                        # spool assigned through the Bambuddy UI is resolved at
+                        # completion (#1459), so deleting the row mid-print
+                        # loses the runout segment's usage — the same failure
+                        # the internal inventory's auto-unlink had.
+                        if not printing_now:
+                            empty_slots.append((ams_id, tray_id_raw))
                         _clear_unknown_tag_dedup(printer_id, ams_id, tray_id_raw)
                         continue
 
@@ -2787,16 +2829,30 @@ async def on_print_start(printer_id: int, data: dict):
     except Exception:
         pass  # Don't fail print start callback if MQTT fails
 
-    # Capture AMS tray remain% for filament consumption tracking (skip if Spoolman handles usage)
+    # Capture AMS tray remain%, the assignment snapshot, the dispatched plate
+    # and mapping, and the seeded tray-change log.
+    #
+    # Unconditional, for both inventory backends. This only *captures* — the
+    # writing is still split, with the internal tracker skipped at completion
+    # when Spoolman owns usage. Spoolman's own durable row (#1820) already
+    # carries its plate-scoped 3MF figures and stored mapping, but not the
+    # tray-change log, and that log is the only record of which spool fed
+    # which layers when AMS Filament Backup swaps trays mid-print. Capturing
+    # it on one side only would leave Spoolman users with the mid-print
+    # restart bug this fixes for everyone else.
     try:
         async with async_session() as db:
             from backend.app.api.routes.settings import get_setting
+            from backend.app.services.usage_tracker import on_print_start as usage_on_print_start
 
             _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
-
-                await usage_on_print_start(printer_id, data, printer_manager, db=db)
+            await usage_on_print_start(
+                printer_id,
+                data,
+                printer_manager,
+                db=db,
+                spoolman_owns_usage=bool(_spoolman_on) and _spoolman_on.lower() == "true",
+            )
     except Exception as e:
         logger.warning("Usage tracker on_print_start failed: %s", e)
 
@@ -4426,6 +4482,87 @@ async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Pat
     await ws_manager.send_archive_updated({"id": archive_id, "photo_added": filename})
 
 
+async def _restore_usage_tracking_session(printer_id: int, state, db, logger) -> None:
+    """Put the filament-attribution context back after a restart mid-print.
+
+    ``usage_tracker._active_sessions`` and ``PrinterState.tray_change_log``
+    both die with the process. The print keeps running, so at completion the
+    tracker would fall back to whatever the printer reports *now* — and AMS
+    filament backup makes "now" the substitute tray, charging the whole print
+    to the spool that only finished it.
+
+    The persisted row is only trusted when its print name still matches what
+    the printer says it is running: a row left behind by a completion we never
+    saw must not attach itself to the next print.
+    """
+    try:
+        from backend.app.api.routes.settings import get_setting
+        from backend.app.services.usage_tracker import (
+            clear_persisted_session,
+            get_persisted_print_name,
+            restore_session,
+        )
+
+        persisted_name = await get_persisted_print_name(db, printer_id)
+        current_name = (state.subtask_name or "").strip()
+        if persisted_name and current_name and persisted_name.strip() != current_name:
+            logger.info(
+                "[RESTART] Discarding stale print session for printer %s (%r != running %r)",
+                printer_id,
+                persisted_name,
+                current_name,
+            )
+            await clear_persisted_session(db, printer_id)
+            # Fall through to seeding: the print on the printer is real, it just
+            # isn't the one the row described.
+            persisted_log = None
+        else:
+            # Spoolman users get the tray-change log back but no in-memory
+            # session — see ``on_print_start`` on why that dict is load-bearing
+            # for the remain%-sync guard.
+            _spoolman_on = await get_setting(db, "spoolman_enabled")
+            persisted_log = await restore_session(
+                db,
+                printer_id,
+                register_active=not (bool(_spoolman_on) and _spoolman_on.lower() == "true"),
+            )
+        if persisted_log:
+            restored = [tuple(entry) for entry in persisted_log if isinstance(entry, (list, tuple)) and len(entry) == 2]
+            # Anything this process already observed goes after the persisted
+            # history — the log is ordered by layer, and a fresh process can
+            # only have seen changes from later in the print.
+            for entry in state.tray_change_log or []:
+                if tuple(entry) not in restored:
+                    restored.append(tuple(entry))
+            state.tray_change_log = restored
+
+        tray_now = state.tray_now
+        if 0 <= tray_now <= 254:
+            if not state.tray_change_log:
+                # No persisted history — a print that started before this build,
+                # or before the row existed. Seed with the tray feeding right
+                # now so the remainder of the print is at least attributable to
+                # the right spool.
+                state.tray_change_log = [(tray_now, state.layer_num)]
+                logger.info(
+                    "[RESTART] Seeded tray change log for printer %s: tray=%d at layer=%d",
+                    printer_id,
+                    tray_now,
+                    state.layer_num,
+                )
+            # The tray handler updates ``last_loaded_tray`` on every push
+            # regardless of whether it logged a change, so re-align it to avoid
+            # a duplicate entry on the next push. Only ever with a real tray:
+            # ``last_loaded_tray`` is the "survives the end-of-print retract to
+            # 255" fallback, and writing 255 into it would defeat that.
+            state.last_loaded_tray = tray_now
+    except Exception:
+        # Never let attribution recovery cost the caller its timelapse
+        # baseline — that capture has to happen before the printer uploads
+        # the in-flight MP4 and there is no second chance at it.
+        logger.exception("[RESTART] Failed to restore usage-tracking session for printer %s", printer_id)
+
+
 async def on_print_running_observed(printer_id: int, data: dict):
     """Restart-recovery for a print that started before Bambuddy came up.
 
@@ -4450,6 +4587,8 @@ async def on_print_running_observed(printer_id: int, data: dict):
             if authorization is True:
                 logger.info("[RESTART] Restored active Bambuddy print for printer %s", printer_id)
 
+            await _restore_usage_tracking_session(printer_id, state, db, logger)
+
         result = await db.execute(select(Printer).where(Printer.id == printer_id))
         printer = result.scalar_one_or_none()
         if not printer:
@@ -5611,6 +5750,18 @@ async def on_print_complete(printer_id: int, data: dict):
     except Exception as e:
         logger.warning("Usage tracker on_print_complete failed: %s", e)
 
+    # Drop the print-start context unconditionally — the Spoolman branch above
+    # skips the internal tracker entirely, so nothing else would clear what
+    # print start captured, and a row surviving its print would be restored
+    # onto the next one after a restart.
+    try:
+        from backend.app.services.usage_tracker import discard_session
+
+        async with async_session() as db:
+            await discard_session(db, printer_id)
+    except Exception as e:
+        logger.warning("Failed to clear persisted print session for printer %s: %s", printer_id, e)
+
     # Spoolman: report filament usage (requires archive_id for tracking data lookup)
     if archive_id:
         if data.get("status") == "completed":
@@ -7765,6 +7916,30 @@ async def lifespan(app: FastAPI):
 
     printer_manager.set_assignment_verified_callback(on_assignment_verified)
 
+    async def on_tray_change(printer_id: int, tray_global: int, layer_num: int):
+        """Persist a mid-print tray change for completion-time attribution.
+
+        AMS filament backup switches trays without telling the slicer, so the
+        tray-change log is the only record of which spool fed which layers.
+        Keeping it only in memory meant a restart mid-print charged everything
+        to the tray that finished the job.
+        """
+        try:
+            from backend.app.services.usage_tracker import record_tray_change
+
+            async with async_session() as db:
+                await record_tray_change(db, printer_id, tray_global, layer_num)
+        except Exception as e:
+            logging.getLogger(__name__).warning(
+                "Failed to persist tray change for printer %d (tray=%d, layer=%d): %s",
+                printer_id,
+                tray_global,
+                layer_num,
+                e,
+            )
+
+    printer_manager.set_tray_change_callback(on_tray_change)
+
     # Initialize MQTT relay from settings
     async with async_session() as db:
         from backend.app.api.routes.settings import get_setting

backend/app/models/active_print_session.py

@@ -0,0 +1,57 @@
+"""Durable copy of the filament-attribution context for an in-flight print."""
+
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class ActivePrintSession(Base):
+    """Print-start context the completion path needs, persisted per printer.
+
+    ``usage_tracker._active_sessions`` holds the same data in memory, and the
+    tray-change log lives on ``PrinterState``. Both are lost when Bambuddy
+    restarts mid-print, which on a long print silently destroys filament
+    attribution: without the plate the 3MF parser sums every plate, without the
+    assignment snapshot a spool unlinked at runout can't be resolved, and
+    without the tray-change log an AMS-backup switch charges the whole print to
+    whichever tray happened to finish it.
+
+    One row per printer — a printer runs one print at a time. Written at print
+    start, appended to on every tray change, deleted at completion. A leaked
+    row (completion missed entirely) is harmless: print start overwrites it,
+    and the completion path ignores a row whose ``started_at`` doesn't line up
+    with the print it is closing.
+
+    The Spoolman writer has had an equivalent durable row since #1820
+    (``active_print_spoolman``); this is the internal-inventory counterpart.
+    """
+
+    __tablename__ = "active_print_sessions"
+
+    printer_id: Mapped[int] = mapped_column(ForeignKey("printers.id", ondelete="CASCADE"), primary_key=True)
+
+    print_name: Mapped[str] = mapped_column(default="")
+    started_at: Mapped[datetime] = mapped_column(DateTime)
+
+    # tray_now at print start — reliable, unlike at completion where the
+    # printer has usually retracted and reports 255.
+    tray_now_at_start: Mapped[int] = mapped_column(default=-1)
+
+    # Queue item's plate for multi-plate 3MFs dispatched one plate at a time.
+    plate_id: Mapped[int | None] = mapped_column(nullable=True)
+
+    # Slicer slot -> global tray, as dispatched: [2]
+    ams_mapping: Mapped[list | None] = mapped_column(JSON, nullable=True)
+
+    # {"<ams_id>-<tray_id>": spool_id} — the assignment map as it stood before
+    # the print could disturb it.
+    spool_assignments: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
+    # {"<ams_id>-<tray_id>": remain%} for the remain-delta fallback path.
+    tray_remain_start: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
+    # [[global_tray_id, layer_num], ...] mirroring PrinterState.tray_change_log.
+    tray_change_log: Mapped[list | None] = mapped_column(JSON, nullable=True)

backend/app/services/bambu_mqtt.py

@@ -813,6 +813,7 @@ class BambuMQTTClient:
         on_print_running_observed: Callable[[dict], None] | None = None,
         on_finish_photo_moment: Callable[[dict], None] | None = None,
         on_assignment_verified: Callable[[int, int, bool, dict], None] | None = None,
+        on_tray_change: Callable[[int, int], None] | None = None,
     ):
         self.ip_address = ip_address
         self.serial_number = serial_number
@@ -845,6 +846,12 @@ class BambuMQTTClient:
         # the same shape as on_print_start (filename / subtask_name /
         # remaining_time / raw_data / ams_mapping).
         self.on_print_running_observed = on_print_running_observed
+        # Fired for every entry appended to ``state.tray_change_log`` so main.py
+        # can mirror it into ``active_print_sessions``. The in-memory log dies
+        # with the process, and a long print outliving a restart would
+        # otherwise lose the segment boundaries the usage tracker splits on.
+        # Receives (global_tray_id, layer_num).
+        self.on_tray_change = on_tray_change
         # #1721: fired the moment the printer enters the end-of-print
         # "Filament unloading" phase (stg_cur=22 while progress>=99 or
         # we've hit the last layer / remaining_time<=0). This is the
@@ -2670,6 +2677,8 @@ class BambuMQTTClient:
                             tn,
                             self.state.layer_num,
                         )
+                        if self.on_tray_change:
+                            self.on_tray_change(tn, self.state.layer_num)
                     self.state.last_loaded_tray = self.state.tray_now
 
                 self._debug_on_change(

backend/app/services/printer_manager.py

@@ -409,6 +409,7 @@ class PrinterManager:
         self._on_bed_temp_update: Callable[[int, float], None] | None = None
         self._on_drying_complete: Callable[[int, int], None] | None = None
         self._on_assignment_verified: Callable[[int, int, int, bool, dict], None] | None = None
+        self._on_tray_change: Callable[[int, int, int], None] | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
         # Track who started the current print (Issue #206)
         self._current_print_user: dict[int, dict] = {}  # {printer_id: {"user_id": int, "username": str}}
@@ -659,6 +660,15 @@ class PrinterManager:
         """
         self._on_assignment_verified = callback
 
+    def set_tray_change_callback(self, callback: Callable[[int, int, int], None]):
+        """Set callback for mid-print tray changes.
+
+        Receives ``(printer_id, global_tray_id, layer_num)`` for every entry
+        appended to the printer's tray-change log, so it can be persisted for
+        the completion-time weight split.
+        """
+        self._on_tray_change = callback
+
     def _schedule_async(self, coro):
         """Schedule an async coroutine from a sync context.
 
@@ -730,6 +740,10 @@ class PrinterManager:
             if self._on_assignment_verified:
                 self._schedule_async(self._on_assignment_verified(printer_id, ams_id, tray_id, verified, detail))
 
+        def on_tray_change(tray_global: int, layer_num: int):
+            if self._on_tray_change:
+                self._schedule_async(self._on_tray_change(printer_id, tray_global, layer_num))
+
         client = BambuMQTTClient(
             ip_address=printer.ip_address,
             serial_number=printer.serial_number,
@@ -746,6 +760,7 @@ class PrinterManager:
             on_print_running_observed=on_print_running_observed,
             on_finish_photo_moment=on_finish_photo_moment,
             on_assignment_verified=on_assignment_verified,
+            on_tray_change=on_tray_change,
         )
 
         client.connect()

backend/app/services/spoolman_tracking.py

@@ -361,7 +361,7 @@ async def store_print_data(
         )
         filament_usage = extract_filament_usage_from_3mf(full_path, effective_plate_id) or None
 
-        layer_usage = extract_layer_filament_usage_from_3mf(full_path)
+        layer_usage = extract_layer_filament_usage_from_3mf(full_path, effective_plate_id)
         if layer_usage:
             # Convert int keys to string for JSON serialization
             layer_usage_json = {str(k): v for k, v in layer_usage.items()}

backend/app/services/usage_tracker.py

@@ -7,6 +7,7 @@ Primary tracking uses 3MF slicer estimates (precise per-filament data).
 AMS remain% delta is the fallback for trays not covered by 3MF data.
 """
 
+import asyncio
 import json
 import logging
 from dataclasses import dataclass, field
@@ -261,9 +262,173 @@ class PrintSession:
     plate_id: int | None = None
 
 
-# Module-level storage, keyed by printer_id
+# Module-level storage, keyed by printer_id. Mirrored to the
+# ``active_print_sessions`` table so a restart mid-print doesn't lose the
+# context — see ``persist_session`` / ``restore_session``.
 _active_sessions: dict[int, PrintSession] = {}
 
+# Serialises the read-modify-write on the persisted tray-change log, per printer.
+_tray_change_locks: dict[int, asyncio.Lock] = {}
+
+
+def _tray_key_to_str(key: tuple[int, int]) -> str:
+    return f"{key[0]}-{key[1]}"
+
+
+def _tray_key_from_str(key: str) -> tuple[int, int] | None:
+    ams_str, _, tray_str = key.partition("-")
+    try:
+        return int(ams_str), int(tray_str)
+    except ValueError:
+        return None
+
+
+def _tray_map_to_json(mapping: dict[tuple[int, int], int]) -> dict[str, int]:
+    return {_tray_key_to_str(k): v for k, v in mapping.items()}
+
+
+def _tray_map_from_json(mapping: dict | None) -> dict[tuple[int, int], int]:
+    result: dict[tuple[int, int], int] = {}
+    for raw_key, value in (mapping or {}).items():
+        key = _tray_key_from_str(str(raw_key))
+        if key is not None and isinstance(value, int):
+            result[key] = value
+    return result
+
+
+async def persist_session(
+    db: AsyncSession,
+    session: PrintSession,
+    tray_change_log: list | None = None,
+) -> None:
+    """Mirror ``session`` into ``active_print_sessions`` for restart recovery.
+
+    Overwrites any existing row for the printer: a printer runs one print at a
+    time, and a row left behind by a completion we never saw must not outlive
+    the next print start.
+    """
+    from backend.app.models.active_print_session import ActivePrintSession
+
+    row = await db.get(ActivePrintSession, session.printer_id)
+    if row is None:
+        row = ActivePrintSession(printer_id=session.printer_id)
+        db.add(row)
+
+    row.print_name = session.print_name or ""
+    row.started_at = session.started_at.replace(tzinfo=None)
+    row.tray_now_at_start = session.tray_now_at_start
+    row.plate_id = session.plate_id
+    row.ams_mapping = list(session.ams_mapping) if session.ams_mapping else None
+    row.spool_assignments = _tray_map_to_json(session.spool_assignments) or None
+    row.tray_remain_start = _tray_map_to_json(session.tray_remain_start) or None
+    row.tray_change_log = [list(entry) for entry in (tray_change_log or [])] or None
+
+    await db.commit()
+
+
+async def record_tray_change(db: AsyncSession, printer_id: int, tray_global: int, layer_num: int) -> None:
+    """Append one tray change to the persisted log.
+
+    No-op when no print-start row exists — a tray change outside a tracked
+    print has nothing to attribute.
+    """
+    from backend.app.models.active_print_session import ActivePrintSession
+
+    # Read-modify-write on a JSON column: two changes close together (a runout
+    # parks the extruder and the backup tray loads moments later) would
+    # otherwise race and drop a segment boundary.
+    async with _tray_change_locks.setdefault(printer_id, asyncio.Lock()):
+        row = await db.get(ActivePrintSession, printer_id)
+        if row is None:
+            return
+
+        log = [list(entry) for entry in (row.tray_change_log or [])]
+        entry = [tray_global, layer_num]
+        if log and log[-1] == entry:
+            # print-start seeds the log from PrinterState, which may already
+            # hold a change this callback is also reporting.
+            return
+        log.append(entry)
+        row.tray_change_log = log
+        await db.commit()
+
+
+async def get_persisted_print_name(db: AsyncSession, printer_id: int) -> str | None:
+    """Print name on the persisted row, for identity-checking a restored session."""
+    from backend.app.models.active_print_session import ActivePrintSession
+
+    row = await db.get(ActivePrintSession, printer_id)
+    return row.print_name if row is not None else None
+
+
+async def restore_session(db: AsyncSession, printer_id: int, register_active: bool = True) -> list[list[int]] | None:
+    """Rebuild the in-memory session for ``printer_id`` from the persisted row.
+
+    Returns the persisted tray-change log so the caller can put it back on
+    ``PrinterState``, or None when there is nothing to restore.
+
+    ``register_active=False`` returns the log without publishing the session to
+    ``_active_sessions`` — for Spoolman users, who need the tray-change log
+    restored but whose remain%-sync must not be suppressed by it (see
+    ``on_print_start``).
+    """
+    from backend.app.models.active_print_session import ActivePrintSession
+
+    row = await db.get(ActivePrintSession, printer_id)
+    if row is None:
+        return None
+
+    started_at = row.started_at
+    if started_at.tzinfo is None:
+        started_at = started_at.replace(tzinfo=timezone.utc)
+
+    session = PrintSession(
+        printer_id=printer_id,
+        print_name=row.print_name or "",
+        started_at=started_at,
+        tray_remain_start=_tray_map_from_json(row.tray_remain_start),
+        tray_now_at_start=row.tray_now_at_start,
+        spool_assignments=_tray_map_from_json(row.spool_assignments),
+        ams_mapping=list(row.ams_mapping) if row.ams_mapping else None,
+        plate_id=row.plate_id,
+    )
+    if register_active:
+        _active_sessions[printer_id] = session
+
+    log = [list(entry) for entry in (row.tray_change_log or [])]
+    logger.info(
+        "[UsageTracker] Restored print session for printer %d: plate_id=%s, ams_mapping=%s, "
+        "%d assignments, tray_change_log=%s",
+        printer_id,
+        row.plate_id,
+        row.ams_mapping,
+        len(row.spool_assignments or {}),
+        log,
+    )
+    return log
+
+
+async def clear_persisted_session(db: AsyncSession, printer_id: int) -> None:
+    """Drop the persisted print-start row once the print is closed out."""
+    from backend.app.models.active_print_session import ActivePrintSession
+
+    row = await db.get(ActivePrintSession, printer_id)
+    if row is not None:
+        await db.delete(row)
+        await db.commit()
+
+
+async def discard_session(db: AsyncSession, printer_id: int) -> None:
+    """Forget a printer's print-start context, in memory and on disk.
+
+    The completion path calls this for every print, including the ones whose
+    usage Spoolman owns: the context is captured for both backends, but only
+    the internal tracker's ``on_print_complete`` consumes (and pops) it.
+    """
+    _active_sessions.pop(printer_id, None)
+    _tray_change_locks.pop(printer_id, None)
+    await clear_persisted_session(db, printer_id)
+
 
 def _to_epoch_seconds(value: datetime | None) -> float | None:
     """Convert datetime to epoch seconds, assuming UTC for naive values."""
@@ -329,8 +494,27 @@ async def _resolve_spool_id_for_tray(
     return None
 
 
-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."""
+async def on_print_start(
+    printer_id: int,
+    data: dict,
+    printer_manager,
+    db: AsyncSession | None = None,
+    spoolman_owns_usage: bool = False,
+) -> None:
+    """Capture AMS tray remain% and spool assignments at print start.
+
+    The capture runs for both inventory backends — the persisted row carries
+    the tray-change log, which is the only record of which spool fed which
+    layers when AMS Filament Backup swaps trays, and Spoolman's own durable
+    row (#1820) does not hold it.
+
+    ``spoolman_owns_usage`` keeps the in-memory session out of
+    ``_active_sessions`` when Spoolman is writing the usage. That dict doubles
+    as ``on_ams_change``'s "a print is running, so skip the remain%-based
+    weight sync because the internal tracker will deduct precisely" flag
+    (#880); registering a session the internal tracker will never complete
+    would suppress a sync those users still need.
+    """
     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)
@@ -452,7 +636,19 @@ async def on_print_start(printer_id: int, data: dict, printer_manager, db: Async
         ams_mapping=data.get("ams_mapping"),
         plate_id=plate_id,
     )
-    _active_sessions[printer_id] = session
+    if spoolman_owns_usage:
+        _active_sessions.pop(printer_id, None)
+    else:
+        _active_sessions[printer_id] = session
+
+    # Mirror to the DB so a restart mid-print doesn't lose the context. The
+    # tray-change log has already been cleared and seeded with the starting
+    # tray by bambu_mqtt before this callback fires.
+    if db:
+        try:
+            await persist_session(db, session, getattr(state, "tray_change_log", None))
+        except Exception:
+            logger.exception("[UsageTracker] Failed to persist print session for printer %d", printer_id)
 
     if tray_remain_start:
         logger.info(
@@ -487,6 +683,16 @@ async def on_print_complete(
     from backend.app.models.spool_usage_history import SpoolUsageHistory
 
     session = _active_sessions.pop(printer_id, None)
+    if session is None:
+        # Restart mid-print: the in-memory session is gone but the print-start
+        # row survived. Without this the completion path loses the plate, the
+        # dispatched mapping and the assignment snapshot, and attributes the
+        # whole print to whichever tray happened to finish it.
+        try:
+            await restore_session(db, printer_id)
+        except Exception:
+            logger.exception("[UsageTracker] Failed to restore print session for printer %d", printer_id)
+        session = _active_sessions.pop(printer_id, None)
     status = data.get("status", "completed")
     results = []
     handled_trays: set[tuple[int, int]] = set()
@@ -972,7 +1178,59 @@ async def _track_from_3mf(
         logger.info("[UsageTracker] 3MF: no file available for archive %s, skipping", archive_id)
         return []
 
+    # The queue item carries both the plate and the dispatched mapping; look it
+    # up at most once. ``.first()`` rather than ``.scalar_one_or_none()``
+    # because a batch dispatches one archive as several queue items, and
+    # raising there would cost the print all of its usage tracking.
+    _queue_item_lookup: list = []
+
+    async def _dispatch_queue_item():
+        if not _queue_item_lookup:
+            if not archive_id:
+                _queue_item_lookup.append(None)
+            else:
+                queue_result = await db.execute(
+                    select(PrintQueueItem)
+                    .where(PrintQueueItem.archive_id == archive_id)
+                    .where(PrintQueueItem.status.in_(["printing", "completed", "failed"]))
+                )
+                _queue_item_lookup.append(queue_result.scalars().first())
+        return _queue_item_lookup[0]
+
+    # The caller's plate_id comes from the in-memory session, which a restart
+    # mid-print destroys. Both the archive and the queue item recorded the
+    # plate at dispatch — without falling back to them the parser sums every
+    # plate of a multi-plate file and charges the lot to one spool.
+    if plate_id is None:
+        if archive is not None and archive.plate_id is not None:
+            plate_id = archive.plate_id
+            logger.info("[UsageTracker] 3MF: plate_id=%s recovered from archive %s", plate_id, archive_id)
+        else:
+            plate_queue_item = await _dispatch_queue_item()
+            if plate_queue_item is not None and plate_queue_item.plate_id is not None:
+                plate_id = plate_queue_item.plate_id
+                logger.info(
+                    "[UsageTracker] 3MF: plate_id=%s recovered from queue item %s",
+                    plate_id,
+                    plate_queue_item.id,
+                )
+
     filament_usage = extract_filament_usage_from_3mf(file_path, plate_id)
+    if not filament_usage and plate_id is not None:
+        # The plate isn't in this file. That happens when the archive's own 3MF
+        # is gone and `_resolve_3mf_fallback` substituted a same-named file from
+        # the library that was sliced with different plates. Summing the whole
+        # file is wrong for a single-plate run, but it is closer than recording
+        # nothing at all — and unlike the silent whole-file sum this replaces,
+        # it says so.
+        filament_usage = extract_filament_usage_from_3mf(file_path, None)
+        if filament_usage:
+            logger.warning(
+                "[UsageTracker] 3MF: plate %s not present in %s — falling back to the whole-file total",
+                plate_id,
+                file_path,
+            )
+            plate_id = None
     if not filament_usage:
         logger.info("[UsageTracker] 3MF: no filament usage data in %s", file_path)
         return []
@@ -987,7 +1245,23 @@ async def _track_from_3mf(
     if slot_to_tray:
         mapping_source = "print_cmd"
 
-    # 2. Try MQTT mapping field from printer state (universal, all print sources)
+    # 2. Try queue item ams_mapping (queue-initiated prints store the exact mapping)
+    #
+    # Ranked above the live MQTT field on purpose: `mapping` reports the tray
+    # the printer is feeding from *now*, and AMS filament backup rewrites it to
+    # the substitute tray when a spool runs dry. Read at completion it names
+    # the tray that finished the print, not the one the slicer assigned — the
+    # queue item's copy is the mapping the print was actually dispatched with.
+    if not slot_to_tray and archive_id:
+        queue_item = await _dispatch_queue_item()
+        if queue_item and queue_item.ams_mapping:
+            try:
+                slot_to_tray = json.loads(queue_item.ams_mapping)
+                mapping_source = "queue"
+            except (json.JSONDecodeError, TypeError):
+                pass
+
+    # 3. Try MQTT mapping field from printer state (universal, all print sources)
     if not slot_to_tray:
         state = printer_manager.get_status(printer_id)
         raw_data = getattr(state, "raw_data", None) if state else None
@@ -998,21 +1272,6 @@ async def _track_from_3mf(
                 slot_to_tray = decoded
                 mapping_source = "mqtt"
 
-    # 3. Try queue item ams_mapping (queue-initiated prints store the exact mapping)
-    if not slot_to_tray and archive_id:
-        queue_result = await db.execute(
-            select(PrintQueueItem)
-            .where(PrintQueueItem.archive_id == archive_id)
-            .where(PrintQueueItem.status.in_(["printing", "completed", "failed"]))
-        )
-        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)
-                mapping_source = "queue"
-            except (json.JSONDecodeError, TypeError):
-                pass
-
     # 4. Color-match 3MF filament slots to AMS trays (for printers without mapping field)
     if not slot_to_tray:
         state = printer_manager.get_status(printer_id)
@@ -1043,6 +1302,22 @@ async def _track_from_3mf(
     state = printer_manager.get_status(printer_id) if len(nonzero_slots) == 1 else None
     if state is not None:
         tray_changes = getattr(state, "tray_change_log", []) or []
+    elif len(nonzero_slots) > 1:
+        # Multi-material print: every filament change moves tray_now, so the
+        # log can't be read as "this slot moved to that tray" and splitting
+        # would attribute worse than the mapping does. Say so rather than
+        # silently dropping the evidence — a runout mid-print on a
+        # multi-material job still lands entirely on the mapped tray.
+        _multi_state = printer_manager.get_status(printer_id)
+        if len(getattr(_multi_state, "tray_change_log", []) or []) > 1:
+            logger.warning(
+                "[UsageTracker] 3MF: %d tray changes observed but %d filament slots used — "
+                "splitting needs a single slot, attributing by mapping alone (printer %d, archive %s)",
+                len(_multi_state.tray_change_log),
+                len(nonzero_slots),
+                printer_id,
+                archive_id,
+            )
 
     if len(tray_changes) > 1:
         # Multi-tray usage detected — splitting takes over regardless of slot_to_tray.
@@ -1101,7 +1376,7 @@ async def _track_from_3mf(
                     mm_to_grams,
                 )
 
-                layer_usage = extract_layer_filament_usage_from_3mf(file_path)
+                layer_usage = extract_layer_filament_usage_from_3mf(file_path, plate_id)
                 if layer_usage:
                     cumulative_mm = get_cumulative_usage_at_layer(layer_usage, current_layer)
                     filament_props = extract_filament_properties_from_3mf(file_path)
@@ -1147,7 +1422,7 @@ async def _track_from_3mf(
                     extract_layer_filament_usage_from_3mf,
                 )
 
-                split_layer_usage = extract_layer_filament_usage_from_3mf(file_path)
+                split_layer_usage = extract_layer_filament_usage_from_3mf(file_path, plate_id)
                 filament_props = extract_filament_properties_from_3mf(file_path)
                 split_props = filament_props.get(slot_id, {})
             except Exception:

backend/app/utils/threemf_tools.py

@@ -161,11 +161,20 @@ def mm_to_grams(
     return volume_cm3 * density_g_cm3
 
 
-def extract_layer_filament_usage_from_3mf(file_path: Path) -> dict[int, dict[int, float]] | None:
+def extract_layer_filament_usage_from_3mf(
+    file_path: Path, plate_id: int | None = None
+) -> dict[int, dict[int, float]] | None:
     """Extract per-layer filament usage from a 3MF file's embedded G-code.
 
     Args:
         file_path: Path to the 3MF file
+        plate_id: Plate to read. Required for multi-plate files — zip member
+            order is whatever the slicer wrote, and Bambu Studio stores
+            ``plate_2.gcode`` ahead of ``plate_1.gcode``, so the old
+            "first member" behaviour read a different plate's layers than
+            the one that printed. Returns None rather than silently falling
+            back to another plate when the requested plate isn't in the
+            file; callers degrade to linear scaling, which is bounded.
 
     Returns:
         Dictionary mapping layers to filament usage, or None if parsing fails.
@@ -173,13 +182,18 @@ def extract_layer_filament_usage_from_3mf(file_path: Path) -> dict[int, dict[int
     """
     try:
         with zipfile.ZipFile(file_path, "r") as zf:
-            # Find G-code file(s) - usually plate_1.gcode or Metadata/plate_1.gcode
-            gcode_files = [f for f in zf.namelist() if f.endswith(".gcode")]
-            if not gcode_files:
+            names = zf.namelist()
+            gcode_path = select_plate_gcode_name(names, plate_id)
+            if gcode_path is None:
+                # No plate asked for, or a file whose single G-code member
+                # doesn't follow the plate_N naming convention (non-Bambu
+                # slicers) — the lone member is unambiguous either way.
+                gcode_files = [f for f in names if f.endswith(".gcode")]
+                if plate_id is None or len(gcode_files) == 1:
+                    gcode_path = default_plate_gcode_name(names)
+            if gcode_path is None:
                 return None
 
-            # Use the first G-code file (typically only one per 3MF export)
-            gcode_path = gcode_files[0]
             gcode_content = zf.read(gcode_path).decode("utf-8", errors="ignore")
 
             return parse_gcode_layer_filament_usage(gcode_content)

backend/tests/conftest.py

@@ -188,6 +188,7 @@ async def test_engine():
 
     # Import all models to register them
     from backend.app.models import (
+        active_print_session,  # noqa: F401
         ams_history,
         ams_label,
         api_key,

backend/tests/integration/test_inventory_assign.py

@@ -1358,3 +1358,293 @@ class TestAssignSpoolPfcnCloudPreset:
             # original PFCN (which the slicer needs separately).
             assert call_kwargs.kwargs["tray_info_idx"] == "GFL05"
             assert call_kwargs.kwargs["setting_id"] == "PFCN80e80c1f79db85"
+
+
+def _make_printing_status(ams_data, state="RUNNING"):
+    """Printer status carrying an explicit gcode state for the runout guard."""
+    status = _make_mock_status(ams_data=ams_data)
+    status.state = state
+    return status
+
+
+class TestAutoUnlinkDuringRunout:
+    """A slot that reports empty mid-print is a filament runout, not a spool
+    swap — the spool is still in the AMS, just consumed.
+
+    Unlinking there erased the only record of which spool fed the print, so the
+    completion path had nothing to charge the runout segment to. With AMS
+    filament backup that is the normal course of events, not an edge case.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cleared_tray_data_keeps_the_assignment_while_printing(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(material="ABS", rgba="616777FF")
+
+        assignment = SpoolAssignment(
+            spool_id=spool.id,
+            printer_id=printer.id,
+            ams_id=0,
+            tray_id=2,
+            fingerprint_color="616777FF",
+            fingerprint_type="ABS",
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+
+        # The firmware clears colour and type when it unloads a spool it just
+        # emptied (state 26 = "not loaded").
+        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = _make_printing_status(ams_data)
+            mock_pm_main.get_model.return_value = "H2D"
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer.id, ams_data)
+
+        # on_ams_change committed through its own session — drop this one's
+        # identity map so the assertion reads the database, not a cached row.
+        db_session.expunge_all()
+        remaining = await db_session.get(SpoolAssignment, assignment.id)
+        assert remaining is not None, "runout must not unlink the spool that fed the print"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cleared_tray_data_still_unlinks_when_idle(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """Off the print, an emptied slot really does mean the spool is gone."""
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(material="ABS", rgba="616777FF")
+
+        assignment = SpoolAssignment(
+            spool_id=spool.id,
+            printer_id=printer.id,
+            ams_id=0,
+            tray_id=2,
+            fingerprint_color="616777FF",
+            fingerprint_type="ABS",
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        assignment_id = assignment.id
+
+        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = _make_printing_status(ams_data, state="IDLE")
+            mock_pm_main.get_model.return_value = "H2D"
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer.id, ams_data)
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, assignment_id) is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_genuinely_different_filament_still_unlinks_while_printing(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        """The guard is for blank tray data only — a real swap must still
+        reconcile, or the wrong spool gets charged."""
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(material="ABS", rgba="616777FF")
+
+        assignment = SpoolAssignment(
+            spool_id=spool.id,
+            printer_id=printer.id,
+            ams_id=0,
+            tray_id=2,
+            fingerprint_color="616777FF",
+            fingerprint_type="ABS",
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        assignment_id = assignment.id
+
+        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "PETG", "tray_color": "6EE53CFF", "state": 11}]}]
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = _make_printing_status(ams_data)
+            mock_pm_main.get_model.return_value = "H2D"
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer.id, ams_data)
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, assignment_id) is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_slot_missing_from_ams_data_keeps_the_assignment_while_printing(
+        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
+    ):
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+        from backend.app.models.spool_assignment import SpoolAssignment
+
+        printer = await printer_factory(name="H2D")
+        spool = await spool_factory(material="ABS", rgba="616777FF")
+
+        assignment = SpoolAssignment(
+            spool_id=spool.id,
+            printer_id=printer.id,
+            ams_id=0,
+            tray_id=2,
+            fingerprint_color="616777FF",
+            fingerprint_type="ABS",
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        assignment_id = assignment.id
+
+        # Tray 2 dropped out of the payload entirely.
+        ams_data = [{"id": 0, "tray": [{"id": 0, "tray_type": "ABS", "tray_color": "FFFFFFFF", "state": 11}]}]
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = _make_printing_status(ams_data)
+            mock_pm_main.get_model.return_value = "H2D"
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer.id, ams_data)
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolAssignment, assignment_id) is not None
+
+
+class TestSpoolmanSlotAssignmentDuringRunout:
+    """`spoolman_slot_assignments` is how a tag-less spool assigned through the
+    Bambuddy UI is resolved at completion (#1459). Deleting the row when a slot
+    empties mid-print loses the runout segment's usage — the same failure the
+    internal inventory's auto-unlink had, so it needs the same guard."""
+
+    async def _enable_spoolman(self, db_session):
+        from backend.app.models.settings import Settings
+
+        for key, value in (
+            ("spoolman_enabled", "true"),
+            ("spoolman_sync_mode", "auto"),
+            ("spoolman_url", "http://spoolman.test"),
+        ):
+            db_session.add(Settings(key=key, value=value))
+        await db_session.commit()
+
+    async def _run(self, printer_id, state):
+        from unittest.mock import AsyncMock
+
+        from backend.app.main import on_ams_change
+
+        # A tray the firmware has cleared: parse_ams_tray returns None, which
+        # is what marks the slot empty for the cleanup pass.
+        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]
+
+        spoolman_client = MagicMock()
+        spoolman_client.health_check = AsyncMock(return_value=True)
+        spoolman_client.get_spools = AsyncMock(return_value=[])
+        spoolman_client.sync_ams_tray = AsyncMock(return_value=None)
+        # None is what marks the slot empty for the cleanup pass.
+        spoolman_client.parse_ams_tray.return_value = None
+
+        with (
+            patch("backend.app.main.printer_manager") as mock_pm_main,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.ws_manager") as mock_ws,
+            patch("backend.app.main.get_spoolman_client", new=AsyncMock(return_value=spoolman_client)),
+        ):
+            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
+            mock_pm_main.get_status.return_value = state
+            mock_pm_main.get_model.return_value = "H2D"
+            mock_relay.on_ams_change = AsyncMock()
+            mock_ws.send_printer_status = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+
+            await on_ams_change(printer_id, ams_data)
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_slot_row_survives_a_runout(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+
+        await self._enable_spoolman(db_session)
+        printer = await printer_factory(name="H2D")
+        row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41)
+        db_session.add(row)
+        await db_session.commit()
+        row_id = row.id
+
+        await self._run(printer.id, _make_printing_status(None))
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_slot_row_is_still_cleaned_up_when_idle(
+        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
+    ):
+        """Proves the guard is what saved the row above, not an unreachable
+        code path."""
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+
+        await self._enable_spoolman(db_session)
+        printer = await printer_factory(name="H2D")
+        row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41)
+        db_session.add(row)
+        await db_session.commit()
+        row_id = row.id
+
+        await self._run(printer.id, _make_printing_status(None, state="IDLE"))
+
+        db_session.expunge_all()
+        assert await db_session.get(SpoolmanSlotAssignment, row_id) is None

backend/tests/unit/services/test_tray_change_callback.py

@@ -0,0 +1,74 @@
+"""The tray-change log has to leave the process.
+
+``PrinterState.tray_change_log`` is what the usage tracker splits filament
+weight on when AMS filament backup swaps in a fresh spool mid-print. It lived
+only in memory, so a restart during a long print erased the segment boundaries
+and the whole job got charged to the tray that finished it. The client now
+reports every appended entry so main.py can persist it.
+"""
+
+import pytest
+
+from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+
+def _tray_msg(tray_now: int):
+    """A partial AMS update carrying only tray_now, as P-series and H2D send."""
+    return {"print": {"ams": {"tray_now": str(tray_now)}}}
+
+
+class TestTrayChangeCallback:
+    @pytest.fixture
+    def mqtt_client(self):
+        client = BambuMQTTClient(
+            ip_address="192.168.1.100",
+            serial_number="TEST123",
+            access_code="12345678",
+        )
+        client._was_running = True
+        client._completion_triggered = False
+        return client
+
+    def test_every_logged_change_is_reported(self, mqtt_client):
+        seen: list[tuple[int, int]] = []
+        mqtt_client.on_tray_change = lambda tray, layer: seen.append((tray, layer))
+
+        mqtt_client.state.layer_num = 0
+        mqtt_client._process_message(_tray_msg(2))
+        mqtt_client.state.layer_num = 670
+        mqtt_client._process_message(_tray_msg(254))
+        mqtt_client.state.layer_num = 675
+        mqtt_client._process_message(_tray_msg(3))
+
+        assert seen == [(2, 0), (254, 670), (3, 675)]
+        assert mqtt_client.state.tray_change_log == [(2, 0), (254, 670), (3, 675)]
+
+    def test_repeat_of_the_same_tray_is_not_reported(self, mqtt_client):
+        """The printer republishes tray_now on every push; only transitions
+        are segment boundaries."""
+        seen: list[tuple[int, int]] = []
+        mqtt_client.on_tray_change = lambda tray, layer: seen.append((tray, layer))
+
+        mqtt_client._process_message(_tray_msg(2))
+        mqtt_client.state.layer_num = 40
+        mqtt_client._process_message(_tray_msg(2))
+
+        assert seen == [(2, 0)]
+
+    def test_no_callback_outside_a_running_print(self, mqtt_client):
+        seen: list[tuple[int, int]] = []
+        mqtt_client.on_tray_change = lambda tray, layer: seen.append((tray, layer))
+        mqtt_client._was_running = False
+
+        mqtt_client._process_message(_tray_msg(2))
+
+        assert seen == []
+        assert mqtt_client.state.tray_change_log == []
+
+    def test_missing_callback_does_not_break_logging(self, mqtt_client):
+        """The callback is optional — the in-memory log still has to work."""
+        mqtt_client.on_tray_change = None
+
+        mqtt_client._process_message(_tray_msg(2))
+
+        assert mqtt_client.state.tray_change_log == [(2, 0)]

backend/tests/unit/test_threemf_tools.py

@@ -9,11 +9,14 @@ import json
 import math
 import zipfile
 
+import pytest
+
 from backend.app.utils.threemf_tools import (
     expand_to_project_slots,
     extract_bed_type_from_3mf,
     extract_embedded_presets_from_3mf,
     extract_filament_usage_from_3mf,
+    extract_layer_filament_usage_from_3mf,
     extract_max_z_height_from_3mf,
     extract_plate_extruder_set_from_3mf,
     extract_print_time_from_3mf,
@@ -1392,3 +1395,87 @@ class TestExtractMaxZHeightFrom3mf:
         gcode = _header(max_z_height="99.9") + "\n" + ("G1 X1 Y1 E0.1\n" * 400_000)
         path = _make_plate_3mf(tmp_path, {"Metadata/plate_1.gcode": gcode})
         assert extract_max_z_height_from_3mf(path, 1) == 99.9
+
+
+def _layer_gcode(per_layer_mm: float, layers: int) -> str:
+    """G-code extruding a fixed amount on filament 0 for each of ``layers``."""
+    lines = ["M620 S0"]
+    for layer in range(1, layers + 1):
+        lines.append(f"M73 L{layer}")
+        lines.append(f"G1 X1 Y1 E{per_layer_mm}")
+    return "\n".join(lines)
+
+
+class TestExtractLayerFilamentUsagePlateSelection:
+    """The per-layer extract feeds the mid-print tray split and partial-print
+    scaling, so reading a different plate than the one that printed silently
+    misattributes filament.
+
+    Bambu Studio writes ``plate_2.gcode`` ahead of ``plate_1.gcode`` in the
+    zip, so "first member" is not "first plate".
+    """
+
+    def test_reads_the_requested_plate_not_the_first_member(self, tmp_path):
+        path = _make_plate_3mf(
+            tmp_path,
+            {
+                "Metadata/plate_2.gcode": _layer_gcode(1.0, 5),
+                "Metadata/plate_1.gcode": _layer_gcode(10.0, 8),
+            },
+        )
+        usage = extract_layer_filament_usage_from_3mf(path, 1)
+        assert usage is not None
+        assert get_cumulative_usage_at_layer(usage, 8)[0] == pytest.approx(80.0)
+
+    def test_plate_two_reads_plate_two(self, tmp_path):
+        path = _make_plate_3mf(
+            tmp_path,
+            {
+                "Metadata/plate_2.gcode": _layer_gcode(1.0, 5),
+                "Metadata/plate_1.gcode": _layer_gcode(10.0, 8),
+            },
+        )
+        usage = extract_layer_filament_usage_from_3mf(path, 2)
+        assert usage is not None
+        assert get_cumulative_usage_at_layer(usage, 5)[0] == pytest.approx(5.0)
+
+    def test_no_plate_asked_for_takes_the_lowest_numbered_plate(self, tmp_path):
+        path = _make_plate_3mf(
+            tmp_path,
+            {
+                "Metadata/plate_2.gcode": _layer_gcode(1.0, 5),
+                "Metadata/plate_1.gcode": _layer_gcode(10.0, 8),
+            },
+        )
+        usage = extract_layer_filament_usage_from_3mf(path)
+        assert usage is not None
+        assert get_cumulative_usage_at_layer(usage, 8)[0] == pytest.approx(80.0)
+
+    def test_requested_plate_missing_returns_none_rather_than_another_plate(self, tmp_path):
+        """Callers degrade to linear scaling, which is bounded. Silently
+        reading a different plate's layers is not."""
+        path = _make_plate_3mf(
+            tmp_path,
+            {
+                "Metadata/plate_1.gcode": _layer_gcode(10.0, 8),
+                "Metadata/plate_2.gcode": _layer_gcode(1.0, 5),
+            },
+        )
+        assert extract_layer_filament_usage_from_3mf(path, 3) is None
+
+    def test_single_unnumbered_gcode_is_used_for_any_plate(self, tmp_path):
+        """Slicers outside Bambu's plate_N convention export one G-code member;
+        it is unambiguous whatever plate the queue recorded."""
+        path = _make_plate_3mf(tmp_path, {"whatever.gcode": _layer_gcode(2.0, 4)})
+        usage = extract_layer_filament_usage_from_3mf(path, 1)
+        assert usage is not None
+        assert get_cumulative_usage_at_layer(usage, 4)[0] == pytest.approx(8.0)
+
+    def test_no_gcode_member_returns_none(self, tmp_path):
+        path = _make_plate_3mf(tmp_path, {"Metadata/slice_info.config": "<config/>"})
+        assert extract_layer_filament_usage_from_3mf(path, 1) is None
+
+    def test_unreadable_file_returns_none(self, tmp_path):
+        path = tmp_path / "broken.3mf"
+        path.write_text("not a zip")
+        assert extract_layer_filament_usage_from_3mf(path, 1) is None

backend/tests/unit/test_usage_tracker.py

@@ -87,10 +87,12 @@ def _mock_db_sequential(responses):
         idx = call_count[0]
         call_count[0] += 1
         result = MagicMock()
-        if idx < len(responses):
-            result.scalar_one_or_none.return_value = responses[idx]
-        else:
-            result.scalar_one_or_none.return_value = None
+        value = responses[idx] if idx < len(responses) else None
+        result.scalar_one_or_none.return_value = value
+        # Lookups that may legitimately match more than one row (an archive
+        # dispatched as several queue items in a batch) use scalars().first()
+        # instead of scalar_one_or_none() — same response sequence either way.
+        result.scalars.return_value.first.return_value = value
         # For cost aggregation queries that use .scalar() instead of .scalar_one_or_none()
         result.scalar.return_value = None
         return result
@@ -1768,11 +1770,13 @@ class TestMqttMappingIntegration:
         assign_red = _make_assignment(spool_id=3, ams_id=128, tray_id=0)
         archive = _make_archive(archive_id=12)
 
-        # db: archive, then 3 pairs of (assignment, spool)
-        # No queue lookup because MQTT mapping is found first
+        # db: archive, queue lookup (no queue item — this is a direct print),
+        # then 3 pairs of (assignment, spool). The queue mapping is consulted
+        # before the MQTT field because AMS backup rewrites the live one.
         db = _mock_db_sequential(
             [
                 archive,
+                None,
                 assign_white,
                 spool_white,
                 assign_black,

backend/tests/unit/test_usage_tracker_restart_recovery.py

@@ -0,0 +1,803 @@
+"""Filament attribution has to survive a restart mid-print.
+
+A 14-hour print that spans a Bambuddy restart used to lose everything the
+completion path needs: the plate (so the 3MF parser summed every plate of a
+multi-plate file), the dispatched slot-to-tray mapping (so it fell back to the
+live MQTT ``mapping`` field, which AMS filament backup rewrites to the
+substitute tray), the spool-assignment snapshot, and the tray-change log that
+splits weight across a runout. The whole print was then charged to whichever
+spool happened to finish it, while the spool that actually ran dry was charged
+nothing.
+
+These tests cover the durable ``active_print_sessions`` row that fixes that,
+plus the plate and mapping fallbacks the completion path now applies.
+"""
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.active_print_session import ActivePrintSession
+from backend.app.models.printer import Printer
+from backend.app.services.usage_tracker import (
+    PrintSession,
+    _active_sessions,
+    _track_from_3mf,
+    clear_persisted_session,
+    get_persisted_print_name,
+    on_print_complete,
+    persist_session,
+    record_tray_change,
+    restore_session,
+)
+
+
+def _make_spool(spool_id=1, label_weight=1000, weight_used=0):
+    spool = MagicMock()
+    spool.id = spool_id
+    spool.label_weight = label_weight
+    spool.weight_used = weight_used
+    spool.tag_uid = None
+    spool.tray_uuid = None
+    spool.last_used = None
+    spool.cost_per_kg = None
+    spool.material = "ABS"
+    spool.rgba = "616777FF"
+    return spool
+
+
+def _make_assignment(spool_id=1, ams_id=0, tray_id=0):
+    assignment = MagicMock()
+    assignment.spool_id = spool_id
+    assignment.printer_id = 1
+    assignment.ams_id = ams_id
+    assignment.tray_id = tray_id
+    assignment.created_at = None
+    return assignment
+
+
+def _make_archive(archive_id=1, plate_id=None, file_path="archives/1/multi_plate.3mf"):
+    archive = MagicMock()
+    archive.id = archive_id
+    archive.file_path = file_path
+    archive.plate_id = plate_id
+    archive.extra_data = None
+    return archive
+
+
+def _make_queue_item(item_id=629, ams_mapping=None, plate_id=None):
+    item = MagicMock()
+    item.id = item_id
+    item.ams_mapping = ams_mapping
+    item.plate_id = plate_id
+    item.status = "printing"
+    return item
+
+
+def _mock_db_sequential(responses):
+    """Mock db whose execute() yields the given rows in order."""
+    db = AsyncMock()
+    call_count = [0]
+
+    async def mock_execute(*args, **kwargs):
+        idx = call_count[0]
+        call_count[0] += 1
+        result = MagicMock()
+        value = responses[idx] if idx < len(responses) else None
+        result.scalar_one_or_none.return_value = value
+        result.scalars.return_value.first.return_value = value
+        result.scalar.return_value = None
+        return result
+
+    db.execute = mock_execute
+    return db
+
+
+def _patched_3mf(filament_usage, capture=None):
+    """Patch the 3MF extract, optionally recording the plate_id it was given."""
+
+    def _extract(path, plate_id=None):
+        if capture is not None:
+            capture.append(plate_id)
+        return filament_usage
+
+    return patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", side_effect=_extract)
+
+
+def _settings_patch():
+    mock_settings = patch("backend.app.core.config.settings")
+    return mock_settings
+
+
+class TestPersistedSessionRoundTrip:
+    """The row is the only thing that outlives the process."""
+
+    @pytest.fixture(autouse=True)
+    def _clear_sessions(self):
+        _active_sessions.clear()
+        yield
+        _active_sessions.clear()
+
+    @pytest.fixture
+    async def printer(self, db_session):
+        row = Printer(name="H2D-1", ip_address="192.168.0.10", access_code="1234", serial_number="TESTSERIAL")
+        db_session.add(row)
+        await db_session.commit()
+        return row
+
+    def _session(self, printer_id):
+        return PrintSession(
+            printer_id=printer_id,
+            print_name="AMS_Rack",
+            started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
+            tray_remain_start={(0, 2): 84, (0, 3): 100},
+            tray_now_at_start=2,
+            spool_assignments={(0, 2): 69, (0, 3): 68},
+            ams_mapping=[2],
+            plate_id=1,
+        )
+
+    @pytest.mark.asyncio
+    async def test_restore_rebuilds_the_session_and_returns_the_tray_log(self, db_session, printer):
+        await persist_session(db_session, self._session(printer.id), [(2, 0)])
+        _active_sessions.clear()
+
+        log = await restore_session(db_session, printer.id)
+
+        assert log == [[2, 0]]
+        restored = _active_sessions[printer.id]
+        assert restored.plate_id == 1
+        assert restored.ams_mapping == [2]
+        assert restored.tray_now_at_start == 2
+        # Tuple keys survive the JSON round trip — the completion path indexes
+        # the snapshot by (ams_id, tray_id).
+        assert restored.spool_assignments == {(0, 2): 69, (0, 3): 68}
+        assert restored.tray_remain_start == {(0, 2): 84, (0, 3): 100}
+        assert restored.started_at.tzinfo is not None
+
+    @pytest.mark.asyncio
+    async def test_tray_changes_accumulate_in_order(self, db_session, printer):
+        await persist_session(db_session, self._session(printer.id), [(2, 0)])
+
+        # The runout sequence from the reported print: A3 empties, the AMS
+        # parks, then filament backup brings A4 in.
+        await record_tray_change(db_session, printer.id, 254, 670)
+        await record_tray_change(db_session, printer.id, 3, 675)
+
+        assert await restore_session(db_session, printer.id) == [[2, 0], [254, 670], [3, 675]]
+
+    @pytest.mark.asyncio
+    async def test_tray_change_without_a_session_is_a_noop(self, db_session, printer):
+        await record_tray_change(db_session, printer.id, 3, 675)
+
+        row = await db_session.get(ActivePrintSession, printer.id)
+        assert row is None
+
+    @pytest.mark.asyncio
+    async def test_print_start_overwrites_a_row_left_by_a_missed_completion(self, db_session, printer):
+        await persist_session(db_session, self._session(printer.id), [(2, 0), (3, 675)])
+
+        second = self._session(printer.id)
+        second.print_name = "Cover"
+        second.plate_id = 2
+        second.ams_mapping = [5]
+        second.spool_assignments = {(1, 0): 60}
+        await persist_session(db_session, second, [(5, 0)])
+
+        rows = (await db_session.execute(select(ActivePrintSession))).scalars().all()
+        assert len(rows) == 1
+        log = await restore_session(db_session, printer.id)
+        assert log == [[5, 0]]
+        assert _active_sessions[printer.id].plate_id == 2
+        assert _active_sessions[printer.id].spool_assignments == {(1, 0): 60}
+
+    @pytest.mark.asyncio
+    async def test_clear_removes_the_row(self, db_session, printer):
+        await persist_session(db_session, self._session(printer.id), [(2, 0)])
+
+        await clear_persisted_session(db_session, printer.id)
+
+        assert await restore_session(db_session, printer.id) is None
+        assert await get_persisted_print_name(db_session, printer.id) is None
+
+    @pytest.mark.asyncio
+    async def test_clear_is_safe_without_a_row(self, db_session, printer):
+        await clear_persisted_session(db_session, printer.id)
+
+    @pytest.mark.asyncio
+    async def test_print_name_is_readable_for_the_identity_check(self, db_session, printer):
+        await persist_session(db_session, self._session(printer.id), None)
+
+        assert await get_persisted_print_name(db_session, printer.id) == "AMS_Rack"
+
+    @pytest.mark.asyncio
+    async def test_completion_falls_back_to_the_persisted_row(self, db_session, printer):
+        """No in-memory session (the restart case): the plate, the mapping and
+        the assignment snapshot must still reach the 3MF path."""
+        await persist_session(db_session, self._session(printer.id), [(2, 0), (3, 675)])
+        _active_sessions.clear()
+
+        captured = {}
+
+        async def _fake_track(*args, **kwargs):
+            captured.update(kwargs)
+            return []
+
+        with (
+            patch("backend.app.api.routes.settings.get_setting", new_callable=AsyncMock, return_value=None),
+            patch("backend.app.services.usage_tracker._track_from_3mf", side_effect=_fake_track),
+        ):
+            await on_print_complete(
+                printer.id,
+                {"status": "completed", "subtask_name": "AMS_Rack"},
+                MagicMock(),
+                db_session,
+                archive_id=312,
+            )
+
+        assert captured["plate_id"] == 1
+        assert captured["ams_mapping"] == [2]
+        assert captured["tray_now_at_start"] == 2
+        assert captured["spool_assignments"] == {(0, 2): 69, (0, 3): 68}
+
+
+class TestPlateIdRecovery:
+    """Without the plate, the 3MF parser sums every plate in the file and the
+    whole multi-plate total lands on one spool."""
+
+    @pytest.mark.asyncio
+    async def test_archive_plate_id_is_used_when_the_session_is_gone(self):
+        archive = _make_archive(archive_id=312, plate_id=1)
+        spool = _make_spool(spool_id=68)
+        assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
+        db = _mock_db_sequential([archive, None, assignment, spool])
+        seen_plate_ids: list = []
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=809,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=809,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            _patched_3mf([{"slot_id": 1, "used_g": 1122.44, "type": "ABS", "color": "#808080"}], seen_plate_ids),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            await _track_from_3mf(
+                printer_id=1,
+                archive_id=312,
+                status="completed",
+                print_name="AMS_Rack",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=None,
+            )
+
+        assert seen_plate_ids == [1]
+
+    @pytest.mark.asyncio
+    async def test_queue_item_plate_id_is_used_when_the_archive_has_none(self):
+        archive = _make_archive(archive_id=312, plate_id=None)
+        queue_item = _make_queue_item(plate_id=2)
+        spool = _make_spool(spool_id=68)
+        assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
+        # db: archive, the single queue lookup (plate + mapping share it),
+        # then assignment and spool
+        db = _mock_db_sequential([archive, queue_item, assignment, spool])
+        seen_plate_ids: list = []
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=361,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=361,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            _patched_3mf([{"slot_id": 1, "used_g": 318.82, "type": "ABS", "color": "#808080"}], seen_plate_ids),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            await _track_from_3mf(
+                printer_id=1,
+                archive_id=312,
+                status="completed",
+                print_name="AMS_Rack",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=None,
+            )
+
+        assert seen_plate_ids == [2]
+
+    @pytest.mark.asyncio
+    async def test_caller_plate_id_wins_over_the_database(self):
+        archive = _make_archive(archive_id=312, plate_id=1)
+        spool = _make_spool(spool_id=68)
+        assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
+        db = _mock_db_sequential([archive, None, assignment, spool])
+        seen_plate_ids: list = []
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=361,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=361,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            _patched_3mf([{"slot_id": 1, "used_g": 318.82, "type": "ABS", "color": "#808080"}], seen_plate_ids),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            await _track_from_3mf(
+                printer_id=1,
+                archive_id=312,
+                status="completed",
+                print_name="AMS_Rack",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=2,
+            )
+
+        assert seen_plate_ids == [2]
+
+
+class TestMappingPriority:
+    """AMS filament backup rewrites the printer's live ``mapping`` field to the
+    substitute tray. Read at completion it names the tray that finished the
+    print, not the one the slicer assigned."""
+
+    @pytest.mark.asyncio
+    async def test_queue_mapping_beats_the_live_mqtt_mapping(self):
+        archive = _make_archive(archive_id=312, plate_id=1)
+        # Dispatched against AMS0-T2 (global tray 2); the printer now reports
+        # tray 3 because backup swapped in the neighbouring spool.
+        queue_item = _make_queue_item(ams_mapping="[2]")
+        spool_69 = _make_spool(spool_id=69)
+        assign_69 = _make_assignment(spool_id=69, ams_id=0, tray_id=2)
+        db = _mock_db_sequential([archive, queue_item, assign_69, spool_69])
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=809,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=809,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            _patched_3mf([{"slot_id": 1, "used_g": 1122.44, "type": "ABS", "color": "#808080"}]),
+        ):
+            mock_settings.base_dir = MagicMock()
+            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=312,
+                status="completed",
+                print_name="AMS_Rack",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=1,
+            )
+
+        assert len(results) == 1
+        assert results[0]["spool_id"] == 69
+        assert (results[0]["ams_id"], results[0]["tray_id"]) == (0, 2)
+
+    @pytest.mark.asyncio
+    async def test_mqtt_mapping_still_used_for_a_direct_print(self):
+        """No queue item — the live field is the only mapping there is."""
+        archive = _make_archive(archive_id=400, plate_id=1)
+        spool = _make_spool(spool_id=68)
+        assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
+        db = _mock_db_sequential([archive, None, assignment, spool])
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=100,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=100,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            _patched_3mf([{"slot_id": 1, "used_g": 50.0, "type": "ABS", "color": "#808080"}]),
+        ):
+            mock_settings.base_dir = MagicMock()
+            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=400,
+                status="completed",
+                print_name="Cover",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=1,
+            )
+
+        assert len(results) == 1
+        assert (results[0]["ams_id"], results[0]["tray_id"]) == (0, 3)
+
+
+class TestRestoreOnRestartRecovery:
+    """``on_print_running_observed`` is the only hook that fires when Bambuddy
+    comes up mid-print — the #1304 guard suppresses ``on_print_start``."""
+
+    @pytest.fixture(autouse=True)
+    def _clear_sessions(self):
+        _active_sessions.clear()
+        yield
+        _active_sessions.clear()
+
+    @pytest.fixture
+    async def printer(self, db_session):
+        row = Printer(name="H2D-1", ip_address="192.168.0.10", access_code="1234", serial_number="TESTSERIAL")
+        db_session.add(row)
+        await db_session.commit()
+        return row
+
+    def _state(self, **overrides):
+        state = SimpleNamespace(
+            subtask_name="AMS_Rack",
+            tray_change_log=[],
+            tray_now=3,
+            layer_num=700,
+            last_loaded_tray=-1,
+        )
+        for key, value in overrides.items():
+            setattr(state, key, value)
+        return state
+
+    def _session(self, printer_id, print_name="AMS_Rack"):
+        return PrintSession(
+            printer_id=printer_id,
+            print_name=print_name,
+            started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
+            tray_now_at_start=2,
+            spool_assignments={(0, 2): 69},
+            ams_mapping=[2],
+            plate_id=1,
+        )
+
+    @pytest.mark.asyncio
+    async def test_persisted_log_comes_back_onto_the_printer_state(self, db_session, printer):
+        from backend.app.main import _restore_usage_tracking_session
+
+        await persist_session(db_session, self._session(printer.id), [(2, 0)])
+        _active_sessions.clear()
+        state = self._state()
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert state.tray_change_log == [(2, 0)]
+        assert _active_sessions[printer.id].plate_id == 1
+
+    @pytest.mark.asyncio
+    async def test_entries_seen_by_this_process_are_kept_after_the_persisted_ones(self, db_session, printer):
+        from backend.app.main import _restore_usage_tracking_session
+
+        await persist_session(db_session, self._session(printer.id), [(2, 0)])
+        state = self._state(tray_change_log=[(3, 675)])
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert state.tray_change_log == [(2, 0), (3, 675)]
+
+    @pytest.mark.asyncio
+    async def test_no_persisted_row_seeds_from_the_tray_feeding_now(self, db_session, printer):
+        """A print that started before this build still gets its remaining
+        segment attributed to the right spool."""
+        from backend.app.main import _restore_usage_tracking_session
+
+        state = self._state()
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert state.tray_change_log == [(3, 700)]
+        assert state.last_loaded_tray == 3
+
+    @pytest.mark.asyncio
+    async def test_unloaded_tray_seeds_nothing(self, db_session, printer):
+        from backend.app.main import _restore_usage_tracking_session
+
+        state = self._state(tray_now=255)
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert state.tray_change_log == []
+
+    @pytest.mark.asyncio
+    async def test_a_row_from_a_different_print_is_discarded(self, db_session, printer):
+        """A completion Bambuddy never saw leaves a row behind; it must not
+        attach itself to whatever is running now."""
+        from backend.app.main import _restore_usage_tracking_session
+
+        await persist_session(db_session, self._session(printer.id, print_name="Old_Print"), [(2, 0)])
+        _active_sessions.clear()
+        state = self._state(subtask_name="AMS_Rack")
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert printer.id not in _active_sessions
+        assert await restore_session(db_session, printer.id) is None
+        # Still seeded, so the rest of the running print stays attributable.
+        assert state.tray_change_log == [(3, 700)]
+
+    @pytest.mark.asyncio
+    async def test_an_unloaded_tray_does_not_clobber_last_loaded_tray(self, db_session, printer):
+        """``last_loaded_tray`` is the fallback that survives the end-of-print
+        retract to 255; writing 255 into it would defeat its whole purpose."""
+        from backend.app.main import _restore_usage_tracking_session
+
+        state = self._state(tray_now=255, last_loaded_tray=2)
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert state.last_loaded_tray == 2
+
+    @pytest.mark.asyncio
+    async def test_a_failure_is_swallowed_so_the_caller_keeps_going(self, db_session, printer):
+        """The caller still has to capture its timelapse baseline before the
+        printer uploads the in-flight recording — there is no second chance."""
+        from backend.app.main import _restore_usage_tracking_session
+
+        broken = SimpleNamespace()  # no subtask_name, no tray fields at all
+
+        await _restore_usage_tracking_session(printer.id, broken, db_session, MagicMock())
+
+
+class TestPlateNotInTheFile:
+    """A recovered plate has to be treated as a hint, not a filter that can
+    silently zero out a print's usage."""
+
+    @pytest.mark.asyncio
+    async def test_falls_back_to_the_whole_file_when_the_plate_is_absent(self):
+        """The archive's own 3MF can be gone, with a same-named library file
+        substituted that was sliced with different plates."""
+        archive = _make_archive(archive_id=312, plate_id=7)
+        spool = _make_spool(spool_id=68)
+        assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
+        db = _mock_db_sequential([archive, None, assignment, spool])
+        calls: list = []
+
+        def _extract(path, plate_id=None):
+            calls.append(plate_id)
+            return [] if plate_id is not None else [{"slot_id": 1, "used_g": 12.0, "type": "ABS", "color": ""}]
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=10,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=10,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", side_effect=_extract),
+        ):
+            mock_settings.base_dir = MagicMock()
+            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=312,
+                status="completed",
+                print_name="AMS_Rack",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=None,
+            )
+
+        assert calls == [7, None]
+        assert len(results) == 1
+        assert results[0]["weight_used"] == 12.0
+
+    @pytest.mark.asyncio
+    async def test_a_file_with_no_usage_at_all_still_records_nothing(self):
+        archive = _make_archive(archive_id=312, plate_id=1)
+        db = _mock_db_sequential([archive, None])
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"mapping": [3]},
+            progress=100,
+            layer_num=10,
+            tray_now=255,
+            tray_change_log=[],
+            total_layers=10,
+        )
+
+        with (
+            _settings_patch() as mock_settings,
+            patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=[]),
+        ):
+            mock_settings.base_dir = MagicMock()
+            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=312,
+                status="completed",
+                print_name="AMS_Rack",
+                handled_trays=set(),
+                printer_manager=printer_manager,
+                db=db,
+                plate_id=1,
+            )
+
+        assert results == []
+
+
+class TestSpoolmanParity:
+    """Both inventory backends need the same restart protection.
+
+    Spoolman's own durable row (#1820) already carries its plate-scoped 3MF
+    figures and the mapping it was dispatched with, but not the tray-change
+    log — and that log is the only record of which spool fed which layers when
+    AMS Filament Backup swaps trays. Capturing it for one backend only would
+    leave Spoolman users with the bug this fixes for everyone else.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _clear_sessions(self):
+        _active_sessions.clear()
+        yield
+        _active_sessions.clear()
+
+    @pytest.fixture
+    async def printer(self, db_session):
+        row = Printer(name="H2D-1", ip_address="192.168.0.10", access_code="1234", serial_number="TESTSERIAL")
+        db_session.add(row)
+        await db_session.commit()
+        return row
+
+    def _printer_manager(self):
+        pm = MagicMock()
+        pm.get_status.return_value = SimpleNamespace(
+            raw_data={
+                "ams": {"ams": [{"id": 0, "tray": [{"id": 2, "remain": 84, "tray_type": "ABS"}]}]},
+                "vt_tray": [],
+                "mapping": [2],
+            },
+            tray_now=2,
+            last_loaded_tray=2,
+            tray_change_log=[(2, 0)],
+        )
+        return pm
+
+    @pytest.mark.asyncio
+    async def test_the_row_is_written_with_spoolman_enabled(self, db_session, printer):
+        from backend.app.services.usage_tracker import on_print_start
+
+        await on_print_start(
+            printer.id,
+            {"subtask_name": "AMS_Rack", "ams_mapping": [2]},
+            self._printer_manager(),
+            db=db_session,
+            spoolman_owns_usage=True,
+        )
+
+        row = await db_session.get(ActivePrintSession, printer.id)
+        assert row is not None
+        assert row.print_name == "AMS_Rack"
+        assert row.tray_change_log == [[2, 0]]
+
+    @pytest.mark.asyncio
+    async def test_spoolman_does_not_get_an_in_memory_session(self, db_session, printer):
+        """``_active_sessions`` doubles as on_ams_change's "skip the remain%
+        weight sync, the internal tracker will deduct precisely" flag (#880).
+        A session the internal tracker will never complete would suppress a
+        sync Spoolman users still need."""
+        from backend.app.services.usage_tracker import on_print_start
+
+        await on_print_start(
+            printer.id,
+            {"subtask_name": "AMS_Rack", "ams_mapping": [2]},
+            self._printer_manager(),
+            db=db_session,
+            spoolman_owns_usage=True,
+        )
+
+        assert printer.id not in _active_sessions
+
+    @pytest.mark.asyncio
+    async def test_the_internal_tracker_still_gets_one(self, db_session, printer):
+        from backend.app.services.usage_tracker import on_print_start
+
+        await on_print_start(
+            printer.id,
+            {"subtask_name": "AMS_Rack", "ams_mapping": [2]},
+            self._printer_manager(),
+            db=db_session,
+            spoolman_owns_usage=False,
+        )
+
+        assert _active_sessions[printer.id].ams_mapping == [2]
+        assert await db_session.get(ActivePrintSession, printer.id) is not None
+
+    @pytest.mark.asyncio
+    async def test_restore_can_return_the_log_without_publishing_a_session(self, db_session, printer):
+        await persist_session(
+            db_session,
+            PrintSession(
+                printer_id=printer.id,
+                print_name="AMS_Rack",
+                started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
+            ),
+            [(2, 0), (3, 675)],
+        )
+        _active_sessions.clear()
+
+        log = await restore_session(db_session, printer.id, register_active=False)
+
+        assert log == [[2, 0], [3, 675]]
+        assert printer.id not in _active_sessions
+
+    @pytest.mark.asyncio
+    async def test_discard_clears_both_halves(self, db_session, printer):
+        from backend.app.services.usage_tracker import discard_session
+
+        session = PrintSession(
+            printer_id=printer.id,
+            print_name="AMS_Rack",
+            started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
+        )
+        _active_sessions[printer.id] = session
+        await persist_session(db_session, session, [(2, 0)])
+
+        await discard_session(db_session, printer.id)
+
+        assert printer.id not in _active_sessions
+        assert await db_session.get(ActivePrintSession, printer.id) is None