fix: a stopped source is not a failing one, and cannot be deep-scanned (4279)

Ebi77 sat in the "1 source is failing" banner for six days with no action
available, reading `stranded by recovery sweep (no terminal status after
time_limit)`. Four things lined up:

1. The membership sweep did its job — saw `former_patron`, disabled the
   source, cleared its failure state. Clean at 02:50.
2. Twenty minutes later a deep scan was armed on it. `/backfill` had a
   credential pre-flight but NO `enabled` guard, while `/check` has carried
   one all along. The two trigger endpoints disagreed, and the ungated one is
   the one that arms the long walk.
3. Without a membership the walk cannot finish, never reaches a terminal
   status, and the recovery sweep strands it with consecutive_failures = 1.
4. Nothing could clear that. A disabled source is never scheduled, so no
   successful run resets the count; `SourceService.update` clears only on an
   explicit disable and it was already disabled; and the banner's Retry routes
   to `/check`, which refuses a disabled source. The card offered a button
   structurally incapable of acting on the only source it was showing.

`failing_sources_clause()` now means "enabled AND erroring". That also settles
a disagreement its two callers already had: the scheduler's count paired it
with `enabled.is_(True)` and `SourceService.list(failing=True)` did not, so
one counted Ebi77 and the other did not — exactly the drift the note above
that function warns about, which is why the test belongs IN the predicate
rather than beside it. The scheduler's now-duplicate clause is dropped so one
place decides.

`/backfill` gains the guard for start/recover/recapture. `stop` stays open on
a disabled source, or arming becomes a one-way door.

