fix(#1128): broadcast printer_status when awaiting_plate_clear flips

  awaiting_plate_clear is a Bambuddy-side flag, not a printer-side one,
  so toggling it does not produce an MQTT push from the printer. Commit
  4e86e8c added the flag to the printer_status payload so MQTT-driven
  broadcasts (e.g. when a print finishes and on_print_complete sets the
  flag to True alongside a state transition to FINISH) carry it. The
  reverse transition didn't: POST /printers/{id}/clear-plate mutated
  PrinterManager._awaiting_plate_clear and persisted to the DB, but
  emitted no printer_status WebSocket update — and the in-main.py
  status-change broadcaster's status_key dedup intentionally excludes
  Bambuddy-side flags, so even a coincidentally-arriving MQTT push
  wouldn't reflect the change.

  The "Mark plate as cleared" button on the printer card disappeared
  "immediately" after a click only because the React Query cache was
  being optimistically updated client-side; clearing the flag through
  any other route (an admin script, a second tab, an automation hitting
  the endpoint directly, the scheduler at print_scheduler.py:1844 when
  dispatching the next queued print) silently left every UI subscriber
  but the originating tab stale until a coincidental status refresh.

  Centralised the broadcast in PrinterManager.set_awaiting_plate_clear
  itself rather than at each call site, so every current AND future
  caller is covered without remembering to wire it up: a new
  _broadcast_status_change(printer_id) private coroutine is scheduled
  alongside the existing _persist_awaiting_plate_clear whenever the flag
  flips under a running event loop. Lazy-imports ws_manager to keep
  printer_manager.py clean of application-layer infra at module-import
  time, short-circuits when get_status returns None (printer
  disconnected — the next reconnect produces a fresh push anyway), and
  swallows ws_manager.send_printer_status failures so the persistence
  path can complete even if the WS layer is temporarily unavailable.

  The same hook is now in place for any other Bambuddy-side flag that
  gets added to printer_state_to_dict later — they'll all need to
  broadcast their own changes for the same reason.

  8 new regression tests in test_printer_manager_status_broadcast.py:
  schedules-on-True/False/loop-running/no-loop/loop-stopped contracts,
  _broadcast_status_change happy path with payload assertion,
  skip-when-no-state, swallow-WS-errors, and an end-to-end live-loop
  test that fires set_awaiting_plate_clear(False) and asserts a
  broadcast lands with awaiting_plate_clear: false in the payload.
  Existing 24 tests in test_scheduler_clear_plate.py continue to pass
  unchanged because they instantiate PrinterManager() without
  attaching a loop (sync unit-test path) — the new _schedule_async
  call short-circuits on the same loop check the existing persistence
  call already used.

backend/app/services/printer_manager.py

@@ -191,6 +191,18 @@ class PrinterManager:
         Persisted so the gate survives Bambuddy/printer restarts (#961): after Auto Off
         cycles the printer, the printer boots into IDLE with no memory of the previous
         finish, and without persistence the queue would bypass the confirmation prompt.
+
+        Also broadcasts an updated ``printer_status`` over the WebSocket (#1128).
+        ``awaiting_plate_clear`` is a Bambuddy-side flag — toggling it does not
+        produce an MQTT push from the printer, so without an explicit broadcast
+        any UI subscriber that's NOT the originating tab would stay stale until
+        the next coincidental status refresh. The plate-clear button on the
+        printer card disappeared "immediately" only because of an optimistic
+        React Query cache update on the click path; clearing the flag through
+        any other route (an admin script, a second tab, an automation that
+        hits ``POST /printers/{id}/clear-plate`` directly) silently broke the
+        UI without it. Centralised here so every current AND future caller is
+        covered without each one having to remember to broadcast.
         """
         if awaiting:
             self._awaiting_plate_clear.add(printer_id)
@@ -200,6 +212,43 @@ class PrinterManager:
         # emits "coroutine was never awaited" warnings (e.g. in sync unit tests).
         if self._loop and self._loop.is_running():
             self._schedule_async(self._persist_awaiting_plate_clear(printer_id, awaiting))
+            self._schedule_async(self._broadcast_status_change(printer_id))
+
+    async def _broadcast_status_change(self, printer_id: int) -> None:
+        """Emit a ``printer_status`` WebSocket update for this printer (#1128).
+
+        Used for state changes that don't come from MQTT — currently just the
+        ``awaiting_plate_clear`` flag, but any future Bambuddy-side flag added
+        to ``printer_state_to_dict`` should plumb through here too. The
+        existing MQTT-driven broadcast in ``main.on_printer_status_change``
+        deduplicates on a status_key that intentionally excludes Bambuddy
+        flags (so e.g. queue-state changes don't get echoed as printer
+        events), which is precisely why those flags need their own emit.
+
+        Lazy-imports ``ws_manager`` to keep ``printer_manager`` clean of
+        application-layer infra at module-import time — the broadcast is the
+        only thing here that needs it.
+        """
+        state = self.get_status(printer_id)
+        if not state:
+            # Printer disconnected or unknown — nothing to broadcast. The
+            # next reconnect will produce a fresh status push anyway, so the
+            # UI eventually catches up without us forcing a stale snapshot
+            # on subscribers now.
+            return
+        try:
+            from backend.app.core.websocket import ws_manager
+
+            await ws_manager.send_printer_status(
+                printer_id,
+                printer_state_to_dict(state, printer_id, self.get_model(printer_id)),
+            )
+        except Exception as e:
+            logger.warning(
+                "Failed to broadcast printer_status after Bambuddy-side state change for printer %d: %s",
+                printer_id,
+                e,
+            )
 
     async def _persist_awaiting_plate_clear(self, printer_id: int, awaiting: bool):
         from backend.app.core.database import async_session

backend/tests/unit/test_printer_manager_status_broadcast.py

@@ -0,0 +1,223 @@
+"""Regression tests for ``PrinterManager._broadcast_status_change`` and
+its wiring from ``set_awaiting_plate_clear`` (#1128).
+
+The bug: ``awaiting_plate_clear`` is a Bambuddy-side flag, so toggling it
+doesn't produce an MQTT push from the printer. Before the fix,
+``set_awaiting_plate_clear()`` mutated state and persisted to DB but never
+notified WebSocket subscribers. The plate-clear button on the printer card
+disappeared "immediately" only because of an optimistic React Query cache
+update on the click path; any other caller (admin script, second tab, an
+automation that hits ``POST /printers/{id}/clear-plate``) silently left
+the UI stale until the next coincidental status refresh.
+
+These tests pin the contract: every flip of the flag must schedule a
+``printer_status`` broadcast, and the broadcast must carry the new flag
+value so subscribers see the right state without polling.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.printer_manager import PrinterManager
+
+
+@pytest.fixture
+def manager():
+    """Fresh manager per test; the awaiting-plate-clear set is per-instance."""
+    return PrinterManager()
+
+
+def _fake_state(**overrides):
+    """Minimal stand-in for a ``PrinterState`` — only the attributes
+    ``printer_state_to_dict`` reads. We use a SimpleNamespace rather than
+    constructing a real PrinterState so this test stays fast and doesn't
+    couple to the (large, evolving) PrinterState dataclass shape."""
+    base = {
+        "connected": True,
+        "state": "FINISH",
+        "raw_data": {},
+        "progress": 100.0,
+    }
+    base.update(overrides)
+    return SimpleNamespace(**base)
+
+
+class TestSchedulingFromSetAwaitingPlateClear:
+    """The hook from the public flag-mutation method into the broadcast."""
+
+    def test_schedules_broadcast_when_loop_running(self, manager):
+        """When a real event loop is attached, every call to
+        ``set_awaiting_plate_clear`` must enqueue both the persistence
+        coroutine and the broadcast coroutine. Both are needed: persist
+        survives restarts, broadcast notifies live subscribers."""
+        manager._loop = MagicMock()
+        manager._loop.is_running.return_value = True
+
+        with patch.object(manager, "_schedule_async") as scheduled:
+            manager.set_awaiting_plate_clear(7, True)
+
+        # Two coroutines: persist + broadcast. Order doesn't matter.
+        assert scheduled.call_count == 2
+
+    def test_does_not_schedule_when_no_loop_attached(self, manager):
+        """Sync unit-test path (no loop attached): nothing must be
+        scheduled, otherwise Python emits 'coroutine was never awaited'
+        runtime warnings and the test suite goes red on harmless flag
+        twiddling."""
+        manager._loop = None
+
+        with patch.object(manager, "_schedule_async") as scheduled:
+            manager.set_awaiting_plate_clear(7, True)
+
+        scheduled.assert_not_called()
+
+    def test_does_not_schedule_when_loop_not_running(self, manager):
+        """A loop attached-but-stopped is the same situation as no loop —
+        scheduling onto a dead loop would never fire."""
+        manager._loop = MagicMock()
+        manager._loop.is_running.return_value = False
+
+        with patch.object(manager, "_schedule_async") as scheduled:
+            manager.set_awaiting_plate_clear(7, True)
+
+        scheduled.assert_not_called()
+
+    def test_both_true_and_false_flips_schedule_broadcast(self, manager):
+        """The bug only became visible on ``False`` flips (clear), but a
+        regression that broadcasts only on ``True`` would re-introduce
+        the original symptom for any future flag mutation that goes
+        ``False → True`` outside the printer-card optimistic-update
+        path. Make both directions a contract."""
+        manager._loop = MagicMock()
+        manager._loop.is_running.return_value = True
+
+        with patch.object(manager, "_schedule_async") as scheduled:
+            manager.set_awaiting_plate_clear(7, True)
+            scheduled.reset_mock()
+            manager.set_awaiting_plate_clear(7, False)
+
+        # Each flip = persist + broadcast = 2 calls.
+        assert scheduled.call_count == 2
+
+
+class TestBroadcastStatusChange:
+    """The broadcast coroutine itself."""
+
+    @pytest.mark.asyncio
+    async def test_emits_ws_update_when_state_present(self, manager):
+        """Happy path: printer has a known status, broadcast goes out
+        with the dict produced by ``printer_state_to_dict``."""
+        state = _fake_state()
+        with (
+            patch.object(manager, "get_status", return_value=state),
+            patch.object(manager, "get_model", return_value="P1S"),
+            patch(
+                "backend.app.core.websocket.ws_manager.send_printer_status",
+                new_callable=AsyncMock,
+            ) as send_status,
+            patch(
+                "backend.app.services.printer_manager.printer_state_to_dict",
+                return_value={"id": 7, "awaiting_plate_clear": False},
+            ) as to_dict,
+        ):
+            await manager._broadcast_status_change(7)
+
+        send_status.assert_awaited_once()
+        # First positional arg is the printer ID, second is the status dict.
+        printer_id_arg, payload_arg = send_status.await_args.args
+        assert printer_id_arg == 7
+        assert payload_arg == {"id": 7, "awaiting_plate_clear": False}
+        # Verify the dict was built from the right inputs (state + id + model).
+        to_dict.assert_called_once_with(state, 7, "P1S")
+
+    @pytest.mark.asyncio
+    async def test_skips_when_status_unknown(self, manager):
+        """Printer not connected / unknown ID → no point broadcasting a
+        snapshot we don't have. A future reconnect will produce a fresh
+        status push anyway, so we'd only be forcing a stale or bogus
+        payload onto subscribers right now."""
+        with (
+            patch.object(manager, "get_status", return_value=None),
+            patch(
+                "backend.app.core.websocket.ws_manager.send_printer_status",
+                new_callable=AsyncMock,
+            ) as send_status,
+        ):
+            await manager._broadcast_status_change(999)
+
+        send_status.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_swallows_websocket_errors(self, manager):
+        """The broadcast is a courtesy, not a correctness path — if the
+        WS layer is down, the flag is already mutated in-memory and
+        persisted. Letting an exception bubble out of
+        ``_broadcast_status_change`` would surface as an
+        ``Exception in scheduled callback`` traceback in the log AND
+        prevent the persistence coroutine from completing if both were
+        gathered together. Swallow + warn instead."""
+        with (
+            patch.object(manager, "get_status", return_value=_fake_state()),
+            patch.object(manager, "get_model", return_value="P1S"),
+            patch(
+                "backend.app.services.printer_manager.printer_state_to_dict",
+                return_value={"id": 7},
+            ),
+            patch(
+                "backend.app.core.websocket.ws_manager.send_printer_status",
+                new_callable=AsyncMock,
+                side_effect=RuntimeError("websocket layer unavailable"),
+            ),
+        ):
+            # Must not raise.
+            await manager._broadcast_status_change(7)
+
+
+class TestEndToEndUnderRunningLoop:
+    """Verify the full flow under a real running event loop — schedule
+    → broadcast → ws_manager.send_printer_status — without mocking
+    ``_schedule_async``. Catches regressions where individual pieces
+    pass but the wiring breaks (e.g. ``_schedule_async`` swallowing the
+    broadcast coroutine)."""
+
+    @pytest.mark.asyncio
+    async def test_set_false_eventually_emits_broadcast(self, manager):
+        """Reproduces the #1128 fix path end-to-end: set the flag to
+        False under a live loop, give the scheduler a tick, the
+        ws broadcast must have fired with the new payload."""
+        loop = asyncio.get_running_loop()
+        manager._loop = loop
+        # Pretend the printer has been seen — without a state present
+        # the broadcast short-circuits before reaching ws_manager.
+        manager._awaiting_plate_clear.add(7)
+
+        with (
+            patch.object(manager, "get_status", return_value=_fake_state()),
+            patch.object(manager, "get_model", return_value="P1S"),
+            patch(
+                "backend.app.services.printer_manager.printer_state_to_dict",
+                return_value={"id": 7, "awaiting_plate_clear": False},
+            ),
+            patch(
+                "backend.app.core.websocket.ws_manager.send_printer_status",
+                new_callable=AsyncMock,
+            ) as send_status,
+            # Persistence path opens a DB session; stub it out so this
+            # stays a pure unit test.
+            patch.object(manager, "_persist_awaiting_plate_clear", new_callable=AsyncMock),
+        ):
+            manager.set_awaiting_plate_clear(7, False)
+            # Yield repeatedly so run_coroutine_threadsafe has a chance
+            # to land its scheduled coroutine on this loop.
+            for _ in range(10):
+                await asyncio.sleep(0)
+
+        send_status.assert_awaited()
+        printer_id_arg, payload_arg = send_status.await_args.args
+        assert printer_id_arg == 7
+        assert payload_arg["awaiting_plate_clear"] is False