feat: store durable main events

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

@@ -66,6 +66,16 @@ pub(crate) fn run_event_seq_prefix(run_id: &RunId, seq: u32) -> SlateKey {
         .with(format!("{seq:06}-"))
 }
 
+pub(crate) fn main_events_prefix() -> SlateKey {
+    SlateKey::new("events").with("main").into_prefix()
+}
+
+pub(crate) fn main_event_key(seq: u32, epoch_ms: i64) -> SlateKey {
+    SlateKey::new("events")
+        .with("main")
+        .with(format!("{seq:06}-{epoch_ms}"))
+}
+
 pub(crate) fn blobs_prefix() -> SlateKey {
     SlateKey::new("blobs").with("sha256").into_prefix()
 }
@@ -90,6 +100,17 @@ pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
     segments.next()?.split_once('-')?.0.parse().ok()
 }
 
+pub(crate) fn parse_main_event_seq(key: &str) -> Option<u32> {
+    let mut segments = SlateKey::segments(key);
+    if segments.next()? != "events" {
+        return None;
+    }
+    if segments.next()? != "main" {
+        return None;
+    }
+    segments.next()?.split_once('-')?.0.parse().ok()
+}
+
 pub(crate) fn parse_blob_id(key: &str) -> Option<RunBlobId> {
     let mut segments = SlateKey::segments(key);
     if segments.next()? != "blobs" {
@@ -136,6 +157,13 @@ mod tests {
         ]);
     }
 
+    #[test]
+    fn main_event_key_segments() {
+        let key = main_event_key(7, 123);
+        let segments: Vec<&str> = SlateKey::segments(key.as_str()).collect();
+        assert_eq!(segments, ["events", "main", "000007-123"]);
+    }
+
     #[test]
     fn blob_key_segments() {
         let blob_id = RunBlobId::new(b"summary");
@@ -165,6 +193,20 @@ mod tests {
         assert_eq!(parse_blob_id(key.as_str()), Some(blob_id));
     }
 
+    #[test]
+    fn parse_main_event_seq_roundtrip() {
+        assert_eq!(parse_main_event_seq(main_event_key(7, 123).as_str()), Some(7));
+        assert_eq!(
+            parse_main_event_seq(
+                SlateKey::new("events")
+                    .with("other")
+                    .with("000007-123")
+                    .as_str()
+            ),
+            None
+        );
+    }
+
     #[test]
     fn parse_helpers_reject_invalid_keys() {
         assert_eq!(

lib/crates/fabro-store/src/lib.rs

@@ -28,7 +28,8 @@ pub use run_state::RunProjectionReducer;
 pub use serializable_projection::SerializableProjection;
 pub use slate::{
     AuthCode, AuthCodeStore, Blob, BlobStore, CachedRunProjection, ConsumeOutcome, Database,
-    RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun,
+    MainEventStore, RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs,
+    UnreadableRun,
 };
 pub use types::EventPayload;
 

lib/crates/fabro-store/src/slate/main_event_store.rs

@@ -0,0 +1,125 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, Ordering};
+
+use bytes::Bytes;
+use chrono::Utc;
+use fabro_types::{MainEvent, MainEventBody, MainEventEnvelope};
+use slatedb::{Db, DbRead};
+use tokio::sync::{Mutex, broadcast};
+use uuid::Uuid;
+
+use crate::{Error, Result, keys};
+
+const DEFAULT_MAIN_EVENT_TAIL_LIMIT: usize = 1024;
+
+#[derive(Clone)]
+pub struct MainEventStore {
+    inner: Arc<MainEventStoreInner>,
+}
+
+impl std::fmt::Debug for MainEventStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MainEventStore")
+            .finish_non_exhaustive()
+    }
+}
+
+struct MainEventStoreInner {
+    db:          Db,
+    event_seq:   AtomicU32,
+    append_lock: Mutex<()>,
+    event_tx:    broadcast::Sender<MainEventEnvelope>,
+}
+
+impl MainEventStore {
+    pub(crate) async fn open(db: Db) -> Result<Self> {
+        let event_seq = recover_next_seq(&db).await?;
+        let (event_tx, _) = broadcast::channel(DEFAULT_MAIN_EVENT_TAIL_LIMIT.max(16));
+        Ok(Self {
+            inner: Arc::new(MainEventStoreInner {
+                db,
+                event_seq: AtomicU32::new(event_seq),
+                append_lock: Mutex::new(()),
+                event_tx,
+            }),
+        })
+    }
+
+    pub async fn append(&self, body: MainEventBody) -> Result<MainEventEnvelope> {
+        let _guard = self.inner.append_lock.lock().await;
+        let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);
+        let now = Utc::now();
+        let event = MainEvent {
+            id: Uuid::now_v7().to_string(),
+            ts: now,
+            body,
+        };
+        self.inner
+            .db
+            .put(
+                keys::main_event_key(seq, now.timestamp_millis()),
+                serde_json::to_vec(&event)?,
+            )
+            .await?;
+        let envelope = MainEventEnvelope { seq, event };
+        let _ = self.inner.event_tx.send(envelope.clone());
+        Ok(envelope)
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<MainEventEnvelope> {
+        self.inner.event_tx.subscribe()
+    }
+
+    pub async fn list_from_with_limit(
+        &self,
+        start_seq: u32,
+        limit: usize,
+    ) -> Result<Vec<MainEventEnvelope>> {
+        let mut events = list_events_from(&self.inner.db, start_seq).await?;
+        events.truncate(limit.saturating_add(1));
+        Ok(events)
+    }
+}
+
+async fn recover_next_seq<R>(db: &R) -> Result<u32>
+where
+    R: DbRead + Sync,
+{
+    let mut iter = db.scan_prefix(keys::main_events_prefix()).await?;
+    let mut max_seq = 0;
+    while let Some(entry) = iter.next().await? {
+        let key = key_to_string(&entry.key)?;
+        if let Some(seq) = keys::parse_main_event_seq(&key) {
+            max_seq = max_seq.max(seq);
+        }
+    }
+    Ok(max_seq.saturating_add(1).max(1))
+}
+
+async fn list_events_from<R>(db: &R, start_seq: u32) -> Result<Vec<MainEventEnvelope>>
+where
+    R: DbRead + Sync,
+{
+    let mut iter = db.scan_prefix(keys::main_events_prefix()).await?;
+    let mut events = Vec::new();
+    while let Some(entry) = iter.next().await? {
+        let key = key_to_string(&entry.key)?;
+        let Some(seq) = keys::parse_main_event_seq(&key) else {
+            continue;
+        };
+        if seq < start_seq {
+            continue;
+        }
+        events.push(MainEventEnvelope {
+            seq,
+            event: serde_json::from_slice(&entry.value)?,
+        });
+    }
+    events.sort_by_key(|event| event.seq);
+    Ok(events)
+}
+
+fn key_to_string(key: &Bytes) -> Result<String> {
+    String::from_utf8(key.to_vec())
+        .map_err(|err| Error::Other(format!("stored key is not valid UTF-8: {err}")))
+}

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

