v2: SQLite cross-scan dedup, LLM failover chain, AI keyword pre-filter

Major dedup overhaul — persistent SQLite database (dedup_db.py) stores normalized URLs and titles across all scans, replacing text-file-only matching that missed query-param variants and same-event rewrites. LLM editor now uses a 3-tier failover chain (Gemini Flash Lite → Grok via OpenRouter → Gemini Flash Preview) instead of single-provider with raw-fallback. Providers alternate to avoid double failure on outages. Inline AI keyword filter in the orchestrator blocks non-AI articles at the RSS extraction stage. Editorial rules updated from "must produce 5" to "up to 7, quality over quantity." Includes 68 unit tests, updated README, changelog, and migration guide.
2026-08-14 00:47:57 +00:00 · 2026-03-04 15:42:43 -05:00
parent d7db73cf2d
commit 43544d2918
8 changed files with 1155 additions and 117 deletions
@@ -0,0 +1,112 @@
+# Changelog
+
+All notable changes to the OpenClaw News Scanner pipeline are documented here.
+
+---
+
+## [v2] — 2026-03-04
+
+### Major: Deduplication Overhaul
+
+**Problem:** Stories were appearing multiple times across scans — the same event reported by different outlets, URLs with different query parameters pointing to the same article, and previously posted stories resurfacing.
+
+**What changed:**
+
+- **New: SQLite dedup database (`dedup_db.py`)** — Persistent cross-scan memory. Every article the pipeline processes is recorded with a normalized URL and title. Before the LLM sees any candidates, the database filters out articles that match a previously seen URL (after normalization) or have a title that is 75%+ similar to a recent article. This replaces the old text-file-only approach that could only match exact URLs.
+
+- **URL normalization** strips query parameters, fragments, `www.` prefixes, and trailing punctuation, then lowercases the domain and normalizes to `https://`. Two URLs that point to the same article but differ only in tracking parameters are now correctly identified as duplicates.
+
+- **Cross-scan title matching** uses `SequenceMatcher` at a 75% threshold over a 2-day window. "Sam Altman tells OpenAI staff decisions are up to government" and "Sam Altman says operational decisions up to US government" are correctly caught as the same story.
+
+- **`quality_score.py` integration** — After within-batch dedup (80% threshold), a new `cross_scan_dedup()` step filters candidates against the SQLite database before they reach the LLM editor.
+
+- **Seeding from history** — Run `python3 dedup_db.py --seed` to import your existing `news_log.md` and `scanner_presented.md` into the database. This gives the dedup system historical context from day one.
+
+### Major: LLM Failover Chain
+
+**Problem:** If the Gemini API was down or timed out, the pipeline fell back to raw scored articles with no editorial curation — often resulting in low-quality or off-topic picks.
+
+**What changed:**
+
+- **3-tier failover chain** in `llm_editor.py`:
+  1. **Gemini 3.1 Flash Lite** (primary — cheapest)
+  2. **Grok 4.1 Fast via OpenRouter** (different provider — avoids double failure if Google is down)
+  3. **Gemini 3 Flash Preview** (last resort)
+
+- The chain intentionally alternates providers. If Google's API fails, the pipeline hits Grok (OpenRouter) on the second try instead of wasting another timeout on a second Google model.
+
+- **Removed raw fallback.** If all 3 LLM providers fail, the pipeline now prints a clean error message and points you to the saved candidates file for a manual re-run — instead of dumping unfiltered articles.
+
+- **New env var: `OPENROUTER_API_KEY`** — Required if you want the Grok failover. Without it, slot 2 is skipped and the chain degrades to Flash Lite → Flash Preview (both Google).
+
+### Major: AI Keyword Pre-Filter
+
+**Problem:** Non-AI articles (sports, energy drinks, phone reviews) were leaking into the candidate pool from RSS feeds, wasting LLM tokens and sometimes slipping through the AI editor's curation.
+
+**What changed:**
+
+- **Inline keyword filter in `news_scan_deduped.sh`** — Applied during RSS extraction, before scoring. Uses word-bounded short keywords (`\bAI\b`, `\bLLM\b`, `\bGPU\b`, etc.) and substring long keywords (`OpenAI`, `Anthropic`, `machine learning`, etc.) to filter articles.
+
+- Word-boundary matching prevents false positives: "foreign affairs" does not match `\bAI\b`, but "new AI system" does.
+
+- Non-AI articles are counted and reported (e.g., "Filtered 57 non-AI articles") but silently dropped from the pipeline.
+
+- The old `filter_ai_news.sh` script still exists for standalone use, but the main pipeline now handles keyword filtering inline.
+
+### Changed: Quality Over Quantity
+
+**Problem:** The editorial rule "Every scan MUST produce at least 5 stories" pressured the LLM to pad results with mediocre or off-topic picks when the candidate pool was thin.
+
+**What changed:**
+
+- **Rule #1 updated** in the editorial profile template and the LLM prompt:
+  > Select UP TO 7 stories per scan. Quality matters more than quantity — 3 great picks are better than 7 mediocre ones. Only select stories that genuinely match the editorial focus. It is perfectly fine to return fewer stories when the candidate pool is thin.
+
+- The LLM prompt now says "UP TO N" instead of "EXACTLY N".
+
+### Added: Unit Test Suite
+
+- **New: `test_components.py`** — 68 tests covering all pipeline components:
+  - `dedup_db.py`: URL normalization (10 tests), DB operations + bulk check (12 tests)
+  - `quality_score.py`: scoring logic (4), within-batch dedup (2), cross-scan dedup (1)
+  - `llm_editor.py`: failover chain config (5), validate_picks (5), prompt wording (2), JSON parsing (5), SQLite pre-filter (1)
+  - AI keyword filter: AI-relevant titles (10), non-AI titles (8), edge cases (3)
+
+- Run with: `cd scripts && python3 test_components.py`
+
+### Migration Guide (v1 → v2)
+
+1. **Copy the new and updated scripts** to your workspace:
+   ```bash
+   cp scripts/dedup_db.py ~/.openclaw/workspace/scripts/
+   cp scripts/test_components.py ~/.openclaw/workspace/scripts/
+   cp scripts/quality_score.py ~/.openclaw/workspace/scripts/
+   cp scripts/llm_editor.py ~/.openclaw/workspace/scripts/
+   cp scripts/news_scan_deduped.sh ~/.openclaw/workspace/scripts/
+   chmod +x ~/.openclaw/workspace/scripts/news_scan_deduped.sh
+   ```
+
+2. **Seed the dedup database** from your existing logs:
+   ```bash
+   cd ~/.openclaw/workspace/scripts
+   python3 dedup_db.py --seed
+   python3 dedup_db.py --stats
+   ```
+
+3. **Set `OPENROUTER_API_KEY`** (optional, for Grok failover):
+   Add to your LaunchAgent plist or export in shell.
+
+4. **Update your editorial profile** — replace rule #1 with the new quality-over-quantity wording from `config/editorial_profile_template.md`.
+
+5. **Verify everything works:**
+   ```bash
+   cd ~/.openclaw/workspace/scripts
+   python3 test_components.py          # 68 tests should pass
+   ./news_scan_deduped.sh --top 5      # manual pipeline test
+   ```
+
+---
+
+## [v1] — 2026-02-28
+
+Initial release. 5-source news scanning pipeline with quality scoring, article enrichment, and Gemini Flash editorial curation.
@@ -23,9 +23,9 @@

 ---

-A complete, automated AI news scanning pipeline for [OpenClaw](https://github.com/openclaw/openclaw). Scans 5 data sources every 2 hours, scores and deduplicates results, enriches top articles with full text, and uses Gemini Flash as an AI editor to curate the best stories for your channel.
+A complete, automated AI news scanning pipeline for [OpenClaw](https://github.com/openclaw/openclaw). Scans 5 data sources every 2 hours, scores and deduplicates results with a persistent SQLite database, enriches top articles with full text, and uses a 3-tier LLM failover chain (Gemini Flash Lite → Grok via OpenRouter → Gemini Flash) to curate the best stories for your channel.

-**Pipeline cost:** ~$5/month (Gemini Flash API + Tavily free tier)
+**Pipeline cost:** ~$5/month (Gemini Flash Lite API + Tavily free tier)

 ---

@@ -38,9 +38,9 @@ OpenClaw Gateway
 ├── Cron scheduler fires every 2 hours
 │   └── Runs news_scan_deduped.sh (the orchestrator)
 │       ├── Calls 5 data source scripts (RSS, Reddit, Twitter, GitHub, Tavily)
-│       ├── Scores + deduplicates via quality_score.py
+│       ├── Scores + deduplicates via quality_score.py + dedup_db.py
 │       ├── Enriches top articles via enrich_top_articles.py
-│       └── Curates via llm_editor.py (Gemini Flash API)
+│       └── Curates via llm_editor.py (3-tier LLM failover)
 │
 ├── Agent receives the pipeline output
 │   └── Formats and delivers to your channel (Telegram, Slack, etc.)
@@ -49,6 +49,7 @@ OpenClaw Gateway
 │   └── Runs update_editorial_profile.py to learn from your approvals/rejections
 │
 └── memory/ directory
+    ├── news_dedup.db             ← SQLite dedup database (cross-scan)
    ├── editorial_profile.md      ← LLM editor reads this for guidance
    ├── editorial_decisions.md    ← Your approval/rejection log
    ├── scanner_presented.md      ← Auto-logged: what was presented
@@ -62,7 +63,7 @@ OpenClaw Gateway
 1. **Scripts live in** `~/.openclaw/workspace/scripts/` — OpenClaw's standard location for agent-callable scripts
 2. **Memory files live in** `~/.openclaw/workspace/memory/` — persistent across sessions
 3. **The cron job** uses `sessionTarget: "isolated"` so each scan gets a clean session (no context contamination)
-4. **The agent model** (e.g., Kimi K2.5) orchestrates the pipeline. The actual AI curation uses Gemini Flash directly via API — so your cron model doesn't need to be expensive
+4. **The agent model** (e.g., Kimi K2.5) orchestrates the pipeline. The actual AI curation uses a 3-tier LLM failover chain (Gemini Flash Lite → Grok → Gemini Flash) via direct API calls — so your cron model doesn't need to be expensive
 5. **Delivery** is handled by OpenClaw's channel system (Telegram, Slack, etc.)

 **Not using OpenClaw?** The scripts work standalone too — just run `./news_scan_deduped.sh` from a regular cron job or shell. The only OpenClaw-specific parts are the cron job setup and channel delivery.
@@ -77,7 +78,7 @@ OpenClaw Gateway
 │                    (Main Orchestrator)                           │
 ├─────────────────────────────────────────────────────────────────┤
 │                                                                 │
-│  [1] RSS Feeds          ──→  filter_ai_news.sh (25 feeds)       │
+│  [1] RSS Feeds          ──→  inline AI keyword filter (25 feeds) │
 │  [2] Reddit JSON API    ──→  fetch_reddit_news.py (13 subs)     │
 │  [3] Twitter/X          ──→  scan_twitter_ai.sh (bird CLI)      │
 │                          ──→  fetch_twitter_api.py (API search)  │
@@ -88,16 +89,21 @@ OpenClaw Gateway
 │                                                                 │
 ├─────────────────────────────────────────────────────────────────┤
 │                                                                 │
-│  quality_score.py   → Score + dedup (80% title similarity)      │
+│  dedup_db.py        → SQLite cross-scan dedup (URL + title)     │
+│                       Persistent memory across all runs         │
+│                                                                 │
+│  quality_score.py   → Score + within-batch dedup (80%)          │
+│                       + cross-scan dedup via SQLite             │
 │                       Output: top 50 scored candidates          │
 │                                                                 │
 │  enrich_top_articles.py → Fetch full text for top 8 articles    │
 │                           CF Markdown preferred, HTML fallback  │
 │                                                                 │
-│  llm_editor.py      → Gemini Flash editorial curation           │
+│  llm_editor.py      → 3-tier LLM failover chain                │
+│                       Flash Lite → Grok (OpenRouter) → Flash   │
 │                       Reads editorial_profile.md for guidance   │
-│                       Checks news_log.md to avoid repeats       │
-│                       Output: top 7 ranked picks (JSON)         │
+│                       SQLite pre-filter before LLM call         │
+│                       Output: up to 7 ranked picks (JSON)       │
 │                                                                 │
 └─────────────────────────────────────────────────────────────────┘
 ```
@@ -114,7 +120,8 @@ OpenClaw Gateway
 ### API Keys (set as environment variables)
 | Key | Required? | Purpose | Free Tier |
 |-----|-----------|---------|-----------|
-| `GEMINI_API_KEY` | Yes | Gemini Flash for LLM editorial curation | Google AI Studio — generous free tier |
+| `GEMINI_API_KEY` | Yes | Gemini Flash Lite / Flash for LLM curation | Google AI Studio — generous free tier |
+| `OPENROUTER_API_KEY` | Recommended | Grok 4.1 Fast failover (via OpenRouter) | Pay-per-token (cheap) |
 | `GH_TOKEN` | Recommended | GitHub API (5000 req/h vs 60/h unauthenticated) | GitHub personal access token (free) |
 | `TAVILY_API_KEY` | Optional | Tavily web search for breaking news | 1000 queries/month free |
 | `TWITTERAPI_IO_KEY` | Optional | twitterapi.io keyword search supplement | Paid (small monthly fee) |
@@ -184,7 +191,19 @@ Edit `~/.openclaw/workspace/memory/editorial_profile.md` to reflect your channel

 This profile is read by the LLM editor on every scan and directly influences story selection.

-### Step 4: Set Environment Variables
+### Step 4: Seed the Dedup Database
+
+Import your existing post history so the dedup system has context from day one:
+
+```bash
+cd ~/.openclaw/workspace/scripts
+python3 dedup_db.py --seed
+python3 dedup_db.py --stats
+```
+
+If this is a fresh install with no history, skip this step — the database will populate automatically as the pipeline runs.
+
+### Step 5: Set Environment Variables

 Add API keys to your OpenClaw LaunchAgent plist (macOS):

@@ -192,6 +211,8 @@ Add API keys to your OpenClaw LaunchAgent plist (macOS):
 # Add to ~/Library/LaunchAgents/ai.openclaw.gateway.plist under EnvironmentVariables:
 # <key>GEMINI_API_KEY</key>
 # <string>your-gemini-api-key</string>
+# <key>OPENROUTER_API_KEY</key>
+# <string>your-openrouter-key</string>
 # <key>GH_TOKEN</key>
 # <string>your-github-token</string>
 # <key>TAVILY_API_KEY</key>
@@ -207,11 +228,12 @@ Or export them in your shell for testing:

 ```bash
 export GEMINI_API_KEY="your-key"
+export OPENROUTER_API_KEY="your-key"
 export GH_TOKEN="your-token"
 export TAVILY_API_KEY="your-key"
 ```

-### Step 5: Create the Cron Job
+### Step 6: Create the Cron Job

 Add the news scan as an OpenClaw cron job:

@@ -231,7 +253,7 @@ openclaw cron add \

 **Model choice:** The cron job uses a cheap/mid-tier model (like Kimi K2.5) to orchestrate the pipeline. The actual AI curation happens via Gemini Flash API directly (called by `llm_editor.py`), so the cron model doesn't need to be expensive.

-### Step 6: Test the Pipeline
+### Step 7: Test the Pipeline

 Run a manual test:

@@ -260,9 +282,11 @@ You should see output like:
 ### 1. `news_scan_deduped.sh` — Main Orchestrator
 The master script that calls everything else in sequence. Collects articles from all 5 sources, pipes through scoring/enrichment/LLM, and formats output. All sources are best-effort — if one fails, the pipeline continues with what it has.

-### 2. `filter_ai_news.sh` — RSS Keyword Filter
+### 2. `filter_ai_news.sh` — RSS Keyword Filter (standalone)
 Reads articles from blogwatcher, filters by AI-related keywords (with word-boundary matching for short keywords like "AI" to avoid false positives), assigns source tiers, and filters out Reddit noise (questions, rants, memes).

+> **Note:** As of v2, the main orchestrator (`news_scan_deduped.sh`) handles AI keyword filtering inline during RSS extraction. This script still exists for standalone use or debugging, but is no longer called by the pipeline.
+
 ### 3. `fetch_reddit_news.py` — Reddit JSON API Scanner
 Fetches posts from 13 AI-related subreddits using Reddit's public JSON API (no auth needed). Features:
 - Per-subreddit score thresholds (30-50 upvotes minimum)
@@ -290,29 +314,37 @@ Maintains state between runs to calculate star velocity.
 ### 7. `fetch_web_news.py` — Tavily Web Search
 Catches breaking news that RSS feeds miss. 5 focused queries, 2-day freshness filter. Skips domains already covered by RSS (Reddit, Twitter, GitHub, YouTube, arxiv). Filters out homepage URLs.

-### 8. `quality_score.py` — Scoring + Deduplication
+### 8. `dedup_db.py` — SQLite Cross-Scan Dedup Database
+Persistent dedup memory shared across all pipeline runs. Stores normalized URLs and titles from every scan in `~/.openclaw/workspace/memory/news_dedup.db`. Features:
+- URL normalization (strips query params, fragments, www prefix, trailing punctuation)
+- Title similarity matching (75% threshold via SequenceMatcher, 2-day window)
+- Bulk check API for efficient pre-filtering
+- CLI for seeding from historical logs, checking URLs/titles, and viewing stats
+
+### 9. `quality_score.py` — Scoring + Deduplication
 Scores every article based on:
 - Source tier (wire services get +5, tech press +3, etc.)
 - High-value keywords (acquisitions, billion, launch, security, etc.)
 - Breaking news signals (exclusive, confirmed, first look, etc.)
 - Title quality (length heuristic)

-Deduplicates by title similarity (80% threshold using SequenceMatcher). Outputs top 50.
+Two-stage dedup: within-batch similarity (80% threshold) followed by cross-scan dedup against the SQLite database. Outputs top 50.

-### 9. `enrich_top_articles.py` — Full Text Fetcher
+### 10. `enrich_top_articles.py` — Full Text Fetcher
 Fetches full article text for the top 8 scored articles. Tries Cloudflare Markdown for Agents first (clean markdown), falls back to HTML extraction. Skips paywalled sites. 1200 character cap per article.

-### 10. `llm_editor.py` — LLM Editorial Curation
-The AI brain of the pipeline. Sends all scored candidates + editorial profile + recent post history to Gemini Flash. The LLM selects the top N stories, ranks them, assigns categories, and writes 1-sentence summaries.
+### 11. `llm_editor.py` — LLM Editorial Curation
+The AI brain of the pipeline. Sends all scored candidates + editorial profile + recent post history to a 3-tier LLM failover chain. The LLM selects up to 7 stories, ranks them, assigns categories, and writes 1-sentence summaries.

 Features:
- Deterministic URL pre-filter (skips already-posted URLs before calling the LLM)
+- **3-tier failover chain:** Gemini 3.1 Flash Lite → Grok 4.1 Fast (OpenRouter) → Gemini 3 Flash Preview. Alternates providers to avoid double failure.
+- SQLite pre-filter (skips already-seen URLs and similar titles before calling the LLM)
 - Editorial profile integration (learns your preferences over time)
- Structured JSON output with validation
- Graceful fallback to raw scoring if LLM fails
+- Structured JSON output with validation and robust parsing (handles markdown fences, dict wrappers, etc.)
+- Records all picks to the SQLite dedup database after selection
 - Logs all presented stories to `scanner_presented.md`

-### 11. `update_editorial_profile.py` — Profile Updater
+### 12. `update_editorial_profile.py` — Profile Updater
 Runs nightly. Analyzes your approval/rejection patterns and updates the editorial profile's stats section. Also identifies "blind spots" — topics you manually seek out but the scanner doesn't catch.

 ---
@@ -343,8 +375,8 @@ Add to the `RELEASE_REPOS` list in `github_trending.py`:
 "owner/repo-name",
 ```

-### Changing the LLM Model
-Edit `GEMINI_MODEL` in `llm_editor.py`. Any Gemini model works. Flash is recommended for cost.
+### Changing the LLM Models
+Edit the `FAILOVER_CHAIN` list in `llm_editor.py`. Each entry specifies a model name, API type (`gemini` or `openrouter`), environment variable for the API key, and timeout. The chain is tried in order — the first provider that responds wins.

 ### Adjusting Scan Frequency
 Edit the cron expression:
@@ -359,18 +391,21 @@ openclaw cron edit <job-id> --cron "0 */3 * * *"  # every 3 hours
 ```
 openclaw-news-scan/
 ├── README.md                              # This file
+├── CHANGELOG.md                           # Version history and migration guide
 ├── scripts/
-│   ├── news_scan_deduped.sh              # Main orchestrator
-│   ├── filter_ai_news.sh                 # RSS keyword filter
+│   ├── news_scan_deduped.sh              # Main orchestrator (inline AI filter)
+│   ├── dedup_db.py                       # SQLite cross-scan dedup database
+│   ├── quality_score.py                  # Scoring + two-stage dedup
+│   ├── enrich_top_articles.py            # Full text fetcher
+│   ├── llm_editor.py                     # LLM curation (3-tier failover)
+│   ├── filter_ai_news.sh                 # RSS keyword filter (standalone)
 │   ├── fetch_reddit_news.py              # Reddit JSON API
 │   ├── scan_twitter_ai.sh               # Twitter bird CLI
 │   ├── fetch_twitter_api.py              # twitterapi.io search
 │   ├── github_trending.py               # GitHub trending + releases
 │   ├── fetch_web_news.py                # Tavily web search
-│   ├── quality_score.py                 # Scoring + dedup
-│   ├── enrich_top_articles.py           # Full text fetcher
-│   ├── llm_editor.py                    # LLM editorial curation
-│   └── update_editorial_profile.py      # Editorial profile updater
+│   ├── update_editorial_profile.py      # Editorial profile updater
+│   └── test_components.py               # Unit tests (68 tests)
 └── config/
    └── editorial_profile_template.md     # Template — customize for your channel
 ```
@@ -380,14 +415,14 @@ openclaw-news-scan/
 ## Pipeline Flow Summary

 ```
-RSS (25 feeds) ─────────┐
-Reddit (13 subs) ───────┤
-Twitter (bird + API) ───┤──→ quality_score.py ──→ enrich_top_articles.py ──→ llm_editor.py ──→ Output
-GitHub (trending+rel) ──┤       (max 50)              (max 8)              (Gemini Flash)
-Tavily (5 queries) ─────┘
+RSS (25 feeds) ─────────┐                                                   ┌─ Gemini Flash Lite
+Reddit (13 subs) ───────┤    AI keyword     quality_score.py   enrich_top   │
+Twitter (bird + API) ───┤──→ pre-filter ──→ + dedup_db.py  ──→ articles ──→ ├─ Grok (OpenRouter) ──→ Output
+GitHub (trending+rel) ──┤   (inline)        (score + dedup)    (max 8)      │  (failover chain)
+Tavily (5 queries) ─────┘                    (max 50)                       └─ Gemini Flash Preview
 ```

-**Typical run:** ~100 raw articles → 50 scored → 8 enriched → 5-7 curated picks
+**Typical run:** ~100 raw → ~50 after AI filter → 50 scored → 8 enriched → 3-7 curated picks

 ---

@@ -395,7 +430,8 @@ Tavily (5 queries) ─────┘

 | Component | Monthly Cost | Notes |
 |-----------|-------------|-------|
-| Gemini Flash API | ~$2-3/month | ~7 calls/day, ~30K tokens each |
+| Gemini Flash Lite API | ~$1-2/month | Primary LLM — ~7 calls/day, ~30K tokens each |
+| OpenRouter (Grok failover) | ~$0-1/month | Only used when Gemini fails |
 | Tavily API | Free | 1000 queries/month free tier covers it |
 | GitHub API | Free | Personal access token, 5000 req/h |
 | twitterapi.io | ~$10/month | Optional — bird CLI is free |
@@ -408,15 +444,17 @@ Tavily (5 queries) ─────┘

 | Issue | Fix |
 |-------|-----|
-| "GEMINI_API_KEY not set" | Add to LaunchAgent plist or export in shell |
+| "GEMINI_API_KEY not set" | Add to LaunchAgent plist or export in shell. Pipeline warns but continues (failover may use OpenRouter). |
 | Reddit 429 (rate limit) | Normal with 2h spacing. Reduce subreddits or increase --hours |
 | Reddit 404 on a sub | Sub may be private/quarantined. Remove from config. |
 | bird CLI not found | Install bird or remove scan_twitter_ai.sh call |
 | "No new stories found" | RSS feeds may all be read. Wait for new articles. |
-| LLM editor timeout | Increase TIMEOUT_SEC in llm_editor.py |
+| All LLM providers failed | Check that `GEMINI_API_KEY` and/or `OPENROUTER_API_KEY` are set. The pipeline saves candidates to a file for manual re-run. |
+| LLM editor timeout | Increase timeout values in the `FAILOVER_CHAIN` in `llm_editor.py` |
 | Pipeline takes too long | Increase cron timeout: `openclaw cron edit <id> --timeout 120` |
 | GitHub rate limit | Set GH_TOKEN env var for 5000 req/h (vs 60/h) |
-| Duplicate stories | Adjust --dedup-threshold in quality_score.py (default 0.80) |
+| Duplicate stories | SQLite dedup handles this automatically. Run `python3 dedup_db.py --seed` to import historical posts. Check DB status: `python3 dedup_db.py --stats` |
+| Non-AI articles leaking | The inline AI keyword filter should catch these. Check the keyword patterns in `news_scan_deduped.sh` and add missing terms. |

 ---

@@ -56,7 +56,7 @@
 - Tier 5 (Primary): Company blogs, GitHub repos, research papers

 ## Story Selection Rules
-1. Every scan MUST produce at least 5 stories (broaden scope if needed)
+1. Select UP TO 7 stories per scan. Quality matters more than quantity — 3 great picks are better than 7 mediocre ones. Only select stories that genuinely match the editorial focus. It is perfectly fine to return fewer stories when the candidate pool is thin.
 2. No exact duplicates of previously posted stories
 3. Same event from different angles = OK if the angle is genuinely new
 4. Prefer concrete news (X acquired Y, X launched Z) over speculation
@@ -0,0 +1,370 @@
+#!/usr/bin/env python3
+"""
+dedup_db.py — SQLite-backed dedup database for the news scan pipeline.
+
+Shared by quality_score.py and llm_editor.py. Stores normalized URLs
+and titles from every scan to prevent cross-scan duplicates.
+
+Database: ~/.openclaw/workspace/memory/news_dedup.db
+
+Usage as module:
+    from dedup_db import DedupDB
+    db = DedupDB()
+    if db.is_seen(url):
+        print("duplicate!")
+    db.record(url, title, source, status="presented")
+
+Usage as CLI (seed from logs):
+    python3 dedup_db.py --seed
+    python3 dedup_db.py --stats
+    python3 dedup_db.py --check-url "https://example.com/article"
+"""
+
+import os
+import re
+import sqlite3
+import sys
+import argparse
+from datetime import datetime, timedelta
+from difflib import SequenceMatcher
+from pathlib import Path
+from typing import List, Tuple, Optional, Dict, Set
+from urllib.parse import urlparse, urlunparse
+
+# ── Paths ────────────────────────────────────────────────────────────
+WORKSPACE = Path(os.environ.get("OPENCLAW_WORKSPACE",
+                                os.path.expanduser("~/.openclaw/workspace")))
+DB_PATH = WORKSPACE / "memory" / "news_dedup.db"
+NEWS_LOG = WORKSPACE / "memory" / "news_log.md"
+SCANNER_PRESENTED = WORKSPACE / "memory" / "scanner_presented.md"
+
+
+# ── URL normalization ────────────────────────────────────────────────
+
+def normalize_url(url):
+    """
+    Normalize a URL for dedup comparison:
+    - Strip query parameters and fragments
+    - Remove www. prefix
+    - Normalize to https://
+    - Remove trailing slashes
+    - Lowercase domain
+    """
+    if not url:
+        return ""
+
+    url = url.strip().rstrip(".,;:)")
+
+    try:
+        parsed = urlparse(url)
+    except Exception:
+        return url.lower()
+
+    # Normalize scheme to https
+    scheme = "https"
+
+    # Lowercase and strip www from domain
+    netloc = parsed.netloc.lower()
+    if netloc.startswith("www."):
+        netloc = netloc[4:]
+
+    # Keep path, strip trailing slash (but keep "/" for root)
+    path = parsed.path.rstrip("/") if parsed.path != "/" else "/"
+
+    # Drop query params and fragment entirely
+    normalized = urlunparse((scheme, netloc, path, "", "", ""))
+
+    return normalized
+
+
+# ── Database class ───────────────────────────────────────────────────
+
+class DedupDB:
+    """SQLite-backed dedup database."""
+
+    def __init__(self, db_path=None):
+        # type: (Optional[str]) -> None
+        self.db_path = db_path or str(DB_PATH)
+        self._ensure_db()
+
+    def _ensure_db(self):
+        """Create database and tables if they don't exist."""
+        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("""
+            CREATE TABLE IF NOT EXISTS seen_articles (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                url_normalized TEXT NOT NULL,
+                url_original TEXT NOT NULL,
+                title TEXT NOT NULL,
+                source TEXT DEFAULT '',
+                status TEXT DEFAULT 'presented',
+                first_seen TEXT NOT NULL,
+                scan_id TEXT DEFAULT ''
+            )
+        """)
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_url_norm
+            ON seen_articles(url_normalized)
+        """)
+        conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_first_seen
+            ON seen_articles(first_seen)
+        """)
+        conn.commit()
+        conn.close()
+
+    def _connect(self):
+        return sqlite3.connect(self.db_path)
+
+    def is_seen(self, url):
+        """Check if a normalized URL already exists in the database."""
+        norm = normalize_url(url)
+        if not norm:
+            return False
+        conn = self._connect()
+        cursor = conn.execute(
+            "SELECT 1 FROM seen_articles WHERE url_normalized = ? LIMIT 1",
+            (norm,)
+        )
+        found = cursor.fetchone() is not None
+        conn.close()
+        return found
+
+    def find_similar_titles(self, title, threshold=0.75, days=7):
+        """
+        Find titles in the DB similar to the given title.
+        Only checks articles from the last N days for performance.
+        Returns list of (db_title, similarity_score, url_normalized).
+        """
+        if not title:
+            return []
+
+        cutoff = (datetime.now() - timedelta(days=days)).isoformat()
+        conn = self._connect()
+        cursor = conn.execute(
+            "SELECT title, url_normalized FROM seen_articles WHERE first_seen > ?",
+            (cutoff,)
+        )
+        rows = cursor.fetchall()
+        conn.close()
+
+        matches = []
+        title_lower = title.lower()
+        for db_title, db_url in rows:
+            sim = SequenceMatcher(None, title_lower, db_title.lower()).ratio()
+            if sim >= threshold:
+                matches.append((db_title, sim, db_url))
+
+        matches.sort(key=lambda x: -x[1])
+        return matches
+
+    def record(self, url, title, source="", status="presented", scan_id=""):
+        """Record an article in the database."""
+        norm = normalize_url(url)
+        if not norm:
+            return
+        now = datetime.now().isoformat()
+        conn = self._connect()
+        conn.execute(
+            """INSERT INTO seen_articles
+               (url_normalized, url_original, title, source, status, first_seen, scan_id)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (norm, url, title, source, status, now, scan_id)
+        )
+        conn.commit()
+        conn.close()
+
+    def bulk_check(self, articles):
+        """
+        Check a list of article dicts against the database.
+        Returns (new_articles, duplicate_articles, url_dupe_count, title_dupe_count).
+
+        Each article dict must have 'url' and 'title' keys.
+        Checks both URL match and title similarity (>75% over last 2 days).
+        """
+        new = []
+        dupes = []
+        url_dupes = 0
+        title_dupes = 0
+
+        for article in articles:
+            url = article.get("url", "")
+            title = article.get("title", "")
+
+            # Check URL first (fast)
+            if self.is_seen(url):
+                url_dupes += 1
+                dupes.append(article)
+                continue
+
+            # Check title similarity (slower, only last 2 days for speed)
+            similar = self.find_similar_titles(title, threshold=0.75, days=2)
+            if similar:
+                title_dupes += 1
+                dupes.append(article)
+                continue
+
+            new.append(article)
+
+        return new, dupes, url_dupes, title_dupes
+
+    def record_batch(self, articles, status="presented", scan_id=""):
+        """Record multiple articles in a single transaction."""
+        if not articles:
+            return
+        now = datetime.now().isoformat()
+        conn = self._connect()
+        for a in articles:
+            url = a.get("url", "")
+            title = a.get("title", "")
+            source = a.get("source", "")
+            norm = normalize_url(url)
+            if norm:
+                conn.execute(
+                    """INSERT INTO seen_articles
+                       (url_normalized, url_original, title, source, status, first_seen, scan_id)
+                       VALUES (?, ?, ?, ?, ?, ?, ?)""",
+                    (norm, url, title, source, status, now, scan_id)
+                )
+        conn.commit()
+        conn.close()
+
+    def stats(self):
+        """Return database statistics."""
+        conn = self._connect()
+        total = conn.execute("SELECT COUNT(*) FROM seen_articles").fetchone()[0]
+        presented = conn.execute(
+            "SELECT COUNT(*) FROM seen_articles WHERE status='presented'"
+        ).fetchone()[0]
+        published = conn.execute(
+            "SELECT COUNT(*) FROM seen_articles WHERE status='published'"
+        ).fetchone()[0]
+        today = datetime.now().strftime("%Y-%m-%d")
+        today_count = conn.execute(
+            "SELECT COUNT(*) FROM seen_articles WHERE first_seen LIKE ?",
+            (today + "%",)
+        ).fetchone()[0]
+        conn.close()
+        return {
+            "total": total,
+            "presented": presented,
+            "published": published,
+            "today": today_count,
+        }
+
+    def seed_from_logs(self, news_log_path=None, scanner_presented_path=None):
+        """
+        One-time import: parse existing news_log.md and scanner_presented.md
+        to populate the database with historical URLs and titles.
+        """
+        news_log = news_log_path or str(NEWS_LOG)
+        scanner = scanner_presented_path or str(SCANNER_PRESENTED)
+        imported = 0
+
+        # Parse news_log.md
+        # Format: DATE | POSTED | TITLE | msg_id:NNN | t.me_url | article_url
+        try:
+            with open(news_log, "r") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line or line.startswith("#"):
+                        continue
+                    parts = line.split("|")
+                    if len(parts) >= 6:
+                        title = parts[2].strip()
+                        article_url = parts[5].strip()
+                        if article_url and not article_url.startswith("http"):
+                            continue
+                        if title and article_url:
+                            if not self.is_seen(article_url):
+                                self.record(article_url, title, status="published")
+                                imported += 1
+        except FileNotFoundError:
+            pass
+
+        # Parse scanner_presented.md
+        # Format: [TIMESTAMP] TITLE | URL
+        url_pattern = re.compile(r'https?://[^\s|>\]\)"\']+')
+        try:
+            with open(scanner, "r") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line or line.startswith("#"):
+                        continue
+                    # Extract timestamp, title, URL
+                    m = re.match(r'\[([^\]]+)\]\s*(.+)', line)
+                    if not m:
+                        continue
+                    rest = m.group(2)
+                    parts = rest.split("|")
+                    title = parts[0].strip() if parts else ""
+                    url = parts[1].strip() if len(parts) > 1 else ""
+                    if not url:
+                        urls = url_pattern.findall(rest)
+                        url = urls[0] if urls else ""
+                    url = url.rstrip(".,;:)")
+                    if title and url and url.startswith("http"):
+                        if "t.me/" in url:
+                            continue
+                        if not self.is_seen(url):
+                            self.record(url, title, status="presented")
+                            imported += 1
+        except FileNotFoundError:
+            pass
+
+        return imported
+
+
+# ── CLI ──────────────────────────────────────────────────────────────
+
+def main():
+    parser = argparse.ArgumentParser(description="News dedup database utility")
+    parser.add_argument("--seed", action="store_true",
+                       help="Seed DB from news_log.md and scanner_presented.md")
+    parser.add_argument("--stats", action="store_true",
+                       help="Show database statistics")
+    parser.add_argument("--check-url",
+                       help="Check if a URL has been seen")
+    parser.add_argument("--check-title",
+                       help="Find similar titles in the DB")
+    args = parser.parse_args()
+
+    db = DedupDB()
+
+    if args.seed:
+        count = db.seed_from_logs()
+        print("Seeded %d articles from existing logs" % count)
+        s = db.stats()
+        print("DB stats: %d total, %d published, %d presented" % (
+            s["total"], s["published"], s["presented"]))
+
+    elif args.stats:
+        s = db.stats()
+        print("Total: %d" % s["total"])
+        print("  Published: %d" % s["published"])
+        print("  Presented: %d" % s["presented"])
+        print("  Today: %d" % s["today"])
+
+    elif args.check_url:
+        norm = normalize_url(args.check_url)
+        seen = db.is_seen(args.check_url)
+        print("URL: %s" % args.check_url)
+        print("Normalized: %s" % norm)
+        print("Seen: %s" % seen)
+
+    elif args.check_title:
+        matches = db.find_similar_titles(args.check_title)
+        if matches:
+            print("Found %d similar titles:" % len(matches))
+            for title, sim, url in matches[:5]:
+                print("  %.0f%% | %s | %s" % (sim * 100, title[:80], url))
+        else:
+            print("No similar titles found")
+
+    else:
+        parser.print_help()
+
+
+if __name__ == "__main__":
+    main()
@@ -31,6 +31,12 @@ import urllib.error
 from datetime import datetime
 from pathlib import Path

+try:
+    from dedup_db import DedupDB, normalize_url
+    HAS_DEDUP_DB = True
+except ImportError:
+    HAS_DEDUP_DB = False
+
 # ── Paths (customize to your workspace) ──────────────────────────────
 WORKSPACE = Path(os.environ.get("OPENCLAW_WORKSPACE",
                                os.path.expanduser("~/.openclaw/workspace")))
@@ -48,6 +54,31 @@ GEMINI_URL = (
 TEMPERATURE = 0.3
 TIMEOUT_SEC = 120
 MAX_ARTICLES = 500
+
+# ── Failover LLM chain ──────────────────────────────────────────────
+FAILOVER_CHAIN = [
+    {
+        "name": "Gemini 3.1 Flash Lite",
+        "model": "gemini-3.1-flash-lite-preview",
+        "api": "gemini",
+        "env_key": "GEMINI_API_KEY",
+        "timeout": 120,
+    },
+    {
+        "name": "OpenRouter (Grok 4.1 Fast)",
+        "model": "x-ai/grok-4.1-fast",
+        "api": "openrouter",
+        "env_key": "OPENROUTER_API_KEY",
+        "timeout": 90,
+    },
+    {
+        "name": "Gemini 3 Flash Preview",
+        "model": "gemini-3-flash-preview",
+        "api": "gemini",
+        "env_key": "GEMINI_API_KEY",
+        "timeout": 120,
+    },
+]
 VALID_CATEGORIES = {
    "ai_product", "m_and_a", "model_release", "security", "geopolitics",
    "github_trending", "gaming", "fintech", "hardware", "open_source", "other"
@@ -101,9 +132,18 @@ def load_file_safe(path, tail_lines=None):

 def filter_already_posted(articles):
    """
-    Deterministic URL pre-filter: remove candidates whose URL already
-    appears in news_log.md or scanner_presented.md.
+    Deterministic pre-filter using SQLite dedup database.
+    Falls back to text-file URL matching if dedup_db unavailable.
    """
+    if HAS_DEDUP_DB:
+        db = DedupDB()
+        new, dupes, url_dupes, title_dupes = db.bulk_check(articles)
+        if dupes:
+            log("Pre-filtered %d candidates via SQLite (%d URL, %d title matches)" % (
+                len(dupes), url_dupes, title_dupes))
+        return new
+
+    # Fallback: original text-file matching
    full_log = load_file_safe(NEWS_LOG)
    if not full_log:
        return articles
@@ -115,9 +155,6 @@ def filter_already_posted(articles):
    for text in [full_log, presented_log]:
        for url in url_pattern.findall(text):
            url = url.rstrip(".,;:)")
-            # Skip your own channel links (customize this pattern)
-            # if "t.me/yourchannel" in url:
-            #     continue
            posted_urls.add(url)

    if not posted_urls:
@@ -128,12 +165,12 @@ def filter_already_posted(articles):
    for a in articles:
        candidate_url = a["url"].rstrip(".,;:)")
        if candidate_url in posted_urls:
-            log(f"  PRE-FILTERED (already posted): {a['title'][:60]}")
+            log("  PRE-FILTERED (already posted): %s" % a['title'][:60])
            removed += 1
        else:
            filtered.append(a)

-    log(f"Pre-filtered {removed} candidates (already posted)")
+    log("Pre-filtered %d candidates (already posted)" % removed)
    return filtered


@@ -169,11 +206,11 @@ the top {top_n} stories from the candidate list below.
 {github_text}

 ## Your Task
-Select exactly {top_n} stories from the candidates above. Rank them by
+Select UP TO {top_n} stories from the candidates above. Rank them by
 newsworthiness for the target audience.

 ## Rules
-1. Return EXACTLY {top_n} stories — no more, no fewer.
+1. Return UP TO {top_n} stories. Quality matters more than quantity — 3 great picks are better than 7 mediocre ones.
 2. Do NOT pick stories that duplicate recently posted stories (same event).
   If a candidate covers the SAME EVENT as a recently posted story — even
   from a different source or with a different headline — do NOT pick it.
@@ -187,7 +224,7 @@ newsworthiness for the target audience.
   github_trending, gaming, fintech, hardware, open_source, other

 ## Required JSON Output Format
-Return a JSON array of exactly {top_n} objects, each with these fields:
+Return a JSON array of your selected stories (up to {top_n}), each with these fields:
 [
  {{
    "rank": 1,
@@ -266,37 +303,171 @@ def call_gemini(prompt, api_key):
        return None


-def fallback_picks(articles, github_articles, top_n):
-    log("FALLBACK: Using raw article order (no LLM judgment)")
-    all_candidates = articles.copy()
-    if github_articles:
-        all_candidates.extend(github_articles)
-
-    picks = []
-    seen_sources = {}
-    for a in all_candidates:
-        src = a["source"]
-        if seen_sources.get(src, 0) >= 2:
+def call_llm_with_failover(prompt, articles, github_articles, editorial_profile, recent_posts, top_n):
+    """
+    Try LLM providers in sequence: Gemini Flash -> Gemini Flash Lite -> OpenRouter.
+    Each step may reduce candidate count for speed.
+    """
+    for i, provider in enumerate(FAILOVER_CHAIN):
+        api_key = os.environ.get(provider["env_key"])
+        if not api_key:
+            log("  Skipping %s: %s not set" % (provider["name"], provider["env_key"]))
            continue
-        seen_sources[src] = seen_sources.get(src, 0) + 1
-        if "github.com" in a["url"]:
-            article_type = "github"
-        elif "x.com/" in a.get("url", "") or "twitter.com/" in a.get("url", "") or "X/" in a.get("source", ""):
-            article_type = "twitter"
+
+        log("Trying %s (model: %s, timeout: %ds)" % (
+            provider["name"], provider["model"], provider["timeout"]))
+
+        # For later failovers, reduce candidate list for speed
+        current_articles = articles
+        current_github = github_articles
+        if i >= 1:
+            current_articles = articles[:30]
+            current_github = github_articles[:5] if github_articles else []
+
+        # Rebuild prompt with current candidates
+        current_prompt = build_prompt(
+            current_articles, current_github, editorial_profile, recent_posts, top_n
+        )
+
+        if provider["api"] == "gemini":
+            model_url = (
+                "https://generativelanguage.googleapis.com/v1beta/models/"
+                "%s:generateContent" % provider["model"]
+            )
+            picks = _call_gemini_api(current_prompt, api_key, model_url, provider["timeout"])
+        elif provider["api"] == "openrouter":
+            picks = _call_openrouter_api(current_prompt, api_key, provider["model"], provider["timeout"])
        else:
-            article_type = "rss"
-        picks.append({
-            "rank": len(picks) + 1,
-            "title": a["title"],
-            "url": a["url"],
-            "source": a["source"],
-            "type": article_type,
-            "summary": "(Fallback: no AI summary available)",
-            "category": "other",
-        })
-        if len(picks) >= top_n:
-            break
-    return picks
+            continue
+
+        if picks is not None:
+            log("  %s returned %d picks" % (provider["name"], len(picks)))
+            return picks
+
+        log("  %s failed, trying next..." % provider["name"])
+
+    log("ERROR: All LLM providers failed")
+    return None
+
+
+def _call_gemini_api(prompt, api_key, model_url, timeout):
+    """Call a Gemini API model. Returns parsed picks list or None."""
+    url = "%s?key=%s" % (model_url, api_key)
+
+    payload = {
+        "contents": [{"parts": [{"text": prompt}]}],
+        "generationConfig": {
+            "temperature": TEMPERATURE,
+            "responseMimeType": "application/json",
+        }
+    }
+
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        url, data=data,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+
+    token_est = estimate_tokens(prompt)
+    log("  Sending ~%d tokens to Gemini API" % token_est)
+
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            body = resp.read().decode("utf-8")
+            result = json.loads(body)
+    except urllib.error.HTTPError as e:
+        error_body = e.read().decode("utf-8") if e.fp else "(no body)"
+        log("  Gemini API HTTP error %d: %s" % (e.code, error_body[:500]))
+        return None
+    except urllib.error.URLError as e:
+        log("  Gemini API connection error: %s" % e.reason)
+        return None
+    except Exception as e:
+        log("  Gemini API call failed: %s" % e)
+        return None
+
+    try:
+        text = result["candidates"][0]["content"]["parts"][0]["text"]
+    except (KeyError, IndexError) as e:
+        log("  Unexpected Gemini response structure: %s" % e)
+        return None
+
+    return _parse_llm_json(text)
+
+
+def _call_openrouter_api(prompt, api_key, model, timeout):
+    """Call OpenRouter API. Returns parsed picks list or None."""
+    url = "https://openrouter.ai/api/v1/chat/completions"
+
+    payload = {
+        "model": model,
+        "messages": [{"role": "user", "content": prompt}],
+        "temperature": TEMPERATURE,
+    }
+
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(
+        url, data=data,
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": "Bearer %s" % api_key,
+        },
+        method="POST",
+    )
+
+    token_est = estimate_tokens(prompt)
+    log("  Sending ~%d tokens to OpenRouter (%s)" % (token_est, model))
+
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            body = resp.read().decode("utf-8")
+            result = json.loads(body)
+    except urllib.error.HTTPError as e:
+        error_body = e.read().decode("utf-8") if e.fp else "(no body)"
+        log("  OpenRouter HTTP error %d: %s" % (e.code, error_body[:500]))
+        return None
+    except urllib.error.URLError as e:
+        log("  OpenRouter connection error: %s" % e.reason)
+        return None
+    except Exception as e:
+        log("  OpenRouter call failed: %s" % e)
+        return None
+
+    try:
+        text = result["choices"][0]["message"]["content"]
+    except (KeyError, IndexError) as e:
+        log("  Unexpected OpenRouter response structure: %s" % e)
+        return None
+
+    return _parse_llm_json(text)
+
+
+def _parse_llm_json(text):
+    """Parse LLM response text into a list of picks."""
+    try:
+        picks = json.loads(text)
+        if isinstance(picks, list):
+            return picks
+        if isinstance(picks, dict) and "stories" in picks:
+            return picks["stories"]
+        if isinstance(picks, dict):
+            # Try to find a list value in the dict
+            for v in picks.values():
+                if isinstance(v, list):
+                    return v
+        return None
+    except json.JSONDecodeError:
+        match = re.search(r'\[\s*\{.*?\}\s*\]', text, re.DOTALL)
+        if match:
+            try:
+                picks = json.loads(match.group())
+                if isinstance(picks, list):
+                    return picks
+            except json.JSONDecodeError:
+                pass
+        log("  Could not parse LLM response. First 500 chars: %s" % text[:500])
+        return None


 def validate_picks(picks, top_n):
@@ -362,8 +533,7 @@ def main():

    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
-        log("ERROR: GEMINI_API_KEY environment variable not set")
-        sys.exit(1)
+        log("WARNING: GEMINI_API_KEY not set (failover providers may still work)")

    top_n = int(os.environ.get("TOP_N", "7"))
    log(f"Configuration: top_n={top_n}, model={GEMINI_MODEL}")
@@ -418,18 +588,28 @@ def main():
        print(prompt, file=sys.stderr)
        return

-    picks = call_gemini(prompt, api_key)
+    picks = call_llm_with_failover(
+        prompt, articles, github_articles, editorial_profile, recent_posts, top_n
+    )

    if picks is None:
-        picks = fallback_picks(articles, github_articles, top_n)
-    else:
-        log(f"LLM returned {len(picks)} picks")
-        picks = validate_picks(picks, top_n)
+        log("ERROR: All LLM providers failed. No stories to output.")
+        return 1
+
+    picks = validate_picks(picks, top_n)

    for pick in picks:
        print(json.dumps(pick, ensure_ascii=False))

    log_to_scanner_presented(picks)
+
+    # Record picks to SQLite dedup database
+    if HAS_DEDUP_DB:
+        db = DedupDB()
+        pick_articles = [{"url": p["url"], "title": p["title"], "source": p.get("source", "")} for p in picks]
+        db.record_batch(pick_articles, status="presented")
+        log("Recorded %d picks to dedup database" % len(picks))
+
    log(f"Done. {len(picks)} stories selected.")


@@ -96,8 +96,24 @@ except Exception as e:
    print(f"  Warning: Could not run blogwatcher articles: {e}", file=sys.stderr)
    raw = ""

+# ── AI keyword filter (same logic as filter_ai_news.sh) ──────────
+SHORT_KW = re.compile(r"\b(AI|AGI|LLM|GPU|TPU|RAG|API)\b")
+LONG_KW = re.compile(
+    r"artificial intelligence|machine learning|deep learning|"
+    r"language model|GPT|Claude|Gemini|ChatGPT|OpenAI|Anthropic|"
+    r"Google AI|DeepMind|agentic|neural network|transformer|"
+    r"diffusion|generative AI|gen AI|Llama|Mistral|Hugging Face|"
+    r"inference|training|fine-tuning|open.source|NVIDIA|DeepSeek|"
+    r"Grok|xAI|Qwen|Codex|Copilot|Meta AI|Cohere|Perplexity|"
+    r"multimodal|reasoning model|robotics|autonomous|chip|"
+    r"acquisition|funding|valuation|launch|release|"
+    r"OpenClaw|Amazon Q|Bedrock|benchmark",
+    re.IGNORECASE
+)
+
 lines = raw.split("\n")
 articles = []
+filtered_out = 0
 i = 0

 while i < len(lines):
@@ -115,14 +131,17 @@ while i < len(lines):
            elif next_line.startswith("URL:"):
                url = next_line[4:].strip()
        if title and url:
-            articles.append(f"{title}|{url}|{source}")
+            if SHORT_KW.search(title) or LONG_KW.search(title):
+                articles.append(f"{title}|{url}|{source}")
+            else:
+                filtered_out += 1
    i += 1

 with open(outpath, "w") as f:
    for a in articles:
        f.write(a + "\n")

-print(f"  Extracted {len(articles)} new RSS articles", file=sys.stderr)
+print(f"  Extracted {len(articles)} AI-relevant RSS articles ({filtered_out} non-AI filtered out)", file=sys.stderr)
 ' "$ARTICLES_FILE"

 RSS_COUNT=$(wc -l < "$ARTICLES_FILE" | tr -d ' ')
@@ -313,33 +332,11 @@ echo "════════════════════════
 echo ""

 if [ "$LLM_SUCCESS" = false ] || [ ! -s "$PICKS_FILE" ]; then
-  echo "Warning: LLM curation unavailable — showing raw top articles:"
+  echo "All LLM providers failed. No curated stories to display."
+  echo "Check /tmp/llm_editor.log for details."
  echo ""
-  head -"$TOP_N" "$ENRICHED_FILE" | while IFS='|' read -r title url source rest; do
-    is_tweet=""
-    if echo "$source" | grep -q "(tweet)"; then
-      is_tweet="yes"
-    fi
-    if [ -n "$is_tweet" ]; then
-      echo "  [tweet] $title"
-    else
-      echo "  * $title"
-    fi
-    if [ -n "$url" ]; then
-      if [ -n "$is_tweet" ]; then
-        echo "    View tweet: $url"
-      else
-        echo "    Link: $url"
-      fi
-    fi
-    source_clean=$(echo "$source" | sed 's/ (tweet)//')
-    if [ -n "$source_clean" ]; then
-      echo "    Source: $source_clean"
-    fi
-    echo ""
-    echo "---"
-    echo ""
-  done
+  echo "Candidates were saved to: $PERSISTENT_CANDIDATES"
+  echo "Re-run manually: python3 $SCRIPT_DIR/llm_editor.py --file $PERSISTENT_CANDIDATES"
 else
  python3 -c '
 import sys, json
@@ -398,6 +395,34 @@ for i, line in enumerate(lines):
 ' "$PICKS_FILE"
 fi

+# ═════════════════════════════════════════════════════════════════════
+# RECORD ALL SCORED CANDIDATES TO DEDUP DB
+# ═════════════════════════════════════════════════════════════════════
+if [ -s "$SCORED_FILE" ]; then
+  python3 -c '
+import sys
+sys.path.insert(0, sys.argv[2])
+try:
+    from dedup_db import DedupDB
+    db = DedupDB()
+    articles = []
+    with open(sys.argv[1], "r") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            parts = line.split("|")
+            if len(parts) >= 3:
+                articles.append({"url": parts[1], "title": parts[0], "source": parts[2]})
+    db.record_batch(articles, status="scored")
+    print(f"  Recorded {len(articles)} scored candidates to dedup DB", file=sys.stderr)
+except ImportError:
+    print("  Warning: dedup_db not available, skipping DB recording", file=sys.stderr)
+except Exception as e:
+    print(f"  Warning: DB recording failed: {e}", file=sys.stderr)
+' "$SCORED_FILE" "$SCRIPT_DIR" 2>&1
+fi
+
 # ═════════════════════════════════════════════════════════════════════
 # CLEANUP: Mark articles as read in blogwatcher
 # ═════════════════════════════════════════════════════════════════════
@@ -17,6 +17,12 @@ import re
 import argparse
 from difflib import SequenceMatcher

+try:
+    from dedup_db import DedupDB, normalize_url
+    HAS_DEDUP_DB = True
+except ImportError:
+    HAS_DEDUP_DB = False
+
 # ── Source priority scoring ──────────────────────────────────────────
 # Higher = better. Customize to match your blogwatcher feed names.
 PRIORITY_SOURCES = {
@@ -112,6 +118,29 @@ def deduplicate(articles, threshold=0.80):
    return unique


+def cross_scan_dedup(articles):
+    """Remove articles already seen in previous scans (via SQLite DB)."""
+    if not HAS_DEDUP_DB:
+        print("  Warning: dedup_db not available, skipping cross-scan dedup", file=sys.stderr)
+        return articles
+
+    db = DedupDB()
+    article_dicts = [{"url": a["url"], "title": a["title"]} for a in articles]
+    new_dicts, dupe_dicts, url_dupes, title_dupes = db.bulk_check(article_dicts)
+
+    # Build set of new URLs for filtering
+    new_urls = set()
+    for d in new_dicts:
+        new_urls.add(normalize_url(d["url"]))
+    filtered = [a for a in articles if normalize_url(a["url"]) in new_urls]
+
+    removed = len(articles) - len(filtered)
+    if removed > 0:
+        print("  Cross-scan dedup: removed %d (%d URL, %d title matches)" % (removed, url_dupes, title_dupes), file=sys.stderr)
+
+    return filtered
+
+
 def main():
    parser = argparse.ArgumentParser(description="Quality scoring pre-filter")
    parser.add_argument('--input', '-i', required=True, help='Input pipe-delimited file')
@@ -154,6 +183,7 @@ def main():

    articles.sort(key=lambda x: -x['score'])
    unique = deduplicate(articles, args.dedup_threshold)
+    unique = cross_scan_dedup(unique)
    unique.sort(key=lambda x: -x['score'])
    output = unique[:args.max]

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""Unit tests for all pipeline components."""
+import sys
+import os
+import tempfile
+import json
+import re
+
+PASS = 0
+FAIL = 0
+
+
+def test(name, condition, detail=""):
+    global PASS, FAIL
+    if condition:
+        PASS += 1
+        print("  PASS: %s" % name)
+    else:
+        FAIL += 1
+        print("  FAIL: %s -- %s" % (name, detail))
+
+
+# ==============================================================
+print("=" * 60)
+print("COMPONENT 1: dedup_db.py")
+print("=" * 60)
+
+from dedup_db import DedupDB, normalize_url
+
+# --- URL normalization ---
+print("\n[URL Normalization]")
+
+test("Strip query params",
+     normalize_url("https://bloomberg.com/article?accessToken=abc&ref=123") == "https://bloomberg.com/article")
+test("Strip fragment",
+     normalize_url("https://example.com/page#section") == "https://example.com/page")
+test("Remove www prefix",
+     normalize_url("https://www.cnbc.com/story") == "https://cnbc.com/story")
+test("Normalize http to https",
+     normalize_url("http://example.com/path") == "https://example.com/path")
+test("Remove trailing slash",
+     normalize_url("https://example.com/path/") == "https://example.com/path")
+test("Keep root slash",
+     normalize_url("https://example.com/") == "https://example.com/")
+test("Lowercase domain only",
+     normalize_url("https://WWW.Example.COM/CasePath") == "https://example.com/CasePath")
+test("Strip trailing punctuation",
+     normalize_url("https://example.com/article.,;:)") == "https://example.com/article")
+test("Empty URL returns empty",
+     normalize_url("") == "")
+test("Same article different params normalize equal",
+     normalize_url("https://bloomberg.com/news?tok=aaa") == normalize_url("https://bloomberg.com/news?tok=bbb"))
+
+# --- DB + Bulk Check ---
+print("\n[DB Operations + Bulk Check]")
+
+db_path = os.path.join(tempfile.gettempdir(), "test_unit.db")
+if os.path.exists(db_path):
+    os.remove(db_path)
+db = DedupDB(db_path=db_path)
+
+db.record("https://example.com/story1?utm=abc",
+          "Sam Altman tells OpenAI staff operational decisions up to government", "CNBC")
+db.record("https://bloomberg.com/anthropic-20b",
+          "Anthropic nears 20 billion revenue run rate", "Bloomberg")
+
+test("is_seen exact URL", db.is_seen("https://example.com/story1?utm=abc"))
+test("is_seen normalized (diff params)", db.is_seen("https://example.com/story1?utm=xyz"))
+test("is_seen with www prefix", db.is_seen("https://www.example.com/story1"))
+test("is_seen negative", not db.is_seen("https://example.com/totally-new"))
+
+articles = [
+    {"url": "https://example.com/story1?ref=reddit", "title": "OpenAI news"},
+    {"url": "https://yahoo.com/sam-altman", "title": "Sam Altman tells OpenAI staff decisions are up to US government"},
+    {"url": "https://new.com/fresh", "title": "Google releases brand new Gemini model"},
+    {"url": "https://another.com/new", "title": "NVIDIA announces next-gen chip architecture"},
+]
+new, dupes, url_d, title_d = db.bulk_check(articles)
+test("Bulk: 2 new articles", len(new) == 2, "got %d new" % len(new))
+test("Bulk: 2 dupes caught", len(dupes) == 2, "got %d dupes" % len(dupes))
+test("Bulk: 1 URL dupe", url_d == 1, "got %d" % url_d)
+test("Bulk: 1 title dupe", title_d == 1, "got %d" % title_d)
+
+# Record batch
+batch = [
+    {"url": "https://batch.com/a", "title": "Batch A", "source": "Src"},
+    {"url": "https://batch.com/b", "title": "Batch B", "source": "Src"},
+]
+db.record_batch(batch, status="scored")
+test("Batch A recorded", db.is_seen("https://batch.com/a"))
+test("Batch B recorded", db.is_seen("https://batch.com/b"))
+
+# Stats
+s = db.stats()
+test("Stats total > 0", s["total"] > 0, "total=%d" % s["total"])
+test("Stats keys present", all(k in s for k in ["total", "presented", "published", "today"]))
+
+os.remove(db_path)
+
+# ==============================================================
+print("\n" + "=" * 60)
+print("COMPONENT 2: quality_score.py")
+print("=" * 60)
+
+from quality_score import compute_score, deduplicate, title_similarity
+
+print("\n[Scoring Logic]")
+
+score_t1 = compute_score("OpenAI launches GPT-6", "Reuters Tech", "1")
+score_t3 = compute_score("OpenAI launches GPT-6", "AI News", "3")
+test("T1 source scores higher than T3", score_t1 > score_t3,
+     "T1=%d, T3=%d" % (score_t1, score_t3))
+
+score_breaking = compute_score("BREAKING: OpenAI acquires startup", "TechCrunch AI", "2")
+score_normal = compute_score("OpenAI discusses future plans", "TechCrunch AI", "2")
+test("Breaking news scores higher", score_breaking > score_normal,
+     "breaking=%d, normal=%d" % (score_breaking, score_normal))
+
+score_hv = compute_score("Microsoft acquisition of AI company for billion dollars", "Reuters Tech", "1")
+score_no_hv = compute_score("Company releases quarterly earnings report", "Reuters Tech", "1")
+test("High-value keywords boost score", score_hv > score_no_hv,
+     "hv=%d, no_hv=%d" % (score_hv, score_no_hv))
+
+score_short = compute_score("AI news", "AI News", "3")
+score_good = compute_score("OpenAI launches revolutionary new language model for developers", "AI News", "3")
+test("Short title penalized", score_short < score_good,
+     "short=%d, good=%d" % (score_short, score_good))
+
+print("\n[Within-Batch Dedup]")
+
+articles = [
+    {"title": "OpenAI launches GPT-6", "score": 10},
+    {"title": "OpenAI launches GPT-6 model", "score": 8},
+    {"title": "NVIDIA announces new GPU", "score": 7},
+]
+unique = deduplicate(articles, threshold=0.80)
+test("Within-batch dedup removes near-dup", len(unique) == 2,
+     "got %d (expected 2)" % len(unique))
+test("Higher-scored dupe kept", any(a["score"] == 10 for a in unique))
+
+print("\n[Cross-Scan Dedup]")
+try:
+    from quality_score import cross_scan_dedup
+    test("cross_scan_dedup function exists", True)
+except ImportError:
+    test("cross_scan_dedup function exists", False, "not found")
+
+# ==============================================================
+print("\n" + "=" * 60)
+print("COMPONENT 3: llm_editor.py")
+print("=" * 60)
+
+import llm_editor
+
+print("\n[Failover Chain Config]")
+chain = llm_editor.FAILOVER_CHAIN
+test("3 providers in chain", len(chain) == 3, "got %d" % len(chain))
+test("Primary is Flash Lite", "Flash Lite" in chain[0]["name"],
+     "got %s" % chain[0]["name"])
+test("Second is Grok/OpenRouter",
+     "Grok" in chain[1]["name"] or "OpenRouter" in chain[1]["name"],
+     "got %s" % chain[1]["name"])
+test("Third is Flash Preview", "Flash Preview" in chain[2]["name"],
+     "got %s" % chain[2]["name"])
+test("Slots 1-2 use different providers",
+     chain[0]["env_key"] != chain[1]["env_key"],
+     "both use %s" % chain[0]["env_key"])
+
+print("\n[Validate Picks]")
+raw_picks = [
+    {"rank": 1, "title": "Story A", "url": "https://a.com", "source": "Src",
+     "type": "rss", "summary": "Good", "category": "ai_product"},
+    {"rank": 2, "title": "Story B", "url": "https://b.com", "source": "Src",
+     "type": "invalid_type", "summary": "OK", "category": "bad_category"},
+    {"rank": 3, "title": "Story C", "url": "https://c.com"},
+]
+validated = llm_editor.validate_picks(raw_picks, 3)
+test("Validate: 3 picks returned", len(validated) == 3, "got %d" % len(validated))
+test("Validate: invalid type fixed to rss", validated[1]["type"] == "rss",
+     "got %s" % validated[1]["type"])
+test("Validate: invalid category fixed to other", validated[1]["category"] == "other",
+     "got %s" % validated[1]["category"])
+test("Validate: missing fields filled", validated[2]["source"] == "unknown")
+test("Validate: ranks renumbered 1-3",
+     [v["rank"] for v in validated] == [1, 2, 3])
+
+print("\n[Prompt Wording]")
+prompt = llm_editor.build_prompt(
+    [{"title": "Test", "url": "https://x.com", "source": "S"}],
+    [], "Editorial profile", "Recent posts", 5
+)
+test("Prompt says UP TO (not EXACTLY)",
+     "UP TO 5" in prompt and "EXACTLY" not in prompt,
+     "still says EXACTLY" if "EXACTLY" in prompt else "UP TO not found")
+test("Prompt mentions quality", "quality" in prompt.lower())
+
+print("\n[Parse LLM JSON]")
+test("Parse valid array",
+     llm_editor._parse_llm_json('[{"rank":1}]') == [{"rank": 1}])
+test("Parse dict with stories key",
+     llm_editor._parse_llm_json('{"stories":[{"rank":1}]}') == [{"rank": 1}])
+test("Parse dict with arbitrary list value",
+     llm_editor._parse_llm_json('{"results":[{"rank":1}]}') == [{"rank": 1}])
+test("Parse with markdown fences",
+     llm_editor._parse_llm_json('```json\n[{"rank":1}]\n```') == [{"rank": 1}])
+test("Parse garbage returns None",
+     llm_editor._parse_llm_json("this is not json at all") is None)
+
+print("\n[SQLite Pre-Filter]")
+test("HAS_DEDUP_DB is True", llm_editor.HAS_DEDUP_DB is True)
+
+# ==============================================================
+print("\n" + "=" * 60)
+print("COMPONENT 4: AI Keyword Filter (from news_scan_deduped.sh)")
+print("=" * 60)
+
+SHORT_KW = re.compile(r"\b(AI|AGI|LLM|GPU|TPU|RAG|API)\b")
+LONG_KW = re.compile(
+    r"artificial intelligence|machine learning|deep learning|"
+    r"language model|GPT|Claude|Gemini|ChatGPT|OpenAI|Anthropic|"
+    r"Google AI|DeepMind|agentic|neural network|transformer|"
+    r"diffusion|generative AI|gen AI|Llama|Mistral|Hugging Face|"
+    r"inference|training|fine-tuning|open.source|NVIDIA|DeepSeek|"
+    r"Grok|xAI|Qwen|Codex|Copilot|Meta AI|Cohere|Perplexity|"
+    r"multimodal|reasoning model|robotics|autonomous|chip|"
+    r"acquisition|funding|valuation|launch|release|"
+    r"OpenClaw|Amazon Q|Bedrock|benchmark",
+    re.IGNORECASE
+)
+
+
+def is_ai(title):
+    return bool(SHORT_KW.search(title) or LONG_KW.search(title))
+
+
+print("\n[Should PASS filter (AI-relevant)]")
+ai_titles = [
+    "OpenAI launches GPT-6 with breakthrough reasoning",
+    "NVIDIA announces next-gen AI chip",
+    "New LLM benchmark shows surprising results",
+    "Anthropic raises 10 billion in funding round",
+    "Google releases Gemini 4 multimodal model",
+    "Meta AI open-sources Llama 5",
+    "DeepSeek releases new reasoning model",
+    "autonomous driving AI reaches level 4",
+    "GPU shortage impacts cloud AI providers",
+    "New open source transformer beats proprietary models",
+]
+for t in ai_titles:
+    test("AI: %s" % t[:50], is_ai(t))
+
+print("\n[Should FAIL filter (non-AI)]")
+non_ai_titles = [
+    "Best energy drinks for gamers in 2026",
+    "Messi stadium deal falls through",
+    "Samsung Galaxy S27 review - best phone yet",
+    "Netflix earnings beat expectations",
+    "Housing market trends in March 2026",
+    "Top 10 hiking trails in Colorado",
+    "Bitcoin reaches new all-time high",
+    "Best wireless speakers under 200 dollars",
+]
+for t in non_ai_titles:
+    test("Non-AI: %s" % t[:50], not is_ai(t))
+
+print("\n[Edge Cases]")
+test("AI in middle of word (affairs) - should NOT match",
+     not is_ai("Foreign affairs committee meets"))
+test("AI as standalone word - should match",
+     is_ai("New AI system detected underwater mines"))
+test("API standalone - should match",
+     is_ai("REST API design best practices for developers"))
+
+# ==============================================================
+print("\n" + "=" * 60)
+print("SUMMARY")
+print("=" * 60)
+print("Total: %d passed, %d failed" % (PASS, FAIL))
+if FAIL > 0:
+    print("*** FAILURES DETECTED ***")
+    sys.exit(1)
+else:
+    print("All tests passed!")