Merge pull request #817 from fabro-sh/codex/run-record-sql-foundation

Add inactive SQL run event storage foundation

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_summary_store = Arc::new(fabro_store::RunSummaryStore::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_summary_store,
     ));
 
     let inventory = store

lib/apps/fabro-server/src/server.rs

@@ -2447,8 +2447,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
         })
         .context("load environments")?,
     );
-    let run_summaries =
-        store.attach_run_summary_store(Arc::new(RunSummaryStore::new(db_pool.clone())));
+    let run_summaries = store.run_summary_store();
     let auth_codes = Arc::new(AuthCodeStore::new(db_pool.clone()));
     let auth_sessions = Arc::new(AuthSessionStore::new(db_pool.clone()));
     let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);

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<u32>,
+    },
+    #[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}")]

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<T> = std::result::Result<T, Box<dyn std::error::Error>>;
 
@@ -1153,6 +1153,7 @@ mod tests {
                 Duration::from_millis(1),
                 None,
                 Arc::clone(&target),
+                store_test_support::test_run_summary_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_summary_store(),
         );
 
         let mut connection = pool.acquire().await?;

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};
@@ -45,7 +45,7 @@ pub struct Database {
     catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
     projection_cache: Arc<RunProjectionCache>,
     projection_cache_warmed: Arc<OnceCell<()>>,
-    run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
+    run_summary_store: Arc<RunSummaryStore>,
 }
 
 impl std::fmt::Debug for Database {
@@ -65,6 +65,7 @@ impl Database {
         flush_interval: Duration,
         cache_path: Option<PathBuf>,
         blobs: Arc<BlobStore>,
+        run_summary_store: Arc<RunSummaryStore>,
     ) -> 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_summary_store,
         }
     }
 
-    pub fn attach_run_summary_store(&self, store: Arc<RunSummaryStore>) -> Arc<RunSummaryStore> {
-        Arc::clone(self.run_summary_store.get_or_init(|| store))
-    }
-
-    fn run_summary_store(&self) -> Option<Arc<RunSummaryStore>> {
-        self.run_summary_store.get().cloned()
+    #[must_use]
+    pub fn run_summary_store(&self) -> Arc<RunSummaryStore> {
+        Arc::clone(&self.run_summary_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_summary_store(),
         )
         .await
     }
@@ -240,9 +238,7 @@ impl Database {
                         }
                     }
                 }
-                if let Some(store) = self.run_summary_store() {
-                    store.reconcile(&entries).await?;
-                }
+                self.run_summary_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_summary_store.delete(run_id).await?;
         Ok(())
     }
 
@@ -558,6 +552,21 @@ mod tests {
         (object_store, store)
     }
 
+    fn make_store_with_run_summaries(
+        run_summaries: Arc<RunSummaryStore>,
+    ) -> (Arc<dyn ObjectStore>, Database) {
+        let object_store: Arc<dyn ObjectStore> = 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_summaries,
+        );
+        (object_store, store)
+    }
+
     #[tokio::test]
     async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() {
         let (_object_store, store) = make_store();
@@ -596,8 +605,8 @@ mod tests {
         );
     }
 
-    async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
-        let (directory, store) = store_test_support::sqlite_summary_store().await;
+    async fn make_run_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
+        let (directory, store) = store_test_support::sqlite_run_summary_store().await;
         (directory, Arc::new(store))
     }
 
