Fix filament not deducted from backup spool after runout auto-switch (#459)

When a spool runs empty mid-print and the AMS auto-switches to a backup,
the entire print weight was attributed to the original spool. Track tray
changes during the print via tray_change_log on PrinterState, then split
the 3MF weight across trays using per-layer gcode segment boundaries at
print completion. Falls back to linear layer-ratio splitting when gcode
data is unavailable.

CHANGELOG.md

@@ -7,7 +7,7 @@ All notable changes to Bambuddy will be documented in this file.
 ### Fixed
 - **Non-Actionable HMS Errors Triggering Notifications** ([#470](https://github.com/maziggy/bambuddy/issues/470)) — Infrastructure and auth-related HMS error codes (like `0500_0007` "MQTT command verification failed") were triggering printer error notifications even though they don't indicate actual print problems. For example, a device with incorrect bind settings sending unauthorized MQTT commands caused repeated false-alarm nozzle/extruder error notifications with camera snapshots of perfectly fine prints. Now suppresses notifications for known non-actionable error codes: `0500_0007` (MQTT auth failure), `0500_4001` (Bambu Cloud connection failure), and `0500_400E` (print cancelled by user).
 - **Support Bundle Leaking Personal Data** ([#473](https://github.com/maziggy/bambuddy/issues/473)) — The support bundle's log sanitizer only used regex patterns, which can't detect arbitrary user-chosen strings like printer names and usernames. Now queries the database for known sensitive values (printer names, serial numbers, auth usernames, Bambu Cloud email) and does exact-string replacement before the regex pass. Serial number regex no longer leaks the first 3 characters (was using a capture group for partial redaction). Tasmota smart plug credentials embedded in URLs (`http://user:pass@host`) were logged verbatim by httpx; now uses httpx's `auth` parameter for HTTP Basic auth so credentials never appear in the URL. Added `username` and `path` to the settings key filter to redact `smtp_username` and `slicer_binary_path` from the support info JSON. A URL credentials regex provides defense-in-depth for any remaining `user:pass@` patterns in logs. IP addresses are no longer redacted from the bundle as they are needed for connectivity debugging. Updated the frontend privacy disclaimer and wiki documentation to reflect the new behavior.
-- **Spool Usage Lost When Spool Runs Empty Mid-Print** ([#459](https://github.com/maziggy/bambuddy/issues/459)) — When a spool ran empty during a print and the AMS auto-switched to a backup spool, the `on_ams_change` handler eagerly deleted the empty spool's `SpoolAssignment` record (fingerprint mismatch). When `on_print_complete` later ran, it queried `SpoolAssignment` live from the database, found nothing, and silently dropped usage. Now snapshots all spool assignments at print start into the `PrintSession`, so usage is correctly attributed at completion regardless of mid-print AMS changes.
+- **Spool Usage Lost When Spool Runs Empty Mid-Print** ([#459](https://github.com/maziggy/bambuddy/issues/459)) — When a spool ran empty during a print and the AMS auto-switched to a backup spool, two problems caused incorrect tracking. First, the `on_ams_change` handler eagerly deleted the empty spool's `SpoolAssignment` record (fingerprint mismatch), so `on_print_complete` found nothing and silently dropped usage — fixed by snapshotting all spool assignments at print start into the `PrintSession`. Second, even with the snapshot fix, the entire print's filament weight was attributed to the original spool (100%/0% split) because `_track_from_3mf()` only knew about the tray loaded at print start. Now tracks tray changes during the print via `tray_change_log` on `PrinterState`, recording each tray switch with its layer number. At print completion, the usage tracker splits the 3MF weight across trays using per-layer gcode data for precise segment boundaries, with a linear layer-ratio fallback when gcode data isn't available. The last segment always receives the remainder to prevent rounding drift.
 - **K-Profile Response Race Condition Crash** ([#462](https://github.com/maziggy/bambuddy/issues/462)) — An unsolicited or late K-profile MQTT response could crash the MQTT handler with `AttributeError: 'NoneType' object has no attribute 'set'`. The MQTT callback thread checked `self._pending_kprofile_response` (not None) at line 2698, but between that check and the `.set()` call, the asyncio thread's `finally` block in `get_kprofiles()` could clear the attribute to `None` after a timeout — a classic TOCTOU race. Fixed by capturing the event reference in a local variable before the check.
 - **Queue Stuck on "Busy" for "Any Model" Jobs** ([#435](https://github.com/maziggy/bambuddy/issues/435)) — When a print was queued with "Any [Model]" (e.g., "Any P1S"), it was created with `printer_id=NULL` and `target_model="P1S"`. After the assigned printer finished, the queue widget queried only for items matching `printer_id=X`, missing the next pending model-based item (`printer_id IS NULL`). With no next item found, the "Clear Plate & Start Next" button never appeared, leaving the scheduler stuck reporting "Busy". The queue API now accepts an optional `target_model` parameter; when combined with `printer_id`, it uses OR logic to also return unassigned items whose `target_model` matches the printer's model. The frontend passes the printer's model through to this query. Additionally, the backend now resolves the printer's model server-side from the database when the frontend doesn't provide `target_model` (e.g., when the printer was added without selecting a model), ensuring the OR logic works regardless of whether the client knows the printer's model.
 - **Queue "Any Model" Jobs Stuck in "Waiting" After Plate Clear** ([#435](https://github.com/maziggy/bambuddy/issues/435)) — After the queue visibility fix above, "Any Model" jobs were correctly assigned to an idle printer but immediately crashed with `'>=' not supported between instances of 'str' and 'int'` when computing AMS filament mapping. MQTT raw data returns AMS unit and tray IDs as strings, but `_build_loaded_filaments()` compared them to integers without casting. The crash prevented the assignment from committing, so the scheduler retried every 30 seconds in an infinite loop. Cast `ams_id` and `tray_id` to `int()` to match the pattern already used for external spool IDs.
@@ -26,6 +26,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Filament Cost Tracking Test Coverage** — Added 2 backend unit tests for archive cost aggregation (zero-cost guard preserves existing costs, positive-cost updates archive correctly). Added 2 frontend unit tests for spool form cost_per_kg persistence. Fixed missing `archive_id` database migration, SQLAlchemy `is None` → `.is_(None)` in where clauses, duplicate archive cost write, and unconditional zero-cost overwrite.
 - **Spool Assignment Snapshot Test Coverage** — Added 7 backend unit tests covering spool assignment snapshotting at print start, snapshot-preferred spool lookup in both 3MF and AMS delta paths, fallback to live query for pre-upgrade sessions, and the core mid-print unlink scenario from #459.
 - **Background Dispatch Test Coverage** — Added 5 backend unit tests for dispatch cancel races (single-lock TOCTOU fix), batch counter reset re-check, and job lifecycle. Added 2 FTP regression tests for voidresp error handling (upload-loop prevention) and A1 model voidresp skip. Added 1 frontend test for reprint toast suppression.
+- **Tray Change Split Test Coverage** — Added 8 MQTT unit tests for `tray_change_log` lifecycle (default empty, seed on print start, clear on new print, record during RUNNING/PAUSE, ignore during IDLE, deduplicate, multi-change history). Added 6 usage tracker unit tests for weight splitting (per-layer gcode split, linear fallback, no-change normal path, empty log recovery, missing spool skip, triple segment split).
 - **Frontend Pre-Commit Hooks** ([#458](https://github.com/maziggy/bambuddy/issues/458)) — Added `frontend-typecheck` (`tsc --noEmit`) and `frontend-lint` (`eslint .`) hooks to the pre-commit config. Both hooks only trigger when `frontend/src/**/*.{ts,tsx}` files are staged.
 
 ## [0.2.1b] - 2026-02-19

backend/app/services/bambu_mqtt.py

@@ -163,6 +163,9 @@ class PrinterState:
     big_fan1_speed: int | None = None  # Auxiliary fan
     big_fan2_speed: int | None = None  # Chamber/exhaust fan
     heatbreak_fan_speed: int | None = None  # Hotend heatbreak fan
+    # 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)
     # Firmware version info (from info.module[name="ota"].sw_ver)
     firmware_version: str | None = None
 
@@ -1059,6 +1062,15 @@ class BambuMQTTClient:
                 # Valid physical trays: 0-15 (regular AMS), 128-135 (AMS-HT), 254 (external spool)
                 tn = self.state.tray_now
                 if (0 <= tn <= 15) or (128 <= tn <= 135) or tn == 254:
+                    # Log tray change for mid-print usage splitting
+                    if tn != self.state.last_loaded_tray and self.state.state in ("RUNNING", "PAUSE"):
+                        self.state.tray_change_log.append((tn, self.state.layer_num))
+                        logger.info(
+                            "[%s] Tray change during print: tray=%d at layer=%d",
+                            self.serial_number,
+                            tn,
+                            self.state.layer_num,
+                        )
                     self.state.last_loaded_tray = self.state.tray_now
 
                 logger.debug("[%s] tray_now updated: %s", self.serial_number, self.state.tray_now)
@@ -2037,6 +2049,11 @@ class BambuMQTTClient:
             # Reset last valid progress/layer for usage tracking
             self._last_valid_progress = 0.0
             self._last_valid_layer_num = 0
+            # Clear and seed tray change log for mid-print usage splitting
+            self.state.tray_change_log.clear()
+            tn = self.state.tray_now
+            if (0 <= tn <= 15) or (128 <= tn <= 135) or tn == 254:
+                self.state.tray_change_log.append((tn, 0))
             # Initialize timelapse tracking based on current state
             # NOTE: xcam data is parsed BEFORE this code runs in _process_message,
             # so self.state.timelapse may already be set from this message.

backend/app/services/usage_tracker.py

@@ -565,13 +565,20 @@ async def _track_from_3mf(
     )
 
     # 5. For single-filament non-queue prints, use tray_now from printer state
-    #    Priority: tray_now_at_start > current tray_now > last_loaded_tray > vt_tray check
+    #    Priority: tray_change_log (multi-tray split) > tray_now_at_start > current tray_now
+    #              > last_loaded_tray > vt_tray check
     nonzero_slots = [u for u in filament_usage if u.get("used_g", 0) > 0]
     tray_now_override: int | None = None
+    tray_changes: list[tuple[int, int]] = []  # [(global_tray_id, layer_num), ...]
     if not slot_to_tray and len(nonzero_slots) == 1:
         state = printer_manager.get_status(printer_id)
-        # Try tray_now_at_start first (captured at print start)
-        if 0 <= tray_now_at_start <= 254:
+        tray_changes = getattr(state, "tray_change_log", []) if state else []
+
+        if len(tray_changes) > 1:
+            # Multi-tray usage detected — will split in per-slot loop using per-layer gcode
+            logger.info("[UsageTracker] 3MF: tray change log: %s (will split weight)", tray_changes)
+        elif 0 <= tray_now_at_start <= 254:
+            # Try tray_now_at_start first (captured at print start)
             tray_now_override = tray_now_at_start
             logger.info("[UsageTracker] 3MF: using tray_now_at_start=%d (single-filament fallback)", tray_now_at_start)
         elif state and 0 <= state.tray_now <= 254:
@@ -588,7 +595,7 @@ async def _track_from_3mf(
             if any(int(vt.get("id", 0)) == 255 for vt in vt_tray if isinstance(vt, dict)):
                 tray_now_override = state.tray_now
                 logger.info("[UsageTracker] 3MF: using tray_now=255 (H2-series external spool)")
-        if tray_now_override is None:
+        if tray_now_override is None and len(tray_changes) <= 1:
             logger.info(
                 "[UsageTracker] 3MF: no valid tray_now (at_start=%d, current=%s, last_loaded=%s)",
                 tray_now_at_start,
@@ -648,6 +655,164 @@ async def _track_from_3mf(
         if used_g <= 0:
             continue
 
+        # --- Mid-print tray switch: split weight across trays ---
+        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:
+                total_weight = layer_grams[slot_id]
+            else:
+                total_weight = used_g * scale
+
+            if total_weight <= 0:
+                continue
+
+            # Extract per-layer gcode for segment splitting
+            split_layer_usage = None
+            split_props: dict = {}
+            try:
+                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)
+                filament_props = extract_filament_properties_from_3mf(file_path)
+                split_props = filament_props.get(slot_id, {})
+            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)
+
+                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
+                    seg_end_layer = tray_changes[seg_idx + 1][1]
+                    total_layers = state.total_layers if state else 0
+                    if total_layers > 0:
+                        segment_grams = total_weight * (seg_end_layer - seg_start_layer) / total_layers
+                    else:
+                        # Can't compute ratio — assign all to last segment
+                        segment_grams = 0.0
+
+                sum_previous += segment_grams
+                if segment_grams <= 0:
+                    continue
+
+                # Convert global tray ID to (ams_id, tray_id)
+                if tray_global >= 254:
+                    seg_ams_id = 255
+                    seg_tray_id = tray_global - 254
+                elif tray_global >= 128:
+                    seg_ams_id = tray_global
+                    seg_tray_id = 0
+                else:
+                    seg_ams_id = tray_global // 4
+                    seg_tray_id = tray_global % 4
+
+                seg_key = (seg_ams_id, seg_tray_id)
+                if seg_key in handled_trays:
+                    continue
+
+                logger.info(
+                    "[UsageTracker] 3MF split: segment %d tray=%d (AMS%d-T%d) layers %d-%s -> %.1fg",
+                    seg_idx,
+                    tray_global,
+                    seg_ams_id,
+                    seg_tray_id,
+                    seg_start_layer,
+                    tray_changes[seg_idx + 1][1] if not is_last else "end",
+                    segment_grams,
+                )
+
+                # Find spool for this tray
+                seg_spool_id = spool_assignments.get(seg_key) if spool_assignments else None
+                if seg_spool_id is None:
+                    assign_result = await db.execute(
+                        select(SpoolAssignment).where(
+                            SpoolAssignment.printer_id == printer_id,
+                            SpoolAssignment.ams_id == seg_ams_id,
+                            SpoolAssignment.tray_id == seg_tray_id,
+                        )
+                    )
+                    assignment = assign_result.scalar_one_or_none()
+                    if not assignment:
+                        logger.info(
+                            "[UsageTracker] 3MF split: no spool at printer %d AMS%d-T%d, skipping segment",
+                            printer_id,
+                            seg_ams_id,
+                            seg_tray_id,
+                        )
+                        continue
+                    seg_spool_id = assignment.spool_id
+
+                spool_result = await db.execute(select(Spool).where(Spool.id == seg_spool_id))
+                spool = spool_result.scalar_one_or_none()
+                if not spool:
+                    continue
+
+                spool.weight_used = (spool.weight_used or 0) + segment_grams
+                spool.last_used = datetime.now(timezone.utc)
+
+                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,
+                    }
+                )
+
+                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,
+                )
+
+            continue  # Skip normal single-tray processing for this slot
+
         # Map 3MF slot_id to physical (ams_id, tray_id) using resolved mapping
         if tray_now_override is not None:
             # Single-filament non-queue print: use actual tray from printer state

backend/tests/unit/services/test_bambu_mqtt.py

@@ -1863,3 +1863,139 @@ class TestTrayNowDualNozzleH2DFullSequence(_H2DFixtureMixin):
         h2d_client._process_message(_ams_payload(255))
         assert h2d_client.state.tray_now == 255
         assert h2d_client.state.last_loaded_tray == 2
+
+
+class TestTrayChangeLog:
+    """Tests for tray_change_log tracking during prints (mid-print tray switch)."""
+
+    @pytest.fixture
+    def mqtt_client(self):
+        """Create a BambuMQTTClient instance for testing."""
+        from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+        client = BambuMQTTClient(
+            ip_address="192.168.1.100",
+            serial_number="TRAYLOG1",
+            access_code="12345678",
+        )
+        return client
+
+    def test_tray_change_log_defaults_empty(self, mqtt_client):
+        """tray_change_log starts as an empty list."""
+        assert mqtt_client.state.tray_change_log == []
+
+    def test_tray_change_log_seeded_on_print_start(self, mqtt_client):
+        """Print start clears log and seeds with initial tray at layer 0."""
+        mqtt_client.state.tray_now = 2
+        mqtt_client.state.last_loaded_tray = 2
+        mqtt_client._previous_gcode_state = "IDLE"
+
+        # Transition to RUNNING via _process_message
+        mqtt_client._process_message(
+            {
+                "print": {
+                    "gcode_state": "RUNNING",
+                    "gcode_file": "test.3mf",
+                }
+            }
+        )
+
+        assert mqtt_client.state.tray_change_log == [(2, 0)]
+
+    def test_tray_change_log_cleared_on_new_print(self, mqtt_client):
+        """Old log entries are cleared when a new print starts."""
+        mqtt_client.state.tray_change_log = [(5, 0), (3, 100)]
+        mqtt_client.state.tray_now = 1
+        mqtt_client.state.last_loaded_tray = 1
+        mqtt_client._previous_gcode_state = "IDLE"
+
+        mqtt_client._process_message(
+            {
+                "print": {
+                    "gcode_state": "RUNNING",
+                    "gcode_file": "new.3mf",
+                }
+            }
+        )
+
+        assert mqtt_client.state.tray_change_log == [(1, 0)]
+
+    def test_tray_change_recorded_during_running(self, mqtt_client):
+        """Tray change while RUNNING is appended to the log."""
+        mqtt_client.state.state = "RUNNING"
+        mqtt_client.state.layer_num = 50
+        mqtt_client.state.last_loaded_tray = 0
+        mqtt_client.state.tray_change_log = [(0, 0)]
+
+        # Simulate tray_now update via AMS data
+        mqtt_client.state.tray_now = 1
+        # Trigger the tracking code path
+        tn = mqtt_client.state.tray_now
+        if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
+            mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
+        mqtt_client.state.last_loaded_tray = tn
+
+        assert mqtt_client.state.tray_change_log == [(0, 0), (1, 50)]
+
+    def test_tray_change_not_recorded_when_idle(self, mqtt_client):
+        """Tray changes while IDLE are NOT logged."""
+        mqtt_client.state.state = "IDLE"
+        mqtt_client.state.layer_num = 0
+        mqtt_client.state.last_loaded_tray = 0
+        mqtt_client.state.tray_change_log = []
+
+        mqtt_client.state.tray_now = 3
+        tn = mqtt_client.state.tray_now
+        if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
+            mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
+        mqtt_client.state.last_loaded_tray = tn
+
+        assert mqtt_client.state.tray_change_log == []
+
+    def test_tray_change_recorded_during_pause(self, mqtt_client):
+        """Tray change while PAUSE is also logged (AMS can swap during pause)."""
+        mqtt_client.state.state = "PAUSE"
+        mqtt_client.state.layer_num = 75
+        mqtt_client.state.last_loaded_tray = 2
+        mqtt_client.state.tray_change_log = [(2, 0)]
+
+        mqtt_client.state.tray_now = 5
+        tn = mqtt_client.state.tray_now
+        if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
+            mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
+        mqtt_client.state.last_loaded_tray = tn
+
+        assert mqtt_client.state.tray_change_log == [(2, 0), (5, 75)]
+
+    def test_same_tray_not_logged_twice(self, mqtt_client):
+        """Same tray value doesn't create duplicate log entries."""
+        mqtt_client.state.state = "RUNNING"
+        mqtt_client.state.layer_num = 30
+        mqtt_client.state.last_loaded_tray = 2
+        mqtt_client.state.tray_change_log = [(2, 0)]
+
+        # Same tray again
+        mqtt_client.state.tray_now = 2
+        tn = mqtt_client.state.tray_now
+        if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
+            mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
+        mqtt_client.state.last_loaded_tray = tn
+
+        assert mqtt_client.state.tray_change_log == [(2, 0)]
+
+    def test_multiple_tray_changes(self, mqtt_client):
+        """Multiple tray changes create a full history."""
+        mqtt_client.state.state = "RUNNING"
+        mqtt_client.state.last_loaded_tray = 0
+        mqtt_client.state.tray_change_log = [(0, 0)]
+
+        changes = [(1, 50), (3, 120), (0, 200)]
+        for tray, layer in changes:
+            mqtt_client.state.tray_now = tray
+            mqtt_client.state.layer_num = layer
+            tn = mqtt_client.state.tray_now
+            if tn != mqtt_client.state.last_loaded_tray and mqtt_client.state.state in ("RUNNING", "PAUSE"):
+                mqtt_client.state.tray_change_log.append((tn, mqtt_client.state.layer_num))
+            mqtt_client.state.last_loaded_tray = tn
+
+        assert mqtt_client.state.tray_change_log == [(0, 0), (1, 50), (3, 120), (0, 200)]

backend/tests/unit/test_usage_tracker.py

@@ -854,6 +854,386 @@ class TestTrackFrom3mf:
         assert results[0]["tray_id"] == 1
 
 
+class TestTrayChangeSplit:
+    """Tests for mid-print tray switch weight splitting in _track_from_3mf()."""
+
+    @pytest.mark.asyncio
+    async def test_tray_switch_splits_weight_with_gcode(self):
+        """Two-tray runout: weight split using per-layer gcode data."""
+        spool_a = _make_spool(spool_id=10, label_weight=1000)
+        spool_b = _make_spool(spool_id=20, label_weight=1000)
+        assign_a = _make_assignment(spool_id=10, ams_id=0, tray_id=1)
+        assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=0)
+        archive = _make_archive(archive_id=100)
+
+        # db: archive, queue_item(None), then for each segment: assignment, spool
+        db = _mock_db_sequential([archive, None, assign_a, spool_a, assign_b, spool_b])
+
+        # Tray change log: started on tray 1, switched to tray 0 at layer 60
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=100,
+            tray_now=0,
+            last_loaded_tray=0,
+            total_layers=100,
+            tray_change_log=[(1, 0), (0, 60)],
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 30.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
+                return_value={30: {0: 3000.0}, 60: {0: 6000.0}, 100: {0: 10000.0}},
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.get_cumulative_usage_at_layer",
+                side_effect=lambda data, layer: {0: {0: 0.0, 60: 6000.0, 100: 10000.0}.get(layer, 0.0)},
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_properties_from_3mf",
+                return_value={1: {"density": 1.24, "diameter": 1.75}},
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.mm_to_grams",
+                side_effect=lambda mm, d, dens: round(mm * 0.003, 1),  # Simple conversion
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=100,
+                status="completed",
+                print_name="Runout Test",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+            )
+
+        # Two results: one per tray segment
+        assert len(results) == 2
+        # First segment: tray 1 (AMS0-T1), layers 0→60
+        assert results[0]["ams_id"] == 0
+        assert results[0]["tray_id"] == 1
+        assert results[0]["spool_id"] == 10
+        assert results[0]["weight_used"] == 18.0  # 6000mm * 0.003
+        # Second segment: tray 0 (AMS0-T0), layers 60→end = 30.0 - 18.0 = 12.0
+        assert results[1]["ams_id"] == 0
+        assert results[1]["tray_id"] == 0
+        assert results[1]["spool_id"] == 20
+        assert results[1]["weight_used"] == 12.0
+        # Both trays handled
+        assert (0, 1) in handled_trays
+        assert (0, 0) in handled_trays
+
+    @pytest.mark.asyncio
+    async def test_tray_switch_linear_fallback(self):
+        """Two-tray runout without per-layer gcode: linear split by layer ratio."""
+        spool_a = _make_spool(spool_id=10, label_weight=1000)
+        spool_b = _make_spool(spool_id=20, label_weight=1000)
+        assign_a = _make_assignment(spool_id=10, ams_id=0, tray_id=2)
+        assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=1)
+        archive = _make_archive(archive_id=101)
+
+        db = _mock_db_sequential([archive, None, assign_a, spool_a, assign_b, spool_b])
+
+        # Tray 2 from layer 0, switched to tray 1 at layer 40 (of 100 total)
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=100,
+            tray_now=1,
+            last_loaded_tray=1,
+            total_layers=100,
+            tray_change_log=[(2, 0), (1, 40)],
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 50.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
+                return_value=None,  # No per-layer gcode available
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=101,
+                status="completed",
+                print_name="Linear Fallback",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+            )
+
+        assert len(results) == 2
+        # Linear split: tray 2 for 40/100 layers = 20g
+        assert results[0]["ams_id"] == 0
+        assert results[0]["tray_id"] == 2
+        assert results[0]["weight_used"] == 20.0
+        # Last segment gets remainder: 50 - 20 = 30g
+        assert results[1]["ams_id"] == 0
+        assert results[1]["tray_id"] == 1
+        assert results[1]["weight_used"] == 30.0
+
+    @pytest.mark.asyncio
+    async def test_no_tray_change_uses_normal_path(self):
+        """Single-entry tray_change_log falls through to normal tray_now_at_start logic."""
+        spool = _make_spool(spool_id=1, label_weight=1000)
+        assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=2)
+        archive = _make_archive(archive_id=102)
+
+        db = _mock_db_sequential([archive, None, assignment, spool])
+
+        # Only one entry = no switch, should use normal path
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=100,
+            tray_now=2,
+            last_loaded_tray=2,
+            total_layers=100,
+            tray_change_log=[(2, 0)],
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 15.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=102,
+                status="completed",
+                print_name="No Switch",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+                tray_now_at_start=2,
+            )
+
+        # Normal path: single result, full weight
+        assert len(results) == 1
+        assert results[0]["weight_used"] == 15.0
+        assert results[0]["ams_id"] == 0
+        assert results[0]["tray_id"] == 2
+
+    @pytest.mark.asyncio
+    async def test_empty_tray_change_log_uses_normal_path(self):
+        """Empty tray_change_log (e.g. server restart) falls through to existing logic."""
+        spool = _make_spool(spool_id=1, label_weight=1000)
+        assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
+        archive = _make_archive(archive_id=103)
+
+        db = _mock_db_sequential([archive, None, assignment, spool])
+
+        # Empty log (server restarted mid-print)
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=100,
+            tray_now=0,
+            last_loaded_tray=0,
+            total_layers=100,
+            tray_change_log=[],
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=103,
+                status="completed",
+                print_name="Restart Recovery",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+                tray_now_at_start=0,
+            )
+
+        assert len(results) == 1
+        assert results[0]["weight_used"] == 10.0
+
+    @pytest.mark.asyncio
+    async def test_tray_switch_segment_no_spool(self):
+        """Segment with no spool assignment is skipped; other segments still tracked."""
+        spool_b = _make_spool(spool_id=20, label_weight=1000)
+        assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=3)
+        archive = _make_archive(archive_id=104)
+
+        # db: archive, queue_item(None), 1st segment: no assignment, 2nd segment: assignment, spool
+        db = _mock_db_sequential([archive, None, None, assign_b, spool_b])
+
+        # Tray 5 (no spool) from layer 0, switched to tray 3 at layer 50
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=100,
+            tray_now=3,
+            last_loaded_tray=3,
+            total_layers=100,
+            tray_change_log=[(5, 0), (3, 50)],
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 40.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
+                return_value=None,  # No per-layer data
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=104,
+                status="completed",
+                print_name="Missing Spool",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+            )
+
+        # Only the second segment (tray 3) tracked; first segment (tray 5) skipped
+        assert len(results) == 1
+        assert results[0]["ams_id"] == 0
+        assert results[0]["tray_id"] == 3
+        assert results[0]["spool_id"] == 20
+
+    @pytest.mark.asyncio
+    async def test_tray_switch_three_segments(self):
+        """Three-segment switch (rare): A→B→C split by linear fallback."""
+        spool_a = _make_spool(spool_id=1, label_weight=1000)
+        spool_b = _make_spool(spool_id=2, label_weight=1000)
+        spool_c = _make_spool(spool_id=3, label_weight=1000)
+        assign_a = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
+        assign_b = _make_assignment(spool_id=2, ams_id=0, tray_id=1)
+        assign_c = _make_assignment(spool_id=3, ams_id=0, tray_id=2)
+        archive = _make_archive(archive_id=105)
+
+        db = _mock_db_sequential(
+            [
+                archive,
+                None,
+                assign_a,
+                spool_a,
+                assign_b,
+                spool_b,
+                assign_c,
+                spool_c,
+            ]
+        )
+
+        # 3 segments: tray 0 (0-30), tray 1 (30-70), tray 2 (70-end)
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            progress=100,
+            layer_num=100,
+            tray_now=2,
+            last_loaded_tray=2,
+            total_layers=100,
+            tray_change_log=[(0, 0), (1, 30), (2, 70)],
+        )
+
+        filament_usage = [{"slot_id": 1, "used_g": 100.0, "type": "PLA", "color": ""}]
+        handled_trays: set[tuple[int, int]] = set()
+
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=filament_usage,
+            ),
+            patch(
+                "backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
+                return_value=None,
+            ),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            results = await _track_from_3mf(
+                printer_id=1,
+                archive_id=105,
+                status="completed",
+                print_name="Triple Switch",
+                handled_trays=handled_trays,
+                printer_manager=printer_manager,
+                db=db,
+            )
+
+        assert len(results) == 3
+        # Tray 0: 30/100 * 100g = 30g
+        assert results[0]["weight_used"] == 30.0
+        assert results[0]["ams_id"] == 0
+        assert results[0]["tray_id"] == 0
+        # Tray 1: 40/100 * 100g = 40g
+        assert results[1]["weight_used"] == 40.0
+        assert results[1]["ams_id"] == 0
+        assert results[1]["tray_id"] == 1
+        # Tray 2: remainder = 100 - 30 - 40 = 30g
+        assert results[2]["weight_used"] == 30.0
+        assert results[2]["ams_id"] == 0
+        assert results[2]["tray_id"] == 2
+
+
 class TestDecodeMqttMapping:
     """Tests for _decode_mqtt_mapping() — snow-encoded MQTT mapping to global tray IDs."""