Give the two stock alerts something that fires them (issue #2955) (#3196)

backend/app/core/database.py

@@ -5128,6 +5128,12 @@ async def run_migrations(conn):
     # on fresh installs only — this covers databases whose table predates it.
     await _migrate_location_ha_sensor_unique_binding(conn)
 
+    # Migration: the stock alert templates name the colour and subtype (#2955).
+    # The forecast groups by colour, so two colours of one product would
+    # otherwise send the same message. A plain UPDATE guarded on the old text, so
+    # a template an admin has edited is left alone.
+    await _migrate_stock_alert_template_sku_variables(conn)
+
     # Migration: supplier master list + spool assignments (#2988).
     # create_all() covers fresh installs; this covers upgrades.
     await _migrate_create_supplier_tables(conn)
@@ -5422,6 +5428,35 @@ async def _migrate_rename_ha_sensor_alert_template(conn) -> None:
     )
 
 
+_STOCK_ALERT_TEMPLATE_BODIES = {
+    "stock_reorder_alert": (
+        "{material} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
+        "{material} {subtype} {color} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
+    ),
+    "stock_break_alert": (
+        "{material} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
+        "{material} {subtype} {color} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
+    ),
+}
+
+
+async def _migrate_stock_alert_template_sku_variables(conn) -> None:
+    """Give the stock alert templates {subtype} and {color} (#2955).
+
+    Updates a body only while it is still the shipped default, so an admin's own
+    wording is kept.
+    """
+    from sqlalchemy import text
+
+    for event_type, (old, new) in _STOCK_ALERT_TEMPLATE_BODIES.items():
+        await conn.execute(
+            text(
+                "UPDATE notification_templates SET body_template = :new WHERE event_type = :et AND body_template = :old"
+            ),
+            {"new": new, "et": event_type, "old": old},
+        )
+
+
 async def _migrate_location_ha_sensor_unique_binding(conn) -> None:
     """Unique index on location_ha_sensors (location_id, entity_id) (#2824).
 

backend/app/models/notification_template.py

@@ -235,13 +235,13 @@ DEFAULT_TEMPLATES = [
         "event_type": "stock_reorder_alert",
         "name": "Stock Reorder Alert",
         "title_template": "Reorder Alert: {material}",
-        "body_template": "{material} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
+        "body_template": "{material} {subtype} {color} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
     },
     {
         "event_type": "stock_break_alert",
         "name": "Stock Break Alert",
         "title_template": "Stock Break Risk: {material}",
-        "body_template": "{material} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
+        "body_template": "{material} {subtype} {color} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
     },
     # User email notification templates (sent to the print job owner).
     # Names include " Email" so they aren't confused with the provider-level

backend/app/schemas/notification_template.py

@@ -101,6 +101,29 @@ EVENT_VARIABLES: dict[str, list[str]] = {
     "queue_job_skipped": ["printer", "job_name", "reason", "timestamp", "app_name"],
     "queue_job_failed": ["printer", "job_name", "reason", "timestamp", "app_name"],
     "queue_completed": ["completed_count", "timestamp", "app_name"],
+    "stock_reorder_alert": [
+        "material",
+        "subtype",
+        "brand",
+        "color",
+        "stock_g",
+        "rate_g_day",
+        "days_left",
+        "timestamp",
+        "app_name",
+    ],
+    "stock_break_alert": [
+        "material",
+        "subtype",
+        "brand",
+        "color",
+        "stock_g",
+        "rate_g_day",
+        "days_left",
+        "lead_time_days",
+        "timestamp",
+        "app_name",
+    ],
     # User management notifications
     "user_created": ["username", "password", "login_url", "app_name", "timestamp"],
     "password_reset": ["username", "password", "login_url", "app_name", "timestamp"],
@@ -303,6 +326,29 @@ SAMPLE_DATA: dict[str, dict[str, str]] = {
         "timestamp": "2024-01-15 18:30",
         "app_name": "Bambuddy",
     },
+    "stock_reorder_alert": {
+        "material": "PLA",
+        "subtype": "Matte",
+        "brand": "Bambu Lab",
+        "color": "Charcoal",
+        "stock_g": "420",
+        "rate_g_day": "18.5",
+        "days_left": "22",
+        "timestamp": "2024-01-15 14:30",
+        "app_name": "Bambuddy",
+    },
+    "stock_break_alert": {
+        "material": "PLA",
+        "subtype": "Matte",
+        "brand": "Bambu Lab",
+        "color": "Charcoal",
+        "stock_g": "160",
+        "rate_g_day": "18.5",
+        "days_left": "8",
+        "lead_time_days": "14",
+        "timestamp": "2024-01-15 14:30",
+        "app_name": "Bambuddy",
+    },
     # User management notifications
     "user_created": {
         "username": "john_doe",

backend/app/services/notification_service.py

@@ -2589,15 +2589,31 @@ class NotificationService:
         rate_g_day: float,
         days_left: int,
         db: AsyncSession,
+        *,
+        subtype: str | None = None,
+        color: str | None = None,
+        skip_break_subscribers: bool = False,
     ):
-        """Fire when an inventory SKU reaches its reorder point."""
+        """Fire when an inventory SKU reaches its reorder point.
+
+        ``subtype`` and ``color`` are what tell apart the messages for two colours of one
+        product, which the forecast (grouped by colour as well) reports separately.
+
+        A SKU in stock break has also reached its reorder point. ``skip_break_subscribers``
+        leaves out the providers that have the break alert on, so the producer can send
+        the reorder event for a break SKU without telling those providers twice.
+        """
         providers = await self._get_providers_for_event(db, "on_stock_reorder_alert", None)
+        if skip_break_subscribers:
+            providers = [p for p in providers if not p.on_stock_break_alert]
         if not providers:
             return
 
         variables = {
             "material": material,
+            "subtype": subtype or "",
             "brand": brand or "",
+            "color": color or "",
             "stock_g": f"{stock_g:.0f}",
             "rate_g_day": f"{rate_g_day:.1f}",
             "days_left": str(days_left),
@@ -2615,15 +2631,23 @@ class NotificationService:
         days_left: int,
         lead_time_days: int,
         db: AsyncSession,
+        *,
+        subtype: str | None = None,
+        color: str | None = None,
     ):
-        """Fire when a stock break is detected (stock runs out before lead time)."""
+        """Fire when a stock break is detected (stock runs out before lead time).
+
+        ``subtype`` and ``color`` as for :meth:`on_stock_reorder_alert`.
+        """
         providers = await self._get_providers_for_event(db, "on_stock_break_alert", None)
         if not providers:
             return
 
         variables = {
             "material": material,
+            "subtype": subtype or "",
             "brand": brand or "",
+            "color": color or "",
             "stock_g": f"{stock_g:.0f}",
             "rate_g_day": f"{rate_g_day:.1f}",
             "days_left": str(days_left),

backend/app/services/print_scheduler.py

@@ -30,7 +30,7 @@ from backend.app.models.settings import Settings
 from backend.app.models.smart_plug import SmartPlug
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
-from backend.app.services import drying_preflight, print_dispatch_context
+from backend.app.services import drying_preflight, print_dispatch_context, stock_forecast
 from backend.app.services.bambu_ftp import (
     FtpFailureReport,
     UploadCancelled,
@@ -111,6 +111,17 @@ def _ams_slot_label(ams_id: int, tray_id: int) -> str:
 # low spool does not need answering at that resolution (#2913).
 _FILAMENT_LOW_MIN_INTERVAL = 30.0
 
+# Minimum seconds between stock forecast checks (#2955). The forecast is in whole
+# days, so a finer check would only repeat the same answer.
+_STOCK_FORECAST_MIN_INTERVAL = 3600.0
+
+# The settings row that holds which stock alerts have already been sent (#2955).
+_STOCK_ALERTS_SETTING_KEY = "stock_alerts_notified"
+
+# The Inventory page asks for this many usage records and the forecast panel
+# builds its rates from them; the alert reads the same window so the two agree.
+_STOCK_FORECAST_HISTORY_LIMIT = 5000
+
 
 def _remaining_percent(label_weight: int | float | None, weight_used: float | None) -> float | None:
     """Remaining filament as a percentage of the label weight.
@@ -1081,6 +1092,18 @@ class PrintScheduler:
         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
+        # SKU key -> (condition, events told) for a SKU that is alerting (#2955): the
+        # condition is "reorder" or "break", the events are those that had a
+        # subscriber when it was sent. Absent means not alerting. Cleared when the
+        # condition clears, so it can alert again, and when nobody wants either
+        # event. Mirrored to one settings row (_STOCK_ALERTS_SETTING_KEY), because the
+        # first check runs as soon as Bambuddy starts and a restart would otherwise
+        # re-send one message per SKU that is still low.
+        self._notified_stock_alerts: dict[stock_forecast.SkuKey, tuple[str, frozenset[str]]] = {}
+        # The JSON last read from or written to that row; None until it has been read.
+        self._stock_alerts_persisted: str | None = None
+        # Earliest monotonic time the next stock forecast check may run (#2955).
+        self._stock_forecast_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;
@@ -1420,6 +1443,7 @@ class PrintScheduler:
                 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)
