diff --git a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs index afd616500..8981049be 100644 --- a/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs +++ b/lib/apps/fabro-server/migrations/2026082301_sqlite_blob_activation.rs @@ -125,12 +125,14 @@ pub(crate) async fn activate_blob_storage( ); let blob_store = Arc::new(fabro_store::BlobStore::new(database.clone_pool())); + let run_record_store = Arc::new(fabro_store::RunRecordStore::new(database.clone_pool())); let store = Arc::new(fabro_store::Database::new( object_store, slatedb_prefix, flush_interval, cache_path, Arc::clone(&blob_store), + run_record_store, )); let inventory = store diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index 833082a2d..5c1afc8fd 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -829,7 +829,7 @@ where .runs .warm_projection_cache() .await - .context("warming run projection cache and reconciling run summaries")?; + .context("warming run projection cache and reconciling run records")?; let reconciled = reconcile_incomplete_runs_on_startup(&state).await?; if reconciled > 0 { info!( diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index a9f58b3d3..06d3083de 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -86,7 +86,7 @@ use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, CachedRunProjection, Database, - EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, + EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunRecordStore, StageArtifactEntry, StageId, }; #[cfg(test)] @@ -1153,7 +1153,7 @@ pub struct AppState { pub(crate) struct AppStores { pub(crate) runs: Arc, - pub(crate) run_summaries: Arc, + pub(crate) run_records: Arc, pub(crate) auth_codes: Arc, pub(crate) auth_sessions: Arc, pub(crate) automations: Arc, @@ -2447,8 +2447,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result anyhow::Result, ) -> fabro_store::Result> { - let Some(mut summary) = state.stores.run_summaries.get(run_id, now).await? else { + let Some(mut summary) = state.stores.run_records.get(run_id, now).await? else { return Ok(None); }; if summary.timestamps.completed_at.is_none() { @@ -361,7 +361,7 @@ pub(super) async fn run_summary_page_response( state: &AppState, query: &RunSummaryListQuery, ) -> Response { - match state.stores.run_summaries.list(query, Utc::now()).await { + match state.stores.run_records.list(query, Utc::now()).await { Ok(page) => { let data = state.decorate_run_summaries(page.data).await; ( @@ -415,7 +415,7 @@ async fn resolve_run( State(state): State>, Query(query): Query, ) -> Response { - let identities = match state.stores.run_summaries.list_identities().await { + let identities = match state.stores.run_records.list_identities().await { Ok(identities) => identities, Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 68c01bdf1..10666634c 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -210,6 +210,15 @@ methods = ["dev-token"] ) } +#[test] +fn run_record_store_is_database_owned() { + let state = test_app_state(); + assert!(Arc::ptr_eq( + &state.stores.run_records, + &state.stores.runs.run_record_store(), + )); +} + async fn body_json(body: Body) -> serde_json::Value { let bytes = to_bytes(body, usize::MAX).await.unwrap(); serde_json::from_slice(&bytes).unwrap() @@ -3915,7 +3924,7 @@ async fn post_runs_run_intent_rejects_invalid_folder_paths_before_persistence() assert!( state .stores - .run_summaries + .run_records .list_identities() .await .unwrap() @@ -3947,7 +3956,7 @@ async fn post_runs_run_intent_applies_the_folder_target_environment_matrix() { assert!( state .stores - .run_summaries + .run_records .list_identities() .await .unwrap() @@ -3982,7 +3991,7 @@ enabled = false assert!( disabled_state .stores - .run_summaries + .run_records .list_identities() .await .unwrap() @@ -4224,7 +4233,7 @@ async fn post_runs_run_intent_rejects_none_target_with_local_environment_before_ assert!( state .stores - .run_summaries + .run_records .list_identities() .await .unwrap() @@ -4269,7 +4278,7 @@ async fn assert_run_intent_targets_unavailable(state: &Arc) { assert!( state .stores - .run_summaries + .run_records .list_identities() .await .unwrap() diff --git a/lib/components/fabro-store/src/error.rs b/lib/components/fabro-store/src/error.rs index c75784716..6251954cf 100644 --- a/lib/components/fabro-store/src/error.rs +++ b/lib/components/fabro-store/src/error.rs @@ -56,6 +56,18 @@ pub enum Error { run_id: String, field: &'static str, }, + #[error("run {run_id} head mismatch: expected {expected_last_seq}, stored {actual_last_seq:?}")] + RunHeadMismatch { + run_id: String, + expected_last_seq: u32, + actual_last_seq: Option, + }, + #[error("stored run event {run_id} sequence {seq} has inconsistent field {field}")] + RunEventMismatch { + run_id: String, + seq: u32, + field: &'static str, + }, #[error(transparent)] InvalidTransition(#[from] fabro_types::InvalidTransition), #[error("{0}")] diff --git a/lib/components/fabro-store/src/legacy_blob_import.rs b/lib/components/fabro-store/src/legacy_blob_import.rs index cc4739db7..69b04fcff 100644 --- a/lib/components/fabro-store/src/legacy_blob_import.rs +++ b/lib/components/fabro-store/src/legacy_blob_import.rs @@ -1128,7 +1128,7 @@ mod tests { PASSIVE_CHECKPOINT_BYTES, set_automatic_checkpoint, }; use crate::keys::SlateKey; - use crate::{BlobStore, Database}; + use crate::{BlobStore, Database, test_support as store_test_support}; type TestResult = std::result::Result>; @@ -1153,6 +1153,7 @@ mod tests { Duration::from_millis(1), None, Arc::clone(&target), + store_test_support::test_run_record_store(), ); let source_db = source.open_db().await?; Ok(Self { @@ -1853,6 +1854,7 @@ mod tests { Duration::from_millis(1), None, Arc::clone(&target), + store_test_support::test_run_record_store(), ); let mut connection = pool.acquire().await?; diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 3fd34521d..fe2078ed6 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -9,9 +9,9 @@ mod keyed_mutex; mod keys; mod legacy_blob_import; mod record; +mod run_record_store; mod run_sessions; mod run_state; -mod run_summary_store; mod serializable_projection; mod slate; mod sqlite_row; @@ -37,15 +37,15 @@ pub use legacy_blob_import::{ LegacyBlobImportError, LegacyBlobImportReport, LegacyBlobInventory, LegacyBlobInventoryError, LegacyBlobVerificationError, LegacyBlobVerificationReport, }; +pub use run_record_store::{ + RunRecordStore, RunSummaryIdentity, RunSummaryListQuery, RunSummaryPage, RunSummarySort, + RunSummarySortDirection, RunSummaryVisibility, +}; pub use run_sessions::{ ProjectedRunSession, project_run_session, project_run_session_with_context, project_run_sessions, }; pub use run_state::RunProjectionReducer; -pub use run_summary_store::{ - RunSummaryIdentity, RunSummaryListQuery, RunSummaryPage, RunSummarySort, - RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility, -}; pub use serializable_projection::SerializableProjection; pub use slate::{CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs, UnreadableRun}; pub use types::EventPayload; diff --git a/lib/components/fabro-store/src/run_record_store.rs b/lib/components/fabro-store/src/run_record_store.rs new file mode 100644 index 000000000..3068185b7 --- /dev/null +++ b/lib/components/fabro-store/src/run_record_store.rs @@ -0,0 +1,1978 @@ +use std::collections::{HashMap, HashSet}; +use std::fmt::Write as _; +use std::sync::LazyLock; + +use chrono::{DateTime, Utc}; +use fabro_types::{ + BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming, + SessionId, StageId, timing, +}; +use sqlx::sqlite::{SqliteConnection, SqliteRow}; +use sqlx::{QueryBuilder, Row as _, Sqlite, SqlitePool}; +use strum::VariantArray as _; + +use crate::run_state::projected_billing; +use crate::slate::CachedRunProjection; +use crate::{Error, EventPayload, Result, keys}; + +const INSERT_RUN_SQL: &str = r" +INSERT INTO runs ( + id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms, + status, archived_at_ms, parent_id, title, workflow_slug, workflow_name, + repository_name, automation_id, diff_files_changed, diff_additions, diff_deletions, + input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens, + total_usd_micros, summary_json +) VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? +) +"; + +const UPSERT_RUN_SQL: &str = r" +INSERT INTO runs ( + id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms, + status, archived_at_ms, parent_id, title, workflow_slug, workflow_name, + repository_name, automation_id, diff_files_changed, diff_additions, diff_deletions, + input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens, + total_usd_micros, summary_json +) VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? +) +ON CONFLICT(id) DO UPDATE SET + source_last_seq = excluded.source_last_seq, + created_at_ms = excluded.created_at_ms, + started_at_ms = excluded.started_at_ms, + last_event_at_ms = excluded.last_event_at_ms, + completed_at_ms = excluded.completed_at_ms, + status = excluded.status, + archived_at_ms = excluded.archived_at_ms, + parent_id = excluded.parent_id, + title = excluded.title, + workflow_slug = excluded.workflow_slug, + workflow_name = excluded.workflow_name, + repository_name = excluded.repository_name, + automation_id = excluded.automation_id, + diff_files_changed = excluded.diff_files_changed, + diff_additions = excluded.diff_additions, + diff_deletions = excluded.diff_deletions, + input_tokens = excluded.input_tokens, + output_tokens = excluded.output_tokens, + reasoning_tokens = excluded.reasoning_tokens, + cache_read_tokens = excluded.cache_read_tokens, + cache_write_tokens = excluded.cache_write_tokens, + total_usd_micros = excluded.total_usd_micros, + summary_json = excluded.summary_json +WHERE excluded.source_last_seq > runs.source_last_seq +"; + +const UPDATE_RUN_SQL: &str = r" +UPDATE runs SET + source_last_seq = ?, + created_at_ms = ?, + started_at_ms = ?, + last_event_at_ms = ?, + completed_at_ms = ?, + status = ?, + archived_at_ms = ?, + parent_id = ?, + title = ?, + workflow_slug = ?, + workflow_name = ?, + repository_name = ?, + automation_id = ?, + diff_files_changed = ?, + diff_additions = ?, + diff_deletions = ?, + input_tokens = ?, + output_tokens = ?, + reasoning_tokens = ?, + cache_read_tokens = ?, + cache_write_tokens = ?, + total_usd_micros = ?, + summary_json = ? +WHERE id = ? AND source_last_seq = ? +"; + +const SELECT_EVENT_COLUMNS: &str = + "SELECT run_id, seq, event_name, node_id, stage_id, session_id, event_json FROM run_events"; + +const SELECT_RUN_SUMMARIES_SQL: &str = r" +SELECT runs.id, runs.summary_json, + (SELECT COUNT(*) FROM runs AS child WHERE child.parent_id = runs.id) AS children_count +FROM runs"; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunSummarySort { + #[default] + CreatedAt, + UpdatedAt, + Status, + Elapsed, + #[serde(rename = "repo")] + Repository, + Title, + Workflow, + Changes, + Size, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum RunSummarySortDirection { + Asc, + #[default] + Desc, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RunSummaryVisibility { + All, + Default { + include_archived: bool, + }, + Selected { + statuses: Vec, + archived: bool, + }, +} + +impl Default for RunSummaryVisibility { + fn default() -> Self { + Self::Default { + include_archived: false, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RunSummaryListQuery { + pub parent_id: Option, + pub automation_id: Option, + pub visibility: RunSummaryVisibility, + pub sort: RunSummarySort, + pub direction: RunSummarySortDirection, + pub limit: u32, + pub offset: u32, +} + +impl Default for RunSummaryListQuery { + fn default() -> Self { + Self { + parent_id: None, + automation_id: None, + visibility: RunSummaryVisibility::default(), + sort: RunSummarySort::default(), + direction: RunSummarySortDirection::default(), + limit: 100, + offset: 0, + } + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RunSummaryPage { + pub data: Vec, + pub total: u64, + pub has_more: bool, +} + +#[derive(Clone)] +pub struct RunRecordStore { + pool: SqlitePool, +} + +impl std::fmt::Debug for RunRecordStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RunRecordStore").finish_non_exhaustive() + } +} + +impl RunRecordStore { + #[must_use] + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> { + let record = PreparedRunSummary::from_entry(entry); + let mut connection = self.pool.acquire().await?; + upsert_run_on_connection(&mut connection, &record).await?; + Ok(()) + } + + #[cfg(test)] + pub(crate) async fn close_pool(&self) { + self.pool.close().await; + } + + pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> { + let mut transaction = self.pool.begin().await?; + let stored_seqs: HashMap = + sqlx::query_as::<_, (String, i64)>("SELECT id, source_last_seq FROM runs") + .fetch_all(&mut *transaction) + .await? + .into_iter() + .collect(); + + let mut authoritative_ids = HashSet::new(); + for entry in entries { + let run_id = entry.run_id.to_string(); + let up_to_date = stored_seqs + .get(&run_id) + .is_some_and(|stored_seq| *stored_seq >= i64::from(entry.last_seq)); + authoritative_ids.insert(run_id); + if up_to_date { + continue; + } + upsert_run_on_connection(&mut transaction, &PreparedRunSummary::from_entry(entry)) + .await?; + } + + let stale_ids = stored_seqs + .keys() + .filter(|stored_id| !authoritative_ids.contains(stored_id.as_str())) + .collect::>(); + for chunk in stale_ids.chunks(500) { + let mut delete = QueryBuilder::::new("DELETE FROM runs WHERE id IN ("); + let mut separated = delete.separated(", "); + for stale_id in chunk { + separated.push_bind(stale_id.as_str()); + } + delete.push(")"); + delete.build().execute(&mut *transaction).await?; + } + transaction.commit().await?; + Ok(()) + } + + pub async fn get(&self, run_id: &RunId, now: DateTime) -> Result> { + let mut query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); + query + .push(" WHERE runs.id = ") + .push_bind(run_id.to_string()); + let row = query.build().fetch_optional(&self.pool).await?; + row.map(|row| decode_run_row(&row, now)).transpose() + } + + /// Identity fields for every stored run, for selector resolution without + /// decoding full summaries. + pub async fn list_identities(&self) -> Result> { + let rows = sqlx::query( + r" +SELECT id, workflow_slug, + json_extract(summary_json, '$.workflow.name') AS workflow_name, + json_extract(summary_json, '$.repository.origin_url') AS repository_origin_url +FROM runs", + ) + .fetch_all(&self.pool) + .await?; + rows.iter() + .map(|row| { + let stored_id: String = row.try_get("id")?; + let id = stored_id + .parse::() + .map_err(|_| Error::RunSummaryMismatch { + run_id: stored_id, + field: "id", + })?; + Ok(RunSummaryIdentity { + id, + workflow_slug: row.try_get("workflow_slug")?, + workflow_name: row.try_get("workflow_name")?, + repository_origin_url: row.try_get("repository_origin_url")?, + }) + }) + .collect() + } + + pub async fn list( + &self, + query: &RunSummaryListQuery, + now: DateTime, + ) -> Result { + let mut transaction = self.pool.begin().await?; + + let mut count_query = QueryBuilder::::new("SELECT COUNT(*) FROM runs"); + push_filters(&mut count_query, query); + let total: i64 = count_query + .build_query_scalar() + .fetch_one(&mut *transaction) + .await?; + + let mut rows_query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); + push_filters(&mut rows_query, query); + push_order(&mut rows_query, query.sort, query.direction, now); + rows_query.push(" LIMIT ").push_bind(i64::from(query.limit)); + rows_query + .push(" OFFSET ") + .push_bind(i64::from(query.offset)); + let rows = rows_query.build().fetch_all(&mut *transaction).await?; + transaction.commit().await?; + + let data = rows + .iter() + .map(|row| decode_run_row(row, now)) + .collect::>>()?; + let total = u64::try_from(total).expect("COUNT(*) is non-negative"); + let consumed = u64::from(query.offset).saturating_add(data.len() as u64); + Ok(RunSummaryPage { + data, + total, + has_more: consumed < total, + }) + } + + pub async fn delete(&self, run_id: &RunId) -> Result<()> { + sqlx::query("DELETE FROM runs WHERE id = ?") + .bind(run_id.to_string()) + .execute(&self.pool) + .await?; + Ok(()) + } +} + +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "the atomic SQL run path stays inactive until the authority cutover" + ) +)] +impl RunRecordStore { + pub(crate) async fn insert_first_event_on_connection( + connection: &mut SqliteConnection, + entry: &CachedRunProjection, + payload: &EventPayload, + ) -> Result { + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, 1)?; + ensure_prepared_head(&record, 1)?; + let envelope = validate_event_for_record(&record, payload, 1)?; + if envelope.event.event_name() != "run.created" { + return Err(run_event_mismatch(&record.run.id, 1, "event_name")); + } + + insert_run_on_connection(connection, &record).await?; + insert_event_on_connection(connection, &record, payload, &envelope).await?; + Ok(envelope) + } + + pub(crate) async fn append_event_on_connection( + connection: &mut SqliteConnection, + expected_last_seq: u32, + entry: &CachedRunProjection, + payload: &EventPayload, + ) -> Result { + let next_seq = expected_last_seq + .checked_add(1) + .filter(|seq| *seq <= keys::MAX_EVENT_SEQ) + .ok_or(Error::EventSequenceExhausted { + max_seq: keys::MAX_EVENT_SEQ, + })?; + let record = PreparedRunSummary::from_entry(entry); + ensure_entry_identity(entry, &record, next_seq)?; + ensure_prepared_head(&record, next_seq)?; + let envelope = validate_event_for_record(&record, payload, next_seq)?; + + update_run_on_connection(connection, &record, expected_last_seq).await?; + insert_event_on_connection(connection, &record, payload, &envelope).await?; + Ok(envelope) + } + + pub(crate) async fn list_events_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" ORDER BY seq ASC"); + let rows = query.build().fetch_all(&mut *connection).await?; + let events = decode_event_rows(&rows, run_id)?; + let expected_last_seq = select_run_head(connection, run_id) + .await? + .ok_or_else(|| Error::RunNotFound(run_id.to_string()))?; + let actual_last_seq = events.last().map(|event| event.seq); + if actual_last_seq != Some(expected_last_seq) { + return Err(Error::RunHeadMismatch { + run_id: run_id.to_string(), + expected_last_seq, + actual_last_seq, + }); + } + Ok(events) + } + + pub(crate) async fn list_events_from_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + start_seq: u32, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)) + .push(" ORDER BY seq ASC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn list_events_before_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + before_seq: Option, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query.push(" WHERE run_id = ").push_bind(run_id.to_string()); + if let Some(before_seq) = before_seq { + query.push(" AND seq < ").push_bind(i64::from(before_seq)); + } + query + .push(" ORDER BY seq DESC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn get_event_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + seq: u32, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq = ") + .push_bind(i64::from(seq)); + let row = query.build().fetch_optional(&mut *connection).await?; + row.as_ref() + .map(|row| decode_event_row(row, run_id)) + .transpose() + } + + pub(crate) async fn list_events_for_stage_from_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + stage_id: &StageId, + start_seq: u32, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)); + if stage_id.visit() == 1 { + query + .push(" AND (stage_id = ") + .push_bind(stage_id.to_string()) + .push(" OR (stage_id IS NULL AND node_id = ") + .push_bind(stage_id.node_id().to_string()) + .push("))"); + } else { + query + .push(" AND stage_id = ") + .push_bind(stage_id.to_string()); + } + query + .push(" ORDER BY seq ASC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } + + pub(crate) async fn list_events_for_session_from_with_limit_on_connection( + connection: &mut SqliteConnection, + run_id: &RunId, + session_id: &SessionId, + start_seq: u32, + limit: usize, + ) -> Result> { + let mut query = QueryBuilder::::new(SELECT_EVENT_COLUMNS); + query + .push(" WHERE run_id = ") + .push_bind(run_id.to_string()) + .push(" AND seq >= ") + .push_bind(i64::from(start_seq)) + .push(" AND session_id = ") + .push_bind(session_id.to_string()) + .push(" AND event_name GLOB 'run.session.*' ORDER BY seq ASC LIMIT ") + .push_bind(sql_limit(limit)); + let rows = query.build().fetch_all(&mut *connection).await?; + decode_event_rows(&rows, run_id) + } +} + +/// Identity fields of a stored run summary, cheap to list for selector +/// resolution. +#[derive(Debug, Clone)] +pub struct RunSummaryIdentity { + pub id: RunId, + pub workflow_slug: Option, + pub workflow_name: Option, + pub repository_origin_url: Option, +} + +#[derive(Debug)] +struct PreparedRunSummary { + run: Run, + last_seq: u32, + workflow_name: Option, + repository_name: Option, + input_tokens: i64, + output_tokens: i64, + reasoning_tokens: i64, + cache_read_tokens: i64, + cache_write_tokens: i64, + total_usd_micros: Option, +} + +impl PreparedRunSummary { + fn from_entry(entry: &CachedRunProjection) -> Self { + let mut run = entry.summary.clone(); + if run.timing.is_none() { + let at = run + .timestamps + .last_event_at + .unwrap_or(run.timestamps.created_at); + run.timing = entry.projection.live_run_timing(at); + } + let billing = normalize_billing_for_read_model(projected_billing(&entry.projection)); + let workflow_name = run.workflow.display_name().map(str::to_string); + let repository_name = run + .repository + .as_ref() + .map(|repository| repository.name.clone()); + + Self { + run, + last_seq: entry.last_seq, + workflow_name, + repository_name, + input_tokens: billing.input_tokens, + output_tokens: billing.output_tokens, + reasoning_tokens: billing.reasoning_tokens, + cache_read_tokens: billing.cache_read_tokens, + cache_write_tokens: billing.cache_write_tokens, + total_usd_micros: billing.total_usd_micros, + } + } +} + +fn ensure_entry_identity( + entry: &CachedRunProjection, + record: &PreparedRunSummary, + seq: u32, +) -> Result<()> { + if entry.run_id != record.run.id || entry.projection.spec.run_id != entry.run_id { + return Err(run_event_mismatch(&entry.run_id, seq, "run_id")); + } + Ok(()) +} + +fn ensure_prepared_head(record: &PreparedRunSummary, expected_last_seq: u32) -> Result<()> { + if record.last_seq != expected_last_seq { + return Err(Error::RunHeadMismatch { + run_id: record.run.id.to_string(), + expected_last_seq, + actual_last_seq: Some(record.last_seq), + }); + } + Ok(()) +} + +fn validate_event_for_record( + record: &PreparedRunSummary, + payload: &EventPayload, + seq: u32, +) -> Result { + if seq == 0 || seq > keys::MAX_EVENT_SEQ { + return Err(run_event_mismatch(&record.run.id, seq, "seq")); + } + payload.validate(&record.run.id)?; + let event = RunEvent::try_from(payload)?; + if event.run_id != record.run.id { + return Err(run_event_mismatch(&record.run.id, seq, "run_id")); + } + Ok(EventEnvelope { seq, event }) +} + +fn run_event_mismatch(run_id: &RunId, seq: u32, field: &'static str) -> Error { + Error::RunEventMismatch { + run_id: run_id.to_string(), + seq, + field, + } +} + +async fn insert_event_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, + payload: &EventPayload, + envelope: &EventEnvelope, +) -> Result<()> { + ensure_prepared_head(record, envelope.seq)?; + let event_json = serde_json::to_string(payload)?; + sqlx::query( + r" +INSERT INTO run_events (run_id, seq, event_name, node_id, stage_id, session_id, event_json) +VALUES (?, ?, ?, ?, ?, ?, ?) +", + ) + .bind(record.run.id.to_string()) + .bind(i64::from(envelope.seq)) + .bind(envelope.event.event_name()) + .bind(envelope.event.node_id.as_deref()) + .bind(envelope.event.stage_id.as_ref().map(ToString::to_string)) + .bind(envelope.event.session_id.as_deref()) + .bind(event_json) + .execute(connection) + .await?; + Ok(()) +} + +fn sql_limit(limit: usize) -> i64 { + i64::try_from(limit.saturating_add(1)).unwrap_or(i64::MAX) +} + +fn decode_event_rows(rows: &[SqliteRow], run_id: &RunId) -> Result> { + rows.iter() + .map(|row| decode_event_row(row, run_id)) + .collect() +} + +fn decode_event_row(row: &SqliteRow, expected_run_id: &RunId) -> Result { + let stored_run_id: String = row.try_get("run_id")?; + let stored_seq: i64 = row.try_get("seq")?; + let seq = u32::try_from(stored_seq) + .ok() + .filter(|seq| (1..=keys::MAX_EVENT_SEQ).contains(seq)) + .ok_or_else(|| run_event_mismatch(expected_run_id, 0, "seq"))?; + if stored_run_id != expected_run_id.to_string() { + return Err(run_event_mismatch(expected_run_id, seq, "run_id")); + } + + let event_json: String = row.try_get("event_json")?; + let payload: EventPayload = serde_json::from_str(&event_json)?; + let event = RunEvent::try_from(&payload)?; + if event.run_id != *expected_run_id { + return Err(run_event_mismatch(expected_run_id, seq, "run_id")); + } + + let stored_event_name: String = row.try_get("event_name")?; + if stored_event_name != event.event_name() { + return Err(run_event_mismatch(expected_run_id, seq, "event_name")); + } + let stored_node_id: Option = row.try_get("node_id")?; + if stored_node_id.as_deref() != event.node_id.as_deref() { + return Err(run_event_mismatch(expected_run_id, seq, "node_id")); + } + let stored_stage_id: Option = row.try_get("stage_id")?; + let decoded_stage_id = event.stage_id.as_ref().map(ToString::to_string); + if stored_stage_id != decoded_stage_id { + return Err(run_event_mismatch(expected_run_id, seq, "stage_id")); + } + let stored_session_id: Option = row.try_get("session_id")?; + if stored_session_id.as_deref() != event.session_id.as_deref() { + return Err(run_event_mismatch(expected_run_id, seq, "session_id")); + } + + Ok(EventEnvelope { seq, event }) +} + +async fn select_run_head(connection: &mut SqliteConnection, run_id: &RunId) -> Result> { + let stored: Option = sqlx::query_scalar("SELECT source_last_seq FROM runs WHERE id = ?") + .bind(run_id.to_string()) + .fetch_optional(connection) + .await?; + stored + .map(|value| { + u32::try_from(value) + .ok() + .filter(|seq| (1..=keys::MAX_EVENT_SEQ).contains(seq)) + .ok_or_else(|| run_event_mismatch(run_id, 0, "source_last_seq")) + }) + .transpose() +} + +/// Older provider codecs could persist a negative disjoint bucket when a +/// detail count exceeded its inclusive parent total. The SQLite summary is a +/// rebuildable, nonnegative read model, so normalize those legacy values here +/// without rewriting the authoritative run events. +fn normalize_billing_for_read_model(mut billing: BilledTokenCounts) -> BilledTokenCounts { + let input_total = billing + .input_tokens + .saturating_add(billing.cache_read_tokens) + .saturating_add(billing.cache_write_tokens) + .max(0); + billing.cache_read_tokens = billing.cache_read_tokens.clamp(0, input_total); + billing.cache_write_tokens = billing + .cache_write_tokens + .clamp(0, input_total - billing.cache_read_tokens); + billing.input_tokens = input_total - billing.cache_read_tokens - billing.cache_write_tokens; + + let output_total = billing + .output_tokens + .saturating_add(billing.reasoning_tokens) + .max(0); + billing.reasoning_tokens = billing.reasoning_tokens.clamp(0, output_total); + billing.output_tokens = output_total - billing.reasoning_tokens; + billing.total_tokens = input_total.saturating_add(output_total); + billing.total_usd_micros = billing.total_usd_micros.map(|value| value.max(0)); + billing +} + +async fn upsert_run_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, +) -> Result<()> { + write_insert_shaped_run(connection, record, UPSERT_RUN_SQL).await +} + +async fn insert_run_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, +) -> Result<()> { + ensure_prepared_head(record, 1)?; + write_insert_shaped_run(connection, record, INSERT_RUN_SQL).await +} + +async fn write_insert_shaped_run( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, + sql: &'static str, +) -> Result<()> { + let run = &record.run; + let diff = run.diff.unwrap_or_default(); + let summary_json = serde_json::to_string(run)?; + sqlx::query(sql) + .bind(run.id.to_string()) + .bind(i64::from(record.last_seq)) + .bind(run.timestamps.created_at.timestamp_millis()) + .bind( + run.timestamps + .started_at + .map(|value| value.timestamp_millis()), + ) + .bind( + run.timestamps + .last_event_at + .unwrap_or(run.timestamps.created_at) + .timestamp_millis(), + ) + .bind( + run.timestamps + .completed_at + .map(|value| value.timestamp_millis()), + ) + .bind(run.lifecycle.status.kind().to_string()) + .bind( + run.lifecycle + .archived_at + .map(|value| value.timestamp_millis()), + ) + .bind(run.parent_id.map(|value| value.to_string())) + .bind(&run.title) + .bind(&run.workflow.slug) + .bind(&record.workflow_name) + .bind(&record.repository_name) + .bind(run.automation.as_ref().map(|automation| &automation.id)) + .bind(diff.files_changed) + .bind(diff.additions) + .bind(diff.deletions) + .bind(record.input_tokens) + .bind(record.output_tokens) + .bind(record.reasoning_tokens) + .bind(record.cache_read_tokens) + .bind(record.cache_write_tokens) + .bind(record.total_usd_micros) + .bind(summary_json) + .execute(connection) + .await?; + Ok(()) +} + +async fn update_run_on_connection( + connection: &mut SqliteConnection, + record: &PreparedRunSummary, + expected_last_seq: u32, +) -> Result<()> { + let next_seq = expected_last_seq + .checked_add(1) + .filter(|seq| *seq <= keys::MAX_EVENT_SEQ) + .ok_or(Error::EventSequenceExhausted { + max_seq: keys::MAX_EVENT_SEQ, + })?; + ensure_prepared_head(record, next_seq)?; + + let run = &record.run; + let diff = run.diff.unwrap_or_default(); + let summary_json = serde_json::to_string(run)?; + let result = sqlx::query(UPDATE_RUN_SQL) + .bind(i64::from(record.last_seq)) + .bind(run.timestamps.created_at.timestamp_millis()) + .bind( + run.timestamps + .started_at + .map(|value| value.timestamp_millis()), + ) + .bind( + run.timestamps + .last_event_at + .unwrap_or(run.timestamps.created_at) + .timestamp_millis(), + ) + .bind( + run.timestamps + .completed_at + .map(|value| value.timestamp_millis()), + ) + .bind(run.lifecycle.status.kind().to_string()) + .bind( + run.lifecycle + .archived_at + .map(|value| value.timestamp_millis()), + ) + .bind(run.parent_id.map(|value| value.to_string())) + .bind(&run.title) + .bind(&run.workflow.slug) + .bind(&record.workflow_name) + .bind(&record.repository_name) + .bind(run.automation.as_ref().map(|automation| &automation.id)) + .bind(diff.files_changed) + .bind(diff.additions) + .bind(diff.deletions) + .bind(record.input_tokens) + .bind(record.output_tokens) + .bind(record.reasoning_tokens) + .bind(record.cache_read_tokens) + .bind(record.cache_write_tokens) + .bind(record.total_usd_micros) + .bind(summary_json) + .bind(run.id.to_string()) + .bind(i64::from(expected_last_seq)) + .execute(&mut *connection) + .await?; + if result.rows_affected() == 0 { + return Err(Error::RunHeadMismatch { + run_id: run.id.to_string(), + expected_last_seq, + actual_last_seq: select_run_head(connection, &run.id).await?, + }); + } + Ok(()) +} + +fn push_filters(builder: &mut QueryBuilder, query: &RunSummaryListQuery) { + builder.push(" WHERE 1 = 1"); + if let Some(parent_id) = query.parent_id { + builder + .push(" AND parent_id = ") + .push_bind(parent_id.to_string()); + } + if let Some(automation_id) = &query.automation_id { + builder + .push(" AND automation_id = ") + .push_bind(automation_id.clone()); + } + + match &query.visibility { + RunSummaryVisibility::All => {} + RunSummaryVisibility::Default { include_archived } => { + let not_removing = format!("status <> '{}'", RunStatusKind::Removing); + if *include_archived { + builder.push(format!( + " AND (archived_at_ms IS NOT NULL OR {not_removing})" + )); + } else { + builder.push(format!(" AND archived_at_ms IS NULL AND {not_removing}")); + } + } + RunSummaryVisibility::Selected { statuses, archived } => { + builder.push(" AND ("); + let mut has_condition = false; + if *archived { + builder.push("archived_at_ms IS NOT NULL"); + has_condition = true; + } + if !statuses.is_empty() { + if has_condition { + builder.push(" OR "); + } + builder.push("(archived_at_ms IS NULL AND status IN ("); + let mut separated = builder.separated(", "); + for status in statuses { + separated.push_bind(status.to_string()); + } + separated.push_unseparated("))"); + has_condition = true; + } + if !has_condition { + builder.push("0"); + } + builder.push(")"); + } + } +} + +/// Status sort rank derived from [`RunStatusKind::board_rank`], so the SQL +/// order and the board column order share one source. Archived runs rank 7, +/// matching the `archived` board column. +static STATUS_RANK_CASE_SQL: LazyLock = LazyLock::new(|| { + let mut case = String::from("CASE WHEN archived_at_ms IS NOT NULL THEN 7"); + for kind in RunStatusKind::VARIANTS { + let _ = write!(case, " WHEN status = '{kind}' THEN {}", kind.board_rank()); + } + case.push_str(" ELSE 9 END"); + case +}); + +/// Size sort rank derived from [`RunSize::BUCKET_MAX_USD_MICROS`], so the SQL +/// order and the displayed size buckets share one source. +static SIZE_RANK_CASE_SQL: LazyLock = LazyLock::new(|| { + let mut case = String::from("CASE"); + for (rank, (_, max_usd_micros)) in RunSize::BUCKET_MAX_USD_MICROS.iter().enumerate() { + let _ = write!( + case, + " WHEN COALESCE(total_usd_micros, 0) <= {max_usd_micros} THEN {rank}" + ); + } + let _ = write!(case, " ELSE {} END", RunSize::BUCKET_MAX_USD_MICROS.len()); + case +}); + +fn push_order( + builder: &mut QueryBuilder, + sort: RunSummarySort, + direction: RunSummarySortDirection, + now: DateTime, +) { + builder.push(" ORDER BY "); + match sort { + RunSummarySort::CreatedAt => builder.push("created_at_ms"), + RunSummarySort::UpdatedAt => builder.push("last_event_at_ms"), + RunSummarySort::Status => builder.push(STATUS_RANK_CASE_SQL.as_str()), + RunSummarySort::Elapsed => builder + .push("(COALESCE(completed_at_ms, ") + .push_bind(now.timestamp_millis()) + .push(") - COALESCE(started_at_ms, created_at_ms))"), + RunSummarySort::Repository => builder.push("COALESCE(repository_name, '') COLLATE NOCASE"), + RunSummarySort::Title => builder.push("TRIM(title) COLLATE NOCASE"), + RunSummarySort::Workflow => builder.push("COALESCE(workflow_name, '') COLLATE NOCASE"), + RunSummarySort::Changes => builder.push("(diff_additions + diff_deletions)"), + RunSummarySort::Size => builder.push(SIZE_RANK_CASE_SQL.as_str()), + }; + match direction { + RunSummarySortDirection::Asc => builder.push(" ASC"), + RunSummarySortDirection::Desc => builder.push(" DESC"), + }; + builder.push(", id DESC"); +} + +fn decode_run_row(row: &SqliteRow, now: DateTime) -> Result { + let stored_id: String = row.try_get("id")?; + let summary_json: String = row.try_get("summary_json")?; + let children_count: i64 = row.try_get("children_count")?; + let mut run: Run = serde_json::from_str(&summary_json)?; + if stored_id != run.id.to_string() { + return Err(Error::RunSummaryMismatch { + run_id: stored_id, + field: "id", + }); + } + run.children_count = u64::try_from(children_count).map_err(|_| Error::RunSummaryMismatch { + run_id: run.id.to_string(), + field: "children_count", + })?; + overlay_live_wall_time(&mut run, now); + Ok(run) +} + +fn overlay_live_wall_time(run: &mut Run, now: DateTime) { + if run.timestamps.completed_at.is_some() { + return; + } + let Some(started_at) = run.timestamps.started_at else { + return; + }; + let wall_time_ms = timing::elapsed_ms(started_at, now); + run.timing = Some( + run.timing + .unwrap_or_else(|| RunTiming::wall_only(wall_time_ms)) + .with_wall_time(wall_time_ms), + ); +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use chrono::{DateTime, Utc}; + use fabro_types::{ + AutomationRef, BilledTokenCounts, BlockedReason, Conclusion, DiffSummary, FailureReason, + Graph, PendingReason, RunDiff, RunId, RunProjection, RunSize, RunSpec, RunStatus, + RunStatusKind, RunTiming, SessionId, StageId, StageOutcome, SuccessReason, + WorkflowSettings, test_support, + }; + use strum::VariantArray as _; + use ulid::Ulid; + + use super::{ + RunRecordStore, RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, + RunSummaryVisibility, decode_event_row, + }; + use crate::slate::CachedRunProjection; + use crate::{Error, EventPayload, test_support as store_test_support}; + + fn dt(value: &str) -> DateTime { + value.parse().unwrap() + } + + fn run_id(timestamp_ms: u64, random: u128) -> RunId { + RunId::from(Ulid::from_parts(timestamp_ms, random)) + } + + fn projection(run_id: RunId, title: &str, created_at: DateTime) -> RunProjection { + RunProjection::new( + title.to_string(), + RunSpec { + run_id, + settings: WorkflowSettings::default(), + graph: Graph::new("test"), + graph_source: None, + workflow_slug: Some("test-workflow".to_string()), + workflow_version_id: None, + target: None, + automation: None, + source_directory: None, + labels: HashMap::new(), + provenance: test_support::test_run_provenance(), + manifest_blob: None, + definition_blob: None, + spec_blob: None, + git: None, + fork_source_ref: None, + }, + created_at, + ) + } + + fn entry(projection: RunProjection, last_seq: u32) -> CachedRunProjection { + CachedRunProjection::from_projection(projection.spec.run_id, projection, last_seq) + } + + async fn store() -> (tempfile::TempDir, RunRecordStore) { + store_test_support::sqlite_run_record_store().await + } + + #[expect( + clippy::needless_pass_by_value, + reason = "test call sites construct one-off JSON properties" + )] + fn sql_event_payload( + run_id: &RunId, + event: &str, + node_id: Option<&str>, + stage_id: Option<&StageId>, + session_id: Option<&SessionId>, + properties: serde_json::Value, + ) -> EventPayload { + let mut value = serde_json::json!({ + "id": format!("evt-{event}"), + "ts": "2026-08-27T12:00:00Z", + "run_id": run_id.to_string(), + "event": event, + "properties": properties, + }); + let object = value.as_object_mut().unwrap(); + if let Some(node_id) = node_id { + object.insert("node_id".to_string(), node_id.into()); + } + if let Some(stage_id) = stage_id { + object.insert("stage_id".to_string(), stage_id.to_string().into()); + } + if let Some(session_id) = session_id { + object.insert("session_id".to_string(), session_id.to_string().into()); + } + EventPayload::new(value, run_id).unwrap() + } + + fn created_payload(run_id: &RunId) -> EventPayload { + sql_event_payload( + run_id, + "run.created", + None, + None, + None, + serde_json::json!({ + "title": "created", + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "workflow_slug": "test-workflow", + "labels": {}, + "provenance": test_support::test_run_provenance(), + }), + ) + } + + async fn seed_sql_event( + store: &RunRecordStore, + run_id: &RunId, + seq: u32, + payload: &EventPayload, + ) { + let event = fabro_types::RunEvent::try_from(payload).unwrap(); + sqlx::query( + r" +INSERT INTO run_events (run_id, seq, event_name, node_id, stage_id, session_id, event_json) +VALUES (?, ?, ?, ?, ?, ?, ?) +", + ) + .bind(run_id.to_string()) + .bind(i64::from(seq)) + .bind(event.event_name()) + .bind(event.node_id) + .bind(event.stage_id.map(|stage_id| stage_id.to_string())) + .bind(event.session_id) + .bind(serde_json::to_string(payload).unwrap()) + .execute(&store.pool) + .await + .unwrap(); + } + + fn sample_status(kind: RunStatusKind) -> RunStatus { + match kind { + RunStatusKind::Submitted => RunStatus::Submitted, + RunStatusKind::Pending => RunStatus::Pending { + reason: PendingReason::ApprovalRequired, + }, + RunStatusKind::Runnable => RunStatus::Runnable, + RunStatusKind::Starting => RunStatus::Starting, + RunStatusKind::Running => RunStatus::Running, + RunStatusKind::Blocked => RunStatus::Blocked { + blocked_reason: BlockedReason::HumanInputRequired, + }, + RunStatusKind::Paused => RunStatus::Paused { prior_block: None }, + RunStatusKind::Removing => RunStatus::Removing, + RunStatusKind::Succeeded => RunStatus::Succeeded { + reason: SuccessReason::Completed, + }, + RunStatusKind::Failed => RunStatus::Failed { + reason: FailureReason::WorkflowError, + }, + RunStatusKind::Dead => RunStatus::Dead, + } + } + + #[tokio::test] + async fn sql_run_transitions_are_atomic_and_guard_the_current_head() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let first = entry(projection(id, "created", created_at), 1); + let first_payload = created_payload(&id); + + let mut transaction = store.pool.begin().await.unwrap(); + let first_envelope = RunRecordStore::insert_first_event_on_connection( + &mut transaction, + &first, + &first_payload, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert_eq!(first_envelope.seq, 1); + + let stored_first: (i64, String) = + sqlx::query_as("SELECT source_last_seq, event_json FROM runs JOIN run_events ON run_events.run_id = runs.id WHERE runs.id = ? AND run_events.seq = 1") + .bind(id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(stored_first.0, 1); + assert_eq!( + stored_first.1, + serde_json::to_string(&first_payload).unwrap() + ); + + let mut duplicate_first = store.pool.begin().await.unwrap(); + assert!( + RunRecordStore::insert_first_event_on_connection( + &mut duplicate_first, + &first, + &first_payload, + ) + .await + .is_err() + ); + duplicate_first.rollback().await.unwrap(); + + let rolled_back_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let rolled_back = entry(projection(rolled_back_id, "rollback", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + RunRecordStore::insert_first_event_on_connection( + &mut transaction, + &rolled_back, + &created_payload(&rolled_back_id), + ) + .await + .unwrap(); + transaction.rollback().await.unwrap(); + let rolled_back_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM runs WHERE id = ?") + .bind(rolled_back_id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(rolled_back_rows, 0); + + let invalid_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); + let invalid = entry(projection(invalid_id, "invalid", created_at), 1); + let invalid_payload = sql_event_payload( + &invalid_id, + "run.title.updated", + None, + None, + None, + serde_json::json!({ "title": "too early" }), + ); + let mut transaction = store.pool.begin().await.unwrap(); + let error = RunRecordStore::insert_first_event_on_connection( + &mut transaction, + &invalid, + &invalid_payload, + ) + .await + .unwrap_err(); + assert!(matches!(error, Error::RunEventMismatch { + field: "event_name", + .. + })); + transaction.rollback().await.unwrap(); + + let rejected_id = run_id(created_at.timestamp_millis().cast_unsigned() + 3, 4); + sqlx::query( + "CREATE TRIGGER reject_test_event BEFORE INSERT ON run_events BEGIN SELECT RAISE(ABORT, 'rejected'); END", + ) + .execute(&store.pool) + .await + .unwrap(); + let rejected = entry(projection(rejected_id, "rejected", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + assert!( + RunRecordStore::insert_first_event_on_connection( + &mut transaction, + &rejected, + &created_payload(&rejected_id), + ) + .await + .is_err() + ); + transaction.rollback().await.unwrap(); + let rejected_rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM runs WHERE id = ?") + .bind(rejected_id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(rejected_rows, 0); + sqlx::query("DROP TRIGGER reject_test_event") + .execute(&store.pool) + .await + .unwrap(); + + let mut updated_projection = projection(id, "updated", created_at); + updated_projection.last_event_at = created_at + chrono::Duration::seconds(1); + updated_projection.status = RunStatus::Running; + let second = entry(updated_projection, 2); + let second_payload = sql_event_payload( + &id, + "run.title.updated", + None, + None, + None, + serde_json::json!({ "title": "updated" }), + ); + let mut transaction = store.pool.begin().await.unwrap(); + RunRecordStore::append_event_on_connection(&mut transaction, 1, &second, &second_payload) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let updated_row: (i64, String, String, i64) = sqlx::query_as( + "SELECT source_last_seq, status, title, last_event_at_ms FROM runs WHERE id = ?", + ) + .bind(id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!( + updated_row, + ( + 2, + "running".to_string(), + "updated".to_string(), + (created_at + chrono::Duration::seconds(1)).timestamp_millis(), + ) + ); + + let mut stale = store.pool.begin().await.unwrap(); + let stale_error = + RunRecordStore::append_event_on_connection(&mut stale, 1, &second, &second_payload) + .await + .unwrap_err(); + assert!(matches!(stale_error, Error::RunHeadMismatch { + expected_last_seq: 1, + actual_last_seq: Some(2), + .. + })); + stale.rollback().await.unwrap(); + + let third = entry(projection(id, "third", created_at), 3); + let third_payload = sql_event_payload( + &id, + "future.event", + None, + None, + None, + serde_json::json!({ "preserved": true }), + ); + let mut mismatched_value = third_payload.as_value().clone(); + mismatched_value["run_id"] = rolled_back_id.to_string().into(); + let mismatched_payload: EventPayload = serde_json::from_value(mismatched_value).unwrap(); + let mut invalid = store.pool.begin().await.unwrap(); + assert!( + RunRecordStore::append_event_on_connection( + &mut invalid, + 2, + &third, + &mismatched_payload, + ) + .await + .is_err() + ); + invalid.rollback().await.unwrap(); + let mut rollback = store.pool.begin().await.unwrap(); + RunRecordStore::append_event_on_connection(&mut rollback, 2, &third, &third_payload) + .await + .unwrap(); + rollback.rollback().await.unwrap(); + + seed_sql_event(&store, &id, 3, &third_payload).await; + let mut duplicate = store.pool.begin().await.unwrap(); + assert!( + RunRecordStore::append_event_on_connection(&mut duplicate, 2, &third, &third_payload,) + .await + .is_err() + ); + duplicate.rollback().await.unwrap(); + let head_after_failures: i64 = + sqlx::query_scalar("SELECT source_last_seq FROM runs WHERE id = ?") + .bind(id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(head_after_failures, 2); + + sqlx::query("DELETE FROM run_events WHERE run_id = ? AND seq = 3") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut transaction = store.pool.begin().await.unwrap(); + RunRecordStore::append_event_on_connection(&mut transaction, 2, &third, &third_payload) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let sequences = sqlx::query_scalar::<_, i64>( + "SELECT seq FROM run_events WHERE run_id = ? ORDER BY seq", + ) + .bind(id.to_string()) + .fetch_all(&store.pool) + .await + .unwrap(); + assert_eq!(sequences, vec![1, 2, 3]); + } + + #[tokio::test] + async fn sql_run_reads_preserve_paging_filters_json_and_legacy_gaps() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 11); + let first = entry(projection(id, "created", created_at), 1); + let mut transaction = store.pool.begin().await.unwrap(); + RunRecordStore::insert_first_event_on_connection( + &mut transaction, + &first, + &created_payload(&id), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let visit_one = StageId::new("work", 1); + let visit_two = StageId::new("work", 2); + let session_id = SessionId::new(); + let payloads = [ + sql_event_payload( + &id, + "future.stage", + Some("work"), + Some(&visit_one), + None, + serde_json::json!({ "kind": "visit-one" }), + ), + sql_event_payload( + &id, + "future.stage", + Some("work"), + Some(&visit_two), + None, + serde_json::json!({ "kind": "visit-two" }), + ), + sql_event_payload( + &id, + "future.legacy", + Some("work"), + None, + None, + serde_json::json!({ "kind": "legacy" }), + ), + sql_event_payload( + &id, + "run.session.future", + None, + None, + Some(&session_id), + serde_json::json!({ "redacted": "[REDACTED]" }), + ), + sql_event_payload( + &id, + "future.non_session", + None, + None, + Some(&session_id), + serde_json::json!({ "same_session": true }), + ), + ]; + for (index, payload) in payloads.iter().enumerate() { + seed_sql_event(&store, &id, u32::try_from(index).unwrap() + 2, payload).await; + } + sqlx::query("UPDATE runs SET source_last_seq = 6 WHERE id = ?") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + + let mut connection = store.pool.acquire().await.unwrap(); + let all = RunRecordStore::list_events_on_connection(&mut connection, &id) + .await + .unwrap(); + assert_eq!(all.iter().map(|event| event.seq).collect::>(), vec![ + 1, 2, 3, 4, 5, 6 + ]); + let forward = + RunRecordStore::list_events_from_with_limit_on_connection(&mut connection, &id, 2, 2) + .await + .unwrap(); + assert_eq!( + forward.iter().map(|event| event.seq).collect::>(), + vec![2, 3, 4] + ); + let reverse = RunRecordStore::list_events_before_with_limit_on_connection( + &mut connection, + &id, + Some(5), + 2, + ) + .await + .unwrap(); + assert_eq!( + reverse.iter().map(|event| event.seq).collect::>(), + vec![4, 3, 2] + ); + let exact = RunRecordStore::get_event_on_connection(&mut connection, &id, 5) + .await + .unwrap() + .unwrap(); + assert_eq!(exact.event.event_name(), "run.session.future"); + assert_eq!( + serde_json::to_value(&exact.event).unwrap(), + payloads[3].as_value().clone() + ); + + let visit_one_events = RunRecordStore::list_events_for_stage_from_with_limit_on_connection( + &mut connection, + &id, + &visit_one, + 1, + 10, + ) + .await + .unwrap(); + assert_eq!( + visit_one_events + .iter() + .map(|event| event.seq) + .collect::>(), + vec![2, 4] + ); + let visit_two_events = RunRecordStore::list_events_for_stage_from_with_limit_on_connection( + &mut connection, + &id, + &visit_two, + 1, + 10, + ) + .await + .unwrap(); + assert_eq!( + visit_two_events + .iter() + .map(|event| event.seq) + .collect::>(), + vec![3] + ); + let session_events = RunRecordStore::list_events_for_session_from_with_limit_on_connection( + &mut connection, + &id, + &session_id, + 1, + 10, + ) + .await + .unwrap(); + assert_eq!( + session_events + .iter() + .map(|event| event.seq) + .collect::>(), + vec![5] + ); + drop(connection); + + sqlx::query("DELETE FROM run_events WHERE run_id = ? AND seq = 3") + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut connection = store.pool.acquire().await.unwrap(); + let gapped = RunRecordStore::list_events_on_connection(&mut connection, &id) + .await + .unwrap(); + assert_eq!(gapped.last().unwrap().seq, 6); + assert_eq!(gapped.len(), 5); + } + + #[tokio::test] + async fn sql_run_reads_reject_extracted_column_and_identity_corruption() { + let (_directory, store) = store().await; + let created_at = dt("2026-08-27T12:00:00Z"); + let id = run_id(created_at.timestamp_millis().cast_unsigned(), 21); + let other_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 22); + store + .upsert_projection(&entry(projection(id, "one", created_at), 2)) + .await + .unwrap(); + store + .upsert_projection(&entry(projection(other_id, "two", created_at), 1)) + .await + .unwrap(); + let stage_id = StageId::new("work", 1); + let session_id = SessionId::new(); + let payload = sql_event_payload( + &id, + "run.session.future", + Some("work"), + Some(&stage_id), + Some(&session_id), + serde_json::json!({ "safe": true }), + ); + seed_sql_event(&store, &id, 2, &payload).await; + + for (sql, field) in [ + ( + "UPDATE run_events SET event_name = 'wrong' WHERE run_id = ? AND seq = 2", + "event_name", + ), + ( + "UPDATE run_events SET node_id = 'wrong' WHERE run_id = ? AND seq = 2", + "node_id", + ), + ( + "UPDATE run_events SET stage_id = 'wrong@1' WHERE run_id = ? AND seq = 2", + "stage_id", + ), + ( + "UPDATE run_events SET session_id = 'wrong' WHERE run_id = ? AND seq = 2", + "session_id", + ), + ] { + sqlx::query(sql) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut connection = store.pool.acquire().await.unwrap(); + let error = RunRecordStore::get_event_on_connection(&mut connection, &id, 2) + .await + .unwrap_err(); + assert!(matches!( + &error, + Error::RunEventMismatch { + field: mismatch_field, + .. + } if *mismatch_field == field + )); + assert!(!error.to_string().contains(&session_id.to_string())); + drop(connection); + seed_sql_event_restore(&store, &id, 2, &payload).await; + } + + let mut wrong_json = payload.as_value().clone(); + wrong_json["run_id"] = other_id.to_string().into(); + sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2") + .bind(serde_json::to_string(&wrong_json).unwrap()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let mut connection = store.pool.acquire().await.unwrap(); + assert!(matches!( + RunRecordStore::get_event_on_connection(&mut connection, &id, 2) + .await + .unwrap_err(), + Error::RunEventMismatch { + field: "run_id", + .. + } + )); + drop(connection); + + seed_sql_event_restore(&store, &id, 2, &payload).await; + sqlx::query("UPDATE run_events SET run_id = ? WHERE run_id = ? AND seq = 2") + .bind(other_id.to_string()) + .bind(id.to_string()) + .execute(&store.pool) + .await + .unwrap(); + let row = sqlx::query(super::SELECT_EVENT_COLUMNS) + .fetch_one(&store.pool) + .await + .unwrap(); + assert!(matches!( + decode_event_row(&row, &id).unwrap_err(), + Error::RunEventMismatch { + field: "run_id", + .. + } + )); + } + + async fn seed_sql_event_restore( + store: &RunRecordStore, + run_id: &RunId, + seq: u32, + payload: &EventPayload, + ) { + sqlx::query("DELETE FROM run_events WHERE run_id = ? AND seq = ?") + .bind(run_id.to_string()) + .bind(i64::from(seq)) + .execute(&store.pool) + .await + .unwrap(); + seed_sql_event(store, run_id, seq, payload).await; + } + + /// The migration's `CHECK (status IN (...))` freezes the status strings; + /// prove every `RunStatusKind` variant passes it so an enum change that + /// forgets a follow-up migration fails in CI instead of at runtime. + #[tokio::test] + async fn every_status_kind_upserts_within_schema_check() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + for (index, kind) in RunStatusKind::VARIANTS.iter().enumerate() { + let id = run_id( + created_at.timestamp_millis().cast_unsigned(), + u128::try_from(index).unwrap() + 1, + ); + let mut projected = projection(id, "status", created_at); + projected.status = sample_status(*kind); + store.upsert_projection(&entry(projected, 1)).await.unwrap(); + } + } + + #[tokio::test] + async fn upsert_is_monotonic_and_get_applies_children_count() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let parent_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let child_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + + let parent = entry(projection(parent_id, "parent", created_at), 1); + store.upsert_projection(&parent).await.unwrap(); + + let mut child_projection = projection(child_id, "new title", created_at); + child_projection.parent_id = Some(parent_id); + child_projection.last_event_at = created_at + chrono::Duration::seconds(2); + store + .upsert_projection(&entry(child_projection, 2)) + .await + .unwrap(); + + let mut stale = projection(child_id, "stale title", created_at); + stale.parent_id = Some(parent_id); + store.upsert_projection(&entry(stale, 1)).await.unwrap(); + + let parent = store.get(&parent_id, created_at).await.unwrap().unwrap(); + let child = store.get(&child_id, created_at).await.unwrap().unwrap(); + assert_eq!(parent.children_count, 1); + assert_eq!(child.title, "new title"); + } + + #[tokio::test] + async fn list_filters_sorts_and_paginates_in_sqlite() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let first_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let second_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let archived_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); + + let mut first = projection(first_id, "bravo", created_at); + first.spec.automation = Some(AutomationRef { + id: "nightly".to_string(), + name: None, + trigger_id: None, + }); + let mut second = projection(second_id, "alpha", created_at); + second.spec.automation = Some(AutomationRef { + id: "nightly".to_string(), + name: None, + trigger_id: None, + }); + let mut archived = projection(archived_id, "charlie", created_at); + archived.archived_at = Some(created_at); + for projected in [first, second, archived] { + store.upsert_projection(&entry(projected, 1)).await.unwrap(); + } + + let page = store + .list( + &RunSummaryListQuery { + automation_id: Some("nightly".to_string()), + sort: RunSummarySort::Title, + direction: RunSummarySortDirection::Asc, + limit: 1, + ..RunSummaryListQuery::default() + }, + created_at, + ) + .await + .unwrap(); + assert_eq!(page.total, 2); + assert!(page.has_more); + assert_eq!(page.data[0].title, "alpha"); + + let archived = store + .list( + &RunSummaryListQuery { + visibility: RunSummaryVisibility::Selected { + statuses: Vec::new(), + archived: true, + }, + ..RunSummaryListQuery::default() + }, + created_at, + ) + .await + .unwrap(); + assert_eq!(archived.data.len(), 1); + assert_eq!(archived.data[0].id, archived_id); + } + + #[tokio::test] + async fn projection_persists_billing_diff_and_derived_size() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let run_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let mut projection = projection(run_id, "billed", created_at); + projection.spec.automation = Some(AutomationRef { + id: "nightly".to_string(), + name: None, + trigger_id: None, + }); + projection.status = RunStatus::Succeeded { + reason: SuccessReason::Completed, + }; + projection.last_event_at = created_at + chrono::Duration::minutes(1); + projection.conclusion = Some(Conclusion { + timestamp: projection.last_event_at, + status: StageOutcome::Succeeded, + timing: RunTiming::wall_only(60_000), + failure: None, + final_git_commit_sha: None, + stages: Vec::new(), + billing: Some(BilledTokenCounts { + input_tokens: 100, + output_tokens: 20, + total_tokens: 135, + reasoning_tokens: 5, + cache_read_tokens: 10, + cache_write_tokens: 0, + total_usd_micros: Some(21_000_000), + }), + total_retries: 0, + diff: RunDiff { + patch: None, + summary: Some(DiffSummary { + files_changed: 2, + additions: 10, + deletions: 3, + }), + }, + }); + store + .upsert_projection(&entry(projection, 4)) + .await + .unwrap(); + + let row = sqlx::query( + "SELECT source_last_seq, created_at_ms, last_event_at_ms, status, title, workflow_slug, \ + automation_id, input_tokens, reasoning_tokens, cache_read_tokens, total_usd_micros, \ + diff_files_changed, diff_additions, diff_deletions FROM runs WHERE id = ?", + ) + .bind(run_id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(sqlx::Row::get::(&row, "source_last_seq"), 4); + assert_eq!( + sqlx::Row::get::(&row, "created_at_ms"), + created_at.timestamp_millis() + ); + assert_eq!( + sqlx::Row::get::(&row, "last_event_at_ms"), + (created_at + chrono::Duration::minutes(1)).timestamp_millis() + ); + assert_eq!(sqlx::Row::get::(&row, "status"), "succeeded"); + assert_eq!(sqlx::Row::get::(&row, "title"), "billed"); + assert_eq!( + sqlx::Row::get::(&row, "workflow_slug"), + "test-workflow" + ); + assert_eq!( + sqlx::Row::get::(&row, "automation_id"), + "nightly" + ); + assert_eq!(sqlx::Row::get::(&row, "input_tokens"), 100); + assert_eq!(sqlx::Row::get::(&row, "reasoning_tokens"), 5); + assert_eq!(sqlx::Row::get::(&row, "cache_read_tokens"), 10); + assert_eq!( + sqlx::Row::get::(&row, "total_usd_micros"), + 21_000_000 + ); + assert_eq!(sqlx::Row::get::(&row, "diff_files_changed"), 2); + assert_eq!(sqlx::Row::get::(&row, "diff_additions"), 10); + assert_eq!(sqlx::Row::get::(&row, "diff_deletions"), 3); + + let run = store.get(&run_id, created_at).await.unwrap().unwrap(); + assert_eq!(run.size, RunSize::S); + } + + #[tokio::test] + async fn projection_normalizes_legacy_overlapping_reasoning_tokens() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let run_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let mut projection = projection(run_id, "legacy billing", created_at); + projection.conclusion = Some(Conclusion { + timestamp: created_at, + status: StageOutcome::Succeeded, + timing: RunTiming::default(), + failure: None, + final_git_commit_sha: None, + stages: Vec::new(), + billing: Some(BilledTokenCounts { + input_tokens: 53, + output_tokens: -7, + total_tokens: 112, + reasoning_tokens: 66, + ..BilledTokenCounts::default() + }), + total_retries: 0, + diff: RunDiff::default(), + }); + + store + .upsert_projection(&entry(projection, 1)) + .await + .unwrap(); + + let row = sqlx::query( + "SELECT input_tokens, output_tokens, reasoning_tokens FROM runs WHERE id = ?", + ) + .bind(run_id.to_string()) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(sqlx::Row::get::(&row, "input_tokens"), 53); + assert_eq!(sqlx::Row::get::(&row, "output_tokens"), 0); + assert_eq!(sqlx::Row::get::(&row, "reasoning_tokens"), 59); + } + + #[tokio::test] + async fn reconcile_removes_rows_absent_from_authoritative_entries() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let kept_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let removed_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let kept = entry(projection(kept_id, "kept", created_at), 1); + let removed = entry(projection(removed_id, "removed", created_at), 1); + store.upsert_projection(&kept).await.unwrap(); + store.upsert_projection(&removed).await.unwrap(); + + store.reconcile(std::slice::from_ref(&kept)).await.unwrap(); + + assert!(store.get(&kept_id, created_at).await.unwrap().is_some()); + assert!(store.get(&removed_id, created_at).await.unwrap().is_none()); + } + + #[tokio::test] + async fn failed_reconcile_rolls_back_and_can_be_retried() { + let (_directory, store) = store().await; + let created_at = dt("2026-07-11T12:00:00Z"); + let stale_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); + let good_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); + let recovered_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); + store + .upsert_projection(&entry(projection(stale_id, "stale", created_at), 1)) + .await + .unwrap(); + + let good = entry(projection(good_id, "good", created_at), 1); + let recovered_projection = projection(recovered_id, "recovered", created_at); + let invalid = entry(recovered_projection.clone(), 0); + + assert!(store.reconcile(&[good.clone(), invalid]).await.is_err()); + assert!(store.get(&stale_id, created_at).await.unwrap().is_some()); + assert!(store.get(&good_id, created_at).await.unwrap().is_none()); + + let recovered = entry(recovered_projection, 1); + store.reconcile(&[good, recovered]).await.unwrap(); + + assert!(store.get(&stale_id, created_at).await.unwrap().is_none()); + assert!(store.get(&good_id, created_at).await.unwrap().is_some()); + assert!( + store + .get(&recovered_id, created_at) + .await + .unwrap() + .is_some() + ); + } +} diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs deleted file mode 100644 index a84283f83..000000000 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ /dev/null @@ -1,932 +0,0 @@ -use std::collections::{HashMap, HashSet}; -use std::fmt::Write as _; -use std::sync::LazyLock; - -use chrono::{DateTime, Utc}; -use fabro_types::{BilledTokenCounts, Run, RunId, RunSize, RunStatusKind, RunTiming, timing}; -use sqlx::sqlite::{SqliteConnection, SqliteRow}; -use sqlx::{QueryBuilder, Row as _, Sqlite, SqlitePool}; -use strum::VariantArray as _; - -use crate::run_state::projected_billing; -use crate::slate::CachedRunProjection; -use crate::{Error, Result}; - -const UPSERT_RUN_SQL: &str = r" -INSERT INTO runs ( - id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms, - status, archived_at_ms, parent_id, title, workflow_slug, workflow_name, - repository_name, automation_id, diff_files_changed, diff_additions, diff_deletions, - input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_write_tokens, - total_usd_micros, summary_json -) VALUES ( - ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? -) -ON CONFLICT(id) DO UPDATE SET - source_last_seq = excluded.source_last_seq, - created_at_ms = excluded.created_at_ms, - started_at_ms = excluded.started_at_ms, - last_event_at_ms = excluded.last_event_at_ms, - completed_at_ms = excluded.completed_at_ms, - status = excluded.status, - archived_at_ms = excluded.archived_at_ms, - parent_id = excluded.parent_id, - title = excluded.title, - workflow_slug = excluded.workflow_slug, - workflow_name = excluded.workflow_name, - repository_name = excluded.repository_name, - automation_id = excluded.automation_id, - diff_files_changed = excluded.diff_files_changed, - diff_additions = excluded.diff_additions, - diff_deletions = excluded.diff_deletions, - input_tokens = excluded.input_tokens, - output_tokens = excluded.output_tokens, - reasoning_tokens = excluded.reasoning_tokens, - cache_read_tokens = excluded.cache_read_tokens, - cache_write_tokens = excluded.cache_write_tokens, - total_usd_micros = excluded.total_usd_micros, - summary_json = excluded.summary_json -WHERE excluded.source_last_seq > runs.source_last_seq -"; - -const SELECT_RUN_SUMMARIES_SQL: &str = r" -SELECT runs.id, runs.summary_json, - (SELECT COUNT(*) FROM runs AS child WHERE child.parent_id = runs.id) AS children_count -FROM runs"; - -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum RunSummarySort { - #[default] - CreatedAt, - UpdatedAt, - Status, - Elapsed, - #[serde(rename = "repo")] - Repository, - Title, - Workflow, - Changes, - Size, -} - -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)] -#[serde(rename_all = "lowercase")] -pub enum RunSummarySortDirection { - Asc, - #[default] - Desc, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum RunSummaryVisibility { - All, - Default { - include_archived: bool, - }, - Selected { - statuses: Vec, - archived: bool, - }, -} - -impl Default for RunSummaryVisibility { - fn default() -> Self { - Self::Default { - include_archived: false, - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RunSummaryListQuery { - pub parent_id: Option, - pub automation_id: Option, - pub visibility: RunSummaryVisibility, - pub sort: RunSummarySort, - pub direction: RunSummarySortDirection, - pub limit: u32, - pub offset: u32, -} - -impl Default for RunSummaryListQuery { - fn default() -> Self { - Self { - parent_id: None, - automation_id: None, - visibility: RunSummaryVisibility::default(), - sort: RunSummarySort::default(), - direction: RunSummarySortDirection::default(), - limit: 100, - offset: 0, - } - } -} - -#[derive(Debug, Clone, PartialEq)] -pub struct RunSummaryPage { - pub data: Vec, - pub total: u64, - pub has_more: bool, -} - -#[derive(Clone)] -pub struct RunSummaryStore { - pool: SqlitePool, -} - -impl std::fmt::Debug for RunSummaryStore { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("RunSummaryStore").finish_non_exhaustive() - } -} - -impl RunSummaryStore { - #[must_use] - pub fn new(pool: SqlitePool) -> Self { - Self { pool } - } - - pub(crate) async fn upsert_projection(&self, entry: &CachedRunProjection) -> Result<()> { - let record = ProjectedRunSummary::from_entry(entry); - let mut connection = self.pool.acquire().await?; - upsert_run(&mut connection, &record).await?; - Ok(()) - } - - #[cfg(test)] - pub(crate) async fn close_pool(&self) { - self.pool.close().await; - } - - pub(crate) async fn reconcile(&self, entries: &[CachedRunProjection]) -> Result<()> { - let mut transaction = self.pool.begin().await?; - let stored_seqs: HashMap = - sqlx::query_as::<_, (String, i64)>("SELECT id, source_last_seq FROM runs") - .fetch_all(&mut *transaction) - .await? - .into_iter() - .collect(); - - let mut authoritative_ids = HashSet::new(); - for entry in entries { - let run_id = entry.run_id.to_string(); - let up_to_date = stored_seqs - .get(&run_id) - .is_some_and(|stored_seq| *stored_seq >= i64::from(entry.last_seq)); - authoritative_ids.insert(run_id); - if up_to_date { - continue; - } - upsert_run(&mut transaction, &ProjectedRunSummary::from_entry(entry)).await?; - } - - let stale_ids = stored_seqs - .keys() - .filter(|stored_id| !authoritative_ids.contains(stored_id.as_str())) - .collect::>(); - for chunk in stale_ids.chunks(500) { - let mut delete = QueryBuilder::::new("DELETE FROM runs WHERE id IN ("); - let mut separated = delete.separated(", "); - for stale_id in chunk { - separated.push_bind(stale_id.as_str()); - } - delete.push(")"); - delete.build().execute(&mut *transaction).await?; - } - transaction.commit().await?; - Ok(()) - } - - pub async fn get(&self, run_id: &RunId, now: DateTime) -> Result> { - let mut query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); - query - .push(" WHERE runs.id = ") - .push_bind(run_id.to_string()); - let row = query.build().fetch_optional(&self.pool).await?; - row.map(|row| decode_run_row(&row, now)).transpose() - } - - /// Identity fields for every stored run, for selector resolution without - /// decoding full summaries. - pub async fn list_identities(&self) -> Result> { - let rows = sqlx::query( - r" -SELECT id, workflow_slug, - json_extract(summary_json, '$.workflow.name') AS workflow_name, - json_extract(summary_json, '$.repository.origin_url') AS repository_origin_url -FROM runs", - ) - .fetch_all(&self.pool) - .await?; - rows.iter() - .map(|row| { - let stored_id: String = row.try_get("id")?; - let id = stored_id - .parse::() - .map_err(|_| Error::RunSummaryMismatch { - run_id: stored_id, - field: "id", - })?; - Ok(RunSummaryIdentity { - id, - workflow_slug: row.try_get("workflow_slug")?, - workflow_name: row.try_get("workflow_name")?, - repository_origin_url: row.try_get("repository_origin_url")?, - }) - }) - .collect() - } - - pub async fn list( - &self, - query: &RunSummaryListQuery, - now: DateTime, - ) -> Result { - let mut transaction = self.pool.begin().await?; - - let mut count_query = QueryBuilder::::new("SELECT COUNT(*) FROM runs"); - push_filters(&mut count_query, query); - let total: i64 = count_query - .build_query_scalar() - .fetch_one(&mut *transaction) - .await?; - - let mut rows_query = QueryBuilder::::new(SELECT_RUN_SUMMARIES_SQL); - push_filters(&mut rows_query, query); - push_order(&mut rows_query, query.sort, query.direction, now); - rows_query.push(" LIMIT ").push_bind(i64::from(query.limit)); - rows_query - .push(" OFFSET ") - .push_bind(i64::from(query.offset)); - let rows = rows_query.build().fetch_all(&mut *transaction).await?; - transaction.commit().await?; - - let data = rows - .iter() - .map(|row| decode_run_row(row, now)) - .collect::>>()?; - let total = u64::try_from(total).expect("COUNT(*) is non-negative"); - let consumed = u64::from(query.offset).saturating_add(data.len() as u64); - Ok(RunSummaryPage { - data, - total, - has_more: consumed < total, - }) - } - - pub async fn delete(&self, run_id: &RunId) -> Result<()> { - sqlx::query("DELETE FROM runs WHERE id = ?") - .bind(run_id.to_string()) - .execute(&self.pool) - .await?; - Ok(()) - } -} - -/// Identity fields of a stored run summary, cheap to list for selector -/// resolution. -#[derive(Debug, Clone)] -pub struct RunSummaryIdentity { - pub id: RunId, - pub workflow_slug: Option, - pub workflow_name: Option, - pub repository_origin_url: Option, -} - -#[derive(Debug)] -struct ProjectedRunSummary { - run: Run, - last_seq: u32, - workflow_name: Option, - repository_name: Option, - input_tokens: i64, - output_tokens: i64, - reasoning_tokens: i64, - cache_read_tokens: i64, - cache_write_tokens: i64, - total_usd_micros: Option, -} - -impl ProjectedRunSummary { - fn from_entry(entry: &CachedRunProjection) -> Self { - let mut run = entry.summary.clone(); - if run.timing.is_none() { - let at = run - .timestamps - .last_event_at - .unwrap_or(run.timestamps.created_at); - run.timing = entry.projection.live_run_timing(at); - } - let billing = normalize_billing_for_read_model(projected_billing(&entry.projection)); - let workflow_name = run.workflow.display_name().map(str::to_string); - let repository_name = run - .repository - .as_ref() - .map(|repository| repository.name.clone()); - - Self { - run, - last_seq: entry.last_seq, - workflow_name, - repository_name, - input_tokens: billing.input_tokens, - output_tokens: billing.output_tokens, - reasoning_tokens: billing.reasoning_tokens, - cache_read_tokens: billing.cache_read_tokens, - cache_write_tokens: billing.cache_write_tokens, - total_usd_micros: billing.total_usd_micros, - } - } -} - -/// Older provider codecs could persist a negative disjoint bucket when a -/// detail count exceeded its inclusive parent total. The SQLite summary is a -/// rebuildable, nonnegative read model, so normalize those legacy values here -/// without rewriting the authoritative run events. -fn normalize_billing_for_read_model(mut billing: BilledTokenCounts) -> BilledTokenCounts { - let input_total = billing - .input_tokens - .saturating_add(billing.cache_read_tokens) - .saturating_add(billing.cache_write_tokens) - .max(0); - billing.cache_read_tokens = billing.cache_read_tokens.clamp(0, input_total); - billing.cache_write_tokens = billing - .cache_write_tokens - .clamp(0, input_total - billing.cache_read_tokens); - billing.input_tokens = input_total - billing.cache_read_tokens - billing.cache_write_tokens; - - let output_total = billing - .output_tokens - .saturating_add(billing.reasoning_tokens) - .max(0); - billing.reasoning_tokens = billing.reasoning_tokens.clamp(0, output_total); - billing.output_tokens = output_total - billing.reasoning_tokens; - billing.total_tokens = input_total.saturating_add(output_total); - billing.total_usd_micros = billing.total_usd_micros.map(|value| value.max(0)); - billing -} - -async fn upsert_run(connection: &mut SqliteConnection, record: &ProjectedRunSummary) -> Result<()> { - let run = &record.run; - let diff = run.diff.unwrap_or_default(); - let summary_json = serde_json::to_string(run)?; - sqlx::query(UPSERT_RUN_SQL) - .bind(run.id.to_string()) - .bind(i64::from(record.last_seq)) - .bind(run.timestamps.created_at.timestamp_millis()) - .bind( - run.timestamps - .started_at - .map(|value| value.timestamp_millis()), - ) - .bind( - run.timestamps - .last_event_at - .unwrap_or(run.timestamps.created_at) - .timestamp_millis(), - ) - .bind( - run.timestamps - .completed_at - .map(|value| value.timestamp_millis()), - ) - .bind(run.lifecycle.status.kind().to_string()) - .bind( - run.lifecycle - .archived_at - .map(|value| value.timestamp_millis()), - ) - .bind(run.parent_id.map(|value| value.to_string())) - .bind(&run.title) - .bind(&run.workflow.slug) - .bind(&record.workflow_name) - .bind(&record.repository_name) - .bind(run.automation.as_ref().map(|automation| &automation.id)) - .bind(diff.files_changed) - .bind(diff.additions) - .bind(diff.deletions) - .bind(record.input_tokens) - .bind(record.output_tokens) - .bind(record.reasoning_tokens) - .bind(record.cache_read_tokens) - .bind(record.cache_write_tokens) - .bind(record.total_usd_micros) - .bind(summary_json) - .execute(connection) - .await?; - Ok(()) -} - -fn push_filters(builder: &mut QueryBuilder, query: &RunSummaryListQuery) { - builder.push(" WHERE 1 = 1"); - if let Some(parent_id) = query.parent_id { - builder - .push(" AND parent_id = ") - .push_bind(parent_id.to_string()); - } - if let Some(automation_id) = &query.automation_id { - builder - .push(" AND automation_id = ") - .push_bind(automation_id.clone()); - } - - match &query.visibility { - RunSummaryVisibility::All => {} - RunSummaryVisibility::Default { include_archived } => { - let not_removing = format!("status <> '{}'", RunStatusKind::Removing); - if *include_archived { - builder.push(format!( - " AND (archived_at_ms IS NOT NULL OR {not_removing})" - )); - } else { - builder.push(format!(" AND archived_at_ms IS NULL AND {not_removing}")); - } - } - RunSummaryVisibility::Selected { statuses, archived } => { - builder.push(" AND ("); - let mut has_condition = false; - if *archived { - builder.push("archived_at_ms IS NOT NULL"); - has_condition = true; - } - if !statuses.is_empty() { - if has_condition { - builder.push(" OR "); - } - builder.push("(archived_at_ms IS NULL AND status IN ("); - let mut separated = builder.separated(", "); - for status in statuses { - separated.push_bind(status.to_string()); - } - separated.push_unseparated("))"); - has_condition = true; - } - if !has_condition { - builder.push("0"); - } - builder.push(")"); - } - } -} - -/// Status sort rank derived from [`RunStatusKind::board_rank`], so the SQL -/// order and the board column order share one source. Archived runs rank 7, -/// matching the `archived` board column. -static STATUS_RANK_CASE_SQL: LazyLock = LazyLock::new(|| { - let mut case = String::from("CASE WHEN archived_at_ms IS NOT NULL THEN 7"); - for kind in RunStatusKind::VARIANTS { - let _ = write!(case, " WHEN status = '{kind}' THEN {}", kind.board_rank()); - } - case.push_str(" ELSE 9 END"); - case -}); - -/// Size sort rank derived from [`RunSize::BUCKET_MAX_USD_MICROS`], so the SQL -/// order and the displayed size buckets share one source. -static SIZE_RANK_CASE_SQL: LazyLock = LazyLock::new(|| { - let mut case = String::from("CASE"); - for (rank, (_, max_usd_micros)) in RunSize::BUCKET_MAX_USD_MICROS.iter().enumerate() { - let _ = write!( - case, - " WHEN COALESCE(total_usd_micros, 0) <= {max_usd_micros} THEN {rank}" - ); - } - let _ = write!(case, " ELSE {} END", RunSize::BUCKET_MAX_USD_MICROS.len()); - case -}); - -fn push_order( - builder: &mut QueryBuilder, - sort: RunSummarySort, - direction: RunSummarySortDirection, - now: DateTime, -) { - builder.push(" ORDER BY "); - match sort { - RunSummarySort::CreatedAt => builder.push("created_at_ms"), - RunSummarySort::UpdatedAt => builder.push("last_event_at_ms"), - RunSummarySort::Status => builder.push(STATUS_RANK_CASE_SQL.as_str()), - RunSummarySort::Elapsed => builder - .push("(COALESCE(completed_at_ms, ") - .push_bind(now.timestamp_millis()) - .push(") - COALESCE(started_at_ms, created_at_ms))"), - RunSummarySort::Repository => builder.push("COALESCE(repository_name, '') COLLATE NOCASE"), - RunSummarySort::Title => builder.push("TRIM(title) COLLATE NOCASE"), - RunSummarySort::Workflow => builder.push("COALESCE(workflow_name, '') COLLATE NOCASE"), - RunSummarySort::Changes => builder.push("(diff_additions + diff_deletions)"), - RunSummarySort::Size => builder.push(SIZE_RANK_CASE_SQL.as_str()), - }; - match direction { - RunSummarySortDirection::Asc => builder.push(" ASC"), - RunSummarySortDirection::Desc => builder.push(" DESC"), - }; - builder.push(", id DESC"); -} - -fn decode_run_row(row: &SqliteRow, now: DateTime) -> Result { - let stored_id: String = row.try_get("id")?; - let summary_json: String = row.try_get("summary_json")?; - let children_count: i64 = row.try_get("children_count")?; - let mut run: Run = serde_json::from_str(&summary_json)?; - if stored_id != run.id.to_string() { - return Err(Error::RunSummaryMismatch { - run_id: stored_id, - field: "id", - }); - } - run.children_count = u64::try_from(children_count).map_err(|_| Error::RunSummaryMismatch { - run_id: run.id.to_string(), - field: "children_count", - })?; - overlay_live_wall_time(&mut run, now); - Ok(run) -} - -fn overlay_live_wall_time(run: &mut Run, now: DateTime) { - if run.timestamps.completed_at.is_some() { - return; - } - let Some(started_at) = run.timestamps.started_at else { - return; - }; - let wall_time_ms = timing::elapsed_ms(started_at, now); - run.timing = Some( - run.timing - .unwrap_or_else(|| RunTiming::wall_only(wall_time_ms)) - .with_wall_time(wall_time_ms), - ); -} - -#[cfg(test)] -mod tests { - use std::collections::HashMap; - - use chrono::{DateTime, Utc}; - use fabro_types::{ - AutomationRef, BilledTokenCounts, BlockedReason, Conclusion, DiffSummary, FailureReason, - Graph, PendingReason, RunDiff, RunId, RunProjection, RunSize, RunSpec, RunStatus, - RunStatusKind, RunTiming, StageOutcome, SuccessReason, WorkflowSettings, test_support, - }; - use strum::VariantArray as _; - use ulid::Ulid; - - use super::{ - RunSummaryListQuery, RunSummarySort, RunSummarySortDirection, RunSummaryStore, - RunSummaryVisibility, - }; - use crate::slate::CachedRunProjection; - use crate::test_support as store_test_support; - - fn dt(value: &str) -> DateTime { - value.parse().unwrap() - } - - fn run_id(timestamp_ms: u64, random: u128) -> RunId { - RunId::from(Ulid::from_parts(timestamp_ms, random)) - } - - fn projection(run_id: RunId, title: &str, created_at: DateTime) -> RunProjection { - RunProjection::new( - title.to_string(), - RunSpec { - run_id, - settings: WorkflowSettings::default(), - graph: Graph::new("test"), - graph_source: None, - workflow_slug: Some("test-workflow".to_string()), - workflow_version_id: None, - target: None, - automation: None, - source_directory: None, - labels: HashMap::new(), - provenance: test_support::test_run_provenance(), - manifest_blob: None, - definition_blob: None, - spec_blob: None, - git: None, - fork_source_ref: None, - }, - created_at, - ) - } - - fn entry(projection: RunProjection, last_seq: u32) -> CachedRunProjection { - CachedRunProjection::from_projection(projection.spec.run_id, projection, last_seq) - } - - async fn store() -> (tempfile::TempDir, RunSummaryStore) { - store_test_support::sqlite_summary_store().await - } - - fn sample_status(kind: RunStatusKind) -> RunStatus { - match kind { - RunStatusKind::Submitted => RunStatus::Submitted, - RunStatusKind::Pending => RunStatus::Pending { - reason: PendingReason::ApprovalRequired, - }, - RunStatusKind::Runnable => RunStatus::Runnable, - RunStatusKind::Starting => RunStatus::Starting, - RunStatusKind::Running => RunStatus::Running, - RunStatusKind::Blocked => RunStatus::Blocked { - blocked_reason: BlockedReason::HumanInputRequired, - }, - RunStatusKind::Paused => RunStatus::Paused { prior_block: None }, - RunStatusKind::Removing => RunStatus::Removing, - RunStatusKind::Succeeded => RunStatus::Succeeded { - reason: SuccessReason::Completed, - }, - RunStatusKind::Failed => RunStatus::Failed { - reason: FailureReason::WorkflowError, - }, - RunStatusKind::Dead => RunStatus::Dead, - } - } - - /// The migration's `CHECK (status IN (...))` freezes the status strings; - /// prove every `RunStatusKind` variant passes it so an enum change that - /// forgets a follow-up migration fails in CI instead of at runtime. - #[tokio::test] - async fn every_status_kind_upserts_within_schema_check() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - for (index, kind) in RunStatusKind::VARIANTS.iter().enumerate() { - let id = run_id( - created_at.timestamp_millis().cast_unsigned(), - u128::try_from(index).unwrap() + 1, - ); - let mut projected = projection(id, "status", created_at); - projected.status = sample_status(*kind); - store.upsert_projection(&entry(projected, 1)).await.unwrap(); - } - } - - #[tokio::test] - async fn upsert_is_monotonic_and_get_applies_children_count() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - let parent_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); - let child_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); - - let parent = entry(projection(parent_id, "parent", created_at), 1); - store.upsert_projection(&parent).await.unwrap(); - - let mut child_projection = projection(child_id, "new title", created_at); - child_projection.parent_id = Some(parent_id); - child_projection.last_event_at = created_at + chrono::Duration::seconds(2); - store - .upsert_projection(&entry(child_projection, 2)) - .await - .unwrap(); - - let mut stale = projection(child_id, "stale title", created_at); - stale.parent_id = Some(parent_id); - store.upsert_projection(&entry(stale, 1)).await.unwrap(); - - let parent = store.get(&parent_id, created_at).await.unwrap().unwrap(); - let child = store.get(&child_id, created_at).await.unwrap().unwrap(); - assert_eq!(parent.children_count, 1); - assert_eq!(child.title, "new title"); - } - - #[tokio::test] - async fn list_filters_sorts_and_paginates_in_sqlite() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - let first_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); - let second_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); - let archived_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); - - let mut first = projection(first_id, "bravo", created_at); - first.spec.automation = Some(AutomationRef { - id: "nightly".to_string(), - name: None, - trigger_id: None, - }); - let mut second = projection(second_id, "alpha", created_at); - second.spec.automation = Some(AutomationRef { - id: "nightly".to_string(), - name: None, - trigger_id: None, - }); - let mut archived = projection(archived_id, "charlie", created_at); - archived.archived_at = Some(created_at); - for projected in [first, second, archived] { - store.upsert_projection(&entry(projected, 1)).await.unwrap(); - } - - let page = store - .list( - &RunSummaryListQuery { - automation_id: Some("nightly".to_string()), - sort: RunSummarySort::Title, - direction: RunSummarySortDirection::Asc, - limit: 1, - ..RunSummaryListQuery::default() - }, - created_at, - ) - .await - .unwrap(); - assert_eq!(page.total, 2); - assert!(page.has_more); - assert_eq!(page.data[0].title, "alpha"); - - let archived = store - .list( - &RunSummaryListQuery { - visibility: RunSummaryVisibility::Selected { - statuses: Vec::new(), - archived: true, - }, - ..RunSummaryListQuery::default() - }, - created_at, - ) - .await - .unwrap(); - assert_eq!(archived.data.len(), 1); - assert_eq!(archived.data[0].id, archived_id); - } - - #[tokio::test] - async fn projection_persists_billing_diff_and_derived_size() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - let run_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); - let mut projection = projection(run_id, "billed", created_at); - projection.spec.automation = Some(AutomationRef { - id: "nightly".to_string(), - name: None, - trigger_id: None, - }); - projection.status = RunStatus::Succeeded { - reason: SuccessReason::Completed, - }; - projection.last_event_at = created_at + chrono::Duration::minutes(1); - projection.conclusion = Some(Conclusion { - timestamp: projection.last_event_at, - status: StageOutcome::Succeeded, - timing: RunTiming::wall_only(60_000), - failure: None, - final_git_commit_sha: None, - stages: Vec::new(), - billing: Some(BilledTokenCounts { - input_tokens: 100, - output_tokens: 20, - total_tokens: 135, - reasoning_tokens: 5, - cache_read_tokens: 10, - cache_write_tokens: 0, - total_usd_micros: Some(21_000_000), - }), - total_retries: 0, - diff: RunDiff { - patch: None, - summary: Some(DiffSummary { - files_changed: 2, - additions: 10, - deletions: 3, - }), - }, - }); - store - .upsert_projection(&entry(projection, 4)) - .await - .unwrap(); - - let row = sqlx::query( - "SELECT source_last_seq, created_at_ms, last_event_at_ms, status, title, workflow_slug, \ - automation_id, input_tokens, reasoning_tokens, cache_read_tokens, total_usd_micros, \ - diff_files_changed, diff_additions, diff_deletions FROM runs WHERE id = ?", - ) - .bind(run_id.to_string()) - .fetch_one(&store.pool) - .await - .unwrap(); - assert_eq!(sqlx::Row::get::(&row, "source_last_seq"), 4); - assert_eq!( - sqlx::Row::get::(&row, "created_at_ms"), - created_at.timestamp_millis() - ); - assert_eq!( - sqlx::Row::get::(&row, "last_event_at_ms"), - (created_at + chrono::Duration::minutes(1)).timestamp_millis() - ); - assert_eq!(sqlx::Row::get::(&row, "status"), "succeeded"); - assert_eq!(sqlx::Row::get::(&row, "title"), "billed"); - assert_eq!( - sqlx::Row::get::(&row, "workflow_slug"), - "test-workflow" - ); - assert_eq!( - sqlx::Row::get::(&row, "automation_id"), - "nightly" - ); - assert_eq!(sqlx::Row::get::(&row, "input_tokens"), 100); - assert_eq!(sqlx::Row::get::(&row, "reasoning_tokens"), 5); - assert_eq!(sqlx::Row::get::(&row, "cache_read_tokens"), 10); - assert_eq!( - sqlx::Row::get::(&row, "total_usd_micros"), - 21_000_000 - ); - assert_eq!(sqlx::Row::get::(&row, "diff_files_changed"), 2); - assert_eq!(sqlx::Row::get::(&row, "diff_additions"), 10); - assert_eq!(sqlx::Row::get::(&row, "diff_deletions"), 3); - - let run = store.get(&run_id, created_at).await.unwrap().unwrap(); - assert_eq!(run.size, RunSize::S); - } - - #[tokio::test] - async fn projection_normalizes_legacy_overlapping_reasoning_tokens() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - let run_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); - let mut projection = projection(run_id, "legacy billing", created_at); - projection.conclusion = Some(Conclusion { - timestamp: created_at, - status: StageOutcome::Succeeded, - timing: RunTiming::default(), - failure: None, - final_git_commit_sha: None, - stages: Vec::new(), - billing: Some(BilledTokenCounts { - input_tokens: 53, - output_tokens: -7, - total_tokens: 112, - reasoning_tokens: 66, - ..BilledTokenCounts::default() - }), - total_retries: 0, - diff: RunDiff::default(), - }); - - store - .upsert_projection(&entry(projection, 1)) - .await - .unwrap(); - - let row = sqlx::query( - "SELECT input_tokens, output_tokens, reasoning_tokens FROM runs WHERE id = ?", - ) - .bind(run_id.to_string()) - .fetch_one(&store.pool) - .await - .unwrap(); - assert_eq!(sqlx::Row::get::(&row, "input_tokens"), 53); - assert_eq!(sqlx::Row::get::(&row, "output_tokens"), 0); - assert_eq!(sqlx::Row::get::(&row, "reasoning_tokens"), 59); - } - - #[tokio::test] - async fn reconcile_removes_rows_absent_from_authoritative_entries() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - let kept_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); - let removed_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); - let kept = entry(projection(kept_id, "kept", created_at), 1); - let removed = entry(projection(removed_id, "removed", created_at), 1); - store.upsert_projection(&kept).await.unwrap(); - store.upsert_projection(&removed).await.unwrap(); - - store.reconcile(std::slice::from_ref(&kept)).await.unwrap(); - - assert!(store.get(&kept_id, created_at).await.unwrap().is_some()); - assert!(store.get(&removed_id, created_at).await.unwrap().is_none()); - } - - #[tokio::test] - async fn failed_reconcile_rolls_back_and_can_be_retried() { - let (_directory, store) = store().await; - let created_at = dt("2026-07-11T12:00:00Z"); - let stale_id = run_id(created_at.timestamp_millis().cast_unsigned(), 1); - let good_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 2); - let recovered_id = run_id(created_at.timestamp_millis().cast_unsigned() + 2, 3); - store - .upsert_projection(&entry(projection(stale_id, "stale", created_at), 1)) - .await - .unwrap(); - - let good = entry(projection(good_id, "good", created_at), 1); - let recovered_projection = projection(recovered_id, "recovered", created_at); - let invalid = entry(recovered_projection.clone(), 0); - - assert!(store.reconcile(&[good.clone(), invalid]).await.is_err()); - assert!(store.get(&stale_id, created_at).await.unwrap().is_some()); - assert!(store.get(&good_id, created_at).await.unwrap().is_none()); - - let recovered = entry(recovered_projection, 1); - store.reconcile(&[good, recovered]).await.unwrap(); - - assert!(store.get(&stale_id, created_at).await.unwrap().is_none()); - assert!(store.get(&good_id, created_at).await.unwrap().is_some()); - assert!( - store - .get(&recovered_id, created_at) - .await - .unwrap() - .is_some() - ); - } -} diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 0e7d48d90..972f5eab5 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -4,7 +4,7 @@ mod run_store; use std::collections::HashMap; use std::path::PathBuf; -use std::sync::{Arc, OnceLock}; +use std::sync::Arc; use std::time::Duration; use chrono::{DateTime, Utc}; @@ -19,7 +19,7 @@ use slatedb::config::{CompressionCodec, Settings}; use tokio::sync::{Mutex, OnceCell}; use tracing::warn; -use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys}; +use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunRecordStore, keys}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct UnreadableRun { @@ -45,7 +45,7 @@ pub struct Database { catalog_index: Arc>>, projection_cache: Arc, projection_cache_warmed: Arc>, - run_summary_store: Arc>>, + run_record_store: Arc, } impl std::fmt::Debug for Database { @@ -65,6 +65,7 @@ impl Database { flush_interval: Duration, cache_path: Option, blobs: Arc, + run_record_store: Arc, ) -> Self { Self { object_store, @@ -77,16 +78,13 @@ impl Database { catalog_index: Arc::new(OnceCell::new()), projection_cache: Arc::new(RunProjectionCache::default()), projection_cache_warmed: Arc::new(OnceCell::new()), - run_summary_store: Arc::new(OnceLock::new()), + run_record_store, } } - pub fn attach_run_summary_store(&self, store: Arc) -> Arc { - Arc::clone(self.run_summary_store.get_or_init(|| store)) - } - - fn run_summary_store(&self) -> Option> { - self.run_summary_store.get().cloned() + #[must_use] + pub fn run_record_store(&self) -> Arc { + Arc::clone(&self.run_record_store) } fn shared_db_prefix(&self) -> String { @@ -142,7 +140,7 @@ impl Database { read_only, self.blobs(), Arc::clone(&self.projection_cache), - Arc::clone(&self.run_summary_store), + self.run_record_store(), ) .await } @@ -240,9 +238,7 @@ impl Database { } } } - if let Some(store) = self.run_summary_store() { - store.reconcile(&entries).await?; - } + self.run_record_store.reconcile(&entries).await?; self.projection_cache.replace_all(entries).await; Ok::<_, Error>(()) }) @@ -389,9 +385,7 @@ impl Database { self.delete_session_indexes_for_run(run_id).await?; self.catalog_index().await?.remove(run_id).await?; self.remove_cached_run(run_id).await; - if let Some(store) = self.run_summary_store() { - store.delete(run_id).await?; - } + self.run_record_store.delete(run_id).await?; Ok(()) } @@ -558,6 +552,31 @@ mod tests { (object_store, store) } + fn make_store_with_run_records( + run_records: Arc, + ) -> (Arc, Database) { + let object_store: Arc = Arc::new(InMemory::new()); + let store = store_test_support::test_database_with_stores( + object_store.clone(), + "runs/", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + run_records, + ); + (object_store, store) + } + + #[tokio::test] + async fn required_run_record_store_is_shared_with_run_handles() { + let (_object_store, store) = make_store(); + let records = store.run_record_store(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); + + assert!(run.shares_run_record_store(&records)); + assert!(Arc::ptr_eq(&records, &store.clone().run_record_store())); + } + #[tokio::test] async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() { let (_object_store, store) = make_store(); @@ -596,8 +615,8 @@ mod tests { ); } - async fn make_summary_store() -> (tempfile::TempDir, Arc) { - let (directory, store) = store_test_support::sqlite_summary_store().await; + async fn make_run_record_store() -> (tempfile::TempDir, Arc) { + let (directory, store) = store_test_support::sqlite_run_record_store().await; (directory, Arc::new(store)) } @@ -972,9 +991,8 @@ mod tests { #[tokio::test] async fn rejected_transition_leaves_reconciled_summary_present() { - let (_object_store, store) = make_store(); - let (_directory, summaries) = make_summary_store().await; - store.attach_run_summary_store(Arc::clone(&summaries)); + let (_directory, summaries) = make_run_record_store().await; + let (_object_store, store) = make_store_with_run_records(Arc::clone(&summaries)); let run_id = test_run_id("run-1"); let run = store.create_run(&run_id).await.unwrap(); append_runnable(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; @@ -995,10 +1013,9 @@ mod tests { } #[tokio::test] - async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() { - let (object_store, store) = make_store(); - let (directory, summaries) = make_summary_store().await; - store.attach_run_summary_store(Arc::clone(&summaries)); + async fn best_effort_run_record_update_failure_keeps_slate_append_repairable() { + let (directory, summaries) = make_run_record_store().await; + let (object_store, store) = make_store_with_run_records(Arc::clone(&summaries)); let run_id = test_run_id("run-1"); let run = store.create_run(&run_id).await.unwrap(); append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; @@ -1022,7 +1039,7 @@ mod tests { assert_eq!(stored.event, result.unwrap().event); let repaired_summaries = - Arc::new(store_test_support::sqlite_summary_store_at(directory.path()).await); + Arc::new(store_test_support::sqlite_run_record_store_at(directory.path()).await); let stale = repaired_summaries .get(&run_id, Utc::now()) .await @@ -1030,13 +1047,14 @@ mod tests { .unwrap(); assert_ne!(stale.title, "Committed title"); - let reopened = store_test_support::test_database( + let reopened = store_test_support::test_database_with_stores( object_store, "runs/", Duration::from_millis(1), None, + store_test_support::test_blob_store(), + Arc::clone(&repaired_summaries), ); - reopened.attach_run_summary_store(Arc::clone(&repaired_summaries)); reopened.warm_projection_cache().await.unwrap(); let repaired = repaired_summaries .get(&run_id, Utc::now()) @@ -1657,10 +1675,9 @@ mod tests { } #[tokio::test] - async fn append_event_refreshes_projection_cache_and_delete_removes_it() { - let (_object_store, store) = make_store(); - let (_directory, summaries) = make_summary_store().await; - store.attach_run_summary_store(Arc::clone(&summaries)); + async fn required_run_record_append_refreshes_cache_and_delete_removes_rows() { + let (_directory, summaries) = make_run_record_store().await; + let (_object_store, store) = make_store_with_run_records(Arc::clone(&summaries)); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; store.warm_projection_cache().await.unwrap(); @@ -1834,15 +1851,20 @@ mod tests { } #[tokio::test] - async fn projection_cache_warmup_backfills_sqlite_run_summaries() { + async fn required_run_record_warmup_backfills_sqlite_run_records() { let (object_store, store) = make_store(); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await; - let reopened = - store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None); - let (_directory, summaries) = make_summary_store().await; - reopened.attach_run_summary_store(Arc::clone(&summaries)); + let (_directory, summaries) = make_run_record_store().await; + let reopened = store_test_support::test_database_with_stores( + object_store, + "runs", + Duration::from_millis(1), + None, + store_test_support::test_blob_store(), + Arc::clone(&summaries), + ); reopened.warm_projection_cache().await.unwrap(); let summary = summaries diff --git a/lib/components/fabro-store/src/slate/run_store.rs b/lib/components/fabro-store/src/slate/run_store.rs index c01ab0904..6ab5fc1a4 100644 --- a/lib/components/fabro-store/src/slate/run_store.rs +++ b/lib/components/fabro-store/src/slate/run_store.rs @@ -1,6 +1,6 @@ use std::collections::VecDeque; +use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; -use std::sync::{Arc, OnceLock}; use bytes::Bytes; use chrono::Utc; @@ -14,7 +14,7 @@ use tracing::warn; use super::projection_cache::{CachedRunProjection, RunProjectionCache}; use crate::run_state::{EventProjectionCache, RunProjectionReducer}; use crate::{ - BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummaryStore, StageId, + BlobStore, Error, EventEnvelope, EventPayload, Result, RunProjection, RunRecordStore, StageId, keys, }; @@ -45,9 +45,7 @@ pub(crate) struct RunDatabaseInner { state_lock: Mutex<()>, projection_cache: Mutex, shared_projection_cache: Arc, - // Shared cell rather than a snapshot so a summary store attached after - // this writer opened is still picked up by later appends. - run_summary_store: Arc>>, + run_record_store: Arc, recent_events: Mutex>, recent_event_limit: usize, event_tx: broadcast::Sender, @@ -60,7 +58,7 @@ impl RunDatabase { read_only: bool, blob_store: Arc, shared_projection_cache: Arc, - run_summary_store: Arc>>, + run_record_store: Arc, ) -> Result { let cached_projection = shared_projection_cache.projection_snapshot(&run_id).await; let projection_cache = cached_projection.as_ref().map_or_else( @@ -92,7 +90,7 @@ impl RunDatabase { state_lock: Mutex::new(()), projection_cache: Mutex::new(projection_cache), shared_projection_cache, - run_summary_store, + run_record_store, recent_events: Mutex::new(VecDeque::with_capacity(DEFAULT_EVENT_TAIL_LIMIT)), recent_event_limit: DEFAULT_EVENT_TAIL_LIMIT, event_tx, @@ -123,6 +121,11 @@ impl RunDatabase { self.inner.run_id } + #[cfg(test)] + pub(crate) fn shares_run_record_store(&self, store: &Arc) -> bool { + Arc::ptr_eq(&self.inner.run_record_store, store) + } + pub fn subscribe(&self) -> broadcast::Receiver { self.inner.event_tx.subscribe() } @@ -231,15 +234,13 @@ impl RunDatabase { } async fn update_summary_after_committed_append(&self, cached: &CachedRunProjection) { - if let Some(store) = self.inner.run_summary_store.get() { - if let Err(err) = store.upsert_projection(cached).await { - warn!( - run_id = %self.inner.run_id, - source_last_seq = cached.last_seq, - error = ?err, - "failed to update SQLite run summary after committed append" - ); - } + if let Err(err) = self.inner.run_record_store.upsert_projection(cached).await { + warn!( + run_id = %self.inner.run_id, + source_last_seq = cached.last_seq, + error = ?err, + "failed to update SQLite run record after committed append" + ); } } diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index d1a7f4848..3124f8ebe 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -8,8 +8,8 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use crate::keys::SlateKey; #[cfg(test)] -use crate::{AuthCodeStore, AuthSessionStore, RunSummaryStore}; -use crate::{BlobStore, Database, Result}; +use crate::{AuthCodeStore, AuthSessionStore}; +use crate::{BlobStore, Database, Result, RunRecordStore}; /// Returns an isolated SQLite blob authority backed by its own in-memory /// database. @@ -46,6 +46,32 @@ pub fn test_blob_store() -> Arc { Arc::new(BlobStore::new(pool)) } +/// Returns an isolated SQLite run-record store backed by its own in-memory +/// database and the production `runs` and `run_events` schemas. +#[must_use] +pub fn test_run_record_store() -> Arc { + let options = SqliteConnectOptions::new() + .filename(":memory:") + .foreign_keys(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .max_lifetime(None) + .idle_timeout(None) + .after_connect(|connection, _metadata| { + Box::pin(async move { + sqlx::raw_sql(fabro_db::RUNS_MIGRATION_SQL) + .execute(&mut *connection) + .await?; + sqlx::raw_sql(fabro_db::RUN_EVENTS_MIGRATION_SQL) + .execute(&mut *connection) + .await?; + Ok(()) + }) + }) + .connect_lazy_with(options); + Arc::new(RunRecordStore::new(pool)) +} + /// Returns the SQLite file backing [`test_blob_store_at`] for `store_dir`. #[must_use] pub fn test_blob_store_path(store_dir: &Path) -> PathBuf { @@ -119,7 +145,37 @@ pub fn test_database_with_blobs( cache_path: Option, blobs: Arc, ) -> Database { - Database::new(object_store, base_prefix, flush_interval, cache_path, blobs) + test_database_with_stores( + object_store, + base_prefix, + flush_interval, + cache_path, + blobs, + test_run_record_store(), + ) +} + +/// Builds a Slate-backed run database with explicit shared SQLite stores. +/// +/// Use this only when a test needs a failing, persistent, or shared store; +/// ordinary fixtures should use [`test_database`]. +#[must_use] +pub fn test_database_with_stores( + object_store: Arc, + base_prefix: impl Into, + flush_interval: Duration, + cache_path: Option, + blobs: Arc, + run_records: Arc, +) -> Database { + Database::new( + object_store, + base_prefix, + flush_interval, + cache_path, + blobs, + run_records, + ) } /// Seeds one canonical row in the legacy SlateDB blob keyspace. @@ -171,13 +227,13 @@ pub(crate) async fn sqlite_auth_code_store() -> (tempfile::TempDir, AuthCodeStor } #[cfg(test)] -pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) { +pub(crate) async fn sqlite_run_record_store() -> (tempfile::TempDir, RunRecordStore) { let directory = tempfile::tempdir().unwrap(); - let store = sqlite_summary_store_at(directory.path()).await; + let store = sqlite_run_record_store_at(directory.path()).await; (directory, store) } #[cfg(test)] -pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore { - RunSummaryStore::new(sqlite_test_pool(directory).await) +pub(crate) async fn sqlite_run_record_store_at(directory: &Path) -> RunRecordStore { + RunRecordStore::new(sqlite_test_pool(directory).await) } diff --git a/lib/foundation/fabro-db/migrations/2026082701_run_events.sql b/lib/foundation/fabro-db/migrations/2026082701_run_events.sql new file mode 100644 index 000000000..1fca0563c --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026082701_run_events.sql @@ -0,0 +1,30 @@ +CREATE TABLE run_events ( + run_id TEXT NOT NULL, + seq INTEGER NOT NULL, + event_name TEXT NOT NULL, + node_id TEXT, + stage_id TEXT, + session_id TEXT, + event_json TEXT NOT NULL, + PRIMARY KEY (run_id, seq), + FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE, + CHECK (seq BETWEEN 1 AND 999999), + CHECK (json_valid(event_json)) +); + +CREATE INDEX run_events_by_stage +ON run_events(run_id, stage_id, seq) +WHERE stage_id IS NOT NULL; + +CREATE INDEX run_events_by_legacy_node +ON run_events(run_id, node_id, seq) +WHERE stage_id IS NULL AND node_id IS NOT NULL; + +CREATE INDEX run_events_by_session +ON run_events(run_id, session_id, seq) +WHERE session_id IS NOT NULL + AND event_name GLOB 'run.session.*'; + +CREATE INDEX run_events_by_pull_request_creation_request +ON run_events(run_id, seq) +WHERE event_name = 'pull_request.creation_requested'; diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index 26567d2c9..87ad39447 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -20,6 +20,14 @@ static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); /// the production blob schema without a filesystem path into this crate. pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blobs.sql"); +/// The run-record migration, exposed so fixtures in other crates can install +/// the production schema without a filesystem path into this crate. +pub const RUNS_MIGRATION_SQL: &str = include_str!("../migrations/2026071104_runs.sql"); + +/// The run-event migration, exposed so fixtures in other crates can install +/// the production schema without a filesystem path into this crate. +pub const RUN_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/2026082701_run_events.sql"); + #[derive(Clone)] pub struct Database { pool: DbPool, diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index 62cd1fcd7..07e6673b4 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -475,6 +475,276 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul Ok(()) } +#[tokio::test] +async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + let run_columns = sqlx::query("PRAGMA table_info(runs)") + .fetch_all(database.pool()) + .await?; + assert_eq!( + run_columns.len(), + 24, + "the existing runs row must stay unchanged" + ); + + let event_columns = sqlx::query("PRAGMA table_info(run_events)") + .fetch_all(database.pool()) + .await?; + let event_column_contract = event_columns + .iter() + .map(|column| { + ( + column.get::("name"), + column.get::("type"), + column.get::("notnull"), + column.get::("pk"), + ) + }) + .collect::>(); + assert_eq!(event_column_contract, vec![ + ("run_id".to_string(), "TEXT".to_string(), 1, 1), + ("seq".to_string(), "INTEGER".to_string(), 1, 2), + ("event_name".to_string(), "TEXT".to_string(), 1, 0), + ("node_id".to_string(), "TEXT".to_string(), 0, 0), + ("stage_id".to_string(), "TEXT".to_string(), 0, 0), + ("session_id".to_string(), "TEXT".to_string(), 0, 0), + ("event_json".to_string(), "TEXT".to_string(), 1, 0), + ]); + + let foreign_keys = sqlx::query("PRAGMA foreign_key_list(run_events)") + .fetch_all(database.pool()) + .await?; + assert_eq!(foreign_keys.len(), 1); + assert_eq!(foreign_keys[0].get::("table"), "runs"); + assert_eq!(foreign_keys[0].get::("from"), "run_id"); + assert_eq!(foreign_keys[0].get::("to"), "id"); + assert_eq!(foreign_keys[0].get::("on_delete"), "CASCADE"); + + let indexes = sqlx::query("PRAGMA index_list(run_events)") + .fetch_all(database.pool()) + .await?; + let named_indexes = indexes + .iter() + .filter_map(|index| { + let name = index.get::("name"); + name.starts_with("run_events_by_").then_some(( + name, + index.get::("unique"), + index.get::("partial"), + )) + }) + .collect::>(); + assert_eq!(named_indexes, vec![ + ( + "run_events_by_pull_request_creation_request".to_string(), + 0, + 1, + ), + ("run_events_by_session".to_string(), 0, 1), + ("run_events_by_legacy_node".to_string(), 0, 1), + ("run_events_by_stage".to_string(), 0, 1), + ]); + assert!(indexes.iter().all(|index| { + index.get::("unique") == 0 + || index.get::("name") == "sqlite_autoindex_run_events_1" + })); + + insert_run_with_id(database.pool(), "parent", None).await?; + insert_run_with_id(database.pool(), "child", Some("parent")).await?; + insert_run_event( + database.pool(), + "parent", + 1, + "run.created", + None, + None, + None, + ) + .await?; + + for invalid in [ + insert_run_event( + database.pool(), + "parent", + 1, + "run.created", + None, + None, + None, + ) + .await, + insert_run_event( + database.pool(), + "missing", + 1, + "run.created", + None, + None, + None, + ) + .await, + insert_run_event( + database.pool(), + "parent", + 0, + "run.created", + None, + None, + None, + ) + .await, + insert_run_event( + database.pool(), + "parent", + 1_000_000, + "run.created", + None, + None, + None, + ) + .await, + ] { + assert!(invalid.is_err()); + } + let invalid_json = sqlx::query( + "INSERT INTO run_events (run_id, seq, event_name, event_json) VALUES (?, ?, ?, ?)", + ) + .bind("parent") + .bind(2_i64) + .bind("run.started") + .bind("not-json") + .execute(database.pool()) + .await; + assert!(invalid_json.is_err()); + + sqlx::query("INSERT INTO blobs (hash, data) VALUES (?, ?)") + .bind("a".repeat(64)) + .bind(vec![1_u8]) + .execute(database.pool()) + .await?; + sqlx::query("DELETE FROM runs WHERE id = ?") + .bind("parent") + .execute(database.pool()) + .await?; + let event_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM run_events WHERE run_id = 'parent'") + .fetch_one(database.pool()) + .await?; + let child_parent: Option = + sqlx::query_scalar("SELECT parent_id FROM runs WHERE id = 'child'") + .fetch_one(database.pool()) + .await?; + let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM blobs") + .fetch_one(database.pool()) + .await?; + assert_eq!(event_count, 0); + assert_eq!(child_parent.as_deref(), Some("parent")); + assert_eq!(blob_count, 1); + + Ok(()) +} + +#[tokio::test] +async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + for (sql, expected_index) in [ + ( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?", + "sqlite_autoindex_run_events_1", + ), + ( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq = ?", + "sqlite_autoindex_run_events_1", + ), + ( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND stage_id = ? ORDER BY seq ASC LIMIT ?", + "run_events_by_stage", + ), + ( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND stage_id IS NULL AND node_id = ? ORDER BY seq ASC LIMIT ?", + "run_events_by_legacy_node", + ), + ( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND session_id = ? AND event_name GLOB 'run.session.*' ORDER BY seq ASC LIMIT ?", + "run_events_by_session", + ), + ( + "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE event_name = 'pull_request.creation_requested' ORDER BY run_id, seq", + "run_events_by_pull_request_creation_request", + ), + ] { + let details = sqlx::query(sql) + .bind("run") + .bind("value") + .bind(10_i64) + .fetch_all(database.pool()) + .await? + .into_iter() + .map(|row| row.get::("detail")) + .collect::>() + .join("; "); + assert!( + details.contains(expected_index), + "expected {expected_index} in query plan: {details}" + ); + } + + Ok(()) +} + +async fn insert_run_with_id( + pool: &fabro_db::DbPool, + id: &str, + parent_id: Option<&str>, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" +INSERT INTO runs ( + id, source_last_seq, created_at_ms, last_event_at_ms, status, parent_id, title, + input_tokens, summary_json +) VALUES (?, 1, 0, 0, 'submitted', ?, 'title', 0, ?) +", + ) + .bind(id) + .bind(parent_id) + .bind(format!(r#"{{"id":"{id}"}}"#)) + .execute(pool) + .await?; + Ok(()) +} + +async fn insert_run_event( + pool: &fabro_db::DbPool, + run_id: &str, + seq: i64, + event_name: &str, + node_id: Option<&str>, + stage_id: Option<&str>, + session_id: Option<&str>, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" +INSERT INTO run_events (run_id, seq, event_name, node_id, stage_id, session_id, event_json) +VALUES (?, ?, ?, ?, ?, ?, '{}') +", + ) + .bind(run_id) + .bind(seq) + .bind(event_name) + .bind(node_id) + .bind(stage_id) + .bind(session_id) + .execute(pool) + .await?; + Ok(()) +} + async fn insert_minimal_run( pool: &fabro_db::DbPool, status: &str,