use std::collections::HashMap;
use std::error::Error;
use std::fs::{File, OpenOptions, TryLockError};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::time::Duration;
use rusqlite::{Connection, OptionalExtension, params};
use crate::class::Class;
use crate::complexity::Features;
use crate::route::{Approval, Policy};
use crate::scan::{Ci, Facts};
use crate::todo::Priority;
use crate::worker::{Parser, Worker};
pub type Result<T> = std::result::Result<T, Box<dyn Error>>;
const VERSION: i64 = 35;
const SCHEMA: &str = "
CREATE TABLE config (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE roots (path TEXT PRIMARY KEY);
CREATE TABLE projects (
name TEXT PRIMARY KEY,
path TEXT NOT NULL,
tier INTEGER CHECK (tier BETWEEN 1 AND 5),
scanned_at INTEGER,
has_todo INTEGER NOT NULL DEFAULT 0,
lint_errors INTEGER NOT NULL DEFAULT 0,
dirty INTEGER NOT NULL DEFAULT 0,
ahead INTEGER,
last_activity INTEGER,
ci TEXT NOT NULL DEFAULT 'unknown',
ci_detail TEXT NOT NULL DEFAULT '',
scan_error TEXT
);
CREATE TABLE tasks (
project TEXT NOT NULL REFERENCES projects(name) ON DELETE CASCADE,
key TEXT NOT NULL,
line INTEGER NOT NULL,
priority TEXT NOT NULL,
text TEXT NOT NULL,
tags TEXT NOT NULL,
due TEXT,
gh INTEGER,
heading TEXT,
added_at INTEGER,
first_seen INTEGER NOT NULL,
PRIMARY KEY (project, key)
);
";
const RUNS: &str = "
CREATE TABLE runs (
id INTEGER PRIMARY KEY,
project TEXT NOT NULL,
task_key TEXT NOT NULL,
text TEXT NOT NULL,
gh INTEGER,
quadrant TEXT,
agent TEXT NOT NULL,
repo TEXT NOT NULL,
branch TEXT NOT NULL,
worktree TEXT NOT NULL,
default_branch TEXT NOT NULL,
base TEXT NOT NULL,
prompt TEXT NOT NULL,
state TEXT NOT NULL,
feedback TEXT,
started_at INTEGER,
seconds INTEGER,
cost_usd REAL,
summary TEXT,
commits INTEGER,
diffstat TEXT,
verify TEXT,
verify_ok INTEGER,
error TEXT,
outcome TEXT
);
";
const DEPS_AND_NOTES: &str = "
ALTER TABLE projects ADD COLUMN deps INTEGER;
ALTER TABLE projects ADD COLUMN deps_detail TEXT NOT NULL DEFAULT '';
ALTER TABLE projects ADD COLUMN deps_at INTEGER;
CREATE TABLE notes (
id INTEGER PRIMARY KEY,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
text TEXT NOT NULL
);
";
const LEFTOVER: &str = "
ALTER TABLE projects ADD COLUMN leftover INTEGER NOT NULL DEFAULT 0;
";
const PR_OPEN: &str = "
UPDATE runs SET state = 'pr-open' WHERE state = 'shipped' AND outcome LIKE 'https://%/pull/%';
";
const ATTEMPTS: &str = "
CREATE TABLE attempts (
id INTEGER PRIMARY KEY,
run_id INTEGER NOT NULL REFERENCES runs(id),
n INTEGER NOT NULL,
agent TEXT NOT NULL,
prompt TEXT NOT NULL,
feedback TEXT,
started_at INTEGER NOT NULL,
seconds INTEGER,
cost_usd REAL,
summary TEXT,
verify TEXT,
verify_ok INTEGER,
error TEXT,
outcome TEXT,
UNIQUE (run_id, n)
);
ALTER TABLE runs ADD COLUMN dispatched_at INTEGER;
ALTER TABLE runs ADD COLUMN ready_at INTEGER;
ALTER TABLE runs ADD COLUMN decided_at INTEGER;
ALTER TABLE runs ADD COLUMN published_at INTEGER;
ALTER TABLE runs ADD COLUMN review_seconds INTEGER;
INSERT INTO attempts (run_id, n, agent, prompt, feedback, started_at, seconds,
cost_usd, summary, verify, verify_ok, error, outcome)
SELECT id, 1, agent, prompt, feedback, started_at, seconds, cost_usd, summary,
verify, verify_ok, error, NULL
FROM runs WHERE started_at IS NOT NULL;
UPDATE runs SET dispatched_at = started_at WHERE started_at IS NOT NULL;
ALTER TABLE runs DROP COLUMN started_at;
";
const SNAPSHOT: &str = "
ALTER TABLE runs ADD COLUMN class TEXT;
ALTER TABLE runs ADD COLUMN scope TEXT;
ALTER TABLE runs ADD COLUMN tier INTEGER;
ALTER TABLE runs ADD COLUMN description TEXT;
ALTER TABLE runs ADD COLUMN agent_budget REAL;
ALTER TABLE runs ADD COLUMN timeout_minutes INTEGER;
";
const VERIFY_BASE: &str = "
CREATE TABLE verify_base (
project TEXT NOT NULL,
base TEXT NOT NULL,
command TEXT NOT NULL,
timeout_minutes INTEGER NOT NULL,
ok INTEGER,
seconds INTEGER NOT NULL,
measured_at INTEGER NOT NULL,
PRIMARY KEY (project, base, command, timeout_minutes)
);
ALTER TABLE runs ADD COLUMN verify_base_ok INTEGER;
ALTER TABLE runs ADD COLUMN verify_base_seconds INTEGER;
";
const CHANGED_PATHS: &str = "
ALTER TABLE runs ADD COLUMN changed_paths TEXT;
ALTER TABLE runs ADD COLUMN scope_error TEXT;
";
const EXHAUSTION: &str = "
CREATE TABLE exhaustion (
project TEXT NOT NULL,
revision TEXT NOT NULL,
attempts INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (project, revision)
);
";
const AGENTS: &str = "
CREATE TABLE agents (
name TEXT PRIMARY KEY,
command TEXT NOT NULL,
args TEXT NOT NULL,
allow TEXT,
parse TEXT NOT NULL,
reports_cost INTEGER NOT NULL,
enforces_budget INTEGER NOT NULL,
sandbox INTEGER NOT NULL,
resumes INTEGER NOT NULL
);
ALTER TABLE runs ADD COLUMN model TEXT;
ALTER TABLE attempts ADD COLUMN model TEXT;
";
const RETIRE_KEYS: &str = "
DELETE FROM config WHERE key IN (
'dispatch_quadrants', 'overflow_quadrants',
'stale_after.1', 'stale_after.2', 'stale_after.3', 'stale_after.4', 'stale_after.5'
);
";
const COMPLEXITY: &str = "
ALTER TABLE runs ADD COLUMN complexity INTEGER;
ALTER TABLE runs ADD COLUMN features TEXT;
ALTER TABLE runs ADD COLUMN estimator TEXT;
";
const ROUTES: &str = "
CREATE TABLE routes (
revision INTEGER PRIMARY KEY,
document TEXT NOT NULL,
proposed_by TEXT NOT NULL,
created_at INTEGER NOT NULL,
activated_at INTEGER,
activated_by TEXT,
shadow INTEGER NOT NULL DEFAULT 0
);
ALTER TABLE runs ADD COLUMN route_revision INTEGER;
ALTER TABLE runs ADD COLUMN route TEXT;
ALTER TABLE runs ADD COLUMN approval TEXT;
";
const APPROVAL_EVIDENCE: &str = "
ALTER TABLE runs ADD COLUMN approved_tree TEXT;
ALTER TABLE runs ADD COLUMN approved_head TEXT;
ALTER TABLE runs ADD COLUMN approved_by TEXT;
";
const CAMPAIGNS: &str = "
CREATE TABLE campaigns (
name TEXT PRIMARY KEY,
text TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
class TEXT NOT NULL,
created_at INTEGER NOT NULL
);
CREATE TABLE campaign_members (
campaign TEXT NOT NULL REFERENCES campaigns(name) ON DELETE CASCADE,
project TEXT NOT NULL,
run_id INTEGER,
PRIMARY KEY (campaign, project)
);
";
const SETTINGS_SYNC: &str = "
CREATE TABLE IF NOT EXISTS settings_sync (text TEXT NOT NULL);
";
const READ_RUNS_DONE: &str = "
UPDATE runs SET state = 'done'
WHERE state = 'ready' AND branch = '' AND workflow_instance IS NOT NULL;
";
const JOB_APPROVAL_AND_STREAM: &str = r#"
ALTER TABLE jobs ADD COLUMN awaiting TEXT;
UPDATE agents
SET args = '["-p","{prompt}","--output-format","stream-json","--verbose","--permission-mode","acceptEdits","--model","{model}","--max-budget-usd","{budget}","{extra}"]'
WHERE name = 'claude'
AND args = '["-p","{prompt}","--output-format","json","--permission-mode","acceptEdits","--model","{model}","--max-budget-usd","{budget}","{extra}"]';
"#;
const JOBS: &str = "
CREATE TABLE jobs (
id INTEGER PRIMARY KEY,
args TEXT NOT NULL,
cwd TEXT NOT NULL,
state TEXT NOT NULL,
created_at INTEGER NOT NULL,
started_at INTEGER,
finished_at INTEGER,
pid INTEGER,
exit_code INTEGER
);
";
const INFRA_REJECTION: &str = "
ALTER TABLE runs ADD COLUMN infra INTEGER NOT NULL DEFAULT 0;
";
const VERIFY_BASE_SANDBOX: &str = "
DROP TABLE verify_base;
CREATE TABLE verify_base (
project TEXT NOT NULL,
base TEXT NOT NULL,
command TEXT NOT NULL,
timeout_minutes INTEGER NOT NULL,
sandbox INTEGER NOT NULL,
ok INTEGER,
seconds INTEGER NOT NULL,
measured_at INTEGER NOT NULL,
PRIMARY KEY (project, base, command, timeout_minutes, sandbox)
);
";
const REVIEWED_TREE: &str = "
ALTER TABLE runs ADD COLUMN reviewed_tree TEXT;
";
const PUSH_OR_PR: &str = "
UPDATE runs SET state = 'merged' WHERE state = 'shipped' AND outcome LIKE '%/pull/%';
UPDATE runs SET state = 'pushed' WHERE state = 'shipped';
UPDATE runs SET state = 'closed' WHERE state = 'rejected' AND outcome LIKE '%/pull/%';
DELETE FROM config WHERE key = 'publish' OR (key LIKE 'projects.%' AND key LIKE '%.publish');
";
const RETIRE_WEIGHTS: &str = "
DELETE FROM config WHERE key LIKE 'weights.%';
";
const GIT_SNAPSHOT: &str = "
ALTER TABLE runs ADD COLUMN git_snapshot TEXT;
";
const SANDUK_WORKER: &str = "-- no schema change; the seed is in Rust";
const PRESETS_REPLACE_WORKER_MODEL: &str = "
INSERT OR IGNORE INTO presets (name, agent, model, args, created_at)
SELECT name, name, model, '[]', strftime('%s', 'now') FROM agents
WHERE model IS NOT NULL AND model <> '';
INSERT OR REPLACE INTO config (key, value)
SELECT 'preset', a.name FROM agents a
WHERE a.model IS NOT NULL AND a.model <> ''
AND a.name = COALESCE((SELECT value FROM config WHERE key = 'agent'), 'claude')
AND NOT EXISTS (SELECT 1 FROM config WHERE key = 'preset' AND value <> '');
ALTER TABLE agents DROP COLUMN model;
";
const PRESETS: &str = "
CREATE TABLE presets (
name TEXT PRIMARY KEY,
agent TEXT NOT NULL,
model TEXT,
args TEXT NOT NULL DEFAULT '[]',
created_at INTEGER NOT NULL
);
ALTER TABLE runs ADD COLUMN preset TEXT;
ALTER TABLE runs ADD COLUMN extra_args TEXT;
";
const WORKER_MODEL: &str = "
ALTER TABLE agents ADD COLUMN model TEXT;
UPDATE agents SET model = (SELECT value FROM config WHERE key = 'model')
WHERE name = COALESCE((SELECT value FROM config WHERE key = 'agent'), 'claude')
AND EXISTS (SELECT 1 FROM config WHERE key = 'model' AND value <> '');
DELETE FROM config WHERE key = 'model';
";
const WORKER_ENV: &str = "
ALTER TABLE agents ADD COLUMN env TEXT NOT NULL DEFAULT '{}';
";
const WORKFLOWS: &str = "
CREATE TABLE workflows (
revision INTEGER PRIMARY KEY,
document TEXT NOT NULL,
source TEXT,
proposed_by TEXT NOT NULL,
created_at INTEGER NOT NULL,
activated_at INTEGER,
activated_by TEXT,
worst_case_runs INTEGER,
worst_case_cost REAL
);
CREATE TABLE workflow_instances (
id INTEGER PRIMARY KEY,
workflow TEXT NOT NULL,
revision INTEGER NOT NULL REFERENCES workflows(revision),
args TEXT NOT NULL,
target TEXT NOT NULL,
started_at INTEGER NOT NULL,
finished_at INTEGER,
outcome TEXT
);
CREATE TABLE workflow_units (
instance INTEGER NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE,
id TEXT NOT NULL,
type TEXT NOT NULL,
node TEXT NOT NULL,
parent TEXT,
root TEXT NOT NULL,
depth INTEGER NOT NULL DEFAULT 0,
lap INTEGER NOT NULL DEFAULT 0,
project TEXT,
data TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (instance, id)
);
CREATE INDEX idx_workflow_units_root ON workflow_units(instance, root);
CREATE TABLE workflow_moves (
instance INTEGER NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE,
unit TEXT NOT NULL,
edge INTEGER NOT NULL,
taken INTEGER NOT NULL,
reason TEXT,
at INTEGER NOT NULL,
PRIMARY KEY (instance, unit, edge)
);
CREATE TABLE workflow_verdicts (
instance INTEGER NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE,
unit TEXT NOT NULL,
check_name TEXT NOT NULL,
verdict TEXT NOT NULL,
detail TEXT,
at INTEGER NOT NULL,
PRIMARY KEY (instance, unit, check_name)
);
ALTER TABLE runs ADD COLUMN workflow_instance INTEGER;
ALTER TABLE runs ADD COLUMN node TEXT;
ALTER TABLE runs ADD COLUMN unit TEXT;
ALTER TABLE runs ADD COLUMN lap INTEGER NOT NULL DEFAULT 0;
";
const PROJECT_TAGS: &str = "
CREATE TABLE project_tags (
project TEXT NOT NULL,
tag TEXT NOT NULL,
PRIMARY KEY (project, tag)
);
CREATE INDEX idx_project_tags_tag ON project_tags(tag);
";
const ABSENCE: &str = "
ALTER TABLE projects ADD COLUMN absent_since INTEGER;
";
const SLUG: &str = "
ALTER TABLE projects ADD COLUMN slug TEXT;
";
pub struct Store {
conn: Connection,
settings: PathBuf,
}
#[derive(Debug, Clone, PartialEq)]
pub struct ProjectRow {
pub name: String,
pub path: PathBuf,
pub slug: Option<String>,
pub absent_since: Option<i64>,
pub tier: Option<u8>,
pub scanned_at: Option<i64>,
pub has_todo: bool,
pub lint_errors: i64,
pub dirty: i64,
pub leftover: i64,
pub ahead: Option<i64>,
pub last_activity: Option<i64>,
pub ci: Ci,
pub scan_error: Option<String>,
pub deps: Option<i64>,
pub deps_detail: String,
pub deps_at: Option<i64>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct Footprint {
pub tasks: i64,
pub tags: i64,
pub campaigns: i64,
pub runs: i64,
pub open_runs: Vec<i64>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct TaskRow {
pub project: String,
pub key: String,
pub line: i64,
pub priority: Priority,
pub text: String,
pub tags: Vec<String>,
pub due: Option<String>,
pub gh: Option<i64>,
pub group: Option<String>,
pub added_at: Option<i64>,
pub first_seen: i64,
}
pub struct Session {
_file: File,
}
impl Session {
pub fn try_acquire(dir: &Path) -> Result<Option<Session>> {
Session::try_lock(dir, "session.lock")
}
pub fn try_lock(dir: &Path, name: &str) -> Result<Option<Session>> {
let path = dir.join(name);
let io = |e: std::io::Error| format!("{}: {e}", path.display());
let mut file = OpenOptions::new()
.write(true)
.create(true)
.truncate(false)
.open(&path)
.map_err(io)?;
match file.try_lock() {
Ok(()) => {}
Err(TryLockError::WouldBlock) => return Ok(None),
Err(TryLockError::Error(e)) => return Err(io(e).into()),
}
file.set_len(0).map_err(io)?;
write!(file, "{}", std::process::id()).map_err(io)?;
Ok(Some(Session { _file: file }))
}
pub fn acquire(dir: &Path) -> Result<Session> {
const WAIT: Duration = Duration::from_secs(2);
let started = std::time::Instant::now();
loop {
if let Some(session) = Session::try_acquire(dir)? {
return Ok(session);
}
if started.elapsed() >= WAIT {
break;
}
std::thread::sleep(Duration::from_millis(50));
}
Err({
let pid = std::fs::read_to_string(dir.join("session.lock")).unwrap_or_default();
let holder = match pid.trim() {
"" => String::new(),
pid => format!(" (pid {pid})"),
};
format!(
"another pma session{holder} is dispatching, reworking, rejecting or publishing; wait for it to finish"
)
.into()
})
}
}
pub fn home() -> Result<PathBuf> {
if let Some(dir) = std::env::var_os("PMA_HOME") {
return Ok(PathBuf::from(dir));
}
let home = std::env::var_os("HOME").ok_or("HOME is not set")?;
Ok(PathBuf::from(home).join(".config").join("pma"))
}
pub fn state_home() -> Result<PathBuf> {
let var = |name: &str| std::env::var_os(name).filter(|v| !v.is_empty());
if let Some(dir) = var("PMA_STATE") {
return Ok(PathBuf::from(dir));
}
if let Some(dir) = var("PMA_HOME") {
return Ok(PathBuf::from(dir).join("state"));
}
if let Some(dir) = var("XDG_STATE_HOME") {
return Ok(PathBuf::from(dir).join("pma"));
}
let home = std::env::var_os("HOME").ok_or("HOME is not set")?;
Ok(PathBuf::from(home).join(".local").join("state").join("pma"))
}
impl Store {
pub fn open_default() -> Result<Store> {
let dir = home()?;
std::fs::create_dir_all(&dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
Store::open(&dir.join("projects.db"))
}
pub fn open(path: &Path) -> Result<Store> {
let conn = Connection::open(path).map_err(|e| format!("open {}: {e}", path.display()))?;
conn.query_row("PRAGMA journal_mode = DELETE", [], |r| {
r.get::<_, String>(0)
})?;
conn.pragma_update(None, "foreign_keys", "ON")?;
conn.busy_timeout(Duration::from_secs(5))?;
let version: i64 = conn.pragma_query_value(None, "user_version", |r| r.get(0))?;
match version {
0..VERSION => {
let steps = [
SCHEMA,
RUNS,
DEPS_AND_NOTES,
LEFTOVER,
PR_OPEN,
ATTEMPTS,
SNAPSHOT,
VERIFY_BASE,
CHANGED_PATHS,
EXHAUSTION,
AGENTS,
RETIRE_KEYS,
COMPLEXITY,
ROUTES,
APPROVAL_EVIDENCE,
CAMPAIGNS,
SLUG,
ABSENCE,
PROJECT_TAGS,
WORKFLOWS,
WORKER_ENV,
WORKER_MODEL,
PRESETS,
PRESETS_REPLACE_WORKER_MODEL,
SANDUK_WORKER,
GIT_SNAPSHOT,
RETIRE_WEIGHTS,
PUSH_OR_PR,
REVIEWED_TREE,
VERIFY_BASE_SANDBOX,
INFRA_REJECTION,
JOBS,
JOB_APPROVAL_AND_STREAM,
READ_RUNS_DONE,
SETTINGS_SYNC,
];
let tx = conn.unchecked_transaction()?;
for step in &steps[version as usize..] {
tx.execute_batch(step)?;
}
if version < 11 {
seed_agent(&tx, &crate::worker::Worker::claude())?;
}
let mut seeded = Vec::new();
if version < 21 {
seeded.extend([
crate::worker::Worker::opencode(),
crate::worker::Worker::omp(),
]);
}
if version < 25 {
seeded.push(crate::worker::Worker::sanduk());
}
for w in seeded {
if tx.query_row(
"SELECT count(*) FROM agents WHERE name = ?1",
[&w.name],
|r| r.get::<_, i64>(0),
)? == 0
{
seed_agent(&tx, &w)?;
}
}
tx.pragma_update(None, "user_version", VERSION)?;
tx.commit()?;
}
VERSION => {}
v => {
return Err(format!(
"{} has schema version {v}; this pma reads {VERSION}",
path.display()
)
.into());
}
}
let store = Store {
conn,
settings: path.with_file_name(crate::settings::FILE),
};
store.load_settings()?;
Ok(store)
}
fn load_settings(&self) -> Result<()> {
if !self.settings.exists() {
crate::settings::create(&self.settings, &self.settings_from_tables()?)?;
}
let (text, settings) = crate::settings::read(&self.settings)?;
let synced: Option<String> = self
.conn
.query_row("SELECT text FROM settings_sync", [], |r| r.get(0))
.optional()?;
if synced.as_deref() != Some(text.as_str()) {
self.apply_settings(&text, &settings)?;
}
Ok(())
}
fn settings_from_tables(&self) -> Result<crate::settings::Settings> {
let mut stmt = self
.conn
.prepare("SELECT name, tier FROM projects WHERE tier IS NOT NULL ORDER BY name")?;
let tiers = stmt
.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
.collect::<rusqlite::Result<_>>()?;
Ok(crate::settings::Settings {
config: self.config_rows()?,
roots: self.roots()?,
tiers,
tags: self.project_tags()?,
agents: self.agents()?,
presets: self.presets()?,
})
}
fn apply_settings(&self, text: &str, s: &crate::settings::Settings) -> Result<()> {
let tx = self.conn.unchecked_transaction()?;
tx.execute_batch(
"DELETE FROM config; DELETE FROM roots; DELETE FROM agents; DELETE FROM presets;
DELETE FROM project_tags; DELETE FROM settings_sync;
UPDATE projects SET tier = NULL;",
)?;
for (key, value) in &s.config {
tx.execute(
"INSERT OR REPLACE INTO config (key, value) VALUES (?1, ?2)",
params![key, value],
)?;
}
for root in &s.roots {
tx.execute(
"INSERT OR IGNORE INTO roots (path) VALUES (?1)",
[path_str(root)?],
)?;
}
for (name, tier) in &s.tiers {
tx.execute(
"UPDATE projects SET tier = ?2 WHERE name = ?1",
params![name, tier],
)?;
}
for (project, tag) in &s.tags {
tx.execute(
"INSERT OR IGNORE INTO project_tags (project, tag) VALUES (?1, ?2)",
params![project, tag],
)?;
}
for w in &s.agents {
seed_agent(&tx, w)?;
}
for p in &s.presets {
tx.execute(
"INSERT OR REPLACE INTO presets (name, agent, model, args, created_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![
p.name,
p.agent,
p.model,
serde_json::to_string(&p.args).unwrap_or_else(|_| "[]".into()),
crate::dates::now()
],
)?;
}
tx.execute("INSERT INTO settings_sync (text) VALUES (?1)", [text])?;
tx.commit()?;
Ok(())
}
fn change_settings<T>(&self, f: impl FnOnce(&mut toml_edit::DocumentMut) -> T) -> Result<T> {
let out = crate::settings::edit(&self.settings, f)?;
let (text, settings) = crate::settings::read(&self.settings)?;
self.apply_settings(&text, &settings)?;
Ok(out)
}
fn reapply_settings(&self) -> Result<()> {
let (text, settings) = crate::settings::read(&self.settings)?;
self.apply_settings(&text, &settings)
}
pub fn config_rows(&self) -> Result<Vec<(String, String)>> {
let mut stmt = self
.conn
.prepare("SELECT key, value FROM config ORDER BY key")?;
let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn set_config(&self, key: &str, value: &str) -> Result<()> {
self.change_settings(|doc| crate::settings::set_config(doc, key, value))
}
pub fn reset_config(&self, key: &str) -> Result<bool> {
self.change_settings(|doc| crate::settings::reset_config(doc, key))
}
pub fn roots(&self) -> Result<Vec<PathBuf>> {
let mut stmt = self.conn.prepare("SELECT path FROM roots ORDER BY path")?;
let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
Ok(rows
.map(|r| r.map(PathBuf::from))
.collect::<rusqlite::Result<_>>()?)
}
pub fn add_root(&self, path: &Path) -> Result<bool> {
let path = path_str(path)?;
self.change_settings(|doc| crate::settings::add_root(doc, path))
}
pub fn remove_root(&self, path: &Path) -> Result<bool> {
let path = path_str(path)?;
self.change_settings(|doc| crate::settings::remove_root(doc, path))
}
pub fn set_tier(&self, name: &str, path: &Path, tier: Option<u8>) -> Result<()> {
self.conn.execute(
"INSERT INTO projects (name, path, tier) VALUES (?1, ?2, ?3)
ON CONFLICT (name) DO UPDATE SET tier = excluded.tier",
params![name, path_str(path)?, tier],
)?;
self.change_settings(|doc| crate::settings::set_tier(doc, name, tier))
}
pub fn last_scan(&self) -> Result<Option<i64>> {
Ok(self
.conn
.query_row("SELECT MAX(scanned_at) FROM projects", [], |r| r.get(0))?)
}
pub fn project(&self, name: &str) -> Result<Option<ProjectRow>> {
Ok(self.projects()?.into_iter().find(|p| p.name == name))
}
pub fn projects(&self) -> Result<Vec<ProjectRow>> {
let mut stmt = self.conn.prepare(
"SELECT name, path, tier, scanned_at, has_todo, lint_errors, dirty, ahead,
last_activity, ci, ci_detail, scan_error, deps, deps_detail, deps_at,
leftover, slug, absent_since
FROM projects ORDER BY name",
)?;
let rows = stmt.query_map([], |r| {
Ok(ProjectRow {
name: r.get(0)?,
path: PathBuf::from(r.get::<_, String>(1)?),
tier: r.get(2)?,
scanned_at: r.get(3)?,
has_todo: r.get(4)?,
lint_errors: r.get(5)?,
dirty: r.get(6)?,
ahead: r.get(7)?,
last_activity: r.get(8)?,
ci: Ci::from_columns(&r.get::<_, String>(9)?, r.get::<_, String>(10)?),
scan_error: r.get(11)?,
deps: r.get(12)?,
deps_detail: r.get(13)?,
deps_at: r.get(14)?,
leftover: r.get(15)?,
slug: r.get(16)?,
absent_since: r.get(17)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn tasks(&self) -> Result<Vec<TaskRow>> {
let mut stmt = self.conn.prepare(
"SELECT project, line, priority, text, tags, due, gh, added_at, first_seen, heading, key
FROM tasks ORDER BY project, line",
)?;
let rows = stmt.query_map([], |r| {
let priority: String = r.get(2)?;
let tags: String = r.get(4)?;
Ok(TaskRow {
project: r.get(0)?,
line: r.get(1)?,
priority: Priority::parse(&priority).unwrap_or(Priority::Low),
text: r.get(3)?,
tags: tags.split_whitespace().map(String::from).collect(),
due: r.get(5)?,
gh: r.get(6)?,
added_at: r.get(7)?,
first_seen: r.get(8)?,
group: r.get(9)?,
key: r.get(10)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn save_scan(&mut self, facts: &[Facts], full: bool, now: i64) -> Result<Vec<String>> {
let open_runs: Vec<Run> = self
.runs()?
.into_iter()
.filter(|r| !r.state.is_final())
.collect();
let tx = self.conn.transaction()?;
for f in facts {
if f.held {
tx.execute(
"UPDATE projects SET scan_error = ?2, absent_since = NULL WHERE name = ?1",
params![f.name, f.error],
)?;
continue;
}
let (ci, ci_detail) = f.ci.to_columns();
let leftover = f
.pma_branches
.iter()
.filter(|b| {
!open_runs
.iter()
.any(|r| r.project == f.name && &r.branch == *b)
})
.count() as i64;
tx.execute(
"INSERT INTO projects (name, path, scanned_at, has_todo, lint_errors, dirty, ahead,
last_activity, ci, ci_detail, scan_error, leftover, slug)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
ON CONFLICT (name) DO UPDATE SET
path = excluded.path, scanned_at = excluded.scanned_at,
has_todo = CASE WHEN ?14 THEN projects.has_todo ELSE excluded.has_todo END,
lint_errors = CASE WHEN ?14 THEN projects.lint_errors
ELSE excluded.lint_errors END,
dirty = excluded.dirty, ahead = excluded.ahead,
last_activity = excluded.last_activity, ci = excluded.ci,
ci_detail = excluded.ci_detail, scan_error = excluded.scan_error,
leftover = excluded.leftover, slug = excluded.slug,
absent_since = NULL",
params![
f.name,
path_str(&f.path)?,
now,
f.todo.is_some() || f.todo_unread,
f.todo.as_ref().map_or(0, |t| t.lint_errors),
f.dirty,
f.ahead,
f.last_activity,
ci,
ci_detail,
f.error,
leftover,
f.slug,
f.todo_unread,
],
)?;
if let Some(d) = &f.deps {
tx.execute(
"UPDATE projects SET deps = ?2, deps_detail = ?3, deps_at = ?4 WHERE name = ?1",
params![f.name, d.outdated, d.detail, now],
)?;
}
if f.todo_unread {
continue;
}
type Seen = (i64, Option<i64>);
let (by_key, by_text): (HashMap<String, Seen>, HashMap<String, Seen>) = {
let mut stmt = tx.prepare(
"SELECT key, text, first_seen, added_at FROM tasks WHERE project = ?1",
)?;
let rows: Vec<(String, String, i64, Option<i64>)> = stmt
.query_map([&f.name], |r| {
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))
})?
.collect::<rusqlite::Result<_>>()?;
let by_text = rows
.iter()
.map(|(_, text, seen, added)| (crate::todo::normal_text(text), (*seen, *added)))
.collect();
(
rows.into_iter()
.map(|(k, _, seen, added)| (k, (seen, added)))
.collect(),
by_text,
)
};
tx.execute("DELETE FROM tasks WHERE project = ?1", [&f.name])?;
for item in f.todo.iter().flat_map(|t| &t.items) {
let before = by_key
.get(&item.key)
.or_else(|| by_text.get(&crate::todo::normal_text(&item.text)));
let first_seen = before.map_or(now, |(seen, _)| *seen);
let added_at = match (item.added_at, before.and_then(|(_, added)| *added)) {
(Some(a), Some(b)) => Some(a.min(b)),
(a, b) => a.or(b),
};
tx.execute(
"INSERT INTO tasks (project, key, line, priority, text, tags, due, gh, added_at, first_seen, heading)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
params![
f.name,
item.key,
item.line,
item.priority.name(),
item.text,
item.tags.join(" "),
item.due,
item.gh,
added_at,
first_seen,
item.group,
],
)?;
}
}
let mut absent = Vec::new();
if full {
let scanned: std::collections::HashSet<&str> =
facts.iter().map(|f| f.name.as_str()).collect();
let mut stmt = tx.prepare("SELECT name FROM projects WHERE absent_since IS NULL")?;
let present: Vec<String> = stmt
.query_map([], |r| r.get(0))?
.collect::<rusqlite::Result<_>>()?;
drop(stmt);
for name in present {
if !scanned.contains(name.as_str()) {
tx.execute(
"UPDATE projects SET absent_since = ?2 WHERE name = ?1",
params![name, now],
)?;
absent.push(name);
}
}
}
tx.commit()?;
self.reapply_settings()?;
Ok(absent)
}
pub fn project_footprint(&self, name: &str) -> Result<Footprint> {
let count =
|sql: &str| -> Result<i64> { Ok(self.conn.query_row(sql, [name], |r| r.get(0))?) };
let open_runs = self
.runs()?
.into_iter()
.filter(|r| r.project == name && !r.state.is_final())
.map(|r| r.id)
.collect();
Ok(Footprint {
tasks: count("SELECT count(*) FROM tasks WHERE project = ?1")?,
tags: count("SELECT count(*) FROM project_tags WHERE project = ?1")?,
campaigns: count("SELECT count(*) FROM campaign_members WHERE project = ?1")?,
runs: count("SELECT count(*) FROM runs WHERE project = ?1")?,
open_runs,
})
}
pub fn forget_project(&mut self, name: &str) -> Result<()> {
let tx = self.conn.transaction()?;
for sql in [
"DELETE FROM project_tags WHERE project = ?1",
"DELETE FROM campaign_members WHERE project = ?1",
"DELETE FROM exhaustion WHERE project = ?1",
"DELETE FROM verify_base WHERE project = ?1",
"DELETE FROM projects WHERE name = ?1",
] {
tx.execute(sql, [name])?;
}
tx.commit()?;
self.change_settings(|doc| crate::settings::forget_project(doc, name))
}
pub fn project_tags(&self) -> Result<Vec<(String, String)>> {
let mut stmt = self
.conn
.prepare("SELECT project, tag FROM project_tags ORDER BY project, tag")?;
let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn add_project_tag(&self, project: &str, tag: &str) -> Result<bool> {
self.change_settings(|doc| crate::settings::tag_project(doc, project, tag, true))
}
pub fn remove_project_tag(&self, project: &str, tag: &str) -> Result<bool> {
self.change_settings(|doc| crate::settings::tag_project(doc, project, tag, false))
}
pub fn projects_tagged(&self, tags: &[String]) -> Result<Vec<String>> {
let mut found: Vec<String> = Vec::new();
for (project, tag) in self.project_tags()? {
if tags.iter().any(|t| t == &tag) && !found.contains(&project) {
found.push(project);
}
}
found.sort();
Ok(found)
}
pub fn notes(&self) -> Result<Vec<Note>> {
let mut stmt = self
.conn
.prepare("SELECT id, created_at, updated_at, text FROM notes ORDER BY id")?;
let rows = stmt.query_map([], |r| {
Ok(Note {
id: r.get(0)?,
created_at: r.get(1)?,
updated_at: r.get(2)?,
text: r.get(3)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn add_job(&self, args: &[String], cwd: &Path) -> Result<i64> {
self.conn.execute(
"INSERT INTO jobs (args, cwd, state, created_at) VALUES (?1, ?2, 'queued', ?3)",
params![
serde_json::to_string(args)?,
cwd.to_string_lossy(),
crate::dates::now()
],
)?;
Ok(self.conn.last_insert_rowid())
}
pub fn jobs(&self, limit: i64) -> Result<Vec<Job>> {
let mut stmt = self.conn.prepare(&format!(
"SELECT {JOB_COLUMNS} FROM jobs ORDER BY id DESC LIMIT ?1"
))?;
let rows = stmt.query_map([limit], job_row)?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn job(&self, id: i64) -> Result<Job> {
self.conn
.query_row(
&format!("SELECT {JOB_COLUMNS} FROM jobs WHERE id = ?1"),
[id],
job_row,
)
.optional()?
.ok_or_else(|| format!("no job {id}; `pma job` lists them").into())
}
pub fn claim_job(&self) -> Result<Option<Job>> {
let Some(id) = self
.conn
.query_row(
"SELECT id FROM jobs WHERE state = 'queued' ORDER BY id LIMIT 1",
[],
|r| r.get::<_, i64>(0),
)
.optional()?
else {
return Ok(None);
};
self.conn.execute(
"UPDATE jobs SET state = 'running', started_at = ?2 WHERE id = ?1 AND state = 'queued'",
params![id, crate::dates::now()],
)?;
self.job(id).map(Some)
}
pub fn set_job_awaiting(&self, id: i64, args: &[String]) -> Result<()> {
self.conn.execute(
"UPDATE jobs SET awaiting = ?2 WHERE id = ?1",
params![id, serde_json::to_string(args)?],
)?;
Ok(())
}
pub fn approve_job(&self, id: i64) -> Result<bool> {
Ok(self.conn.execute(
"UPDATE jobs SET state = 'approved' WHERE id = ?1 AND state = 'awaiting-approval'",
[id],
)? > 0)
}
pub fn settle_jobs_awaiting(&self, plan: &str) -> Result<()> {
self.conn.execute(
"UPDATE jobs SET state = 'approved'
WHERE state = 'awaiting-approval' AND instr(awaiting, '\"' || ?1 || '\"') > 0",
[plan],
)?;
Ok(())
}
pub fn set_job_pid(&self, id: i64, pid: u32) -> Result<()> {
self.conn
.execute("UPDATE jobs SET pid = ?2 WHERE id = ?1", params![id, pid])?;
Ok(())
}
pub fn finish_job(&self, id: i64, exit_code: Option<i32>) -> Result<()> {
self.conn.execute(
"UPDATE jobs SET finished_at = ?2, exit_code = ?3,
state = CASE WHEN state = 'stopped' THEN 'stopped'
WHEN ?3 = 0 AND awaiting IS NOT NULL THEN 'awaiting-approval'
WHEN ?3 = 0 THEN 'done' ELSE 'failed' END
WHERE id = ?1",
params![id, crate::dates::now(), exit_code],
)?;
Ok(())
}
pub fn stop_job(&self, id: i64) -> Result<String> {
let job = self.job(id)?;
if matches!(job.state.as_str(), "queued" | "running") {
self.conn.execute(
"UPDATE jobs SET state = 'stopped', finished_at = ?2 WHERE id = ?1",
params![id, crate::dates::now()],
)?;
}
Ok(job.state)
}
pub fn fail_running_jobs(&self) -> Result<usize> {
Ok(self.conn.execute(
"UPDATE jobs SET state = 'failed', finished_at = ?1 WHERE state = 'running'",
[crate::dates::now()],
)?)
}
pub fn has_queued_jobs(&self) -> Result<bool> {
Ok(self.conn.query_row(
"SELECT count(*) FROM jobs WHERE state = 'queued'",
[],
|r| r.get::<_, i64>(0),
)? > 0)
}
pub fn add_note(&self, text: &str, now: i64) -> Result<i64> {
self.conn.execute(
"INSERT INTO notes (created_at, updated_at, text) VALUES (?1, ?1, ?2)",
params![now, text],
)?;
Ok(self.conn.last_insert_rowid())
}
pub fn edit_note(&self, id: i64, text: &str, now: i64) -> Result<bool> {
Ok(self.conn.execute(
"UPDATE notes SET text = ?2, updated_at = ?3 WHERE id = ?1",
params![id, text, now],
)? > 0)
}
pub fn remove_note(&self, id: i64) -> Result<bool> {
Ok(self.conn.execute("DELETE FROM notes WHERE id = ?1", [id])? > 0)
}
pub fn insert_run(&self, run: &mut Run) -> Result<i64> {
run.dispatched_at = Some(crate::dates::now());
self.conn.execute(
"INSERT INTO runs (project, task_key, text, gh, quadrant, agent, branch, worktree,
default_branch, base, prompt, state, repo, dispatched_at,
class, scope, tier, description, agent_budget, timeout_minutes,
verify, verify_base_ok, verify_base_seconds, model,
complexity, features, estimator,
route_revision, route, approval,
workflow_instance, node, unit, lap, preset, extra_args,
git_snapshot)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16,
?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27, ?28, ?29, ?30,
?31, ?32, ?33, ?34, ?35, ?36, ?37)",
params![
run.project,
run.task_key,
run.text,
run.gh,
run.quadrant,
run.agent,
run.branch,
path_str(&run.worktree)?,
run.default_branch,
run.base,
run.prompt,
run.state.name(),
path_str(&run.repo)?,
run.dispatched_at,
run.class.map(Class::name),
run.scope.join(","),
run.tier,
run.description,
run.agent_budget,
run.timeout_minutes,
run.verify,
run.verify_base_ok,
run.verify_base_seconds,
run.model,
run.complexity,
run.features.map(Features::to_json),
run.complexity.map(|_| crate::complexity::ESTIMATOR),
run.route_revision,
run.route,
run.approval.map(Approval::name),
run.workflow_instance,
run.node,
run.unit,
run.lap,
run.preset,
serde_json::to_string(&run.extra_args).unwrap_or_else(|_| "[]".into()),
run.git_snapshot,
],
)?;
run.id = self.conn.last_insert_rowid();
Ok(run.id)
}
pub fn update_run(&self, run: &Run) -> Result<()> {
self.write_run(run, None)
}
pub fn update_run_from(&self, run: &Run, read: RunState) -> Result<()> {
self.write_run(run, Some(read))
}
fn write_run(&self, run: &Run, read: Option<RunState>) -> Result<()> {
let written = self.conn.execute(
"UPDATE runs SET state = ?2, feedback = ?3, seconds = ?4, cost_usd = ?5,
summary = ?6, commits = ?7, diffstat = ?8, verify_ok = ?9,
error = ?10, outcome = ?11, prompt = ?12, ready_at = ?13,
decided_at = ?14, published_at = ?15, review_seconds = ?16,
changed_paths = ?17, scope_error = ?18, approved_tree = ?19,
approved_head = ?20, approved_by = ?21, reviewed_tree = ?22, infra = ?23
WHERE id = ?1 AND (?24 IS NULL OR state = ?24)",
params![
run.id,
run.state.name(),
run.feedback,
run.seconds,
run.cost_usd,
run.summary,
run.commits,
run.diffstat,
run.verify_ok,
run.error,
run.outcome,
run.prompt,
run.ready_at,
run.decided_at,
run.published_at,
run.review_seconds,
run.changed_paths
.as_ref()
.map(|p| serde_json::to_string(p).unwrap_or_default()),
run.scope_error,
run.approved_tree,
run.approved_head,
run.approved_by,
run.reviewed_tree,
run.infra,
read.map(RunState::name),
],
)?;
if written == 0 {
let now = self.run(run.id)?.state;
return Err(format!(
"run #{} is now {}; read it again with `pma review {}`",
run.id,
now.name(),
run.id
)
.into());
}
Ok(())
}
pub fn add_review_seconds(&self, id: i64, seconds: i64) -> Result<()> {
self.conn.execute(
"UPDATE runs SET review_seconds = coalesce(review_seconds, 0) + ?2 WHERE id = ?1",
params![id, seconds],
)?;
Ok(())
}
pub fn set_reviewed_tree(&self, id: i64, tree: &str) -> Result<()> {
self.conn.execute(
"UPDATE runs SET reviewed_tree = ?2 WHERE id = ?1",
params![id, tree],
)?;
Ok(())
}
pub fn runs(&self) -> Result<Vec<Run>> {
let mut stmt = self.conn.prepare(
"SELECT id, project, task_key, text, gh, quadrant, agent, branch, worktree,
default_branch, base, prompt, state, feedback, seconds,
cost_usd, summary, commits, diffstat, verify, verify_ok, error, outcome, repo,
dispatched_at, ready_at, decided_at, published_at, review_seconds,
class, scope, tier, description, agent_budget, timeout_minutes,
verify_base_ok, verify_base_seconds, changed_paths, scope_error, model,
complexity, features, estimator, route_revision, route, approval,
approved_tree, approved_head, approved_by,
workflow_instance, node, unit, lap, preset, extra_args, git_snapshot,
reviewed_tree, infra
FROM runs ORDER BY id",
)?;
let rows = stmt.query_map([], |r| {
let state: String = r.get(12)?;
Ok(Run {
id: r.get(0)?,
project: r.get(1)?,
task_key: r.get(2)?,
text: r.get(3)?,
gh: r.get(4)?,
quadrant: r.get(5)?,
agent: r.get(6)?,
branch: r.get(7)?,
worktree: PathBuf::from(r.get::<_, String>(8)?),
default_branch: r.get(9)?,
base: r.get(10)?,
prompt: r.get(11)?,
state: RunState::parse(&state).unwrap_or(RunState::Failed),
feedback: r.get(13)?,
seconds: r.get(14)?,
cost_usd: r.get(15)?,
summary: r.get(16)?,
commits: r.get(17)?,
diffstat: r.get(18)?,
verify: r.get(19)?,
verify_ok: r.get(20)?,
error: r.get(21)?,
outcome: r.get(22)?,
repo: PathBuf::from(r.get::<_, String>(23)?),
dispatched_at: r.get(24)?,
ready_at: r.get(25)?,
decided_at: r.get(26)?,
published_at: r.get(27)?,
review_seconds: r.get(28)?,
class: r
.get::<_, Option<String>>(29)?
.as_deref()
.and_then(Class::parse),
scope: r
.get::<_, Option<String>>(30)?
.unwrap_or_default()
.split(',')
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect(),
tier: r.get(31)?,
description: r.get(32)?,
agent_budget: r.get(33)?,
timeout_minutes: r.get(34)?,
verify_base_ok: r.get(35)?,
verify_base_seconds: r.get(36)?,
changed_paths: r
.get::<_, Option<String>>(37)?
.and_then(|s| serde_json::from_str(&s).ok()),
scope_error: r.get(38)?,
model: r.get(39)?,
complexity: r.get(40)?,
features: r
.get::<_, Option<String>>(41)?
.as_deref()
.and_then(Features::from_json),
estimator: r.get(42)?,
route_revision: r.get(43)?,
route: r.get(44)?,
approval: r
.get::<_, Option<String>>(45)?
.as_deref()
.and_then(Approval::parse),
approved_tree: r.get(46)?,
approved_head: r.get(47)?,
approved_by: r.get(48)?,
workflow_instance: r.get(49)?,
node: r.get(50)?,
unit: r.get(51)?,
lap: r.get(52)?,
preset: r.get(53)?,
extra_args: r
.get::<_, Option<String>>(54)?
.and_then(|s| serde_json::from_str(&s).ok())
.unwrap_or_default(),
git_snapshot: r.get(55)?,
reviewed_tree: r.get(56)?,
infra: r.get(57)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn run(&self, id: i64) -> Result<Run> {
self.runs()?
.into_iter()
.find(|r| r.id == id)
.ok_or_else(|| format!("no run #{id}").into())
}
pub fn add_campaign(&self, c: &Campaign) -> Result<()> {
if c.members.is_empty() {
return Err(format!("campaign `{}` selects no project", c.name).into());
}
let tx = self.conn.unchecked_transaction()?;
tx.execute(
"INSERT INTO campaigns (name, text, description, class, created_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![c.name, c.text, c.description, c.class.name(), c.created_at],
)
.map_err(|e| format!("campaign `{}`: {e}", c.name))?;
for (project, _) in &c.members {
tx.execute(
"INSERT INTO campaign_members (campaign, project) VALUES (?1, ?2)",
params![c.name, project],
)?;
}
tx.commit()?;
Ok(())
}
pub fn campaigns(&self) -> Result<Vec<Campaign>> {
let mut stmt = self.conn.prepare(
"SELECT name, text, description, class, created_at FROM campaigns ORDER BY name",
)?;
let rows: Vec<Campaign> = stmt
.query_map([], |r| {
let class: String = r.get(3)?;
Ok(Campaign {
name: r.get(0)?,
text: r.get(1)?,
description: r.get(2)?,
class: Class::parse(&class).unwrap_or(Class::Specified),
created_at: r.get(4)?,
members: Vec::new(),
})
})?
.collect::<rusqlite::Result<_>>()?;
let mut stmt = self
.conn
.prepare("SELECT campaign, project, run_id FROM campaign_members ORDER BY project")?;
let members: Vec<(String, String, Option<i64>)> = stmt
.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
.collect::<rusqlite::Result<_>>()?;
Ok(rows
.into_iter()
.map(|mut c| {
c.members = members
.iter()
.filter(|(name, ..)| *name == c.name)
.map(|(_, p, run)| (p.clone(), *run))
.collect();
c
})
.collect())
}
pub fn campaign(&self, name: &str) -> Result<Campaign> {
self.campaigns()?
.into_iter()
.find(|c| c.name == name)
.ok_or_else(|| format!("unknown campaign `{name}`; see `pma campaign`").into())
}
pub fn set_campaign_run(&self, campaign: &str, project: &str, run: i64) -> Result<()> {
self.conn.execute(
"UPDATE campaign_members SET run_id = ?3 WHERE campaign = ?1 AND project = ?2",
params![campaign, project, run],
)?;
Ok(())
}
pub fn remove_campaign(&self, name: &str) -> Result<bool> {
self.conn
.execute("DELETE FROM campaign_members WHERE campaign = ?1", [name])?;
Ok(self
.conn
.execute("DELETE FROM campaigns WHERE name = ?1", [name])?
> 0)
}
pub fn add_route_revision(&self, document: &str, by: &str) -> Result<i64> {
let document = Policy::parse(document)?.to_json();
let next: i64 = self.conn.query_row(
"SELECT COALESCE(MAX(revision), 0) + 1 FROM routes",
[],
|r| r.get(0),
)?;
self.conn.execute(
"INSERT INTO routes (revision, document, proposed_by, created_at)
VALUES (?1, ?2, ?3, ?4)",
params![next, document, by, crate::dates::now()],
)?;
Ok(next)
}
pub fn activate_route(&self, revision: i64, by: &str, shadow: bool) -> Result<()> {
let tx = self.conn.unchecked_transaction()?;
let found = tx.execute(
"UPDATE routes SET activated_at = ?2, activated_by = ?3, shadow = ?4
WHERE revision = ?1",
params![revision, crate::dates::now(), by, shadow],
)?;
if found == 0 {
return Err(format!("no policy revision {revision}").into());
}
tx.execute(
"UPDATE routes SET activated_at = NULL, activated_by = NULL WHERE revision <> ?1",
[revision],
)?;
tx.commit()?;
Ok(())
}
pub fn route_revisions(&self) -> Result<Vec<RouteRevision>> {
let mut stmt = self.conn.prepare(
"SELECT revision, document, proposed_by, created_at, activated_at, activated_by,
shadow
FROM routes ORDER BY revision",
)?;
let rows = stmt.query_map([], |r| {
Ok(RouteRevision {
revision: r.get(0)?,
document: r.get(1)?,
proposed_by: r.get(2)?,
created_at: r.get(3)?,
activated_at: r.get(4)?,
activated_by: r.get(5)?,
shadow: r.get(6)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn active_route(&self) -> Result<Option<RouteRevision>> {
Ok(self
.route_revisions()?
.into_iter()
.find(|r| r.activated_at.is_some()))
}
pub fn add_workflow_revision(
&self,
doc: &crate::workflow::Document,
source: Option<&str>,
by: &str,
worst_case_runs: i64,
worst_case_cost: f64,
) -> Result<i64> {
let next: i64 = self.conn.query_row(
"SELECT COALESCE(MAX(revision), 0) + 1 FROM workflows",
[],
|r| r.get(0),
)?;
self.conn.execute(
"INSERT INTO workflows
(revision, document, source, proposed_by, created_at,
worst_case_runs, worst_case_cost)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
next,
doc.to_json(),
source,
by,
crate::dates::now(),
worst_case_runs,
worst_case_cost
],
)?;
Ok(next)
}
pub fn activate_workflow(&self, revision: i64, by: &str) -> Result<()> {
let tx = self.conn.unchecked_transaction()?;
let found = tx.execute(
"UPDATE workflows SET activated_at = ?2, activated_by = ?3 WHERE revision = ?1",
params![revision, crate::dates::now(), by],
)?;
if found == 0 {
return Err(format!("no workflow revision {revision}").into());
}
tx.execute(
"UPDATE workflows SET activated_at = NULL, activated_by = NULL WHERE revision <> ?1",
[revision],
)?;
tx.commit()?;
Ok(())
}
pub fn workflow_revisions(&self) -> Result<Vec<WorkflowRevision>> {
let mut stmt = self.conn.prepare(
"SELECT revision, document, source, proposed_by, created_at, activated_at,
activated_by, worst_case_runs, worst_case_cost
FROM workflows ORDER BY revision",
)?;
let rows = stmt.query_map([], |r| {
Ok(WorkflowRevision {
revision: r.get(0)?,
document: r.get(1)?,
source: r.get(2)?,
proposed_by: r.get(3)?,
created_at: r.get(4)?,
activated_at: r.get(5)?,
activated_by: r.get(6)?,
worst_case_runs: r.get(7)?,
worst_case_cost: r.get(8)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn active_workflow(&self) -> Result<Option<WorkflowRevision>> {
Ok(self
.workflow_revisions()?
.into_iter()
.find(|r| r.activated_at.is_some()))
}
pub fn workflow_revision(&self, revision: i64) -> Result<Option<WorkflowRevision>> {
Ok(self
.workflow_revisions()?
.into_iter()
.find(|r| r.revision == revision))
}
pub fn add_workflow_instance(
&self,
workflow: &str,
revision: i64,
args: &str,
target: &str,
) -> Result<i64> {
self.conn.execute(
"INSERT INTO workflow_instances (workflow, revision, args, target, started_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![workflow, revision, args, target, crate::dates::now()],
)?;
Ok(self.conn.last_insert_rowid())
}
pub fn workflow_instances(&self) -> Result<Vec<WorkflowInstance>> {
let mut stmt = self.conn.prepare(
"SELECT id, workflow, revision, args, target, started_at, finished_at, outcome
FROM workflow_instances ORDER BY id",
)?;
let rows = stmt.query_map([], |r| {
Ok(WorkflowInstance {
id: r.get(0)?,
workflow: r.get(1)?,
revision: r.get(2)?,
args: r.get(3)?,
target: r.get(4)?,
started_at: r.get(5)?,
finished_at: r.get(6)?,
outcome: r.get(7)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn atomically<T>(&self, work: impl FnOnce() -> Result<T>) -> Result<T> {
let tx = rusqlite::Transaction::new_unchecked(
&self.conn,
rusqlite::TransactionBehavior::Immediate,
)?;
let out = work()?;
tx.commit()?;
Ok(out)
}
pub fn workflow_instance(&self, instance: i64) -> Result<Option<WorkflowInstance>> {
Ok(self
.workflow_instances()?
.into_iter()
.find(|i| i.id == instance))
}
pub fn note_instance_outcome(&self, instance: i64, outcome: &str) -> Result<()> {
self.conn.execute(
"UPDATE workflow_instances SET outcome = ?2 WHERE id = ?1",
params![instance, outcome],
)?;
Ok(())
}
pub fn reopen_instance(&self, instance: i64, args: &str) -> Result<()> {
self.conn.execute(
"UPDATE workflow_instances SET args = ?2, outcome = NULL WHERE id = ?1",
params![instance, args],
)?;
Ok(())
}
pub fn finish_instance(&self, instance: i64, outcome: &str) -> Result<()> {
self.conn.execute(
"UPDATE workflow_instances SET finished_at = ?2, outcome = ?3 WHERE id = ?1",
params![instance, crate::dates::now(), outcome],
)?;
Ok(())
}
pub fn add_workflow_unit(&self, instance: i64, u: &WorkflowUnit) -> Result<()> {
self.conn.execute(
"INSERT INTO workflow_units
(instance, id, type, node, parent, root, depth, lap, project, data, created_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
params![
instance,
u.id,
u.ty,
u.node,
u.parent,
u.root,
u.depth,
u.lap,
u.project,
u.data,
crate::dates::now()
],
)?;
Ok(())
}
pub fn workflow_units(&self, instance: i64) -> Result<Vec<WorkflowUnit>> {
let mut stmt = self.conn.prepare(
"SELECT id, type, node, parent, root, depth, lap, project, data
FROM workflow_units WHERE instance = ?1 ORDER BY rowid",
)?;
let rows = stmt.query_map([instance], |r| {
Ok(WorkflowUnit {
id: r.get(0)?,
ty: r.get(1)?,
node: r.get(2)?,
parent: r.get(3)?,
root: r.get(4)?,
depth: r.get(5)?,
lap: r.get(6)?,
project: r.get(7)?,
data: r.get(8)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn add_workflow_move(
&self,
instance: i64,
unit: &str,
edge: usize,
taken: bool,
reason: Option<&str>,
) -> Result<()> {
self.conn.execute(
"INSERT OR REPLACE INTO workflow_moves (instance, unit, edge, taken, reason, at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
instance,
unit,
edge as i64,
taken,
reason,
crate::dates::now()
],
)?;
Ok(())
}
pub fn workflow_moves(&self, instance: i64) -> Result<Vec<(String, usize, bool)>> {
let mut stmt = self
.conn
.prepare("SELECT unit, edge, taken FROM workflow_moves WHERE instance = ?1")?;
let rows = stmt.query_map([instance], |r| {
Ok((
r.get::<_, String>(0)?,
r.get::<_, i64>(1)? as usize,
r.get::<_, bool>(2)?,
))
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn add_workflow_verdict(
&self,
instance: i64,
unit: &str,
check: &str,
verdict: &str,
detail: Option<&str>,
) -> Result<()> {
self.conn.execute(
"INSERT OR REPLACE INTO workflow_verdicts
(instance, unit, check_name, verdict, detail, at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![instance, unit, check, verdict, detail, crate::dates::now()],
)?;
Ok(())
}
pub fn workflow_verdicts(&self, instance: i64) -> Result<Vec<(String, String, String)>> {
let mut stmt = self.conn.prepare(
"SELECT unit, check_name, verdict FROM workflow_verdicts WHERE instance = ?1",
)?;
let rows = stmt.query_map([instance], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn set_preset(&self, p: &Preset) -> Result<()> {
if self.agent(&p.agent).is_err() {
return Err(format!("no worker `{}`; `pma agent` lists them", p.agent).into());
}
self.change_settings(|doc| crate::settings::set_preset(doc, p))
}
pub fn presets(&self) -> Result<Vec<Preset>> {
let mut stmt = self
.conn
.prepare("SELECT name, agent, model, args FROM presets ORDER BY name")?;
let rows = stmt.query_map([], |r| {
Ok(Preset {
name: r.get(0)?,
agent: r.get(1)?,
model: r.get::<_, Option<String>>(2)?.filter(|m| !m.is_empty()),
args: serde_json::from_str(&r.get::<_, String>(3)?).unwrap_or_default(),
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn preset(&self, name: &str) -> Result<Preset> {
self.presets()?
.into_iter()
.find(|p| p.name == name)
.ok_or_else(|| format!("no preset `{name}`; `pma preset` lists them").into())
}
pub fn remove_preset(&self, name: &str) -> Result<bool> {
self.change_settings(|doc| crate::settings::remove_preset(doc, name))
}
pub fn agents(&self) -> Result<Vec<Worker>> {
let mut stmt = self.conn.prepare(
"SELECT name, command, args, allow, parse, reports_cost, enforces_budget,
sandbox, resumes, env
FROM agents ORDER BY name",
)?;
let rows = stmt.query_map([], |r| {
let args: String = r.get(2)?;
let parse: String = r.get(4)?;
Ok(Worker {
name: r.get(0)?,
command: r.get(1)?,
args: serde_json::from_str(&args).unwrap_or_default(),
allow: r.get::<_, Option<String>>(3)?.filter(|a| !a.is_empty()),
parse: Parser::parse(&parse).unwrap_or(Parser::TextTail),
reports_cost: r.get(5)?,
enforces_budget: r.get(6)?,
sandbox: r.get(7)?,
resumes: r.get(8)?,
env: serde_json::from_str(&r.get::<_, String>(9)?).unwrap_or_default(),
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn agent(&self, name: &str) -> Result<Worker> {
self.agents()?
.into_iter()
.find(|w| w.name == name)
.ok_or_else(|| format!("unknown agent `{name}`; see `pma agent`").into())
}
pub fn set_agent(&self, w: &Worker) -> Result<()> {
self.change_settings(|doc| crate::settings::set_agent(doc, w))
}
pub fn remove_agent(&self, name: &str) -> Result<bool> {
self.change_settings(|doc| crate::settings::remove_agent(doc, name))
}
pub fn decided_runs(&self, project: &str) -> Result<(i64, i64)> {
let mut decided = 0;
let mut accepted = 0;
for run in self.runs()?.iter().filter(|r| r.project == project) {
match run.state {
RunState::Approved | RunState::Pushed | RunState::Merged => {
decided += 1;
accepted += 1;
}
RunState::Rejected if run.infra => {}
RunState::Rejected | RunState::Closed => decided += 1,
_ => {}
}
}
Ok((decided, accepted))
}
pub fn rekey_runs(&self, project: &str, old: &str, text: &str, new: &str) -> Result<()> {
let text = crate::todo::normal_text(text);
for run in self.runs()? {
if run.project == project
&& !run.state.is_final()
&& (run.task_key == old || crate::todo::normal_text(&run.text) == text)
{
self.conn.execute(
"UPDATE runs SET task_key = ?2 WHERE id = ?1",
params![run.id, new],
)?;
}
}
Ok(())
}
pub fn consumed_attempts(&self, project: &str, revision: &str) -> Result<i64> {
Ok(self
.conn
.query_row(
"SELECT attempts FROM exhaustion WHERE project = ?1 AND revision = ?2",
params![project, revision],
|r| r.get(0),
)
.optional()?
.unwrap_or(0))
}
pub fn consume_attempt(&self, project: &str, revision: &str) -> Result<i64> {
self.conn.execute(
"INSERT INTO exhaustion (project, revision, attempts, updated_at)
VALUES (?1, ?2, 1, ?3)
ON CONFLICT (project, revision)
DO UPDATE SET attempts = attempts + 1, updated_at = ?3",
params![project, revision, crate::dates::now()],
)?;
self.consumed_attempts(project, revision)
}
pub fn refund_attempts(&self, project: &str, revision: &str, n: i64) -> Result<()> {
self.conn.execute(
"UPDATE exhaustion SET attempts = max(attempts - ?3, 0), updated_at = ?4
WHERE project = ?1 AND revision = ?2",
params![project, revision, n, crate::dates::now()],
)?;
Ok(())
}
pub fn reset_attempts(&self, project: &str, revision: &str) -> Result<bool> {
Ok(self.conn.execute(
"DELETE FROM exhaustion WHERE project = ?1 AND revision = ?2",
params![project, revision],
)? > 0)
}
pub fn verify_base(
&self,
project: &str,
base: &str,
command: &str,
timeout_minutes: i64,
sandbox: bool,
) -> Result<Option<(Option<bool>, i64)>> {
let mut stmt = self.conn.prepare(
"SELECT ok, seconds FROM verify_base
WHERE project = ?1 AND base = ?2 AND command = ?3 AND timeout_minutes = ?4
AND sandbox = ?5",
)?;
let mut rows = stmt.query_map(
params![project, base, command, timeout_minutes, sandbox],
|r| Ok((r.get(0)?, r.get(1)?)),
)?;
Ok(rows.next().transpose()?)
}
#[allow(clippy::too_many_arguments)]
pub fn set_verify_base(
&self,
project: &str,
base: &str,
command: &str,
timeout_minutes: i64,
sandbox: bool,
ok: Option<bool>,
seconds: i64,
) -> Result<()> {
self.conn.execute(
"INSERT OR REPLACE INTO verify_base
(project, base, command, timeout_minutes, sandbox, ok, seconds, measured_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
project,
base,
command,
timeout_minutes,
sandbox,
ok,
seconds,
crate::dates::now()
],
)?;
Ok(())
}
pub fn insert_attempt(&self, a: &Attempt) -> Result<i64> {
self.conn.execute(
"INSERT INTO attempts (run_id, n, agent, prompt, feedback, started_at, seconds,
cost_usd, summary, verify, verify_ok, error, outcome, model)
VALUES (?1, (SELECT COALESCE(MAX(n), 0) + 1 FROM attempts WHERE run_id = ?1),
?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
params![
a.run_id,
a.agent,
a.prompt,
a.feedback,
a.started_at,
a.seconds,
a.cost_usd,
a.summary,
a.verify,
a.verify_ok,
a.error,
a.outcome,
a.model,
],
)?;
Ok(self.conn.last_insert_rowid())
}
pub fn attempts(&self, run: Option<i64>) -> Result<Vec<Attempt>> {
let mut stmt = self.conn.prepare(
"SELECT id, run_id, n, agent, prompt, feedback, started_at, seconds, cost_usd,
summary, verify, verify_ok, error, outcome, model
FROM attempts WHERE (?1 IS NULL OR run_id = ?1) ORDER BY run_id, n",
)?;
let rows = stmt.query_map([run], |r| {
Ok(Attempt {
id: r.get(0)?,
run_id: r.get(1)?,
n: r.get(2)?,
agent: r.get(3)?,
prompt: r.get(4)?,
feedback: r.get(5)?,
started_at: r.get(6)?,
seconds: r.get(7)?,
cost_usd: r.get(8)?,
summary: r.get(9)?,
verify: r.get(10)?,
verify_ok: r.get(11)?,
error: r.get(12)?,
outcome: r.get(13)?,
model: r.get(14)?,
})
})?;
Ok(rows.collect::<rusqlite::Result<_>>()?)
}
pub fn fail_interrupted_runs(&self, _held: &Session) -> Result<usize> {
Ok(self.conn.execute(
"UPDATE runs SET state = 'failed', error = 'interrupted: pma exited during the run'
WHERE state IN ('queued', 'running')",
[],
)?)
}
#[cfg(test)]
fn journal_mode(&self) -> String {
self.conn
.query_row("PRAGMA journal_mode", [], |r| r.get(0))
.unwrap()
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct Job {
pub id: i64,
pub args: Vec<String>,
pub cwd: PathBuf,
pub state: String,
pub created_at: i64,
pub started_at: Option<i64>,
pub finished_at: Option<i64>,
pub pid: Option<i64>,
pub exit_code: Option<i64>,
pub awaiting: Option<Vec<String>>,
}
const JOB_COLUMNS: &str =
"id, args, cwd, state, created_at, started_at, finished_at, pid, exit_code, awaiting";
fn job_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Job> {
let args: String = r.get(1)?;
Ok(Job {
id: r.get(0)?,
args: serde_json::from_str(&args).unwrap_or_default(),
cwd: PathBuf::from(r.get::<_, String>(2)?),
state: r.get(3)?,
created_at: r.get(4)?,
started_at: r.get(5)?,
finished_at: r.get(6)?,
pid: r.get(7)?,
exit_code: r.get(8)?,
awaiting: r
.get::<_, Option<String>>(9)?
.and_then(|a| serde_json::from_str(&a).ok()),
})
}
#[derive(Debug, Clone, PartialEq)]
pub struct Note {
pub id: i64,
pub created_at: i64,
pub updated_at: i64,
pub text: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum RunState {
#[default]
Queued,
Running,
Ready,
Failed,
Approved,
PrOpen,
Rejected,
Pushed,
Merged,
Closed,
Applied,
Done,
}
impl RunState {
const ALL: [(&'static str, RunState); 12] = [
("queued", RunState::Queued),
("running", RunState::Running),
("ready", RunState::Ready),
("failed", RunState::Failed),
("approved", RunState::Approved),
("pr-open", RunState::PrOpen),
("rejected", RunState::Rejected),
("pushed", RunState::Pushed),
("merged", RunState::Merged),
("closed", RunState::Closed),
("applied", RunState::Applied),
("done", RunState::Done),
];
pub fn name(self) -> &'static str {
Self::ALL
.iter()
.find(|(_, s)| *s == self)
.map_or("", |(n, _)| n)
}
fn parse(s: &str) -> Option<RunState> {
Self::ALL.iter().find(|(n, _)| *n == s).map(|(_, v)| *v)
}
pub fn is_final(self) -> bool {
matches!(
self,
RunState::Rejected
| RunState::Pushed
| RunState::Merged
| RunState::Closed
| RunState::Applied
| RunState::Done
)
}
pub fn landed(self) -> bool {
matches!(self, RunState::Pushed | RunState::Merged)
}
}
#[derive(Debug, Clone, PartialEq, Default)]
pub struct Run {
pub id: i64,
pub project: String,
pub task_key: String,
pub text: String,
pub gh: Option<i64>,
pub quadrant: Option<String>,
pub agent: String,
pub repo: PathBuf,
pub branch: String,
pub worktree: PathBuf,
pub default_branch: String,
pub base: String,
pub prompt: String,
pub state: RunState,
pub feedback: Option<String>,
pub seconds: Option<i64>,
pub cost_usd: Option<f64>,
pub summary: Option<String>,
pub commits: Option<i64>,
pub diffstat: Option<String>,
pub verify: Option<String>,
pub verify_ok: Option<bool>,
pub error: Option<String>,
pub outcome: Option<String>,
pub dispatched_at: Option<i64>,
pub ready_at: Option<i64>,
pub decided_at: Option<i64>,
pub published_at: Option<i64>,
pub review_seconds: Option<i64>,
pub class: Option<Class>,
pub scope: Vec<String>,
pub tier: Option<u8>,
pub description: Option<String>,
pub agent_budget: Option<f64>,
pub timeout_minutes: Option<i64>,
pub verify_base_ok: Option<bool>,
pub verify_base_seconds: Option<i64>,
pub changed_paths: Option<Vec<String>>,
pub scope_error: Option<String>,
pub model: Option<String>,
pub complexity: Option<i64>,
pub features: Option<Features>,
pub estimator: Option<String>,
pub route_revision: Option<i64>,
pub route: Option<String>,
pub approval: Option<Approval>,
pub approved_tree: Option<String>,
pub approved_head: Option<String>,
pub approved_by: Option<String>,
pub workflow_instance: Option<i64>,
pub node: Option<String>,
pub unit: Option<String>,
pub lap: i64,
pub preset: Option<String>,
pub extra_args: Vec<String>,
pub git_snapshot: Option<String>,
pub reviewed_tree: Option<String>,
pub infra: bool,
}
impl Run {
#[cfg(test)]
pub fn blank() -> Run {
Run {
id: 0,
project: "p".into(),
task_key: "fix it".into(),
text: "Fix it".into(),
gh: None,
quadrant: None,
agent: "claude".into(),
repo: PathBuf::from("/r/p"),
branch: "pma/fix-it".into(),
worktree: PathBuf::from("/w/p/fix-it"),
default_branch: "main".into(),
base: "abc".into(),
prompt: "do it".into(),
state: RunState::Queued,
feedback: None,
seconds: None,
cost_usd: None,
summary: None,
commits: None,
diffstat: None,
verify: None,
verify_ok: None,
error: None,
outcome: None,
dispatched_at: None,
ready_at: None,
decided_at: None,
published_at: None,
review_seconds: None,
class: None,
scope: Vec::new(),
tier: None,
description: None,
agent_budget: None,
timeout_minutes: None,
verify_base_ok: None,
verify_base_seconds: None,
changed_paths: None,
scope_error: None,
model: None,
complexity: None,
features: None,
estimator: None,
route_revision: None,
route: None,
approval: None,
approved_tree: None,
approved_head: None,
approved_by: None,
workflow_instance: None,
node: None,
unit: None,
lap: 0,
preset: None,
extra_args: Vec::new(),
git_snapshot: None,
reviewed_tree: None,
infra: false,
}
}
pub fn enter(&mut self, state: RunState) {
let now = crate::dates::now();
match state {
RunState::Ready => self.ready_at = Some(now),
RunState::Approved | RunState::Rejected => {
self.decided_at.get_or_insert(now);
}
RunState::Pushed | RunState::PrOpen => {
self.published_at.get_or_insert(now);
}
_ => {}
}
self.state = state;
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct Attempt {
pub id: i64,
pub run_id: i64,
pub n: i64,
pub agent: String,
pub prompt: String,
pub feedback: Option<String>,
pub started_at: i64,
pub seconds: Option<i64>,
pub cost_usd: Option<f64>,
pub summary: Option<String>,
pub verify: Option<String>,
pub verify_ok: Option<bool>,
pub error: Option<String>,
pub outcome: Option<String>,
pub model: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct Campaign {
pub name: String,
pub text: String,
pub description: String,
pub class: Class,
pub created_at: i64,
pub members: Vec<(String, Option<i64>)>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct RouteRevision {
pub revision: i64,
pub document: String,
pub proposed_by: String,
pub created_at: i64,
pub activated_at: Option<i64>,
pub activated_by: Option<String>,
pub shadow: bool,
}
#[derive(Debug, Clone, PartialEq, Default)]
pub struct Preset {
pub name: String,
pub agent: String,
pub model: Option<String>,
pub args: Vec<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowInstance {
pub id: i64,
pub workflow: String,
pub revision: i64,
pub args: String,
pub target: String,
pub started_at: i64,
pub finished_at: Option<i64>,
pub outcome: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowUnit {
pub id: String,
pub ty: String,
pub node: String,
pub parent: Option<String>,
pub root: String,
pub depth: i64,
pub lap: i64,
pub project: Option<String>,
pub data: String,
}
#[derive(Debug, Clone, PartialEq)]
pub struct WorkflowRevision {
pub revision: i64,
pub document: String,
pub source: Option<String>,
pub proposed_by: String,
pub created_at: i64,
pub activated_at: Option<i64>,
pub activated_by: Option<String>,
pub worst_case_runs: Option<i64>,
pub worst_case_cost: Option<f64>,
}
impl WorkflowRevision {
pub fn document(&self) -> Result<crate::workflow::Document> {
Ok(crate::workflow::Document::parse(&self.document)?)
}
}
impl RouteRevision {
pub fn policy(&self) -> Result<Policy> {
Ok(Policy::parse(&self.document)?)
}
}
fn seed_agent(conn: &Connection, w: &Worker) -> Result<()> {
conn.execute(
"INSERT OR REPLACE INTO agents
(name, command, args, allow, parse, reports_cost, enforces_budget, sandbox,
resumes, env)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
w.name,
w.command,
serde_json::to_string(&w.args).unwrap_or_default(),
w.allow,
w.parse.name(),
w.reports_cost,
w.enforces_budget,
w.sandbox,
w.resumes,
serde_json::to_string(&w.env).unwrap_or_else(|_| "{}".into()),
],
)?;
Ok(())
}
fn path_str(path: &Path) -> Result<&str> {
path.to_str()
.ok_or_else(|| format!("{} is not valid UTF-8", path.display()).into())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::scan::{DepsFacts, ScannedItem, TodoFacts};
fn scratch(name: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!("pma-store-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
fn item(key: &str, line: i64) -> ScannedItem {
ScannedItem {
key: key.into(),
line,
priority: Priority::High,
text: key.into(),
tags: vec!["a".into(), "b".into()],
due: Some("2026-10-01".into()),
gh: Some(4),
group: Some("Security".into()),
added_at: Some(100),
}
}
fn facts(name: &str, items: Vec<ScannedItem>) -> Facts {
Facts {
name: name.into(),
path: PathBuf::from(format!("/r/{name}")),
slug: Some(format!("shakfu/{name}")),
todo: Some(TodoFacts {
lint_errors: 1,
items,
}),
dirty: 3,
pma_branches: vec![],
ahead: None,
last_activity: Some(42),
ci: Ci::Failing(vec!["test".into(), "wheels".into()]),
deps: None,
error: None,
todo_unread: false,
held: false,
}
}
#[test]
fn opens_with_rollback_journal_and_reopens() {
let dir = scratch("open");
let db = dir.join("p.db");
let store = Store::open(&db).unwrap();
assert_eq!(store.journal_mode(), "delete");
store.set_config("tiers.1", "0.9").unwrap();
drop(store);
let store = Store::open(&db).unwrap();
assert_eq!(
store.config_rows().unwrap(),
[("tiers.1".to_string(), "0.9".to_string())]
);
assert!(store.reset_config("tiers.1").unwrap());
assert!(!store.reset_config("tiers.1").unwrap());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn shipped_splits_into_pushed_merged_and_closed() {
let dir = scratch("push-or-pr");
let db = dir.join("v27.db");
let conn = Connection::open(&db).unwrap();
let steps = [
SCHEMA,
RUNS,
DEPS_AND_NOTES,
LEFTOVER,
PR_OPEN,
ATTEMPTS,
SNAPSHOT,
VERIFY_BASE,
CHANGED_PATHS,
EXHAUSTION,
AGENTS,
RETIRE_KEYS,
COMPLEXITY,
ROUTES,
APPROVAL_EVIDENCE,
CAMPAIGNS,
SLUG,
ABSENCE,
PROJECT_TAGS,
WORKFLOWS,
WORKER_ENV,
WORKER_MODEL,
PRESETS,
PRESETS_REPLACE_WORKER_MODEL,
SANDUK_WORKER,
GIT_SNAPSHOT,
RETIRE_WEIGHTS,
];
for step in steps {
conn.execute_batch(step).unwrap();
}
conn.pragma_update(None, "user_version", 27).unwrap();
for (state, outcome) in [
("shipped", "https://github.com/o/r/pull/3"),
("shipped", "pushed abc to main"),
("rejected", "https://github.com/o/r/pull/4"),
("rejected", ""),
] {
conn.execute(
"INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
default_branch, base, prompt, state, outcome)
VALUES ('p', 'k', 't', 'claude', '/k', 'pma/t', '/w', 'main', 'b', 'p', ?1,
NULLIF(?2, ''))",
params![state, outcome],
)
.unwrap();
}
for (key, value) in [
("publish", "push"),
("projects.a.publish", "pr"),
("timeout", "5"),
] {
conn.execute(
"INSERT INTO config (key, value) VALUES (?1, ?2)",
params![key, value],
)
.unwrap();
}
drop(conn);
let store = Store::open(&db).unwrap();
let states: Vec<_> = store.runs().unwrap().iter().map(|r| r.state).collect();
assert_eq!(
states,
[
RunState::Merged,
RunState::Pushed,
RunState::Closed,
RunState::Rejected
]
);
let keys: Vec<String> = store
.config_rows()
.unwrap()
.into_iter()
.map(|(k, _)| k)
.collect();
assert_eq!(keys, ["timeout"]);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn older_databases_are_upgraded_in_place() {
let dir = scratch("migrate");
for (version, steps) in [
(1, &[SCHEMA][..]),
(2, &[SCHEMA, RUNS][..]),
(3, &[SCHEMA, RUNS, DEPS_AND_NOTES][..]),
(4, &[SCHEMA, RUNS, DEPS_AND_NOTES, LEFTOVER][..]),
] {
let db = dir.join(format!("v{version}.db"));
let conn = Connection::open(&db).unwrap();
for step in steps {
conn.execute_batch(step).unwrap();
}
conn.pragma_update(None, "user_version", version).unwrap();
conn.execute(
"INSERT INTO projects (name, path) VALUES ('kept', '/k')",
[],
)
.unwrap();
if version >= 2 {
for outcome in ["https://github.com/o/r/pull/3", "pushed abc to main"] {
conn.execute(
"INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
default_branch, base, prompt, state, outcome)
VALUES ('kept', 'k', 't', 'claude', '/k', 'pma/t', '/w', 'main', 'b',
'p', 'shipped', ?1)",
[outcome],
)
.unwrap();
}
}
drop(conn);
let store = Store::open(&db).unwrap();
let p = store.project("kept").unwrap().unwrap();
assert_eq!(
(p.deps, p.deps_detail.as_str(), p.leftover),
(None, "", 0),
"v{version}"
);
let states: Vec<_> = store.runs().unwrap().iter().map(|r| r.state).collect();
if version >= 2 {
assert_eq!(
states,
[RunState::PrOpen, RunState::Pushed],
"v{version}: a published pull request is open again, and a push is `pushed`"
);
}
assert!(store.notes().unwrap().is_empty());
drop(store);
assert!(Store::open(&db).is_ok(), "v{version} reopens");
}
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_v5_database_gains_an_attempt_per_run() {
let dir = scratch("attempts-migrate");
let db = dir.join("v5.db");
let conn = Connection::open(&db).unwrap();
for step in [SCHEMA, RUNS, DEPS_AND_NOTES, LEFTOVER, PR_OPEN] {
conn.execute_batch(step).unwrap();
}
conn.pragma_update(None, "user_version", 5).unwrap();
conn.execute_batch(
"INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
default_branch, base, prompt, state, started_at, seconds,
cost_usd, summary, verify, verify_ok)
VALUES ('p', 'k', 't', 'claude', '/r', 'pma/t', '/w', 'main', 'b', 'do it',
'ready', 100, 42, 0.5, 'first summary', 'make test', 1);
INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
default_branch, base, prompt, state)
VALUES ('p', 'k2', 't2', 'claude', '/r', 'pma/t2', '/w2', 'main', 'b', 'do it',
'queued');",
)
.unwrap();
drop(conn);
let store = Store::open(&db).unwrap();
let runs = store.runs().unwrap();
assert_eq!(runs[0].dispatched_at, Some(100));
assert_eq!(runs[1].dispatched_at, None);
let attempts = store.attempts(None).unwrap();
assert_eq!(attempts.len(), 1, "only the run that started");
let a = &attempts[0];
assert_eq!((a.run_id, a.n, a.started_at), (runs[0].id, 1, 100));
assert_eq!(a.summary.as_deref(), Some("first summary"));
assert_eq!(
(a.seconds, a.cost_usd, a.verify_ok),
(Some(42), Some(0.5), Some(true))
);
assert_eq!(a.outcome, None, "the old row did not record one");
assert_eq!(runs[0].class, None);
assert_eq!(runs[0].tier, None);
assert!(runs[0].scope.is_empty());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn upgrading_seeds_the_worker_that_was_compiled_in() {
let dir = scratch("agents");
let store = Store::open(&dir.join("p.db")).unwrap();
assert_eq!(
store.agents().unwrap(),
[
Worker::claude(),
Worker::omp(),
Worker::opencode(),
Worker::sanduk()
]
);
let mut w = Worker::claude();
w.name = "codex".into();
w.command = "codex".into();
w.args = vec![
"exec".into(),
"-C".into(),
"{dir}".into(),
"{prompt}".into(),
];
w.allow = None;
w.parse = Parser::TextTail;
w.reports_cost = false;
w.sandbox = true;
store.set_agent(&w).unwrap();
assert_eq!(
store.agents().unwrap(),
[
Worker::claude(),
w.clone(),
Worker::omp(),
Worker::opencode(),
Worker::sanduk()
]
);
assert_eq!(store.agent("codex").unwrap(), w);
assert!(store.agent("cursor").is_err());
assert!(store.remove_agent("codex").unwrap());
assert!(!store.remove_agent("codex").unwrap());
assert_eq!(
store.agents().unwrap(),
[
Worker::claude(),
Worker::omp(),
Worker::opencode(),
Worker::sanduk()
]
);
let mut edited = Worker::opencode();
edited.env = std::collections::BTreeMap::from([(
"OPENAI_BASE_URL".to_string(),
"http://localhost:11434/v1".to_string(),
)]);
store.set_agent(&edited).unwrap();
drop(store);
let store = Store::open(&dir.join("p.db")).unwrap();
assert_eq!(store.agent("opencode").unwrap(), edited);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn the_claude_template_moves_to_stream_json_unless_edited() {
let old = r#"["-p","{prompt}","--output-format","json","--permission-mode","acceptEdits","--model","{model}","--max-budget-usd","{budget}","{extra}"]"#;
for (args, streams) in [(old, true), (r#"["-p","{prompt}"]"#, false)] {
let dir = scratch("stream-json");
let path = dir.join("p.db");
{
let store = Store::open(&path).unwrap();
store
.conn
.execute_batch("ALTER TABLE jobs DROP COLUMN awaiting;")
.unwrap();
store
.conn
.execute("UPDATE agents SET args = ?1 WHERE name = 'claude'", [args])
.unwrap();
store.conn.pragma_update(None, "user_version", 32).unwrap();
}
let store = Store::open(&path).unwrap();
let got = store.agent("claude").unwrap().args;
assert_eq!(got.contains(&"stream-json".to_string()), streams, "{got:?}");
if streams {
assert_eq!(got, Worker::claude().args);
}
let _ = std::fs::remove_dir_all(dir);
}
}
#[test]
fn a_ready_read_run_is_done_after_the_migration() {
let dir = scratch("read-runs");
let path = dir.join("p.db");
{
let store = Store::open(&path).unwrap();
for (branch, instance) in [("", Some(1)), ("pma/t", Some(1)), ("pma/u", None)] {
store
.conn
.execute(
"INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
default_branch, base, prompt, state, workflow_instance)
VALUES ('p', 'k', 't', 'claude', '/k', ?1, '/w', 'main', 'b', 'p',
'ready', ?2)",
params![branch, instance],
)
.unwrap();
}
store.conn.pragma_update(None, "user_version", 33).unwrap();
}
let store = Store::open(&path).unwrap();
let states: Vec<_> = store.runs().unwrap().iter().map(|r| r.state).collect();
assert_eq!(states, [RunState::Done, RunState::Ready, RunState::Ready]);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn retiring_a_workers_model_leaves_a_preset_in_its_place() {
let dir = scratch("worker-model");
let path = dir.join("p.db");
{
let store = Store::open(&path).unwrap();
store.set_config("agent", "claude").unwrap();
store
.conn
.execute_batch(
"ALTER TABLE agents ADD COLUMN model TEXT;
ALTER TABLE runs DROP COLUMN git_snapshot;
ALTER TABLE runs DROP COLUMN reviewed_tree;
ALTER TABLE runs DROP COLUMN infra;
DROP TABLE jobs;",
)
.unwrap();
store
.conn
.execute("UPDATE agents SET model = 'opus' WHERE name = 'claude'", [])
.unwrap();
store.conn.pragma_update(None, "user_version", 23).unwrap();
}
let store = Store::open(&path).unwrap();
assert_eq!(
store.preset("claude").unwrap(),
Preset {
name: "claude".into(),
agent: "claude".into(),
model: Some("opus".into()),
args: Vec::new(),
}
);
assert_eq!(
store
.config_rows()
.unwrap()
.iter()
.find(|(k, _)| k == "preset"),
Some(&("preset".to_string(), "claude".to_string())),
"and it is the default, because that worker was the configured one"
);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn consumed_attempts_are_kept_per_task_revision() {
let dir = scratch("exhaustion");
let store = Store::open(&dir.join("p.db")).unwrap();
assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 0);
assert_eq!(store.consume_attempt("p", "fix the parser").unwrap(), 1);
assert_eq!(store.consume_attempt("p", "fix the parser").unwrap(), 2);
assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 2);
assert_eq!(
store
.consumed_attempts("p", "fix the parser on empty input")
.unwrap(),
0
);
assert_eq!(store.consumed_attempts("q", "fix the parser").unwrap(), 0);
store.refund_attempts("p", "fix the parser", 1).unwrap();
assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 1);
store.refund_attempts("p", "fix the parser", 3).unwrap();
assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 0);
store.consume_attempt("p", "fix the parser").unwrap();
assert!(store.reset_attempts("p", "fix the parser").unwrap());
assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 0);
assert!(!store.reset_attempts("p", "fix the parser").unwrap());
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn base_verification_is_cached_per_command_and_timeout() {
let dir = scratch("verify-base");
let store = Store::open(&dir.join("p.db")).unwrap();
assert_eq!(
store
.verify_base("p", "abc", "make test", 30, true)
.unwrap(),
None
);
store
.set_verify_base("p", "abc", "make test", 30, true, Some(false), 12)
.unwrap();
assert_eq!(
store
.verify_base("p", "abc", "make test", 30, true)
.unwrap(),
Some((Some(false), 12))
);
for (base, command, timeout, sandbox) in [
("def", "make test", 30, true),
("abc", "cargo test", 30, true),
("abc", "make test", 10, true),
("abc", "make test", 30, false),
] {
assert_eq!(
store
.verify_base("p", base, command, timeout, sandbox)
.unwrap(),
None,
"{base} {command} {timeout} {sandbox}"
);
}
store
.set_verify_base("p", "def", "make test", 30, true, None, 0)
.unwrap();
assert_eq!(
store
.verify_base("p", "def", "make test", 30, true)
.unwrap(),
Some((None, 0))
);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_guarded_write_refuses_a_run_that_moved_on() {
let dir = scratch("guarded");
let store = Store::open(&dir.join("p.db")).unwrap();
let mut run = Run::blank();
run.state = RunState::Ready;
store.insert_run(&mut run).unwrap();
let mut stale = store.run(run.id).unwrap();
let mut published = store.run(run.id).unwrap();
published.state = RunState::Pushed;
store.update_run(&published).unwrap();
stale.state = RunState::Approved;
let err = store
.update_run_from(&stale, RunState::Ready)
.unwrap_err()
.to_string();
assert!(err.contains("is now pushed"), "{err}");
assert_eq!(store.run(run.id).unwrap().state, RunState::Pushed);
store.update_run_from(&stale, RunState::Pushed).unwrap();
store.add_review_seconds(run.id, 60).unwrap();
store.add_review_seconds(run.id, 60).unwrap();
assert_eq!(store.run(run.id).unwrap().review_seconds, Some(120));
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn atomically_waits_for_another_writer() {
let dir = scratch("immediate");
let path = dir.join("p.db");
let store = Store::open(&path).unwrap();
let (held, holding) = std::sync::mpsc::channel();
let other = path.clone();
let writer = std::thread::spawn(move || {
let conn = Connection::open(&other).unwrap();
conn.execute_batch("BEGIN IMMEDIATE; INSERT INTO config VALUES ('a', '1');")
.unwrap();
held.send(()).unwrap();
std::thread::sleep(Duration::from_millis(300));
conn.execute_batch("COMMIT").unwrap();
});
holding.recv().unwrap();
store
.atomically(|| {
let n: i64 = store
.conn
.query_row("SELECT count(*) FROM config", [], |r| r.get(0))?;
store
.conn
.execute("INSERT INTO config VALUES ('b', ?1)", [n.to_string()])?;
Ok(())
})
.unwrap();
writer.join().unwrap();
assert_eq!(store.config_rows().unwrap().len(), 2);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn an_infrastructure_rejection_is_not_decided() {
let dir = scratch("decided");
let store = Store::open(&dir.join("p.db")).unwrap();
for (state, infra) in [
(RunState::Merged, false),
(RunState::Rejected, false),
(RunState::Rejected, true),
] {
let mut run = Run::blank();
run.project = "p".into();
store.insert_run(&mut run).unwrap();
run.state = state;
run.infra = infra;
store.update_run(&run).unwrap();
}
assert_eq!(store.decided_runs("p").unwrap(), (2, 1));
assert!(store.runs().unwrap()[2].infra, "stored and read back");
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn attempts_are_appended_never_replaced() {
let dir = scratch("attempts");
let store = Store::open(&dir.join("p.db")).unwrap();
let mut run = Run::blank();
store.insert_run(&mut run).unwrap();
assert!(run.dispatched_at.is_some(), "stamped on insert");
let mut a = Attempt {
id: 0,
run_id: run.id,
n: 0,
agent: "claude".into(),
prompt: "do it".into(),
feedback: None,
started_at: 10,
seconds: Some(30),
cost_usd: Some(0.2),
summary: Some("first".into()),
verify: Some("make test".into()),
verify_ok: Some(false),
error: None,
outcome: Some("ready".into()),
model: Some("haiku".into()),
};
store.insert_attempt(&a).unwrap();
a.feedback = Some("try again".into());
a.summary = Some("second".into());
a.verify_ok = Some(true);
store.insert_attempt(&a).unwrap();
let rows = store.attempts(Some(run.id)).unwrap();
assert_eq!(rows.iter().map(|r| r.n).collect::<Vec<_>>(), [1, 2]);
assert_eq!(rows[0].summary.as_deref(), Some("first"));
assert_eq!(rows[0].verify_ok, Some(false));
assert_eq!(rows[1].feedback.as_deref(), Some("try again"));
assert!(store.attempts(None).unwrap().len() == 2);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn transitions_are_stamped_once() {
let mut run = Run::blank();
run.enter(RunState::Ready);
let ready = run.ready_at.expect("ready stamped");
run.enter(RunState::Approved);
let decided = run.decided_at.expect("approval stamped");
run.enter(RunState::PrOpen);
let published = run.published_at.expect("publication stamped");
run.enter(RunState::Merged);
assert_eq!(
run.published_at,
Some(published),
"a merge is not a publish"
);
run.enter(RunState::Rejected);
assert_eq!(run.decided_at, Some(decided), "the approval stands");
assert_eq!(run.ready_at, Some(ready));
}
#[test]
fn notes_add_edit_and_remove() {
let dir = scratch("notes");
let store = Store::open(&dir.join("p.db")).unwrap();
let a = store
.add_note("move CI to one reusable workflow", 10)
.unwrap();
let b = store.add_note("second", 20).unwrap();
assert!(
store
.edit_note(a, "move CI to reusable workflows", 30)
.unwrap()
);
assert!(!store.edit_note(99, "x", 30).unwrap());
assert!(store.remove_note(b).unwrap());
assert!(!store.remove_note(b).unwrap());
assert_eq!(
store.notes().unwrap(),
[Note {
id: a,
created_at: 10,
updated_at: 30,
text: "move CI to reusable workflows".into()
}]
);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn runs_round_trip_and_interrupted_runs_fail() {
let dir = scratch("runs");
let mut store = Store::open(&dir.join("p.db")).unwrap();
let mut run = Run {
id: 0,
project: "p".into(),
task_key: "fix it".into(),
text: "Fix it".into(),
gh: Some(3),
quadrant: Some("Q1".into()),
agent: "claude".into(),
repo: PathBuf::from("/r/p"),
branch: "pma/fix-it".into(),
worktree: PathBuf::from("/w/p/fix-it"),
default_branch: "main".into(),
base: "abc".into(),
prompt: "do it".into(),
state: RunState::Running,
feedback: None,
seconds: None,
cost_usd: None,
summary: None,
commits: None,
diffstat: None,
verify: Some("make test".into()),
verify_ok: None,
error: None,
outcome: None,
dispatched_at: None,
ready_at: None,
decided_at: None,
published_at: None,
review_seconds: None,
class: Some(Class::Mechanical),
scope: vec!["Cargo.lock".into(), "Cargo.toml".into()],
tier: Some(2),
description: Some("within the constraints\n".into()),
agent_budget: Some(1.0),
timeout_minutes: Some(30),
verify_base_ok: Some(true),
verify_base_seconds: Some(12),
changed_paths: None,
scope_error: None,
model: Some("sonnet".into()),
complexity: Some(2),
features: Some(Features {
text_words: 3,
..Features::default()
}),
estimator: Some(crate::complexity::ESTIMATOR.into()),
route_revision: Some(2),
route: Some("chores".into()),
approval: Some(Approval::Batch),
approved_tree: None,
approved_head: None,
approved_by: None,
workflow_instance: None,
node: None,
unit: None,
lap: 0,
preset: None,
extra_args: Vec::new(),
git_snapshot: Some("hook pre-commit abc executable".into()),
reviewed_tree: None,
infra: false,
};
store.insert_run(&mut run).unwrap();
assert_eq!(store.run(run.id).unwrap(), run);
run.state = RunState::Ready;
run.cost_usd = Some(0.25);
run.changed_paths = Some(vec!["Cargo.lock".into(), "a b/c\nd".into()]);
run.verify_ok = Some(false);
run.summary = Some("done".into());
store.update_run(&run).unwrap();
assert_eq!(store.run(run.id).unwrap(), run);
run.verify = Some("make something-else".into());
store.update_run(&run).unwrap();
assert_eq!(
store.run(run.id).unwrap().verify.as_deref(),
Some("make test")
);
run.verify = Some("make test".into());
run.state = RunState::PrOpen;
store.update_run(&run).unwrap();
assert_eq!(store.run(run.id).unwrap().state, RunState::PrOpen);
assert!(!run.state.is_final(), "an open pull request holds its task");
run.state = RunState::Running;
store.update_run(&run).unwrap();
let session = Session::acquire(&dir).unwrap();
assert_eq!(store.fail_interrupted_runs(&session).unwrap(), 1);
let failed = store.run(run.id).unwrap();
assert_eq!(failed.state, RunState::Failed);
assert!(failed.error.unwrap().starts_with("interrupted"));
assert!(store.run(99).is_err());
let mut scanned = facts("p", vec![]);
scanned.pma_branches = vec!["pma/fix-it".into(), "pma/stray".into()];
let mut other = facts("q", vec![]);
other.pma_branches = vec!["pma/fix-it".into()];
let leftover = |store: &mut Store, facts: &[Facts]| {
store.save_scan(facts, false, 1).unwrap();
let projects = store.projects().unwrap();
projects.iter().map(|p| p.leftover).collect::<Vec<_>>()
};
assert_eq!(
leftover(&mut store, &[scanned.clone(), other.clone()]),
[1, 1]
);
run.state = RunState::Pushed;
store.update_run(&run).unwrap();
assert_eq!(leftover(&mut store, &[scanned, other]), [2, 1]);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn one_session_holds_the_lock_until_dropped() {
let dir = scratch("session");
let first = Session::acquire(&dir).unwrap();
assert!(Session::try_acquire(&dir).unwrap().is_none());
let err = Session::acquire(&dir).err().unwrap().to_string();
assert!(
err.contains(&format!("(pid {})", std::process::id())),
"{err}"
);
drop(first);
assert!(Session::try_acquire(&dir).unwrap().is_some());
let brief = Session::acquire(&dir).unwrap();
let release = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
drop(brief);
});
assert!(Session::acquire(&dir).is_ok(), "a brief hold is waited out");
release.join().unwrap();
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn the_last_scan_owns_the_slug() {
let dir = scratch("slug");
let mut store = Store::open(&dir.join("p.db")).unwrap();
store
.save_scan(&[facts("one", vec![])], true, 1000)
.unwrap();
assert_eq!(
store.project("one").unwrap().unwrap().slug.as_deref(),
Some("shakfu/one")
);
let mut moved = facts("one", vec![]);
moved.slug = None;
store.save_scan(&[moved], true, 2000).unwrap();
assert_eq!(store.project("one").unwrap().unwrap().slug, None);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn forgetting_a_project_keeps_its_runs() {
let dir = scratch("forget");
let mut store = Store::open(&dir.join("p.db")).unwrap();
store
.save_scan(
&[facts("a", vec![item("t", 1)]), facts("b", vec![])],
true,
1,
)
.unwrap();
store.set_tier("a", Path::new("/r/a"), Some(1)).unwrap();
store.add_project_tag("a", "ai").unwrap();
store.add_project_tag("b", "ai").unwrap();
let mut run = Run::blank();
run.project = "a".into();
store.insert_run(&mut run).unwrap();
store.consume_attempt("a", "t").unwrap();
let f = store.project_footprint("a").unwrap();
assert_eq!((f.tasks, f.tags, f.runs), (1, 1, 1));
assert_eq!(f.open_runs, [run.id], "an open run holds a worktree");
let mut open = store.run(run.id).unwrap();
open.state = RunState::Merged;
store.update_run(&open).unwrap();
assert!(store.project_footprint("a").unwrap().open_runs.is_empty());
store.forget_project("a").unwrap();
assert!(store.project("a").unwrap().is_none());
assert!(store.tasks().unwrap().is_empty(), "its tasks go");
assert_eq!(
store.projects_tagged(&["ai".into()]).unwrap(),
["b"],
"another project keeps the tag"
);
assert_eq!(store.consumed_attempts("a", "t").unwrap(), 0);
assert_eq!(store.run(run.id).unwrap().project, "a", "the run is kept");
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_project_carries_several_tags() {
let dir = scratch("tags");
let store = Store::open(&dir.join("p.db")).unwrap();
assert!(store.add_project_tag("cyllama", "ai").unwrap());
assert!(
!store.add_project_tag("cyllama", "ai").unwrap(),
"idempotent"
);
store.add_project_tag("cyllama", "audio").unwrap();
store.add_project_tag("cysox", "audio").unwrap();
assert_eq!(
store.project_tags().unwrap(),
[
("cyllama".to_string(), "ai".to_string()),
("cyllama".to_string(), "audio".to_string()),
("cysox".to_string(), "audio".to_string()),
]
);
assert_eq!(
store.projects_tagged(&["audio".into()]).unwrap(),
["cyllama", "cysox"]
);
assert_eq!(
store
.projects_tagged(&["ai".into(), "audio".into()])
.unwrap(),
["cyllama", "cysox"],
"a project in both tags is listed once"
);
assert!(store.remove_project_tag("cyllama", "audio").unwrap());
assert!(!store.remove_project_tag("cyllama", "audio").unwrap());
assert_eq!(store.projects_tagged(&["audio".into()]).unwrap(), ["cysox"]);
assert_eq!(
store.projects_tagged(&["ai".into()]).unwrap(),
["cyllama"],
"removing one tag leaves the other"
);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn settings_toml_takes_over_the_configuration_tables() {
let dir = scratch("settings");
let path = dir.join("p.db");
{
let store = Store::open(&path).unwrap();
store
.conn
.execute_batch(
"INSERT INTO config (key, value) VALUES ('agent_budget', '3');
INSERT INTO roots (path) VALUES ('/r');
INSERT INTO project_tags (project, tag) VALUES ('a', 'py');",
)
.unwrap();
store
.conn
.execute_batch("DELETE FROM settings_sync")
.unwrap();
}
std::fs::remove_file(dir.join(crate::settings::FILE)).unwrap();
let mut store = Store::open(&path).unwrap();
let text = std::fs::read_to_string(dir.join(crate::settings::FILE)).unwrap();
assert!(
text.contains("agent_budget = 3") && text.contains("roots = [\"/r\"]"),
"{text}"
);
assert!(
text.contains("[agents.claude]"),
"the seeded workers: {text}"
);
std::fs::write(
dir.join(crate::settings::FILE),
text.replace("agent_budget = 3", "agent_budget = 4") + "\n[projects.b]\ntier = 2\n",
)
.unwrap();
drop(store);
store = Store::open(&path).unwrap();
assert!(
store
.config_rows()
.unwrap()
.contains(&("agent_budget".into(), "4".into()))
);
assert!(store.project("b").unwrap().is_none());
store.save_scan(&[facts("b", vec![])], true, 1).unwrap();
assert_eq!(store.project("b").unwrap().unwrap().tier, Some(2));
std::fs::write(
dir.join(crate::settings::FILE),
"[config]\nmax_parallel = 0\n",
)
.unwrap();
drop(store);
let e = Store::open(&path).err().expect("refused").to_string();
assert!(e.contains("config.max_parallel"), "{e}");
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn roots_add_and_remove() {
let dir = scratch("roots");
let store = Store::open(&dir.join("p.db")).unwrap();
assert!(store.add_root(Path::new("/a")).unwrap());
assert!(!store.add_root(Path::new("/a")).unwrap());
store.add_root(Path::new("/b")).unwrap();
assert_eq!(
store.roots().unwrap(),
[PathBuf::from("/a"), PathBuf::from("/b")]
);
assert!(store.remove_root(Path::new("/a")).unwrap());
assert_eq!(store.roots().unwrap(), [PathBuf::from("/b")]);
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn scan_round_trips_and_keeps_tier_and_first_seen() {
let dir = scratch("scan");
let mut store = Store::open(&dir.join("p.db")).unwrap();
store.set_tier("one", Path::new("/r/one"), Some(2)).unwrap();
store
.save_scan(
&[facts("one", vec![item("x", 5), item("y", 6)])],
true,
1000,
)
.unwrap();
store
.save_scan(
&[facts("one", vec![item("y", 7), item("z", 8)])],
true,
2000,
)
.unwrap();
let p = store.project("one").unwrap().unwrap();
assert_eq!(p.tier, Some(2), "a scan keeps the tier");
assert_eq!((p.deps, p.deps_at), (None, None));
assert_eq!(p.slug.as_deref(), Some("shakfu/one"));
let mut measured = facts("one", vec![item("y", 7), item("z", 8)]);
measured.deps = Some(DepsFacts {
outdated: Some(2),
detail: "uv: ruff 0.14 -> 0.16".into(),
});
store.save_scan(&[measured], true, 3000).unwrap();
store
.save_scan(
&[facts("one", vec![item("y", 7), item("z", 8)])],
true,
4000,
)
.unwrap();
let p = store.project("one").unwrap().unwrap();
assert_eq!(
(p.deps, p.deps_detail.as_str(), p.deps_at),
(Some(2), "uv: ruff 0.14 -> 0.16", Some(3000)),
"a scan that does not measure deps keeps the last measurement"
);
assert_eq!(
(p.scanned_at, p.has_todo, p.lint_errors, p.dirty),
(Some(4000), true, 1, 3)
);
assert_eq!((p.ahead, p.last_activity), (None, Some(42)));
assert_eq!(p.ci, Ci::Failing(vec!["test".into(), "wheels".into()]));
let tasks = store.tasks().unwrap();
let seen: Vec<_> = tasks
.iter()
.map(|t| (t.text.as_str(), t.line, t.first_seen))
.collect();
assert_eq!(
seen,
[("y", 7, 1000), ("z", 8, 2000)],
"x is gone, y keeps its first scan"
);
assert_eq!(tasks[0].tags, ["a", "b"]);
assert_eq!(tasks[0].group.as_deref(), Some("Security"));
assert_eq!(
(tasks[0].due.as_deref(), tasks[0].gh, tasks[0].added_at),
(Some("2026-10-01"), Some(4), Some(100))
);
assert_eq!(store.last_scan().unwrap(), Some(4000));
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_held_project_keeps_its_last_scan() {
let dir = scratch("held");
let mut store = Store::open(&dir.join("p.db")).unwrap();
store
.save_scan(&[facts("one", vec![item("x", 5)])], true, 1000)
.unwrap();
let held = Facts::held("one", Path::new("/r/one"), "settings changed".into());
assert!(store.save_scan(&[held], true, 2000).unwrap().is_empty());
let tasks: Vec<_> = store.tasks().unwrap().into_iter().map(|t| t.key).collect();
assert_eq!(tasks, ["x"]);
let p = store.project("one").unwrap().unwrap();
assert_eq!((p.dirty, p.absent_since), (3, None));
assert!(matches!(p.ci, Ci::Failing(_)), "{:?}", p.ci);
assert_eq!(p.scan_error.as_deref(), Some("settings changed"));
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn an_unreadable_todo_keeps_the_tasks() {
let dir = scratch("unread");
let mut store = Store::open(&dir.join("p.db")).unwrap();
store
.save_scan(&[facts("one", vec![item("x", 5)])], true, 1000)
.unwrap();
let unread = Facts {
todo: None,
todo_unread: true,
error: Some("TODO.md: permission denied".into()),
..facts("one", vec![])
};
store.save_scan(&[unread], true, 2000).unwrap();
let kept: Vec<_> = store
.tasks()
.unwrap()
.into_iter()
.map(|t| (t.key, t.first_seen))
.collect();
assert_eq!(kept, [("x".to_string(), 1000)]);
let p = store.project("one").unwrap().unwrap();
assert_eq!((p.has_todo, p.lint_errors), (true, 1));
assert_eq!(p.scan_error.as_deref(), Some("TODO.md: permission denied"));
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn a_task_keeps_its_first_seen_when_its_key_changes() {
let dir = scratch("rekey");
let mut store = Store::open(&dir.join("p.db")).unwrap();
let unlinked = ScannedItem {
gh: None,
..item("fix it", 5)
};
store
.save_scan(&[facts("one", vec![unlinked])], true, 1000)
.unwrap();
let linked = ScannedItem {
key: "gh:9".into(),
text: "Fix it".into(),
gh: Some(9),
..item("fix it", 5)
};
store
.save_scan(&[facts("one", vec![linked])], true, 2000)
.unwrap();
let tasks = store.tasks().unwrap();
assert_eq!((tasks[0].key.as_str(), tasks[0].first_seen), ("gh:9", 1000));
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn full_scan_marks_projects_no_longer_found() {
let dir = scratch("full");
let mut store = Store::open(&dir.join("p.db")).unwrap();
store
.save_scan(
&[facts("a", vec![item("t", 1)]), facts("b", vec![])],
true,
1,
)
.unwrap();
store.set_tier("a", Path::new("/r/a"), Some(1)).unwrap();
assert!(
store
.save_scan(&[facts("b", vec![])], false, 2)
.unwrap()
.is_empty(),
"a partial scan marks nothing"
);
assert_eq!(
store.projects().unwrap().len(),
2,
"a partial scan removes nothing"
);
let absent = store.save_scan(&[facts("b", vec![])], true, 3).unwrap();
assert_eq!(absent, ["a"]);
let a = store.project("a").unwrap().unwrap();
assert_eq!(a.absent_since, Some(3));
assert_eq!(a.tier, Some(1), "an absent project keeps its tier");
assert_eq!(
store.tasks().unwrap().len(),
1,
"its tasks and their first_seen are kept"
);
assert!(
store
.save_scan(&[facts("b", vec![])], true, 4)
.unwrap()
.is_empty()
);
assert_eq!(store.project("a").unwrap().unwrap().absent_since, Some(3));
store
.save_scan(&[facts("a", vec![item("t", 1)])], false, 5)
.unwrap();
assert_eq!(store.project("a").unwrap().unwrap().absent_since, None);
assert_eq!(store.tasks().unwrap()[0].first_seen, 1);
let _ = std::fs::remove_dir_all(dir);
}
}