feat(sponsor-prompt): in-app toast at earned milestones

ghcr.io pull baseline (~10k/day rising → ~8-12k active installs) puts
      sponsor conversion at 0.08% — roughly an order of magnitude under
      industry-benchmark for OSS with visible CTA. The Settings banner from
      0d4b9d4e gives passive every-visit visibility on one page; this adds
      opt-out-able active visibility at moments where the user has just
      earned something with Bambuddy.

      Five trigger families with a 14-day cross-family cooldown: prints
      (100/500/1000/2500/5000), cost (100/500/1000 tracked filament +
      energy), archives (50/250/1000), anniversary (1 year), version-update
      (re-armable on each major bump). New sponsor_toast_state table with
      nullable user_id so auth-disabled installs get the same trigger logic
      through one code path (NULL-keyed install-default row).

backend/app/api/routes/sponsor_prompt.py

@@ -0,0 +1,55 @@
+"""API routes for the in-app sponsor toast."""
+
+import logging
+
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.auth import RequirePermissionIfAuthEnabled
+from backend.app.core.database import get_db
+from backend.app.core.permissions import Permission
+from backend.app.models.user import User
+from backend.app.schemas.sponsor_prompt import (
+    SponsorPromptCheckResponse,
+    SponsorPromptDismissRequest,
+)
+from backend.app.services import sponsor_prompt as service
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/sponsor-prompt", tags=["sponsor-prompt"])
+
+
+def _user_id(current_user: User | None) -> int | None:
+    return current_user.id if current_user is not None else None
+
+
+@router.get("/check", response_model=SponsorPromptCheckResponse)
+async def check_sponsor_prompt(
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    db: AsyncSession = Depends(get_db),
+):
+    """Return the next eligible sponsor-toast trigger, or `{show: false}`."""
+    trigger = await service.evaluate(db, _user_id(current_user))
+    await db.commit()
+    if trigger is None:
+        return SponsorPromptCheckResponse(show=False)
+    return SponsorPromptCheckResponse(
+        show=True,
+        milestone=trigger.milestone,
+        family=trigger.family,
+        threshold=trigger.threshold,
+        payload=trigger.payload,
+    )
+
+
+@router.post("/dismiss", status_code=status.HTTP_204_NO_CONTENT)
+async def dismiss_sponsor_prompt(
+    data: SponsorPromptDismissRequest,
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    db: AsyncSession = Depends(get_db),
+):
+    """Anchor the 14-day cooldown and record the milestone as shown."""
+    await service.dismiss(db, _user_id(current_user), data.milestone)
+    await db.commit()
+    return None

backend/app/main.py