@@ -1,6 +1,7 @@
 mod auth_codes;
 mod auth_tokens;
 mod blob_store;
+mod main_event_store;
 mod projection_cache;
 mod run_catalog_index;
 mod run_store;
@@ -15,6 +16,7 @@ pub use auth_tokens::{ConsumeOutcome, RefreshToken, RefreshTokenStore};
 pub use blob_store::{Blob, BlobStore};
 use chrono::{DateTime, Utc};
 use fabro_types::{Run, RunId, SessionId};
+pub use main_event_store::MainEventStore;
 use object_store::ObjectStore;
 pub use projection_cache::CachedRunProjection;
 use projection_cache::RunProjectionCache;
@@ -49,6 +51,7 @@ pub struct Database {
     active_runs: Arc<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
     blobs: Arc<OnceCell<Arc<BlobStore>>>,
     catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
+    main_events: Arc<OnceCell<Arc<MainEventStore>>>,
     auth_codes: Arc<OnceCell<Arc<AuthCodeStore>>>,
     refresh_tokens: Arc<OnceCell<Arc<RefreshTokenStore>>>,
     projection_cache: Arc<RunProjectionCache>,
@@ -81,6 +84,7 @@ impl Database {
             active_runs: Arc::new(Mutex::new(HashMap::new())),
             blobs: Arc::new(OnceCell::new()),
             catalog_index: Arc::new(OnceCell::new()),
+            main_events: Arc::new(OnceCell::new()),
             auth_codes: Arc::new(OnceCell::new()),
             refresh_tokens: Arc::new(OnceCell::new()),
             projection_cache: Arc::new(RunProjectionCache::default()),
@@ -410,6 +414,15 @@ impl Database {
         Ok(Arc::clone(store))
     }
 
+    pub async fn main_events(&self) -> Result<Arc<MainEventStore>> {
+        let db = self.open_db().await?;
+        let store = self
+            .main_events
+            .get_or_try_init(|| async { Ok::<_, Error>(Arc::new(MainEventStore::open(db).await?)) })
+            .await?;
+        Ok(Arc::clone(store))
+    }
+
     pub async fn refresh_tokens(&self) -> Result<Arc<RefreshTokenStore>> {
         let store = self
             .refresh_tokens
@@ -471,8 +484,9 @@ fn active_run_from(
 mod tests {
     use chrono::{DateTime, Utc};
     use fabro_types::{
-        AttrValue, FailureReason, Graph, RunControlAction, RunSpec, RunStatus, StageId,
-        SuccessReason, WorkflowSettings, test_support,
+        AttrValue, FailureReason, Graph, MainEventBody, MainEventLifecycleProps, MainEventSource,
+        MainEventStartedProps, RunControlAction, RunSpec, RunStatus, StageId, SuccessReason,
+        WorkflowSettings, test_support,
     };
     use futures::TryStreamExt;
     use object_store::memory::InMemory;
@@ -608,6 +622,30 @@ mod tests {
         .unwrap();
     }
 
+    fn main_event_body(run_id: RunId, source_event_name: &str) -> MainEventBody {
+        let source = MainEventSource {
+            run_id,
+            source_event_id: format!("evt-{source_event_name}"),
+            source_event_ts: dt("2026-06-21T12:00:00Z"),
+            source_event_name: source_event_name.to_string(),
+            actor: None,
+        };
+        match source_event_name {
+            "run.started" => MainEventBody::RunStarted(MainEventStartedProps {
+                source,
+                name: "test".to_string(),
+                base_branch: None,
+                base_sha: None,
+                run_branch: None,
+                worktree_dir: None,
+                goal: None,
+            }),
+            "run.running" => MainEventBody::RunRunning(MainEventLifecycleProps { source }),
+            "run.paused" => MainEventBody::RunPaused(MainEventLifecycleProps { source }),
+            other => panic!("unsupported test main event source {other}"),
+        }
+    }
+
     async fn append_created_with_parent(
         run: &RunDatabase,
         label: &str,
@@ -729,6 +767,95 @@ mod tests {
         assert!(!list_paths(object_store, "runs/").await.is_empty());
     }
 
+    #[tokio::test]
+    async fn main_events_append_list_and_subscribe_in_order() {
+        let (_object_store, store) = make_store();
+        let main_events = store.main_events().await.unwrap();
+        let run_id = test_run_id("run-1");
+        let mut rx = main_events.subscribe();
+
+        let first = main_events
+            .append(main_event_body(run_id, "run.started"))
+            .await
+            .unwrap();
+        let second = main_events
+            .append(main_event_body(run_id, "run.running"))
+            .await
+            .unwrap();
+
+        assert_eq!(first.seq, 1);
+        assert_eq!(second.seq, 2);
+        assert_eq!(first.event.event_name(), "fabro.run.started");
+
+        assert_eq!(rx.recv().await.unwrap().seq, 1);
+        assert_eq!(rx.recv().await.unwrap().seq, 2);
+
+        let listed = main_events.list_from_with_limit(1, 10).await.unwrap();
+        assert_eq!(
+            listed.iter().map(|event| event.seq).collect::<Vec<_>>(),
+            vec![1, 2]
+        );
+    }
+
+    #[tokio::test]
+    async fn main_events_list_returns_limit_plus_one_from_start_seq() {
+        let (_object_store, store) = make_store();
+        let main_events = store.main_events().await.unwrap();
+        let run_id = test_run_id("run-1");
+
+        for source in ["run.started", "run.running", "run.paused"] {
+            main_events
+                .append(main_event_body(run_id, source))
+                .await
+                .unwrap();
+        }
+
+        let listed = main_events.list_from_with_limit(2, 1).await.unwrap();
+        assert_eq!(
+            listed.iter().map(|event| event.seq).collect::<Vec<_>>(),
+            vec![2, 3]
+        );
+    }
+
+    #[tokio::test]
+    async fn main_events_recover_next_sequence_after_reopen() {
+        let (object_store, store) = make_store();
+        let run_id = test_run_id("run-1");
+        let main_events = store.main_events().await.unwrap();
+        main_events
+            .append(main_event_body(run_id, "run.started"))
+            .await
+            .unwrap();
+
+        let reopened = Database::new(object_store, "runs", Duration::from_millis(1), None);
+        let reopened_main_events = reopened.main_events().await.unwrap();
+        let appended = reopened_main_events
+            .append(main_event_body(run_id, "run.running"))
+            .await
+            .unwrap();
+
+        assert_eq!(appended.seq, 2);
+    }
+
+    #[tokio::test]
+    async fn delete_run_keeps_main_events() {
+        let (_object_store, store) = make_store();
+        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
+        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
+
+        let main_events = store.main_events().await.unwrap();
+        main_events
+            .append(main_event_body(test_run_id("run-1"), "run.started"))
+            .await
+            .unwrap();
+
+        store.delete_run(&test_run_id("run-1")).await.unwrap();
+
+        let listed = main_events.list_from_with_limit(1, 10).await.unwrap();
+        assert_eq!(listed.len(), 1);
+        assert_eq!(listed[0].event.event_name(), "fabro.run.started");
+    }
+
     #[tokio::test]
     async fn delete_run_keeps_global_cas_blobs() {
         let (_object_store, store) = make_store();