feat: apply fix/adapter-health-1-2-9 changes (unified retry, failure_class, RSS enabled)

- adapters/__init__.py: add http_get() unified retry helper + AdapterHTTPError
  + failure_class classification (429/5xx/4xx/error/zero_fetch/ok)
  + last_failure_class on SourceAdapter for pipeline capture
- pipeline.py: ENABLED_SOURCES now includes RSS
  + failure_class rollup in run_log (most severe across all adapters)
  + per-source failure_class in source_stats
- schema.sql: add failure_class column to run_log table

Backport of fix/adapter-health-1-2-9 branch (issues #1, #2, #9).

adapters/__init__.py

@@ -1,11 +1,18 @@
 """Source adapters for AI Research Oracle."""
 
+import urllib.request
+import urllib.error
+import time
 from abc import ABC, abstractmethod
 
 
 class SourceAdapter(ABC):
     """Base class for all ingestion adapters."""
 
+    def __init__(self):
+        # Set by http_get when a request fails (issue #2 classification)
+        self.last_failure_class = None
+
     @abstractmethod
     def name(self) -> str:
         """Source name: 'github', 'arxiv', 'reddit'."""
@@ -15,3 +22,81 @@ class SourceAdapter(ABC):
     def fetch(self, query: str = "", limit: int = 20) -> list[dict]:
         """Return entries matching DB schema fields."""
         pass
+
+
+class AdapterHTTPError(Exception):
+    """Raised by http_get on non-retryable or exhausted HTTP failures.
+
+    failure_class is one of: '429', '5xx', '4xx', 'error'.
+    Consumed by pipeline.py to populate run_log.failure_class (issue #2).
+    """
+
+    def __init__(self, status, failure_class, message=""):
+        super().__init__(message)
+        self.status = status
+        self.failure_class = failure_class
+
+
+def classify_http_status(code):
+    """Map an HTTP status code to a run_log failure_class."""
+    if code == 429:
+        return "429"
+    if 500 <= code < 600:
+        return "5xx"
+    return "4xx"  # 401/403/404 etc = client/config problem, not retried
+
+
+def http_get(url, headers=None, timeout=15, max_retries=2,
+             backoff_base=2, retry_403_ratelimit=False, return_headers=False,
+             owner=None):
+    """GET with a unified retry policy shared by all adapters (issue #1).
+
+    Retries ONLY on 429 and 5xx (transient). 4xx other than 429 are NOT
+    retried (config/client problems). Optional GitHub-style 403 rate-limit
+    handling via retry_403_ratelimit (waits for X-RateLimit-Reset header).
+
+    Returns response body bytes (or (body, headers) tuple if return_headers).
+    Raises AdapterHTTPError on non-retryable or exhausted failures, carrying
+    .failure_class for run_log classification. If owner is provided, sets
+    owner.last_failure_class so the pipeline can record it (issue #2).
+    """
+    def _fail(code, fclass, msg):
+        if owner is not None:
+            owner.last_failure_class = fclass
+        raise AdapterHTTPError(code, fclass, msg)
+
+    req = urllib.request.Request(url, headers=headers or {})
+    last_err = None
+    for attempt in range(max_retries + 1):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                body = resp.read()
+                if return_headers:
+                    return body, resp.headers
+                return body
+        except urllib.error.HTTPError as e:
+            last_err = e
+            code = e.code
+            if code == 429 or (code == 403 and retry_403_ratelimit):
+                if attempt < max_retries:
+                    if code == 403:
+                        reset = int(e.headers.get("X-RateLimit-Reset", 0))
+                        wait = (max(reset - int(time.time()), 0) + 1
+                                if reset else backoff_base * (attempt + 1) * 15)
+                        time.sleep(min(wait, 300))
+                    else:
+                        time.sleep(min(backoff_base * (attempt + 1), 30))
+                    continue
+                fclass = "429" if code == 429 else "4xx"
+                _fail(code, fclass, f"HTTP {code} for {url} (exhausted)")
+            if 500 <= code < 600:
+                if attempt < max_retries:
+                    time.sleep(min(backoff_base * (attempt + 1), 30))
+                    continue
+                _fail(code, "5xx", f"HTTP {code} for {url} (exhausted)")
+            # other 4xx (401/403 w/o flag/404) — not retried
+            _fail(code, classify_http_status(code), f"HTTP {code} for {url}")
+        except Exception as e:  # non-HTTP (timeout, DNS, conn reset)
+            _fail(None, "error", f"Request error: {e}")
+    # Defensive: loop should always raise or return
+    _fail(getattr(last_err, "code", None), "error", str(last_err))

pipeline.py

@@ -37,7 +37,7 @@ ADAPTERS = {
 }
 
 # Default enabled sources
-ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface"]
+ENABLED_SOURCES = ["github", "arxiv", "reddit", "hackernews", "huggingface", "rss"]
 
 
 def init_db(db_path: str, schema_path: str) -> sqlite3.Connection:
@@ -252,9 +252,14 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
             entries = adapter.fetch(limit=limit)
         except Exception as e:
             print(f"  ✗ {source_name} failed: {e}")
-            source_stats[source_name] = {"fetched": 0, "stored": 0, "error": str(e)}
+            source_stats[source_name] = {"fetched": 0, "stored": 0,
+                                         "error": str(e),
+                                         "failure_class": "error"}
             continue
 
+        # Capture classification from the adapter (set by http_get on failure)
+        fc = getattr(adapter, "last_failure_class", None)
+
         # Add adapter_version to metadata
         for entry in entries:
             meta = json.loads(entry["raw_metadata"]) if isinstance(entry["raw_metadata"], str) else entry["raw_metadata"]
@@ -262,7 +267,8 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
             entry["raw_metadata"] = json.dumps(meta)
 
         all_entries.extend(entries)
-        source_stats[source_name] = {"fetched": len(entries), "stored": 0}
+        source_stats[source_name] = {"fetched": len(entries), "stored": 0,
+                                     "failure_class": fc or "ok"}
         print(f"    Fetched: {len(entries)} entries")
 
         # Small spacing between sources
@@ -294,16 +300,32 @@ def run_pipeline(sources: list[str] | None = None, limit: int = 20, dry_run: boo
         # Zero-fetch (e.g. Reddit fully rate-limited) raises no exception but
         # is still a degraded run — record it so run_log can tell
         # "intermittent vs consistently-broken" apart over time.
-        zero = [s for s, st in source_stats.items() if st.get("fetched", 0) == 0 and not st.get("error")]
+        zero = [s for s, st in source_stats.items()
+                if st.get("fetched", 0) == 0 and not st.get("error")]
         notes_parts = [f"{s}: {st['error']}" for s, st in source_stats.items() if st.get("error")]
         if zero:
             notes_parts.append(f"no-fetch (degraded): {', '.join(zero)}")
         notes = "; ".join(notes_parts) or "all sources ok"
+
+        # Rollup failure_class (issue #2): most severe across sources.
+        # Priority: 5xx > 4xx > 429 > error > zero_fetch > ok
+        rank = {"5xx": 5, "4xx": 4, "429": 3, "error": 2, "zero_fetch": 1, "ok": 0}
+        classes = [st.get("failure_class", "ok") for st in source_stats.values()]
+        if any(c in ("5xx", "4xx", "429", "error") for c in classes):
+            run_fc = max((c for c in classes if c in rank),
+                         key=lambda c: rank[c])
+        elif zero:
+            run_fc = "zero_fetch"
+        else:
+            run_fc = "ok"
+
         try:
             conn.execute("""
-                INSERT INTO run_log (total_fetched, total_stored, sources_ok, sources_failed, notes)
+                INSERT INTO run_log (total_fetched, total_stored, sources_ok,
-                VALUES (?, ?, ?, ?, ?)
+                                     sources_failed, failure_class, notes)
-            """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed), notes))
+                VALUES (?, ?, ?, ?, ?, ?)
+            """, (len(all_entries), stored, json.dumps(ok), json.dumps(failed),
+                  run_fc, notes))
             conn.commit()
         except Exception as e:
             print(f"  ⚠ run_log write failed: {e}")

schema.sql

@@ -22,6 +22,8 @@ CREATE INDEX IF NOT EXISTS idx_entries_category ON entries(category_tags);
 -- Partial failures (e.g. Reddit rate-limited) are detectable here, not hidden
 -- as a "complete" run. Also enables future pruning decisions (entries older
 -- than N days with no re-fetch can be archived).
+-- failure_class (issue #2): one of 4xx / 5xx / 429 / zero_fetch / ok, derived
+-- from the real HTTP response via adapters.http_get, not guessed after the fact.
 CREATE TABLE IF NOT EXISTS run_log (
     id INTEGER PRIMARY KEY AUTOINCREMENT,
     run_time TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
@@ -29,6 +31,7 @@ CREATE TABLE IF NOT EXISTS run_log (
     total_stored INTEGER DEFAULT 0,
     sources_ok TEXT,       -- JSON list of sources that succeeded
     sources_failed TEXT,   -- JSON list of sources that errored/skipped
+    failure_class TEXT,    -- 4xx / 5xx / 429 / zero_fetch / ok
     notes TEXT
 );