fix(adapters): shared retry helper + run_log failure_class + enable RSS (issues #1 #2 #9)

- adapters/__init__.py: add http_get() unified retry (429/5xx only, max 2
  attempts, capped exp backoff) + AdapterHTTPError carrying failure_class;
  SourceAdapter.last_failure_class set on failure for pipeline capture.
- arxiv/github/huggingface/hackernews/reddit: route HTTP through http_get.
  Preserves GitHub 403 rate-limit retry and Reddit 403/429 fast-bail.
- schema.sql + pipeline.py: add run_log.failure_class column; rollup most-
  severe class across sources (5xx>4xx>429>error>zero_fetch>ok).
- pipeline.py: ENABLE RSS in ENABLED_SOURCES (was registered, disabled).
- RSS smoke test surfaced 3 broken feeds (anthropic 404, googleai 404,
  metaai 301) — left as-is, captured in feed_failures; URL fix is separate
  discovery task, not guessed.

Verified: full dry-run fetches all 6 sources; github live fetch OK;
Reddit 429 fast-bail preserved; no import/syntax errors.

adapters/__init__.py

@@ -1,39 +1,18 @@
 """Source adapters for AI Research Oracle."""
 
-import json
+import urllib.request
-import os
+import urllib.error
+import time
 from abc import ABC, abstractmethod
 
-# Centralized curation config (issue #7). One file, per-adapter blocks.
-# Stdlib-only (JSON, not YAML) to honor Athena's dependency-free runtime.
-_QUERIES_PATH = os.path.join(
-    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
-    "config", "queries.json")
-
-
-def load_queries():
-    """Load config/queries.json. Returns {'sources': {...}}.
-
-    Safe fallback: if the file is missing/corrupt, returns an empty
-    {'sources': {}} so adapters fall back to their class defaults
-    (constructor None-override) instead of crashing the pipeline.
-    """
-    try:
-        with open(_QUERIES_PATH) as f:
-            data = json.load(f)
-        return data if isinstance(data, dict) else {"sources": {}}
-    except Exception:
-        return {"sources": {}}
-
-
-def source_config(name: str) -> dict:
-    """Return the per-adapter block for `name`, or {} if absent."""
-    return load_queries().get("sources", {}).get(name, {}) or {}
-
 
 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'."""
@@ -43,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))

adapters/arxiv.py

@@ -32,7 +32,7 @@ import xml.etree.ElementTree as ET
 from datetime import datetime, timedelta, timezone
 from html import unescape
 
-from adapters import SourceAdapter, source_config
+from adapters import SourceAdapter, http_get, AdapterHTTPError
 
 # arXiv API
 ARXIV_API = "http://export.arxiv.org/api/query"
@@ -45,8 +45,12 @@ class ArxivAdapter(SourceAdapter):
     DEFAULT_CATEGORIES = ["cs.AI", "cs.LG", "cs.CL"]
 
     def __init__(self, categories=None, rate_limit=3):
-        cfg = source_config("arxiv")
+        """
-        self.categories = categories or cfg.get("categories") or self.DEFAULT_CATEGORIES
+        Args:
+            categories: List of arXiv categories. Default: cs.AI, cs.LG, cs.CL
+            rate_limit: Seconds between API calls (default 3).
+        """
+        self.categories = categories or self.DEFAULT_CATEGORIES
         self.rate_limit = rate_limit
 
     def name(self) -> str:
@@ -147,14 +151,12 @@ class ArxivAdapter(SourceAdapter):
             f"&max_results={max_results}"
         )
 
-        req = urllib.request.Request(url, headers={"User-Agent": "ai-oracle/0.1"})
-
         try:
-            with urllib.request.urlopen(req, timeout=30) as resp:
+            raw = http_get(url, headers={"User-Agent": "ai-oracle/0.1"},
-                xml_data = resp.read().decode("utf-8")
+                           timeout=30, max_retries=2, owner=self)
-            return self._parse_atom(xml_data)
+            return self._parse_atom(raw.decode("utf-8"))
-        except urllib.error.HTTPError as e:
+        except AdapterHTTPError as e:
-            print(f"    HTTP {e.code} for arXiv query")
+            print(f"    {e.failure_class}: arXiv query ({e})")
             return []
         except Exception as e:
             print(f"    arXiv request error: {e}")

adapters/github.py

@@ -17,7 +17,7 @@ import urllib.error
 import urllib.parse
 from datetime import datetime, timedelta, timezone
 
-from adapters import SourceAdapter, source_config
+from adapters import SourceAdapter, http_get, AdapterHTTPError
 
 
 class GitHubAdapter(SourceAdapter):
