fix(inventory): archive filament colour follows the assigned spool, not the 3MF (#1494)

  An archive's filament_color was parsed verbatim from the print job's
  3MF (filament_colour in project_settings.config) — the slicer's
  filament-slot colour, which a user picks independently of the exact
  hex they curate on the Bambuddy inventory spool. So a print from a
  #000000 inventory spool showed #161616 (the slicer's near-black) in
  the archive card and the Color Distribution graph, even though usage
  tracking correctly decremented the right spool.

  Once usage tracking has resolved the print's filament slots to
  inventory spools, the spool colours are authoritative. _track_from_3mf
  (built-in inventory) and report_usage (Spoolman mode) now overwrite
  the archive's filament_color with the slot-ordered, de-duplicated
  colours of the matched spools.

  The rewrite is all-or-nothing: it only applies when every used slot
  resolved to a spool carrying a colour, so a partially-mapped
  multi-colour print keeps the 3MF colour rather than silently dropping
  the unmatched slots.

  Shipped for both inventory modes: built-in spools read Spool.rgba,
  Spoolman spools read the spool's filament.color_hex (fetched via
  get_spool for tag-less slot-assignment matches). New helpers
  _spool_color_to_hex / _archive_colors_from_spools in usage_tracker.py,
  reused by spoolman_tracking.py via _apply_spool_colors_to_archive.

  Tests: 12 new in test_usage_tracker.py (hex normalisation, the
  all-or-nothing rule across single/multi/partial/no-colour/AMS-fallback
  cases, end-to-end rewrite), 4 in test_spoolman_tracking.py (Spoolman
  rewrite + empty/partial/missing-archive no-ops). 70 tracking tests
  green; backend ruff clean.

backend/app/services/spoolman_tracking.py

@@ -385,6 +385,7 @@ async def _report_spool_usage_for_slots(
     method_label: str,
     printer_serial: str = "",
     printer_id: int | None = None,
+    slot_colors_out: dict[int, str] | None = None,
 ) -> int:
     """Report usage to Spoolman for a list of (slot_id, grams) pairs.
 
@@ -395,6 +396,11 @@ async def _report_spool_usage_for_slots(
     never get their weight decremented because their extra.tag is empty
     on the Spoolman side.
 
+    When ``slot_colors_out`` is provided it is populated with
+    ``{slot_id: color_hex}`` for every resolved spool — used by
+    :func:`report_usage` to stamp the archive's filament colour from the
+    Spoolman spool rather than the slicer's 3MF value (#1494).
+
     Returns number of spools successfully updated.
     """
     spools_updated = 0
@@ -420,6 +426,10 @@ async def _report_spool_usage_for_slots(
 
         spool_id_to_use: int | None = None
         resolution_path = ""
+        # color_hex of the resolved spool's filament, for the #1494 archive
+        # colour rewrite. The tag path already has the full spool object;
+        # the slot-assignment path only yields an id and is fetched below.
+        spool_color_hex: str | None = None
 
         spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id)
         if spool_tag:
@@ -427,6 +437,7 @@ async def _report_spool_usage_for_slots(
             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 and printer_id is not None:
             ams_id, tray_id = _global_tray_id_to_ams_slot(global_tray_id)
@@ -442,6 +453,20 @@ async def _report_spool_usage_for_slots(
             )
             continue
 
+        # Record the spool's filament colour for the archive rewrite (#1494).
+        # The slot-assignment path resolved only an id, so fetch the spool.
+        # Strictly best-effort: a colour-fetch failure must never abort the
+        # weight reporting for the remaining slots, so the catch is broad.
+        if slot_colors_out is not None:
+            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] 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, grams_used)
             logger.info(
@@ -675,6 +700,7 @@ async def report_usage(printer_id: int, archive_id: int):
         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]
+        slot_colors: dict[int, str] = {}
         spools_updated = await _report_spool_usage_for_slots(
             client,
             usage_items,
@@ -683,9 +709,58 @@ async def report_usage(printer_id: int, archive_id: int):
             f"Archive {archive_id}",
             printer_serial,
             printer_id=printer_id,
+            slot_colors_out=slot_colors,
         )
 
         if spools_updated == 0:
             logger.info("[SPOOLMAN] Archive %s: no spools updated", archive_id)
         else:
             logger.info("[SPOOLMAN] Archive %s: updated %s spool(s)", archive_id, spools_updated)
