fix(scheduler): use inventory weight for "Prefer Lowest Filament" sort (#1508)

  Reporter has a P1S with an inventory spool cloned to slot 1 and the
  original (much further used) in slot 4, the preference enabled, and
  the dispatch picked slot 1 every time. The sort's been blind to
  Bambuddy inventory weights — it reads MQTT `tray.remain`, the
  printer firmware's RFID-decremented value, which has two limitations:

  - Bambu RFID only. Non-RFID spools report -1 and get clamped to a
    sentinel; multiple non-RFID trays then tie in the sort and Python's
    stable sort collapses to AMS-slot insertion order, so slot 1 wins.
  - Even when set, it's the printer's counter, not Bambuddy's
    `label_weight - weight_used` (internal) or Spoolman's
    `remaining_weight`. The two diverge whenever the user re-spools,
    swaps cardboard, or runs a print outside Bambuddy.

  The reporter is on internal-inventory mode with non-RFID spools — both
  failure modes apply, hence slot 1 every time.

  Fix: when a slot is bound to an inventory spool the inventory record's
  remaining weight becomes the sort signal. New async helper
  `_build_inventory_remain_overrides(db, printer_id, loaded)` returns
  `{global_tray_id: remaining_grams}` for bound slots — internal mode
  joins SpoolAssignment → Spool once per dispatch; Spoolman mode joins
  SpoolmanSlotAssignment then reuses `_spoolman_remaining_grams` from
  filament_deficit.py for parity.

  New `_prefer_lowest_sort_key` does a two-tier comparison: inventory-
  tracked spools sort BEFORE MQTT-only spools, then ascending by
  remaining within each tier, then ascending by ams_id*4+tray_id as the
  deterministic slot tie-breaker. The tier flag dominates so grams
  (inventory) and percent (MQTT) never get cross-compared — no unit
  conversion needed.

  MQTT-only behaviour is preserved exactly: remain=-1 still maps to the
  101 sentinel and slot order still decides on ties. Users who haven't
  bound any inventory spool see no change. The DB lookup runs only when
  prefer_lowest_filament is enabled.

  External / VT slots are skipped (tracked separately from AMS bindings).

backend/app/services/print_scheduler.py

@@ -9,6 +9,7 @@ from pathlib import Path
 
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 
 from backend.app.core.config import settings
 from backend.app.core.database import async_session, run_with_retry
@@ -18,6 +19,8 @@ from backend.app.models.print_queue import PrintQueueItem
 from backend.app.models.printer import Printer
 from backend.app.models.settings import Settings
 from backend.app.models.smart_plug import SmartPlug
+from backend.app.models.spool_assignment import SpoolAssignment
+from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
 from backend.app.services.bambu_ftp import (
     cache_3mf_download,
     delete_file_async,
@@ -855,8 +858,19 @@ class PrintScheduler:
         # Check if user prefers lowest remaining filament when multiple spools match
         prefer_lowest = await self._get_bool_setting(db, "prefer_lowest_filament")
 
+        # When the preference is on, surface Bambuddy's inventory-side
+        # remaining for each slot that's bound to a tracked spool, so the
+        # sort beats the MQTT-only blind spot (#1508). Skip the lookup
+        # entirely when the preference is off — no behaviour change for
+        # users who haven't opted in.
+        inventory_remain_overrides: dict[int, float] | None = None
+        if prefer_lowest:
+            inventory_remain_overrides = await self._build_inventory_remain_overrides(db, printer_id, loaded_filaments)
+
         # Compute mapping: match required filaments to available slots
-        return self._match_filaments_to_slots(filament_reqs, loaded_filaments, prefer_lowest)
+        return self._match_filaments_to_slots(
+            filament_reqs, loaded_filaments, prefer_lowest, inventory_remain_overrides
+        )
 
     def _build_override_direct_mapping(self, force_overrides: list[dict], status) -> list[int] | None:
         """Build an AMS mapping directly from force-color overrides without a 3MF.
@@ -1015,8 +1029,156 @@ class PrintScheduler:
         except ValueError:
             return False
 
+    async def _build_inventory_remain_overrides(
+        self, db: AsyncSession, printer_id: int, loaded: list[dict]
+    ) -> dict[int, float]:
+        """Return ``{global_tray_id: remaining_grams}`` for AMS slots the user
+        has bound to an inventory spool — Bambuddy-side or Spoolman-side.
+
+        The MQTT ``remain`` field on a tray is the printer firmware's
+        RFID-decremented value, which has two limitations the "Prefer Lowest
+        Remaining Filament" feature has been ignoring (#1508):
+
+        - it's only meaningful for Bambu RFID spools; everything else reports
+          ``-1`` (then clamped to a sentinel), so multiple non-RFID trays
+          compare equal and the sort collapses to AMS-slot order — the user
+          who's curating inventory weights gets the lower-slot pick instead
+          of the lower-remaining pick;
+        - even when set, it's the *printer's* counter, not Bambuddy's
+          ``label_weight - weight_used`` (internal mode) or Spoolman's
+          ``remaining_weight`` (Spoolman mode) — the two diverge any time the
+          user re-spools, swaps cardboard, or runs a print outside Bambuddy.
+
+        When the user has bound a spool to a slot, their own inventory
+        tracking is authoritative; this helper surfaces that value so the
+        sort can prefer it. Slots without a binding are absent from the
+        returned map — the caller then falls back to MQTT ``remain`` for
+        those, preserving the pre-#1508 behaviour for un-tracked spools.
+
+        Returns an empty map on any failure (no inventory bindings, DB
+        error, Spoolman unreachable). A best-effort lookup; "Prefer Lowest"
+        is a preference, not a guarantee.
+        """
+        if not loaded:
+            return {}
+        # External / virtual-tray slots are tracked separately from AMS — skip
+        # them so a VT-loaded spool doesn't accidentally inherit a tracked
+        # AMS binding (the tables use ams_id 254/255 for VT, but the cross
+        # match is fiddly and out of scope for this fix).
+        tracked_slots = [(f["ams_id"], f["tray_id"], f["global_tray_id"]) for f in loaded if not f.get("is_external")]
+        if not tracked_slots:
+            return {}
+
+        is_spoolman = await self._is_spoolman_mode(db)
+        overrides: dict[int, float] = {}
+
+        if is_spoolman:
+            result = await db.execute(
+                select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == printer_id)
+            )
+            assignments = list(result.scalars().all())
+            by_slot = {(a.ams_id, a.tray_id): a.spoolman_spool_id for a in assignments}
+            from backend.app.services.filament_deficit import _spoolman_remaining_grams
+
+            for ams_id, tray_id, gtid in tracked_slots:
+                spoolman_id = by_slot.get((ams_id, tray_id))
+                if spoolman_id is None:
+                    continue
+                grams = await _spoolman_remaining_grams(spoolman_id)
+                if grams is not None:
+                    overrides[gtid] = grams
+            return overrides
+
+        # Internal inventory mode (default). selectinload matches the pattern
+        # used elsewhere (inventory.py, spoolman.py routes) — a single query
+        # plus an eager-loaded relationship rather than an explicit join, so
+        # the row-attribute shape is exactly what those routes already rely on.
+        result = await db.execute(
+            select(SpoolAssignment)
+            .options(selectinload(SpoolAssignment.spool))
+            .where(SpoolAssignment.printer_id == printer_id)
+        )
+        assignments = list(result.scalars().all())
+        by_slot = {(a.ams_id, a.tray_id): a.spool for a in assignments}
+        for ams_id, tray_id, gtid in tracked_slots:
+            spool = by_slot.get((ams_id, tray_id))
+            if spool is None:
+                continue
+            label = float(spool.label_weight or 0)
+            used = float(spool.weight_used or 0)
+            overrides[gtid] = max(0.0, label - used)
+        return overrides
+
+    @staticmethod
+    async def _is_spoolman_mode(db: AsyncSession) -> bool:
+        """Mirror of ``filament_deficit._is_spoolman_mode`` — kept private
+        here to avoid making this module import-dependent on that private
+        helper's signature."""
+        try:
+            from backend.app.api.routes.settings import get_setting
+
+            v = await get_setting(db, "spoolman_enabled")
+            return bool(v) and v.lower() == "true"
+        except Exception:
+            return False
+
+    @staticmethod
+    def _slot_priority(ams_id: int | None, tray_id: int | None) -> int:
+        """Deterministic slot-position tie-breaker for the prefer-lowest sort.
+
+        Three bands, matched to the emission order in ``_build_loaded_filaments``
+        so a tied sort produces the same physical-position order the pre-#1508
+        stable sort did (preserves the regression-free baseline):
+
+        - Regular AMS (``ams_id`` 0..7): ``ams_id * 4 + tray_id`` → 0..31
+        - AMS-HT (``ams_id`` >= 128, single tray): ``1000 + (ams_id - 128) * 4``
+        - External / VT (``ams_id`` < 0, or ``None``): ``10_000``
+
+        Banding ensures regular AMS < AMS-HT < external on ties, regardless of
+        what the raw ``ams_id`` happens to be (in particular, ``ams_id = -1``
+        for VT must NOT sort to a negative number or it would beat AMS slot 0).
+        """
+        if ams_id is None or ams_id < 0:
+            return 10_000
+        if ams_id >= 128:
+            return 1_000 + (ams_id - 128) * 4 + (tray_id or 0)
+        return ams_id * 4 + (tray_id or 0)
+
+    @staticmethod
+    def _prefer_lowest_sort_key(f: dict, overrides: dict[int, float] | None) -> tuple[int, float, int]:
+        """Sort key for the "Prefer Lowest Remaining Filament" preference.
+
+        Two-tier ordering: inventory-tracked spools always sort BEFORE
+        non-tracked spools (the user has told us they care about these
+        specifically), then ascending by remaining within each tier, then
+        ascending by AMS slot position as the deterministic tie-breaker.
+
+        Tiers are flagged by the first tuple element (0 = inventory-tracked,
+        1 = MQTT-only / unknown). Cross-tier value comparisons never run
+        because the tier flag dominates — which is what lets us mix grams
+        (inventory) and percent (MQTT) without a unit conversion.
+
+        Within the MQTT tier ``remain = -1`` (unknown) is mapped to 101 so
+        spools the printer DOES know something about sort ahead of those
+        it knows nothing about — preserves pre-#1508 behaviour for the
+        no-inventory-binding case.
+
+        Slot tie-breaker via ``_slot_priority`` so regular AMS < AMS-HT <
+        external on ties, matching the legacy emission-order stable sort.
+        """
+        gtid = f.get("global_tray_id")
+        slot_order = PrintScheduler._slot_priority(f.get("ams_id"), f.get("tray_id"))
+        if overrides and gtid in overrides:
+            return (0, overrides[gtid], slot_order)
+        remain = f.get("remain", -1)
+        return (1, float(remain) if remain is not None and remain >= 0 else 101.0, slot_order)
+
     def _match_filaments_to_slots(
-        self, required: list[dict], loaded: list[dict], prefer_lowest: bool = False
+        self,
+        required: list[dict],
+        loaded: list[dict],
+        prefer_lowest: bool = False,
+        inventory_remain_overrides: dict[int, float] | None = None,
     ) -> list[int] | None:
         """Match required filaments to loaded filaments and build AMS mapping.
 
@@ -1063,9 +1225,11 @@ class PrintScheduler:
             if req_nozzle_id is not None:
                 available = [f for f in available if f.get("extruder_id") == req_nozzle_id]
 
-            # Sort by remaining filament (ascending) so lowest-remain spool wins .find()
+            # Sort by remaining filament (ascending) so lowest-remain spool wins .find().
+            # Inventory-tracked spools sort before MQTT-only ones (#1508); see
+            # _prefer_lowest_sort_key for the full rationale.
             if prefer_lowest:
-                available.sort(key=lambda f: f.get("remain", -1) if f.get("remain", -1) >= 0 else 101)
+                available.sort(key=lambda f: self._prefer_lowest_sort_key(f, inventory_remain_overrides))
 
             # Check if tray_info_idx is unique among available trays
             if req_tray_info_idx:
@@ -1084,7 +1248,7 @@ class PrintScheduler:
                         f"using color matching among trays: {[f['global_tray_id'] for f in idx_matches]}"
                     )
                     if prefer_lowest:
-                        idx_matches.sort(key=lambda f: f.get("remain", -1) if f.get("remain", -1) >= 0 else 101)
+                        idx_matches.sort(key=lambda f: self._prefer_lowest_sort_key(f, inventory_remain_overrides))
                     # Use color matching within this subset
                     for f in idx_matches:
                         f_color = f.get("color", "")

backend/tests/unit/test_scheduler_ams_mapping.py

@@ -505,6 +505,270 @@ class TestPreferLowestFilament:
         assert result == [254]  # Should pick external spool (10%) over AMS (80%)
 
 
+class TestPreferLowestInventoryOverride:
+    """Tests for the #1508 inventory-aware sort: when the user has bound a
+    Bambuddy inventory spool to an AMS slot, that spool's remaining weight
+    becomes the sort signal instead of the MQTT ``remain`` percentage.
+
+    The fix is two-tier: inventory-tracked spools always sort before
+    MQTT-only ones, then ascending by remaining within each tier, then
+    ascending by AMS slot position. See
+    ``print_scheduler._prefer_lowest_sort_key`` for the rationale.
+    """
+
+    @pytest.fixture
+    def scheduler(self):
+        return PrintScheduler()
+
+    def test_inventory_override_beats_mqtt_remain(self, scheduler):
+        """Slot 4's inventory shows 50 g remaining; slot 1's clone has 950 g.
+        MQTT ``remain`` is -1 for both (non-RFID spools), so without the
+        override the sort collapses to AMS-slot order and slot 1 wins.
+        With the override slot 4 (the original, nearly empty) wins. This is
+        the literal reporter scenario in #1508.
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": -1,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 3,
+                "global_tray_id": 3,
+                "remain": -1,
+            },
+        ]
+        # Slot 1 (gtid 0) is the fresh clone at 950 g; slot 4 (gtid 3) is the
+        # nearly-empty original at 50 g.
+        overrides = {0: 950.0, 3: 50.0}
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
+        )
+        assert result == [3]
+
+    def test_inventory_override_with_zero_grams_still_wins(self, scheduler):
+        """An inventory-tracked spool at 0 g must still sort first within
+        its tier — the user wants to finish what's left (or be told there's
+        a deficit) rather than skip to the fresh one. The clamp-to-101
+        legacy logic only fires on negative values, so 0 g stays 0.0.
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": -1,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 1,
+                "global_tray_id": 1,
+                "remain": -1,
+            },
+        ]
+        overrides = {0: 500.0, 1: 0.0}
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
+        )
+        assert result == [1]
+
+    def test_inventory_tier_beats_mqtt_tier_regardless_of_value(self, scheduler):
+        """Mixed mode: one slot is inventory-tracked at 800 g (high), the
+        other has only a MQTT remain of 10 (very low). The inventory tier
+        wins because the user has explicitly told us to manage that slot —
+        unit-mixing across tiers is intentional and resolved by the tier
+        flag, not value comparison.
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": 10,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 1,
+                "global_tray_id": 1,
+                "remain": -1,
+            },
+        ]
+        overrides = {1: 800.0}  # only slot 2 has an inventory binding
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
+        )
+        assert result == [1]
+
+    def test_two_tracked_spools_tied_at_same_grams_lower_slot_wins(self, scheduler):
+        """Two inventory-tracked spools with genuinely equal remaining
+        weight — slot tie-breaker decides. Lower AMS slot wins to match
+        the user's mental model of "use the lower slot first when equal."
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 1,
+                "global_tray_id": 1,
+                "remain": -1,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": -1,
+            },
+        ]
+        overrides = {0: 500.0, 1: 500.0}
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=overrides
+        )
+        assert result == [0]
+
+    def test_no_override_falls_back_to_mqtt_remain(self, scheduler):
+        """When no slots are inventory-bound the override map is empty
+        and behaviour is identical to the pre-#1508 MQTT-only sort.
+        Regression guard for the un-tracked-spool case.
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": 80,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 1,
+                "global_tray_id": 1,
+                "remain": 30,
+            },
+        ]
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides={}
+        )
+        assert result == [1]
+
+    def test_no_override_unknown_remain_sorts_after_known(self, scheduler):
+        """In the no-binding case, ``remain = -1`` (non-RFID, unknown) must
+        still sort *after* a slot the printer knows something about — the
+        legacy sentinel-101 behaviour is preserved.
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": -1,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 1,
+                "global_tray_id": 1,
+                "remain": 50,
+            },
+        ]
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=None
+        )
+        assert result == [1]
+
+    def test_external_with_negative_ams_id_does_not_outrank_ams_on_tie(self, scheduler):
+        """``_build_loaded_filaments`` emits external/VT trays with
+        ``ams_id = -1``. A naive ``ams_id * 4 + tray_id`` slot-priority
+        formula would compute -4 for an external and 0 for AMS slot 0 —
+        flipping the legacy stable-sort baseline (which kept AMS first
+        because it's emitted before externals). When ``remain`` ties
+        between the two, AMS slot 0 must still win.
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": -1,
+                "is_external": False,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": -1,
+                "tray_id": 0,
+                "global_tray_id": 254,
+                "remain": -1,
+                "is_external": True,
+            },
+        ]
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=None
+        )
+        assert result == [0]
+
+    def test_ams_ht_does_not_outrank_regular_ams_on_tie(self, scheduler):
+        """AMS-HT units use ``ams_id`` >= 128 with a single tray. On a tied
+        ``remain`` value, regular AMS slot 0 (slot_priority 0) must beat
+        AMS-HT (slot_priority 1000+).
+        """
+        required = [{"slot_id": 1, "type": "PLA", "color": "#FF0000"}]
+        loaded = [
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 0,
+                "tray_id": 0,
+                "global_tray_id": 0,
+                "remain": 50,
+            },
+            {
+                "type": "PLA",
+                "color": "#FF0000",
+                "ams_id": 128,
+                "tray_id": 0,
+                "global_tray_id": 128,
+                "remain": 50,
+            },
+        ]
+        result = scheduler._match_filaments_to_slots(
+            required, loaded, prefer_lowest=True, inventory_remain_overrides=None
+        )
+        assert result == [0]
+
+
 class TestBuildLoadedFilamentsTrayInfoIdx:
     """Test tray_info_idx extraction in _build_loaded_filaments."""
 

