A queued print that never starts no longer marks the printer's next print as started by Bambuddy.

backend/app/api/routes/printers.py

@@ -1074,7 +1074,9 @@ async def get_current_print_user(
         raise HTTPException(404, "Printer not found")
 
     user_info = printer_manager.get_current_print_user(printer_id)
-    return user_info or {}
+    if not user_info:
+        return {}
+    return {"user_id": user_info["user_id"], "username": user_info["username"]}
 
 
 @router.post("/{printer_id}/refresh-status")

backend/app/services/print_scheduler.py

@@ -1246,6 +1246,17 @@ class PrintScheduler:
         # unsuccessful exit; a successful start removes the item id and leaves
         # the reservation for finance_billing to consume with the archive.
         self._unconfirmed_budget_reservations: set[int] = set()
+        # The print user `_start_print` handed to printer_manager for an item
+        # whose print command has not been sent yet (item_id -> printer_id).
+        # `_dispatch_one` takes it back on every unsuccessful exit, like the
+        # expected print above: it marks the printer's next print as Bambuddy's.
+        self._unconfirmed_print_user: dict[int, int] = {}
+        # Prints whose command went out but which the printer has not been
+        # seen starting yet: item_id -> (printer_id, remote_filename, archive_id),
+        # the file and archive None for a library print. The watchdog drops the
+        # entry once the print runs, and withdraws what it stands for when the
+        # print never starts (see `_forget_unstarted_print`).
+        self._sent_unstarted_print: dict[int, tuple[int, str | None, int | None]] = {}
         # Chamber temperature history for smart soak-time reduction.
         # printer_id -> deque of (monotonic_timestamp, celsius) sampled each scheduler tick.
         # Entries older than _chamber_history_ttl are pruned on write.
@@ -1379,6 +1390,8 @@ class PrintScheduler:
                     item.status = status
                     item.completed_at = datetime.now(timezone.utc)
                     closed = True
+                    # The printer is idle: nothing it prints next is this item's.
+                    self._forget_unstarted_print(item.id, unless_printing=True)
                     logger.warning(
                         "Queue item %s was still 'printing' after printer %s reported %s for %.0fs — "
                         "closing it as %s. Its completion event was never matched to it, which blocks "
@@ -1465,6 +1478,8 @@ class PrintScheduler:
         for keyed in (self._auto_dry_units, self._auto_dry_above):
             for key in [key for key in keyed if key[0] == printer_id]:
                 del keyed[key]
+        for item_id in [i for i, sent in self._sent_unstarted_print.items() if sent[0] == printer_id]:
+            del self._sent_unstarted_print[item_id]
 
     async def check_queue(self) -> bool:
         """Check for prints ready to start.
@@ -2630,6 +2645,9 @@ class PrintScheduler:
                 # A confirmed send removes the entry itself, so this is a no-op
                 # on the happy path.
                 self._rollback_unconfirmed_expected_print(item_id)
+                # Likewise the print user it set: the printer's next print is
+                # not this item's.
+                self._rollback_unconfirmed_print_user(item_id)
                 # Mirror the pre-#1625 background-dispatch lifecycle: a
                 # reservation survives only after start_print() accepted the
                 # command. Failure, cancellation, deferral, and exceptions all
@@ -2707,6 +2725,43 @@ class PrintScheduler:
         except Exception:
             pass  # toast is best-effort
 
+    def _rollback_unconfirmed_print_user(self, item_id: int) -> None:
+        printer_id = self._unconfirmed_print_user.pop(item_id, None)
+        if printer_id is not None:
+            printer_manager.clear_current_print_user_for_item(printer_id, item_id)
+
+    def _forget_unstarted_print(self, item_id: int, *, unless_printing: bool, keep_expected: bool = False) -> None:
+        """Withdraw the expected print and print user of a sent print that never started.
+
+        Both tell the rest of Bambuddy that the printer's next print is this
+        item's: left behind, a print started by hand later is taken for it.
+        With ``unless_printing``, a printer that is printing after all keeps
+        them, as that print is most likely this one starting late.
+
+        ``keep_expected`` withdraws only the print user, which vouches for any
+        print on the printer. The expected print names this item's file, so a
+        late start of it is still matched to its archive.
+        """
+        sent = self._sent_unstarted_print.get(item_id)
+        if sent is None:
+            return
+        printer_id, remote_filename, archive_id = sent
+        if unless_printing:
+            status = printer_manager.get_status(printer_id)
+            if status is not None and status.state in _ACTIVE_PRINT_STATES:
+                return
+        printer_manager.clear_current_print_user_for_item(printer_id, item_id)
+        if keep_expected:
+            return
+        del self._sent_unstarted_print[item_id]
+        if remote_filename is not None and archive_id is not None:
+            try:
+                from backend.app.main import unregister_expected_print
+
+                unregister_expected_print(printer_id, remote_filename, archive_id)
+            except Exception:
+                logger.warning("Queue item %s: failed to unregister expected print", item_id)
+
     def _rollback_unconfirmed_expected_print(self, item_id: int) -> None:
         """Drop an expectation for a print command that was never sent.
 
@@ -7051,6 +7106,8 @@ class PrintScheduler:
         """
         if item_id in self._inflight:
             self._cancelled_dispatches.add(item_id)
+        # A print sent for it earlier that never started is not one any more.
+        self._forget_unstarted_print(item_id, unless_printing=False)
 
     async def _preheat_sleep(self, item_id: int, seconds: float) -> bool:
         """Sleep in slices, returning False as soon as the item stops wanting preheat.
@@ -8120,7 +8177,9 @@ class PrintScheduler:
             )
         ).first()
         if owner:
-            printer_manager.set_current_print_user(item.printer_id, owner.id, owner.username)
+            printer_manager.set_current_print_user(item.printer_id, owner.id, owner.username, queue_item_id=item.id)
+            if item.printer_id is not None:
+                self._unconfirmed_print_user[item.id] = item.printer_id
 
     async def _start_print(self, db: AsyncSession, item: PrintQueueItem):
         """Upload file and start print for a queue item.
@@ -9144,6 +9203,13 @@ class PrintScheduler:
             # rolled back.
             self._unconfirmed_expected_print.pop(item.id, None)
             self._unconfirmed_budget_reservations.discard(item.id)
+            self._unconfirmed_print_user.pop(item.id, None)
+            if item.printer_id is not None:
+                self._sent_unstarted_print[item.id] = (
+                    item.printer_id,
+                    remote_filename if archive else None,
+                    archive.id if archive else None,
+                )
             # Handoff to the print's own gcode: keep whatever preheat set (bed
             # target, chamber target, airduct heating) — the gcode owns
             # heater/flap control from here. Clearing the pin prevents
@@ -9376,6 +9442,7 @@ class PrintScheduler:
                 # would otherwise look like "command landed" and leave the
                 # queue item stuck in 'printing' forever (#1370).
                 scheduler._release_dispatch_hold(printer_id)
+                scheduler._sent_unstarted_print.pop(queue_item_id, None)
                 try:
                     await ws_manager.send_queue_item_acked(
                         user_id=created_by_id,
@@ -9413,6 +9480,7 @@ class PrintScheduler:
                 last_status = status
                 if status.state in _ACTIVE_PRINT_STATES:
                     scheduler._release_dispatch_hold(printer_id)
+                    scheduler._sent_unstarted_print.pop(queue_item_id, None)
                     try:
                         await ws_manager.send_queue_item_acked(
                             user_id=created_by_id,
@@ -9543,6 +9611,17 @@ class PrintScheduler:
             )
             revert_outcome = "revert_failed"
 
+        if revert_outcome in ("command_rejected", "gave_up"):
+            # The item is failed: its print user and expected print must not
+            # pass the printer's next print off as Bambuddy's.
+            scheduler._forget_unstarted_print(queue_item_id, unless_printing=True)
+        elif revert_outcome == "reverted":
+            # Another try claims the printer again. Until then no print on it
+            # is vouched for as this item's, except a late start of its file.
+            scheduler._forget_unstarted_print(queue_item_id, unless_printing=True, keep_expected=True)
+        elif revert_outcome == "already_moved_on":
+            scheduler._sent_unstarted_print.pop(queue_item_id, None)
+
         if revert_outcome == "already_moved_on":
             # Preserves the pre-#1370 early-return: if on_print_complete (or any
             # other path) already moved the item past 'printing', don't run the

backend/app/services/printer_manager.py

@@ -422,9 +422,13 @@ class PrinterManager:
         """Get printer info by ID."""
         return self._printer_info.get(printer_id)
 
-    def set_current_print_user(self, printer_id: int, user_id: int, username: str):
-        """Track who started the current print (Issue #206)."""
-        self._current_print_user[printer_id] = {"user_id": user_id, "username": username}
+    def set_current_print_user(self, printer_id: int, user_id: int, username: str, queue_item_id: int | None = None):
+        """Track who started the current print (Issue #206), and for which queue item."""
+        self._current_print_user[printer_id] = {
+            "user_id": user_id,
+            "username": username,
+            "queue_item_id": queue_item_id,
+        }
 
     def get_current_print_user(self, printer_id: int) -> dict | None:
         """Get the user who started the current print (Issue #206)."""
@@ -434,6 +438,15 @@ class PrinterManager:
         """Clear the current print user when print completes (Issue #206)."""
         self._current_print_user.pop(printer_id, None)
 
+    def clear_current_print_user_for_item(self, printer_id: int, queue_item_id: int) -> None:
+        """Clear the print user if a queue item's dispatch set it and that print never started.
+
+        A later dispatch's user, set for another item, stays.
+        """
+        user = self._current_print_user.get(printer_id)
+        if user is not None and user.get("queue_item_id") == queue_item_id:
+            del self._current_print_user[printer_id]
+
     def forget_user(self, user_id: int) -> None:
         """A deleted user is credited with none of the prints still running."""
         for printer_id, user in list(self._current_print_user.items()):

backend/tests/integration/test_printers_api.py

@@ -1160,14 +1160,12 @@ class TestPrinterDataIntegrity:
         printer = await printer_factory(name="Test Printer")
 
         with patch("backend.app.api.routes.printers.printer_manager") as mock_pm:
-            mock_pm.get_current_print_user.return_value = {"user_id": 42, "username": "testuser"}
+            mock_pm.get_current_print_user.return_value = {"user_id": 42, "username": "testuser", "queue_item_id": 5}
 
             response = await async_client.get(f"/api/v1/printers/{printer.id}/current-print-user")
 
             assert response.status_code == 200
-            result = response.json()
-            assert result["user_id"] == 42
-            assert result["username"] == "testuser"
+            assert response.json() == {"user_id": 42, "username": "testuser"}
 
 
 class TestPrintControlAPI:

backend/tests/integration/test_queue_start_user_attribution.py

@@ -196,7 +196,7 @@ class TestSchedulerPropagatesOwnerToPrinterManager:
         monkeypatch.setattr(
             scheduler_module.printer_manager,
             "set_current_print_user",
-            lambda printer_id, uid, username: captured.append((printer_id, uid, username)),
+            lambda printer_id, uid, username, **_: captured.append((printer_id, uid, username)),
         )
 
         await PrintScheduler()._propagate_owner_to_printer_manager(db_session, queue_item)
@@ -218,7 +218,7 @@ class TestSchedulerPropagatesOwnerToPrinterManager:
         monkeypatch.setattr(
             scheduler_module.printer_manager,
             "set_current_print_user",
-            lambda *args: captured.append(args),
+            lambda *args, **_: captured.append(args),
         )
 
         await PrintScheduler()._propagate_owner_to_printer_manager(db_session, queue_item)
@@ -243,7 +243,7 @@ class TestSchedulerPropagatesOwnerToPrinterManager:
         monkeypatch.setattr(
             scheduler_module.printer_manager,
             "set_current_print_user",
-            lambda *args: captured.append(args),
+            lambda *args, **_: captured.append(args),
         )
 
         # Must not raise — the dispatch loop would otherwise lose the whole

backend/tests/unit/test_failed_start_releases_printer.py

@@ -0,0 +1,251 @@
+"""A queued print that never starts leaves no claim on the printer's next print.
+
+Dispatching a queue item tells the rest of Bambuddy that the printer's next
+print is the item's: the item's owner is recorded as the print's user, and an
+archive print is registered as expected. When that print never starts, both
+must go, or a print started on the printer afterwards is taken for it.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.services.print_scheduler import DISPATCH_MAX_ATTEMPTS, PrintScheduler, scheduler
+from backend.app.services.printer_manager import printer_manager
+
+PRINTER = 42
+ITEM = 1
+
+
+@pytest.fixture
+async def db_session():
+    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+    import backend.app.models  # noqa: F401
+    from backend.app.core.database import Base
+
+    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)
+    async with session_maker() as db:
+        db.add(PrintQueueItem(id=ITEM, printer_id=PRINTER, archive_id=99, status="printing"))
+        await db.commit()
+    try:
+        yield session_maker
+    finally:
+        await engine.dispose()
+
+
+@pytest.fixture(autouse=True)
+def clean_state():
+    from backend.app import main
+
+    tables = ("_expected_prints", "_expected_print_creators", "_expected_print_registered_at")
+    saved = {name: dict(getattr(main, name)) for name in tables}
+    for name in tables:
+        getattr(main, name).clear()
+    printer_manager.clear_current_print_user(PRINTER)
+    scheduler._sent_unstarted_print.clear()
+    yield main
+    printer_manager.clear_current_print_user(PRINTER)
+    scheduler._sent_unstarted_print.clear()
+    for name in tables:
+        getattr(main, name).clear()
+        getattr(main, name).update(saved[name])
+
+
+def _sent(main, item_id: int = ITEM, *, sched=scheduler):
+    """What a dispatch whose print command went out leaves behind."""
+    main.register_expected_print(PRINTER, "part.3mf", 99, created_by_id=7)
+    printer_manager.set_current_print_user(PRINTER, 7, "anna", queue_item_id=item_id)
+    sched._sent_unstarted_print[item_id] = (PRINTER, "part.3mf", 99)
+
+
+def _claimed(main) -> bool:
+    return printer_manager.get_current_print_user(PRINTER) is not None or bool(main._expected_prints)
+
+
+async def _watch(db_session, status):
+    with (
+        patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=status)),
+        patch("backend.app.services.print_scheduler.printer_manager.get_client", MagicMock(return_value=MagicMock())),
+        patch("backend.app.services.print_scheduler.async_session", db_session),
+        patch("backend.app.core.database.async_session", db_session),
+        patch("backend.app.services.notification_service.notification_service.on_queue_job_failed", AsyncMock()),
+    ):
+        await PrintScheduler._watchdog_print_start(
+            queue_item_id=ITEM,
+            printer_id=PRINTER,
+            pre_state="IDLE",
+            pre_subtask_id="OLD",
+            pre_gcode_file="/old.3mf",
+            timeout=0.2,
+            phase_b_timeout=0.2,
+            poll_interval=0.05,
+        )
+
+
+async def _set_printing(db_session):
+    async with db_session() as db:
+        (await db.get(PrintQueueItem, ITEM)).status = "printing"
+        await db.commit()
+
+
+@pytest.mark.asyncio
+async def test_giving_up_on_a_print_releases_the_printer(db_session, clean_state):
+    for _ in range(DISPATCH_MAX_ATTEMPTS):
+        await _set_printing(db_session)
+        _sent(clean_state)
+        await _watch(db_session, SimpleNamespace(state="IDLE", subtask_id="NEW", gcode_file="/new.3mf"))
+
+    async with db_session() as db:
+        assert (await db.get(PrintQueueItem, ITEM)).status == "failed"
+    assert not _claimed(clean_state)
+
+
+@pytest.mark.asyncio
+async def test_a_rejected_print_command_releases_the_printer(db_session, clean_state):
+    from backend.app.services.bambu_mqtt import HMS_MQTT_VERIFY_FAILED
+
+    _sent(clean_state)
+    rejected = SimpleNamespace(
+        state="IDLE",
+        subtask_id="OLD",
+        gcode_file="/new.3mf",
+        hms_errors=[SimpleNamespace(full_code=HMS_MQTT_VERIFY_FAILED)],
+    )
+    await _watch(db_session, rejected)
+
+    assert not _claimed(clean_state)
+
+
+@pytest.mark.asyncio
+async def test_a_print_that_starts_keeps_its_claim(db_session, clean_state):
+    _sent(clean_state)
+    await _watch(db_session, SimpleNamespace(state="RUNNING", subtask_id="NEW", gcode_file="/new.3mf"))
+
+    assert printer_manager.get_current_print_user(PRINTER)["user_id"] == 7
+    assert clean_state._expected_prints
+    assert ITEM not in scheduler._sent_unstarted_print
+
+
+def test_a_print_that_started_late_keeps_its_claim(clean_state):
+    _sent(clean_state)
+    with patch(
+        "backend.app.services.print_scheduler.printer_manager.get_status",
+        MagicMock(return_value=SimpleNamespace(state="RUNNING")),
+    ):
+        scheduler._forget_unstarted_print(ITEM, unless_printing=True)
+
+    assert _claimed(clean_state)
+
+
+def test_another_item_keeps_the_printer(clean_state):
+    _sent(clean_state)
+    printer_manager.set_current_print_user(PRINTER, 8, "ben", queue_item_id=2)
+
+    scheduler._forget_unstarted_print(ITEM, unless_printing=False)
+
+    assert printer_manager.get_current_print_user(PRINTER)["user_id"] == 8
+
+
+def test_cancelling_an_item_after_a_failed_start_releases_the_printer(clean_state):
+    _sent(clean_state)
+
+    scheduler.notify_dispatch_cancelled(ITEM)
+
+    assert not _claimed(clean_state)
+
+
+@pytest.mark.asyncio
+async def test_a_dispatch_that_never_sends_releases_the_print_user(clean_state):
+    sched = PrintScheduler()
+
+    async def fake_start_print(db, item):
+        # What `_start_print` has done by the time the cancel wins the race.
+        printer_manager.set_current_print_user(PRINTER, 7, "anna", queue_item_id=597)
+        sched._unconfirmed_print_user[597] = PRINTER
+
+    db = MagicMock()
+    db.get = AsyncMock(return_value=MagicMock(id=597, printer_id=PRINTER))
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=db)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    with (
+        patch("backend.app.services.print_scheduler.async_session", return_value=ctx),
+        patch.object(sched, "_claim_for_dispatch", AsyncMock(return_value=True)),
+        patch.object(sched, "_start_print", side_effect=fake_start_print),
+        patch.object(sched, "_clear_dispatch_claim", AsyncMock()),
+    ):
+        await sched._dispatch_one(597)
+
+    assert printer_manager.get_current_print_user(PRINTER) is None
+    assert sched._unconfirmed_print_user == {}
+
+
+@pytest.mark.asyncio
+async def test_a_dispatch_records_whose_print_it_claims(db_session, clean_state):
+    from backend.app.models.printer import Printer
+    from backend.app.models.user import User
+
+    async with db_session() as db:
+        db.add(User(id=7, username="anna", password_hash="x"))
+        db.add(Printer(id=PRINTER, name="P", serial_number="00M09A000000042", ip_address="192.0.2.1", access_code="x"))
+        item = await db.get(PrintQueueItem, ITEM)
+        item.created_by_id = 7
+        await db.commit()
+        sched = PrintScheduler()
+        await sched._propagate_owner_to_printer_manager(db, item)
+
+    assert printer_manager.get_current_print_user(PRINTER)["queue_item_id"] == ITEM
+    assert sched._unconfirmed_print_user == {ITEM: PRINTER}
+
+
+@pytest.mark.asyncio
+async def test_a_print_sent_back_for_another_try_vouches_only_for_its_own_file(db_session, clean_state):
+    _sent(clean_state)
+    await _watch(db_session, SimpleNamespace(state="IDLE", subtask_id="NEW", gcode_file="/new.3mf"))
+
+    async with db_session() as db:
+        assert (await db.get(PrintQueueItem, ITEM)).status == "pending"
+    assert printer_manager.get_current_print_user(PRINTER) is None
+    assert clean_state._expected_prints, "a late start of its own file must still match its archive"
+
+    # Cancelling it then withdraws the rest.
+    scheduler.notify_dispatch_cancelled(ITEM)
+    assert not _claimed(clean_state)
+
+
+@pytest.mark.asyncio
+async def test_closing_a_stuck_item_releases_the_printer(db_session, clean_state):
+    _sent(clean_state)
+    sched = PrintScheduler()
+    sched._sent_unstarted_print[ITEM] = scheduler._sent_unstarted_print[ITEM]
+    sched._terminal_since[PRINTER] = -10_000.0
+    with (
+        patch("backend.app.services.print_scheduler.async_session", db_session),
+        patch(
+            "backend.app.services.print_scheduler.printer_manager.get_status",
+            MagicMock(return_value=SimpleNamespace(state="IDLE", connected=True)),
+        ),
+    ):
+        await sched._close_stranded_printing_items()
+
+    async with db_session() as db:
+        assert (await db.get(PrintQueueItem, ITEM)).status != "printing"
+    assert not _claimed(clean_state)
+
+
+def test_deleting_a_printer_drops_its_sent_prints(clean_state):
+    sched = PrintScheduler()
+    sched._sent_unstarted_print[ITEM] = (PRINTER, "part.3mf", 99)
+    sched._sent_unstarted_print[2] = (7, None, None)
+
+    sched.forget_printer(PRINTER)
+
+    assert sched._sent_unstarted_print == {2: (7, None, None)}