Simplify session owner lookup and migration preflight

- Drop the post-decode run_id/session_id/event-body checks in
  find_session_owner: decode_event_row already verifies every stored
  column against the decoded envelope, and the WHERE clause pins
  session_id and event_name to the requested values.
- Build the lookup query from SELECT_EVENT_COLUMNS like the sibling
  event queries instead of duplicating the column list.
- Carry the stored run_id text in the unparseable-id error instead of
  an "<invalid>" placeholder.
- Fetch applied migration versions once per migrate() and share the
  set between the session-owner preflight and the pre-migration
  snapshot; check the applied version first so steady-state startups
  skip the sqlite_master probe. Mark the preflight as removable with
  the run-history compatibility window.
- Restore session_by_id_key as a #[cfg(test)] helper so tests stop
  hand-rolling the legacy reverse-index key shape.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

lib/components/fabro-store/src/keys.rs

@@ -120,6 +120,11 @@ pub(crate) fn sessions_by_id_prefix() -> SlateKey {
     SlateKey::new("sessions").with("by-id").into_prefix()
 }
 
+#[cfg(test)]
+pub(crate) fn session_by_id_key(session_id: &fabro_types::SessionId) -> SlateKey {
+    SlateKey::new("sessions").with("by-id").with(session_id)
+}
+
 // --- Parsing ---
 
 #[cfg(test)]
@@ -186,9 +191,7 @@ mod tests {
         // Sibling namespaces of the same run sort outside the range.
         assert!(!contains(&SlateKey::new("runs").with(run_id).with("state")));
         assert!(!contains(
-            &SlateKey::new("sessions")
-                .with("by-id")
-                .with(fabro_types::SessionId::new())
+            &session_by_id_key(&fabro_types::SessionId::new())
         ));
     }
 

lib/components/fabro-store/src/legacy_run_history_import.rs

@@ -1574,7 +1574,7 @@ mod tests {
             .await?;
         context
             .put_raw(
-                SlateKey::new("sessions").with("by-id").with(session_id),
+                keys::session_by_id_key(&session_id),
                 b"opaque legacy reverse row",
             )
             .await?;

lib/components/fabro-store/src/run_summary_store.rs

@@ -3,8 +3,8 @@ use std::sync::LazyLock;
 
 use chrono::{DateTime, Utc};
 use fabro_types::{
-    BilledTokenCounts, EventBody, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind,
-    RunTiming, SessionId, StageId, timing,
+    BilledTokenCounts, EventEnvelope, Run, RunEvent, RunId, RunSize, RunStatusKind, RunTiming,
+    SessionId, StageId, timing,
 };
 use sqlx::pool::PoolConnection;
 use sqlx::query::Query;
@@ -436,17 +436,12 @@ ON CONFLICT(singleton) DO NOTHING
     }
 
     pub(crate) async fn find_session_owner(&self, session_id: &SessionId) -> Result<Option<RunId>> {
-        let row = sqlx::query(
-            r"
-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'
-",
-        )
-        .bind(session_id.to_string())
-        .fetch_optional(&self.pool)
-        .await?;
+        let mut query = QueryBuilder::<Sqlite>::new(SELECT_EVENT_COLUMNS);
+        query
+            .push(" WHERE session_id = ")
+            .push_bind(session_id.to_string())
+            .push(" AND event_name = 'run.session.created'");
+        let row = query.build().fetch_optional(&self.pool).await?;
         let Some(row) = row else {
             return Ok(None);
         };
@@ -455,21 +450,14 @@ WHERE session_id = ?
         let run_id = stored_run_id
             .parse::<RunId>()
             .map_err(|_| Error::RunEventMismatch {
-                run_id: "<invalid>".to_string(),
+                run_id: stored_run_id.clone(),
                 seq:    0,
                 field:  "run_id",
             })?;
-        let envelope = decode_event_row(&row, &run_id, &stored_run_id)?;
-        if envelope.event.run_id != run_id {
-            return Err(run_event_mismatch(&run_id, envelope.seq, "run_id"));
-        }
-        let requested_session_id = session_id.to_string();
-        if envelope.event.session_id.as_deref() != Some(requested_session_id.as_str()) {
-            return Err(run_event_mismatch(&run_id, envelope.seq, "session_id"));
-        }
-        if !matches!(envelope.event.body, EventBody::RunSessionCreated(_)) {
-            return Err(run_event_mismatch(&run_id, envelope.seq, "event_name"));
-        }
+        // The WHERE clause pins the row's session_id and event_name columns
+        // to the requested values, and decoding verifies the envelope against
+        // every stored column, so a successful decode proves ownership.
+        decode_event_row(&row, &run_id, &stored_run_id)?;
         Ok(Some(run_id))
     }
 

lib/components/fabro-store/src/slate/mod.rs

@@ -888,11 +888,7 @@ mod tests {
             Some(first_id)
         );
 
-        let legacy_key = keys::SlateKey::new("sessions")
-            .with("by-id")
-            .with(session_id)
-            .as_ref()
-            .to_vec();
+        let legacy_key = keys::session_by_id_key(&session_id).as_ref().to_vec();
         let legacy = store.open_db().await.unwrap();
         assert!(legacy.get(&legacy_key).await.unwrap().is_none());
         legacy

lib/foundation/fabro-db/src/lib.rs

@@ -73,10 +73,11 @@ impl Database {
     }
 
     pub async fn migrate(&self) -> anyhow::Result<()> {
-        self.preflight_session_owner_index()
+        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()
+        self.snapshot_before_new_migrations(&applied)
             .await
             .context("snapshotting SQLite database before migrations")?;
         MIGRATOR
@@ -88,7 +89,16 @@ impl Database {
     /// 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.
-    async fn preflight_session_owner_index(&self) -> anyhow::Result<()> {
+    ///
+    /// 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<i64>) -> 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')",
         )
@@ -99,11 +109,6 @@ impl Database {
             return Ok(());
         }
 
-        let applied = applied_migration_versions(&self.pool).await?;
-        if applied.contains(&SESSION_OWNER_INDEX_MIGRATION_VERSION) {
-            return Ok(());
-        }
-
         let collision_groups: i64 = sqlx::query_scalar(
             r"
 SELECT COUNT(*)
@@ -145,8 +150,7 @@ FROM (
     /// from immediately before the most recent schema change. Failing to
     /// write the snapshot fails the migration: no rollback artifact, no
     /// schema change.
-    async fn snapshot_before_new_migrations(&self) -> anyhow::Result<()> {
-        let applied = applied_migration_versions(&self.pool).await?;
+    async fn snapshot_before_new_migrations(&self, applied: &HashSet<i64>) -> anyhow::Result<()> {
         let has_pending = MIGRATOR
             .iter()
             .any(|migration| !applied.contains(&migration.version));