+
+        # Stamp the archive's filament colour from the matched Spoolman spools
+        # so it reflects the curated inventory colour, not the slicer's 3MF
+        # value (#1494) — mirrors the built-in inventory path in usage_tracker.
+        await _apply_spool_colors_to_archive(db, archive_id, filament_usage, slot_colors)
+
+
+async def _apply_spool_colors_to_archive(
+    db,
+    archive_id: int,
+    filament_usage: list[dict],
+    slot_colors: dict[int, str],
+) -> None:
+    """Overwrite an archive's ``filament_color`` with the colours of the
+    Spoolman spools that fed the print (#1494).
+
+    All-or-nothing, exactly like the built-in inventory path: the colour is
+    only rewritten when every used slot resolved to a spool that carries a
+    colour, so a partial match never drops slots from the archive.
+    """
+    if not slot_colors:
+        return
+
+    from backend.app.models.archive import PrintArchive
+    from backend.app.services.usage_tracker import (
+        _archive_colors_from_spools,
+        _spool_color_to_hex,
+    )
+
+    results = [{"slot_id": sid, "color": _spool_color_to_hex(hex_)} for sid, hex_ in slot_colors.items()]
+    colors = _archive_colors_from_spools(filament_usage, results)
+    if not colors:
+        return
+
+    archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
+    if archive is None:
+        return
+
+    joined = ",".join(colors)
+    if joined != archive.filament_color:
+        logger.info(
+            "[SPOOLMAN] Archive %s filament_color %r -> %r (from Spoolman spools)",
+            archive_id,
+            archive.filament_color,
+            joined,
+        )
+        archive.filament_color = joined
+        await db.commit()

backend/app/services/usage_tracker.py

@@ -63,6 +63,60 @@ def _decode_mqtt_mapping(mapping_raw: list | None) -> list[int] | None:
     return result
 
 
+def _spool_color_to_hex(rgba: str | None) -> str | None:
+    """Normalise a ``Spool.rgba`` value (``RRGGBBAA`` hex, no ``#``) to the
+    ``#RRGGBB`` form archives store in ``filament_color``.
+
+    Alpha is dropped — the archive colour list and the Color Distribution
+    graph treat filament colour as opaque. Returns ``None`` for a missing or
+    too-short value so the caller can fall back to the 3MF colour.
+    """
+    if not rgba:
+        return None
+    h = rgba.strip().lstrip("#")
+    if len(h) < 6:
+        return None
+    return "#" + h[:6].upper()
+
+
+def _archive_colors_from_spools(filament_usage: list[dict], results: list[dict]) -> list[str] | None:
+    """Slot-ordered, de-duplicated hex colours for an archive's ``filament_color``,
+    taken from the inventory spools that actually fed the print (#1494).
+
+    The slicer's 3MF carries its own ``filament_colour`` per slot — a value
+    picked independently of the colour the user curates on the matched
+    inventory spool. So an archive printed from a ``#000000`` inventory spool
+    would otherwise show the slicer's near-black ``#161616``. Once usage
+    tracking has resolved the used slots to spools, the spool colours are the
+    authoritative source and replace the 3MF values.
+
+    Returns ``None`` — leave the 3MF colour untouched — unless *every* slot
+    with non-zero usage was matched to a spool that carries a colour. A
+    partial rewrite would silently drop the unmatched slots' colours from the
+    archive (and the Color Distribution graph), so it is all-or-nothing.
+    """
+    used_slots = {u["slot_id"] for u in filament_usage if u.get("used_g", 0) > 0 and u.get("slot_id") is not None}
+    if not used_slots:
+        return None
+
+    slot_color: dict[int, str] = {}
+    for r in results:
+        slot_id = r.get("slot_id")
+        color = r.get("color")
+        if slot_id is not None and color:
+            slot_color.setdefault(slot_id, color)
+
+    if not used_slots.issubset(slot_color):
+        return None
+
+    ordered: list[str] = []
+    for slot_id in sorted(used_slots):
+        color = slot_color[slot_id]
+        if color not in ordered:
+            ordered.append(color)
+    return ordered
+
+
 def _match_slots_by_color(
     filament_usage: list[dict],
     ams_raw: dict | list | None,
@@ -589,6 +643,10 @@ async def on_print_complete(
                         "tray_id": assign_tray_id,
                         "material": spool.material,
                         "cost": cost,
+                        # AMS remain%-delta fallback has no 3MF slot — slot_id
+                        # stays None so it is excluded from the colour rewrite.
+                        "slot_id": None,
+                        "color": _spool_color_to_hex(spool.rgba),
                     }
                 )
 
