Put back a slot's K-profile when the printer loses it (#3219)

backend/app/api/routes/printers.py

@@ -49,7 +49,7 @@ from backend.app.schemas.printer import (
     PrintOptionsResponse,
     hms_error_responses,
 )
-from backend.app.services import drying_preflight
+from backend.app.services import drying_preflight, kprofile_drift
 from backend.app.services.bambu_ftp import (
     cache_3mf_download,
     delete_file_async,
@@ -3138,6 +3138,9 @@ async def configure_ams_slot(
         filament_id=filament_id_for_kprofile,
         nozzle_diameter=nozzle_diameter,
     )
+    # A deliberate Default pick must not be "restored" to the spool's stored
+    # profile by the lost-selection check (#3219).
+    kprofile_drift.note_slot_configured(printer_id, ams_id, tray_id, cali_idx)
 
     # Method 2: Only send extrusion_cali_set when NO existing profile was selected
     # (cali_idx == -1). When cali_idx >= 0, extrusion_cali_sel already selected the

backend/app/main.py

@@ -88,7 +88,7 @@ from backend.app.core.config import APP_VERSION, settings as app_settings
 from backend.app.core.database import async_session, engine, init_db
 from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
-from backend.app.services import print_dispatch_context, slot_unlink_grace
+from backend.app.services import kprofile_drift, print_dispatch_context, slot_unlink_grace
 from backend.app.services.archive import ArchiveService, peek_plate_index_in_3mf, swap_plate_suffix
 from backend.app.services.archive_purge import archive_purge_service
 from backend.app.services.bambu_ftp import (
@@ -1615,6 +1615,22 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
     elif not state.connected and _printer_kprofiles_primed_since_connect.get(printer_id, False):
         _printer_kprofiles_primed_since_connect[printer_id] = False
 
+    # A slot can lose its K-profile selection (cali_idx back to -1) with nothing
+    # else in the AMS report changing -- a power cycle does it to every slot --
+    # and on_ams_change never hears of it (#3219). Checked here, on every push
+    # while the printer is idle; needs_check is cheap and throttled per slot.
+    # Guarded: nothing here may stop the status broadcast below.
+    try:
+        if not state.connected:
+            kprofile_drift.forget_printer(printer_id)
+        elif kprofile_drift.needs_check(printer_id, state):
+            spawn_background_task(
+                kprofile_drift.reapply_lost_kprofiles(printer_id),
+                name=f"reapply-kprofiles-{printer_id}",
+            )
+    except Exception:
+        logging.getLogger(__name__).exception("[Printer %s] K-profile check failed", printer_id)
+
     # Offline-notification edge (#1752): schedule `on_printer_offline` on
     # connected → disconnected. The "back online" channel is already covered
     # by the print-failure notification (firmware reports gcode_state=FAILED

backend/app/services/kprofile_drift.py

@@ -0,0 +1,300 @@
+"""Put a slot's stored K-profile back when the printer has lost the selection.
+
+A tray's ``cali_idx`` can fall back to -1 without anything else in the AMS
+report changing: an X1 Carbon power-cycled mid-print came back with every slot
+on the default K while spools, tags and remain% were exactly as before (#3219).
+``on_ams_change`` is keyed on tray_type / tag_uid / remain, so it never saw
+that, and two queued jobs went out on the default K.
+
+This module re-selects the stored profile for such a slot. It acts only when
+the selection is *lost* (``cali_idx`` -1 or missing), never when the slot
+holds a different real profile: that one was picked on purpose, in Bambu
+Studio for example. A profile picked in Bambuddy's Configure Slot dialog is
+saved as the spool's stored profile anyway, but a "Default" pick there is not
+saved anywhere, so it is recorded through ``note_slot_configured`` and left
+alone too.
+
+Two callers: the status handler, on every push while the printer is idle, and
+the print scheduler, just before it sends a job.
+"""
+
+import logging
+import time
+from dataclasses import dataclass
+
+from backend.app.core.database import async_session
+from backend.app.services.printer_manager import printer_manager
+from backend.app.services.slot_kprofile import find_slot_kprofile_for_extruder
+from backend.app.services.slot_nozzle import resolve_slot_nozzle
+from backend.app.services.spool_filament_preset import printer_safe_filament_id
+from backend.app.utils.printer_models import is_dual_nozzle_model
+
+logger = logging.getLogger(__name__)
+
+# States in which re-selecting a profile cannot touch a print. PREPARE and
+# SLICING are left out on purpose: the printer is starting a job there.
+IDLE_STATES = frozenset({"IDLE", "FINISH", "FAILED"})
+
+# Time between two re-selections of one slot: the push right after a send can
+# still show the old value, so an immediate resend would only be noise.
+RETRY_INTERVAL_S = 30.0
+# Time before a slot with no stored profile is looked up again. Most slots on
+# the default K are there on purpose, and on an install with many printers a
+# lookup per slot every 30 seconds would add up for nothing. A reconnect (the
+# #3219 case) clears this, so a power cycle is still caught at once.
+NO_PROFILE_RECHECK_S = 600.0
+# Re-selections of one profile before giving up on it. A stored cali_idx the
+# printer no longer has will never stick; this keeps that from repeating forever.
+MAX_ATTEMPTS = 3
+
+SlotKey = tuple[int, int, int]
+
+
+@dataclass
+class _SlotWatch:
+    next_check: float = 0.0
+    attempts: int = 0
+    gave_up_logged: bool = False
+
+
+_watches: dict[SlotKey, _SlotWatch] = {}
+_default_chosen: set[SlotKey] = set()
+_running: set[int] = set()
+
+
+def note_slot_configured(printer_id: int, ams_id: int, tray_id: int, cali_idx: int) -> None:
+    """Record a K-profile picked for a slot in Bambuddy's Configure Slot dialog.
+
+    A default pick (-1) is respected until the slot is emptied or configured
+    again. It is not cleared by the slot reporting a real profile, because the
+    push right after the pick can still carry the old one.
+    """
+    key = (printer_id, ams_id, tray_id)
+    if cali_idx < 0:
+        _default_chosen.add(key)
+    else:
+        _default_chosen.discard(key)
+
+
+def forget_printer(printer_id: int) -> None:
+    """Drop the retry bookkeeping for a printer, e.g. when it disconnects.
+
+    A reconnect is a fresh start: a profile given up on before may stick now.
+    A deliberate default pick is kept -- it still describes the slot.
+    """
+    for key in [k for k in _watches if k[0] == printer_id]:
+        del _watches[key]
+
+
+def _selection_lost(tray: dict) -> bool:
+    cali_idx = tray.get("cali_idx")
+    if cali_idx is None:
+        return True
+    try:
+        return int(cali_idx) < 0
+    except (TypeError, ValueError):
+        return False
+
+
+def _loaded_trays(state) -> list[tuple[int, int, int, dict]]:
+    """``(ams_id, tray_id, global_tray_id, tray)`` for every loaded slot.
+
+    External spools use the assignment addressing (ams 255, tray 0/1) and the
+    global id the print command carries for them (254/255).
+    """
+    raw = getattr(state, "raw_data", None) or {}
+    ams_raw = raw.get("ams")
+    units = ams_raw.get("ams", []) if isinstance(ams_raw, dict) else ams_raw if isinstance(ams_raw, list) else []
+    trays: list[tuple[int, int, int, dict]] = []
+    for unit in units:
+        if not isinstance(unit, dict):
+            continue
+        try:
+            ams_id = int(unit.get("id", 0))
+        except (TypeError, ValueError):
+            continue
+        for tray in unit.get("tray") or []:
+            if not isinstance(tray, dict):
+                continue
+            try:
+                tray_id = int(tray.get("id", 0))
+            except (TypeError, ValueError):
+                continue
+            global_id = ams_id if ams_id >= 128 else ams_id * 4 + tray_id
+            trays.append((ams_id, tray_id, global_id, tray))
+
+    vt_raw = raw.get("vt_tray")
+    if isinstance(vt_raw, dict):
+        vt_raw = [vt_raw]
+    for vt in vt_raw if isinstance(vt_raw, list) else []:
+        if not isinstance(vt, dict):
+            continue
+        try:
+            vt_id = int(vt.get("id", 254))
+        except (TypeError, ValueError):
+            continue
+        if vt_id in (254, 255):
+            trays.append((255, vt_id - 254, vt_id, vt))
+    return trays
+
+
+def _settle(key: SlotKey, tray: dict) -> bool:
+    """Clear bookkeeping a slot no longer needs; True if it is a candidate.
+
+    A candidate is a loaded slot whose selection is lost and was not set to
+    the default on purpose.
+    """
+    if not tray.get("tray_type"):
+        _watches.pop(key, None)
+        _default_chosen.discard(key)
+        return False
+    if not _selection_lost(tray):
+        _watches.pop(key, None)
+        return False
+    return key not in _default_chosen
+
+
+def _due(printer_id: int, key: SlotKey, now: float) -> bool:
+    watch = _watches.get(key)
+    if watch is None:
+        return True
+    if now < watch.next_check:
+        return False
+    if watch.attempts < MAX_ATTEMPTS:
+        return True
+    # The last re-selection had its interval to land and the slot is still
+    # lost: say so once, then leave the slot alone until it changes.
+    if not watch.gave_up_logged:
+        watch.gave_up_logged = True
+        logger.warning(
+            "[Printer %s] AMS%d-T%d still has no K-profile after %d re-selections of the stored "
+            "one; giving up until the slot changes. Check that the profile still exists on the printer.",
+            printer_id,
+            key[1],
+            key[2],
+            MAX_ATTEMPTS,
+        )
+    return False
+
+
+def needs_check(printer_id: int, state) -> bool:
+    """Cheap test the status handler runs on every push before spawning a check."""
+    if printer_id in _running:
+        return False
+    if not getattr(state, "connected", False):
+        return False
+    if (getattr(state, "state", "") or "").upper() not in IDLE_STATES:
+        return False
+    now = time.monotonic()
+    found = False
+    for ams_id, tray_id, _global_id, tray in _loaded_trays(state):
+        key = (printer_id, ams_id, tray_id)
+        if _settle(key, tray) and _due(printer_id, key, now):
+            found = True
+    return found
+
+
+async def reapply_lost_kprofiles(
+    printer_id: int, global_trays: set[int] | None = None, *, throttle: bool = True
+) -> int:
+    """Re-select the stored K-profile on every slot whose selection was lost.
+
+    ``global_trays`` limits the pass to the trays a print uses. ``throttle``
+    off is for the dispatch guard: a job about to start must not wait out a
+    retry interval, and that runs once per job anyway. Returns how many slots
+    were re-selected.
+    """
+    # The dispatch guard runs regardless: a pass already in flight may have
+    # skipped this job's trays on its retry interval.
+    if throttle and printer_id in _running:
+        return 0
+    state = printer_manager.get_status(printer_id)
+    client = printer_manager.get_client(printer_id)
+    if state is None or client is None or not getattr(state, "connected", False):
+        return 0
+
+    model = printer_manager.get_model(printer_id)
+    nozzles = getattr(state, "nozzles", None) or []
+    dual_nozzle = is_dual_nozzle_model(model) or (len(nozzles) > 1 and bool(getattr(nozzles[1], "nozzle_diameter", "")))
+    now = time.monotonic()
+    candidates = []
+    for ams_id, tray_id, global_id, tray in _loaded_trays(state):
+        key = (printer_id, ams_id, tray_id)
+        if not _settle(key, tray):
+            continue
+        if global_trays is not None and global_id not in global_trays:
+            continue
+        if throttle and not _due(printer_id, key, now):
+            continue
+        # Claimed before the first await, so a push arriving meanwhile does
+        # not pick the same slot up a second time.
+        _watches.setdefault(key, _SlotWatch()).next_check = now + RETRY_INTERVAL_S
+        candidates.append((key, ams_id, tray_id, tray))
+    if not candidates:
+        return 0
+
+    applied = 0
+    if throttle:
+        _running.add(printer_id)
+    try:
+        async with async_session() as db:
+            for key, ams_id, tray_id, tray in candidates:
+                slot_nozzle = resolve_slot_nozzle(state, ams_id, tray_id, model)
+                if dual_nozzle:
+                    if slot_nozzle.extruder is None:
+                        # Unknown nozzle on a dual machine: picking extruder 0
+                        # would be a guess, and a wrong guess binds the other
+                        # hotend's calibration.
+                        continue
+                    extruders = [slot_nozzle.extruder]
+                else:
+                    # One nozzle, but not one extruder number: the external
+                    # holder resolves to 1 (it is H2D's Ext-L), and Configure
+                    # Slot stores the profile under that, while the spool form
+                    # stores 0. Both mean the only nozzle there is.
+                    extruders = list(dict.fromkeys([0, slot_nozzle.extruder_or_default]))
+                profile = None
+                for extruder in extruders:
+                    profile = await find_slot_kprofile_for_extruder(
+                        db,
+                        printer_id,
+                        ams_id,
+                        tray_id,
+                        extruder,
+                        slot_nozzle.diameter,
+                        model,
+                        slot_nozzle.flow,
+                    )
+                    if profile is not None:
+                        break
+                if profile is None or profile.cali_idx is None or profile.cali_idx < 0:
+                    _watches.setdefault(key, _SlotWatch()).next_check = now + NO_PROFILE_RECHECK_S
+                    continue
+
+                client.extrusion_cali_sel(
+                    ams_id=ams_id,
+                    tray_id=tray_id,
+                    cali_idx=profile.cali_idx,
+                    filament_id=printer_safe_filament_id(profile.filament_id, tray.get("tray_info_idx")),
+                    nozzle_diameter=slot_nozzle.diameter,
+                )
+                applied += 1
+                watch = _watches.setdefault(key, _SlotWatch())
+                watch.attempts += 1
+                logger.info(
+                    "[Printer %s] AMS%d-T%d lost its K-profile (cali_idx=%s): re-selected stored "
+                    "profile %r (cali_idx=%d), attempt %d",
+                    printer_id,
+                    ams_id,
+                    tray_id,
+                    tray.get("cali_idx"),
+                    profile.name,
+                    profile.cali_idx,
+                    watch.attempts,
+                )
+    except Exception:
+        logger.exception("[Printer %s] K-profile re-selection failed", printer_id)
+    finally:
+        if throttle:
+            _running.discard(printer_id)
+    return applied

backend/app/services/print_scheduler.py

@@ -30,7 +30,7 @@ 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 import drying_preflight, print_dispatch_context, stock_forecast
+from backend.app.services import drying_preflight, kprofile_drift, print_dispatch_context, stock_forecast
 from backend.app.services.bambu_ftp import (
     FtpFailureKind,
     FtpFailureReport,
@@ -8197,6 +8197,16 @@ class PrintScheduler:
                     ams_mapping,
                 )
 
+        # A slot that lost its K-profile selection (a power cycle resets every
+        # slot to the default K) would print on the default. The idle check in
+        # the status handler normally restores it first, but nothing guarantees
+        # it ran before this dispatch (#3219). Unthrottled: once per job.
+        # Never allowed to stop the print it is protecting.
+        try:
+            await kprofile_drift.reapply_lost_kprofiles(item.printer_id, _used_global_tray_ids(item), throttle=False)
+        except Exception:
+            logger.exception("Queue item %s: K-profile check before dispatch failed", item.id)
+
         # Start the print with AMS mapping, plate_id and print options.
         # nozzle_mapping rides through verbatim — JSON string captured from
         # Bambu Studio's project_file on VP intake (#1780); the MQTT layer

backend/tests/integration/test_printers_api.py

@@ -1630,6 +1630,41 @@ class TestConfigureAMSSlotAPI:
             call_kwargs = mock_client.ams_set_filament_setting.call_args
             assert call_kwargs.kwargs["tray_info_idx"] == "GFL05"
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    @pytest.mark.parametrize("cali_idx", [-1, 9354])
+    async def test_configure_records_the_k_profile_pick(self, async_client: AsyncClient, printer_factory, cali_idx):
+        """A Default pick must not be undone by the lost-selection check (#3219)."""
+        printer = await printer_factory(name="X1C")
+
+        mock_client = MagicMock()
+        mock_client.ams_set_filament_setting.return_value = True
+        mock_client.extrusion_cali_sel.return_value = True
+        mock_client.request_status_update.return_value = True
+
+        with (
+            patch("backend.app.api.routes.printers.printer_manager") as mock_pm,
+            patch("backend.app.api.routes.printers.kprofile_drift.note_slot_configured") as note,
+        ):
+            mock_pm.get_client.return_value = mock_client
+            mock_pm.get_status.return_value = None
+
+            response = await async_client.post(
+                f"/api/v1/printers/{printer.id}/slots/1/2/configure",
+                params={
+                    "tray_info_idx": "GFL05",
+                    "tray_type": "PLA",
+                    "tray_sub_brands": "PLA Basic",
+                    "tray_color": "FFFFFFFF",
+                    "nozzle_temp_min": 190,
+                    "nozzle_temp_max": 230,
+                    "cali_idx": cali_idx,
+                },
+            )
+
+            assert response.status_code == 200
+            note.assert_called_once_with(printer.id, 1, 2, cali_idx)
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_configure_builtin_empty_setting_id_is_derived(self, async_client: AsyncClient, printer_factory):

backend/tests/unit/services/test_kprofile_drift_3219.py

@@ -0,0 +1,436 @@
+"""A slot that loses its K-profile selection gets the stored one back (#3219).
+
+An X1 Carbon power-cycled mid-print came back with every AMS slot on
+``cali_idx: -1`` while tags, spools and remain% were unchanged. ``on_ams_change``
+never fired, so nothing restored the selections, and two queued jobs printed
+on the default K.
+"""
+
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services import kprofile_drift
+from backend.app.services.slot_kprofile import SlotKProfile
+
+PRINTER = 7
+
+
+def _tray(tray_id, cali_idx=-1, tray_type="PLA"):
+    return {"id": str(tray_id), "tray_type": tray_type, "cali_idx": cali_idx, "tray_info_idx": "GFA00"}
+
+
+def _state(trays, *, state="IDLE", connected=True, vt_tray=None, model_nozzles=1):
+    raw = {"ams": {"ams": [{"id": "0", "tray": trays}]}}
+    if vt_tray is not None:
+        raw["vt_tray"] = vt_tray
+    nozzles = [SimpleNamespace(nozzle_diameter="0.4", nozzle_type="") for _ in range(model_nozzles)]
+    return SimpleNamespace(
+        raw_data=raw,
+        connected=connected,
+        state=state,
+        nozzles=nozzles,
+        ams_extruder_map=None,
+        ams_switch_inlet=None,
+    )
+
+
+def _profile(cali_idx, name="PLA Basic 0.025"):
+    return SlotKProfile(cali_idx=cali_idx, k_value=0.025, name=name, extruder=0, filament_id="GFA00")
+
+
+@pytest.fixture(autouse=True)
+def _clean_bookkeeping():
+    kprofile_drift._watches.clear()
+    kprofile_drift._default_chosen.clear()
+    kprofile_drift._running.clear()
+    yield
+    kprofile_drift._watches.clear()
+    kprofile_drift._default_chosen.clear()
+    kprofile_drift._running.clear()
+
+
+@pytest.fixture
+def env():
+    """Patch the printer, the database and the stored-profile lookup."""
+    client = MagicMock()
+    stored: dict[tuple[int, int], SlotKProfile] = {}
+    holder = SimpleNamespace(state=None, client=client, stored=stored, model="X1C")
+
+    async def lookup(_db, _printer_id, ams_id, tray_id, *_args, **_kwargs):
+        return stored.get((ams_id, tray_id))
+
+    @asynccontextmanager
+    async def session():
+        yield MagicMock()
+
+    pm = MagicMock()
+    pm.get_status.side_effect = lambda _pid: holder.state
+    pm.get_client.side_effect = lambda _pid: holder.client
+    pm.get_model.side_effect = lambda _pid: holder.model
+
+    with (
+        patch.object(kprofile_drift, "printer_manager", pm),
+        patch.object(kprofile_drift, "async_session", session),
+        patch.object(kprofile_drift, "find_slot_kprofile_for_extruder", AsyncMock(side_effect=lookup)) as finder,
+    ):
+        holder.finder = finder
+        yield holder
+
+
+def _selected(client):
+    return [
+        (c.kwargs["ams_id"], c.kwargs["tray_id"], c.kwargs["cali_idx"])
+        for c in client.extrusion_cali_sel.call_args_list
+    ]
+
+
+@pytest.mark.asyncio
+async def test_lost_selection_is_restored(env):
+    env.state = _state([_tray(0), _tray(1)])
+    env.stored[(0, 0)] = _profile(9354)
+    env.stored[(0, 1)] = _profile(3175)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 2
+
+    assert sorted(_selected(env.client)) == [(0, 0, 9354), (0, 1, 3175)]
+    call = env.client.extrusion_cali_sel.call_args_list[0]
+    assert call.kwargs["filament_id"] == "GFA00"
+    assert call.kwargs["nozzle_diameter"] == "0.4"
+
+
+@pytest.mark.asyncio
+async def test_missing_cali_idx_counts_as_lost(env):
+    tray = _tray(0)
+    del tray["cali_idx"]
+    env.state = _state([tray])
+    env.stored[(0, 0)] = _profile(9354)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+
+
+@pytest.mark.asyncio
+async def test_a_different_real_profile_is_left_alone(env):
+    # Picked on purpose, in Bambu Studio or Bambuddy's Configure Slot dialog.
+    env.state = _state([_tray(0, cali_idx=5)])
+    env.stored[(0, 0)] = _profile(9354)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+    env.client.extrusion_cali_sel.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_slot_without_stored_profile_is_left_on_default(env):
+    env.state = _state([_tray(0)])
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+    env.client.extrusion_cali_sel.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_empty_slot_is_skipped(env):
+    env.state = _state([_tray(0, tray_type="")])
+    env.stored[(0, 0)] = _profile(9354)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+    env.finder.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_deliberate_default_pick_is_respected(env):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+    kprofile_drift.note_slot_configured(PRINTER, 0, 0, -1)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+
+    # Picking a real profile again in the dialog lifts it.
+    kprofile_drift.note_slot_configured(PRINTER, 0, 0, 9354)
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+
+
+@pytest.mark.asyncio
+async def test_default_pick_survives_the_push_that_still_shows_the_old_profile(env):
+    env.stored[(0, 0)] = _profile(9354)
+    kprofile_drift.note_slot_configured(PRINTER, 0, 0, -1)
+
+    # The push right after the pick can still carry the previous selection.
+    env.state = _state([_tray(0, cali_idx=9354)])
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+    env.state = _state([_tray(0)])
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+
+
+@pytest.mark.asyncio
+async def test_emptying_the_slot_clears_a_default_pick(env):
+    env.stored[(0, 0)] = _profile(9354)
+    kprofile_drift.note_slot_configured(PRINTER, 0, 0, -1)
+
+    env.state = _state([_tray(0, tray_type="")])
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+    env.state = _state([_tray(0)])
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+
+
+@pytest.mark.asyncio
+async def test_retries_are_spaced_and_capped(env):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+    clock = [1000.0]
+
+    with patch.object(kprofile_drift.time, "monotonic", side_effect=lambda: clock[0]):
+        assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+        # The next push, before the printer has applied it: no resend.
+        assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+
+        for _ in range(kprofile_drift.MAX_ATTEMPTS + 2):
+            clock[0] += kprofile_drift.RETRY_INTERVAL_S
+            await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+
+    assert len(_selected(env.client)) == kprofile_drift.MAX_ATTEMPTS
+
+
+@pytest.mark.asyncio
+async def test_a_selection_that_sticks_resets_the_retry_count(env):
+    env.stored[(0, 0)] = _profile(9354)
+    env.state = _state([_tray(0)])
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+
+    env.state = _state([_tray(0, cali_idx=9354)])
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+    # Lost again later, e.g. the next power cycle: restored at once.
+    env.state = _state([_tray(0)])
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+
+
+@pytest.mark.asyncio
+async def test_disconnect_forgets_retries(env):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+
+    kprofile_drift.forget_printer(PRINTER)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+
+
+@pytest.mark.asyncio
+async def test_dispatch_guard_ignores_the_retry_interval(env):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER, throttle=False) == 1
+
+
+@pytest.mark.asyncio
+async def test_dispatch_guard_runs_while_an_idle_pass_is_in_flight(env):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+    kprofile_drift._running.add(PRINTER)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER, throttle=False) == 1
+
+
+@pytest.mark.asyncio
+async def test_dispatch_guard_only_touches_the_trays_the_job_uses(env):
+    env.state = _state([_tray(0), _tray(1), _tray(2)])
+    for tray_id in range(3):
+        env.stored[(0, tray_id)] = _profile(100 + tray_id)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER, {2}, throttle=False) == 1
+    assert _selected(env.client) == [(0, 2, 102)]
+
+
+@pytest.mark.asyncio
+async def test_external_spool_is_covered(env):
+    env.state = _state([], vt_tray=[{"id": "254", "tray_type": "PETG", "cali_idx": -1}])
+    env.stored[(255, 0)] = _profile(42)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER, {254}, throttle=False) == 1
+    assert _selected(env.client) == [(255, 0, 42)]
+
+
+@pytest.mark.asyncio
+async def test_dual_nozzle_with_unknown_routing_is_not_guessed(env):
+    env.model = "H2D"
+    env.state = _state([_tray(0)], model_nozzles=2)
+    env.stored[(0, 0)] = _profile(9354)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+    env.finder.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_disconnected_printer_is_skipped(env):
+    env.state = _state([_tray(0)], connected=False)
+    env.stored[(0, 0)] = _profile(9354)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 0
+
+
+@pytest.mark.asyncio
+async def test_single_nozzle_external_spool_finds_a_profile_under_either_extruder(env):
+    """X1C external holder: slot_extruder says 1, the spool form stores 0."""
+    env.state = _state([], vt_tray=[{"id": "254", "tray_type": "PETG", "cali_idx": -1}])
+    seen = []
+
+    async def lookup(_db, _pid, ams_id, tray_id, extruder, *_a, **_k):
+        seen.append(extruder)
+        return _profile(42) if extruder == 1 else None
+
+    env.finder.side_effect = lookup
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+    assert seen == [0, 1]
+
+
+@pytest.mark.asyncio
+async def test_single_nozzle_ams_slot_asks_for_extruder_zero_only(env):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+
+    await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+
+    assert [c.args[4] for c in env.finder.await_args_list] == [0]
+
+
+@pytest.mark.asyncio
+async def test_dual_nozzle_uses_only_the_slot_nozzle(env):
+    env.model = "H2D"
+    env.state = _state([_tray(0)], model_nozzles=2)
+    env.state.ams_extruder_map = {"0": 1}
+    env.stored[(0, 0)] = _profile(9354)
+
+    assert await kprofile_drift.reapply_lost_kprofiles(PRINTER) == 1
+    assert [c.args[4] for c in env.finder.await_args_list] == [1]
+
+
+@pytest.mark.asyncio
+async def test_slot_with_nothing_stored_is_looked_up_rarely(env):
+    env.state = _state([_tray(0)])
+    clock = [1000.0]
+
+    with patch.object(kprofile_drift.time, "monotonic", side_effect=lambda: clock[0]):
+        await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+        clock[0] += kprofile_drift.RETRY_INTERVAL_S * 2
+        assert kprofile_drift.needs_check(PRINTER, env.state) is False
+        clock[0] += kprofile_drift.NO_PROFILE_RECHECK_S
+        assert kprofile_drift.needs_check(PRINTER, env.state) is True
+
+    assert env.finder.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_giving_up_is_logged_once_after_the_last_attempt_had_time_to_land(env, caplog):
+    env.state = _state([_tray(0)])
+    env.stored[(0, 0)] = _profile(9354)
+    clock = [1000.0]
+
+    with patch.object(kprofile_drift.time, "monotonic", side_effect=lambda: clock[0]):
+        for _ in range(kprofile_drift.MAX_ATTEMPTS):
+            await kprofile_drift.reapply_lost_kprofiles(PRINTER)
+            # Not yet: the last send may still stick.
+            assert "giving up" not in caplog.text
+            clock[0] += kprofile_drift.RETRY_INTERVAL_S
+        for _ in range(3):
+            kprofile_drift.needs_check(PRINTER, env.state)
+            clock[0] += kprofile_drift.RETRY_INTERVAL_S
+
+    assert caplog.text.count("giving up") == 1
+
+
+class TestNeedsCheck:
+    def test_idle_with_a_lost_slot(self):
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0)])) is True
+
+    @pytest.mark.parametrize("gcode_state", ["RUNNING", "PAUSE", "PREPARE", "SLICING", "unknown"])
+    def test_never_while_a_print_may_be_running(self, gcode_state):
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0)], state=gcode_state)) is False
+
+    @pytest.mark.parametrize("gcode_state", ["IDLE", "FINISH", "FAILED"])
+    def test_idle_states(self, gcode_state):
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0)], state=gcode_state)) is True
+
+    def test_not_when_every_slot_has_a_profile(self):
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0, cali_idx=3)])) is False
+
+    def test_not_while_disconnected(self):
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0)], connected=False)) is False
+
+    def test_not_while_a_pass_is_running(self):
+        kprofile_drift._running.add(PRINTER)
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0)])) is False
+
+    def test_not_again_within_the_retry_interval(self):
+        kprofile_drift._watches[(PRINTER, 0, 0)] = kprofile_drift._SlotWatch(
+            next_check=kprofile_drift.time.monotonic() + 30
+        )
+        assert kprofile_drift.needs_check(PRINTER, _state([_tray(0)])) is False
+
+
+@pytest.mark.asyncio
+async def test_status_handler_spawns_the_check_and_forgets_on_disconnect():
+    """The #3219 trigger lives in on_printer_status_change: every push, idle only."""
+    from backend.app import main
+
+    spawned = []
+
+    def fake_spawn(coro, name=None):
+        spawned.append(name)
+        coro.close()
+
+    with (
+        patch.object(main, "spawn_background_task", side_effect=fake_spawn),
+        patch.object(main.kprofile_drift, "needs_check", return_value=True) as needs_check,
+        patch.object(main.kprofile_drift, "forget_printer") as forget,
+        patch.object(main, "ws_manager") as ws,
+    ):
+        ws.send_printer_status = AsyncMock()
+        state = MagicMock()
+        state.connected = True
+        state.state = "IDLE"
+        state.nozzles = []
+        try:
+            await main.on_printer_status_change(PRINTER, state)
+        except Exception:
+            pass  # Later parts of the handler need more of a real state.
+        assert f"reapply-kprofiles-{PRINTER}" in spawned
+        needs_check.assert_called_once_with(PRINTER, state)
+
+        spawned.clear()
+        state.connected = False
+        try:
+            await main.on_printer_status_change(PRINTER, state)
+        except Exception:
+            pass
+        assert f"reapply-kprofiles-{PRINTER}" not in spawned
+        forget.assert_called_with(PRINTER)
+
+
+@pytest.mark.asyncio
+async def test_a_failing_check_does_not_stop_the_status_broadcast():
+    from backend.app import main
+
+    with (
+        patch.object(main.kprofile_drift, "needs_check", side_effect=RuntimeError("boom")),
+        patch.object(main, "ws_manager") as ws,
+        patch.object(main, "spawn_background_task", side_effect=lambda coro, name=None: coro.close()),
+    ):
+        ws.send_printer_status = AsyncMock()
+        state = MagicMock()
+        state.connected = True
+        state.state = "IDLE"
+        state.nozzles = []
+        try:
+            await main.on_printer_status_change(PRINTER, state)
+        except RuntimeError as exc:
+            pytest.fail(f"K-profile check escaped the status handler: {exc}")
+        except Exception:
+            pass  # Later parts of the handler need more of a real state.