+                await self._check_stock_forecast(db)
                 return bool(self._inflight)
 
             logger.info(
@@ -2205,6 +2229,10 @@ class PrintScheduler:
             # the spools in the printers stopped mattering.
             await self._check_filament_low(db)
 
+            # Stock forecast: alert when a filament SKU reaches its reorder point or
+            # is about to run out before a replenishment could arrive (#2955).
+            await self._check_stock_forecast(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
@@ -4554,6 +4582,287 @@ class PrintScheduler:
             except Exception as e:
                 logger.warning("Low-filament notification failed for slot %s: %s", key, e)
 
+    async def _check_stock_forecast(self, db: AsyncSession) -> None:
+        """Alert when a filament SKU reaches its reorder point or is about to break (#2955).
+
+        ``on_stock_reorder_alert`` and ``on_stock_break_alert`` have had a column,
+        a schema field, a template and a UI toggle, and nothing that computes the
+        condition -- the Forecast panel does it in the browser, so with no page
+        open nothing could ever alert. This runs the same arithmetic
+        (``stock_forecast``) from the scheduler loop.
+
+        An alert is sent when a SKU *moves into* a condition, once, and again
+        after the condition has cleared. A SKU that worsens from reorder to break
+        alerts again as a break, because the panel treats the two as exclusive; one
+        that eases from break back to reorder does not alert again.
+        A SKU with alerts snoozed is treated as not alerting, so un-snoozing it
+        while it is still low tells you.
+
+        Time-gated to hourly for the reasons ``_check_filament_low`` is gated: the
+        loop runs every 3 s while an upload is in flight, and the whole spool
+        collection is read over HTTP in Spoolman mode. Both events default to off
+        on every provider, so nothing is read unless a provider wants one.
+        """
+        now = time.monotonic()
+        if now < self._stock_forecast_next_check:
+            return
+        self._stock_forecast_next_check = now + _STOCK_FORECAST_MIN_INTERVAL
+
+        try:
+            if self._stock_alerts_persisted is None:
+                await self._load_stock_alerts(db)
+            wanted = {
+                "reorder": bool(await notification_service._get_providers_for_event(db, "on_stock_reorder_alert")),
+                "break": bool(await notification_service._get_providers_for_event(db, "on_stock_break_alert")),
+            }
+            if not any(wanted.values()):
+                # Nobody wants either event. Forget what was sent, so whoever
+                # switches one on is told about the SKUs that are low now.
+                self._notified_stock_alerts.clear()
+                await self._save_stock_alerts(db)
+                return
+
+            forecasts = await self._stock_forecasts(db)
+            if forecasts is None:
+                # Spoolman unreachable. Deliberately not clearing the notified
+                # set: a brief outage would otherwise re-alert every SKU when it
+                # came back.
+                return
+
+            await self._emit_stock_alerts(db, forecasts, wanted)
+            await self._save_stock_alerts(db)
+        except Exception as e:
+            logger.warning("Stock forecast check failed: %s", e, exc_info=True)
+
+    def _serialize_stock_alerts(self) -> str:
+        """The notified set as JSON, in a fixed order so an unchanged set compares equal."""
+        rows = sorted([list(key), kind, sorted(events)] for key, (kind, events) in self._notified_stock_alerts.items())
+        return json.dumps(rows)
+
+    async def _load_stock_alerts(self, db: AsyncSession) -> None:
+        """Read the notified set saved by the last run, once per process.
+
+        A missing or unreadable row is an empty set: the worst that does is one
+        repeat of what a restart used to cause every time.
+        """
+        raw = (await db.execute(select(Settings).where(Settings.key == _STOCK_ALERTS_SETTING_KEY))).scalar_one_or_none()
+        loaded: dict[stock_forecast.SkuKey, tuple[str, frozenset[str]]] = {}
+        if raw and raw.value:
+            try:
+                for key, kind, events in json.loads(raw.value):
+                    if len(key) == 4 and kind in ("reorder", "break"):
+                        loaded[tuple(str(part) for part in key)] = (kind, frozenset(str(e) for e in events))
+            except (ValueError, TypeError):
+                logger.warning("Ignoring unreadable %s setting", _STOCK_ALERTS_SETTING_KEY)
+                loaded = {}
+        self._notified_stock_alerts = loaded
+        self._stock_alerts_persisted = self._serialize_stock_alerts()
+
+    async def _save_stock_alerts(self, db: AsyncSession) -> None:
+        """Write the notified set to its settings row if it changed since it was last read or written."""
+        serialized = self._serialize_stock_alerts()
+        if serialized == self._stock_alerts_persisted:
+            return
+        from backend.app.core.db_dialect import upsert_setting
+
+        await upsert_setting(db, Settings, _STOCK_ALERTS_SETTING_KEY, serialized)
+        await db.commit()
+        self._stock_alerts_persisted = serialized
+
+    async def _stock_forecasts(self, db: AsyncSession) -> stock_forecast.ForecastMap | None:
+        """Forecast every active SKU in whichever inventory mode is on. None if it cannot be read."""
+        from backend.app.models.filament_sku_settings import FilamentSkuSettings
+
+        global_lead_time = max(0, await self._get_int_setting(db, "forecast_global_lead_time_days", default=0))
+        sku_rows = (await db.execute(select(FilamentSkuSettings))).scalars().all()
+        sku_settings = {
+            stock_forecast.sku_key(row.material, row.subtype, row.brand, row.color_name): stock_forecast.SkuSettings(
+                lead_time_days=row.lead_time_days,
+                safety_margin_value=row.safety_margin_value,
+                safety_margin_unit=row.safety_margin_unit,
+                alerts_snoozed=bool(row.alerts_snoozed),
+            )
+            for row in sku_rows
+        }
+
+        if await self._get_bool_setting(db, "spoolman_enabled"):
+            spools = await self._stock_spools_spoolman()
+            if spools is None:
+                return None
+            # Spoolman owns the usage in this mode and Bambuddy's own history
+            # table holds nothing for those spools, so the rate is always the
+            # delta rate. Bambuddy's table is deliberately not consulted: its ids
+            # are local spool ids, and a Spoolman id that happens to match one
+            # (left over from before Spoolman was switched on) would borrow an
+            # unrelated spool's history.
+            history: dict[int, list[stock_forecast.UsageRecord]] = {}
+        else:
+            spools = await self._stock_spools_internal(db)
+            history = await self._stock_usage_history(db)
+
+        return stock_forecast.forecast_all(spools, history, sku_settings, global_lead_time, utcnow_naive())
+
+    async def _stock_spools_internal(self, db: AsyncSession) -> list[stock_forecast.StockSpool]:
+        from backend.app.models.spool import Spool
+
+        rows = (await db.execute(select(Spool).where(Spool.archived_at.is_(None)))).scalars().all()
+        return [
+            stock_forecast.StockSpool(
+                id=spool.id,
+                material=spool.material,
+                subtype=spool.subtype,
+                brand=spool.brand,
+                color_name=spool.color_name,
+                label_weight=float(spool.label_weight or 0),
+                weight_used=float(spool.weight_used or 0),
+                weight_used_baseline=float(spool.weight_used_baseline or 0),
+                created_at=spool.created_at,
+            )
+            for spool in rows
+        ]
+
+    async def _stock_usage_history(self, db: AsyncSession) -> dict[int, list[stock_forecast.UsageRecord]]:
+        from backend.app.models.spool_usage_history import SpoolUsageHistory
+
+        rows = (
+            await db.execute(
+                select(SpoolUsageHistory.spool_id, SpoolUsageHistory.created_at, SpoolUsageHistory.weight_used)
+                .order_by(SpoolUsageHistory.created_at.desc())
+                .limit(_STOCK_FORECAST_HISTORY_LIMIT)
+            )
+        ).all()
+        history: dict[int, list[stock_forecast.UsageRecord]] = {}
+        for spool_id, created_at, weight_used in rows:
+            if created_at is None:
+                continue
+            history.setdefault(spool_id, []).append(stock_forecast.UsageRecord(created_at, float(weight_used or 0)))
+        return history
+
+    async def _stock_spools_spoolman(self) -> list[stock_forecast.StockSpool] | None:
+        """Spoolman's spools in the forecast's shape, or None when Spoolman cannot be reached.
+
+        ``get_all_spools`` leaves archived spools out unless asked, which is the
+        exclusion the internal query applies explicitly. ``_map_spoolman_spool``
+        is the same mapping the Inventory page (and so the panel) reads.
+        """
+        from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool
+        from backend.app.services.spoolman import SpoolmanUnavailableError, get_spoolman_client
+
+        client = await get_spoolman_client()
+        if client is None:
+            logger.debug("Stock forecast skipped, no Spoolman client (spoolman_enabled without a URL?)")
+            return None
+        try:
+            raw_spools = await client.get_all_spools()
+        except SpoolmanUnavailableError as e:
+            logger.debug("Stock forecast skipped, Spoolman unreachable: %s", e)
+            return None
+
+        spools: list[stock_forecast.StockSpool] = []
+        for raw in raw_spools:
+            try:
+                mapped = _map_spoolman_spool(raw)
+            except ValueError:
+                continue
+            created_at = None
+            if mapped.get("created_at"):
+                try:
+                    created_at = datetime.fromisoformat(str(mapped["created_at"]).replace("Z", "+00:00"))
+                except ValueError:
+                    created_at = None
+            spools.append(
+                stock_forecast.StockSpool(
+                    id=mapped["id"],
+                    material=mapped["material"],
+                    subtype=mapped.get("subtype"),
+                    brand=mapped.get("brand"),
+                    color_name=mapped.get("color_name"),
+                    label_weight=float(mapped.get("label_weight") or 0),
+                    weight_used=float(mapped.get("weight_used") or 0),
+                    weight_used_baseline=float(mapped.get("weight_used_baseline") or 0),
+                    created_at=created_at,
+                    color_name_is_synthesized=bool(mapped.get("color_name_is_synthesized")),
+                )
+            )
+        return spools
+
+    async def _emit_stock_alerts(
+        self,
+        db: AsyncSession,
+        forecasts: stock_forecast.ForecastMap,
+        wanted: dict[str, bool],
+    ) -> None:
+        """Send the notifications for each SKU that has moved into a stock alert condition.
+
+        A SKU in break has also reached its reorder point, so both events apply to
+        it: the break event goes to providers that have it on, and the reorder event
+        to the providers that have only that one on. A provider with both on gets the
+        break and no second message.
+
+        What is remembered per SKU is the condition and which events had a
+        subscriber when it was sent. An event with no subscriber is forgotten, so
+        switching it on later reports the SKUs already in that condition.
+        """
+        active_events = {event for event, on in wanted.items() if on}
+        for key, (forecast, spool) in forecasts.items():
+            kind: str | None = None
+            if not forecast.snoozed:
+                if forecast.stock_break_alert:
+                    kind = "break"
+                elif forecast.reorder_alert:
+                    kind = "reorder"
+            applicable = {"break": {"break", "reorder"}, "reorder": {"reorder"}}.get(kind or "", set())
+            events = applicable & active_events
+            if not events:
+                self._notified_stock_alerts.pop(key, None)
+                continue
+
+            previous = self._notified_stock_alerts.get(key)
+            told = (previous[1] & active_events) if previous else frozenset()
+            # Trimmed to the events this condition can send. That is what makes easing from
+            # break back to reorder silent (the reorder event was told with the break) and
+            # worsening again news again (the break event is no longer remembered).
+            self._notified_stock_alerts[key] = (kind, frozenset((told | events) & applicable))
+            # Recorded before sending, for the reason _emit_filament_low gives: a
+            # provider that is down should lose this alert, not repeat it every hour.
+            to_send = events - told
+
+            # Both flags imply a positive rate and so a day count.
+            days_left = forecast.days_remaining if forecast.days_remaining is not None else 0
+            rate = forecast.daily_rate_g if forecast.daily_rate_g is not None else 0.0
+            color = None if spool.color_name_is_synthesized else spool.color_name
+            try:
+                if "break" in to_send:
+                    await notification_service.on_stock_break_alert(
+                        spool.material,
+                        spool.brand,
+                        forecast.remaining_g,
+                        rate,
+                        days_left,
+                        forecast.effective_lead_time_days,
+                        db,
+                        subtype=spool.subtype,
+                        color=color,
+                    )
+                if "reorder" in to_send:
+                    await notification_service.on_stock_reorder_alert(
+                        spool.material,
+                        spool.brand,
+                        forecast.remaining_g,
+                        rate,
+                        days_left,
+                        db,
+                        subtype=spool.subtype,
+                        color=color,
+                        skip_break_subscribers=kind == "break",
+                    )
+            except Exception as e:
+                logger.warning("Stock %s notification failed for %s: %s", kind, key, e)
+
+        # A SKU with no spools left is no longer alerting.
+        for key in [k for k in self._notified_stock_alerts if k not in forecasts]:
+            del self._notified_stock_alerts[key]
+
     async def _check_auto_drying(
         self,
         db: AsyncSession,

backend/app/services/stock_forecast.py

@@ -0,0 +1,284 @@
+"""Reorder-point forecast for inventory SKUs (#2955).
+
+This is the arithmetic ``frontend/src/components/ForecastPanel.tsx`` runs in the
+browser, as pure functions, so something that is not a browser can decide when
+a SKU has reached its reorder point or is about to break. ``test_stock_forecast_2955.py``
+pins the two to the same numbers on the same inputs.
+
+The pieces, in the order the panel applies them:
+
+* Spools are grouped by SKU -- ``(material, subtype, brand, color_name)``.
+* The daily rate comes from usage history when there are at least two distinct
+  days of it (weighted by a 30-day half-life, so recent prints dominate), and
+  from consumption since the spool baseline over the age of the oldest spool
+  otherwise.
+* ``reorder point = rate * lead time + safety stock``, where the safety stock
+  is a statistical term (``Z_95 * sigma * sqrt(lead time)``) plus a margin the
+  user sets in days or grams.
+* A *stock break* is "the stock runs out before a replenishment can arrive"; a
+  *reorder* is "the reorder point has been reached". They are exclusive: a SKU
+  that is already breaking is not also reported as reorderable.
+
+Nothing here touches the database. The caller supplies spools, their usage
+history and the settings, and decides what to do with the answer.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+
+# One-sided 95 % z-score, as in the panel.
+Z_95 = 1.65
+
+# Prints from this many days ago count half as much as one from today.
+_HALF_LIFE_DAYS = 30.0
+_LAMBDA = math.log(2) / _HALF_LIFE_DAYS
+
+DEFAULT_SAFETY_MARGIN_VALUE = 14
+DEFAULT_SAFETY_MARGIN_UNIT = "days"
+
+# With no measured spread the panel assumes 20 % of the rate, and with no rate at
+# all it assumes 5 g/day so a margin given in days still means something.
+_FALLBACK_SIGMA_FRACTION = 0.2
+_FALLBACK_RATE_G_PER_DAY = 5
+
+_SECONDS_PER_DAY = 86400.0
+
+SkuKey = tuple[str, str, str, str]
+
+
+def sku_key(material: str, subtype: str | None, brand: str | None, color_name: str | None) -> SkuKey:
+    """The grouping key. ``None`` and an empty string are the same SKU, as in the panel."""
+    return (material, subtype or "", brand or "", color_name or "")
+
+
+@dataclass(frozen=True)
+class StockSpool:
+    """The fields of a spool the forecast reads, whichever inventory mode it came from."""
+
+    id: int
+    material: str
+    subtype: str | None
+    brand: str | None
+    color_name: str | None
+    label_weight: float
+    weight_used: float
+    weight_used_baseline: float
+    created_at: datetime | None
+    # True when color_name is not a name the spool carries but the subtype standing in
+    # for one (Spoolman mode does this so the list has no blank colours). The SKU key
+    # keeps it, as the panel does; anything shown to a person should not.
+    color_name_is_synthesized: bool = False
+
+    @property
+    def key(self) -> SkuKey:
+        return sku_key(self.material, self.subtype, self.brand, self.color_name)
+
+    @property
+    def remaining_g(self) -> float:
+        return max(0.0, self.label_weight - self.weight_used)
+
+    @property
+    def consumed_g(self) -> float:
+        """Consumed since the baseline, so "Reset usage to 0" restarts the forecast (#1390)."""
+        return max(0.0, self.weight_used - self.weight_used_baseline)
+
+
+@dataclass(frozen=True)
+class UsageRecord:
+    """One consumption event: how much a print took from a spool, and when."""
+
+    created_at: datetime
+    weight_used: float
+
+
+@dataclass(frozen=True)
+class SkuSettings:
+    """A SKU's user-set overrides. The defaults are the panel's for a SKU with no row."""
+
+    lead_time_days: int = 0
+    safety_margin_value: float = DEFAULT_SAFETY_MARGIN_VALUE
+    safety_margin_unit: str = DEFAULT_SAFETY_MARGIN_UNIT
+    alerts_snoozed: bool = False
+
+
+@dataclass(frozen=True)
+class SkuForecast:
+    key: SkuKey
+    remaining_g: float
+    daily_rate_g: float | None
+    effective_lead_time_days: int
+    reorder_point_g: float
+    days_remaining: int | None
+    days_until_reorder_point: int | None
+    stock_break_alert: bool
+    reorder_alert: bool
+    snoozed: bool
+
+
+def _utc(moment: datetime) -> datetime:
+    """A naive timestamp is UTC: that is what the database stores."""
+    return moment.replace(tzinfo=timezone.utc) if moment.tzinfo is None else moment.astimezone(timezone.utc)
+
+
+def history_rate(records: Sequence[UsageRecord], now: datetime) -> tuple[float, float] | None:
+    """Daily consumption rate and its standard deviation from usage history.
+
+    Records are summed per UTC calendar day, so concurrent multi-spool prints on
+    one day count together. Each day after the first gives one observation: the
+    grams printed that day over the gap since the previous day with any usage,
+    weighted by ``exp(-lambda * age)``. Returns None with fewer than two
+    distinct days -- there is no gap to measure -- and the caller falls back to
+    :func:`delta_rate`.
+    """
+    if len(records) < 2:
+        return None
+
+    by_day: dict[date, float] = {}
+    for record in records:
+        day = _utc(record.created_at).date()
+        by_day[day] = by_day.get(day, 0.0) + record.weight_used
+    if len(by_day) < 2:
+        return None
+
+    now = _utc(now)
+    days = sorted(by_day.items())
+    observations: list[tuple[float, float]] = []  # (rate, weight)
+    for i in range(1, len(days)):
+        elapsed_days = max((days[i][0] - days[i - 1][0]).days, 1)
+        midnight = datetime(days[i][0].year, days[i][0].month, days[i][0].day, tzinfo=timezone.utc)
+        age_days = (now - midnight).total_seconds() / _SECONDS_PER_DAY
+        observations.append((days[i][1] / elapsed_days, math.exp(-_LAMBDA * age_days)))
+
+    total_weight = sum(w for _, w in observations)
+    if total_weight == 0:
+        return None
+    mean = sum(r * w for r, w in observations) / total_weight
+    variance = sum(w * (r - mean) ** 2 for r, w in observations) / total_weight
+    return mean, math.sqrt(variance)
+
+
+def delta_rate(spools: Sequence[StockSpool], now: datetime) -> float | None:
+    """Daily rate from consumption since baseline over the age of the oldest spool.
+
+    The fallback when there is not enough usage history, which is always the
+    case in Spoolman mode: Spoolman owns the usage there and Bambuddy's own
+    ``spool_usage_history`` table holds nothing for those spools. Returns None
+    with nothing consumed, or with under a day of age to divide by.
+    """
+    total_used = sum(s.consumed_g for s in spools)
+    if total_used == 0:
+        return None
+    now = _utc(now)
+    oldest = now
+    for spool in spools:
+        if spool.created_at is not None:
+            created = _utc(spool.created_at)
+            if created < oldest:
+                oldest = created
+    age_days = (now - oldest).total_seconds() / _SECONDS_PER_DAY
+    if age_days < 1:
+        return None
+    return total_used / age_days
+
+
+def forecast_sku(
+    spools: Sequence[StockSpool],
+    history_by_spool: Mapping[int, Sequence[UsageRecord]],
+    settings: SkuSettings,
+    global_lead_time_days: int,
+    now: datetime,
+) -> SkuForecast:
+    """Forecast one SKU from its spools. ``spools`` must be non-empty and share a key."""
+    lead_time = max(global_lead_time_days, settings.lead_time_days)
+    remaining = sum(s.remaining_g for s in spools)
+
+    # Only history from spools that were never reset: a reset spool's earlier
+    # events have no anchor and would inflate the rate. A spool with no
+    # baseline is clean and keeps its records.
+    history: list[UsageRecord] = []
+    for spool in spools:
+        if spool.weight_used_baseline == 0:
+            history.extend(history_by_spool.get(spool.id, ()))
+
+    rate: float | None = None
+    std_dev: float | None = None
+    measured = history_rate(history, now)
+    if measured is not None:
+        rate, std_dev = measured
+    else:
+        rate = delta_rate(spools, now)
+
+    sigma = std_dev if std_dev is not None else (rate * _FALLBACK_SIGMA_FRACTION if rate is not None else 0.0)
+    statistical_safety_g = Z_95 * sigma * math.sqrt(lead_time)
+    if settings.safety_margin_unit == "g":
+        margin_g = settings.safety_margin_value
+    elif rate is not None:
+        margin_g = rate * settings.safety_margin_value
+    else:
+        margin_g = settings.safety_margin_value * _FALLBACK_RATE_G_PER_DAY
+    safety_stock_g = statistical_safety_g + margin_g
+    reorder_point_g = rate * lead_time + safety_stock_g if rate is not None else 0.0
+
+    days_remaining: int | None = None
+    days_until_rop: int | None = None
+    if rate is not None and rate > 0:
+        days_remaining = math.floor(remaining / rate)
+        days_until_rop = math.floor((remaining - reorder_point_g) / rate)
+
+    stock_break = days_remaining is not None and lead_time > 0 and days_remaining <= lead_time
+    reorder = not stock_break and days_until_rop is not None and days_until_rop <= 0
+
+    return SkuForecast(
+        key=spools[0].key,
+        remaining_g=remaining,
+        daily_rate_g=rate,
+        effective_lead_time_days=lead_time,
+        reorder_point_g=reorder_point_g,
+        days_remaining=days_remaining,
+        days_until_reorder_point=days_until_rop,
+        stock_break_alert=stock_break,
+        reorder_alert=reorder,
+        snoozed=settings.alerts_snoozed,
+    )
+
+
+ForecastMap = dict[SkuKey, tuple[SkuForecast, StockSpool]]
+"""Each SKU's forecast with the first spool of its group, which names it."""
+
+
+def forecast_all(
+    spools: Sequence[StockSpool],
+    history_by_spool: Mapping[int, Sequence[UsageRecord]],
+    sku_settings: Mapping[SkuKey, SkuSettings],
+    global_lead_time_days: int,
+    now: datetime,
+) -> ForecastMap:
+    """Forecast every SKU present in ``spools``.
+
+    Returns each SKU's forecast with the first spool of its group, which is what
+    a caller needs to name the SKU (material, subtype, brand, colour) without a
+    second lookup. The caller passes only spools that count as stock -- archived
+    ones excluded -- because what is archived differs by inventory mode.
+
+    A colour-specific SKU with no settings row of its own falls back to the
+    colourless row, which is where settings saved before colour became part of
+    the key still live.
+    """
+    groups: dict[SkuKey, list[StockSpool]] = {}
+    for spool in spools:
+        groups.setdefault(spool.key, []).append(spool)
+
+    out: ForecastMap = {}
+    for key, members in groups.items():
+        settings = sku_settings.get(key)
+        if settings is None and key[3] != "":
+            settings = sku_settings.get((key[0], key[1], key[2], ""))
+        out[key] = (
+            forecast_sku(members, history_by_spool, settings or SkuSettings(), global_lead_time_days, now),
+            members[0],
+        )
+    return out

backend/tests/unit/services/test_stock_alert_producer_2955.py

@@ -0,0 +1,989 @@
+"""The two Inventory stock alerts have a producer now (issue #2955).
+
+``on_stock_reorder_alert`` and ``on_stock_break_alert`` had a column, a schema
+field, a route, a template and a UI toggle -- and, until #2945, could not even be
+saved -- but nothing that decides a SKU has reached its reorder point: the Forecast
+panel does that in the browser, so with no page open nothing could ever alert.
+``PrintScheduler._check_stock_forecast`` runs the panel's arithmetic
+(``stock_forecast``) from the scheduler loop.
+
+The arithmetic itself is pinned against the panel in ``test_stock_forecast_2955``;
+these tests are about *when* it alerts and what it sends.
+
+The scenarios use the delta rate (grams consumed over the age of the spool)
+because it needs no usage history: a 1000 g spool created 100 days ago with
+``used`` grams consumed runs at ``used / 100`` g/day. With a 6 day lead time and
+the default 14 day margin:
+
+    used  920 -> 80 g left, 9.2 g/day, 8 days of stock, reorder point ~90 g : reorder
+    used  960 -> 40 g left, 9.6 g/day, 4 days of stock                       : break
+    used  500 -> 500 g left, 5 g/day, 100 days of stock                      : nothing
+"""
+
+import json
+import logging
+import time
+from datetime import datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.models.filament_sku_settings import FilamentSkuSettings
+from backend.app.models.notification import NotificationProvider
+from backend.app.models.settings import Settings
+from backend.app.models.spool import Spool
+from backend.app.services.print_scheduler import _STOCK_ALERTS_SETTING_KEY, PrintScheduler
+from backend.app.services.spoolman import SpoolmanUnavailableError
+from backend.app.utils.local_time import utcnow_naive
+
+REORDER_USED = 920.0
+BREAK_USED = 960.0
+QUIET_USED = 500.0
+
+
+async def _spool(
+    db,
+    *,
+    used: float,
+    color: str | None = "Black",
+    subtype: str | None = "Basic",
+    brand: str | None = "Bambu Lab",
+    age_days: int = 100,
+    archived: bool = False,
+    baseline: float = 0.0,
+) -> Spool:
+    spool = Spool(
+        material="PLA",
+        subtype=subtype,
+        brand=brand,
+        color_name=color,
+        label_weight=1000,
+        core_weight=250,
+        weight_used=used,
+        weight_used_baseline=baseline,
+        created_at=utcnow_naive() - timedelta(days=age_days),
+        archived_at=utcnow_naive() if archived else None,
+    )
+    spool.k_profiles = []
+    spool.assignments = []
+    db.add(spool)
+    await db.flush()
+    return spool
+
+
+async def _sku(db, *, lead: int = 6, color: str | None = "Black", snoozed: bool = False, margin: int = 14):
+    row = FilamentSkuSettings(
+        material="PLA",
+        subtype="Basic",
+        brand="Bambu Lab",
+        color_name=color,
+        lead_time_days=lead,
+        safety_margin_value=margin,
+        safety_margin_unit="days",
+        alerts_snoozed=snoozed,
+    )
+    db.add(row)
+    await db.flush()
+    return row
+
+
+@pytest.fixture
+def scheduler():
+    return PrintScheduler()
+
+
+async def _pass(scheduler, db):
+    """Run one check, stepping over the hourly gate (which has its own test)."""
+    scheduler._stock_forecast_next_check = 0.0
+    await scheduler._check_stock_forecast(db)
+
+
+@pytest.fixture
+def notify():
+    """The service, replaced, with a provider that wants both events.
+
+    Every test about the producer assumes someone wants the alerts; the guard in
+    front of it has its own tests further down, against the real provider query.
+    """
+    with patch("backend.app.services.print_scheduler.notification_service") as ns:
+        ns.on_stock_reorder_alert = AsyncMock()
+        ns.on_stock_break_alert = AsyncMock()
+        ns._get_providers_for_event = AsyncMock(return_value=[MagicMock()])
+        yield ns
+
+
+# -- a restart does not start it over ----------------------------------------
+
+
+async def _stored(db) -> list | None:
+    """What the settings row holds, parsed. None when there is no row."""
+    from sqlalchemy import select
+
+    row = (await db.execute(select(Settings).where(Settings.key == _STOCK_ALERTS_SETTING_KEY))).scalar_one_or_none()
+    return json.loads(row.value) if row else None
+
+
+@pytest.mark.asyncio
+async def test_a_fresh_scheduler_does_not_re_send_what_is_already_stored(db_session, scheduler, notify):
+    """The first check runs as soon as Bambuddy starts. Without the stored set, every update
+    re-sent one message per SKU that was still low."""
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+    restarted = PrintScheduler()
+    await _pass(restarted, db_session)
+
+    assert notify.on_stock_reorder_alert.await_count == 1
+    notify.on_stock_break_alert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_a_stored_reorder_still_lets_the_worsening_to_a_break_through_after_a_restart(
+    db_session, scheduler, notify
+):
+    await _sku(db_session)
+    spool = await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    spool.weight_used = BREAK_USED
+    await db_session.commit()
+    restarted = PrintScheduler()
+    await _pass(restarted, db_session)
+
+    notify.on_stock_break_alert.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_a_sku_that_cleared_while_it_was_down_alerts_again_after_a_restart(db_session, scheduler, notify):
+    await _sku(db_session)
+    spool = await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    spool.weight_used = QUIET_USED
+    await db_session.commit()
+    await _pass(PrintScheduler(), db_session)
+    spool.weight_used = REORDER_USED
+    await db_session.commit()
+    await _pass(PrintScheduler(), db_session)
+
+    assert notify.on_stock_reorder_alert.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_the_stored_set_drops_a_sku_that_no_longer_exists(db_session, scheduler, notify):
+    await _sku(db_session)
+    spool = await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert len(await _stored(db_session)) == 1
+
+    spool.archived_at = utcnow_naive()
+    await db_session.commit()
+    await _pass(PrintScheduler(), db_session)
+
+    assert await _stored(db_session) == []
+
+
+@pytest.mark.asyncio
+async def test_the_stored_set_is_emptied_when_nobody_wants_either_event(db_session, scheduler, real_providers):
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    provider_row = NotificationProvider(
+        name="p", provider_type="ntfy", config="{}", enabled=True, on_stock_reorder_alert=True
+    )
+    db_session.add(provider_row)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert len(await _stored(db_session)) == 1
+
+    provider_row.on_stock_reorder_alert = False
+    await db_session.commit()
+    await _pass(PrintScheduler(), db_session)
+
+    assert await _stored(db_session) == []
+
+
+@pytest.mark.asyncio
+async def test_an_unchanged_set_is_not_written_again(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    with patch("backend.app.core.db_dialect.upsert_setting", AsyncMock()) as upsert:
+        await _pass(scheduler, db_session)
+        await _pass(PrintScheduler(), db_session)
+
+    upsert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_an_unreadable_stored_set_is_treated_as_empty(db_session, scheduler, notify):
+    db_session.add(Settings(key=_STOCK_ALERTS_SETTING_KEY, value="not json {"))
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    assert len(await _stored(db_session)) == 1
+
+
+# -- when it alerts ----------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_sku_at_its_reorder_point_alerts_once_with_its_numbers(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_break_alert.assert_not_awaited()
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    material, brand, stock_g, rate, days_left, _db = notify.on_stock_reorder_alert.await_args.args
+    assert (material, brand) == ("PLA", "Bambu Lab")
+    assert stock_g == pytest.approx(80.0)
+    assert rate == pytest.approx(9.2, rel=1e-3)
+    assert days_left == 8
+    kwargs = notify.on_stock_reorder_alert.await_args.kwargs
+    assert kwargs == {"subtype": "Basic", "color": "Black", "skip_break_subscribers": False}
+
+
+@pytest.mark.asyncio
+async def test_a_sku_that_runs_out_before_the_lead_time_alerts_as_a_break(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _spool(db_session, used=BREAK_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_break_alert.assert_awaited_once()
+    args = notify.on_stock_break_alert.await_args.args
+    assert args[2] == pytest.approx(40.0)
+    assert args[4] == 4  # days left
+    assert args[5] == 6  # lead time
+    assert notify.on_stock_break_alert.await_args.kwargs == {"subtype": "Basic", "color": "Black"}
+    # A breaking SKU has also reached its reorder point: the reorder event goes too, minus the
+    # providers that have the break alert on (the service applies that filter; see below).
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    assert notify.on_stock_reorder_alert.await_args.kwargs["skip_break_subscribers"] is True
+
+
+@pytest.mark.asyncio
+async def test_a_sku_with_plenty_of_stock_stays_quiet(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _spool(db_session, used=QUIET_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_not_awaited()
+    notify.on_stock_break_alert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_a_sku_that_has_not_been_used_has_no_rate_and_stays_quiet(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _spool(db_session, used=0.0)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_not_awaited()
+    notify.on_stock_break_alert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_the_global_lead_time_counts(db_session, scheduler, notify):
+    """No SKU row at all: the lead time is the global setting, and 40 g at 9.6 g/day is 4 days."""
+    db_session.add(Settings(key="forecast_global_lead_time_days", value="30"))
+    await _spool(db_session, used=BREAK_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_break_alert.assert_awaited_once()
+    assert notify.on_stock_break_alert.await_args.args[5] == 30
+
+
+@pytest.mark.asyncio
+async def test_usage_history_beats_the_delta_rate_when_there_is_enough_of_it(db_session, scheduler, notify):
+    """300 g used over 100 days is 3 g/day and 700 g left is months of stock. But the
+    last three days took 300 g each: the history rate says two days, so it is a break.
+
+    Also the only test that reads ``spool_usage_history`` through the scheduler.
+    """
+    from backend.app.models.spool_usage_history import SpoolUsageHistory
+
+    await _sku(db_session)
+    spool = await _spool(db_session, used=300.0)
+    for days_ago in (3, 2, 1):
+        db_session.add(
+            SpoolUsageHistory(
+                spool_id=spool.id, weight_used=300.0, created_at=utcnow_naive() - timedelta(days=days_ago)
+            )
+        )
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_break_alert.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_history_of_a_reset_spool_is_left_out(db_session, scheduler, notify):
+    """Same history, but the spool was reset (baseline set): its earlier events have no
+    anchor, so the rate falls back to the delta rate -- 3 g/day, nothing to alert."""
+    from backend.app.models.spool_usage_history import SpoolUsageHistory
+
+    await _sku(db_session)
+    spool = await _spool(db_session, used=300.0, baseline=100.0)
+    for days_ago in (3, 2, 1):
+        db_session.add(
+            SpoolUsageHistory(
+                spool_id=spool.id, weight_used=300.0, created_at=utcnow_naive() - timedelta(days=days_ago)
+            )
+        )
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_break_alert.assert_not_awaited()
+    notify.on_stock_reorder_alert.assert_not_awaited()
+
+
+# -- once per transition -----------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_does_not_repeat_on_the_next_pass(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    await _pass(scheduler, db_session)
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_re_arms_once_the_condition_clears(db_session, scheduler, notify):
+    """Restocked (a fresh roll brings the SKU back above the line), then low again."""
+    await _sku(db_session)
+    spool = await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+    spool.weight_used = QUIET_USED
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+    spool.weight_used = REORDER_USED
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_worsening_from_reorder_to_break_alerts_again_as_a_break(db_session, scheduler, notify):
+    await _sku(db_session)
+    spool = await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    spool.weight_used = BREAK_USED
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    notify.on_stock_break_alert.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_easing_from_break_to_reorder_is_not_announced_again(db_session, scheduler, notify):
+    """Ordered a few rolls' worth, so it is no longer a break but still under the reorder point.
+    "Reached the reorder point" straight after "order immediately" is the opposite news."""
+    await _sku(db_session)
+    spool = await _spool(db_session, used=BREAK_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    spool.weight_used = REORDER_USED
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    assert notify.on_stock_break_alert.await_count == 1
+    assert notify.on_stock_reorder_alert.await_count == 1  # the break's own, from before it eased
+
+    # Worse again is news again.
+    spool.weight_used = BREAK_USED
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_break_alert.await_count == 2
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_two_colours_of_one_product_alert_separately_and_say_which(db_session, scheduler, notify):
+    """The SKU is material/subtype/brand/colour, so the messages must differ by colour."""
+    await _sku(db_session, color="Black")
+    await _sku(db_session, color="White")
+    await _spool(db_session, used=REORDER_USED, color="Black")
+    await _spool(db_session, used=REORDER_USED, color="White")
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    assert notify.on_stock_reorder_alert.await_count == 2
+    colours = {call.kwargs["color"] for call in notify.on_stock_reorder_alert.await_args_list}
+    assert colours == {"Black", "White"}
+
+
+@pytest.mark.asyncio
+async def test_a_sku_that_disappears_and_comes_back_low_alerts_again(db_session, scheduler, notify):
+    """The roll runs out and is archived, so the SKU has no spools; a new roll is later low.
+
+    The old alert must not silence the new roll: nothing has cleared the SKU's
+    entry, because it was not in the forecast to be cleared.
+    """
+    await _sku(db_session)
+    spool = await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+    spool.archived_at = utcnow_naive()
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    assert notify.on_stock_reorder_alert.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_spools_of_one_sku_are_added_together(db_session, scheduler, notify):
+    """Two half-full spools are one SKU with 1000 g, not two SKUs with 500 g each."""
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)  # 80 g left, on its own a reorder
+    await _spool(db_session, used=QUIET_USED)  # 500 g left; together 580 g at 14.2 g/day, 40 days
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_not_awaited()
+    notify.on_stock_break_alert.assert_not_awaited()
+
+
+# -- snooze and archive ------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_snoozed_sku_does_not_alert_and_alerts_when_unsnoozed(db_session, scheduler, notify):
+    row = await _sku(db_session, snoozed=True)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    notify.on_stock_reorder_alert.assert_not_awaited()
+
+    row.alerts_snoozed = False
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    notify.on_stock_reorder_alert.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_a_colour_with_no_row_of_its_own_uses_the_colourless_row(db_session, scheduler, notify):
+    """Settings saved before colour became part of the key live on the colourless row."""
+    await _sku(db_session, color=None, snoozed=True)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_an_archived_spool_is_not_stock(db_session, scheduler, notify):
+    """It would supply 900 g of stock the SKU does not have."""
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await _spool(db_session, used=0.0, archived=True)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    assert notify.on_stock_reorder_alert.await_args.args[2] == pytest.approx(80.0)
+
+
+# -- failure and cost --------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_failing_provider_does_not_break_the_pass(db_session, scheduler, notify):
+    await _sku(db_session)
+    await _sku(db_session, color="White")
+    await _spool(db_session, used=REORDER_USED, color="Black")
+    await _spool(db_session, used=REORDER_USED, color="White")
+    await db_session.commit()
+    notify.on_stock_reorder_alert.side_effect = [RuntimeError("provider down"), None]
+
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 2
+
+    # The failed alert is lost, not retried every hour: that repetition is what the
+    # debounce is for, and a provider that is down is the case where it would never stop.
+    notify.on_stock_reorder_alert.side_effect = None
+    await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_a_second_pass_inside_the_interval_does_no_work(db_session, scheduler, notify):
+    """The gate. run() re-enters the checks every 3 s while an upload is in flight."""
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    with patch.object(scheduler, "_stock_forecasts", new_callable=AsyncMock) as work:
+        work.return_value = {}
+        await scheduler._check_stock_forecast(db_session)
+        wait = scheduler._stock_forecast_next_check - time.monotonic()
+        await scheduler._check_stock_forecast(db_session)
+        await scheduler._check_stock_forecast(db_session)
+
+    assert work.await_count == 1
+    # Hourly: the forecast is in whole days, so a finer check repeats the same answer.
+    assert 3500 < wait <= 3600
+
+
+# -- the provider guard ------------------------------------------------------
+
+
+async def _provider(db, *, enabled: bool = True, reorder: bool = False, brk: bool = False) -> None:
+    db.add(
+        NotificationProvider(
+            name=f"p-{enabled}-{reorder}-{brk}",
+            provider_type="ntfy",
+            config="{}",
+            enabled=enabled,
+            on_stock_reorder_alert=reorder,
+            on_stock_break_alert=brk,
+        )
+    )
+    await db.flush()
+
+
+@pytest.fixture
+def real_providers():
+    """The real service, with only the two sends replaced, so the guard runs its real query."""
+    from backend.app.services.print_scheduler import notification_service
+
+    with (
+        patch.object(notification_service, "on_stock_reorder_alert", new_callable=AsyncMock) as reorder,
+        patch.object(notification_service, "on_stock_break_alert", new_callable=AsyncMock) as brk,
+    ):
+        yield reorder, brk
+
+
+@pytest.mark.asyncio
+async def test_no_work_when_no_provider_wants_either_event(db_session, scheduler, real_providers):
+    """Both toggles default to off, so on most installs nobody wants this.
+
+    Unguarded, every install read its whole spool collection every hour (in
+    Spoolman mode, over HTTP) for alerts that were switched off. The two
+    providers are the two ways of not wanting them: enabled with both events off,
+    and the events on but the provider disabled.
+    """
+    await _provider(db_session, enabled=True)
+    await _provider(db_session, enabled=False, reorder=True, brk=True)
+    await _spool(db_session, used=REORDER_USED)
+    await db_session.commit()
+
+    with patch.object(scheduler, "_stock_forecasts", new_callable=AsyncMock) as work:
+        await _pass(scheduler, db_session)
+
+    work.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_an_event_nobody_wants_is_not_recorded_as_sent(db_session, scheduler, real_providers):
+    """Only the break event is on, and the SKU is at its reorder point. Switching
+    the reorder event on later must report it -- it was never sent."""
+    reorder, brk = real_providers
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    provider_row = NotificationProvider(
+        name="p", provider_type="ntfy", config="{}", enabled=True, on_stock_break_alert=True
+    )
+    db_session.add(provider_row)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    reorder.assert_not_awaited()
+    brk.assert_not_awaited()
+
+    provider_row.on_stock_reorder_alert = True
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    reorder.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_a_reorder_only_provider_still_hears_about_a_break(db_session, scheduler, real_providers):
+    """A SKU that is breaking has certainly reached its reorder point, which is what the
+    Reorder Alert toggle promises. Without this, the worst state is the one a provider
+    with only that toggle on is never told about."""
+    reorder, brk = real_providers
+    await _sku(db_session)
+    await _spool(db_session, used=BREAK_USED)
+    await _provider(db_session, reorder=True)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    reorder.assert_awaited_once()
+    assert reorder.await_args.kwargs["skip_break_subscribers"] is True
+    brk.assert_not_awaited()  # nobody has it on
+
+
+@pytest.mark.asyncio
+async def test_switching_the_break_alert_on_later_reports_the_break(db_session, scheduler, real_providers):
+    """The reorder-only provider was told when the SKU broke. A provider that turns the break
+    alert on afterwards is told too -- and the reorder-only one is not told a second time."""
+    reorder, brk = real_providers
+    await _sku(db_session)
+    await _spool(db_session, used=BREAK_USED)
+    await _provider(db_session, reorder=True)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert reorder.await_count == 1
+
+    await _provider(db_session, brk=True)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    brk.assert_awaited_once()
+    assert reorder.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_switching_one_event_off_and_on_again_reports_it_while_the_other_stays_on(
+    db_session, scheduler, real_providers
+):
+    """A wants reorders, B wants breaks, and the SKU is breaking: both were told. B switches
+    the break alert off (A keeps the check running) and on again: B is told again."""
+    reorder, brk = real_providers
+    await _sku(db_session)
+    await _spool(db_session, used=BREAK_USED)
+    await _provider(db_session, reorder=True)
+    b_provider = NotificationProvider(
+        name="b", provider_type="ntfy", config="{}", enabled=True, on_stock_break_alert=True
+    )
+    db_session.add(b_provider)
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    assert (reorder.await_count, brk.await_count) == (1, 1)
+
+    b_provider.on_stock_break_alert = False
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    b_provider.on_stock_break_alert = True
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    assert brk.await_count == 2
+    assert reorder.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_a_break_alert_only_provider_is_not_told_about_a_reorder(db_session, scheduler, real_providers):
+    reorder, brk = real_providers
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    await _provider(db_session, brk=True)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+
+    reorder.assert_not_awaited()
+    brk.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_a_provider_with_both_on_gets_the_break_and_no_second_message(db_session, scheduler):
+    """Through the real service: A has both toggles on, B only reorder. A break SKU goes to A as
+    a break and to B as a reorder -- and A does not also get the reorder."""
+    from backend.app.services.notification_service import NotificationService
+    from backend.app.services.print_scheduler import notification_service
+
+    await _sku(db_session)
+    await _spool(db_session, used=BREAK_USED)
+    both = NotificationProvider(
+        name="both",
+        provider_type="ntfy",
+        config="{}",
+        enabled=True,
+        on_stock_reorder_alert=True,
+        on_stock_break_alert=True,
+    )
+    only_reorder = NotificationProvider(
+        name="only-reorder", provider_type="ntfy", config="{}", enabled=True, on_stock_reorder_alert=True
+    )
+    db_session.add_all([both, only_reorder])
+    await db_session.commit()
+
+    sent: list[tuple[str, list[str]]] = []
+
+    async def record(providers, title, message, db, event_type, **kwargs):
+        sent.append((event_type, sorted(p.name for p in providers)))
+
+    with (
+        patch.object(notification_service, "_send_to_providers", side_effect=record),
+        patch.object(notification_service, "_build_message_from_template", AsyncMock(return_value=("t", "b"))),
+    ):
+        await _pass(scheduler, db_session)
+
+    assert isinstance(notification_service, NotificationService)
+    assert sorted(sent) == [("stock_break_alert", ["both"]), ("stock_reorder_alert", ["only-reorder"])]
+
+
+@pytest.mark.asyncio
+async def test_switching_every_event_off_and_on_again_reports_what_is_low_now(db_session, scheduler, real_providers):
+    reorder, _brk = real_providers
+    await _sku(db_session)
+    await _spool(db_session, used=REORDER_USED)
+    provider_row = NotificationProvider(
+        name="p", provider_type="ntfy", config="{}", enabled=True, on_stock_reorder_alert=True
+    )
+    db_session.add(provider_row)
+    await db_session.commit()
+
+    await _pass(scheduler, db_session)
+    assert reorder.await_count == 1
+
+    provider_row.on_stock_reorder_alert = False
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+    provider_row.on_stock_reorder_alert = True
+    await db_session.commit()
+    await _pass(scheduler, db_session)
+
+    assert reorder.await_count == 2
+
+
+# -- Spoolman mode -----------------------------------------------------------
+
+
+def _spoolman_spool(spool_id: int, used: float, *, color: str = "Black", age_days: int = 100) -> dict:
+    """A raw Spoolman spool, in the shape _map_spoolman_spool reads."""
+    registered = (utcnow_naive() - timedelta(days=age_days)).isoformat() + "Z"
+    return {
+        "id": spool_id,
+        "remaining_weight": 1000 - used,
+        "used_weight": used,
+        "registered": registered,
+        "filament": {
+            "id": 1,
+            "name": "PLA Basic",
+            "material": "PLA",
+            "weight": 1000,
+            "color_name": color,
+            "vendor": {"name": "Bambu Lab"},
+        },
+    }
+
+
+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_one_collection_read(db_session, scheduler, notify):
+    db_session.add(Settings(key="spoolman_enabled", value="true"))
+    await db_session.commit()
+    client = _spoolman_client([_spoolman_spool(7, used=REORDER_USED), _spoolman_spool(8, used=0.0, color="White")])
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
+        await _pass(scheduler, db_session)
+
+    # Black has 80 g at 9.2 g/day and no SKU row: lead time 0, so a reorder needs the
+    # 14 day margin -- 80 g is under the ~128 g that gives.
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    assert notify.on_stock_reorder_alert.await_args.kwargs["color"] == "Black"
+    client.get_all_spools.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_an_unreachable_spoolman_stays_quiet_and_does_not_forget(db_session, scheduler, notify, caplog):
+    """An outage is an ordinary state for a third-party service. It must not raise, and it
+    must not clear what was sent: the SKU would alert again the moment Spoolman came back."""
+    db_session.add(Settings(key="spoolman_enabled", value="true"))
+    await db_session.commit()
+    spools = [_spoolman_spool(7, used=REORDER_USED)]
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client(spools))):
+        await _pass(scheduler, db_session)
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+    down = _spoolman_client([], 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=down)),
+    ):
+        await _pass(scheduler, db_session)
+    # An ordinary state for a third-party service, not a traceback every hour.
+    assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client(spools))):
+        await _pass(scheduler, db_session)
+
+    assert notify.on_stock_reorder_alert.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_a_colour_spoolman_made_up_from_the_subtype_is_not_sent_as_a_colour(db_session, scheduler, notify):
+    """With no colour of its own, Spoolman mode shows the subtype in its place so the list has
+    no blank colours. In a message that reads "PLA Basic Basic", so the colour is left out."""
+    db_session.add(Settings(key="spoolman_enabled", value="true"))
+    await db_session.commit()
+    raw = _spoolman_spool(7, used=REORDER_USED)
+    del raw["filament"]["color_name"]
+    raw["filament"]["name"] = "Basic"
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client([raw]))):
+        await _pass(scheduler, db_session)
+
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    assert notify.on_stock_reorder_alert.await_args.kwargs["color"] is None
+
+
+@pytest.mark.asyncio
+async def test_spoolman_mode_does_not_borrow_a_local_spools_history(db_session, scheduler, notify):
+    """A Spoolman id that matches a local spool id (left from before Spoolman was switched on)
+    must not pick up that spool's usage history. Here the local spool 1 has a history that would
+    make this a break; Spoolman's spool 1 is quiet by its own delta rate."""
+    from backend.app.models.spool_usage_history import SpoolUsageHistory
+
+    local = await _spool(db_session, used=300.0)
+    for days_ago in (3, 2, 1):
+        db_session.add(
+            SpoolUsageHistory(
+                spool_id=local.id, weight_used=300.0, created_at=utcnow_naive() - timedelta(days=days_ago)
+            )
+        )
+    db_session.add(Settings(key="spoolman_enabled", value="true"))
+    await _sku(db_session)
+    await db_session.commit()
+    client = _spoolman_client([_spoolman_spool(local.id, used=300.0)])
+
+    with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
+        await _pass(scheduler, db_session)
+
+    notify.on_stock_break_alert.assert_not_awaited()
+    notify.on_stock_reorder_alert.assert_not_awaited()
+
+
+# -- the service -------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_the_events_reach_a_provider_with_subtype_and_colour():
+    """Both events pass subtype and colour through to the template variables a provider is sent."""
+    from backend.app.services.notification_service import NotificationService
+
+    service = NotificationService()
+    provider = MagicMock()
+    provider.id = 1
+
+    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 = ("t", "b")
+
+        await service.on_stock_reorder_alert(
+            "PLA", "Bambu Lab", 80.0, 9.2, 8, AsyncMock(), subtype="Matte", color="Charcoal"
+        )
+        await service.on_stock_break_alert(
+            "PLA", "Bambu Lab", 40.0, 9.6, 4, 6, AsyncMock(), subtype="Matte", color="Charcoal"
+        )
+
+    assert [c.args[1] for c in mock_get.await_args_list] == ["on_stock_reorder_alert", "on_stock_break_alert"]
+    assert mock_send.await_count == 2
+    for call in mock_send.await_args_list:
+        variables = call.kwargs["variables"]
+        assert variables["subtype"] == "Matte"
+        assert variables["color"] == "Charcoal"
+        assert variables["material"] == "PLA"
+
+
+@pytest.mark.asyncio
+async def test_subtype_and_colour_are_optional_so_the_old_call_shape_still_works():
+    """Neither method had a caller, but the new arguments are keyword-only with defaults regardless."""
+    from backend.app.services.notification_service import NotificationService
+
+    service = NotificationService()
+    with (
+        patch.object(service, "_get_providers_for_event", new_callable=AsyncMock, return_value=[MagicMock()]),
+        patch.object(service, "_send_to_providers", new_callable=AsyncMock) as mock_send,
+        patch.object(service, "_build_message_from_template", new_callable=AsyncMock, return_value=("t", "b")),
+    ):
+        await service.on_stock_reorder_alert("PLA", None, 80.0, 9.2, 8, AsyncMock())
+
+    variables = mock_send.await_args.kwargs["variables"]
+    assert variables["subtype"] == ""
+    assert variables["color"] == ""
+
+
+@pytest.mark.asyncio
+async def test_a_spoolman_registered_date_with_a_trailing_z_still_gives_the_spool_an_age(db_session, scheduler, notify):
+    """Python 3.10's fromisoformat rejects a trailing "Z" (pyproject still allows 3.10), which
+    silently dropped the spool's age and with it the delta rate. Simulated here so the test
+    means the same on every interpreter."""
+
+    class _Py310Datetime(datetime):
+        @classmethod
+        def fromisoformat(cls, value):
+            if value.endswith("Z"):
+                raise ValueError(f"Invalid isoformat string: {value!r}")
+            return super().fromisoformat(value)
+
+    db_session.add(Settings(key="spoolman_enabled", value="true"))
+    await db_session.commit()
+    client = _spoolman_client([_spoolman_spool(7, used=REORDER_USED)])
+
+    with (
+        patch("backend.app.services.print_scheduler.datetime", _Py310Datetime),
+        patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)),
+    ):
+        await _pass(scheduler, db_session)
+
+    # 920 g in 100 days is 9.2 g/day: a reorder. With the age dropped the rate falls back to a
+    # different number and this SKU is not at its reorder point.
+    notify.on_stock_reorder_alert.assert_awaited_once()
+    assert notify.on_stock_reorder_alert.await_args.args[3] == pytest.approx(9.2, abs=0.05)

backend/tests/unit/services/test_stock_forecast_2955.py

@@ -0,0 +1,234 @@
+"""The backend forecast against ForecastPanel.tsx, on the same inputs (#2955).
+
+``stock_forecast`` is the panel's arithmetic moved out of the browser so the
+stock alerts have something to run on. The two exist side by side until the
+panel reads the backend's values, so these tests pin them: ``EXPECTED`` was
+produced by running the panel's own functions (``computeHistoryRate``,
+``computeDeltaRate`` and the block that turns them into a reorder point and the
+two alert flags, copied verbatim from ``ForecastPanel.tsx``) under Node with
+``Date.now()`` fixed to ``NOW`` and ``TZ=UTC``, on the scenarios below. A change
+to either side that moves a number shows up here.
+
+``TZ=UTC`` matters for exactly one thing: the panel parses a timezone-less
+``created_at`` as browser-local time, the backend reads it as UTC, and the delta
+rate divides by an age in days, so the two differ by the browser's offset. The
+database stores UTC, so UTC is what the timestamps mean.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from backend.app.services.stock_forecast import (
+    SkuSettings,
+    StockSpool,
+    UsageRecord,
+    delta_rate,
+    forecast_all,
+    forecast_sku,
+    history_rate,
+    sku_key,
+)
+
+NOW = datetime(2026, 9, 30, tzinfo=timezone.utc)
+
+
+def _ago(days: float) -> datetime:
+    """A naive UTC timestamp ``days`` before NOW, the shape the database returns."""
+    return (NOW - timedelta(days=days)).replace(tzinfo=None)
+
+
+def _spool(spool_id: int, used: float, age_days: float, baseline: float = 0) -> StockSpool:
+    return StockSpool(
+        id=spool_id,
+        material="PLA",
+        subtype="Basic",
+        brand="Bambu Lab",
+        color_name="Black",
+        label_weight=1000,
+        weight_used=used,
+        weight_used_baseline=baseline,
+        created_at=_ago(age_days),
+    )
+
+
+def _history(*events: tuple[int, float, float]) -> dict[int, list[UsageRecord]]:
+    """(spool id, days ago, grams) triples, grouped by spool the way the caller does."""
+    out: dict[int, list[UsageRecord]] = {}
+    for spool_id, ago, grams in events:
+        out.setdefault(spool_id, []).append(UsageRecord(created_at=_ago(ago), weight_used=grams))
+    return out
+
+
+# name -> (spools, history, settings, global lead time). The same scenarios, in the
+# same order, as the Node run that produced EXPECTED.
+SCENARIOS = {
+    # History rate, no settings row, lead time from the global setting.
+    "A": (
+        [_spool(1, 600, 60), _spool(2, 0, 10)],
+        _history((1, 40, 120), (1, 25, 90), (1, 12, 200), (1, 3, 150)),
+        None,
+        7,
+    ),
+    # No history: the delta rate, with the margin given in days.
+    "B": ([_spool(1, 300, 20)], {}, SkuSettings(lead_time_days=14, safety_margin_value=10), 0),
+    # A reset spool's history is left out; the clean spool's is kept.
+    "C": (
+        [_spool(1, 700, 50, baseline=200), _spool(2, 300, 30)],
+        _history((1, 20, 400), (1, 10, 50), (2, 20, 100), (2, 6, 100), (2, 1, 100)),
+        SkuSettings(lead_time_days=5),
+        0,
+    ),
+    # Stock break from the delta rate: 100 g left, 30 days of lead time.
+    "D": ([_spool(1, 900, 30)], {}, SkuSettings(lead_time_days=30), 0),
+    # Margin in grams.
+    "E": (
+        [_spool(1, 500, 90)],
+        _history((1, 30, 80), (1, 20, 60), (1, 10, 100)),
+        SkuSettings(lead_time_days=10, safety_margin_value=150, safety_margin_unit="g"),
+        0,
+    ),
+    # Nothing consumed: no rate, so no dates and no alerts.
+    "F": ([_spool(1, 0, 40)], {}, None, 7),
+    # Younger than a day: the delta rate is not measurable. The global lead time wins over the SKU's.
+    "G": ([_spool(1, 200, 0.5)], {}, SkuSettings(lead_time_days=3), 9),
+    # Under the reorder point with more than the lead time of stock left: a reorder.
+    "H": (
+        [_spool(1, 850, 100)],
+        _history((1, 50, 200), (1, 40, 200), (1, 20, 200)),
+        SkuSettings(lead_time_days=6, safety_margin_value=3),
+        0,
+    ),
+    # The same history with less left: a stock break, not also a reorder.
+    "I": (
+        [_spool(1, 930, 100)],
+        _history((1, 50, 200), (1, 40, 200), (1, 20, 200)),
+        SkuSettings(lead_time_days=6, safety_margin_value=3),
+        0,
+    ),
+    # Two spools of one SKU: the delta rate divides by the age of the oldest.
+    "J": ([_spool(1, 400, 400), _spool(2, 100, 5)], {}, SkuSettings(lead_time_days=10), 0),
+}
+
+EXPECTED = {
+    "A": {
+        "remaining_g": 1400,
+        "daily_rate_g": 13.57710201710711,
+        "effective_lead_time_days": 7,
+        "reorder_point_g": 304.32790456992615,
+        "days_remaining": 103,
+        "days_until_reorder_point": 80,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+    "B": {
+        "remaining_g": 700,
+        "daily_rate_g": 15,
+        "effective_lead_time_days": 14,
+        "reorder_point_g": 378.521204064531,
+        "days_remaining": 46,
+        "days_until_reorder_point": 21,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+    "C": {
+        "remaining_g": 1000,
+        "daily_rate_g": 13.942344991570476,
+        "effective_lead_time_days": 5,
+        "reorder_point_g": 288.583334452561,
+        "days_remaining": 71,
+        "days_until_reorder_point": 51,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+    "D": {
+        "remaining_g": 100,
+        "daily_rate_g": 30,
+        "effective_lead_time_days": 30,
+        "reorder_point_g": 1374.2245331930114,
+        "days_remaining": 3,
+        "days_until_reorder_point": -43,
+        "stock_break_alert": True,
+        "reorder_alert": False,
+    },
+    "E": {
+        "remaining_g": 500,
+        "daily_rate_g": 8.230026663902231,
+        "effective_lead_time_days": 10,
+        "reorder_point_g": 242.666532290416,
+        "days_remaining": 60,
+        "days_until_reorder_point": 31,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+    "F": {
+        "remaining_g": 1000,
+        "daily_rate_g": None,
+        "effective_lead_time_days": 7,
+        "reorder_point_g": 0,
+        "days_remaining": None,
+        "days_until_reorder_point": None,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+    "G": {
+        "remaining_g": 800,
+        "daily_rate_g": None,
+        "effective_lead_time_days": 9,
+        "reorder_point_g": 0,
+        "days_remaining": None,
+        "days_until_reorder_point": None,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+    "H": {
+        "remaining_g": 150,
+        "daily_rate_g": 13.864882095643093,
+        "effective_lead_time_days": 6,
+        "reorder_point_g": 144.46457585381498,
+        "days_remaining": 10,
+        "days_until_reorder_point": 0,
+        "stock_break_alert": False,
+        "reorder_alert": True,
+    },
+    "I": {
+        "remaining_g": 70,
+        "daily_rate_g": 13.864882095643093,
+        "effective_lead_time_days": 6,
+        "reorder_point_g": 144.46457585381498,
+        "days_remaining": 5,
+        "days_until_reorder_point": -6,
+        "stock_break_alert": True,
+        "reorder_alert": False,
+    },
+    "J": {
+        "remaining_g": 1500,
+        "daily_rate_g": 1.25,
+        "effective_lead_time_days": 10,
+        "reorder_point_g": 31.304439534819455,
+        "days_remaining": 1200,
+        "days_until_reorder_point": 1174,
+        "stock_break_alert": False,
+        "reorder_alert": False,
+    },
+}
+
+
+@pytest.mark.parametrize("name", sorted(SCENARIOS))
+def test_backend_matches_the_panels_numbers(name):
+    spools, history, settings, global_lead = SCENARIOS[name]
+    got = forecast_sku(spools, history, settings or SkuSettings(), global_lead, NOW)
+    expected = EXPECTED[name]
+
+    for field, want in expected.items():
+        have = getattr(got, field)
+        if isinstance(want, float):
+            assert have == pytest.approx(want, rel=1e-9), field
+        else:
+            assert have == want, field
+
+
+def test_the_scenarios_cover_both_alerts_and_neither():
+    """The pin means little if no scenario ever alerts."""
+    flags = {(e["reorder_alert"], e["stock_break_alert"]) for e in EXPECTED.values()}
+    assert flags == {(False, False), (True, False), (False, True)}

backend/tests/unit/test_stock_alert_template_migration_2955.py

@@ -0,0 +1,92 @@
+"""The stock alert templates name the colour and subtype (#2955).
+
+The forecast groups by colour, so two colours of one product would otherwise send
+the same message. The migration rewrites a template body IF AND ONLY IF it is
+still the shipped default -- an admin who edited the wording keeps it, the same
+guard as the ha_sensor_alert rename (#2824).
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from backend.app.core.database import _STOCK_ALERT_TEMPLATE_BODIES, _migrate_stock_alert_template_sku_variables
+from backend.app.models.notification_template import DEFAULT_TEMPLATES
+
+
+@pytest.fixture
+async def engine():
+    """In-memory SQLite with just the notification_templates table."""
+    from backend.app.models.notification_template import NotificationTemplate
+
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with engine.begin() as conn:
+        await conn.run_sync(NotificationTemplate.__table__.create)
+    try:
+        yield engine
+    finally:
+        await engine.dispose()
+
+
+async def _insert(conn, event_type: str, body: str) -> None:
+    await conn.execute(
+        text(
+            "INSERT INTO notification_templates (event_type, name, title_template, body_template, is_default) "
+            "VALUES (:et, 'n', 't', :b, 1)"
+        ),
+        {"et": event_type, "b": body},
+    )
+
+
+async def _body(conn, event_type: str) -> str:
+    return (
+        await conn.execute(
+            text("SELECT body_template FROM notification_templates WHERE event_type = :et"), {"et": event_type}
+        )
+    ).scalar_one()
+
+
+@pytest.mark.parametrize("event_type", sorted(_STOCK_ALERT_TEMPLATE_BODIES))
+async def test_rewrites_the_shipped_default(engine, event_type):
+    old, _new = _STOCK_ALERT_TEMPLATE_BODIES[event_type]
+    async with engine.begin() as conn:
+        await _insert(conn, event_type, old)
+
+    async with engine.begin() as conn:
+        await _migrate_stock_alert_template_sku_variables(conn)
+        body = await _body(conn, event_type)
+
+    assert "{subtype}" in body
+    assert "{color}" in body
+
+
+@pytest.mark.parametrize("event_type", sorted(_STOCK_ALERT_TEMPLATE_BODIES))
+async def test_leaves_an_edited_template_alone(engine, event_type):
+    async with engine.begin() as conn:
+        await _insert(conn, event_type, "Running low on {material}!")
+
+    async with engine.begin() as conn:
+        await _migrate_stock_alert_template_sku_variables(conn)
+        assert await _body(conn, event_type) == "Running low on {material}!"
+
+
+@pytest.mark.parametrize("event_type", sorted(_STOCK_ALERT_TEMPLATE_BODIES))
+async def test_the_rewrite_is_exactly_what_a_fresh_install_gets(engine, event_type):
+    """The migrated body and the DEFAULT_TEMPLATES body must not drift apart."""
+    default = next(t for t in DEFAULT_TEMPLATES if t["event_type"] == event_type)
+    assert _STOCK_ALERT_TEMPLATE_BODIES[event_type][1] == default["body_template"]
+
+
+async def test_running_it_twice_changes_nothing(engine):
+    old, new = _STOCK_ALERT_TEMPLATE_BODIES["stock_reorder_alert"]
+    async with engine.begin() as conn:
+        await _insert(conn, "stock_reorder_alert", old)
+
+    for _ in range(2):
+        async with engine.begin() as conn:
+            await _migrate_stock_alert_template_sku_variables(conn)
+
+    async with engine.begin() as conn:
+        assert await _body(conn, "stock_reorder_alert") == new