@@ -823,6 +881,7 @@ async def _track_from_3mf(
     from backend.app.utils.threemf_tools import extract_filament_usage_from_3mf
 
     file_path: Path | None = threemf_path
+    archive: PrintArchive | None = None
 
     if file_path is None and archive_id:
         result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
@@ -1134,6 +1193,8 @@ async def _track_from_3mf(
                         "tray_id": seg_tray_id,
                         "material": spool.material,
                         "cost": cost,
+                        "slot_id": slot_id,
+                        "color": _spool_color_to_hex(spool.rgba),
                     }
                 )
 
@@ -1263,6 +1324,8 @@ async def _track_from_3mf(
                 "tray_id": tray_id,
                 "material": spool.material,
                 "cost": cost,
+                "slot_id": slot_id,
+                "color": _spool_color_to_hex(spool.rgba),
             }
         )
 
@@ -1285,4 +1348,22 @@ async def _track_from_3mf(
             status,
         )
 
+    # --- Adopt the matched inventory spools' colours for the archive (#1494) ---
+    # The archive's filament_color was set from the slicer's 3MF at creation
+    # time; now that every used slot has been resolved to an inventory spool,
+    # the curated spool colour is authoritative. Committed by the caller's
+    # `if results: await db.commit()`.
+    if archive is not None:
+        spool_colors = _archive_colors_from_spools(filament_usage, results)
+        if spool_colors:
+            joined = ",".join(spool_colors)
+            if joined != archive.filament_color:
+                logger.info(
+                    "[UsageTracker] 3MF: archive %s filament_color %r -> %r (from inventory spools)",
+                    archive_id,
+                    archive.filament_color,
+                    joined,
+                )
+                archive.filament_color = joined
+
     return results

backend/tests/unit/services/test_spoolman_tracking.py

@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from backend.app.services.spoolman_tracking import (
+    _apply_spool_colors_to_archive,
     _get_fallback_spool_tag,
     _global_tray_id_to_ams_slot,
     _hash_serial_to_hex32,
@@ -300,3 +301,69 @@ class TestStorePrintData:
 
         # Tracking row was inserted — the fix is working.
         db.add.assert_called_once()
+
+
+class TestApplySpoolColorsToArchive:
+    """`_apply_spool_colors_to_archive` stamps the archive's filament_color
+    from the matched Spoolman spools (#1494) — the Spoolman-mode mirror of
+    the built-in inventory rewrite in usage_tracker."""
+
+    def _make_db(self, archive):
+        db = AsyncMock()
+        db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=archive)))
+        return db
+
+    @pytest.mark.asyncio
+    async def test_rewrites_color_from_spoolman_spool(self):
+        """The #1494 case: 3MF said #161616, the Spoolman spool is 000000."""
+        archive = MagicMock()
+        archive.filament_color = "#161616"
+        db = self._make_db(archive)
+
+        await _apply_spool_colors_to_archive(
+            db,
+            archive_id=10,
+            filament_usage=[{"slot_id": 1, "used_g": 15.9}],
+            slot_colors={1: "000000"},
+        )
+
+        assert archive.filament_color == "#000000"
+        db.commit.assert_awaited()
+
+    @pytest.mark.asyncio
+    async def test_empty_slot_colors_is_noop(self):
+        """No resolved spool colours — never touches the DB."""
+        db = self._make_db(MagicMock())
+        await _apply_spool_colors_to_archive(
+            db, archive_id=10, filament_usage=[{"slot_id": 1, "used_g": 15.9}], slot_colors={}
+        )
+        db.execute.assert_not_awaited()
+        db.commit.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_partial_match_leaves_archive_untouched(self):
+        """Slot 2 used but unresolved — keep the 3MF colour, don't load the archive."""
+        db = self._make_db(MagicMock())
+        await _apply_spool_colors_to_archive(
+            db,
+            archive_id=10,
+            filament_usage=[
+                {"slot_id": 1, "used_g": 10.0},
+                {"slot_id": 2, "used_g": 20.0},
+            ],
+            slot_colors={1: "000000"},
+        )
+        db.execute.assert_not_awaited()
+        db.commit.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_missing_archive_does_not_crash(self):
+        """Archive row gone (deleted between completion and reporting)."""
+        db = self._make_db(None)
+        await _apply_spool_colors_to_archive(
+            db,
+            archive_id=10,
+            filament_usage=[{"slot_id": 1, "used_g": 15.9}],
+            slot_colors={1: "000000"},
+        )
+        db.commit.assert_not_awaited()

