"""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), )