fix(queue): two-phase dispatch watchdog so a printer that accepts project_file but never starts doesn't wedge the queue (#1678)

  _watchdog_print_start now treats subtask_id-advance as Phase A
  "command landed", not as final success. Phase B (180s) keeps watching
  for the active-state transition; if it never arrives — printer
  accepted the file but stalled (cloud+LAN re-auth after a power cycle
  on old firmware was the reported trigger) — revert the queue item to
  'pending' instead of leaving it stuck in 'printing' until container
  restart. Phase B skips force_reconnect because subtask_id-advance
  proves the project_file landed; a reconnect mid-parse would trigger
  0500_4003 (#1150). Phase A's H2D 50s FINISH→PREPARE tolerance (#1078)
  is preserved by Phase B's 180s headroom.

backend/app/services/print_scheduler.py

@@ -2315,31 +2315,39 @@ class PrintScheduler:
         pre_subtask_id: str | None = None,
         pre_gcode_file: str | None = None,
         timeout: float = 90.0,
+        phase_b_timeout: float = 180.0,
         poll_interval: float = 3.0,
     ) -> None:
         """Revert a queue item if the printer never acknowledges the start command.
 
         Bambuddy optimistically marks the queue item as "printing" right after the
-        MQTT project_file publish succeeds locally. If the printer drops/ignores the
-        command (half-broken MQTT session — #887/#936), the state never transitions
-        and the item would otherwise stay stuck in "printing" forever (#967).
-
-        Exit paths (printer picked up the job — no revert):
-          - gcode_state changed from pre_state, OR
-          - subtask_id advanced past pre_subtask_id — the printer echoes our
-            per-dispatch identity back on push_status, so a subtask_id change is
-            a definitive "command landed" signal even while state is still FINISH.
-            H2D can sit at FINISH for ~50 s after accepting project_file before
-            transitioning to PREPARE, which used to trip the state-only watchdog
-            and caused the scheduler to revert + re-dispatch the item; the next
-            successful dispatch then looked like a reprint of the just-finished
-            job (#1078).
-
-        Timeout raised from 45 s → 90 s as belt-and-braces for slow transitions
-        that also don't emit an early subtask_id tick.
+        MQTT project_file publish succeeds locally. The watchdog runs in two phases:
+
+        Phase A (up to ``timeout``): wait for either an active-state transition
+        or a ``subtask_id`` advance past ``pre_subtask_id``. State alone is the
+        primary signal; subtask_id advance handles the H2D case where state can
+        sit at FINISH for ~50 s after the printer accepted ``project_file``
+        before flipping to PREPARE (#1078). If neither happens, the MQTT publish
+        was lost on a half-broken session (#887/#936) — revert and force
+        reconnect (the #967 recovery path).
+
+        Phase B (up to ``phase_b_timeout``, only if Phase A exited on subtask_id
+        alone): keep watching for the active-state transition. subtask_id alone
+        proves the file landed but not that the printer started — and a printer
+        that accepts the command but stays at IDLE/FINISH indefinitely (e.g.
+        cloud+LAN re-auth dance after a power cycle on old firmware, #1678)
+        used to leave the queue item stuck in 'printing' forever because the
+        old watchdog returned success as soon as subtask_id advanced. If Phase
+        B times out, revert the queue item so the user can retry without
+        restarting Bambuddy. Skip ``force_reconnect`` here: the file landed and
+        a forced reconnect mid-parse triggers 0500_4003 (#1150).
+
+        Phase A timeout raised from 45 s → 90 s as belt-and-braces for slow
+        transitions that also don't emit an early subtask_id tick.
         """
-        deadline = time.monotonic() + timeout
         last_status = None
+        landed_on_subtask = False
+        deadline = time.monotonic() + timeout
         while time.monotonic() < deadline:
             await asyncio.sleep(poll_interval)
             status = printer_manager.get_status(printer_id)
@@ -2362,14 +2370,28 @@ class PrintScheduler:
                 scheduler._release_dispatch_hold(printer_id)
                 return
             if pre_subtask_id is not None and status.subtask_id is not None and status.subtask_id != pre_subtask_id:
-                # Printer picked up the job (subtask_id advanced). H2D can
-                # sit at FINISH for ~50 s after accepting project_file
-                # before transitioning to PREPARE, but the subtask_id flips
-                # to our submission_id almost immediately (#1078).
-                scheduler._release_dispatch_hold(printer_id)
-                return
-
-        # No transition. Revert the item so the scheduler can retry.
+                # Phase A exit — printer accepted the file (subtask_id flipped
+                # to our submission id). Don't return yet: the printer may
+                # have accepted the command but never actually start (e.g.
+                # cloud+LAN re-auth dance after a power cycle, #1678). Phase
+                # B watches for the active-state transition.
+                landed_on_subtask = True
+                break
+
+        if landed_on_subtask:
+            phase_b_deadline = time.monotonic() + phase_b_timeout
+            while time.monotonic() < phase_b_deadline:
+                await asyncio.sleep(poll_interval)
+                status = printer_manager.get_status(printer_id)
+                if not status:
+                    scheduler._release_dispatch_hold(printer_id)
+                    return
+                last_status = status
+                if status.state in _ACTIVE_PRINT_STATES:
+                    scheduler._release_dispatch_hold(printer_id)
+                    return
+
+        # No active-state transition. Revert the item so the scheduler can retry.
         # Drop the in-memory hold so the retry isn't blocked by it.
         scheduler._release_dispatch_hold(printer_id)
 
@@ -2410,24 +2432,44 @@ class PrintScheduler:
             # session breaks ongoing prints on the same printer.
             return
 
+        total_timeout = timeout + (phase_b_timeout if landed_on_subtask else 0.0)
         if revert_outcome == "reverted":
-            logger.warning(
-                "Queue item %s: printer %d did not respond to print command within "
-                "%.0fs (state still %s, subtask_id still %s) — reverted to 'pending' "
-                "for retry (#967)",
-                queue_item_id,
-                printer_id,
-                timeout,
-                pre_state,
-                pre_subtask_id,
-            )
+            if landed_on_subtask:
+                logger.warning(
+                    "Queue item %s: printer %d accepted project_file (subtask_id "
+                    "advanced) but never transitioned to an active state within "
+                    "%.0fs — printer wedged post-acceptance; reverted to 'pending' "
+                    "for retry (#1678)",
+                    queue_item_id,
+                    printer_id,
+                    total_timeout,
+                )
+            else:
+                logger.warning(
+                    "Queue item %s: printer %d did not respond to print command within "
+                    "%.0fs (state still %s, subtask_id still %s) — reverted to 'pending' "
+                    "for retry (#967)",
+                    queue_item_id,
+                    printer_id,
+                    timeout,
+                    pre_state,
+                    pre_subtask_id,
+                )
+
+        # Phase B was entered iff subtask_id advanced, which means the
+        # project_file landed on the printer. A forced reconnect at this point
+        # would interrupt the printer's parse and trigger 0500_4003 (#1150) —
+        # skip the recovery entirely.
+        if landed_on_subtask:
+            return
 
-        # Same #1150 / #887/#936 discriminator as background_dispatch: if the
-        # printer's gcode_file changed since pre-dispatch, the project_file
-        # command landed and the printer is parsing — a forced reconnect
-        # mid-parse triggers 0500_4003. If gcode_file is unchanged, the
-        # publish was silently swallowed (#887/#936) and the original
-        # force_reconnect recovery is what we want.
+        # Phase A timeout path — same #1150 / #887/#936 discriminator as
+        # background_dispatch: if the printer's gcode_file changed since
+        # pre-dispatch, the project_file command landed and the printer is
+        # parsing — a forced reconnect mid-parse triggers 0500_4003. If
+        # gcode_file is unchanged, the publish was silently swallowed
+        # (#887/#936) and the original force_reconnect recovery is what we
+        # want.
         client = printer_manager.get_client(printer_id)
         current_gcode_file = getattr(last_status, "gcode_file", None) if last_status else None
         publish_landed = current_gcode_file is not None and current_gcode_file != pre_gcode_file

backend/tests/unit/test_scheduler_watchdog.py

@@ -76,11 +76,25 @@ class TestWatchdogExitsEarlyOnPickup:
             assert item.status == "printing"
 
     @pytest.mark.asyncio
