Simplify run handle construction and blob store test fixtures

- Collapse RunDatabase::open_writer/open_reader wrappers into one
  pub(crate) build, with a Database::open_run_database helper that
  gathers the shared-store dependencies in one place
- Stop fetching the blob store on open_run's active-cache hit path
- Share a raw-db test fixture between the two BlobStore raw-key tests
- Evict the cached writer in open_run_reader_is_read_only so the test
  exercises the real reader construction path

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

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

@@ -103,6 +103,16 @@ mod tests {
         db.blobs().await.unwrap()
     }
 
+    async fn raw_store(name: &str) -> (Arc<slatedb::Db>, BlobStore) {
+        let raw_db = Arc::new(
+            slatedb::Db::open(name, Arc::new(InMemory::new()))
+                .await
+                .unwrap(),
+        );
+        let store = BlobStore::new(Arc::clone(&raw_db));
+        (raw_db, store)
+    }
+
     #[tokio::test]
     async fn writes_reads_and_checks_existence() {
         let store = store().await;
@@ -141,12 +151,7 @@ mod tests {
 
     #[tokio::test]
     async fn list_skips_malformed_blob_ids() {
-        let raw_db = Arc::new(
-            slatedb::Db::open("blob-store-list-tests", Arc::new(InMemory::new()))
-                .await
-                .unwrap(),
-        );
-        let store = BlobStore::new(Arc::clone(&raw_db));
+        let (raw_db, store) = raw_store("blob-store-list-tests").await;
         let id = store.write(b"valid").await.unwrap();
 
         raw_db
@@ -162,12 +167,7 @@ mod tests {
 
     #[tokio::test]
     async fn raw_db_reads_exact_blob_bytes() {
-        let raw_db = Arc::new(
-            slatedb::Db::open("blob-store-tests", Arc::new(InMemory::new()))
-                .await
-                .unwrap(),
-        );
-        let store = BlobStore::new(Arc::clone(&raw_db));
+        let (raw_db, store) = raw_store("blob-store-tests").await;
         let bytes = b"{\"ok\":true}";
         let id = store.write(bytes).await.unwrap();
 

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

@@ -143,20 +143,23 @@ impl Database {
             .map(RunDatabase::from_inner)
     }
 
-    pub async fn create_run(&self, run_id: &RunId) -> Result<RunDatabase> {
-        self.warm_projection_cache().await?;
-        let db = self.open_db().await?;
-        let blob_store = self.blobs().await?;
-
-        self.catalog_index().await?.add(run_id).await?;
-        let run_store = RunDatabase::open_writer(
+    /// Builds a run handle wired to the Database-owned shared stores.
+    async fn open_run_database(&self, run_id: &RunId, read_only: bool) -> Result<RunDatabase> {
+        RunDatabase::build(
             *run_id,
-            db,
-            blob_store,
+            self.open_db().await?,
+            read_only,
+            self.blobs().await?,
             Arc::clone(&self.projection_cache),
             Arc::clone(&self.run_summary_store),
         )
-        .await?;
+        .await
+    }
+
+    pub async fn create_run(&self, run_id: &RunId) -> Result<RunDatabase> {
+        self.warm_projection_cache().await?;
+        self.catalog_index().await?.add(run_id).await?;
+        let run_store = self.open_run_database(run_id, false).await?;
         let mut active_runs = self.active_runs.lock().await;
         Self::cache_active_run(&mut active_runs, &run_store);
         Ok(run_store)
@@ -165,7 +168,6 @@ impl Database {
     pub async fn open_run(&self, run_id: &RunId) -> Result<RunDatabase> {
         self.warm_projection_cache().await?;
         let db = self.open_db().await?;
-        let blob_store = self.blobs().await?;
         // Keep the active-writer miss and insert atomic. Otherwise concurrent
         // callers can create independent writers with the same recovered seq.
         let mut active_runs = self.active_runs.lock().await;
@@ -181,14 +183,7 @@ impl Database {
         if !RunDatabase::has_any_events(&db, run_id).await? {
             return Err(Error::RunNotFound(run_id.to_string()));
         }
-        let run_store = RunDatabase::open_writer(
-            *run_id,
-            db,
-            blob_store,
-            Arc::clone(&self.projection_cache),
-            Arc::clone(&self.run_summary_store),
-        )
-        .await?;
+        let run_store = self.open_run_database(run_id, false).await?;
         Self::cache_active_run(&mut active_runs, &run_store);
         Ok(run_store)
     }
@@ -206,15 +201,7 @@ impl Database {
         if !RunDatabase::has_any_events(&db, run_id).await? {
             return Err(Error::RunNotFound(run_id.to_string()));
         }
-        let blob_store = self.blobs().await?;
-        RunDatabase::open_reader(
-            *run_id,
-            db,
-            blob_store,
-            Arc::clone(&self.projection_cache),
-            Arc::clone(&self.run_summary_store),
-        )
-        .await
+        self.open_run_database(run_id, true).await
     }
 
     pub async fn list_runs(&self, query: &ListRunsQuery, now: DateTime<Utc>) -> Result<Vec<Run>> {
@@ -866,6 +853,10 @@ mod tests {
         let blob = br#"{"summary":"readable"}"#;
         let blob_id = run.write_blob(blob).await.unwrap();
 
+        // Evict the cached writer so the reader is built through the real
+        // `open_run_reader` construction path, not a clone of the writer.
+        let _ = store.remove_active_run(&test_run_id("run-1")).await;
+
         let reader = store.open_run_reader(&test_run_id("run-1")).await.unwrap();
         assert_eq!(
             reader.read_blob(&blob_id).await.unwrap().as_deref(),

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

@@ -54,43 +54,7 @@ pub(crate) struct RunDatabaseInner {
 }
 
 impl RunDatabase {
-    pub(crate) async fn open_writer(
-        run_id: RunId,
-        db: Db,
-        blob_store: Arc<BlobStore>,
-        shared_projection_cache: Arc<RunProjectionCache>,
-        run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
-    ) -> Result<Self> {
-        Self::build(
-            run_id,
-            db,
-            false,
-            blob_store,
-            shared_projection_cache,
-            run_summary_store,
-        )
-        .await
-    }
-
-    pub(crate) async fn open_reader(
-        run_id: RunId,
-        db: Db,
-        blob_store: Arc<BlobStore>,
-        shared_projection_cache: Arc<RunProjectionCache>,
-        run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
-    ) -> Result<Self> {
-        Self::build(
-            run_id,
-            db,
-            true,
-            blob_store,
-            shared_projection_cache,
-            run_summary_store,
-        )
-        .await
-    }
-
-    async fn build(
+    pub(crate) async fn build(
         run_id: RunId,
         db: Db,
         read_only: bool,