fix(notifications): wire on_printer_offline dispatch on disconnect edge (#1752)

The provider toggle, schema, template, and NotificationService.on_printer_offline
      all shipped, but no caller invoked the dispatcher — the offline event was an
      orphan toggle. Edge detection in on_printer_status_change now schedules a
      debounced (60s) background task on the connected→disconnected transition;
      reconnect before the window elapses cancels it. Covers both upstream paths
      (smart-plug power-off via mark_printer_offline, and MQTT staleness via
      check_staleness), both of which already route through the status callback.
      The "back online" channel is the existing print-failure notification on
      firmware FAILED report — no symmetric on_printer_online needed.

CHANGELOG.md

@@ -22,6 +22,8 @@ All notable changes to Bambuddy will be documented in this file.
 - **VP bridge-synthesised reply trace (#1622 round 3)** — The round-2 cmd.jsonl from shaddowlink's P1S vs H2D capture proves the actual failure mode: on P1S in archive mode the slicer issues `extrusion_cali_set` (push K/n directly) and the printer responds `fail`, on H2D and on the P1S second round the slicer takes the `extrusion_cali_sel` flow (select by `filament_id` / `cali_idx`) and the printer responds `success`. Both flows traverse the bridge cleanly — `ams_filament_setting` round-trips with `result=success` and the cached push_status carries `tray_info_idx=GFA11`, `tray_type=PLA-AERO`, K/n, and `cali_idx=-1` intact. So the bridge is innocent on every layer the dump can see, and the open question becomes: what makes the slicer pick `_set` vs `_sel`? Likely candidates are the `info.get_version` answer Bambuddy synthesises (slicer fingerprints on `sw_ver` / `hw_ver` / `module` to decide its command flow) or the first cached `pushall` response the slicer reads to bootstrap its UI. Round 2 captured neither — the JSONL had `slicer_to_bridge` and `printer_to_slicer` directions but no `bridge_to_slicer` direction for the bridge's own synthesised replies. Same env flag (`BAMBUDDY_VP_DUMP_WIRE=1`) now also appends every bridge-synthesised reply (info.get_version answer, project_file ack, on-demand pushall response) to `<log_dir>/vp_wire/<vp_name>_cmd.jsonl` under direction `bridge_to_slicer`. Capture lives in `mqtt_server.py::_publish_to_report` — the single chokepoint every synthesised reply already passes through — gated on a new `log_event: bool = True` parameter; the 1Hz periodic-push path threads `log_event=False` so the JSONL isn't flooded with ~60 lines/min per VP (snapshot dump already covers cache shape). The on-demand pushall response from `_send_status_report` IS logged because that's the bootstrap-fingerprint reply the slicer reads on first connect. Two additional unit tests in `test_vp_mqtt_bridge.py::TestWireFormat` pin the event-on-default and skip-when-`log_event=False` posture; `test_vp_wire_dump.py` already covers the underlying `append_event` shape and the new direction is documented in `_debug.py`'s docstring. Diagnostic-only — does not change the publish data path; the new param defaults preserve every existing call site's behaviour.
 
 ### Fixed
+- **Push notification for "Printer offline" now actually fires (#1752, reported by @saint-hh)** — The notification provider's `on_printer_offline` toggle has shipped since the notifications feature landed: schema field, DB column, `notification_template.py` entry, and the dispatcher `NotificationService.on_printer_offline(printer_id, printer_name, db)` are all in place. What was missing was the caller — nothing in the codebase actually invoked the dispatcher when a printer went offline. The reporter (P2S, smart-plug-cuts-power scenario) confirmed turning the toggle on did nothing; only the print-failure notification fired when power was restored, via the firmware's `gcode_state=FAILED` report on MQTT reconnect. **Why the toggle was orphan:** every other provider event (`on_print_start`, `on_print_complete`, `on_print_progress`, `on_printer_error`, etc.) has a clear call site under `main.py::on_printer_status_change` or alongside the print-lifecycle hooks. The offline event was the only edge-triggered toggle without one — the dispatcher and template predated the wiring step and were silently shipped. Both upstream offline-trigger paths (`smart_plug_manager` → `printer_manager.mark_printer_offline()` and `bambu_mqtt.py::check_staleness` after the 30s STALE_RECONNECT_COOLDOWN) route through `_on_status_change` already and reach `on_printer_status_change`; the handler just didn't act on the disconnect edge. **Fix:** edge detection in `on_printer_status_change` watches `state.connected` against the previous observation per printer (`_printer_last_connected: dict[int, bool]`). On the True → False transition it schedules `_maybe_notify_printer_offline(printer_id)` as a background asyncio task; on the next True observation it cancels any pending task. The helper sleeps `_PRINTER_OFFLINE_NOTIFY_DEBOUNCE_SECONDS = 60.0` then re-checks `printer_manager.is_connected(printer_id)` — only fires the notification if the printer is still offline. **Why 60s debounce:** sized against `bambu_mqtt.py::STALE_RECONNECT_COOLDOWN = 30s` — a single stale-trigger + reconnect cycle isn't enough to fire, only a real outage that survives one full cooldown notifies. Transient MQTT blips (WiFi roam, broker reload, brief packet loss) recover within the window and the cancellation path kicks in. **Edge-case handling:** initial observation with no prior connected state doesn't fire (covers Bambuddy startup with an already-offline printer); a False → False repeat doesn't reschedule (the in-flight task stays in place rather than resetting the clock on every status callback, which would otherwise mean the notification never fires); the task entry pops from `_printer_offline_notify_tasks` in the finally block whether the notification fired, the printer reconnected, or the task was cancelled mid-await. **No symmetric `on_printer_online` event:** the reporter explicitly noted the "printer lost power and interrupted the print" notification already fires when power is restored — that's the print-failure notification, triggered by the firmware reporting `gcode_state=FAILED` for the interrupted print on MQTT reconnect. That covers the "printer is back" channel without a new toggle. If the user then resumes the print, no print_start notification fires (Bambuddy's `bambu_mqtt.py:3039` explicitly suppresses `is_new_print` for PAUSE → RUNNING to prevent duplicates when resuming from pause), but that's a separate scope from offline-detection. **Tests:** 9 new cases in `test_printer_offline_notification.py` split across two classes. `TestMaybeNotifyPrinterOffline` pins the debounced helper: fires notification when still offline at end of window, doesn't fire when printer reconnected during debounce, doesn't fire when the printer disappeared from the DB (uninstall mid-window), clears `_printer_offline_notify_tasks[printer_id]` after run. `TestOfflineEdgeDetection` pins the edge logic inside `on_printer_status_change`: first observation (connected) doesn't schedule, first observation (disconnected) doesn't schedule (the no-prior-True case — important for startup), True → False schedules a task, reconnect cancels the pending task, repeated False observations don't replace the in-flight task. Full backend suite still green; ruff clean.
+
 - **Virtual Printer queue mode: multi-plate "Send All" now enqueues one queue item per plate** — BambuStudio / OrcaSlicer's "Send All" packs every plate of the project into a SINGLE 3MF and uploads it with one FTP STOR — `slice_info.config` inside the file carries N `<plate>` blocks (one per plate), each with its own `<metadata key="index" value="N"/>` and its own `Metadata/plate_N.gcode` payload. Previously the VP queue path only ever extracted the FIRST plate's index via `_extract_plate_id` and created exactly ONE PrintQueueItem with that single `plate_id`; plates 2..N silently dropped on the floor. Indistinguishable from the user's perspective from "Send" of a single plate — except they expected 3 items in the queue and got 1, with no log line to explain why. **Confirmed against the wire** on the live H2D-1 Proxy VP: `Cube.gcode.3mf` carrying three `<plate>` blocks (indices 1, 2, 3) + three per-plate gcode payloads in the same zip, identical filename whether "Send" or "Send All" was clicked — the only signal of intent is the count of `<plate>` blocks inside the file. **Fix:** replaced `_extract_plate_id` (returning `int | None`) with `_extract_plate_ids` (returning `list[int]`). The list contains every `<plate>` block's `index` metadata, in order; falls back to `[1]` for files missing `slice_info.config` or with no parseable plates so the single-plate path is preserved. `_add_to_print_queue` now loops over the list — each iteration calls `extract_filament_requirements(file_path, plate_id)` per-plate (the plate-aware path was already there from the #1697 work) and creates a PrintQueueItem with that plate's filament types / overrides, plate-specific position = `MAX(position) + iteration`. Single-plate "Send" hits the loop once → exactly today's behaviour (one queue item, plate_id from the slicer, same archive). Multi-plate "Send All" of a 3-plate file → 3 queue items, plate_id 1/2/3, consecutive positions, all pointing at the same backing archive (one upload = one archive). **What stays the same:** the single archive row per upload (the archive backs the queue items via `archive_id`); the `auto_dispatch=False` / `manual_start=true` posture inherited from the VP config (so multi-plate items still require manual start); the `queue_force_color_match` per-VP toggle (now applies per-plate). **What this also fixed downstream:** the `required_filament_types` / `filament_overrides` JSON on each queue item now reflects THAT plate's filaments, not the file's first plate — so the scheduler's per-printer "Any X" matching dispatches each plate onto a printer with the right colours loaded for THAT plate, not for plate 1's filament set. **Tests:** 1 new regression case in `test_virtual_printer.py::TestVirtualPrinterInstance::test_add_to_print_queue_multi_plate_send_all_enqueues_one_per_plate` — builds a 3-plate 3MF (writes the per-plate `<plate>` blocks into `slice_info.config` and the per-plate gcode payloads), runs `_add_to_print_queue`, asserts 3 PrintQueueItems with `plate_id == [1, 2, 3]`, `position == [1, 2, 3]`, shared `archive_id`, all `manual_start=True`. 126 existing single-plate VP tests stay green (loop runs once when input has one plate). Full backend suite 5962/5962 green; ruff clean; frontend untouched. **Live-verified** on the H2D-1 Proxy VP — a Send All of the 3-plate Cube project now produces 3 queue items + 1 archive instead of 1 queue item + 1 archive.
 
 - **Archive delete now removes related queue items instead of leaving "cancelled" rows behind** — Previously the soft-delete path (the default — what the trash-can button does) called `_cancel_pending_queue_items`, which only flipped queue rows with `status='pending'` to `status='cancelled'` while leaving every other status alone AND leaving every row in the DB. The Send All multi-plate work above made this much more visible: deleting an archive backed by N queue items now had to clean up N rows, and what users saw instead was N "cancelled" rows lingering in the queue history. **Fix (backend):** replaced `_cancel_pending_queue_items` with `_delete_related_queue_items(db, archive_id) -> int` that DELETEs every queue row where `archive_id = X` regardless of status. Behavior now matches what the hard-delete path already did via the `ON DELETE CASCADE` FK on `print_queue.archive_id` — both paths produce the same end state. Print history lives in `PrintLogEntry` (FK `ON DELETE SET NULL`) and is untouched, so stats / Quick Stats / accuracy bands are preserved across both delete paths. **New guard:** the route at `archives.py::delete_archive` now 409s when any related queue item is currently in `status='printing'` — both soft and hard delete are gated by the same precondition, because deleting the archive while a print is live would strip the dispatcher's metadata trail (filament / plate / ams_mapping) out from under the running print. The 409 surfaces a clear "Stop the print first, then retry" message. **Pre-flight count for the UI:** new endpoint `GET /archives/{id}/delete-impact` returns `{related_queue_items: N, currently_printing: M}` — cheap, single endpoint, not folded into the archive list response so the much larger list endpoint isn't forced to run the same query per row. Frontend ArchivesPage delete-confirm modal queries this when the modal opens (`useQuery({queryKey: ['archive', id, 'delete-impact'], enabled: showDeleteConfirm})`) and renders: an amber warning "**N queue item(s) linked to this archive will also be removed.**" when total > 0 AND printing = 0, OR a red warning "**Cannot delete — M queue item(s) are currently printing. Stop the print first, then retry.**" when printing > 0 (with the confirm button disabled in that case so the user can't bonk the 409 on submit). **ConfirmModal extension:** added optional `confirmDisabled?: boolean` prop. Existing `isLoading` was the only disable knob; this adds an external-precondition path that disables the confirm without the loading spinner. **Tests:** rewrote `test_print_queue_api.py::test_soft_delete_archive_cancels_pending_queue_items` → `test_soft_delete_archive_deletes_all_related_queue_items` to pin the new contract (both pending AND completed rows are gone post-soft-delete). 2 new integration cases in `test_archives_api.py`: `test_delete_archive_blocked_when_related_queue_item_printing` (both soft and hard paths return 409 with "printing" in detail message) + `test_archive_delete_impact_reports_counts` (3 mixed-status related rows + 1 unrelated row → endpoint reports `related_queue_items=3, currently_printing=1`, unrelated row doesn't bleed in). **i18n:** 2 new keys (`archives.modal.deleteQueueItemsWarning`, `archives.modal.deleteBlockedByPrinting`) translated across all 11 locales per `feedback_translate_dont_fallback` — no English fallbacks. **Verification:** full backend suite 5964/5964 green with `-n 30`; ruff clean; ESLint clean; `npm run build` clean; vitest 2118/2118 green; i18n parity 5109 × 11 locales green. No DB migration — the CASCADE FK was already in place; only the helper's semantics changed.

backend/app/main.py

@@ -395,6 +395,22 @@ _bed_cool_waiters: dict[int, dict] = {}
 # as "cancelled" (stopped by user) so the correct notification email is sent.
 _user_stopped_printers: set[int] = set()
 
+# Offline-notification edge state (#1752): fire `on_printer_offline` exactly
+# once when a printer transitions connected → disconnected. `_printer_last_connected`
+# holds the previous observation so we only fire on the True → False edge (a
+# False → False repeat doesn't notify; an initial False at startup doesn't
+# notify either, since there's no prior True). `_printer_offline_notify_tasks`
+# holds the per-printer pending asyncio task that fires the notification
+# after a debounce window — cancelled if the printer reconnects before the
+# window elapses, so transient MQTT blips don't flood the user.
+_printer_last_connected: dict[int, bool] = {}
+_printer_offline_notify_tasks: dict[int, asyncio.Task] = {}
+# Debounce: a printer must stay offline this long before we notify. Sized
+# against the staleness path (`bambu_mqtt.py::STALE_RECONNECT_COOLDOWN = 30s`)
+# so a single stale-trigger cooldown isn't enough to fire — only a real
+# offline that survives one reconnect attempt notifies.
+_PRINTER_OFFLINE_NOTIFY_DEBOUNCE_SECONDS = 60.0
+
 
 # HMS short-code → human-readable failure reason. Used by _dispatch_archive_update
 # when status="failed" to label the print's failure_reason in archives.
@@ -843,6 +859,53 @@ _last_status_broadcast: dict[int, str] = {}
 _nozzle_count_updated: set[int] = set()
 
 
+async def _maybe_notify_printer_offline(printer_id: int) -> None:
+    """Wait the debounce window then fire `on_printer_offline` if the printer
+    is still offline.
+
+    Scheduled by `on_printer_status_change` on the connected → disconnected
+    edge (#1752). Cancelled by the same handler if the printer reconnects
+    before the window elapses, so a single MQTT blip + recovery doesn't
+    notify. Both the staleness-detector path (`bambu_mqtt.py::check_staleness`)
+    and the smart-plug power-off path (`printer_manager.mark_printer_offline`)
+    route through the same status-change callback, so this covers both.
+    """
+    logger = logging.getLogger(__name__)
+    try:
+        await asyncio.sleep(_PRINTER_OFFLINE_NOTIFY_DEBOUNCE_SECONDS)
+        still_offline = not printer_manager.is_connected(printer_id)
+        logger.info(
+            "[#1752] Printer %s offline debounce elapsed: still_offline=%s",
+            printer_id,
+            still_offline,
+        )
+        if not still_offline:
+            return
+        async with async_session() as db:
+            from backend.app.models.printer import Printer
+
+            result = await db.execute(select(Printer).where(Printer.id == printer_id))
+            printer = result.scalar_one_or_none()
+            if not printer:
+                logger.warning(
+                    "[#1752] Printer %s missing from DB at offline-notify time; skipping",
+                    printer_id,
+                )
+                return
+            logger.info(
+                "[#1752] Dispatching on_printer_offline for printer %s (%s)",
+                printer_id,
+                printer.name,
+            )
+            await notification_service.on_printer_offline(printer_id, printer.name, db)
+    except asyncio.CancelledError:
+        raise
+    except Exception as e:
+        logger.warning("Printer offline notification failed for printer %s: %s", printer_id, e)
+    finally:
+        _printer_offline_notify_tasks.pop(printer_id, None)
+
+
 async def on_printer_status_change(printer_id: int, state: PrinterState):
     """Handle printer status changes - broadcast via WebSocket."""
     # Connected-edge reconciliation (#1542 follow-up). When the printer
@@ -883,6 +946,34 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
         # Re-arm so the next reconnect triggers reconciliation again.
         _printer_reconciled_since_connect[printer_id] = False
 
+    # Offline-notification edge (#1752): schedule `on_printer_offline` on
+    # connected → disconnected. The "back online" channel is already covered
+    # by the print-failure notification (firmware reports gcode_state=FAILED
+    # on reconnect of an interrupted print), so we don't add a symmetric
+    # online event here.
+    prev_connected = _printer_last_connected.get(printer_id)
+    _printer_last_connected[printer_id] = state.connected
+    if prev_connected is True and not state.connected:
+        existing = _printer_offline_notify_tasks.get(printer_id)
+        if existing is None or existing.done():
+            logging.getLogger(__name__).info(
+                "[#1752] Printer %s connected→disconnected edge; scheduling offline notification in %.0fs",
+                printer_id,
+                _PRINTER_OFFLINE_NOTIFY_DEBOUNCE_SECONDS,
+            )
+            _printer_offline_notify_tasks[printer_id] = asyncio.create_task(
+                _maybe_notify_printer_offline(printer_id),
+                name=f"printer-offline-notify-{printer_id}",
+            )
+    elif state.connected:
+        pending = _printer_offline_notify_tasks.pop(printer_id, None)
+        if pending is not None and not pending.done():
+            logging.getLogger(__name__).info(
+                "[#1752] Printer %s reconnected before debounce; cancelling pending offline notification",
+                printer_id,
+            )
+            pending.cancel()
+
     # Only broadcast if something meaningful changed (reduce WebSocket spam)
     # Include rounded temperatures to detect meaningful temp changes (within 1 degree)
     temps = state.temperatures or {}

backend/tests/unit/test_printer_offline_notification.py

@@ -0,0 +1,257 @@
+"""Tests for the connected → disconnected edge that fires the
+`on_printer_offline` notification (#1752).
+
+The provider toggle, schema, and dispatcher already existed; what was missing
+was a caller that fires `notification_service.on_printer_offline` when a
+printer goes offline. These tests pin both layers:
+
+  * `_maybe_notify_printer_offline` — the debounced background task. Must
+    fire when the printer is still offline at the end of the window, and
+    must NOT fire if the printer reconnected during the window.
+
+  * Edge detection inside `on_printer_status_change` — schedules the task
+    only on the True → False transition, cancels any pending task on
+    reconnect, and stays silent on startup (no prior connected state).
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app import main as main_module
+
+
+def _state(connected: bool, state: str = "IDLE") -> SimpleNamespace:
+    """Minimal PrinterState stub. `state="IDLE"` keeps the reconcile-edge
+    branch quiescent (it only fires on `connected=True` with a non-unknown
+    state-string, which we exercise separately) but otherwise lets the
+    handler thread through without doing extra DB / WS work."""
+    return SimpleNamespace(
+        connected=connected,
+        state=state,
+        progress=0,
+        layer_num=0,
+        temperatures={},
+        raw_data={},
+        stg_cur=0,
+        cooling_fan_speed=0,
+        big_fan1_speed=0,
+        big_fan2_speed=0,
+        chamber_light="",
+        active_extruder=0,
+        tray_now=0,
+        door_open=False,
+        subtask_name="",
+    )
+
+
+@pytest.fixture(autouse=True)
+def _reset_edge_state():
+    """Clear the module-level edge dicts between tests so one test's
+    True-edge doesn't leak into the next."""
+    main_module._printer_last_connected.clear()
+    for task in list(main_module._printer_offline_notify_tasks.values()):
+        if not task.done():
+            task.cancel()
+    main_module._printer_offline_notify_tasks.clear()
+    main_module._printer_reconciled_since_connect.clear()
+    main_module._last_status_broadcast.clear()
+    yield
+    main_module._printer_last_connected.clear()
+    for task in list(main_module._printer_offline_notify_tasks.values()):
+        if not task.done():
+            task.cancel()
+    main_module._printer_offline_notify_tasks.clear()
+
+
+class TestMaybeNotifyPrinterOffline:
+    """The debounced background task — fires notification at the end of the
+    window only if the printer is still offline."""
+
+    @pytest.mark.asyncio
+    async def test_fires_notification_when_still_offline_after_debounce(self):
+        printer = SimpleNamespace(id=1, name="Workshop")
+        scalar = MagicMock()
+        scalar.scalar_one_or_none.return_value = printer
+        db = AsyncMock()
+        db.execute = AsyncMock(return_value=scalar)
+        session_cm = MagicMock()
+        session_cm.__aenter__ = AsyncMock(return_value=db)
+        session_cm.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
+            patch("backend.app.main.printer_manager") as mock_pm,
+            patch("backend.app.main.async_session", return_value=session_cm),
+            patch("backend.app.main.notification_service") as mock_notif,
+        ):
+            mock_pm.is_connected.return_value = False
+            mock_notif.on_printer_offline = AsyncMock()
+
+            await main_module._maybe_notify_printer_offline(printer_id=1)
+
+            mock_notif.on_printer_offline.assert_awaited_once_with(1, "Workshop", db)
+
+    @pytest.mark.asyncio
+    async def test_does_not_fire_when_printer_reconnected_during_debounce(self):
+        with (
+            patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
+            patch("backend.app.main.printer_manager") as mock_pm,
+            patch("backend.app.main.notification_service") as mock_notif,
+        ):
+            mock_pm.is_connected.return_value = True
+            mock_notif.on_printer_offline = AsyncMock()
+
+            await main_module._maybe_notify_printer_offline(printer_id=1)
+
+            mock_notif.on_printer_offline.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_does_not_fire_when_printer_missing_from_db(self):
+        scalar = MagicMock()
+        scalar.scalar_one_or_none.return_value = None
+        db = AsyncMock()
+        db.execute = AsyncMock(return_value=scalar)
+        session_cm = MagicMock()
+        session_cm.__aenter__ = AsyncMock(return_value=db)
+        session_cm.__aexit__ = AsyncMock(return_value=False)
+
+        with (
+            patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
+            patch("backend.app.main.printer_manager") as mock_pm,
+            patch("backend.app.main.async_session", return_value=session_cm),
+            patch("backend.app.main.notification_service") as mock_notif,
+        ):
+            mock_pm.is_connected.return_value = False
+            mock_notif.on_printer_offline = AsyncMock()
+
+            await main_module._maybe_notify_printer_offline(printer_id=1)
+
+            mock_notif.on_printer_offline.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_clears_task_entry_after_run(self):
+        with (
+            patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
+            patch("backend.app.main.printer_manager") as mock_pm,
+        ):
+            mock_pm.is_connected.return_value = True  # No notification path
+            main_module._printer_offline_notify_tasks[1] = MagicMock()
+            await main_module._maybe_notify_printer_offline(printer_id=1)
+            assert 1 not in main_module._printer_offline_notify_tasks
+
+
+class TestOfflineEdgeDetection:
+    """Edge detection inside `on_printer_status_change` — only the
+    True → False transition schedules a task. Reconnects cancel pending
+    tasks. Startup-with-disconnected does not fire."""
+
+    @staticmethod
+    def _patch_handler_deps():
+        """Patch out the heavy side-effects of `on_printer_status_change`
+        (MQTT relay, WebSocket broadcast, state serializer) so we can focus
+        on edge state."""
+        ws_mgr = MagicMock()
+        ws_mgr.send_printer_status = AsyncMock()
+        relay = MagicMock()
+        relay.on_printer_status = AsyncMock()
+        pm = MagicMock()
+        pm.get_printer.return_value = None  # Skip the relay payload branch.
+        pm.get_model.return_value = ""
+        return ws_mgr, relay, pm
+
+    @pytest.mark.asyncio
+    async def test_first_call_connected_does_not_schedule(self):
+        ws_mgr, relay, pm = self._patch_handler_deps()
+        with (
+            patch("backend.app.main.ws_manager", ws_mgr),
+            patch("backend.app.main.mqtt_relay", relay),
+            patch("backend.app.main.printer_manager", pm),
+            patch("backend.app.main.spawn_background_task"),
+            patch("backend.app.main.printer_state_to_dict", return_value={}),
+        ):
+            await main_module.on_printer_status_change(1, _state(connected=True))
+        assert 1 not in main_module._printer_offline_notify_tasks
+        assert main_module._printer_last_connected[1] is True
+
+    @pytest.mark.asyncio
+    async def test_first_call_disconnected_does_not_schedule(self):
+        """Startup with an already-offline printer must not fire — there's
+        no prior True observation, so we have no edge to trigger on."""
+        ws_mgr, relay, pm = self._patch_handler_deps()
+        with (
+            patch("backend.app.main.ws_manager", ws_mgr),
+            patch("backend.app.main.mqtt_relay", relay),
+            patch("backend.app.main.printer_manager", pm),
+            patch("backend.app.main.spawn_background_task"),
+            patch("backend.app.main.printer_state_to_dict", return_value={}),
+        ):
+            await main_module.on_printer_status_change(1, _state(connected=False))
+        assert 1 not in main_module._printer_offline_notify_tasks
+        assert main_module._printer_last_connected[1] is False
+
+    @pytest.mark.asyncio
+    async def test_connected_to_disconnected_schedules_task(self):
+        ws_mgr, relay, pm = self._patch_handler_deps()
+        with (
+            patch("backend.app.main.ws_manager", ws_mgr),
+            patch("backend.app.main.mqtt_relay", relay),
+            patch("backend.app.main.printer_manager", pm),
+            patch("backend.app.main.spawn_background_task"),
+            patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()),
+            patch("backend.app.main.printer_state_to_dict", return_value={}),
+        ):
+            await main_module.on_printer_status_change(1, _state(connected=True))
+            await main_module.on_printer_status_change(1, _state(connected=False))
+
+        task = main_module._printer_offline_notify_tasks.get(1)
+        assert task is not None
+        task.cancel()
+
+    @pytest.mark.asyncio
+    async def test_reconnect_cancels_pending_task(self):
+        ws_mgr, relay, pm = self._patch_handler_deps()
+        with (
+            patch("backend.app.main.ws_manager", ws_mgr),
+            patch("backend.app.main.mqtt_relay", relay),
+            patch("backend.app.main.printer_manager", pm),
+            patch("backend.app.main.spawn_background_task"),
+            patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()),
+            patch("backend.app.main.printer_state_to_dict", return_value={}),
+        ):
+            await main_module.on_printer_status_change(1, _state(connected=True))
+            await main_module.on_printer_status_change(1, _state(connected=False))
+            scheduled = main_module._printer_offline_notify_tasks.get(1)
+            assert scheduled is not None
+            await main_module.on_printer_status_change(1, _state(connected=True))
+            # Yield so the cancellation propagates through the event loop.
+            await asyncio.sleep(0)
+
+        assert 1 not in main_module._printer_offline_notify_tasks
+        assert scheduled.cancelled() or scheduled.done()
+
+    @pytest.mark.asyncio
+    async def test_repeated_disconnected_does_not_reschedule(self):
+        """A second False observation while a task is already pending must
+        not replace the in-flight task — otherwise the debounce clock
+        resets on every status callback and the notification never fires."""
+        ws_mgr, relay, pm = self._patch_handler_deps()
+        with (
+            patch("backend.app.main.ws_manager", ws_mgr),
+            patch("backend.app.main.mqtt_relay", relay),
+            patch("backend.app.main.printer_manager", pm),
+            patch("backend.app.main.spawn_background_task"),
+            patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()),
+            patch("backend.app.main.printer_state_to_dict", return_value={}),
+        ):
+            await main_module.on_printer_status_change(1, _state(connected=True))
+            await main_module.on_printer_status_change(1, _state(connected=False))
+            first_task = main_module._printer_offline_notify_tasks.get(1)
+            await main_module.on_printer_status_change(1, _state(connected=False))
+            second_task = main_module._printer_offline_notify_tasks.get(1)
+
+        assert first_task is second_task
+        if first_task is not None:
+            first_task.cancel()