From da368366f0d8cf3a8f9cc32b2f276005eacf7216 Mon Sep 17 00:00:00 2001
From: Bryan Helmkamp <bryan@brynary.com>
Date: Sun, 10 May 2026 13:38:06 -0400
Subject: [PATCH] fix(store): hydrate projection cache before appending later
 events

---
 lib/crates/fabro-store/src/slate/mod.rs       | 73 +++++++++++++++++++
 lib/crates/fabro-store/src/slate/run_store.rs | 32 +++++++-
 2 files changed, 103 insertions(+), 2 deletions(-)

diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs
index 989d3fc5e..c422ee244 100644
--- a/lib/crates/fabro-store/src/slate/mod.rs
+++ b/lib/crates/fabro-store/src/slate/mod.rs
@@ -1083,4 +1083,77 @@ mod tests {
                 .is_empty()
         );
     }
+
+    #[tokio::test]
+    async fn append_event_hydrates_local_projection_cache_for_fresh_writer() {
+        let (object_store, store) = make_store();
+        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;
+        run.append_event(&event_payload(
+            "run-1",
+            "2026-03-27T12:00:01Z",
+            "run.queued",
+            &serde_json::json!({}),
+        ))
+        .await
+        .unwrap();
+        run.append_event(&event_payload(
+            "run-1",
+            "2026-03-27T12:00:02Z",
+            "run.starting",
+            &serde_json::json!({}),
+        ))
+        .await
+        .unwrap();
+        run.append_event(&event_payload(
+            "run-1",
+            "2026-03-27T12:00:03Z",
+            "run.running",
+            &serde_json::json!({}),
+        ))
+        .await
+        .unwrap();
+        run.append_event(&event_payload(
+            "run-1",
+            "2026-03-27T12:00:04Z",
+            "run.failed",
+            &serde_json::json!({
+                "error": "workflow failed",
+                "duration_ms": 1,
+                "reason": "workflow_error",
+            }),
+        ))
+        .await
+        .unwrap();
+
+        let reopened = Database::new(
+            Arc::clone(&object_store),
+            "runs/",
+            Duration::from_millis(1),
+            None,
+        );
+        let fresh_writer = reopened.open_run(&run_id).await.unwrap();
+        fresh_writer
+            .append_event(&event_payload(
+                "run-1",
+                "2026-03-27T12:00:05Z",
+                "run.title.updated",
+                &serde_json::json!({ "title": "Renamed failed run" }),
+            ))
+            .await
+            .unwrap();
+
+        let state = fresh_writer.state().await.unwrap();
+        assert_eq!(state.title, "Renamed failed run");
+        assert_eq!(state.status, RunStatus::Failed {
+            reason: FailureReason::WorkflowError,
+        });
+
+        let cached = reopened.get_cached_run(&run_id).await.unwrap().unwrap();
+        assert_eq!(cached.summary.title, "Renamed failed run");
+        assert_eq!(cached.summary.status, RunStatus::Failed {
+            reason: FailureReason::WorkflowError,
+        });
+    }
 }
diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs
index e001fe607..6051063c3 100644
--- a/lib/crates/fabro-store/src/slate/run_store.rs
+++ b/lib/crates/fabro-store/src/slate/run_store.rs
@@ -174,8 +174,14 @@ impl RunDatabase {
     async fn cache_event(&self, event: &EventEnvelope) -> Result<()> {
         {
             let mut projection_cache = self.inner.projection_cache.lock().await;
-            apply_cached_projection_event(&mut projection_cache.state, event)?;
-            projection_cache.last_seq = event.seq;
+            if projection_cache.state.is_none() && event.seq > 1 {
+                drop(projection_cache);
+                self.rebuild_local_projection_cache_through(event.seq)
+                    .await?;
+            } else {
+                apply_cached_projection_event(&mut projection_cache.state, event)?;
+                projection_cache.last_seq = event.seq;
+            }
         }
         let mut recent_events = self.inner.recent_events.lock().await;
         recent_events.push_back(event.clone());
@@ -186,6 +192,28 @@ impl RunDatabase {
         Ok(())
     }
 
+    async fn rebuild_local_projection_cache_through(&self, seq: u32) -> Result<()> {
+        let events = list_events_from(&self.inner.db, &self.inner.run_id, 1).await?;
+        let Some(last_seq) = events.last().map(|event| event.seq) else {
+            return Err(Error::InvalidEvent(format!(
+                "run {} has no events while rebuilding projection cache",
+                self.inner.run_id
+            )));
+        };
+        if last_seq < seq {
+            return Err(Error::InvalidEvent(format!(
+                "run {} projection cache rebuild stopped at seq {last_seq}, before appended seq {seq}",
+                self.inner.run_id
+            )));
+        }
+
+        let state = RunProjection::apply_events(&events)?;
+        let mut projection_cache = self.inner.projection_cache.lock().await;
+        projection_cache.state = Some(state);
+        projection_cache.last_seq = last_seq;
+        Ok(())
+    }
+
     async fn cached_events_from(&self, start_seq: u32, limit: usize) -> Option<Vec<EventEnvelope>> {
         let recent_events = self.inner.recent_events.lock().await;
         let oldest_seq = recent_events.front().map(|event| event.seq)?;