-    async def test_exits_on_subtask_id_change_even_if_state_still_finish(self, db_session):
-        """Regression for #1078: H2D keeps state=FINISH for ~50 s after accepting
-        project_file, but subtask_id flips to our new submission_id almost
-        immediately. That must short-circuit the revert."""
-        get_status = MagicMock(return_value=_status("FINISH", "NEW_SUBTASK_12345"))
+    async def test_h2d_finish_to_running_via_subtask_id_then_active_state(self, db_session):
+        """Regression for #1078 (preserved through the two-phase rewrite for #1678):
+
+        H2D keeps state=FINISH for ~50 s after accepting project_file, but
+        subtask_id flips to our new submission_id almost immediately. The
+        watchdog must NOT revert on the basis of state staying at FINISH —
+        Phase A exits on the subtask_id advance, Phase B then keeps watching
+        and exits SUCCESS as soon as the printer transitions to PREPARE /
+        RUNNING within the longer Phase B window.
+        """
+        # First poll: state still FINISH, subtask_id advanced (Phase A → B).
+        # Second poll: state has flipped to RUNNING (Phase B success).
+        get_status = MagicMock(
+            side_effect=[
+                _status("FINISH", "NEW_SUBTASK_12345"),
+                _status("RUNNING", "NEW_SUBTASK_12345"),
+            ]
+            + [_status("RUNNING", "NEW_SUBTASK_12345")] * 10,
+        )
         with (
             patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
             patch("backend.app.services.print_scheduler.async_session", db_session),
@@ -91,15 +105,16 @@ class TestWatchdogExitsEarlyOnPickup:
                 pre_state="FINISH",
                 pre_subtask_id="OLD_SUBTASK_99999",
                 timeout=0.3,
+                phase_b_timeout=0.3,
                 poll_interval=0.05,
             )
 
         async with db_session() as db:
             item = await db.get(PrintQueueItem, 1)
             assert item.status == "printing", (
-                "subtask_id advanced past pre_subtask_id — the printer accepted our "
-                "project_file and the watchdog must not revert the queue item even "
-                "though state is still FINISH (#1078)"
+                "Phase A exit on subtask_id advance + Phase B observing the "
+                "active-state transition is the H2D success path — watchdog "
+                "must keep the item 'printing' (#1078)"
             )
 
 
@@ -228,6 +243,65 @@ class TestWatchdogRevertsWhenStuck:
         sig = inspect.signature(PrintScheduler._watchdog_print_start)
         assert sig.parameters["timeout"].default == 90.0
 
+    @pytest.mark.asyncio
+    async def test_default_phase_b_timeout_is_180_seconds(self):
+        """Phase B (subtask_id advanced, waiting for active state) must
+        comfortably exceed the H2D FINISH→PREPARE delay (~50 s observed)
+        before declaring a printer-side wedge. 180 s gives ~3.5× headroom
+        and reverts the queue item in well under the previous 2-hour
+        expected_print TTL (#1678)."""
+        import inspect
+
+        sig = inspect.signature(PrintScheduler._watchdog_print_start)
+        assert sig.parameters["phase_b_timeout"].default == 180.0
+
+    @pytest.mark.asyncio
+    async def test_reverts_when_subtask_advanced_but_state_never_active(self, db_session):
+        """Regression for #1678: P1S on old firmware, power-cycled mid-print,
+        cloud+LAN re-auth dance in flight. Printer accepts project_file
+        (gcode_file updates, subtask_id advances to our submission id) but
+        never transitions from IDLE/FINISH to PREPARE/RUNNING. The pre-fix
+        watchdog returned SUCCESS as soon as subtask_id advanced and the
+        queue item stayed in 'printing' until container restart. Phase B now
+        keeps watching; if the active-state transition never arrives, the
+        item reverts to 'pending' so the user can retry without restarting.
+        """
+        get_status = MagicMock(
+            return_value=_status("IDLE", "NEW_SUBTASK_12345", gcode_file="/new.3mf"),
+        )
+        client = MagicMock()  # NOT None — must verify reconnect isn't called
+        get_client = MagicMock(return_value=client)
+
+        with (
+            patch("backend.app.services.print_scheduler.printer_manager.get_status", get_status),
+            patch("backend.app.services.print_scheduler.printer_manager.get_client", get_client),
+            patch("backend.app.services.print_scheduler.async_session", db_session),
+            patch("backend.app.core.database.async_session", db_session),
+        ):
+            await PrintScheduler._watchdog_print_start(
+                queue_item_id=1,
+                printer_id=42,
+                pre_state="IDLE",
+                pre_subtask_id="OLD_SUBTASK_99999",
+                pre_gcode_file="/old.3mf",
+                timeout=0.2,
+                phase_b_timeout=0.2,
+                poll_interval=0.05,
+            )
+
+        async with db_session() as db:
+            item = await db.get(PrintQueueItem, 1)
+            assert item.status == "pending", (
+                "subtask_id advanced (Phase A → B) but state never reached an "
+                "active value — printer-side wedge; the queue item must be "
+                "reverted to 'pending' (#1678)"
+            )
+            assert item.started_at is None
+
+        # File landed (subtask_id advance proves this), so a forced reconnect
+        # would trigger 0500_4003 mid-parse (#1150) — skip.
+        client.force_reconnect_stale_session.assert_not_called()
+
 
 class TestWatchdogFallbackBehaviour:
     """Backwards-compat and defensive behaviour around missing data."""