@@ -972,9 +981,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_summary_store().await;
+        let (_object_store, store) = make_store_with_run_summaries(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 +1003,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_summary_update_failure_keeps_slate_append_repairable() {
+        let (directory, summaries) = make_run_summary_store().await;
+        let (object_store, store) = make_store_with_run_summaries(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 +1029,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_summary_store_at(directory.path()).await);
         let stale = repaired_summaries
             .get(&run_id, Utc::now())
             .await
@@ -1030,13 +1037,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 +1665,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_summary_append_refreshes_cache_and_delete_removes_rows() {
+        let (_directory, summaries) = make_run_summary_store().await;
+        let (_object_store, store) = make_store_with_run_summaries(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 +1841,20 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn projection_cache_warmup_backfills_sqlite_run_summaries() {
+    async fn required_run_summary_warmup_backfills_sqlite_run_summaries() {
         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_summary_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

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;
@@ -45,9 +45,7 @@ pub(crate) struct RunDatabaseInner {
     state_lock: Mutex<()>,
     projection_cache: Mutex<EventProjectionCache>,
     shared_projection_cache: Arc<RunProjectionCache>,
-    // 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<OnceLock<Arc<RunSummaryStore>>>,
+    run_summary_store: Arc<RunSummaryStore>,
     recent_events: Mutex<VecDeque<EventEnvelope>>,
     recent_event_limit: usize,
     event_tx: broadcast::Sender<EventEnvelope>,
@@ -60,7 +58,7 @@ impl RunDatabase {
         read_only: bool,
         blob_store: Arc<BlobStore>,
         shared_projection_cache: Arc<RunProjectionCache>,
-        run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
+        run_summary_store: Arc<RunSummaryStore>,
     ) -> Result<Self> {
         let cached_projection = shared_projection_cache.projection_snapshot(&run_id).await;
         let projection_cache = cached_projection.as_ref().map_or_else(
@@ -231,15 +229,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_summary_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"
+            );
         }
     }
 

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, RunSummaryStore};
 
 /// Returns an isolated SQLite blob authority backed by its own in-memory
 /// database.
@@ -18,32 +18,48 @@ use crate::{BlobStore, Database, Result};
 /// by other tests in the same process. Reopen-style tests that model one
 /// process-wide blob authority across several store handles should call this
 /// once and share the result through [`test_database_with_blobs`].
-///
-/// The pool connects lazily so synchronous fixture builders can remain
-/// synchronous. Its single connection installs the production blob schema on
-/// first use.
 #[must_use]
 pub fn test_blob_store() -> Arc<BlobStore> {
+    Arc::new(BlobStore::new(lazy_in_memory_pool(&[
+        fabro_db::BLOBS_MIGRATION_SQL,
+    ])))
+}
+
+/// Returns an isolated SQLite run-summary store backed by its own in-memory
+/// database and the production `runs` and `run_events` schemas.
+#[must_use]
+pub fn test_run_summary_store() -> Arc<RunSummaryStore> {
+    Arc::new(RunSummaryStore::new(lazy_in_memory_pool(&[
+        fabro_db::RUNS_MIGRATION_SQL,
+        fabro_db::RUN_EVENTS_MIGRATION_SQL,
+    ])))
+}
+
+/// Builds a single-connection in-memory SQLite pool that installs
+/// `migrations` on first use.
+///
+/// The pool connects lazily so synchronous fixture builders can remain
+/// synchronous.
+fn lazy_in_memory_pool(migrations: &'static [&'static str]) -> sqlx::SqlitePool {
     let options = SqliteConnectOptions::new()
         .filename(":memory:")
         .foreign_keys(true);
-    let pool = SqlitePoolOptions::new()
+    SqlitePoolOptions::new()
         .max_connections(1)
         // A single in-memory test connection never needs reaping. Disabling
         // both timers also keeps this lazy fixture constructible from sync
         // tests, where SQLx has no Tokio runtime for maintenance tasks.
         .max_lifetime(None)
         .idle_timeout(None)
-        .after_connect(|connection, _metadata| {
+        .after_connect(move |connection, _metadata| {
             Box::pin(async move {
-                sqlx::query(fabro_db::BLOBS_MIGRATION_SQL)
-                    .execute(&mut *connection)
-                    .await?;
+                for migration in migrations {
+                    sqlx::raw_sql(*migration).execute(&mut *connection).await?;
+                }
                 Ok(())
             })
         })
-        .connect_lazy_with(options);
-    Arc::new(BlobStore::new(pool))
+        .connect_lazy_with(options)
 }
 
 /// Returns the SQLite file backing [`test_blob_store_at`] for `store_dir`.
@@ -119,7 +135,37 @@ pub fn test_database_with_blobs(
     cache_path: Option<PathBuf>,
     blobs: Arc<BlobStore>,
 ) -> 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_summary_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<dyn ObjectStore>,
+    base_prefix: impl Into<String>,
+    flush_interval: Duration,
+    cache_path: Option<PathBuf>,
+    blobs: Arc<BlobStore>,
+    run_summaries: Arc<RunSummaryStore>,
+) -> Database {
+    Database::new(
+        object_store,
+        base_prefix,
+        flush_interval,
+        cache_path,
+        blobs,
+        run_summaries,
+    )
 }
 
 /// Seeds one canonical row in the legacy SlateDB blob keyspace.
@@ -171,13 +217,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_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
     let directory = tempfile::tempdir().unwrap();
-    let store = sqlite_summary_store_at(directory.path()).await;
+    let store = sqlite_run_summary_store_at(directory.path()).await;
     (directory, store)
 }
 
 #[cfg(test)]
-pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore {
+pub(crate) async fn sqlite_run_summary_store_at(directory: &Path) -> RunSummaryStore {
     RunSummaryStore::new(sqlite_test_pool(directory).await)
 }

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';

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 summary 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,

lib/foundation/fabro-db/tests/sqlite.rs

@@ -639,22 +639,283 @@ 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::<String, _>("name"),
+                column.get::<String, _>("type"),
+                column.get::<i64, _>("notnull"),
+                column.get::<i64, _>("pk"),
+            )
+        })
+        .collect::<Vec<_>>();
+    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::<String, _>("table"), "runs");
+    assert_eq!(foreign_keys[0].get::<String, _>("from"), "run_id");
+    assert_eq!(foreign_keys[0].get::<String, _>("to"), "id");
+    assert_eq!(foreign_keys[0].get::<String, _>("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::<String, _>("name");
+            name.starts_with("run_events_by_").then_some((
+                name,
+                index.get::<i64, _>("unique"),
+                index.get::<i64, _>("partial"),
+            ))
+        })
+        .collect::<Vec<_>>();
+    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::<i64, _>("unique") == 0
+            || index.get::<String, _>("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,
+    ] {
+        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<String> =
+        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::<String, _>("detail"))
+            .collect::<Vec<_>>()
+            .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::<String, _>("detail"))
+    .collect::<Vec<_>>()
+    .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(())
+}
+
+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 (?, ?, ?, '{}')
+",
+    )
+    .bind(run_id)
+    .bind(seq)
+    .bind(event_name)
+    .execute(pool)
+    .await?;
+    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,
+    summary_json: &str,
 ) -> Result<(), sqlx::Error> {
     sqlx::query(
         r"
 INSERT INTO runs (
-    id, source_last_seq, created_at_ms, last_event_at_ms, status, title,
+    id, source_last_seq, created_at_ms, last_event_at_ms, status, parent_id, title,
     input_tokens, summary_json
-) VALUES (?, 1, 0, 0, ?, 'title', ?, ?)
+) VALUES (?, 1, 0, 0, ?, ?, 'title', ?, ?)
 ",
     )
-    .bind(format!("run-{status}-{input_tokens}"))
+    .bind(id)
     .bind(status)
+    .bind(parent_id)
     .bind(input_tokens)
     .bind(summary_json)
     .execute(pool)