diff --git a/docs/superpowers/plans/2026-06-21-main-event-stream.md b/docs/superpowers/plans/2026-06-21-main-event-stream.md new file mode 100644 index 0000000000..cabb8d5165 --- /dev/null +++ b/docs/superpowers/plans/2026-06-21-main-event-stream.md @@ -0,0 +1,1663 @@ +# Main Event Stream Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build the first internal chunk of Fabro's durable main event stream by storing `MainEvent`s in SlateDB and mirroring selected run lifecycle events into it. + +**Architecture:** Add shared `MainEvent` types in `fabro-types`, a focused `MainEventStore` in `fabro-store`, and a server-owned mapper in `fabro-server`. The store owns append/list/subscribe mechanics under the global `events/main/` logical prefix, while `fabro-server` owns the product policy for which `RunEvent`s become main events. Mirror `run.created` at the server creation append site so it is captured before live execution subscriptions exist; mirror later lifecycle events from the active-run forwarding path. + +**Tech Stack:** Rust, serde, chrono, uuid v7, tokio broadcast, SlateDB, object_store in-memory tests, cargo nextest. + +--- + +## File Structure + +- Create `lib/crates/fabro-types/src/main_event.rs` + - Defines `MainEventEnvelope`, `MainEvent`, `MainEventBody`, and compact lifecycle property structs. + - Implements the same explicit `event` plus `properties` serde shape and unknown-event fallback used by `RunEvent`. +- Modify `lib/crates/fabro-types/src/lib.rs` + - Exports the new shared main event vocabulary. +- Create `lib/crates/fabro-types/tests/main_event_serde.rs` + - Covers v1 event names, round trips, and unknown event preservation. +- Modify `lib/crates/fabro-store/src/keys.rs` + - Adds `events\0main\0` key helpers and parser for the global main event prefix. +- Create `lib/crates/fabro-store/src/slate/main_event_store.rs` + - Implements append, list-from-with-limit, sequence recovery, and live subscribe. +- Modify `lib/crates/fabro-store/src/slate/mod.rs` + - Adds a `main_events` `OnceCell` on `Database` and exposes `Database::main_events()`. +- Modify `lib/crates/fabro-store/src/lib.rs` + - Re-exports `MainEventStore`. +- Modify `lib/crates/fabro-store/src/slate/mod.rs` tests + - Adds storage tests beside existing SlateDB integration-style unit tests. +- Create `lib/crates/fabro-server/src/server/main_event_mirror.rs` + - Converts selected `RunEvent`s into `MainEventBody` values. +- Modify `lib/crates/fabro-server/src/server.rs` + - Registers the new submodule, adds shared mirroring helpers, and mirrors later lifecycle events from `forward_run_events_to_global`. +- Modify `lib/crates/fabro-server/src/server/handler/runs.rs` + - Mirrors the persisted `run.created` event immediately after successful server run creation. +- Modify `lib/crates/fabro-server/src/server/tests.rs` + - Adds private server tests for active-run mirroring and ignored non-v1 events. + +## Task 1: Shared Main Event Types + +**Files:** +- Create: `lib/crates/fabro-types/src/main_event.rs` +- Modify: `lib/crates/fabro-types/src/lib.rs` +- Test: `lib/crates/fabro-types/tests/main_event_serde.rs` + +- [ ] **Step 1: Write failing serde tests** + +Create `lib/crates/fabro-types/tests/main_event_serde.rs`: + +```rust +use chrono::{TimeZone, Utc}; +use fabro_types::{ + AuthMethod, FailureCategory, FailureReason, IdpIdentity, MainEvent, MainEventBody, + MainEventCancelledProps, MainEventCompletedProps, MainEventCreatedProps, + MainEventFailedProps, MainEventLifecycleProps, MainEventSource, MainEventStartedProps, + Principal, RunId, SuccessReason, +}; +use serde_json::json; + +fn run_id() -> RunId { + "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap() +} + +fn actor() -> Principal { + Principal::user( + IdpIdentity { + provider: "github".to_string(), + subject: "123".to_string(), + email: Some("fabian@example.com".to_string()), + name: Some("Fabian".to_string()), + avatar_url: None, + }, + Some("fabian".to_string()), + AuthMethod::Github, + ) +} + +fn source(source_event_name: &str) -> MainEventSource { + MainEventSource { + run_id: run_id(), + source_event_id: format!("evt-{source_event_name}"), + source_event_ts: Utc.with_ymd_and_hms(2026, 6, 21, 10, 0, 0).unwrap(), + source_event_name: source_event_name.to_string(), + actor: Some(actor()), + } +} + +fn main_event(body: MainEventBody) -> MainEvent { + MainEvent { + id: "019796d3-d2d0-7c81-a65a-6f3ddc67d4a4".to_string(), + ts: Utc.with_ymd_and_hms(2026, 6, 21, 10, 1, 0).unwrap(), + body, + } +} + +#[test] +fn v1_event_names_are_stable() { + let cases = [ + ( + MainEventBody::RunCreated(MainEventCreatedProps { + source: source("run.created"), + title: Some("Triage bug".to_string()), + workflow_slug: Some("triage".to_string()), + automation: None, + git: None, + parent_id: None, + provenance_subject: Some(actor()), + }), + "fabro.run.created", + ), + ( + MainEventBody::RunStarted(MainEventStartedProps { + source: source("run.started"), + name: "triage".to_string(), + base_branch: Some("main".to_string()), + base_sha: Some("abc123".to_string()), + run_branch: Some("fabro/run-1".to_string()), + worktree_dir: None, + goal: Some("Triage bug".to_string()), + }), + "fabro.run.started", + ), + ( + MainEventBody::RunRunning(MainEventLifecycleProps { + source: source("run.running"), + }), + "fabro.run.running", + ), + ( + MainEventBody::RunCompleted(MainEventCompletedProps { + source: source("run.completed"), + status: "succeeded".to_string(), + reason: SuccessReason::Completed, + total_usd_micros: Some(12), + final_git_commit_sha: Some("abc123".to_string()), + automation: None, + billing: None, + }), + "fabro.run.completed", + ), + ( + MainEventBody::RunFailed(MainEventFailedProps { + source: source("run.failed"), + reason: FailureReason::WorkflowError, + category: FailureCategory::Deterministic, + message: "boom".to_string(), + final_git_commit_sha: None, + automation: None, + billing: None, + }), + "fabro.run.failed", + ), + ( + MainEventBody::RunCancelled(MainEventCancelledProps { + source: source("run.failed"), + reason: FailureReason::Cancelled, + category: FailureCategory::Canceled, + message: "cancelled".to_string(), + final_git_commit_sha: None, + automation: None, + billing: None, + }), + "fabro.run.cancelled", + ), + ( + MainEventBody::RunPaused(MainEventLifecycleProps { + source: source("run.paused"), + }), + "fabro.run.paused", + ), + ( + MainEventBody::RunUnpaused(MainEventLifecycleProps { + source: source("run.unpaused"), + }), + "fabro.run.unpaused", + ), + ]; + + for (body, expected) in cases { + assert_eq!(body.event_name(), expected); + assert_eq!(main_event(body).event_name(), expected); + } +} + +#[test] +fn main_event_round_trips_flat_event_properties() { + let event = main_event(MainEventBody::RunCompleted(MainEventCompletedProps { + source: source("run.completed"), + status: "succeeded".to_string(), + reason: SuccessReason::Completed, + total_usd_micros: Some(42), + final_git_commit_sha: Some("deadbeef".to_string()), + automation: None, + billing: None, + })); + + let value = serde_json::to_value(&event).unwrap(); + assert_eq!(value["event"], "fabro.run.completed"); + assert_eq!(value["properties"]["run_id"], run_id().to_string()); + assert_eq!(value["properties"]["source_event_name"], "run.completed"); + assert_eq!(value["properties"]["status"], "succeeded"); + + let parsed: MainEvent = serde_json::from_value(value.clone()).unwrap(); + assert_eq!(serde_json::to_value(parsed).unwrap(), value); +} + +#[test] +fn unknown_main_event_preserves_name_and_properties() { + let value = json!({ + "id": "019796d3-d2d0-7c81-a65a-6f3ddc67d4a4", + "ts": "2026-06-21T10:01:00Z", + "event": "github.issue_comment.created", + "properties": { + "delivery_id": "delivery-1", + "action": "created" + } + }); + + let parsed: MainEvent = serde_json::from_value(value.clone()).unwrap(); + match parsed.body { + MainEventBody::Unknown { name, properties } => { + assert_eq!(name, "github.issue_comment.created"); + assert_eq!(properties["delivery_id"], "delivery-1"); + } + other => panic!("expected unknown event, got {other:?}"), + } + assert_eq!(serde_json::to_value(parsed).unwrap(), value); +} +``` + +- [ ] **Step 2: Run tests and verify the expected failure** + +Run: + +```bash +cargo nextest run -p fabro-types main_event_serde +``` + +Expected: compile failure with unresolved imports such as `fabro_types::MainEvent`. + +- [ ] **Step 3: Add the main event type module** + +Create `lib/crates/fabro-types/src/main_event.rs`: + +```rust +use chrono::{DateTime, Utc}; +use serde::de::Error as DeError; +use serde::ser::Error as SerError; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::{Map, Value, json}; + +use crate::{ + AutomationRef, BilledTokenCounts, FailureCategory, FailureReason, GitContext, Principal, RunId, + SuccessReason, +}; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventEnvelope { + pub seq: u32, + pub event: MainEvent, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct MainEvent { + pub id: String, + pub ts: DateTime, + pub body: MainEventBody, +} + +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +#[serde(tag = "event", content = "properties")] +pub enum MainEventBody { + #[serde(rename = "fabro.run.created")] + RunCreated(MainEventCreatedProps), + #[serde(rename = "fabro.run.started")] + RunStarted(MainEventStartedProps), + #[serde(rename = "fabro.run.running")] + RunRunning(MainEventLifecycleProps), + #[serde(rename = "fabro.run.completed")] + RunCompleted(MainEventCompletedProps), + #[serde(rename = "fabro.run.failed")] + RunFailed(MainEventFailedProps), + #[serde(rename = "fabro.run.cancelled")] + RunCancelled(MainEventCancelledProps), + #[serde(rename = "fabro.run.paused")] + RunPaused(MainEventLifecycleProps), + #[serde(rename = "fabro.run.unpaused")] + RunUnpaused(MainEventLifecycleProps), + Unknown { + name: String, + properties: Value, + }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventSource { + pub run_id: RunId, + pub source_event_id: String, + pub source_event_ts: DateTime, + pub source_event_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub actor: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventCreatedProps { + #[serde(flatten)] + pub source: MainEventSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workflow_slug: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub git: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provenance_subject: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventLifecycleProps { + #[serde(flatten)] + pub source: MainEventSource, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventStartedProps { + #[serde(flatten)] + pub source: MainEventSource, + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_branch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_branch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worktree_dir: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub goal: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventCompletedProps { + #[serde(flatten)] + pub source: MainEventSource, + pub status: String, + pub reason: SuccessReason, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_usd_micros: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub billing: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventFailedProps { + #[serde(flatten)] + pub source: MainEventSource, + pub reason: FailureReason, + pub category: FailureCategory, + pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub billing: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventCancelledProps { + #[serde(flatten)] + pub source: MainEventSource, + pub reason: FailureReason, + pub category: FailureCategory, + pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub billing: Option, +} + +#[derive(Debug, Clone, Deserialize)] +struct MainEventRaw { + id: String, + ts: DateTime, + event: String, + #[serde(default = "default_properties")] + properties: Value, +} + +struct MainEventParts<'a> { + id: String, + ts: DateTime, + event: &'a str, + properties: &'a Value, +} + +impl MainEventBody { + pub fn event_name(&self) -> &str { + match self { + Self::RunCreated(_) => "fabro.run.created", + Self::RunStarted(_) => "fabro.run.started", + Self::RunRunning(_) => "fabro.run.running", + Self::RunCompleted(_) => "fabro.run.completed", + Self::RunFailed(_) => "fabro.run.failed", + Self::RunCancelled(_) => "fabro.run.cancelled", + Self::RunPaused(_) => "fabro.run.paused", + Self::RunUnpaused(_) => "fabro.run.unpaused", + Self::Unknown { name, .. } => name.as_str(), + } + } + + fn properties_value(&self) -> serde_json::Result { + if let Self::Unknown { properties, .. } = self { + return Ok(properties.clone()); + } + + match serde_json::to_value(self)? { + Value::Object(mut map) => { + Ok(map.remove("properties").unwrap_or_else(default_properties)) + } + _ => Ok(default_properties()), + } + } +} + +fn is_known_main_event_name(event: &str) -> bool { + matches!( + event, + "fabro.run.created" + | "fabro.run.started" + | "fabro.run.running" + | "fabro.run.completed" + | "fabro.run.failed" + | "fabro.run.cancelled" + | "fabro.run.paused" + | "fabro.run.unpaused" + ) +} + +impl MainEvent { + pub fn from_value(value: Value) -> serde_json::Result { + let raw: MainEventRaw = serde_json::from_value(value)?; + Self::from_parts(MainEventParts { + id: raw.id, + ts: raw.ts, + event: &raw.event, + properties: &raw.properties, + }) + } + + pub fn from_ref(value: &Value) -> serde_json::Result { + let obj = value.as_object().ok_or_else(|| { + ::custom("main event must be a JSON object") + })?; + let id = obj.get("id").and_then(Value::as_str).ok_or_else(|| { + ::custom("missing or non-string field: id") + })?; + let ts = obj + .get("ts") + .ok_or_else(|| ::custom("missing field: ts")) + .and_then(DateTime::::deserialize)?; + let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| { + ::custom("missing or non-string field: event") + })?; + let properties = obj + .get("properties") + .cloned() + .unwrap_or_else(default_properties); + Self::from_parts(MainEventParts { + id: id.to_string(), + ts, + event, + properties: &properties, + }) + } + + fn from_parts(parts: MainEventParts<'_>) -> serde_json::Result { + let body_payload = json!({ + "event": parts.event, + "properties": parts.properties, + }); + let body: MainEventBody = match serde_json::from_value(body_payload) { + Ok(body) => body, + Err(err) if is_known_main_event_name(parts.event) => return Err(err), + Err(_) => MainEventBody::Unknown { + name: parts.event.to_string(), + properties: parts.properties.clone(), + }, + }; + Ok(Self { + id: parts.id, + ts: parts.ts, + body, + }) + } + + pub fn to_value(&self) -> serde_json::Result { + let mut map = Map::new(); + map.insert("id".to_string(), Value::String(self.id.clone())); + map.insert("ts".to_string(), serde_json::to_value(self.ts)?); + map.insert( + "event".to_string(), + Value::String(self.body.event_name().to_string()), + ); + map.insert("properties".to_string(), self.body.properties_value()?); + Ok(Value::Object(map)) + } + + pub fn event_name(&self) -> &str { + self.body.event_name() + } + + pub fn properties(&self) -> serde_json::Result { + self.body.properties_value() + } +} + +impl Serialize for MainEvent { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.to_value() + .map_err(S::Error::custom)? + .serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for MainEvent { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + Self::from_value(value).map_err(D::Error::custom) + } +} + +fn default_properties() -> Value { + Value::Object(Map::new()) +} +``` + +- [ ] **Step 4: Export the types** + +Modify `lib/crates/fabro-types/src/lib.rs`. + +Add the module near the other shared modules: + +```rust +pub mod main_event; +``` + +Add exports near the other `pub use` blocks: + +```rust +pub use main_event::{ + MainEvent, MainEventBody, MainEventCancelledProps, MainEventCompletedProps, + MainEventCreatedProps, MainEventEnvelope, MainEventFailedProps, MainEventLifecycleProps, + MainEventSource, MainEventStartedProps, +}; +``` + +- [ ] **Step 5: Run the focused tests** + +Run: + +```bash +cargo nextest run -p fabro-types main_event_serde +``` + +Expected: all `main_event_serde` tests pass. + +- [ ] **Step 6: Commit** + +```bash +git add lib/crates/fabro-types/src/main_event.rs lib/crates/fabro-types/src/lib.rs lib/crates/fabro-types/tests/main_event_serde.rs +git commit -m "feat: add main event types" +``` + +## Task 2: Durable Main Event Store + +**Files:** +- Modify: `lib/crates/fabro-store/src/keys.rs` +- Create: `lib/crates/fabro-store/src/slate/main_event_store.rs` +- Modify: `lib/crates/fabro-store/src/slate/mod.rs` +- Modify: `lib/crates/fabro-store/src/lib.rs` + +- [ ] **Step 1: Add failing key tests** + +Modify `lib/crates/fabro-store/src/keys.rs` in the existing `#[cfg(test)] mod tests`: + +```rust + #[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 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 + ); + } +``` + +- [ ] **Step 2: Run key tests and verify failure** + +Run: + +```bash +cargo nextest run -p fabro-store keys +``` + +Expected: compile failure because `main_event_key` and `parse_main_event_seq` are missing. + +- [ ] **Step 3: Add main event key helpers** + +Modify `lib/crates/fabro-store/src/keys.rs` after the run event helpers: + +```rust +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}")) +} +``` + +Modify `lib/crates/fabro-store/src/keys.rs` after `parse_event_seq`: + +```rust +pub(crate) fn parse_main_event_seq(key: &str) -> Option { + 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() +} +``` + +- [ ] **Step 4: Run key tests** + +Run: + +```bash +cargo nextest run -p fabro-store keys +``` + +Expected: key tests pass. + +- [ ] **Step 5: Add failing store tests** + +Modify `lib/crates/fabro-store/src/slate/mod.rs` in its test module. Add imports to the existing test imports: + +```rust +use fabro_types::{MainEventBody, MainEventLifecycleProps, MainEventSource, MainEventStartedProps}; +``` + +Add helper functions inside the test module: + +```rust +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}"), + } +} +``` + +Add tests in the same module: + +```rust + #[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![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![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"); + } + +``` + +- [ ] **Step 6: Run store tests and verify failure** + +Run: + +```bash +cargo nextest run -p fabro-store main_events +``` + +Expected: compile failure because `Database::main_events` and `MainEventStore` do not exist. + +- [ ] **Step 7: Implement `MainEventStore`** + +Create `lib/crates/fabro-store/src/slate/main_event_store.rs`: + +```rust +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, +} + +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, +} + +impl MainEventStore { + pub(crate) async fn open(db: Db) -> Result { + 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 { + 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 { + self.inner.event_tx.subscribe() + } + + pub async fn list_from_with_limit( + &self, + start_seq: u32, + limit: usize, + ) -> Result> { + 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(db: &R) -> Result +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(db: &R, start_seq: u32) -> Result> +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::from_utf8(key.to_vec()) + .map_err(|err| Error::Other(format!("stored key is not valid UTF-8: {err}"))) +} +``` + +- [ ] **Step 8: Wire `MainEventStore` into `Database`** + +Modify `lib/crates/fabro-store/src/slate/mod.rs`. + +Add the module: + +```rust +mod main_event_store; +``` + +Add the export: + +```rust +pub use main_event_store::MainEventStore; +``` + +Add a field to `Database`: + +```rust + main_events: Arc>>, +``` + +Initialize it in `Database::new`: + +```rust + main_events: Arc::new(OnceCell::new()), +``` + +Add this method in `impl Database` near `blob_store` and `catalog_index` accessors: + +```rust + pub async fn main_events(&self) -> Result> { + 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)) + } +``` + +Modify `lib/crates/fabro-store/src/lib.rs` in the SlateDB re-export list: + +```rust + RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun, + MainEventStore, +``` + +- [ ] **Step 9: Run focused store tests** + +Run: + +```bash +cargo nextest run -p fabro-store main_events +``` + +Expected: all `main_events` tests pass. + +- [ ] **Step 10: Run broader store tests** + +Run: + +```bash +cargo nextest run -p fabro-store +``` + +Expected: all `fabro-store` tests pass. + +- [ ] **Step 11: Commit** + +```bash +git add lib/crates/fabro-store/src/keys.rs lib/crates/fabro-store/src/slate/main_event_store.rs lib/crates/fabro-store/src/slate/mod.rs lib/crates/fabro-store/src/lib.rs +git commit -m "feat: store durable main events" +``` + +## Task 3: Server Run Lifecycle Mirroring + +**Files:** +- Create: `lib/crates/fabro-server/src/server/main_event_mirror.rs` +- Modify: `lib/crates/fabro-server/src/server.rs` +- Modify: `lib/crates/fabro-server/src/server/handler/runs.rs` +- Modify: `lib/crates/fabro-server/src/server/tests.rs` + +- [ ] **Step 1: Add failing mapper unit tests** + +Create `lib/crates/fabro-server/src/server/main_event_mirror.rs` with tests first: + +```rust +use fabro_types::{EventBody, MainEventBody, RunEvent}; + +pub(crate) fn main_event_body_from_run_event(_event: &RunEvent) -> Option { + None +} + +#[cfg(test)] +mod tests { + use chrono::{TimeZone, Utc}; + use fabro_types::{ + EventBody, FailureCategory, FailureDetail, FailureReason, Graph, MainEventBody, RunEvent, + RunFailedProps, RunFailure, RunTiming, WorkflowSettings, fixtures, test_support, + }; + use serde_json::json; + + use super::main_event_body_from_run_event; + + fn run_event(event_name: &str, properties: serde_json::Value) -> RunEvent { + RunEvent::from_value(json!({ + "id": format!("evt-{event_name}"), + "ts": "2026-06-21T12:00:00Z", + "run_id": fixtures::RUN_1, + "event": event_name, + "properties": properties + })) + .unwrap() + } + + #[test] + fn maps_created_with_compact_identity_fields() { + let event = run_event( + "run.created", + json!({ + "title": "Triage bug", + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "run_dir": "/tmp/test", + "workflow_slug": "triage", + "provenance": test_support::test_run_provenance(), + "parent_id": fixtures::RUN_2, + }), + ); + + let mapped = main_event_body_from_run_event(&event).unwrap(); + match mapped { + MainEventBody::RunCreated(props) => { + assert_eq!(props.source.run_id, fixtures::RUN_1); + assert_eq!(props.source.source_event_name, "run.created"); + assert_eq!(props.title.as_deref(), Some("Triage bug")); + assert_eq!(props.workflow_slug.as_deref(), Some("triage")); + assert_eq!(props.parent_id, Some(fixtures::RUN_2)); + assert!(props.provenance_subject.is_some()); + } + other => panic!("expected created, got {other:?}"), + } + } + + #[test] + fn maps_failed_and_cancelled_separately() { + let failed = RunEvent { + id: "evt-failed".to_string(), + ts: Utc.with_ymd_and_hms(2026, 6, 21, 12, 0, 0).unwrap(), + run_id: fixtures::RUN_1, + node_id: None, + node_label: None, + stage_id: None, + parallel_group_id: None, + parallel_branch_id: None, + session_id: None, + parent_session_id: None, + tool_call_id: None, + actor: None, + body: EventBody::RunFailed(RunFailedProps { + failure: RunFailure { + reason: FailureReason::WorkflowError, + detail: FailureDetail { + message: "boom".to_string(), + category: FailureCategory::Deterministic, + causes: Vec::new(), + signature: None, + exec_output_tail: None, + }, + }, + timing: RunTiming::default(), + final_git_commit_sha: Some("abc123".to_string()), + final_patch: None, + diff_summary: None, + billing: None, + }), + }; + assert!(matches!( + main_event_body_from_run_event(&failed), + Some(MainEventBody::RunFailed(_)) + )); + + let mut cancelled = failed.clone(); + cancelled.id = "evt-cancelled".to_string(); + cancelled.body = EventBody::RunFailed(RunFailedProps { + failure: RunFailure { + reason: FailureReason::Cancelled, + detail: FailureDetail { + message: "cancelled".to_string(), + category: FailureCategory::Canceled, + causes: Vec::new(), + signature: None, + exec_output_tail: None, + }, + }, + timing: RunTiming::default(), + final_git_commit_sha: None, + final_patch: None, + diff_summary: None, + billing: None, + }); + assert!(matches!( + main_event_body_from_run_event(&cancelled), + Some(MainEventBody::RunCancelled(_)) + )); + } + + #[test] + fn ignores_non_v1_events_and_cancel_requested() { + let stage = run_event( + "stage.started", + json!({ + "index": 0, + "handler_type": "command", + "attempt": 1, + "max_attempts": 1 + }), + ); + assert!(main_event_body_from_run_event(&stage).is_none()); + + let cancel_requested = run_event("run.cancel.requested", json!({ "action": "cancel" })); + assert!(main_event_body_from_run_event(&cancel_requested).is_none()); + } +} +``` + +Modify `lib/crates/fabro-server/src/server.rs` near the other server submodules: + +```rust +mod main_event_mirror; +``` + +- [ ] **Step 2: Run mapper tests and verify failure** + +Run: + +```bash +cargo nextest run -p fabro-server main_event_mirror +``` + +Expected: tests compile and fail because the mapper returns `None`. + +- [ ] **Step 3: Implement the mapper** + +Replace the placeholder contents in `lib/crates/fabro-server/src/server/main_event_mirror.rs` above the tests with: + +```rust +use fabro_types::{ + EventBody, FailureReason, MainEventBody, MainEventCancelledProps, MainEventCompletedProps, + MainEventCreatedProps, MainEventFailedProps, MainEventLifecycleProps, MainEventSource, + MainEventStartedProps, RunEvent, +}; + +pub(crate) fn main_event_body_from_run_event(event: &RunEvent) -> Option { + let source = source_from_run_event(event); + match &event.body { + EventBody::RunCreated(props) => Some(MainEventBody::RunCreated(MainEventCreatedProps { + source, + title: props.title.clone(), + workflow_slug: props.workflow_slug.clone(), + automation: props.automation.clone(), + git: props.git.clone(), + parent_id: props.parent_id, + provenance_subject: Some(props.provenance.subject.clone()), + })), + EventBody::RunStarted(props) => Some(MainEventBody::RunStarted(MainEventStartedProps { + source, + name: props.name.clone(), + base_branch: props.base_branch.clone(), + base_sha: props.base_sha.clone(), + run_branch: props.run_branch.clone(), + worktree_dir: props.worktree_dir.clone(), + goal: props.goal.clone(), + })), + EventBody::RunRunning(_) => Some(MainEventBody::RunRunning(MainEventLifecycleProps { + source, + })), + EventBody::RunCompleted(props) => { + Some(MainEventBody::RunCompleted(MainEventCompletedProps { + source, + status: props.status.clone(), + reason: props.reason, + total_usd_micros: props.total_usd_micros, + final_git_commit_sha: props.final_git_commit_sha.clone(), + automation: None, + billing: props.billing.clone(), + })) + } + EventBody::RunFailed(props) if props.failure.reason == FailureReason::Cancelled => { + Some(MainEventBody::RunCancelled(MainEventCancelledProps { + source, + reason: props.failure.reason, + category: props.failure.detail.category, + message: props.failure.detail.message.clone(), + final_git_commit_sha: props.final_git_commit_sha.clone(), + automation: None, + billing: props.billing.clone(), + })) + } + EventBody::RunFailed(props) => Some(MainEventBody::RunFailed(MainEventFailedProps { + source, + reason: props.failure.reason, + category: props.failure.detail.category, + message: props.failure.detail.message.clone(), + final_git_commit_sha: props.final_git_commit_sha.clone(), + automation: None, + billing: props.billing.clone(), + })), + EventBody::RunPaused(_) => Some(MainEventBody::RunPaused(MainEventLifecycleProps { + source, + })), + EventBody::RunUnpaused(_) => Some(MainEventBody::RunUnpaused(MainEventLifecycleProps { + source, + })), + _ => None, + } +} + +fn source_from_run_event(event: &RunEvent) -> MainEventSource { + MainEventSource { + run_id: event.run_id, + source_event_id: event.id.clone(), + source_event_ts: event.ts, + source_event_name: event.event_name().to_string(), + actor: event.actor.clone(), + } +} +``` + +- [ ] **Step 4: Run mapper tests** + +Run: + +```bash +cargo nextest run -p fabro-server main_event_mirror +``` + +Expected: mapper tests pass. + +- [ ] **Step 5: Add failing forwarding tests** + +Modify `lib/crates/fabro-server/src/server/tests.rs`. + +Add `MainEventBody` to the existing `use fabro_types::{}` list: + +```rust + MainEventBody, +``` + +Add helper functions near the existing test helpers: + +```rust +async fn wait_for_main_events( + main_events: Arc, + expected_len: usize, +) -> Vec { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + let listed = main_events.list_from_with_limit(1, 100).await.unwrap(); + if listed.len() == expected_len { + return listed; + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for {expected_len} main events, saw {}", + listed.len() + ); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } +} + +fn run_event_value(run_id: RunId, event_name: &str, properties: serde_json::Value) -> serde_json::Value { + json!({ + "id": format!("evt-{event_name}"), + "ts": "2026-06-21T12:00:00Z", + "run_id": run_id, + "event": event_name, + "properties": properties + }) +} + +async fn append_run_event( + run_store: &fabro_store::RunDatabase, + run_id: RunId, + event_name: &str, + properties: serde_json::Value, +) { + let payload = EventPayload::new(run_event_value(run_id, event_name, properties), &run_id) + .unwrap(); + run_store.append_event(&payload).await.unwrap(); +} +``` + +Add tests near other lifecycle tests: + +```rust +#[tokio::test] +async fn persisted_run_created_mirrors_to_main_stream_without_forwarder() { + let state = test_app_state(); + let run_id = fixtures::RUN_1; + let run_store = state.store.create_run(&run_id).await.unwrap(); + + append_run_event( + &run_store, + run_id, + "run.created", + json!({ + "title": "Triage bug", + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "run_dir": "/tmp/test", + "workflow_slug": "triage", + "provenance": test_support::test_run_provenance() + }), + ) + .await; + + mirror_persisted_run_created_to_main(state.as_ref(), run_id).await.unwrap(); + + let main_events = state.store.main_events().await.unwrap(); + let listed = wait_for_main_events(Arc::clone(&main_events), 1).await; + + match &listed[0].event.body { + MainEventBody::RunCreated(props) => { + assert_eq!(props.source.run_id, run_id); + assert_eq!(props.title.as_deref(), Some("Triage bug")); + assert_eq!(props.workflow_slug.as_deref(), Some("triage")); + } + other => panic!("expected created, got {other:?}"), + } +} + +#[tokio::test] +async fn forward_run_events_mirrors_live_lifecycle_events_to_main_stream() { + let state = test_app_state(); + let run_id = fixtures::RUN_1; + let run_store = state.store.create_run(&run_id).await.unwrap(); + let forwarder = tokio::spawn(forward_run_events_to_global( + Arc::clone(&state), + run_id, + run_store.subscribe(), + )); + + append_run_event( + &run_store, + run_id, + "run.started", + json!({ + "name": "triage", + "base_branch": "main", + "run_branch": "fabro/run-1", + "goal": "Triage bug" + }), + ).await; + append_run_event(&run_store, run_id, "run.running", json!({})).await; + + let main_events = state.store.main_events().await.unwrap(); + let listed = wait_for_main_events(Arc::clone(&main_events), 2).await; + + assert!(matches!(listed[0].event.body, MainEventBody::RunStarted(_))); + assert!(matches!(listed[1].event.body, MainEventBody::RunRunning(_))); + + drop(run_store); + forwarder.abort(); +} + +#[tokio::test] +async fn forward_run_events_ignores_non_v1_and_maps_cancelled() { + let state = test_app_state(); + let run_id = fixtures::RUN_1; + let run_store = state.store.create_run(&run_id).await.unwrap(); + let forwarder = tokio::spawn(forward_run_events_to_global( + Arc::clone(&state), + run_id, + run_store.subscribe(), + )); + + append_run_event( + &run_store, + run_id, + "stage.started", + json!({ + "index": 0, + "handler_type": "command", + "attempt": 1, + "max_attempts": 1 + }), + ) + .await; + append_run_event( + &run_store, + run_id, + "run.failed", + json!({ + "failure": { + "reason": "cancelled", + "detail": { + "message": "cancelled", + "category": "canceled" + } + }, + "timing": {"wall_time_ms": 1, "inference_time_ms": 0, "tool_time_ms": 0, "active_time_ms": 0} + }), + ) + .await; + + let main_events = state.store.main_events().await.unwrap(); + let listed = wait_for_main_events(Arc::clone(&main_events), 1).await; + + assert_eq!( + listed.iter().map(|event| event.event.event_name()).collect::>(), + vec!["fabro.run.cancelled"] + ); + + drop(run_store); + forwarder.abort(); +} +``` + +- [ ] **Step 6: Run forwarding tests and verify failure** + +Run: + +```bash +cargo nextest run -p fabro-server main_stream forward_run_events_ +``` + +Expected: tests fail because the server does not mirror persisted `run.created` events or live forwarded lifecycle events. + +- [ ] **Step 7: Add a shared server-side mirroring helper** + +Modify `lib/crates/fabro-server/src/server.rs`. + +Add this helper near `forward_run_events_to_global`: + +```rust +pub(super) async fn append_main_event_for_run_event(state: &AppState, event: &RunEvent) { + if let Some(body) = main_event_mirror::main_event_body_from_run_event(event) { + match state.store.main_events().await { + Ok(main_events) => { + if let Err(err) = main_events.append(body).await { + tracing::error!( + run_id = %event.run_id, + source_event_id = %event.id, + source_event_name = event.event_name(), + error = %err, + "Failed to append mirrored main event" + ); + } + } + Err(err) => { + tracing::error!( + run_id = %event.run_id, + source_event_id = %event.id, + source_event_name = event.event_name(), + error = %err, + "Failed to open main event store for mirroring" + ); + } + } + } +} +``` + +Add this helper near the server run creation helpers: + +```rust +pub(super) async fn mirror_persisted_run_created_to_main( + state: &AppState, + run_id: RunId, +) -> anyhow::Result<()> { + let run_store = state.store.open_run(&run_id).await?; + let events = run_store.list_events_from_with_limit(1, 1).await?; + let Some(created) = events.into_iter().find(|event| { + matches!(event.event.body, EventBody::RunCreated(_)) + }) else { + anyhow::bail!("run {run_id} has no run.created event to mirror"); + }; + append_main_event_for_run_event(state, &created.event).await; + Ok(()) +} +``` + +Modify `lib/crates/fabro-server/src/server/handler/runs.rs` to import `mirror_persisted_run_created_to_main` from `super::super`, then call `mirror_persisted_run_created_to_main(state.as_ref(), created.run_id).await` immediately after `operations::create(...)` succeeds in `create_run_from_manifest`. If reading the persisted `run.created` event fails, return `500` before scheduling the run. Main-event store open/append errors are already logged inside `append_main_event_for_run_event`. Keep ordinary run event append behavior everywhere else in this chunk; later execution lifecycle events are mirrored by the live forwarder. + +- [ ] **Step 8: Mirror live lifecycle events from `forward_run_events_to_global`** + +Modify `lib/crates/fabro-server/src/server.rs`. + +Replace the `Ok(event)` branch in `forward_run_events_to_global` with: + +```rust + Ok(event) => { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + if let Some(managed_run) = runs.get_mut(&run_id) { + reconcile_live_interview_state_for_event(managed_run, &event.event); + } + drop(runs); + + append_main_event_for_run_event(state.as_ref(), &event.event).await; + + let _ = state.global_event_tx.send(event); + } +``` + +Replace the `RecvError::Lagged(_)` branch in the same loop with: + +```rust + Err(RecvError::Lagged(skipped)) => { + tracing::warn!( + run_id = %run_id, + skipped, + "Run event forwarder lagged; skipped live run events cannot be mirrored to the main stream" + ); + } +``` + +- [ ] **Step 9: Run forwarding tests** + +Run: + +```bash +cargo nextest run -p fabro-server main_stream forward_run_events_ +``` + +Expected: forwarding tests pass. + +- [ ] **Step 10: Run focused server lifecycle tests** + +Run: + +```bash +cargo nextest run -p fabro-server main_event_mirror main_stream forward_run_events_ +``` + +Expected: mapper and forwarding tests pass. + +- [ ] **Step 11: Commit** + +```bash +git add lib/crates/fabro-server/src/server/main_event_mirror.rs lib/crates/fabro-server/src/server.rs lib/crates/fabro-server/src/server/handler/runs.rs lib/crates/fabro-server/src/server/tests.rs +git commit -m "feat: mirror run lifecycle events" +``` + +## Task 4: Whole-Workspace Verification + +**Files:** +- No new source files. + +- [ ] **Step 1: Format** + +Run: + +```bash +cargo +nightly-2026-04-14 fmt --all +``` + +Expected: command exits successfully. + +- [ ] **Step 2: Run focused crate tests** + +Run: + +```bash +cargo nextest run -p fabro-types main_event_serde +cargo nextest run -p fabro-store main_events +cargo nextest run -p fabro-server main_event_mirror main_stream forward_run_events_ +``` + +Expected: all focused tests pass. + +- [ ] **Step 3: Run full workspace tests** + +Run: + +```bash +cargo nextest run --workspace +``` + +Expected: all workspace tests pass. If macOS returns `Too many open files (os error 24)`, run: + +```bash +ulimit -n 4096 && cargo nextest run --workspace +``` + +Expected: all workspace tests pass with the raised file descriptor limit. + +- [ ] **Step 4: Run clippy** + +Run: + +```bash +cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings +``` + +Expected: clippy exits successfully with no warnings. + +- [ ] **Step 5: Commit verification-only changes if formatting touched files** + +Run: + +```bash +git status --short +``` + +If only formatting changes from this feature are present, run: + +```bash +git add lib/crates/fabro-types lib/crates/fabro-store lib/crates/fabro-server +git commit -m "style: format main event stream changes" +``` + +If `git status --short` is clean, skip this commit. + +## Self-Review + +- Spec coverage: the plan adds `MainEvent`, stores events under the global `events/main/` logical prefix, recovers sequence numbers, lists from a sequence with bounded `limit + 1` behavior, subscribes to live appends, keeps main events outside run deletion, mirrors v1 run lifecycle names, maps cancelled terminal failures to `fabro.run.cancelled`, ignores non-v1 events, logs live-forwarder lag, and logs mirror append/open failures without failing the run forwarder. +- API scope: no OpenAPI, HTTP, CLI, or SSE surface is added. +- Type consistency: `MainEventBody`, `MainEventSource`, `MainEventStore`, `Database::main_events`, and `main_event_body_from_run_event` use the same names across tasks. +- Compatibility: unknown main event names preserve raw name and properties. +- Test shape: tests are added at `fabro-types`, `fabro-store`, and `fabro-server`, matching the design spec. diff --git a/docs/superpowers/specs/2026-06-21-main-event-stream-design.md b/docs/superpowers/specs/2026-06-21-main-event-stream-design.md new file mode 100644 index 0000000000..9734aa93e2 --- /dev/null +++ b/docs/superpowers/specs/2026-06-21-main-event-stream-design.md @@ -0,0 +1,194 @@ +# Fabro Main Event Stream Design + +## Context + +Fabro currently has durable workflow run event streams. Each run stores canonical +`RunEvent` records under a run-scoped SlateDB prefix: + +```text +runs//events/ +``` + +Those per-run events drive run projections, run event APIs, attach/SSE, CLI +progress, and live server broadcasts. They are also deleted when the run is +deleted. + +The server also has an in-memory `global_event_tx`, but that is only a live +broadcast of per-run events for active server consumers. It is not durable and +is not the "main event stream" needed for future automation triggers. + +This design creates the first chunk of the main event stream: a durable, +server-wide event log for coarse internal Fabro lifecycle events. It does not +include GitHub webhook ingestion, automation trigger matching, public APIs, or +listener cursor persistence. + +## Goals + +- Add a durable main event stream in the existing SlateDB database. +- Store main events under a global prefix that survives run deletion. +- Mirror selected coarse run lifecycle events into the main stream. +- Give main events their own identity, sequence, and timestamp. +- Preserve enough compact event details for future automation matching and audit + even after a run is deleted. +- Keep storage generic and place lifecycle mirroring policy in `fabro-server`. + +## Non-Goals + +- Do not ingest GitHub webhook deliveries in this chunk. +- Do not add event-triggered automations or matcher syntax. +- Do not add public API, SSE, or CLI surfaces for the main stream. +- Do not persist listener cursors or retry state. +- Do not mirror stage, agent, tool, transcript, sandbox, or detailed run events. +- Do not duplicate large run payloads such as full graphs, settings, patches, + transcripts, or tool output. + +## Main Event Model + +Add a new shared type module in `fabro-types` named `main_event`. + +`MainEventEnvelope` contains: + +- `seq`: monotonically increasing main-stream sequence number. +- `event`: the `MainEvent`. + +`MainEvent` contains: + +- `id`: new UUIDv7-style event id for the main stream record. +- `ts`: append timestamp for the main stream record. +- `event`: the main event name through a typed `MainEventBody`. +- `properties`: typed body data through serde's tagged enum shape. + +Each mirrored run lifecycle event gets a distinct main event name: + +- `fabro.run.created` +- `fabro.run.started` +- `fabro.run.running` +- `fabro.run.completed` +- `fabro.run.failed` +- `fabro.run.cancelled` +- `fabro.run.paused` +- `fabro.run.unpaused` + +`run.cancel.requested` is intentionally not mirrored in v1. Terminal +cancellation is represented today by `run.failed` with cancellation semantics, +so the main stream emits `fabro.run.cancelled` for that terminal condition. + +All mirrored events include source metadata: + +- `run_id` +- `source_event_id` +- `source_event_ts` +- `source_event_name` +- `actor`, when present + +Lifecycle bodies include compact standalone summaries: + +- Created: title, workflow slug, automation ref, git context, parent id, and + provenance subject or actor information. +- Started/running: run id, source metadata, and lightweight start metadata when + present. +- Completed: success reason/status, final commit SHA, compact billing summary, + automation ref when available, and source metadata. +- Failed: failure reason/category/message summary, final commit SHA, compact + billing summary, automation ref when available, and source metadata. +- Cancelled: same source shape as failed, but named as a cancellation event and + containing cancellation reason/message summary. +- Paused/unpaused: run id and source metadata. + +The main event stream must stand on its own after run deletion, but it should +not become a second full run event log. + +## Storage + +Add a main event store to `fabro-store::Database` using the same SlateDB +database as runs. Events are stored under: + +```text +events/main/ +``` + +The store supports internal operations: + +- append a `MainEventBody` and return a `MainEventEnvelope`. +- list events from `since_seq` with a bounded limit. +- recover the next sequence number on open by scanning the prefix. +- subscribe to live appended main events for future internal consumers. + +Run deletion remains scoped to `runs//...`, so it does not remove main +events. + +`fabro-store` should not decide which run events are worth mirroring. It only +stores and replays main events. + +## Server Mirroring + +`fabro-server` owns the mirroring policy. + +The first implementation point is the existing active-run forwarding path that +subscribes to each active run's event stream and forwards events into +`global_event_tx`. When a run event arrives: + +1. Reconcile the server's live run state as it does today. +2. Attempt to map the `RunEvent` to a `MainEventBody`. +3. Append the main event if the run event is one of the selected lifecycle + events. +4. Ignore all non-v1 run events. +5. Continue broadcasting the run event through the existing in-memory bus. + +This keeps product policy in the server and leaves `fabro-store` as the durable +storage layer. + +## Error Handling + +This chunk follows the current practical behavior of run event persistence: + +- Server/API lifecycle append paths already fail when their run event append + fails. +- Workflow runtime event logging can be warning-only through the async run event + logger. + +Main event mirroring from the active-run forwarding path logs loudly on append +failure and does not stop or fail the running workflow. This is acceptable for +the first chunk because there is no automation listener yet. A later automation +trigger chunk can introduce stricter fail-closed wrappers for critical server +transitions if trigger correctness requires them. + +## Testing + +Add focused coverage at three layers. + +`fabro-types`: + +- `MainEvent` serde round trips for each v1 event name. +- Event name mapping is explicit and stable. +- Unknown event names deserialize into an `Unknown` body that preserves the + original name and raw properties, matching the compatibility style of + `RunEvent`. + +`fabro-store`: + +- Append and list main events under `events/main/`. +- Sequence numbers are stable and recover after reopening the database. +- `delete_run` does not delete main events. +- Bounded list-from behavior returns ordered events from `since_seq`. + +`fabro-server`: + +- Active run lifecycle events mirror into the main event store. +- Non-v1 events such as stage or agent events are ignored. +- Terminal cancelled `run.failed` maps to `fabro.run.cancelled`. +- Normal `run.failed` maps to `fabro.run.failed`. +- Mirroring failures are logged and do not stop forwarding the original run + event. + +## Deferred Follow-Up Work + +- GitHub webhook producer: convert verified webhook deliveries into durable main + events with raw provider payloads and idempotency metadata. +- Automation event triggers: add deterministic matcher syntax and an internal + replay-plus-tail listener over the main event store. +- Listener cursors: persist per-listener processing state and retry behavior. +- Public read/stream APIs: expose main events to operators or external + consumers only after the internal model has settled. +- Stronger failure semantics: add server wrappers for critical lifecycle paths + if automation correctness requires fail-closed mirroring. diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 2bc7cb946f..6666a2dcbe 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -169,6 +169,7 @@ use crate::{ mod automation_scheduler; mod handler; +mod main_event_mirror; mod resource_sampler; mod session_runtime; @@ -3199,18 +3200,72 @@ async fn forward_run_events_to_global( loop { match run_events.recv().await { Ok(event) => { - let mut runs = state.runs.lock().expect("runs lock poisoned"); - if let Some(managed_run) = runs.get_mut(&run_id) { - reconcile_live_interview_state_for_event(managed_run, &event.event); + { + let mut runs = state.runs.lock().expect("runs lock poisoned"); + if let Some(managed_run) = runs.get_mut(&run_id) { + reconcile_live_interview_state_for_event(managed_run, &event.event); + } } + + append_main_event_for_run_event(state.as_ref(), &event.event).await; + let _ = state.global_event_tx.send(event); } - Err(RecvError::Lagged(_)) => {} + Err(RecvError::Lagged(skipped)) => { + tracing::warn!( + run_id = %run_id, + skipped, + "Run event forwarder lagged; skipped live run events cannot be mirrored to the main stream" + ); + } Err(RecvError::Closed) => break, } } } +pub(super) async fn append_main_event_for_run_event(state: &AppState, event: &RunEvent) { + if let Some(body) = main_event_mirror::main_event_body_from_run_event(event) { + match state.store.main_events().await { + Ok(main_events) => { + if let Err(err) = main_events.append(body).await { + tracing::error!( + run_id = %event.run_id, + source_event_id = %event.id, + source_event_name = event.event_name(), + error = %err, + "Failed to append mirrored main event" + ); + } + } + Err(err) => { + tracing::error!( + run_id = %event.run_id, + source_event_id = %event.id, + source_event_name = event.event_name(), + error = %err, + "Failed to open main event store for mirroring" + ); + } + } + } +} + +pub(super) async fn mirror_persisted_run_created_to_main( + state: &AppState, + run_id: RunId, +) -> anyhow::Result<()> { + let run_store = state.store.open_run(&run_id).await?; + let events = run_store.list_events_from_with_limit(1, 1).await?; + let Some(created) = events + .into_iter() + .find(|event| matches!(event.event.body, EventBody::RunCreated(_))) + else { + anyhow::bail!("run {run_id} has no run.created event to mirror"); + }; + append_main_event_for_run_event(state, &created.event).await; + Ok(()) +} + fn managed_run( dot_source: String, status: RunStatus, diff --git a/lib/crates/fabro-server/src/server/handler/runs.rs b/lib/crates/fabro-server/src/server/handler/runs.rs index 621b0bf5cc..5662dd5574 100644 --- a/lib/crates/fabro-server/src/server/handler/runs.rs +++ b/lib/crates/fabro-server/src/server/handler/runs.rs @@ -35,8 +35,9 @@ use tracing::info; use super::super::{ AppState, DeleteRunOutcome, ListResponse, PaginationParams, RunExecutionMode, answer_from_request, api_question_from_pending_interview, default_page_limit, - delete_run_internal, load_pending_interview, managed_run, paginate_items, parse_run_id_path, - parse_stage_id_path, reject_if_archived, submit_pending_interview_answer, workflow_event, + delete_run_internal, load_pending_interview, managed_run, mirror_persisted_run_created_to_main, + paginate_items, parse_run_id_path, parse_stage_id_path, reject_if_archived, + submit_pending_interview_answer, workflow_event, }; use crate::error::ApiError; use crate::principal_middleware::{ @@ -717,6 +718,13 @@ pub(crate) async fn create_run_from_manifest( .into_response(); } }; + if let Err(err) = mirror_persisted_run_created_to_main(state.as_ref(), created.run_id).await { + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to mirror run creation to main event stream: {err}"), + ) + .into_response(); + } let created_at = created.run_id.created_at(); let summary = match state .store diff --git a/lib/crates/fabro-server/src/server/main_event_mirror.rs b/lib/crates/fabro-server/src/server/main_event_mirror.rs new file mode 100644 index 0000000000..88d2b103ed --- /dev/null +++ b/lib/crates/fabro-server/src/server/main_event_mirror.rs @@ -0,0 +1,215 @@ +use fabro_types::{ + EventBody, FailureReason, MainEventBody, MainEventCancelledProps, MainEventCompletedProps, + MainEventCreatedProps, MainEventFailedProps, MainEventLifecycleProps, MainEventSource, + MainEventStartedProps, RunEvent, +}; + +pub(crate) fn main_event_body_from_run_event(event: &RunEvent) -> Option { + let source = source_from_run_event(event); + match &event.body { + EventBody::RunCreated(props) => Some(MainEventBody::RunCreated(MainEventCreatedProps { + source, + title: props.title.clone(), + workflow_slug: props.workflow_slug.clone(), + automation: props.automation.clone(), + git: props.git.clone(), + parent_id: props.parent_id, + provenance_subject: Some(props.provenance.subject.clone()), + })), + EventBody::RunStarted(props) => Some(MainEventBody::RunStarted(MainEventStartedProps { + source, + name: props.name.clone(), + base_branch: props.base_branch.clone(), + base_sha: props.base_sha.clone(), + run_branch: props.run_branch.clone(), + worktree_dir: props.worktree_dir.clone(), + goal: props.goal.clone(), + })), + EventBody::RunRunning(_) => Some(MainEventBody::RunRunning(MainEventLifecycleProps { + source, + })), + EventBody::RunCompleted(props) => { + Some(MainEventBody::RunCompleted(MainEventCompletedProps { + source, + status: props.status.clone(), + reason: props.reason, + total_usd_micros: props.total_usd_micros, + final_git_commit_sha: props.final_git_commit_sha.clone(), + automation: None, + billing: props.billing.clone(), + })) + } + EventBody::RunFailed(props) if props.failure.reason == FailureReason::Cancelled => { + Some(MainEventBody::RunCancelled(MainEventCancelledProps { + source, + reason: props.failure.reason, + category: props.failure.detail.category, + message: props.failure.detail.message.clone(), + final_git_commit_sha: props.final_git_commit_sha.clone(), + automation: None, + billing: props.billing.clone(), + })) + } + EventBody::RunFailed(props) => Some(MainEventBody::RunFailed(MainEventFailedProps { + source, + reason: props.failure.reason, + category: props.failure.detail.category, + message: props.failure.detail.message.clone(), + final_git_commit_sha: props.final_git_commit_sha.clone(), + automation: None, + billing: props.billing.clone(), + })), + EventBody::RunPaused(_) => { + Some(MainEventBody::RunPaused(MainEventLifecycleProps { source })) + } + EventBody::RunUnpaused(_) => Some(MainEventBody::RunUnpaused(MainEventLifecycleProps { + source, + })), + _ => None, + } +} + +fn source_from_run_event(event: &RunEvent) -> MainEventSource { + MainEventSource { + run_id: event.run_id, + source_event_id: event.id.clone(), + source_event_ts: event.ts, + source_event_name: event.event_name().to_string(), + actor: event.actor.clone(), + } +} + +#[cfg(test)] +mod tests { + use chrono::{TimeZone, Utc}; + use fabro_types::run_event::RunFailedProps; + use fabro_types::{ + EventBody, FailureCategory, FailureDetail, FailureReason, Graph, MainEventBody, RunEvent, + RunFailure, RunTiming, WorkflowSettings, fixtures, test_support, + }; + use serde_json::json; + + use super::main_event_body_from_run_event; + + fn run_event(event_name: &str, properties: &serde_json::Value) -> RunEvent { + RunEvent::from_value(json!({ + "id": format!("evt-{event_name}"), + "ts": "2026-06-21T12:00:00Z", + "run_id": fixtures::RUN_1, + "event": event_name, + "properties": properties.clone() + })) + .unwrap() + } + + #[test] + fn maps_created_with_compact_identity_fields() { + let event = run_event( + "run.created", + &json!({ + "title": "Triage bug", + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "run_dir": "/tmp/test", + "workflow_slug": "triage", + "provenance": test_support::test_run_provenance(), + "parent_id": fixtures::RUN_2, + }), + ); + + let mapped = main_event_body_from_run_event(&event).unwrap(); + match mapped { + MainEventBody::RunCreated(props) => { + assert_eq!(props.source.run_id, fixtures::RUN_1); + assert_eq!(props.source.source_event_name, "run.created"); + assert_eq!(props.title.as_deref(), Some("Triage bug")); + assert_eq!(props.workflow_slug.as_deref(), Some("triage")); + assert_eq!(props.parent_id, Some(fixtures::RUN_2)); + assert!(props.provenance_subject.is_some()); + } + other => panic!("expected created, got {other:?}"), + } + } + + #[test] + fn maps_failed_and_cancelled_separately() { + let failed = RunEvent { + id: "evt-failed".to_string(), + ts: Utc.with_ymd_and_hms(2026, 6, 21, 12, 0, 0).unwrap(), + run_id: fixtures::RUN_1, + node_id: None, + node_label: None, + stage_id: None, + parallel_group_id: None, + parallel_branch_id: None, + session_id: None, + parent_session_id: None, + tool_call_id: None, + actor: None, + body: EventBody::RunFailed(RunFailedProps { + failure: RunFailure { + reason: FailureReason::WorkflowError, + detail: FailureDetail { + message: "boom".to_string(), + category: FailureCategory::Deterministic, + causes: Vec::new(), + system_actor: None, + signature: None, + exec_output_tail: None, + }, + }, + timing: RunTiming::default(), + final_git_commit_sha: Some("abc123".to_string()), + final_patch: None, + diff_summary: None, + billing: None, + }), + }; + assert!(matches!( + main_event_body_from_run_event(&failed), + Some(MainEventBody::RunFailed(_)) + )); + + let mut cancelled = failed.clone(); + cancelled.id = "evt-cancelled".to_string(); + cancelled.body = EventBody::RunFailed(RunFailedProps { + failure: RunFailure { + reason: FailureReason::Cancelled, + detail: FailureDetail { + message: "cancelled".to_string(), + category: FailureCategory::Canceled, + causes: Vec::new(), + system_actor: None, + signature: None, + exec_output_tail: None, + }, + }, + timing: RunTiming::default(), + final_git_commit_sha: None, + final_patch: None, + diff_summary: None, + billing: None, + }); + assert!(matches!( + main_event_body_from_run_event(&cancelled), + Some(MainEventBody::RunCancelled(_)) + )); + } + + #[test] + fn ignores_non_v1_events_and_cancel_requested() { + let stage = run_event( + "stage.started", + &json!({ + "index": 0, + "handler_type": "command", + "attempt": 1, + "max_attempts": 1 + }), + ); + assert!(main_event_body_from_run_event(&stage).is_none()); + + let cancel_requested = run_event("run.cancel.requested", &json!({ "action": "cancel" })); + assert!(main_event_body_from_run_event(&cancel_requested).is_none()); + } +} diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index b6333b3174..a12ef0628c 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -26,7 +26,7 @@ use fabro_types::settings::ServerAuthMethod; use fabro_types::settings::run::EnvironmentProvider; use fabro_types::{ AgentBackend, AttrValue, AuthMethod, CommandTermination, FailureCategory, FailureDetail, Graph, - InterviewQuestionRecord, Node, Outcome, QuestionType, RunBlobId, RunId, RunSpec, + InterviewQuestionRecord, MainEventBody, Node, Outcome, QuestionType, RunBlobId, RunId, RunSpec, SandboxProviderKind, StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness, StageContextWindowWarning, StageModelUsage, StageTiming, SuccessReason, SystemActorKind, @@ -160,6 +160,211 @@ fn run_json_archived(run: &serde_json::Value) -> bool { run["lifecycle"]["archived"].as_bool().unwrap_or(false) } +async fn wait_for_main_events( + main_events: StdArc, + expected_len: usize, +) -> Vec { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + let listed = main_events.list_from_with_limit(1, 100).await.unwrap(); + if listed.len() == expected_len { + return listed; + } + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for {expected_len} main events, saw {}", + listed.len() + ); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } +} + +fn run_event_value( + run_id: RunId, + event_name: &str, + properties: &serde_json::Value, +) -> serde_json::Value { + json!({ + "id": format!("evt-{event_name}"), + "ts": "2026-06-21T12:00:00Z", + "run_id": run_id, + "event": event_name, + "properties": properties.clone() + }) +} + +async fn append_run_event( + run_store: &fabro_store::RunDatabase, + run_id: RunId, + event_name: &str, + properties: &serde_json::Value, +) { + let payload = + fabro_store::EventPayload::new(run_event_value(run_id, event_name, properties), &run_id) + .unwrap(); + run_store.append_event(&payload).await.unwrap(); +} + +#[tokio::test] +async fn persisted_run_created_mirrors_to_main_stream_without_forwarder() { + let state = test_app_state(); + let run_id = fixtures::RUN_1; + let run_store = state.store.create_run(&run_id).await.unwrap(); + + append_run_event( + &run_store, + run_id, + "run.created", + &json!({ + "title": "Triage bug", + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "run_dir": "/tmp/test", + "workflow_slug": "triage", + "provenance": test_support::test_run_provenance() + }), + ) + .await; + + mirror_persisted_run_created_to_main(state.as_ref(), run_id) + .await + .unwrap(); + + let main_events = state.store.main_events().await.unwrap(); + let listed = wait_for_main_events(StdArc::clone(&main_events), 1).await; + + match &listed[0].event.body { + MainEventBody::RunCreated(props) => { + assert_eq!(props.source.run_id, run_id); + assert_eq!(props.title.as_deref(), Some("Triage bug")); + assert_eq!(props.workflow_slug.as_deref(), Some("triage")); + } + other => panic!("expected created, got {other:?}"), + } +} + +#[tokio::test] +async fn forward_run_events_mirrors_live_lifecycle_events_to_main_stream() { + let state = test_app_state(); + let run_id = fixtures::RUN_1; + let run_store = state.store.create_run(&run_id).await.unwrap(); + append_run_event( + &run_store, + run_id, + "run.created", + &json!({ + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "run_dir": "/tmp/test", + "provenance": test_support::test_run_provenance() + }), + ) + .await; + let forwarder = tokio::spawn(forward_run_events_to_global( + StdArc::clone(&state), + run_id, + run_store.subscribe(), + )); + + append_run_event( + &run_store, + run_id, + "run.started", + &json!({ + "name": "triage", + "base_branch": "main", + "run_branch": "fabro/run-1", + "goal": "Triage bug" + }), + ) + .await; + append_run_event( + &run_store, + run_id, + "run.runnable", + &json!({ "source": "start_requested" }), + ) + .await; + append_run_event(&run_store, run_id, "run.starting", &json!({})).await; + append_run_event(&run_store, run_id, "run.running", &json!({})).await; + + let main_events = state.store.main_events().await.unwrap(); + let listed = wait_for_main_events(StdArc::clone(&main_events), 2).await; + + assert!(matches!(listed[0].event.body, MainEventBody::RunStarted(_))); + assert!(matches!(listed[1].event.body, MainEventBody::RunRunning(_))); + + drop(run_store); + forwarder.abort(); +} + +#[tokio::test] +async fn forward_run_events_ignores_non_v1_and_maps_cancelled() { + let state = test_app_state(); + let run_id = fixtures::RUN_1; + let run_store = state.store.create_run(&run_id).await.unwrap(); + append_run_event( + &run_store, + run_id, + "run.created", + &json!({ + "settings": WorkflowSettings::default(), + "graph": Graph::new("test"), + "run_dir": "/tmp/test", + "provenance": test_support::test_run_provenance() + }), + ) + .await; + let forwarder = tokio::spawn(forward_run_events_to_global( + StdArc::clone(&state), + run_id, + run_store.subscribe(), + )); + + append_run_event( + &run_store, + run_id, + "stage.started", + &json!({ + "index": 0, + "handler_type": "command", + "attempt": 1, + "max_attempts": 1 + }), + ) + .await; + append_run_event( + &run_store, + run_id, + "run.failed", + &json!({ + "failure": { + "reason": "cancelled", + "detail": { + "message": "cancelled", + "category": "canceled" + } + }, + "timing": {"wall_time_ms": 1, "inference_time_ms": 0, "tool_time_ms": 0, "active_time_ms": 0} + }), + ) + .await; + + let main_events = state.store.main_events().await.unwrap(); + let listed = wait_for_main_events(StdArc::clone(&main_events), 1).await; + + assert_eq!( + listed + .iter() + .map(|event| event.event.event_name()) + .collect::>(), + vec!["fabro.run.cancelled"] + ); + + drop(run_store); + forwarder.abort(); +} + async fn mock_daytona_auth_probe(server: &MockServer) -> httpmock::Mock<'_> { server .mock_async(|when, then| { diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index 2451df6329..07ed71dc8f 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/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 { segments.next()?.split_once('-')?.0.parse().ok() } +pub(crate) fn parse_main_event_seq(key: &str) -> Option { + 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 { 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,23 @@ 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!( diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index cfc8e1c0c1..cf68098787 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/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; diff --git a/lib/crates/fabro-store/src/slate/main_event_store.rs b/lib/crates/fabro-store/src/slate/main_event_store.rs new file mode 100644 index 0000000000..0960233b58 --- /dev/null +++ b/lib/crates/fabro-store/src/slate/main_event_store.rs @@ -0,0 +1,124 @@ +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, +} + +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, +} + +impl MainEventStore { + pub(crate) async fn open(db: Db) -> Result { + 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 { + 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 { + self.inner.event_tx.subscribe() + } + + pub async fn list_from_with_limit( + &self, + start_seq: u32, + limit: usize, + ) -> Result> { + 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(db: &R) -> Result +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(db: &R, start_seq: u32) -> Result> +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::from_utf8(key.to_vec()) + .map_err(|err| Error::Other(format!("stored key is not valid UTF-8: {err}"))) +} diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 8cc1bd41d5..09fbd82cce 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/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>>>, blobs: Arc>>, catalog_index: Arc>>, + main_events: Arc>>, auth_codes: Arc>>, refresh_tokens: Arc>>, projection_cache: Arc, @@ -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> { + 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> { 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![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![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(); diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index ce31e51399..7c67b51f35 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -15,6 +15,7 @@ pub mod graph; mod id; pub mod interview; pub mod llm_backend; +pub mod main_event; pub mod manifest_path; pub mod outcome; pub mod pair; @@ -74,6 +75,11 @@ pub use graph::{ }; pub use interview::{InterviewQuestionRecord, QuestionType}; pub use llm_backend::AgentBackend; +pub use main_event::{ + MainEvent, MainEventBody, MainEventCancelledProps, MainEventCompletedProps, + MainEventCreatedProps, MainEventEnvelope, MainEventFailedProps, MainEventLifecycleProps, + MainEventSource, MainEventStartedProps, +}; pub use manifest_path::{ManifestPath, ManifestPathParseError}; pub use outcome::{ FailureCategory, FailureDetail, NodeResult, Outcome, OutcomeMeta, StageOutcome, StageState, diff --git a/lib/crates/fabro-types/src/main_event.rs b/lib/crates/fabro-types/src/main_event.rs new file mode 100644 index 0000000000..5db4dc9443 --- /dev/null +++ b/lib/crates/fabro-types/src/main_event.rs @@ -0,0 +1,311 @@ +use chrono::{DateTime, Utc}; +use serde::de::Error as DeError; +use serde::ser::Error as SerError; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::{Map, Value, json}; + +use crate::{ + AutomationRef, BilledTokenCounts, FailureCategory, FailureReason, GitContext, Principal, RunId, + SuccessReason, +}; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventEnvelope { + pub seq: u32, + pub event: MainEvent, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct MainEvent { + pub id: String, + pub ts: DateTime, + pub body: MainEventBody, +} + +#[allow( + clippy::large_enum_variant, + reason = "Main event bodies stay inline to match the tagged wire format." +)] +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +#[serde(tag = "event", content = "properties")] +pub enum MainEventBody { + #[serde(rename = "fabro.run.created")] + RunCreated(MainEventCreatedProps), + #[serde(rename = "fabro.run.started")] + RunStarted(MainEventStartedProps), + #[serde(rename = "fabro.run.running")] + RunRunning(MainEventLifecycleProps), + #[serde(rename = "fabro.run.completed")] + RunCompleted(MainEventCompletedProps), + #[serde(rename = "fabro.run.failed")] + RunFailed(MainEventFailedProps), + #[serde(rename = "fabro.run.cancelled")] + RunCancelled(MainEventCancelledProps), + #[serde(rename = "fabro.run.paused")] + RunPaused(MainEventLifecycleProps), + #[serde(rename = "fabro.run.unpaused")] + RunUnpaused(MainEventLifecycleProps), + Unknown { + name: String, + properties: Value, + }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventSource { + pub run_id: RunId, + pub source_event_id: String, + pub source_event_ts: DateTime, + pub source_event_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub actor: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventCreatedProps { + #[serde(flatten)] + pub source: MainEventSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub title: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub workflow_slug: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub git: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provenance_subject: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventLifecycleProps { + #[serde(flatten)] + pub source: MainEventSource, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventStartedProps { + #[serde(flatten)] + pub source: MainEventSource, + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_branch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_branch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worktree_dir: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub goal: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventCompletedProps { + #[serde(flatten)] + pub source: MainEventSource, + pub status: String, + pub reason: SuccessReason, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_usd_micros: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub billing: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventFailedProps { + #[serde(flatten)] + pub source: MainEventSource, + pub reason: FailureReason, + pub category: FailureCategory, + pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub billing: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MainEventCancelledProps { + #[serde(flatten)] + pub source: MainEventSource, + pub reason: FailureReason, + pub category: FailureCategory, + pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_git_commit_sha: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub automation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub billing: Option, +} + +#[derive(Debug, Clone, Deserialize)] +struct MainEventRaw { + id: String, + ts: DateTime, + event: String, + #[serde(default = "default_properties")] + properties: Value, +} + +struct MainEventParts<'a> { + id: String, + ts: DateTime, + event: &'a str, + properties: &'a Value, +} + +impl MainEventBody { + pub fn event_name(&self) -> &str { + match self { + Self::RunCreated(_) => "fabro.run.created", + Self::RunStarted(_) => "fabro.run.started", + Self::RunRunning(_) => "fabro.run.running", + Self::RunCompleted(_) => "fabro.run.completed", + Self::RunFailed(_) => "fabro.run.failed", + Self::RunCancelled(_) => "fabro.run.cancelled", + Self::RunPaused(_) => "fabro.run.paused", + Self::RunUnpaused(_) => "fabro.run.unpaused", + Self::Unknown { name, .. } => name.as_str(), + } + } + + fn properties_value(&self) -> serde_json::Result { + if let Self::Unknown { properties, .. } = self { + return Ok(properties.clone()); + } + + match serde_json::to_value(self)? { + Value::Object(mut map) => { + Ok(map.remove("properties").unwrap_or_else(default_properties)) + } + _ => Ok(default_properties()), + } + } +} + +fn is_known_main_event_name(event: &str) -> bool { + matches!( + event, + "fabro.run.created" + | "fabro.run.started" + | "fabro.run.running" + | "fabro.run.completed" + | "fabro.run.failed" + | "fabro.run.cancelled" + | "fabro.run.paused" + | "fabro.run.unpaused" + ) +} + +impl MainEvent { + pub fn from_value(value: Value) -> serde_json::Result { + let raw: MainEventRaw = serde_json::from_value(value)?; + Self::from_parts(MainEventParts { + id: raw.id, + ts: raw.ts, + event: &raw.event, + properties: &raw.properties, + }) + } + + pub fn from_ref(value: &Value) -> serde_json::Result { + let obj = value.as_object().ok_or_else(|| { + ::custom("main event must be a JSON object") + })?; + let id = obj.get("id").and_then(Value::as_str).ok_or_else(|| { + ::custom("missing or non-string field: id") + })?; + let ts = obj + .get("ts") + .ok_or_else(|| ::custom("missing field: ts")) + .and_then(DateTime::::deserialize)?; + let event = obj.get("event").and_then(Value::as_str).ok_or_else(|| { + ::custom("missing or non-string field: event") + })?; + let properties = obj + .get("properties") + .cloned() + .unwrap_or_else(default_properties); + Self::from_parts(MainEventParts { + id: id.to_string(), + ts, + event, + properties: &properties, + }) + } + + fn from_parts(parts: MainEventParts<'_>) -> serde_json::Result { + let body_payload = json!({ + "event": parts.event, + "properties": parts.properties, + }); + let body: MainEventBody = match serde_json::from_value(body_payload) { + Ok(body) => body, + Err(err) if is_known_main_event_name(parts.event) => return Err(err), + Err(_) => MainEventBody::Unknown { + name: parts.event.to_string(), + properties: parts.properties.clone(), + }, + }; + Ok(Self { + id: parts.id, + ts: parts.ts, + body, + }) + } + + pub fn to_value(&self) -> serde_json::Result { + let mut map = Map::new(); + map.insert("id".to_string(), Value::String(self.id.clone())); + map.insert("ts".to_string(), serde_json::to_value(self.ts)?); + map.insert( + "event".to_string(), + Value::String(self.body.event_name().to_string()), + ); + map.insert("properties".to_string(), self.body.properties_value()?); + Ok(Value::Object(map)) + } + + pub fn event_name(&self) -> &str { + self.body.event_name() + } + + pub fn properties(&self) -> serde_json::Result { + self.body.properties_value() + } +} + +impl Serialize for MainEvent { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + self.to_value() + .map_err(S::Error::custom)? + .serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for MainEvent { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + Self::from_value(value).map_err(D::Error::custom) + } +} + +fn default_properties() -> Value { + Value::Object(Map::new()) +} diff --git a/lib/crates/fabro-types/tests/main_event_serde.rs b/lib/crates/fabro-types/tests/main_event_serde.rs new file mode 100644 index 0000000000..b444ddae74 --- /dev/null +++ b/lib/crates/fabro-types/tests/main_event_serde.rs @@ -0,0 +1,188 @@ +use chrono::{TimeZone, Utc}; +use fabro_types::{ + AuthMethod, FailureCategory, FailureReason, IdpIdentity, MainEvent, MainEventBody, + MainEventCancelledProps, MainEventCompletedProps, MainEventCreatedProps, MainEventFailedProps, + MainEventLifecycleProps, MainEventSource, MainEventStartedProps, Principal, RunId, + SuccessReason, +}; +use serde_json::json; + +fn run_id() -> RunId { + "01JT56VE4Z5NZ814GZN2JZD65A" + .parse() + .expect("test run id should parse") +} + +fn actor() -> Principal { + Principal::user( + IdpIdentity::new("https://github.com", "123").expect("test identity should construct"), + "fabian".to_string(), + AuthMethod::Github, + ) +} + +fn source(source_event_name: &str) -> MainEventSource { + MainEventSource { + run_id: run_id(), + source_event_id: format!("evt-{source_event_name}"), + source_event_ts: Utc + .with_ymd_and_hms(2026, 6, 21, 10, 0, 0) + .single() + .expect("test source timestamp should be valid"), + source_event_name: source_event_name.to_string(), + actor: Some(actor()), + } +} + +fn main_event(body: MainEventBody) -> MainEvent { + MainEvent { + id: "019796d3-d2d0-7c81-a65a-6f3ddc67d4a4".to_string(), + ts: Utc + .with_ymd_and_hms(2026, 6, 21, 10, 1, 0) + .single() + .expect("test main event timestamp should be valid"), + body, + } +} + +#[test] +fn v1_event_names_are_stable() { + let cases = [ + ( + MainEventBody::RunCreated(MainEventCreatedProps { + source: source("run.created"), + title: Some("Triage bug".to_string()), + workflow_slug: Some("triage".to_string()), + automation: None, + git: None, + parent_id: None, + provenance_subject: Some(actor()), + }), + "fabro.run.created", + ), + ( + MainEventBody::RunStarted(MainEventStartedProps { + source: source("run.started"), + name: "triage".to_string(), + base_branch: Some("main".to_string()), + base_sha: Some("abc123".to_string()), + run_branch: Some("fabro/run-1".to_string()), + worktree_dir: None, + goal: Some("Triage bug".to_string()), + }), + "fabro.run.started", + ), + ( + MainEventBody::RunRunning(MainEventLifecycleProps { + source: source("run.running"), + }), + "fabro.run.running", + ), + ( + MainEventBody::RunCompleted(MainEventCompletedProps { + source: source("run.completed"), + status: "succeeded".to_string(), + reason: SuccessReason::Completed, + total_usd_micros: Some(12), + final_git_commit_sha: Some("abc123".to_string()), + automation: None, + billing: None, + }), + "fabro.run.completed", + ), + ( + MainEventBody::RunFailed(MainEventFailedProps { + source: source("run.failed"), + reason: FailureReason::WorkflowError, + category: FailureCategory::Deterministic, + message: "boom".to_string(), + final_git_commit_sha: None, + automation: None, + billing: None, + }), + "fabro.run.failed", + ), + ( + MainEventBody::RunCancelled(MainEventCancelledProps { + source: source("run.failed"), + reason: FailureReason::Cancelled, + category: FailureCategory::Canceled, + message: "cancelled".to_string(), + final_git_commit_sha: None, + automation: None, + billing: None, + }), + "fabro.run.cancelled", + ), + ( + MainEventBody::RunPaused(MainEventLifecycleProps { + source: source("run.paused"), + }), + "fabro.run.paused", + ), + ( + MainEventBody::RunUnpaused(MainEventLifecycleProps { + source: source("run.unpaused"), + }), + "fabro.run.unpaused", + ), + ]; + + for (body, expected) in cases { + assert_eq!(body.event_name(), expected); + assert_eq!(main_event(body).event_name(), expected); + } +} + +#[test] +fn main_event_round_trips_flat_event_properties() { + let event = main_event(MainEventBody::RunCompleted(MainEventCompletedProps { + source: source("run.completed"), + status: "succeeded".to_string(), + reason: SuccessReason::Completed, + total_usd_micros: Some(42), + final_git_commit_sha: Some("deadbeef".to_string()), + automation: None, + billing: None, + })); + + let value = serde_json::to_value(&event).expect("main event should serialize"); + assert_eq!(value["event"], "fabro.run.completed"); + assert_eq!(value["properties"]["run_id"], run_id().to_string()); + assert_eq!(value["properties"]["source_event_name"], "run.completed"); + assert_eq!(value["properties"]["status"], "succeeded"); + + let parsed: MainEvent = + serde_json::from_value(value.clone()).expect("main event should deserialize"); + assert_eq!( + serde_json::to_value(parsed).expect("main event should serialize after round trip"), + value + ); +} + +#[test] +fn unknown_main_event_preserves_name_and_properties() { + let value = json!({ + "id": "019796d3-d2d0-7c81-a65a-6f3ddc67d4a4", + "ts": "2026-06-21T10:01:00Z", + "event": "github.issue_comment.created", + "properties": { + "delivery_id": "delivery-1", + "action": "created" + } + }); + + let parsed: MainEvent = + serde_json::from_value(value.clone()).expect("unknown main event should deserialize"); + match &parsed.body { + MainEventBody::Unknown { name, properties } => { + assert_eq!(name, "github.issue_comment.created"); + assert_eq!(properties["delivery_id"], "delivery-1"); + } + other => panic!("expected unknown event, got {other:?}"), + } + assert_eq!( + serde_json::to_value(parsed).expect("unknown main event should serialize after round trip"), + value + ); +}