fix(dispatch): don't send start-print to a busy printer; scheduler defers (#2598)

start_print() published project_file guarding only on connection state, so a
re-dispatch onto a printer that had already started — e.g. a watchdog revert
(#2555) after the printer sat in FINISH past accepting the job — collided with
the live print. The firmware answers 0500_4004 ("Device is busy and cannot
start a new task"), which on an A1 mini cancels the running job.

Defense-in-depth at the paths that can reach a busy printer:

- bambu_mqtt: refuse to publish project_file when gcode_state is
  PREPARE/SLICING/RUNNING/PAUSE and return without sending. This is the one
  publish choke point every dispatch path funnels through (queue scheduler,
  manual start, webhook, Virtual-Printer forward). IDLE/FINISH/FAILED still
  start.
- print_scheduler: re-check the live printer state right before the FTP upload
  and defer a busy printer (leave the item pending for a later tick) instead of
  uploading and dispatching. If the printer goes busy in the upload window and
  the start is refused, revert the item to pending rather than marking it
  failed — a busy printer is a deferral, not a failure.

A transport-level MQTT QoS-1 replay on reconnect would bypass the client guard,
but the dispatch/watchdog reconnect path already hard-resets the client with a
fresh session, so it has no inflight project_file to replay.

backend/app/services/bambu_mqtt.py

@@ -32,6 +32,14 @@ logger = logging.getLogger(__name__)
 #   "n3s/<id>"  – AMS HT (H2D Pro and similar; IDs typically start at 128)
 _AMS_MODULE_PREFIXES = ("ams/", "n3f/", "n3s/")
 
+# gcode_state values that mean the printer is not idle and must not be handed a
+# new start-print (#2598). The firmware rejects a project_file while busy with
+# 0500_4004 "Device is busy and cannot start a new task", and on some models
+# (A1 mini reported) that error cancels the RUNNING job. IDLE / FINISH / FAILED
+# are valid start targets and are deliberately excluded. Mirrors
+# printer_manager.ACTIVE_PRINT_STATES and print_scheduler._ACTIVE_PRINT_STATES.
+_ACTIVE_PRINT_STATES = frozenset({"PREPARE", "SLICING", "RUNNING", "PAUSE"})
+
 
 def parse_ams_filament_backup_from_cfg(cfg_raw: object) -> bool | None:
     """Extract AMS Filament Backup state from a Bambu push_status ``print.cfg`` value.
@@ -3851,7 +3859,31 @@ class BambuMQTTClient:
                 firmware honours the user's slicer pick instead of falling
                 back to "last matching nozzle" auto-pick. Silently ignored
                 on single-nozzle printers.
+
+        Returns True when the start command was published, False otherwise
+        (not connected, or the printer is already busy — see the run-state
+        guard below).
         """
+        # Never dispatch project_file to a printer that is not idle (#2598).
+        # This is the single publish choke point for every dispatch path — the
+        # queue scheduler, a manual start, a webhook, and a Virtual-Printer
+        # forwarded job all funnel through here — so one guard covers them all.
+        # The firmware rejects a start while busy with 0500_4004 ("Device is
+        # busy and cannot start a new task"), and on an A1 mini that error
+        # cancels the RUNNING job (#2598). IDLE / FINISH / FAILED are valid
+        # start targets; only the active-print states are refused. (A
+        # transport-level QoS-1 replay on reconnect would bypass this guard,
+        # but the dispatch/watchdog reconnect path hard-resets the client with a
+        # fresh client_id, so paho has no inflight project_file to replay there.)
+        if self.state.state in _ACTIVE_PRINT_STATES:
+            logger.warning(
+                "[%s] start_print refused: printer busy (gcode_state=%s) — not publishing project_file for %s",
+                self.serial_number,
+                self.state.state,
+                filename,
+            )
+            return False
+
         if self._client and self.state.connected:
             # Bambu print command format — matches Bambu Studio's format.
             # The calibration/leveling fields (timelapse, bed_leveling,

backend/app/services/print_scheduler.py

@@ -2868,6 +2868,28 @@ class PrintScheduler:
             )
             return
 
+        # Busy-printer guard (#2598). check_queue gates dispatch on
+        # _is_printer_idle(), but that treats FINISH as idle and a printer can
+        # keep reporting FINISH for tens of seconds *after* it accepted a
+        # project_file (see the watchdog's phase-B note). A watchdog revert
+        # (#2555) also releases the dispatch hold, so a re-selected item can
+        # reach here while its printer has actually started printing. Uploading
+        # and dispatching then collides with the live job — the firmware answers
+        # 0500_4004 and, on an A1 mini, cancels the running print. Re-check the
+        # live state right before the expensive FTP upload: if the printer is
+        # busy, leave the item pending and let a later tick dispatch it once the
+        # printer is genuinely idle. No wasted upload, no collision.
+        pre_dispatch_state = getattr(printer_manager.get_status(item.printer_id), "state", None)
+        if pre_dispatch_state in _ACTIVE_PRINT_STATES:
+            logger.info(
+                "Queue item %s: printer %s is busy (state=%s) — deferring dispatch, "
+                "leaving item pending for a later tick (#2598)",
+                item.id,
+                item.printer_id,
+                pre_dispatch_state,
+            )
+            return
+
         # Determine source: archive or library file
         archive = None
         library_file = None
@@ -3447,6 +3469,29 @@ class PrintScheduler:
             except Exception:
                 pass  # Best-effort — don't fail the error handler
 
+            # Busy-refusal is a deferral, not a failure (#2598). The printer's
+            # state can flip from idle to active in the window between the
+            # pre-dispatch check above and this publish (the FTP upload takes
+            # seconds); start_print() then refuses to send project_file to the
+            # now-busy printer and returns False. Failing the item here would be
+            # wrong — the printer is fine, it is simply busy — so revert to
+            # pending and let a later tick dispatch it once the printer is idle,
+            # exactly like the pre-dispatch guard. Only a start_print() False on
+            # an idle/unknown printer is a genuine command failure.
+            post_dispatch_state = getattr(printer_manager.get_status(item.printer_id), "state", None)
+            if post_dispatch_state in _ACTIVE_PRINT_STATES:
+                logger.info(
+                    "Queue item %s: printer %s became busy (state=%s) before the start "
+                    "command was sent — deferring, reverting item to pending (#2598)",
+                    item.id,
+                    item.printer_id,
+                    post_dispatch_state,
+                )
+                item.status = "pending"
+                item.started_at = None
+                await db.commit()
+                return
+
             # Print command failed - revert status
             item.status = "failed"
             item.error_message = "Failed to send print command to printer"

backend/tests/unit/services/test_start_print_busy_guard_2598.py

@@ -0,0 +1,60 @@
+"""start_print() must not publish project_file to a busy printer (#2598).
+
+The firmware rejects a start command while the printer is not idle with
+0500_4004 ("Device is busy and cannot start a new task"), and on an A1 mini
+that error cancels the RUNNING job. Because every dispatch path (queue
+scheduler, manual start, webhook, Virtual-Printer forward) funnels through
+BambuMQTTClient.start_print, a run-state guard here covers them all.
+
+IDLE / FINISH / FAILED are valid start targets; only PREPARE / SLICING /
+RUNNING / PAUSE are refused.
+"""
+
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+from backend.app.services.bambu_mqtt import BambuMQTTClient
+
+
+def _connected_client() -> BambuMQTTClient:
+    client = BambuMQTTClient(ip_address="127.0.0.1", serial_number="TEST123", access_code="12345678")
+    client._client = MagicMock()
+    client.state.connected = True
+    return client
+
+
+@pytest.mark.parametrize("busy_state", ["RUNNING", "PREPARE", "PAUSE", "SLICING"])
+def test_start_print_refused_when_printer_busy(busy_state):
+    client = _connected_client()
+    client.state.state = busy_state
+
+    result = client.start_print("job.3mf")
+
+    assert result is False, f"start_print should refuse while {busy_state}"
+    client._client.publish.assert_not_called()
+
+
+@pytest.mark.parametrize("idle_state", ["IDLE", "FINISH", "FAILED"])
+def test_start_print_publishes_when_printer_idle(idle_state):
+    client = _connected_client()
+    client.state.state = idle_state
+
+    result = client.start_print("job.3mf")
+
+    assert result is True, f"start_print should proceed while {idle_state}"
+    client._client.publish.assert_called_once()
+    topic, payload = client._client.publish.call_args.args[:2]
+    assert json.loads(payload)["print"]["command"] == "project_file"
+
+
+def test_busy_guard_takes_precedence_over_disconnected():
+    """A busy printer is refused even if the connection flag is stale/false —
+    the guard runs before the connection check, so no publish is attempted."""
+    client = _connected_client()
+    client.state.connected = False
+    client.state.state = "RUNNING"
+
+    assert client.start_print("job.3mf") is False
+    client._client.publish.assert_not_called()

backend/tests/unit/test_scheduler_busy_defer_2598.py

@@ -0,0 +1,159 @@
+"""The scheduler defers (never fails) a dispatch that hits a busy printer (#2598).
+
+check_queue gates dispatch on _is_printer_idle(), but that treats FINISH as
+idle and a printer can keep reporting FINISH for tens of seconds after it
+accepted a project_file; a watchdog revert (#2555) also releases the dispatch
+hold. So a re-selected item can reach _start_print while its printer has
+actually started printing. Two guards keep that from cancelling the live job:
+
+* pre-dispatch — before the FTP upload, a busy printer defers (item stays
+  pending), so there is no wasted upload and no start command;
+* post-dispatch — if the printer goes busy in the upload window and
+  start_print() returns False, the item is reverted to pending (deferred), not
+  marked failed.
+"""
+
+from contextlib import ExitStack
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+import backend.app.models  # noqa: F401 - populate Base.metadata
+import backend.app.services.print_scheduler as scheduler_module
+from backend.app.core.database import Base
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+async def dispatch_case(tmp_path):
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    base_dir = tmp_path / "case"
+    base_dir.mkdir()
+    archive_rel = Path("archives") / "job.3mf"
+    archive_abs = base_dir / archive_rel
+    archive_abs.parent.mkdir(parents=True, exist_ok=True)
+    archive_abs.write_bytes(b"archive payload")
+
+    async with session_maker() as db:
+        printer = Printer(
+            name="Printer",
+            serial_number="SERIAL",
+            ip_address="127.0.0.1",
+            access_code="access-code",
+            model="A1MINI",
+        )
+        db.add(printer)
+        await db.flush()
+        archive = PrintArchive(
+            printer_id=printer.id,
+            filename="job.3mf",
+            file_path=str(archive_rel),
+            file_size=archive_abs.stat().st_size,
+            status="completed",
+        )
+        db.add(archive)
+        await db.flush()
+        item = PrintQueueItem(printer_id=printer.id, archive_id=archive.id, status="pending")
+        db.add(item)
+        await db.commit()
+        ids = SimpleNamespace(printer_id=printer.id, archive_id=archive.id, item_id=item.id)
+
+    try:
+        yield SimpleNamespace(session_maker=session_maker, base_dir=base_dir, ids=ids)
+    finally:
+        await engine.dispose()
+
+
+def _base_patches(scheduler, ctx, upload_mock, start_print_mock, get_status):
+    return [
+        patch.object(scheduler_module.settings, "base_dir", ctx.base_dir),
+        patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
+        patch("backend.app.services.print_scheduler.printer_manager.start_print", start_print_mock),
+        patch("backend.app.services.print_scheduler.printer_manager.set_awaiting_plate_clear", MagicMock()),
+        patch(
+            "backend.app.services.print_scheduler.get_ftp_retry_settings",
+            AsyncMock(return_value=(False, 0, 0, 1.0)),
+        ),
+        patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True)),
+        patch("backend.app.services.print_scheduler.upload_file_async", upload_mock),
+        patch("backend.app.services.print_scheduler.cache_3mf_download", MagicMock()),
+        patch("backend.app.services.print_scheduler.spawn_background_task", MagicMock()),
+        patch.object(scheduler, "_propagate_owner_to_printer_manager", AsyncMock()),
+        patch.object(scheduler, "_power_off_if_needed", AsyncMock()),
+        patch.object(scheduler, "_preheat_and_soak", AsyncMock()),
+    ]
+
+
+async def _final_item(ctx):
+    async with ctx.session_maker() as db:
+        return await db.get(PrintQueueItem, ctx.ids.item_id)
+
+
+@pytest.mark.asyncio
+async def test_pre_dispatch_busy_defers_without_upload(dispatch_case):
+    """Printer already RUNNING when _start_print begins → defer, no upload/start."""
+    scheduler = PrintScheduler()
+    upload = AsyncMock(return_value=True)
+    start_print = MagicMock(return_value=True)
+    get_status = MagicMock(return_value=SimpleNamespace(state="RUNNING", subtask_id=None, gcode_file=None))
+
+    async with dispatch_case.session_maker() as db:
+        item = await db.get(PrintQueueItem, dispatch_case.ids.item_id)
+        with ExitStack() as stack:
+            for p in _base_patches(scheduler, dispatch_case, upload, start_print, get_status):
+                stack.enter_context(p)
+            await scheduler._start_print(db, item)
+
+    upload.assert_not_awaited()
+    start_print.assert_not_called()
+    final = await _final_item(dispatch_case)
+    assert final.status == "pending", "a busy printer must defer the item, not consume it"
+
+
+@pytest.mark.asyncio
+async def test_post_dispatch_busy_reverts_to_pending_not_failed(dispatch_case):
+    """Printer goes busy in the upload window; start_print returns False → defer."""
+    scheduler = PrintScheduler()
+    upload = AsyncMock(return_value=True)
+
+    holder = {"state": "IDLE"}
+
+    class _Status:
+        subtask_id = None
+        gcode_file = None
+
+        @property
+        def state(self):
+            return holder["state"]
+
+    def _start_print(*args, **kwargs):
+        # The printer became busy between the pre-dispatch check and the publish.
+        holder["state"] = "RUNNING"
+        return False  # start_print() refused: busy
+
+    start_print = MagicMock(side_effect=_start_print)
+    get_status = MagicMock(return_value=_Status())
+
+    async with dispatch_case.session_maker() as db:
+        item = await db.get(PrintQueueItem, dispatch_case.ids.item_id)
+        with ExitStack() as stack:
+            for p in _base_patches(scheduler, dispatch_case, upload, start_print, get_status):
+                stack.enter_context(p)
+            await scheduler._start_print(db, item)
+
+    upload.assert_awaited_once()  # it proceeded past the (idle) pre-dispatch check
+    start_print.assert_called_once()
+    final = await _final_item(dispatch_case)
+    assert final.status == "pending", "a busy-refused start must defer, not fail the item"
+    assert final.started_at is None