1use std::collections::HashMap;
9use std::error::Error;
10use std::fs::{File, OpenOptions, TryLockError};
11use std::io::Write;
12use std::path::{Path, PathBuf};
13use std::time::Duration;
14
15use rusqlite::{Connection, OptionalExtension, params};
16
17use crate::class::Class;
18use crate::complexity::Features;
19use crate::route::{Approval, Policy};
20use crate::scan::{Ci, Facts};
21use crate::todo::Priority;
22use crate::worker::{Parser, Worker};
23
24pub type Result<T> = std::result::Result<T, Box<dyn Error>>;
25
26const VERSION: i64 = 35;
27
28const SCHEMA: &str = "
29CREATE TABLE config (key TEXT PRIMARY KEY, value TEXT NOT NULL);
30CREATE TABLE roots (path TEXT PRIMARY KEY);
31CREATE TABLE projects (
32 name TEXT PRIMARY KEY,
33 path TEXT NOT NULL,
34 tier INTEGER CHECK (tier BETWEEN 1 AND 5),
35 scanned_at INTEGER,
36 has_todo INTEGER NOT NULL DEFAULT 0,
37 lint_errors INTEGER NOT NULL DEFAULT 0,
38 dirty INTEGER NOT NULL DEFAULT 0,
39 ahead INTEGER,
40 last_activity INTEGER,
41 ci TEXT NOT NULL DEFAULT 'unknown',
42 ci_detail TEXT NOT NULL DEFAULT '',
43 scan_error TEXT
44);
45CREATE TABLE tasks (
46 project TEXT NOT NULL REFERENCES projects(name) ON DELETE CASCADE,
47 key TEXT NOT NULL,
48 line INTEGER NOT NULL,
49 priority TEXT NOT NULL,
50 text TEXT NOT NULL,
51 tags TEXT NOT NULL,
52 due TEXT,
53 gh INTEGER,
54 heading TEXT,
55 added_at INTEGER,
56 first_seen INTEGER NOT NULL,
57 PRIMARY KEY (project, key)
58);
59";
60
61const RUNS: &str = "
64CREATE TABLE runs (
65 id INTEGER PRIMARY KEY,
66 project TEXT NOT NULL,
67 task_key TEXT NOT NULL,
68 text TEXT NOT NULL,
69 gh INTEGER,
70 quadrant TEXT,
71 agent TEXT NOT NULL,
72 repo TEXT NOT NULL,
73 branch TEXT NOT NULL,
74 worktree TEXT NOT NULL,
75 default_branch TEXT NOT NULL,
76 base TEXT NOT NULL,
77 prompt TEXT NOT NULL,
78 state TEXT NOT NULL,
79 feedback TEXT,
80 started_at INTEGER,
81 seconds INTEGER,
82 cost_usd REAL,
83 summary TEXT,
84 commits INTEGER,
85 diffstat TEXT,
86 verify TEXT,
87 verify_ok INTEGER,
88 error TEXT,
89 outcome TEXT
90);
91";
92
93const DEPS_AND_NOTES: &str = "
96ALTER TABLE projects ADD COLUMN deps INTEGER;
97ALTER TABLE projects ADD COLUMN deps_detail TEXT NOT NULL DEFAULT '';
98ALTER TABLE projects ADD COLUMN deps_at INTEGER;
99CREATE TABLE notes (
100 id INTEGER PRIMARY KEY,
101 created_at INTEGER NOT NULL,
102 updated_at INTEGER NOT NULL,
103 text TEXT NOT NULL
104);
105";
106
107const LEFTOVER: &str = "
109ALTER TABLE projects ADD COLUMN leftover INTEGER NOT NULL DEFAULT 0;
110";
111
112const PR_OPEN: &str = "
116UPDATE runs SET state = 'pr-open' WHERE state = 'shipped' AND outcome LIKE 'https://%/pull/%';
117";
118
119const ATTEMPTS: &str = "
125CREATE TABLE attempts (
126 id INTEGER PRIMARY KEY,
127 run_id INTEGER NOT NULL REFERENCES runs(id),
128 n INTEGER NOT NULL,
129 agent TEXT NOT NULL,
130 prompt TEXT NOT NULL,
131 feedback TEXT,
132 started_at INTEGER NOT NULL,
133 seconds INTEGER,
134 cost_usd REAL,
135 summary TEXT,
136 verify TEXT,
137 verify_ok INTEGER,
138 error TEXT,
139 outcome TEXT,
140 UNIQUE (run_id, n)
141);
142ALTER TABLE runs ADD COLUMN dispatched_at INTEGER;
143ALTER TABLE runs ADD COLUMN ready_at INTEGER;
144ALTER TABLE runs ADD COLUMN decided_at INTEGER;
145ALTER TABLE runs ADD COLUMN published_at INTEGER;
146ALTER TABLE runs ADD COLUMN review_seconds INTEGER;
147INSERT INTO attempts (run_id, n, agent, prompt, feedback, started_at, seconds,
148 cost_usd, summary, verify, verify_ok, error, outcome)
149SELECT id, 1, agent, prompt, feedback, started_at, seconds, cost_usd, summary,
150 verify, verify_ok, error, NULL
151FROM runs WHERE started_at IS NOT NULL;
152UPDATE runs SET dispatched_at = started_at WHERE started_at IS NOT NULL;
153ALTER TABLE runs DROP COLUMN started_at;
154";
155
156const SNAPSHOT: &str = "
162ALTER TABLE runs ADD COLUMN class TEXT;
163ALTER TABLE runs ADD COLUMN scope TEXT;
164ALTER TABLE runs ADD COLUMN tier INTEGER;
165ALTER TABLE runs ADD COLUMN description TEXT;
166ALTER TABLE runs ADD COLUMN agent_budget REAL;
167ALTER TABLE runs ADD COLUMN timeout_minutes INTEGER;
168";
169
170const VERIFY_BASE: &str = "
176CREATE TABLE verify_base (
177 project TEXT NOT NULL,
178 base TEXT NOT NULL,
179 command TEXT NOT NULL,
180 timeout_minutes INTEGER NOT NULL,
181 ok INTEGER,
182 seconds INTEGER NOT NULL,
183 measured_at INTEGER NOT NULL,
184 PRIMARY KEY (project, base, command, timeout_minutes)
185);
186ALTER TABLE runs ADD COLUMN verify_base_ok INTEGER;
187ALTER TABLE runs ADD COLUMN verify_base_seconds INTEGER;
188";
189
190const CHANGED_PATHS: &str = "
197ALTER TABLE runs ADD COLUMN changed_paths TEXT;
198ALTER TABLE runs ADD COLUMN scope_error TEXT;
199";
200
201const EXHAUSTION: &str = "
208CREATE TABLE exhaustion (
209 project TEXT NOT NULL,
210 revision TEXT NOT NULL,
211 attempts INTEGER NOT NULL,
212 updated_at INTEGER NOT NULL,
213 PRIMARY KEY (project, revision)
214);
215";
216
217const AGENTS: &str = "
223CREATE TABLE agents (
224 name TEXT PRIMARY KEY,
225 command TEXT NOT NULL,
226 args TEXT NOT NULL,
227 allow TEXT,
228 parse TEXT NOT NULL,
229 reports_cost INTEGER NOT NULL,
230 enforces_budget INTEGER NOT NULL,
231 sandbox INTEGER NOT NULL,
232 resumes INTEGER NOT NULL
233);
234ALTER TABLE runs ADD COLUMN model TEXT;
235ALTER TABLE attempts ADD COLUMN model TEXT;
236";
237
238const RETIRE_KEYS: &str = "
243DELETE FROM config WHERE key IN (
244 'dispatch_quadrants', 'overflow_quadrants',
245 'stale_after.1', 'stale_after.2', 'stale_after.3', 'stale_after.4', 'stale_after.5'
246);
247";
248
249const COMPLEXITY: &str = "
254ALTER TABLE runs ADD COLUMN complexity INTEGER;
255ALTER TABLE runs ADD COLUMN features TEXT;
256ALTER TABLE runs ADD COLUMN estimator TEXT;
257";
258
259const ROUTES: &str = "
265CREATE TABLE routes (
266 revision INTEGER PRIMARY KEY,
267 document TEXT NOT NULL,
268 proposed_by TEXT NOT NULL,
269 created_at INTEGER NOT NULL,
270 activated_at INTEGER,
271 activated_by TEXT,
272 shadow INTEGER NOT NULL DEFAULT 0
273);
274ALTER TABLE runs ADD COLUMN route_revision INTEGER;
275ALTER TABLE runs ADD COLUMN route TEXT;
276ALTER TABLE runs ADD COLUMN approval TEXT;
277";
278
279const APPROVAL_EVIDENCE: &str = "
285ALTER TABLE runs ADD COLUMN approved_tree TEXT;
286ALTER TABLE runs ADD COLUMN approved_head TEXT;
287ALTER TABLE runs ADD COLUMN approved_by TEXT;
288";
289
290const CAMPAIGNS: &str = "
295CREATE TABLE campaigns (
296 name TEXT PRIMARY KEY,
297 text TEXT NOT NULL,
298 description TEXT NOT NULL DEFAULT '',
299 class TEXT NOT NULL,
300 created_at INTEGER NOT NULL
301);
302CREATE TABLE campaign_members (
303 campaign TEXT NOT NULL REFERENCES campaigns(name) ON DELETE CASCADE,
304 project TEXT NOT NULL,
305 run_id INTEGER,
306 PRIMARY KEY (campaign, project)
307);
308";
309
310const SETTINGS_SYNC: &str = "
314CREATE TABLE IF NOT EXISTS settings_sync (text TEXT NOT NULL);
315";
316
317const READ_RUNS_DONE: &str = "
320UPDATE runs SET state = 'done'
321WHERE state = 'ready' AND branch = '' AND workflow_instance IS NOT NULL;
322";
323
324const JOB_APPROVAL_AND_STREAM: &str = r#"
328ALTER TABLE jobs ADD COLUMN awaiting TEXT;
329UPDATE agents
330SET args = '["-p","{prompt}","--output-format","stream-json","--verbose","--permission-mode","acceptEdits","--model","{model}","--max-budget-usd","{budget}","{extra}"]'
331WHERE name = 'claude'
332 AND args = '["-p","{prompt}","--output-format","json","--permission-mode","acceptEdits","--model","{model}","--max-budget-usd","{budget}","{extra}"]';
333"#;
334
335const JOBS: &str = "
338CREATE TABLE jobs (
339 id INTEGER PRIMARY KEY,
340 args TEXT NOT NULL,
341 cwd TEXT NOT NULL,
342 state TEXT NOT NULL,
343 created_at INTEGER NOT NULL,
344 started_at INTEGER,
345 finished_at INTEGER,
346 pid INTEGER,
347 exit_code INTEGER
348);
349";
350
351const INFRA_REJECTION: &str = "
354ALTER TABLE runs ADD COLUMN infra INTEGER NOT NULL DEFAULT 0;
355";
356
357const VERIFY_BASE_SANDBOX: &str = "
362DROP TABLE verify_base;
363CREATE TABLE verify_base (
364 project TEXT NOT NULL,
365 base TEXT NOT NULL,
366 command TEXT NOT NULL,
367 timeout_minutes INTEGER NOT NULL,
368 sandbox INTEGER NOT NULL,
369 ok INTEGER,
370 seconds INTEGER NOT NULL,
371 measured_at INTEGER NOT NULL,
372 PRIMARY KEY (project, base, command, timeout_minutes, sandbox)
373);
374";
375
376const REVIEWED_TREE: &str = "
379ALTER TABLE runs ADD COLUMN reviewed_tree TEXT;
380";
381
382const PUSH_OR_PR: &str = "
389UPDATE runs SET state = 'merged' WHERE state = 'shipped' AND outcome LIKE '%/pull/%';
390UPDATE runs SET state = 'pushed' WHERE state = 'shipped';
391UPDATE runs SET state = 'closed' WHERE state = 'rejected' AND outcome LIKE '%/pull/%';
392DELETE FROM config WHERE key = 'publish' OR (key LIKE 'projects.%' AND key LIKE '%.publish');
393";
394
395const RETIRE_WEIGHTS: &str = "
399DELETE FROM config WHERE key LIKE 'weights.%';
400";
401
402const GIT_SNAPSHOT: &str = "
407ALTER TABLE runs ADD COLUMN git_snapshot TEXT;
408";
409
410const SANDUK_WORKER: &str = "-- no schema change; the seed is in Rust";
415
416const PRESETS_REPLACE_WORKER_MODEL: &str = "
424INSERT OR IGNORE INTO presets (name, agent, model, args, created_at)
425 SELECT name, name, model, '[]', strftime('%s', 'now') FROM agents
426 WHERE model IS NOT NULL AND model <> '';
427INSERT OR REPLACE INTO config (key, value)
428 SELECT 'preset', a.name FROM agents a
429 WHERE a.model IS NOT NULL AND a.model <> ''
430 AND a.name = COALESCE((SELECT value FROM config WHERE key = 'agent'), 'claude')
431 AND NOT EXISTS (SELECT 1 FROM config WHERE key = 'preset' AND value <> '');
432ALTER TABLE agents DROP COLUMN model;
433";
434
435const PRESETS: &str = "
445CREATE TABLE presets (
446 name TEXT PRIMARY KEY,
447 agent TEXT NOT NULL,
448 model TEXT,
449 args TEXT NOT NULL DEFAULT '[]',
450 created_at INTEGER NOT NULL
451);
452ALTER TABLE runs ADD COLUMN preset TEXT;
453ALTER TABLE runs ADD COLUMN extra_args TEXT;
454";
455
456const WORKER_MODEL: &str = "
461ALTER TABLE agents ADD COLUMN model TEXT;
462UPDATE agents SET model = (SELECT value FROM config WHERE key = 'model')
463 WHERE name = COALESCE((SELECT value FROM config WHERE key = 'agent'), 'claude')
464 AND EXISTS (SELECT 1 FROM config WHERE key = 'model' AND value <> '');
465DELETE FROM config WHERE key = 'model';
466";
467
468const WORKER_ENV: &str = "
474ALTER TABLE agents ADD COLUMN env TEXT NOT NULL DEFAULT '{}';
475";
476
477const WORKFLOWS: &str = "
485CREATE TABLE workflows (
486 revision INTEGER PRIMARY KEY,
487 document TEXT NOT NULL,
488 source TEXT,
489 proposed_by TEXT NOT NULL,
490 created_at INTEGER NOT NULL,
491 activated_at INTEGER,
492 activated_by TEXT,
493 worst_case_runs INTEGER,
494 worst_case_cost REAL
495);
496CREATE TABLE workflow_instances (
497 id INTEGER PRIMARY KEY,
498 workflow TEXT NOT NULL,
499 revision INTEGER NOT NULL REFERENCES workflows(revision),
500 args TEXT NOT NULL,
501 target TEXT NOT NULL,
502 started_at INTEGER NOT NULL,
503 finished_at INTEGER,
504 outcome TEXT
505);
506CREATE TABLE workflow_units (
507 instance INTEGER NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE,
508 id TEXT NOT NULL,
509 type TEXT NOT NULL,
510 node TEXT NOT NULL,
511 parent TEXT,
512 root TEXT NOT NULL,
513 depth INTEGER NOT NULL DEFAULT 0,
514 lap INTEGER NOT NULL DEFAULT 0,
515 project TEXT,
516 data TEXT NOT NULL,
517 created_at INTEGER NOT NULL,
518 PRIMARY KEY (instance, id)
519);
520CREATE INDEX idx_workflow_units_root ON workflow_units(instance, root);
521CREATE TABLE workflow_moves (
522 instance INTEGER NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE,
523 unit TEXT NOT NULL,
524 edge INTEGER NOT NULL,
525 taken INTEGER NOT NULL,
526 reason TEXT,
527 at INTEGER NOT NULL,
528 PRIMARY KEY (instance, unit, edge)
529);
530CREATE TABLE workflow_verdicts (
531 instance INTEGER NOT NULL REFERENCES workflow_instances(id) ON DELETE CASCADE,
532 unit TEXT NOT NULL,
533 check_name TEXT NOT NULL,
534 verdict TEXT NOT NULL,
535 detail TEXT,
536 at INTEGER NOT NULL,
537 PRIMARY KEY (instance, unit, check_name)
538);
539ALTER TABLE runs ADD COLUMN workflow_instance INTEGER;
540ALTER TABLE runs ADD COLUMN node TEXT;
541ALTER TABLE runs ADD COLUMN unit TEXT;
542ALTER TABLE runs ADD COLUMN lap INTEGER NOT NULL DEFAULT 0;
543";
544
545const PROJECT_TAGS: &str = "
550CREATE TABLE project_tags (
551 project TEXT NOT NULL,
552 tag TEXT NOT NULL,
553 PRIMARY KEY (project, tag)
554);
555CREATE INDEX idx_project_tags_tag ON project_tags(tag);
556";
557
558const ABSENCE: &str = "
563ALTER TABLE projects ADD COLUMN absent_since INTEGER;
564";
565
566const SLUG: &str = "
570ALTER TABLE projects ADD COLUMN slug TEXT;
571";
572
573pub struct Store {
574 conn: Connection,
575 settings: PathBuf,
578}
579
580#[derive(Debug, Clone, PartialEq)]
581pub struct ProjectRow {
582 pub name: String,
583 pub path: PathBuf,
584 pub slug: Option<String>,
587 pub absent_since: Option<i64>,
590 pub tier: Option<u8>,
591 pub scanned_at: Option<i64>,
592 pub has_todo: bool,
593 pub lint_errors: i64,
594 pub dirty: i64,
595 pub leftover: i64,
597 pub ahead: Option<i64>,
598 pub last_activity: Option<i64>,
599 pub ci: Ci,
600 pub scan_error: Option<String>,
601 pub deps: Option<i64>,
603 pub deps_detail: String,
605 pub deps_at: Option<i64>,
606}
607
608#[derive(Debug, Clone, PartialEq)]
610pub struct Footprint {
611 pub tasks: i64,
612 pub tags: i64,
613 pub campaigns: i64,
614 pub runs: i64,
616 pub open_runs: Vec<i64>,
618}
619
620#[derive(Debug, Clone, PartialEq)]
621pub struct TaskRow {
622 pub project: String,
623 pub key: String,
624 pub line: i64,
625 pub priority: Priority,
626 pub text: String,
627 pub tags: Vec<String>,
628 pub due: Option<String>,
629 pub gh: Option<i64>,
630 pub group: Option<String>,
631 pub added_at: Option<i64>,
632 pub first_seen: i64,
633}
634
635pub struct Session {
638 _file: File,
639}
640
641impl Session {
642 pub fn try_acquire(dir: &Path) -> Result<Option<Session>> {
644 Session::try_lock(dir, "session.lock")
645 }
646
647 pub fn try_lock(dir: &Path, name: &str) -> Result<Option<Session>> {
650 let path = dir.join(name);
651 let io = |e: std::io::Error| format!("{}: {e}", path.display());
652 let mut file = OpenOptions::new()
653 .write(true)
654 .create(true)
655 .truncate(false)
656 .open(&path)
657 .map_err(io)?;
658 match file.try_lock() {
659 Ok(()) => {}
660 Err(TryLockError::WouldBlock) => return Ok(None),
661 Err(TryLockError::Error(e)) => return Err(io(e).into()),
662 }
663 file.set_len(0).map_err(io)?;
664 write!(file, "{}", std::process::id()).map_err(io)?;
665 Ok(Some(Session { _file: file }))
666 }
667
668 pub fn acquire(dir: &Path) -> Result<Session> {
672 const WAIT: Duration = Duration::from_secs(2);
673 let started = std::time::Instant::now();
674 loop {
675 if let Some(session) = Session::try_acquire(dir)? {
676 return Ok(session);
677 }
678 if started.elapsed() >= WAIT {
679 break;
680 }
681 std::thread::sleep(Duration::from_millis(50));
682 }
683 Err({
684 let pid = std::fs::read_to_string(dir.join("session.lock")).unwrap_or_default();
685 let holder = match pid.trim() {
686 "" => String::new(),
687 pid => format!(" (pid {pid})"),
688 };
689 format!(
690 "another pma session{holder} is dispatching, reworking, rejecting or publishing; wait for it to finish"
691 )
692 .into()
693 })
694 }
695}
696
697pub fn home() -> Result<PathBuf> {
699 if let Some(dir) = std::env::var_os("PMA_HOME") {
700 return Ok(PathBuf::from(dir));
701 }
702 let home = std::env::var_os("HOME").ok_or("HOME is not set")?;
703 Ok(PathBuf::from(home).join(".config").join("pma"))
704}
705
706pub fn state_home() -> Result<PathBuf> {
710 let var = |name: &str| std::env::var_os(name).filter(|v| !v.is_empty());
711 if let Some(dir) = var("PMA_STATE") {
712 return Ok(PathBuf::from(dir));
713 }
714 if let Some(dir) = var("PMA_HOME") {
715 return Ok(PathBuf::from(dir).join("state"));
716 }
717 if let Some(dir) = var("XDG_STATE_HOME") {
718 return Ok(PathBuf::from(dir).join("pma"));
719 }
720 let home = std::env::var_os("HOME").ok_or("HOME is not set")?;
721 Ok(PathBuf::from(home).join(".local").join("state").join("pma"))
722}
723
724impl Store {
725 pub fn open_default() -> Result<Store> {
726 let dir = home()?;
727 std::fs::create_dir_all(&dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
728 Store::open(&dir.join("projects.db"))
729 }
730
731 pub fn open(path: &Path) -> Result<Store> {
732 let conn = Connection::open(path).map_err(|e| format!("open {}: {e}", path.display()))?;
733 conn.query_row("PRAGMA journal_mode = DELETE", [], |r| {
734 r.get::<_, String>(0)
735 })?;
736 conn.pragma_update(None, "foreign_keys", "ON")?;
737 conn.busy_timeout(Duration::from_secs(5))?;
739 let version: i64 = conn.pragma_query_value(None, "user_version", |r| r.get(0))?;
740 match version {
741 0..VERSION => {
742 let steps = [
743 SCHEMA,
744 RUNS,
745 DEPS_AND_NOTES,
746 LEFTOVER,
747 PR_OPEN,
748 ATTEMPTS,
749 SNAPSHOT,
750 VERIFY_BASE,
751 CHANGED_PATHS,
752 EXHAUSTION,
753 AGENTS,
754 RETIRE_KEYS,
755 COMPLEXITY,
756 ROUTES,
757 APPROVAL_EVIDENCE,
758 CAMPAIGNS,
759 SLUG,
760 ABSENCE,
761 PROJECT_TAGS,
762 WORKFLOWS,
763 WORKER_ENV,
764 WORKER_MODEL,
765 PRESETS,
766 PRESETS_REPLACE_WORKER_MODEL,
767 SANDUK_WORKER,
768 GIT_SNAPSHOT,
769 RETIRE_WEIGHTS,
770 PUSH_OR_PR,
771 REVIEWED_TREE,
772 VERIFY_BASE_SANDBOX,
773 INFRA_REJECTION,
774 JOBS,
775 JOB_APPROVAL_AND_STREAM,
776 READ_RUNS_DONE,
777 SETTINGS_SYNC,
778 ];
779 let tx = conn.unchecked_transaction()?;
780 for step in &steps[version as usize..] {
781 tx.execute_batch(step)?;
782 }
783 if version < 11 {
784 seed_agent(&tx, &crate::worker::Worker::claude())?;
785 }
786 let mut seeded = Vec::new();
789 if version < 21 {
790 seeded.extend([
791 crate::worker::Worker::opencode(),
792 crate::worker::Worker::omp(),
793 ]);
794 }
795 if version < 25 {
796 seeded.push(crate::worker::Worker::sanduk());
797 }
798 for w in seeded {
799 if tx.query_row(
800 "SELECT count(*) FROM agents WHERE name = ?1",
801 [&w.name],
802 |r| r.get::<_, i64>(0),
803 )? == 0
804 {
805 seed_agent(&tx, &w)?;
806 }
807 }
808 tx.pragma_update(None, "user_version", VERSION)?;
809 tx.commit()?;
810 }
811 VERSION => {}
812 v => {
813 return Err(format!(
814 "{} has schema version {v}; this pma reads {VERSION}",
815 path.display()
816 )
817 .into());
818 }
819 }
820 let store = Store {
821 conn,
822 settings: path.with_file_name(crate::settings::FILE),
823 };
824 store.load_settings()?;
825 Ok(store)
826 }
827
828 fn load_settings(&self) -> Result<()> {
834 if !self.settings.exists() {
835 crate::settings::create(&self.settings, &self.settings_from_tables()?)?;
836 }
837 let (text, settings) = crate::settings::read(&self.settings)?;
838 let synced: Option<String> = self
839 .conn
840 .query_row("SELECT text FROM settings_sync", [], |r| r.get(0))
841 .optional()?;
842 if synced.as_deref() != Some(text.as_str()) {
843 self.apply_settings(&text, &settings)?;
844 }
845 Ok(())
846 }
847
848 fn settings_from_tables(&self) -> Result<crate::settings::Settings> {
849 let mut stmt = self
850 .conn
851 .prepare("SELECT name, tier FROM projects WHERE tier IS NOT NULL ORDER BY name")?;
852 let tiers = stmt
853 .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
854 .collect::<rusqlite::Result<_>>()?;
855 Ok(crate::settings::Settings {
856 config: self.config_rows()?,
857 roots: self.roots()?,
858 tiers,
859 tags: self.project_tags()?,
860 agents: self.agents()?,
861 presets: self.presets()?,
862 })
863 }
864
865 fn apply_settings(&self, text: &str, s: &crate::settings::Settings) -> Result<()> {
868 let tx = self.conn.unchecked_transaction()?;
869 tx.execute_batch(
870 "DELETE FROM config; DELETE FROM roots; DELETE FROM agents; DELETE FROM presets;
871 DELETE FROM project_tags; DELETE FROM settings_sync;
872 UPDATE projects SET tier = NULL;",
873 )?;
874 for (key, value) in &s.config {
875 tx.execute(
876 "INSERT OR REPLACE INTO config (key, value) VALUES (?1, ?2)",
877 params![key, value],
878 )?;
879 }
880 for root in &s.roots {
881 tx.execute(
882 "INSERT OR IGNORE INTO roots (path) VALUES (?1)",
883 [path_str(root)?],
884 )?;
885 }
886 for (name, tier) in &s.tiers {
887 tx.execute(
888 "UPDATE projects SET tier = ?2 WHERE name = ?1",
889 params![name, tier],
890 )?;
891 }
892 for (project, tag) in &s.tags {
893 tx.execute(
894 "INSERT OR IGNORE INTO project_tags (project, tag) VALUES (?1, ?2)",
895 params![project, tag],
896 )?;
897 }
898 for w in &s.agents {
899 seed_agent(&tx, w)?;
900 }
901 for p in &s.presets {
902 tx.execute(
903 "INSERT OR REPLACE INTO presets (name, agent, model, args, created_at)
904 VALUES (?1, ?2, ?3, ?4, ?5)",
905 params![
906 p.name,
907 p.agent,
908 p.model,
909 serde_json::to_string(&p.args).unwrap_or_else(|_| "[]".into()),
910 crate::dates::now()
911 ],
912 )?;
913 }
914 tx.execute("INSERT INTO settings_sync (text) VALUES (?1)", [text])?;
915 tx.commit()?;
916 Ok(())
917 }
918
919 fn change_settings<T>(&self, f: impl FnOnce(&mut toml_edit::DocumentMut) -> T) -> Result<T> {
921 let out = crate::settings::edit(&self.settings, f)?;
922 let (text, settings) = crate::settings::read(&self.settings)?;
923 self.apply_settings(&text, &settings)?;
924 Ok(out)
925 }
926
927 fn reapply_settings(&self) -> Result<()> {
929 let (text, settings) = crate::settings::read(&self.settings)?;
930 self.apply_settings(&text, &settings)
931 }
932
933 pub fn config_rows(&self) -> Result<Vec<(String, String)>> {
934 let mut stmt = self
935 .conn
936 .prepare("SELECT key, value FROM config ORDER BY key")?;
937 let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
938 Ok(rows.collect::<rusqlite::Result<_>>()?)
939 }
940
941 pub fn set_config(&self, key: &str, value: &str) -> Result<()> {
942 self.change_settings(|doc| crate::settings::set_config(doc, key, value))
943 }
944
945 pub fn reset_config(&self, key: &str) -> Result<bool> {
947 self.change_settings(|doc| crate::settings::reset_config(doc, key))
948 }
949
950 pub fn roots(&self) -> Result<Vec<PathBuf>> {
951 let mut stmt = self.conn.prepare("SELECT path FROM roots ORDER BY path")?;
952 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
953 Ok(rows
954 .map(|r| r.map(PathBuf::from))
955 .collect::<rusqlite::Result<_>>()?)
956 }
957
958 pub fn add_root(&self, path: &Path) -> Result<bool> {
960 let path = path_str(path)?;
961 self.change_settings(|doc| crate::settings::add_root(doc, path))
962 }
963
964 pub fn remove_root(&self, path: &Path) -> Result<bool> {
965 let path = path_str(path)?;
966 self.change_settings(|doc| crate::settings::remove_root(doc, path))
967 }
968
969 pub fn set_tier(&self, name: &str, path: &Path, tier: Option<u8>) -> Result<()> {
971 self.conn.execute(
972 "INSERT INTO projects (name, path, tier) VALUES (?1, ?2, ?3)
973 ON CONFLICT (name) DO UPDATE SET tier = excluded.tier",
974 params![name, path_str(path)?, tier],
975 )?;
976 self.change_settings(|doc| crate::settings::set_tier(doc, name, tier))
977 }
978
979 pub fn last_scan(&self) -> Result<Option<i64>> {
980 Ok(self
981 .conn
982 .query_row("SELECT MAX(scanned_at) FROM projects", [], |r| r.get(0))?)
983 }
984
985 pub fn project(&self, name: &str) -> Result<Option<ProjectRow>> {
986 Ok(self.projects()?.into_iter().find(|p| p.name == name))
987 }
988
989 pub fn projects(&self) -> Result<Vec<ProjectRow>> {
990 let mut stmt = self.conn.prepare(
991 "SELECT name, path, tier, scanned_at, has_todo, lint_errors, dirty, ahead,
992 last_activity, ci, ci_detail, scan_error, deps, deps_detail, deps_at,
993 leftover, slug, absent_since
994 FROM projects ORDER BY name",
995 )?;
996 let rows = stmt.query_map([], |r| {
997 Ok(ProjectRow {
998 name: r.get(0)?,
999 path: PathBuf::from(r.get::<_, String>(1)?),
1000 tier: r.get(2)?,
1001 scanned_at: r.get(3)?,
1002 has_todo: r.get(4)?,
1003 lint_errors: r.get(5)?,
1004 dirty: r.get(6)?,
1005 ahead: r.get(7)?,
1006 last_activity: r.get(8)?,
1007 ci: Ci::from_columns(&r.get::<_, String>(9)?, r.get::<_, String>(10)?),
1008 scan_error: r.get(11)?,
1009 deps: r.get(12)?,
1010 deps_detail: r.get(13)?,
1011 deps_at: r.get(14)?,
1012 leftover: r.get(15)?,
1013 slug: r.get(16)?,
1014 absent_since: r.get(17)?,
1015 })
1016 })?;
1017 Ok(rows.collect::<rusqlite::Result<_>>()?)
1018 }
1019
1020 pub fn tasks(&self) -> Result<Vec<TaskRow>> {
1021 let mut stmt = self.conn.prepare(
1022 "SELECT project, line, priority, text, tags, due, gh, added_at, first_seen, heading, key
1023 FROM tasks ORDER BY project, line",
1024 )?;
1025 let rows = stmt.query_map([], |r| {
1026 let priority: String = r.get(2)?;
1027 let tags: String = r.get(4)?;
1028 Ok(TaskRow {
1029 project: r.get(0)?,
1030 line: r.get(1)?,
1031 priority: Priority::parse(&priority).unwrap_or(Priority::Low),
1032 text: r.get(3)?,
1033 tags: tags.split_whitespace().map(String::from).collect(),
1034 due: r.get(5)?,
1035 gh: r.get(6)?,
1036 added_at: r.get(7)?,
1037 first_seen: r.get(8)?,
1038 group: r.get(9)?,
1039 key: r.get(10)?,
1040 })
1041 })?;
1042 Ok(rows.collect::<rusqlite::Result<_>>()?)
1043 }
1044
1045 pub fn save_scan(&mut self, facts: &[Facts], full: bool, now: i64) -> Result<Vec<String>> {
1052 let open_runs: Vec<Run> = self
1053 .runs()?
1054 .into_iter()
1055 .filter(|r| !r.state.is_final())
1056 .collect();
1057 let tx = self.conn.transaction()?;
1058 for f in facts {
1059 if f.held {
1060 tx.execute(
1061 "UPDATE projects SET scan_error = ?2, absent_since = NULL WHERE name = ?1",
1062 params![f.name, f.error],
1063 )?;
1064 continue;
1065 }
1066 let (ci, ci_detail) = f.ci.to_columns();
1067 let leftover = f
1068 .pma_branches
1069 .iter()
1070 .filter(|b| {
1071 !open_runs
1072 .iter()
1073 .any(|r| r.project == f.name && &r.branch == *b)
1074 })
1075 .count() as i64;
1076 tx.execute(
1077 "INSERT INTO projects (name, path, scanned_at, has_todo, lint_errors, dirty, ahead,
1078 last_activity, ci, ci_detail, scan_error, leftover, slug)
1079 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
1080 ON CONFLICT (name) DO UPDATE SET
1081 path = excluded.path, scanned_at = excluded.scanned_at,
1082 has_todo = CASE WHEN ?14 THEN projects.has_todo ELSE excluded.has_todo END,
1083 lint_errors = CASE WHEN ?14 THEN projects.lint_errors
1084 ELSE excluded.lint_errors END,
1085 dirty = excluded.dirty, ahead = excluded.ahead,
1086 last_activity = excluded.last_activity, ci = excluded.ci,
1087 ci_detail = excluded.ci_detail, scan_error = excluded.scan_error,
1088 leftover = excluded.leftover, slug = excluded.slug,
1089 absent_since = NULL",
1090 params![
1091 f.name,
1092 path_str(&f.path)?,
1093 now,
1094 f.todo.is_some() || f.todo_unread,
1095 f.todo.as_ref().map_or(0, |t| t.lint_errors),
1096 f.dirty,
1097 f.ahead,
1098 f.last_activity,
1099 ci,
1100 ci_detail,
1101 f.error,
1102 leftover,
1103 f.slug,
1104 f.todo_unread,
1105 ],
1106 )?;
1107
1108 if let Some(d) = &f.deps {
1109 tx.execute(
1110 "UPDATE projects SET deps = ?2, deps_detail = ?3, deps_at = ?4 WHERE name = ?1",
1111 params![f.name, d.outdated, d.detail, now],
1112 )?;
1113 }
1114
1115 if f.todo_unread {
1118 continue;
1119 }
1120 type Seen = (i64, Option<i64>);
1126 let (by_key, by_text): (HashMap<String, Seen>, HashMap<String, Seen>) = {
1127 let mut stmt = tx.prepare(
1128 "SELECT key, text, first_seen, added_at FROM tasks WHERE project = ?1",
1129 )?;
1130 let rows: Vec<(String, String, i64, Option<i64>)> = stmt
1131 .query_map([&f.name], |r| {
1132 Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))
1133 })?
1134 .collect::<rusqlite::Result<_>>()?;
1135 let by_text = rows
1136 .iter()
1137 .map(|(_, text, seen, added)| (crate::todo::normal_text(text), (*seen, *added)))
1138 .collect();
1139 (
1140 rows.into_iter()
1141 .map(|(k, _, seen, added)| (k, (seen, added)))
1142 .collect(),
1143 by_text,
1144 )
1145 };
1146 tx.execute("DELETE FROM tasks WHERE project = ?1", [&f.name])?;
1147 for item in f.todo.iter().flat_map(|t| &t.items) {
1148 let before = by_key
1149 .get(&item.key)
1150 .or_else(|| by_text.get(&crate::todo::normal_text(&item.text)));
1151 let first_seen = before.map_or(now, |(seen, _)| *seen);
1152 let added_at = match (item.added_at, before.and_then(|(_, added)| *added)) {
1153 (Some(a), Some(b)) => Some(a.min(b)),
1154 (a, b) => a.or(b),
1155 };
1156 tx.execute(
1157 "INSERT INTO tasks (project, key, line, priority, text, tags, due, gh, added_at, first_seen, heading)
1158 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
1159 params![
1160 f.name,
1161 item.key,
1162 item.line,
1163 item.priority.name(),
1164 item.text,
1165 item.tags.join(" "),
1166 item.due,
1167 item.gh,
1168 added_at,
1169 first_seen,
1170 item.group,
1171 ],
1172 )?;
1173 }
1174 }
1175
1176 let mut absent = Vec::new();
1177 if full {
1178 let scanned: std::collections::HashSet<&str> =
1179 facts.iter().map(|f| f.name.as_str()).collect();
1180 let mut stmt = tx.prepare("SELECT name FROM projects WHERE absent_since IS NULL")?;
1181 let present: Vec<String> = stmt
1182 .query_map([], |r| r.get(0))?
1183 .collect::<rusqlite::Result<_>>()?;
1184 drop(stmt);
1185 for name in present {
1186 if !scanned.contains(name.as_str()) {
1187 tx.execute(
1188 "UPDATE projects SET absent_since = ?2 WHERE name = ?1",
1189 params![name, now],
1190 )?;
1191 absent.push(name);
1192 }
1193 }
1194 }
1195 tx.commit()?;
1196 self.reapply_settings()?;
1198 Ok(absent)
1199 }
1200
1201 pub fn project_footprint(&self, name: &str) -> Result<Footprint> {
1203 let count =
1204 |sql: &str| -> Result<i64> { Ok(self.conn.query_row(sql, [name], |r| r.get(0))?) };
1205 let open_runs = self
1206 .runs()?
1207 .into_iter()
1208 .filter(|r| r.project == name && !r.state.is_final())
1209 .map(|r| r.id)
1210 .collect();
1211 Ok(Footprint {
1212 tasks: count("SELECT count(*) FROM tasks WHERE project = ?1")?,
1213 tags: count("SELECT count(*) FROM project_tags WHERE project = ?1")?,
1214 campaigns: count("SELECT count(*) FROM campaign_members WHERE project = ?1")?,
1215 runs: count("SELECT count(*) FROM runs WHERE project = ?1")?,
1216 open_runs,
1217 })
1218 }
1219
1220 pub fn forget_project(&mut self, name: &str) -> Result<()> {
1224 let tx = self.conn.transaction()?;
1225 for sql in [
1226 "DELETE FROM project_tags WHERE project = ?1",
1227 "DELETE FROM campaign_members WHERE project = ?1",
1228 "DELETE FROM exhaustion WHERE project = ?1",
1229 "DELETE FROM verify_base WHERE project = ?1",
1230 "DELETE FROM projects WHERE name = ?1",
1232 ] {
1233 tx.execute(sql, [name])?;
1234 }
1235 tx.commit()?;
1236 self.change_settings(|doc| crate::settings::forget_project(doc, name))
1237 }
1238
1239 pub fn project_tags(&self) -> Result<Vec<(String, String)>> {
1241 let mut stmt = self
1242 .conn
1243 .prepare("SELECT project, tag FROM project_tags ORDER BY project, tag")?;
1244 let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
1245 Ok(rows.collect::<rusqlite::Result<_>>()?)
1246 }
1247
1248 pub fn add_project_tag(&self, project: &str, tag: &str) -> Result<bool> {
1250 self.change_settings(|doc| crate::settings::tag_project(doc, project, tag, true))
1251 }
1252
1253 pub fn remove_project_tag(&self, project: &str, tag: &str) -> Result<bool> {
1255 self.change_settings(|doc| crate::settings::tag_project(doc, project, tag, false))
1256 }
1257
1258 pub fn projects_tagged(&self, tags: &[String]) -> Result<Vec<String>> {
1261 let mut found: Vec<String> = Vec::new();
1262 for (project, tag) in self.project_tags()? {
1263 if tags.iter().any(|t| t == &tag) && !found.contains(&project) {
1264 found.push(project);
1265 }
1266 }
1267 found.sort();
1268 Ok(found)
1269 }
1270
1271 pub fn notes(&self) -> Result<Vec<Note>> {
1272 let mut stmt = self
1273 .conn
1274 .prepare("SELECT id, created_at, updated_at, text FROM notes ORDER BY id")?;
1275 let rows = stmt.query_map([], |r| {
1276 Ok(Note {
1277 id: r.get(0)?,
1278 created_at: r.get(1)?,
1279 updated_at: r.get(2)?,
1280 text: r.get(3)?,
1281 })
1282 })?;
1283 Ok(rows.collect::<rusqlite::Result<_>>()?)
1284 }
1285
1286 pub fn add_job(&self, args: &[String], cwd: &Path) -> Result<i64> {
1287 self.conn.execute(
1288 "INSERT INTO jobs (args, cwd, state, created_at) VALUES (?1, ?2, 'queued', ?3)",
1289 params![
1290 serde_json::to_string(args)?,
1291 cwd.to_string_lossy(),
1292 crate::dates::now()
1293 ],
1294 )?;
1295 Ok(self.conn.last_insert_rowid())
1296 }
1297
1298 pub fn jobs(&self, limit: i64) -> Result<Vec<Job>> {
1300 let mut stmt = self.conn.prepare(&format!(
1301 "SELECT {JOB_COLUMNS} FROM jobs ORDER BY id DESC LIMIT ?1"
1302 ))?;
1303 let rows = stmt.query_map([limit], job_row)?;
1304 Ok(rows.collect::<rusqlite::Result<_>>()?)
1305 }
1306
1307 pub fn job(&self, id: i64) -> Result<Job> {
1308 self.conn
1309 .query_row(
1310 &format!("SELECT {JOB_COLUMNS} FROM jobs WHERE id = ?1"),
1311 [id],
1312 job_row,
1313 )
1314 .optional()?
1315 .ok_or_else(|| format!("no job {id}; `pma job` lists them").into())
1316 }
1317
1318 pub fn claim_job(&self) -> Result<Option<Job>> {
1320 let Some(id) = self
1321 .conn
1322 .query_row(
1323 "SELECT id FROM jobs WHERE state = 'queued' ORDER BY id LIMIT 1",
1324 [],
1325 |r| r.get::<_, i64>(0),
1326 )
1327 .optional()?
1328 else {
1329 return Ok(None);
1330 };
1331 self.conn.execute(
1332 "UPDATE jobs SET state = 'running', started_at = ?2 WHERE id = ?1 AND state = 'queued'",
1333 params![id, crate::dates::now()],
1334 )?;
1335 self.job(id).map(Some)
1336 }
1337
1338 pub fn set_job_awaiting(&self, id: i64, args: &[String]) -> Result<()> {
1340 self.conn.execute(
1341 "UPDATE jobs SET awaiting = ?2 WHERE id = ?1",
1342 params![id, serde_json::to_string(args)?],
1343 )?;
1344 Ok(())
1345 }
1346
1347 pub fn approve_job(&self, id: i64) -> Result<bool> {
1349 Ok(self.conn.execute(
1350 "UPDATE jobs SET state = 'approved' WHERE id = ?1 AND state = 'awaiting-approval'",
1351 [id],
1352 )? > 0)
1353 }
1354
1355 pub fn settle_jobs_awaiting(&self, plan: &str) -> Result<()> {
1358 self.conn.execute(
1359 "UPDATE jobs SET state = 'approved'
1360 WHERE state = 'awaiting-approval' AND instr(awaiting, '\"' || ?1 || '\"') > 0",
1361 [plan],
1362 )?;
1363 Ok(())
1364 }
1365
1366 pub fn set_job_pid(&self, id: i64, pid: u32) -> Result<()> {
1367 self.conn
1368 .execute("UPDATE jobs SET pid = ?2 WHERE id = ?1", params![id, pid])?;
1369 Ok(())
1370 }
1371
1372 pub fn finish_job(&self, id: i64, exit_code: Option<i32>) -> Result<()> {
1374 self.conn.execute(
1375 "UPDATE jobs SET finished_at = ?2, exit_code = ?3,
1376 state = CASE WHEN state = 'stopped' THEN 'stopped'
1377 WHEN ?3 = 0 AND awaiting IS NOT NULL THEN 'awaiting-approval'
1378 WHEN ?3 = 0 THEN 'done' ELSE 'failed' END
1379 WHERE id = ?1",
1380 params![id, crate::dates::now(), exit_code],
1381 )?;
1382 Ok(())
1383 }
1384
1385 pub fn stop_job(&self, id: i64) -> Result<String> {
1387 let job = self.job(id)?;
1388 if matches!(job.state.as_str(), "queued" | "running") {
1389 self.conn.execute(
1390 "UPDATE jobs SET state = 'stopped', finished_at = ?2 WHERE id = ?1",
1391 params![id, crate::dates::now()],
1392 )?;
1393 }
1394 Ok(job.state)
1395 }
1396
1397 pub fn fail_running_jobs(&self) -> Result<usize> {
1400 Ok(self.conn.execute(
1401 "UPDATE jobs SET state = 'failed', finished_at = ?1 WHERE state = 'running'",
1402 [crate::dates::now()],
1403 )?)
1404 }
1405
1406 pub fn has_queued_jobs(&self) -> Result<bool> {
1407 Ok(self.conn.query_row(
1408 "SELECT count(*) FROM jobs WHERE state = 'queued'",
1409 [],
1410 |r| r.get::<_, i64>(0),
1411 )? > 0)
1412 }
1413
1414 pub fn add_note(&self, text: &str, now: i64) -> Result<i64> {
1415 self.conn.execute(
1416 "INSERT INTO notes (created_at, updated_at, text) VALUES (?1, ?1, ?2)",
1417 params![now, text],
1418 )?;
1419 Ok(self.conn.last_insert_rowid())
1420 }
1421
1422 pub fn edit_note(&self, id: i64, text: &str, now: i64) -> Result<bool> {
1424 Ok(self.conn.execute(
1425 "UPDATE notes SET text = ?2, updated_at = ?3 WHERE id = ?1",
1426 params![id, text, now],
1427 )? > 0)
1428 }
1429
1430 pub fn remove_note(&self, id: i64) -> Result<bool> {
1432 Ok(self.conn.execute("DELETE FROM notes WHERE id = ?1", [id])? > 0)
1433 }
1434
1435 pub fn insert_run(&self, run: &mut Run) -> Result<i64> {
1438 run.dispatched_at = Some(crate::dates::now());
1439 self.conn.execute(
1440 "INSERT INTO runs (project, task_key, text, gh, quadrant, agent, branch, worktree,
1441 default_branch, base, prompt, state, repo, dispatched_at,
1442 class, scope, tier, description, agent_budget, timeout_minutes,
1443 verify, verify_base_ok, verify_base_seconds, model,
1444 complexity, features, estimator,
1445 route_revision, route, approval,
1446 workflow_instance, node, unit, lap, preset, extra_args,
1447 git_snapshot)
1448 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16,
1449 ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27, ?28, ?29, ?30,
1450 ?31, ?32, ?33, ?34, ?35, ?36, ?37)",
1451 params![
1452 run.project,
1453 run.task_key,
1454 run.text,
1455 run.gh,
1456 run.quadrant,
1457 run.agent,
1458 run.branch,
1459 path_str(&run.worktree)?,
1460 run.default_branch,
1461 run.base,
1462 run.prompt,
1463 run.state.name(),
1464 path_str(&run.repo)?,
1465 run.dispatched_at,
1466 run.class.map(Class::name),
1467 run.scope.join(","),
1468 run.tier,
1469 run.description,
1470 run.agent_budget,
1471 run.timeout_minutes,
1472 run.verify,
1473 run.verify_base_ok,
1474 run.verify_base_seconds,
1475 run.model,
1476 run.complexity,
1477 run.features.map(Features::to_json),
1478 run.complexity.map(|_| crate::complexity::ESTIMATOR),
1479 run.route_revision,
1480 run.route,
1481 run.approval.map(Approval::name),
1482 run.workflow_instance,
1483 run.node,
1484 run.unit,
1485 run.lap,
1486 run.preset,
1487 serde_json::to_string(&run.extra_args).unwrap_or_else(|_| "[]".into()),
1488 run.git_snapshot,
1489 ],
1490 )?;
1491 run.id = self.conn.last_insert_rowid();
1492 Ok(run.id)
1493 }
1494
1495 pub fn update_run(&self, run: &Run) -> Result<()> {
1497 self.write_run(run, None)
1498 }
1499
1500 pub fn update_run_from(&self, run: &Run, read: RunState) -> Result<()> {
1504 self.write_run(run, Some(read))
1505 }
1506
1507 fn write_run(&self, run: &Run, read: Option<RunState>) -> Result<()> {
1508 let written = self.conn.execute(
1509 "UPDATE runs SET state = ?2, feedback = ?3, seconds = ?4, cost_usd = ?5,
1512 summary = ?6, commits = ?7, diffstat = ?8, verify_ok = ?9,
1513 error = ?10, outcome = ?11, prompt = ?12, ready_at = ?13,
1514 decided_at = ?14, published_at = ?15, review_seconds = ?16,
1515 changed_paths = ?17, scope_error = ?18, approved_tree = ?19,
1516 approved_head = ?20, approved_by = ?21, reviewed_tree = ?22, infra = ?23
1517 WHERE id = ?1 AND (?24 IS NULL OR state = ?24)",
1518 params![
1519 run.id,
1520 run.state.name(),
1521 run.feedback,
1522 run.seconds,
1523 run.cost_usd,
1524 run.summary,
1525 run.commits,
1526 run.diffstat,
1527 run.verify_ok,
1528 run.error,
1529 run.outcome,
1530 run.prompt,
1531 run.ready_at,
1532 run.decided_at,
1533 run.published_at,
1534 run.review_seconds,
1535 run.changed_paths
1536 .as_ref()
1537 .map(|p| serde_json::to_string(p).unwrap_or_default()),
1538 run.scope_error,
1539 run.approved_tree,
1540 run.approved_head,
1541 run.approved_by,
1542 run.reviewed_tree,
1543 run.infra,
1544 read.map(RunState::name),
1545 ],
1546 )?;
1547 if written == 0 {
1548 let now = self.run(run.id)?.state;
1549 return Err(format!(
1550 "run #{} is now {}; read it again with `pma review {}`",
1551 run.id,
1552 now.name(),
1553 run.id
1554 )
1555 .into());
1556 }
1557 Ok(())
1558 }
1559
1560 pub fn add_review_seconds(&self, id: i64, seconds: i64) -> Result<()> {
1563 self.conn.execute(
1564 "UPDATE runs SET review_seconds = coalesce(review_seconds, 0) + ?2 WHERE id = ?1",
1565 params![id, seconds],
1566 )?;
1567 Ok(())
1568 }
1569
1570 pub fn set_reviewed_tree(&self, id: i64, tree: &str) -> Result<()> {
1573 self.conn.execute(
1574 "UPDATE runs SET reviewed_tree = ?2 WHERE id = ?1",
1575 params![id, tree],
1576 )?;
1577 Ok(())
1578 }
1579
1580 pub fn runs(&self) -> Result<Vec<Run>> {
1581 let mut stmt = self.conn.prepare(
1582 "SELECT id, project, task_key, text, gh, quadrant, agent, branch, worktree,
1583 default_branch, base, prompt, state, feedback, seconds,
1584 cost_usd, summary, commits, diffstat, verify, verify_ok, error, outcome, repo,
1585 dispatched_at, ready_at, decided_at, published_at, review_seconds,
1586 class, scope, tier, description, agent_budget, timeout_minutes,
1587 verify_base_ok, verify_base_seconds, changed_paths, scope_error, model,
1588 complexity, features, estimator, route_revision, route, approval,
1589 approved_tree, approved_head, approved_by,
1590 workflow_instance, node, unit, lap, preset, extra_args, git_snapshot,
1591 reviewed_tree, infra
1592 FROM runs ORDER BY id",
1593 )?;
1594 let rows = stmt.query_map([], |r| {
1595 let state: String = r.get(12)?;
1596 Ok(Run {
1597 id: r.get(0)?,
1598 project: r.get(1)?,
1599 task_key: r.get(2)?,
1600 text: r.get(3)?,
1601 gh: r.get(4)?,
1602 quadrant: r.get(5)?,
1603 agent: r.get(6)?,
1604 branch: r.get(7)?,
1605 worktree: PathBuf::from(r.get::<_, String>(8)?),
1606 default_branch: r.get(9)?,
1607 base: r.get(10)?,
1608 prompt: r.get(11)?,
1609 state: RunState::parse(&state).unwrap_or(RunState::Failed),
1610 feedback: r.get(13)?,
1611 seconds: r.get(14)?,
1612 cost_usd: r.get(15)?,
1613 summary: r.get(16)?,
1614 commits: r.get(17)?,
1615 diffstat: r.get(18)?,
1616 verify: r.get(19)?,
1617 verify_ok: r.get(20)?,
1618 error: r.get(21)?,
1619 outcome: r.get(22)?,
1620 repo: PathBuf::from(r.get::<_, String>(23)?),
1621 dispatched_at: r.get(24)?,
1622 ready_at: r.get(25)?,
1623 decided_at: r.get(26)?,
1624 published_at: r.get(27)?,
1625 review_seconds: r.get(28)?,
1626 class: r
1627 .get::<_, Option<String>>(29)?
1628 .as_deref()
1629 .and_then(Class::parse),
1630 scope: r
1631 .get::<_, Option<String>>(30)?
1632 .unwrap_or_default()
1633 .split(',')
1634 .filter(|s| !s.is_empty())
1635 .map(str::to_string)
1636 .collect(),
1637 tier: r.get(31)?,
1638 description: r.get(32)?,
1639 agent_budget: r.get(33)?,
1640 timeout_minutes: r.get(34)?,
1641 verify_base_ok: r.get(35)?,
1642 verify_base_seconds: r.get(36)?,
1643 changed_paths: r
1644 .get::<_, Option<String>>(37)?
1645 .and_then(|s| serde_json::from_str(&s).ok()),
1646 scope_error: r.get(38)?,
1647 model: r.get(39)?,
1648 complexity: r.get(40)?,
1649 features: r
1650 .get::<_, Option<String>>(41)?
1651 .as_deref()
1652 .and_then(Features::from_json),
1653 estimator: r.get(42)?,
1654 route_revision: r.get(43)?,
1655 route: r.get(44)?,
1656 approval: r
1657 .get::<_, Option<String>>(45)?
1658 .as_deref()
1659 .and_then(Approval::parse),
1660 approved_tree: r.get(46)?,
1661 approved_head: r.get(47)?,
1662 approved_by: r.get(48)?,
1663 workflow_instance: r.get(49)?,
1664 node: r.get(50)?,
1665 unit: r.get(51)?,
1666 lap: r.get(52)?,
1667 preset: r.get(53)?,
1668 extra_args: r
1669 .get::<_, Option<String>>(54)?
1670 .and_then(|s| serde_json::from_str(&s).ok())
1671 .unwrap_or_default(),
1672 git_snapshot: r.get(55)?,
1673 reviewed_tree: r.get(56)?,
1674 infra: r.get(57)?,
1675 })
1676 })?;
1677 Ok(rows.collect::<rusqlite::Result<_>>()?)
1678 }
1679
1680 pub fn run(&self, id: i64) -> Result<Run> {
1681 self.runs()?
1682 .into_iter()
1683 .find(|r| r.id == id)
1684 .ok_or_else(|| format!("no run #{id}").into())
1685 }
1686
1687 pub fn add_campaign(&self, c: &Campaign) -> Result<()> {
1690 if c.members.is_empty() {
1691 return Err(format!("campaign `{}` selects no project", c.name).into());
1692 }
1693 let tx = self.conn.unchecked_transaction()?;
1694 tx.execute(
1695 "INSERT INTO campaigns (name, text, description, class, created_at)
1696 VALUES (?1, ?2, ?3, ?4, ?5)",
1697 params![c.name, c.text, c.description, c.class.name(), c.created_at],
1698 )
1699 .map_err(|e| format!("campaign `{}`: {e}", c.name))?;
1700 for (project, _) in &c.members {
1701 tx.execute(
1702 "INSERT INTO campaign_members (campaign, project) VALUES (?1, ?2)",
1703 params![c.name, project],
1704 )?;
1705 }
1706 tx.commit()?;
1707 Ok(())
1708 }
1709
1710 pub fn campaigns(&self) -> Result<Vec<Campaign>> {
1711 let mut stmt = self.conn.prepare(
1712 "SELECT name, text, description, class, created_at FROM campaigns ORDER BY name",
1713 )?;
1714 let rows: Vec<Campaign> = stmt
1715 .query_map([], |r| {
1716 let class: String = r.get(3)?;
1717 Ok(Campaign {
1718 name: r.get(0)?,
1719 text: r.get(1)?,
1720 description: r.get(2)?,
1721 class: Class::parse(&class).unwrap_or(Class::Specified),
1722 created_at: r.get(4)?,
1723 members: Vec::new(),
1724 })
1725 })?
1726 .collect::<rusqlite::Result<_>>()?;
1727 let mut stmt = self
1728 .conn
1729 .prepare("SELECT campaign, project, run_id FROM campaign_members ORDER BY project")?;
1730 let members: Vec<(String, String, Option<i64>)> = stmt
1731 .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?
1732 .collect::<rusqlite::Result<_>>()?;
1733 Ok(rows
1734 .into_iter()
1735 .map(|mut c| {
1736 c.members = members
1737 .iter()
1738 .filter(|(name, ..)| *name == c.name)
1739 .map(|(_, p, run)| (p.clone(), *run))
1740 .collect();
1741 c
1742 })
1743 .collect())
1744 }
1745
1746 pub fn campaign(&self, name: &str) -> Result<Campaign> {
1747 self.campaigns()?
1748 .into_iter()
1749 .find(|c| c.name == name)
1750 .ok_or_else(|| format!("unknown campaign `{name}`; see `pma campaign`").into())
1751 }
1752
1753 pub fn set_campaign_run(&self, campaign: &str, project: &str, run: i64) -> Result<()> {
1756 self.conn.execute(
1757 "UPDATE campaign_members SET run_id = ?3 WHERE campaign = ?1 AND project = ?2",
1758 params![campaign, project, run],
1759 )?;
1760 Ok(())
1761 }
1762
1763 pub fn remove_campaign(&self, name: &str) -> Result<bool> {
1765 self.conn
1766 .execute("DELETE FROM campaign_members WHERE campaign = ?1", [name])?;
1767 Ok(self
1768 .conn
1769 .execute("DELETE FROM campaigns WHERE name = ?1", [name])?
1770 > 0)
1771 }
1772
1773 pub fn add_route_revision(&self, document: &str, by: &str) -> Result<i64> {
1776 let document = Policy::parse(document)?.to_json();
1782 let next: i64 = self.conn.query_row(
1783 "SELECT COALESCE(MAX(revision), 0) + 1 FROM routes",
1784 [],
1785 |r| r.get(0),
1786 )?;
1787 self.conn.execute(
1788 "INSERT INTO routes (revision, document, proposed_by, created_at)
1789 VALUES (?1, ?2, ?3, ?4)",
1790 params![next, document, by, crate::dates::now()],
1791 )?;
1792 Ok(next)
1793 }
1794
1795 pub fn activate_route(&self, revision: i64, by: &str, shadow: bool) -> Result<()> {
1798 let tx = self.conn.unchecked_transaction()?;
1799 let found = tx.execute(
1800 "UPDATE routes SET activated_at = ?2, activated_by = ?3, shadow = ?4
1801 WHERE revision = ?1",
1802 params![revision, crate::dates::now(), by, shadow],
1803 )?;
1804 if found == 0 {
1805 return Err(format!("no policy revision {revision}").into());
1806 }
1807 tx.execute(
1808 "UPDATE routes SET activated_at = NULL, activated_by = NULL WHERE revision <> ?1",
1809 [revision],
1810 )?;
1811 tx.commit()?;
1812 Ok(())
1813 }
1814
1815 pub fn route_revisions(&self) -> Result<Vec<RouteRevision>> {
1817 let mut stmt = self.conn.prepare(
1818 "SELECT revision, document, proposed_by, created_at, activated_at, activated_by,
1819 shadow
1820 FROM routes ORDER BY revision",
1821 )?;
1822 let rows = stmt.query_map([], |r| {
1823 Ok(RouteRevision {
1824 revision: r.get(0)?,
1825 document: r.get(1)?,
1826 proposed_by: r.get(2)?,
1827 created_at: r.get(3)?,
1828 activated_at: r.get(4)?,
1829 activated_by: r.get(5)?,
1830 shadow: r.get(6)?,
1831 })
1832 })?;
1833 Ok(rows.collect::<rusqlite::Result<_>>()?)
1834 }
1835
1836 pub fn active_route(&self) -> Result<Option<RouteRevision>> {
1838 Ok(self
1839 .route_revisions()?
1840 .into_iter()
1841 .find(|r| r.activated_at.is_some()))
1842 }
1843
1844 pub fn add_workflow_revision(
1849 &self,
1850 doc: &crate::workflow::Document,
1851 source: Option<&str>,
1852 by: &str,
1853 worst_case_runs: i64,
1854 worst_case_cost: f64,
1855 ) -> Result<i64> {
1856 let next: i64 = self.conn.query_row(
1857 "SELECT COALESCE(MAX(revision), 0) + 1 FROM workflows",
1858 [],
1859 |r| r.get(0),
1860 )?;
1861 self.conn.execute(
1862 "INSERT INTO workflows
1863 (revision, document, source, proposed_by, created_at,
1864 worst_case_runs, worst_case_cost)
1865 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
1866 params![
1867 next,
1868 doc.to_json(),
1869 source,
1870 by,
1871 crate::dates::now(),
1872 worst_case_runs,
1873 worst_case_cost
1874 ],
1875 )?;
1876 Ok(next)
1877 }
1878
1879 pub fn activate_workflow(&self, revision: i64, by: &str) -> Result<()> {
1882 let tx = self.conn.unchecked_transaction()?;
1883 let found = tx.execute(
1884 "UPDATE workflows SET activated_at = ?2, activated_by = ?3 WHERE revision = ?1",
1885 params![revision, crate::dates::now(), by],
1886 )?;
1887 if found == 0 {
1888 return Err(format!("no workflow revision {revision}").into());
1889 }
1890 tx.execute(
1891 "UPDATE workflows SET activated_at = NULL, activated_by = NULL WHERE revision <> ?1",
1892 [revision],
1893 )?;
1894 tx.commit()?;
1895 Ok(())
1896 }
1897
1898 pub fn workflow_revisions(&self) -> Result<Vec<WorkflowRevision>> {
1900 let mut stmt = self.conn.prepare(
1901 "SELECT revision, document, source, proposed_by, created_at, activated_at,
1902 activated_by, worst_case_runs, worst_case_cost
1903 FROM workflows ORDER BY revision",
1904 )?;
1905 let rows = stmt.query_map([], |r| {
1906 Ok(WorkflowRevision {
1907 revision: r.get(0)?,
1908 document: r.get(1)?,
1909 source: r.get(2)?,
1910 proposed_by: r.get(3)?,
1911 created_at: r.get(4)?,
1912 activated_at: r.get(5)?,
1913 activated_by: r.get(6)?,
1914 worst_case_runs: r.get(7)?,
1915 worst_case_cost: r.get(8)?,
1916 })
1917 })?;
1918 Ok(rows.collect::<rusqlite::Result<_>>()?)
1919 }
1920
1921 pub fn active_workflow(&self) -> Result<Option<WorkflowRevision>> {
1923 Ok(self
1924 .workflow_revisions()?
1925 .into_iter()
1926 .find(|r| r.activated_at.is_some()))
1927 }
1928
1929 pub fn workflow_revision(&self, revision: i64) -> Result<Option<WorkflowRevision>> {
1932 Ok(self
1933 .workflow_revisions()?
1934 .into_iter()
1935 .find(|r| r.revision == revision))
1936 }
1937
1938 pub fn add_workflow_instance(
1940 &self,
1941 workflow: &str,
1942 revision: i64,
1943 args: &str,
1944 target: &str,
1945 ) -> Result<i64> {
1946 self.conn.execute(
1947 "INSERT INTO workflow_instances (workflow, revision, args, target, started_at)
1948 VALUES (?1, ?2, ?3, ?4, ?5)",
1949 params![workflow, revision, args, target, crate::dates::now()],
1950 )?;
1951 Ok(self.conn.last_insert_rowid())
1952 }
1953
1954 pub fn workflow_instances(&self) -> Result<Vec<WorkflowInstance>> {
1955 let mut stmt = self.conn.prepare(
1956 "SELECT id, workflow, revision, args, target, started_at, finished_at, outcome
1957 FROM workflow_instances ORDER BY id",
1958 )?;
1959 let rows = stmt.query_map([], |r| {
1960 Ok(WorkflowInstance {
1961 id: r.get(0)?,
1962 workflow: r.get(1)?,
1963 revision: r.get(2)?,
1964 args: r.get(3)?,
1965 target: r.get(4)?,
1966 started_at: r.get(5)?,
1967 finished_at: r.get(6)?,
1968 outcome: r.get(7)?,
1969 })
1970 })?;
1971 Ok(rows.collect::<rusqlite::Result<_>>()?)
1972 }
1973
1974 pub fn atomically<T>(&self, work: impl FnOnce() -> Result<T>) -> Result<T> {
1983 let tx = rusqlite::Transaction::new_unchecked(
1987 &self.conn,
1988 rusqlite::TransactionBehavior::Immediate,
1989 )?;
1990 let out = work()?;
1991 tx.commit()?;
1992 Ok(out)
1993 }
1994
1995 pub fn workflow_instance(&self, instance: i64) -> Result<Option<WorkflowInstance>> {
1997 Ok(self
1998 .workflow_instances()?
1999 .into_iter()
2000 .find(|i| i.id == instance))
2001 }
2002
2003 pub fn note_instance_outcome(&self, instance: i64, outcome: &str) -> Result<()> {
2007 self.conn.execute(
2008 "UPDATE workflow_instances SET outcome = ?2 WHERE id = ?1",
2009 params![instance, outcome],
2010 )?;
2011 Ok(())
2012 }
2013
2014 pub fn reopen_instance(&self, instance: i64, args: &str) -> Result<()> {
2017 self.conn.execute(
2018 "UPDATE workflow_instances SET args = ?2, outcome = NULL WHERE id = ?1",
2019 params![instance, args],
2020 )?;
2021 Ok(())
2022 }
2023
2024 pub fn finish_instance(&self, instance: i64, outcome: &str) -> Result<()> {
2025 self.conn.execute(
2026 "UPDATE workflow_instances SET finished_at = ?2, outcome = ?3 WHERE id = ?1",
2027 params![instance, crate::dates::now(), outcome],
2028 )?;
2029 Ok(())
2030 }
2031
2032 pub fn add_workflow_unit(&self, instance: i64, u: &WorkflowUnit) -> Result<()> {
2034 self.conn.execute(
2035 "INSERT INTO workflow_units
2036 (instance, id, type, node, parent, root, depth, lap, project, data, created_at)
2037 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
2038 params![
2039 instance,
2040 u.id,
2041 u.ty,
2042 u.node,
2043 u.parent,
2044 u.root,
2045 u.depth,
2046 u.lap,
2047 u.project,
2048 u.data,
2049 crate::dates::now()
2050 ],
2051 )?;
2052 Ok(())
2053 }
2054
2055 pub fn workflow_units(&self, instance: i64) -> Result<Vec<WorkflowUnit>> {
2056 let mut stmt = self.conn.prepare(
2057 "SELECT id, type, node, parent, root, depth, lap, project, data
2058 FROM workflow_units WHERE instance = ?1 ORDER BY rowid",
2059 )?;
2060 let rows = stmt.query_map([instance], |r| {
2061 Ok(WorkflowUnit {
2062 id: r.get(0)?,
2063 ty: r.get(1)?,
2064 node: r.get(2)?,
2065 parent: r.get(3)?,
2066 root: r.get(4)?,
2067 depth: r.get(5)?,
2068 lap: r.get(6)?,
2069 project: r.get(7)?,
2070 data: r.get(8)?,
2071 })
2072 })?;
2073 Ok(rows.collect::<rusqlite::Result<_>>()?)
2074 }
2075
2076 pub fn add_workflow_move(
2079 &self,
2080 instance: i64,
2081 unit: &str,
2082 edge: usize,
2083 taken: bool,
2084 reason: Option<&str>,
2085 ) -> Result<()> {
2086 self.conn.execute(
2087 "INSERT OR REPLACE INTO workflow_moves (instance, unit, edge, taken, reason, at)
2088 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
2089 params![
2090 instance,
2091 unit,
2092 edge as i64,
2093 taken,
2094 reason,
2095 crate::dates::now()
2096 ],
2097 )?;
2098 Ok(())
2099 }
2100
2101 pub fn workflow_moves(&self, instance: i64) -> Result<Vec<(String, usize, bool)>> {
2103 let mut stmt = self
2104 .conn
2105 .prepare("SELECT unit, edge, taken FROM workflow_moves WHERE instance = ?1")?;
2106 let rows = stmt.query_map([instance], |r| {
2107 Ok((
2108 r.get::<_, String>(0)?,
2109 r.get::<_, i64>(1)? as usize,
2110 r.get::<_, bool>(2)?,
2111 ))
2112 })?;
2113 Ok(rows.collect::<rusqlite::Result<_>>()?)
2114 }
2115
2116 pub fn add_workflow_verdict(
2117 &self,
2118 instance: i64,
2119 unit: &str,
2120 check: &str,
2121 verdict: &str,
2122 detail: Option<&str>,
2123 ) -> Result<()> {
2124 self.conn.execute(
2125 "INSERT OR REPLACE INTO workflow_verdicts
2126 (instance, unit, check_name, verdict, detail, at)
2127 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
2128 params![instance, unit, check, verdict, detail, crate::dates::now()],
2129 )?;
2130 Ok(())
2131 }
2132
2133 pub fn workflow_verdicts(&self, instance: i64) -> Result<Vec<(String, String, String)>> {
2135 let mut stmt = self.conn.prepare(
2136 "SELECT unit, check_name, verdict FROM workflow_verdicts WHERE instance = ?1",
2137 )?;
2138 let rows = stmt.query_map([instance], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
2139 Ok(rows.collect::<rusqlite::Result<_>>()?)
2140 }
2141
2142 pub fn set_preset(&self, p: &Preset) -> Result<()> {
2145 if self.agent(&p.agent).is_err() {
2146 return Err(format!("no worker `{}`; `pma agent` lists them", p.agent).into());
2147 }
2148 self.change_settings(|doc| crate::settings::set_preset(doc, p))
2149 }
2150
2151 pub fn presets(&self) -> Result<Vec<Preset>> {
2152 let mut stmt = self
2153 .conn
2154 .prepare("SELECT name, agent, model, args FROM presets ORDER BY name")?;
2155 let rows = stmt.query_map([], |r| {
2156 Ok(Preset {
2157 name: r.get(0)?,
2158 agent: r.get(1)?,
2159 model: r.get::<_, Option<String>>(2)?.filter(|m| !m.is_empty()),
2160 args: serde_json::from_str(&r.get::<_, String>(3)?).unwrap_or_default(),
2161 })
2162 })?;
2163 Ok(rows.collect::<rusqlite::Result<_>>()?)
2164 }
2165
2166 pub fn preset(&self, name: &str) -> Result<Preset> {
2167 self.presets()?
2168 .into_iter()
2169 .find(|p| p.name == name)
2170 .ok_or_else(|| format!("no preset `{name}`; `pma preset` lists them").into())
2171 }
2172
2173 pub fn remove_preset(&self, name: &str) -> Result<bool> {
2175 self.change_settings(|doc| crate::settings::remove_preset(doc, name))
2176 }
2177
2178 pub fn agents(&self) -> Result<Vec<Worker>> {
2180 let mut stmt = self.conn.prepare(
2181 "SELECT name, command, args, allow, parse, reports_cost, enforces_budget,
2182 sandbox, resumes, env
2183 FROM agents ORDER BY name",
2184 )?;
2185 let rows = stmt.query_map([], |r| {
2186 let args: String = r.get(2)?;
2187 let parse: String = r.get(4)?;
2188 Ok(Worker {
2189 name: r.get(0)?,
2190 command: r.get(1)?,
2191 args: serde_json::from_str(&args).unwrap_or_default(),
2192 allow: r.get::<_, Option<String>>(3)?.filter(|a| !a.is_empty()),
2193 parse: Parser::parse(&parse).unwrap_or(Parser::TextTail),
2194 reports_cost: r.get(5)?,
2195 enforces_budget: r.get(6)?,
2196 sandbox: r.get(7)?,
2197 resumes: r.get(8)?,
2198 env: serde_json::from_str(&r.get::<_, String>(9)?).unwrap_or_default(),
2199 })
2200 })?;
2201 Ok(rows.collect::<rusqlite::Result<_>>()?)
2202 }
2203
2204 pub fn agent(&self, name: &str) -> Result<Worker> {
2205 self.agents()?
2206 .into_iter()
2207 .find(|w| w.name == name)
2208 .ok_or_else(|| format!("unknown agent `{name}`; see `pma agent`").into())
2209 }
2210
2211 pub fn set_agent(&self, w: &Worker) -> Result<()> {
2212 self.change_settings(|doc| crate::settings::set_agent(doc, w))
2213 }
2214
2215 pub fn remove_agent(&self, name: &str) -> Result<bool> {
2217 self.change_settings(|doc| crate::settings::remove_agent(doc, name))
2218 }
2219
2220 pub fn decided_runs(&self, project: &str) -> Result<(i64, i64)> {
2223 let mut decided = 0;
2224 let mut accepted = 0;
2225 for run in self.runs()?.iter().filter(|r| r.project == project) {
2226 match run.state {
2227 RunState::Approved | RunState::Pushed | RunState::Merged => {
2228 decided += 1;
2229 accepted += 1;
2230 }
2231 RunState::Rejected if run.infra => {}
2232 RunState::Rejected | RunState::Closed => decided += 1,
2233 _ => {}
2234 }
2235 }
2236 Ok((decided, accepted))
2237 }
2238
2239 pub fn rekey_runs(&self, project: &str, old: &str, text: &str, new: &str) -> Result<()> {
2243 let text = crate::todo::normal_text(text);
2244 for run in self.runs()? {
2245 if run.project == project
2246 && !run.state.is_final()
2247 && (run.task_key == old || crate::todo::normal_text(&run.text) == text)
2248 {
2249 self.conn.execute(
2250 "UPDATE runs SET task_key = ?2 WHERE id = ?1",
2251 params![run.id, new],
2252 )?;
2253 }
2254 }
2255 Ok(())
2256 }
2257
2258 pub fn consumed_attempts(&self, project: &str, revision: &str) -> Result<i64> {
2260 Ok(self
2261 .conn
2262 .query_row(
2263 "SELECT attempts FROM exhaustion WHERE project = ?1 AND revision = ?2",
2264 params![project, revision],
2265 |r| r.get(0),
2266 )
2267 .optional()?
2270 .unwrap_or(0))
2271 }
2272
2273 pub fn consume_attempt(&self, project: &str, revision: &str) -> Result<i64> {
2275 self.conn.execute(
2276 "INSERT INTO exhaustion (project, revision, attempts, updated_at)
2277 VALUES (?1, ?2, 1, ?3)
2278 ON CONFLICT (project, revision)
2279 DO UPDATE SET attempts = attempts + 1, updated_at = ?3",
2280 params![project, revision, crate::dates::now()],
2281 )?;
2282 self.consumed_attempts(project, revision)
2283 }
2284
2285 pub fn refund_attempts(&self, project: &str, revision: &str, n: i64) -> Result<()> {
2287 self.conn.execute(
2288 "UPDATE exhaustion SET attempts = max(attempts - ?3, 0), updated_at = ?4
2289 WHERE project = ?1 AND revision = ?2",
2290 params![project, revision, n, crate::dates::now()],
2291 )?;
2292 Ok(())
2293 }
2294
2295 pub fn reset_attempts(&self, project: &str, revision: &str) -> Result<bool> {
2297 Ok(self.conn.execute(
2298 "DELETE FROM exhaustion WHERE project = ?1 AND revision = ?2",
2299 params![project, revision],
2300 )? > 0)
2301 }
2302
2303 pub fn verify_base(
2306 &self,
2307 project: &str,
2308 base: &str,
2309 command: &str,
2310 timeout_minutes: i64,
2311 sandbox: bool,
2312 ) -> Result<Option<(Option<bool>, i64)>> {
2313 let mut stmt = self.conn.prepare(
2314 "SELECT ok, seconds FROM verify_base
2315 WHERE project = ?1 AND base = ?2 AND command = ?3 AND timeout_minutes = ?4
2316 AND sandbox = ?5",
2317 )?;
2318 let mut rows = stmt.query_map(
2319 params![project, base, command, timeout_minutes, sandbox],
2320 |r| Ok((r.get(0)?, r.get(1)?)),
2321 )?;
2322 Ok(rows.next().transpose()?)
2323 }
2324
2325 #[allow(clippy::too_many_arguments)]
2326 pub fn set_verify_base(
2327 &self,
2328 project: &str,
2329 base: &str,
2330 command: &str,
2331 timeout_minutes: i64,
2332 sandbox: bool,
2333 ok: Option<bool>,
2334 seconds: i64,
2335 ) -> Result<()> {
2336 self.conn.execute(
2337 "INSERT OR REPLACE INTO verify_base
2338 (project, base, command, timeout_minutes, sandbox, ok, seconds, measured_at)
2339 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
2340 params![
2341 project,
2342 base,
2343 command,
2344 timeout_minutes,
2345 sandbox,
2346 ok,
2347 seconds,
2348 crate::dates::now()
2349 ],
2350 )?;
2351 Ok(())
2352 }
2353
2354 pub fn insert_attempt(&self, a: &Attempt) -> Result<i64> {
2357 self.conn.execute(
2358 "INSERT INTO attempts (run_id, n, agent, prompt, feedback, started_at, seconds,
2359 cost_usd, summary, verify, verify_ok, error, outcome, model)
2360 VALUES (?1, (SELECT COALESCE(MAX(n), 0) + 1 FROM attempts WHERE run_id = ?1),
2361 ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
2362 params![
2363 a.run_id,
2364 a.agent,
2365 a.prompt,
2366 a.feedback,
2367 a.started_at,
2368 a.seconds,
2369 a.cost_usd,
2370 a.summary,
2371 a.verify,
2372 a.verify_ok,
2373 a.error,
2374 a.outcome,
2375 a.model,
2376 ],
2377 )?;
2378 Ok(self.conn.last_insert_rowid())
2379 }
2380
2381 pub fn attempts(&self, run: Option<i64>) -> Result<Vec<Attempt>> {
2383 let mut stmt = self.conn.prepare(
2384 "SELECT id, run_id, n, agent, prompt, feedback, started_at, seconds, cost_usd,
2385 summary, verify, verify_ok, error, outcome, model
2386 FROM attempts WHERE (?1 IS NULL OR run_id = ?1) ORDER BY run_id, n",
2387 )?;
2388 let rows = stmt.query_map([run], |r| {
2389 Ok(Attempt {
2390 id: r.get(0)?,
2391 run_id: r.get(1)?,
2392 n: r.get(2)?,
2393 agent: r.get(3)?,
2394 prompt: r.get(4)?,
2395 feedback: r.get(5)?,
2396 started_at: r.get(6)?,
2397 seconds: r.get(7)?,
2398 cost_usd: r.get(8)?,
2399 summary: r.get(9)?,
2400 verify: r.get(10)?,
2401 verify_ok: r.get(11)?,
2402 error: r.get(12)?,
2403 outcome: r.get(13)?,
2404 model: r.get(14)?,
2405 })
2406 })?;
2407 Ok(rows.collect::<rusqlite::Result<_>>()?)
2408 }
2409
2410 pub fn fail_interrupted_runs(&self, _held: &Session) -> Result<usize> {
2413 Ok(self.conn.execute(
2414 "UPDATE runs SET state = 'failed', error = 'interrupted: pma exited during the run'
2415 WHERE state IN ('queued', 'running')",
2416 [],
2417 )?)
2418 }
2419
2420 #[cfg(test)]
2421 fn journal_mode(&self) -> String {
2422 self.conn
2423 .query_row("PRAGMA journal_mode", [], |r| r.get(0))
2424 .unwrap()
2425 }
2426}
2427
2428#[derive(Debug, Clone, PartialEq)]
2430pub struct Job {
2431 pub id: i64,
2432 pub args: Vec<String>,
2433 pub cwd: PathBuf,
2434 pub state: String,
2436 pub created_at: i64,
2437 pub started_at: Option<i64>,
2438 pub finished_at: Option<i64>,
2439 pub pid: Option<i64>,
2441 pub exit_code: Option<i64>,
2442 pub awaiting: Option<Vec<String>>,
2445}
2446
2447const JOB_COLUMNS: &str =
2448 "id, args, cwd, state, created_at, started_at, finished_at, pid, exit_code, awaiting";
2449
2450fn job_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<Job> {
2451 let args: String = r.get(1)?;
2452 Ok(Job {
2453 id: r.get(0)?,
2454 args: serde_json::from_str(&args).unwrap_or_default(),
2455 cwd: PathBuf::from(r.get::<_, String>(2)?),
2456 state: r.get(3)?,
2457 created_at: r.get(4)?,
2458 started_at: r.get(5)?,
2459 finished_at: r.get(6)?,
2460 pid: r.get(7)?,
2461 exit_code: r.get(8)?,
2462 awaiting: r
2463 .get::<_, Option<String>>(9)?
2464 .and_then(|a| serde_json::from_str(&a).ok()),
2465 })
2466}
2467
2468#[derive(Debug, Clone, PartialEq)]
2469pub struct Note {
2470 pub id: i64,
2471 pub created_at: i64,
2472 pub updated_at: i64,
2473 pub text: String,
2474}
2475
2476#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
2477pub enum RunState {
2478 #[default]
2479 Queued,
2480 Running,
2481 Ready,
2482 Failed,
2483 Approved,
2484 PrOpen,
2486 Rejected,
2488 Pushed,
2490 Merged,
2492 Closed,
2494 Applied,
2497 Done,
2500}
2501
2502impl RunState {
2503 const ALL: [(&'static str, RunState); 12] = [
2504 ("queued", RunState::Queued),
2505 ("running", RunState::Running),
2506 ("ready", RunState::Ready),
2507 ("failed", RunState::Failed),
2508 ("approved", RunState::Approved),
2509 ("pr-open", RunState::PrOpen),
2510 ("rejected", RunState::Rejected),
2511 ("pushed", RunState::Pushed),
2512 ("merged", RunState::Merged),
2513 ("closed", RunState::Closed),
2514 ("applied", RunState::Applied),
2515 ("done", RunState::Done),
2516 ];
2517
2518 pub fn name(self) -> &'static str {
2519 Self::ALL
2520 .iter()
2521 .find(|(_, s)| *s == self)
2522 .map_or("", |(n, _)| n)
2523 }
2524
2525 fn parse(s: &str) -> Option<RunState> {
2526 Self::ALL.iter().find(|(n, _)| *n == s).map(|(_, v)| *v)
2527 }
2528
2529 pub fn is_final(self) -> bool {
2531 matches!(
2532 self,
2533 RunState::Rejected
2534 | RunState::Pushed
2535 | RunState::Merged
2536 | RunState::Closed
2537 | RunState::Applied
2538 | RunState::Done
2539 )
2540 }
2541
2542 pub fn landed(self) -> bool {
2544 matches!(self, RunState::Pushed | RunState::Merged)
2545 }
2546}
2547
2548#[derive(Debug, Clone, PartialEq, Default)]
2552pub struct Run {
2553 pub id: i64,
2554 pub project: String,
2555 pub task_key: String,
2557 pub text: String,
2558 pub gh: Option<i64>,
2559 pub quadrant: Option<String>,
2561 pub agent: String,
2562 pub repo: PathBuf,
2564 pub branch: String,
2565 pub worktree: PathBuf,
2566 pub default_branch: String,
2567 pub base: String,
2569 pub prompt: String,
2570 pub state: RunState,
2571 pub feedback: Option<String>,
2573 pub seconds: Option<i64>,
2575 pub cost_usd: Option<f64>,
2577 pub summary: Option<String>,
2578 pub commits: Option<i64>,
2580 pub diffstat: Option<String>,
2581 pub verify: Option<String>,
2583 pub verify_ok: Option<bool>,
2584 pub error: Option<String>,
2585 pub outcome: Option<String>,
2587 pub dispatched_at: Option<i64>,
2591 pub ready_at: Option<i64>,
2593 pub decided_at: Option<i64>,
2596 pub published_at: Option<i64>,
2599 pub review_seconds: Option<i64>,
2602 pub class: Option<Class>,
2605 pub scope: Vec<String>,
2608 pub tier: Option<u8>,
2610 pub description: Option<String>,
2612 pub agent_budget: Option<f64>,
2615 pub timeout_minutes: Option<i64>,
2616 pub verify_base_ok: Option<bool>,
2620 pub verify_base_seconds: Option<i64>,
2621 pub changed_paths: Option<Vec<String>>,
2625 pub scope_error: Option<String>,
2627 pub model: Option<String>,
2629 pub complexity: Option<i64>,
2631 pub features: Option<Features>,
2633 pub estimator: Option<String>,
2634 pub route_revision: Option<i64>,
2637 pub route: Option<String>,
2638 pub approval: Option<Approval>,
2639 pub approved_tree: Option<String>,
2642 pub approved_head: Option<String>,
2645 pub approved_by: Option<String>,
2646 pub workflow_instance: Option<i64>,
2649 pub node: Option<String>,
2651 pub unit: Option<String>,
2653 pub lap: i64,
2655 pub preset: Option<String>,
2657 pub extra_args: Vec<String>,
2659 pub git_snapshot: Option<String>,
2662 pub reviewed_tree: Option<String>,
2666 pub infra: bool,
2668}
2669
2670impl Run {
2671 #[cfg(test)]
2673 pub fn blank() -> Run {
2674 Run {
2675 id: 0,
2676 project: "p".into(),
2677 task_key: "fix it".into(),
2678 text: "Fix it".into(),
2679 gh: None,
2680 quadrant: None,
2681 agent: "claude".into(),
2682 repo: PathBuf::from("/r/p"),
2683 branch: "pma/fix-it".into(),
2684 worktree: PathBuf::from("/w/p/fix-it"),
2685 default_branch: "main".into(),
2686 base: "abc".into(),
2687 prompt: "do it".into(),
2688 state: RunState::Queued,
2689 feedback: None,
2690 seconds: None,
2691 cost_usd: None,
2692 summary: None,
2693 commits: None,
2694 diffstat: None,
2695 verify: None,
2696 verify_ok: None,
2697 error: None,
2698 outcome: None,
2699 dispatched_at: None,
2700 ready_at: None,
2701 decided_at: None,
2702 published_at: None,
2703 review_seconds: None,
2704 class: None,
2705 scope: Vec::new(),
2706 tier: None,
2707 description: None,
2708 agent_budget: None,
2709 timeout_minutes: None,
2710 verify_base_ok: None,
2711 verify_base_seconds: None,
2712 changed_paths: None,
2713 scope_error: None,
2714 model: None,
2715 complexity: None,
2716 features: None,
2717 estimator: None,
2718 route_revision: None,
2719 route: None,
2720 approval: None,
2721 approved_tree: None,
2722 approved_head: None,
2723 approved_by: None,
2724 workflow_instance: None,
2725 node: None,
2726 unit: None,
2727 lap: 0,
2728 preset: None,
2729 extra_args: Vec::new(),
2730 git_snapshot: None,
2731 reviewed_tree: None,
2732 infra: false,
2733 }
2734 }
2735
2736 pub fn enter(&mut self, state: RunState) {
2739 let now = crate::dates::now();
2740 match state {
2741 RunState::Ready => self.ready_at = Some(now),
2742 RunState::Approved | RunState::Rejected => {
2743 self.decided_at.get_or_insert(now);
2744 }
2745 RunState::Pushed | RunState::PrOpen => {
2746 self.published_at.get_or_insert(now);
2747 }
2748 _ => {}
2749 }
2750 self.state = state;
2751 }
2752}
2753
2754#[derive(Debug, Clone, PartialEq)]
2758pub struct Attempt {
2759 pub id: i64,
2760 pub run_id: i64,
2761 pub n: i64,
2763 pub agent: String,
2764 pub prompt: String,
2765 pub feedback: Option<String>,
2767 pub started_at: i64,
2768 pub seconds: Option<i64>,
2770 pub cost_usd: Option<f64>,
2772 pub summary: Option<String>,
2773 pub verify: Option<String>,
2774 pub verify_ok: Option<bool>,
2775 pub error: Option<String>,
2776 pub outcome: Option<String>,
2779 pub model: Option<String>,
2781}
2782
2783#[derive(Debug, Clone, PartialEq)]
2785pub struct Campaign {
2786 pub name: String,
2787 pub text: String,
2788 pub description: String,
2789 pub class: Class,
2790 pub created_at: i64,
2791 pub members: Vec<(String, Option<i64>)>,
2793}
2794
2795#[derive(Debug, Clone, PartialEq)]
2797pub struct RouteRevision {
2798 pub revision: i64,
2799 pub document: String,
2800 pub proposed_by: String,
2801 pub created_at: i64,
2802 pub activated_at: Option<i64>,
2803 pub activated_by: Option<String>,
2804 pub shadow: bool,
2806}
2807
2808#[derive(Debug, Clone, PartialEq, Default)]
2811pub struct Preset {
2812 pub name: String,
2813 pub agent: String,
2814 pub model: Option<String>,
2815 pub args: Vec<String>,
2816}
2817
2818#[derive(Debug, Clone, PartialEq)]
2819pub struct WorkflowInstance {
2820 pub id: i64,
2821 pub workflow: String,
2822 pub revision: i64,
2823 pub args: String,
2826 pub target: String,
2828 pub started_at: i64,
2829 pub finished_at: Option<i64>,
2830 pub outcome: Option<String>,
2831}
2832
2833#[derive(Debug, Clone, PartialEq)]
2835pub struct WorkflowUnit {
2836 pub id: String,
2837 pub ty: String,
2838 pub node: String,
2840 pub parent: Option<String>,
2841 pub root: String,
2842 pub depth: i64,
2843 pub lap: i64,
2844 pub project: Option<String>,
2845 pub data: String,
2847}
2848
2849#[derive(Debug, Clone, PartialEq)]
2850pub struct WorkflowRevision {
2851 pub revision: i64,
2852 pub document: String,
2854 pub source: Option<String>,
2856 pub proposed_by: String,
2857 pub created_at: i64,
2858 pub activated_at: Option<i64>,
2859 pub activated_by: Option<String>,
2860 pub worst_case_runs: Option<i64>,
2863 pub worst_case_cost: Option<f64>,
2864}
2865
2866impl WorkflowRevision {
2867 pub fn document(&self) -> Result<crate::workflow::Document> {
2868 Ok(crate::workflow::Document::parse(&self.document)?)
2869 }
2870}
2871
2872impl RouteRevision {
2873 pub fn policy(&self) -> Result<Policy> {
2874 Ok(Policy::parse(&self.document)?)
2875 }
2876}
2877
2878fn seed_agent(conn: &Connection, w: &Worker) -> Result<()> {
2879 conn.execute(
2880 "INSERT OR REPLACE INTO agents
2881 (name, command, args, allow, parse, reports_cost, enforces_budget, sandbox,
2882 resumes, env)
2883 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
2884 params![
2885 w.name,
2886 w.command,
2887 serde_json::to_string(&w.args).unwrap_or_default(),
2888 w.allow,
2889 w.parse.name(),
2890 w.reports_cost,
2891 w.enforces_budget,
2892 w.sandbox,
2893 w.resumes,
2894 serde_json::to_string(&w.env).unwrap_or_else(|_| "{}".into()),
2895 ],
2896 )?;
2897 Ok(())
2898}
2899
2900fn path_str(path: &Path) -> Result<&str> {
2901 path.to_str()
2902 .ok_or_else(|| format!("{} is not valid UTF-8", path.display()).into())
2903}
2904
2905#[cfg(test)]
2906mod tests {
2907 use super::*;
2908 use crate::scan::{DepsFacts, ScannedItem, TodoFacts};
2909
2910 fn scratch(name: &str) -> PathBuf {
2911 let dir = std::env::temp_dir().join(format!("pma-store-{}-{name}", std::process::id()));
2912 let _ = std::fs::remove_dir_all(&dir);
2913 std::fs::create_dir_all(&dir).unwrap();
2914 dir
2915 }
2916
2917 fn item(key: &str, line: i64) -> ScannedItem {
2918 ScannedItem {
2919 key: key.into(),
2920 line,
2921 priority: Priority::High,
2922 text: key.into(),
2923 tags: vec!["a".into(), "b".into()],
2924 due: Some("2026-10-01".into()),
2925 gh: Some(4),
2926 group: Some("Security".into()),
2927 added_at: Some(100),
2928 }
2929 }
2930
2931 fn facts(name: &str, items: Vec<ScannedItem>) -> Facts {
2932 Facts {
2933 name: name.into(),
2934 path: PathBuf::from(format!("/r/{name}")),
2935 slug: Some(format!("shakfu/{name}")),
2936 todo: Some(TodoFacts {
2937 lint_errors: 1,
2938 items,
2939 }),
2940 dirty: 3,
2941 pma_branches: vec![],
2942 ahead: None,
2943 last_activity: Some(42),
2944 ci: Ci::Failing(vec!["test".into(), "wheels".into()]),
2945 deps: None,
2946 error: None,
2947 todo_unread: false,
2948 held: false,
2949 }
2950 }
2951
2952 #[test]
2953 fn opens_with_rollback_journal_and_reopens() {
2954 let dir = scratch("open");
2955 let db = dir.join("p.db");
2956 let store = Store::open(&db).unwrap();
2957 assert_eq!(store.journal_mode(), "delete");
2958 store.set_config("tiers.1", "0.9").unwrap();
2959 drop(store);
2960 let store = Store::open(&db).unwrap();
2961 assert_eq!(
2962 store.config_rows().unwrap(),
2963 [("tiers.1".to_string(), "0.9".to_string())]
2964 );
2965 assert!(store.reset_config("tiers.1").unwrap());
2966 assert!(!store.reset_config("tiers.1").unwrap());
2967 let _ = std::fs::remove_dir_all(dir);
2968 }
2969
2970 #[test]
2974 fn shipped_splits_into_pushed_merged_and_closed() {
2975 let dir = scratch("push-or-pr");
2976 let db = dir.join("v27.db");
2977 let conn = Connection::open(&db).unwrap();
2978 let steps = [
2979 SCHEMA,
2980 RUNS,
2981 DEPS_AND_NOTES,
2982 LEFTOVER,
2983 PR_OPEN,
2984 ATTEMPTS,
2985 SNAPSHOT,
2986 VERIFY_BASE,
2987 CHANGED_PATHS,
2988 EXHAUSTION,
2989 AGENTS,
2990 RETIRE_KEYS,
2991 COMPLEXITY,
2992 ROUTES,
2993 APPROVAL_EVIDENCE,
2994 CAMPAIGNS,
2995 SLUG,
2996 ABSENCE,
2997 PROJECT_TAGS,
2998 WORKFLOWS,
2999 WORKER_ENV,
3000 WORKER_MODEL,
3001 PRESETS,
3002 PRESETS_REPLACE_WORKER_MODEL,
3003 SANDUK_WORKER,
3004 GIT_SNAPSHOT,
3005 RETIRE_WEIGHTS,
3006 ];
3007 for step in steps {
3008 conn.execute_batch(step).unwrap();
3009 }
3010 conn.pragma_update(None, "user_version", 27).unwrap();
3011 for (state, outcome) in [
3012 ("shipped", "https://github.com/o/r/pull/3"),
3013 ("shipped", "pushed abc to main"),
3014 ("rejected", "https://github.com/o/r/pull/4"),
3015 ("rejected", ""),
3016 ] {
3017 conn.execute(
3018 "INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
3019 default_branch, base, prompt, state, outcome)
3020 VALUES ('p', 'k', 't', 'claude', '/k', 'pma/t', '/w', 'main', 'b', 'p', ?1,
3021 NULLIF(?2, ''))",
3022 params![state, outcome],
3023 )
3024 .unwrap();
3025 }
3026 for (key, value) in [
3027 ("publish", "push"),
3028 ("projects.a.publish", "pr"),
3029 ("timeout", "5"),
3030 ] {
3031 conn.execute(
3032 "INSERT INTO config (key, value) VALUES (?1, ?2)",
3033 params![key, value],
3034 )
3035 .unwrap();
3036 }
3037 drop(conn);
3038 let store = Store::open(&db).unwrap();
3039 let states: Vec<_> = store.runs().unwrap().iter().map(|r| r.state).collect();
3040 assert_eq!(
3041 states,
3042 [
3043 RunState::Merged,
3044 RunState::Pushed,
3045 RunState::Closed,
3046 RunState::Rejected
3047 ]
3048 );
3049 let keys: Vec<String> = store
3050 .config_rows()
3051 .unwrap()
3052 .into_iter()
3053 .map(|(k, _)| k)
3054 .collect();
3055 assert_eq!(keys, ["timeout"]);
3056 let _ = std::fs::remove_dir_all(dir);
3057 }
3058
3059 #[test]
3060 fn older_databases_are_upgraded_in_place() {
3061 let dir = scratch("migrate");
3062 for (version, steps) in [
3063 (1, &[SCHEMA][..]),
3064 (2, &[SCHEMA, RUNS][..]),
3065 (3, &[SCHEMA, RUNS, DEPS_AND_NOTES][..]),
3066 (4, &[SCHEMA, RUNS, DEPS_AND_NOTES, LEFTOVER][..]),
3067 ] {
3068 let db = dir.join(format!("v{version}.db"));
3069 let conn = Connection::open(&db).unwrap();
3070 for step in steps {
3071 conn.execute_batch(step).unwrap();
3072 }
3073 conn.pragma_update(None, "user_version", version).unwrap();
3074 conn.execute(
3075 "INSERT INTO projects (name, path) VALUES ('kept', '/k')",
3076 [],
3077 )
3078 .unwrap();
3079 if version >= 2 {
3080 for outcome in ["https://github.com/o/r/pull/3", "pushed abc to main"] {
3081 conn.execute(
3082 "INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
3083 default_branch, base, prompt, state, outcome)
3084 VALUES ('kept', 'k', 't', 'claude', '/k', 'pma/t', '/w', 'main', 'b',
3085 'p', 'shipped', ?1)",
3086 [outcome],
3087 )
3088 .unwrap();
3089 }
3090 }
3091 drop(conn);
3092 let store = Store::open(&db).unwrap();
3093 let p = store.project("kept").unwrap().unwrap();
3094 assert_eq!(
3095 (p.deps, p.deps_detail.as_str(), p.leftover),
3096 (None, "", 0),
3097 "v{version}"
3098 );
3099 let states: Vec<_> = store.runs().unwrap().iter().map(|r| r.state).collect();
3100 if version >= 2 {
3101 assert_eq!(
3102 states,
3103 [RunState::PrOpen, RunState::Pushed],
3104 "v{version}: a published pull request is open again, and a push is `pushed`"
3105 );
3106 }
3107 assert!(store.notes().unwrap().is_empty());
3108 drop(store);
3109 assert!(Store::open(&db).is_ok(), "v{version} reopens");
3110 }
3111 let _ = std::fs::remove_dir_all(dir);
3112 }
3113
3114 #[test]
3119 fn a_v5_database_gains_an_attempt_per_run() {
3120 let dir = scratch("attempts-migrate");
3121 let db = dir.join("v5.db");
3122 let conn = Connection::open(&db).unwrap();
3123 for step in [SCHEMA, RUNS, DEPS_AND_NOTES, LEFTOVER, PR_OPEN] {
3124 conn.execute_batch(step).unwrap();
3125 }
3126 conn.pragma_update(None, "user_version", 5).unwrap();
3127 conn.execute_batch(
3128 "INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
3129 default_branch, base, prompt, state, started_at, seconds,
3130 cost_usd, summary, verify, verify_ok)
3131 VALUES ('p', 'k', 't', 'claude', '/r', 'pma/t', '/w', 'main', 'b', 'do it',
3132 'ready', 100, 42, 0.5, 'first summary', 'make test', 1);
3133 INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
3134 default_branch, base, prompt, state)
3135 VALUES ('p', 'k2', 't2', 'claude', '/r', 'pma/t2', '/w2', 'main', 'b', 'do it',
3136 'queued');",
3137 )
3138 .unwrap();
3139 drop(conn);
3140
3141 let store = Store::open(&db).unwrap();
3142 let runs = store.runs().unwrap();
3143 assert_eq!(runs[0].dispatched_at, Some(100));
3144 assert_eq!(runs[1].dispatched_at, None);
3146
3147 let attempts = store.attempts(None).unwrap();
3148 assert_eq!(attempts.len(), 1, "only the run that started");
3149 let a = &attempts[0];
3150 assert_eq!((a.run_id, a.n, a.started_at), (runs[0].id, 1, 100));
3151 assert_eq!(a.summary.as_deref(), Some("first summary"));
3152 assert_eq!(
3153 (a.seconds, a.cost_usd, a.verify_ok),
3154 (Some(42), Some(0.5), Some(true))
3155 );
3156 assert_eq!(a.outcome, None, "the old row did not record one");
3157 assert_eq!(runs[0].class, None);
3159 assert_eq!(runs[0].tier, None);
3160 assert!(runs[0].scope.is_empty());
3161 let _ = std::fs::remove_dir_all(dir);
3162 }
3163
3164 #[test]
3167 fn upgrading_seeds_the_worker_that_was_compiled_in() {
3168 let dir = scratch("agents");
3169 let store = Store::open(&dir.join("p.db")).unwrap();
3170 assert_eq!(
3173 store.agents().unwrap(),
3174 [
3175 Worker::claude(),
3176 Worker::omp(),
3177 Worker::opencode(),
3178 Worker::sanduk()
3179 ]
3180 );
3181
3182 let mut w = Worker::claude();
3183 w.name = "codex".into();
3184 w.command = "codex".into();
3185 w.args = vec![
3186 "exec".into(),
3187 "-C".into(),
3188 "{dir}".into(),
3189 "{prompt}".into(),
3190 ];
3191 w.allow = None;
3192 w.parse = Parser::TextTail;
3193 w.reports_cost = false;
3194 w.sandbox = true;
3195 store.set_agent(&w).unwrap();
3196 assert_eq!(
3197 store.agents().unwrap(),
3198 [
3199 Worker::claude(),
3200 w.clone(),
3201 Worker::omp(),
3202 Worker::opencode(),
3203 Worker::sanduk()
3204 ]
3205 );
3206 assert_eq!(store.agent("codex").unwrap(), w);
3207 assert!(store.agent("cursor").is_err());
3208
3209 assert!(store.remove_agent("codex").unwrap());
3210 assert!(!store.remove_agent("codex").unwrap());
3211 assert_eq!(
3212 store.agents().unwrap(),
3213 [
3214 Worker::claude(),
3215 Worker::omp(),
3216 Worker::opencode(),
3217 Worker::sanduk()
3218 ]
3219 );
3220
3221 let mut edited = Worker::opencode();
3224 edited.env = std::collections::BTreeMap::from([(
3225 "OPENAI_BASE_URL".to_string(),
3226 "http://localhost:11434/v1".to_string(),
3227 )]);
3228 store.set_agent(&edited).unwrap();
3229 drop(store);
3230 let store = Store::open(&dir.join("p.db")).unwrap();
3231 assert_eq!(store.agent("opencode").unwrap(), edited);
3232 let _ = std::fs::remove_dir_all(dir);
3233 }
3234
3235 #[test]
3238 fn the_claude_template_moves_to_stream_json_unless_edited() {
3239 let old = r#"["-p","{prompt}","--output-format","json","--permission-mode","acceptEdits","--model","{model}","--max-budget-usd","{budget}","{extra}"]"#;
3240 for (args, streams) in [(old, true), (r#"["-p","{prompt}"]"#, false)] {
3241 let dir = scratch("stream-json");
3242 let path = dir.join("p.db");
3243 {
3244 let store = Store::open(&path).unwrap();
3245 store
3246 .conn
3247 .execute_batch("ALTER TABLE jobs DROP COLUMN awaiting;")
3248 .unwrap();
3249 store
3250 .conn
3251 .execute("UPDATE agents SET args = ?1 WHERE name = 'claude'", [args])
3252 .unwrap();
3253 store.conn.pragma_update(None, "user_version", 32).unwrap();
3254 }
3255 let store = Store::open(&path).unwrap();
3256 let got = store.agent("claude").unwrap().args;
3257 assert_eq!(got.contains(&"stream-json".to_string()), streams, "{got:?}");
3258 if streams {
3259 assert_eq!(got, Worker::claude().args);
3260 }
3261 let _ = std::fs::remove_dir_all(dir);
3262 }
3263 }
3264
3265 #[test]
3268 fn a_ready_read_run_is_done_after_the_migration() {
3269 let dir = scratch("read-runs");
3270 let path = dir.join("p.db");
3271 {
3272 let store = Store::open(&path).unwrap();
3273 for (branch, instance) in [("", Some(1)), ("pma/t", Some(1)), ("pma/u", None)] {
3274 store
3275 .conn
3276 .execute(
3277 "INSERT INTO runs (project, task_key, text, agent, repo, branch, worktree,
3278 default_branch, base, prompt, state, workflow_instance)
3279 VALUES ('p', 'k', 't', 'claude', '/k', ?1, '/w', 'main', 'b', 'p',
3280 'ready', ?2)",
3281 params![branch, instance],
3282 )
3283 .unwrap();
3284 }
3285 store.conn.pragma_update(None, "user_version", 33).unwrap();
3286 }
3287 let store = Store::open(&path).unwrap();
3288 let states: Vec<_> = store.runs().unwrap().iter().map(|r| r.state).collect();
3289 assert_eq!(states, [RunState::Done, RunState::Ready, RunState::Ready]);
3290 let _ = std::fs::remove_dir_all(dir);
3291 }
3292
3293 #[test]
3297 fn retiring_a_workers_model_leaves_a_preset_in_its_place() {
3298 let dir = scratch("worker-model");
3299 let path = dir.join("p.db");
3300 {
3302 let store = Store::open(&path).unwrap();
3303 store.set_config("agent", "claude").unwrap();
3306 store
3307 .conn
3308 .execute_batch(
3309 "ALTER TABLE agents ADD COLUMN model TEXT;
3310 ALTER TABLE runs DROP COLUMN git_snapshot;
3311 ALTER TABLE runs DROP COLUMN reviewed_tree;
3312 ALTER TABLE runs DROP COLUMN infra;
3313 DROP TABLE jobs;",
3314 )
3315 .unwrap();
3316 store
3317 .conn
3318 .execute("UPDATE agents SET model = 'opus' WHERE name = 'claude'", [])
3319 .unwrap();
3320 store.conn.pragma_update(None, "user_version", 23).unwrap();
3321 }
3322 let store = Store::open(&path).unwrap();
3323 assert_eq!(
3324 store.preset("claude").unwrap(),
3325 Preset {
3326 name: "claude".into(),
3327 agent: "claude".into(),
3328 model: Some("opus".into()),
3329 args: Vec::new(),
3330 }
3331 );
3332 assert_eq!(
3333 store
3334 .config_rows()
3335 .unwrap()
3336 .iter()
3337 .find(|(k, _)| k == "preset"),
3338 Some(&("preset".to_string(), "claude".to_string())),
3339 "and it is the default, because that worker was the configured one"
3340 );
3341 let _ = std::fs::remove_dir_all(dir);
3342 }
3343
3344 #[test]
3347 fn consumed_attempts_are_kept_per_task_revision() {
3348 let dir = scratch("exhaustion");
3349 let store = Store::open(&dir.join("p.db")).unwrap();
3350 assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 0);
3351
3352 assert_eq!(store.consume_attempt("p", "fix the parser").unwrap(), 1);
3353 assert_eq!(store.consume_attempt("p", "fix the parser").unwrap(), 2);
3354 assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 2);
3355 assert_eq!(
3358 store
3359 .consumed_attempts("p", "fix the parser on empty input")
3360 .unwrap(),
3361 0
3362 );
3363 assert_eq!(store.consumed_attempts("q", "fix the parser").unwrap(), 0);
3364
3365 store.refund_attempts("p", "fix the parser", 1).unwrap();
3367 assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 1);
3368 store.refund_attempts("p", "fix the parser", 3).unwrap();
3369 assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 0);
3370 store.consume_attempt("p", "fix the parser").unwrap();
3371
3372 assert!(store.reset_attempts("p", "fix the parser").unwrap());
3373 assert_eq!(store.consumed_attempts("p", "fix the parser").unwrap(), 0);
3374 assert!(!store.reset_attempts("p", "fix the parser").unwrap());
3375 let _ = std::fs::remove_dir_all(dir);
3376 }
3377
3378 #[test]
3383 fn base_verification_is_cached_per_command_and_timeout() {
3384 let dir = scratch("verify-base");
3385 let store = Store::open(&dir.join("p.db")).unwrap();
3386 assert_eq!(
3387 store
3388 .verify_base("p", "abc", "make test", 30, true)
3389 .unwrap(),
3390 None
3391 );
3392
3393 store
3394 .set_verify_base("p", "abc", "make test", 30, true, Some(false), 12)
3395 .unwrap();
3396 assert_eq!(
3397 store
3398 .verify_base("p", "abc", "make test", 30, true)
3399 .unwrap(),
3400 Some((Some(false), 12))
3401 );
3402 for (base, command, timeout, sandbox) in [
3403 ("def", "make test", 30, true),
3404 ("abc", "cargo test", 30, true),
3405 ("abc", "make test", 10, true),
3406 ("abc", "make test", 30, false),
3407 ] {
3408 assert_eq!(
3409 store
3410 .verify_base("p", base, command, timeout, sandbox)
3411 .unwrap(),
3412 None,
3413 "{base} {command} {timeout} {sandbox}"
3414 );
3415 }
3416 store
3418 .set_verify_base("p", "def", "make test", 30, true, None, 0)
3419 .unwrap();
3420 assert_eq!(
3421 store
3422 .verify_base("p", "def", "make test", 30, true)
3423 .unwrap(),
3424 Some((None, 0))
3425 );
3426 let _ = std::fs::remove_dir_all(dir);
3427 }
3428
3429 #[test]
3431 fn a_guarded_write_refuses_a_run_that_moved_on() {
3432 let dir = scratch("guarded");
3433 let store = Store::open(&dir.join("p.db")).unwrap();
3434 let mut run = Run::blank();
3435 run.state = RunState::Ready;
3436 store.insert_run(&mut run).unwrap();
3437 let mut stale = store.run(run.id).unwrap();
3438
3439 let mut published = store.run(run.id).unwrap();
3440 published.state = RunState::Pushed;
3441 store.update_run(&published).unwrap();
3442
3443 stale.state = RunState::Approved;
3444 let err = store
3445 .update_run_from(&stale, RunState::Ready)
3446 .unwrap_err()
3447 .to_string();
3448 assert!(err.contains("is now pushed"), "{err}");
3449 assert_eq!(store.run(run.id).unwrap().state, RunState::Pushed);
3450 store.update_run_from(&stale, RunState::Pushed).unwrap();
3451
3452 store.add_review_seconds(run.id, 60).unwrap();
3453 store.add_review_seconds(run.id, 60).unwrap();
3454 assert_eq!(store.run(run.id).unwrap().review_seconds, Some(120));
3455 let _ = std::fs::remove_dir_all(dir);
3456 }
3457
3458 #[test]
3462 fn atomically_waits_for_another_writer() {
3463 let dir = scratch("immediate");
3464 let path = dir.join("p.db");
3465 let store = Store::open(&path).unwrap();
3466 let (held, holding) = std::sync::mpsc::channel();
3467 let other = path.clone();
3468 let writer = std::thread::spawn(move || {
3469 let conn = Connection::open(&other).unwrap();
3470 conn.execute_batch("BEGIN IMMEDIATE; INSERT INTO config VALUES ('a', '1');")
3471 .unwrap();
3472 held.send(()).unwrap();
3473 std::thread::sleep(Duration::from_millis(300));
3474 conn.execute_batch("COMMIT").unwrap();
3475 });
3476 holding.recv().unwrap();
3477 store
3478 .atomically(|| {
3479 let n: i64 = store
3480 .conn
3481 .query_row("SELECT count(*) FROM config", [], |r| r.get(0))?;
3482 store
3483 .conn
3484 .execute("INSERT INTO config VALUES ('b', ?1)", [n.to_string()])?;
3485 Ok(())
3486 })
3487 .unwrap();
3488 writer.join().unwrap();
3489 assert_eq!(store.config_rows().unwrap().len(), 2);
3490 let _ = std::fs::remove_dir_all(dir);
3491 }
3492
3493 #[test]
3496 fn an_infrastructure_rejection_is_not_decided() {
3497 let dir = scratch("decided");
3498 let store = Store::open(&dir.join("p.db")).unwrap();
3499 for (state, infra) in [
3500 (RunState::Merged, false),
3501 (RunState::Rejected, false),
3502 (RunState::Rejected, true),
3503 ] {
3504 let mut run = Run::blank();
3505 run.project = "p".into();
3506 store.insert_run(&mut run).unwrap();
3507 run.state = state;
3508 run.infra = infra;
3509 store.update_run(&run).unwrap();
3510 }
3511 assert_eq!(store.decided_runs("p").unwrap(), (2, 1));
3512 assert!(store.runs().unwrap()[2].infra, "stored and read back");
3513 let _ = std::fs::remove_dir_all(dir);
3514 }
3515
3516 #[test]
3518 fn attempts_are_appended_never_replaced() {
3519 let dir = scratch("attempts");
3520 let store = Store::open(&dir.join("p.db")).unwrap();
3521 let mut run = Run::blank();
3522 store.insert_run(&mut run).unwrap();
3523 assert!(run.dispatched_at.is_some(), "stamped on insert");
3524
3525 let mut a = Attempt {
3526 id: 0,
3527 run_id: run.id,
3528 n: 0,
3529 agent: "claude".into(),
3530 prompt: "do it".into(),
3531 feedback: None,
3532 started_at: 10,
3533 seconds: Some(30),
3534 cost_usd: Some(0.2),
3535 summary: Some("first".into()),
3536 verify: Some("make test".into()),
3537 verify_ok: Some(false),
3538 error: None,
3539 outcome: Some("ready".into()),
3540 model: Some("haiku".into()),
3541 };
3542 store.insert_attempt(&a).unwrap();
3543 a.feedback = Some("try again".into());
3544 a.summary = Some("second".into());
3545 a.verify_ok = Some(true);
3546 store.insert_attempt(&a).unwrap();
3547
3548 let rows = store.attempts(Some(run.id)).unwrap();
3549 assert_eq!(rows.iter().map(|r| r.n).collect::<Vec<_>>(), [1, 2]);
3550 assert_eq!(rows[0].summary.as_deref(), Some("first"));
3551 assert_eq!(rows[0].verify_ok, Some(false));
3552 assert_eq!(rows[1].feedback.as_deref(), Some("try again"));
3553 assert!(store.attempts(None).unwrap().len() == 2);
3554 let _ = std::fs::remove_dir_all(dir);
3555 }
3556
3557 #[test]
3560 fn transitions_are_stamped_once() {
3561 let mut run = Run::blank();
3562 run.enter(RunState::Ready);
3563 let ready = run.ready_at.expect("ready stamped");
3564 run.enter(RunState::Approved);
3565 let decided = run.decided_at.expect("approval stamped");
3566 run.enter(RunState::PrOpen);
3567 let published = run.published_at.expect("publication stamped");
3568
3569 run.enter(RunState::Merged);
3570 assert_eq!(
3571 run.published_at,
3572 Some(published),
3573 "a merge is not a publish"
3574 );
3575 run.enter(RunState::Rejected);
3576 assert_eq!(run.decided_at, Some(decided), "the approval stands");
3577 assert_eq!(run.ready_at, Some(ready));
3578 }
3579
3580 #[test]
3581 fn notes_add_edit_and_remove() {
3582 let dir = scratch("notes");
3583 let store = Store::open(&dir.join("p.db")).unwrap();
3584 let a = store
3585 .add_note("move CI to one reusable workflow", 10)
3586 .unwrap();
3587 let b = store.add_note("second", 20).unwrap();
3588 assert!(
3589 store
3590 .edit_note(a, "move CI to reusable workflows", 30)
3591 .unwrap()
3592 );
3593 assert!(!store.edit_note(99, "x", 30).unwrap());
3594 assert!(store.remove_note(b).unwrap());
3595 assert!(!store.remove_note(b).unwrap());
3596 assert_eq!(
3597 store.notes().unwrap(),
3598 [Note {
3599 id: a,
3600 created_at: 10,
3601 updated_at: 30,
3602 text: "move CI to reusable workflows".into()
3603 }]
3604 );
3605 let _ = std::fs::remove_dir_all(dir);
3606 }
3607
3608 #[test]
3609 fn runs_round_trip_and_interrupted_runs_fail() {
3610 let dir = scratch("runs");
3611 let mut store = Store::open(&dir.join("p.db")).unwrap();
3612 let mut run = Run {
3613 id: 0,
3614 project: "p".into(),
3615 task_key: "fix it".into(),
3616 text: "Fix it".into(),
3617 gh: Some(3),
3618 quadrant: Some("Q1".into()),
3619 agent: "claude".into(),
3620 repo: PathBuf::from("/r/p"),
3621 branch: "pma/fix-it".into(),
3622 worktree: PathBuf::from("/w/p/fix-it"),
3623 default_branch: "main".into(),
3624 base: "abc".into(),
3625 prompt: "do it".into(),
3626 state: RunState::Running,
3627 feedback: None,
3628 seconds: None,
3629 cost_usd: None,
3630 summary: None,
3631 commits: None,
3632 diffstat: None,
3633 verify: Some("make test".into()),
3634 verify_ok: None,
3635 error: None,
3636 outcome: None,
3637 dispatched_at: None,
3638 ready_at: None,
3639 decided_at: None,
3640 published_at: None,
3641 review_seconds: None,
3642 class: Some(Class::Mechanical),
3643 scope: vec!["Cargo.lock".into(), "Cargo.toml".into()],
3644 tier: Some(2),
3645 description: Some("within the constraints\n".into()),
3646 agent_budget: Some(1.0),
3647 timeout_minutes: Some(30),
3648 verify_base_ok: Some(true),
3649 verify_base_seconds: Some(12),
3650 changed_paths: None,
3651 scope_error: None,
3652 model: Some("sonnet".into()),
3653 complexity: Some(2),
3654 features: Some(Features {
3655 text_words: 3,
3656 ..Features::default()
3657 }),
3658 estimator: Some(crate::complexity::ESTIMATOR.into()),
3659 route_revision: Some(2),
3660 route: Some("chores".into()),
3661 approval: Some(Approval::Batch),
3662 approved_tree: None,
3663 approved_head: None,
3664 approved_by: None,
3665 workflow_instance: None,
3666 node: None,
3667 unit: None,
3668 lap: 0,
3669 preset: None,
3670 extra_args: Vec::new(),
3671 git_snapshot: Some("hook pre-commit abc executable".into()),
3672 reviewed_tree: None,
3673 infra: false,
3674 };
3675 store.insert_run(&mut run).unwrap();
3676 assert_eq!(store.run(run.id).unwrap(), run);
3677
3678 run.state = RunState::Ready;
3679 run.cost_usd = Some(0.25);
3680 run.changed_paths = Some(vec!["Cargo.lock".into(), "a b/c\nd".into()]);
3681 run.verify_ok = Some(false);
3682 run.summary = Some("done".into());
3683 store.update_run(&run).unwrap();
3684 assert_eq!(store.run(run.id).unwrap(), run);
3685
3686 run.verify = Some("make something-else".into());
3688 store.update_run(&run).unwrap();
3689 assert_eq!(
3690 store.run(run.id).unwrap().verify.as_deref(),
3691 Some("make test")
3692 );
3693 run.verify = Some("make test".into());
3694
3695 run.state = RunState::PrOpen;
3696 store.update_run(&run).unwrap();
3697 assert_eq!(store.run(run.id).unwrap().state, RunState::PrOpen);
3698 assert!(!run.state.is_final(), "an open pull request holds its task");
3699
3700 run.state = RunState::Running;
3701 store.update_run(&run).unwrap();
3702 let session = Session::acquire(&dir).unwrap();
3703 assert_eq!(store.fail_interrupted_runs(&session).unwrap(), 1);
3704 let failed = store.run(run.id).unwrap();
3705 assert_eq!(failed.state, RunState::Failed);
3706 assert!(failed.error.unwrap().starts_with("interrupted"));
3707 assert!(store.run(99).is_err());
3708
3709 let mut scanned = facts("p", vec![]);
3711 scanned.pma_branches = vec!["pma/fix-it".into(), "pma/stray".into()];
3712 let mut other = facts("q", vec![]);
3713 other.pma_branches = vec!["pma/fix-it".into()];
3714 let leftover = |store: &mut Store, facts: &[Facts]| {
3715 store.save_scan(facts, false, 1).unwrap();
3716 let projects = store.projects().unwrap();
3717 projects.iter().map(|p| p.leftover).collect::<Vec<_>>()
3718 };
3719 assert_eq!(
3720 leftover(&mut store, &[scanned.clone(), other.clone()]),
3721 [1, 1]
3722 );
3723 run.state = RunState::Pushed;
3724 store.update_run(&run).unwrap();
3725 assert_eq!(leftover(&mut store, &[scanned, other]), [2, 1]);
3726 let _ = std::fs::remove_dir_all(dir);
3727 }
3728
3729 #[test]
3730 fn one_session_holds_the_lock_until_dropped() {
3731 let dir = scratch("session");
3732 let first = Session::acquire(&dir).unwrap();
3733 assert!(Session::try_acquire(&dir).unwrap().is_none());
3734 let err = Session::acquire(&dir).err().unwrap().to_string();
3735 assert!(
3736 err.contains(&format!("(pid {})", std::process::id())),
3737 "{err}"
3738 );
3739 drop(first);
3740 assert!(Session::try_acquire(&dir).unwrap().is_some());
3741
3742 let brief = Session::acquire(&dir).unwrap();
3743 let release = std::thread::spawn(move || {
3744 std::thread::sleep(Duration::from_millis(200));
3745 drop(brief);
3746 });
3747 assert!(Session::acquire(&dir).is_ok(), "a brief hold is waited out");
3748 release.join().unwrap();
3749 let _ = std::fs::remove_dir_all(dir);
3750 }
3751
3752 #[test]
3755 fn the_last_scan_owns_the_slug() {
3756 let dir = scratch("slug");
3757 let mut store = Store::open(&dir.join("p.db")).unwrap();
3758 store
3759 .save_scan(&[facts("one", vec![])], true, 1000)
3760 .unwrap();
3761 assert_eq!(
3762 store.project("one").unwrap().unwrap().slug.as_deref(),
3763 Some("shakfu/one")
3764 );
3765
3766 let mut moved = facts("one", vec![]);
3767 moved.slug = None;
3768 store.save_scan(&[moved], true, 2000).unwrap();
3769 assert_eq!(store.project("one").unwrap().unwrap().slug, None);
3770 let _ = std::fs::remove_dir_all(dir);
3771 }
3772
3773 #[test]
3776 fn forgetting_a_project_keeps_its_runs() {
3777 let dir = scratch("forget");
3778 let mut store = Store::open(&dir.join("p.db")).unwrap();
3779 store
3780 .save_scan(
3781 &[facts("a", vec![item("t", 1)]), facts("b", vec![])],
3782 true,
3783 1,
3784 )
3785 .unwrap();
3786 store.set_tier("a", Path::new("/r/a"), Some(1)).unwrap();
3787 store.add_project_tag("a", "ai").unwrap();
3788 store.add_project_tag("b", "ai").unwrap();
3789 let mut run = Run::blank();
3790 run.project = "a".into();
3791 store.insert_run(&mut run).unwrap();
3792 store.consume_attempt("a", "t").unwrap();
3793
3794 let f = store.project_footprint("a").unwrap();
3795 assert_eq!((f.tasks, f.tags, f.runs), (1, 1, 1));
3796 assert_eq!(f.open_runs, [run.id], "an open run holds a worktree");
3797
3798 let mut open = store.run(run.id).unwrap();
3799 open.state = RunState::Merged;
3800 store.update_run(&open).unwrap();
3801 assert!(store.project_footprint("a").unwrap().open_runs.is_empty());
3802
3803 store.forget_project("a").unwrap();
3804 assert!(store.project("a").unwrap().is_none());
3805 assert!(store.tasks().unwrap().is_empty(), "its tasks go");
3806 assert_eq!(
3807 store.projects_tagged(&["ai".into()]).unwrap(),
3808 ["b"],
3809 "another project keeps the tag"
3810 );
3811 assert_eq!(store.consumed_attempts("a", "t").unwrap(), 0);
3812 assert_eq!(store.run(run.id).unwrap().project, "a", "the run is kept");
3813 let _ = std::fs::remove_dir_all(dir);
3814 }
3815
3816 #[test]
3817 fn a_project_carries_several_tags() {
3818 let dir = scratch("tags");
3819 let store = Store::open(&dir.join("p.db")).unwrap();
3820 assert!(store.add_project_tag("cyllama", "ai").unwrap());
3821 assert!(
3822 !store.add_project_tag("cyllama", "ai").unwrap(),
3823 "idempotent"
3824 );
3825 store.add_project_tag("cyllama", "audio").unwrap();
3826 store.add_project_tag("cysox", "audio").unwrap();
3827
3828 assert_eq!(
3829 store.project_tags().unwrap(),
3830 [
3831 ("cyllama".to_string(), "ai".to_string()),
3832 ("cyllama".to_string(), "audio".to_string()),
3833 ("cysox".to_string(), "audio".to_string()),
3834 ]
3835 );
3836 assert_eq!(
3837 store.projects_tagged(&["audio".into()]).unwrap(),
3838 ["cyllama", "cysox"]
3839 );
3840 assert_eq!(
3841 store
3842 .projects_tagged(&["ai".into(), "audio".into()])
3843 .unwrap(),
3844 ["cyllama", "cysox"],
3845 "a project in both tags is listed once"
3846 );
3847
3848 assert!(store.remove_project_tag("cyllama", "audio").unwrap());
3849 assert!(!store.remove_project_tag("cyllama", "audio").unwrap());
3850 assert_eq!(store.projects_tagged(&["audio".into()]).unwrap(), ["cysox"]);
3851 assert_eq!(
3852 store.projects_tagged(&["ai".into()]).unwrap(),
3853 ["cyllama"],
3854 "removing one tag leaves the other"
3855 );
3856 let _ = std::fs::remove_dir_all(dir);
3857 }
3858
3859 #[test]
3862 fn settings_toml_takes_over_the_configuration_tables() {
3863 let dir = scratch("settings");
3864 let path = dir.join("p.db");
3865 {
3866 let store = Store::open(&path).unwrap();
3867 store
3868 .conn
3869 .execute_batch(
3870 "INSERT INTO config (key, value) VALUES ('agent_budget', '3');
3871 INSERT INTO roots (path) VALUES ('/r');
3872 INSERT INTO project_tags (project, tag) VALUES ('a', 'py');",
3873 )
3874 .unwrap();
3875 store
3876 .conn
3877 .execute_batch("DELETE FROM settings_sync")
3878 .unwrap();
3879 }
3880 std::fs::remove_file(dir.join(crate::settings::FILE)).unwrap();
3881 let mut store = Store::open(&path).unwrap();
3882 let text = std::fs::read_to_string(dir.join(crate::settings::FILE)).unwrap();
3883 assert!(
3884 text.contains("agent_budget = 3") && text.contains("roots = [\"/r\"]"),
3885 "{text}"
3886 );
3887 assert!(
3888 text.contains("[agents.claude]"),
3889 "the seeded workers: {text}"
3890 );
3891
3892 std::fs::write(
3894 dir.join(crate::settings::FILE),
3895 text.replace("agent_budget = 3", "agent_budget = 4") + "\n[projects.b]\ntier = 2\n",
3896 )
3897 .unwrap();
3898 drop(store);
3899 store = Store::open(&path).unwrap();
3900 assert!(
3901 store
3902 .config_rows()
3903 .unwrap()
3904 .contains(&("agent_budget".into(), "4".into()))
3905 );
3906 assert!(store.project("b").unwrap().is_none());
3907 store.save_scan(&[facts("b", vec![])], true, 1).unwrap();
3908 assert_eq!(store.project("b").unwrap().unwrap().tier, Some(2));
3909
3910 std::fs::write(
3912 dir.join(crate::settings::FILE),
3913 "[config]\nmax_parallel = 0\n",
3914 )
3915 .unwrap();
3916 drop(store);
3917 let e = Store::open(&path).err().expect("refused").to_string();
3918 assert!(e.contains("config.max_parallel"), "{e}");
3919 let _ = std::fs::remove_dir_all(dir);
3920 }
3921
3922 #[test]
3923 fn roots_add_and_remove() {
3924 let dir = scratch("roots");
3925 let store = Store::open(&dir.join("p.db")).unwrap();
3926 assert!(store.add_root(Path::new("/a")).unwrap());
3927 assert!(!store.add_root(Path::new("/a")).unwrap());
3928 store.add_root(Path::new("/b")).unwrap();
3929 assert_eq!(
3930 store.roots().unwrap(),
3931 [PathBuf::from("/a"), PathBuf::from("/b")]
3932 );
3933 assert!(store.remove_root(Path::new("/a")).unwrap());
3934 assert_eq!(store.roots().unwrap(), [PathBuf::from("/b")]);
3935 let _ = std::fs::remove_dir_all(dir);
3936 }
3937
3938 #[test]
3939 fn scan_round_trips_and_keeps_tier_and_first_seen() {
3940 let dir = scratch("scan");
3941 let mut store = Store::open(&dir.join("p.db")).unwrap();
3942 store.set_tier("one", Path::new("/r/one"), Some(2)).unwrap();
3943
3944 store
3945 .save_scan(
3946 &[facts("one", vec![item("x", 5), item("y", 6)])],
3947 true,
3948 1000,
3949 )
3950 .unwrap();
3951 store
3952 .save_scan(
3953 &[facts("one", vec![item("y", 7), item("z", 8)])],
3954 true,
3955 2000,
3956 )
3957 .unwrap();
3958
3959 let p = store.project("one").unwrap().unwrap();
3960 assert_eq!(p.tier, Some(2), "a scan keeps the tier");
3961 assert_eq!((p.deps, p.deps_at), (None, None));
3962 assert_eq!(p.slug.as_deref(), Some("shakfu/one"));
3963
3964 let mut measured = facts("one", vec![item("y", 7), item("z", 8)]);
3965 measured.deps = Some(DepsFacts {
3966 outdated: Some(2),
3967 detail: "uv: ruff 0.14 -> 0.16".into(),
3968 });
3969 store.save_scan(&[measured], true, 3000).unwrap();
3970 store
3971 .save_scan(
3972 &[facts("one", vec![item("y", 7), item("z", 8)])],
3973 true,
3974 4000,
3975 )
3976 .unwrap();
3977 let p = store.project("one").unwrap().unwrap();
3978 assert_eq!(
3979 (p.deps, p.deps_detail.as_str(), p.deps_at),
3980 (Some(2), "uv: ruff 0.14 -> 0.16", Some(3000)),
3981 "a scan that does not measure deps keeps the last measurement"
3982 );
3983 assert_eq!(
3984 (p.scanned_at, p.has_todo, p.lint_errors, p.dirty),
3985 (Some(4000), true, 1, 3)
3986 );
3987 assert_eq!((p.ahead, p.last_activity), (None, Some(42)));
3988 assert_eq!(p.ci, Ci::Failing(vec!["test".into(), "wheels".into()]));
3989
3990 let tasks = store.tasks().unwrap();
3991 let seen: Vec<_> = tasks
3992 .iter()
3993 .map(|t| (t.text.as_str(), t.line, t.first_seen))
3994 .collect();
3995 assert_eq!(
3996 seen,
3997 [("y", 7, 1000), ("z", 8, 2000)],
3998 "x is gone, y keeps its first scan"
3999 );
4000 assert_eq!(tasks[0].tags, ["a", "b"]);
4001 assert_eq!(tasks[0].group.as_deref(), Some("Security"));
4002 assert_eq!(
4003 (tasks[0].due.as_deref(), tasks[0].gh, tasks[0].added_at),
4004 (Some("2026-10-01"), Some(4), Some(100))
4005 );
4006 assert_eq!(store.last_scan().unwrap(), Some(4000));
4007 let _ = std::fs::remove_dir_all(dir);
4008 }
4009
4010 #[test]
4013 fn a_held_project_keeps_its_last_scan() {
4014 let dir = scratch("held");
4015 let mut store = Store::open(&dir.join("p.db")).unwrap();
4016 store
4017 .save_scan(&[facts("one", vec![item("x", 5)])], true, 1000)
4018 .unwrap();
4019 let held = Facts::held("one", Path::new("/r/one"), "settings changed".into());
4020 assert!(store.save_scan(&[held], true, 2000).unwrap().is_empty());
4021 let tasks: Vec<_> = store.tasks().unwrap().into_iter().map(|t| t.key).collect();
4022 assert_eq!(tasks, ["x"]);
4023 let p = store.project("one").unwrap().unwrap();
4024 assert_eq!((p.dirty, p.absent_since), (3, None));
4025 assert!(matches!(p.ci, Ci::Failing(_)), "{:?}", p.ci);
4026 assert_eq!(p.scan_error.as_deref(), Some("settings changed"));
4027 let _ = std::fs::remove_dir_all(dir);
4028 }
4029
4030 #[test]
4033 fn an_unreadable_todo_keeps_the_tasks() {
4034 let dir = scratch("unread");
4035 let mut store = Store::open(&dir.join("p.db")).unwrap();
4036 store
4037 .save_scan(&[facts("one", vec![item("x", 5)])], true, 1000)
4038 .unwrap();
4039 let unread = Facts {
4040 todo: None,
4041 todo_unread: true,
4042 error: Some("TODO.md: permission denied".into()),
4043 ..facts("one", vec![])
4044 };
4045 store.save_scan(&[unread], true, 2000).unwrap();
4046 let kept: Vec<_> = store
4047 .tasks()
4048 .unwrap()
4049 .into_iter()
4050 .map(|t| (t.key, t.first_seen))
4051 .collect();
4052 assert_eq!(kept, [("x".to_string(), 1000)]);
4053 let p = store.project("one").unwrap().unwrap();
4054 assert_eq!((p.has_todo, p.lint_errors), (true, 1));
4055 assert_eq!(p.scan_error.as_deref(), Some("TODO.md: permission denied"));
4056 let _ = std::fs::remove_dir_all(dir);
4057 }
4058
4059 #[test]
4062 fn a_task_keeps_its_first_seen_when_its_key_changes() {
4063 let dir = scratch("rekey");
4064 let mut store = Store::open(&dir.join("p.db")).unwrap();
4065 let unlinked = ScannedItem {
4066 gh: None,
4067 ..item("fix it", 5)
4068 };
4069 store
4070 .save_scan(&[facts("one", vec![unlinked])], true, 1000)
4071 .unwrap();
4072 let linked = ScannedItem {
4073 key: "gh:9".into(),
4074 text: "Fix it".into(),
4075 gh: Some(9),
4076 ..item("fix it", 5)
4077 };
4078 store
4079 .save_scan(&[facts("one", vec![linked])], true, 2000)
4080 .unwrap();
4081 let tasks = store.tasks().unwrap();
4082 assert_eq!((tasks[0].key.as_str(), tasks[0].first_seen), ("gh:9", 1000));
4083 let _ = std::fs::remove_dir_all(dir);
4084 }
4085
4086 #[test]
4087 fn full_scan_marks_projects_no_longer_found() {
4088 let dir = scratch("full");
4089 let mut store = Store::open(&dir.join("p.db")).unwrap();
4090 store
4091 .save_scan(
4092 &[facts("a", vec![item("t", 1)]), facts("b", vec![])],
4093 true,
4094 1,
4095 )
4096 .unwrap();
4097 store.set_tier("a", Path::new("/r/a"), Some(1)).unwrap();
4098
4099 assert!(
4100 store
4101 .save_scan(&[facts("b", vec![])], false, 2)
4102 .unwrap()
4103 .is_empty(),
4104 "a partial scan marks nothing"
4105 );
4106 assert_eq!(
4107 store.projects().unwrap().len(),
4108 2,
4109 "a partial scan removes nothing"
4110 );
4111
4112 let absent = store.save_scan(&[facts("b", vec![])], true, 3).unwrap();
4113 assert_eq!(absent, ["a"]);
4114 let a = store.project("a").unwrap().unwrap();
4115 assert_eq!(a.absent_since, Some(3));
4116 assert_eq!(a.tier, Some(1), "an absent project keeps its tier");
4117 assert_eq!(
4118 store.tasks().unwrap().len(),
4119 1,
4120 "its tasks and their first_seen are kept"
4121 );
4122
4123 assert!(
4125 store
4126 .save_scan(&[facts("b", vec![])], true, 4)
4127 .unwrap()
4128 .is_empty()
4129 );
4130 assert_eq!(store.project("a").unwrap().unwrap().absent_since, Some(3));
4131
4132 store
4134 .save_scan(&[facts("a", vec![item("t", 1)])], false, 5)
4135 .unwrap();
4136 assert_eq!(store.project("a").unwrap().unwrap().absent_since, None);
4137 assert_eq!(store.tasks().unwrap()[0].first_seen, 1);
4138 let _ = std::fs::remove_dir_all(dir);
4139 }
4140}