feat: a source stops pulling once its membership ends, and resumes on resubscribe (3995)

Operator, 2026-09-13: "if I kill a subscription on patreon I would like the pulling to stop on curator as well", with auto-resume chosen. This reverses the 2026-09-11 "report only" decision for lapsed sources.

membership_reconcile.apply_membership_lapses runs in sync_memberships right after each platform's successful sync, so it only ever acts on the roster just written.

It stops a source (enabled=false, with the same failure-state reset as a manual disable, #1285) only when all of these hold:
- the roster is fresh
- the source's matched membership says has_paid_access is False (lapsed, or a free follow)
- the paid-through date has passed, where the platform gives one (Patreon's member.access_expires_at; SubscribeStar gives none, so it stops at once)
- the source is enabled
- the operator hasn't chosen to keep it

It never acts on absence. A source with no matched membership keeps pulling, because a rename or a never-walked source produces the same absence. An unrecognised status is never a lapse either.

It resumes only sources carrying its own `_membership_stopped` marker, once the membership is paid again.

The operator outranks the sweep both ways (SourceService.update):
- turning a stopped source back on marks it `_membership_kept`, so the next sweep leaves it alone until it's paid again
- turning a source off by hand drops the marker, so the sweep never switches it back on

Both are `_`-prefixed app-managed config keys, which operator edits already preserve. No migration.

The roster/fetch line holds. This is a source-level action by the sweep. No download path reads the roster, and the scheduler still selects on `enabled` alone. test_no_fetch_path_can_read_the_roster is unchanged.

UI: SourceRow shows a neutral "Membership ended" chip, with the status and the resume/keep explanation, ahead of the other chips. The sweep's task summary reports stopped/resumed counts.

Tests (tests/test_membership_lapses.py):
- a lapse stops the source with a clean slate and keeps the id cache
- paid-through is honoured
- absence, an unknown status and a stale roster never stop anything
- a resume touches only what the sweep stopped
- a manual on sticks, a manual off drops the marker, and a kept source is released once paid

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SHQB1YukL3VyvMK8rcbmV9

backend/app/services/membership_reconcile.py

@@ -8,9 +8,10 @@ trustworthy enough to act on.
 1. `subscribed_not_tracked` — you pay for this and FC does not follow it. The
    adoption win, and the only bucket carrying an action.
 2. `tracked_not_subscribed` — FC follows this and the roster does not show you
-   paying for it. REPORT ONLY, by the operator's decision (2026-09-11): it says
-   what it sees and links to the existing Subscriptions row, and offers no
-   one-click disable.
+   paying for it. No longer shown on the card: the operator reversed the
+   2026-09-11 "report only" call on 2026-09-13. The lapsed half of it now ACTS,
+   in `apply_membership_lapses` below (#3995). The absent half still only
+   reports, because absence proves nothing.
 3. `matched` — the healthy set. Counted, not listed loudly.
 4. `unidentified` — sources this join cannot speak to at all. Reported as
    exactly that, because the alternative is filing them under a verdict.
@@ -39,7 +40,7 @@ rendered as lapsed. That is the whole reason it returns a tri-state.
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import UTC, datetime
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -233,3 +234,126 @@ async def reconcile_all(session: AsyncSession, now: datetime | None = None) -> d
             await reconcile(session, platform=p, now=now) for p in sorted(platforms)
         ]
     }
+
+
+# ---------------------------------------------------------------------------
+# Stop pulling what the account no longer pays for (#3995)
+# ---------------------------------------------------------------------------
+#
+# Operator decision, 2026-09-13, reversing the 2026-09-11 "report only" call
+# for this direction: "if I kill a subscription on patreon I would like the
+# pulling to stop on curator as well", with automatic resume on resubscribing.
+#
+# This is a SOURCE-level action taken by the daily sweep, visible on the source
+# row and reversible there. It is not a fetch-path decision. The line C5 draws,
+# that the roster never decides a POST is inaccessible, still holds: nothing
+# here reads per-post access, and no download path reads the roster
+# (`test_no_fetch_path_can_read_the_roster`). The scheduler keeps selecting on
+# `enabled` alone.
+#
+# Acts ONLY on positive evidence. A source whose matched membership says access
+# has ended is stopped. A source with NO matched membership is left alone,
+# because absence has innocent causes: a creator rename, a source never walked
+# so no id is cached, a membership the platform stopped listing. Stopping on
+# absence would switch off things the operator still pays for.
+#
+# Two app-managed config_overrides keys carry the state. The `_` prefix is
+# already the "FC writes this, an operator edit preserves it" family.
+#   _membership_stopped  set when the sweep stops a source; the sweep resumes
+#                        ONLY sources carrying it, so a source the operator
+#                        switched off by hand is never switched back on
+#   _membership_kept     set by SourceService.update when the operator turns a
+#                        stopped source back ON: a deliberate choice to keep
+#                        pulling a lapsed creator, which the next sweep must
+#                        not undo. Cleared when the membership is paid again.
+STOPPED_KEY = "_membership_stopped"
+KEPT_KEY = "_membership_kept"
+
+
+def _access_expires_at(m: PlatformMembership) -> datetime | None:
+    """When paid access actually ends, if the platform says.
+
+    Patreon keeps a cancelled membership's access until the end of the billing
+    period and reports that date (`member.access_expires_at`, note #3992).
+    SubscribeStar's page gives no such date, so a cancelled SubscribeStar
+    membership stops at once. Returns None when there is no usable date.
+    """
+    details = m.details or {}
+    raw = details.get("access_expires_at") or (details.get("member") or {}).get("access_expires_at")
+    if not isinstance(raw, str) or not raw:
+        return None
+    try:
+        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
+
+
+async def apply_membership_lapses(
+    session: AsyncSession, *, platform: str, now: datetime | None = None,
+) -> dict:
+    """Stop sources whose paid access has ended; resume the ones this stopped.
+
+    Refuses to act on a roster that isn't fresh, for the same reason C4 refuses
+    to draw conclusions from one.
+    """
+    now = now or datetime.now(UTC)
+    state = await get_sync_state(session, platform)
+    if not roster_is_fresh(state, now=now):
+        return {"platform": platform, "skipped": "roster not fresh", "stopped": 0, "resumed": 0}
+
+    memberships = (await session.execute(
+        select(PlatformMembership).where(PlatformMembership.platform == platform)
+    )).scalars().all()
+    sources = (await session.execute(
+        select(Source).where(Source.platform == platform)
+    )).scalars().all()
+    pairs = pair_sources_with_memberships(list(sources), list(memberships))
+
+    stopped: list[int] = []
+    resumed: list[int] = []
+    for source in sources:
+        pair = pairs.get(source.id)
+        if pair is None:
+            continue  # absence is never acted on, see above
+        m, _kind = pair
+        paid = has_paid_access(
+            m.platform, m.status,
+            is_free_member=bool((m.details or {}).get("is_free_member")),
+        )
+        co = dict(source.config_overrides or {})
+
+        if paid is True:
+            changed = co.pop(KEPT_KEY, None) is not None
+            if STOPPED_KEY in co:
+                co.pop(STOPPED_KEY)
+                source.enabled = True
+                resumed.append(source.id)
+                changed = True
+            if changed:
+                source.config_overrides = co
+            continue
+
+        # Unknown status: never a reason to stop something (has_paid_access's
+        # tri-state exists for exactly this).
+        if paid is None:
+            continue
+        if not source.enabled or co.get(KEPT_KEY):
+            continue
+        expires = _access_expires_at(m)
+        if expires is not None and expires > now:
+            continue  # still inside the paid-through period
+
+        co[STOPPED_KEY] = {"at": now.isoformat(), "status": m.status}
+        source.config_overrides = co
+        source.enabled = False
+        # The same clean slate a manual disable gives (SourceService.update,
+        # #1285), so a stopped source doesn't linger as failing or gated.
+        source.last_error = None
+        source.error_type = None
+        source.consecutive_failures = 0
+        stopped.append(source.id)
+
+    await session.commit()
+    return {"platform": platform, "stopped": len(stopped), "resumed": len(resumed)}
+

backend/app/services/source_service.py

@@ -19,6 +19,7 @@ from ..models import (
 )
 from .db_helpers import failing_sources_clause
 from .gallery_dl import ErrorType
+from .membership_reconcile import KEPT_KEY, STOPPED_KEY
 from .membership_roster import gated_reasons_for_sources
 from .platforms import known_platform_keys
 from .scheduler_service import compute_next_check_at
@@ -171,6 +172,25 @@ def arm_backfill(source: Source) -> None:
     source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
 
 
+def _record_manual_enable_choice(source: Source, *, enabled: bool) -> None:
+    """Keep the membership sweep (#3995) from overriding the operator.
+
+    Turning a source the sweep STOPPED back on is a deliberate choice to keep
+    pulling a lapsed creator, so it is marked kept and the next sweep leaves it
+    alone. Turning a source off by hand drops any sweep marker, so the sweep
+    never switches back on something the operator switched off themselves.
+    """
+    co = dict(source.config_overrides or {})
+    if enabled and STOPPED_KEY in co:
+        co.pop(STOPPED_KEY)
+        co[KEPT_KEY] = True
+    elif not enabled:
+        co.pop(STOPPED_KEY, None)
+    else:
+        return
+    source.config_overrides = co
+
+
 class SourceService:
     def __init__(self, session: AsyncSession):
         self.session = session
@@ -428,6 +448,9 @@ class SourceService:
         for key, value in fields.items():
             setattr(source, key, value)
 
+        if "enabled" in fields:
+            _record_manual_enable_choice(source, enabled=bool(fields["enabled"]))
+
         if url_changed:
             # Repointing a source at a different creator makes a cached campaign
             # id WRONG, not merely stale, and `patreon_resolver` consults that

backend/app/tasks/maintenance.py

@@ -1247,6 +1247,7 @@ def sync_memberships() -> str:
     from ..services.artist_membership_service import rescan as membership_rescan
     from ..services.credential_crypto import CredentialCrypto
     from ..services.credential_service import CredentialService
+    from ..services.membership_reconcile import apply_membership_lapses
     from ..services.membership_roster import roster_user_id, sync_platform
     from ..services.patreon_client import PatreonClient
     from ..services.subscribestar_client import SubscribeStarClient
@@ -1298,9 +1299,18 @@ def sync_memberships() -> str:
                     )
 
                 async with async_factory() as session:
-                    results.append(
-                        await sync_platform(session, platform=platform, fetch=fetch)
-                    )
+                    result = await sync_platform(session, platform=platform, fetch=fetch)
+                results.append(result)
+
+                # #3995: stop pulling sources whose paid access has ended, and
+                # resume the ones this stopped once they are paid again. Only
+                # right after a successful sync, so it always acts on the roster
+                # just written, never on a stale one.
+                if result.get("ok"):
+                    async with async_factory() as session:
+                        result["lapses"] = await apply_membership_lapses(
+                            session, platform=platform,
+                        )
 
             # #388 E4: offer the freshly-synced roster to the artists FC already
             # tracks. Chained here rather than given its own beat entry because
@@ -1321,7 +1331,12 @@ def sync_memberships() -> str:
         if "skipped" in r:
             parts.append(f"{r['platform']}=skipped({r['skipped']})")
         elif r.get("ok"):
-            parts.append(f"{r['platform']}={r['count']}")
+            lapses = r.get("lapses") or {}
+            detail = (
+                f"(stopped={lapses['stopped']},resumed={lapses['resumed']})"
+                if lapses.get("stopped") or lapses.get("resumed") else ""
+            )
+            parts.append(f"{r['platform']}={r['count']}{detail}")
         else:
             parts.append(f"{r['platform']}=FAILED({r['error']})")
     if res.get("suggested") is not None: