diff --git a/skills/last30days/scripts/store.py b/skills/last30days/scripts/store.py index bb308da..5effa82 100644 --- a/skills/last30days/scripts/store.py +++ b/skills/last30days/scripts/store.py @@ -159,7 +159,30 @@ _UPDATABLE_FINDING_COLUMNS = frozenset({ }) # Future migrations keyed by version number -MIGRATIONS: Dict[int, str] = {} +MIGRATIONS: Dict[int, str] = { + 2: """ +CREATE TABLE IF NOT EXISTS finding_sightings ( + id INTEGER PRIMARY KEY, + finding_id INTEGER REFERENCES findings(id) ON DELETE CASCADE, + run_id INTEGER REFERENCES research_runs(id) ON DELETE CASCADE, + topic_id INTEGER REFERENCES topics(id) ON DELETE CASCADE, + source TEXT NOT NULL, + source_url TEXT NOT NULL, + source_title TEXT, + engagement_score REAL, + relevance_score REAL, + seen_at TEXT DEFAULT (datetime('now')), + UNIQUE(run_id, finding_id) +); + +CREATE INDEX IF NOT EXISTS idx_finding_sightings_run + ON finding_sightings(run_id, topic_id); +CREATE INDEX IF NOT EXISTS idx_finding_sightings_topic_seen + ON finding_sightings(topic_id, seen_at); +CREATE INDEX IF NOT EXISTS idx_finding_sightings_url + ON finding_sightings(source_url); +""", +} def _connect(db_path: Optional[Path] = None) -> sqlite3.Connection: @@ -423,6 +446,7 @@ def store_findings( new_count = len(insert_rows) updated_count = len(update_rows) + _record_sightings(conn, run_id, topic_id, with_urls) conn.execute( "UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?", (new_count, updated_count, run_id), @@ -434,6 +458,68 @@ def store_findings( return {"new": new_count, "updated": updated_count} +def _record_sightings( + conn: sqlite3.Connection, + run_id: int, + topic_id: int, + findings_with_urls: List[tuple[str, Dict[str, Any]]], +) -> None: + """Record the findings observed during this run. + + The aggregate findings table keeps one row per URL and updates that row on + re-sighting. This ledger preserves the run/topic membership needed for + watchlist deltas and dossiers. + """ + if not findings_with_urls: + return + + by_url = {url: finding for url, finding in findings_with_urls} + placeholders = ",".join("?" for _ in by_url) + rows = conn.execute( + f"SELECT id, source_url FROM findings WHERE source_url IN ({placeholders})", + list(by_url), + ).fetchall() + sighting_rows = [] + for row in rows: + finding = by_url[row["source_url"]] + sighting_rows.append(( + row["id"], + run_id, + topic_id, + finding.get("source", "unknown"), + row["source_url"], + finding.get("source_title") or finding.get("title", ""), + finding.get("engagement_score", 0), + finding.get("relevance_score", 0), + )) + + if not sighting_rows: + return + + conn.executemany( + """INSERT OR IGNORE INTO finding_sightings + (finding_id, run_id, topic_id, source, source_url, source_title, + engagement_score, relevance_score) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + sighting_rows, + ) + + +def get_sightings_for_run(topic_id: int, run_id: int) -> List[Dict[str, Any]]: + """Return findings observed for a topic during a specific run.""" + conn = _connect() + try: + rows = conn.execute( + """SELECT * FROM finding_sightings + WHERE topic_id = ? AND run_id = ? + ORDER BY id""", + (topic_id, run_id), + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + def get_new_findings( topic_id: int, since: Optional[str] = None, diff --git a/tests/test_store.py b/tests/test_store.py index aaacfa5..b0662c3 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -59,7 +59,68 @@ def sample_report(): "source_weights": {}, }, "clusters": [], - "ranked_candidates": [], + "ranked_candidates": [ + { + "candidate_id": "c-r1", + "item_id": "R1", + "source": "reddit", + "title": "Test Reddit Post", + "url": "https://reddit.com/r/test/1", + "snippet": "Reddit snippet", + "subquery_labels": ["primary"], + "native_ranks": {"reddit": 1}, + "local_relevance": 0.8, + "freshness": 100, + "engagement": 50.0, + "source_quality": 0.8, + "rrf_score": 1.0, + "final_score": 0.8, + "explanation": "Reddit snippet", + "source_items": [ + { + "item_id": "R1", + "source": "reddit", + "title": "Test Reddit Post", + "body": "Reddit discussion content", + "url": "https://reddit.com/r/test/1", + "author": "testuser", + "engagement_score": 50.0, + "local_relevance": 0.8, + "snippet": "Reddit snippet", + } + ], + }, + { + "candidate_id": "c-x1", + "item_id": "X1", + "source": "x", + "title": "Test X Post", + "url": "https://x.com/test/status/1", + "snippet": "X snippet", + "subquery_labels": ["primary"], + "native_ranks": {"x": 1}, + "local_relevance": 0.85, + "freshness": 100, + "engagement": 75.0, + "source_quality": 0.8, + "rrf_score": 1.0, + "final_score": 0.85, + "explanation": "X snippet", + "source_items": [ + { + "item_id": "X1", + "source": "x", + "title": "Test X Post", + "body": "X post content", + "url": "https://x.com/test/status/1", + "author": "xuser", + "engagement_score": 75.0, + "local_relevance": 0.85, + "snippet": "X snippet", + } + ], + }, + ], "items_by_source": { "reddit": [ { @@ -236,13 +297,13 @@ def test_findings_from_report_handles_missing_fields(): "clusters": [], "ranked_candidates": [], "items_by_source": { - "reddit": [ + "hackernews": [ { "item_id": "R1", - "source": "reddit", + "source": "hackernews", "title": "Test", "body": "Content", - "url": "https://reddit.com/1", + "url": "https://news.ycombinator.com/item?id=1", "author": None, # Missing author "engagement_score": None, # Missing engagement "local_relevance": None, # Missing relevance @@ -384,6 +445,96 @@ def test_store_findings_skips_items_without_url(temp_db): assert counts["new"] == 1 +def test_init_db_creates_finding_sightings_table(temp_db): + """Test that the per-run sightings ledger is available on fresh databases.""" + conn = sqlite3.connect(str(temp_db)) + table = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='finding_sightings'" + ).fetchone() + conn.close() + + assert table is not None + + +def test_store_findings_records_sightings_for_new_findings(temp_db): + """Test that each stored finding is linked to the run that observed it.""" + topic = store.add_topic("Test Topic") + run_id = store.record_run(topic["id"], source_mode="v3") + findings = [ + { + "source": "reddit", + "source_url": "https://reddit.com/1", + "source_title": "Reddit 1", + "content": "Content 1", + "engagement_score": 10.0, + "relevance_score": 0.7, + }, + { + "source": "x", + "source_url": "https://x.com/a/status/1", + "source_title": "X 1", + "content": "Content 2", + "engagement_score": 20.0, + "relevance_score": 0.8, + }, + ] + + store.store_findings(run_id, topic["id"], findings) + + sightings = store.get_sightings_for_run(topic["id"], run_id) + assert [s["source_url"] for s in sightings] == [ + "https://reddit.com/1", + "https://x.com/a/status/1", + ] + assert {s["source"] for s in sightings} == {"reddit", "x"} + + +def test_store_findings_records_sightings_for_resighted_findings(temp_db): + """Test that a re-seen finding is recorded for each run that observes it.""" + topic = store.add_topic("Test Topic") + first_run_id = store.record_run(topic["id"], source_mode="v3") + second_run_id = store.record_run(topic["id"], source_mode="v3") + finding = { + "source": "reddit", + "source_url": "https://reddit.com/1", + "source_title": "Reddit 1", + "content": "Content", + "engagement_score": 10.0, + "relevance_score": 0.7, + } + + store.store_findings(first_run_id, topic["id"], [finding]) + store.store_findings(second_run_id, topic["id"], [{**finding, "engagement_score": 15.0}]) + + first_sightings = store.get_sightings_for_run(topic["id"], first_run_id) + second_sightings = store.get_sightings_for_run(topic["id"], second_run_id) + + assert len(first_sightings) == 1 + assert len(second_sightings) == 1 + assert first_sightings[0]["source_url"] == second_sightings[0]["source_url"] + assert second_sightings[0]["engagement_score"] == 15.0 + + +def test_store_findings_sightings_are_idempotent_per_run(temp_db): + """Test that storing the same finding twice for one run does not duplicate sightings.""" + topic = store.add_topic("Test Topic") + run_id = store.record_run(topic["id"], source_mode="v3") + finding = { + "source": "reddit", + "source_url": "https://reddit.com/1", + "source_title": "Reddit 1", + "content": "Content", + "engagement_score": 10.0, + "relevance_score": 0.7, + } + + store.store_findings(run_id, topic["id"], [finding]) + store.store_findings(run_id, topic["id"], [finding]) + + sightings = store.get_sightings_for_run(topic["id"], run_id) + assert len(sightings) == 1 + + def test_update_validates_allowed_columns(temp_db, sample_report): """Test update_run/update_finding accept valid keys and reject invalid keys.""" topic = store.add_topic("Test Topic") diff --git a/tests/test_watchlist_commands.py b/tests/test_watchlist_commands.py index b86a5ee..28ff488 100644 --- a/tests/test_watchlist_commands.py +++ b/tests/test_watchlist_commands.py @@ -251,7 +251,38 @@ def test_run_topic_success(mock_subprocess, temp_db): "source_weights": {}, }, "clusters": [], - "ranked_candidates": [], + "ranked_candidates": [ + { + "candidate_id": "c-r1", + "item_id": "R1", + "source": "reddit", + "title": "Test", + "url": "https://reddit.com/1", + "snippet": "Snippet", + "subquery_labels": ["primary"], + "native_ranks": {"reddit": 1}, + "local_relevance": 0.8, + "freshness": 100, + "engagement": 50.0, + "source_quality": 0.8, + "rrf_score": 1.0, + "final_score": 0.8, + "explanation": "Snippet", + "source_items": [ + { + "item_id": "R1", + "source": "reddit", + "title": "Test", + "body": "Content", + "url": "https://reddit.com/1", + "author": "user", + "engagement_score": 50.0, + "local_relevance": 0.8, + "snippet": "Snippet", + } + ], + } + ], "items_by_source": { "reddit": [ { @@ -338,7 +369,38 @@ def test_run_topic_calls_delivery(mock_deliver, mock_subprocess, temp_db): "source_weights": {}, }, "clusters": [], - "ranked_candidates": [], + "ranked_candidates": [ + { + "candidate_id": "c-r1", + "item_id": "R1", + "source": "reddit", + "title": "Test", + "url": "https://reddit.com/1", + "snippet": "Snippet", + "subquery_labels": ["primary"], + "native_ranks": {"reddit": 1}, + "local_relevance": 0.8, + "freshness": 100, + "engagement": 50.0, + "source_quality": 0.8, + "rrf_score": 1.0, + "final_score": 0.8, + "explanation": "Snippet", + "source_items": [ + { + "item_id": "R1", + "source": "reddit", + "title": "Test", + "body": "Content", + "url": "https://reddit.com/1", + "author": "user", + "engagement_score": 50.0, + "local_relevance": 0.8, + "snippet": "Snippet", + } + ], + } + ], "items_by_source": { "reddit": [ {