Migration 0101 clears failure state on sources that are already disabled — the
predicate fixes what the surfaces report, not what the rows carry, and the
rows are why the operator had no way out (lesson #4202). It matches what
`update` already does on an explicit disable, so rows disabled by any other
path come into line. Enabled sources are untouched: a real failure on a live
source must keep showing, which the second new test pins.

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

alembic/versions/0101_clear_failure_state_on_disabled_sources.py

@@ -0,0 +1,66 @@
+"""Clear failure state on sources that are disabled (#4279).
+
+`failing_sources_clause()` now means "enabled AND erroring", so a disabled
+source no longer counts as failing. That fixes what the surfaces REPORT; it
+does not touch what the rows already CARRY, and the rows are the reason the
+operator saw a banner for six days with no way to act on it (lesson #4202 —
+a guard does not undo the value already stored).
+
+## The row this exists for
+
+Ebi77 (source 19): the membership sweep stopped it as `former_patron` at
+02:50 on 2026-09-15 and correctly cleared its failure state. A deep scan was
+armed twenty minutes later — `/backfill` had no `enabled` guard, which this
+release also fixes — and could not complete without access, so the recovery
+sweep stranded it:
+
+    consecutive_failures = 1
+    last_error = "stranded by recovery sweep (no terminal status after time_limit)"
+
+Nothing could clear that. A disabled source is never scheduled, so no
+successful run resets the counter; `SourceService.update` clears failure
+state only on an explicit disable, and the source was already disabled; and
+the card's Retry routes to `/check`, which refuses a disabled source.
+
+## Why every disabled source, not just that one
+
+The clear matches what `SourceService.update` already does when a source is
+disabled through the app — "disable the subs you're not paying for without
+them lingering as failing" — so this brings rows disabled by any OTHER path
+(the membership sweep, a retired platform in 0097) into line with the rows
+disabled by hand. Same shape as 0097: a repair migration reaches the live
+instance on deploy rather than waiting for someone to find the row.
+
+Enabled sources are untouched — a real failure on a live source must keep
+showing.
+
+Revision ID: 0101
+Revises: 0100
+Create Date: 2026-09-21
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0101"
+down_revision: Union[str, None] = "0100"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "UPDATE source SET last_error = NULL, error_type = NULL, "
+        "consecutive_failures = 0 "
+        "WHERE NOT enabled "
+        "AND (last_error IS NOT NULL OR error_type IS NOT NULL "
+        "     OR consecutive_failures <> 0)"
+    )
+
+
+def downgrade() -> None:
+    # Irreversible by design: the cleared strings and counts are not recorded
+    # anywhere, and restoring a failure state nobody can act on would only
+    # re-create the banner this removes. Rule #22 owes no story backwards.
+    pass

backend/app/api/sources.py

@@ -201,6 +201,16 @@ async def set_backfill(source_id: int):
             rec = await SourceService(session).get(source_id)
             if rec is None:
                 return _bad("not_found", status=404)
+            # A disabled source must not be armable for a deep walk — the same
+            # rule /check has carried all along (see `source_disabled` below).
+            # Arming one anyway is how #4279 happened: the membership sweep had
+            # stopped Ebi77 as `former_patron`, a deep scan was armed twenty
+            # minutes later, the walk could not complete without access, and
+            # the recovery sweep stranded it with a failure count no surface
+            # could clear — a disabled source is never scheduled again, and
+            # Retry routes to /check, which refuses it.
+            if not rec.enabled:
+                return _bad("source_disabled", detail="enable the source first")
             native = uses_native_ingester(rec.platform)
             if native:
                 cred = CredentialService(session, _get_crypto())

backend/app/services/db_helpers.py

@@ -16,7 +16,7 @@ from __future__ import annotations
 
 from collections.abc import Awaitable, Callable
 
-from sqlalchemy import Select
+from sqlalchemy import Select, and_
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -67,12 +67,26 @@ async def get_or_create[T](
 
 
 def failing_sources_clause():
-    """A source is FAILING when its runs are actually erroring.
+    """A source is FAILING when it is ENABLED and its runs are 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.
+
+    The `enabled` half was folded in 2026-09-21 (#4279). A disabled source is
+    one FC deliberately stopped — most often because the membership sweep saw
+    `former_patron` — and "stopped because you no longer subscribe" is not
+    "failing". Worse, it is a failure nobody can clear: a disabled source is
+    never scheduled, so no successful run ever resets the counter, and the
+    card's Retry button routes to `/check`, which refuses a disabled source
+    outright. Ebi77 sat in the banner for six days with no action available.
+
+    This also settles a disagreement the two callers already had. The
+    scheduler's status count paired this clause with `enabled.is_(True)`;
+    `SourceService.list(failing=True)` did not. One counted Ebi77, the other
+    did not — the exact drift the note above this function warns about, which
+    is why the `enabled` test belongs IN the predicate rather than beside it.
     """
-    return Source.consecutive_failures > 0
+    return and_(Source.enabled.is_(True), Source.consecutive_failures > 0)
 
 
 def no_access_sources_clause():

backend/app/services/scheduler_service.py

@@ -229,7 +229,7 @@ async def scheduler_status(session: AsyncSession) -> dict:
     # 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())
+        .where(failing_sources_clause())
     )).scalar_one()
     no_access_sources = (await session.execute(
         select(func.count()).select_from(Source)

tests/test_api_sources.py

@@ -274,6 +274,47 @@ async def test_backfill_endpoint_start_and_stop(client, artist, db):
     assert (await stopped.get_json())["backfill_state"] is None
 
 
+@pytest.mark.asyncio
+async def test_backfill_endpoint_refuses_a_disabled_source(client, artist, db):
+    """#4279: a source FC deliberately stopped must not be armable for a deep
+    walk. Arming one is how Ebi77 got a failure nobody could clear — the walk
+    cannot complete without access, the recovery sweep strands it, and a
+    disabled source is never scheduled again to reset the count."""
+    src = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/alice-stopped", enabled=False,
+    )
+    db.add(src)
+    await db.commit()
+
+    for action in ("start", "recover", "recapture"):
+        resp = await client.post(
+            f"/api/sources/{src.id}/backfill", json={"action": action},
+        )
+        assert resp.status_code == 400, action
+        assert (await resp.get_json())["error"] == "source_disabled"
+
+
+@pytest.mark.asyncio
+async def test_backfill_stop_still_works_on_a_disabled_source(client, artist, db):
+    """Only the ARMING actions are gated. Cancelling a walk on a source that
+    was disabled mid-backfill must stay available, or the arm becomes a
+    one-way door."""
+    src = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/alice-stopping", enabled=False,
+        config_overrides={"_backfill_state": "running"},
+    )
+    db.add(src)
+    await db.commit()
+
+    resp = await client.post(
+        f"/api/sources/{src.id}/backfill", json={"action": "stop"},
+    )
+    assert resp.status_code == 200
+    assert (await resp.get_json())["backfill_state"] is None
+
+
 @pytest.mark.asyncio
 async def test_backfill_endpoint_defaults_to_start(client, artist, db):
     src = Source(

tests/test_source_service.py

@@ -682,3 +682,83 @@ async def test_new_disabled_source_skips_backfill(db):
         enabled=False,
     )
     assert rec.backfill_runs_remaining == 0
