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"]
-    }
-  }
-}

docs/Deployment-Plan.md

@@ -0,0 +1,42 @@
+# Deployment Plan — Athena MVP
+
+## Prerequisites
+- Docker installed on the target host
+- Network access to the 6 source APIs/feeds (arxiv, github, huggingface, hackernews, reddit, rss_feeds)
+- `GITHUB_TOKEN` available as an environment variable (optional, but raises the GitHub rate limit from 60/hr to 5000/hr)
+- `HUGGINGFACE_TOKEN` available as an environment variable
+- A local Ollama instance running `llama3.2:1b`, or an alternate reachable inference backend if swapping — per the model-agnostic `summarize(text) -> (summary, model)` contract
+- Hermes cron infrastructure available and able to invoke `oracle-pipeline.sh`
+
+## Setup
+1. Clone `main` (not a milestone branch) onto the target host
+2. `pip install -r requirements.txt` if present; otherwise confirm stdlib + `requests` are available
+3. Run `schema.sql` against a fresh `oracle.db` — this file is git-ignored and created locally, never committed
+4. Export required environment variables (`GITHUB_TOKEN`, `HUGGINGFACE_TOKEN`) — never hardcode these
+5. Build and run inside Docker with the 150MB memory cap and non-root user enforced, per the PRD platform requirements
+6. Manual smoke test: run `python3 pipeline.py` once and confirm ingest → store → summarize → score completes without errors before handing off to cron
+
+## Cron / Scheduling
+1. Confirm `oracle-pipeline.sh` is the entry point Hermes cron calls
+2. Schedule for 13:00 UTC daily
+3. Add a lock file or PID check so overlapping runs can't happen if a prior run is still in progress
+4. Confirm the cron environment actually carries the exported tokens — cron environments are frequently minimal and won't inherit an interactive shell's exports
+
+## Hermes Integration
+1. Confirm the wrapper script's exit codes are meaningful (0 = success, non-zero = failure) so Hermes can act on them
+2. Define where Hermes should look for pipeline output/logs
+3. Decide on a failure-notification path (Hermes alert, log flag, etc.) — **not yet specified, needs a decision**
+
+## Monitoring
+1. Aggregate logs from each pipeline stage (ingest, store, summarize, score)
+2. Track theme-scan new-arrival counts per cycle per theme — this is the core signal the falsification logic depends on, so it deserves visibility beyond raw logs
+3. Add a heartbeat/dead-man's-switch alert if a scheduled run doesn't fire, rather than relying on someone noticing missing data days later
+4. Watch memory usage against the 150MB cap under real production load, not just dev conditions
+
+## Rollback
+1. Back up `oracle.db` before any schema change — it's git-ignored and not recoverable from the repo itself
+2. If a bad deploy breaks the pipeline, revert to the last known-good commit on `main` and redeploy the Docker image
+3. Check falsification state (new-arrivals counters) after any rollback — rolling back mid-window could distort the 7-day dead-thesis calculation if not handled carefully
+
+---
+*Draft prepared by Claude from the README, whitepaper falsification logic, and MVP-PRD platform requirements on `main`. The Hermes failure-notification path is the one open decision blocking this from being final.*

docs/MVP-PRD.md

@@ -0,0 +1,161 @@
+## Chapter 1: Vision and Scope
+
+### Elevator Pitch
+
+Athena is an autonomous research intelligence engine that cuts through high-volume, fragmented signals by ingesting from multiple sources, surfacing cross-source convergence, and using falsification to distinguish real momentum from noise. While the initial focus is on AI signals, the system is designed to work with any class of signals. It delivers actionable insight into emerging trends and capability gaps while remaining model-agnostic and lightweight enough to run autonomously.
+
+### 1.1 Vision
+
+#### Why are we building it?
+
+The AI space produces an overwhelming volume of new research, tools, discussions, and model releases every day. Individual sources only provide partial views, making it difficult to distinguish genuine, sustained trends from one-day spikes. Without a system that can detect convergence across sources and validate momentum over time, real opportunities tied to emerging capability gaps are missed.
+
+#### What happens if we don’t build it?
+
+Without this capability, builders and researchers will continue to operate with fragmented, noisy signals. Early indicators of meaningful trends will remain hidden, decisions will stay reactive, and the ability to spot validated cross-source momentum before it becomes obvious will be lost.
+
+#### When must it be done?
+
+The foundational ability to reliably ingest, score, and validate signals through falsification must be established before meaningful trend detection and opportunity mapping can occur. This forms the core of the MVP and must be in place to enable the system to deliver on its intended value.
+
+### 1.2 Personas and Archetypes
+
+See committed document:  
+**`docs/Personas-and-Archetypes.md`** (on `MVP-milestone` branch)
+
+**Summary of scoped personas and archetypes for MVP:**
+
+**Personas**
+- Pers-1 (Bob) – Sector Trend Tracker (New to AI)
+- Pers-2 (Alice) – Content Creator
+- Pers-3 (Sam) – Hermes Research Agent
+
+**Archetypes**
+- Arch-1 (Small Scrappy VPS)
+- Arch-2 (Research Consumption Layer)
+
+All user stories in this PRD are scoped to combinations of the above.
+
+### 1.3 Use Case Priority Taxonomy
+
+This PRD focuses on defining the core functionality required for MVP. It also catalogs use cases and requirements across V1.0 – V1.5 to maintain context. The primary goal is to deliver a working MVP, with future PRDs derived from the remaining prioritized content.
+
+We will use the following prioritization model:
+
+- **MVP**: The short list of P1 use cases required to prove the concept with a working prototype.
+- **P1**: Use cases that are fundamental to successfully implementing the product vision.
+- **P2**: Use cases that add strength, convenience, and quality to the product vision.
+- **P3**: Use cases that bring additional value but can be cut if time or resource constrained.
+
+## Chapter 2: User Stories (Bob)
+These user stories are based on the personas and archetypes document contained in this repo.  
+
+Chapter 2.1 - Bob's user stories
+
+**As Bob, I want to…**
+
+**Bob-1.** Automatically receive daily updates on new AI innovations without having to manually check multiple sources.
+
+**Bob-5.** See emerging trends and differentiate durable signal from temporary or artificial hype.
+
+**Bob-10.** See when the same idea or pattern is appearing across multiple independent sources (GitHub, arXiv, Reddit, HN, HF).
+
+**Bob-15.** Identify emerging capability gaps or opportunities early, before they become widely obvious.
+
+**Bob-20.** Have research that gives me confidence it is exhaustive and vetted.
+
+**Bob-25.** Adjust or alter the underlying data feeds and weights so I can tune the accuracy and relevance of the output.
+
+**Bob-30.** Understand why a particular signal is considered strong or weak (e.g., cross-source convergence or falsification results).
+
+Chapter 2.2 - Alice's user stories
+As Alice, I want to…
+Alice-1. Integrate deep, vetted research directly into my existing content production pipeline so I can reduce manual research time.
+Alice-5. Query the research system with follow-up questions to explore specific angles or topics on demand.
+Alice-10. Have my tools automatically receive curated, high-signal research so I can focus on content creation instead of information filtering.
+Alice-15. Get research outputs in a structured format that my existing AI tools and workflows can consume without manual reformatting.
+Alice-20. Quickly surface non-obvious insights and patterns from research data to develop more compelling content angles.
+Alice-25. Control which research sources and signals are prioritized so the output stays aligned with my content focus and audience.
+Alice-30. Understand the reasoning and supporting evidence behind key research findings so I can speak to them confidently in my content.
+
+## Chapter 3: Requirements
+Requirements defined as what the product / system must do, differentiated from what the persona can accomplish. Requirements are defined to meet the needs of use cases as well as the architectural system design.
+
+High level design (refer to ***TBD_Design.MD for full design details)
+High-Level Design
+.
+├── Runtime Environment
+│   ├── Linux
+│   └── Docker (containerized)
+│
+├── Core Components
+│   ├── Database: SQLite
+│   ├── Scheduling: Cron
+│   └── Runtime: Python
+│
+├── Connectivity
+│   ├── Outbound (Internet)
+│   │   ├── HTTP client for data feeds (RSS, cURL, optional Playwright)
+│   │   └── OpenAI-compatible inference endpoints
+│   ├── Inbound (Internet)
+│   │   └── HTTP server endpoint (MCP + external consumers)
+│   └── Internal (Intranet)
+│       └── HTTP client for local inference (e.g. Hermes)
+│
+├── Storage
+│   ├── File system (daily digest artifacts stored outside container)
+│   └── Temporary working storage during pipeline execution
+│
+├── Configuration & Secrets
+│   ├── Research topic manifest (feeds, URLs, declarations)
+│   ├── System settings (YAML)
+│   └── Secrets (.env)
+│
+├── Business Logic / Pipeline Flow
+│   ├── Starting trigger
+│   ├── Preflight checks
+│   ├── Query feeds → Temporary result storage
+│   ├── Vet and promote final results to database
+│   ├── Optional daily digest generation
+│   └── Cleanup and sleep
+│
+└── Observability
+├── Structured logging
+├── Diagnostics and instrumentation (inside Docker)
+└── Health/status reporting
+
+3.1 Setup and configuration (REQ-SNC-XX)
+Requirements for initial setup, deployment configs, updating, and uninstall
+    REQ-SNC-05 -, with outbound access to the internet and in/outbound access to the underlying OS network
+    REQ-SNC-10 - The installation process shall be a single command which can be run interactively or silently
+    REQ-SNC-15 - The insallation shall utilize best-practice settings and secrets storage
+    REQ-SNC-20 - 
+
+3.2 Platform requirements (REQ-PLT-XX)
+    REQ-PLT-05 -  All processes will run as standard user (no admin / sudo elevation necessary)
+    REQ-PLT-10 -  ...
+    REQ-PLT-15 -  The system shall be Docker based limited to 150MB of memory
+Requirements addressing what OS and hardware support is in scope
+3.3 Performance and scalability (REQ-PERF-XX)
+3.4 Instrumenation and diagnostics (REQ-DIAG-XX)
+3.5 
+### 3.1 Reliability
+
+REQ-REL-05: Once setup and configured, the system will reliably operate without interaction from the user.
+REQ-REL-10: The system shall automatically retry failed source fetches with exponential backoff.
+REQ-REL-15: The system shall not lose previously stored data on restart or failure.
+REQ-REL-20: The daily pipeline shall complete successfully even if one or more sources are unavailable.
+
+3.3 Observability and Diagnostics
+REQ-DIAG-05: The system shall produce structured logs with timestamps and severity levels.
+REQ-DIAG-10: A health check endpoint or command shall report overall system status and last successful run.
+REQ-DIAG-15: Run logs shall capture per-source success/failure and basic metrics (items fetched, stored, failed).
+
+3.4 Security 
+REQ-SEC-05: All external HTTP calls shall use TLS.
+REQ-SEC-10: No secrets shall be hardcoded or stored in plaintext.
+REQ-SEC-15: The system shall support least-privilege access for outbound API calls.
+
+3.5 Integration and extensibility 
+REQ-INT-05: The system shall expose research output in a structured, machine-readable format (e.g., JSON files or API) consumable by external tools.
+REQ-INT-10: The adapter layer shall support adding new sources without modifying core pipeline logic.

docs/Personas-and-Archetypes.md

@@ -0,0 +1,59 @@
+# Personas and Archetypes (Derived from /main)
+
+**Status**: First Draft – Derived from existing docs on `main`  
+**Source**: README.md + whitepaper.md (branch: main)  
+**Date**: 2026-07-08
+
+## Overview
+
+This document extracts the implied users and operating contexts directly from the current documentation on the `main` branch. It serves as the baseline before we expand or refine.
+
+---
+
+## Personas
+
+### Pers-1 (Bob) – Sector Trend Tracker (New to AI)
+- Is relatively new to AI and the broader space.
+- Wants to stay current with trends in AI (and potentially other sectors) without getting overwhelmed.
+- Needs a way to keep up with the high volume of new research, tools, and discussions with minimal ongoing effort.
+- Benefits from a system that filters noise and surfaces what actually matters.
+
+### Pers-2 (Alice) – Content Creator
+- Runs a YouTube channel and an X account with 25k followers.
+- Goal is to grow her audience significantly (targeting 1M followers).
+- Needs help doing research across AI and related topics.
+- Wants to convert research signals into interesting, timely, and compelling content for her audience.
+
+### Pers-3 (Sam) – Hermes Research Agent
+- Is a Hermes agent profile with its own memory and endpoint connection.
+- Acts as the dedicated research team member for an AI-first development team.
+- Needs to stay current on a defined market segment (AI for the MVP; extensible to other segments later).
+- Consumes structured signals from Athena to support ongoing research and decision-making within the team.
+
+---
+
+## Archetypes (Operating Environments)
+
+### Arch-1 (Small Scrappy VPS)
+- Small, low-budget, and scrappy VPS environment.
+- Used primarily for learning and early prototyping.
+- Requires a small-footprint workload that can be memory-constrained so it doesn’t destabilize the host system.
+
+### Arch-2 (Research Consumption Layer)
+- Functions as a consumption layer for Athena’s research output.
+- Designed to support downstream AI systems (examples: MCP tools, LoRA adapters, or other agent profiles).
+- Focuses on making Athena’s signals and summaries easily consumable by other systems rather than direct human use.
+
+---
+
+## Notes & Limitations (from /main)
+
+- The current documentation does **not** describe team or multi-user usage.
+- Emphasis is on autonomous operation and signal integrity.
+- Polished human-facing interfaces (e.g., daily digest) are not yet built.
+
+---
+
+## Next Steps
+
+This version incorporates the updated personas and archetypes.

