From 9d0418688c4685964866134ee2b26fc507f72b53 Mon Sep 17 00:00:00 2001
From: maziggy <mz@v8w.de>
Date: Sun, 26 Apr 2026 08:58:08 +0200
Subject: [PATCH]   fix(#1134): propagate background-dispatch watchdog timeout
 as job failure
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

  Follow-up to #1042. The post-dispatch watchdog _verify_print_response was
  fire-and-forget — it correctly detected when the printer never transitioned
  (HMS error pending, half-broken MQTT session, plate-clear gate, SD card
  fault) and force-reconnected the MQTT session, but the dispatch job had
  already been marked successful on the optimistic MQTT-publish-acknowledged
  path. The UI carried on showing "Print started successfully" while the
  printer sat idle.

  The watchdog now returns bool and is awaited inline by both call sites in
  _run_reprint_archive and _run_print_library_file. On False the call sites
  raise a RuntimeError carrying a user-actionable message ("Printer did not
  acknowledge print command — state still {pre_state}. Check the printer for
  a pending error...") which routes through the existing _run_active_job →
  _mark_job_finished(failed=True) → background_dispatch WS broadcast path.
  Library-file flow rolls back the freshly-created archive on timeout so no
  phantom row is left behind for a print that never started.

  The watchdog now also accepts subtask_id advancing past pre_subtask_id as a
  definitive "command landed" signal — same as the queue-side watchdog at
  print_scheduler.py:1992 — so slow H2D FINISH→PREPARE transitions (~50 s
  observed) don't false-fail when the printer has clearly accepted the
  project_file but is still in FINISH. Default timeout raised from 15 s to
  90 s to match the queue-side watchdog and give the same headroom on both
  dispatch paths. Brief mid-window MQTT disconnects keep polling instead of
  immediately failing — matches what the queue watchdog already does and
  avoids false-failing on transient telemetry gaps.

  11 new tests in test_background_dispatch_watchdog.py: state-change pickup,
  subtask_id-change pickup with state still FINISH, neither-changed timeout
  plus force_reconnect_stale_session call, pre_subtask_id=None backwards-
  compat, post-dispatch subtask_id=None not counting as a change, brief
  disconnect not short-circuiting the window, persistent disconnect for the
  full window returning False, default-timeout=90s contract, _run_reprint_archive
  raises RuntimeError with the captured pre-state args on watchdog False,
  _run_reprint_archive happy path doesn't rollback, _run_active_job marks
  the job failed with the message when _process_job raises RuntimeError.
---
 CHANGELOG.md                                  |   2 +
 backend/app/services/background_dispatch.py   |  91 ++-
 .../test_background_dispatch_watchdog.py      | 521 ++++++++++++++++++
 3 files changed, 600 insertions(+), 14 deletions(-)
 create mode 100644 backend/tests/unit/services/test_background_dispatch_watchdog.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a562e773c..205b3a1c7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -28,6 +28,8 @@ All notable changes to Bambuddy will be documented in this file.
 - **Settings page: permission-gated instead of admin-only** — the Settings sidebar entry has always been visible to any user holding `settings:read`, but the route guard required admin role, so a non-admin with `settings:read` would see the entry, click it, and get silently redirected back to the dashboard. The route guard now matches the sidebar: any user with `settings:read` can open the page, and the individual tabs / cards continue to enforce their own per-feature permissions (`users:read`, `groups:update`, `oidc:*`, etc. — many of them admin-only, some not). Group editor routes moved to permission-based guards too (`groups:create` for `/groups/new`, `groups:update` for `/groups/:id/edit`), so permission delegation works end-to-end. Admins retain full access since admins implicitly hold every permission.
 
 ### Fixed
+- **Background-dispatch reported "Print started successfully" when the printer never actually transitioned** ([#1134](https://github.com/maziggy/bambuddy/issues/1134), follow-up to [#1042](https://github.com/maziggy/bambuddy/issues/1042)) — The int32 `task_id` modulo fix that was the original root cause of #1042 is verified working in the reporter's most recent support pack (the published `task_id` values are well below 2^31-1 and match the `int(time.time() * 1000) % 2_147_483_647` formula exactly). The remaining residual — "the UI reports despatch success which is slightly misleading" — was a real second bug class: the post-dispatch watchdog `_verify_print_response` in `services/background_dispatch.py` was *fire-and-forget*. It would correctly detect that the printer never transitioned (e.g. P1S sitting in `gcode_state: FAILED` with HMS `0300_400C` "task was canceled", a half-broken MQTT session, an SD card error, or any other pre-print blocker), log a `did not respond to print command within 15s` warning, force-reconnect the MQTT session — and then return without touching the dispatch job state. The dispatch job had already been marked successful on the optimistic MQTT-publish-acknowledged path, so the UI carried on showing "Print started successfully" while the printer sat idle. The watchdog now returns a `bool` and is awaited inline by both call sites (`_run_reprint_archive` at line 687, `_run_print_library_file` at line 860); on `False` (timeout) the call sites raise a `RuntimeError` carrying a user-actionable message ("Printer did not acknowledge print command — state still {pre_state}. Check the printer for a pending error (HMS code, plate-clear prompt, SD card) and try again."), which routes through the existing `_mark_job_finished(failed=True, …)` path so the dispatch UI shows a real failure toast and the library-file flow's freshly-created archive is `db.rollback()`'d (no orphan rows for prints that never started). The watchdog now also accepts `subtask_id` advancing past the captured `pre_subtask_id` as a definitive "command landed" signal — same as the queue-side watchdog at `print_scheduler.py:1992` (#1078) — so slow H2D `FINISH→PREPARE` transitions (~50 s observed) don't false-fail when the printer has clearly accepted the project_file but is still in FINISH. Default timeout raised from 15 s to 90 s to match the queue-side watchdog (#967 / #1078) and give the same headroom on both dispatch paths. Brief mid-window MQTT disconnects (`get_status() is None` for one tick) now keep polling instead of immediately failing — matches what the queue watchdog already does and avoids false-failing on transient telemetry gaps. The existing `force_reconnect_stale_session` recovery is preserved on the timeout path. 8 new regression tests in `test_background_dispatch_watchdog.py` cover state-change pickup, subtask_id-change pickup with state still FINISH (the H2D case), neither-signal-changed timeout + force-reconnect, pre_subtask_id=None backwards-compat, post-dispatch subtask_id=None not counting as a change (avoids false-pass on transient reconnect), brief disconnect not short-circuiting the window, persistent disconnect for the full window returning False, and a contract test that the default timeout is 90 s. Thanks to @EdwardChamberlain for the detailed retest with logs that pinpointed the watchdog's no-propagation gap.
+
 - **Bambu RFID auto-match created duplicate inventory rows for Quick-Add and non-Bambu-branded spools** ([#918](https://github.com/maziggy/bambuddy/issues/918)) — `find_matching_untagged_spool` is supposed to attach a Bambu RFID UID to a pre-existing manually-logged spool of the same material/color so users who log inventory before scanning don't end up with a duplicate row on first AMS read. Two bugs in the matcher meant it almost never worked for the actual reporting workflow: **(1)** the subtype filter was strict — when the AMS tray reports `tray_sub_brands="PLA Basic"` the matcher required `Spool.subtype = 'Basic'` exactly, so any Quick-Add row (Quick-Add only requires `material`, leaving `subtype=NULL`) was excluded and duplicated on first AMS read. **(2)** the docstring claimed it filtered on brand but the WHERE clause didn't, so a same-color *Polymaker* untagged spool would silently acquire a Bambu Lab tray UUID, leaving the user with `brand="Polymaker"` but a Bambu UUID — silent data corruption. Both bugs are addressed in the same query: subtype now prefers an exact match but accepts a NULL-subtype row as fallback (with a `CASE` in `ORDER BY` so an exact match still wins when both exist), and brand is now restricted to "contains 'bambu' (case-insensitive)" or NULL — matching `'Bambu'` (the form's `DEFAULT_BRANDS` value), `'Bambu Lab'` (the catalog value), `'BambuLab'`, `'bambu lab'`, etc., while rejecting any explicitly-named third-party brand. 6 new regression tests in `test_spool_tag_matcher.py` cover the NULL-subtype fallback, exact-subtype-wins-over-NULL ordering, non-Bambu brand rejection, NULL brand acceptance, all four Bambu brand spelling variants, and the full Quick-Add scenario (`brand=NULL` + `subtype=NULL`). The broader UI proposals in #918 (manual override / merge / disambiguation prompt) are intentionally out of scope — once the matcher works, the duplicate-on-RFID complaint that motivated those proposals goes away. Thanks to @ViridityCorn for the report and pointing at the right function, and to @Arn0uDz for confirming with a 20-spool repro.
 - **Swagger UI link in Settings → API Keys rendered a blank page** — the global CSP applied by `security_headers_middleware` set `script-src 'self'` and `style-src 'self' 'unsafe-inline' https://fonts.googleapis.com`, which blocked both the inline `<script>` that boots Swagger and the `cdn.jsdelivr.net` URL that ships `swagger-ui-bundle.js` / `swagger-ui.css`. FastAPI's `/docs` page therefore loaded a 1 KB shell with no JS executed, leaving an empty white page. The middleware now emits a docs-scoped CSP for `/docs`, `/redoc`, and `/docs/oauth2-redirect` that allows `https://cdn.jsdelivr.net` for scripts + styles, the FastAPI/Redoc favicon hosts for images, and `'unsafe-inline'` for the Swagger boot script — every other route keeps the unchanged stricter SPA policy.
 - **Camera stream second viewer fails / kicks the first off** ([#1089](https://github.com/maziggy/bambuddy/issues/1089)) — Most Bambu Lab printers only allow one concurrent camera connection (RTSP socket on X1/H2/P2, port-6000 chamber-image socket on A1/P1), but `GET /printers/{id}/camera/stream` opened a fresh upstream per viewer keyed on a per-request `stream_id`. Two browser tabs / two dashboard cards → the second viewer either failed silently or kicked the first one off. New `services/camera_fanout.py::MjpegBroadcaster` owns a single upstream per printer and fans pre-formatted MJPEG chunks out to N subscriber queues; new viewers tap the existing connection. When the last subscriber leaves, the upstream stays alive for a 5 s grace window so a tab refresh or "open in new tab" doesn't pay an ffmpeg/RTSP reconnect, then tears down cleanly. Per-subscriber queues are bounded (depth 4) so a slow viewer drops frames for itself rather than blocking the broadcaster — live video, old frames have no value. Stop endpoint and app-shutdown both call into the broadcaster's force-shutdown path so subscribers wake up via an upstream-gone sentinel instead of hanging on `queue.get()`. External-camera path is unchanged (user-supplied MJPEG/RTSP servers handle multi-viewer themselves). The upstream uses a deterministic `{printer_id}-fanout` stream id so every existing prefix-match in `cleanup_orphaned_streams`, `camera_status`, the snapshot fall-through in `main.py`, and the `stop` endpoint continues to find it without changes. Two follow-up correctness fixes from the audit pass: (1) `_stream_start_times[printer_id]` is now set with `setdefault()` so `/camera/status` reports the SHARED upstream's age — previously each new viewer overwrote it, making `stream_uptime` jump backward whenever a second viewer attached; (2) the route now retries `subscribe()` once on `RuntimeError` to close a tiny race where the grace teardown can flip the broadcaster to `stopped` between the registry lookup and the subscribe call (the retry forces the registry to mint a fresh broadcaster). Detach log line shows the post-unsubscribe count returned atomically by `unsubscribe()` — no more two viewers leaving simultaneously both reporting `subscribers=0`. Permission gates unchanged: `/camera/stream` still requires the existing token (minted by `POST /camera/stream-token` with `CAMERA_VIEW`); `/camera/stop` still requires `CAMERA_VIEW`; the broadcaster is internal infra with no FastAPI surface. 13 unit tests for the broadcaster (single subscriber, multi-subscriber-shares-one-pump, slow-subscriber-doesn't-block-fast, grace-window teardown, grace-cancelled-on-rejoin, force-shutdown sentinel, `iter_subscriber` exits on upstream-gone and on client-disconnect, registry replaces stopped broadcasters, `subscribe()` raises on stopped broadcaster, `unsubscribe()` returns post-removal count atomically across concurrent leavers, double-unsubscribe is idempotent, and the route's force-shutdown-then-fresh-subscribe retry path) plus 2 new integration tests on the stop endpoint covering the deterministic fan-out stream id and the `shutdown_broadcaster` wiring. Thanks to @swheettaos for the diagnosis and broadcaster sketch.
diff --git a/backend/app/services/background_dispatch.py b/backend/app/services/background_dispatch.py
index 2a7aff350..f9e72bc56 100644
--- a/backend/app/services/background_dispatch.py
+++ b/backend/app/services/background_dispatch.py
@@ -684,9 +684,34 @@ class BackgroundDispatchService:
                     )
                     raise RuntimeError("Failed to start print")
 
-                pre_state = getattr(printer_manager.get_status(job.printer_id), "state", None)
+                # Wait for the printer to actually pick up the command before
+                # marking the dispatch job complete (#1042). MQTT-publish success
+                # only proves the command queued locally; the printer can still
+                # reject it (HMS error pending, half-broken session, SD card
+                # missing) and never transition. Until #1042 this watchdog was
+                # fire-and-forget — the job was reported successful and the
+                # user had no signal that the print never started. The uploaded
+                # file is intentionally left on the printer's SD card on
+                # timeout: the next dispatch will overwrite it via the existing
+                # delete-then-upload step, and the printer may still be in the
+                # middle of reading it if it picked up just past the timeout.
+                pre_status = printer_manager.get_status(job.printer_id)
+                pre_state = getattr(pre_status, "state", None) if pre_status else None
+                pre_subtask_id = getattr(pre_status, "subtask_id", None) if pre_status else None
                 if pre_state:
-                    asyncio.create_task(self._verify_print_response(job.printer_id, printer_name, pre_state))
+                    await self._set_active_message(job, f"Waiting for {printer_name} to acknowledge print...")
+                    transitioned = await self._verify_print_response(
+                        job.printer_id,
+                        printer_name,
+                        pre_state,
+                        pre_subtask_id=pre_subtask_id,
+                    )
+                    if not transitioned:
+                        raise RuntimeError(
+                            f"Printer did not acknowledge print command — state still {pre_state}. "
+                            f"Check the printer for a pending error (HMS code, plate-clear prompt, "
+                            f"SD card) and try again."
+                        )
 
                 if job.requested_by_user_id and job.requested_by_username:
                     printer_manager.set_current_print_user(
@@ -857,9 +882,28 @@ class BackgroundDispatchService:
                     await db.rollback()
                     raise RuntimeError("Failed to start print")
 
-                pre_state = getattr(printer_manager.get_status(job.printer_id), "state", None)
+                # See _run_reprint_archive for rationale (#1042). On timeout
+                # also rolls back the freshly-created archive so the library
+                # flow doesn't leave behind a phantom row for a print that
+                # never started.
+                pre_status = printer_manager.get_status(job.printer_id)
+                pre_state = getattr(pre_status, "state", None) if pre_status else None
+                pre_subtask_id = getattr(pre_status, "subtask_id", None) if pre_status else None
                 if pre_state:
-                    asyncio.create_task(self._verify_print_response(job.printer_id, printer_name, pre_state))
+                    await self._set_active_message(job, f"Waiting for {printer_name} to acknowledge print...")
+                    transitioned = await self._verify_print_response(
+                        job.printer_id,
+                        printer_name,
+                        pre_state,
+                        pre_subtask_id=pre_subtask_id,
+                    )
+                    if not transitioned:
+                        await db.rollback()
+                        raise RuntimeError(
+                            f"Printer did not acknowledge print command — state still {pre_state}. "
+                            f"Check the printer for a pending error (HMS code, plate-clear prompt, "
+                            f"SD card) and try again."
+                        )
 
                 if job.requested_by_user_id and job.requested_by_username:
                     printer_manager.set_current_print_user(
@@ -899,38 +943,57 @@ class BackgroundDispatchService:
         printer_id: int,
         printer_name: str,
         pre_state: str,
-        timeout: float = 15.0,
+        pre_subtask_id: str | None = None,
+        timeout: float = 90.0,
         poll_interval: float = 3.0,
-    ):
-        """Check if the printer responded to a print command.
+    ) -> bool:
+        """Wait for the printer to acknowledge a print command.
 
-        Runs as a fire-and-forget background task after start_print() succeeds.
-        If the printer's gcode_state hasn't changed within the timeout, logs a
-        warning for diagnostics (visible in support packages).
+        Returns True if the printer transitioned (state advanced past pre_state
+        or subtask_id advanced past pre_subtask_id). Returns False on timeout —
+        in that case logs a warning and forces an MQTT reconnect, mirroring the
+        queue-side watchdog (`_watchdog_print_start`). Caller is responsible
+        for surfacing the False result to the user (typically by raising so the
+        dispatch job is marked failed).
+
+        Both transition signals are checked because H2D can sit at FINISH for
+        ~50 s after accepting `project_file` before flipping to PREPARE; the
+        printer echoes our per-dispatch identity back as `subtask_id` on
+        `push_status` first, so a subtask_id change is a definitive "command
+        landed" signal even while state is still FINISH (#1078).
         """
         deadline = time.monotonic() + timeout
         while time.monotonic() < deadline:
             await asyncio.sleep(poll_interval)
             state = printer_manager.get_status(printer_id)
             if not state:
-                return  # Printer disconnected
+                # Printer momentarily not reporting — could be a brief MQTT
+                # disconnect mid-window. Keep polling rather than declaring
+                # failure on the first missed tick; the printer may reconnect
+                # within the remaining timeout and still surface a transition.
+                continue
             if state.state != pre_state:
-                return  # Printer responded
+                return True
+            if pre_subtask_id is not None and state.subtask_id is not None and state.subtask_id != pre_subtask_id:
+                return True
         logger.warning(
-            "Printer %s (%d) did not respond to print command within %.0fs (state still %s) — printer may need restart",
+            "Printer %s (%d) did not respond to print command within %.0fs "
+            "(state still %s, subtask_id still %s) — printer may need restart",
             printer_name,
             printer_id,
             timeout,
             pre_state,
+            pre_subtask_id,
         )
         # Strong signal the MQTT session is half-broken (#887, #936): telemetry
         # still arrives but our publishes don't reach the printer. Force a fresh
         # session so the next dispatch can land without a power cycle.
         client = printer_manager.get_client(printer_id)
-        if client:
+        if client and hasattr(client, "force_reconnect_stale_session"):
             client.force_reconnect_stale_session(
                 f"print command unacknowledged after {timeout:.0f}s (state still {pre_state})"
             )
+        return False
 
     @staticmethod
     async def _cleanup_sd_card_file(
diff --git a/backend/tests/unit/services/test_background_dispatch_watchdog.py b/backend/tests/unit/services/test_background_dispatch_watchdog.py
new file mode 100644
index 000000000..39fc72637
--- /dev/null
+++ b/backend/tests/unit/services/test_background_dispatch_watchdog.py
@@ -0,0 +1,521 @@
+"""Regression tests for ``BackgroundDispatchService._verify_print_response``.
+
+The background-dispatch watchdog used to be fire-and-forget — it logged a
+warning and force-reconnected MQTT, but the dispatch job had already been
+marked successful. The user therefore saw "Print started successfully" while
+the printer never actually transitioned (#1042 follow-up). The watchdog now
+returns a bool so the caller can fail the dispatch job when the printer
+doesn't acknowledge the command, mirroring what `_watchdog_print_start` does
+on the queue side.
+
+Both transition signals are accepted: ``state`` advancing past ``pre_state``
+*or* ``subtask_id`` advancing past ``pre_subtask_id`` — H2D firmware can sit
+at FINISH for ~50 s after accepting ``project_file`` while echoing the new
+subtask_id back almost immediately (#1078).
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from backend.app.services.background_dispatch import BackgroundDispatchService
+
+
+def _status(state: str, subtask_id: str | None = None):
+    """Minimal stand-in for PrinterState — only the two fields the watchdog reads."""
+    return SimpleNamespace(state=state, subtask_id=subtask_id)
+
+
+class TestReturnsTrueOnPickup:
+    @pytest.mark.asyncio
+    async def test_returns_true_on_state_change(self):
+        get_status = MagicMock(return_value=_status("RUNNING", "OLD_SUBTASK"))
+        with patch(
+            "backend.app.services.background_dispatch.printer_manager.get_status",
+            get_status,
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="P1S",
+                pre_state="FINISH",
+                pre_subtask_id="OLD_SUBTASK",
+                timeout=0.3,
+                poll_interval=0.05,
+            )
+
+        assert result is True
+
+    @pytest.mark.asyncio
+    async def test_returns_true_on_subtask_id_change_even_if_state_still_finish(self):
+        """#1078: H2D keeps state=FINISH for ~50 s after accepting project_file
+        but flips subtask_id immediately. Must be accepted as a pickup signal."""
+        get_status = MagicMock(return_value=_status("FINISH", "NEW_SUBTASK_12345"))
+        with patch(
+            "backend.app.services.background_dispatch.printer_manager.get_status",
+            get_status,
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="H2D",
+                pre_state="FINISH",
+                pre_subtask_id="OLD_SUBTASK_99999",
+                timeout=0.3,
+                poll_interval=0.05,
+            )
+
+        assert result is True
+
+
+class TestReturnsFalseOnTimeout:
+    @pytest.mark.asyncio
+    async def test_returns_false_when_neither_state_nor_subtask_id_changes(self):
+        """The exact #1042 scenario: P1S sits in FAILED with HMS pending,
+        accepts the MQTT publish, never transitions. Watchdog must report
+        failure so the caller fails the dispatch job."""
+        get_status = MagicMock(return_value=_status("FINISH", "OLD_SUBTASK"))
+        client = MagicMock()
+        get_client = MagicMock(return_value=client)
+
+        with (
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_status",
+                get_status,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_client",
+                get_client,
+            ),
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="P1S",
+                pre_state="FINISH",
+                pre_subtask_id="OLD_SUBTASK",
+                timeout=0.2,
+                poll_interval=0.05,
+            )
+
+        assert result is False
+        client.force_reconnect_stale_session.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_returns_false_when_pre_subtask_id_none_and_state_unchanged(self):
+        """Backward-compat: callers without a captured pre_subtask_id (e.g. the
+        printer never reported one) must still get the timeout failure path
+        based on state alone."""
+        get_status = MagicMock(return_value=_status("FINISH", "ANYTHING"))
+        get_client = MagicMock(return_value=None)
+
+        with (
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_status",
+                get_status,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_client",
+                get_client,
+            ),
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="P1S",
+                pre_state="FINISH",
+                pre_subtask_id=None,
+                timeout=0.2,
+                poll_interval=0.05,
+            )
+
+        assert result is False
+
+    @pytest.mark.asyncio
+    async def test_subtask_id_none_post_dispatch_does_not_count_as_change(self):
+        """If the printer transiently reports subtask_id=None during the
+        watchdog window (e.g. mid-reconnect), that must not be treated as
+        "advanced past pre_subtask_id" — otherwise we'd false-pass and mark
+        a never-started print as successful."""
+        get_status = MagicMock(return_value=_status("FINISH", None))
+        get_client = MagicMock(return_value=None)
+
+        with (
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_status",
+                get_status,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_client",
+                get_client,
+            ),
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="P1S",
+                pre_state="FINISH",
+                pre_subtask_id="OLD_SUBTASK",
+                timeout=0.2,
+                poll_interval=0.05,
+            )
+
+        assert result is False
+
+
+class TestDisconnectHandling:
+    @pytest.mark.asyncio
+    async def test_disconnect_does_not_short_circuit_window(self):
+        """A momentary ``get_status() is None`` (brief MQTT disconnect mid-window)
+        must not immediately fail the dispatch — the printer may reconnect and
+        still produce a valid transition before timeout. Falsely failing on the
+        first missed tick is the previous bug class we're moving away from."""
+        # First call: disconnected. Second call onward: reconnected and transitioned.
+        get_status = MagicMock(side_effect=[None, _status("RUNNING")])
+        with patch(
+            "backend.app.services.background_dispatch.printer_manager.get_status",
+            get_status,
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="P1S",
+                pre_state="FINISH",
+                pre_subtask_id="OLD_SUBTASK",
+                timeout=0.3,
+                poll_interval=0.05,
+            )
+
+        assert result is True
+        assert get_status.call_count >= 2
+
+    @pytest.mark.asyncio
+    async def test_disconnect_for_full_window_returns_false(self):
+        """Persistent disconnect for the full window is treated as failure.
+        Better to false-fail and let the user retry than to false-succeed and
+        leave them watching an idle printer (#1042)."""
+        get_status = MagicMock(return_value=None)
+        get_client = MagicMock(return_value=None)
+
+        with (
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_status",
+                get_status,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_client",
+                get_client,
+            ),
+        ):
+            result = await BackgroundDispatchService._verify_print_response(
+                printer_id=42,
+                printer_name="P1S",
+                pre_state="FINISH",
+                pre_subtask_id="OLD_SUBTASK",
+                timeout=0.2,
+                poll_interval=0.05,
+            )
+
+        assert result is False
+
+
+class TestDefaults:
+    def test_default_timeout_matches_queue_watchdog(self):
+        """Queue and background watchdogs need the same 90 s default to give
+        slow H2D FINISH→PREPARE transitions the same headroom on both paths."""
+        import inspect
+
+        sig = inspect.signature(BackgroundDispatchService._verify_print_response)
+        assert sig.parameters["timeout"].default == 90.0
+
+
+# ---------------------------------------------------------------------------
+# Integration tests: the call sites in _run_reprint_archive and
+# _run_print_library_file must (a) await the watchdog instead of fire-and-
+# forget, (b) raise RuntimeError on watchdog False so _run_active_job marks
+# the job failed, (c) rollback the library-file flow's freshly-created
+# archive on timeout. Heavy mocking — the goal is to verify the new wiring,
+# not to re-test the dependencies.
+# ---------------------------------------------------------------------------
+
+from contextlib import asynccontextmanager  # noqa: E402
+from unittest.mock import AsyncMock  # noqa: E402
+
+from backend.app.services.background_dispatch import (  # noqa: E402
+    ActiveDispatchState,
+    PrintDispatchJob,
+)
+
+
+def _make_session_factory(db_mock):
+    """Build an async-session factory whose context manager yields ``db_mock``.
+
+    Mirrors the ``async with async_session() as db`` shape used by both
+    ``_run_*`` methods so the test can intercept ``db.rollback`` / ``db.scalar``.
+    """
+
+    @asynccontextmanager
+    async def _factory():
+        yield db_mock
+
+    return _factory
+
+
+def _printer_namespace():
+    return SimpleNamespace(
+        id=10,
+        name="P1S",
+        ip_address="1.2.3.4",
+        access_code="abc",
+        model="P1S",
+    )
+
+
+def _make_dispatch_job(kind: str = "reprint_archive") -> PrintDispatchJob:
+    return PrintDispatchJob(
+        id=1,
+        kind=kind,
+        source_id=99,
+        source_name="Test.gcode.3mf",
+        printer_id=10,
+        printer_name="P1S",
+        options={},
+        requested_by_user_id=None,
+        requested_by_username=None,
+    )
+
+
+@pytest.fixture
+def reprint_archive_mocks(tmp_path):
+    """Mock harness for ``_run_reprint_archive`` covering every external
+    dependency up to (and including) ``start_print``. The watchdog is left
+    real so the caller can patch ``_verify_print_response`` per-test."""
+    archive_file = tmp_path / "test.3mf"
+    archive_file.write_bytes(b"fake-3mf-content")
+
+    archive = SimpleNamespace(
+        id=99,
+        filename="Test.gcode.3mf",
+        file_path=str(archive_file),
+    )
+
+    db = MagicMock()
+    db.scalar = AsyncMock(return_value=_printer_namespace())
+    db.rollback = AsyncMock()
+
+    archive_service = MagicMock()
+    archive_service.get_archive = AsyncMock(return_value=archive)
+
+    return {
+        "archive": archive,
+        "archive_file": archive_file,
+        "db": db,
+        "archive_service": archive_service,
+        "session_factory": _make_session_factory(db),
+    }
+
+
+@pytest.fixture
+def library_file_mocks(tmp_path):
+    """Mock harness for ``_run_print_library_file`` — separate from the
+    reprint fixture because the library flow creates its archive via
+    ``archive_service.archive_print(...)`` rather than fetching one."""
+    src_file = tmp_path / "lib_src.3mf"
+    src_file.write_bytes(b"fake-3mf-content")
+
+    lib_file = SimpleNamespace(
+        id=22,
+        filename="cube.gcode.3mf",
+        file_path=str(src_file.relative_to(tmp_path)),
+    )
+    lib_file.active = staticmethod(lambda: lib_file)  # mimic LibraryFile.active() chainable
+
+    new_archive = SimpleNamespace(id=500, filename="cube.gcode.3mf", file_path=str(src_file))
+
+    db = MagicMock()
+    db.scalar = AsyncMock()  # configured per-test
+    db.flush = AsyncMock()
+    db.commit = AsyncMock()
+    db.rollback = AsyncMock()
+
+    archive_service = MagicMock()
+    archive_service.archive_print = AsyncMock(return_value=new_archive)
+
+    return {
+        "lib_file": lib_file,
+        "src_file": src_file,
+        "new_archive": new_archive,
+        "db": db,
+        "archive_service": archive_service,
+        "session_factory": _make_session_factory(db),
+    }
+
+
+class TestReprintArchiveDispatchWiring:
+    """Verify ``_run_reprint_archive`` (a) awaits the watchdog inline and
+    (b) raises RuntimeError on False so the dispatch job is marked failed."""
+
+    @pytest.mark.asyncio
+    async def test_raises_runtime_error_when_watchdog_returns_false(self, reprint_archive_mocks):
+        """The exact #1042 propagation gap: watchdog detects non-transition,
+        _run_reprint_archive must surface it as a RuntimeError so the surrounding
+        _run_active_job marks the job failed (instead of silently completing)."""
+        from backend.app.services.background_dispatch import BackgroundDispatchService
+
+        m = reprint_archive_mocks
+        service = BackgroundDispatchService()
+        job = _make_dispatch_job(kind="reprint_archive")
+
+        watchdog = AsyncMock(return_value=False)
+
+        with (
+            patch("backend.app.services.background_dispatch.async_session", m["session_factory"]),
+            patch(
+                "backend.app.services.background_dispatch.ArchiveService",
+                return_value=m["archive_service"],
+            ),
+            patch.object(BackgroundDispatchService, "_verify_print_response", watchdog),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.is_connected",
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_status",
+                return_value=SimpleNamespace(state="FINISH", subtask_id="OLD_SUBTASK"),
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.start_print",
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.delete_file_async",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.with_ftp_retry",
+                new_callable=AsyncMock,
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.get_ftp_retry_settings",
+                new_callable=AsyncMock,
+                return_value=(False, 0, 0, 30.0),
+            ),
+            patch(
+                "backend.app.services.background_dispatch.upload_file_async",
+                new_callable=AsyncMock,
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.ws_manager.broadcast",
+                new_callable=AsyncMock,
+            ),
+            patch("backend.app.main.register_expected_print"),
+            pytest.raises(RuntimeError, match="did not acknowledge print command"),
+        ):
+            await service._run_reprint_archive(job)
+
+        # Watchdog received the captured pre-state and pre_subtask_id.
+        watchdog.assert_awaited_once()
+        kwargs = watchdog.await_args.kwargs
+        args = watchdog.await_args.args
+        assert "FINISH" in args  # pre_state
+        assert kwargs["pre_subtask_id"] == "OLD_SUBTASK"
+
+    @pytest.mark.asyncio
+    async def test_succeeds_when_watchdog_returns_true(self, reprint_archive_mocks):
+        """Happy path: watchdog confirms pickup; _run_reprint_archive returns
+        without raising. Guards against the wiring accidentally raising on True."""
+        from backend.app.services.background_dispatch import BackgroundDispatchService
+
+        m = reprint_archive_mocks
+        service = BackgroundDispatchService()
+        job = _make_dispatch_job(kind="reprint_archive")
+
+        with (
+            patch("backend.app.services.background_dispatch.async_session", m["session_factory"]),
+            patch(
+                "backend.app.services.background_dispatch.ArchiveService",
+                return_value=m["archive_service"],
+            ),
+            patch.object(
+                BackgroundDispatchService,
+                "_verify_print_response",
+                AsyncMock(return_value=True),
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.is_connected",
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.get_status",
+                return_value=SimpleNamespace(state="FINISH", subtask_id="OLD_SUBTASK"),
+            ),
+            patch(
+                "backend.app.services.background_dispatch.printer_manager.start_print",
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.delete_file_async",
+                new_callable=AsyncMock,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.with_ftp_retry",
+                new_callable=AsyncMock,
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.get_ftp_retry_settings",
+                new_callable=AsyncMock,
+                return_value=(False, 0, 0, 30.0),
+            ),
+            patch(
+                "backend.app.services.background_dispatch.upload_file_async",
+                new_callable=AsyncMock,
+                return_value=True,
+            ),
+            patch(
+                "backend.app.services.background_dispatch.ws_manager.broadcast",
+                new_callable=AsyncMock,
+            ),
+            patch("backend.app.main.register_expected_print"),
+        ):
+            await service._run_reprint_archive(job)  # must not raise
+
+        # Reprint flow does not touch the existing archive — no rollback expected.
+        m["db"].rollback.assert_not_called()
+
+
+class TestRunActiveJobMarksFailedOnRuntimeError:
+    """End-to-end: a watchdog-driven RuntimeError must reach
+    `_mark_job_finished(failed=True)` via the existing ``_run_active_job``
+    catch-all, so the dispatch UI shows a real failure (not "Done")."""
+
+    @pytest.mark.asyncio
+    async def test_runtime_error_from_process_job_marks_failed_with_message(self):
+        from backend.app.services.background_dispatch import BackgroundDispatchService
+
+        service = BackgroundDispatchService()
+        job = _make_dispatch_job()
+        # Place the job into _active_jobs so _set_active_message has a target.
+        service._active_jobs[job.id] = ActiveDispatchState(job=job, message="")
+
+        failure_message = (
+            "Printer did not acknowledge print command — state still FINISH. "
+            "Check the printer for a pending error (HMS code, plate-clear prompt, "
+            "SD card) and try again."
+        )
+
+        with (
+            patch.object(
+                BackgroundDispatchService,
+                "_process_job",
+                AsyncMock(side_effect=RuntimeError(failure_message)),
+            ),
+            patch.object(
+                BackgroundDispatchService,
+                "_mark_job_finished",
+                new_callable=AsyncMock,
+            ) as mark_finished,
+        ):
+            await service._run_active_job(job)
+
+        mark_finished.assert_awaited_once()
+        kwargs = mark_finished.await_args.kwargs
+        assert kwargs["failed"] is True
+        assert "did not acknowledge print command" in kwargs["message"]