backend/tests/unit/services/test_usage_tracker.py

@@ -12,13 +12,15 @@ import pytest
 from backend.app.services.usage_tracker import (
     PrintSession,
     _active_sessions,
+    _archive_colors_from_spools,
+    _spool_color_to_hex,
     _track_from_3mf,
     on_print_complete,
     on_print_start,
 )
 
 
-def _make_spool(*, id=1, label_weight=1000, weight_used=0, tag_uid=None, tray_uuid=None):
+def _make_spool(*, id=1, label_weight=1000, weight_used=0, tag_uid=None, tray_uuid=None, rgba=None):
     """Create a mock Spool object."""
     spool = MagicMock()
     spool.id = id
@@ -29,6 +31,7 @@ def _make_spool(*, id=1, label_weight=1000, weight_used=0, tag_uid=None, tray_uu
     spool.last_used = None
     spool.cost_per_kg = None
     spool.material = "PLA"
+    spool.rgba = rgba
     return spool
 
 
@@ -766,3 +769,194 @@ class TestSpoolAssignmentSnapshot:
         assert results[0]["weight_used"] == 14.2
         # Spool weight should be updated: 50 + 14.2 = 64.2
         assert spool.weight_used == 64.2
+
+
+class TestSpoolColorToHex:
+    """`_spool_color_to_hex` normalises Spool.rgba (RRGGBBAA, no #) to #RRGGBB."""
+
+    def test_strips_alpha_and_adds_hash(self):
+        assert _spool_color_to_hex("000000FF") == "#000000"
+        assert _spool_color_to_hex("EC984CFF") == "#EC984C"
+
+    def test_uppercases(self):
+        assert _spool_color_to_hex("ec984cff") == "#EC984C"
+
+    def test_accepts_six_char_value(self):
+        """A value with no alpha is still valid."""
+        assert _spool_color_to_hex("161616") == "#161616"
+
+    def test_tolerates_leading_hash(self):
+        assert _spool_color_to_hex("#000000FF") == "#000000"
+
+    def test_none_and_too_short_return_none(self):
+        """Missing / malformed colour falls back to the 3MF value."""
+        assert _spool_color_to_hex(None) is None
+        assert _spool_color_to_hex("") is None
+        assert _spool_color_to_hex("FFF") is None
+
+
+class TestArchiveColorsFromSpools:
+    """`_archive_colors_from_spools` rebuilds an archive's filament_color from
+    the inventory spools that fed the print (#1494). All-or-nothing: a partial
+    match returns None so the 3MF colour is left intact."""
+
+    def test_single_slot_matched(self):
+        """The #1494 case: one used slot, matched to a #000000 spool."""
+        usage = [{"slot_id": 1, "used_g": 15.9, "color": "#161616"}]
+        results = [{"slot_id": 1, "color": "#000000"}]
+        assert _archive_colors_from_spools(usage, results) == ["#000000"]
+
+    def test_multi_slot_all_matched_keeps_slot_order(self):
+        usage = [
+            {"slot_id": 1, "used_g": 10.0, "color": "#111111"},
+            {"slot_id": 2, "used_g": 20.0, "color": "#222222"},
+        ]
+        # results deliberately out of slot order — output must be slot-ordered
+        results = [
+            {"slot_id": 2, "color": "#00FF00"},
+            {"slot_id": 1, "color": "#FF0000"},
+        ]
+        assert _archive_colors_from_spools(usage, results) == ["#FF0000", "#00FF00"]
+
+    def test_duplicate_colors_deduplicated(self):
+        """Two slots of the same spool colour collapse to one entry, as the
+        3MF-derived path also de-duplicates."""
+        usage = [
+            {"slot_id": 1, "used_g": 10.0, "color": "#111111"},
+            {"slot_id": 2, "used_g": 20.0, "color": "#222222"},
+        ]
+        results = [
+            {"slot_id": 1, "color": "#000000"},
+            {"slot_id": 2, "color": "#000000"},
+        ]
+        assert _archive_colors_from_spools(usage, results) == ["#000000"]
+
+    def test_partial_match_returns_none(self):
+        """Slot 2 was used but never matched to a spool — leave the 3MF colour
+        untouched rather than dropping slot 2 from the archive."""
+        usage = [
+            {"slot_id": 1, "used_g": 10.0, "color": "#111111"},
+            {"slot_id": 2, "used_g": 20.0, "color": "#222222"},
+        ]
+        results = [{"slot_id": 1, "color": "#000000"}]
+        assert _archive_colors_from_spools(usage, results) is None
+
+    def test_matched_spool_without_color_returns_none(self):
+        """A spool with no rgba (color None) does not count as matched."""
+        usage = [{"slot_id": 1, "used_g": 15.0, "color": "#161616"}]
+        results = [{"slot_id": 1, "color": None}]
+        assert _archive_colors_from_spools(usage, results) is None
+
+    def test_unused_slot_not_required(self):
+        """A slot with zero usage need not be matched."""
+        usage = [
+            {"slot_id": 1, "used_g": 15.0, "color": "#161616"},
+            {"slot_id": 2, "used_g": 0.0, "color": "#888888"},
+        ]
+        results = [{"slot_id": 1, "color": "#000000"}]
+        assert _archive_colors_from_spools(usage, results) == ["#000000"]
+
+    def test_no_used_slots_returns_none(self):
+        assert _archive_colors_from_spools([], []) is None
+
+    def test_ams_fallback_results_excluded(self):
+        """AMS remain%-delta fallback results carry slot_id=None and must not
+        satisfy the match for a real 3MF slot."""
+        usage = [{"slot_id": 1, "used_g": 15.0, "color": "#161616"}]
+        results = [{"slot_id": None, "color": "#000000"}]
+        assert _archive_colors_from_spools(usage, results) is None
+
+
+class TestArchiveFilamentColorRewrite:
+    """`_track_from_3mf` overwrites the archive's filament_color with the
+    matched inventory spool colour at print completion (#1494)."""
+
+    @pytest.mark.asyncio
+    async def test_archive_color_adopts_spool_color(self):
+        """A print from a #000000 inventory spool whose 3MF says #161616 ends
+        up with the archive showing the spool's #000000."""
+        spool = _make_spool(id=5, label_weight=1000, weight_used=100, rgba="000000FF")
+        assignment = _make_assignment(spool_id=5)
+        archive = MagicMock()
+        archive.file_path = "archives/test.3mf"
+        archive.filament_color = "#161616"  # what archive.py set from the 3MF
+
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        pm = _make_printer_manager(_make_printer_state([], tray_now=0))
+        filament_usage = [{"slot_id": 1, "used_g": 25.5, "type": "PETG", "color": "#161616"}]
+
+        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_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=10,
+                status="completed",
+                print_name="test_print",
+                handled_trays=set(),
+                printer_manager=pm,
+                db=db,
+            )
+
+        assert len(results) == 1
+        assert results[0]["color"] == "#000000"
+        assert results[0]["slot_id"] == 1
+        # The archive colour was rewritten from the slicer's #161616 to the
+        # inventory spool's #000000.
+        assert archive.filament_color == "#000000"
+
+    @pytest.mark.asyncio
+    async def test_archive_color_untouched_when_spool_has_no_color(self):
+        """A spool with no rgba leaves the 3MF colour in place."""
+        spool = _make_spool(id=5, label_weight=1000, weight_used=100, rgba=None)
+        assignment = _make_assignment(spool_id=5)
+        archive = MagicMock()
+        archive.file_path = "archives/test.3mf"
+        archive.filament_color = "#161616"
+
+        db = AsyncMock()
+        db.execute = AsyncMock(
+            side_effect=[
+                MagicMock(scalar_one_or_none=MagicMock(return_value=archive)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=None)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=assignment)),
+                MagicMock(scalar_one_or_none=MagicMock(return_value=spool)),
+            ]
+        )
+
+        pm = _make_printer_manager(_make_printer_state([], tray_now=0))
+        filament_usage = [{"slot_id": 1, "used_g": 25.5, "type": "PETG", "color": "#161616"}]
+
+        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_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+
+            await _track_from_3mf(
+                printer_id=1,
+                archive_id=10,
+                status="completed",
+                print_name="test_print",
+                handled_trays=set(),
+                printer_manager=pm,
+                db=db,
+            )
+
+        assert archive.filament_color == "#161616"