Spaces:
Sleeping
Sleeping
| """storage/threads_db.py — persist named issues ("threads") across brief regenerations. | |
| A thread is a specific tension, risk, or theme worth tracking across quarters | |
| (e.g. "Cloud segment margin compression"). Each regeneration classifies every | |
| currently open thread as ongoing/escalated/resolved; a thread unmentioned for | |
| STALE_AFTER_MISSED_PERIODS consecutive periods is deterministically retired | |
| as stale rather than left to silently vanish. See | |
| docs/superpowers/specs/2026-08-27-continuity-threads-design.md. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sqlite3 | |
| import uuid | |
| from pathlib import Path | |
| from typing import Optional | |
| DB_PATH = Path("data/threads.db") | |
| STALE_AFTER_MISSED_PERIODS = 2 | |
| _OPEN_STATUSES = ("new", "ongoing", "escalated") | |
| def _connect() -> sqlite3.Connection: | |
| conn = sqlite3.connect(DB_PATH, timeout=10) | |
| conn.execute("PRAGMA journal_mode=WAL") | |
| conn.execute("PRAGMA busy_timeout=10000") | |
| return conn | |
| def init_db() -> None: | |
| """Create the threads table if it doesn't exist. Safe to call multiple times.""" | |
| DB_PATH.parent.mkdir(parents=True, exist_ok=True) | |
| with _connect() as conn: | |
| conn.execute(""" | |
| CREATE TABLE IF NOT EXISTS threads ( | |
| thread_id TEXT PRIMARY KEY, | |
| ticker TEXT NOT NULL, | |
| topic TEXT NOT NULL, | |
| category TEXT NOT NULL, | |
| status TEXT NOT NULL, | |
| first_seen_period TEXT NOT NULL, | |
| last_updated_period TEXT NOT NULL, | |
| missed_count INTEGER NOT NULL DEFAULT 0, | |
| history_json TEXT NOT NULL | |
| ) | |
| """) | |
| conn.execute(""" | |
| CREATE INDEX IF NOT EXISTS idx_threads_ticker_status | |
| ON threads (ticker, status) | |
| """) | |
| def _row_to_dict(row: tuple) -> dict: | |
| (thread_id, ticker, topic, category, status, first_seen_period, | |
| last_updated_period, missed_count, history_json) = row | |
| return { | |
| "thread_id": thread_id, | |
| "ticker": ticker, | |
| "topic": topic, | |
| "category": category, | |
| "status": status, | |
| "first_seen_period": first_seen_period, | |
| "last_updated_period": last_updated_period, | |
| "missed_count": missed_count, | |
| "history": json.loads(history_json), | |
| } | |
| _SELECT_COLUMNS = ( | |
| "thread_id, ticker, topic, category, status, first_seen_period, " | |
| "last_updated_period, missed_count, history_json" | |
| ) | |
| def get_open_threads(ticker: str) -> list[dict]: | |
| """Return threads not yet resolved/stale for *ticker*, most recently updated first.""" | |
| if not DB_PATH.exists(): | |
| return [] | |
| init_db() | |
| placeholders = ",".join("?" for _ in _OPEN_STATUSES) | |
| with _connect() as conn: | |
| rows = conn.execute( | |
| f"SELECT {_SELECT_COLUMNS} FROM threads " | |
| f"WHERE ticker = ? AND status IN ({placeholders}) " | |
| f"ORDER BY last_updated_period DESC", | |
| (ticker.upper(), *_OPEN_STATUSES), | |
| ).fetchall() | |
| return [_row_to_dict(r) for r in rows] | |
| def list_all_threads(ticker: str) -> list[dict]: | |
| """Return every thread for *ticker* regardless of status. Powers the UI timeline.""" | |
| if not DB_PATH.exists(): | |
| return [] | |
| init_db() | |
| with _connect() as conn: | |
| rows = conn.execute( | |
| f"SELECT {_SELECT_COLUMNS} FROM threads WHERE ticker = ? " | |
| f"ORDER BY last_updated_period DESC", | |
| (ticker.upper(),), | |
| ).fetchall() | |
| return [_row_to_dict(r) for r in rows] | |
| def create_thread( | |
| ticker: str, period: str, topic: str, category: str, status: str, | |
| note: str, evidence_ref: Optional[dict], evidence_snippet: Optional[str], | |
| ) -> str: | |
| """Insert a brand-new thread and return its generated thread_id.""" | |
| init_db() | |
| thread_id = uuid.uuid4().hex | |
| history = [{ | |
| "period": period, "status": status, "note": note, | |
| "evidence_ref": evidence_ref, "evidence_snippet": evidence_snippet, | |
| }] | |
| with _connect() as conn: | |
| conn.execute( | |
| """ | |
| INSERT INTO threads ( | |
| thread_id, ticker, topic, category, status, | |
| first_seen_period, last_updated_period, missed_count, history_json | |
| ) VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?) | |
| """, | |
| (thread_id, ticker.upper(), topic, category, status, period, period, json.dumps(history)), | |
| ) | |
| return thread_id | |
| def continue_thread( | |
| thread_id: str, period: str, status: str, note: str, | |
| evidence_ref: Optional[dict], evidence_snippet: Optional[str], | |
| ) -> None: | |
| """Append a history entry to an existing thread and update its status. | |
| Idempotent per period: a user can regenerate a brief for the same filing | |
| period more than once (a normal Streamlit action). If the last history | |
| entry is already for *period*, replace it in place instead of appending a | |
| second entry for the identical period -- otherwise the Storyline UI | |
| timeline would show a duplicate row for one real quarter. | |
| """ | |
| init_db() | |
| with _connect() as conn: | |
| row = conn.execute("SELECT history_json FROM threads WHERE thread_id = ?", (thread_id,)).fetchone() | |
| if row is None: | |
| raise ValueError(f"no thread with id {thread_id!r}") | |
| history = json.loads(row[0]) | |
| entry = { | |
| "period": period, "status": status, "note": note, | |
| "evidence_ref": evidence_ref, "evidence_snippet": evidence_snippet, | |
| } | |
| if history and history[-1].get("period") == period: | |
| history[-1] = entry | |
| else: | |
| history.append(entry) | |
| conn.execute( | |
| """ | |
| UPDATE threads | |
| SET status = ?, last_updated_period = ?, missed_count = 0, history_json = ? | |
| WHERE thread_id = ? | |
| """, | |
| (status, period, json.dumps(history), thread_id), | |
| ) | |
| def mark_missed(thread_id: str, period: str) -> None: | |
| """Increment missed_count; retire to 'stale' once STALE_AFTER_MISSED_PERIODS is reached. | |
| Idempotent per period: `last_updated_period` records the period this | |
| thread was last touched by *any* operation (created, continued, or | |
| previously marked missed). If it already equals *period*, this call is a | |
| no-op -- otherwise regenerating a brief twice for the same filing period | |
| (with the thread unmentioned both times) would increment missed_count | |
| twice, retiring the thread to 'stale' after only one real quarter of | |
| silence. This also correctly blocks a same-period mark_missed right after | |
| create_thread, since a freshly created thread's `last_updated_period` | |
| already equals its `first_seen_period`. | |
| """ | |
| init_db() | |
| with _connect() as conn: | |
| row = conn.execute( | |
| "SELECT missed_count, history_json, last_updated_period FROM threads WHERE thread_id = ?", | |
| (thread_id,), | |
| ).fetchone() | |
| if row is None: | |
| raise ValueError(f"no thread with id {thread_id!r}") | |
| missed_count, history_json, last_updated_period = row | |
| if last_updated_period == period: | |
| return | |
| missed_count += 1 | |
| history = json.loads(history_json) | |
| if missed_count >= STALE_AFTER_MISSED_PERIODS: | |
| history.append({ | |
| "period": period, "status": "stale", | |
| "note": f"Not mentioned for {STALE_AFTER_MISSED_PERIODS} consecutive quarters.", | |
| "evidence_ref": None, "evidence_snippet": None, | |
| }) | |
| conn.execute( | |
| """ | |
| UPDATE threads | |
| SET status = 'stale', missed_count = ?, last_updated_period = ?, history_json = ? | |
| WHERE thread_id = ? | |
| """, | |
| (missed_count, period, json.dumps(history), thread_id), | |
| ) | |
| else: | |
| conn.execute( | |
| "UPDATE threads SET missed_count = ?, last_updated_period = ? WHERE thread_id = ?", | |
| (missed_count, period, thread_id), | |
| ) | |