fix(spoolman): split mid-print usage across AMS backup switch (#1793)

usage_tracker's tray-switch split has never had a Spoolman peer.
An AMS same-material runout switch mid-print charged the whole slot
to the origin spool via the (via tag) path and double-credited the
backup via remain-delta — origin exceeded initial_weight.

Extract the segment-math into utils/tray_split.compute_tray_split_grams
and call it from both writers so the two inventory backends attribute
mid-print switches identically. spoolman_tracking gains
_report_spool_usage_split_by_tray_changes; the Path 2 remain-delta
fallback now skips trays the split path covered, killing the
double-count.

backend/app/services/spoolman_tracking.py

@@ -569,6 +569,135 @@ async def _report_spool_usage_for_slots(
     return spools_updated
 
 
+async def _report_spool_usage_split_by_tray_changes(
+    client,
+    filament_usage: list[dict],
+    tray_changes: list[tuple[int, int]],
+    ams_trays: dict[int, dict],
+    layer_usage: dict[int, dict[int, float]] | None,
+    filament_properties: dict | None,
+    total_layers: int,
+    last_layer_num: int,
+    method_label: str,
+    printer_serial: str,
+    printer_id: int,
+    slot_colors_out: dict[int, str] | None = None,
+) -> tuple[int, set[int]]:
+    """Split each slot's grams across ``tray_changes`` and charge per-segment.
+
+    Mirrors ``usage_tracker`` Path 1's tray-switch branch so Spoolman and
+    the internal Spool inventory attribute mid-print AMS-backup switches
+    identically (#1793 — reporter's origin spool was over-charged the
+    whole print because this path didn't exist). ``compute_tray_split_grams``
+    holds the shared segment-math; this function wraps the per-segment
+    spool resolution + ``use_spool`` sink for the Spoolman side.
+
+    Returns ``(spools_updated, handled_global_tray_ids)`` — the caller
+    passes ``handled_global_tray_ids`` into the remain-delta fallback so
+    a tray attributed here is not double-charged there.
+    """
+    from backend.app.utils.tray_split import compute_tray_split_grams
+
+    spools_updated = 0
+    handled_global_tray_ids: set[int] = set()
+
+    for usage in filament_usage:
+        slot_id = usage.get("slot_id", 0)
+        total_weight = usage.get("used_g", 0)
+        if total_weight <= 0 or slot_id <= 0:
+            continue
+
+        props = (filament_properties or {}).get(str(slot_id)) or (filament_properties or {}).get(slot_id) or {}
+        segments = compute_tray_split_grams(
+            tray_changes=tray_changes,
+            total_weight=float(total_weight),
+            slot_id=slot_id,
+            layer_usage=layer_usage,
+            density=float(props.get("density", 1.24)),
+            diameter=float(props.get("diameter", 1.75)),
+            total_layers=total_layers,
+            last_layer_num=last_layer_num,
+        )
+
+        for seg_idx, tray_global, segment_grams in segments:
+            if segment_grams <= 0:
+                continue
+
+            # Mark this tray as handled BEFORE the resolution attempt so
+            # remain-delta doesn't double-charge it, even if we fail to
+            # find a spool below. Matches usage_tracker behaviour: the
+            # tray was physically fed from during this print, whether or
+            # not Spoolman happens to have a matching row.
+            handled_global_tray_ids.add(tray_global)
+
+            tray_info = ams_trays.get(tray_global) or {}
+            spool_id_to_use: int | None = None
+            resolution_path = ""
+            spool_color_hex: str | None = None
+
+            spool_tag = _resolve_spool_tag(tray_info, printer_serial, tray_global) if tray_info else ""
+            if spool_tag:
+                spool = await client.find_spool_by_tag(spool_tag)
+                if spool:
+                    spool_id_to_use = spool["id"]
+                    resolution_path = "tag"
+                    spool_color_hex = (spool.get("filament") or {}).get("color_hex")
+
+            if spool_id_to_use is None:
+                seg_ams_id, seg_tray_id = _global_tray_id_to_ams_slot(tray_global)
+                spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, seg_ams_id, seg_tray_id)
+                if spool_id_to_use is not None:
+                    resolution_path = "slot-assignment"
+
+            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",
+                    slot_id,
+                    seg_idx,
+                    tray_global,
+                    segment_grams,
+                )
+                continue
+
+            # Colour rewrite (#1494) — first segment for a slot wins. The
+            # UI displays a single colour per slot, so later segments on the
+            # same slot don't overwrite (a backup swap can be a different
+            # colour but the archive card stays consistent with the origin).
+            if slot_colors_out is not None and slot_id not in slot_colors_out:
+                if spool_color_hex is None:
+                    try:
+                        full_spool = await client.get_spool(spool_id_to_use)
+                        spool_color_hex = (full_spool.get("filament") or {}).get("color_hex")
+                    except Exception as exc:  # noqa: BLE001 — colour is non-critical
+                        logger.debug("[SPOOLMAN] Split slot %s: could not fetch spool colour: %s", slot_id, exc)
+                if spool_color_hex:
+                    slot_colors_out[slot_id] = spool_color_hex
+
+            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
+            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,
+                )
+
+    return spools_updated, handled_global_tray_ids
+
+
 async def _report_partial_usage(
     printer_id: int,
     tracking,
@@ -789,6 +918,13 @@ async def report_usage(printer_id: int, archive_id: int):
         ams_trays = {int(k): v for k, v in (tracking.ams_trays or {}).items()}
         slot_to_tray = tracking.slot_to_tray
         tray_remain_start = tracking.tray_remain_start or {}
+        # ``layer_usage`` and ``filament_properties`` were added later than
+        # the base tracking fields; use ``getattr`` so tests that stub
+        # ``tracking`` as a lightweight SimpleNamespace stay valid, and
+        # historic ORM rows loaded without these columns can't AttributeError
+        # on read.
+        layer_usage_raw = getattr(tracking, "layer_usage", None) or {}
+        filament_properties = getattr(tracking, "filament_properties", None) or {}
         printer_serial = await _get_printer_serial(printer_id)
 
         # Delete tracking row (we're done with it)
@@ -809,29 +945,104 @@ async def report_usage(printer_id: int, archive_id: int):
             logger.warning("[SPOOLMAN] Not reachable for usage reporting")
             return
 
+        # Consult the live printer state for the tray-change log written by
+        # ``bambu_mqtt.py`` on every mid-print ``tray_now`` change (#957).
+        # When there's more than one entry, the print traversed >1 AMS tray
+        # and the split path attributes each segment to the tray that was
+        # loaded at the time — matches the internal Spool inventory writer
+        # in ``usage_tracker.py``. Without this, an AMS-backup runout switch
+        # charges the whole slot to the origin spool and pushes it past
+        # ``initial_weight`` (#1793).
+        #
+        # Split only for SINGLE-slot prints — same gate as
+        # ``usage_tracker.py:1002``. Multi-slot (multi-colour) prints
+        # naturally cycle trays for every colour change, so splitting each
+        # slot's grams across ALL tray_change_log entries would attribute
+        # slot 1's grams to the segments where slot 2's tray was loaded and
+        # vice versa. Multi-slot prints fall through to the existing
+        # single-tray path (which uses the stable ``slot_to_tray`` mapping).
+        nonzero_slots = [u for u in filament_usage if u.get("used_g", 0) > 0]
+        tray_changes: list[tuple[int, int]] = []
+        _state = None
+        if len(nonzero_slots) == 1:
+            from backend.app.services.printer_manager import printer_manager as _pm
+
+            _state = _pm.get_status(printer_id)
+            if _state is not None:
+                tray_changes = list(getattr(_state, "tray_change_log", []) or [])
+        _total_layers = int(getattr(_state, "total_layers", 0) or 0) if _state else 0
+        _current_layer = int(getattr(_state, "layer_num", 0) or 0) if _state else 0
+        # For the linear-fallback denominator when total_layers is 0 (P1S
+        # firmware resets it at print end). At completion the current layer
+        # is the print's last valid layer.
+        _layer_denom_hint = _total_layers or _current_layer
+
         slot_colors: dict[int, str] = {}
         handled_global_tray_ids: set[int] = set()
         spools_updated = 0
 
         # --- Path 1: 3MF per-slot estimates -----------------------------
         if filament_usage:
-            logger.info("[SPOOLMAN] Reporting per-filament usage for archive %s", archive_id)
-            usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage]
-            spools_updated = await _report_spool_usage_for_slots(
-                client,
-                usage_items,
-                ams_trays,
-                slot_to_tray,
-                f"Archive {archive_id}",
-                printer_serial,
-                printer_id=printer_id,
-                slot_colors_out=slot_colors,
-            )
-            # Track which physical slots the 3MF path already covered so
-            # Path 2 doesn't double-charge them.
-            for u in filament_usage:
-                slot_id = u.get("slot_id", 0)
-                handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays))
+            if len(tray_changes) > 1:
+                # Tray-split path — attribute per-segment to the tray that
+                # was loaded at that time.
+                logger.info(
+                    "[SPOOLMAN] Reporting per-filament usage for archive %s with tray-split "
+                    "(tray_change_log=%s, denom_layers=%d)",
+                    archive_id,
+                    tray_changes,
+                    _layer_denom_hint,
+                )
+                # ``tracking.layer_usage`` was serialized to JSON so int keys
+                # come back as strings. Restore them for the split math.
+                layer_usage = None
+                if layer_usage_raw:
+                    try:
+                        layer_usage = {
+                            int(layer): {int(fid): mm for fid, mm in filaments.items()}
+                            for layer, filaments in layer_usage_raw.items()
+                        }
+                    except (TypeError, ValueError, AttributeError):
+                        # AttributeError catches ``inner.items()`` when the
+                        # inner value isn't dict-shaped (corrupt JSON row).
+                        # Missing gcode falls through to the linear-ratio
+                        # branch inside ``compute_tray_split_grams`` — still
+                        # gives a correct split, just less precise.
+                        layer_usage = None
+                split_updated, split_handled = await _report_spool_usage_split_by_tray_changes(
+                    client,
+                    filament_usage,
+                    tray_changes,
+                    ams_trays,
+                    layer_usage,
+                    filament_properties,
+                    _total_layers,
+                    _layer_denom_hint,
+                    f"Archive {archive_id}",
+                    printer_serial,
+                    printer_id=printer_id,
+                    slot_colors_out=slot_colors,
+                )
+                spools_updated += split_updated
+                handled_global_tray_ids |= split_handled
+            else:
+                logger.info("[SPOOLMAN] Reporting per-filament usage for archive %s", archive_id)
+                usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage]
+                spools_updated = await _report_spool_usage_for_slots(
+                    client,
+                    usage_items,
+                    ams_trays,
+                    slot_to_tray,
+                    f"Archive {archive_id}",
+                    printer_serial,
+                    printer_id=printer_id,
+                    slot_colors_out=slot_colors,
+                )
+                # Track which physical slots the 3MF path already covered so
+                # Path 2 doesn't double-charge them.
+                for u in filament_usage:
+                    slot_id = u.get("slot_id", 0)
+                    handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays))
 
         # --- Path 2: AMS remain%-delta for slots 3MF didn't cover -------
         # Triggered for no-3MF "Untitled" prints (#1820) AND for partial

