feat: the front door says whether ingestion is working (milestone 387 step B3)

The step phase A was building toward. A1 made the gated count true, A2
made it a durable state, A3 made it visible in Subscriptions — but
Subscriptions is where you go once you already suspect something. This
is the line that reaches someone who wasn't looking.

A thin grey strip above the feed, front door only: last check, sources
failing, sources you can't see. Only the actionable items take a
colour, and nothing renders at zero — a permanent "0 failing" trains
you to skip the line, which would hide the real number when it appears.

Two predicates, defined once. The ribbon counts and the surfaces it
links to have to agree on what "failing" and "no access" MEAN, or the
ribbon says 3 and the card shows 4. They live in db_helpers, which
exists for exactly this reason (its docstring: divergent copies are how
the race bugs crept in). Not in source_service, because
scheduler_service needs them too and source_service already imports
scheduler_service — the other direction is a cycle.

Counting deliberately spans all ENABLED sources rather than the
auto_check subset scheduler_status already walks: a source erroring on
a manual-only artist is still erroring. Disabled sources count for
nothing, which is what makes issue 1285 the real escape hatch for a sub
you stopped paying for.

Extends the existing schedule-status endpoint rather than adding a
parallel aggregate — the store already fetches it. Two scalar COUNTs.

The status filter is now URL-addressable, which it had to be for the
ribbon's links to land anywhere: a count that drops you on an
unfiltered list makes the reader redo the filtering the ribbon just
did. Mirrors how artistFilter already reads from route.query.

Front-door-only via a route prop, not a route.name check, so the view
doesn't need to know what it's mounted as and the router states the
intent in one place. Inside Browse's Posts tab you're looking FOR
something and the hub is one click away.

The fetch is swallowed on mount by design (rule 164): this is an aside,
and the feed must render whether or not the status call succeeds.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LNXXULQDjVZmbuNa2G9mD9

backend/app/services/db_helpers.py

@@ -20,6 +20,9 @@ from sqlalchemy import Select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from ..models import Source
+from .gallery_dl import ErrorType
+
 
 async def get_or_create[T](
     session: AsyncSession,
@@ -50,3 +53,31 @@ async def get_or_create[T](
     except IntegrityError:
         await sp.rollback()
         return (await session.execute(select_stmt)).scalar_one(), False
+
+
+# --- shared Source health predicates ----------------------------------------
+#
+# The subscriptions rollup, the front-door status ribbon and the list endpoint
+# all have to agree on what "failing" and "no access" MEAN, or the ribbon says
+# 3 and the card it links to shows 4. Same reasoning as get_or_create above:
+# divergent copies of one predicate are how the drift creeps in. Defined here
+# rather than in source_service because scheduler_service needs them too, and
+# source_service already imports scheduler_service (the other direction would
+# be a cycle).
+
+
+def failing_sources_clause():
+    """A source is FAILING when its runs are actually erroring.
+
+    Deliberately not `last_error IS NOT NULL` — a tier-limited source clears
+    last_error and keeps a chip, and must never be counted as broken.
+    """
+    return Source.consecutive_failures > 0
+
+
+def no_access_sources_clause():
+    """A source we can't see the content of: the walk works, the tier doesn't
+    grant it (#874 / milestone #387 phase A). Not a failure — kept separate
+    from failing_sources_clause on purpose, and the two are disjoint because
+    an informational class only ever rides an otherwise-OK run."""
+    return Source.error_type == ErrorType.TIER_LIMITED

backend/app/services/scheduler_service.py

@@ -8,12 +8,13 @@ from __future__ import annotations
 
 from datetime import UTC, datetime, timedelta
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from ..models import AppSetting, Artist, ImportSettings, Source
+from .db_helpers import failing_sources_clause, no_access_sources_clause
 
 MIN_INTERVAL_SECONDS = 60
 MAX_INTERVAL_SECONDS = 86400
@@ -219,10 +220,28 @@ async def scheduler_status(session: AsyncSession) -> dict:
 
     cooldowns = await active_platform_cooldowns(session)
 
+    # Ingestion health for the front-door ribbon (#387 B3). Counted over ENABLED
+    # sources rather than the auto_check subset walked above: a source that is
+    # erroring or paywalled is worth surfacing whether or not a schedule happens
+    # to poll it. Two scalar COUNTs, not a second pass over `rows`.
+    #
+    # Both predicates are the shared ones, so the ribbon and the surfaces it
+    # links to cannot disagree about what they are counting.
+    failing_sources = (await session.execute(
+        select(func.count()).select_from(Source)
+        .where(Source.enabled.is_(True), failing_sources_clause())
+    )).scalar_one()
+    no_access_sources = (await session.execute(
+        select(func.count()).select_from(Source)
+        .where(Source.enabled.is_(True), no_access_sources_clause())
+    )).scalar_one()
+
     return {
         "last_tick_at": last_tick_at,
         "next_due_at": next_due_at.isoformat() if next_due_at else None,
         "due_now": due_now,
         "auto_sources": len(rows),
+        "failing_sources": failing_sources,
+        "no_access_sources": no_access_sources,
         "platform_cooldowns": {p: dt.isoformat() for p, dt in cooldowns.items()},
     }

backend/app/services/source_service.py

@@ -17,6 +17,7 @@ from ..models import (
     Post,
     Source,
 )
+from .db_helpers import failing_sources_clause
 from .gallery_dl import ErrorType
 from .platforms import known_platform_keys
 from .scheduler_service import compute_next_check_at
@@ -250,7 +251,9 @@ class SourceService:
             stmt = stmt.where(~Source.url.like("sidecar:%"))
         if failing:
             # Worst-first so the rollup card surfaces the loudest failures.
-            stmt = stmt.where(Source.consecutive_failures > 0).order_by(
+            # Shared clause: the front-door ribbon counts with the same one, so
+            # it can never report a number this list then contradicts.
+            stmt = stmt.where(failing_sources_clause()).order_by(
                 Source.consecutive_failures.desc(), Artist.name.asc(),
             )
         else: