Source management console: pause/resume, flag/clear, decision metrics

Turn the Sources tab into a real management console (per Codex):

* source_health now lists ALL sources (active + paused) with backing metrics:
  served / accepted_total / total_articles / duplicates + acceptance & duplicate
  rates + review_reason, alongside last success/attempt, next poll, failures.
* Admin endpoints (gated, 404 on missing): POST sources/{id}/active (pause/
  resume) and /review (flag/clear with reason).
* Pausing only stops future polling — the feed query has no active filter, so a
  paused source's accepted articles stay live.
* Frontend: metric table + Paused filter + per-row Pause/Resume & Flag/Clear
  (optimistic, revert on failure). Attention 'resting' now scoped to active.

Retire/Delete intentionally deferred (distinct lifecycle state, later).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

goodnews/api.py

@@ -381,6 +381,15 @@ class FeedbackReadBody(BaseModel):
     read: bool = True
 
 
+class SourceActiveBody(BaseModel):
+    active: bool = True
+
+
+class SourceReviewBody(BaseModel):
+    flag: bool = False
+    reason: str | None = None
+
+
 _FEEDBACK_CATEGORIES = {"idea", "concern", "bug", "praise", "other"}
 
 # The only event kinds we record. All aggregate, non-personal.
@@ -813,6 +822,31 @@ def create_app() -> FastAPI:
             conn.commit()
         return {"ok": True}
 
+    @app.post("/api/admin/sources/{sid}/active")
+    def admin_source_active(sid: int, body: SourceActiveBody, request: Request) -> dict:
+        # Pause/resume a feed. Pausing only stops future polling — its already-
+        # accepted articles stay live (the feed query doesn't filter on active).
+        with get_conn() as conn:
+            _require_admin(conn, request)
+            cur = conn.execute("UPDATE sources SET active = ? WHERE id = ?", (1 if body.active else 0, sid))
+            if cur.rowcount == 0:
+                raise HTTPException(status_code=404, detail="source not found")
+            conn.commit()
+        return {"ok": True, "active": body.active}
+
+    @app.post("/api/admin/sources/{sid}/review")
+    def admin_source_review(sid: int, body: SourceReviewBody, request: Request) -> dict:
+        with get_conn() as conn:
+            _require_admin(conn, request)
+            cur = conn.execute(
+                "UPDATE sources SET review_flag = ?, review_reason = ? WHERE id = ?",
+                (1 if body.flag else 0, (body.reason or None) if body.flag else None, sid),
+            )
+            if cur.rowcount == 0:
+                raise HTTPException(status_code=404, detail="source not found")
+            conn.commit()
+        return {"ok": True, "flag": body.flag}
+
     @app.get("/api/admin/stats")
     def admin_stats(request: Request) -> dict:
         with get_conn() as conn:

goodnews/queries.py

@@ -282,21 +282,28 @@ def content_stats(conn: sqlite3.Connection) -> dict:
 
 
 def source_health(conn: sqlite3.Connection) -> list[dict]:
-    """Per active source: failure streak, last error, accepted contribution, and
-    the computed next-poll time (so the backoff/'resting until' state is visible).
+    """Per source (active AND paused), the data an operator needs to manage feeds:
+    failure streak, last error/success/attempt, computed next poll, and the
+    backing metrics (served/accepted/total counts + acceptance & duplicate rates)
+    so Pause/Flag decisions aren't vibes-based.
 
     next_due_at = last attempt + MIN(cap, interval * (1 + consecutive_failures)),
-    mirroring feeds.due_source_rows; NULL last attempt means "due now".
+    mirroring feeds.due_source_rows; NULL last attempt means "due now". Paused
+    sources are included (their articles stay live; only polling stops).
     """
     rows = conn.execute(
         """
         SELECT
             s.id, s.name, s.default_category AS category, s.active,
-            s.consecutive_failures AS failures, s.review_flag,
+            s.consecutive_failures AS failures, s.review_flag, s.review_reason,
             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
              WHERE r.source_id = s.id AND r.finished_at IS NOT NULL) AS last_attempt,
+            (SELECT COUNT(*) FROM articles a WHERE a.source_id = s.id) AS total_articles,
+            (SELECT COUNT(*) FROM articles a JOIN article_scores sc ON sc.article_id = a.id
+             WHERE a.source_id = s.id AND sc.accepted = 1) AS accepted_total,
+            (SELECT COUNT(*) FROM articles a WHERE a.source_id = s.id AND a.duplicate_of IS NOT NULL) AS duplicates,
             (SELECT COUNT(*) FROM articles a JOIN article_scores sc ON sc.article_id = a.id
              WHERE a.source_id = s.id AND sc.accepted = 1 AND a.duplicate_of IS NULL) AS served,
             datetime(
@@ -305,12 +312,18 @@ def source_health(conn: sqlite3.Connection) -> list[dict]:
                 '+' || MIN(?, s.poll_interval_minutes * (1 + s.consecutive_failures)) || ' minutes'
             ) AS next_due_at
         FROM sources s
-        WHERE s.active = 1
-        ORDER BY s.consecutive_failures DESC, served DESC, s.name
+        ORDER BY s.active DESC, s.consecutive_failures DESC, served DESC, s.name
         """,
         (MAX_BACKOFF_MINUTES,),
     ).fetchall()
-    return [dict(r) for r in rows]
+    out = []
+    for r in rows:
+        d = dict(r)
+        total = d["total_articles"] or 0
+        d["acceptance_rate"] = round(100 * d["accepted_total"] / total) if total else None
+        d["duplicate_rate"] = round(100 * d["duplicates"] / total) if total else None
+        out.append(d)
+    return out
 
 
 def _attention(content: dict, sources: list[dict], feedback_unread: int) -> list[dict]:
@@ -320,7 +333,7 @@ def _attention(content: dict, sources: list[dict], feedback_unread: int) -> list
     items: list[dict] = []
     n = lambda c: "" if c == 1 else "s"  # noqa: E731 — tiny pluralizer
 
-    resting = [s for s in sources if (s.get("failures") or 0) > 0]
+    resting = [s for s in sources if s.get("active") and (s.get("failures") or 0) > 0]
     if resting:
         items.append({"level": "warn", "text": f"{len(resting)} source{n(len(resting))} backing off after failures"})
     flagged = [s for s in sources if s.get("review_flag")]