From f794f82af552e1e1a746008bf2b7e2f23c69b538 Mon Sep 17 00:00:00 2001 From: Hiten Shah Date: Sat, 9 May 2026 19:58:54 -0700 Subject: [PATCH 1/3] feat(store): record per-run finding sightings --- skills/last30days/scripts/store.py | 88 +++++++++++++++- tests/test_store.py | 159 ++++++++++++++++++++++++++++- tests/test_watchlist_commands.py | 66 +++++++++++- 3 files changed, 306 insertions(+), 7 deletions(-) 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": [ { From 92d65723e41b4ac308bf691fd00d85222085a8c2 Mon Sep 17 00:00:00 2001 From: Hiten Shah Date: Sat, 9 May 2026 23:31:56 -0700 Subject: [PATCH 2/3] fix(watchlist): refresh sighting retries --- skills/last30days/scripts/store.py | 38 +++++++++++++++++++++--------- tests/test_store.py | 26 ++++++++++++++++++++ 2 files changed, 53 insertions(+), 11 deletions(-) diff --git a/skills/last30days/scripts/store.py b/skills/last30days/scripts/store.py index 5effa82..df0d5b7 100644 --- a/skills/last30days/scripts/store.py +++ b/skills/last30days/scripts/store.py @@ -446,7 +446,7 @@ def store_findings( new_count = len(insert_rows) updated_count = len(update_rows) - _record_sightings(conn, run_id, topic_id, with_urls) + _record_sightings(conn, run_id, topic_id, with_urls, existing_by_url) conn.execute( "UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?", (new_count, updated_count, run_id), @@ -463,6 +463,7 @@ def _record_sightings( run_id: int, topic_id: int, findings_with_urls: List[tuple[str, Dict[str, Any]]], + existing_by_url: Optional[Dict[str, sqlite3.Row]] = None, ) -> None: """Record the findings observed during this run. @@ -474,20 +475,28 @@ def _record_sightings( 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() + rows_by_url = dict(existing_by_url or {}) + + missing_urls = [url for url in by_url if url not in rows_by_url] + if missing_urls: + placeholders = ",".join("?" for _ in missing_urls) + rows = conn.execute( + f"SELECT id, source_url FROM findings WHERE source_url IN ({placeholders})", + missing_urls, + ).fetchall() + rows_by_url.update({row["source_url"]: row for row in rows}) + sighting_rows = [] - for row in rows: - finding = by_url[row["source_url"]] + for url, finding in by_url.items(): + row = rows_by_url.get(url) + if row is None: + continue sighting_rows.append(( row["id"], run_id, topic_id, finding.get("source", "unknown"), - row["source_url"], + url, finding.get("source_title") or finding.get("title", ""), finding.get("engagement_score", 0), finding.get("relevance_score", 0), @@ -497,10 +506,17 @@ def _record_sightings( return conn.executemany( - """INSERT OR IGNORE INTO finding_sightings + """INSERT INTO finding_sightings (finding_id, run_id, topic_id, source, source_url, source_title, engagement_score, relevance_score) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(run_id, finding_id) DO UPDATE SET + topic_id = excluded.topic_id, + source = excluded.source, + source_url = excluded.source_url, + source_title = excluded.source_title, + engagement_score = excluded.engagement_score, + relevance_score = excluded.relevance_score""", sighting_rows, ) diff --git a/tests/test_store.py b/tests/test_store.py index b0662c3..b74dba1 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -535,6 +535,32 @@ def test_store_findings_sightings_are_idempotent_per_run(temp_db): assert len(sightings) == 1 +def test_store_findings_updates_existing_sighting_for_same_run(temp_db): + """Test that retrying a run refreshes its sighting snapshot instead of freezing it.""" + 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, "source_title": "Reddit 1 updated", "engagement_score": 15.0}], + ) + + sightings = store.get_sightings_for_run(topic["id"], run_id) + assert len(sightings) == 1 + assert sightings[0]["source_title"] == "Reddit 1 updated" + assert sightings[0]["engagement_score"] == 15.0 + + 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") From 375fd0bcc0201128c8ac60033a7849780ff967a9 Mon Sep 17 00:00:00 2001 From: Hiten Shah Date: Sat, 16 May 2026 22:57:32 -0700 Subject: [PATCH 3/3] fix(store): enforce sighting finding id invariant --- skills/last30days/scripts/store.py | 2 +- tests/test_store.py | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/skills/last30days/scripts/store.py b/skills/last30days/scripts/store.py index df0d5b7..39f8286 100644 --- a/skills/last30days/scripts/store.py +++ b/skills/last30days/scripts/store.py @@ -163,7 +163,7 @@ 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, + finding_id INTEGER NOT NULL 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, diff --git a/tests/test_store.py b/tests/test_store.py index b74dba1..f5c1eb3 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -451,9 +451,14 @@ def test_init_db_creates_finding_sightings_table(temp_db): table = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='finding_sightings'" ).fetchone() + columns = { + row[1]: row[3] + for row in conn.execute("PRAGMA table_info(finding_sightings)").fetchall() + } conn.close() assert table is not None + assert columns["finding_id"] == 1 def test_store_findings_records_sightings_for_new_findings(temp_db):