Sources: per-source paywall override (3-state) — fix domain-rule mis-flags

The Articles inspector revealed paywall is domain-coarse: nytimes.com is flagged,
so NY Times Learning's free Word-of-the-Day inherits 🔒 — and that flag isn't
cosmetic, it deprioritizes the content in feed sort + lead selection. Add a
per-source override so admins can correct it after inspecting.

- sources.paywall_override: NULL (domain rule) | 'free' | 'paywalled'.
- paywall.py: keep low-level is_paywalled(url) (domain); add is_paywalled_for_source
  (url, override) for the EFFECTIVE decision — never patched the domain helper
  globally (per Codex), so "domain says X" stays distinguishable from "overridden".
- Threaded everywhere ranking/UI touches paywall, via src.paywall_override on the
  shared _ARTICLE_COLUMNS + the source-aware helper: feed sort, /api/since, replace,
  lead selection, Article badge, brief composition (briefs.py), digest, source_health
  (table 🔒), the Articles inspector, and the review/attention check — so ranking and
  UI always agree.
- Endpoint POST /api/admin/sources/{id}/paywall {override}; admin UI: a select in the
  inspector header (Use domain rule / Treat as free / Treat as paywalled) + the basis
  ("ON (domain)" / "OFF (override)"), optimistic so the panel stays open.

Test: domain rule → paywalled in table+inspector+feed badge; 'free' → off in all
three; validation 422 + 404. 242 pytest + 11 vitest.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

goodnews/queries.py

@@ -11,7 +11,7 @@ import sqlite3
 from datetime import UTC, datetime, timedelta
 
 from .feeds import MAX_BACKOFF_MINUTES
-from .paywall import is_paywalled
+from .paywall import is_paywalled, is_paywalled_for_source
 
 # UA substrings that mark automated clients. Crawlers run JS on a throttled
 # budget and trip the boot-failure beacon routinely — without this filter they
@@ -53,6 +53,7 @@ _ARTICLE_COLUMNS = f"""
     s.reason_code,
     s.reason_text,
     s.model_name,
+    src.paywall_override AS paywall_override,
     (SELECT group_concat(t.tag) FROM article_tags t WHERE t.article_id = a.id) AS tags,
     {RANK_SCORE_SQL} AS rank_score
 """
@@ -335,7 +336,7 @@ def source_health(conn: sqlite3.Connection) -> list[dict]:
         SELECT
             s.id, s.name, s.feed_url, s.homepage_url, s.default_category AS category, s.active,
             s.status, s.content_visible, s.retry_after_at,
-            s.consecutive_failures AS failures, s.review_flag, s.review_reason,
+            s.consecutive_failures AS failures, s.review_flag, s.review_reason, s.paywall_override,
             s.poll_interval_minutes AS interval_minutes,
             s.last_success_at, s.last_error_at, substr(s.last_error, 1, 160) AS last_error,
             (SELECT MAX(r.finished_at) FROM ingest_runs r
@@ -370,8 +371,8 @@ def source_health(conn: sqlite3.Connection) -> list[dict]:
         # duplicate of content already served (accepted_total − served = accepted dupes).
         d["accepted_dup_rate"] = round(100 * (accepted - d["served"]) / accepted) if accepted else None
         d["image_coverage"] = round(100 * (d["images"] or 0) / d["served"]) if d["served"] else None
-        # Paywall is a domain-level hint, so it's a per-source flag (not a rate).
-        d["paywalled"] = is_paywalled(d.get("homepage_url") or d.get("feed_url"))
+        # Paywall is a domain-level hint + a per-source override; show the EFFECTIVE flag.
+        d["paywalled"] = is_paywalled_for_source(d.get("homepage_url") or d.get("feed_url"), d.get("paywall_override"))
         # Match the REAL scheduler gate: due = the later of the streak-backoff time
         # and any retry_after_at rest (UTC strings sort chronologically).
         due_times = [t for t in (d["next_due_at"], d["retry_after_at"]) if t]
@@ -468,6 +469,8 @@ def source_articles(conn: sqlite3.Connection, source_id: int, filter: str = "all
                     limit: int = 25, offset: int = 0) -> list[dict]:
     """The actual ingested articles for a source, newest first — so admins can
     verify the metric (paywall/image/acceptance) against real evidence."""
+    ov = conn.execute("SELECT paywall_override FROM sources WHERE id = ?", (source_id,)).fetchone()
+    override = ov["paywall_override"] if ov else None
     where = _SRC_ART_FILTERS.get(filter, "")
     rows = conn.execute(
         f"""
@@ -492,7 +495,7 @@ def source_articles(conn: sqlite3.Connection, source_id: int, filter: str = "all
             "reason": r["reason_text"] or r["reason_code"],  # the "why" behind accept/reject
             "topic": r["topic"],
             "flavor": r["flavor"],
-            "paywalled": is_paywalled(r["canonical_url"]),    # domain rule — same for the source
+            "paywalled": is_paywalled_for_source(r["canonical_url"], override),  # effective (domain rule + override)
             "has_image": bool(r["image_url"]),
             "duplicate": r["duplicate_of"] is not None,
         }
@@ -515,11 +518,15 @@ def source_articles_summary(conn: sqlite3.Connection, source_id: int) -> dict:
         """,
         (source_id,),
     ).fetchone()
-    one = conn.execute("SELECT canonical_url FROM articles WHERE source_id = ? LIMIT 1", (source_id,)).fetchone()
+    srow = conn.execute("SELECT homepage_url, feed_url, paywall_override FROM sources WHERE id = ?", (source_id,)).fetchone()
+    override = srow["paywall_override"] if srow else None
+    url = (srow["homepage_url"] or srow["feed_url"]) if srow else None
     return {
         "total": agg["total"], "accepted": agg["accepted"], "rejected": agg["rejected"],
         "no_image": agg["no_image"], "duplicates": agg["duplicates"],
-        "paywalled": is_paywalled(one["canonical_url"]) if one else False,
+        "paywalled": is_paywalled_for_source(url, override),  # effective
+        "paywall_domain": is_paywalled(url),                  # what the domain rule alone says
+        "paywall_override": override,                         # null | 'free' | 'paywalled' — the basis
     }