1"""SQLite state store (single writer: the controller). All access under one lock (sqlite + threads gotcha, WP29)."""2from __future__ import annotations3import json, os, sqlite3, threading4from .contracts import utcnow56SCHEMA = """7CREATE TABLE IF NOT EXISTS runs (8 run_id TEXT PRIMARY KEY, task_id TEXT, occupation TEXT, arm TEXT, rep INTEGER, model TEXT, stage TEXT,9 node TEXT, session_id TEXT, status TEXT, assigned_utc TEXT, started_utc TEXT, ended_utc TEXT,10 wall_s REAL, timed_out INTEGER, exit_code INTEGER, result_subtype TEXT, cost_usd REAL, num_turns INTEGER,11 recovery_attempted INTEGER DEFAULT 0, recovery_success INTEGER, recovery_cost_usd REAL, recovery_wall_s REAL,12 selfcheck_pass_first INTEGER, selfcheck_pass_final INTEGER,13 n_events INTEGER, claude_version TEXT, order_index INTEGER, seed INTEGER, notes TEXT14);15CREATE TABLE IF NOT EXISTS events (16 event_id TEXT PRIMARY KEY, run_id TEXT, seq INTEGER, received_utc TEXT, elapsed_s REAL, session_id TEXT,17 event_type TEXT, subtype TEXT, tool_name TEXT, summary TEXT, status TEXT, error_class TEXT, usage TEXT, payload_sha256 TEXT18);19CREATE INDEX IF NOT EXISTS ix_events_run ON events(run_id, seq);20CREATE TABLE IF NOT EXISTS artifacts (run_id TEXT, path TEXT, sha256 TEXT, bytes INTEGER, PRIMARY KEY(run_id, path));21CREATE TABLE IF NOT EXISTS receipts (run_id TEXT PRIMARY KEY, receipt TEXT, evaluated_utc TEXT, det_completion INTEGER, det_score REAL, n_violations INTEGER, abstention_status TEXT, abstention_justified INTEGER);22CREATE TABLE IF NOT EXISTS judgments (run_id TEXT, judge_idx INTEGER, judge_model TEXT, score REAL, scores TEXT, rationale TEXT, cost_usd REAL, judged_utc TEXT, raw TEXT, PRIMARY KEY(run_id, judge_idx));23CREATE TABLE IF NOT EXISTS metrics (run_id TEXT PRIMARY KEY, metrics TEXT);24CREATE TABLE IF NOT EXISTS interventions (id INTEGER PRIMARY KEY AUTOINCREMENT, utc TEXT, run_id TEXT, minutes REAL, description TEXT);25CREATE TABLE IF NOT EXISTS ledger (id INTEGER PRIMARY KEY AUTOINCREMENT, utc TEXT, run_id TEXT, kind TEXT, cost_usd REAL);26"""2728class Store:29 def __init__(self, path: str):30 os.makedirs(os.path.dirname(path), exist_ok=True)31 self.path = path; self.lock = threading.RLock()32 self.con = sqlite3.connect(path, check_same_thread=False, timeout=60)33 self.con.row_factory = sqlite3.Row34 with self.lock:35 self.con.executescript(SCHEMA); self.con.commit()3637 # --- generic helpers ---38 def exec(self, sql, args=()):39 with self.lock:40 cur = self.con.execute(sql, args); self.con.commit(); return cur41 def q(self, sql, args=()):42 with self.lock:43 return [dict(r) for r in self.con.execute(sql, args).fetchall()]44 def one(self, sql, args=()):45 r = self.q(sql, args); return r[0] if r else None4647 # --- runs ---48 def add_run(self, **kw):49 kw.setdefault("assigned_utc", utcnow()); kw.setdefault("status", "assigned")50 cols = ",".join(kw); ph = ",".join("?" * len(kw))51 self.exec(f"INSERT OR IGNORE INTO runs ({cols}) VALUES ({ph})", tuple(kw.values()))52 def update_run(self, run_id, **kw):53 sets = ",".join(f"{k}=?" for k in kw)54 self.exec(f"UPDATE runs SET {sets} WHERE run_id=?", (*kw.values(), run_id))55 def get_run(self, run_id):56 return self.one("SELECT * FROM runs WHERE run_id=?", (run_id,))57 def runs(self, where="1=1", args=()):58 return self.q(f"SELECT * FROM runs WHERE {where} ORDER BY order_index", args)5960 # --- events ---61 def add_event(self, ev: dict):62 with self.lock:63 self.con.execute("INSERT OR IGNORE INTO events (event_id,run_id,seq,received_utc,elapsed_s,session_id,event_type,subtype,tool_name,summary,status,error_class,usage,payload_sha256) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",64 (ev["event_id"], ev["run_id"], ev["seq"], ev["received_utc"], ev["elapsed_s"], ev.get("session_id"), ev.get("event_type"), ev.get("subtype"), ev.get("tool_name"), ev.get("summary"), ev.get("status"), ev.get("error_class"), json.dumps(ev.get("usage")) if ev.get("usage") is not None else None, ev.get("payload_sha256")))65 self.con.commit()66 def events(self, run_id):67 return self.q("SELECT * FROM events WHERE run_id=? ORDER BY seq", (run_id,))6869 def add_artifacts(self, run_id, items):70 with self.lock:71 self.con.executemany("INSERT OR REPLACE INTO artifacts VALUES (?,?,?,?)", [(run_id, p, h, b) for p, h, b in items]); self.con.commit()72 def set_receipt(self, run_id, receipt: dict):73 self.exec("INSERT OR REPLACE INTO receipts VALUES (?,?,?,?,?,?,?,?)", (run_id, json.dumps(receipt, default=str), utcnow(), int(bool(receipt.get("det_completion"))), float(receipt.get("det_score") or 0), len(receipt.get("critical_violations") or []), (receipt.get("abstention") or {}).get("status"), None if (receipt.get("abstention") or {}).get("justified") is None else int(bool(receipt["abstention"]["justified"]))))74 def set_judgment(self, run_id, idx, model, score, scores, rationale, cost, raw):75 self.exec("INSERT OR REPLACE INTO judgments VALUES (?,?,?,?,?,?,?,?,?)", (run_id, idx, model, score, json.dumps(scores), rationale, cost, utcnow(), raw[:20000] if raw else None))76 def set_metrics(self, run_id, m: dict):77 self.exec("INSERT OR REPLACE INTO metrics VALUES (?,?)", (run_id, json.dumps(m, default=str)))78 def ledger_add(self, run_id, kind, cost):79 self.exec("INSERT INTO ledger (utc,run_id,kind,cost_usd) VALUES (?,?,?,?)", (utcnow(), run_id, kind, float(cost or 0)))80 def ledger_total(self) -> float:81 r = self.one("SELECT COALESCE(SUM(cost_usd),0) AS t FROM ledger"); return float(r["t"]) if r else 0.082 def add_intervention(self, run_id, minutes, description):83 self.exec("INSERT INTO interventions (utc,run_id,minutes,description) VALUES (?,?,?,?)", (utcnow(), run_id, minutes, description))84