@@ -57,6 +57,7 @@ from backend.app.api.routes import (
     slice_jobs,
     slicer_presets,
     smart_plugs,
+    sponsor_prompt,
     spoolbuddy,
     spoolman,
     spoolman_inventory,
@@ -6525,6 +6526,7 @@ app.include_router(user_notifications.router, prefix=app_settings.api_prefix)
 app.include_router(spoolman.router, prefix=app_settings.api_prefix)
 app.include_router(spoolman_inventory.router, prefix=app_settings.api_prefix)
 app.include_router(updates.router, prefix=app_settings.api_prefix)
+app.include_router(sponsor_prompt.router, prefix=app_settings.api_prefix)
 app.include_router(maintenance.router, prefix=app_settings.api_prefix)
 app.include_router(camera.router, prefix=app_settings.api_prefix)
 app.include_router(external_links.router, prefix=app_settings.api_prefix)

backend/app/models/__init__.py

@@ -25,6 +25,7 @@ from backend.app.models.project import Project
 from backend.app.models.settings import Settings
 from backend.app.models.smart_plug import SmartPlug
 from backend.app.models.smart_plug_energy_snapshot import SmartPlugEnergySnapshot
+from backend.app.models.sponsor_toast_state import SponsorToastState
 from backend.app.models.spool import Spool
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spool_catalog import SpoolCatalogEntry
@@ -75,6 +76,7 @@ __all__ = [
     "SpoolUsageHistory",
     "ColorCatalogEntry",
     "SpoolBuddyDevice",
+    "SponsorToastState",
     "UserEmailPreference",
     "UserOTPCode",
     "UserTOTP",

backend/app/models/sponsor_toast_state.py

@@ -0,0 +1,39 @@
+"""Per-user (or install-default) state for the sponsor-prompt toast.
+
+A single row stores which sponsor-toast milestones have already fired for a
+given user, when the most recent toast was shown (for the 14-day cooldown),
+and the app version last seen so we can fire the "version-update" trigger
+exactly once per major bump.
+
+``user_id`` is nullable: in auth-disabled installs (no user concept), the
+service stores everything against a single NULL-keyed row.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class SponsorToastState(Base):
+    __tablename__ = "sponsor_toast_state"
+    __table_args__ = (UniqueConstraint("user_id", name="uq_sponsor_toast_state_user_id"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int | None] = mapped_column(
+        Integer,
+        ForeignKey("users.id", ondelete="CASCADE"),
+        nullable=True,
+        index=True,
+    )
+    last_shown_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    # JSON-serialised list[str] of milestone keys already fired (e.g. ["prints-100", "cost-100"]).
+    # Stored as Text for SQLite/Postgres uniformity; the service serialises with json.dumps.
+    milestones_seen: Mapped[str] = mapped_column(Text, nullable=False, default="[]")
+    last_seen_version: Mapped[str | None] = mapped_column(String(50), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())

backend/app/schemas/sponsor_prompt.py

@@ -0,0 +1,23 @@
+"""Pydantic schemas for the sponsor-prompt API."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class SponsorPromptCheckResponse(BaseModel):
+    """Result of GET /sponsor-prompt/check."""
+
+    show: bool
+    milestone: str | None = None
+    family: str | None = None
+    threshold: int | None = None
+    payload: dict[str, Any] = Field(default_factory=dict)
+
+
+class SponsorPromptDismissRequest(BaseModel):
+    """Body of POST /sponsor-prompt/dismiss."""
+
+    milestone: str

backend/app/services/sponsor_prompt.py

@@ -0,0 +1,256 @@
+"""Sponsor-prompt trigger evaluator and dismiss handler.
+
+Drives the in-app "support keeps Bambuddy independent" toast. Trigger families
+fire at milestones the user has earned (prints, archives, filament cost,
+anniversary) plus a soft version-update nudge after a major upgrade.
+
+A 14-day cooldown applies across ALL families: if any toast fired in the last
+14 days, no new toast fires. Each individual milestone is shown at most once
+per user (or once per install in auth-disabled mode); version-update is the
+exception — it re-arms every time the running version is newer than the one
+last acknowledged.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.config import APP_VERSION
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.sponsor_toast_state import SponsorToastState
+from backend.app.models.user import User
+
+logger = logging.getLogger(__name__)
+
+COOLDOWN_DAYS = 14
+
+PRINT_MILESTONES = (100, 500, 1000, 2500, 5000)
+COST_MILESTONES = (100, 500, 1000)
+ARCHIVE_MILESTONES = (50, 250, 1000)
+ANNIVERSARY_YEARS = 1
+
+
+@dataclass
+class Trigger:
+    """Evaluated trigger result returned to the frontend."""
+
+    milestone: str  # e.g. "prints-500", "anniversary-1", "version-update"
+    family: str  # "prints" | "cost" | "archives" | "anniversary" | "version-update"
+    threshold: int | None = None
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# State helpers
+# ---------------------------------------------------------------------------
+
+
+async def _get_or_create_state(db: AsyncSession, user_id: int | None) -> SponsorToastState:
+    """Fetch the state row for this user (or the install-default NULL row).
+
+    Creates the row lazily on first access so the migration doesn't need to
+    seed anything.
+    """
+    if user_id is None:
+        stmt = select(SponsorToastState).where(SponsorToastState.user_id.is_(None))
+    else:
+        stmt = select(SponsorToastState).where(SponsorToastState.user_id == user_id)
+    result = await db.execute(stmt)
+    state = result.scalar_one_or_none()
+    if state is None:
+        state = SponsorToastState(user_id=user_id, milestones_seen="[]")
+        db.add(state)
+        await db.flush()
+    return state
+
+
+def _within_cooldown(state: SponsorToastState) -> bool:
+    if state.last_shown_at is None:
+        return False
+    cutoff = datetime.now(timezone.utc) - timedelta(days=COOLDOWN_DAYS)
+    last = state.last_shown_at
+    if last.tzinfo is None:
+        last = last.replace(tzinfo=timezone.utc)
+    return last >= cutoff
+
+
+def _seen_milestones(state: SponsorToastState) -> set[str]:
+    try:
+        raw = json.loads(state.milestones_seen or "[]")
+        return set(raw) if isinstance(raw, list) else set()
+    except (json.JSONDecodeError, TypeError):
+        logger.warning(
+            "sponsor_toast_state.milestones_seen for user=%s was not valid JSON; resetting",
+            state.user_id,
+        )
+        return set()
+
+
+# ---------------------------------------------------------------------------
+# Per-family checks
+# ---------------------------------------------------------------------------
+
+
+def _user_filter(column, user_id: int | None):
+    return column.is_(None) if user_id is None else column == user_id
+
+
+async def _check_anniversary(
+    db: AsyncSession, user_id: int | None, seen: set[str], _state: SponsorToastState
+) -> Trigger | None:
+    milestone = f"anniversary-{ANNIVERSARY_YEARS}"
+    if milestone in seen:
+        return None
+    if user_id is None:
+        # Install-anchor = earliest users.created_at (the first admin row).
+        result = await db.execute(select(func.min(User.created_at)))
+        anchor = result.scalar()
+    else:
+        result = await db.execute(select(User.created_at).where(User.id == user_id))
+        anchor = result.scalar()
+    if anchor is None:
+        return None
+    if anchor.tzinfo is None:
+        anchor = anchor.replace(tzinfo=timezone.utc)
+    if datetime.now(timezone.utc) - anchor < timedelta(days=365 * ANNIVERSARY_YEARS):
+        return None
+    return Trigger(milestone=milestone, family="anniversary")
+
+
+async def _check_prints(
+    db: AsyncSession, user_id: int | None, seen: set[str], _state: SponsorToastState
+) -> Trigger | None:
+    stmt = (
+        select(func.count())
+        .select_from(PrintLogEntry)
+        .where(
+            PrintLogEntry.status == "completed",
+            _user_filter(PrintLogEntry.created_by_id, user_id),
+        )
+    )
+    completed = (await db.execute(stmt)).scalar() or 0
+    # Pick the LARGEST milestone the user has crossed but not yet seen.
+    for threshold in sorted(PRINT_MILESTONES, reverse=True):
+        key = f"prints-{threshold}"
+        if completed >= threshold and key not in seen:
+            return Trigger(
+                milestone=key,
+                family="prints",
+                threshold=threshold,
+                payload={"count": completed},
+            )
+    return None
+
+
+async def _check_archives(
+    db: AsyncSession, user_id: int | None, seen: set[str], _state: SponsorToastState
+) -> Trigger | None:
+    stmt = select(func.count()).select_from(PrintArchive).where(_user_filter(PrintArchive.created_by_id, user_id))
+    archived = (await db.execute(stmt)).scalar() or 0
+    for threshold in sorted(ARCHIVE_MILESTONES, reverse=True):
+        key = f"archives-{threshold}"
+        if archived >= threshold and key not in seen:
+            return Trigger(
+                milestone=key,
+                family="archives",
+                threshold=threshold,
+                payload={"count": archived},
+            )
+    return None
+
+
+async def _check_cost(
+    db: AsyncSession, user_id: int | None, seen: set[str], _state: SponsorToastState
+) -> Trigger | None:
+    stmt = (
+        select(func.coalesce(func.sum(PrintLogEntry.cost), 0) + func.coalesce(func.sum(PrintLogEntry.energy_cost), 0))
+        .select_from(PrintLogEntry)
+        .where(_user_filter(PrintLogEntry.created_by_id, user_id))
+    )
+    total = float((await db.execute(stmt)).scalar() or 0)
+    for threshold in sorted(COST_MILESTONES, reverse=True):
+        key = f"cost-{threshold}"
+        if total >= threshold and key not in seen:
+            return Trigger(
+                milestone=key,
+                family="cost",
+                threshold=threshold,
+                payload={"total": round(total, 2)},
+            )
+    return None
+
+
+async def _check_version_update(
+    _db: AsyncSession, _user_id: int | None, _seen: set[str], state: SponsorToastState
+) -> Trigger | None:
+    # version-update is NOT in milestones_seen — it has its own state column
+    # so it can re-fire on each major bump.
+    if not APP_VERSION:
+        return None
+    last = state.last_seen_version
+    if last is None:
+        # First-ever read; treat as already-acknowledged so we don't toast
+        # immediately on a brand-new install. Persist current version silently.
+        state.last_seen_version = APP_VERSION
+        return None
+    if last == APP_VERSION:
+        return None
+    return Trigger(
+        milestone="version-update",
+        family="version-update",
+        payload={"from": last, "to": APP_VERSION},
+    )
+
+
+# Priority order: most emotional / earned first; version-update is the soft fallback.
+_CHECKS = (
+    _check_anniversary,
+    _check_prints,
+    _check_archives,
+    _check_cost,
+    _check_version_update,
+)
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+async def evaluate(db: AsyncSession, user_id: int | None) -> Trigger | None:
+    """Return the next eligible sponsor-toast trigger, or None."""
+    state = await _get_or_create_state(db, user_id)
+    if _within_cooldown(state):
+        return None
+    seen = _seen_milestones(state)
+    for check in _CHECKS:
+        trigger = await check(db, user_id, seen, state)
+        if trigger is not None:
+            return trigger
+    # No triggers eligible — still commit any in-progress state changes
+    # (e.g. version-update's first-touch persistence).
+    await db.flush()
+    return None
+
+
+async def dismiss(db: AsyncSession, user_id: int | None, milestone: str) -> None:
+    """Mark a milestone as shown (sets cooldown anchor + records seen)."""
+    state = await _get_or_create_state(db, user_id)
+    if milestone == "version-update":
+        # Re-armable: just update last_seen_version, don't add to seen-list.
+        state.last_seen_version = APP_VERSION
+    else:
+        seen = _seen_milestones(state)
+        if milestone not in seen:
+            seen.add(milestone)
+            state.milestones_seen = json.dumps(sorted(seen))
+    state.last_shown_at = datetime.now(timezone.utc)
+    await db.flush()

backend/tests/conftest.py

@@ -139,6 +139,7 @@ async def test_engine():
         slot_preset,
         smart_plug,
         smart_plug_energy_snapshot,  # noqa: F401
+        sponsor_toast_state,  # noqa: F401
         spool,
         spool_assignment,
         spool_catalog,

backend/tests/integration/test_sponsor_prompt_api.py

@@ -0,0 +1,48 @@
+"""Integration tests for /sponsor-prompt routes."""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+
+
+class TestSponsorPromptAPI:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_check_returns_show_false_for_empty_install(self, async_client: AsyncClient):
+        response = await async_client.get("/api/v1/sponsor-prompt/check")
+        assert response.status_code == 200
+        body = response.json()
+        assert body["show"] is False
+        assert body.get("milestone") is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_dismiss_requires_milestone(self, async_client: AsyncClient):
+        response = await async_client.post("/api/v1/sponsor-prompt/dismiss", json={})
+        # Pydantic missing-field → 422.
+        assert response.status_code == 422
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_dismiss_returns_204(self, async_client: AsyncClient):
+        response = await async_client.post(
+            "/api/v1/sponsor-prompt/dismiss",
+            json={"milestone": "version-update"},
+        )
+        assert response.status_code == 204
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_check_then_dismiss_then_recheck_is_silent(self, async_client: AsyncClient):
+        """End-to-end: even if no trigger is currently eligible, dismissing
+        anchors the cooldown, so a subsequent check stays {show: false}."""
+        first = await async_client.get("/api/v1/sponsor-prompt/check")
+        assert first.json()["show"] is False
+        dismiss = await async_client.post(
+            "/api/v1/sponsor-prompt/dismiss",
+            json={"milestone": "version-update"},
+        )
+        assert dismiss.status_code == 204
+        second = await async_client.get("/api/v1/sponsor-prompt/check")
+        assert second.json()["show"] is False

backend/tests/unit/test_sponsor_prompt_service.py

@@ -0,0 +1,327 @@
+"""Unit tests for the sponsor-prompt trigger evaluator."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.sponsor_toast_state import SponsorToastState
+from backend.app.models.user import User
+from backend.app.services import sponsor_prompt as service
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+async def _make_user(db: AsyncSession, *, username: str = "alice", created_days_ago: int = 0) -> User:
+    user = User(username=username, role="admin")
+    db.add(user)
+    await db.flush()
+    if created_days_ago:
+        user.created_at = datetime.now(timezone.utc) - timedelta(days=created_days_ago)
+        await db.flush()
+    return user
+
+
+async def _add_completed_prints(db: AsyncSession, *, user_id: int | None, count: int, cost_each: float = 0.0) -> None:
+    for _ in range(count):
+        db.add(
+            PrintLogEntry(
+                status="completed",
+                created_by_id=user_id,
+                cost=cost_each if cost_each else None,
+            )
+        )
+    await db.flush()
+
+
+async def _add_archives(db: AsyncSession, *, user_id: int | None, count: int) -> None:
+    for i in range(count):
+        db.add(
+            PrintArchive(
+                filename=f"archive-{i}.zip",
+                file_path=f"/tmp/archive-{i}.zip",
+                file_size=1024,
+                created_by_id=user_id,
+            )
+        )
+    await db.flush()
+
+
+# ---------------------------------------------------------------------------
+# Empty / no-eligibility cases
+# ---------------------------------------------------------------------------
+
+
+class TestEmptyState:
+    @pytest.mark.asyncio
+    async def test_evaluate_returns_none_for_fresh_user(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is None
+
+    @pytest.mark.asyncio
+    async def test_state_row_is_created_lazily(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await service.evaluate(db_session, user.id)
+        from sqlalchemy import select
+
+        row = (
+            await db_session.execute(select(SponsorToastState).where(SponsorToastState.user_id == user.id))
+        ).scalar_one_or_none()
+        assert row is not None
+        assert row.milestones_seen == "[]"
+
+
+# ---------------------------------------------------------------------------
+# Cooldown
+# ---------------------------------------------------------------------------
+
+
+class TestCooldown:
+    @pytest.mark.asyncio
+    async def test_no_toast_within_14d_window(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=200)
+        # Pre-populate state with a recent last_shown_at
+        state = SponsorToastState(
+            user_id=user.id,
+            last_shown_at=datetime.now(timezone.utc) - timedelta(days=3),
+        )
+        db_session.add(state)
+        await db_session.flush()
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is None
+
+    @pytest.mark.asyncio
+    async def test_toast_eligible_after_14d_window(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=200)
+        state = SponsorToastState(
+            user_id=user.id,
+            last_shown_at=datetime.now(timezone.utc) - timedelta(days=15),
+        )
+        db_session.add(state)
+        await db_session.flush()
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.family == "prints"
+
+
+# ---------------------------------------------------------------------------
+# Per-family triggers
+# ---------------------------------------------------------------------------
+
+
+class TestPrintMilestones:
+    @pytest.mark.asyncio
+    async def test_fires_at_100(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=100)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.milestone == "prints-100"
+        assert trigger.threshold == 100
+
+    @pytest.mark.asyncio
+    async def test_picks_highest_unseen_milestone(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=600)
+        trigger = await service.evaluate(db_session, user.id)
+        # 500 is the highest crossed milestone (1000 not reached).
+        assert trigger is not None
+        assert trigger.milestone == "prints-500"
+
+    @pytest.mark.asyncio
+    async def test_skips_already_seen(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=600)
+        # Mark prints-500 as already seen — but NOT prints-100.
+        # Service should fall through to the next-largest unseen, which is prints-100.
+        state = SponsorToastState(
+            user_id=user.id,
+            milestones_seen=json.dumps(["prints-500"]),
+        )
+        db_session.add(state)
+        await db_session.flush()
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.milestone == "prints-100"
+
+    @pytest.mark.asyncio
+    async def test_failed_prints_dont_count(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=50)
+        for _ in range(60):
+            db_session.add(PrintLogEntry(status="failed", created_by_id=user.id))
+        await db_session.flush()
+        trigger = await service.evaluate(db_session, user.id)
+        # Only 50 completed → below 100 threshold → no print trigger.
+        # Anniversary not reached either; no other counter populated.
+        assert trigger is None
+
+
+class TestArchiveMilestones:
+    @pytest.mark.asyncio
+    async def test_fires_at_50(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_archives(db_session, user_id=user.id, count=50)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.milestone == "archives-50"
+
+
+class TestCostMilestones:
+    @pytest.mark.asyncio
+    async def test_fires_when_cost_sum_crosses_100(self, db_session: AsyncSession):
+        # Prints with cost = ~3.5 each, 30 prints → 105.
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=30, cost_each=3.5)
+        # 30 < 100 prints, so prints-100 not eligible. cost = 105 ≥ 100 → fires.
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.family == "cost"
+        assert trigger.milestone == "cost-100"
+
+
+class TestAnniversary:
+    @pytest.mark.asyncio
+    async def test_fires_after_1_year(self, db_session: AsyncSession):
+        user = await _make_user(db_session, created_days_ago=370)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.milestone == "anniversary-1"
+        assert trigger.family == "anniversary"
+
+    @pytest.mark.asyncio
+    async def test_does_not_fire_before_1_year(self, db_session: AsyncSession):
+        user = await _make_user(db_session, created_days_ago=300)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is None
+
+
+class TestVersionUpdate:
+    @pytest.mark.asyncio
+    async def test_first_read_silently_anchors(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        with patch.object(service, "APP_VERSION", "0.3.0"):
+            trigger = await service.evaluate(db_session, user.id)
+        assert trigger is None
+        from sqlalchemy import select
+
+        state = (
+            await db_session.execute(select(SponsorToastState).where(SponsorToastState.user_id == user.id))
+        ).scalar_one()
+        assert state.last_seen_version == "0.3.0"
+
+    @pytest.mark.asyncio
+    async def test_fires_on_version_bump(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        state = SponsorToastState(user_id=user.id, last_seen_version="0.2.0")
+        db_session.add(state)
+        await db_session.flush()
+        with patch.object(service, "APP_VERSION", "0.3.0"):
+            trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.milestone == "version-update"
+        assert trigger.payload == {"from": "0.2.0", "to": "0.3.0"}
+
+
+# ---------------------------------------------------------------------------
+# Priority order
+# ---------------------------------------------------------------------------
+
+
+class TestPriorityOrder:
+    @pytest.mark.asyncio
+    async def test_anniversary_beats_prints(self, db_session: AsyncSession):
+        # User old enough for anniversary AND with 100+ prints.
+        user = await _make_user(db_session, created_days_ago=400)
+        await _add_completed_prints(db_session, user_id=user.id, count=200)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.family == "anniversary"
+
+    @pytest.mark.asyncio
+    async def test_prints_beats_archives(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=200)
+        await _add_archives(db_session, user_id=user.id, count=100)
+        trigger = await service.evaluate(db_session, user.id)
+        assert trigger is not None
+        assert trigger.family == "prints"
+
+
+# ---------------------------------------------------------------------------
+# Dismiss
+# ---------------------------------------------------------------------------
+
+
+class TestDismiss:
+    @pytest.mark.asyncio
+    async def test_dismiss_adds_to_seen_and_anchors_cooldown(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        await _add_completed_prints(db_session, user_id=user.id, count=100)
+        await service.evaluate(db_session, user.id)
+        await service.dismiss(db_session, user.id, "prints-100")
+        from sqlalchemy import select
+
+        state = (
+            await db_session.execute(select(SponsorToastState).where(SponsorToastState.user_id == user.id))
+        ).scalar_one()
+        assert "prints-100" in json.loads(state.milestones_seen)
+        assert state.last_shown_at is not None
+        # Re-evaluation must now return None (cooldown).
+        next_trigger = await service.evaluate(db_session, user.id)
+        assert next_trigger is None
+
+    @pytest.mark.asyncio
+    async def test_version_update_dismiss_updates_version_not_seen_list(self, db_session: AsyncSession):
+        user = await _make_user(db_session)
+        state = SponsorToastState(user_id=user.id, last_seen_version="0.2.0")
+        db_session.add(state)
+        await db_session.flush()
+        with patch.object(service, "APP_VERSION", "0.3.0"):
+            await service.dismiss(db_session, user.id, "version-update")
+        from sqlalchemy import select
+
+        state = (
+            await db_session.execute(select(SponsorToastState).where(SponsorToastState.user_id == user.id))
+        ).scalar_one()
+        assert state.last_seen_version == "0.3.0"
+        assert json.loads(state.milestones_seen) == []
+
+
+# ---------------------------------------------------------------------------
+# Auth-disabled (user_id = None) — NULL-keyed install-default row
+# ---------------------------------------------------------------------------
+
+
+class TestAuthDisabledMode:
+    @pytest.mark.asyncio
+    async def test_uses_install_anchor_for_anniversary(self, db_session: AsyncSession):
+        # In auth-disabled mode, anniversary anchor = MIN(users.created_at).
+        # Seed a user from >1 year ago.
+        await _make_user(db_session, username="root", created_days_ago=400)
+        # Prints written without created_by_id.
+        await _add_completed_prints(db_session, user_id=None, count=10)
+        trigger = await service.evaluate(db_session, None)
+        assert trigger is not None
+        assert trigger.family == "anniversary"
+
+    @pytest.mark.asyncio
+    async def test_null_keyed_counters_isolated_from_per_user(self, db_session: AsyncSession):
+        # A user-attributed prints set should NOT show up in the install-default count.
+        user = await _make_user(db_session, username="alice")
+        await _add_completed_prints(db_session, user_id=user.id, count=200)
+        # NULL-keyed install has zero prints.
+        trigger = await service.evaluate(db_session, None)
+        # No anniversary either (user only just created).
+        assert trigger is None