backend/tests/unit/test_scheduler_inventory_remain.py

@@ -0,0 +1,194 @@
+"""Tests for the inventory-remain override builder in print_scheduler (#1508).
+
+The MQTT ``remain`` field on an AMS tray is the printer firmware's
+RFID-tracked value, which is ``-1`` for non-Bambu spools (and even when
+set diverges from Bambuddy's inventory). When the user has bound an
+inventory spool to an AMS slot, that inventory record's
+``label_weight - weight_used`` (or Spoolman's ``remaining_weight``) is
+the authoritative remaining-weight signal. These tests verify
+``_build_inventory_remain_overrides`` surfaces those values keyed by
+``global_tray_id`` so the "Prefer Lowest Remaining Filament" sort can
+consume them.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+def scheduler():
+    return PrintScheduler()
+
+
+def _make_async_session_returning(rows: list):
+    """Build a stub AsyncSession whose .execute() returns an object whose
+    .all() (and .scalars().all()) yield ``rows``."""
+    result = MagicMock()
+    result.all.return_value = rows
+    scalars = MagicMock()
+    scalars.all.return_value = rows
+    result.scalars.return_value = scalars
+    db = MagicMock()
+    db.execute = AsyncMock(return_value=result)
+    return db
+
+
+class TestInternalInventoryOverrides:
+    @pytest.mark.asyncio
+    async def test_returns_remaining_grams_for_bound_slots(self, scheduler):
+        """Two slots bound; both come back keyed by global_tray_id with the
+        correct ``label_weight - weight_used`` in grams. This is the
+        reporter scenario in #1508: slot 1 has a 950 g clone, slot 4 has
+        a 50 g original — the sort can now actually pick the 50 g spool.
+
+        The override builder uses ``select(SpoolAssignment).options(
+        selectinload(SpoolAssignment.spool))`` (matching the rest of the
+        codebase), so the rows it iterates expose ``.ams_id``, ``.tray_id``
+        and ``.spool`` directly — the test stubs the same shape.
+        """
+        spool_a = SimpleNamespace(label_weight=1000, weight_used=50)  # 950 g remaining
+        spool_b = SimpleNamespace(label_weight=1000, weight_used=950)  # 50 g remaining
+        rows = [
+            SimpleNamespace(ams_id=0, tray_id=0, spool=spool_a),
+            SimpleNamespace(ams_id=0, tray_id=3, spool=spool_b),
+        ]
+        loaded = [
+            {"ams_id": 0, "tray_id": 0, "global_tray_id": 0, "is_external": False},
+            {"ams_id": 0, "tray_id": 3, "global_tray_id": 3, "is_external": False},
+        ]
+        db = _make_async_session_returning(rows)
+        with patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=False)):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=loaded)
+        assert out == {0: 950.0, 3: 50.0}
+
+    @pytest.mark.asyncio
+    async def test_skips_external_slots(self, scheduler):
+        """VT / external slots are tracked separately from AMS inventory
+        bindings — the override builder must not assign them an inventory
+        remaining value even if (somehow) an assignment row exists.
+        """
+        loaded = [
+            {"ams_id": -1, "tray_id": 0, "global_tray_id": 254, "is_external": True},
+        ]
+        db = _make_async_session_returning([])
+        with patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=False)):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=loaded)
+        # DB shouldn't even be queried — nothing AMS-side to look up.
+        db.execute.assert_not_called()
+        assert out == {}
+
+    @pytest.mark.asyncio
+    async def test_empty_loaded_returns_empty(self, scheduler):
+        """No loaded filaments → no overrides. The scheduler short-circuits
+        before this is called in practice, but the function must be
+        defensive — it's used in any prefer_lowest dispatch path."""
+        db = _make_async_session_returning([])
+        with patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=False)):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=[])
+        assert out == {}
+        db.execute.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_negative_remaining_clamped_to_zero(self, scheduler):
+        """An over-consumed spool (weight_used > label_weight) shouldn't
+        produce a negative grams value — clamped to 0 so the sort treats
+        it as fully empty rather than "more empty than zero."
+        """
+        spool = SimpleNamespace(label_weight=1000, weight_used=1100)
+        rows = [
+            SimpleNamespace(ams_id=0, tray_id=0, spool=spool),
+        ]
+        loaded = [{"ams_id": 0, "tray_id": 0, "global_tray_id": 0, "is_external": False}]
+        db = _make_async_session_returning(rows)
+        with patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=False)):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=loaded)
+        assert out == {0: 0.0}
+
+    @pytest.mark.asyncio
+    async def test_slot_without_binding_absent_from_overrides(self, scheduler):
+        """A slot that has loaded filament but no inventory binding must
+        not appear in the override map — the sort then falls back to MQTT
+        ``remain`` for that one slot, preserving pre-#1508 behaviour.
+        """
+        rows = [
+            SimpleNamespace(
+                ams_id=0,
+                tray_id=0,
+                spool=SimpleNamespace(label_weight=1000, weight_used=100),
+            ),
+        ]
+        loaded = [
+            {"ams_id": 0, "tray_id": 0, "global_tray_id": 0, "is_external": False},
+            {"ams_id": 0, "tray_id": 1, "global_tray_id": 1, "is_external": False},
+        ]
+        db = _make_async_session_returning(rows)
+        with patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=False)):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=loaded)
+        assert out == {0: 900.0}
+        assert 1 not in out
+
+
+class TestSpoolmanModeOverrides:
+    @pytest.mark.asyncio
+    async def test_spoolman_remaining_grams_used_when_available(self, scheduler):
+        """Spoolman mode: each bound slot's spoolman_spool_id is fetched
+        through ``_spoolman_remaining_grams``; the result is the same
+        global-tray-id-keyed grams map. Parity rule with internal mode
+        (feedback_inventory_modes_parity).
+        """
+        rows = [
+            SimpleNamespace(printer_id=1, ams_id=0, tray_id=0, spoolman_spool_id=42),
+            SimpleNamespace(printer_id=1, ams_id=0, tray_id=2, spoolman_spool_id=99),
+        ]
+        loaded = [
+            {"ams_id": 0, "tray_id": 0, "global_tray_id": 0, "is_external": False},
+            {"ams_id": 0, "tray_id": 2, "global_tray_id": 2, "is_external": False},
+        ]
+        db = _make_async_session_returning(rows)
+
+        async def _fake_grams(spool_id: int):
+            return {42: 720.0, 99: 80.0}[spool_id]
+
+        with (
+            patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=True)),
+            patch(
+                "backend.app.services.filament_deficit._spoolman_remaining_grams",
+                new=AsyncMock(side_effect=_fake_grams),
+            ),
+        ):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=loaded)
+        assert out == {0: 720.0, 2: 80.0}
+
+    @pytest.mark.asyncio
+    async def test_spoolman_unreachable_skips_silently(self, scheduler):
+        """If Spoolman is unreachable for one spool, ``_spoolman_remaining_grams``
+        returns None and that slot is omitted from the override map —
+        sorting then falls back to MQTT remain for that slot only.
+        """
+        rows = [
+            SimpleNamespace(printer_id=1, ams_id=0, tray_id=0, spoolman_spool_id=42),
+            SimpleNamespace(printer_id=1, ams_id=0, tray_id=1, spoolman_spool_id=99),
+        ]
+        loaded = [
+            {"ams_id": 0, "tray_id": 0, "global_tray_id": 0, "is_external": False},
+            {"ams_id": 0, "tray_id": 1, "global_tray_id": 1, "is_external": False},
+        ]
+        db = _make_async_session_returning(rows)
+
+        async def _fake_grams(spool_id: int):
+            return 500.0 if spool_id == 42 else None
+
+        with (
+            patch.object(PrintScheduler, "_is_spoolman_mode", new=AsyncMock(return_value=True)),
+            patch(
+                "backend.app.services.filament_deficit._spoolman_remaining_grams",
+                new=AsyncMock(side_effect=_fake_grams),
+            ),
+        ):
+            out = await scheduler._build_inventory_remain_overrides(db, printer_id=1, loaded=loaded)
+        assert out == {0: 500.0}
+        assert 1 not in out