Skip to main content

pma_core/
store.rs

1//! The portfolio database: roots, projects with tiers, settings, the results
2//! of the last scan, agent runs, and portfolio notes.
3//!
4//! One user and one session at a time is assumed. The rollback journal
5//! (`journal_mode=DELETE`) keeps the file complete between transactions, so
6//! a copy of it is a usable backup.
7
8use 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
61/// Version 2. Runs outlive a project's removal from the scan: their worktrees
62/// still exist and must be published or rejected.
63const 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
93/// Version 3. `deps` is `NULL` until measured, and keeps its value across
94/// scans that do not measure it.
95const 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
107/// Version 4. `pma/` branches that no open run owned at the last scan.
108const LEFTOVER: &str = "
109ALTER TABLE projects ADD COLUMN leftover INTEGER NOT NULL DEFAULT 0;
110";
111
112/// Version 5. Runs may be `pr-open`, which an older `pma` would read as
113/// `failed`. Runs shipped as pull requests before it return to `pr-open`, so
114/// the next settle checks whether they were merged.
115const PR_OPEN: &str = "
116UPDATE runs SET state = 'pr-open' WHERE state = 'shipped' AND outcome LIKE 'https://%/pull/%';
117";
118
119/// Version 6. `runs` holds one mutable row per task, so a rework overwrites
120/// the previous attempt's summary, cost, verify result and start time. An
121/// attempt is now its own append-only row, and `runs` keeps the lifecycle
122/// summary plus a timestamp per transition. Existing runs contribute their
123/// last attempt, whose outcome was not recorded and stays null.
124const 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
156/// Version 7. The decision a dispatch made, snapshotted on the run and never
157/// updated. A rescan or a reworded item changes the scanned task, and the
158/// project's tier and the budgets change under `pma config`, so none of them
159/// can be read back later from the task. Runs dispatched before this version
160/// have no snapshot and stay null.
161const 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
170/// Version 8. One verify run at the head proves the tree is green now. The
171/// same command at the base separates three cases: already broken, broken by
172/// the agent, fixed by the agent. The result is cached because every task of
173/// a project in one batch shares a base. Timeout is part of the key: a pass
174/// under a longer limit says nothing about a shorter one.
175const 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
190/// Version 9. The paths a run changed, as a JSON array. Stored rather than
191/// the violations derived from them, so a later change to the class rules
192/// re-reads the evidence instead of trusting a verdict recorded under rules
193/// nobody can name any more. Null means the paths could not be enumerated,
194/// which `scope_error` explains; an empty array means the run changed
195/// nothing.
196const CHANGED_PATHS: &str = "
197ALTER TABLE runs ADD COLUMN changed_paths TEXT;
198ALTER TABLE runs ADD COLUMN scope_error TEXT;
199";
200
201/// Version 10. Attempts consumed per task revision, so `--auto` stops
202/// choosing a task an agent cannot close. Keyed by the normalised task text
203/// rather than by the run, because a counter on a run resets when the run
204/// ends, and keyed by text rather than by a permanent id, because a reworded
205/// task is a different specification and a recurring `ci` incident names its
206/// failing workflows.
207const 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
217/// Version 11. A worker is a record, not a code path. `config` is a flat
218/// table of scalars with a closed key list, and an agent record is an
219/// open-ended name holding a list, so it gets its own table rather than a
220/// dotted key. `claude` is seeded as one entry with exactly the behaviour it
221/// had when it was compiled in.
222const 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
238/// Version 12. Drops the settings phase 3 retired. A stored row for an
239/// unknown key makes every command fail, so retiring one is a migration, not
240/// only a code change. The names stay in `config::RETIRED`, which explains
241/// them to `pma config`.
242const 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
249/// Version 13. The complexity estimate and the features it read, on the run
250/// that used them, with the version of the rule that produced it. Routing
251/// matches on the value, so replaying a decision means re-running the rule
252/// named here rather than today's.
253const 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
259/// Version 14. Routing policy as a versioned artifact with provenance.
260/// Editing a revision and putting it into effect are separate acts: a
261/// revision is a draft until someone activates it, and the run records the
262/// revision it was dispatched under, so a later activation does not rewrite
263/// what a past dispatch decided.
264const 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
279/// Version 15. What an approval was given for. Ship rebases, edits
280/// `TODO.md`, amends and publishes without rerunning anything, so an
281/// approval that names no tree can publish content no check ever saw. The
282/// base, the verify command and the scope are already immutable on the run;
283/// the tree and the approver are what was missing.
284const 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
290/// Version 16. One task definition across many repositories. Membership is
291/// stored rather than recomputed, so a rescan cannot change the set under a
292/// campaign that is already running, and a restart can tell which members
293/// already have a run instead of opening a second pull request for them.
294const 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
310/// Version 35. The `settings.toml` text the configuration tables were last
311/// filled from, so an unchanged file is not applied again. `IF NOT EXISTS`
312/// because tests replay migrations over a store that has it.
313const SETTINGS_SYNC: &str = "
314CREATE TABLE IF NOT EXISTS settings_sync (text TEXT NOT NULL);
315";
316
317/// Version 34. A workflow node that is not an `edit` changes no files, so a
318/// run of one that finished is `done`, not `ready` for a review it cannot have.
319const READ_RUNS_DONE: &str = "
320UPDATE runs SET state = 'done'
321WHERE state = 'ready' AND branch = '' AND workflow_instance IS NOT NULL;
322";
323
324/// Version 33. The command a job that stopped at a workflow plan would run
325/// to approve it, as a JSON array. And the `claude` worker streams its
326/// events, unless its row was edited: `pma watch` reads them.
327const 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
335/// Version 32. A `pma` command queued to run detached, with what became of
336/// it. `args` is a JSON array, run as `pma <args>` in `cwd`.
337const 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
351/// Version 31. A rejection the reviewer put down to infrastructure: the
352/// check or the machine failed, not the agent. Such a run is in no share.
353const INFRA_REJECTION: &str = "
354ALTER TABLE runs ADD COLUMN infra INTEGER NOT NULL DEFAULT 0;
355";
356
357/// Version 30. Whether verify ran confined is part of the base cache's key. A
358/// base measured unconfined says nothing about a confined head: a check that
359/// fails only under the sandbox would read as the agent's regression. The
360/// cached rows are dropped and measured again.
361const 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
376/// Version 29. The tree a reviewer was last shown, which an approval must
377/// match: what the agent left, or what `pma review <id>` printed since.
378const REVIEWED_TREE: &str = "
379ALTER TABLE runs ADD COLUMN reviewed_tree TEXT;
380";
381
382/// Version 28. `shipped` named two different things: a push to the default
383/// branch, and a pull request merged. `pma push` and `pma pr` are separate
384/// commands now, and a run's state says which happened, in GitHub's words. A
385/// pull request closed without merging is `closed`, not `rejected`, which is
386/// kept for a run rejected in review and never published. The `publish`
387/// settings are dropped: the command says where a run goes.
388const 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
395/// Version 27. Drops the health score's weights. `pma status` sorts by the
396/// worst state a project is in, which needs no weight, and a stored row for a
397/// key nothing reads is one `pma config` would have to explain forever.
398const RETIRE_WEIGHTS: &str = "
399DELETE FROM config WHERE key LIKE 'weights.%';
400";
401
402/// Version 26. What makes git run a program in a run's repository, or push
403/// elsewhere, as it stood at dispatch: hooks and the settings that name a
404/// command or rewrite a URL. Publishing refuses when it has changed, since neither
405/// is in the diff a reviewer reads.
406const GIT_SNAPSHOT: &str = "
407ALTER TABLE runs ADD COLUMN git_snapshot TEXT;
408";
409
410/// Version 25. Seeds `sanduk`, the worker that runs the same agent in a
411/// disposable container. No schema: a worker is a record, and this one exists
412/// so the flags that have to be right are right without being typed. An
413/// existing row of that name is left alone, as the other templates are.
414const SANDUK_WORKER: &str = "-- no schema change; the seed is in Rust";
415
416/// Version 24. Retires a worker's own default model in favour of presets. A
417/// preset says the same thing and more -- a worker, a model and the arguments
418/// that configure it -- and two ways to name a default model is one too many.
419/// Each worker that named one gets a preset of its own name, which becomes the
420/// default when that worker was the configured one. The column is dropped
421/// rather than left unread: a retirement that leaves the old place writable is
422/// not one.
423const 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
435/// Version 23. Presets: a named worker, model and configuration. Effort is not
436/// a field, because what expresses it differs per agent -- `--thinking` for
437/// one, `--variant` for another -- so a preset carries arguments and the worker
438/// record says where they go. A preset takes part in no matching, so it
439/// competes with no route; what it buys is a name for a combination worth
440/// returning to, and a list of the ones in use.
441///
442/// The name and the arguments are recorded on each run, because a replay must
443/// read back what ran rather than what the preset says today.
444const 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
456/// Version 22. A worker's default model. A model name is only meaningful to
457/// the worker that understands it, so storing it beside the choice of worker
458/// meant naming another worker had to drop it. The value the `model` setting
459/// held is carried onto the configured worker, and the setting is retired.
460const 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
468/// Version 21. Environment a worker needs, so reaching OpenAI, OpenRouter or
469/// any OpenAI-compatible endpoint is a record rather than a code path: a base
470/// URL or a config location belongs to the worker, and the key comes from the
471/// session. Seeds the two other workers `pma` ships a template for, without
472/// replacing one the user has already edited.
473const WORKER_ENV: &str = "
474ALTER TABLE agents ADD COLUMN env TEXT NOT NULL DEFAULT '{}';
475";
476
477/// Version 20. A workflow is a typed function over bags of units, stored as a
478/// revision and activated like a routing policy. Units are immutable and carry
479/// their lineage, so a lap, an annotation and a decomposition each leave a
480/// readable chain; a node's bag is derived from the units it wrote and the
481/// frontier from the moves recorded, so nothing holds a cursor a crash could
482/// lose. The worst case is stored with the revision, because what `activate`
483/// weighs against the budget must be what `propose` printed.
484const 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
545/// Version 19. A project's private tags, for grouping and for selecting a set
546/// to act on. A project carries several, so the tags cannot live in a column
547/// on its row. Kept out of `projects` for a second reason: a tag is the user's
548/// own, and every column there is overwritten by the next scan.
549const 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
558/// Version 18. When a full scan stopped finding a project under any root.
559/// The row used to be deleted, which also dropped its tier and every task's
560/// `first_seen` -- neither of which a later scan can rebuild, so a project
561/// that left a root and came back lost the age of all its work.
562const ABSENCE: &str = "
563ALTER TABLE projects ADD COLUMN absent_since INTEGER;
564";
565
566/// Version 17. `owner/name` on GitHub, recorded by the scan. `sync` derived it
567/// per invocation from the origin URL and no other command could reach it, so
568/// an item's `gh:N` was a number with no repository to resolve it against.
569const SLUG: &str = "
570ALTER TABLE projects ADD COLUMN slug TEXT;
571";
572
573pub struct Store {
574    conn: Connection,
575    /// `settings.toml`, beside the database. The configuration tables mirror
576    /// it; every change to them goes through it first.
577    settings: PathBuf,
578}
579
580#[derive(Debug, Clone, PartialEq)]
581pub struct ProjectRow {
582    pub name: String,
583    pub path: PathBuf,
584    /// `owner/name` on GitHub; `None` without a GitHub origin, or until the
585    /// project is scanned again.
586    pub slug: Option<String>,
587    /// When a full scan stopped finding the project under any root; `None`
588    /// while it is present. `path` is then where it was last seen.
589    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    /// `pma/` branches that no open run owned at the last scan.
596    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    /// Outdated dependencies at the last measurement; `None` if never measured.
602    pub deps: Option<i64>,
603    /// One `ecosystem: name current -> latest` per line.
604    pub deps_detail: String,
605    pub deps_at: Option<i64>,
606}
607
608/// What a project's record consists of, for `pma project forget`.
609#[derive(Debug, Clone, PartialEq)]
610pub struct Footprint {
611    pub tasks: i64,
612    pub tags: i64,
613    pub campaigns: i64,
614    /// Runs are kept, so this is reported rather than deleted.
615    pub runs: i64,
616    /// Runs that are not final; each may own a worktree.
617    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
635/// Exclusive right to run agents and change worktrees. `flock` releases it
636/// when the process exits, so a crash leaves no stale lock.
637pub struct Session {
638    _file: File,
639}
640
641impl Session {
642    /// `None` when another process holds the lock.
643    pub fn try_acquire(dir: &Path) -> Result<Option<Session>> {
644        Session::try_lock(dir, "session.lock")
645    }
646
647    /// The same exclusive lock on another file in `dir`, such as the job
648    /// runner's.
649    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    /// Like `try_acquire`, but waits out a brief hold, such as `pma review`
669    /// checking for a live session. A lock still held is an error naming its
670    /// holder.
671    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
697/// `~/.config/pma`, or `PMA_HOME` when set.
698pub 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
706/// Where worktrees, artifacts and logs go: `PMA_STATE`, else `$PMA_HOME/state`
707/// when `PMA_HOME` is set, else `$XDG_STATE_HOME/pma` or `~/.local/state/pma`.
708/// Kept apart from the database, whose directory holds nothing else.
709pub 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        // Other commands may write while a session runs agents.
738        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                // Not replacing an edited row: a template is a starting
787                // point, and the store is the record.
788                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    /// Fills the configuration tables from `settings.toml` when its text
829    /// changed since they were last filled. With no file, writes one from the
830    /// tables: the first run after they held the configuration. A migration
831    /// that changes a configuration row must change the file too, or the next
832    /// edit of the file undoes it.
833    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    /// Replaces the configuration tables with what the file says. A tier for a
866    /// project no scan has recorded waits for the scan.
867    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    /// Edits `settings.toml` with `f`, then refills the tables from it.
920    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    /// Re-applies the file's tiers and tags, for projects a scan just added.
928    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    /// Returns whether a stored value was removed.
946    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    /// Returns false when the root was already present.
959    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    /// Records the project first if no scan has, so the tier has a row.
970    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    /// Records scan results. Each task keeps the `first_seen` it had in the
1046    /// previous scan. With `full`, projects not in `facts` are deleted and
1047    /// returned with the tier they had.
1048    /// Records the scan. A full scan marks every project it did not find as
1049    /// absent and returns their names; the rows stay, since a project that
1050    /// leaves a root is usually a move, not a deletion.
1051    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            // An unreadable file says nothing about the tasks; the last good
1116            // scan's rows and their ages stand.
1117            if f.todo_unread {
1118                continue;
1119            }
1120            // By key, else by text: `sync` adding `gh:N` changes an item's key
1121            // but not its text, and an `id:` keeps an item's key when its text
1122            // is reworded. Either way the item keeps its age: when it was first
1123            // seen, and when its text first appeared in git, which a reworded
1124            // text no longer finds in the history.
1125            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        // A project this scan added may have a tier or tags in the file.
1197        self.reapply_settings()?;
1198        Ok(absent)
1199    }
1200
1201    /// What forgetting a project would delete, and what would hold it back.
1202    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    /// Deletes a project's record and everything keyed to it but its runs,
1221    /// whose worktrees may still exist and which the design keeps beyond a
1222    /// project's removal. Caller checks that nothing holds it back.
1223    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            // `tasks` follows by ON DELETE CASCADE.
1231            "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    /// Every `(project, tag)` pair, ordered by project then tag.
1240    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    /// Returns false when the project already had the tag.
1249    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    /// Returns false when the project did not have the tag.
1254    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    /// The projects carrying any of `tags`, in name order. A project in
1259    /// several of them appears once.
1260    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    /// The newest `limit` jobs, newest first.
1299    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    /// Marks the oldest queued job running and returns it.
1319    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    /// Records what would approve the plan a job stopped at.
1339    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    /// Marks an awaiting job approved. False if it was not awaiting.
1348    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    /// Marks every job awaiting `plan` approved, once that plan has run by
1356    /// any route. Plan ids are hex, so the quoted match is exact.
1357    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    /// Records how a job ended. A job stopped while it ran stays `stopped`.
1373    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    /// Returns the state the job was in. Only a queued or running job stops.
1386    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    /// Jobs left `running` by a runner that exited. Called under the runner
1398    /// lock, so no live runner holds them.
1399    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    /// Returns false when no note has the id.
1423    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    /// Returns false when no note has the id.
1431    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    /// Assigns the run's id and stamps `dispatched_at`, in the row and in
1436    /// `run`.
1437    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    /// Writes every mutable column of the run with `run.id`.
1496    pub fn update_run(&self, run: &Run) -> Result<()> {
1497        self.write_run(run, None)
1498    }
1499
1500    /// `update_run`, only while the row is still in `read`, the state the copy
1501    /// was read in. For a command that holds no session lock, so its stale
1502    /// copy cannot undo a publish that ran meanwhile.
1503    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            // `verify` is frozen at dispatch: a command re-detected at rework
1510            // could differ from the one the base was checked with.
1511            "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    /// Adds review time in one column, so a command that holds no session
1561    /// lock writes nothing else.
1562    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    /// Records the tree `pma review <id>` showed. One column, so a reading
1571    /// command does not write back a copy of the whole run.
1572    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    /// Creates a campaign with a fixed membership. The set does not change
1688    /// afterwards: a rescan must not move work under a campaign in flight.
1689    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    /// Records which run took a member, so a restart does not dispatch it
1754    /// again and open a second pull request.
1755    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    /// Returns false when no campaign has the name.
1764    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    /// Stores a policy revision as a draft and returns its number. Storing
1774    /// is not activating: nothing routes by it until someone says so.
1775    pub fn add_route_revision(&self, document: &str, by: &str) -> Result<i64> {
1776        // Refused here, so an unreadable document never reaches the table.
1777        // Stored as `pma` read it, with every default written out, so two
1778        // revisions diff by what they mean rather than by how they were
1779        // typed, and `pma route show` states the route names a reader will
1780        // see in a replay.
1781        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    /// Puts a revision into effect, in shadow or for real, and retires any
1796    /// other. `by` is recorded: raising autonomy is somebody's decision.
1797    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    /// Every revision, oldest first.
1816    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    /// The revision in effect, if any.
1837    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    /// Stores a document as a draft. Refused here, so an unreadable one never
1845    /// reaches the table, and stored normalised, so two revisions diff by what
1846    /// they mean rather than by how they were typed -- or by whether a script
1847    /// or a person wrote them. `source` keeps the script for provenance.
1848    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    /// Puts a revision into effect and retires any other. `by` is recorded:
1880    /// what a workflow may spend is somebody's decision.
1881    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    /// Every revision, oldest first.
1899    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    /// The revision in effect, if any.
1922    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    /// One revision by number. What an instance resumes under, which is the
1930    /// revision it recorded rather than whichever is in effect now.
1931    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    /// Starts an instance: the frozen argument bag and what it was run over.
1939    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    /// Runs `work` in one transaction, committing only where it succeeds.
1975    ///
1976    /// A workflow node writes several rows per unit -- the units it produced,
1977    /// the moves that route them, the marker saying it is done with its
1978    /// input -- and a failure part way through would leave a unit routed by a
1979    /// node that had not finished with it. The boundary is one unit, not one
1980    /// node: a sink writes a file, and a unit whose file is written and whose
1981    /// move is not would have its sink run twice.
1982    pub fn atomically<T>(&self, work: impl FnOnce() -> Result<T>) -> Result<T> {
1983        // Immediate takes the write lock first, so `busy_timeout` waits for
1984        // it. A deferred read upgraded to a write fails at once when another
1985        // writer holds the lock.
1986        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    /// One instance by id.
1996    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    /// Records why an instance cannot go on, without closing it. A pass that
2004    /// hit a cap has work recorded and a reason it stopped, and neither is a
2005    /// finished workflow.
2006    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    /// Records new arguments on an instance and clears the outcome that
2015    /// stopped it, so a raised cap resumes it.
2016    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    /// Units are immutable, so this only ever inserts.
2033    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    /// One row per unit and edge: which way it went, or the guard that
2077    /// refused it. A unit that stopped is therefore answerable for.
2078    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    /// Every move recorded, as `(unit, edge, taken)`.
2102    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    /// Verdicts by unit and check name.
2134    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    /// Names a worker, a model and the arguments that configure it. Replaces
2143    /// one of the same name.
2144    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    /// Returns false when no preset has the name.
2174    pub fn remove_preset(&self, name: &str) -> Result<bool> {
2175        self.change_settings(|doc| crate::settings::remove_preset(doc, name))
2176    }
2177
2178    /// Every worker, by name.
2179    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    /// Returns false when no agent has the name.
2216    pub fn remove_agent(&self, name: &str) -> Result<bool> {
2217        self.change_settings(|doc| crate::settings::remove_agent(doc, name))
2218    }
2219
2220    /// Decided runs in a project and how many were accepted. An open pull
2221    /// request is in neither: no one has judged it.
2222    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    /// Moves the runs still holding a task onto its new key, once `pma` has
2240    /// written an id into its line: those keyed by its old key or by its text.
2241    /// A finished run keeps the key it was dispatched under.
2242    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    /// Attempts consumed against a task revision, across every run of it.
2259    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            // No row is no attempt; any other error is an error, since the
2268            // limit it guards would otherwise be lifted.
2269            .optional()?
2270            .unwrap_or(0))
2271    }
2272
2273    /// Records one consumed attempt and returns the new total.
2274    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    /// Takes back `n` consumed attempts, never below none.
2286    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    /// Returns false when the revision had consumed none.
2296    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    /// A base verification of this exact repository, commit, command,
2304    /// timeout and sandbox, or `None` when it has not been measured.
2305    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    /// Appends an attempt. `n` is assigned here, so two reworks cannot race
2355    /// to the same number.
2356    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    /// Every attempt, oldest first. `run` limits it to one run.
2382    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    /// Marks runs left queued or running by a session that ended as failed.
2411    /// Only a `Session` holder runs agents, so while it is held none is live.
2412    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/// A `pma` command queued to run detached.
2429#[derive(Debug, Clone, PartialEq)]
2430pub struct Job {
2431    pub id: i64,
2432    pub args: Vec<String>,
2433    pub cwd: PathBuf,
2434    /// `queued`, `running`, `done`, `failed` or `stopped`.
2435    pub state: String,
2436    pub created_at: i64,
2437    pub started_at: Option<i64>,
2438    pub finished_at: Option<i64>,
2439    /// The process group leader of a running job.
2440    pub pid: Option<i64>,
2441    pub exit_code: Option<i64>,
2442    /// For a job that stopped at a workflow plan: the arguments that approve
2443    /// it, which `pma job approve` queues.
2444    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    /// `pma pr` opened a pull request that is neither merged nor closed.
2485    PrOpen,
2486    /// Rejected in `pma review`: never published.
2487    Rejected,
2488    /// `pma push` pushed it to the default branch.
2489    Pushed,
2490    /// Its pull request was merged.
2491    Merged,
2492    /// Its pull request was closed without merging.
2493    Closed,
2494    /// Written into its clone as unstaged changes, for the developer to
2495    /// stage and commit, or discard.
2496    Applied,
2497    /// A workflow node that is not an `edit` finished; its output became
2498    /// units. It changed no files, so there is nothing to review.
2499    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    /// A run in a final state no longer holds its task or its worktree.
2530    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    /// Whether its change reached the default branch: pushed, or merged.
2543    pub fn landed(self) -> bool {
2544        matches!(self, RunState::Pushed | RunState::Merged)
2545    }
2546}
2547
2548/// One task dispatched to an agent, from its worktree to publishing. `Default`
2549/// is every field empty, which a workflow node fills differently from a
2550/// dispatch: it has no branch to push and no item to tick.
2551#[derive(Debug, Clone, PartialEq, Default)]
2552pub struct Run {
2553    pub id: i64,
2554    pub project: String,
2555    /// The item's key, or a signal: `ci` or `deps`.
2556    pub task_key: String,
2557    pub text: String,
2558    pub gh: Option<i64>,
2559    /// Quadrant at dispatch; `None` for an untiered project.
2560    pub quadrant: Option<String>,
2561    pub agent: String,
2562    /// The user's clone the worktree belongs to.
2563    pub repo: PathBuf,
2564    pub branch: String,
2565    pub worktree: PathBuf,
2566    pub default_branch: String,
2567    /// Commit the worktree started from.
2568    pub base: String,
2569    pub prompt: String,
2570    pub state: RunState,
2571    /// Reviewer feedback for the latest rework.
2572    pub feedback: Option<String>,
2573    /// Agent and verify time, summed over reworks.
2574    pub seconds: Option<i64>,
2575    /// Summed over reworks; `None` when the agent reports no cost.
2576    pub cost_usd: Option<f64>,
2577    pub summary: Option<String>,
2578    /// Commits the agent made beyond `base`; it was told to make none.
2579    pub commits: Option<i64>,
2580    pub diffstat: Option<String>,
2581    /// The verify command, or `None` when there is none.
2582    pub verify: Option<String>,
2583    pub verify_ok: Option<bool>,
2584    pub error: Option<String>,
2585    /// What publishing did: a pushed commit, or a pull request URL.
2586    pub outcome: Option<String>,
2587    /// When the run row was created. Never updated, so a per-day count and a
2588    /// digest window have a stable date. Per-attempt start times are in
2589    /// `attempts`.
2590    pub dispatched_at: Option<i64>,
2591    /// When the run last became ready. A rework moves it forward.
2592    pub ready_at: Option<i64>,
2593    /// The first approval or rejection. A pull request closed later does not
2594    /// replace the approval that published it.
2595    pub decided_at: Option<i64>,
2596    /// When `pma` published: the push, or the pull request opening. A merge
2597    /// days later does not move it.
2598    pub published_at: Option<i64>,
2599    /// Review time the user reported with `pma review --minutes`, summed over
2600    /// reviews of this run.
2601    pub review_seconds: Option<i64>,
2602    /// The class predicted at dispatch. `None` for a run dispatched before
2603    /// classes existed.
2604    pub class: Option<Class>,
2605    /// Globs the class allowed, resolved at dispatch. Empty means the class
2606    /// stated no bound; the privileged paths still apply.
2607    pub scope: Vec<String>,
2608    /// The project's tier at dispatch. Tiers change, and routes match on them.
2609    pub tier: Option<u8>,
2610    /// The item's description lines as dispatched.
2611    pub description: Option<String>,
2612    /// `agent_budget` at dispatch, so an attempt's cost can be read against
2613    /// the limit that applied to it.
2614    pub agent_budget: Option<f64>,
2615    pub timeout_minutes: Option<i64>,
2616    /// `verify` at the base commit, before the agent ran. `None` when there
2617    /// is no command, or when the run predates the check; `Some(false)` also
2618    /// covers a base run that timed out.
2619    pub verify_base_ok: Option<bool>,
2620    pub verify_base_seconds: Option<i64>,
2621    /// Every path the run changed against its base, both sides of a rename
2622    /// included. `None` when they could not be enumerated: an empty list
2623    /// would read as a clean run.
2624    pub changed_paths: Option<Vec<String>>,
2625    /// Why the paths could not be enumerated.
2626    pub scope_error: Option<String>,
2627    /// The model chosen at dispatch, or `None` for the worker's own default.
2628    pub model: Option<String>,
2629    /// 1 to 5, from the rule named in `estimator`.
2630    pub complexity: Option<i64>,
2631    /// What that rule read.
2632    pub features: Option<Features>,
2633    pub estimator: Option<String>,
2634    /// The policy revision in effect at dispatch, and the route it matched.
2635    /// A later activation does not change either.
2636    pub route_revision: Option<i64>,
2637    pub route: Option<String>,
2638    pub approval: Option<Approval>,
2639    /// The tree the approver saw, as `git write-tree` names it. Publishing refuses
2640    /// to publish a worktree that no longer matches.
2641    pub approved_tree: Option<String>,
2642    /// The commit the worktree was on. Each publish attempt starts again from
2643    /// it and `approved_tree`.
2644    pub approved_head: Option<String>,
2645    pub approved_by: Option<String>,
2646    /// The workflow instance that dispatched this run, or `None` for a task
2647    /// dispatched on its own.
2648    pub workflow_instance: Option<i64>,
2649    /// The qualified node, which a route matches on and a report groups by.
2650    pub node: Option<String>,
2651    /// The unit that drove it.
2652    pub unit: Option<String>,
2653    /// Laps the unit had completed. 0 outside a workflow.
2654    pub lap: i64,
2655    /// The preset this run was dispatched under, where one named it.
2656    pub preset: Option<String>,
2657    /// Arguments the preset added, recorded so a replay reads back what ran.
2658    pub extra_args: Vec<String>,
2659    /// `dispatch::git_snapshot` at dispatch. `None` for a run from before it
2660    /// was taken.
2661    pub git_snapshot: Option<String>,
2662    /// The tree the reviewer was last shown: what the agent left, or what
2663    /// `pma review <id>` printed since. An approval must match it. `None` for
2664    /// a run from before it was recorded.
2665    pub reviewed_tree: Option<String>,
2666    /// Rejected for a fault of the check or the machine, not the agent.
2667    pub infra: bool,
2668}
2669
2670impl Run {
2671    /// A run with every field empty, for tests in this crate to override.
2672    #[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    /// Moves the run to `state` and stamps the transition. Call it instead of
2737    /// assigning `state`, so no transition goes unrecorded.
2738    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/// One agent invocation and the verification that followed it. Rows are
2755/// appended and never updated, so a rework cannot overwrite what the previous
2756/// attempt did.
2757#[derive(Debug, Clone, PartialEq)]
2758pub struct Attempt {
2759    pub id: i64,
2760    pub run_id: i64,
2761    /// 1 for the first attempt, then one per rework.
2762    pub n: i64,
2763    pub agent: String,
2764    pub prompt: String,
2765    /// The reviewer feedback this attempt was given, if any.
2766    pub feedback: Option<String>,
2767    pub started_at: i64,
2768    /// Agent and verify time for this attempt alone.
2769    pub seconds: Option<i64>,
2770    /// `None` when the agent reports no cost. Never zero for unknown.
2771    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    /// The state the run reached: `ready` or `failed`. Null for the attempt
2777    /// reconstructed from a run that predates this table.
2778    pub outcome: Option<String>,
2779    /// The model this attempt ran, or `None` for the worker's own default.
2780    pub model: Option<String>,
2781}
2782
2783/// One task definition applied across a fixed set of repositories.
2784#[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    /// Project, and the run that took it, if any.
2792    pub members: Vec<(String, Option<i64>)>,
2793}
2794
2795/// One stored policy document and who put it where it is.
2796#[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    /// Computed and recorded on each run, but not applied.
2805    pub shadow: bool,
2806}
2807
2808/// A named worker, model and configuration. `model` unset means the worker's
2809/// own default, and `args` are inserted where its record puts `{extra}`.
2810#[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    /// The arguments the pass was given, as JSON, so a replay reads the same
2824    /// instantiation rather than the document's defaults.
2825    pub args: String,
2826    /// What the argument bag was taken from.
2827    pub target: String,
2828    pub started_at: i64,
2829    pub finished_at: Option<i64>,
2830    pub outcome: Option<String>,
2831}
2832
2833/// A unit as stored: immutable, carrying its lineage and its data.
2834#[derive(Debug, Clone, PartialEq)]
2835pub struct WorkflowUnit {
2836    pub id: String,
2837    pub ty: String,
2838    /// The node that wrote it; `@input` for a root unit.
2839    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    /// The declared fields, as JSON.
2846    pub data: String,
2847}
2848
2849#[derive(Debug, Clone, PartialEq)]
2850pub struct WorkflowRevision {
2851    pub revision: i64,
2852    /// The document as `pma` read it.
2853    pub document: String,
2854    /// The script that built it, where one did.
2855    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    /// Over one unit of input, so what a pass may spend is this times the
2861    /// argument bag.
2862    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    /// `shipped` said two things. A run with a pull request became `merged`,
2971    /// a push became `pushed`, and a pull request closed without merging is
2972    /// `closed`, where it was `rejected`. The `publish` settings go.
2973    #[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    /// Migration 6 turns each existing run's single recorded attempt into an
3115    /// `attempts` row, so upgrading loses no history, and moves the run's
3116    /// start time to `dispatched_at`. Migration 7 adds the decision snapshot,
3117    /// which an existing run cannot supply.
3118    #[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        // Never started, so it contributes no attempt and no dispatch time.
3145        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        // Migration 7 cannot reconstruct a decision that was never recorded.
3158        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    /// Upgrading seeds `claude` with exactly the behaviour it had when it
3165    /// was compiled in, so a database made before the adapter keeps working.
3166    #[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        // Four templates, alphabetically: the one that was compiled in, the
3171        // two installed separately, and the same agent in a container.
3172        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        // A template is a starting point: an edited row survives reopening,
3222        // which is what seeding only an absent name buys.
3223        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    /// Schema 33 moves a `claude` row still on the shipped template to
3236    /// `stream-json`, which `pma watch` reads, and leaves an edited row alone.
3237    #[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    /// Only a workflow run with no branch, which a non-`edit` node made, is
3266    /// moved from `ready` to `done`; a dispatched run still waits for review.
3267    #[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    /// A worker that named a model before presets existed keeps that pairing:
3294    /// it becomes a preset of the worker's name, and the default when that
3295    /// worker was the configured one.
3296    #[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        // A store as schema 23 left it: the column exists and holds a model.
3301        {
3302            let store = Store::open(&path).unwrap();
3303            // First: it refills the tables from settings.toml, which knows no
3304            // `model` column.
3305            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    /// The counter survives the run that raised it, so rejecting a task and
3345    /// dispatching it again does not start from zero. Rewording it does.
3346    #[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        // A reworded task, another project, and a `ci` incident naming other
3356        // workflows are each their own revision.
3357        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        // A refund never takes the counter below none.
3366        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    /// The cache answers only for the exact repository, commit, command,
3379    /// timeout and sandbox it measured. A pass under a longer limit says
3380    /// nothing about a shorter one, a confined result nothing about an
3381    /// unconfined one, and a changed command nothing at all.
3382    #[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        // A base that could not be measured is unknown, not failing.
3417        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    /// A copy read before a publish must not write its older state over it.
3430    #[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    /// A transaction that reads and then writes waits for another writer.
3459    /// Deferred, the read lock it took could not be upgraded, and SQLite
3460    /// fails such an upgrade at once rather than waiting.
3461    #[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    /// A rejection put down to infrastructure says nothing about the agent, so
3494    /// the project's record that routing reads leaves it out.
3495    #[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    /// A rework must not overwrite what the previous attempt did.
3517    #[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    /// Each transition stamps its own time, and neither a merge nor a closure
3558    /// moves the time the run was published.
3559    #[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        // The command the base was checked with cannot change under the run.
3687        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        // A branch is leftover unless an open run of the same project owns it.
3710        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    /// Unlike `deps`, the slug is read on every scan, so the last scan wins
3753    /// and a project whose GitHub origin is gone loses it.
3754    #[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    /// Forgetting clears the record but not the runs, whose worktrees may
3774    /// still exist.
3775    #[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    /// The first open after the tables held the configuration writes them to
3860    /// `settings.toml`; from then on the file is what the tables say.
3861    #[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        // A hand edit, applied on the next open; a tier waits for its scan.
3893        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        // An invalid file stops the store from opening, naming the key.
3911        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    /// A project left unread keeps everything the last scan recorded, and is
4011    /// not absent: only its scan error changes.
4012    #[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    /// A file that could not be read is not an empty file: the tasks and their
4031    /// ages stand until a scan can read it again.
4032    #[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    /// `sync` writes `gh:N` into an item, which changes its key from its text
4060    /// to the issue number. Its age is the item's, not the key's.
4061    #[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        // Reported once, not on every later scan.
4124        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        // Finding it again makes it present, without a second first_seen.
4133        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}