sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #373 from hnshah/ren/watchlist-sightings

Browse Source 
feat(store): record per-run finding sightings
This commit is contained in:
Trevin Chow
2026-05-16 23:24:09 -07:00
committed by GitHub
parent 03043da407 375fd0bcc0
commit 9f39d10bc5
3 changed files with 353 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/last30days/scripts/store.py
+103 -1
View File
@@ -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 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,
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, existing_by_url)
conn.execute(
"UPDATE research_runs SET findings_new = ?, findings_updated = ? WHERE id = ?",
(new_count, updated_count, run_id),
@@ -434,6 +458,84 @@ 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]]],
existing_by_url: Optional[Dict[str, sqlite3.Row]] = None,
) -> 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}
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 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"),
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 INTO finding_sightings
(finding_id, run_id, topic_id, source, source_url, source_title,
engagement_score, relevance_score)
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,
)
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,
tests/test_store.py
+186 -4
View File
@@ -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,127 @@ 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()
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):
"""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_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")
tests/test_watchlist_commands.py
+64 -2
View File
@@ -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": [
{