artifactize_core/store/
receipts.rs

1use crate::types::{
2    Fingerprint, RequestId, RequestStatus, ReuseKey, RunId, RunStatus, SessionId, Timestamp,
3};
4use std::{
5    collections::{BTreeMap, BTreeSet},
6    path::{Path, PathBuf},
7    time::Duration,
8};
9
10use rusqlite::{OptionalExtension, params};
11use serde::{Deserialize, Serialize};
12
13use tokio_rusqlite::Connection;
14
15use super::{ExecutionOptions, Origin, Producer, Provenance, STATE_SCHEMA_VERSION};
16use crate::{
17    config::ProfileKind,
18    workspace::{canonical_target, outside_workspace, prepare_directory},
19};
20
21pub const DATABASE: &str = "state.sqlite";
22/// Yield between WAL initialization races instead of spinning, still within the
23/// shared SQLite contention deadline used by the outer initialization loop.
24const INITIALIZATION_RETRY_INTERVAL: Duration = Duration::from_millis(20);
25
26/// The four tables of a state and the indexes its lookups use.
27///
28/// - `runs`: one row per Run.
29/// - `requests`: one row per eval of a Run, at its `ordinal` in the Run's selection order;
30///   `claimed_by` and `claimed_at` hold the Human claim of a waiting request.
31/// - `executions`: one row per execution. `backend` names the Agent backend it runs on, and
32///   a RUNNING execution with a backend holds one of that backend's machine-wide slots. A
33///   completed GREEN/RED execution with a key is a record of the key's history while
34///   `completed_at` (sortable), `bytes` and `last_used` are set; the cache GC and `cache rm`
35///   clear them, and the execution stays.
36/// - `state_meta`: named values of the state itself: `id`, a random UUID made when the
37///   database is created, which identifies the state in Agent session references.
38const SCHEMA: &str =
39    "CREATE TABLE runs(id TEXT PRIMARY KEY, repo TEXT NOT NULL, status TEXT NOT NULL, data TEXT NOT NULL);
40    CREATE TABLE requests(id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES runs(id), eval_id TEXT NOT NULL, ordinal INTEGER NOT NULL, execution_id TEXT REFERENCES executions(id), status TEXT NOT NULL, claimed_by TEXT, claimed_at TEXT, data TEXT NOT NULL, UNIQUE(run_id, eval_id), UNIQUE(run_id, ordinal));
41    CREATE TABLE executions(id TEXT PRIMARY KEY, key TEXT, eval_def_hash TEXT NOT NULL, status TEXT NOT NULL, owner_pid INTEGER NOT NULL, owner_start_time INTEGER NOT NULL, backend TEXT, completed_at TEXT, bytes INTEGER, last_used TEXT, data TEXT NOT NULL);
42    CREATE TABLE state_meta(name TEXT PRIMARY KEY, value TEXT NOT NULL);
43    CREATE INDEX active_request_execution ON requests(execution_id) WHERE status IN ('QUEUED','RUNNING','WAITING_HUMAN');
44    CREATE UNIQUE INDEX active_key ON executions(key) WHERE key IS NOT NULL AND status IN ('RUNNING','WAITING_HUMAN');
45    CREATE INDEX key_history ON executions(key, completed_at);
46    CREATE INDEX definition_history ON executions(eval_def_hash, completed_at);
47    CREATE INDEX running_backend ON executions(backend) WHERE status='RUNNING';";
48
49/// Why a state written by an earlier artifactize is refused: no version migrates.
50pub const EARLIER_STATE: &str = "This state was written by an earlier artifactize. Start a new state (set ARTIFACTIZE_STATE_HOME or move the old one away). artifactize does not migrate it.";
51
52/// Why a state database of schema `version`, neither this one nor uninitialized, is refused.
53pub fn schema_error(version: u32) -> String {
54    match version {
55        0 => "Unsupported state schema version: 0.".into(),
56        version if version < STATE_SCHEMA_VERSION => EARLIER_STATE.into(),
57        version => {
58            format!("Unsupported state schema version: {version}; a newer artifactize wrote it.")
59        }
60    }
61}
62
63#[derive(Debug, thiserror::Error)]
64pub enum Error {
65    #[error(transparent)]
66    Sql(#[from] rusqlite::Error),
67    #[error(transparent)]
68    Json(#[from] serde_json::Error),
69    #[error("{0}")]
70    Invalid(String),
71}
72
73#[derive(Debug, Clone, Serialize, Deserialize)]
74#[serde(try_from = "RunWire", into = "RunWire")]
75pub struct Run {
76    pub id: RunId,
77    pub repo_path: PathBuf,
78    /// Optional display metadata; schema 5 Runs written before this remain readable.
79    pub repository: crate::repository::Identity,
80    pub state_dir: PathBuf,
81    pub state: super::RunState,
82    pub created_at: Timestamp,
83    pub selection: crate::project::selection::Selection,
84    pub profile: Option<crate::project::selection::ProfileSelection>,
85    pub definitions: super::definitions::Definitions,
86    pub recursive: bool,
87    pub force: bool,
88    pub ignore_gates: bool,
89    pub jobs: usize,
90    /// Fingerprints computed at once; absent in Runs saved before 0.5.
91    pub fingerprint_jobs: Option<usize>,
92    pub max_executions: Option<u64>,
93    pub executions_started: u64,
94    pub wait_timeout_ms: Option<Duration>,
95    pub wait_timed_out: bool,
96    /// Kinds whose evals only reuse a result; one with nothing to reuse is not executed.
97    pub reuse_only: BTreeSet<ProfileKind>,
98    pub validation: super::Validation,
99    /// Agent backends this Run stopped admitting reviews on, in the order they stopped.
100    pub stopped_backends: Vec<StoppedBackend>,
101    /// Verdicts the Run took from saved results for evals it has no request for, such as the
102    /// dependencies of a partial Run, so readers judge gates on the Run's own evidence.
103    pub evidence: BTreeMap<crate::types::EvalId, crate::types::RequestStatus>,
104}
105
106#[derive(Debug, Clone, Serialize, Deserialize)]
107#[serde(rename_all = "camelCase")]
108pub(crate) struct RunWire {
109    pub id: RunId,
110    #[serde(with = "crate::platform::path_serde")]
111    pub repo_path: PathBuf,
112    /// Optional display metadata; schema 5 Runs written before this remain readable.
113    #[serde(flatten)]
114    pub repository: crate::repository::Identity,
115    #[serde(with = "crate::platform::path_serde")]
116    pub state_dir: PathBuf,
117    pub status: RunStatus,
118    pub created_at: Timestamp,
119    pub completed_at: Option<Timestamp>,
120    pub selection: crate::project::selection::Selection,
121    #[serde(default)]
122    pub profile: Option<crate::project::selection::ProfileSelection>,
123    #[serde(default)]
124    pub definitions: super::definitions::Definitions,
125    #[serde(default)]
126    pub recursive: bool,
127    #[serde(default)]
128    pub force: bool,
129    #[serde(default)]
130    pub ignore_gates: bool,
131    #[serde(default = "default_jobs")]
132    pub jobs: usize,
133    /// Fingerprints computed at once; absent in Runs saved before 0.5.
134    #[serde(default, skip_serializing_if = "Option::is_none")]
135    pub fingerprint_jobs: Option<usize>,
136    #[serde(default)]
137    pub max_executions: Option<u64>,
138    #[serde(default)]
139    pub executions_started: u64,
140    #[serde(default, with = "super::wait_timeout")]
141    pub wait_timeout_ms: Option<Duration>,
142    #[serde(default)]
143    pub wait_timed_out: bool,
144    /// Kinds whose evals only reuse a result; one with nothing to reuse is not executed.
145    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
146    pub reuse_only: BTreeSet<ProfileKind>,
147    pub validation: super::Validation,
148    pub error: Option<String>,
149    /// Agent backends this Run stopped admitting reviews on, in the order they stopped.
150    #[serde(default, skip_serializing_if = "Vec::is_empty")]
151    pub stopped_backends: Vec<StoppedBackend>,
152    /// Verdicts the Run took from saved results for evals it has no request for, such as the
153    /// dependencies of a partial Run, so readers judge gates on the Run's own evidence.
154    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
155    pub evidence: BTreeMap<crate::types::EvalId, crate::types::RequestStatus>,
156}
157
158impl TryFrom<RunWire> for Run {
159    type Error = String;
160    fn try_from(wire: RunWire) -> Result<Self, Self::Error> {
161        let state = super::RunState::from_parts(wire.status, wire.completed_at, wire.error)?;
162        Ok(Self {
163            state,
164            id: wire.id,
165            repo_path: wire.repo_path,
166            repository: wire.repository,
167            state_dir: wire.state_dir,
168            created_at: wire.created_at,
169            selection: wire.selection,
170            profile: wire.profile,
171            definitions: wire.definitions,
172            recursive: wire.recursive,
173            force: wire.force,
174            ignore_gates: wire.ignore_gates,
175            jobs: wire.jobs,
176            fingerprint_jobs: wire.fingerprint_jobs,
177            max_executions: wire.max_executions,
178            executions_started: wire.executions_started,
179            wait_timeout_ms: wire.wait_timeout_ms,
180            wait_timed_out: wire.wait_timed_out,
181            reuse_only: wire.reuse_only,
182            validation: wire.validation,
183            stopped_backends: wire.stopped_backends,
184            evidence: wire.evidence,
185        })
186    }
187}
188impl From<Run> for RunWire {
189    fn from(record: Run) -> Self {
190        Self {
191            status: record.status(),
192            completed_at: record.completed_at(),
193            error: record.error().map(str::to_owned),
194            id: record.id,
195            repo_path: record.repo_path,
196            repository: record.repository,
197            state_dir: record.state_dir,
198            created_at: record.created_at,
199            selection: record.selection,
200            profile: record.profile,
201            definitions: record.definitions,
202            recursive: record.recursive,
203            force: record.force,
204            ignore_gates: record.ignore_gates,
205            jobs: record.jobs,
206            fingerprint_jobs: record.fingerprint_jobs,
207            max_executions: record.max_executions,
208            executions_started: record.executions_started,
209            wait_timeout_ms: record.wait_timeout_ms,
210            wait_timed_out: record.wait_timed_out,
211            reuse_only: record.reuse_only,
212            validation: record.validation,
213            stopped_backends: record.stopped_backends,
214            evidence: record.evidence,
215        }
216    }
217}
218impl Run {
219    pub fn status(&self) -> RunStatus {
220        self.state.status()
221    }
222    pub fn completed_at(&self) -> Option<Timestamp> {
223        self.state.completed_at()
224    }
225    pub fn error(&self) -> Option<&str> {
226        self.state.error()
227    }
228}
229
230/// An Agent backend a Run stopped after an AUTHENTICATION or QUOTA failure, which every
231/// later review on it would repeat.
232#[derive(Debug, Clone, Serialize, Deserialize)]
233#[serde(rename_all = "camelCase")]
234pub struct StoppedBackend {
235    pub backend: crate::config::Backend,
236    pub error_code: crate::types::BackendStopCode,
237    pub eval_id: crate::types::EvalId,
238    pub request_id: RequestId,
239    pub error: String,
240}
241
242fn default_jobs() -> usize {
243    crate::project::DEFAULT_JOBS
244}
245
246#[derive(Debug, Clone, Serialize, Deserialize)]
247#[serde(try_from = "RequestWire", into = "RequestWire")]
248pub struct Request {
249    pub state: super::RequestState,
250    pub id: RequestId,
251    pub run_id: RunId,
252    pub eval_id: crate::types::EvalId,
253    pub target: crate::types::ArtifactName,
254    pub title: String,
255    pub profile: crate::config::StoredProfile,
256    pub requested_profile: crate::config::StoredProfile,
257    pub eval_def_hash: crate::types::DefinitionHash,
258    /// The requested execution options, or a reused result's.
259    pub options: ExecutionOptions,
260    pub execution_id: Option<crate::types::ExecutionId>,
261    pub provenance: Option<Provenance>,
262    /// Usage spent by this request; a reused request spent none.
263    pub usage: Option<Vec<crate::llm::Attempt>>,
264    /// The reused execution's original usage, never counted as spent.
265    pub reused_usage: Option<Vec<crate::llm::Attempt>>,
266    /// The reused result came from the live execution this request waited for, not a
267    /// completed record. Saved only; output reports it as `source.kind` ([`crate::query::source`]).
268    pub joined: bool,
269    /// Who produced a reused result.
270    pub producer: Option<Producer>,
271    /// The Human claimant who submitted a reused result.
272    pub reviewer: Option<crate::types::ReviewerId>,
273    /// The remote store, publisher and publication time of a result reused from a mirror.
274    pub origin: Option<Origin>,
275    /// The Agent review's session id ([`crate::agent::session_id`]), which every request of
276    /// the review sends as its prompt-cache identity. Absent when no Agent review ran, as on
277    /// reuse, and on requests saved before it existed.
278    pub session_id: Option<SessionId>,
279    /// Where the saved conversation behind this request's result lives: its own review's,
280    /// or for a reused result the producing review's. Absent when none was saved.
281    pub session: Option<crate::agent::session::SessionRef>,
282    pub human_definition: Option<super::HumanDefinition>,
283    pub payload: crate::config::StoredPayload,
284    pub references: BTreeMap<crate::config::ReferenceName, crate::types::ArtifactName>,
285    pub deps: Vec<crate::types::ArtifactName>,
286    pub force: bool,
287    /// The target Artifact's fingerprint.
288    pub fingerprint: Option<Fingerprint>,
289    /// The reuse key; absent without a fingerprint on every Artifact the eval depends on.
290    pub key: Option<ReuseKey>,
291    /// Each Artifact the key covers, with its fingerprint.
292    pub fingerprints: std::collections::BTreeMap<crate::types::ArtifactName, Fingerprint>,
293    pub created_at: Timestamp,
294    pub started_at: Option<Timestamp>,
295    pub cwd: PathBuf,
296    pub run_dir: Option<PathBuf>,
297    pub argv: Option<Vec<String>>,
298    pub child: Option<super::ChildIdentity>,
299    pub blocked_reason: Option<String>,
300    /// Unfulfilled Artifacts and Evals behind a derived dependency verdict.
301    pub blocked_by: Vec<super::Blocker>,
302    /// What a queued request waits for besides a job; `blocked_reason` says it in words.
303    pub queue: Option<QueueReason>,
304}
305
306/// What a queued request waits for besides a job slot of its Run.
307#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
308#[serde(tag = "kind", rename_all = "camelCase")]
309pub enum QueueReason {
310    /// A free machine-wide slot of this backend (`limits.json`).
311    Slot { backend: crate::config::Backend },
312}
313
314#[derive(Debug, Clone, Serialize, Deserialize)]
315#[serde(rename_all = "camelCase")]
316pub(crate) struct RequestWire {
317    pub id: RequestId,
318    pub run_id: RunId,
319    pub eval_id: crate::types::EvalId,
320    pub target: crate::types::ArtifactName,
321    pub title: String,
322    pub profile: crate::config::StoredProfile,
323    pub requested_profile: crate::config::StoredProfile,
324    pub eval_def_hash: crate::types::DefinitionHash,
325    /// The requested execution options, or a reused result's.
326    #[serde(default)]
327    pub options: ExecutionOptions,
328    pub execution_id: Option<crate::types::ExecutionId>,
329    pub provenance: Option<Provenance>,
330    /// Usage spent by this request; a reused request spent none.
331    pub usage: Option<Vec<crate::llm::Attempt>>,
332    /// The reused execution's original usage, never counted as spent.
333    #[serde(default, skip_serializing_if = "Option::is_none")]
334    pub reused_usage: Option<Vec<crate::llm::Attempt>>,
335    /// The reused result came from the live execution this request waited for, not a
336    /// completed record. Saved only; output reports it as `source.kind` ([`crate::query::source`]).
337    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
338    pub joined: bool,
339    /// Who produced a reused result.
340    #[serde(default, skip_serializing_if = "Option::is_none")]
341    pub producer: Option<Producer>,
342    /// The Human claimant who submitted a reused result.
343    #[serde(default, skip_serializing_if = "Option::is_none")]
344    pub reviewer: Option<crate::types::ReviewerId>,
345    /// The remote store, publisher and publication time of a result reused from a mirror.
346    #[serde(default, skip_serializing_if = "Option::is_none")]
347    pub origin: Option<Origin>,
348    /// The Agent review's session id ([`crate::agent::session_id`]), which every request of
349    /// the review sends as its prompt-cache identity. Absent when no Agent review ran, as on
350    /// reuse, and on requests saved before it existed.
351    #[serde(default, skip_serializing_if = "Option::is_none")]
352    pub session_id: Option<SessionId>,
353    /// Where the saved conversation behind this request's result lives: its own review's,
354    /// or for a reused result the producing review's. Absent when none was saved.
355    #[serde(default, skip_serializing_if = "Option::is_none")]
356    pub session: Option<crate::agent::session::SessionRef>,
357    #[serde(default, skip_serializing_if = "Option::is_none")]
358    pub human_definition: Option<super::HumanDefinition>,
359    pub payload: crate::config::StoredPayload,
360    pub references: BTreeMap<crate::config::ReferenceName, crate::types::ArtifactName>,
361    pub deps: Vec<crate::types::ArtifactName>,
362    #[serde(default)]
363    pub force: bool,
364    /// The target Artifact's fingerprint.
365    #[serde(default)]
366    pub fingerprint: Option<Fingerprint>,
367    /// The reuse key; absent without a fingerprint on every Artifact the eval depends on.
368    #[serde(default)]
369    pub key: Option<ReuseKey>,
370    /// Each Artifact the key covers, with its fingerprint.
371    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
372    pub fingerprints: std::collections::BTreeMap<crate::types::ArtifactName, Fingerprint>,
373    pub status: RequestStatus,
374    pub created_at: Timestamp,
375    pub started_at: Option<Timestamp>,
376    pub completed_at: Option<Timestamp>,
377    #[serde(with = "crate::platform::path_serde")]
378    pub cwd: PathBuf,
379    #[serde(default, with = "crate::platform::path_serde::option")]
380    pub run_dir: Option<PathBuf>,
381    pub argv: Option<Vec<String>>,
382    pub child: Option<super::ChildIdentity>,
383    pub result: Option<super::ExecutionResult>,
384    pub error: Option<String>,
385    pub error_code: Option<crate::types::FailureCode>,
386    pub blocked_reason: Option<String>,
387    /// Unfulfilled Artifacts and Evals behind a derived dependency verdict.
388    #[serde(default, skip_serializing_if = "Vec::is_empty")]
389    pub blocked_by: Vec<super::Blocker>,
390    /// Absent from requests saved before it existed, which read as waiting for a job.
391    #[serde(default, skip_serializing_if = "Option::is_none")]
392    pub queue: Option<QueueReason>,
393}
394
395impl TryFrom<RequestWire> for Request {
396    type Error = String;
397    fn try_from(wire: RequestWire) -> Result<Self, Self::Error> {
398        let state = super::RequestState::from_parts(
399            wire.status,
400            wire.result,
401            wire.error,
402            wire.error_code,
403            wire.completed_at,
404        )?;
405        Ok(Self {
406            state,
407            id: wire.id,
408            run_id: wire.run_id,
409            eval_id: wire.eval_id,
410            target: wire.target,
411            title: wire.title,
412            profile: wire.profile,
413            requested_profile: wire.requested_profile,
414            eval_def_hash: wire.eval_def_hash,
415            options: wire.options,
416            execution_id: wire.execution_id,
417            provenance: wire.provenance,
418            usage: wire.usage,
419            reused_usage: wire.reused_usage,
420            joined: wire.joined,
421            producer: wire.producer,
422            reviewer: wire.reviewer,
423            origin: wire.origin,
424            session_id: wire.session_id,
425            session: wire.session,
426            human_definition: wire.human_definition,
427            payload: wire.payload,
428            references: wire.references,
429            deps: wire.deps,
430            force: wire.force,
431            fingerprint: wire.fingerprint,
432            key: wire.key,
433            fingerprints: wire.fingerprints,
434            created_at: wire.created_at,
435            started_at: wire.started_at,
436            cwd: wire.cwd,
437            run_dir: wire.run_dir,
438            argv: wire.argv,
439            child: wire.child,
440            blocked_reason: wire.blocked_reason,
441            blocked_by: wire.blocked_by,
442            queue: wire.queue,
443        })
444    }
445}
446impl From<Request> for RequestWire {
447    fn from(record: Request) -> Self {
448        Self {
449            status: record.status(),
450            result: record.result().cloned(),
451            error: record.error().map(str::to_owned),
452            error_code: record.error_code(),
453            completed_at: record.completed_at(),
454            id: record.id,
455            run_id: record.run_id,
456            eval_id: record.eval_id,
457            target: record.target,
458            title: record.title,
459            profile: record.profile,
460            requested_profile: record.requested_profile,
461            eval_def_hash: record.eval_def_hash,
462            options: record.options,
463            execution_id: record.execution_id,
464            provenance: record.provenance,
465            usage: record.usage,
466            reused_usage: record.reused_usage,
467            joined: record.joined,
468            producer: record.producer,
469            reviewer: record.reviewer,
470            origin: record.origin,
471            session_id: record.session_id,
472            session: record.session,
473            human_definition: record.human_definition,
474            payload: record.payload,
475            references: record.references,
476            deps: record.deps,
477            force: record.force,
478            fingerprint: record.fingerprint,
479            key: record.key,
480            fingerprints: record.fingerprints,
481            created_at: record.created_at,
482            started_at: record.started_at,
483            cwd: record.cwd,
484            run_dir: record.run_dir,
485            argv: record.argv,
486            child: record.child,
487            blocked_reason: record.blocked_reason,
488            blocked_by: record.blocked_by,
489            queue: record.queue,
490        }
491    }
492}
493impl Request {
494    pub fn status(&self) -> crate::types::RequestStatus {
495        self.state.status()
496    }
497    pub fn result(&self) -> Option<&super::ExecutionResult> {
498        self.state.result()
499    }
500    pub fn error(&self) -> Option<&str> {
501        self.state.error()
502    }
503    pub fn error_code(&self) -> Option<crate::types::FailureCode> {
504        self.state.error_code()
505    }
506    pub fn completed_at(&self) -> Option<crate::types::Timestamp> {
507        self.state.completed_at()
508    }
509}
510
511#[derive(Debug, Serialize, Deserialize)]
512pub struct RunView {
513    #[serde(flatten)]
514    pub run: Run,
515    pub requests: Vec<Request>,
516    #[serde(default, skip_serializing_if = "Vec::is_empty")]
517    pub unreadable: Vec<super::Unreadable>,
518}
519
520#[derive(Clone)]
521pub struct Receipts {
522    pub(super) connection: NotifyingConnection,
523}
524
525/// Notify at the DB-worker completion boundary, even if the awaiting caller was cancelled.
526/// total_changes is conservative (rolled-back changes can invalidate harmlessly). Only a
527/// finished autocommit edge can publish; no event is an audit record or proof of success.
528#[derive(Clone)]
529pub(crate) struct NotifyingConnection {
530    inner: Connection,
531    publisher: crate::changes::Publisher,
532}
533impl NotifyingConnection {
534    pub(crate) fn new(inner: Connection, state: &Path) -> Self {
535        Self {
536            inner,
537            publisher: crate::changes::Publisher::new(state),
538        }
539    }
540    pub(crate) async fn call<F, R, E>(&self, function: F) -> Result<R, tokio_rusqlite::Error<E>>
541    where
542        F: FnOnce(&mut rusqlite::Connection) -> Result<R, E> + Send + 'static,
543        R: Send + 'static,
544        E: Send + 'static,
545    {
546        let publisher = self.publisher.clone();
547        self.inner
548            .call(move |db| {
549                let before = db.total_changes();
550                let result = function(db);
551                if db.is_autocommit() && db.total_changes() != before {
552                    publisher.notify(crate::changes::Change::StateInvalidated);
553                }
554                result
555            })
556            .await
557    }
558}
559
560impl Receipts {
561    pub async fn open(state: &Path, repo: &Path) -> Result<Self, String> {
562        let state = prepare_directory(state, repo).map_err(|e| e.to_string())?;
563        check_files(&state)?;
564        let connection = Connection::open(state.join(DATABASE))
565            .await
566            .map_err(|e| e.to_string())?;
567        let connection = NotifyingConnection::new(connection, &state);
568        // WAL initialization retries share the database contention budget.
569        let deadline = tokio::time::Instant::now() + super::SQLITE_BUSY_TIMEOUT;
570        let state_id: crate::types::StateId = crate::agent::uuid()?.parse()?;
571        loop {
572            let state_id = state_id.clone();
573            let initialized = connection
574                .call(move |db| -> Result<(), Error> {
575                    db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
576                    db.pragma_update(None, "foreign_keys", true)?;
577                    let version: u32 =
578                        db.pragma_query_value(None, "user_version", |row| row.get(0))?;
579                    if version != 0 && version != STATE_SCHEMA_VERSION {
580                        return Err(Error::Invalid(schema_error(version)));
581                    }
582                    db.pragma_update(None, "journal_mode", "WAL")?;
583                    let transaction =
584                        db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
585                    // Publish the schema, the state's id and the version together; readers see
586                    // an empty snapshot until commit.
587                    if !schema_initialized(&transaction)? {
588                        transaction.execute_batch(SCHEMA)?;
589                        transaction.execute(
590                            "INSERT INTO state_meta(name,value) VALUES ('id',?)",
591                            [&state_id],
592                        )?;
593                        transaction.pragma_update(None, "user_version", STATE_SCHEMA_VERSION)?;
594                    }
595                    transaction.commit()?;
596                    Ok(())
597                })
598                .await;
599            match initialized {
600                Ok(()) => break,
601                Err(tokio_rusqlite::Error::Error(Error::Sql(rusqlite::Error::SqliteFailure(
602                    error,
603                    _,
604                )))) if error.code == rusqlite::ErrorCode::DatabaseBusy
605                    && tokio::time::Instant::now() < deadline =>
606                {
607                    // Concurrent first opens can race when enabling WAL despite busy_timeout.
608                    tokio::time::sleep(INITIALIZATION_RETRY_INTERVAL).await;
609                }
610                Err(error) => return Err(error.to_string()),
611            }
612        }
613        Ok(Self { connection })
614    }
615
616    /// This state's stable id ([`read_state_id`]).
617    pub async fn state_id(&self) -> Result<crate::types::StateId, String> {
618        self.connection
619            .call(|db| -> Result<crate::types::StateId, Error> {
620                Ok(
621                    db.query_row("SELECT value FROM state_meta WHERE name='id'", [], |row| {
622                        row.get(0)
623                    })?,
624                )
625            })
626            .await
627            .map_err(|e| e.to_string())
628    }
629
630    pub async fn create_run(&self, run: &Run, requests: &[Request]) -> Result<(), String> {
631        for request in requests {
632            if request.run_id != run.id {
633                return Err("Request belongs to another Run.".into());
634            }
635        }
636        let run = run.clone();
637        let requests = requests.to_vec();
638        self.connection
639            .call(move |db| -> Result<(), Error> {
640                let transaction = db.transaction()?;
641                transaction.execute(
642                    "INSERT INTO runs(id,repo,status,data) VALUES (?,?,?,?)",
643                    params![
644                        run.id,
645                        crate::platform::path_text(&run.repo_path),
646                        run.status(),
647                        serde_json::to_string(&run)?
648                    ],
649                )?;
650                for (ordinal, request) in requests.iter().enumerate() {
651                    transaction.execute(
652                        "INSERT INTO requests(id,run_id,eval_id,ordinal,status,data) VALUES (?,?,?,?,?,?)",
653                        params![
654                            request.id,
655                            run.id,
656                            request.eval_id,
657                            ordinal as i64,
658                            request.status(),
659                            serde_json::to_string(request)?
660                        ],
661                    )?;
662                }
663                transaction.commit()?;
664                Ok(())
665            })
666            .await
667            .map_err(|e| e.to_string())
668    }
669
670    pub async fn save_request(&self, request: &Request) -> Result<(), String> {
671        let request = request.clone();
672        self.connection
673            .call(move |db| update_request(db, &request))
674            .await
675            .map_err(|e| e.to_string())
676    }
677
678    pub async fn save_run(&self, run: &Run) -> Result<(), String> {
679        let run = run.clone();
680        self.connection
681            .call(move |db| -> Result<(), Error> {
682                db.execute(
683                    "UPDATE runs SET status=?,data=? WHERE id=?",
684                    params![run.status(), serde_json::to_string(&run)?, run.id],
685                )?;
686                Ok(())
687            })
688            .await
689            .map_err(|e| e.to_string())
690    }
691
692    pub async fn finish(&self, run: &Run, requests: &[Request]) -> Result<(), String> {
693        let run = run.clone();
694        let requests = requests.to_vec();
695        self.connection
696            .call(move |db| -> Result<(), Error> {
697                let transaction = db.transaction()?;
698                for request in requests {
699                    // A concurrent submission may already have settled this saved request.
700                    if request.status() == crate::types::RequestStatus::WaitingHuman {
701                        continue;
702                    }
703                    update_request(&transaction, &request)?;
704                }
705                transaction.execute(
706                    "UPDATE runs SET status=?,data=? WHERE id=?",
707                    params![run.status(), serde_json::to_string(&run)?, run.id],
708                )?;
709                transaction.commit()?;
710                Ok(())
711            })
712            .await
713            .map_err(|e| e.to_string())
714    }
715}
716
717pub(super) fn update_request(db: &rusqlite::Connection, request: &Request) -> Result<(), Error> {
718    if db.execute(
719        "UPDATE requests SET status=?,data=?,execution_id=? WHERE id=?",
720        params![
721            request.status(),
722            serde_json::to_string(request)?,
723            request.execution_id,
724            request.id
725        ],
726    )? != 1
727    {
728        return Err(Error::Invalid("Review request not found.".into()));
729    }
730    Ok(())
731}
732
733pub(super) fn schema_initialized(transaction: &rusqlite::Transaction<'_>) -> Result<bool, Error> {
734    let version: u32 = transaction.pragma_query_value(None, "user_version", |row| row.get(0))?;
735    // Check both in one snapshot: initialization may commit between separate reads.
736    if version == 0
737        && !transaction.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master)", [], |row| {
738            row.get::<_, bool>(0)
739        })?
740    {
741        return Ok(false);
742    }
743    if version != STATE_SCHEMA_VERSION {
744        return Err(Error::Invalid(schema_error(version)));
745    }
746    Ok(true)
747}
748
749/// Reject non-regular state files and a database of another schema before any read or write.
750pub(super) fn check_files(state: &Path) -> Result<(), String> {
751    match state_schema(state)? {
752        Some(version) if version != 0 && version != STATE_SCHEMA_VERSION => {
753            Err(schema_error(version))
754        }
755        _ => Ok(()),
756    }
757}
758
759/// The state database's schema version, read without creating it; `None` without a database.
760/// `doctor` reports it.
761pub fn state_schema(state: &Path) -> Result<Option<u32>, String> {
762    let state = canonical_target(state).map_err(|e| e.to_string())?;
763    regular_files(&state)?;
764    let database = state.join(DATABASE);
765    if !database.try_exists().map_err(|e| e.to_string())? {
766        return Ok(None);
767    }
768    let read = || -> Result<u32, rusqlite::Error> {
769        let db = rusqlite::Connection::open_with_flags(
770            &database,
771            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
772        )?;
773        db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
774        db.pragma_query_value(None, "user_version", |row| row.get(0))
775    };
776    read().map(Some).map_err(|e| e.to_string())
777}
778
779/// The state's stable id, read without creating anything; `None` until a command that writes
780/// the state has opened it.
781pub fn read_state_id(state: &Path) -> Result<Option<crate::types::StateId>, String> {
782    let state = canonical_target(state).map_err(|e| e.to_string())?;
783    check_files(&state)?;
784    let database = state.join(DATABASE);
785    if !database.try_exists().map_err(|e| e.to_string())? {
786        return Ok(None);
787    }
788    let read = || -> Result<Option<crate::types::StateId>, Error> {
789        let mut db = rusqlite::Connection::open_with_flags(
790            &database,
791            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
792        )?;
793        db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
794        let transaction = db.transaction()?;
795        if !schema_initialized(&transaction)? {
796            return Ok(None);
797        }
798        Ok(transaction
799            .query_row("SELECT value FROM state_meta WHERE name='id'", [], |row| {
800                row.get(0)
801            })
802            .optional()?)
803    };
804    read().map_err(|e| e.to_string())
805}
806
807pub(crate) fn regular_files(state: &Path) -> Result<(), String> {
808    regular_files_with_retry(state, || {
809        std::thread::sleep(super::TRANSIENT_ACCESS_BACKOFF);
810    })
811}
812
813/// The probe and retry path is shared with the seam test; only the between-attempt effect
814/// differs, so a fixture can release a pending-delete handle exactly after a failed probe.
815pub(super) fn regular_files_with_retry(
816    state: &Path,
817    mut before_retry: impl FnMut(),
818) -> Result<(), String> {
819    for suffix in ["", "-wal", "-shm"] {
820        let path = state.join(format!("{DATABASE}{suffix}"));
821        match super::retry_transient_access(|| crate::platform::path_kind(&path), &mut before_retry)
822        {
823            Ok(kind) if kind != crate::platform::FileKind::File => {
824                return Err(format!(
825                    "State files must be regular files: {}",
826                    crate::platform::path_text(&path)
827                ));
828            }
829            Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
830                return Err(error.to_string());
831            }
832            _ => {}
833        }
834    }
835    Ok(())
836}
837
838#[derive(Debug, Serialize)]
839#[serde(rename_all = "camelCase")]
840pub struct LastRequest {
841    pub run_id: RunId,
842    pub verdict: RequestStatus,
843    #[serde(skip_serializing_if = "Option::is_none")]
844    pub fingerprint: Option<Fingerprint>,
845}
846
847/// Last attempts are audit pointers, never evidence for a new current-input query.
848pub async fn read_latest_requests(
849    state: &Path,
850    repo: &Path,
851) -> Result<std::collections::BTreeMap<crate::types::EvalId, LastRequest>, String> {
852    let state = canonical_target(state).map_err(|e| e.to_string())?;
853    outside_workspace(repo, &state).map_err(|e| e.to_string())?;
854    check_files(&state)?;
855    if !state
856        .join(DATABASE)
857        .try_exists()
858        .map_err(|e| e.to_string())?
859    {
860        return Ok(Default::default());
861    }
862    let connection = Connection::open_with_flags(
863        state.join(DATABASE),
864        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
865    )
866    .await
867    .map_err(|e| e.to_string())?;
868    let repo = repo.to_path_buf();
869    connection
870        .call(move |db| -> Result<_, Error> {
871            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
872            let transaction = db.transaction()?;
873            if !schema_initialized(&transaction)? {
874                return Ok(Default::default());
875            }
876            let latest = {
877                let mut statement = transaction.prepare(
878                    "SELECT data,run_data FROM (
879                SELECT q.eval_id,q.data,r.data AS run_data,
880                    row_number() OVER (PARTITION BY q.eval_id ORDER BY r.rowid DESC) AS rank
881                FROM requests q JOIN runs r ON r.id=q.run_id
882                WHERE r.repo=?
883            ) WHERE rank=1 ORDER BY eval_id",
884                )?;
885                statement
886                    .query_map([crate::platform::path_text(&repo)], |row| {
887                        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
888                    })?
889                    .filter_map(|row| match row {
890                        Err(error) => Some(Err(error)),
891                        Ok((data, run_data)) => {
892                            super::unreadable::evidence::<Run>("run", &run_data)
893                                .and_then(|_| {
894                                    super::unreadable::evidence::<Request>("request", &data)
895                                })
896                                .map(|request| {
897                                    Ok((
898                                        request.eval_id.clone(),
899                                        LastRequest {
900                                            run_id: request.run_id.clone(),
901                                            verdict: request.status(),
902                                            fingerprint: request.fingerprint,
903                                        },
904                                    ))
905                                })
906                        }
907                    })
908                    .collect::<Result<_, _>>()?
909            };
910            transaction.commit()?;
911            Ok(latest)
912        })
913        .await
914        .map_err(|e| e.to_string())
915}
916
917/// Existing store only: no discovery, schema creation, or execution.
918pub async fn read_run(state: &Path, id: &str) -> Result<RunView, String> {
919    let state = canonical_target(state).map_err(|e| e.to_string())?;
920    check_files(&state)?;
921    let connection = Connection::open_with_flags(
922        state.join(DATABASE),
923        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
924    )
925    .await
926    .map_err(|e| e.to_string())?;
927    let id: RunId = id.parse()?;
928    connection
929        .call(move |db| -> Result<RunView, Error> {
930            db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)?;
931            let transaction = db.transaction()?;
932            if !schema_initialized(&transaction)? {
933                return Err(Error::Invalid("Run not found.".into()));
934            }
935            let saved: Option<(String, String)> = transaction
936                .query_row("SELECT repo,data FROM runs WHERE id=?", [&id], |row| {
937                    Ok((row.get(0)?, row.get(1)?))
938                })
939                .optional()?;
940            let (repo, data) = saved.ok_or_else(|| Error::Invalid("Run not found.".into()))?;
941            outside_workspace(Path::new(&repo), &state)
942                .map_err(|e| Error::Invalid(e.to_string()))?;
943            let run = super::unreadable::decode("run", &id, &data)
944                .map_err(|error| Error::Invalid(error.to_string()))?;
945            let requests = {
946                let mut statement = transaction
947                    .prepare("SELECT data FROM requests WHERE run_id=? ORDER BY ordinal")?;
948                statement
949                    .query_map([&id], |row| row.get::<_, String>(0))?
950                    .filter_map(|row| match row {
951                        Ok(data) => super::unreadable::evidence("request", &data).map(Ok),
952                        Err(error) => Some(Err(Error::Sql(error))),
953                    })
954                    .collect::<Result<Vec<Request>, Error>>()?
955            };
956            let unreadable = super::unreadable::in_run(&transaction, &id)?;
957            transaction.commit()?;
958            Ok(RunView {
959                run,
960                requests,
961                unreadable,
962            })
963        })
964        .await
965        .map_err(|e| e.to_string())
966}
artifactize_core/store/receipts.rs

artifactize_core/store/
receipts.rs
