Merge pull request #829 from fabro-sh/codex/session-ownership-index

Index session ownership from creation events

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

@@ -218,9 +218,9 @@ async fn append_run_event(
     if event.run_id != id {
         return ApiError::bad_request("Event run_id does not match path run ID.").into_response();
     }
-    if let Some(denied) = denied_lifecycle_event_name(&event.body) {
+    if let Some(denied) = denied_dedicated_operation_event_name(&event.body) {
         return ApiError::bad_request(format!(
-            "{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event"
+            "{denied} must be performed through its dedicated operation endpoint instead of injecting it via append_run_event"
         ))
         .into_response();
     }
@@ -576,7 +576,7 @@ async fn attach_run_events(
 /// (e.g. "archive only from terminal") that a direct event append would
 /// bypass. Other run-lifecycle events flow through this endpoint legitimately:
 /// the worker subprocess emits state transitions during execution.
-fn denied_lifecycle_event_name(body: &EventBody) -> Option<&str> {
+fn denied_dedicated_operation_event_name(body: &EventBody) -> Option<&str> {
     match body {
         EventBody::RunArchived(_)
         | EventBody::RunUnarchived(_)
@@ -585,7 +585,8 @@ fn denied_lifecycle_event_name(body: &EventBody) -> Option<&str> {
         | EventBody::RunPauseRequested(_)
         | EventBody::RunUnpauseRequested(_)
         | EventBody::PullRequestLinked(_)
-        | EventBody::PullRequestUnlinked(_) => Some(body.event_name()),
+        | EventBody::PullRequestUnlinked(_)
+        | EventBody::RunSessionCreated(_) => Some(body.event_name()),
         _ => None,
     }
 }

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

@@ -184,14 +184,6 @@ async fn create_run_session(
 
     let session_id = SessionId::new();
     let now = Utc::now();
-    if let Err(err) = state
-        .store_ref()
-        .put_session_run_index(&session_id, &run_id)
-        .await
-    {
-        return store_error(&err).into_response();
-    }
-
     let event = match append_run_session_event(
         &run_store,
         run_id,
@@ -1388,7 +1380,7 @@ async fn load_session(
     state: &AppState,
     session_id: SessionId,
 ) -> Result<(RunId, RunDatabase, ProjectedRunSession), Response> {
-    let run_id = match state.store_ref().get_session_run_id(&session_id).await {
+    let run_id = match state.store_ref().find_session_owner(&session_id).await {
         Ok(Some(run_id)) => run_id,
         Ok(None) => return Err(ApiError::not_found("Session not found.").into_response()),
         Err(err) => return Err(store_error(&err).into_response()),
@@ -1408,7 +1400,7 @@ async fn load_session_read(
     state: &AppState,
     session_id: SessionId,
 ) -> Result<(RunId, ProjectedRunSession), Response> {
-    let run_id = match state.store_ref().get_session_run_id(&session_id).await {
+    let run_id = match state.store_ref().find_session_owner(&session_id).await {
         Ok(Some(run_id)) => run_id,
         Ok(None) => return Err(ApiError::not_found("Session not found.").into_response()),
         Err(err) => return Err(store_error(&err).into_response()),
@@ -1428,7 +1420,7 @@ async fn load_session_run_reader(
     state: &AppState,
     session_id: SessionId,
 ) -> Result<(RunId, RunDatabase), Response> {
-    let run_id = match state.store_ref().get_session_run_id(&session_id).await {
+    let run_id = match state.store_ref().find_session_owner(&session_id).await {
         Ok(Some(run_id)) => run_id,
         Ok(None) => return Err(ApiError::not_found("Session not found.").into_response()),
         Err(err) => return Err(store_error(&err).into_response()),

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

@@ -12188,10 +12188,11 @@ async fn append_run_event_rejects_reserved_archive_event() {
     let response = app.oneshot(req).await.unwrap();
     let body = response_json!(response, StatusCode::BAD_REQUEST).await;
     assert!(
-        body["errors"][0]["detail"]
-            .as_str()
-            .is_some_and(|message| message.contains("run.archived is a lifecycle event")),
-        "expected lifecycle rejection, got: {body}"
+        body["errors"][0]["detail"].as_str().is_some_and(|message| {
+            message
+                .contains("run.archived must be performed through its dedicated operation endpoint")
+        }),
+        "expected dedicated-operation rejection, got: {body}"
     );
 }
 

lib/apps/fabro-server/tests/it/api/sessions.rs

@@ -133,6 +133,65 @@ async fn run_bound_session_is_created_as_run_event_and_resolves_by_flat_id() {
     );
 }
 
+#[tokio::test]
+async fn generic_session_creation_requires_the_dedicated_operation_without_advancing_history() {
+    let app = fabro_server::test_support::build_test_router(test_app_state());
+    let run_id = create_run(&app).await;
+    let before_request = Request::builder()
+        .method("GET")
+        .uri(api(&format!("/runs/{run_id}/events")))
+        .body(Body::empty())
+        .expect("run-events request should build");
+    let before = response_json(
+        app.clone().oneshot(before_request).await.unwrap(),
+        StatusCode::OK,
+        format!("GET /api/v1/runs/{run_id}/events before rejected append"),
+    )
+    .await;
+
+    let session_id = fabro_types::SessionId::new();
+    let request = Request::builder()
+        .method("POST")
+        .uri(api(&format!("/runs/{run_id}/events")))
+        .header("content-type", "application/json")
+        .body(Body::from(
+            serde_json::to_string(&serde_json::json!({
+                "id": ulid::Ulid::new().to_string(),
+                "ts": "2026-08-31T12:00:00Z",
+                "run_id": run_id,
+                "event": "run.session.created",
+                "session_id": session_id,
+                "properties": { "title": "Injected session" },
+            }))
+            .expect("session creation event should serialize"),
+        ))
+        .expect("append-event request should build");
+    let rejected = response_json(
+        app.clone().oneshot(request).await.unwrap(),
+        StatusCode::BAD_REQUEST,
+        format!("POST /api/v1/runs/{run_id}/events with session creation"),
+    )
+    .await;
+    let detail = rejected["errors"][0]["detail"]
+        .as_str()
+        .expect("error response should include detail");
+    assert!(detail.contains("dedicated operation endpoint"));
+    assert!(detail.contains("run.session.created"));
+
+    let after_request = Request::builder()
+        .method("GET")
+        .uri(api(&format!("/runs/{run_id}/events")))
+        .body(Body::empty())
+        .expect("run-events request should build");
+    let after = response_json(
+        app.clone().oneshot(after_request).await.unwrap(),
+        StatusCode::OK,
+        format!("GET /api/v1/runs/{run_id}/events after rejected append"),
+    )
+    .await;
+    assert_eq!(after, before);
+}
+
 #[tokio::test]
 async fn sessions_are_listed_only_under_their_owning_run() {
     let app = fabro_server::test_support::build_test_router(test_app_state());

lib/apps/fabro-server/tests/it/scenario/archive.rs

@@ -183,8 +183,8 @@ async fn appending_run_archived_event_directly_is_rejected() {
     .await;
     let detail = body["errors"][0]["detail"].as_str().unwrap_or_default();
     assert!(
-        detail.contains("lifecycle event"),
-        "expected lifecycle rejection, got: {body}"
+        detail.contains("run.archived must be performed through its dedicated operation endpoint"),
+        "expected dedicated-operation rejection, got: {body}"
     );
 }
 

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

@@ -2,7 +2,7 @@ use std::fmt::{self, Write};
 #[cfg(test)]
 use std::ops::Range;
 
-use fabro_types::{RunId, SessionId};
+use fabro_types::RunId;
 
 pub(crate) const MAX_EVENT_SEQ: u32 = 999_999;
 
@@ -120,7 +120,8 @@ pub(crate) fn sessions_by_id_prefix() -> SlateKey {
     SlateKey::new("sessions").with("by-id").into_prefix()
 }
 
-pub(crate) fn session_by_id_key(session_id: &SessionId) -> SlateKey {
+#[cfg(test)]
+pub(crate) fn session_by_id_key(session_id: &fabro_types::SessionId) -> SlateKey {
     SlateKey::new("sessions").with("by-id").with(session_id)
 }
 

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

@@ -1396,6 +1396,24 @@ mod tests {
         event_value(run_id, seq, "run.submitted", &serde_json::json!({}))
     }
 
+    fn session_created_value(
+        run_id: &RunId,
+        seq: u32,
+        session_id: &SessionId,
+    ) -> serde_json::Value {
+        let mut value = event_value(
+            run_id,
+            seq,
+            "run.session.created",
+            &serde_json::json!({ "title": "Imported session" }),
+        );
+        value
+            .as_object_mut()
+            .unwrap()
+            .insert("session_id".to_string(), session_id.to_string().into());
+        value
+    }
+
     #[tokio::test]
     async fn legacy_source_identity_covers_exact_keys_and_json_bytes() -> TestResult<()> {
         let context = TestContext::new().await?;
@@ -1534,17 +1552,20 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn legacy_run_history_imports_exact_json_gaps_and_count_only_diagnostics()
+    async fn session_owner_imports_typed_event_and_counts_opaque_legacy_reverse_rows()
     -> TestResult<()> {
         let context = TestContext::new().await?;
         let first = run_id(1);
         let second = run_id(2);
         let empty_marker = run_id(3);
         let stale = run_id(4);
+        let session_id = SessionId::new();
         let first_created = serde_json::to_string_pretty(&created_value(&first, "first"))?;
+        let first_session = serde_json::to_string(&session_created_value(&first, 3, &session_id))?;
         let first_submitted = serde_json::to_string(&submitted_value(&first, 4))?;
         let second_created = serde_json::to_string(&created_value(&second, "second"))?;
         context.put_event(&first, 1, 10, &first_created).await?;
+        context.put_event(&first, 3, 30, &first_session).await?;
         context.put_event(&first, 4, 40, &first_submitted).await?;
         context.put_event(&second, 1, 20, &second_created).await?;
         context.put_raw(keys::run_catalog_key(&first), b"").await?;
@@ -1552,8 +1573,10 @@ mod tests {
             .put_raw(keys::run_catalog_key(&empty_marker), b"")
             .await?;
         context
-            .source
-            .put_session_run_index(&SessionId::new(), &first)
+            .put_raw(
+                keys::session_by_id_key(&session_id),
+                b"opaque legacy reverse row",
+            )
             .await?;
 
         let stale_created = serde_json::to_string(&created_value(&stale, "stale"))?;
@@ -1572,9 +1595,9 @@ mod tests {
 
         assert_eq!(report, LegacyRunHistoryImportReport {
             scanned_source_runs: 2,
-            scanned_source_events: 3,
+            scanned_source_events: 4,
             imported_runs: 2,
-            imported_events: 3,
+            imported_events: 4,
             discarded_projection_only_rows: 1,
             committed_run_transactions: 2,
             diagnostics: LegacyRunHistoryDiagnostics {
@@ -1589,7 +1612,16 @@ mod tests {
                 .bind(first.to_string())
                 .fetch_all(&context.sqlite)
                 .await?;
-        assert_eq!(stored, vec![(1, first_created), (4, first_submitted)]);
+        assert_eq!(stored, vec![
+            (1, first_created),
+            (3, first_session),
+            (4, first_submitted)
+        ]);
+        let summaries = RunSummaryStore::new(context.sqlite.clone());
+        assert_eq!(
+            summaries.find_session_owner(&session_id).await?,
+            Some(first)
+        );
         assert_eq!(
             sqlx::query_scalar::<_, i64>("SELECT source_last_seq FROM runs WHERE id = ?")
                 .bind(first.to_string())
@@ -1603,7 +1635,7 @@ mod tests {
             .verify_legacy_run_history_in(&context.sqlite)
             .await?;
         assert_eq!(verification.target_runs, 2);
-        assert_eq!(verification.target_events, 3);
+        assert_eq!(verification.target_events, 4);
         Ok(())
     }
 

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

@@ -435,6 +435,32 @@ ON CONFLICT(singleton) DO NOTHING
         .await
     }
 
+    pub(crate) async fn find_session_owner(&self, session_id: &SessionId) -> Result<Option<RunId>> {
+        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);
+        };
+
+        let stored_run_id: String = row.try_get("run_id")?;
+        let run_id = stored_run_id
+            .parse::<RunId>()
+            .map_err(|_| Error::RunEventMismatch {
+                run_id: stored_run_id.clone(),
+                seq:    0,
+                field:  "run_id",
+            })?;
+        // 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))
+    }
+
     pub(crate) async fn delete_canonical(&self, run_id: &RunId, deleted_at_ms: i64) -> Result<()> {
         // This transaction reads the activation marker before it writes the
         // tombstone and run deletion. A deferred SQLite transaction can fail
@@ -1536,6 +1562,17 @@ mod tests {
         )
     }
 
+    fn session_created_payload(run_id: &RunId, session_id: &SessionId) -> EventPayload {
+        sql_event_payload(
+            run_id,
+            "run.session.created",
+            None,
+            None,
+            Some(session_id),
+            serde_json::json!({ "title": "Owned session" }),
+        )
+    }
+
     async fn seed_sql_event(
         store: &RunSummaryStore,
         run_id: &RunId,
@@ -1807,6 +1844,122 @@ mod tests {
         assert_eq!(sequences, vec![1, 2, 3]);
     }
 
+    #[tokio::test]
+    async fn session_owner_lookup_resolves_only_typed_creation_events() {
+        let (_directory, store) = store().await;
+        let created_at = dt("2026-08-27T12:00:00Z");
+        let id = run_id(created_at.timestamp_millis().cast_unsigned(), 31);
+        store
+            .upsert_projection(&entry(projection(id, "owner", created_at), 3))
+            .await
+            .unwrap();
+
+        let owner_session = SessionId::new();
+        seed_sql_event(
+            &store,
+            &id,
+            2,
+            &session_created_payload(&id, &owner_session),
+        )
+        .await;
+        assert_eq!(
+            store.find_session_owner(&owner_session).await.unwrap(),
+            Some(id)
+        );
+
+        let non_owner_session = SessionId::new();
+        let non_creation = sql_event_payload(
+            &id,
+            "run.session.future",
+            None,
+            None,
+            Some(&non_owner_session),
+            serde_json::json!({ "kind": "future" }),
+        );
+        seed_sql_event(&store, &id, 3, &non_creation).await;
+        assert_eq!(
+            store.find_session_owner(&non_owner_session).await.unwrap(),
+            None
+        );
+        assert_eq!(
+            store.find_session_owner(&SessionId::new()).await.unwrap(),
+            None
+        );
+    }
+
+    #[tokio::test]
+    async fn session_owner_lookup_rejects_corrupt_selected_events() {
+        let (_directory, store) = store().await;
+        let created_at = dt("2026-08-27T12:00:00Z");
+        let id = run_id(created_at.timestamp_millis().cast_unsigned(), 32);
+        let other_id = run_id(created_at.timestamp_millis().cast_unsigned() + 1, 33);
+        store
+            .upsert_projection(&entry(projection(id, "owner", created_at), 2))
+            .await
+            .unwrap();
+        let session_id = SessionId::new();
+        let payload = session_created_payload(&id, &session_id);
+        seed_sql_event(&store, &id, 2, &payload).await;
+
+        let mut wrong_event_name = payload.as_value().clone();
+        wrong_event_name["event"] = "run.session.future".into();
+        sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2")
+            .bind(serde_json::to_string(&wrong_event_name).unwrap())
+            .bind(id.to_string())
+            .execute(&store.pool)
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.find_session_owner(&session_id).await.unwrap_err(),
+            Error::RunEventMismatch {
+                field: "event_name",
+                ..
+            }
+        ));
+
+        seed_sql_event_restore(&store, &id, 2, &payload).await;
+        let mut wrong_run = payload.as_value().clone();
+        wrong_run["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_run).unwrap())
+            .bind(id.to_string())
+            .execute(&store.pool)
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.find_session_owner(&session_id).await.unwrap_err(),
+            Error::RunEventMismatch {
+                field: "run_id",
+                ..
+            }
+        ));
+
+        seed_sql_event_restore(&store, &id, 2, &payload).await;
+        let mut wrong_session = payload.as_value().clone();
+        wrong_session["session_id"] = SessionId::new().to_string().into();
+        sqlx::query("UPDATE run_events SET event_json = ? WHERE run_id = ? AND seq = 2")
+            .bind(serde_json::to_string(&wrong_session).unwrap())
+            .bind(id.to_string())
+            .execute(&store.pool)
+            .await
+            .unwrap();
+        assert!(matches!(
+            store.find_session_owner(&session_id).await.unwrap_err(),
+            Error::RunEventMismatch {
+                field: "session_id",
+                ..
+            }
+        ));
+
+        seed_sql_event_restore(&store, &id, 2, &payload).await;
+        sqlx::query("UPDATE run_events SET event_json = '{}' WHERE run_id = ? AND seq = 2")
+            .bind(id.to_string())
+            .execute(&store.pool)
+            .await
+            .unwrap();
+        assert!(store.find_session_owner(&session_id).await.is_err());
+    }
+
     #[tokio::test]
     async fn canonical_delete_waits_for_a_concurrent_writer() {
         let (_directory, store) = store().await;

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

@@ -28,11 +28,6 @@ pub struct UnreadableRun {
     pub error:      String,
 }
 
-#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
-struct SessionRunIndexEntry {
-    run_id: RunId,
-}
-
 #[derive(Clone)]
 pub struct Database {
     object_store: Arc<dyn ObjectStore>,
@@ -374,27 +369,10 @@ impl Database {
         Ok(self.projection_cache.pending_pull_request_creations())
     }
 
-    pub async fn put_session_run_index(
-        &self,
-        session_id: &SessionId,
-        run_id: &RunId,
-    ) -> Result<()> {
-        let db = self.open_db().await?;
-        db.put(
-            keys::session_by_id_key(session_id),
-            serde_json::to_vec(&SessionRunIndexEntry { run_id: *run_id })?,
-        )
-        .await?;
-        Ok(())
-    }
-
-    pub async fn get_session_run_id(&self, session_id: &SessionId) -> Result<Option<RunId>> {
-        let db = self.open_db().await?;
-        if let Some(bytes) = db.get(keys::session_by_id_key(session_id)).await? {
-            let entry: SessionRunIndexEntry = serde_json::from_slice(&bytes)?;
-            return Ok(Some(entry.run_id));
-        }
-        Ok(None)
+    /// Resolves the run that owns `session_id` from the canonical typed
+    /// creation event stored in SQLite.
+    pub async fn find_session_owner(&self, session_id: &SessionId) -> Result<Option<RunId>> {
+        self.run_summary_store.find_session_owner(session_id).await
     }
 
     pub(crate) fn remove_cached_run(&self, run_id: &RunId) {
@@ -413,31 +391,6 @@ impl Database {
             .await?;
         active_runs.remove(run_id);
         self.remove_cached_run(run_id);
-        if let Err(err) = self.delete_session_indexes_for_run(run_id).await {
-            warn!(
-                run_id = %run_id,
-                error = %err,
-                "Failed to remove retired session reverse indexes after deleting run"
-            );
-        }
-        Ok(())
-    }
-
-    async fn delete_session_indexes_for_run(&self, run_id: &RunId) -> Result<()> {
-        let db = self.open_db().await?;
-        let mut keys_to_delete = Vec::new();
-        let mut iter = db.scan_prefix(keys::sessions_by_id_prefix()).await?;
-        while let Some(entry) = iter.next().await? {
-            let index: SessionRunIndexEntry = serde_json::from_slice(&entry.value)?;
-            if index.run_id == *run_id {
-                keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| {
-                    Error::Other(format!("stored key is not valid UTF-8: {err}"))
-                })?);
-            }
-        }
-        for key in keys_to_delete {
-            db.delete(key).await?;
-        }
         Ok(())
     }
 
@@ -695,6 +648,21 @@ mod tests {
         .unwrap()
     }
 
+    fn session_created_payload(label: &str, session_id: &SessionId) -> EventPayload {
+        EventPayload::new(
+            serde_json::json!({
+                "id": format!("evt-{label}-session-created"),
+                "ts": "2026-03-27T12:00:05Z",
+                "run_id": test_run_id(label).to_string(),
+                "event": "run.session.created",
+                "session_id": session_id,
+                "properties": { "title": "Owned session" },
+            }),
+            &test_run_id(label),
+        )
+        .unwrap()
+    }
+
     async fn append_created(run: &RunDatabase, label: &str, created_at: DateTime<Utc>) {
         let run_spec = sample_run_spec(label);
         run.append_event(&event_payload(
@@ -850,7 +818,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn create_open_list_and_delete_full_lifecycle_in_shared_db() {
+    async fn create_open_list_and_delete_full_lifecycle_without_legacy_slate_writes() {
         let (object_store, store) = make_store();
         let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
         let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
@@ -883,7 +851,100 @@ mod tests {
             .unwrap();
         assert_eq!(remaining.len(), 1);
         assert_eq!(remaining[0].id, test_run_id("run-2"));
-        assert!(!list_paths(object_store, "runs/").await.is_empty());
+        assert!(
+            list_paths(object_store, "runs/").await.is_empty(),
+            "canonical run lifecycle must not open SlateDB solely for retired session indexes"
+        );
+    }
+
+    #[tokio::test]
+    async fn session_owner_claims_are_atomic_durable_and_ignore_legacy_reverse_rows() {
+        let directory = tempfile::tempdir().unwrap();
+        let object_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+        let store = store_test_support::test_database_at(
+            Arc::clone(&object_store),
+            "session-owner",
+            Duration::from_millis(1),
+            None,
+            directory.path(),
+        );
+        let first_id = test_run_id("run-1");
+        let second_id = test_run_id("run-2");
+        let first = store.create_run(&first_id).await.unwrap();
+        let second = store.create_run(&second_id).await.unwrap();
+        append_created(&first, "run-1", dt("2026-03-27T12:00:00Z")).await;
+        append_created(&second, "run-2", dt("2026-03-27T12:00:10Z")).await;
+
+        let session_id = SessionId::new();
+        assert_eq!(
+            first
+                .append_event(&session_created_payload("run-1", &session_id))
+                .await
+                .unwrap(),
+            2
+        );
+        assert_eq!(
+            store.find_session_owner(&session_id).await.unwrap(),
+            Some(first_id)
+        );
+
+        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
+            .put(
+                &legacy_key,
+                serde_json::to_vec(&serde_json::json!({ "run_id": second_id })).unwrap(),
+            )
+            .await
+            .unwrap();
+        legacy.flush().await.unwrap();
+        assert_eq!(
+            store.find_session_owner(&session_id).await.unwrap(),
+            Some(first_id),
+            "legacy reverse rows must not influence ownership"
+        );
+
+        assert!(
+            first
+                .append_event(&session_created_payload("run-1", &session_id))
+                .await
+                .is_err()
+        );
+        assert_eq!(first.last_event_seq().await.unwrap(), Some(2));
+        assert!(
+            second
+                .append_event(&session_created_payload("run-2", &session_id))
+                .await
+                .is_err()
+        );
+        assert_eq!(second.last_event_seq().await.unwrap(), Some(1));
+        assert_eq!(
+            store.find_session_owner(&session_id).await.unwrap(),
+            Some(first_id)
+        );
+
+        let reopened = store_test_support::test_database_at(
+            object_store,
+            "session-owner",
+            Duration::from_millis(1),
+            None,
+            directory.path(),
+        );
+        assert_eq!(
+            reopened.find_session_owner(&session_id).await.unwrap(),
+            Some(first_id)
+        );
+
+        reopened.delete_run(&first_id).await.unwrap();
+        assert_eq!(
+            reopened.find_session_owner(&session_id).await.unwrap(),
+            None
+        );
+        assert!(
+            legacy.get(&legacy_key).await.unwrap().is_some(),
+            "legacy reverse rows remain diagnostic-only during the support window"
+        );
     }
 
     #[tokio::test]

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

@@ -33,6 +33,7 @@ pub fn test_run_summary_store() -> Arc<RunSummaryStore> {
         fabro_db::RUNS_MIGRATION_SQL,
         fabro_db::RUN_EVENTS_MIGRATION_SQL,
         fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL,
+        fabro_db::RUN_EVENT_SESSION_OWNER_MIGRATION_SQL,
     ])))
 }
 
@@ -106,6 +107,7 @@ pub fn test_run_summary_store_at(store_dir: &Path) -> Arc<RunSummaryStore> {
             fabro_db::RUNS_MIGRATION_SQL,
             fabro_db::RUN_EVENTS_MIGRATION_SQL,
             fabro_db::RUN_HISTORY_ACTIVATION_MIGRATION_SQL,
+            fabro_db::RUN_EVENT_SESSION_OWNER_MIGRATION_SQL,
         ],
     )))
 }

lib/foundation/fabro-db/migrations/2026083101_run_event_session_owner.sql

@@ -0,0 +1,4 @@
+CREATE UNIQUE INDEX run_events_by_session_owner
+ON run_events(session_id)
+WHERE session_id IS NOT NULL
+  AND event_name = 'run.session.created';

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

@@ -16,6 +16,8 @@ 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");
@@ -28,6 +30,11 @@ pub const RUNS_MIGRATION_SQL: &str = include_str!("../migrations/2026071104_runs
 /// 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 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 =
@@ -66,7 +73,11 @@ impl Database {
     }
 
     pub async fn migrate(&self) -> anyhow::Result<()> {
-        self.snapshot_before_new_migrations()
+        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")?;
         MIGRATOR
@@ -75,6 +86,53 @@ 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<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')",
+        )
+        .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
@@ -92,8 +150,7 @@ impl Database {
     /// 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));

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

@@ -785,7 +785,7 @@ async fn runs_schema_creates_indexes_and_rejects_invalid_rows() -> anyhow::Resul
 }
 
 #[tokio::test]
-async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::Result<()> {
+async fn session_owner_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?;
@@ -847,6 +847,7 @@ async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::
         })
         .collect::<Vec<_>>();
     assert_eq!(named_indexes, vec![
+        ("run_events_by_session_owner".to_string(), 1, 1),
         (
             "run_events_by_pull_request_creation_request".to_string(),
             0,
@@ -858,6 +859,7 @@ async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::
     ]);
     assert!(indexes.iter().all(|index| {
         index.get::<i64, _>("unique") == 0
+            || index.get::<String, _>("name") == "run_events_by_session_owner"
             || index.get::<String, _>("name") == "sqlite_autoindex_run_events_1"
     }));
 
@@ -912,7 +914,8 @@ async fn run_events_schema_has_final_shape_constraints_and_indexes() -> anyhow::
 }
 
 #[tokio::test]
-async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result<()> {
+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?;
@@ -938,6 +941,10 @@ async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result
             "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 * FROM run_events WHERE event_name = 'pull_request.creation_requested' ORDER BY run_id, seq",
             "run_events_by_pull_request_creation_request",
@@ -989,6 +996,99 @@ async fn run_events_schema_query_plans_use_candidate_indexes() -> anyhow::Result
     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,