bambuddy/backend/app/services/spoolman_tracking.py

"""Spoolman per-filament usage tracking for active prints.

Captures AMS tray state and G-code data at print start, then reports
per-filament usage to the correct Spoolman spools at print completion.
Supports accurate partial usage reporting for failed/cancelled prints.
"""

import json
import logging

from sqlalchemy import delete, select

from backend.app.core.config import settings as app_settings
from backend.app.core.database import async_session
from backend.app.services.spoolman import (
    SpoolmanClientError,
    SpoolmanNotFoundError,
    SpoolmanUnavailableError,
    get_spoolman_client,
    init_spoolman_client,
)

logger = logging.getLogger(__name__)

# Zero UUID used by Bambu printers for empty/unset tray_uuid
_ZERO_UUID = "00000000000000000000000000000000"
_ZERO_TAG_UID = "0000000000000000"


def _is_non_zero_identifier(value: str) -> bool:
    """Return True when identifier is non-empty and not all zeros."""
    if not value:
        return False
    return set(value) != {"0"}


def _to_fixed_hex(value: int, width: int) -> str:
    """Mirror frontend toFixedHex(): uppercase, zero-padded, fixed width."""
    safe = max(0, int(value))
    return format(safe, "X").zfill(width)[-width:]


def _hash_serial_to_hex32(serial: str) -> str:
    """Mirror frontend hashSerialToHex32() exactly (32-bit FNV-1a)."""
    input_str = (serial or "").strip().upper()
    hash_value = 0x811C9DC5
    for char in input_str:
        hash_value ^= ord(char)
        hash_value = (hash_value * 0x01000193) & 0xFFFFFFFF
    return format(hash_value, "X").zfill(8)


def _global_tray_id_to_ams_slot(global_tray_id: int) -> tuple[int, int]:
    """Convert global tray id to (ams_id, tray_id) tuple for fallback tag generation."""
    # External spool slots use IDs 254/255 and map to ams_id=255 tray_id=0/1.
    if global_tray_id >= 254:
        return 255, max(0, global_tray_id - 254)
    # AMS-HT units are addressed by ams_id directly and have a single tray.
    if global_tray_id >= 128:
        return global_tray_id, 0
    # Standard AMS units: four trays each.
    return global_tray_id // 4, global_tray_id % 4


def _get_fallback_spool_tag(printer_serial: str, global_tray_id: int) -> str:
    """Mirror frontend getFallbackSpoolTag(serial, amsId, trayId) exactly."""
    if not printer_serial:
        return ""
    ams_id, tray_id = _global_tray_id_to_ams_slot(global_tray_id)
    return get_fallback_spool_tag_for_slot(printer_serial, ams_id, tray_id)


def get_fallback_spool_tag_for_slot(printer_serial: str, ams_id: int, tray_id: int) -> str:
    """Public helper matching frontend getFallbackSpoolTag(serial, amsId, trayId).

    Used by stale-tag cleanup (#1457) to detect Spoolman spools still holding
    this slot's deterministic fallback tag in extra.tag.
    """
    if not printer_serial:
        return ""
    return f"{_hash_serial_to_hex32(printer_serial)}{_to_fixed_hex(ams_id, 4)}{_to_fixed_hex(tray_id, 4)}"


def _resolve_spool_tag(tray_info: dict, printer_serial: str = "", global_tray_id: int | None = None) -> str:
    """Get the best spool identifier from tray info (prefer tray_uuid over tag_uid).

    Returns empty string if no usable identifier is found.
    """
    tray_uuid = str(tray_info.get("tray_uuid", "") or "")
    tag_uid = str(tray_info.get("tag_uid", "") or "")

    if tray_uuid and tray_uuid != _ZERO_UUID and _is_non_zero_identifier(tray_uuid):
        return tray_uuid
    if tag_uid and tag_uid != _ZERO_TAG_UID and _is_non_zero_identifier(tag_uid):
        return tag_uid
    if global_tray_id is not None:
        return _get_fallback_spool_tag(printer_serial, global_tray_id)
    return ""


async def _get_printer_serial(printer_id: int) -> str:
    """Get printer serial for deterministic fallback tag generation."""
    from backend.app.models.printer import Printer
    from backend.app.services.printer_manager import printer_manager

    printer_info = printer_manager.get_printer(printer_id)
    if printer_info and printer_info.serial_number:
        return printer_info.serial_number

    async with async_session() as db:
        result = await db.execute(select(Printer.serial_number).where(Printer.id == printer_id))
        serial_number = result.scalar_one_or_none()
        return serial_number or ""


def _resolve_global_tray_id(slot_id: int, slot_to_tray: list | None, ams_trays: dict | None = None) -> int:
    """Map a 1-based slot_id to a global_tray_id using optional custom mapping.

    Custom mapping: slot_to_tray[slot_id - 1] is used when >= 0.
    A value of -1 in the custom mapping means the slicer routed this slot to
    the external spool. BambuStudio converts virtual tray IDs (254/255) to -1
    in the flat ams_mapping array before sending to the printer — see
    start_print() in bambu_mqtt.py which documents this convention. We mirror
    it here: when -1 is seen, look up the external spool's actual
    global_tray_id (254/255) in ams_trays rather than falling through to the
    position-based default (which would map slot_id=1 to the first AMS tray
    and credit an unrelated spool — see #1276, regression of #853).
    Position-based default: uses sorted ams_trays keys so external spools (ID 254/255)
    naturally follow standard AMS trays, matching the slicer's slot numbering.
    Final fallback: slot_id - 1 (legacy, works for pure AMS without external spools).
    """
    if slot_to_tray and slot_id <= len(slot_to_tray):
        mapped_tray = slot_to_tray[slot_id - 1]
        if mapped_tray >= 0:
            return mapped_tray
        if mapped_tray == -1 and ams_trays:
            # -1 means external spool. 254 = VIRTUAL_TRAY_DEPUTY_ID (main on
            # single-nozzle, left/deputy on H2D dual-nozzle); 255 =
            # VIRTUAL_TRAY_MAIN_ID. Prefer 254 when both exist since that's
            # what single-nozzle printers report via tray_now.
            for ext_id in (254, 255):
                if ext_id in ams_trays:
                    return ext_id
    # Position-based default: sort available tray IDs so external spools (254/255)
    # come after standard AMS trays, matching the slicer's slot assignment order.
    if ams_trays:
        sorted_tray_ids = sorted(ams_trays.keys())
        if slot_id <= len(sorted_tray_ids):
            return sorted_tray_ids[slot_id - 1]
    return slot_id - 1