backend/app/services/usage_tracker.py

@@ -1083,6 +1083,10 @@ async def _track_from_3mf(
             continue
 
         # --- Mid-print tray switch: split weight across trays ---
+        # Split math is shared with the Spoolman writer via
+        # ``utils.tray_split.compute_tray_split_grams`` (#1793) — both
+        # inventory backends must attribute segments identically or a
+        # user running dual-mode sees divergent totals.
         if len(tray_changes) > 1:
             # Compute total weight for this slot (same logic as normal path)
             if layer_grams and slot_id in layer_grams:
@@ -1100,8 +1104,6 @@ async def _track_from_3mf(
                 from backend.app.utils.threemf_tools import (
                     extract_filament_properties_from_3mf,
                     extract_layer_filament_usage_from_3mf,
-                    get_cumulative_usage_at_layer,
-                    mm_to_grams,
                 )
 
                 split_layer_usage = extract_layer_filament_usage_from_3mf(file_path)
@@ -1110,46 +1112,20 @@ async def _track_from_3mf(
             except Exception:
                 pass  # Fall back to linear splitting
 
-            density = split_props.get("density", 1.24)
-            diameter = split_props.get("diameter", 1.75)
-            filament_id = slot_id - 1  # 0-based for gcode
-
-            sum_previous = 0.0
-            for seg_idx, (tray_global, seg_start_layer) in enumerate(tray_changes):
-                is_last = seg_idx + 1 >= len(tray_changes)
+            from backend.app.utils.tray_split import compute_tray_split_grams
+
+            segments = compute_tray_split_grams(
+                tray_changes=tray_changes,
+                total_weight=total_weight,
+                slot_id=slot_id,
+                layer_usage=split_layer_usage,
+                density=split_props.get("density", 1.24),
+                diameter=split_props.get("diameter", 1.75),
+                total_layers=(state.total_layers if state else 0) or 0,
+                last_layer_num=last_layer_num,
+            )
 
-                if is_last:
-                    # Last segment: remainder to avoid rounding drift
-                    segment_grams = total_weight - sum_previous
-                elif split_layer_usage:
-                    seg_end_layer = tray_changes[seg_idx + 1][1]
-                    mm_at_start = get_cumulative_usage_at_layer(split_layer_usage, seg_start_layer).get(filament_id, 0)
-                    mm_at_end = get_cumulative_usage_at_layer(split_layer_usage, seg_end_layer).get(filament_id, 0)
-                    segment_grams = mm_to_grams(mm_at_end - mm_at_start, diameter, density)
-                else:
-                    # No per-layer data: linear fallback by layer ratio (#1771).
-                    # Cascade denominators because firmware on some models (P1S
-                    # observed) resets `total_layer_num` to 0 at print end —
-                    # `last_layer_num` is the print's last-valid layer captured
-                    # mid-print and survives that reset (same shape as the
-                    # `last_progress` fallback at line 1040). Equal-split is the
-                    # last-resort fence: still wrong, but bounded — never dumps
-                    # the entire print onto the last segment, which was the
-                    # original #1771 symptom for the reporter (P1S, AMS Backup
-                    # fed from spool 1 then spool 2, all 260 g credited to
-                    # spool 2 even though spool 1 had given up its 180 g).
-                    seg_end_layer = tray_changes[seg_idx + 1][1]
-                    denom = (state.total_layers if state else 0) or last_layer_num
-                    if denom > 0:
-                        segment_grams = total_weight * (seg_end_layer - seg_start_layer) / denom
-                    else:
-                        # No layer information available from any source —
-                        # spread evenly across segments. The last segment will
-                        # get the rounding remainder via the `is_last` branch
-                        # above on its own iteration.
-                        segment_grams = total_weight / len(tray_changes)
-
-                sum_previous += segment_grams
+            for seg_idx, tray_global, segment_grams in segments:
                 if segment_grams <= 0:
                     continue
 
@@ -1168,6 +1144,8 @@ async def _track_from_3mf(
                 if seg_key in handled_trays:
                     continue
 
+                seg_start_layer = tray_changes[seg_idx][1]
+                is_last = seg_idx + 1 >= len(tray_changes)
                 logger.info(
                     "[UsageTracker] 3MF split: segment %d tray=%d (AMS%d-T%d) layers %d-%s -> %.1fg",
                     seg_idx,

backend/app/utils/tray_split.py

@@ -0,0 +1,94 @@
+"""Weight-split math for prints that traversed >1 AMS tray mid-print.
+
+`state.tray_change_log` records `(global_tray_id, layer_num)` tuples every
+time `tray_now` changes during a print (see `bambu_mqtt.py:1861`). At
+completion, both the internal Spool inventory (`usage_tracker.py`) and the
+Spoolman writer (`spoolman_tracking.py`) need to split a slot's total
+weight across the segments those changes define — one call site per
+inventory backend, one identical splitting algorithm.
+
+The algorithm lives here so the two callers cannot drift: #1793 came from
+`spoolman_tracking` never carrying the split at all, while `usage_tracker`
+had shipped it since #957 and refined it in #1771. Sharing the helper is
+the structural fix; each caller wraps its own "resolve segment tray →
+charge N grams" side effect.
+"""
+
+from __future__ import annotations
+
+# Qualified-name access (``threemf_tools.mm_to_grams(...)`` rather than
+# ``from … import mm_to_grams``) so ``unittest.mock.patch`` on the
+# threemf_tools module lands in the helper too — the pre-refactor
+# ``usage_tracker`` call site imported at call-time inside a try block,
+# which had the same testability property.
+from backend.app.utils import threemf_tools
+
+
+def compute_tray_split_grams(
+    tray_changes: list[tuple[int, int]],
+    total_weight: float,
+    slot_id: int,
+    layer_usage: dict[int, dict[int, float]] | None,
+    density: float,
+    diameter: float,
+    total_layers: int,
+    last_layer_num: int,
+) -> list[tuple[int, int, float]]:
+    """Split ``total_weight`` for a single slice slot across tray segments.
+
+    ``tray_changes`` is the ordered list ``[(global_tray_id, seg_start_layer), ...]``
+    exactly as it appears in ``state.tray_change_log``. The last segment
+    runs to the end of the print; every other segment ends at the next
+    entry's ``seg_start_layer``.
+
+    Preference order for per-segment grams — matches ``usage_tracker`` so
+    both inventory backends split identically:
+
+    1. **G-code cumulative extrusion** (``layer_usage``, indexed by 0-based
+       filament id). Precise: uses the mm actually consumed between
+       ``seg_start_layer`` and the next segment's start, then converts via
+       Spoolman-authoritative ``density`` / ``diameter``.
+    2. **Linear layer-ratio** — ``total_weight * segment_layers / denom``,
+       with ``denom = total_layers or last_layer_num``. Firmware on P1S
+       (observed) resets ``total_layer_num`` to 0 at print end, so the
+       captured ``last_layer_num`` is the durable denominator (see
+       ``usage_tracker.py:1132``). #1771 addressed the pre-fix behaviour
+       of dumping everything onto the last segment.
+    3. **Equal-split** — when neither denominator is available (server
+       restart mid-print, missing state). Wrong but bounded — the last
+       segment absorbs any rounding drift via the ``is_last`` branch.
+
+    Returns ``[(seg_idx, global_tray_id, segment_grams)]``. Empty when
+    ``tray_changes`` is empty; the caller decides whether to fall through
+    to single-tray attribution (``len(tray_changes) <= 1``).
+    """
+    if not tray_changes:
+        return []
+
+    filament_id = slot_id - 1
+    n_segments = len(tray_changes)
+    denom = total_layers or last_layer_num
+    results: list[tuple[int, int, float]] = []
+    sum_previous = 0.0
+
+    for seg_idx, (tray_global, seg_start_layer) in enumerate(tray_changes):
+        is_last = seg_idx + 1 >= n_segments
+
+        if is_last:
+            segment_grams = total_weight - sum_previous
+        elif layer_usage:
+            seg_end_layer = tray_changes[seg_idx + 1][1]
+            mm_at_start = threemf_tools.get_cumulative_usage_at_layer(layer_usage, seg_start_layer).get(filament_id, 0)
+            mm_at_end = threemf_tools.get_cumulative_usage_at_layer(layer_usage, seg_end_layer).get(filament_id, 0)
+            segment_grams = threemf_tools.mm_to_grams(mm_at_end - mm_at_start, diameter, density)
+        else:
+            seg_end_layer = tray_changes[seg_idx + 1][1]
+            if denom > 0:
+                segment_grams = total_weight * (seg_end_layer - seg_start_layer) / denom
+            else:
+                segment_grams = total_weight / n_segments
+
+        sum_previous += segment_grams
+        results.append((seg_idx, tray_global, segment_grams))
+
+    return results

backend/tests/unit/services/test_spoolman_tray_split.py

@@ -0,0 +1,350 @@
+"""Spoolman-side mid-print tray-split accounting (#1793).
+
+Reporter (@ojimpo) shipped the OP shape:
+- H2S, AMS filament backup ON, two same-material spools loaded
+- Single-slot print (72.56g on slot 1)
+- Origin ran dry at layer 37, AMS auto-switched to backup, print finished
+- Pre-fix: whole 72.56g charged to origin (via tag path) + separate 30g to
+  backup (via remain-delta) — origin exceeded initial_weight, backup double-count
+
+The fix ports usage_tracker's split path to spoolman_tracking so both
+inventory backends attribute segments identically. These tests pin the
+OP's shape plus the Path 2 (remain-delta) skip guarantee so it can't
+double-charge tray IDs the split path already covered.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+class _AsyncCtx:
+    """async_session() shim — same shape as test_spoolman_no3mf_remain_fallback."""
+
+    def __init__(self, db):
+        self._db = db
+
+    async def __aenter__(self):
+        return self._db
+
+    async def __aexit__(self, *_):
+        return False
+
+
+def _make_db(tracking):
+    db = AsyncMock()
+    select_result = MagicMock()
+    select_result.scalar_one_or_none.return_value = tracking
+    db.execute = AsyncMock(return_value=select_result)
+    db.delete = AsyncMock()
+    db.commit = AsyncMock()
+    return db
+
+
+class TestReportUsageTraySplit:
+    """report_usage must consult state.tray_change_log and split per-segment."""
+
+    @pytest.mark.asyncio
+    async def test_op_sample_a_seamless_switch_splits_origin_to_backup(self):
+        """Sample A from the reporter, verbatim: 72.56g single-slot print,
+        AMS runout switch tray 0 → tray 1 at layer 37 of ~100 total.
+
+        No gcode layer_usage is provided → linear-by-layer-ratio fallback:
+        - seg 0 (tray 0, layers 0-37) = 72.56 * 37/100 = 26.85g → spool 8
+        - seg 1 (tray 1, layers 37-end) = 72.56 - 26.85 = 45.71g → spool 7
+        Path 2 (remain-delta) must NOT run against either tray — the split
+        path already covered them.
+        """
+        from backend.app.services.spoolman_tracking import report_usage
+
+        tracking = SimpleNamespace(
+            filament_usage=[{"slot_id": 1, "used_g": 72.56}],
+            ams_trays={
+                0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"},
+                1: {"tray_uuid": "BBBB", "tag_uid": "T2TAG", "tray_type": "PLA"},
+            },
+            slot_to_tray=[0],
+            tray_remain_start={
+                "0-0": {"remain": 3, "tray_uuid": "AAAA"},  # origin near-empty at start-of-completion snapshot
+                "0-1": {"remain": 73, "tray_uuid": "BBBB"},
+            },
+            layer_usage={},
+            filament_properties={},
+        )
+
+        db = _make_db(tracking)
+
+        client = AsyncMock()
+
+        async def _find_spool_by_tag(tag):
+            return (
+                {"id": 8, "filament": {"color_hex": "000000"}}
+                if tag == "AAAA"
+                else {"id": 7, "filament": {"color_hex": "000000"}}
+                if tag == "BBBB"
+                else None
+            )
+
+        client.find_spool_by_tag = AsyncMock(side_effect=_find_spool_by_tag)
+        client.use_spool = AsyncMock()
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            tray_change_log=[(0, 0), (1, 37)],
+            total_layers=100,
+            layer_num=100,
+            raw_data={
+                "ams": [
+                    {
+                        "id": 0,
+                        "tray": [
+                            {"id": 0, "tray_uuid": "AAAA", "remain": 0},
+                            {"id": 1, "tray_uuid": "BBBB", "remain": 70},
+                        ],
+                    }
+                ]
+            },
+        )
+
+        with (
+            patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+            patch(
+                "backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
+                AsyncMock(return_value=client),
+            ),
+            patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
+            patch(
+                "backend.app.services.spoolman_tracking._apply_spool_colors_to_archive",
+                AsyncMock(),
+            ),
+            patch("backend.app.services.printer_manager.printer_manager", printer_manager),
+        ):
+            await report_usage(printer_id=1, archive_id=143)
+
+        # Exactly two use_spool calls — one per segment. Origin (spool 8)
+        # gets the layers-0-37 slice, backup (spool 7) gets the remainder.
+        calls = client.use_spool.await_args_list
+        assert len(calls) == 2, f"expected 2 use_spool calls, got {len(calls)}: {calls}"
+
+        by_spool = {c.args[0]: c.args[1] for c in calls}
+        assert set(by_spool.keys()) == {8, 7}
+
+        # Sum must equal the OP's total — no phantom grams created or lost.
+        assert round(sum(by_spool.values()), 2) == 72.56
+
+        # Origin (spool 8) should carry roughly the layers-0-37 fraction.
+        # Linear: 72.56 * 37/100 = 26.85g. Allow small rounding wiggle.
+        assert 26.0 < by_spool[8] < 28.0
+        # Backup (spool 7) carries the remainder.
+        assert 44.0 < by_spool[7] < 46.6
+
+    @pytest.mark.asyncio
+    async def test_path_2_remain_delta_skips_tray_handled_by_split(self):
+        """After the split path attributes segments to tray 0 AND tray 1,
+        the Path 2 remain-delta iterator must skip BOTH — otherwise backup
+        would get charged twice (~30g double-count in the OP's Sample A).
+        """
+        from backend.app.services.spoolman_tracking import report_usage
+
+        tracking = SimpleNamespace(
+            filament_usage=[{"slot_id": 1, "used_g": 100.0}],
+            ams_trays={
+                0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"},
+                1: {"tray_uuid": "BBBB", "tag_uid": "T2TAG", "tray_type": "PLA"},
+            },
+            slot_to_tray=[0],
+            tray_remain_start={
+                "0-0": {"remain": 20, "tray_uuid": "AAAA"},
+                "0-1": {"remain": 80, "tray_uuid": "BBBB"},
+            },
+            layer_usage={},
+            filament_properties={},
+        )
+
+        db = _make_db(tracking)
+        client = AsyncMock()
+
+        async def _find_spool_by_tag(tag):
+            return {"id": 8, "filament": {}} if tag == "AAAA" else {"id": 7, "filament": {}}
+
+        client.find_spool_by_tag = AsyncMock(side_effect=_find_spool_by_tag)
+        client.use_spool = AsyncMock()
+        # If Path 2 ever runs, it needs a filament.weight to compute grams.
+        # Making it valid means a failure to guard = extra use_spool calls,
+        # not a silent skip. Combined with a truthy slot-assignment result
+        # below, this is what actually proves the double-count guard works.
+        client.get_spool = AsyncMock(return_value={"filament": {"weight": 1000.0}})
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            tray_change_log=[(0, 0), (1, 50)],
+            total_layers=100,
+            layer_num=100,
+            raw_data={
+                "ams": [
+                    {
+                        "id": 0,
+                        "tray": [
+                            {"id": 0, "tray_uuid": "AAAA", "remain": 0},
+                            {"id": 1, "tray_uuid": "BBBB", "remain": 60},
+                        ],
+                    }
+                ]
+            },
+        )
+
+        with (
+            patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+            patch(
+                "backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
+                AsyncMock(return_value=client),
+            ),
+            patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
+            patch(
+                # Path 2 uses this to resolve trays. Return valid IDs so
+                # the ONLY thing stopping Path 2 from double-charging is
+                # ``handled_global_tray_ids``. If the guard is broken,
+                # Path 2 would successfully call ``use_spool`` two more
+                # times and this test would fail with 4 calls, not 2.
+                "backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
+                AsyncMock(side_effect=lambda pid, ams, tray: 999 if (ams, tray) == (0, 0) else 888),
+            ),
+            patch("backend.app.services.printer_manager.printer_manager", printer_manager),
+        ):
+            await report_usage(printer_id=1, archive_id=200)
+
+        # EXACTLY 2 — one per segment; Path 2 must not add a third.
+        assert client.use_spool.await_count == 2, (
+            f"Path 2 leaked past the split — expected 2 use_spool calls, got "
+            f"{client.use_spool.await_count}: {client.use_spool.await_args_list}"
+        )
+
+    @pytest.mark.asyncio
+    async def test_multi_slot_print_does_not_activate_split_even_with_tray_changes(self):
+        """Multi-colour prints normally cycle trays every colour change, so
+        ``tray_change_log`` has many entries — but splitting each slot's
+        grams across all of them would attribute slot 1's usage to segments
+        where slot 2's tray was loaded (and vice versa).
+
+        Mirrors ``usage_tracker.py:1002``'s gate: split only when there's
+        exactly one nonzero slot. Multi-slot prints fall through to the
+        existing single-tray path with its stable ``slot_to_tray`` mapping.
+        """
+        from backend.app.services.spoolman_tracking import report_usage
+
+        # Two nonzero slots — regular multi-colour print
+        tracking = SimpleNamespace(
+            filament_usage=[
+                {"slot_id": 1, "used_g": 30.0},
+                {"slot_id": 2, "used_g": 20.0},
+            ],
+            ams_trays={
+                0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"},
+                1: {"tray_uuid": "BBBB", "tag_uid": "T2TAG", "tray_type": "PLA"},
+            },
+            slot_to_tray=[0, 1],
+            tray_remain_start={
+                "0-0": {"remain": 90, "tray_uuid": "AAAA"},
+                "0-1": {"remain": 80, "tray_uuid": "BBBB"},
+            },
+            layer_usage={},
+            filament_properties={},
+        )
+
+        db = _make_db(tracking)
+        client = AsyncMock()
+
+        async def _find_spool_by_tag(tag):
+            return {"id": 100, "filament": {}} if tag == "AAAA" else {"id": 200, "filament": {}}
+
+        client.find_spool_by_tag = AsyncMock(side_effect=_find_spool_by_tag)
+        client.use_spool = AsyncMock()
+
+        printer_manager = MagicMock()
+        # Multi-colour print naturally cycles between trays many times.
+        # If we don't gate on single-slot, my split would attribute slot 1's
+        # grams to every segment — including segments where tray 1 was loaded.
+        printer_manager.get_status.return_value = SimpleNamespace(
+            tray_change_log=[(0, 0), (1, 10), (0, 20), (1, 30), (0, 40)],
+            total_layers=50,
+            layer_num=50,
+            raw_data={
+                "ams": [
+                    {
+                        "id": 0,
+                        "tray": [
+                            {"id": 0, "tray_uuid": "AAAA", "remain": 87},
+                            {"id": 1, "tray_uuid": "BBBB", "remain": 78},
+                        ],
+                    }
+                ]
+            },
+        )
+
+        with (
+            patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+            patch(
+                "backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
+                AsyncMock(return_value=client),
+            ),
+            patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
+            patch("backend.app.services.printer_manager.printer_manager", printer_manager),
+        ):
+            await report_usage(printer_id=1, archive_id=42)
+
+        # Split path must NOT engage. The single-tray path charges each
+        # slot to its stable slot_to_tray mapping: slot 1 → tray 0 → spool
+        # 100 (30g), slot 2 → tray 1 → spool 200 (20g). Two calls, exact
+        # weights from the 3MF (not split).
+        assert client.use_spool.await_count == 2
+        by_spool = {c.args[0]: c.args[1] for c in client.use_spool.await_args_list}
+        assert by_spool == {100: 30.0, 200: 20.0}
+
+    @pytest.mark.asyncio
+    async def test_single_tray_change_entry_uses_normal_path(self):
+        """Only ONE entry in tray_change_log (start-of-print seed, no
+        switch) must fall through to the existing single-tray charging
+        path — not accidentally split when there's nothing to split.
+        """
+        from backend.app.services.spoolman_tracking import report_usage
+
+        tracking = SimpleNamespace(
+            filament_usage=[{"slot_id": 1, "used_g": 50.0}],
+            ams_trays={0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"}},
+            slot_to_tray=[0],
+            tray_remain_start={"0-0": {"remain": 80, "tray_uuid": "AAAA"}},
+            layer_usage={},
+            filament_properties={},
+        )
+
+        db = _make_db(tracking)
+        client = AsyncMock()
+        client.find_spool_by_tag = AsyncMock(return_value={"id": 8, "filament": {}})
+        client.use_spool = AsyncMock()
+
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            tray_change_log=[(0, 0)],  # just the start-of-print seed
+            total_layers=100,
+            layer_num=100,
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "remain": 75}]}]},
+        )
+
+        with (
+            patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
+            patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+            patch(
+                "backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
+                AsyncMock(return_value=client),
+            ),
+            patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
+            patch("backend.app.services.printer_manager.printer_manager", printer_manager),
+        ):
+            await report_usage(printer_id=1, archive_id=42)
+
+        # Single-tray path: exactly one use_spool call, all 50g to spool 8.
+        client.use_spool.assert_awaited_once_with(8, 50.0)

backend/tests/unit/utils/test_tray_split.py

@@ -0,0 +1,179 @@
+"""Pure-logic tests for the mid-print tray-split math (#1793).
+
+The helper lives in ``backend/app/utils/tray_split.py`` and is exercised
+by both inventory backends (``usage_tracker`` and ``spoolman_tracking``).
+These tests pin the algorithm so a change in one caller can't silently
+break the other — cross-inventory parity is a HARD RULE for this project.
+"""
+
+from __future__ import annotations
+
+from backend.app.utils.tray_split import compute_tray_split_grams
+
+
+class TestComputeTraySplitGrams:
+    """Segment-attribution algorithm — gcode preferred, linear fallback, equal split."""
+
+    def test_empty_tray_changes_returns_empty(self):
+        assert (
+            compute_tray_split_grams(
+                tray_changes=[],
+                total_weight=100.0,
+                slot_id=1,
+                layer_usage=None,
+                density=1.24,
+                diameter=1.75,
+                total_layers=200,
+                last_layer_num=200,
+            )
+            == []
+        )
+
+    def test_single_segment_charges_everything_to_that_tray(self):
+        segments = compute_tray_split_grams(
+            tray_changes=[(0, 0)],
+            total_weight=72.56,
+            slot_id=1,
+            layer_usage=None,
+            density=1.24,
+            diameter=1.75,
+            total_layers=100,
+            last_layer_num=100,
+        )
+        assert segments == [(0, 0, 72.56)]
+
+    def test_two_segments_linear_split_by_layer_ratio(self):
+        # Runout at layer 37 of 100 total; no gcode available → linear.
+        # Segment 0 (tray 0, layers 0-37) = 100 * 37/100 = 37g
+        # Segment 1 (tray 1, layers 37-end) = 100 - 37 = 63g (remainder)
+        segments = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 37)],
+            total_weight=100.0,
+            slot_id=1,
+            layer_usage=None,
+            density=1.24,
+            diameter=1.75,
+            total_layers=100,
+            last_layer_num=100,
+        )
+        assert segments == [(0, 0, 37.0), (1, 1, 63.0)]
+
+    def test_two_segments_gcode_preferred_over_linear(self):
+        # layer_usage stores mm of filament extruded per (layer, filament_id).
+        # Values are cumulative-per-key inside get_cumulative_usage_at_layer.
+        # 20 layers, filament_id=0 (slot_id=1 → filament_id 0):
+        #   layer 10 → 100mm cumulative
+        #   layer 20 → 300mm cumulative
+        # tray change at layer 10 → seg 0 spans layers 0-10 (mm 0 → 100),
+        #                            seg 1 spans layers 10-end.
+        # mm_to_grams(100, 1.75, 1.24) ≈ 0.298g; last segment absorbs the rest.
+        layer_usage = {
+            5: {0: 50.0},
+            10: {0: 100.0},
+            15: {0: 200.0},
+            20: {0: 300.0},
+        }
+        segments = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 10)],
+            total_weight=1.0,  # sentinel — we assert the seg1 remainder
+            slot_id=1,
+            layer_usage=layer_usage,
+            density=1.24,
+            diameter=1.75,
+            total_layers=20,
+            last_layer_num=20,
+        )
+        # Seg 0 charged from gcode delta (mm 0 → 100).
+        # Seg 1 gets total_weight - seg0 as remainder.
+        assert segments[0][0] == 0
+        assert segments[0][1] == 0  # tray 0
+        assert segments[0][2] > 0  # non-zero gcode contribution
+        assert segments[1][0] == 1
+        assert segments[1][1] == 1  # tray 1
+        # Sum equals the input total by construction (last segment absorbs).
+        assert round(segments[0][2] + segments[1][2], 6) == 1.0
+
+    def test_three_segments_last_absorbs_rounding_drift(self):
+        # 100g over three segments at layers 30 and 60 of 90; linear fallback.
+        # Seg 0: 100 * 30/90 = 33.3333...
+        # Seg 1: 100 * 30/90 = 33.3333...
+        # Seg 2: remainder = 100 - 66.6666... = 33.3333... — exact by construction
+        segments = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 30), (2, 60)],
+            total_weight=100.0,
+            slot_id=1,
+            layer_usage=None,
+            density=1.24,
+            diameter=1.75,
+            total_layers=90,
+            last_layer_num=90,
+        )
+        assert len(segments) == 3
+        assert round(sum(g for _, _, g in segments), 6) == 100.0
+        assert segments[0][1] == 0
+        assert segments[1][1] == 1
+        assert segments[2][1] == 2
+
+    def test_no_layer_info_at_all_falls_to_equal_split(self):
+        # Denominator 0 → last-resort equal-split; last segment absorbs remainder.
+        segments = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 50)],
+            total_weight=90.0,
+            slot_id=1,
+            layer_usage=None,
+            density=1.24,
+            diameter=1.75,
+            total_layers=0,
+            last_layer_num=0,
+        )
+        # 90g / 2 = 45g each; sum still 90 by remainder mechanic.
+        assert segments == [(0, 0, 45.0), (1, 1, 45.0)]
+
+    def test_last_layer_num_used_when_total_layers_zero(self):
+        # P1S firmware-reset scenario: total_layers=0 at completion, but the
+        # captured last_layer_num survives. Should give the same linear split
+        # as if total_layers had held its value (#1771 cascade).
+        segments_captured = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 30)],
+            total_weight=100.0,
+            slot_id=1,
+            layer_usage=None,
+            density=1.24,
+            diameter=1.75,
+            total_layers=0,
+            last_layer_num=100,
+        )
+        segments_normal = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 30)],
+            total_weight=100.0,
+            slot_id=1,
+            layer_usage=None,
+            density=1.24,
+            diameter=1.75,
+            total_layers=100,
+            last_layer_num=100,
+        )
+        assert segments_captured == segments_normal
+
+    def test_slot_id_maps_to_zero_based_filament_id_in_gcode(self):
+        # slot_id 2 → filament_id 1 in layer_usage. If we mistakenly used
+        # slot_id as-is, we'd read filament_id 2 which is absent → 0mm delta
+        # → seg 0 gets 0, seg 1 (remainder) gets the whole total. Guard
+        # against that regression.
+        layer_usage = {
+            5: {0: 0.0, 1: 40.0},
+            10: {0: 0.0, 1: 80.0},
+            20: {0: 0.0, 1: 160.0},
+        }
+        segments = compute_tray_split_grams(
+            tray_changes=[(0, 0), (1, 10)],
+            total_weight=1.0,
+            slot_id=2,
+            layer_usage=layer_usage,
+            density=1.24,
+            diameter=1.75,
+            total_layers=20,
+            last_layer_num=20,
+        )
+        # Seg 0 gcode delta on filament_id=1 is non-zero → not 0g.
+        assert segments[0][2] > 0