fix(inventory): spool replaced mid-print is charged its part (issue #3325)

- A spool that ran out and was replaced in the same slot (external
  spool, AMS slot without backup) and assigned while the print ran
  had the whole print charged to the new spool, with the built-in
  inventory and with Spoolman. The tray never changes, so the
  tray-change split could not see it.
- New services/spool_swaps: each print logs the spool on every
  assigned tray at start and every different spool assigned since,
  with the layer. Kept on PrinterState, persisted in
  active_print_sessions.spool_change_log (new JSON column) and
  restored after a restart. All six assignment writers record it in
  their own transaction; the state takes it only once that commits.
- At completion a swapped tray's grams are split by the G-code feed
  between the swap layers (layer count without it), proportionally,
  for completed and failed prints, inside AMS-backup segments too.
  A tray empty at start keeps its first assignment for the whole
  print; a correction before layer 1 is not a swap; remain% across a
  real swap is not charged.
- Tests for the split, recording, rollback, restart, both backends'
  charge paths and routes, and a guard that every assignment writer
  logs swaps.

CHANGELOG.md

@@ -174,6 +174,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **A spool replaced during a print is charged only for its part (#3325, reported by @HarryCoff2)** — When a spool ran out and a new one was loaded in the same slot (an external spool, or an AMS slot without a backup) and assigned in Bambuddy while the print ran, the whole print was charged to the new spool. Bambuddy now notes the layer of the new assignment and, when the print ends, splits the filament between the two spools by what the G-code feeds before and after it, for completed and failed prints, with the built-in inventory and with Spoolman, also after a restart mid-print. A slot that had no spool when the print started is still charged to the spool assigned to it, for the whole print.
 - **A write is saved before its response is sent (#3320, reported by @MorganMLGman)** — Many API routes left the final save to the end of the request, and that ran only after the response had gone out. A client reading straight after a successful write could miss it (a new project came back as 404, an edit still showed the old value), and a save that failed at that point was undone behind a success the client already had. Every route now saves before it answers, and a failed save is reported as an error.
 - **A failed or cancelled print's card says its cost is for one full print (#3318, reported by @ojimpo)** — After a print failed or was cancelled, the archive card showed the cost and grams of the whole file with no hint, as if that were what the run cost. The archive keeps one full print's figures on purpose (reprints are estimated and billed from them); the cost and grams now carry a tooltip saying so, pointing to the archive's Print Log for what the run used. With Spoolman, that run's cost is priced from the spools it was charged to.
 - **A print that fails or is cancelled early is charged only for the filament it used (#3317, reported by @ojimpo)** — Two errors added up to a failed print being charged to Spoolman as if it had finished, sometimes for more than the finished print would have used.

backend/app/api/routes/inventory.py

@@ -60,7 +60,7 @@ from backend.app.schemas.supplier import (
     SupplierStats,
     SupplierUpdate,
 )
-from backend.app.services import slot_unlink_grace
+from backend.app.services import slot_unlink_grace, spool_swaps
 from backend.app.services.ams_assignment_lock import get_ams_assignment_lock
 from backend.app.services.ams_slot_presence import spool_present
 from backend.app.services.location_service import (
@@ -2322,6 +2322,9 @@ async def assign_spool(
             fingerprint_type=fingerprint_type,
         )
         db.add(assignment)
+        await spool_swaps.record_assignment(
+            db, data.printer_id, data.ams_id, data.tray_id, data.spool_id, spool_swaps.INVENTORY
+        )
         await db.commit()
     await db.refresh(assignment)
     slot_unlink_grace.forget_slot(data.printer_id, data.ams_id, data.tray_id)

backend/app/api/routes/spoolman.py

@@ -27,7 +27,7 @@ from backend.app.models.spoolman_k_profile import SpoolmanKProfile
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
 from backend.app.models.user import User
 from backend.app.schemas.spoolman import SlotAmsId, SlotTrayId
-from backend.app.services import slot_unlink_grace
+from backend.app.services import slot_unlink_grace, spool_swaps
 from backend.app.services.ams_slot_presence import spool_present
 from backend.app.services.inventory_mode import spoolman_owns_assignments
 from backend.app.services.printer_manager import PrinterManager, printer_manager
@@ -965,6 +965,7 @@ async def link_spool(
                 ),
                 {"printer_id": p_id, "ams_id": a_id, "tray_id": t_id, "spool_id": spool_id},
             )
+            await spool_swaps.record_assignment(db, p_id, a_id, t_id, spool_id, spool_swaps.SPOOLMAN)
             await db.commit()
             slot_unlink_grace.forget_slot(p_id, a_id, t_id)
         except Exception as e:
@@ -1409,6 +1410,9 @@ async def create_spool_from_slot(
                     "spool_id": sync_result["id"],
                 },
             )
+            await spool_swaps.record_assignment(
+                db, req.printer_id, req.ams_id, req.tray_id, sync_result["id"], spool_swaps.SPOOLMAN
+            )
             await db.commit()
             slot_unlink_grace.forget_slot(req.printer_id, req.ams_id, req.tray_id)
         except Exception as exc:

backend/app/api/routes/spoolman_inventory.py

@@ -55,7 +55,7 @@ from backend.app.models.user import User
 from backend.app.schemas.spool import SpoolFilamentPresetBase, SpoolKProfileBase, naive_utc
 from backend.app.schemas.spoolman import SlotAmsId, SlotTrayId, SpoolmanFilamentPatch, SpoolmanSlotAssignmentEnriched
 from backend.app.schemas.supplier import SpoolSupplierLinkInput
-from backend.app.services import slot_unlink_grace
+from backend.app.services import slot_unlink_grace, spool_swaps
 from backend.app.services.location_service import (
     enrich_spool_dicts_with_location_id,
     maybe_sync_spoolman_locations,
@@ -1814,6 +1814,9 @@ async def assign_spoolman_slot(
                 "spool_id": body.spoolman_spool_id,
             },
         )
+        await spool_swaps.record_assignment(
+            db, body.printer_id, body.ams_id, body.tray_id, body.spoolman_spool_id, spool_swaps.SPOOLMAN
+        )
         await db.commit()
         slot_unlink_grace.forget_slot(body.printer_id, body.ams_id, body.tray_id)
     except Exception as exc:

backend/app/core/database.py

@@ -5863,6 +5863,11 @@ async def _run_migrations(conn):
     # on SQLite and Postgres.
     await _safe_execute(conn, "ALTER TABLE print_queue ADD COLUMN color_substitutions TEXT")
 
+    # Migration: spools assigned to a tray while a print runs, so a hand swap
+    # splits the print between them across a restart (#3325). Nullable JSON, no
+    # default -- identical DDL on SQLite and Postgres.
+    await _safe_execute(conn, "ALTER TABLE active_print_sessions ADD COLUMN spool_change_log JSON")
+
     await _migrate_auth_datetimes_to_timestamp(conn)
     await _migrate_add_missing_set_null_fks(conn)
     await _migrate_wallet_transactions_keep_deleted_users(conn)

backend/app/main.py

@@ -6938,6 +6938,7 @@ async def _restore_usage_tracking_session(printer_id: int, state, db, logger) ->
             clear_persisted_session,
             get_persisted_ams_mapping,
             get_persisted_print_name,
+            get_persisted_spool_change_log,
             restore_session,
         )
 
@@ -6969,6 +6970,12 @@ async def _restore_usage_tracking_session(printer_id: int, state, db, logger) ->
             # has captured since is newer.
             if state.print_ams_mapping is None:
                 state.print_ams_mapping = await get_persisted_ams_mapping(db, printer_id)
+            # Spools swapped on a tray before the restart (#3325); anything this
+            # process has recorded since goes after them.
+            persisted_swaps = await get_persisted_spool_change_log(db, printer_id)
+            if persisted_swaps:
+                seen = getattr(state, "spool_change_log", None) or []
+                state.spool_change_log = persisted_swaps + [item for item in seen if item not in persisted_swaps]
         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

backend/app/models/active_print_session.py

@@ -55,3 +55,7 @@ class ActivePrintSession(Base):
 
     # [[global_tray_id, layer_num], ...] mirroring PrinterState.tray_change_log.
     tray_change_log: Mapped[list | None] = mapped_column(JSON, nullable=True)
+
+    # Spool on each assigned tray at print start and every spool assigned to a
+    # tray since, mirroring PrinterState.spool_change_log (services.spool_swaps).
+    spool_change_log: Mapped[list | None] = mapped_column(JSON, nullable=True)

backend/app/services/bambu_mqtt.py

@@ -1132,6 +1132,9 @@ class PrinterState:
     # Tray change history during current print: [(global_tray_id, layer_num), ...]
     # Used by usage tracker to split filament weight on mid-print tray switch
     tray_change_log: list = field(default_factory=list)
+    # Spools assigned to a tray during the current print, and the spool each
+    # assigned tray held at its start (services.spool_swaps, #3325).
+    spool_change_log: list = field(default_factory=list)
     # Firmware version info (from info.module[name="ota"].sw_ver)
     firmware_version: str | None = None
     # Developer LAN mode: parsed from MQTT "fun" field bit 0x20000000
@@ -5786,6 +5789,8 @@ class BambuMQTTClient:
             self._last_valid_layer_num = 0
             # Clear and seed tray change log for mid-print usage splitting
             self.state.tray_change_log.clear()
+            # Seeded with the assigned spools by usage_tracker.on_print_start.
+            self.state.spool_change_log = []
             tn = self.state.tray_now
             if (
                 (0 <= tn <= 15)

backend/app/services/spool_swaps.py

@@ -0,0 +1,246 @@
+"""A spool swapped on one tray while a print runs (#3325).
+
+The tray-change log (``utils.tray_split``) splits a print across trays when the
+AMS switches to a backup on a runout. A spool replaced by hand in the *same*
+tray leaves no trace there: the external spool on a printer without an AMS is
+tray 254 before and after, and so is an AMS slot refilled with a new roll. The
+only record of the swap is the operator assigning the new spool, and without a
+layer to put it at, completion charged the whole print to whichever spool the
+tray held at the end.
+
+So a print keeps a second log: the spool on every assigned tray at print start,
+and every later assignment of a different spool to a tray, with the layer the
+printer was on. At completion a tray's grams are divided between its spools in
+proportion to the filament the G-code feeds between those layers (by layer
+count without G-code data). Proportional, so the shares always add up to the
+tray's total and the last spool is never left with nothing.
+
+A tray with no spool at print start is not split by its first assignment:
+that is the operator catching up on a forgotten one, and that spool takes the
+print from its start, as before. Only a later, different spool splits it.
+
+The log lives on ``PrinterState`` (Spoolman reports after the print session
+row is gone) and is mirrored to ``active_print_sessions`` for a restart.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from sqlalchemy import event
+
+from backend.app.utils import threemf_tools
+
+logger = logging.getLogger(__name__)
+
+INVENTORY = "inventory"
+SPOOLMAN = "spoolman"
+
+
+def entry(backend: str, ams_id: int, tray_id: int, spool_id: int, layer: int, start: bool = False) -> dict:
+    return {"backend": backend, "ams": ams_id, "tray": tray_id, "spool": spool_id, "layer": layer, "start": start}
+
+
+def start_entries(backend: str, assignments: dict[tuple[int, int], int]) -> list[dict]:
+    """The spool on every assigned tray as the print starts."""
+    return [
+        entry(backend, ams_id, tray_id, spool_id, 0, start=True) for (ams_id, tray_id), spool_id in assignments.items()
+    ]
+
+
+def _valid(item) -> bool:
+    return isinstance(item, dict) and {"backend", "ams", "tray", "spool", "layer"} <= item.keys()
+
+
+def segments(log: list | None, backend: str, ams_id: int, tray_id: int) -> list[tuple[int, int]]:
+    """``[(spool_id, from_layer), ...]`` for one tray, or ``[]`` when it was not swapped.
+
+    On a tray that had no spool at print start, the first assignment is a
+    forgotten one caught up: that spool covers the print from its start, and
+    only a later, different spool is a swap. Repeats of the same spool
+    collapse; a swap back to an earlier spool is its own segment.
+    """
+    items = [
+        item
+        for item in (log or [])
+        if _valid(item) and item["backend"] == backend and item["ams"] == ams_id and item["tray"] == tray_id
+    ]
+    result: list[tuple[int, int]] = []
+    for item in items:
+        if result and result[-1][0] == item["spool"]:
+            continue
+        result.append((item["spool"], max(0, int(item["layer"] or 0)) if result else 0))
+    return result if len(result) > 1 else []
+
+
+def split(
+    segs: list[tuple[int, int]],
+    grams: float,
+    filament_id: int | None,
+    layer_usage: dict[int, dict[int, float]] | None,
+    end_layer: int,
+    from_layer: int = 0,
+) -> list[tuple[int, float]]:
+    """Divide ``grams`` fed between ``from_layer`` and ``end_layer`` among the spools of ``segs``.
+
+    Returns ``[(spool_id, grams), ...]`` in order, one entry per spool, the
+    shares summing to ``grams``. With nothing to weigh by (no G-code data, no
+    layer range) everything goes to the spool loaded at ``end_layer``.
+    """
+    end_layer = max(end_layer, from_layer)
+    bounds: list[tuple[int, int]] = []  # (spool_id, from_layer) restricted to the range
+    for spool_id, layer in segs:
+        if layer <= from_layer:
+            bounds = [(spool_id, from_layer)]
+        elif layer < end_layer:
+            bounds.append((spool_id, layer))
+    if not bounds:
+        bounds = [(segs[-1][0], from_layer)]
+
+    edges = [layer for _, layer in bounds] + [end_layer]
+    weights: list[float] = []
+    if layer_usage and filament_id is not None:
+
+        def mm_at(layer: int) -> float:
+            return threemf_tools.get_cumulative_usage_at_layer(layer_usage, layer).get(filament_id, 0.0)
+
+        weights = [max(0.0, mm_at(edges[i + 1]) - mm_at(edges[i])) for i in range(len(bounds))]
+    if sum(weights) <= 0:
+        weights = [float(edges[i + 1] - edges[i]) for i in range(len(bounds))]
+    total = sum(weights)
+
+    shares: dict[int, float] = {}
+    if total <= 0:
+        shares[bounds[-1][0]] = grams
+    else:
+        for (spool_id, _), weight in zip(bounds, weights, strict=True):
+            shares[spool_id] = shares.get(spool_id, 0.0) + grams * weight / total
+    return list(shares.items())
+
+
+class Splitter:
+    """One print's swap log, ready to share a tray's grams out (one per backend).
+
+    ``layer_usage`` is the G-code's per-layer usage, or a callable returning it,
+    read only when a tray was actually swapped. ``end_layer`` is where the
+    print stopped; for a completed print the G-code's last layer wins over it.
+    """
+
+    def __init__(self, log: list | None, backend: str, layer_usage, end_layer: int, completed: bool) -> None:
+        self.log = list(log or [])
+        self.backend = backend
+        # The per-layer usage, or until first read, the callable that returns it.
+        self._layer_usage: Any = layer_usage
+        self.end_layer = end_layer
+        self.completed = completed
+
+    def _usage(self) -> dict[int, dict[int, float]] | None:
+        if callable(self._layer_usage):
+            try:
+                self._layer_usage = self._layer_usage()
+            except Exception:  # noqa: BLE001 -- the split falls back to layer counts
+                logger.debug("[SPOOL-SWAP] No per-layer usage for the split", exc_info=True)
+                self._layer_usage = None
+        return self._layer_usage
+
+    def swapped(self, ams_id: int, tray_id: int) -> bool:
+        return bool(segments(self.log, self.backend, ams_id, tray_id))
+
+    def shares(
+        self,
+        ams_id: int,
+        tray_id: int,
+        slot_id: int,
+        grams: float,
+        from_layer: int = 0,
+        to_layer: int | None = None,
+    ) -> list[tuple[int, float]] | None:
+        """``[(spool_id, grams), ...]`` for a swapped tray, None for one that kept its spool."""
+        segs = segments(self.log, self.backend, ams_id, tray_id)
+        if not segs:
+            return None
+        usage = self._usage()
+        if to_layer is not None:
+            end = to_layer
+        elif self.completed and usage:
+            end = max(usage)
+        else:
+            end = self.end_layer
+        result = split(segs, grams, slot_id - 1, usage, end, from_layer)
+        logger.info(
+            "[SPOOL-SWAP] %s AMS%d-T%d swapped during the print %s: %.2fg split %s",
+            self.backend,
+            ams_id,
+            tray_id,
+            segs,
+            grams,
+            [(spool, round(share, 2)) for spool, share in result],
+        )
+        return result
+
+
+def _on_commit(db, apply) -> None:
+    """Run ``apply`` once ``db`` commits; never if it rolls back first."""
+    sync = getattr(db, "sync_session", None)
+    if sync is None:
+        apply()
+        return
+    settled: list[bool] = []
+
+    def committed(_session) -> None:
+        if not settled:
+            settled.append(True)
+            apply()
+
+    def rolled_back(_session) -> None:
+        settled.append(False)
+
+    event.listen(sync, "after_commit", committed, once=True)
+    event.listen(sync, "after_rollback", rolled_back, once=True)
+
+
+async def record_assignment(db, printer_id: int, ams_id: int, tray_id: int, spool_id: int, backend: str) -> None:
+    """Note that ``spool_id`` was assigned to a tray, if that printer is printing.
+
+    Runs on the caller's session, before the caller commits the assignment:
+    a second session would wait on the SQLite write lock the assignment holds.
+    The log is the persisted row's merged with ``PrinterState``'s: after a
+    restart the state is empty until the printer reports RUNNING again, and a
+    runout leaves it paused, so the row alone holds the print's start spools.
+    ``PrinterState`` takes the new entry only once the assignment commits; one
+    that fails must not split the print. Best-effort: an assignment must never
+    fail because of this.
+    """
+    try:
+        from backend.app.models.active_print_session import ActivePrintSession
+        from backend.app.services.printer_manager import printer_manager
+
+        state = printer_manager.get_status(printer_id)
+        if state is None:
+            return
+        # The row exists from print start to completion: a print Bambuddy is
+        # tracking is running.
+        row = await db.get(ActivePrintSession, printer_id)
+        if row is None:
+            return
+        log = [item for item in (row.spool_change_log or []) if _valid(item)]
+        log += [item for item in (getattr(state, "spool_change_log", None) or []) if _valid(item) and item not in log]
+        same_tray = [i for i in log if _valid(i) and (i["backend"], i["ams"], i["tray"]) == (backend, ams_id, tray_id)]
+        if same_tray and same_tray[-1]["spool"] == spool_id:
+            return
+        layer = int(getattr(state, "layer_num", 0) or 0)
+        log.append(entry(backend, ams_id, tray_id, spool_id, layer))
+        row.spool_change_log = list(log)
+        _on_commit(db, lambda: setattr(state, "spool_change_log", log))
+        logger.info(
+            "[SPOOL-SWAP] Printer %d %s AMS%d-T%d: spool %d assigned at layer %d during the print",
+            printer_id,
+            backend,
+            ams_id,
+            tray_id,
+            spool_id,
+            layer,
+        )
+    except Exception:  # noqa: BLE001 -- never fail an assignment over its usage note
+        logger.warning("[SPOOL-SWAP] Could not record a mid-print assignment on printer %d", printer_id, exc_info=True)

backend/app/services/spool_tag_matcher.py

@@ -9,7 +9,7 @@ from sqlalchemy.orm import selectinload
 from backend.app.models.spool import Spool
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.schemas.spool import normalize_effect_type
-from backend.app.services import slot_unlink_grace
+from backend.app.services import slot_unlink_grace, spool_swaps
 from backend.app.services.color_catalog_lookup import resolve_bambu_color
 from backend.app.services.slot_nozzle import resolve_slot_nozzle
 from backend.app.services.spool_filament_preset import printer_safe_filament_id, resolve_spool_preset
@@ -576,6 +576,7 @@ async def auto_assign_spool(
     )
     db.add(assignment)
     await db.flush()
+    await spool_swaps.record_assignment(db, printer_id, ams_id, tray_id, spool.id, spool_swaps.INVENTORY)
     slot_unlink_grace.forget_slot(printer_id, ams_id, tray_id)
 
     # Apply K-profile via MQTT (if available)

backend/app/services/spoolman_slot_sync.py

@@ -21,7 +21,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.core.websocket import ws_manager
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
-from backend.app.services import slot_unlink_grace
+from backend.app.services import slot_unlink_grace, spool_swaps
 
 logger = logging.getLogger(__name__)
 
@@ -75,6 +75,7 @@ async def write_slot_results(
         if cast(CursorResult, await db.execute(statement)).rowcount:
             changed.append(slot)
             assigned.append(slot)
+            await spool_swaps.record_assignment(db, printer_id, ams_id, tray_id, spool_id, spool_swaps.SPOOLMAN)
     for slot in emptied:
         # A slot the sync read without a spool matches no row below: there is
         # nothing of the sync's own to remove.
@@ -114,6 +115,7 @@ async def _write_unconditionally(
             ).bindparams(printer_id=printer_id, ams_id=ams_id, tray_id=tray_id, spool_id=spool_id)
         )
         assigned.append((printer_id, ams_id, tray_id))
+        await spool_swaps.record_assignment(db, printer_id, ams_id, tray_id, spool_id, spool_swaps.SPOOLMAN)
     removed: list[Slot] = []
     for printer_id, ams_id, tray_id in emptied:
         await db.execute(

backend/app/services/spoolman_tracking.py

@@ -17,6 +17,7 @@ from sqlalchemy.engine import CursorResult
 from backend.app.api.routes._spoolman_helpers import spoolman_net_weight
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session
+from backend.app.services import spool_swaps
 from backend.app.services.spoolman import (
     SpoolmanClientError,
     SpoolmanNotFoundError,
@@ -935,6 +936,46 @@ class _PrintCost:
         self.priced += 1
 
 
+def _swap_splitter(printer_id: int, state, layer_usage_raw, end_layer: int, completed: bool) -> spool_swaps.Splitter:
+    """The print's spool swaps (services.spool_swaps) for the Spoolman charges.
+
+    ``layer_usage_raw`` is the tracking row's JSON, keys as strings; parsed only
+    when a tray was actually swapped.
+    """
+    if state is None:
+        from backend.app.services.printer_manager import printer_manager
+
+        state = printer_manager.get_status(printer_id)
+    if not isinstance(end_layer, int) or end_layer <= 0:
+        # The caller reads the printer's layers for single-slot prints only;
+        # a multi-colour print's swap needs them as much.
+        total = getattr(state, "total_layers", 0)
+        current = getattr(state, "layer_num", 0)
+        end_layer = max(total if isinstance(total, int) else 0, current if isinstance(current, int) else 0)
+
+    def layer_usage():
+        if not layer_usage_raw:
+            return None
+        return {int(layer): {int(fid): mm for fid, mm in f.items()} for layer, f in layer_usage_raw.items()}
+
+    return spool_swaps.Splitter(
+        getattr(state, "spool_change_log", None) if state is not None else None,
+        spool_swaps.SPOOLMAN,
+        layer_usage,
+        end_layer,
+        completed=completed,
+    )
+
+
+async def _fetch_spool_for_price(client, spool_id: int) -> dict | None:
+    """The spool row a swapped-in spool's share is priced from; None if it can't be read."""
+    try:
+        return await client.get_spool(spool_id)
+    except Exception as exc:  # noqa: BLE001 -- pricing is best-effort, the charge is not
+        logger.debug("[SPOOLMAN] Could not fetch spool %s for its price: %s", spool_id, exc)
+        return None
+
+
 async def _report_spool_usage_for_slots(
     client,
     filament_usage_items: list[tuple[int, float]],
@@ -946,6 +987,7 @@ async def _report_spool_usage_for_slots(
     slot_colors_out: dict[int, str] | None = None,
     slot_materials_out: dict[int, str] | None = None,
     cost_out: _PrintCost | None = None,
+    swaps: spool_swaps.Splitter | None = None,
 ) -> int:
     """Report usage to Spoolman for a list of (slot_id, grams) pairs.
 
@@ -1012,6 +1054,16 @@ async def _report_spool_usage_for_slots(
             if spool_id_to_use is not None:
                 resolution_path = "slot-assignment"
 
+        # A spool swapped by hand on this tray mid-print shares the slot's
+        # grams with the one it replaced (#3325); the first is the archive's.
+        shares = None
+        if swaps is not None:
+            shares = swaps.shares(*_global_tray_id_to_ams_slot(global_tray_id), slot_id, grams_used)
+            if shares and shares[0][0] != spool_id_to_use:
+                spool_id_to_use = shares[0][0]
+                resolution_path = "swap log"
+                spool_obj = spool_color_hex = spool_material = None
+
         if spool_id_to_use is None:
             logger.debug(
                 "[SPOOLMAN] Slot %s: no spool resolved (tag=%s, no slot-assignment)",
@@ -1044,23 +1096,31 @@ async def _report_spool_usage_for_slots(
             if slot_materials_out is not None and spool_material:
                 slot_materials_out[slot_id] = spool_material
 
-        try:
-            await client.use_spool(spool_id_to_use, grams_used)
-            logger.info(
-                "[SPOOLMAN] %s: slot %s: %sg -> spool %s (via %s)",
-                method_label,
-                slot_id,
-                grams_used,
-                spool_id_to_use,
-                resolution_path,
-            )
-            spools_updated += 1
-            # Priced only after the charge landed, so a spool Spoolman refused
-            # cannot contribute to what the print is said to have cost.
-            if cost_out is not None:
-                cost_out.add(grams_used, spool_obj, f"Slot {slot_id}")
-        except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
-            logger.warning("[SPOOLMAN] Failed to record usage for spool %s: %s", spool_id_to_use, exc)
+        for charge_spool_id, charge_grams in shares or [(spool_id_to_use, grams_used)]:
+            charge_obj = spool_obj
+            if shares:
+                charge_grams = round(charge_grams, 2)
+                if charge_grams <= 0:
+                    continue
+                if charge_spool_id != spool_id_to_use:
+                    charge_obj = await _fetch_spool_for_price(client, charge_spool_id) if cost_out is not None else None
+            try:
+                await client.use_spool(charge_spool_id, charge_grams)
+                logger.info(
+                    "[SPOOLMAN] %s: slot %s: %sg -> spool %s (via %s)",
+                    method_label,
+                    slot_id,
+                    charge_grams,
+                    charge_spool_id,
+                    resolution_path,
+                )
+                spools_updated += 1
+                # Priced only after the charge landed, so a spool Spoolman refused
+                # cannot contribute to what the print is said to have cost.
+                if cost_out is not None:
+                    cost_out.add(charge_grams, charge_obj, f"Slot {slot_id}")
+            except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
+                logger.warning("[SPOOLMAN] Failed to record usage for spool %s: %s", charge_spool_id, exc)
 
     return spools_updated
 
@@ -1080,6 +1140,7 @@ async def _report_spool_usage_split_by_tray_changes(
     slot_colors_out: dict[int, str] | None = None,
     slot_materials_out: dict[int, str] | None = None,
     cost_out: _PrintCost | None = None,
+    swaps: spool_swaps.Splitter | None = None,
 ) -> tuple[int, set[int]]:
     """Split each slot's grams across ``tray_changes`` and charge per-segment.
 
@@ -1151,6 +1212,22 @@ async def _report_spool_usage_split_by_tray_changes(
                 if spool_id_to_use is not None:
                     resolution_path = "slot-assignment"
 
+            # A spool swapped by hand on this tray within the segment (#3325).
+            seg_shares = None
+            if swaps is not None:
+                is_last = seg_idx + 1 >= len(tray_changes)
+                seg_shares = swaps.shares(
+                    *_global_tray_id_to_ams_slot(tray_global),
+                    slot_id,
+                    segment_grams,
+                    tray_changes[seg_idx][1],
+                    None if is_last else tray_changes[seg_idx + 1][1],
+                )
+                if seg_shares and seg_shares[0][0] != spool_id_to_use:
+                    spool_id_to_use = seg_shares[0][0]
+                    resolution_path = "swap log"
+                    spool_obj = spool_color_hex = spool_material = None
+
             if spool_id_to_use is None:
                 logger.info(
                     "[SPOOLMAN] Split slot %s seg %s tray=%d: no spool resolved — %.2fg lost from split accounting",
@@ -1188,29 +1265,35 @@ async def _report_spool_usage_split_by_tray_changes(
             if slot_materials_out is not None and slot_id not in slot_materials_out and spool_material:
                 slot_materials_out[slot_id] = spool_material
 
-            try:
-                await client.use_spool(spool_id_to_use, round(segment_grams, 2))
-                logger.info(
-                    "[SPOOLMAN] %s: slot %s seg %s tray=%d: %.2fg -> spool %s (via %s)",
-                    method_label,
-                    slot_id,
-                    seg_idx,
-                    tray_global,
-                    segment_grams,
-                    spool_id_to_use,
-                    resolution_path,
-                )
-                spools_updated += 1
-                if cost_out is not None:
-                    cost_out.add(round(segment_grams, 2), spool_obj, f"Split slot {slot_id} seg {seg_idx}")
-            except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
-                logger.warning(
-                    "[SPOOLMAN] Split slot %s seg %s: failed to record usage for spool %s: %s",
-                    slot_id,
-                    seg_idx,
-                    spool_id_to_use,
-                    exc,
-                )
+            for charge_spool_id, charge_grams in seg_shares or [(spool_id_to_use, segment_grams)]:
+                if charge_grams <= 0:
+                    continue
+                charge_obj = spool_obj
+                if charge_spool_id != spool_id_to_use:
+                    charge_obj = await _fetch_spool_for_price(client, charge_spool_id) if cost_out is not None else None
+                try:
+                    await client.use_spool(charge_spool_id, round(charge_grams, 2))
+                    logger.info(
+                        "[SPOOLMAN] %s: slot %s seg %s tray=%d: %.2fg -> spool %s (via %s)",
+                        method_label,
+                        slot_id,
+                        seg_idx,
+                        tray_global,
+                        charge_grams,
+                        charge_spool_id,
+                        resolution_path,
+                    )
+                    spools_updated += 1
+                    if cost_out is not None:
+                        cost_out.add(round(charge_grams, 2), charge_obj, f"Split slot {slot_id} seg {seg_idx}")
+                except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
+                    logger.warning(
+                        "[SPOOLMAN] Split slot %s seg %s: failed to record usage for spool %s: %s",
+                        slot_id,
+                        seg_idx,
+                        charge_spool_id,
+                        exc,
+                    )
 
     return spools_updated, handled_global_tray_ids
 
@@ -1339,6 +1422,8 @@ async def _report_partial_usage(
     # (#1793). The G-code split is not used here: its last segment runs to the
     # end of the whole print. A multi-material print keeps the mapping.
     tray_changes = list(getattr(state, "tray_change_log", None) or []) if state else []
+    # Spools swapped by hand on a tray, shared out up to where the print stopped (#3325).
+    swaps = _swap_splitter(printer_id, state, layer_usage, current_layer, completed=False)
 
     async def _charge(usage_items: list[tuple[int, float]], method_label: str) -> int:
         used = [(slot_id, grams) for slot_id, grams in usage_items if grams > 0]
@@ -1359,6 +1444,7 @@ async def _report_partial_usage(
                 printer_serial,
                 printer_id=printer_id,
                 cost_out=cost_out,
+                swaps=swaps,
             )
             return updated
         return await _report_spool_usage_for_slots(
@@ -1370,6 +1456,7 @@ async def _report_partial_usage(
             printer_serial,
             printer_id=printer_id,
             cost_out=cost_out,
+            swaps=swaps,
         )
 
     # Try to use accurate G-code parsed data
@@ -1583,6 +1670,7 @@ async def report_usage(printer_id: int, archive_id: int):
         # Priced as each charge lands, so the figure the archive ends up with
         # describes the same grams Spoolman actually had deducted (#2591).
         print_cost = _PrintCost()
+        swaps = _swap_splitter(printer_id, _state, layer_usage_raw, _layer_denom_hint, completed=True)
         handled_global_tray_ids: set[int] = set()
         spools_updated = 0
 
@@ -1629,6 +1717,7 @@ async def report_usage(printer_id: int, archive_id: int):
                     slot_colors_out=slot_colors,
                     slot_materials_out=slot_materials,
                     cost_out=print_cost,
+                    swaps=swaps,
                 )
                 spools_updated += split_updated
                 handled_global_tray_ids |= split_handled
@@ -1646,6 +1735,7 @@ async def report_usage(printer_id: int, archive_id: int):
                     slot_colors_out=slot_colors,
                     slot_materials_out=slot_materials,
                     cost_out=print_cost,
+                    swaps=swaps,
                 )
                 # Track which physical slots the 3MF path already covered so
                 # Path 2 doesn't double-charge them.

backend/app/services/usage_tracker.py

@@ -20,6 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.app.models.spool import Spool
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spool_usage_history import SpoolUsageHistory
+from backend.app.services import spool_swaps
 
 logger = logging.getLogger(__name__)
 
@@ -358,6 +359,7 @@ async def persist_session(
     db: AsyncSession,
     session: PrintSession,
     tray_change_log: list | None = None,
+    spool_change_log: list | None = None,
 ) -> None:
     """Mirror ``session`` into ``active_print_sessions`` for restart recovery.
 
@@ -380,6 +382,7 @@ async def persist_session(
     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
+    row.spool_change_log = list(spool_change_log or []) or None
 
     await db.commit()
 
@@ -419,6 +422,14 @@ async def get_persisted_print_name(db: AsyncSession, printer_id: int) -> str | N
     return row.print_name if row is not None else None
 
 
+async def get_persisted_spool_change_log(db: AsyncSession, printer_id: int) -> list[dict]:
+    """The mid-print spool assignments on the persisted row (services.spool_swaps)."""
+    from backend.app.models.active_print_session import ActivePrintSession
+
+    row = await db.get(ActivePrintSession, printer_id)
+    return [item for item in (row.spool_change_log or []) if isinstance(item, dict)] if row is not None else []
+
+
 async def get_persisted_ams_mapping(db: AsyncSession, printer_id: int) -> list[int] | None:
     """AMS mapping on the persisted row: the one captured from the print command."""
     from backend.app.models.active_print_session import ActivePrintSession
@@ -685,6 +696,22 @@ async def on_print_start(
                 {f"{k[0]}-{k[1]}": v for k, v in spool_assignments.items()},
             )
 
+    # The spool each assigned tray starts the print with, in the backend that
+    # writes the usage: a spool swapped on a tray mid-print is split against it
+    # (#3325).
+    if spoolman_owns_usage and db:
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+
+        spoolman_slots = {
+            (row.ams_id, row.tray_id): row.spoolman_spool_id
+            for row in (
+                await db.execute(select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == printer_id))
+            ).scalars()
+        }
+        state.spool_change_log = spool_swaps.start_entries(spool_swaps.SPOOLMAN, spoolman_slots)
+    else:
+        state.spool_change_log = spool_swaps.start_entries(spool_swaps.INVENTORY, spool_assignments)
+
     # Capture the queue item's plate_id so 3MF parsing at completion is scoped to
     # the plate that actually ran, not the whole multi-plate file (#1697).
     plate_id: int | None = None
@@ -722,7 +749,12 @@ async def on_print_start(
     # tray by bambu_mqtt before this callback fires.
     if db:
         try:
-            await persist_session(db, session, getattr(state, "tray_change_log", None))
+            await persist_session(
+                db,
+                session,
+                getattr(state, "tray_change_log", None),
+                getattr(state, "spool_change_log", None),
+            )
         except Exception:
             logger.exception("[UsageTracker] Failed to persist print session for printer %d", printer_id)
 
@@ -969,6 +1001,21 @@ async def on_print_complete(
                     )
                     continue
 
+                swap_segs = spool_swaps.segments(
+                    getattr(state, "spool_change_log", None), spool_swaps.INVENTORY, assign_ams_id, assign_tray_id
+                )
+                if any(layer > 0 for _, layer in swap_segs[1:]):
+                    # Start and end remain% belong to two different spools: the
+                    # difference is neither one's usage (#3325). An assignment
+                    # corrected before the first layer is not a swap; the live
+                    # spool takes the delta below, as it always did.
+                    logger.warning(
+                        "[UsageTracker] %s: spool swapped during the print, remain%% delta not charged for printer %d",
+                        tray_label,
+                        printer_id,
+                    )
+                    continue
+
                 spool_id = await _resolve_spool_id_for_tray(
                     printer_id=printer_id,
                     ams_id=assign_ams_id,
@@ -1721,6 +1768,29 @@ async def _track_from_3mf(
     charged_slots: set[int] = set()
     covered_before = frozenset(handled_trays)
 
+    # Spools swapped by hand on one tray mid-print (#3325). The G-code's
+    # per-layer usage is read only when a tray was actually swapped.
+    _swap_state = printer_manager.get_status(printer_id)
+    _swap_total = getattr(_swap_state, "total_layers", 0)
+    _swap_layer = getattr(_swap_state, "layer_num", 0)
+    if status == "completed":
+        _swap_end = max(_swap_total if isinstance(_swap_total, int) else 0, last_layer_num)
+    else:
+        _swap_end = _swap_layer if isinstance(_swap_layer, int) and _swap_layer > 0 else last_layer_num
+
+    def _read_layer_usage():
+        from backend.app.utils.threemf_tools import extract_layer_filament_usage_from_3mf
+
+        return extract_layer_filament_usage_from_3mf(file_path, plate_id)
+
+    _swap_shares = spool_swaps.Splitter(
+        getattr(_swap_state, "spool_change_log", None) if _swap_state else None,
+        spool_swaps.INVENTORY,
+        _read_layer_usage,
+        _swap_end,
+        completed=status == "completed",
+    ).shares
+
     for usage in filament_usage:
         slot_id = usage.get("slot_id", 0)
         used_g = usage.get("used_g", 0)
@@ -1805,6 +1875,15 @@ async def _track_from_3mf(
                     segment_grams,
                 )
 
+                # A spool swapped by hand on this tray within the segment (#3325).
+                seg_shares = _swap_shares(
+                    seg_ams_id,
+                    seg_tray_id,
+                    slot_id,
+                    segment_grams,
+                    seg_start_layer,
+                    None if is_last else tray_changes[seg_idx + 1][1],
+                )
                 seg_spool_id = await _resolve_spool_id_for_tray(
                     printer_id=printer_id,
                     ams_id=seg_ams_id,
@@ -1813,7 +1892,7 @@ async def _track_from_3mf(
                     spool_assignments_snapshot=spool_assignments,
                     print_started_at=print_started_at,
                 )
-                if seg_spool_id is None:
+                if seg_shares is None and seg_spool_id is None:
                     logger.info(
                         "[UsageTracker] 3MF split: no spool at printer %d AMS%d-T%d, skipping segment",
                         printer_id,
@@ -1822,58 +1901,59 @@ async def _track_from_3mf(
                     )
                     continue
 
-                spool_result = await db.execute(select(Spool).where(Spool.id == seg_spool_id))
-                spool = spool_result.scalar_one_or_none()
-                if not spool:
-                    continue
-
-                if not await _charge_spool(db, spool, segment_grams):
-                    continue
-
-                percent = round(segment_grams / (spool.label_weight or 1000) * 100)
-
-                cost = None
-                cost_per_kg = spool.cost_per_kg if spool.cost_per_kg is not None else default_filament_cost
-                if cost_per_kg > 0:
-                    cost = round((segment_grams / 1000.0) * cost_per_kg, 2)
-
-                history = SpoolUsageHistory(
-                    spool_id=spool.id,
-                    printer_id=printer_id,
-                    print_name=print_name,
-                    weight_used=round(segment_grams, 1),
-                    percent_used=percent,
-                    status=status,
-                    cost=cost,
-                    archive_id=archive_id,
-                )
-                db.add(history)
-
-                handled_trays.add(seg_key)
-                results.append(
-                    {
-                        "spool_id": spool.id,
-                        "weight_used": round(segment_grams, 1),
-                        "percent_used": percent,
-                        "ams_id": seg_ams_id,
-                        "tray_id": seg_tray_id,
-                        "material": spool.material,
-                        "cost": cost,
-                        "slot_id": slot_id,
-                        "color": _spool_color_to_hex(spool.rgba),
-                    }
-                )
+                for share_spool_id, share_grams in seg_shares or [(seg_spool_id, segment_grams)]:
+                    spool_result = await db.execute(select(Spool).where(Spool.id == share_spool_id))
+                    spool = spool_result.scalar_one_or_none()
+                    if not spool:
+                        continue
+
+                    if not await _charge_spool(db, spool, share_grams):
+                        continue
+
+                    percent = round(share_grams / (spool.label_weight or 1000) * 100)
+
+                    cost = None
+                    cost_per_kg = spool.cost_per_kg if spool.cost_per_kg is not None else default_filament_cost
+                    if cost_per_kg > 0:
+                        cost = round((share_grams / 1000.0) * cost_per_kg, 2)
+
+                    history = SpoolUsageHistory(
+                        spool_id=spool.id,
+                        printer_id=printer_id,
+                        print_name=print_name,
+                        weight_used=round(share_grams, 1),
+                        percent_used=percent,
+                        status=status,
+                        cost=cost,
+                        archive_id=archive_id,
+                    )
+                    db.add(history)
+
+                    handled_trays.add(seg_key)
+                    results.append(
+                        {
+                            "spool_id": spool.id,
+                            "weight_used": round(share_grams, 1),
+                            "percent_used": percent,
+                            "ams_id": seg_ams_id,
+                            "tray_id": seg_tray_id,
+                            "material": spool.material,
+                            "cost": cost,
+                            "slot_id": slot_id,
+                            "color": _spool_color_to_hex(spool.rgba),
+                        }
+                    )
 
-                logger.info(
-                    "[UsageTracker] Spool %d consumed %.1fg (3MF split seg%d) on printer %d AMS%d-T%d (%s)",
-                    spool.id,
-                    segment_grams,
-                    seg_idx,
-                    printer_id,
-                    seg_ams_id,
-                    seg_tray_id,
-                    status,
-                )
+                    logger.info(
+                        "[UsageTracker] Spool %d consumed %.1fg (3MF split seg%d) on printer %d AMS%d-T%d (%s)",
+                        spool.id,
+                        share_grams,
+                        seg_idx,
+                        printer_id,
+                        seg_ams_id,
+                        seg_tray_id,
+                        status,
+                    )
 
             continue  # Skip normal single-tray processing for this slot
 
@@ -1967,7 +2047,15 @@ async def _track_from_3mf(
             spool_assignments_snapshot=spool_assignments,
             print_started_at=print_started_at,
         )
-        if spool_id is None:
+        # Use per-layer grams if available, otherwise linear scale
+        if layer_grams and slot_id in layer_grams:
+            slot_grams = layer_grams[slot_id]
+        else:
+            slot_grams = used_g * scale
+
+        # A spool swapped by hand on this tray mid-print shares it (#3325).
+        shares = _swap_shares(ams_id, tray_id, slot_id, slot_grams) if slot_grams > 0 else None
+        if shares is None and spool_id is None:
             # WARNING, not INFO: everything upstream of this line succeeded --
             # the 3MF was found, the grams were read, the tray resolved -- and
             # the print will still report success while this filament is never
@@ -1983,79 +2071,76 @@ async def _track_from_3mf(
             unassigned_global_trays.append(global_tray_id)
             continue
 
-        # Load spool
-        spool_result = await db.execute(select(Spool).where(Spool.id == spool_id))
-        spool = spool_result.scalar_one_or_none()
-        if not spool:
-            continue
-
-        # Use per-layer grams if available, otherwise linear scale
-        if layer_grams and slot_id in layer_grams:
-            weight_grams = layer_grams[slot_id]
-        else:
-            weight_grams = used_g * scale
-
-        if weight_grams <= 0:
-            continue
-
-        # Update spool
-        if not await _charge_spool(db, spool, weight_grams):
+        if slot_grams <= 0:
             continue
 
-        percent = round(weight_grams / (spool.label_weight or 1000) * 100)
+        for share_spool_id, weight_grams in shares or [(spool_id, slot_grams)]:
+            # Load spool
+            spool_result = await db.execute(select(Spool).where(Spool.id == share_spool_id))
+            spool = spool_result.scalar_one_or_none()
+            if not spool:
+                continue
 
-        # Calculate cost for this usage
-        cost = None
-        cost_per_kg = spool.cost_per_kg if spool.cost_per_kg is not None else default_filament_cost
-        if cost_per_kg > 0:
-            cost = round((weight_grams / 1000.0) * cost_per_kg, 2)
+            # Update spool
+            if not await _charge_spool(db, spool, weight_grams):
+                continue
 
-        # Insert usage history record
-        history = SpoolUsageHistory(
-            spool_id=spool.id,
-            printer_id=printer_id,
-            print_name=print_name,
-            weight_used=round(weight_grams, 1),
-            percent_used=percent,
-            status=status,
-            cost=cost,
-            archive_id=archive_id,
-        )
-        db.add(history)
-
-        handled_trays.add(key)
-        results.append(
-            {
-                "spool_id": spool.id,
-                "weight_used": round(weight_grams, 1),
-                "percent_used": percent,
-                "ams_id": ams_id,
-                "tray_id": tray_id,
-                "material": spool.material,
-                "cost": cost,
-                "slot_id": slot_id,
-                "color": _spool_color_to_hex(spool.rgba),
-            }
-        )
+            percent = round(weight_grams / (spool.label_weight or 1000) * 100)
+
+            # Calculate cost for this usage
+            cost = None
+            cost_per_kg = spool.cost_per_kg if spool.cost_per_kg is not None else default_filament_cost
+            if cost_per_kg > 0:
+                cost = round((weight_grams / 1000.0) * cost_per_kg, 2)
+
+            # Insert usage history record
+            history = SpoolUsageHistory(
+                spool_id=spool.id,
+                printer_id=printer_id,
+                print_name=print_name,
+                weight_used=round(weight_grams, 1),
+                percent_used=percent,
+                status=status,
+                cost=cost,
+                archive_id=archive_id,
+            )
+            db.add(history)
+
+            handled_trays.add(key)
+            results.append(
+                {
+                    "spool_id": spool.id,
+                    "weight_used": round(weight_grams, 1),
+                    "percent_used": percent,
+                    "ams_id": ams_id,
+                    "tray_id": tray_id,
+                    "material": spool.material,
+                    "cost": cost,
+                    "slot_id": slot_id,
+                    "color": _spool_color_to_hex(spool.rgba),
+                }
+            )
 
-        # Determine mapping source for debug logging
-        if tray_now_override is not None:
-            map_src = ", tray_now"
-        elif mapping_source:
-            map_src = f", {mapping_source}_map"
-        else:
-            map_src = ""
-        logger.info(
-            "[UsageTracker] Spool %d consumed %.1fg (3MF%s%s) on printer %d AMS%d-T%d (%s)",
-            spool.id,
-            weight_grams,
-            " per-layer" if (layer_grams and slot_id in layer_grams) else (f" scaled {scale:.0%}" if scale < 1 else ""),
-            map_src,
-            printer_id,
-            ams_id,
-            tray_id,
-            status,
-        )
+            # Determine mapping source for debug logging
+            if tray_now_override is not None:
+                map_src = ", tray_now"
+            elif mapping_source:
+                map_src = f", {mapping_source}_map"
+            else:
+                map_src = ""
+            logger.info(
+                "[UsageTracker] Spool %d consumed %.1fg (3MF%s%s) on printer %d AMS%d-T%d (%s)",
+                spool.id,
+                weight_grams,
+                " per-layer"
+                if (layer_grams and slot_id in layer_grams)
+                else (f" scaled {scale:.0%}" if scale < 1 else ""),
+                map_src,
+                printer_id,
+                ams_id,
+                tray_id,
+                status,
+            )
 
     # --- Adopt the matched inventory spools' colours for the archive (#1494) ---
     # The archive's filament_color was set from the slicer's 3MF at creation

backend/tests/integration/test_mid_print_spool_swap_routes_3325.py

@@ -0,0 +1,136 @@
+"""Assigning a spool while the printer prints logs the swap (#3325).
+
+Both inventory backends' assignment routes note the new spool and the layer the
+printer is on, so completion can split the print between it and the spool the
+tray started with. Outside a print nothing is logged.
+"""
+
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.active_print_session import ActivePrintSession
+from backend.app.models.printer import Printer
+from backend.app.models.spool import Spool
+from backend.app.services import spool_swaps
+
+
+def _status(layer_num: int, log: list):
+    status = MagicMock()
+    status.raw_data = {"ams": {"ams": []}, "vt_tray": [{"id": "254", "tray_type": "PLA"}]}
+    status.nozzles = [MagicMock(nozzle_diameter="0.4")]
+    status.ams_extruder_map = None
+    status.kprofiles = []
+    status.layer_num = layer_num
+    status.spool_change_log = log
+    return status
+
+
+def _mqtt():
+    client = MagicMock()
+    client.ams_set_filament_setting.return_value = True
+    client.extrusion_cali_sel.return_value = True
+    return client
+
+
+async def _printer(db_session, serial: str, printing: bool, start_log: list | None = None) -> Printer:
+    printer = Printer(name="P2S", serial_number=serial, ip_address="192.168.1.80", access_code="12345678", model="P2S")
+    db_session.add(printer)
+    await db_session.commit()
+    await db_session.refresh(printer)
+    if printing:
+        db_session.add(ActivePrintSession(printer_id=printer.id, started_at=datetime.now(), spool_change_log=start_log))
+        await db_session.commit()
+    return printer
+
+
+class TestBuiltInInventory:
+    async def _assign(self, async_client, db_session, printing: bool):
+        old = Spool(material="PLA", rgba="000000ff", label_weight=1000, weight_used=0)
+        new = Spool(material="PLA", rgba="000000ff", label_weight=1000, weight_used=0)
+        db_session.add_all([old, new])
+        await db_session.commit()
+        start = [spool_swaps.entry(spool_swaps.INVENTORY, 255, 0, old.id, 0, start=True)]
+        printer = await _printer(db_session, f"P2S3325{int(printing)}", printing, start)
+        status = _status(57, list(start) if printing else [])
+        with patch("backend.app.services.printer_manager.printer_manager") as pm:
+            pm.get_client.return_value = _mqtt()
+            pm.get_status.return_value = status
+            response = await async_client.post(
+                "/api/v1/inventory/assignments",
+                json={"spool_id": new.id, "printer_id": printer.id, "ams_id": 255, "tray_id": 0},
+            )
+        assert response.status_code == 200
+        return printer, status, old, new
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_swap_mid_print_is_logged(self, async_client: AsyncClient, db_session: AsyncSession):
+        printer, status, old, new = await self._assign(async_client, db_session, printing=True)
+
+        assert spool_swaps.segments(status.spool_change_log, spool_swaps.INVENTORY, 255, 0) == [
+            (old.id, 0),
+            (new.id, 57),
+        ]
+        row = await db_session.get(ActivePrintSession, printer.id)
+        await db_session.refresh(row)
+        assert row.spool_change_log == status.spool_change_log
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_no_print_no_log(self, async_client: AsyncClient, db_session: AsyncSession):
+        _, status, _, _ = await self._assign(async_client, db_session, printing=False)
+
+        assert status.spool_change_log == []
+
+
+class TestSpoolman:
+    @pytest.fixture
+    async def settings(self, db_session):
+        from backend.app.models.settings import Settings
+
+        db_session.add(Settings(key="spoolman_enabled", value="true"))
+        db_session.add(Settings(key="spoolman_url", value="http://localhost:7912"))
+        await db_session.commit()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_swap_mid_print_is_logged(self, async_client: AsyncClient, db_session: AsyncSession, settings):
+        start = [spool_swaps.entry(spool_swaps.SPOOLMAN, 255, 0, 10, 0, start=True)]
+        printer = await _printer(db_session, "P2S3325SM", True, start)
+        status = _status(80, list(start))
+        spool = {
+            "id": 11,
+            "filament": {"id": 1, "name": "Black", "material": "PLA", "color_hex": "000000", "weight": 1000},
+            "remaining_weight": 1000.0,
+            "used_weight": 0.0,
+            "archived": False,
+            "extra": {},
+        }
+        client = MagicMock()
+        client.has_tag_api = AsyncMock(return_value=False)
+        client.add_native_tags = AsyncMock(return_value=0)
+        client.unlink_all_native_tags = AsyncMock()
+        client.base_url = "http://localhost:7912"
+        client.health_check = AsyncMock(return_value=True)
+        client.get_spool = AsyncMock(return_value=spool)
+        client.get_spools = AsyncMock(return_value=[spool])
+        client.merge_spool_extra = AsyncMock(return_value=spool)
+        with (
+            patch("backend.app.api.routes.spoolman_inventory.printer_manager") as route_pm,
+            patch("backend.app.services.printer_manager.printer_manager") as pm,
+            patch("backend.app.api.routes.spoolman_inventory.get_spoolman_client", AsyncMock(return_value=client)),
+        ):
+            route_pm.get_client.return_value = _mqtt()
+            route_pm.get_status.return_value = status
+            pm.get_status.return_value = status
+            response = await async_client.post(
+                "/api/v1/spoolman/inventory/slot-assignments",
+                json={"spoolman_spool_id": 11, "printer_id": printer.id, "ams_id": 255, "tray_id": 0},
+            )
+
+        assert response.status_code == 200
+        assert spool_swaps.segments(status.spool_change_log, spool_swaps.SPOOLMAN, 255, 0) == [(10, 0), (11, 80)]

backend/tests/unit/test_mid_print_spool_swap_3325.py

@@ -0,0 +1,541 @@
+"""A spool swapped by hand on one tray mid-print shares the print (#3325).
+
+The reporter's P2S has no AMS: the external spool ran out, they loaded a new
+roll and assigned it in Bambuddy while the print was paused, and the whole
+print was charged to the new spool. The tray never changed (254 before and
+after), so the tray-change split had nothing to work with; the assignment was
+the only trace of the swap.
+"""
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.models.active_print_session import ActivePrintSession
+from backend.app.models.archive import PrintArchive
+from backend.app.models.spool import Spool
+from backend.app.services import spool_swaps
+from backend.app.services.usage_tracker import _track_from_3mf
+from backend.app.utils.threemf_tools import mm_to_grams
+
+pytestmark = pytest.mark.unit
+
+INV = spool_swaps.INVENTORY
+EXT = (255, 0)  # the external spool, as assignments key it
+
+
+def _log(*swaps, start=4, backend=INV, key=EXT):
+    """A print that started with spool ``start`` on ``key`` and the given (spool, layer) swaps."""
+    log = [spool_swaps.entry(backend, *key, start, 0, start=True)]
+    log += [spool_swaps.entry(backend, *key, spool, layer) for spool, layer in swaps]
+    return log
+
+
+# 100 layers of filament 0, evenly: the whole print is 50 g of PLA, as the
+# slicer says, so the G-code and the slicer agree.
+LAYER_MM = 50.0 / mm_to_grams(1.0, 1.75, 1.24) / 100
+LAYER_USAGE = {layer: {0: layer * LAYER_MM} for layer in range(1, 101)}
+
+
+class TestSegments:
+    def test_a_swap_after_the_start(self):
+        assert spool_swaps.segments(_log((9, 60)), INV, *EXT) == [(4, 0), (9, 60)]
+
+    def test_no_swap_is_no_split(self):
+        assert spool_swaps.segments(_log(), INV, *EXT) == []
+
+    def test_reassigning_the_same_spool_is_no_swap(self):
+        assert spool_swaps.segments(_log((4, 30)), INV, *EXT) == []
+
+    def test_a_tray_empty_at_the_start_is_not_split(self):
+        # Assigning a spool mid-print to a tray that had none is catching up on
+        # a forgotten assignment: the whole print goes to it, as before.
+        log = [spool_swaps.entry(INV, *EXT, 9, 40)]
+        assert spool_swaps.segments(log, INV, *EXT) == []
+
+    def test_a_caught_up_assignment_covers_the_start_of_a_later_swap(self):
+        # Spool 9 assigned late at layer 40, then swapped for 12 at 70: 9 fed
+        # the print from its start, not from layer 40.
+        log = [spool_swaps.entry(INV, *EXT, 9, 40), spool_swaps.entry(INV, *EXT, 12, 70)]
+        assert spool_swaps.segments(log, INV, *EXT) == [(9, 0), (12, 70)]
+
+    def test_other_trays_and_backends_are_ignored(self):
+        log = _log((9, 60), key=(0, 1)) + _log((7, 20), backend=spool_swaps.SPOOLMAN)
+        assert spool_swaps.segments(log, INV, *EXT) == []
+
+    def test_swapping_back_is_a_segment_of_its_own(self):
+        assert spool_swaps.segments(_log((9, 30), (4, 70)), INV, *EXT) == [(4, 0), (9, 30), (4, 70)]
+
+
+class TestSplit:
+    def test_by_gcode_extrusion(self):
+        shares = spool_swaps.split([(4, 0), (9, 60)], 50.0, 0, LAYER_USAGE, 100)
+        assert [(s, round(g, 6)) for s, g in shares] == [(4, 30.0), (9, 20.0)]
+
+    def test_by_layer_count_without_gcode(self):
+        shares = spool_swaps.split([(4, 0), (9, 25)], 40.0, 0, None, 100)
+        assert [(s, round(g, 6)) for s, g in shares] == [(4, 10.0), (9, 30.0)]
+
+    def test_shares_always_add_up_and_never_starve_the_last_spool(self):
+        # Front-loaded G-code: nearly all filament before the swap. The new
+        # spool still gets its proportional share, and the total holds.
+        usage = {1: {0: 9000.0}, 50: {0: 9900.0}, 100: {0: 10000.0}}
+        shares = dict(spool_swaps.split([(4, 0), (9, 50)], 12.0, 0, usage, 100))
+        assert shares[9] > 0
+        assert sum(shares.values()) == pytest.approx(12.0)
+
+    def test_a_swap_back_adds_up_per_spool(self):
+        shares = dict(spool_swaps.split([(4, 0), (9, 25), (4, 75)], 100.0, 0, None, 100))
+        assert shares == pytest.approx({4: 50.0, 9: 50.0})
+
+    def test_within_a_range(self):
+        # An AMS-backup segment from layer 40 to 80, the spool swapped at 60.
+        shares = spool_swaps.split([(4, 0), (9, 60)], 8.0, 0, LAYER_USAGE, 80, from_layer=40)
+        assert [(s, round(g, 6)) for s, g in shares] == [(4, 4.0), (9, 4.0)]
+
+    def test_a_swap_after_the_print_stopped_goes_nowhere(self):
+        # A failed print stopped at layer 30; the spool was swapped at 60.
+        assert spool_swaps.split([(4, 0), (9, 60)], 5.0, 0, LAYER_USAGE, 30) == [(4, 5.0)]
+
+    def test_nothing_to_weigh_by_goes_to_the_last_spool(self):
+        assert spool_swaps.split([(4, 0), (9, 0)], 5.0, 0, None, 0) == [(9, 5.0)]
+
+
+class TestRecordAssignment:
+    async def _record(self, db_session, printer_id, state, spool_id=9):
+        with patch("backend.app.services.printer_manager.printer_manager.get_status", return_value=state):
+            await spool_swaps.record_assignment(db_session, printer_id, *EXT, spool_id, INV)
+        await db_session.commit()
+
+    @pytest.mark.asyncio
+    async def test_a_swap_while_printing_is_logged_at_the_current_layer(self, db_session, printer_factory):
+        printer = await printer_factory()
+        db_session.add(ActivePrintSession(printer_id=printer.id, started_at=datetime.now(), spool_change_log=_log()))
+        await db_session.commit()
+        state = SimpleNamespace(layer_num=57, spool_change_log=_log())
+
+        await self._record(db_session, printer.id, state)
+
+        assert spool_swaps.segments(state.spool_change_log, INV, *EXT) == [(4, 0), (9, 57)]
+        row = await db_session.get(ActivePrintSession, printer.id)
+        await db_session.refresh(row)
+        assert row.spool_change_log == state.spool_change_log
+
+    @pytest.mark.asyncio
+    async def test_nothing_is_logged_without_a_print(self, db_session, printer_factory):
+        printer = await printer_factory()
+        state = SimpleNamespace(layer_num=0, spool_change_log=[])
+
+        await self._record(db_session, printer.id, state)
+
+        assert state.spool_change_log == []
+
+    @pytest.mark.asyncio
+    async def test_an_assignment_that_rolls_back_is_not_logged(self, db_session, printer_factory):
+        printer = await printer_factory()
+        db_session.add(ActivePrintSession(printer_id=printer.id, started_at=datetime.now(), spool_change_log=_log()))
+        await db_session.commit()
+        state = SimpleNamespace(layer_num=57, spool_change_log=_log())
+
+        with patch("backend.app.services.printer_manager.printer_manager.get_status", return_value=state):
+            await spool_swaps.record_assignment(db_session, printer.id, *EXT, 9, INV)
+        await db_session.rollback()
+
+        assert state.spool_change_log == _log()
+
+    @pytest.mark.asyncio
+    async def test_after_a_restart_the_start_spool_is_kept(self, db_session, printer_factory):
+        # Bambuddy restarted during the runout pause: the state's log is empty
+        # until the printer reports RUNNING, the row still has the start spool.
+        printer = await printer_factory()
+        db_session.add(ActivePrintSession(printer_id=printer.id, started_at=datetime.now(), spool_change_log=_log()))
+        await db_session.commit()
+        state = SimpleNamespace(layer_num=57, spool_change_log=[])
+
+        await self._record(db_session, printer.id, state)
+
+        assert spool_swaps.segments(state.spool_change_log, INV, *EXT) == [(4, 0), (9, 57)]
+        row = await db_session.get(ActivePrintSession, printer.id)
+        await db_session.refresh(row)
+        assert row.spool_change_log == state.spool_change_log
+
+    @pytest.mark.asyncio
+    async def test_the_same_spool_again_is_not_logged(self, db_session, printer_factory):
+        printer = await printer_factory()
+        db_session.add(ActivePrintSession(printer_id=printer.id, started_at=datetime.now()))
+        await db_session.commit()
+        state = SimpleNamespace(layer_num=20, spool_change_log=_log())
+
+        await self._record(db_session, printer.id, state, spool_id=4)
+
+        assert state.spool_change_log == _log()
+
+
+# --- The built-in inventory -------------------------------------------------
+
+
+def _spool(spool_id: int):
+    spool = MagicMock()
+    spool.id = spool_id
+    spool.label_weight = 1000
+    spool.weight_used = 0
+    spool.cost_per_kg = None
+    spool.material = "PLA"
+    spool.rgba = None
+    return spool
+
+
+def _db(spools: dict[int, MagicMock]):
+    archive = MagicMock()
+    archive.id = 7
+    archive.file_path = "archives/7/test.3mf"
+    archive.extra_data = None
+    archive.plate_id = None
+
+    async def execute(stmt, *args, **kwargs):
+        entity = stmt.column_descriptions[0].get("entity")
+        result = MagicMock()
+        value = None
+        if entity is PrintArchive:
+            value = archive
+        elif entity is Spool:
+            value = spools.get(stmt.whereclause.right.value)
+        result.scalar_one_or_none.return_value = value
+        result.scalars.return_value.first.return_value = None
+        result.scalar.return_value = None
+        return result
+
+    db = AsyncMock()
+    db.execute = execute
+    db.add = MagicMock()
+    return db
+
+
+async def _track(
+    spool_change_log, status="completed", layer_num=100, tray_change_log=None, last_layer_num=0, total_layers=100
+):
+    printer_manager = MagicMock()
+    printer_manager.get_status.return_value = SimpleNamespace(
+        progress=100 if status == "completed" else 40,
+        layer_num=layer_num,
+        total_layers=total_layers,
+        tray_now=254,
+        last_loaded_tray=254,
+        tray_change_log=tray_change_log or [],
+        spool_change_log=spool_change_log,
+        raw_data={"ams": [], "vt_tray": [{"id": "254", "tray_type": "PLA"}]},
+    )
+
+    async def resolve(printer_id, ams_id, tray_id, **kwargs):
+        # AMS slot 0 holds spool 20; the external spool's live assignment is
+        # spool 9, the one loaded at the end.
+        return 20 if (ams_id, tray_id) == (0, 0) else 9
+
+    charged: list[tuple[int, float]] = []
+
+    async def charge(db, spool, grams):
+        charged.append((spool.id, round(grams, 2)))
+        return True
+
+    with (
+        patch("backend.app.core.config.settings") as mock_settings,
+        patch(
+            "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+            return_value=[{"slot_id": 1, "used_g": 50.0, "type": "PLA", "color": "#000000"}],
+        ),
+        patch("backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf", return_value=LAYER_USAGE),
+        patch("backend.app.utils.threemf_tools.extract_filament_properties_from_3mf", return_value={}),
+        patch("backend.app.services.usage_tracker._resolve_spool_id_for_tray", side_effect=resolve),
+        patch("backend.app.services.usage_tracker._charge_spool", side_effect=charge),
+    ):
+        mock_path = MagicMock()
+        mock_path.exists.return_value = True
+        mock_settings.base_dir = MagicMock()
+        mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+        results = await _track_from_3mf(
+            printer_id=1,
+            archive_id=7,
+            status=status,
+            print_name="swap",
+            handled_trays=set(),
+            printer_manager=printer_manager,
+            db=_db({4: _spool(4), 9: _spool(9), 20: _spool(20)}),
+            ams_mapping=[254],
+            last_layer_num=last_layer_num,
+            print_started_at=datetime.now(timezone.utc),
+        )
+    return results, charged
+
+
+class TestBuiltInInventory:
+    @pytest.mark.asyncio
+    async def test_the_print_is_split_between_the_two_spools(self):
+        results, charged = await _track(_log((9, 60)))
+        assert charged == [(4, 30.0), (9, 20.0)]
+        assert [r["spool_id"] for r in results] == [4, 9]
+
+    @pytest.mark.asyncio
+    async def test_a_printer_that_forgot_its_layer_count_still_splits(self):
+        # P1S firmware zeroes total_layer_num at print end; the G-code's own
+        # last layer bounds the split instead.
+        _, charged = await _track(_log((9, 60)), layer_num=0, total_layers=0)
+        assert charged == [(4, 30.0), (9, 20.0)]
+
+    @pytest.mark.asyncio
+    async def test_without_a_swap_the_live_spool_takes_it_all(self):
+        _, charged = await _track(_log())
+        assert charged == [(9, 50.0)]
+
+    @pytest.mark.asyncio
+    async def test_a_failed_print_splits_up_to_where_it_stopped(self):
+        # Stopped at layer 80 of 100: 40 g of the slot's 50 g by the G-code,
+        # the swap at 60 splits it 30 / 10.
+        _, charged = await _track(_log((9, 60)), status="failed", layer_num=0, last_layer_num=80)
+        assert charged == [(4, 30.0), (9, 10.0)]
+
+    @pytest.mark.asyncio
+    async def test_an_ams_backup_segment_is_split_too(self):
+        # AMS backup moved the print from AMS slot 0 to the external spool at
+        # layer 40, and that spool was swapped by hand at 60.
+        _, charged = await _track(_log((9, 60)), tray_change_log=[(0, 0), (254, 40)])
+        assert charged == [(20, 20.0), (4, 10.0), (9, 20.0)]
+
+
+# --- Spoolman ---------------------------------------------------------------
+
+
+def _client():
+    client = MagicMock()
+    client.find_spool_by_tag = AsyncMock(return_value=None)
+    client.use_spool = AsyncMock()
+    client.get_spool = AsyncMock(side_effect=lambda spool_id: {"id": spool_id, "price": 20.0, "initial_weight": 1000})
+    return client
+
+
+def _splitter(log, end_layer=100, completed=True):
+    return spool_swaps.Splitter(log, spool_swaps.SPOOLMAN, LAYER_USAGE, end_layer, completed=completed)
+
+
+class TestSpoolman:
+    @pytest.mark.asyncio
+    async def test_the_print_is_split_between_the_two_spools(self):
+        from backend.app.services.spoolman_tracking import _PrintCost, _report_spool_usage_for_slots
+
+        client = _client()
+        cost = _PrintCost()
+        with patch(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment", AsyncMock(return_value=9)
+        ):
+            updated = await _report_spool_usage_for_slots(
+                client,
+                [(1, 50.0)],
+                {254: {"tray_type": "PLA"}},
+                [254],
+                "Archive 7",
+                "SERIAL",
+                printer_id=1,
+                cost_out=cost,
+                swaps=_splitter(_log((9, 60), backend=spool_swaps.SPOOLMAN)),
+            )
+
+        assert updated == 2
+        assert [c.args for c in client.use_spool.await_args_list] == [(4, 30.0), (9, 20.0)]
+        assert cost.priced_grams == pytest.approx(50.0)
+
+    @pytest.mark.asyncio
+    async def test_without_a_swap_nothing_changes(self):
+        from backend.app.services.spoolman_tracking import _report_spool_usage_for_slots
+
+        client = _client()
+        with patch(
+            "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment", AsyncMock(return_value=9)
+        ):
+            await _report_spool_usage_for_slots(
+                client,
+                [(1, 50.0)],
+                {254: {"tray_type": "PLA"}},
+                [254],
+                "Archive 7",
+                "SERIAL",
+                printer_id=1,
+                swaps=_splitter(_log(backend=spool_swaps.SPOOLMAN)),
+            )
+
+        assert [c.args for c in client.use_spool.await_args_list] == [(9, 50.0)]
+
+    def test_a_multi_colour_print_without_gcode_layers_splits_by_layer_count(self):
+        # report_usage reads the printer's layers for single-slot prints only.
+        from backend.app.services.spoolman_tracking import _swap_splitter
+
+        log = _log((9, 25), backend=spool_swaps.SPOOLMAN)
+        state = SimpleNamespace(total_layers=100, layer_num=100, spool_change_log=log)
+        with patch("backend.app.services.printer_manager.printer_manager.get_status", return_value=state):
+            swaps = _swap_splitter(1, None, None, 0, completed=True)
+
+        assert swaps.shares(*EXT, 1, 40.0) == pytest.approx([(4, 10.0), (9, 30.0)])
+
+    @pytest.mark.asyncio
+    async def test_a_tray_change_segment_is_split_too(self):
+        from backend.app.services.spoolman_tracking import _report_spool_usage_split_by_tray_changes
+
+        client = _client()
+
+        async def slot(printer_id, ams_id, tray_id):
+            return 9 if ams_id == 255 else 20
+
+        with patch("backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment", side_effect=slot):
+            await _report_spool_usage_split_by_tray_changes(
+                client,
+                [{"slot_id": 1, "used_g": 50.0}],
+                [(0, 0), (254, 40)],
+                {0: {"tray_type": "PLA"}, 254: {"tray_type": "PLA"}},
+                LAYER_USAGE,
+                {},
+                100,
+                100,
+                "Archive 7",
+                "SERIAL",
+                printer_id=1,
+                swaps=_splitter(_log((9, 60), backend=spool_swaps.SPOOLMAN)),
+            )
+
+        charged = [c.args for c in client.use_spool.await_args_list]
+        assert charged[0][0] == 20
+        assert charged[1:] == [(4, 10.0), (9, 20.0)]
+
+
+# --- Print start and restart --------------------------------------------------
+
+
+class TestPrintStartAndRestart:
+    def _state(self):
+        state = MagicMock()
+        state.raw_data = {"ams": [], "vt_tray": [{"id": "254", "tray_type": "PLA"}]}
+        state.tray_now = 254
+        state.tray_change_log = [(254, 0)]
+        state.spool_change_log = []
+        return state
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("spoolman", [False, True])
+    async def test_print_start_logs_the_spool_each_tray_starts_with(self, db_session, printer_factory, spoolman):
+        from backend.app.models.spool_assignment import SpoolAssignment
+        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+        from backend.app.services.usage_tracker import _active_sessions, on_print_start
+
+        printer = await printer_factory()
+        spool = Spool(material="PLA", label_weight=1000, weight_used=0)
+        db_session.add(spool)
+        await db_session.commit()
+        db_session.add(SpoolAssignment(spool_id=spool.id, printer_id=printer.id, ams_id=255, tray_id=0))
+        db_session.add(SpoolmanSlotAssignment(printer_id=printer.id, ams_id=255, tray_id=0, spoolman_spool_id=77))
+        await db_session.commit()
+        state = self._state()
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = state
+
+        await on_print_start(
+            printer.id, {"subtask_name": "swap"}, printer_manager, db=db_session, spoolman_owns_usage=spoolman
+        )
+        _active_sessions.pop(printer.id, None)
+
+        backend, spool_id = (spool_swaps.SPOOLMAN, 77) if spoolman else (INV, spool.id)
+        assert state.spool_change_log == [spool_swaps.entry(backend, 255, 0, spool_id, 0, start=True)]
+        row = await db_session.get(ActivePrintSession, printer.id)
+        assert row.spool_change_log == state.spool_change_log
+
+    @pytest.mark.asyncio
+    async def test_a_restart_puts_the_log_back(self, db_session, printer_factory):
+        from backend.app.main import _restore_usage_tracking_session
+
+        printer = await printer_factory()
+        before_restart = _log((9, 40))
+        db_session.add(
+            ActivePrintSession(
+                printer_id=printer.id,
+                print_name="swap",
+                started_at=datetime.now(),
+                spool_change_log=before_restart,
+            )
+        )
+        await db_session.commit()
+        since = spool_swaps.entry(INV, *EXT, 12, 70)
+        state = SimpleNamespace(
+            subtask_name="swap",
+            tray_change_log=[],
+            tray_now=254,
+            layer_num=75,
+            last_loaded_tray=254,
+            print_ams_mapping=None,
+            spool_change_log=[since],
+        )
+
+        await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
+
+        assert state.spool_change_log == before_restart + [since]
+        assert spool_swaps.segments(state.spool_change_log, INV, *EXT) == [(4, 0), (9, 40), (12, 70)]
+
+
+class TestBuiltInRemainDelta:
+    """Without a 3MF the built-in tracker charges a tray's remain% drop. Across a
+    swap the start and end readings belong to two different spools."""
+
+    @pytest.fixture(autouse=True)
+    def _clean(self):
+        from backend.app.services.usage_tracker import _active_sessions
+
+        _active_sessions.clear()
+        with patch("backend.app.api.routes.settings.get_setting", new_callable=AsyncMock, return_value=None):
+            yield
+        _active_sessions.clear()
+
+    async def _complete(self, spool_change_log):
+        from backend.app.services.usage_tracker import PrintSession, _active_sessions, on_print_complete
+
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="Test",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 80},
+        )
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]},
+            tray_now=0,
+            last_loaded_tray=-1,
+            spool_change_log=spool_change_log,
+        )
+        assignment = MagicMock(spool_id=2, printer_id=1, ams_id=0, tray_id=0)
+        spool = _spool(2)
+        responses = iter([None, None, assignment, spool])
+
+        async def execute(*args, **kwargs):
+            result = MagicMock()
+            value = next(responses, 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 = AsyncMock()
+        db.execute = execute
+        db.add = MagicMock()
+        return await on_print_complete(
+            printer_id=1, data={"status": "completed"}, printer_manager=printer_manager, db=db, archive_id=None
+        )
+
+    @pytest.mark.asyncio
+    async def test_a_swapped_tray_is_not_charged_its_remain_drop(self):
+        assert await self._complete(_log((2, 40), start=5, key=(0, 0))) == []
+
+    @pytest.mark.asyncio
+    async def test_an_assignment_corrected_before_the_first_layer_is_still_charged(self):
+        results = await self._complete(_log((2, 0), start=5, key=(0, 0)))
+        assert [(r["spool_id"], r["weight_used"]) for r in results] == [(2, 100.0)]
+
+    @pytest.mark.asyncio
+    async def test_a_tray_that_kept_its_spool_still_is(self):
+        results = await self._complete(_log(start=2, key=(0, 0)))
+        assert [(r["spool_id"], r["weight_used"]) for r in results] == [(2, 100.0)]

backend/tests/unit/test_spool_swap_writers_3325.py

@@ -0,0 +1,54 @@
+"""Every spool assignment writer logs a mid-print swap (#3325).
+
+A spool assigned to a tray while a print runs splits the print between it and
+the spool the tray started with, but only if the writer told
+``spool_swaps.record_assignment`` about it. A writer added later without that
+call would put the whole print on the new spool again, so each function that
+writes an assignment row must make the call, or be listed below with why not.
+"""
+
+import ast
+import re
+from pathlib import Path
+
+BACKEND_DIR = Path(__file__).parent.parent.parent / "app"
+
+WRITES_ASSIGNMENT = re.compile(
+    r"INSERT INTO spoolman_slot_assignments|INSERT INTO spool_assignment|SpoolAssignment\(|"
+    r"update\(SpoolmanSlotAssignment\)|update\(SpoolAssignment\)"
+)
+
+EXEMPT = {
+    ("core/database.py", "_migrate_widen_spoolman_slot_ams_id_range"): "startup migration copying existing rows",
+}
+
+
+def _writers():
+    for path in BACKEND_DIR.rglob("*.py"):
+        source = path.read_text()
+        if not WRITES_ASSIGNMENT.search(source):
+            continue
+        for node in ast.walk(ast.parse(source)):
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                continue
+            body = ast.get_source_segment(source, node) or ""
+            nested = [
+                n for n in ast.walk(node) if n is not node and isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
+            ]
+            if WRITES_ASSIGNMENT.search(body) and not any(
+                WRITES_ASSIGNMENT.search(ast.get_source_segment(source, n) or "") for n in nested
+            ):
+                yield str(path.relative_to(BACKEND_DIR)), node.name, "record_assignment" in body
+
+
+def test_every_assignment_writer_logs_a_swap():
+    writers = list(_writers())
+    assert len(writers) >= 7  # the scan finds the writers at all
+
+    missing = sorted(f"{path}:{name}" for path, name, records in writers if not records and (path, name) not in EXEMPT)
+    assert missing == [], "call spool_swaps.record_assignment before committing:\n" + "\n".join(missing)
+
+
+def test_exemptions_still_exist():
+    found = {(path, name) for path, name, _ in _writers()}
+    assert set(EXEMPT) <= found