diff --git a/lib/components/fabro-petri/tests/projection.rs b/lib/components/fabro-petri/tests/projection.rs index 73cac347a..a8c5f2e78 100644 --- a/lib/components/fabro-petri/tests/projection.rs +++ b/lib/components/fabro-petri/tests/projection.rs @@ -124,6 +124,7 @@ fn pool() -> DbPool { test_support::in_memory_pool_with(&[ fabro_db::BLOBS_MIGRATION_SQL, fabro_db::RUNS_MIGRATION_SQL, + fabro_db::DROP_RUN_EVENTS_MIGRATION_SQL, fabro_db::PETRI_RECORDS_MIGRATION_SQL, fabro_db::PETRI_PROJECTION_MIGRATION_SQL, ]) diff --git a/lib/components/fabro-store/src/run_summary_store.rs b/lib/components/fabro-store/src/run_summary_store.rs index 0a55e2d9c..14cce156e 100644 --- a/lib/components/fabro-store/src/run_summary_store.rs +++ b/lib/components/fabro-store/src/run_summary_store.rs @@ -17,17 +17,15 @@ use crate::run_summary::{build_summary, projected_usage}; use crate::{Error, Result, RunProjection}; /// The `runs` row of a Petri run, written by its projector: every column the -/// list views and the scheduler read, and never `source_last_seq`, which the -/// legacy event path owns while it still writes the row. +/// list views and the scheduler read. const UPSERT_PETRI_RUN_SQL: &str = r" INSERT INTO runs ( - id, source_last_seq, created_at_ms, started_at_ms, last_event_at_ms, completed_at_ms, + id, 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, + repository_name, automation_id, diff_additions, diff_deletions, total_usd_micros, summary_json ) VALUES ( - ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ) ON CONFLICT(id) DO UPDATE SET created_at_ms = excluded.created_at_ms, @@ -403,15 +401,10 @@ pub struct RunSummaryIdentity { #[derive(Debug)] struct PreparedRunSummary { - run: Run, - 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, + run: Run, + workflow_name: Option, + repository_name: Option, + total_usd_micros: Option, } impl PreparedRunSummary { @@ -435,11 +428,6 @@ impl PreparedRunSummary { run, workflow_name, repository_name, - input_tokens: column_count(usage.tokens.input), - output_tokens: column_count(usage.tokens.output), - reasoning_tokens: column_count(usage.tokens.reasoning), - cache_read_tokens: column_count(usage.tokens.cache_read), - cache_write_tokens: column_count(usage.tokens.cache_write), total_usd_micros: usage.cost.map(|cost| column_count(cost.usd_micros)), } } @@ -451,9 +439,8 @@ fn column_count(count: u64) -> i64 { i64::try_from(count).unwrap_or(i64::MAX) } -/// Binds the `runs` columns shared by the insert, upsert, and update -/// statements, in the positional order those statements declare them -/// (`source_last_seq` through `summary_json`). +/// Binds the `runs` columns after `id`, in the positional order the upsert +/// declares them (`created_at_ms` through `summary_json`). fn bind_run_columns<'q>( query: Query<'q, Sqlite, SqliteArguments>, record: &'q PreparedRunSummary, @@ -462,9 +449,6 @@ fn bind_run_columns<'q>( let diff = run.diff.unwrap_or_default(); let summary_json = serde_json::to_string(run)?; Ok(query - // `source_last_seq`: a column the legacy event log owned; `1` until - // the migration that drops it. - .bind(1_i64) .bind(run.timestamps.created_at.timestamp_millis()) .bind( run.timestamps @@ -494,14 +478,8 @@ fn bind_run_columns<'q>( .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)) } @@ -1121,8 +1099,8 @@ mod tests { let row = sqlx::query( "SELECT 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 = ?", + automation_id, total_usd_micros, diff_additions, diff_deletions \ + FROM runs WHERE id = ?", ) .bind(run_id.to_string()) .fetch_one(&store.pool) @@ -1146,14 +1124,10 @@ mod tests { 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); diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index 5689e1605..17d35f941 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -25,6 +25,7 @@ pub fn test_blob_store() -> Arc { /// platform records and the projection tables. const RUN_SUMMARY_MIGRATIONS: &[&str] = &[ fabro_db::RUNS_MIGRATION_SQL, + fabro_db::DROP_RUN_EVENTS_MIGRATION_SQL, fabro_db::PETRI_PROJECTION_MIGRATION_SQL, ]; diff --git a/lib/foundation/fabro-db/migrations/2026091803_drop_run_events.sql b/lib/foundation/fabro-db/migrations/2026091803_drop_run_events.sql new file mode 100644 index 000000000..49c61e182 --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026091803_drop_run_events.sql @@ -0,0 +1,69 @@ +-- The legacy executor's run event log is gone: every run is a Petri run +-- whose history is `petri_records` and `platform_records`. Fabro is +-- greenfield here, so the rows are dropped, not converted, and the +-- one-time activation bookkeeping of that log goes with them. +DROP TABLE IF EXISTS run_events; +DROP TABLE IF EXISTS legacy_run_history_activation; +DROP TABLE IF EXISTS legacy_run_history_deletions; + +-- The `runs` row loses the columns only that log wrote or read: +-- `source_last_seq`, its write-path concurrency guard (the projection's +-- per-log positions replace it), and the six token and file-count columns +-- nothing read (`summary_json` keeps the values). SQLite cannot drop a +-- column a table CHECK names, so the table is rebuilt. +CREATE TABLE runs_next ( + id TEXT PRIMARY KEY NOT NULL, + created_at_ms INTEGER NOT NULL, + started_at_ms INTEGER, + last_event_at_ms INTEGER NOT NULL, + completed_at_ms INTEGER, + status TEXT NOT NULL, + archived_at_ms INTEGER, + parent_id TEXT, + title TEXT NOT NULL, + workflow_slug TEXT, + workflow_name TEXT, + repository_name TEXT, + automation_id TEXT, + diff_additions INTEGER NOT NULL DEFAULT 0, + diff_deletions INTEGER NOT NULL DEFAULT 0, + total_usd_micros INTEGER, + summary_json TEXT NOT NULL, + CHECK (status IN ( + 'submitted', + 'pending', + 'runnable', + 'starting', + 'running', + 'blocked', + 'paused', + 'removing', + 'succeeded', + 'failed', + 'dead' + )), + CHECK (diff_additions >= 0), + CHECK (diff_deletions >= 0), + CHECK (total_usd_micros IS NULL OR total_usd_micros >= 0), + CHECK (json_valid(summary_json)) +); + +INSERT INTO runs_next ( + id, 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_additions, diff_deletions, total_usd_micros, summary_json +) +SELECT + id, 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_additions, diff_deletions, total_usd_micros, summary_json +FROM runs; + +DROP TABLE runs; +ALTER TABLE runs_next RENAME TO runs; + +CREATE INDEX runs_by_created_at ON runs(created_at_ms DESC, id DESC); +CREATE INDEX runs_by_updated_at ON runs(last_event_at_ms DESC, id DESC); +CREATE INDEX runs_by_status ON runs(archived_at_ms, status, last_event_at_ms DESC, id DESC); +CREATE INDEX runs_by_parent ON runs(parent_id, created_at_ms DESC, id DESC); +CREATE INDEX runs_by_automation ON runs(automation_id, created_at_ms DESC, id DESC); diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index d710ae1cd..12387705c 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -16,8 +16,6 @@ pub type DbPool = sqlx::SqlitePool; static MIGRATOR: Migrator = sqlx::migrate!("./migrations"); -const SESSION_OWNER_INDEX_MIGRATION_VERSION: i64 = 2_026_083_101; - /// The blob-table migration, exposed so fixtures in other crates can install /// the production blob schema without a filesystem path into this crate. pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blobs.sql"); @@ -26,14 +24,11 @@ pub const BLOBS_MIGRATION_SQL: &str = include_str!("../migrations/2026081301_blo /// 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"); - -/// The run-session owner index migration, exposed so fixtures in other crates -/// can install the production run-history indexes. -pub const RUN_EVENT_SESSION_OWNER_MIGRATION_SQL: &str = - include_str!("../migrations/2026083101_run_event_session_owner.sql"); +/// The migration that drops the legacy run event log and narrows the `runs` +/// row to its final shape, exposed so fixtures that install +/// [`RUNS_MIGRATION_SQL`] can apply it next and get the production schema. +pub const DROP_RUN_EVENTS_MIGRATION_SQL: &str = + include_str!("../migrations/2026091803_drop_run_events.sql"); /// The Ask Fabro session record migration, exposed so fixtures in other /// crates can install the production schema without a filesystem path into @@ -59,11 +54,6 @@ pub const PETRI_RECORDS_MIGRATION_SQL: &str = pub const PETRI_PROJECTION_MIGRATION_SQL: &str = include_str!("../migrations/2026091801_petri_projection.sql"); -/// The temporary run-history activation migration, exposed so fixtures in -/// other crates can install the production compatibility schema. -pub const RUN_HISTORY_ACTIVATION_MIGRATION_SQL: &str = - include_str!("../migrations/2026082802_run_history_activation.sql"); - #[derive(Clone)] pub struct Database { pool: DbPool, @@ -98,9 +88,6 @@ impl Database { pub async fn migrate(&self) -> anyhow::Result<()> { let applied = applied_migration_versions(&self.pool).await?; - self.preflight_session_owner_index(&applied) - .await - .context("checking session ownership before SQLite migrations")?; self.snapshot_before_new_migrations(&applied) .await .context("snapshotting SQLite database before migrations")?; @@ -110,53 +97,6 @@ impl Database { .context("running SQLite migrations") } - /// Refuse the unique owner index when old event history contains - /// collisions. The diagnostic is deliberately count-only because session - /// identifiers and event contents are not safe startup-log fields. - /// - /// Temporary compatibility guard: once every supported database has - /// applied the session-owner index migration the version check below - /// always short-circuits, and this preflight can be deleted along with - /// the run-history compatibility window. - async fn preflight_session_owner_index(&self, applied: &HashSet) -> anyhow::Result<()> { - if applied.contains(&SESSION_OWNER_INDEX_MIGRATION_VERSION) { - return Ok(()); - } - - let run_events_exists: bool = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'run_events')", - ) - .fetch_one(&self.pool) - .await - .context("checking for the run event table")?; - if !run_events_exists { - return Ok(()); - } - - let collision_groups: i64 = sqlx::query_scalar( - r" -SELECT COUNT(*) -FROM ( - SELECT session_id - FROM run_events - WHERE session_id IS NOT NULL - AND event_name = 'run.session.created' - GROUP BY session_id - HAVING COUNT(*) > 1 -) -", - ) - .fetch_one(&self.pool) - .await - .context("counting duplicate session ownership groups")?; - if collision_groups > 0 { - anyhow::bail!( - "cannot create the unique session owner index: found {collision_groups} duplicate session ownership groups" - ); - } - Ok(()) - } - /// Copy the database aside before applying migrations it has not seen. /// /// A binary downgrade after new migrations have been applied fails sqlx's diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index efc11e885..576888941 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -769,13 +769,13 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul assert_eq!(index_count, 5); insert_minimal_run(database.pool(), "submitted", 0, r#"{"id":"run"}"#).await?; - for (status, input_tokens, summary_json) in [ + for (status, diff_additions, summary_json) in [ ("unknown", 0, r#"{"id":"run-2"}"#), ("submitted", -1, r#"{"id":"run-3"}"#), ("submitted", 0, "not-json"), ] { assert!( - insert_minimal_run(database.pool(), status, input_tokens, summary_json) + insert_minimal_run(database.pool(), status, diff_additions, summary_json) .await .is_err() ); @@ -784,384 +784,148 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul Ok(()) } +/// The `runs` row is the projection's summary of a Petri run: the columns +/// the list views filter and sort by, and the JSON the API serves. The +/// legacy event log's tables and the columns only it wrote are gone. #[tokio::test] -async fn session_owner_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> { +async fn runs_schema_has_its_final_shape_without_the_legacy_event_log() -> anyhow::Result<()> { let dir = tempfile::tempdir()?; let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; database.migrate().await?; + for table in [ + "run_events", + "legacy_run_history_activation", + "legacy_run_history_deletions", + "runs_next", + ] { + assert!( + !table_exists(database.pool(), table).await?, + "{table} must not exist" + ); + } + let run_columns = sqlx::query("PRAGMA table_info(runs)") .fetch_all(database.pool()) - .await?; + .await? + .iter() + .map(|column| column.get::("name")) + .collect::>(); + assert_eq!(run_columns, [ + "id", + "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_additions", + "diff_deletions", + "total_usd_micros", + "summary_json", + ]); + + let index_names = sqlx::query("PRAGMA index_list(runs)") + .fetch_all(database.pool()) + .await? + .iter() + .map(|index| index.get::("name")) + .filter(|name| name.starts_with("runs_by_")) + .collect::>(); assert_eq!( - run_columns.len(), - 24, - "the existing runs row must stay unchanged" + index_names.iter().map(String::as_str).collect::>(), + [ + "runs_by_automation", + "runs_by_created_at", + "runs_by_parent", + "runs_by_status", + "runs_by_updated_at", + ] ); - let event_columns = sqlx::query("PRAGMA table_info(run_events)") - .fetch_all(database.pool()) + Ok(()) +} + +/// A database written before the event log was dropped keeps its run rows: +/// the migration rebuilds the table and copies every surviving column. +#[tokio::test] +async fn dropping_the_event_log_keeps_the_run_rows() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + // Rewind to the schema an older binary left: the `runs` table with its + // legacy columns, the event log referencing it and the activation + // bookkeeping, with only the drop migration pending again. Those + // migrations' own rows stay applied, so sqlx's checksum validation + // still passes. + sqlx::raw_sql("DROP TABLE runs; DELETE FROM _sqlx_migrations WHERE version = 2026091803;") + .execute(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_session_owner".to_string(), 1, 1), - ( - "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") == "run_events_by_session_owner" - || 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").await?; - - for invalid in [ - insert_run_event(database.pool(), "parent", 1, "run.created").await, - insert_run_event(database.pool(), "missing", 1, "run.created").await, - insert_run_event(database.pool(), "parent", 0, "run.created").await, - insert_run_event(database.pool(), "parent", 1_000_000, "run.created").await, + for migration in [ + fabro_db::RUNS_MIGRATION_SQL, + include_str!("../migrations/2026082701_run_events.sql"), + include_str!("../migrations/2026082802_run_history_activation.sql"), + include_str!("../migrations/2026083101_run_event_session_owner.sql"), ] { - assert!(invalid.is_err()); + sqlx::raw_sql(migration).execute(database.pool()).await?; } - let invalid_json = sqlx::query( - "INSERT INTO run_events (run_id, seq, event_name, event_json) VALUES (?, ?, ?, ?)", + sqlx::query( + r#" +INSERT INTO runs ( + id, source_last_seq, created_at_ms, last_event_at_ms, status, title, input_tokens, + diff_additions, total_usd_micros, summary_json +) VALUES ('kept', 7, 1, 2, 'succeeded', 'Kept run', 99, 3, 4, '{"id":"kept"}') +"#, ) - .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_including_session_owner() --> 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 run_id, seq, event_name, node_id, stage_id, session_id, event_json FROM run_events WHERE session_id = ? AND event_name = 'run.session.created'", - "run_events_by_session_owner", - ), - ( - "EXPLAIN QUERY PLAN SELECT DISTINCT run_id FROM run_events WHERE event_name = 'pull_request.creation_requested'", - "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}" - ); - } - - // The first-visit stage listing unions both shapes so each arm keeps its - // own partial index instead of scanning the run's primary key range. - let details = sqlx::query( - "EXPLAIN QUERY PLAN SELECT * FROM run_events WHERE run_id = ? AND seq >= ? AND stage_id = ? \ - UNION ALL SELECT * FROM run_events WHERE run_id = ? AND seq >= ? AND stage_id IS NULL AND node_id = ? \ - ORDER BY seq ASC LIMIT ?", - ) - .bind("run") - .bind(1_i64) - .bind("stage") - .bind("run") - .bind(1_i64) - .bind("node") - .bind(10_i64) - .fetch_all(database.pool()) - .await? - .into_iter() - .map(|row| row.get::("detail")) - .collect::>() - .join("; "); - for expected_index in ["run_events_by_stage", "run_events_by_legacy_node"] { - assert!( - details.contains(expected_index), - "expected {expected_index} in query plan: {details}" - ); - } - - Ok(()) -} - -#[tokio::test] -async fn session_owner_migration_preflight_is_count_only_retriable_and_idempotent() --> anyhow::Result<()> { - let dir = tempfile::tempdir()?; - let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; - database.migrate().await?; - - sqlx::query("DROP INDEX IF EXISTS run_events_by_session_owner") - .execute(database.pool()) - .await?; - sqlx::query("DELETE FROM _sqlx_migrations WHERE version = 2026083101") - .execute(database.pool()) - .await?; - - for run_id in ["first", "second", "third", "fourth"] { - insert_run_with_id(database.pool(), run_id, None).await?; - } - for (run_id, session_id) in [ - ("first", "collision-alpha"), - ("second", "collision-alpha"), - ("third", "collision-beta"), - ("fourth", "collision-beta"), - ] { - insert_session_creation_claim(database.pool(), run_id, session_id).await?; - } - - let error = database - .migrate() - .await - .expect_err("duplicate session owners must abort migration"); - let rendered = format!("{error:#}"); - assert!(rendered.contains("2 duplicate session ownership groups")); - assert!(!rendered.contains("collision-alpha")); - assert!(!rendered.contains("collision-beta")); - assert!(!rendered.contains("sensitive event contents")); - assert_eq!( - sqlx::query_scalar::<_, i64>( - "SELECT COUNT(*) FROM run_events WHERE event_name = 'run.session.created'" - ) - .fetch_one(database.pool()) - .await?, - 4 - ); - assert_eq!( - sqlx::query_scalar::<_, i64>( - "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = 'run_events_by_session_owner'" - ) - .fetch_one(database.pool()) - .await?, - 0 - ); - - sqlx::query("DELETE FROM run_events WHERE run_id IN ('second', 'fourth')") - .execute(database.pool()) - .await?; - database.migrate().await?; - database.migrate().await?; - assert_eq!( - sqlx::query_scalar::<_, i64>( - "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = 'run_events_by_session_owner'" - ) - .fetch_one(database.pool()) - .await?, - 1 - ); - Ok(()) -} - -async fn insert_session_creation_claim( - pool: &fabro_db::DbPool, - run_id: &str, - session_id: &str, -) -> Result<(), sqlx::Error> { - sqlx::query( - r" -INSERT INTO run_events (run_id, seq, event_name, session_id, event_json) -VALUES (?, 1, 'run.session.created', ?, json_object( - 'run_id', ?, - 'event', 'run.session.created', - 'session_id', ?, - 'properties', json_object('note', 'sensitive event contents') -)) -", - ) - .bind(run_id) - .bind(session_id) - .bind(run_id) - .bind(session_id) - .execute(pool) .await?; - Ok(()) -} - -async fn insert_run_event( - pool: &fabro_db::DbPool, - run_id: &str, - seq: i64, - event_name: &str, -) -> Result<(), sqlx::Error> { sqlx::query( - r" -INSERT INTO run_events (run_id, seq, event_name, event_json) -VALUES (?, ?, ?, '{}') -", + "INSERT INTO run_events (run_id, seq, event_name, event_json) VALUES ('kept', 1, 'run.created', '{}')", ) - .bind(run_id) - .bind(seq) - .bind(event_name) - .execute(pool) + .execute(database.pool()) .await?; + + database.migrate().await?; + + assert!(!table_exists(database.pool(), "run_events").await?); + let row = sqlx::query( + "SELECT created_at_ms, last_event_at_ms, status, title, diff_additions, total_usd_micros, summary_json FROM runs WHERE id = 'kept'", + ) + .fetch_one(database.pool()) + .await?; + assert_eq!(row.get::("created_at_ms"), 1); + assert_eq!(row.get::("last_event_at_ms"), 2); + assert_eq!(row.get::("status"), "succeeded"); + assert_eq!(row.get::("title"), "Kept run"); + assert_eq!(row.get::("diff_additions"), 3); + assert_eq!(row.get::("total_usd_micros"), 4); + assert_eq!(row.get::("summary_json"), r#"{"id":"kept"}"#); Ok(()) } async fn insert_minimal_run( pool: &fabro_db::DbPool, status: &str, - input_tokens: i64, - summary_json: &str, -) -> Result<(), sqlx::Error> { - insert_run_row( - pool, - &format!("run-{status}-{input_tokens}"), - None, - status, - input_tokens, - summary_json, - ) - .await -} - -async fn insert_run_with_id( - pool: &fabro_db::DbPool, - id: &str, - parent_id: Option<&str>, -) -> Result<(), sqlx::Error> { - insert_run_row( - pool, - id, - parent_id, - "submitted", - 0, - &format!(r#"{{"id":"{id}"}}"#), - ) - .await -} - -async fn insert_run_row( - pool: &fabro_db::DbPool, - id: &str, - parent_id: Option<&str>, - status: &str, - input_tokens: i64, + diff_additions: i64, summary_json: &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, ?, ?, 'title', ?, ?) + id, created_at_ms, last_event_at_ms, status, title, diff_additions, summary_json +) VALUES (?, 0, 0, ?, 'title', ?, ?) ", ) - .bind(id) + .bind(format!("run-{status}-{diff_additions}")) .bind(status) - .bind(parent_id) - .bind(input_tokens) + .bind(diff_additions) .bind(summary_json) .execute(pool) .await?;