def build_ams_tray_lookup(raw_data: dict) -> dict[int, dict]:
    """Build lookup of global_tray_id -> tray info from printer state.

    Returns: {0: {"tray_uuid": "...", "tag_uid": "...", "tray_type": "..."}, ...}
    """
    lookup = {}
    ams_data = raw_data.get("ams", [])
    for ams_unit in ams_data:
        ams_id = int(ams_unit.get("id", 0))
        for tray in ams_unit.get("tray", []):
            tray_id = int(tray.get("id", 0))
            # AMS-HT units have IDs starting at 128 with a single tray
            global_tray_id = ams_id if ams_id >= 128 else ams_id * 4 + tray_id
            lookup[global_tray_id] = {
                "tray_uuid": tray.get("tray_uuid", ""),
                "tag_uid": tray.get("tag_uid", ""),
                "tray_type": tray.get("tray_type", ""),
            }

    # External spool(s) (vt_tray is a list, global_tray_id from each entry's "id")
    for vt in raw_data.get("vt_tray") or []:
        if vt.get("tray_type"):
            tray_id = int(vt.get("id", 254))
            lookup[tray_id] = {
                "tray_uuid": vt.get("tray_uuid", ""),
                "tag_uid": vt.get("tag_uid", ""),
                "tray_type": vt.get("tray_type", ""),
            }

    return lookup


def _snapshot_tray_remain(raw_data: dict) -> dict[str, dict]:
    """Capture per-slot ``remain%`` + ``tray_uuid`` at print start so the
    completion path can compute a remain-delta when 3MF data doesn't cover
    the slot (or there's no 3MF at all — #1820).

    Returns ``{"<ams_id>-<tray_id>": {"remain": int, "tray_uuid": str}}``.
    Only slots whose ``remain`` is a valid 0..100 int are included; invalid
    values mean the AMS hasn't read the spool yet and a delta would be
    meaningless. Mirrors the gate in
    ``usage_tracker.on_print_start:309``.
    """
    snapshot: dict[str, dict] = {}
    ams_raw = raw_data.get("ams", [])
    ams_data = ams_raw.get("ams", []) if isinstance(ams_raw, dict) else ams_raw if isinstance(ams_raw, list) else []
    for ams_unit in ams_data:
        if not isinstance(ams_unit, dict):
            continue
        ams_id = int(ams_unit.get("id", 0))
        for tray in ams_unit.get("tray", []):
            if not isinstance(tray, dict):
                continue
            tray_id = int(tray.get("id", 0))
            remain = tray.get("remain", -1)
            if isinstance(remain, int) and 0 <= remain <= 100:
                snapshot[f"{ams_id}-{tray_id}"] = {
                    "remain": remain,
                    "tray_uuid": tray.get("tray_uuid", "") or "",
                }
    vt_tray_raw = raw_data.get("vt_tray") or []
    if isinstance(vt_tray_raw, dict):
        vt_tray_raw = [vt_tray_raw]
    for vt in vt_tray_raw:
        if not isinstance(vt, dict):
            continue
        vt_id = int(vt.get("id", 254))
        # 254 → (255, 0), 255 → (255, 1) — matches usage_tracker's encoding.
        vt_tray_id = vt_id - 254
        remain = vt.get("remain", -1)
        if isinstance(remain, int) and 0 <= remain <= 100:
            snapshot[f"255-{vt_tray_id}"] = {
                "remain": remain,
                "tray_uuid": vt.get("tray_uuid", "") or "",
            }
    return snapshot


async def store_print_data(
    printer_id: int,
    archive_id: int,
    file_path: str,
    db,
    printer_manager,
    ams_mapping: list[int] | None = None,
    plate_id: int | None = None,
):
    """Store Spoolman tracking data at print start (persisted to database).

    Per-print tracking is the primary weight-update path for Spoolman, mirroring
    how the internal Filament Inventory works. The legacy AMS-remain%-based sync
    is no longer used as a weight writer (#1119), so this runs whenever Spoolman
    is enabled regardless of the deprecated `spoolman_disable_weight_sync` flag.

    ``plate_id``, when set, scopes the 3MF filament extract to a single plate so
    queue / direct-Print dispatch of plate N of a multi-plate file doesn't
    attribute every plate's filament to the printed spool (#1697). When unset,
    the queue item's plate_id (if any) is used; otherwise the whole-file sum is
    extracted, which is correct for direct prints that target the first/only
    plate of a single-plate file.
    """
    from backend.app.api.routes.settings import get_setting
    from backend.app.models.active_print_spoolman import ActivePrintSpoolman
    from backend.app.models.print_queue import PrintQueueItem
    from backend.app.utils.threemf_tools import (
        extract_filament_properties_from_3mf,
        extract_filament_usage_from_3mf,
        extract_layer_filament_usage_from_3mf,
    )

    # Check if Spoolman is enabled
    spoolman_enabled = await get_setting(db, "spoolman_enabled")
    if not spoolman_enabled or spoolman_enabled.lower() != "true":
        return

    # Get current AMS tray state up front — needed both for the 3MF path's
    # ams_trays field and for the remain%-delta snapshot (#1820 fallback for
    # no-3MF "Untitled" prints, mirroring usage_tracker.on_print_start).
    state = printer_manager.get_status(printer_id)
    ams_trays: dict[int, dict] = {}
    tray_remain_start: dict[str, dict] = {}
    if state and state.raw_data:
        ams_trays = build_ams_tray_lookup(state.raw_data)
        tray_remain_start = _snapshot_tray_remain(state.raw_data)

    # Try to read per-slot filament estimates from the 3MF. Two paths can
    # leave ``filament_usage`` empty: (1) fallback archive (no .gcode.3mf
    # was downloadable from the printer — "Untitled" prints, see #1820),
    # (2) 3MF present but slice_info missing per-filament estimates.
    # Both fall through to the remain%-delta path at completion.
    filament_usage: list | None = None
    layer_usage_json: dict | None = None
    filament_properties: dict | None = None
    full_path = (
        app_settings.base_dir / file_path
    )  # SEC-PATH-OK: file_path is archive.file_path / library_file.file_path — DB-stored, internally generated
    threemf_available = bool(file_path) and full_path.exists()
    queue_item = None
    if threemf_available:
        # Resolve the queue item once — used both for the plate-scoped 3MF parsing
        # fallback (#1697: multi-plate file dispatched for one plate must only count
        # that plate's filament) and for the ams_mapping fallback below.
        queue_result = await db.execute(
            select(PrintQueueItem)
            .where(PrintQueueItem.archive_id == archive_id)
            .where(PrintQueueItem.status == "printing")
        )
        queue_item = queue_result.scalar_one_or_none()
        # Caller-supplied plate_id wins (direct-Print path); fall back to the queue
        # item's plate_id (queue dispatch path).
        effective_plate_id = (
            plate_id if plate_id is not None else (queue_item.plate_id if queue_item is not None else None)
        )
        filament_usage = extract_filament_usage_from_3mf(full_path, effective_plate_id) or None

        layer_usage = extract_layer_filament_usage_from_3mf(full_path)
        if layer_usage:
            # Convert int keys to string for JSON serialization
            layer_usage_json = {str(k): v for k, v in layer_usage.items()}
            logger.debug("[SPOOLMAN] Parsed %s layers from G-code", len(layer_usage))

        filament_properties = extract_filament_properties_from_3mf(full_path)
    else:
        # No 3MF on disk — common for "Untitled" prints whose .gcode.3mf
        # was never on the printer's FTP. Logged at debug since the
        # fallback path below picks up the slack when remain% is available.
        logger.debug("[SPOOLMAN] 3MF file not available: %s", full_path)

    # If neither path has anything useful, there's nothing to track.
    if not filament_usage and not tray_remain_start:
        if threemf_available:
            logger.debug("[SPOOLMAN] No filament usage data in 3MF for archive %s", archive_id)
        return

    # Prefer the explicit mapping captured from the print command, then fall back
    # to any queue mapping stored for scheduled/reprint jobs.
    slot_to_tray = ams_mapping if ams_mapping is not None else None
    if not slot_to_tray and queue_item and queue_item.ams_mapping:
        try:
            slot_to_tray = json.loads(queue_item.ams_mapping)
        except json.JSONDecodeError:
            pass  # Ignore malformed AMS mapping; fall back to default slot assignment

    # Delete any existing row for this printer/archive (shouldn't exist, but just in case)
    await db.execute(
        delete(ActivePrintSpoolman)
        .where(ActivePrintSpoolman.printer_id == printer_id)
        .where(ActivePrintSpoolman.archive_id == archive_id)
    )

    # Insert new tracking data. ``filament_usage`` may be None for the
    # no-3MF case; report_usage falls back to ``tray_remain_start``.
    tracking = ActivePrintSpoolman(
        printer_id=printer_id,
        archive_id=archive_id,
        filament_usage=filament_usage,
        ams_trays=ams_trays,
        slot_to_tray=slot_to_tray,
        layer_usage=layer_usage_json,
        filament_properties=filament_properties,
        tray_remain_start=tray_remain_start or None,
    )
    db.add(tracking)
    await db.commit()

    logger.info(
        "[SPOOLMAN] Stored tracking data for print: printer=%s, archive=%s (3mf=%s, remain_snapshot=%d slot(s))",
        printer_id,
        archive_id,
        "yes" if filament_usage else "no",
        len(tray_remain_start),
    )
    logger.debug("[SPOOLMAN] Filament usage: %s", filament_usage)
    logger.debug("[SPOOLMAN] AMS trays: %s", list(ams_trays.keys()))
    if slot_to_tray:
        logger.debug("[SPOOLMAN] Custom slot mapping: %s", slot_to_tray)
    if layer_usage_json:
        logger.debug("[SPOOLMAN] Layer usage data available for partial tracking")


async def cleanup_tracking(
    printer_id: int,
    archive_id: int,
    db,
    last_layer_num: int | None = None,
    last_progress: int | None = None,
):
    """Report partial usage and clean up Spoolman tracking data for failed/aborted prints."""
    from backend.app.models.active_print_spoolman import ActivePrintSpoolman

    # Get tracking data first (needed for partial usage reporting)
    result = await db.execute(
        select(ActivePrintSpoolman)
        .where(ActivePrintSpoolman.printer_id == printer_id)
        .where(ActivePrintSpoolman.archive_id == archive_id)
    )
    tracking = result.scalar_one_or_none()

    if not tracking:
        logger.debug("[SPOOLMAN] No tracking data to clean up for printer=%s, archive=%s", printer_id, archive_id)
        return

    # Try to report partial usage before cleanup
    try:
        await _report_partial_usage(
            printer_id,
            tracking,
            last_layer_num=last_layer_num,
            last_progress=last_progress,
        )
    except Exception as e:
        logger.warning("[SPOOLMAN] Partial usage report failed: %s", e)

    # Delete tracking data
    await db.execute(
        delete(ActivePrintSpoolman)
        .where(ActivePrintSpoolman.printer_id == printer_id)
        .where(ActivePrintSpoolman.archive_id == archive_id)
    )
    await db.commit()
    logger.debug("[SPOOLMAN] Cleaned up tracking data for printer=%s, archive=%s", printer_id, archive_id)


async def _get_spoolman_client_with_fallback():
    """Get Spoolman client, initializing from settings if needed.

    Returns (client, is_healthy) tuple. Client may be None.
    """
    client = await get_spoolman_client()
    if not client:
        async with async_session() as db:
            from backend.app.api.routes.settings import get_setting

            spoolman_url = await get_setting(db, "spoolman_url")
            if spoolman_url:
                try:
                    client = await init_spoolman_client(spoolman_url)
                except ValueError as exc:
                    logger.warning("Spoolman URL %r rejected by SSRF guard: %s", spoolman_url, exc)
                    return None

    if not client:
        return None
    if not await client.health_check():
        logger.warning("Spoolman health check failed; skipping usage reporting")
        return None

    return client