docs/Prioritized-Task-List.md

@@ -0,0 +1,42 @@
+# Prioritized Task List — Athena MVP
+
+Tied to the personas and requirements in `MVP-PRD.md`. Ordered by phase; within each phase, roughly in the order they should be tackled.
+
+## Phase 1: Get Running Daily
+- [ ] Verify the cron entry (`oracle-pipeline.sh`) fires reliably at 13:00 UTC under Hermes
+- [ ] Confirm `pipeline.py` runs the full ingest → store → summarize → score cycle without manual intervention
+- [ ] Add a lock/guard so a slow run can't overlap with the next day's cron trigger
+- [ ] Validate all 6 adapters (arxiv, github, huggingface, hackernews, reddit, rss_feeds) independently — one adapter failing shouldn't kill the whole run
+- [ ] Confirm environment-only secrets (`GITHUB_TOKEN`, `HUGGINGFACE_TOKEN`) resolve correctly in the cron context (cron environments are often stripped down compared to an interactive shell)
+
+## Phase 2: Core Functionality
+- [ ] Confirm `schema.sql` initializes `oracle.db` cleanly and stays idempotent across repeated runs
+- [ ] Verify `theme_scan.py`'s 4-theme tagging (tool-call, context, compute, trust) against a few real days of data
+- [ ] Confirm the falsification counter (new arrivals per cycle) is genuinely idempotent — re-running against unchanged data must yield 0 new
+- [ ] Wire `summarize.py` to degrade gracefully when the Ollama endpoint (`llama3.2:1b`) isn't reachable — ingestion, scoring, and theme-scan must keep running without it
+- [ ] Confirm `archive.py`'s cold-storage rotation doesn't delete data still needed inside the 7-day falsification window
+
+## Phase 3: Observability & Reliability
+- [ ] Add structured logging per pipeline stage (ingest, store, summarize, score) with pass/fail per adapter
+- [ ] Surface theme-scan counts (new arrivals per theme per cycle) somewhere inspectable, not just buried in log files
+- [ ] Add a daily heartbeat/health check so a silent failure (e.g. cron didn't fire at all) is detectable rather than just showing up as missing data later
+- [ ] Decide and implement retry/backoff behavior for adapters that hit rate limits (especially GitHub without a token: 60/hr)
+
+## Phase 4: Human Consumption Layer
+- [ ] Extend `query.py` to support Bob's cross-source convergence lookups and Alice's curated-research pulls
+- [ ] Define the output format(s) for a "trend confirmed" vs. "trend killed" verdict (per the 7-day dead-thesis rule)
+- [ ] Decide how Alice's content pipeline actually consumes Athena's output — file drop, API, direct DB read — this is currently undefined
+
+## Phase 5: Validation & UAT
+- [ ] Run the pipeline unattended for at least one full 7-day falsification window
+- [ ] Manually verify at least one theme through to a real "confirmed" or "killed" verdict
+- [ ] Walk Bob's and Alice's user stories from `MVP-PRD.md` end-to-end against real output, not synthetic data
+- [ ] Confirm memory stays under the 150MB cap under real daily load, not just in a light dev test
+
+## Phase 6: Hermes Integration
+- [ ] Confirm `oracle-pipeline.sh`'s contract matches what Hermes cron expects (exit codes, output location)
+- [ ] Decide how Hermes is notified on pipeline failure vs. success — not yet specified
+- [ ] Confirm the non-root execution requirement is actually satisfied inside the Hermes-invoked environment, not just in local Docker testing
+
+---
+*Draft prepared by Claude from the MVP-PRD, Personas doc, and README/whitepaper on `main`. Open items flagged "not yet specified" need a decision before Phase 4–6 can be considered done.*

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