Give the Low Filament notification something that fires it (issue #2913) (#2940)

backend/app/services/notification_service.py

@@ -1814,7 +1814,7 @@ class NotificationService:
         self,
         printer_id: int,
         printer_name: str,
-        slot: int,
+        slot: str,
         remaining_percent: int,
         db: AsyncSession,
         color: str | None = None,
@@ -1826,7 +1826,7 @@ class NotificationService:
 
         variables = {
             "printer": printer_name,
-            "slot": str(slot),
+            "slot": slot,
             "remaining_percent": str(remaining_percent),
             "color": color or "",
         }

backend/app/services/print_scheduler.py

@@ -60,6 +60,10 @@ from backend.app.services.printer_manager import (
     supports_drying_while_printing,
 )
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.spool_assignment_notifications import (
+    _global_tray_from_assignment,
+    _slot_label_from_global_tray,
+)
 from backend.app.utils.ams_drying import is_countdown_parked
 from backend.app.utils.ams_humidity import ams_humidity_percent
 from backend.app.utils.archive_paths import archive_photos_dir
@@ -83,6 +87,47 @@ from backend.app.utils.threemf_tools import (
 
 logger = logging.getLogger(__name__)
 
+
+def _ams_slot_label(ams_id: int, tray_id: int) -> str:
+    """Human slot name for an assignment tuple, in the spelling already in use.
+
+    Composed from the notification side's own pair rather than spelled out again
+    here. The hand-rolled version this replaces got three real slot kinds wrong:
+    ams_id 254 fell into its ``>= 128`` branch and came out as ``HT-`` plus
+    ``chr(191)``; both externals collapsed to one label, because an external
+    assignment stores ams_id 255 with tray_id picking left or right
+    (inventory.py:1771, ``ext_id = data.tray_id + 254``); and an A2L AMS-Lite,
+    normalised to unit 6, read as ``AMS-G`` instead of ``Lite-``.
+
+    Naming the same slot two ways is its own defect once an alert and the
+    Inventory page are meant to be talking about the same tray, so this defers
+    rather than adding a fifth spelling.
+    """
+    return _slot_label_from_global_tray(_global_tray_from_assignment(ams_id, tray_id))
+
+
+# Minimum seconds between low-filament checks. Matches the scheduler's idle
+# interval: the fast path runs every 3 s while an upload is in flight, and a
+# low spool does not need answering at that resolution (#2913).
+_FILAMENT_LOW_MIN_INTERVAL = 30.0
+
+
+def _remaining_percent(label_weight: int | float | None, weight_used: float | None) -> float | None:
+    """Remaining filament as a percentage of the label weight.
+
+    The same arithmetic the Inventory page's Low Stock count uses, and it works
+    unchanged in both inventory modes: internal spools store ``weight_used``
+    directly, and ``_map_spoolman_spool`` derives it from Spoolman's
+    ``remaining_weight`` so the shape matches. Returns None when there is no
+    label weight to be a percentage of -- a spool that cannot say how full it
+    started cannot say how empty it is.
+    """
+    if not label_weight or label_weight <= 0:
+        return None
+    remaining = max(0.0, float(label_weight) - float(weight_used or 0.0))
+    return remaining / float(label_weight) * 100.0
+
+
 # Dispatch-toast progress throttling (#1625 follow-up). Mirrors the legacy
 # background_dispatch.py upload_progress_callback (200 ms time gate + 256 KB
 # byte gate) from before the scheduler unification. Time gate keeps small
@@ -998,6 +1043,23 @@ class PrintScheduler:
         #            streak began
         #   last  -- monotonic stamp of the last pass that observed the streak
         self._auto_dry_above: dict[tuple[int, int], dict[str, float]] = {}
+        # Slots already notified as low on filament:
+        # {(printer_id, ams_id, tray_id, spool_id)}.
+        # Cleared when the slot goes back above its threshold rather than on a
+        # timer, mirroring _notified_hms_errors — a spool hovering at the
+        # boundary must not produce an alert on every pass, and a slot that has
+        # been refilled has to be able to alert again. See #2913.
+        #
+        # The spool id is part of the key because the slot alone cannot clear
+        # itself. Pull a low spool and put a different part-used one in the same
+        # slot: the key survives the pass where the slot resolves to nothing --
+        # deliberately, so a brief Spoolman outage does not re-alert everything
+        # when it returns -- and the replacement never goes above the threshold,
+        # so it can never re-arm. Keyed with the spool, the new spool is simply a
+        # different key and the stale one is inert.
+        self._notified_filament_low: set[tuple[int, int, int, int]] = set()
+        # Earliest monotonic time the next low-filament check may run (#2913).
+        self._filament_low_next_check: float = 0.0
         # Printers with a "running" scheduled drying row (#2638). Rebuilt from the
         # DB on every _check_scheduled_dryings call so route-side cancels show up.
         # Auto-drying's stop-all branches must not stop or untrack these printers;
@@ -1331,6 +1393,7 @@ class PrintScheduler:
                 self._sweep_keep_warm(active_candidates=set(), dispatched=set())
                 inflight_printers = {pid for (_task, pid) in self._inflight.values() if pid is not None}
                 await self._check_auto_drying(db, [], inflight_printers)
+                await self._check_filament_low(db)
                 return bool(self._inflight)
 
             logger.info(
@@ -2109,6 +2172,12 @@ class PrintScheduler:
             # Auto-drying: start drying on idle printers that have no pending queue items
             await self._check_auto_drying(db, items, dispatching_printers)
 
+            # Low filament: alert on assigned spools that have crossed their
+            # low-stock threshold (#2913). Runs on both paths out of this method
+            # for the same reason auto-drying does — an empty queue does not mean
+            # the spools in the printers stopped mattering.
+            await self._check_filament_low(db)
+
             # Keep the loop on the fast interval while any upload is in flight so
             # a slot freed mid-tick refills within seconds rather than after the
             # 30 s idle sleep (#2602). Selecting anything this pass (launched or
@@ -4245,6 +4314,219 @@ class PrintScheduler:
             return None
         return (min_temp, max_hours or 12, filament_type)
 
+    async def _check_filament_low(self, db: AsyncSession) -> None:
+        """Alert on AMS slots whose assigned spool has crossed its low-stock threshold (#2913).
+
+        ``on_filament_low`` has had a column, a schema field, a route, a template
+        and a UI toggle since the notification system was built, and no caller --
+        so the toggle could be switched on and could never fire. This is the
+        producer.
+
+        No new setting. ``low_stock_threshold`` (default 20%) with the per-spool
+        ``low_stock_threshold_pct`` override is already exactly this decision:
+        already configurable, already surfaced, already driving the Inventory
+        page's Low Stock count. The AMS ``remain`` percentage is deliberately not
+        consulted -- it has been measured up to 56 points out against a scale,
+        which is not a number to page someone on.
+
+        Only slots with an assigned spool produce an event. A slot Bambuddy
+        cannot resolve to a spool has no remaining weight it can stand behind,
+        and guessing one is how the remain percentage would have got in.
+        """
+        from backend.app.models.spool import Spool
+
+        # Time-gated rather than run on every pass. run() sleeps
+        # _fast_check_interval -- 3 seconds -- on any productive pass, and the
+        # early-return path counts as productive while an upload is in flight
+        # (#2602), so an unthrottled check would re-read the whole spool
+        # collection every 3 seconds for the length of a batch drain. In
+        # Spoolman mode that is a request storm against a third-party service,
+        # which is the thing this producer's design note argues against; and
+        # when Spoolman is down, _get_with_retry burns ~16 s inside
+        # check_queue holding the scheduler's session, so the outage would
+        # throttle the print queue itself. A low spool is not a 3-second
+        # concern -- the idle interval is the natural resolution.
+        now = time.monotonic()
+        if now < self._filament_low_next_check:
+            return
+        self._filament_low_next_check = now + _FILAMENT_LOW_MIN_INTERVAL
+
+        try:
+            # on_filament_low defaults to off on every provider, so on most
+            # installs nobody wants this alert. Without this the check would
+            # still read every assigned spool each interval -- the whole
+            # collection over HTTP in Spoolman mode -- for an event that is
+            # switched off. One query against the provider table settles it
+            # before any spool work, the same guard the bed-cooled waiter uses.
+            # No printer_id: this asks whether any provider wants the event at
+            # all; per-printer scoping is applied when the event is sent.
+            #
+            # While nobody wants it, no pass sees a slot go back above its
+            # threshold, so nothing can re-arm; a spool refilled under the same
+            # id in that time would stay silenced until a restart. Forgetting
+            # what was sent is right anyway: whoever switches the event back on
+            # is told about the spools that are low now.
+            if not await notification_service._get_providers_for_event(db, "on_filament_low"):
+                self._notified_filament_low.clear()
+                return
+
+            global_threshold = await self._get_low_stock_threshold(db)
+            spoolman_on = await self._get_bool_setting(db, "spoolman_enabled")
+
+            # (printer_id, ams_id, tray_id, spool_id) -> (remaining_pct, threshold, colour name)
+            slots: dict[tuple[int, int, int, int], tuple[float, float, str | None]] = {}
+
+            if spoolman_on:
+                slots = await self._filament_low_slots_spoolman(db, global_threshold)
+            else:
+                rows = (
+                    await db.execute(
+                        select(SpoolAssignment, Spool)
+                        .join(Spool, SpoolAssignment.spool_id == Spool.id)
+                        # An archived spool is not stock. The Inventory page's
+                        # Low Stock count skips them (InventoryPage.tsx:1089)
+                        # and get_all_spools without allow_archived already
+                        # excludes them in Spoolman mode, so without this the
+                        # internal path is the only one that alerts on them.
+                        .where(Spool.archived_at.is_(None))
+                    )
+                ).all()
+                for assignment, spool in rows:
+                    pct = _remaining_percent(spool.label_weight, spool.weight_used)
+                    if pct is None:
+                        continue
+                    threshold = float(spool.low_stock_threshold_pct or global_threshold)
+                    slots[(assignment.printer_id, assignment.ams_id, assignment.tray_id, spool.id)] = (
+                        pct,
+                        threshold,
+                        spool.color_name,
+                    )
+
+            if not slots:
+                # Nothing resolvable this pass. Deliberately not clearing the
+                # notified set: a Spoolman that is briefly unreachable would
+                # otherwise re-alert on every spool as soon as it came back.
+                return
+
+            await self._emit_filament_low(db, slots)
+        except Exception as e:
+            logger.warning("Low-filament check failed: %s", e, exc_info=True)
+
+    async def _get_low_stock_threshold(self, db: AsyncSession) -> float:
+        """The global low-stock percentage, defaulting to the schema's 20.0."""
+        raw = (await db.execute(select(Settings).where(Settings.key == "low_stock_threshold"))).scalar_one_or_none()
+        if raw is None or raw.value is None:
+            return 20.0
+        try:
+            value = float(raw.value)
+        except (TypeError, ValueError):
+            return 20.0
+        return value if 0 < value <= 100 else 20.0
+
+    async def _filament_low_slots_spoolman(
+        self, db: AsyncSession, global_threshold: float
+    ) -> dict[tuple[int, int, int, int], tuple[float, float, str | None]]:
+        """Resolve Spoolman-mode slots to (remaining %, threshold, colour name).
+
+        Spoolman spools carry no per-spool override -- ``low_stock_threshold_pct``
+        is a column on Bambuddy's own spool table and has no Spoolman equivalent,
+        so the global threshold is the only one that applies here. That matches
+        what the Inventory page already does in this mode.
+
+        One ``get_all_spools`` call covers every slot rather than a request per
+        spool. Archived spools do not appear -- ``get_all_spools`` excludes them
+        without ``allow_archived`` -- which is the behaviour the internal path
+        has to ask for explicitly.
+
+        An unreachable Spoolman returns no slots rather than raising. It is an
+        ordinary state for a third-party service, not an error in this pass: the
+        caller's "nothing resolvable" branch already leaves the notified set
+        alone, which is exactly right here, whereas letting it reach the broad
+        ``except`` would write a full traceback every time the check runs.
+        """
+        from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool
+        from backend.app.services.spoolman import SpoolmanUnavailableError, get_spoolman_client
+
+        assignments = (await db.execute(select(SpoolmanSlotAssignment))).scalars().all()
+        if not assignments:
+            return {}
+
+        client = await get_spoolman_client()
+        if client is None:
+            return {}
+
+        try:
+            all_spools = await client.get_all_spools()
+        except SpoolmanUnavailableError as e:
+            logger.debug("Low-filament check skipped, Spoolman unreachable: %s", e)
+            return {}
+
+        by_id: dict[int, dict] = {}
+        for raw in all_spools:
+            raw_id = raw.get("id")
+            if isinstance(raw_id, int):
+                by_id[raw_id] = raw
+
+        slots: dict[tuple[int, int, int, int], tuple[float, float, str | None]] = {}
+        for assignment in assignments:
+            raw = by_id.get(assignment.spoolman_spool_id)
+            if raw is None:
+                continue
+            try:
+                mapped = _map_spoolman_spool(raw)
+            except ValueError:
+                continue
+            pct = _remaining_percent(mapped.get("label_weight"), mapped.get("weight_used"))
+            if pct is None:
+                continue
+            key = (assignment.printer_id, assignment.ams_id, assignment.tray_id, assignment.spoolman_spool_id)
+            slots[key] = (pct, global_threshold, mapped.get("color_name"))
+        return slots
+
+    async def _emit_filament_low(
+        self, db: AsyncSession, slots: dict[tuple[int, int, int, int], tuple[float, float, str | None]]
+    ) -> None:
+        """Send one notification per slot that has newly crossed its threshold."""
+        printer_names: dict[int, str] = {}
+        for key, (pct, threshold, color) in slots.items():
+            printer_id, ams_id, tray_id, _spool_id = key
+            if pct >= threshold:
+                # Back above the line: re-arm rather than expire on a timer, so a
+                # refilled slot can alert again and a spool sitting just under the
+                # threshold stays quiet.
+                self._notified_filament_low.discard(key)
+                continue
+            if key in self._notified_filament_low:
+                continue
+            # Marked before sending, which is a choice rather than the only
+            # option, and it is the opposite failure mode from the one the
+            # debounce argues against: one transient provider failure loses this
+            # alert until the spool goes back above the threshold and crosses it
+            # again. Marking after a successful send would trade that for
+            # re-alerting on every pass while a provider is down -- which is the
+            # repetition the whole debounce exists to prevent, and the louder of
+            # the two failures. A spool that is low stays low, so the next real
+            # signal is not far away; a provider stuck retrying is a signal that
+            # never stops.
+            self._notified_filament_low.add(key)
+
+            if printer_id not in printer_names:
+                printer = (await db.execute(select(Printer).where(Printer.id == printer_id))).scalar_one_or_none()
+                if printer is None:
+                    continue
+                printer_names[printer_id] = printer.name
+            try:
+                await notification_service.on_filament_low(
+                    printer_id,
+                    printer_names[printer_id],
+                    _ams_slot_label(ams_id, tray_id),
+                    int(pct),
+                    db,
+                    color=color,
+                )
+            except Exception as e:
+                logger.warning("Low-filament notification failed for slot %s: %s", key, e)
+
     async def _check_auto_drying(
         self,
         db: AsyncSession,

backend/tests/unit/services/test_filament_low_producer_2913.py

@@ -0,0 +1,627 @@
+"""The "Low Filament" notification has a producer now (issue #2913).
+
+Every layer of this feature existed -- a column on notification_providers, a
+schema field, a route, a message template, thirteen locales and a UI toggle --
+except the thing that fires it. `_get_providers_for_event(db, "on_filament_low")`
+appeared only inside the method that would be called, so the toggle could be
+switched on and could never produce a notification.
+
+The reason it shipped is visible in the old test file: the only occurrence of
+`on_filament_low` in test_notification_service.py is a fixture setting the flag
+to False. Nothing ever asserted the event reaches a provider. The last test here
+does exactly that.
+"""
+
+import logging
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.notification import NotificationProvider
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.models.spool import Spool
+from backend.app.models.spool_assignment import SpoolAssignment
+from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+from backend.app.services.print_scheduler import PrintScheduler, _ams_slot_label, _remaining_percent
+from backend.app.services.spoolman import SpoolmanUnavailableError
+
+
+async def _printer(db, name: str = "X2D") -> Printer:
+    p = Printer(name=name, serial_number=f"S-{name}", ip_address="1.1.1.1", access_code="c", model="X2D")
+    db.add(p)
+    await db.flush()
+    return p
+
+
+async def _assigned_spool(
+    db, printer, *, weight_used: float, threshold_pct: int | None = None, ams=0, tray=0, color_name=None
+):
+    spool = Spool(
+        material="PLA",
+        color_name=color_name,
+        label_weight=1000,
+        core_weight=250,
+        weight_used=weight_used,
+        low_stock_threshold_pct=threshold_pct,
+    )
+    spool.k_profiles = []
+    spool.assignments = []
+    db.add(spool)
+    await db.flush()
+    db.add(SpoolAssignment(printer_id=printer.id, spool_id=spool.id, ams_id=ams, tray_id=tray))
+    await db.flush()
+    return spool
+
+
+# -- the arithmetic ---------------------------------------------------------
+
+
+def test_remaining_percent_matches_the_inventory_page_rule():
+    """`remaining = label - used`, as a percentage of label. The same expression
+    the Low Stock count uses, so the alert and the badge cannot disagree."""
+    assert _remaining_percent(1000, 850.0) == pytest.approx(15.0)
+    assert _remaining_percent(1000, 0.0) == pytest.approx(100.0)
+
+
+def test_remaining_percent_is_none_without_a_label_weight():
+    """A spool that cannot say how full it started cannot say how empty it is."""
+    assert _remaining_percent(0, 100.0) is None
+    assert _remaining_percent(None, 100.0) is None
+
+
+def test_remaining_percent_floors_at_zero_when_overdrawn():
+    assert _remaining_percent(1000, 1200.0) == pytest.approx(0.0)
+
+
+def test_slot_labels_are_the_ones_the_notification_side_already_uses():
+    """Every kind of slot, against the spelling the rest of the app produces.
+
+    The four cases after the first two are the ones the hand-rolled version got
+    wrong: 254 is a real ams_id (InventoryPage.tsx:286, usage_tracker.py:55) and
+    fell into its ``>= 128`` branch as ``HT-`` plus ``chr(191)``; an external
+    assignment stores 255 with tray_id choosing the side, so both externals came
+    out as one label and a two-external printer could not tell them apart; and
+    an A2L AMS-Lite normalises to unit 6, which read as ``AMS-G``.
+    """
+    assert _ams_slot_label(0, 0) == "A1"
+    assert _ams_slot_label(1, 3) == "B4"
+    assert _ams_slot_label(128, 0) == "HT-A"
+    assert _ams_slot_label(254, 0) == "Ext-L"
+    assert _ams_slot_label(255, 0) == "Ext-L"
+    assert _ams_slot_label(255, 1) == "Ext-R"
+    assert _ams_slot_label(6, 0) == "Lite-1"
+
+
+# -- the producer -----------------------------------------------------------
+
+
+@pytest.fixture
+def scheduler():
+    return PrintScheduler()
+
+
+async def _pass(scheduler, db):
+    """Run one low-filament check, stepping over the inter-pass time gate.
+
+    ``_check_filament_low`` refuses to run again within
+    ``_FILAMENT_LOW_MIN_INTERVAL``, so consecutive calls in a test would be
+    swallowed by the gate and every multi-pass assertion below would pass
+    without exercising what it names. The gate has its own test.
+    """
+    scheduler._filament_low_next_check = 0.0
+    await scheduler._check_filament_low(db)
+
+
+@pytest.fixture
+def notify():
+    """The service, replaced, with one provider that wants the event.
+
+    Every test below that is about the producer assumes someone wants the alert;
+    the provider guard in front of the check has its own tests further down,
+    which run the real provider query.
+    """
+    with patch("backend.app.services.print_scheduler.notification_service") as ns:
+        ns.on_filament_low = AsyncMock()
+        ns._get_providers_for_event = AsyncMock(return_value=[MagicMock()])
+        yield ns
+
+
+@pytest.mark.asyncio
+async def test_fires_when_an_assigned_spool_is_below_the_threshold(db_session, scheduler, notify):
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0)  # 15% left, default threshold 20
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_awaited_once()
+    args = notify.on_filament_low.await_args.args
+    assert args[0] == printer.id
+    assert args[1] == "X2D"
+    assert args[2] == "A1"
+    assert args[3] == 15
+
+
+@pytest.mark.asyncio
+async def test_the_spool_colour_is_passed_for_the_template(db_session, scheduler, notify):
+    """on_filament_low has always accepted color= and the filament_low template
+    lists {color} as a variable; with nothing passing it, the variable rendered
+    empty in every custom template."""
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0, color_name="Jade White")
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    assert notify.on_filament_low.await_args.kwargs["color"] == "Jade White"
+
+
+@pytest.mark.asyncio
+async def test_stays_quiet_above_the_threshold(db_session, scheduler, notify):
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=500.0)  # 50% left
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_a_slot_with_no_assigned_spool_produces_nothing(db_session, scheduler, notify):
+    """Guessing a remaining weight for an unassigned slot is how the AMS remain
+    percentage would have got back in."""
+    await _printer(db_session)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_does_not_repeat_on_the_next_pass(db_session, scheduler, notify):
+    """The debounce. A spool sitting under the threshold must not alert every
+    30 seconds until it is changed."""
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    await _pass(scheduler, db_session)
+    await _pass(scheduler, db_session)
+
+    assert notify.on_filament_low.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_re_arms_once_the_slot_goes_back_above_the_threshold(db_session, scheduler, notify):
+    """Cleared by the value going back up, not by a timer -- so a refilled slot
+    can alert again, and a spool hovering at the boundary cannot spam."""
+    printer = await _printer(db_session)
+    spool = await _assigned_spool(db_session, printer, weight_used=850.0)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    assert notify.on_filament_low.await_count == 1
+
+    # Refilled — a fresh spool put on the same slot.
+    spool.weight_used = 0.0
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_filament_low.await_count == 1
+
+    # And down again.
+    spool.weight_used = 900.0
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_filament_low.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_debounce_is_per_slot_not_per_printer(db_session, scheduler, notify):
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0, ams=0, tray=0)
+    await _assigned_spool(db_session, printer, weight_used=900.0, ams=0, tray=1)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    assert notify.on_filament_low.await_count == 2
+    slots = {call.args[2] for call in notify.on_filament_low.await_args_list}
+    assert slots == {"A1", "A2"}
+
+
+@pytest.mark.asyncio
+async def test_per_spool_override_beats_the_global_threshold(db_session, scheduler, notify):
+    """A spool marked as needing 60% left alerts at 50%, where the global 20%
+    would have stayed quiet. Same precedence the Inventory page uses."""
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=500.0, threshold_pct=60)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_awaited_once()
+    assert notify.on_filament_low.await_args.args[3] == 50
+
+
+@pytest.mark.asyncio
+async def test_global_setting_is_honoured(db_session, scheduler, notify):
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=600.0)  # 40% left
+    db_session.add(Settings(key="low_stock_threshold", value="50"))
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_a_failing_provider_does_not_break_the_pass(db_session, scheduler, notify):
+    """A notification provider being down must not take the scheduler with it."""
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0)
+    await db_session.commit()
+    notify.on_filament_low.side_effect = RuntimeError("provider down")
+
+    await _pass(scheduler, db_session)  # must not raise
+
+    notify.on_filament_low.assert_awaited_once()
+
+
+# -- the pass is time-gated -------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_second_pass_inside_the_interval_does_no_work(db_session, scheduler, notify):
+    """The gate, which every multi-pass test above steps over deliberately.
+
+    run() sleeps _fast_check_interval -- 3 seconds -- on any productive pass,
+    and the early-return path counts as productive while an upload is in flight
+    (#2602). Ungated, this check would re-read every assigned spool on that
+    cadence for the length of a batch drain, which in Spoolman mode means the
+    whole collection over HTTP against a third-party service.
+    """
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0)
+    await db_session.commit()
+
+    with patch.object(scheduler, "_get_low_stock_threshold", new_callable=AsyncMock) as work:
+        work.return_value = 20.0
+        await scheduler._check_filament_low(db_session)
+        await scheduler._check_filament_low(db_session)
+        await scheduler._check_filament_low(db_session)
+
+    # Not "no notification" -- the debounce would give that too. No work at all.
+    assert work.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_an_archived_spool_is_not_stock(db_session, scheduler, notify):
+    """Archiving a spool takes it out of the Low Stock count (InventoryPage.tsx:1089).
+
+    Spoolman mode gets this for free -- get_all_spools without allow_archived
+    does not return them -- so without the explicit filter the internal path
+    would be the only mode that alerts on a spool the user has retired.
+    """
+    printer = await _printer(db_session)
+    spool = await _assigned_spool(db_session, printer, weight_used=850.0)
+    spool.archived_at = datetime(2026, 8, 1, 12, 0, 0)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_not_awaited()
+
+
+# -- Spoolman mode ----------------------------------------------------------
+
+
+def _spoolman_spool(spool_id: int, remaining_weight: float, label_weight: int = 1000, color_name=None) -> dict:
+    """A raw Spoolman spool, in the shape _map_spoolman_spool reads."""
+    filament = {"id": 1, "name": "PLA Basic", "material": "PLA", "weight": label_weight, "vendor": {"name": "X"}}
+    if color_name is not None:
+        filament["color_name"] = color_name
+    return {
+        "id": spool_id,
+        "remaining_weight": remaining_weight,
+        "used_weight": label_weight - remaining_weight,
+        "filament": filament,
+    }
+
+
+async def _spoolman_slot(db, printer, spool_id: int, *, ams=0, tray=0):
+    db.add(SpoolmanSlotAssignment(printer_id=printer.id, ams_id=ams, tray_id=tray, spoolman_spool_id=spool_id))
+    db.add(Settings(key="spoolman_enabled", value="true"))
+    await db.flush()
+
+
+def _spoolman_client(spools, *, unavailable: bool = False):
+    client = MagicMock()
+    if unavailable:
+        client.get_all_spools = AsyncMock(side_effect=SpoolmanUnavailableError("Cannot reach Spoolman"))
+    else:
+        client.get_all_spools = AsyncMock(return_value=spools)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_spoolman_mode_alerts_from_the_collection_read(db_session, scheduler, notify):
+    """The half with the client in it, which had no coverage at all.
+
+    The threshold is meant to mean the same thing in both inventory modes, so
+    the mode that reaches a third-party service for its numbers needs its own
+    tests rather than inheriting the internal path's.
+    """
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7, ams=0, tray=1)
+    await db_session.commit()
+    client = _spoolman_client([_spoolman_spool(7, remaining_weight=150.0, color_name="Jade White")])  # 15% left
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
+        await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_awaited_once()
+    args = notify.on_filament_low.await_args.args
+    assert args[2] == "A2"
+    assert args[3] == 15
+    assert notify.on_filament_low.await_args.kwargs["color"] == "Jade White"
+    # One request for the whole collection, not one per assigned slot.
+    client.get_all_spools.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_a_spoolman_slot_whose_spool_is_gone_is_skipped(db_session, scheduler, notify):
+    """A spool deleted in Spoolman leaves the assignment row behind. It has no
+    remaining weight, so it produces nothing rather than a guess."""
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7)
+    await db_session.commit()
+    client = _spoolman_client([_spoolman_spool(99, remaining_weight=150.0)])
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
+        await _pass(scheduler, db_session)
+
+    notify.on_filament_low.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_an_unreachable_spoolman_resolves_nothing_and_stays_quiet(db_session, scheduler, notify, caplog):
+    """An outage is an ordinary state for a third-party service, not an error here.
+
+    The assertion is the log, not the silence. Letting SpoolmanUnavailableError
+    reach _check_filament_low's broad except produces the same absence of
+    notifications -- so "nothing was sent" cannot tell the two apart -- while
+    writing a full traceback on every pass for the length of the outage. What
+    the catch buys is that it is handled where it is expected.
+    """
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7)
+    await db_session.commit()
+    client = _spoolman_client(None, unavailable=True)
+
+    with (
+        caplog.at_level(logging.WARNING, logger="backend.app.services.print_scheduler"),
+        patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)),
+    ):
+        await _pass(scheduler, db_session)  # must not raise
+
+    notify.on_filament_low.assert_not_awaited()
+    assert "Low-filament check failed" not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_an_outage_does_not_re_alert_everything_when_it_clears(db_session, scheduler, notify):
+    """The reason a pass that resolves nothing leaves the notified set alone.
+
+    Clearing the set on an empty pass would make a brief Spoolman outage
+    indistinguishable from every spool being refilled, and the alerts would all
+    fire again the moment it came back.
+    """
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7)
+    await db_session.commit()
+    low = [_spoolman_spool(7, remaining_weight=150.0)]
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client(low))):
+        await _pass(scheduler, db_session)
+    assert notify.on_filament_low.await_count == 1
+
+    outage = _spoolman_client(None, unavailable=True)
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=outage)):
+        await _pass(scheduler, db_session)
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client(low))):
+        await _pass(scheduler, db_session)
+    assert notify.on_filament_low.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_a_replacement_spool_in_the_same_slot_can_alert(db_session, scheduler, notify):
+    """Why the notified key carries the spool id.
+
+    Pull a low spool and put a different part-used one in the same slot. The
+    key survives the pass where the slot resolves to nothing -- deliberately --
+    and the replacement never goes above the threshold, so keyed on the slot
+    alone it could never re-arm and the second spool would run out silently.
+    """
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7)
+    await db_session.commit()
+
+    first = _spoolman_client([_spoolman_spool(7, remaining_weight=150.0)])
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=first)):
+        await _pass(scheduler, db_session)
+    assert notify.on_filament_low.await_count == 1
+
+    # The slot resolves to nothing for a pass while the roll is swapped.
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client([]))):
+        await _pass(scheduler, db_session)
+
+    assignment = (await db_session.execute(select(SpoolmanSlotAssignment))).scalars().one()
+    assignment.spoolman_spool_id = 8
+    await db_session.commit()
+
+    second = _spoolman_client([_spoolman_spool(8, remaining_weight=100.0)])  # 10% left
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=second)):
+        await _pass(scheduler, db_session)
+
+    assert notify.on_filament_low.await_count == 2
+    assert notify.on_filament_low.await_args.args[3] == 10
+
+
+# -- nobody wants the alert ------------------------------------------------
+
+
+async def _provider(db, *, enabled: bool, on_filament_low: bool) -> None:
+    db.add(
+        NotificationProvider(
+            name=f"p-{enabled}-{on_filament_low}",
+            provider_type="ntfy",
+            config="{}",
+            enabled=enabled,
+            on_filament_low=on_filament_low,
+        )
+    )
+    await db.flush()
+
+
+@pytest.fixture
+def real_providers():
+    """The real service, with only the send replaced.
+
+    The guard is one query against notification_providers, so these tests run
+    it for real rather than mocking the answer -- otherwise a guard asking about
+    the wrong event, or ignoring ``enabled``, would pass.
+    """
+    from backend.app.services.print_scheduler import notification_service
+
+    with patch.object(notification_service, "on_filament_low", new_callable=AsyncMock) as send:
+        yield send
+
+
+@pytest.mark.asyncio
+async def test_no_spoolman_call_when_no_provider_wants_the_event(db_session, scheduler, real_providers):
+    """on_filament_low defaults to off, so on most installs nobody wants this.
+
+    Unguarded, every Spoolman install with slot assignments read its whole spool
+    collection every interval, forever, for an alert that was switched off. The
+    two providers here are the two ways of not wanting it: enabled with the
+    event off, and the event on but the provider disabled.
+    """
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7)
+    await _provider(db_session, enabled=True, on_filament_low=False)
+    await _provider(db_session, enabled=False, on_filament_low=True)
+    await db_session.commit()
+    get_client = AsyncMock(return_value=_spoolman_client([_spoolman_spool(7, remaining_weight=150.0)]))
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", get_client):
+        await _pass(scheduler, db_session)
+
+    get_client.assert_not_awaited()
+    real_providers.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_internal_mode_does_no_spool_work_when_no_provider_wants_the_event(db_session, scheduler, real_providers):
+    """Same guard, internal mode: no spool query at all, not just no send."""
+    printer = await _printer(db_session)
+    await _assigned_spool(db_session, printer, weight_used=850.0)
+    await db_session.commit()
+
+    with patch.object(scheduler, "_get_low_stock_threshold", new_callable=AsyncMock) as work:
+        work.return_value = 20.0
+        await _pass(scheduler, db_session)
+
+    work.assert_not_awaited()
+    real_providers.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_one_provider_wanting_the_event_is_enough(db_session, scheduler, real_providers):
+    """The other side of the guard, through the same real query: one enabled
+    provider with the event on and the check runs. Without this, a guard that
+    always returned would pass the two tests above."""
+    printer = await _printer(db_session)
+    await _spoolman_slot(db_session, printer, 7)
+    await _provider(db_session, enabled=True, on_filament_low=False)
+    await _provider(db_session, enabled=True, on_filament_low=True)
+    await db_session.commit()
+    client = _spoolman_client([_spoolman_spool(7, remaining_weight=150.0)])
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
+        await _pass(scheduler, db_session)
+
+    client.get_all_spools.assert_awaited_once()
+    real_providers.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_switching_the_event_off_and_on_again_re_arms(db_session, scheduler, real_providers):
+    """While the event is off no pass runs, so none can see a slot go back above
+    its threshold and re-arm it. A spool refilled under the same id meanwhile
+    would stay silenced once the event is back on. Switching it off forgets what
+    was sent; switching it on reports the spools that are low now."""
+    printer = await _printer(db_session)
+    spool = await _assigned_spool(db_session, printer, weight_used=850.0)
+    provider = NotificationProvider(name="p", provider_type="ntfy", config="{}", enabled=True, on_filament_low=True)
+    db_session.add(provider)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    assert real_providers.await_count == 1
+
+    # Off; refilled and run low again while no pass was looking; on again.
+    provider.on_filament_low = False
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    spool.weight_used = 900.0
+    provider.on_filament_low = True
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    assert real_providers.await_count == 2
+
+
+# -- the event actually reaches a provider ----------------------------------
+
+
+@pytest.mark.asyncio
+async def test_the_event_reaches_a_provider_with_the_toggle_on():
+    """The assertion that was missing, and the reason this shipped unfired.
+
+    Everything above proves the producer calls the service. This proves the
+    service does not drop it: a provider with on_filament_low enabled is handed
+    the rendered message.
+    """
+    from backend.app.services.notification_service import NotificationService
+
+    service = NotificationService()
+    provider = MagicMock()
+    provider.id = 1
+    provider.on_filament_low = True
+
+    with (
+        patch.object(service, "_get_providers_for_event", new_callable=AsyncMock) as mock_get,
+        patch.object(service, "_send_to_providers", new_callable=AsyncMock) as mock_send,
+        patch.object(service, "_build_message_from_template", new_callable=AsyncMock) as mock_build,
+    ):
+        mock_get.return_value = [provider]
+        mock_build.return_value = ("Filament Low", "X2D A1 is at 15%")
+
+        await service.on_filament_low(1, "X2D", "A1", 15, AsyncMock(), color="Jade White")
+
+    mock_get.assert_awaited_once()
+    assert mock_get.await_args.args[1] == "on_filament_low"
+    mock_send.assert_awaited_once()
+    variables = mock_send.await_args.kwargs["variables"]
+    assert variables["printer"] == "X2D"
+    assert variables["slot"] == "A1"
+    assert variables["remaining_percent"] == "15"
+    assert variables["color"] == "Jade White"