+
+
+@pytest.mark.asyncio
+async def test_a_disabled_source_is_not_failing(db):
+    """#4279: "stopped because you no longer subscribe" is not "failing".
+
+    Ebi77 was stopped by the membership sweep as `former_patron`, then a deep
+    scan armed on it got stranded by the recovery sweep. The banner showed it
+    for six days with no action available: a disabled source is never
+    scheduled (so no run clears the count), `update` only clears on an
+    explicit disable (it was already disabled), and Retry routes to /check,
+    which refuses a disabled source.
+    """
+    artist = await _artist(db)
+    svc = SourceService(db)
+    rec = await svc.create(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/stopped",
+    )
+    source = (await db.execute(
+        select(Source).where(Source.id == rec.id)
+    )).scalar_one()
+    source.enabled = False
+    source.consecutive_failures = 1
+    source.last_error = "stranded by recovery sweep (no terminal status after time_limit)"
+    await db.commit()
+
+    assert [r.id for r in await svc.list(failing=True)] == []
+
+
+@pytest.mark.asyncio
+async def test_an_enabled_source_that_errors_is_still_failing(db):
+    """The other half — folding `enabled` in must not hide a real failure on
+    a live source."""
+    artist = await _artist(db)
+    svc = SourceService(db)
+    rec = await svc.create(
+        artist_id=artist.id, platform="patreon", url="https://patreon.com/live",
+    )
+    source = (await db.execute(
+        select(Source).where(Source.id == rec.id)
+    )).scalar_one()
+    source.enabled = True
+    source.consecutive_failures = 2
+    source.last_error = "auth failed"
+    await db.commit()
+
+    assert [r.id for r in await svc.list(failing=True)] == [rec.id]
+
+
+@pytest.mark.asyncio
+async def test_the_failing_list_and_the_ribbon_count_agree(db):
+    """The two callers of `failing_sources_clause` disagreed before #4279:
+    the scheduler's count paired it with `enabled.is_(True)`, the list did
+    not, so one counted Ebi77 and the other did not. The predicate owns the
+    whole definition now — assert the two surfaces match rather than trusting
+    that they were both updated."""
+    from backend.app.services.scheduler_service import scheduler_status
+
+    artist = await _artist(db)
+    svc = SourceService(db)
+    for url, enabled, fails in (
+        ("https://patreon.com/one", True, 3),
+        ("https://patreon.com/two", False, 1),
+        ("https://patreon.com/three", True, 0),
+    ):
+        rec = await svc.create(
+            artist_id=artist.id, platform="patreon", url=url,
+        )
+        s = (await db.execute(
+            select(Source).where(Source.id == rec.id)
+        )).scalar_one()
+        s.enabled = enabled
+        s.consecutive_failures = fails
+    await db.commit()
+
+    listed = len(await svc.list(failing=True))
+    counts = await scheduler_status(db)
+    assert listed == 1
+    assert counts["failing_sources"] == listed