async def _resolve_spool_id_via_slot_assignment(printer_id: int, ams_id: int, tray_id: int) -> int | None:
    """Look up the Spoolman spool ID locally bound to (printer, ams, tray).

    Fallback path for #1459: when a tag-less spool was assigned via the
    Bambuddy UI, the user's deterministic fallback tag is intentionally NOT
    written to Spoolman's extra.tag (kept clean per #1457), so
    find_spool_by_tag misses. The local spoolman_slot_assignments table is
    the authoritative binding for those spools.
    """
    from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment

    async with async_session() as db:
        result = await db.execute(
            select(SpoolmanSlotAssignment.spoolman_spool_id).where(
                SpoolmanSlotAssignment.printer_id == printer_id,
                SpoolmanSlotAssignment.ams_id == ams_id,
                SpoolmanSlotAssignment.tray_id == tray_id,
            )
        )
        return result.scalar_one_or_none()


async def _report_spool_usage_for_slots(
    client,
    filament_usage_items: list[tuple[int, float]],
    ams_trays: dict[int, dict],
    slot_to_tray: list | None,
    method_label: str,
    printer_serial: str = "",
    printer_id: int | None = None,
    slot_colors_out: dict[int, str] | None = None,
    slot_materials_out: dict[int, str] | None = None,
) -> int:
    """Report usage to Spoolman for a list of (slot_id, grams) pairs.

    Resolution order per slot: (1) Spoolman extra.tag match against the
    tray's RFID or deterministic fallback tag, (2) #1459 fallback —
    local spoolman_slot_assignments table keyed by (printer_id, ams_id,
    tray_id). Without (2), tag-less spools assigned via the Bambuddy UI
    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
    for slot_id, grams_used in filament_usage_items:
        if grams_used <= 0:
            continue

        global_tray_id = _resolve_global_tray_id(slot_id, slot_to_tray, ams_trays)
        tray_info = ams_trays.get(global_tray_id)
        if not tray_info:
            logger.debug("[SPOOLMAN] Slot %s: no tray at global_tray_id %s", slot_id, global_tray_id)
            continue

        is_external = global_tray_id >= 254
        tray_type = tray_info.get("tray_type", "")
        logger.debug(
            "[SPOOLMAN] Slot %s resolved to global_tray_id %s (tray_type=%s, external=%s)",
            slot_id,
            global_tray_id,
            tray_type or "unknown",
            is_external,
        )

        spool_id_to_use: int | None = None
        resolution_path = ""
        # color_hex + material of the resolved spool's filament, for the #1494
        # archive colour rewrite and the #2563 type 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_material: str | None = None

        spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id)
        if spool_tag:
            spool = await client.find_spool_by_tag(spool_tag)
            if spool:
                spool_id_to_use = spool["id"]
                resolution_path = "tag"
                spool_color_hex = (spool.get("filament") or {}).get("color_hex")
                spool_material = (spool.get("filament") or {}).get("material")

        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)
            spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id)
            if spool_id_to_use is not None:
                resolution_path = "slot-assignment"

        if spool_id_to_use is None:
            logger.debug(
                "[SPOOLMAN] Slot %s: no spool resolved (tag=%s, no slot-assignment)",
                slot_id,
                spool_tag[:16] if spool_tag else "none",
            )
            continue

        # Record the spool's filament colour + material for the archive
        # rewrites (#1494, #2563). The slot-assignment path resolved only an
        # id, so fetch the spool once for whichever value is still missing.
        # Strictly best-effort: a fetch failure must never abort the weight
        # reporting for the remaining slots, so the catch is broad.
        if slot_colors_out is not None or slot_materials_out is not None:
            need_color = slot_colors_out is not None and spool_color_hex is None
            need_material = slot_materials_out is not None and spool_material is None
            if need_color or need_material:
                try:
                    _fil = (await client.get_spool(spool_id_to_use)).get("filament") or {}
                    if need_color:
                        spool_color_hex = _fil.get("color_hex")
                    if need_material:
                        spool_material = _fil.get("material")
                except Exception as exc:  # noqa: BLE001 — colour/material are non-critical
                    logger.debug("[SPOOLMAN] Slot %s: could not fetch spool filament: %s", slot_id, exc)
            if slot_colors_out is not None and spool_color_hex:
                slot_colors_out[slot_id] = spool_color_hex
            if slot_materials_out is not None and spool_material:
                slot_materials_out[slot_id] = spool_material

        try:
            await client.use_spool(spool_id_to_use, grams_used)
            logger.info(
                "[SPOOLMAN] %s: slot %s: %sg -> spool %s (via %s)",
                method_label,
                slot_id,
                grams_used,
                spool_id_to_use,
                resolution_path,
            )
            spools_updated += 1
        except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
            logger.warning("[SPOOLMAN] Failed to record usage for spool %s: %s", spool_id_to_use, exc)

    return spools_updated


async def _report_spool_usage_split_by_tray_changes(
    client,
    filament_usage: list[dict],
    tray_changes: list[tuple[int, int]],
    ams_trays: dict[int, dict],
    layer_usage: dict[int, dict[int, float]] | None,
    filament_properties: dict | None,
    total_layers: int,
    last_layer_num: int,
    method_label: str,
    printer_serial: str,
    printer_id: int,
    slot_colors_out: dict[int, str] | None = None,
    slot_materials_out: dict[int, str] | None = None,
) -> tuple[int, set[int]]:
    """Split each slot's grams across ``tray_changes`` and charge per-segment.

    Mirrors ``usage_tracker`` Path 1's tray-switch branch so Spoolman and
    the internal Spool inventory attribute mid-print AMS-backup switches
    identically (#1793 — reporter's origin spool was over-charged the
    whole print because this path didn't exist). ``compute_tray_split_grams``
    holds the shared segment-math; this function wraps the per-segment
    spool resolution + ``use_spool`` sink for the Spoolman side.

    Returns ``(spools_updated, handled_global_tray_ids)`` — the caller
    passes ``handled_global_tray_ids`` into the remain-delta fallback so
    a tray attributed here is not double-charged there.
    """
    from backend.app.utils.tray_split import compute_tray_split_grams

    spools_updated = 0
    handled_global_tray_ids: set[int] = set()

    for usage in filament_usage:
        slot_id = usage.get("slot_id", 0)
        total_weight = usage.get("used_g", 0)
        if total_weight <= 0 or slot_id <= 0:
            continue

        props = (filament_properties or {}).get(str(slot_id)) or (filament_properties or {}).get(slot_id) or {}
        segments = compute_tray_split_grams(
            tray_changes=tray_changes,
            total_weight=float(total_weight),
            slot_id=slot_id,
            layer_usage=layer_usage,
            density=float(props.get("density", 1.24)),
            diameter=float(props.get("diameter", 1.75)),
            total_layers=total_layers,
            last_layer_num=last_layer_num,
        )

        for seg_idx, tray_global, segment_grams in segments:
            if segment_grams <= 0:
                continue

            # Mark this tray as handled BEFORE the resolution attempt so
            # remain-delta doesn't double-charge it, even if we fail to
            # find a spool below. Matches usage_tracker behaviour: the
            # tray was physically fed from during this print, whether or
            # not Spoolman happens to have a matching row.
            handled_global_tray_ids.add(tray_global)

            tray_info = ams_trays.get(tray_global) or {}
            spool_id_to_use: int | None = None
            resolution_path = ""
            spool_color_hex: str | None = None
            spool_material: str | None = None

            spool_tag = _resolve_spool_tag(tray_info, printer_serial, tray_global) if tray_info else ""
            if spool_tag:
                spool = await client.find_spool_by_tag(spool_tag)
                if spool:
                    spool_id_to_use = spool["id"]
                    resolution_path = "tag"
                    spool_color_hex = (spool.get("filament") or {}).get("color_hex")
                    spool_material = (spool.get("filament") or {}).get("material")

            if spool_id_to_use is None:
                seg_ams_id, seg_tray_id = _global_tray_id_to_ams_slot(tray_global)
                spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, seg_ams_id, seg_tray_id)
                if spool_id_to_use is not None:
                    resolution_path = "slot-assignment"

            if spool_id_to_use is None:
                logger.info(
                    "[SPOOLMAN] Split slot %s seg %s tray=%d: no spool resolved — %.2fg lost from split accounting",
                    slot_id,
                    seg_idx,
                    tray_global,
                    segment_grams,
                )
                continue

            # Colour (#1494) + material (#2563) rewrite — first segment for a
            # slot wins. The UI displays a single colour/type per slot, so
            # later segments on the same slot don't overwrite (a backup swap
            # can differ but the archive card stays consistent with the origin).
            need_color = slot_colors_out is not None and slot_id not in slot_colors_out and spool_color_hex is None
            need_material = (
                slot_materials_out is not None and slot_id not in slot_materials_out and spool_material is None
            )
            if need_color or need_material:
                try:
                    _fil = (await client.get_spool(spool_id_to_use)).get("filament") or {}
                    if need_color:
                        spool_color_hex = _fil.get("color_hex")
                    if need_material:
                        spool_material = _fil.get("material")
                except Exception as exc:  # noqa: BLE001 — colour/material are non-critical
                    logger.debug("[SPOOLMAN] Split slot %s: could not fetch spool filament: %s", slot_id, exc)
            if slot_colors_out is not None and slot_id not in slot_colors_out and spool_color_hex:
                slot_colors_out[slot_id] = spool_color_hex
            if slot_materials_out is not None and slot_id not in slot_materials_out and spool_material:
                slot_materials_out[slot_id] = spool_material

            try:
                await client.use_spool(spool_id_to_use, round(segment_grams, 2))
                logger.info(
                    "[SPOOLMAN] %s: slot %s seg %s tray=%d: %.2fg -> spool %s (via %s)",
                    method_label,
                    slot_id,
                    seg_idx,
                    tray_global,
                    segment_grams,
                    spool_id_to_use,
                    resolution_path,
                )
                spools_updated += 1
            except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
                logger.warning(
                    "[SPOOLMAN] Split slot %s seg %s: failed to record usage for spool %s: %s",
                    slot_id,
                    seg_idx,
                    spool_id_to_use,
                    exc,
                )

    return spools_updated, handled_global_tray_ids


async def _report_partial_usage(
    printer_id: int,
    tracking,
    last_layer_num: int | None = None,
    last_progress: int | None = None,
):
    """Report partial filament usage based on actual G-code layer data.

    Uses per-layer cumulative extrusion from G-code parsing for accurate
    multi-material tracking. Falls back to linear interpolation if G-code
    data is unavailable.
    """
    from backend.app.services.printer_manager import printer_manager
    from backend.app.utils.threemf_tools import get_cumulative_usage_at_layer, mm_to_grams

    async with async_session() as db:
        from backend.app.api.routes.settings import get_setting

        # Check if partial usage reporting is enabled (default: true)
        report_partial = await get_setting(db, "spoolman_report_partial_usage")
        if report_partial and report_partial.lower() == "false":
            logger.debug("[SPOOLMAN] Partial usage reporting disabled by setting")
            return

        # Check if Spoolman is enabled
        spoolman_enabled = await get_setting(db, "spoolman_enabled")
        if not spoolman_enabled or spoolman_enabled.lower() != "true":
            return

    # Get current printer state for layer progress.
    # On failed/aborted prints the firmware may already reset to IDLE with layer=0,
    # so we fall back to completion-time hints captured from MQTT.
    state = printer_manager.get_status(printer_id)
    current_layer = state.layer_num if state else None
    total_layers = state.total_layers if state else None

    if (not current_layer or current_layer <= 0) and last_layer_num and last_layer_num > 0:
        current_layer = last_layer_num
        logger.debug("[SPOOLMAN] Using captured last_layer_num=%s for partial usage", current_layer)

    progress_ratio_from_event = None
    if last_progress is not None:
        try:
            progress_ratio_from_event = min(max(float(last_progress), 0.0), 100.0) / 100.0
        except (TypeError, ValueError):
            progress_ratio_from_event = None

    if (not current_layer or current_layer <= 0) and progress_ratio_from_event and total_layers and total_layers > 0:
        current_layer = max(1, int(round(total_layers * progress_ratio_from_event)))
        logger.debug(
            "[SPOOLMAN] Estimated layer from last_progress=%s%% and total_layers=%s -> %s",
            last_progress,
            total_layers,
            current_layer,
        )

    if not current_layer or current_layer <= 0:
        logger.debug(
            "[SPOOLMAN] No progress to report (layer 0/unknown, last_layer_num=%s, last_progress=%s)",
            last_layer_num,
            last_progress,
        )
        return

    logger.info("[SPOOLMAN] Reporting partial usage at layer %s/%s", current_layer, total_layers or "?")

    # Get tracking data
    layer_usage = tracking.layer_usage
    filament_properties = tracking.filament_properties or {}
    filament_usage = tracking.filament_usage or []
    ams_trays = {int(k): v for k, v in (tracking.ams_trays or {}).items()}
    slot_to_tray = tracking.slot_to_tray
    tray_remain_start = tracking.tray_remain_start or {}
    printer_serial = await _get_printer_serial(printer_id)

    client = await _get_spoolman_client_with_fallback()
    if not client:
        logger.warning("[SPOOLMAN] Not reachable for partial usage reporting")
        return

    # No-3MF aborted print (#1820 mirror of the completion path): nothing in
    # filament_usage or layer_usage to base partial estimates on, but the
    # remain%-delta snapshot we captured at start still describes consumption
    # up to the abort moment. Write it the same way report_usage's fallback
    # does, then return — there's no 3MF-derived partial to layer on top.
    # ``state`` was already fetched at the top of the function for current_layer.
    if not filament_usage and not layer_usage and tray_remain_start:
        current_lookup = _snapshot_tray_remain(state.raw_data) if state and state.raw_data else {}
        await _report_remain_delta_for_slots(
            client,
            printer_id=printer_id,
            tray_remain_start=tray_remain_start,
            current_lookup=current_lookup,
            handled_global_tray_ids=set(),
            archive_id=getattr(tracking, "archive_id", -1),
        )
        return

    # Try to use accurate G-code parsed data
    if layer_usage:
        layer_usage_int = {
            int(layer): {int(fid): mm for fid, mm in filaments.items()} for layer, filaments in layer_usage.items()
        }
        usage_mm = get_cumulative_usage_at_layer(layer_usage_int, current_layer)

        if usage_mm:
            logger.info("[SPOOLMAN] Using G-code parsed data for layer %s", current_layer)

            # Build (slot_id, grams) list using Spoolman densities with 3MF fallback
            usage_items = []
            for filament_id, mm_used in usage_mm.items():
                slot_id = filament_id + 1  # filament_id is 0-based, slot_id is 1-based

                # Get density from Spoolman (most accurate), fall back to 3MF, then PLA default
                global_tray_id = _resolve_global_tray_id(slot_id, slot_to_tray, ams_trays)
                tray_info = ams_trays.get(global_tray_id)
                density = None
                diameter = 1.75

                if tray_info:
                    spool_tag = _resolve_spool_tag(tray_info, printer_serial, global_tray_id)
                    if spool_tag:
                        spool = await client.find_spool_by_tag(spool_tag)
                        if spool:
                            filament_data = spool.get("filament", {})
                            density = filament_data.get("density")
                            diameter = filament_data.get("diameter", 1.75)

                if not density:
                    props = filament_properties.get(str(slot_id), filament_properties.get(slot_id, {}))
                    density = props.get("density", 1.24)
                    logger.debug("[SPOOLMAN] Using fallback density %s for slot %s", density, slot_id)

                grams_used = round(mm_to_grams(mm_used, diameter, density), 2)
                usage_items.append((slot_id, grams_used))

            spools_updated = await _report_spool_usage_for_slots(
                client,
                usage_items,
                ams_trays,
                slot_to_tray,
                "Partial (G-code)",
                printer_serial,
                printer_id=printer_id,
            )
            if spools_updated > 0:
                logger.info("[SPOOLMAN] Reported partial usage to %s spool(s) using G-code data", spools_updated)
            return

    # Fallback: linear interpolation (if no G-code data available)
    progress_ratio = None
    if total_layers and total_layers > 0:
        progress_ratio = min(current_layer / total_layers, 1.0)
    elif progress_ratio_from_event is not None:
        progress_ratio = progress_ratio_from_event

    if progress_ratio is None:
        logger.debug(
            "[SPOOLMAN] Cannot use linear fallback: total_layers=%s, last_progress=%s",
            total_layers,
            last_progress,
        )
        return

    logger.info("[SPOOLMAN] Falling back to linear interpolation (%s)", progress_ratio)

    usage_items = []
    for usage in filament_usage:
        slot_id = usage.get("slot_id", 0)
        total_used_g = usage.get("used_g", 0)
        if total_used_g > 0:
            partial_used_g = round(total_used_g * progress_ratio, 2)
            usage_items.append((slot_id, partial_used_g))

    spools_updated = await _report_spool_usage_for_slots(
        client,
        usage_items,
        ams_trays,
        slot_to_tray,
        "Partial (linear)",
        printer_serial,
        printer_id=printer_id,
    )
    if spools_updated > 0:
        logger.info("[SPOOLMAN] Reported partial usage to %s spool(s) using linear interpolation", spools_updated)


async def report_usage(printer_id: int, archive_id: int):
    """Report filament usage to Spoolman after print completion.

    Two writers, mirroring the internal-inventory split in usage_tracker:

    1. **3MF path (primary)** — per-filament slice estimates captured at
       print start drive a precise per-slot ``use_spool`` call.
    2. **AMS remain%-delta (fallback)** — for slots the 3MF path didn't
       handle (including the no-3MF "Untitled" case from #1820): compute
       ``start_remain - current_remain``, multiply by the resolved
       Spoolman filament's reference weight, and write the delta. Mirrors
       ``usage_tracker.on_print_complete`` Path 2 (line 517).
    """
    async with async_session() as db:
        from backend.app.api.routes.settings import get_setting
        from backend.app.models.active_print_spoolman import ActivePrintSpoolman

        # Get tracking data stored at print start
        result = await db.execute(
            select(ActivePrintSpoolman)
            .where(ActivePrintSpoolman.printer_id == printer_id)
            .where(ActivePrintSpoolman.archive_id == archive_id)
        )
        tracking = result.scalar_one_or_none()

        if not tracking:
            logger.info("[SPOOLMAN] No tracking data for print (printer=%s, archive=%s)", printer_id, archive_id)
            return

        filament_usage = tracking.filament_usage or []
        ams_trays = {int(k): v for k, v in (tracking.ams_trays or {}).items()}
        slot_to_tray = tracking.slot_to_tray
        tray_remain_start = tracking.tray_remain_start or {}
        # ``layer_usage`` and ``filament_properties`` were added later than
        # the base tracking fields; use ``getattr`` so tests that stub
        # ``tracking`` as a lightweight SimpleNamespace stay valid, and
        # historic ORM rows loaded without these columns can't AttributeError
        # on read.
        layer_usage_raw = getattr(tracking, "layer_usage", None) or {}
        filament_properties = getattr(tracking, "filament_properties", None) or {}
        printer_serial = await _get_printer_serial(printer_id)

        # Delete tracking row (we're done with it)
        await db.delete(tracking)
        await db.commit()

        if not filament_usage and not tray_remain_start:
            logger.debug("[SPOOLMAN] No usage data or remain-snapshot for archive %s", archive_id)
            return

        # Check if Spoolman is enabled
        spoolman_enabled = await get_setting(db, "spoolman_enabled")
        if not spoolman_enabled or spoolman_enabled.lower() != "true":
            return

        client = await _get_spoolman_client_with_fallback()
        if not client:
            logger.warning("[SPOOLMAN] Not reachable for usage reporting")
            return

        # Consult the live printer state for the tray-change log written by
        # ``bambu_mqtt.py`` on every mid-print ``tray_now`` change (#957).
        # When there's more than one entry, the print traversed >1 AMS tray
        # and the split path attributes each segment to the tray that was
        # loaded at the time — matches the internal Spool inventory writer
        # in ``usage_tracker.py``. Without this, an AMS-backup runout switch
        # charges the whole slot to the origin spool and pushes it past
        # ``initial_weight`` (#1793).
        #
        # Split only for SINGLE-slot prints — same gate as
        # ``usage_tracker.py:1002``. Multi-slot (multi-colour) prints
        # naturally cycle trays for every colour change, so splitting each
        # slot's grams across ALL tray_change_log entries would attribute
        # slot 1's grams to the segments where slot 2's tray was loaded and
        # vice versa. Multi-slot prints fall through to the existing
        # single-tray path (which uses the stable ``slot_to_tray`` mapping).
        nonzero_slots = [u for u in filament_usage if u.get("used_g", 0) > 0]
        tray_changes: list[tuple[int, int]] = []
        _state = None
        if len(nonzero_slots) == 1:
            from backend.app.services.printer_manager import printer_manager as _pm

            _state = _pm.get_status(printer_id)
            if _state is not None:
                tray_changes = list(getattr(_state, "tray_change_log", []) or [])
        _total_layers = int(getattr(_state, "total_layers", 0) or 0) if _state else 0
        _current_layer = int(getattr(_state, "layer_num", 0) or 0) if _state else 0
        # For the linear-fallback denominator when total_layers is 0 (P1S
        # firmware resets it at print end). At completion the current layer
        # is the print's last valid layer.
        _layer_denom_hint = _total_layers or _current_layer

        slot_colors: dict[int, str] = {}
        slot_materials: dict[int, str] = {}
        handled_global_tray_ids: set[int] = set()
        spools_updated = 0

        # --- Path 1: 3MF per-slot estimates -----------------------------
        if filament_usage:
            if len(tray_changes) > 1:
                # Tray-split path — attribute per-segment to the tray that
                # was loaded at that time.
                logger.info(
                    "[SPOOLMAN] Reporting per-filament usage for archive %s with tray-split "
                    "(tray_change_log=%s, denom_layers=%d)",
                    archive_id,
                    tray_changes,
                    _layer_denom_hint,
                )
                # ``tracking.layer_usage`` was serialized to JSON so int keys
                # come back as strings. Restore them for the split math.
                layer_usage = None
                if layer_usage_raw:
                    try:
                        layer_usage = {
                            int(layer): {int(fid): mm for fid, mm in filaments.items()}
                            for layer, filaments in layer_usage_raw.items()
                        }
                    except (TypeError, ValueError, AttributeError):
                        # AttributeError catches ``inner.items()`` when the
                        # inner value isn't dict-shaped (corrupt JSON row).
                        # Missing gcode falls through to the linear-ratio
                        # branch inside ``compute_tray_split_grams`` — still
                        # gives a correct split, just less precise.
                        layer_usage = None
                split_updated, split_handled = await _report_spool_usage_split_by_tray_changes(
                    client,
                    filament_usage,
                    tray_changes,
                    ams_trays,
                    layer_usage,
                    filament_properties,
                    _total_layers,
                    _layer_denom_hint,
                    f"Archive {archive_id}",
                    printer_serial,
                    printer_id=printer_id,
                    slot_colors_out=slot_colors,
                    slot_materials_out=slot_materials,
                )
                spools_updated += split_updated
                handled_global_tray_ids |= split_handled
            else:
                logger.info("[SPOOLMAN] Reporting per-filament usage for archive %s", archive_id)
                usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage]
                spools_updated = await _report_spool_usage_for_slots(
                    client,
                    usage_items,
                    ams_trays,
                    slot_to_tray,
                    f"Archive {archive_id}",
                    printer_serial,
                    printer_id=printer_id,
                    slot_colors_out=slot_colors,
                    slot_materials_out=slot_materials,
                )
                # Track which physical slots the 3MF path already covered so
                # Path 2 doesn't double-charge them.
                for u in filament_usage:
                    slot_id = u.get("slot_id", 0)
                    handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays))

        # --- Path 2: AMS remain%-delta for slots 3MF didn't cover -------
        # Triggered for no-3MF "Untitled" prints (#1820) AND for partial
        # 3MF coverage (slots whose filament_id wasn't in slice_info).
        if tray_remain_start:
            from backend.app.services.printer_manager import printer_manager

            current = printer_manager.get_status(printer_id)
            current_lookup = _snapshot_tray_remain(current.raw_data) if current and current.raw_data else {}
            fallback_updates = await _report_remain_delta_for_slots(
                client,
                printer_id=printer_id,
                tray_remain_start=tray_remain_start,
                current_lookup=current_lookup,
                handled_global_tray_ids=handled_global_tray_ids,
                archive_id=archive_id,
                slot_colors_out=slot_colors,
                slot_materials_out=slot_materials,
            )
            spools_updated += fallback_updates

        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)

        # Same for the material: a slot mapped to a differently-typed spool than
        # it was sliced for otherwise records the sliced type (#2563).
        await _apply_spool_types_to_archive(db, archive_id, filament_usage, slot_materials)


async def _report_remain_delta_for_slots(
    client,
    *,
    printer_id: int,
    tray_remain_start: dict[str, dict],
    current_lookup: dict[str, dict],
    handled_global_tray_ids: set[int],
    archive_id: int,
    slot_colors_out: dict[int, str] | None = None,
    slot_materials_out: dict[int, str] | None = None,
) -> int:
    """AMS remain%-delta path: write ``(start - current) * filament.weight``
    grams to Spoolman for slots the 3MF path didn't cover.

    Mirrors ``usage_tracker.on_print_complete`` Path 2: per-slot, gated on a
    valid current ``remain%``, skipped on spool swap (``tray_uuid`` changed),
    using the resolved spool's filament reference weight rather than MQTT's
    unreliable ``tray_weight`` (which is the failure mode #1119 documented).
    """
    spools_updated = 0
    for slot_key, start in tray_remain_start.items():
        try:
            ams_id_str, tray_id_str = slot_key.split("-", 1)
            ams_id, tray_id = int(ams_id_str), int(tray_id_str)
        except (ValueError, AttributeError):
            continue

        # Skip slots already handled by the 3MF path. Encoding mirrors
        # build_ams_tray_lookup: VT trays land at 254/255, AMS-HT keeps
        # its native id (>=128), regular AMS slots are ams_id*4+tray_id.
        if ams_id == 255:
            global_tray_id = 254 + tray_id
        elif ams_id >= 128:
            global_tray_id = ams_id
        else:
            global_tray_id = ams_id * 4 + tray_id
        if global_tray_id in handled_global_tray_ids:
            continue

        current = current_lookup.get(slot_key)
        if not current:
            logger.debug("[SPOOLMAN] AMS%d-T%d: no current remain%% at completion, skipping fallback", ams_id, tray_id)
            continue

        # Spool swap mid-print — tray_uuid changed. We don't know how much
        # of the print went to which spool; skip rather than mis-attribute.
        start_uuid = (start.get("tray_uuid") or "").lower()
        cur_uuid = (current.get("tray_uuid") or "").lower()
        if start_uuid and cur_uuid and start_uuid != cur_uuid:
            logger.info(
                "[SPOOLMAN] AMS%d-T%d: spool swapped mid-print (uuid changed), skipping remain-delta", ams_id, tray_id
            )
            continue

        delta_pct = start["remain"] - current["remain"]
        if delta_pct <= 0:
            continue  # No consumption captured at AMS granularity, or refilled

        spool_id = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id)
        if spool_id is None:
            logger.debug("[SPOOLMAN] AMS%d-T%d: no Spoolman slot assignment, skipping fallback", ams_id, tray_id)
            continue

        # Look up the spool's filament reference weight. Use a fresh GET so
        # we don't depend on a stale cached_spools list. Failure here is
        # silent-skip rather than fatal — other slots can still be written.
        try:
            spool = await client.get_spool(spool_id)
        except Exception as exc:  # noqa: BLE001
            logger.debug("[SPOOLMAN] AMS%d-T%d: get_spool(%s) failed: %s", ams_id, tray_id, spool_id, exc)
            continue
        filament = spool.get("filament") or {}
        ref_weight = filament.get("weight")
        if not ref_weight or ref_weight <= 0:
            logger.debug(
                "[SPOOLMAN] AMS%d-T%d: spool %s has no filament.weight, skipping remain-delta",
                ams_id,
                tray_id,
                spool_id,
            )
            continue

        grams_used = round((delta_pct / 100.0) * ref_weight, 2)
        if grams_used <= 0:
            continue
        try:
            await client.use_spool(spool_id, grams_used)
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "[SPOOLMAN] AMS%d-T%d: use_spool(%s, %.2fg) failed: %s", ams_id, tray_id, spool_id, grams_used, exc
            )
            continue

        spools_updated += 1
        # No 3MF slot_id for this path — use the AMS slot key so the maps can
        # still be inspected by callers if needed. The archive rewrites
        # (#1494 colour, #2563 type) key on 3MF slot_ids, so remain-delta-only
        # prints intentionally don't participate (matches usage_tracker's
        # slot_id=None).
        if slot_colors_out is not None:
            color = filament.get("color_hex")
            if color:
                slot_colors_out[-(global_tray_id + 1)] = color
        if slot_materials_out is not None:
            material = filament.get("material")
            if material:
                slot_materials_out[-(global_tray_id + 1)] = material
        logger.info(
            "[SPOOLMAN] Archive %s AMS%d-T%d: %.2fg via remain-delta (%d%% of %.0fg) -> spool %s",
            archive_id,
            ams_id,
            tray_id,
            grams_used,
            delta_pct,
            ref_weight,
            spool_id,
        )
    return spools_updated


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()


async def _apply_spool_types_to_archive(
    db,
    archive_id: int,
    filament_usage: list[dict],
    slot_materials: dict[int, str],
) -> None:
    """Overwrite an archive's ``filament_type`` with the materials of the
    Spoolman spools that fed the print (#2563).

    All-or-nothing, exactly like the colour path and the built-in inventory
    path: the type is only rewritten when every used slot resolved to a spool
    that carries a material, so a partial match never drops slots from the
    archive or the material statistics.
    """
    if not slot_materials:
        return

    from backend.app.models.archive import PrintArchive
    from backend.app.services.usage_tracker import _archive_types_from_spools

    results = [{"slot_id": sid, "material": material} for sid, material in slot_materials.items()]
    types = _archive_types_from_spools(filament_usage, results)
    if not types:
        return

    archive = (await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))).scalar_one_or_none()
    if archive is None:
        return

    joined = ",".join(types)
    if joined != archive.filament_type:
        logger.info(
            "[SPOOLMAN] Archive %s filament_type %r -> %r (from Spoolman spools)",
            archive_id,
            archive.filament_type,
            joined,
        )
        archive.filament_type = joined
        await db.commit()