sloop/run_store/
mod.rs

1//! Concrete SQLite storage for run supervision state.
2//!
3//! `runs` owns run lifecycle records and their attached notes and activity
4//! events, `evidence` owns verdict and stage facts, and `limits` owns
5//! cooldown and budget state. Scheduler state and shared ID counters stay on
6//! [`RunStore`] because they apply across those families.
7//!
8//! This module must never import the work-state sibling; run persistence is an
9//! independent boundary composed with work state by the daemon.
10
11pub(crate) mod evidence;
12pub(crate) mod limits;
13pub(crate) mod runs;
14
15pub use evidence::{
16    EvidenceRecord, OutputStallEvidence, PanelReportRecord, StagePhase, StageRecord,
17};
18pub use limits::{CooldownRecord, CooldownUpdate};
19pub use runs::{ActiveRun, EventRecord, ProjectNote, RunRecord, RunState, RunTimeline};
20pub use runs::{AdmittedRun, RunAdmission};
21pub(crate) use runs::{NeedsReviewBranch, RecoverableRun, WorktreeCleanupCandidate};
22
23use rusqlite::{OptionalExtension, TransactionBehavior, params};
24
25use crate::db::{Db, StoreError};
26use crate::domain::ticket::TicketState;
27use crate::domain::work::{OwnerId, WorkOutcome};
28use crate::ids::NOTE_ID_PREFIX;
29use crate::outcome::Outcome;
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct ProjectCommitEvidence {
33    pub run_id: String,
34    pub ticket_id: String,
35    pub data_json: String,
36}
37
38/// Whether the caller won the `running` to `driving` transition and with it
39/// ownership of the agent's exit evidence and of the rest of the walk.
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub enum ExitClaim {
42    Claimed,
43    AlreadyClaimed { state: String },
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub enum Start {
48    Granted,
49    Denied(StartDenial),
50}
51
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub enum StartDenial {
54    /// The run left `claimed` before its process was recorded.
55    NotClaimed { state: Option<String> },
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub enum Exit {
60    Granted,
61    Denied(ExitDenial),
62}
63
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub enum ExitDenial {
66    /// Another path checkpointed this exit first and owns the walk.
67    AlreadyClaimed { state: String },
68}
69
70/// Facts about a launched agent process, recorded when the run turns running.
71pub struct RunStart<'a> {
72    pub run_id: &'a str,
73    pub branch: &'a str,
74    pub worktree_path: &'a str,
75    pub pid: u32,
76    pub pid_start_time: Option<i64>,
77    pub process_group_id: u32,
78    pub worker_token: &'a str,
79    pub worker_socket_path: &'a str,
80}
81
82/// Facts about an agent exit at the checkpoint that returns the run to its
83/// driver.
84pub struct RunExit<'a> {
85    pub run_id: &'a str,
86    /// Which execution of the run's primary agent stage exited. A backward
87    /// edge can re-enter that stage, and the checkpoint has to say which
88    /// execution it speaks for or a resumed run would resolve a re-entry from
89    /// an earlier attempt's exit.
90    pub attempt: u32,
91    pub exit_code: Option<i32>,
92    pub capture_complete: bool,
93    pub commits_json: &'a str,
94    pub vendor_error: Option<&'a crate::vendor::VendorErrorMatch>,
95    pub cooldown_until_ms: Option<i64>,
96}
97
98#[derive(Debug, Clone, PartialEq, Eq)]
99pub struct RecordedOutcome {
100    pub work: WorkOutcome,
101    pub not_before_ms: Option<i64>,
102}
103
104pub struct RunStore {
105    db: Db,
106}
107
108impl RunStore {
109    pub fn from_db(db: Db) -> Self {
110        Self { db }
111    }
112
113    #[cfg(test)]
114    pub(crate) fn db(&self) -> Db {
115        self.db.clone()
116    }
117
118    fn write<T>(
119        &self,
120        behavior: TransactionBehavior,
121        operation: impl FnOnce(&rusqlite::Transaction<'_>) -> rusqlite::Result<T>,
122    ) -> Result<T, StoreError> {
123        let mut connection = self.db.lock();
124        let transaction = connection.transaction_with_behavior(behavior)?;
125        let result = operation(&transaction)?;
126        transaction.commit()?;
127        Ok(result)
128    }
129
130    pub(crate) fn next_note_ordinal(&self) -> Result<i64, StoreError> {
131        self.reserve_ordinal("note", "notes", NOTE_ID_PREFIX)
132    }
133
134    fn reserve_ordinal(&self, kind: &str, table: &str, prefix: &str) -> Result<i64, StoreError> {
135        self.write(TransactionBehavior::Immediate, |transaction| {
136            tx::reserve_ordinal(transaction, kind, table, prefix)
137        })
138    }
139
140    pub(crate) fn paused(&self) -> Result<bool, StoreError> {
141        paused(&self.db.lock()).map_err(StoreError::from)
142    }
143
144    pub(crate) fn clear_restart_draining(&self, now_ms: i64) -> Result<(), StoreError> {
145        self.write(TransactionBehavior::Deferred, |transaction| {
146            tx::clear_restart_draining(transaction, now_ms)
147        })
148    }
149
150    #[cfg(test)]
151    pub(crate) fn restart_draining(&self) -> Result<bool, StoreError> {
152        restart_draining(&self.db.lock()).map_err(StoreError::from)
153    }
154
155    pub(crate) fn set_paused(&self, paused: bool, now_ms: i64) -> Result<(), StoreError> {
156        self.write(TransactionBehavior::Deferred, |transaction| {
157            tx::set_paused(transaction, paused, now_ms)
158        })
159    }
160
161    pub(crate) fn begin_restart_draining(
162        &self,
163        active_runs: usize,
164        now_ms: i64,
165    ) -> Result<bool, StoreError> {
166        self.write(TransactionBehavior::Immediate, |transaction| {
167            tx::begin_restart_draining(transaction, active_runs, now_ms)
168        })
169    }
170
171    pub(crate) fn resume_scheduler(&self, now_ms: i64) -> Result<bool, StoreError> {
172        self.write(TransactionBehavior::Immediate, |transaction| {
173            tx::resume_scheduler(transaction, now_ms)
174        })
175    }
176
177    pub(crate) fn probe_writable(&self, now_ms: i64) -> Result<(), StoreError> {
178        self.write(TransactionBehavior::Deferred, |transaction| {
179            tx::probe_writable(transaction, now_ms)
180        })
181    }
182
183    #[allow(clippy::too_many_arguments)]
184    pub fn settle(
185        &self,
186        run_id: &str,
187        exit_code: Option<i32>,
188        outcome: Outcome,
189        records: &[EvidenceRecord],
190        cooldown: Option<&CooldownUpdate<'_>>,
191        now_ms: i64,
192    ) -> Result<(RecordedOutcome, bool), StoreError> {
193        let mut connection = self.db.lock();
194        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
195        let run_state = RunState::from(outcome);
196        let changed = runs::tx::finish(&transaction, run_id, run_state, exit_code, now_ms)?;
197        if changed == 0 {
198            match runs::tx::state_and_exit(&transaction, run_id)? {
199                Some((_, Some(_))) => {}
200                Some((state, None)) => {
201                    return Err(StoreError::RunStateConflict {
202                        run_id: run_id.into(),
203                        state: Some(state),
204                        requested: run_state.as_str().into(),
205                    });
206                }
207                None => {
208                    return Err(StoreError::RunNotFound {
209                        run_id: run_id.into(),
210                    });
211                }
212            }
213        } else {
214            if let Some(cooldown) = cooldown {
215                limits::tx::upsert_cooldown(&transaction, run_id, cooldown, now_ms)?;
216            }
217            evidence::tx::record_settlement(&transaction, run_id, records, now_ms)?;
218            let ticket_id = runs::tx::ticket_id(&transaction, run_id)?;
219            runs::tx::record_event(
220                &transaction,
221                now_ms,
222                "run_finished",
223                Some(run_id),
224                Some(&ticket_id),
225                &serde_json::json!({
226                    "outcome": outcome.as_str(),
227                    "exit_code": exit_code,
228                    "ticket_state": TicketState::after_outcome(outcome).as_str(),
229                })
230                .to_string(),
231            )?;
232        }
233        let recorded = recorded_outcome(&transaction, run_id)?.ok_or_else(|| {
234            StoreError::RunStateConflict {
235                run_id: run_id.into(),
236                state: runs::tx::state(&transaction, run_id).ok().flatten(),
237                requested: run_state.as_str().into(),
238            }
239        })?;
240        transaction.commit()?;
241        Ok((recorded, changed != 0))
242    }
243
244    pub fn abort(&self, run_id: &str, ticket_id: &str, now_ms: i64) -> Result<bool, StoreError> {
245        let mut connection = self.db.lock();
246        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
247        let changed = runs::tx::abort(&transaction, run_id, now_ms)?;
248        if changed != 0 {
249            runs::tx::record_event(
250                &transaction,
251                now_ms,
252                "run_aborted",
253                Some(run_id),
254                Some(ticket_id),
255                "{}",
256            )?;
257        } else {
258            let state = runs::tx::state(&transaction, run_id)?;
259            if state.as_deref() != Some(RunState::Aborted.as_str()) {
260                return Err(StoreError::RunStateConflict {
261                    run_id: run_id.into(),
262                    state,
263                    requested: RunState::Aborted.as_str().into(),
264                });
265            }
266        }
267        transaction.commit()?;
268        Ok(changed != 0)
269    }
270
271    pub(crate) fn record_external_merge(
272        &self,
273        run_id: &str,
274        ticket_id: &str,
275        branch: &str,
276        branch_tip: &str,
277        observed_default_tip: &str,
278        now_ms: i64,
279    ) -> Result<bool, StoreError> {
280        self.write(TransactionBehavior::Immediate, |transaction| {
281            runs::tx::mark_cleanup_eligible(transaction, run_id, now_ms)?;
282            let data_json = serde_json::json!({
283                "branch": branch,
284                "branch_tip": branch_tip,
285                "observed_default_tip": observed_default_tip,
286            })
287            .to_string();
288            let inserted =
289                evidence::tx::record_external_merge(transaction, run_id, &data_json, now_ms)?;
290            if inserted {
291                runs::tx::record_event(
292                    transaction,
293                    now_ms,
294                    "external_merge_reconciled",
295                    Some(run_id),
296                    Some(ticket_id),
297                    &data_json,
298                )?;
299            }
300            Ok(inserted)
301        })
302    }
303
304    pub(crate) fn recorded_outcome(
305        &self,
306        run_id: &str,
307    ) -> Result<Option<RecordedOutcome>, StoreError> {
308        recorded_outcome(&self.db.lock(), run_id).map_err(StoreError::from)
309    }
310
311    /// Hands a claimed run to its driver, recording the workspace every stage
312    /// will execute in. This is the run's start: the activity feed says so
313    /// here, once, whatever kind of stage the flow opens with.
314    pub fn begin(
315        &self,
316        run_id: &str,
317        branch: &str,
318        worktree_path: &str,
319        now_ms: i64,
320    ) -> Result<Start, StoreError> {
321        let mut connection = self.db.lock();
322        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
323        let changed = runs::tx::mark_driving(&transaction, run_id, branch, worktree_path, now_ms)?;
324        if changed != 1 {
325            let state = runs::tx::state(&transaction, run_id)?;
326            return Ok(Start::Denied(StartDenial::NotClaimed { state }));
327        }
328        let ticket_id = runs::tx::ticket_id(&transaction, run_id)?;
329        runs::tx::record_event(
330            &transaction,
331            now_ms,
332            "run_started",
333            Some(run_id),
334            Some(&ticket_id),
335            "{}",
336        )?;
337        transaction.commit()?;
338        Ok(Start::Granted)
339    }
340
341    /// Records the agent process an agent stage launched. The run is already
342    /// under way; this only says which process the daemon is now supervising.
343    pub fn start(&self, start: &RunStart<'_>, now_ms: i64) -> Result<Start, StoreError> {
344        let mut connection = self.db.lock();
345        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
346        let changed = runs::tx::mark_running(
347            &transaction,
348            start.run_id,
349            start.branch,
350            start.worktree_path,
351            start.pid,
352            start.pid_start_time,
353            start.process_group_id,
354            start.worker_token,
355            start.worker_socket_path,
356            now_ms,
357        )?;
358        if changed != 1 {
359            let state = runs::tx::state(&transaction, start.run_id)?;
360            return Ok(Start::Denied(StartDenial::NotClaimed { state }));
361        }
362        transaction.commit()?;
363        Ok(Start::Granted)
364    }
365
366    #[allow(clippy::too_many_arguments)]
367    pub(crate) fn record_agent_exit(
368        &self,
369        run_id: &str,
370        attempt: u32,
371        exit_code: Option<i32>,
372        capture_complete: bool,
373        commits_json: &str,
374        vendor_error: Option<&crate::vendor::VendorErrorMatch>,
375        cooldown_until_ms: Option<i64>,
376        now_ms: i64,
377    ) -> Result<ExitClaim, StoreError> {
378        let mut connection = self.db.lock();
379        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
380        let changed = runs::tx::claim_agent_exit(&transaction, run_id, exit_code, now_ms)?;
381        if changed == 0 {
382            let state = runs::tx::state(&transaction, run_id)?;
383            return match state {
384                Some(state) => Ok(ExitClaim::AlreadyClaimed { state }),
385                None => Err(StoreError::RunNotFound {
386                    run_id: run_id.into(),
387                }),
388            };
389        }
390        evidence::tx::record_agent_exit(
391            &transaction,
392            run_id,
393            attempt,
394            exit_code,
395            capture_complete,
396            commits_json,
397            vendor_error,
398            cooldown_until_ms,
399            now_ms,
400        )?;
401        transaction.commit()?;
402        Ok(ExitClaim::Claimed)
403    }
404
405    /// Returns a run to its driver after an agent stage that is not the run's
406    /// own attempt at its ticket. The exit code, commits, and vendor reading on
407    /// the run belong to the first agent stage alone, so none of them move
408    /// here; only ownership of the walk does.
409    pub fn release_agent(&self, run_id: &str, now_ms: i64) -> Result<Exit, StoreError> {
410        let mut connection = self.db.lock();
411        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
412        let changed = runs::tx::release_agent(&transaction, run_id, now_ms)?;
413        if changed == 0 {
414            let state = runs::tx::state(&transaction, run_id)?;
415            return match state {
416                Some(state) => Ok(Exit::Denied(ExitDenial::AlreadyClaimed { state })),
417                None => Err(StoreError::RunNotFound {
418                    run_id: run_id.into(),
419                }),
420            };
421        }
422        transaction.commit()?;
423        Ok(Exit::Granted)
424    }
425
426    /// Checkpoints an agent exit, granting one caller ownership of the walk.
427    pub fn record_exit(&self, exit: &RunExit<'_>, now_ms: i64) -> Result<Exit, StoreError> {
428        match self.record_agent_exit(
429            exit.run_id,
430            exit.attempt,
431            exit.exit_code,
432            exit.capture_complete,
433            exit.commits_json,
434            exit.vendor_error,
435            exit.cooldown_until_ms,
436            now_ms,
437        )? {
438            ExitClaim::Claimed => Ok(Exit::Granted),
439            ExitClaim::AlreadyClaimed { state } => {
440                Ok(Exit::Denied(ExitDenial::AlreadyClaimed { state }))
441            }
442        }
443    }
444}
445
446fn recorded_outcome(
447    connection: &rusqlite::Connection,
448    run_id: &str,
449) -> rusqlite::Result<Option<RecordedOutcome>> {
450    let row = connection
451        .query_row(
452            "SELECT ticket_id, state, branch, attempt, exited_at_ms,
453                    (SELECT data_json FROM run_evidence
454                     WHERE run_id = runs.id AND kind = 'commits_observed'
455                     ORDER BY sequence DESC LIMIT 1),
456                    (SELECT data_json FROM run_evidence
457                     WHERE run_id = runs.id AND kind = 'vendor_error_classified'
458                     ORDER BY sequence DESC LIMIT 1),
459                    (SELECT until_ms FROM cooldowns WHERE source_run_id = runs.id
460                     ORDER BY until_ms DESC LIMIT 1),
461                    EXISTS(SELECT 1 FROM run_evidence
462                           WHERE run_id = runs.id AND kind = 'external_merge_observed')
463             FROM runs WHERE id = ?1",
464            params![run_id],
465            |row| {
466                Ok((
467                    row.get::<_, String>(0)?,
468                    row.get::<_, RunState>(1)?,
469                    row.get::<_, Option<String>>(2)?,
470                    row.get::<_, i64>(3)?,
471                    row.get::<_, Option<i64>>(4)?,
472                    row.get::<_, Option<String>>(5)?,
473                    row.get::<_, Option<String>>(6)?,
474                    row.get::<_, Option<i64>>(7)?,
475                    row.get::<_, bool>(8)?,
476                ))
477            },
478        )
479        .optional()?;
480    let Some((
481        ticket_id,
482        state,
483        branch,
484        attempt,
485        finished_at_ms,
486        commits,
487        vendor_error,
488        cooldown,
489        externally_merged,
490    )) = row
491    else {
492        return Ok(None);
493    };
494    let Some(verdict) = externally_merged
495        .then_some(Outcome::Merged)
496        .or_else(|| state.outcome())
497    else {
498        return Ok(None);
499    };
500    let Some(finished_at_ms) = finished_at_ms else {
501        return Ok(None);
502    };
503    let commit_count = commits
504        .as_deref()
505        .and_then(|value| serde_json::from_str::<serde_json::Value>(value).ok())
506        .and_then(|value| value["oids"].as_array().map(Vec::len))
507        .unwrap_or(0)
508        .min(u32::MAX as usize) as u32;
509    let not_before_ms = vendor_error
510        .as_deref()
511        .and_then(|value| serde_json::from_str::<serde_json::Value>(value).ok())
512        .and_then(|value| value["cooldown_until_ms"].as_i64())
513        .or(cooldown);
514    Ok(Some(RecordedOutcome {
515        work: WorkOutcome {
516            ticket_id,
517            owner: OwnerId(run_id.into()),
518            verdict,
519            branch,
520            commit_count,
521            attempt: attempt.clamp(0, i64::from(u32::MAX)) as u32,
522            finished_at_ms,
523        },
524        not_before_ms,
525    }))
526}
527
528fn paused(connection: &rusqlite::Connection) -> rusqlite::Result<bool> {
529    let paused: i64 = connection.query_row(
530        "SELECT paused FROM scheduler_state WHERE singleton = 1",
531        [],
532        |row| row.get(0),
533    )?;
534    Ok(paused != 0)
535}
536
537#[cfg(test)]
538fn restart_draining(connection: &rusqlite::Connection) -> rusqlite::Result<bool> {
539    let draining: i64 = connection.query_row(
540        "SELECT draining FROM scheduler_state WHERE singleton = 1",
541        [],
542        |row| row.get(0),
543    )?;
544    Ok(draining != 0)
545}
546
547pub(crate) mod tx {
548    use rusqlite::{Transaction, params};
549
550    /// `prefix` is what the minted ids of `table` begin with, and it decides
551    /// where the ordinal starts inside them: `TR91` yields 91 only when the
552    /// substring skips both letters. Getting it wrong casts to zero rather than
553    /// erroring, which would silently disarm the high-water mark below.
554    pub(crate) fn reserve_ordinal(
555        transaction: &Transaction<'_>,
556        kind: &str,
557        table: &str,
558        prefix: &str,
559    ) -> rusqlite::Result<i64> {
560        let reserved: i64 = transaction.query_row(
561            "SELECT next_ordinal FROM id_counters WHERE kind = ?1",
562            params![kind],
563            |row| row.get(0),
564        )?;
565        let ordinal_start = prefix.len() + 1;
566        let existing: i64 = transaction.query_row(
567            &format!(
568                "SELECT COALESCE(MAX(CAST(SUBSTR(id, {ordinal_start}) AS INTEGER)), 0) + 1
569                 FROM {table}"
570            ),
571            [],
572            |row| row.get(0),
573        )?;
574        let ordinal = reserved.max(existing);
575        transaction.execute(
576            "UPDATE id_counters SET next_ordinal = ?2 WHERE kind = ?1",
577            params![kind, ordinal + 1],
578        )?;
579        Ok(ordinal)
580    }
581
582    pub(crate) fn clear_restart_draining(
583        transaction: &Transaction<'_>,
584        now_ms: i64,
585    ) -> rusqlite::Result<()> {
586        transaction.execute(
587            "UPDATE scheduler_state SET draining = 0, updated_at_ms = ?1 WHERE singleton = 1",
588            params![now_ms],
589        )?;
590        Ok(())
591    }
592
593    pub(crate) fn set_paused(
594        transaction: &Transaction<'_>,
595        paused: bool,
596        now_ms: i64,
597    ) -> rusqlite::Result<()> {
598        transaction.execute(
599            "UPDATE scheduler_state SET paused = ?1, updated_at_ms = ?2 WHERE singleton = 1",
600            params![i64::from(paused), now_ms],
601        )?;
602        Ok(())
603    }
604
605    pub(crate) fn begin_restart_draining(
606        transaction: &Transaction<'_>,
607        active_runs: usize,
608        now_ms: i64,
609    ) -> rusqlite::Result<bool> {
610        let changed = transaction.execute(
611            "UPDATE scheduler_state SET draining = 1, updated_at_ms = ?1
612             WHERE singleton = 1 AND draining = 0",
613            params![now_ms],
614        )? != 0;
615        if changed {
616            super::runs::tx::record_event(
617                transaction,
618                now_ms,
619                "daemon_restart_requested",
620                None,
621                None,
622                &serde_json::json!({"active_runs": active_runs}).to_string(),
623            )?;
624        }
625        Ok(changed)
626    }
627
628    pub(crate) fn resume_scheduler(
629        transaction: &Transaction<'_>,
630        now_ms: i64,
631    ) -> rusqlite::Result<bool> {
632        let was_draining: bool = transaction.query_row(
633            "SELECT draining FROM scheduler_state WHERE singleton = 1",
634            [],
635            |row| row.get::<_, i64>(0).map(|value| value != 0),
636        )?;
637        transaction.execute(
638            "UPDATE scheduler_state
639             SET paused = 0, draining = 0, updated_at_ms = ?1
640             WHERE singleton = 1",
641            params![now_ms],
642        )?;
643        Ok(was_draining)
644    }
645
646    pub(crate) fn probe_writable(
647        transaction: &Transaction<'_>,
648        now_ms: i64,
649    ) -> rusqlite::Result<()> {
650        transaction.execute(
651            "UPDATE scheduler_state SET updated_at_ms = ?1 WHERE singleton = 1",
652            params![now_ms],
653        )?;
654        Ok(())
655    }
656}
657
658#[cfg(test)]
659pub(crate) mod test_support {
660    use std::path::Path;
661
662    use rusqlite::{TransactionBehavior, params};
663
664    use super::{CooldownUpdate, EvidenceRecord, RunAdmission, RunStore};
665    use crate::db::Db;
666    use crate::outcome::Outcome;
667
668    pub(crate) fn open_seeded(path: &Path) -> RunStore {
669        let db = Db::open(path, 1_000).unwrap();
670        db.lock()
671            .execute_batch(
672                "INSERT INTO projects
673                     (id, file_path, source, source_ref, title, created_at_ms, updated_at_ms)
674                 VALUES
675                     ('default', '.agents/sloop/projects/default.md', 'local', NULL,
676                      'Default', 1000, 1000);
677                 INSERT INTO tickets
678                     (id, project_id, file_path, source, source_ref, state, attempts,
679                      content_hash, name, worktree, target, model, effort, flow, body,
680                      created_at_ms, updated_at_ms)
681                 VALUES
682                     ('T1', 'default', '.agents/sloop/tickets/t1.md', 'local', NULL,
683                      'ready', 0, '', 'Ticket one', 'sloop/T1', 'claude', 'sonnet',
684                      'medium', 'default', '', 1000, 1000);",
685            )
686            .unwrap();
687        crate::work_state::trigger::insert(
688            &db.lock(),
689            &crate::work_state::trigger::NewTrigger {
690                id: "TR1",
691                kind: crate::domain::trigger::TriggerKind::Immediate,
692                ticket_id: Some("T1"),
693                project_id: None,
694                eligible_at_ms: None,
695                interval_ms: None,
696            },
697            1_000,
698        )
699        .unwrap();
700        RunStore::from_db(db)
701    }
702
703    pub(crate) fn claim_run(
704        store: &RunStore,
705        run_id: &str,
706        flow_json: &str,
707        ticket_json: &str,
708        now_ms: i64,
709    ) {
710        store
711            .insert_claimed_run(
712                &RunAdmission {
713                    run_id,
714                    trigger_id: "TR1",
715                    ticket_id: "T1",
716                    flow_json,
717                    ticket_json,
718                },
719                now_ms,
720            )
721            .unwrap();
722        let db = store.db();
723        let mut connection = db.lock();
724        let transaction = connection
725            .transaction_with_behavior(TransactionBehavior::Immediate)
726            .unwrap();
727        transaction
728            .execute(
729                "UPDATE tickets SET state = 'claimed', attempts = attempts + 1,
730                                    updated_at_ms = ?1 WHERE id = 'T1'",
731                params![now_ms],
732            )
733            .unwrap();
734        transaction
735            .execute(
736                "INSERT INTO leases
737                     (ticket_id, run_id, owner_id, acquired_at_ms, renewed_at_ms, expires_at_ms)
738                 VALUES ('T1', ?1, 'daemon-1', ?2, ?2, ?3)",
739                params![run_id, now_ms, now_ms + 60_000],
740            )
741            .unwrap();
742        transaction.commit().unwrap();
743    }
744
745    pub(crate) fn settle_run(
746        store: &RunStore,
747        run_id: &str,
748        exit_code: Option<i32>,
749        outcome: Outcome,
750        evidence: &[EvidenceRecord],
751        cooldown: Option<&CooldownUpdate<'_>>,
752        now_ms: i64,
753    ) -> bool {
754        let (_, applied) = store
755            .settle(run_id, exit_code, outcome, evidence, cooldown, now_ms)
756            .unwrap();
757        let ticket_state = match outcome {
758            Outcome::Merged => "merged",
759            Outcome::Failed => "failed",
760            Outcome::NeedsReview => "needs_review",
761            Outcome::Cancelled | Outcome::RateLimited | Outcome::Orphaned => "ready",
762        };
763        let db = store.db();
764        let mut connection = db.lock();
765        let transaction = connection
766            .transaction_with_behavior(TransactionBehavior::Immediate)
767            .unwrap();
768        transaction
769            .execute("DELETE FROM leases WHERE run_id = ?1", params![run_id])
770            .unwrap();
771        transaction
772            .execute(
773                "UPDATE tickets SET state = ?1, updated_at_ms = ?2 WHERE id = 'T1'",
774                params![ticket_state, now_ms],
775            )
776            .unwrap();
777        transaction.commit().unwrap();
778        applied
779    }
780
781    pub(crate) fn abort_run(store: &RunStore, run_id: &str, now_ms: i64) -> bool {
782        let applied = store.abort(run_id, "T1", now_ms).unwrap();
783        let db = store.db();
784        let mut connection = db.lock();
785        let transaction = connection
786            .transaction_with_behavior(TransactionBehavior::Immediate)
787            .unwrap();
788        transaction
789            .execute("DELETE FROM leases WHERE run_id = ?1", params![run_id])
790            .unwrap();
791        transaction
792            .execute(
793                "UPDATE tickets SET state = 'ready', updated_at_ms = ?1 WHERE id = 'T1'",
794                params![now_ms],
795            )
796            .unwrap();
797        transaction.commit().unwrap();
798        applied
799    }
800}
sloop/run_store/mod.rs

sloop/run_store/
mod.rs