backend/tests/unit/test_scheduler_cleanup_library.py

@@ -718,3 +718,30 @@ async def test_oserror_during_unlink_logs_orphan_path_and_does_not_crash_dispatc
     assert "TRANSIENT_LIBRARY_FILE_ORPHAN" in caplog.text
     assert str(ctx.source_path) in caplog.text
     assert "permission denied" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_lost_k_profiles_are_restored_before_the_print_starts(queue_factory):
+    """A slot reset to the default K must be fixed before the job goes out (#3219)."""
+    ctx = await queue_factory(cleanup=False)
+    order = []
+    reapply = AsyncMock(side_effect=lambda *a, **k: order.append("reapply") or 0)
+    ctx.start_print.side_effect = lambda *a, **k: order.append("start") or True
+
+    with patch.object(scheduler_module.kprofile_drift, "reapply_lost_kprofiles", reapply):
+        await _dispatch_library_item(ctx)
+
+    assert order == ["reapply", "start"]
+    # No stored mapping: every loaded tray is a candidate, and no retry wait.
+    reapply.assert_awaited_once_with(ctx.printer_id, None, throttle=False)
+
+
+@pytest.mark.asyncio
+async def test_a_failing_k_profile_check_does_not_stop_the_print(queue_factory):
+    ctx = await queue_factory(cleanup=False)
+    reapply = AsyncMock(side_effect=RuntimeError("boom"))
+
+    with patch.object(scheduler_module.kprofile_drift, "reapply_lost_kprofiles", reapply):
+        await _dispatch_library_item(ctx)
+
+    ctx.start_print.assert_called_once()