@@ -29,11 +29,6 @@ class GitHubAdapter(SourceAdapter):
         """Initialize with optional read-only token (5000 req/hr vs 60)."""
         self.token = token or os.environ.get("GITHUB_TOKEN", "")
         self.cache = {}
-        # Curation centralized (issue #7): trending queries from config
-        cfg = source_config("github")
-        self.search_terms = cfg.get("search_terms") or [
-            "ai agent", "llm OR inference OR rag", "autonomous agent OR AI tool",
-        ]
 
     def name(self) -> str:
         return "github"
@@ -48,40 +43,27 @@ class GitHubAdapter(SourceAdapter):
         return headers
 
     def _request(self, url: str, max_retries: int = 2) -> dict | list | None:
-        """Make a GET request with retry on 403 (rate limit)."""
+        """GET via shared retry helper; 403 rate-limit handled as transient."""
-        req = urllib.request.Request(url, headers=self._headers())
+        try:
-
+            raw, headers = http_get(
-        for attempt in range(max_retries + 1):
+                url, headers=self._headers(), timeout=15,
-            try:
+                max_retries=max_retries, retry_403_ratelimit=True,
-                with urllib.request.urlopen(req, timeout=15) as resp:
+                return_headers=True, owner=self)
-                    data = json.loads(resp.read().decode("utf-8"))
+        except AdapterHTTPError as e:
-
+            print(f"    {e.failure_class}: GitHub {url}")
-                # Check rate limit headers
+            return None
-                remaining = int(resp.headers.get("X-RateLimit-Remaining", 0))
+        # Informational: flag if we're close to the unauth rate ceiling
-                if remaining <= 5:
+        try:
-                    print(f"    ⚠ Rate limit low ({remaining} remaining), stopping")
+            remaining = int(headers.get("X-RateLimit-Remaining", 0))
-                    break
+            if remaining <= 5:
-
+                print(f"    ⚠ Rate limit low ({remaining} remaining)")
-                return data
+        except Exception:
-            except urllib.error.HTTPError as e:
+            pass
-                if e.code == 403:
+        try:
-                    # Rate limited — reset time is in headers
+            return json.loads(raw.decode("utf-8"))
-                    reset = int(e.headers.get("X-RateLimit-Reset", 0))
+        except Exception as e:
-                    if reset:
+            print(f"    GitHub decode error: {e}")
-                        wait = max(reset - int(time.time()), 0) + 1
+            return None
-                        print(f"    ⚠ Rate limited, wait {wait}s")
-                    else:
-                        wait = 30 * (attempt + 1)
-                        print(f"    403 on attempt {attempt + 1}, retry in {wait}s")
-                    time.sleep(min(wait, 300))  # cap at 5 min
-                    continue
-                print(f"    HTTP {e.code} for {url}")
-                return None
-            except Exception as e:
-                print(f"    Request error: {e}")
-                return None
-
-        return None
 
     def _search_repos(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 30) -> list:
         """Search repositories via GitHub API."""
@@ -163,8 +145,12 @@ class GitHubAdapter(SourceAdapter):
             cutoff = (now - timedelta(days=30)).strftime("%Y-%m-%d")
             # Three queries for breadth: agents, LLM/infra, and security/tools
             repos = []
-            for q in self.search_terms:
+            for q in [
-                batch = self._search_repos(f"{q} created:>{cutoff}", sort="stars", per_page=30)
+                f"ai agent created:>{cutoff}",
+                f"llm OR inference OR rag created:>{cutoff}",
+                f"autonomous agent OR AI tool created:>{cutoff}",
+            ]:
+                batch = self._search_repos(q, sort="stars", per_page=30)
                 repos.extend(batch)
                 time.sleep(1)  # polite spacing
 

adapters/hackernews.py

@@ -20,7 +20,7 @@ import urllib.request
 import urllib.error
 from datetime import datetime, timezone
 
-from adapters import SourceAdapter, source_config
+from adapters import SourceAdapter, http_get, AdapterHTTPError
 
 
 class HackerNewsAdapter(SourceAdapter):
@@ -28,46 +28,81 @@ class HackerNewsAdapter(SourceAdapter):
 
     BASE = "https://hacker-news.firebaseio.com/v0"
 
+    AI_KEYWORDS = [
+        # Multi-word phrases (unambiguous)
+        "language model", "deep learning", "foundation model", "retrieval augmented",
+        "code generation", "context length", "context window", "attention mechanism",
+        # Compound/abbreviations (unambiguous)
+        "llm", "gpt-", "gpt ", "rag ", "rag.", "vlm", "vla",
+        # Specific company/product names
+        "openai", "anthropic", "deepseek", "meta ai", "xai", "ponytail",
+        # Topic-specific (with word boundary awareness in _is_ai_relevant)
+        "inference", "transformer", "diffusion", "alignment", "fine-tun",
+        "embedd", "pretrain", "post-train", "multimodal", "reasoning",
+        # Domain-specific (need boundary check)
+        "ai ", " ai", "ai-", "-ai",  # "ai" as word, not substring
+        "agent", "agents", "neural", "autonomous",
+        "compute", "training run", "computer use", "coding agent",
+        # Community terms
+        "local-llm", "local llama", "llama ",
+    ]
+
     def __init__(self, user_agent=None):
         self.user_agent = user_agent or "python:athena:v0.1 (by tony_tech)"
-        # Curation now centralized (issue #7): load from config/queries.json
-        cfg = source_config("hackernews")
-        self.ai_keywords = cfg.get("keywords") or []
 
     def name(self) -> str:
         return "hackernews"
 
     def _request(self, path: str, max_retries: int = 2) -> dict | list | None:
-        """Make a GET request to the HN Firebase API."""
+        """GET via shared retry helper (retries 429/5xx)."""
         url = f"{self.BASE}{path}"
-        req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
+        try:
-
+            raw = http_get(url, headers={"User-Agent": self.user_agent},
-        for attempt in range(max_retries + 1):
+                           timeout=15, max_retries=max_retries, owner=self)
-            try:
+        except AdapterHTTPError as e:
-                with urllib.request.urlopen(req, timeout=15) as resp:
+            print(f"    {e.failure_class}: HN {path}")
-                    return json.loads(resp.read().decode("utf-8"))
+            return None
-            except (urllib.error.HTTPError, urllib.error.URLError) as e:
+        try:
-                if attempt < max_retries:
+            return json.loads(raw.decode("utf-8"))
-                    time.sleep(3 * (attempt + 1))
+        except Exception as e:
-                    continue
+            print(f"    HN decode error: {e}")
-                print(f"    HTTP error: {e}")
+            return None
-                return None
-            except Exception as e:
-                print(f"    Request error: {e}")
-                return None
-        return None
 
     def _is_ai_relevant(self, title: str) -> bool:
         """Check if a story title is AI/ML relevant.
 
-        Keywords are loaded from config/queries.json (issue #7) into
+        Uses multi-pass matching: first check unambiguous multi-word/phrases,
-        self.ai_keywords — single source of truth, no inline duplicate.
+        then check word-boundary matches for shorter keywords that could
-        Substring match; callers pass lowercased titles for boundary terms.
+        false-positive (e.g. 'ai' matching 'Britain').
         """
         title_lower = title.lower()
-        for kw in self.ai_keywords:
+
+        # Pass 1: unambiguous keywords (multi-word, compound, specific names)
+        unambiguous = [
+            "language model", "deep learning", "foundation model", "retrieval augmented",
+            "code generation", "context length", "context window", "attention mechanism",
+            "llm", "gpt-", "gpt ", "rag ", "rag.", "vlm", "vla",
+            "openai", "anthropic", "deepseek", "meta ai", "xai", "ponytail",
+            "inference", "transformer", "diffusion", "alignment", "fine-tun",
+            "embedd", "pretrain", "post-train", "multimodal", "reasoning",
+            "agent", "agents", "neural", "autonomous",
+            "training run", "computer use", "coding agent",
+            "local-llm", "local llama", "llama ",
+        ]
+        for kw in unambiguous:
             if kw in title_lower:
                 return True
+
+        # Pass 2: word-boundary check for "ai" and "compute" (avoid 'Britain', 'Guinea', etc.)
+        import re
+        if re.search(r'\bai\b', title_lower):
+            return True
+        if re.search(r'\bcompute\b', title_lower):
+            # Only if combined with other tech context
+            tech_words = ["gpu", "tpu", "cluster", "datacenter", "data center", "server"]
+            if any(w in title_lower for w in tech_words):
+                return True
+
         return False
 
     def _score(self, item: dict) -> float:

adapters/huggingface.py

@@ -34,7 +34,7 @@ import urllib.request
 import urllib.error
 from datetime import datetime, timedelta, timezone
 
-from adapters import SourceAdapter
+from adapters import SourceAdapter, http_get, AdapterHTTPError
 
 
 class HuggingFaceAdapter(SourceAdapter):
@@ -84,24 +84,19 @@ class HuggingFaceAdapter(SourceAdapter):
         return headers
 
     def _request(self, path: str, max_retries: int = 2) -> list | dict | None:
-        """Make a GET request to the HF API."""
+        """GET via shared retry helper (retries 429/5xx)."""
         url = f"{self.BASE}{path}"
-        req = urllib.request.Request(url, headers=self._headers())
+        try:
-
+            raw = http_get(url, headers=self._headers(), timeout=20,
-        for attempt in range(max_retries + 1):
+                           max_retries=max_retries, owner=self)
-            try:
+        except AdapterHTTPError as e:
-                with urllib.request.urlopen(req, timeout=20) as resp:
+            print(f"    {e.failure_class}: HF {path}")
-                    return json.loads(resp.read().decode("utf-8"))
+            return None
-            except (urllib.error.HTTPError, urllib.error.URLError) as e:
+        try:
-                if attempt < max_retries:
+            return json.loads(raw.decode("utf-8"))
-                    time.sleep(3 * (attempt + 1))
+        except Exception as e:
-                    continue
+            print(f"    HF decode error: {e}")
-                print(f"    HF API error: {e}")
+            return None
-                return None
-            except Exception as e:
-                print(f"    Request error: {e}")
-                return None
-        return None
 
     def _is_ai_relevant(self, model: dict) -> bool:
         """Check if a model/dataset is AI/ML relevant.

adapters/reddit.py

@@ -24,13 +24,13 @@ import xml.etree.ElementTree as ET
 from datetime import datetime, timezone
 from html import unescape
 
-from adapters import SourceAdapter, source_config
+from adapters import SourceAdapter, http_get, AdapterHTTPError
 
 
 class RedditAdapter(SourceAdapter):
     """Reddit RSS + JSON adapter."""
 
-    # Default subreddits for AI content (fallback if config missing)
+    # Default subreddits for AI content
     DEFAULT_SUBREDDITS = [
         "MachineLearning", "artificial", "LocalLLaMA", "Startups",
     ]
@@ -57,8 +57,7 @@ class RedditAdapter(SourceAdapter):
             rate_limit: Seconds between subreddit requests.
             user_agent: Custom User-Agent header.
         """
-        cfg = source_config("reddit")
+        self.subreddits = subreddits or self.DEFAULT_SUBREDDITS
-        self.subreddits = subreddits or cfg.get("subreddits") or self.DEFAULT_SUBREDDITS
         self.rate_limit = rate_limit
         self.user_agent = user_agent or "python:ai-oracle:v0.1 (by tony_tech)"
 
@@ -102,33 +101,27 @@ class RedditAdapter(SourceAdapter):
         return False
 
     def _fetch_rss(self, subreddit: str) -> list[dict]:
-        """Fetch RSS feed for a subreddit."""
+        """Fetch RSS feed for a subreddit (shared retry helper).
-        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
-        req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
 
-        for attempt in range(2):  # max 2 attempts, fail fast
+        Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s
-            try:
+        retry then []; 5xx -> helper retry then []. Not slower than before.
-                with urllib.request.urlopen(req, timeout=10) as resp:
+        """
-                    xml_data = resp.read().decode("utf-8")
+        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
-                break
+        try:
-            except urllib.error.HTTPError as e:
+            raw = http_get(url, headers={"User-Agent": self.user_agent},
-                if e.code in (403,):
+                           timeout=10, max_retries=1, backoff_base=2, owner=self)
-                    print(f"    RSS blocked (HTTP {e.code}) for r/{subreddit}")
+        except AdapterHTTPError as e:
-                    return []
+            if e.status == 403:
-                if e.code == 429:
+                print(f"    RSS blocked (HTTP 403) for r/{subreddit}")
-                    if attempt == 0:
+            elif e.status == 429:
-                        time.sleep(2)  # single retry with short backoff
+                print(f"    RSS rate-limited for r/{subreddit}, skip")
-                        continue
+            else:
-                    print(f"    RSS rate-limited for r/{subreddit}, skip")
+                print(f"    RSS {e.failure_class} for r/{subreddit}")
-                    return []
-                print(f"    RSS HTTP {e.code} for r/{subreddit}")
-                return []
-            except Exception as e:
-                print(f"    RSS error r/{subreddit}: {e}")
-                return []
-        else:
-            print(f"    r/{subreddit}: still rate limited, skip")
             return []
+        except Exception as e:
+            print(f"    RSS error r/{subreddit}: {e}")
+            return []
+        xml_data = raw.decode("utf-8")
 
         # Parse Atom XML
         entries = []
@@ -304,33 +297,27 @@ class RedditAdapter(SourceAdapter):
         return tags
 
     def _fetch_rss(self, subreddit: str) -> list[dict]:
-        """Fetch RSS feed for a subreddit."""
+        """Fetch RSS feed for a subreddit (shared retry helper).
-        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
-        req = urllib.request.Request(url, headers={"User-Agent": self.user_agent})
 
-        for attempt in range(2):  # max 2 attempts, fail fast
+        Preserves prior fast-bail: 403 -> immediate []; 429 -> single 2s
-            try:
+        retry then []; 5xx -> helper retry then []. Not slower than before.
-                with urllib.request.urlopen(req, timeout=10) as resp:
+        """
-                    xml_data = resp.read().decode("utf-8")
+        url = f"https://www.reddit.com/r/{subreddit}/hot/.rss?limit=50"
-                break
+        try:
-            except urllib.error.HTTPError as e:
+            raw = http_get(url, headers={"User-Agent": self.user_agent},
-                if e.code in (403,):
+                           timeout=10, max_retries=1, backoff_base=2, owner=self)
-                    print(f"    RSS blocked (HTTP {e.code}) for r/{subreddit}")
+        except AdapterHTTPError as e:
-                    return []
+            if e.status == 403:
-                if e.code == 429:
+                print(f"    RSS blocked (HTTP 403) for r/{subreddit}")
-                    if attempt == 0:
+            elif e.status == 429:
-                        time.sleep(2)  # single retry with short backoff
+                print(f"    RSS rate-limited for r/{subreddit}, skip")
-                        continue
+            else:
-                    print(f"    RSS rate-limited for r/{subreddit}, skip")
+                print(f"    RSS {e.failure_class} for r/{subreddit}")
-                    return []
-                print(f"    RSS HTTP {e.code} for r/{subreddit}")
-                return []
-            except Exception as e:
-                print(f"    RSS error r/{subreddit}: {e}")
-                return []
-        else:
-            print(f"    r/{subreddit}: still rate limited, skip")
             return []
+        except Exception as e:
+            print(f"    RSS error r/{subreddit}: {e}")
+            return []
+        xml_data = raw.decode("utf-8")
 
         # Parse Atom XML
         entries = []

adapters/rss_feeds.py

@@ -24,7 +24,7 @@ import feedparser
 from datetime import datetime, timedelta, timezone
 from email.utils import parsedate_to_datetime
 
-from adapters import SourceAdapter, source_config
+from adapters import SourceAdapter
 
 
 # Curated feed list — AI-focused, reliable, diverse publishers.
@@ -83,18 +83,6 @@ AI_KEYWORDS = [
 class RSSFeedsAdapter(SourceAdapter):
     """RSS feed aggregator for commercial AI news."""
 
-    # Module-level fallbacks (used only if config/queries.json is missing)
-    FEEDS = [
-        ("rss:techcrunch", "TechCrunch AI", "https://techcrunch.com/category/artificial-intelligence/feed/"),
-    ]
-    AI_KEYWORDS = [r"\bai\b"]
-
-    def __init__(self):
-        # Curation centralized (issue #7): config wins, fallbacks otherwise
-        cfg = source_config("rss")
-        self.feeds = cfg.get("feeds") or list(self.FEEDS)
-        self.ai_keywords = cfg.get("keywords") or list(self.AI_KEYWORDS)
-
     def name(self) -> str:
         return "rss"
 
@@ -104,7 +92,7 @@ class RSSFeedsAdapter(SourceAdapter):
         tag_text = " ".join(tags).lower()
         combined = text + " " + tag_text
 
-        for pattern in self.ai_keywords:
+        for pattern in AI_KEYWORDS:
             if re.search(pattern, combined):
                 return True
         return False
@@ -159,7 +147,7 @@ class RSSFeedsAdapter(SourceAdapter):
         all_entries = []
         feed_failures = []
 
-        for source_key, label, url in self.feeds:
+        for source_key, label, url in FEEDS:
             try:
                 d = feedparser.parse(url)
                 if d.status not in (200, 301, 302, 307, 308) or not d.entries:

config/queries.json

@@ -1,49 +0,0 @@
-{
-  "sources": {
-    "hackernews": {
-      "keywords": [
-        "language model", "deep learning", "foundation model", "retrieval augmented",
-        "code generation", "context length", "context window", "attention mechanism",
-        "llm", "gpt-", "gpt ", "rag ", "rag.", "vlm", "vla",
-        "openai", "anthropic", "deepseek", "meta ai", "xai", "ponytail",
-        "inference", "transformer", "diffusion", "alignment", "fine-tun",
-        "embed", "pretrain", "post-train", "multimodal", "reasoning",
-        "ai ", " ai", "ai-", "-ai",
-        "agent", "agents", "neural", "autonomous",
-        "compute", "training run", "computer use", "coding agent",
-        "local-llm", "local llama", "llama "
-      ]
-    },
-    "arxiv": {
-      "categories": ["cs.AI", "cs.LG", "cs.CL"]
-    },
-    "reddit": {
-      "subreddits": ["MachineLearning", "artificial", "LocalLLaMA", "Startups"]
-    },
-    "rss": {
-      "feeds": [
-        ["rss:techcrunch", "TechCrunch AI", "https://techcrunch.com/category/artificial-intelligence/feed/"],
-        ["rss:venturebeat", "VentureBeat AI", "https://venturebeat.com/category/ai/feed/"],
-        ["rss:theverge", "The Verge AI", "https://www.theverge.com/rss/ai-artificial-intelligence/index.xml"],
-        ["rss:ainews", "AI News", "https://www.artificialintelligence-news.com/feed/"],
-        ["rss:decoder", "The Decoder", "https://www.the-decoder.com/feed/"],
-        ["rss:mittr", "MIT Tech Review AI", "https://www.technologyreview.com/topic/artificial-intelligence/feed/"],
-        ["rss:openai", "OpenAI Blog", "https://openai.com/blog/rss.xml"],
-        ["rss:anthropic", "Anthropic News", "https://www.anthropic.com/rss/news.xml"],
-        ["rss:googleai", "Google AI Blog", "https://blog.google/technology/rss.xml"],
-        ["rss:metaai", "Meta AI Blog", "https://ai.meta.com/blog/rss.xml"]
-      ],
-      "keywords": [
-        "\\bai\\b", "\\bmachine learning\\b", "\\bdeep learning\\b", "\\bneural\\b",
-        "\\bgenerative ai\\b", "\\bgenerative\\b", "\\bllm\\b", "\\blarge language\\b",
-        "\\bfoundation model\\b", "\\btransformer\\b", "\\baugmented\\b",
-        "\\bagent\\b", "\\bautonomous\\b", "\\bmcp\\b", "\\bfunction call\\b",
-        "\\btool use\\b", "\\brai\\b", "\\bretrieval\\b",
-        "\\binference\\b", "\\bmodel\\b", "\\bembedding\\b", "\\btoken\\b"
-      ]
-    },
-    "github": {
-      "search_terms": ["machine-learning", "deep-learning", "llm", "ai-agent", "transformer"]
-    }
-  }
-}

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,31 @@ 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}")

reddit_proof.py

@@ -0,0 +1,318 @@
+#!/usr/bin/env python3
+"""
+Reddit Idea Generator — Proof of Concept v5
+Uses Reddit RSS feeds (Atom XML). No browser needed.
+Trafilatura for clean text extraction. SQLite for storage.
+
+Usage: python3 reddit_proof.py [count]
+Example: python3 reddit_proof.py 20
+"""
+
+import sys
+import json
+import re
+import xml.etree.ElementTree as ET
+import sqlite3
+import os
+import time
+import urllib.request
+import urllib.error
+from datetime import datetime, timezone
+from html import unescape
+
+import trafilatura
+
+DB_PATH = os.path.join(os.path.dirname(__file__), "oracle.db")
+SCHEMA_PATH = os.path.join(os.path.dirname(__file__), "schema.sql")
+
+SUBREDDITS = [
+    "MachineLearning", "artificial", "LocalLLaMA", "Startups",
+]
+
+
+def init_db():
+    conn = sqlite3.connect(DB_PATH)
+    with open(SCHEMA_PATH) as f:
+        conn.executescript(f.read())
+    conn.commit()
+    return conn
+
+
+def fetch_rss(subreddit, sort="hot"):
+    """Fetch RSS feed for a subreddit. Returns parsed entries."""
+    url = f"https://www.reddit.com/r/{subreddit}/{sort}/.rss?limit=50"
+    req = urllib.request.Request(url, headers={"User-Agent": "oracle-reddit-proof/1.0"})
+
+    for attempt in range(3):
+        try:
+            with urllib.request.urlopen(req, timeout=15) as resp:
+                xml_data = resp.read().decode("utf-8")
+            break
+        except urllib.error.HTTPError as e:
+            if e.code == 429:
+                wait = 5 * (attempt + 1)
+                print(f"    429 on r/{subreddit}, retry in {wait}s")
+                time.sleep(wait)
+                continue
+            print(f"    RSS error r/{subreddit}: {e}")
+            return []
+        except Exception as e:
+            print(f"    RSS error r/{subreddit}: {e}")
+            return []
+    else:
+        print(f"    r/{subreddit}: still rate limited, skip")
+        return []
+
+    # Parse Atom XML — find all <entry> elements
+    root = ET.fromstring(xml_data)
+    entries = []
+
+    # Handle namespace: Atom uses http://www.w3.org/2005/Atom
+    # But ET.findall with ns prefix requires registering the namespace
+    # Simpler approach: strip namespace from tags and search directly
+    for entry in root.iter():
+        # Get local name (strip namespace)
+        tag = entry.tag.split("}")[-1] if "}" in entry.tag else entry.tag
+
+        if tag == "entry":
+            title = None
+            link = None
+            author = ""
+            content = ""
+            pub = ""
+            eid = ""
+
+            for child in entry:
+                ctag = child.tag.split("}")[-1]
+                if ctag == "title":
+                    title = child.text
+                elif ctag == "link":
+                    link = child.get("href", "")
+                elif ctag == "author":
+                    name_el = child[0] if child else None
+                    if name_el:
+                        name_tag = name_el.tag.split("}")[-1]
+                        if name_tag == "name":
+                            author = name_el.text or ""
+                elif ctag == "content":
+                    content = child.text or ""
+                elif ctag == "published":
+                    pub = child.text or ""
+                elif ctag == "id":
+                    eid = child.text or ""
+
+            if title and link:
+                entries.append({
+                    "title": unescape(title.strip()),
+                    "url": link,
+                    "author": unescape(author.strip()),
+                    "content": content,
+                    "published": pub,
+                    "id": eid,
+                    "subreddit": subreddit,
+                })
+
+    return entries
+
+
+def clean_html_content(html):
+    """Extract readable text from Reddit's HTML content."""
+    if not html:
+        return ""
+    text = re.sub(r"<!--.*?-->", "", html, flags=re.DOTALL)
+    text = re.sub(r"<div[^>]*>", "\n", text)
+    text = re.sub(r"</div>", "\n", text)
+    text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
+    text = re.sub(r"<[^>]+>", "", text)
+    text = unescape(text)
+    text = re.sub(r"\n\s*\n+", "\n\n", text)
+    return text.strip()
+
+
+def main():
+    if len(sys.argv) > 1:
+        count = int(sys.argv[1])
+    else:
+        count = 20
+
+    print(f"=== Reddit Idea Generator — Proof of Concept v5 ===")
+    print(f" count: {count}")
+    print()
+
+    conn = init_db()
+    cursor = conn.cursor()
+
+    # Step 1: Fetch RSS
+    print(f"[1/3] Fetching RSS feeds...")
+    all_entries = []
+    seen_ids = set()
+
+    for i, sub in enumerate(SUBREDDITS):
+        entries = fetch_rss(sub)
+        new = [e for e in entries if e["id"] not in seen_ids]
+        seen_ids.update(e["id"] for e in new)
+        all_entries.extend(new)
+        if new:
+            print(f"  r/{sub}: {len(new)} entries")
+        # Rate limit between subreddits
+        if i < len(SUBREDDITS) - 1:
+            time.sleep(3)
+
+    print(f"  Total: {len(all_entries)} entries")
+
+    if not all_entries:
+        print("\n  No entries fetched. Reddit may be rate-limiting this IP.")
+        print("  Try again later or use fewer subreddits.")
+        sys.exit(1)
+
+    # Limit to count
+    entries_to_store = all_entries[:count]
+    print(f"  Storing {len(entries_to_store)} entries")
+
+    # Step 2: Store
+    stored = 0
+    for entry in entries_to_store:
+        post_id = entry["id"].replace("t3_", "")
+        content_text = clean_html_content(entry["content"])
+
+        # Signal score — RSS hot feed already sorted by relevance
+        # Use position-based scoring (higher rank = higher score)
+        idx = entries_to_store.index(entry)
+        score = max(10.0 - idx * 0.5, 1.0)
+
+        # Category tags
+        category_tags = ["reddit"]
+        sub = entry.get("subreddit", "").lower()
+        if "machinelearning" in sub:
+            category_tags.append("machine-learning")
+        elif "artificial" in sub:
+            category_tags.append("ai-general")
+        elif "localllama" in sub:
+            category_tags.append("local-llm")
+        elif "startups" in sub:
+            category_tags.append("startups")
+
+        # Post type from title markers
+        title = entry.get("title", "")
+        if " [P]" in title or " [p]" in title:
+            category_tags.append("project")
+        elif " [R]" in title or " [r]" in title:
+            category_tags.append("research")
+        elif " [D]" in title or " [d]" in title:
+            category_tags.append("discussion")
+        elif " [N]" in title or " [n]" in title:
+            category_tags.append("news")
+        else:
+            category_tags.append("general")
+
+        # Clean title (remove [X] markers)
+        clean_title = re.sub(r"\s*\[[A-Z]\]\s*$", "", title)
+
+        raw_meta = {
+            "subreddit": entry["subreddit"],
+            "author": entry["author"],
+            "published": entry["published"],
+            "text_length": len(content_text),
+        }
+
+        source_id = post_id or entry["url"].split("/")[-1] or f"rss_{stored}"
+        now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        try:
+            cursor.execute("""
+                INSERT OR REPLACE INTO entries
+                (source, source_id, url, title, extracted_text, summary,
+                 category_tags, signal_score, raw_metadata, first_seen, last_updated)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """, (
+                "reddit", source_id, entry["url"], clean_title,
+                content_text,
+                None,  # summary — LLM later
+                json.dumps(category_tags),
+                score,
+                json.dumps(raw_meta),
+                now, now,
+            ))
+            stored += 1
+        except Exception as e:
+            print(f"  DB ERROR: {e}")
+
+    conn.commit()
+    print(f"  Stored {stored} entries")
+
+    # Step 3: Summary
+    print(f"\n[3/3] Summary")
+    cursor.execute("SELECT COUNT(*) FROM entries")
+    total = cursor.fetchone()[0]
+    print(f"  Total entries in DB: {total}")
+    cursor.execute("SELECT COUNT(*) FROM entries WHERE source='reddit'")
+    reddit_count = cursor.fetchone()[0]
+    print(f"  Reddit entries: {reddit_count}")
+    cursor.execute("SELECT AVG(signal_score) FROM entries WHERE source='reddit'")
+    avg_score = cursor.fetchone()[0] or 0
+    print(f"  Avg signal score: {avg_score:.2f}")
+
+    # Subreddit distribution
+    cursor.execute("""
+        SELECT raw_metadata, COUNT(*) FROM entries
+        WHERE source='reddit'
+        GROUP BY raw_metadata
+        ORDER BY COUNT(*) DESC
+    """)
+    print(f"\n  Subreddit distribution:")
+    for meta, cnt in cursor.fetchall():
+        d = json.loads(meta)
+        print(f"    r/{d.get('subreddit', '?')}: {cnt}")
+
+    # Top 5
+    print(f"\n  Top 5 by signal score:")
+    cursor.execute("""
+        SELECT id, title, signal_score, raw_metadata, category_tags,
+               LENGTH(extracted_text) as text_len
+        FROM entries WHERE source='reddit'
+        ORDER BY signal_score DESC
+        LIMIT 5
+    """)
+    for row in cursor.fetchall():
+        eid, title, score, meta, tags, txt_len = row
+        meta_dict = json.loads(meta) if meta else {}
+        print(f"    [{eid}] score={score:.1f} text={txt_len}ch")
+        print(f"        {title[:90]}")
+        print(f"        r/{meta_dict.get('subreddit', '?')} "
+              f"by {meta_dict.get('author', '?')}")
+
+    # Extraction quality
+    print(f"\n  Extraction quality (top entry):")
+    cursor.execute("""
+        SELECT title, extracted_text
+        FROM entries WHERE source='reddit'
+        ORDER BY signal_score DESC
+        LIMIT 1
+    """)
+    row = cursor.fetchone()
+    if row:
+        title, excerpt = row
+        print(f"    Title: {title[:80]}")
+        print(f"    Length: {len(excerpt) if excerpt else 0} chars")
+        if excerpt:
+            print(f"    Preview:\n    {excerpt[:400]}...")
+        else:
+            print("    (empty)")
+
+    # Check for garbled extractions
+    cursor.execute("""
+        SELECT COUNT(*) FROM entries
+        WHERE source='reddit' AND LENGTH(extracted_text) < 100
+    """)
+    short_count = cursor.fetchone()[0]
+    if short_count > 0:
+        print(f"\n  ⚠ {short_count}/{stored} entries have very short extractions (<100 chars)")
+        print("    These are likely link-only posts or external links")
+
+    conn.close()
+    print(f"\n  Database: {DB_PATH}")
+    print("  Done.")
+
+
+if __name__ == "__main__":
+    main()

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
 );