← refacto-feature-scout
from __future__ import annotations
import json
import sqlite3
import time
from pathlib import Path
from typing import Any, Iterable
from .config import DB_PATH, IDEA_STATUSES, LEGACY_PLATFORMS
SCHEMA = """
CREATE TABLE IF NOT EXISTS ideas (
id INTEGER PRIMARY KEY AUTOINCREMENT,
platform TEXT NOT NULL,
title TEXT NOT NULL,
summary TEXT NOT NULL,
user_value TEXT NOT NULL DEFAULT '',
value_score INTEGER, -- 1 (niche) .. 5 (most users, often); NULL = not scored yet
value_reason TEXT NOT NULL DEFAULT '',
plan TEXT NOT NULL DEFAULT '', -- which of the project's plans this belongs in ('' = not said)
plan_reason TEXT NOT NULL DEFAULT '',
auto_ship INTEGER NOT NULL DEFAULT 0, -- 1 = ship it as soon as it is built and tested
after_id INTEGER, -- approve this idea automatically when idea after_id ships
evidence TEXT NOT NULL DEFAULT '[]',
effort TEXT NOT NULL DEFAULT 'M',
category TEXT NOT NULL DEFAULT 'feature',
spec TEXT NOT NULL DEFAULT '',
source TEXT NOT NULL DEFAULT 'scan',
status TEXT NOT NULL DEFAULT 'proposed',
created_at INTEGER NOT NULL,
decided_at INTEGER,
build_branch TEXT,
build_dir TEXT,
build_report TEXT,
built_at INTEGER,
shipped_at INTEGER,
error TEXT
);
CREATE TABLE IF NOT EXISTS notices (
id INTEGER PRIMARY KEY AUTOINCREMENT,
created_at INTEGER NOT NULL,
kind TEXT NOT NULL, -- 'paused', 'shipped', 'problem' ...
text TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
kind TEXT NOT NULL,
started_at INTEGER NOT NULL,
finished_at INTEGER,
ok INTEGER,
detail TEXT NOT NULL DEFAULT '{}'
);
"""
def _now() -> int:
return int(time.time())
def clamp_score(score: Any) -> int | None:
"""A value score is a whole number 1..5; anything else is 'not scored'."""
try:
n = int(round(float(score)))
except (TypeError, ValueError):
return None
return n if 1 <= n <= 5 else None
def by_value(ideas: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Most valuable first. Scored ideas (5 down to 1) come before unscored ones; ties keep
the newest day first and, within a day, the order the researcher ranked them in."""
def key(idea: dict[str, Any]) -> tuple[int, int, int]:
score = clamp_score(idea.get("value_score")) or 0
day = int(float(idea.get("created_at") or 0)) // 86400
return (-score, -day, int(idea["id"]))
return sorted(ideas, key=key)
class Store:
def __init__(self, path: Path | str = DB_PATH, *, readonly: bool = False):
self.path = str(path)
if readonly and self.path != ":memory:" and Path(self.path).exists():
# Just looking. Opening the usual way checks the schema on every connection, and a
# schema check is a write — so a page that only reads would queue behind a build.
self.conn = sqlite3.connect(f"file:{self.path}?mode=ro", uri=True, timeout=15)
self.conn.row_factory = sqlite3.Row
self.conn.execute("PRAGMA busy_timeout=3000")
return
if self.path != ":memory:":
Path(self.path).parent.mkdir(parents=True, exist_ok=True)
self.conn = sqlite3.connect(self.path, timeout=15)
self.conn.row_factory = sqlite3.Row
if self.path != ":memory:":
# The review page and a running watcher both talk to this file at once.
self.conn.execute("PRAGMA journal_mode=WAL")
self.conn.execute("PRAGMA busy_timeout=5000")
self.conn.executescript(SCHEMA)
self._migrate()
def _migrate(self) -> None:
cols = {r[1] for r in self.conn.execute("PRAGMA table_info(ideas)")}
if "venue" not in cols:
self.conn.execute("ALTER TABLE ideas ADD COLUMN venue TEXT NOT NULL DEFAULT 'both'")
if "merge_base_sha" not in cols:
# Where the project stood just before a change was added, so it can be put back.
self.conn.execute("ALTER TABLE ideas ADD COLUMN merge_base_sha TEXT")
if "value_score" not in cols:
# How much users would get out of it, 1..5, so the whole backlog can be ordered by
# value rather than by the day an idea arrived. `scout score` fills in old rows.
self.conn.execute("ALTER TABLE ideas ADD COLUMN value_score INTEGER")
self.conn.execute("ALTER TABLE ideas ADD COLUMN value_reason TEXT NOT NULL DEFAULT ''")
if "plan" not in cols:
# Which plan (Free / Pro ...) an idea belongs in, once a project has plans.
self.conn.execute("ALTER TABLE ideas ADD COLUMN plan TEXT NOT NULL DEFAULT ''")
self.conn.execute("ALTER TABLE ideas ADD COLUMN plan_reason TEXT NOT NULL DEFAULT ''")
if "auto_ship" not in cols:
# A chain of steps that must land one after another without the owner
# clicking between them ("do them all without asking me", 2026-09-07).
self.conn.execute("ALTER TABLE ideas ADD COLUMN auto_ship INTEGER NOT NULL DEFAULT 0")
self.conn.execute("ALTER TABLE ideas ADD COLUMN after_id INTEGER")
if "ship_requested_at" not in cols:
# "Please ship this" used to be written into the error column. Now that both a
# web page and the command line can ask, it gets a column of its own.
self.conn.execute("ALTER TABLE ideas ADD COLUMN ship_requested_at INTEGER")
self.conn.execute("UPDATE ideas SET ship_requested_at=? WHERE error='SHIP_REQUESTED'", (_now(),))
for old, (new, venue) in LEGACY_PLATFORMS.items():
self.conn.execute("UPDATE ideas SET platform=?, venue=? WHERE platform=?", (new, venue, old))
self.conn.commit()
# ---- ideas -----------------------------------------------------------
def add_idea(self, platform: str, title: str, summary: str, *, user_value: str = "",
evidence: Iterable[dict[str, Any]] = (), effort: str = "M",
category: str = "feature", spec: str = "", source: str = "scan",
venue: str = "both", value_score: int | None = None, value_reason: str = "",
plan: str = "", plan_reason: str = "") -> int:
if platform in LEGACY_PLATFORMS:
platform, venue = LEGACY_PLATFORMS[platform]
cur = self.conn.execute(
"INSERT INTO ideas (platform,title,summary,user_value,evidence,effort,category,spec,source,status,"
"created_at,venue,value_score,value_reason,plan,plan_reason)"
" VALUES (?,?,?,?,?,?,?,?,?,'proposed',?,?,?,?,?,?)",
(platform, title.strip(), summary.strip(), user_value.strip(), json.dumps(list(evidence)),
effort, category, spec.strip(), source, _now(), venue, clamp_score(value_score),
(value_reason or "").strip(), (plan or "").strip().lower(), (plan_reason or "").strip()),
)
self.conn.commit()
return int(cur.lastrowid)
def set_value(self, idea_id: int, score: int | None, reason: str = "") -> None:
self.conn.execute("UPDATE ideas SET value_score=?, value_reason=? WHERE id=?",
(clamp_score(score), (reason or "").strip(), idea_id))
self.conn.commit()
def idea(self, idea_id: int) -> dict[str, Any] | None:
row = self.conn.execute("SELECT * FROM ideas WHERE id=?", (idea_id,)).fetchone()
return self._idea_row(row) if row else None
def ideas(self, *, status: str | None = None, platform: str | None = None,
statuses: Iterable[str] | None = None) -> list[dict[str, Any]]:
sql, args = "SELECT * FROM ideas WHERE 1=1", []
if status:
sql += " AND status=?"; args.append(status)
if statuses:
ss = list(statuses)
sql += " AND status IN (%s)" % ",".join("?" * len(ss)); args.extend(ss)
if platform:
sql += " AND platform=?"; args.append(platform)
sql += " ORDER BY id"
return [self._idea_row(r) for r in self.conn.execute(sql, args).fetchall()]
def all_titles(self, platform: str) -> list[tuple[str, str]]:
"""(title, status) for every idea ever proposed for a platform — used to avoid repeats."""
rows = self.conn.execute("SELECT title,status FROM ideas WHERE platform=? ORDER BY id", (platform,))
return [(r["title"], r["status"]) for r in rows]
def chain(self, ids: list[int]) -> None:
"""Make ``ids`` a hands-off sequence: each ships as soon as it is built, and each
one after the first is approved the moment the one before it has shipped."""
prev = None
for iid in ids:
self.conn.execute("UPDATE ideas SET auto_ship=1, after_id=? WHERE id=?", (prev, iid))
prev = iid
self.conn.commit()
def release_dependents(self, shipped_id: int) -> list[int]:
"""Approve every proposed idea that was waiting on ``shipped_id``; returns their ids."""
ids = [int(r[0]) for r in self.conn.execute(
"SELECT id FROM ideas WHERE after_id=? AND status='proposed' ORDER BY id", (shipped_id,))]
for iid in ids:
self.set_status(iid, "approved")
return ids
def send_back_for_rebase(self, idea_id: int, detail: str) -> None:
"""A built feature no longer fits on the latest version: hand it back to the builder
(status approved, place in the queue kept, ship request cleared) with the conflict noted
so the next build attempt starts by rebasing."""
self.conn.execute(
"UPDATE ideas SET status='approved', ship_requested_at=NULL, error=? WHERE id=?",
(f"needs rebasing onto the latest version (conflict): {detail}"[:900], idea_id))
self.conn.commit()
def reorder_queue(self, ids: list[int]) -> None:
"""Make ``ids`` the build order (first builds first). Only approved ideas move; a fixed
early base keeps them ahead of anything approved later, which joins at the back."""
base = 1_700_000_000
for n, iid in enumerate(ids):
self.conn.execute("UPDATE ideas SET decided_at=? WHERE id=? AND status='approved'", (base + n, iid))
self.conn.commit()
def keep_approved(self, idea_id: int, note: str) -> None:
"""Back to approved without losing its place in the queue (decided_at untouched)."""
self.conn.execute("UPDATE ideas SET status='approved', error=? WHERE id=?", (note[:600], idea_id))
self.conn.commit()
def request_ship(self, idea_id: int) -> None:
"""Mark a built idea as one the owner wants merged into their project."""
self.conn.execute("UPDATE ideas SET status='built', ship_requested_at=? WHERE id=?",
(_now(), idea_id))
self.conn.commit()
def clear_ship_request(self, idea_id: int) -> None:
self.conn.execute("UPDATE ideas SET ship_requested_at=NULL WHERE id=?", (idea_id,))
self.conn.commit()
def requeue_interrupted_builds(self) -> list[int]:
"""Put ideas stranded in 'building' back in the queue, keeping their place.
Builds run one at a time under the lock, so when a build pass starts nothing can be
building: an idea still marked so was cut off mid-build (the machine restarted, e.g. on
a deploy) and would otherwise sit there forever — the queue only picks up 'approved'.
decided_at is kept, so it does not lose its turn. Its unfinished run is closed too.
"""
ids = [int(r[0]) for r in self.conn.execute("SELECT id FROM ideas WHERE status='building' ORDER BY id")]
if not ids:
return []
marks = ",".join("?" * len(ids))
self.conn.execute(f"UPDATE ideas SET status='approved', error=? WHERE id IN ({marks})",
["build interrupted by a restart; queued again", *ids])
self.conn.execute(
"UPDATE runs SET finished_at=?, ok=0, detail=? WHERE finished_at IS NULL AND kind LIKE 'build:%'",
(_now(), json.dumps({"status": "interrupted", "note": "cut off by a restart; the idea was queued again"})))
self.conn.commit()
return ids
def set_status(self, idea_id: int, status: str, **fields: Any) -> None:
if status not in IDEA_STATUSES:
raise ValueError(f"unknown status {status!r}")
cols = ["status=?"]; args: list[Any] = [status]
if status in ("approved", "rejected"):
cols.append("decided_at=?"); args.append(_now())
if status == "built":
cols.append("built_at=?"); args.append(_now())
if status == "shipped":
cols.append("shipped_at=?"); args.append(_now())
for k, v in fields.items():
if k not in ("build_branch", "build_dir", "build_report", "error",
"merge_base_sha"):
raise ValueError(f"cannot set {k}")
cols.append(f"{k}=?"); args.append(v)
args.append(idea_id)
self.conn.execute(f"UPDATE ideas SET {', '.join(cols)} WHERE id=?", args)
self.conn.commit()
@staticmethod
def _idea_row(row: sqlite3.Row) -> dict[str, Any]:
d = dict(row)
try:
d["evidence"] = json.loads(d.get("evidence") or "[]")
except json.JSONDecodeError:
d["evidence"] = []
return d
# ---- notices: what used to be an email, now a line on the board --------
def notice(self, kind: str, text: str) -> int:
"""Something the owner should see next time they look at the page."""
cur = self.conn.execute("INSERT INTO notices (created_at,kind,text) VALUES (?,?,?)",
(_now(), kind, text[:2000]))
self.conn.commit()
return int(cur.lastrowid)
def notices(self, *, within_s: int = 7 * 86400, limit: int = 12) -> list[dict[str, Any]]:
"""Newest first."""
rows = self.conn.execute(
"SELECT * FROM notices WHERE created_at>=? ORDER BY id DESC LIMIT ?",
(_now() - within_s, limit)).fetchall()
return [dict(r) for r in rows]
def last_note_recent(self, kind: str, within_s: int) -> bool:
row = self.conn.execute("SELECT MAX(created_at) FROM notices WHERE kind=?", (kind,)).fetchone()
return bool(row and row[0] and int(row[0]) > _now() - within_s)
# ---- runs -------------------------------------------------------------
def start_run(self, kind: str) -> int:
cur = self.conn.execute("INSERT INTO runs (kind,started_at) VALUES (?,?)", (kind, _now()))
self.conn.commit()
return int(cur.lastrowid)
def finish_run(self, run_id: int, ok: bool, detail: dict[str, Any]) -> None:
self.conn.execute("UPDATE runs SET finished_at=?, ok=?, detail=? WHERE id=?",
(_now(), 1 if ok else 0, json.dumps(detail), run_id))
self.conn.commit()
def spend_since(self, since: int, project: str | None = None) -> float:
"""What Claude has been asked to do since a moment, in dollars.
Every run records what it cost; nothing added them up before, which is fine when the
bill is your own and a surprise when it isn't.
"""
total = 0.0
for row in self.conn.execute("SELECT kind, detail FROM runs WHERE started_at>=?", (since,)):
try:
detail = json.loads(row["detail"] or "{}")
except (TypeError, json.JSONDecodeError):
continue
if not isinstance(detail, dict):
continue
if project and detail.get("project") != project:
continue
total += float(detail.get("cost_usd") or 0)
return round(total, 4)
def last_run(self, kind: str) -> dict[str, Any] | None:
row = self.conn.execute("SELECT * FROM runs WHERE kind=? ORDER BY id DESC LIMIT 1", (kind,)).fetchone()
if not row:
return None
d = dict(row); d["detail"] = json.loads(d["detail"] or "{}"); return d