From a7be127fbb95a7c8f70b653442a5d870160e416d Mon Sep 17 00:00:00 2001 From: fkukuck Date: Fri, 16 Oct 2026 20:02:55 +0000 Subject: [PATCH 1/3] feat: define compact main lifecycle events --- lib/foundation/fabro-types/src/lib.rs | 6 + lib/foundation/fabro-types/src/main_event.rs | 311 ++++++++++++++++++ .../fabro-types/tests/main_event_serde.rs | 202 ++++++++++++ 3 files changed, 519 insertions(+) create mode 100644 lib/foundation/fabro-types/src/main_event.rs create mode 100644 lib/foundation/fabro-types/tests/main_event_serde.rs diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index e08f61bfa..6dc0cace2 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -19,6 +19,7 @@ mod id; mod input_scalar; pub mod interview; pub mod llm_backend; +pub mod main_event; pub mod manifest_path; pub mod mcp_store; pub mod model_test; @@ -95,6 +96,11 @@ pub use interview::{ ReviewTargetKind, }; 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 mcp_store::{ McpServerDefinition, McpServerDraft, McpServerId, McpServerReplace, McpServerRevision, diff --git a/lib/foundation/fabro-types/src/main_event.rs b/lib/foundation/fabro-types/src/main_event.rs new file mode 100644 index 000000000..4b6719f9b --- /dev/null +++ b/lib/foundation/fabro-types/src/main_event.rs @@ -0,0 +1,311 @@ +use chrono::{DateTime, Utc}; +use lithos_llm::types::Usage; +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, 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 usage: 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 usage: 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 usage: 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/foundation/fabro-types/tests/main_event_serde.rs b/lib/foundation/fabro-types/tests/main_event_serde.rs new file mode 100644 index 000000000..1983fee54 --- /dev/null +++ b/lib/foundation/fabro-types/tests/main_event_serde.rs @@ -0,0 +1,202 @@ +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, + usage: 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, + usage: 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, + usage: 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); + let event = main_event(body); + assert_eq!(event.event_name(), expected); + let value = serde_json::to_value(&event).unwrap(); + assert_eq!(serde_json::from_value::(value).unwrap(), event); + } +} + +#[test] +fn malformed_known_main_event_is_rejected() { + let value = json!({ + "id": "event-1", + "ts": "2026-06-21T10:01:00Z", + "event": "fabro.run.created", + "properties": {"run_id": 42} + }); + assert!(serde_json::from_value::(value).is_err()); +} + +#[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, + usage: 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 + ); +} From 20638361a8c4a1943ee2a790371d79136730814d Mon Sep 17 00:00:00 2001 From: fkukuck Date: Fri, 16 Oct 2026 20:02:55 +0000 Subject: [PATCH 2/3] feat: persist idempotent main events in SQLite --- lib/components/fabro-store/src/database.rs | 14 +- lib/components/fabro-store/src/lib.rs | 2 + .../fabro-store/src/main_event_store.rs | 353 ++++++++++++++++++ .../fabro-store/src/test_support/mod.rs | 1 + .../migrations/2026101601_main_events.sql | 7 + lib/foundation/fabro-db/src/lib.rs | 4 + 6 files changed, 378 insertions(+), 3 deletions(-) create mode 100644 lib/components/fabro-store/src/main_event_store.rs create mode 100644 lib/foundation/fabro-db/migrations/2026101601_main_events.sql diff --git a/lib/components/fabro-store/src/database.rs b/lib/components/fabro-store/src/database.rs index cf5d4ff5e..3acb62124 100644 --- a/lib/components/fabro-store/src/database.rs +++ b/lib/components/fabro-store/src/database.rs @@ -3,19 +3,21 @@ //! //! A run's history is Petri's records (`petri_records`) and Fabro's //! platform records; its view is the projection the projector commits. -//! This handle carries the two stores every reader of a run reaches, and -//! the run-level operations that span them. +//! This handle carries the stores every reader of a run reaches, the +//! independent main lifecycle stream, and the run-level operations that +//! span them. use std::sync::Arc; use fabro_types::{RunId, RunProjection}; -use crate::{BlobStore, PlatformRecordHook, Result, RunSummaryStore}; +use crate::{BlobStore, MainEventStore, PlatformRecordHook, Result, RunSummaryStore}; #[derive(Clone)] pub struct Database { blobs: Arc, run_summary_store: Arc, + main_events: Arc, } impl std::fmt::Debug for Database { @@ -28,6 +30,7 @@ impl Database { #[must_use] pub fn new(blobs: Arc, run_summary_store: Arc) -> Self { Self { + main_events: Arc::new(MainEventStore::new(run_summary_store.pool())), blobs, run_summary_store, } @@ -43,6 +46,11 @@ impl Database { Arc::clone(&self.blobs) } + #[must_use] + pub fn main_events(&self) -> Arc { + Arc::clone(&self.main_events) + } + /// The run's projection: the one its projector last committed over /// Petri's records and the platform records, or `None` before the /// first view pass commits (or for no such run). diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 9b4a8e305..153ee416b 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -5,6 +5,7 @@ mod blob_store; mod database; mod error; mod keyed_mutex; +mod main_event_store; pub mod platform_records; mod run_session_event_store; mod run_session_record_store; @@ -31,6 +32,7 @@ pub use fabro_types::{ BlobHash, PendingInterviewRecord, Run, RunProjection, StageId, StageProjection, }; pub use keyed_mutex::{KeyedMutex, KeyedMutexGuard}; +pub use main_event_store::{AppendMainEventOutcome, MainEventSourceKey, MainEventStore}; pub use platform_records::{ PlatformRecord, PlatformRecordHook, PlatformRecordKind, PlatformRecordStore, StagePosition, StoredPlatformRecord, diff --git a/lib/components/fabro-store/src/main_event_store.rs b/lib/components/fabro-store/src/main_event_store.rs new file mode 100644 index 000000000..3def2e5ab --- /dev/null +++ b/lib/components/fabro-store/src/main_event_store.rs @@ -0,0 +1,353 @@ +//! The compact server-wide event stream, independent of any run's lifetime. +//! +//! A source key and its event commit together. The run stream is an outbox: +//! retrying a delivery after a crash returns the original event without +//! allocating another global sequence or publishing another live message. + +use std::sync::Arc; + +use chrono::Utc; +use fabro_types::{MainEvent, MainEventBody, MainEventEnvelope, RunId}; +use sqlx::SqlitePool; +use tokio::sync::{Mutex, broadcast}; +use uuid::Uuid; + +use crate::{Error, Result}; + +const LIVE_CAPACITY: usize = 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MainEventSourceKey { + RunStream { run_id: RunId, item_id: String }, +} + +impl MainEventSourceKey { + fn store_key(&self) -> String { + let Self::RunStream { run_id, item_id } = self; + format!("runs:{run_id}:{item_id}") + } +} + +#[derive(Debug, Clone, PartialEq)] +pub enum AppendMainEventOutcome { + Appended(MainEventEnvelope), + Duplicate(MainEventEnvelope), +} + +#[derive(Clone)] +pub struct MainEventStore { + pool: SqlitePool, + append_lock: Arc>, + live: broadcast::Sender, +} + +impl std::fmt::Debug for MainEventStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MainEventStore").finish_non_exhaustive() + } +} + +impl MainEventStore { + #[must_use] + pub fn new(pool: SqlitePool) -> Self { + let (live, _) = broadcast::channel(LIVE_CAPACITY); + Self { + pool, + append_lock: Arc::new(Mutex::new(())), + live, + } + } + + pub async fn append(&self, body: MainEventBody) -> Result { + match self.append_inner(None, body).await? { + AppendMainEventOutcome::Appended(event) | AppendMainEventOutcome::Duplicate(event) => { + Ok(event) + } + } + } + + pub async fn append_once( + &self, + source: MainEventSourceKey, + body: MainEventBody, + ) -> Result { + self.append_inner(Some(source.store_key()), body).await + } + + async fn append_inner( + &self, + source_key: Option, + body: MainEventBody, + ) -> Result { + // Keep broadcasts in commit order for all clones of this handle. + // SQLite's write lock also serializes independent store handles. + let _guard = self.append_lock.lock().await; + let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; + let existing = match &source_key { + Some(source_key) => { + sqlx::query_as::<_, (i64, String)>( + "SELECT seq, event_json FROM main_events WHERE source_key = ?", + ) + .bind(source_key) + .fetch_optional(&mut *transaction) + .await? + } + None => None, + }; + if let Some((seq, json)) = existing { + let event = decode_event(seq, &json)?; + transaction.commit().await?; + return Ok(AppendMainEventOutcome::Duplicate(event)); + } + let last_seq: Option = sqlx::query_scalar("SELECT MAX(seq) FROM main_events") + .fetch_one(&mut *transaction) + .await?; + let seq = u32::try_from(last_seq.unwrap_or(0)) + .ok() + .and_then(|last| last.checked_add(1)) + .ok_or(Error::EventSequenceExhausted { max_seq: u32::MAX })?; + let event = MainEvent { + id: Uuid::now_v7().to_string(), + ts: Utc::now(), + body, + }; + sqlx::query("INSERT INTO main_events (seq, source_key, event_json) VALUES (?, ?, ?)") + .bind(i64::from(seq)) + .bind(source_key) + .bind(serde_json::to_string(&event)?) + .execute(&mut *transaction) + .await?; + transaction.commit().await?; + let envelope = MainEventEnvelope { seq, event }; + let _ = self.live.send(envelope.clone()); + Ok(AppendMainEventOutcome::Appended(envelope)) + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.live.subscribe() + } + + /// Replay from an inclusive sequence, returning at most `limit + 1` + /// events so a consumer can determine whether another page exists. + pub async fn list_from_with_limit( + &self, + start_seq: u32, + limit: usize, + ) -> Result> { + let rows = sqlx::query_as::<_, (i64, String)>( + "SELECT seq, event_json FROM main_events WHERE seq >= ? ORDER BY seq LIMIT ?", + ) + .bind(i64::from(start_seq)) + .bind(i64::try_from(limit.saturating_add(1)).unwrap_or(i64::MAX)) + .fetch_all(&self.pool) + .await?; + rows.into_iter() + .map(|(seq, json)| decode_event(seq, &json)) + .collect() + } +} + +fn decode_event(seq: i64, json: &str) -> Result { + Ok(MainEventEnvelope { + seq: u32::try_from(seq) + .map_err(|_| Error::EventSequenceExhausted { max_seq: u32::MAX })?, + event: serde_json::from_str(json)?, + }) +} + +#[cfg(test)] +mod tests { + use fabro_types::{MainEventLifecycleProps, MainEventSource, fixtures, test_support}; + use tokio::sync::broadcast::error::TryRecvError; + + use super::*; + use crate::platform_records::{PlatformRecord, RunCreatedRecord}; + use crate::test_support as store_test_support; + + fn body(run_id: RunId) -> MainEventBody { + MainEventBody::RunRunning(MainEventLifecycleProps { + source: MainEventSource { + run_id, + source_event_id: "platform:1".into(), + source_event_ts: Utc::now(), + source_event_name: "run.running".into(), + actor: None, + }, + }) + } + + fn source(run_id: RunId) -> MainEventSourceKey { + MainEventSourceKey::RunStream { + run_id, + item_id: "platform:1".into(), + } + } + + #[tokio::test] + async fn main_events_replay_and_broadcast_in_global_order() { + let database = store_test_support::test_database(); + let store = database.main_events(); + let mut live = store.subscribe(); + let first = store.append(body(fixtures::RUN_1)).await.unwrap(); + let second = store.append(body(fixtures::RUN_2)).await.unwrap(); + let third = store.append(body(fixtures::RUN_1)).await.unwrap(); + assert_eq!((first.seq, second.seq, third.seq), (1, 2, 3)); + assert_ne!(first.event.id, second.event.id); + assert_eq!(live.recv().await.unwrap(), first); + assert_eq!(live.recv().await.unwrap(), second); + assert_eq!(live.recv().await.unwrap(), third); + assert_eq!(store.list_from_with_limit(2, 1).await.unwrap(), vec![ + second, third + ]); + assert!(store.list_from_with_limit(4, 10).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn main_event_source_survives_reopen_and_is_scoped_to_run() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("fabro.sqlite3"); + let database = fabro_db::Database::connect(&path).await.unwrap(); + database.migrate().await.unwrap(); + let store = MainEventStore::new(database.clone_pool()); + let AppendMainEventOutcome::Appended(first) = store + .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)) + .await + .unwrap() + else { + panic!("first append"); + }; + database.pool().close().await; + drop(store); + drop(database); + + let database = fabro_db::Database::connect(&path).await.unwrap(); + database.migrate().await.unwrap(); + let reopened = MainEventStore::new(database.clone_pool()); + let mut live = reopened.subscribe(); + let duplicate = reopened + .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)) + .await + .unwrap(); + assert_eq!(duplicate, AppendMainEventOutcome::Duplicate(first.clone())); + assert_eq!(live.try_recv(), Err(TryRecvError::Empty)); + let AppendMainEventOutcome::Appended(second) = reopened + .append_once(source(fixtures::RUN_2), body(fixtures::RUN_2)) + .await + .unwrap() + else { + panic!("a different run has a different source key"); + }; + assert_eq!(second.seq, 2); + assert_eq!(reopened.list_from_with_limit(1, 10).await.unwrap(), vec![ + first, second + ]); + } + + #[tokio::test] + async fn concurrent_main_event_sources_consume_one_sequence_across_handles() { + let directory = tempfile::tempdir().unwrap(); + let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3")) + .await + .unwrap(); + database.migrate().await.unwrap(); + let left = MainEventStore::new(database.clone_pool()); + let right = MainEventStore::new(database.clone_pool()); + let (first, second) = tokio::join!( + left.append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)), + right.append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)), + ); + assert_ne!( + matches!(first.unwrap(), AppendMainEventOutcome::Appended(_)), + matches!(second.unwrap(), AppendMainEventOutcome::Appended(_)) + ); + assert_eq!(left.list_from_with_limit(1, 10).await.unwrap().len(), 1); + assert_eq!(right.append(body(fixtures::RUN_2)).await.unwrap().seq, 2); + } + + #[tokio::test] + async fn failed_main_event_append_leaves_no_receipt_or_broadcast() { + let database = store_test_support::test_database(); + let store = database.main_events(); + let pool = database.run_summary_store().pool(); + // Connect the lazy fixture and then inject a database write failure. + store.list_from_with_limit(1, 1).await.unwrap(); + sqlx::raw_sql("CREATE TRIGGER fail_main_event BEFORE INSERT ON main_events BEGIN SELECT RAISE(ABORT, 'injected failure'); END;") + .execute(&pool).await.unwrap(); + let mut live = store.subscribe(); + assert!( + store + .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)) + .await + .is_err() + ); + assert_eq!(live.try_recv(), Err(TryRecvError::Empty)); + sqlx::query("DROP TRIGGER fail_main_event") + .execute(&pool) + .await + .unwrap(); + let AppendMainEventOutcome::Appended(event) = store + .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)) + .await + .unwrap() + else { + panic!("retry must append"); + }; + assert_eq!(event.seq, 1); + assert_eq!(live.recv().await.unwrap(), event); + } + + #[tokio::test] + async fn main_event_sequence_exhaustion_preserves_existing_events() { + let database = store_test_support::test_database(); + let store = database.main_events(); + let first = store.append(body(fixtures::RUN_1)).await.unwrap(); + sqlx::query("UPDATE main_events SET seq = ? WHERE seq = 1") + .bind(i64::from(u32::MAX)) + .execute(&database.run_summary_store().pool()) + .await + .unwrap(); + assert!(matches!( + store.append(body(fixtures::RUN_2)).await, + Err(Error::EventSequenceExhausted { .. }) + )); + let events = store.list_from_with_limit(u32::MAX, 1).await.unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].event, first.event); + } + + #[tokio::test] + async fn deleting_a_run_keeps_its_main_events_and_source_receipts() { + let database = store_test_support::test_database(); + let spec = test_support::test_run_spec(); + let record = PlatformRecord::RunCreated(RunCreatedRecord { + spec, + title: None, + parent_id: None, + retried_from: None, + web_url: None, + }); + database + .run_summary_store() + .platform_records() + .append(&fixtures::RUN_1, &record, None) + .await + .unwrap(); + let events = database.main_events(); + let first = events + .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)) + .await + .unwrap(); + database.delete_run(&fixtures::RUN_1).await.unwrap(); + let duplicate = events + .append_once(source(fixtures::RUN_1), body(fixtures::RUN_1)) + .await + .unwrap(); + let AppendMainEventOutcome::Appended(first) = first else { + panic!("first append"); + }; + assert_eq!(duplicate, AppendMainEventOutcome::Duplicate(first.clone())); + assert_eq!(events.list_from_with_limit(1, 10).await.unwrap(), vec![ + first + ]); + } +} diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index 17d35f941..46a4a45eb 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -27,6 +27,7 @@ const RUN_SUMMARY_MIGRATIONS: &[&str] = &[ fabro_db::RUNS_MIGRATION_SQL, fabro_db::DROP_RUN_EVENTS_MIGRATION_SQL, fabro_db::PETRI_PROJECTION_MIGRATION_SQL, + fabro_db::MAIN_EVENTS_MIGRATION_SQL, ]; /// Returns an isolated SQLite run-summary store backed by its own in-memory diff --git a/lib/foundation/fabro-db/migrations/2026101601_main_events.sql b/lib/foundation/fabro-db/migrations/2026101601_main_events.sql new file mode 100644 index 000000000..a42732350 --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026101601_main_events.sql @@ -0,0 +1,7 @@ +-- The global lifecycle mirror outlives individual runs. Deliberately no +-- foreign key to runs: deleting a run must keep its events and source keys. +CREATE TABLE main_events ( + seq INTEGER PRIMARY KEY CHECK (seq > 0 AND seq <= 4294967295), + source_key TEXT UNIQUE, + event_json TEXT NOT NULL +); diff --git a/lib/foundation/fabro-db/src/lib.rs b/lib/foundation/fabro-db/src/lib.rs index 12387705c..a5ef597ec 100644 --- a/lib/foundation/fabro-db/src/lib.rs +++ b/lib/foundation/fabro-db/src/lib.rs @@ -42,6 +42,10 @@ pub const RUN_SESSION_RECORDS_MIGRATION_SQL: &str = pub const RUN_SESSION_EVENTS_MIGRATION_SQL: &str = include_str!("../migrations/2026091802_run_session_events.sql"); +/// The global lifecycle stream schema, also used by isolated store fixtures. +pub const MAIN_EVENTS_MIGRATION_SQL: &str = + include_str!("../migrations/2026101601_main_events.sql"); + /// The Petri run record migration (`petri_runs`, `petri_records`), exposed /// so fixtures in other crates can install the production schema without a /// filesystem path into this crate. From 8f4f127b20f539458e81993350d0fb4b7f9384e6 Mon Sep 17 00:00:00 2001 From: fkukuck Date: Fri, 16 Oct 2026 20:07:10 +0000 Subject: [PATCH 3/3] feat: mirror Petri run lifecycle into the main stream --- docs/internal/events-strategy.md | 8 + .../2026-06-21-main-event-stream-design.md | 126 ++++++ lib/apps/fabro-server/src/server.rs | 2 + .../src/server/handler/lineage.rs | 17 +- .../fabro-server/src/server/handler/runs.rs | 11 +- .../src/server/main_event_mirror.rs | 336 ++++++++++++++++ lib/apps/fabro-server/src/server/tests.rs | 165 ++++++++ lib/components/fabro-petri/src/lib.rs | 1 + lib/components/fabro-petri/src/main_events.rs | 360 ++++++++++++++++++ 9 files changed, 1023 insertions(+), 3 deletions(-) create mode 100644 docs/superpowers/specs/2026-06-21-main-event-stream-design.md create mode 100644 lib/apps/fabro-server/src/server/main_event_mirror.rs create mode 100644 lib/components/fabro-petri/src/main_events.rs diff --git a/docs/internal/events-strategy.md b/docs/internal/events-strategy.md index f1e5440c7..8101f6b2e 100644 --- a/docs/internal/events-strategy.md +++ b/docs/internal/events-strategy.md @@ -32,6 +32,14 @@ Tracing logs are separate. Tracing is developer diagnostics; the stream is the product-facing record other systems consume. If something must be visible after a reattach, it has to be a record, not a log line. +The server also mirrors selected lifecycle records into an independent +global **main event stream** (`fabro-store::MainEventStore`). Its compact +`fabro.run.*` events have their own sequence and survive run deletion. +Source-keyed appends deduplicate replay of a run's canonical stream; the +main stream is a cross-run mirror, not another source for run projections. +See [the main event design](../superpowers/specs/2026-06-21-main-event-stream-design.md) +for its event vocabulary, failure policy, and recovery boundary. + ## Recording a fact Petri's own facts need nothing from Fabro: the engine records them and the 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 000000000..71057f641 --- /dev/null +++ b/docs/superpowers/specs/2026-06-21-main-event-stream-design.md @@ -0,0 +1,126 @@ +# Durable main event stream + +Updated 2026-10-16 for upstream's Petri run stream and SQLite storage. + +## Purpose + +A run's canonical history is Petri's execution records and Fabro's platform +records, delivered together as `RunStreamItem`. This design adds an independent, +compact server-wide lifecycle stream for future cross-run consumers. It does +not replace the run stream, its projection, APIs, attach broadcast, or CLI. + +The implementation follows the lifecycle mirror maintained in the ipt fork, +using upstream's current SQLite store in place of the fork's SlateDB store. + +## Event contract + +`fabro-types::MainEventEnvelope` contains a global `seq` and a `MainEvent`. +The main event has its own UUID v7 `id`, append timestamp `ts`, event name, +and compact `properties`. The eight lifecycle names are: + +- `fabro.run.created` +- `fabro.run.started` +- `fabro.run.running` +- `fabro.run.completed` +- `fabro.run.failed` +- `fabro.run.cancelled` +- `fabro.run.paused` +- `fabro.run.unpaused` + +Each lifecycle payload preserves the run ID, source item ID, source timestamp, +and source event name. Source IDs include the stream kind, for example +`platform:1` or `petri:coordinator/3/0`, so the two logs cannot collide. +Native records do not carry a request actor; the optional actor stays absent. +Creation preserves the initiating principal from the immutable run spec. + +Creation includes the original title, workflow slug, automation reference, +Git context, and parent ID. Start includes workflow identity, goal, and Git +baseline/branch metadata when available. Terminal events include the outcome, +final Git commit, automation reference, and recorded aggregate Lithos usage +when available. Completion also retains the recorded cost in USD micros; +missing cost remains absent. Mirroring does not price usage again. + +Unknown event names and their arbitrary JSON properties round-trip unchanged. +Malformed properties for a known event are rejected rather than silently +reinterpreted as an unknown event. + +## Persistence + +`fabro-store::MainEventStore` uses the existing run-summary SQLite pool. The +`main_events` table has: + +- `seq`: a positive global sequence, bounded to `u32`; +- `source_key`: an optional unique key for idempotent source delivery; +- `event_json`: the typed main event serialized as JSON. + +There is no foreign key to a run. Run deletion removes neither these events +nor their deduplication identities. A `BEGIN IMMEDIATE` transaction serializes +writers, checks for an existing source, and commits the next sequence and +payload together. Retrying a source returns the original envelope, consumes +no sequence, and emits no new live message. Failed writes publish nothing. +The next sequence comes from the committed table, including after restart. +Sequence exhaustion returns an error without overwriting existing events. + +The store supports ordinary append, source-keyed append, inclusive replay +from a global sequence, and in-process subscription. Replay queries return +at most `limit + 1` rows in sequence order. Subscribers receive committed +appends; a slow subscriber must replay after a broadcast lag. Broadcasts are +local to the shared store handle, not a cross-process delivery mechanism. + +Storage owns persistence and deduplication. It does not decide which run +records belong in the lifecycle stream. + +## Native lifecycle mapping + +`fabro-petri::main_events` maps native stream items into compact payloads: + +- The platform `run.created` record supplies creation from its immutable + record, including the original title rather than a later renamed title. +- Petri's `run.started` supplies start. +- Platform `run.lifecycle` transitions supply running, paused, unpaused, + succeeded, failed, and dead outcomes. +- A terminal failure whose reason is `Cancelled` becomes + `fabro.run.cancelled`. A cancellation request alone is not terminal. + +Petri's `run.finished` is not mirrored independently: Fabro's terminal +lifecycle record reflects the coordinator result after required finalization. +Stage, agent, tool, transcript, sandbox, graph, settings, patch, and output +records remain exclusively in the canonical run history. + +## Server delivery and recovery + +The server mirrors creation synchronously after persistence and projection, +before returning success from create, retry, fork, or rewind. A failed main +append fails that request; the persisted run is retained and background +reconciliation can repair the missing delivery. The synchronous path and +background path use the same source identity. + +A separate server task follows projector commit signals and scans all retained +runs at startup and every 30 seconds. For each run it reads stream pages and +loads the committed projection after the page, so terminal usage is visible. +It advances its in-memory cursor only after an item has been processed +successfully. Mapping or storage errors are logged with run identity and +retried; they do not interrupt execution or the existing live run broadcast. + +After restart, cursors begin at zero. The canonical run stream acts as the +outbox, and source-keyed appends make replay idempotent. A lost or lagged signal +costs delivery latency; the periodic scan repairs it. Cursors are an +optimization, never evidence that an uncommitted append succeeded. + +The main sequence describes append order. Recovery may interleave historical +runs differently from their original wall-clock order; source timestamps +preserve the time of the underlying records. Events already mirrored survive +run deletion, but a run deleted before reconciliation cannot supply missing +records afterward. This is not an exactly-once trigger-delivery contract. + +## Scope and verification + +This change adds no public main-stream API, CLI command, SSE endpoint, webhook +ingestion, automation listener, persistent consumer cursor, or delivery receipt +for external consumers. + +Tests cover all lifecycle names and JSON round trips, unknown events and +malformed known events, native record filtering, cancellation, terminal usage, +global replay and subscription order, concurrent deduplication, restart +recovery, failed writes, sequence exhaustion, deletion survival, multi-page +reconciliation, and startup reconciliation without a new signal. diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 413c8c608..1baa433c7 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -164,6 +164,7 @@ use crate::{ mod automation_scheduler; mod handler; +mod main_event_mirror; pub(crate) mod petri_runs; mod pull_request_supervisor; pub(crate) mod resource_sampler; @@ -4471,6 +4472,7 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { /// Background task that promotes runnable runs when capacity is available. pub fn spawn_scheduler(state: Arc) { stream_follower::spawn_stream_follower(Arc::clone(&state)); + main_event_mirror::spawn_native_mirror(Arc::clone(&state)); tokio::spawn(async move { loop { tokio::select! { diff --git a/lib/apps/fabro-server/src/server/handler/lineage.rs b/lib/apps/fabro-server/src/server/handler/lineage.rs index 2982ba795..a4fe319e0 100644 --- a/lib/apps/fabro-server/src/server/handler/lineage.rs +++ b/lib/apps/fabro-server/src/server/handler/lineage.rs @@ -34,7 +34,8 @@ use fabro_workflow::operations::{self, ForkTarget, ResolvedForkTarget, RunTimeli use tracing::{error, warn}; use super::super::{ - ApiError, AppState, RequireRunManagementTarget, RequiredUser, parse_run_id_path, run_records, + ApiError, AppState, RequireRunManagementTarget, RequiredUser, main_event_mirror, + parse_run_id_path, run_records, }; use super::lifecycle::{ArchiveAction, queue_run, run_archive_operation, run_response}; use super::runs::run_provenance; @@ -179,6 +180,7 @@ async fn retry_run( )) .await .map_err(workflow_operation_error)?; + mirror_created(state.as_ref(), new_run_id).await?; let projection = run_records::require_projection(state.as_ref(), new_run_id).await?; queue_run(state.as_ref(), new_run_id, &projection, false, actor).await?; Ok::<_, ApiError>(new_run_id) @@ -308,6 +310,7 @@ async fn fork_at( .await .map_err(workflow_operation_error)?; + mirror_created(state, new_run_id).await?; let seeded = petri_fork::fork(ForkRequest { source: id, fork: new_run_id, @@ -350,6 +353,18 @@ async fn fork_at( }) } +async fn mirror_created(state: &AppState, run_id: RunId) -> Result<(), ApiError> { + main_event_mirror::mirror_created(state, run_id) + .await + .map_err(|error| { + error!(%run_id, error = %error, "Failed to mirror persisted run creation"); + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to persist run lifecycle event", + ) + }) +} + /// A refused position is the caller's mistake; anything else is the /// server's. fn fork_error(err: &ForkError) -> ApiError { diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs index 696c8af1e..65e653b66 100644 --- a/lib/apps/fabro-server/src/server/handler/runs.rs +++ b/lib/apps/fabro-server/src/server/handler/runs.rs @@ -47,7 +47,7 @@ use tracing::info; use super::super::{ AppState, DeleteRunOutcome, ListResponse, RunExecutionMode, VariableError, answer_from_request, api_question_from_pending_interview, clamp_page_limit, clamp_page_offset, default_page_limit, - delete_run_internal, load_pending_interview, managed_run, parse_run_id_path, + delete_run_internal, load_pending_interview, main_event_mirror, managed_run, parse_run_id_path, parse_stage_id_path, petri_runs, reject_if_archived, run_records, submit_pending_interview_answer, }; @@ -778,7 +778,14 @@ async fn finalize_created_run( let created_at = created.run_id.created_at(); // The run's summary row is the projector's: wait for the pass that // folds the run's first records before reading the run back. - state.petri_projector.settle(created.run_id).await; + if let Err(error) = main_event_mirror::mirror_created(&state, created.run_id).await { + tracing::error!(run_id = %created.run_id, error = %error, "Failed to mirror persisted run creation"); + return intent_error( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to persist run lifecycle event", + "run_persistence_failed", + ); + } let summary = match state .stores .run_summaries diff --git a/lib/apps/fabro-server/src/server/main_event_mirror.rs b/lib/apps/fabro-server/src/server/main_event_mirror.rs new file mode 100644 index 000000000..0504847fa --- /dev/null +++ b/lib/apps/fabro-server/src/server/main_event_mirror.rs @@ -0,0 +1,336 @@ +//! Replay selected lifecycle records into the independent global stream. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Context as _; +use fabro_petri::main_events; +use fabro_store::MainEventSourceKey; +use fabro_types::{RunId, RunProjection, RunStreamItem, RunStreamItemKind}; +use tokio::sync::broadcast::error::RecvError; +use tokio::time; +use tracing::warn; + +use super::AppState; + +const BATCH_LIMIT: usize = 256; + +/// Mirror a native stream item with a stable source key. The SQL stream is the +/// outbox: replay after a crash retries the append without duplication. +async fn mirror_native_item( + state: &AppState, + item: &RunStreamItem, + projection: &RunProjection, +) -> anyhow::Result<()> { + let Some(body) = main_events::from_stream_item(item, projection)? else { + return Ok(()); + }; + state + .stores + .runs + .main_events() + .append_once( + MainEventSourceKey::RunStream { + run_id: item.run_id, + item_id: format!("{}:{}", item.kind, item.id), + }, + body, + ) + .await?; + Ok(()) +} + +/// Reconcile all native streams on startup and periodically. Signals shorten +/// delivery latency; the scan also repairs gaps when a signal is lost or a +/// process exits between the run record commit and the main event append. +pub(crate) fn spawn_native_mirror(state: Arc) { + tokio::spawn(async move { + let mut cursors = HashMap::new(); + let mut signals = state.petri_projector.subscribe(); + let mut tick = time::interval(Duration::from_secs(30)); + let shutdown = state.shutdown_token(); + loop { + tokio::select! { + () = shutdown.cancelled() => break, + _ = tick.tick() => { + match state.stores.run_summaries.list_identities().await { + Ok(runs) => for run in runs { + let cursor = cursors.entry(run.id).or_insert(0); + if let Err(error) = reconcile_native_run(&state, run.id, cursor).await { + warn!(run_id = %run.id, error = %error, "Native main event mirror will retry"); + } + }, + Err(error) => warn!(error = %error, "Main event mirror could not list runs; will retry"), + } + } + signal = signals.recv() => match signal { + Ok(run_id) => { + let cursor = cursors.entry(run_id).or_insert(0); + if let Err(error) = reconcile_native_run(&state, run_id, cursor).await { + warn!(%run_id, error = %error, "Native main event mirror will retry"); + } + } + Err(RecvError::Closed) => break, + Err(RecvError::Lagged(_)) => {}, + }, + } + } + }); +} + +async fn reconcile_native_run( + state: &AppState, + run_id: RunId, + cursor: &mut u64, +) -> anyhow::Result<()> { + loop { + let items = state + .petri_projector + .stream_after(run_id, *cursor, BATCH_LIMIT) + .await?; + if items.is_empty() { + return Ok(()); + } + // Read after the stream page: the projection transaction that published + // these items must be visible before terminal usage is mirrored. + let Some(projection) = state + .stores + .run_summaries + .load_petri_projection(&run_id) + .await? + else { + return Ok(()); + }; + let drained = items.len() < BATCH_LIMIT; + for item in items { + mirror_native_item(state, &item, &projection).await?; + *cursor = item.stream_seq; + } + if drained { + return Ok(()); + } + } +} + +/// Creation is mirrored before a successful create response. The background +/// mirror uses the same source key, so either delivery can win safely. +pub(crate) async fn mirror_created(state: &AppState, run_id: RunId) -> anyhow::Result<()> { + state.petri_projector.settle(run_id).await; + let items = state.petri_projector.stream_after(run_id, 0, 1).await?; + let created = items + .first() + .filter(|item| { + item.kind == RunStreamItemKind::Platform && item.name() == Some("run.created") + }) + .context("created run has no initial run.created stream item")?; + let projection = state + .stores + .run_summaries + .load_petri_projection(&run_id) + .await? + .context("created run has no projection")?; + mirror_native_item(state, created, &projection).await +} + +#[cfg(test)] +mod tests { + use fabro_store::platform_records::{ + PlatformRecord, RunCreatedRecord, RunLifecycleKind, RunLifecycleRecord, + }; + use fabro_types::{FailureReason, MainEventBody, RunStatus, test_support}; + use tokio::sync::broadcast::error::TryRecvError; + + use super::*; + use crate::server::run_records; + + async fn create(state: &AppState) -> RunId { + let mut spec = test_support::test_run_spec(); + spec.run_id = RunId::new(); + let run_id = spec.run_id; + run_records::append( + state, + run_id, + PlatformRecord::RunCreated(RunCreatedRecord { + spec, + title: Some("Original title".into()), + parent_id: None, + retried_from: None, + web_url: None, + }), + ) + .await + .unwrap(); + run_id + } + + #[tokio::test] + async fn main_event_mirror_replays_lifecycle_once_and_keeps_events_after_deletion() { + let state = crate::test_support::test_app_state(); + let run_id = create(&state).await; + mirror_created(&state, run_id).await.unwrap(); + let events = state.stores.runs.main_events(); + let created = events.list_from_with_limit(1, 10).await.unwrap(); + assert_eq!(created.len(), 1); + let MainEventBody::RunCreated(created) = &created[0].event.body else { + panic!("created"); + }; + assert_eq!(created.title.as_deref(), Some("Original title")); + + for transition in [ + RunLifecycleKind::Running, + RunLifecycleKind::PauseRequested, + RunLifecycleKind::Paused, + RunLifecycleKind::Unpaused, + RunLifecycleKind::CancelRequested, + ] { + run_records::lifecycle(&state, run_id, RunLifecycleRecord::new(transition)) + .await + .unwrap(); + } + run_records::lifecycle( + &state, + run_id, + run_records::failed(FailureReason::Cancelled, "Stopped by user"), + ) + .await + .unwrap(); + let mut cursor = 0; + reconcile_native_run(&state, run_id, &mut cursor) + .await + .unwrap(); + let mirrored = events.list_from_with_limit(1, 20).await.unwrap(); + assert_eq!( + mirrored + .iter() + .map(|event| event.event.event_name()) + .collect::>(), + [ + "fabro.run.created", + "fabro.run.running", + "fabro.run.paused", + "fabro.run.unpaused", + "fabro.run.cancelled" + ] + ); + let MainEventBody::RunCancelled(cancelled) = &mirrored[4].event.body else { + panic!("cancelled"); + }; + assert_eq!(cancelled.message, "Stopped by user"); + let head = state + .petri_projector + .stream_head(run_id) + .await + .unwrap() + .unwrap(); + assert_eq!(cursor, head); + + let mut live = events.subscribe(); + // A new process starts with no in-memory cursor, then replays its outbox. + reconcile_native_run(&state, run_id, &mut 0).await.unwrap(); + assert_eq!(live.try_recv(), Err(TryRecvError::Empty)); + assert_eq!(events.list_from_with_limit(1, 20).await.unwrap(), mirrored); + state.petri_projector.delete_run(run_id).await.unwrap(); + assert_eq!(events.list_from_with_limit(1, 20).await.unwrap(), mirrored); + } + + #[tokio::test] + async fn main_event_mirror_retries_failed_delivery_without_advancing_cursor() { + let state = crate::test_support::test_app_state(); + let run_id = create(&state).await; + let pool = state.stores.run_summaries.pool(); + sqlx::raw_sql("CREATE TRIGGER fail_main_event BEFORE INSERT ON main_events BEGIN SELECT RAISE(ABORT, 'injected failure'); END;") + .execute(&pool).await.unwrap(); + assert!(mirror_created(&state, run_id).await.is_err()); + let mut cursor = 0; + assert!( + reconcile_native_run(&state, run_id, &mut cursor) + .await + .is_err() + ); + assert_eq!(cursor, 0); + sqlx::query("DROP TRIGGER fail_main_event") + .execute(&pool) + .await + .unwrap(); + reconcile_native_run(&state, run_id, &mut cursor) + .await + .unwrap(); + assert!(cursor > 0); + mirror_created(&state, run_id).await.unwrap(); + let events = state + .stores + .runs + .main_events() + .list_from_with_limit(1, 10) + .await + .unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].seq, 1); + } + + #[tokio::test] + async fn main_event_mirror_drains_multiple_pages_without_changing_the_run_stream() { + let state = crate::test_support::test_app_state(); + let run_id = create(&state).await; + let records = state.stores.run_summaries.platform_records(); + for _ in 0..BATCH_LIMIT { + records + .append( + &run_id, + &PlatformRecord::RunLifecycle( + RunLifecycleRecord::new(RunLifecycleKind::Running) + .with_status(RunStatus::Running), + ), + None, + ) + .await + .unwrap(); + } + state.stores.run_summaries.notify_platform_record(run_id); + state.petri_projector.settle(run_id).await; + let original = state + .petri_projector + .stream_after(run_id, 0, BATCH_LIMIT * 2) + .await + .unwrap(); + let mut cursor = 0; + reconcile_native_run(&state, run_id, &mut cursor) + .await + .unwrap(); + assert_eq!(cursor, original.last().unwrap().stream_seq); + assert_eq!( + state + .petri_projector + .stream_after(run_id, 0, BATCH_LIMIT * 2) + .await + .unwrap(), + original + ); + let events = state + .stores + .runs + .main_events() + .list_from_with_limit(1, BATCH_LIMIT * 2) + .await + .unwrap(); + assert_eq!(events.len(), BATCH_LIMIT + 1); + } + + #[tokio::test] + async fn main_event_mirror_startup_reconciles_without_a_new_signal() { + let state = crate::test_support::test_app_state(); + let run_id = create(&state).await; + let events = state.stores.runs.main_events(); + let mut live = events.subscribe(); + spawn_native_mirror(Arc::clone(&state)); + let event = time::timeout(Duration::from_secs(5), live.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!(event.event.body, MainEventBody::RunCreated(_))); + mirror_created(&state, run_id).await.unwrap(); + assert_eq!(events.list_from_with_limit(1, 10).await.unwrap().len(), 1); + state.shutdown_token().cancel(); + } +} diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index a1f978590..24d3cc894 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -3525,6 +3525,171 @@ file = "goal.md" ); } +#[tokio::test] +async fn main_event_mirror_follows_an_executed_petri_workflow() { + let workspace = tempfile::tempdir().unwrap(); + let state = TestAppStateBuilder::new() + .default_environment_provider(Some(SandboxProviderKind::LOCAL)) + .vault_entries([(fabro_static::EnvVars::OPENAI_API_KEY, "test-openai-api-key")]) + .in_process_execution() + .build(); + let app = test_app_with_scheduler(Arc::clone(&state)); + let version = store_workflow_version( + &state, + r#"digraph Command { + graph [goal="Run one command"] + start [shape=Mdiamond] + exit [shape=Msquare] + say [shape=parallelogram, script="echo hello from petri"] + start -> say -> exit + }"#, + None, + ) + .await; + let events = state.stores.runs.main_events(); + let mut live = events.subscribe(); + let created = post_run_intent(&app, folder_intent(version, workspace.path())).await; + let run_id: RunId = created["id"].as_str().unwrap().parse().unwrap(); + let request = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/start"))) + .body(Body::empty()) + .unwrap(); + let response = app.oneshot(request).await.unwrap(); + response_json!(response, StatusCode::OK).await; + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let event = live.recv().await.unwrap(); + if matches!( + event.event.body, + fabro_types::MainEventBody::RunCompleted(_) + ) { + break; + } + assert!(!matches!( + event.event.body, + fabro_types::MainEventBody::RunFailed(_) + )); + } + }) + .await + .expect("workflow lifecycle should reach the main stream"); + let mirrored = events.list_from_with_limit(1, 20).await.unwrap(); + let mut names = mirrored + .iter() + .map(|event| event.event.event_name()) + .collect::>(); + names.sort_unstable(); + assert_eq!(names, [ + "fabro.run.completed", + "fabro.run.created", + "fabro.run.running", + "fabro.run.started" + ]); + let stream = state + .petri_projector + .stream_after(run_id, 0, 100) + .await + .unwrap(); + for event in mirrored { + let properties = event.event.properties().unwrap(); + assert_eq!(properties["run_id"], run_id.to_string()); + assert!( + stream + .iter() + .any(|item| properties["source_event_id"] == format!("{}:{}", item.kind, item.id)) + ); + } + state.shutdown_token().cancel(); +} + +#[tokio::test] +async fn main_event_mirror_create_is_persisted_before_the_api_returns() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await; + let body = post_run_intent( + &app, + json!({ + "workflow_version_id": workflow_version_id, + "target": { "kind": "none" }, + "args": {} + }), + ) + .await; + let run_id: RunId = body["id"].as_str().unwrap().parse().unwrap(); + let events = state + .stores + .runs + .main_events() + .list_from_with_limit(1, 10) + .await + .unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].seq, 1); + let fabro_types::MainEventBody::RunCreated(created) = &events[0].event.body else { + panic!("expected run creation"); + }; + assert_eq!(created.source.run_id, run_id); + assert_eq!(created.source.source_event_id, "platform:1"); + let projection = state.load_run_projection(&run_id).await.unwrap(); + assert_eq!( + created.provenance_subject.as_ref(), + Some(&projection.spec.provenance.subject) + ); +} + +#[tokio::test] +async fn main_event_mirror_create_failure_retains_the_run_for_reconciliation() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await; + let pool = state.stores.run_summaries.pool(); + sqlx::raw_sql("CREATE TRIGGER fail_main_event BEFORE INSERT ON main_events BEGIN SELECT RAISE(ABORT, 'injected failure'); END;") + .execute(&pool).await.unwrap(); + let response = post_run_intent_response( + &app, + json!({ + "workflow_version_id": workflow_version_id, + "target": { "kind": "none" }, + "args": {} + }), + ) + .await; + let body = response_json!(response, StatusCode::INTERNAL_SERVER_ERROR).await; + assert_eq!(body["errors"][0]["code"], "run_persistence_failed"); + let runs = state.stores.run_summaries.list_identities().await.unwrap(); + assert_eq!(runs.len(), 1); + assert!( + state + .stores + .runs + .main_events() + .list_from_with_limit(1, 10) + .await + .unwrap() + .is_empty() + ); + sqlx::query("DROP TRIGGER fail_main_event") + .execute(&pool) + .await + .unwrap(); + main_event_mirror::mirror_created(&state, runs[0].id) + .await + .unwrap(); + assert_eq!( + state + .stores + .runs + .main_events() + .list_from_with_limit(1, 10) + .await + .unwrap() + .len(), + 1 + ); +} + #[tokio::test] async fn post_runs_run_intent_creates_submitted_none_target_without_git_projection() { let state = test_app_state(); diff --git a/lib/components/fabro-petri/src/lib.rs b/lib/components/fabro-petri/src/lib.rs index 15ee10d9a..ef7605cfe 100644 --- a/lib/components/fabro-petri/src/lib.rs +++ b/lib/components/fabro-petri/src/lib.rs @@ -74,6 +74,7 @@ pub mod hooks; pub mod host_tools; pub mod http_store; pub mod interview; +pub mod main_events; pub mod petri; pub mod platform_records; pub mod projection; diff --git a/lib/components/fabro-petri/src/main_events.rs b/lib/components/fabro-petri/src/main_events.rs new file mode 100644 index 000000000..fa4cab9ea --- /dev/null +++ b/lib/components/fabro-petri/src/main_events.rs @@ -0,0 +1,360 @@ +//! Native run records projected into Fabro's compact main lifecycle vocabulary. +//! Storage and idempotent delivery remain the server's main event store +//! boundary. +use chrono::DateTime; +use fabro_types::{ + FailureReason, MainEventBody, MainEventCancelledProps, MainEventCompletedProps, + MainEventCreatedProps, MainEventFailedProps, MainEventLifecycleProps, MainEventSource, + MainEventStartedProps, RunProjection, RunStreamItem, +}; + +pub fn from_stream_item( + item: &RunStreamItem, + projection: &RunProjection, +) -> anyhow::Result> { + use fabro_store::platform_records::{PlatformRecord, RunLifecycleKind, StoredPlatformRecord}; + use fabro_types::{FailureCategory, RunStatus, RunStreamItemKind}; + let timestamp = i64::try_from(item.recorded_at) + .ok() + .and_then(DateTime::from_timestamp_millis) + .ok_or_else(|| { + anyhow::anyhow!( + "invalid timestamp in run {} stream item {}", + item.run_id, + item.id + ) + })?; + let source = |name: &str| MainEventSource { + run_id: item.run_id, + source_event_id: format!("{}:{}", item.kind, item.id), + source_event_ts: timestamp, + source_event_name: name.to_owned(), + actor: None, + }; + if item.kind == RunStreamItemKind::Petri && item.name() == Some("run.started") { + return Ok(Some(MainEventBody::RunStarted(MainEventStartedProps { + source: source("run.started"), + name: projection.spec.graph.name.clone(), + base_branch: projection.spec.git.as_ref().map(|git| git.branch.clone()), + base_sha: projection + .start + .as_ref() + .and_then(|start| start.base_sha.clone()), + run_branch: projection + .start + .as_ref() + .and_then(|start| start.run_branch.clone()), + worktree_dir: None, + goal: Some(projection.spec.graph.goal.clone()).filter(|goal| !goal.is_empty()), + }))); + } + if item.kind != RunStreamItemKind::Platform { + return Ok(None); + } + let stored: StoredPlatformRecord = serde_json::from_value(item.item.clone())?; + Ok(match stored.record { + PlatformRecord::RunCreated(created) => { + Some(MainEventBody::RunCreated(MainEventCreatedProps { + source: source("run.created"), + title: created.title, + workflow_slug: created.spec.workflow_slug, + automation: created.spec.automation, + git: created.spec.git, + parent_id: created.parent_id, + provenance_subject: Some(created.spec.provenance.subject), + })) + } + PlatformRecord::RunLifecycle(record) => { + let conclusion = projection.conclusion.as_ref(); + let usage = conclusion.and_then(|conclusion| conclusion.usage); + let final_git_commit_sha = + conclusion.and_then(|conclusion| conclusion.final_git_commit_sha.clone()); + match record.transition { + RunLifecycleKind::Running => { + Some(MainEventBody::RunRunning(MainEventLifecycleProps { + source: source("run.running"), + })) + } + RunLifecycleKind::Paused => { + Some(MainEventBody::RunPaused(MainEventLifecycleProps { + source: source("run.paused"), + })) + } + RunLifecycleKind::Unpaused => { + Some(MainEventBody::RunUnpaused(MainEventLifecycleProps { + source: source("run.unpaused"), + })) + } + RunLifecycleKind::Succeeded => { + let Some(RunStatus::Succeeded { reason }) = record.status else { + anyhow::bail!("successful lifecycle record has no success reason"); + }; + Some(MainEventBody::RunCompleted(MainEventCompletedProps { + source: source("run.completed"), + status: "succeeded".into(), + reason, + total_usd_micros: usage + .and_then(|usage| usage.cost) + .map(|cost| i64::try_from(cost.usd_micros).unwrap_or(i64::MAX)), + final_git_commit_sha, + automation: projection.spec.automation.clone(), + usage, + })) + } + RunLifecycleKind::Failed | RunLifecycleKind::Dead => { + let failure = conclusion.and_then(|conclusion| conclusion.failure.as_ref()); + let reason = match record.status { + Some(RunStatus::Failed { reason }) => reason, + _ => failure.map_or(FailureReason::WorkflowError, |failure| failure.reason), + }; + let category = failure.map_or( + if reason == FailureReason::Cancelled { + FailureCategory::Canceled + } else { + FailureCategory::Deterministic + }, + |failure| failure.detail.category, + ); + let message = record + .reason + .or_else(|| failure.map(|failure| failure.detail.message.clone())) + .unwrap_or_default(); + if reason == FailureReason::Cancelled { + Some(MainEventBody::RunCancelled(MainEventCancelledProps { + source: source("run.cancelled"), + reason, + category, + message, + final_git_commit_sha, + automation: projection.spec.automation.clone(), + usage, + })) + } else { + Some(MainEventBody::RunFailed(MainEventFailedProps { + source: source("run.failed"), + reason, + category, + message, + final_git_commit_sha, + automation: projection.spec.automation.clone(), + usage, + })) + } + } + _ => None, + } + } + _ => None, + }) +} + +#[cfg(test)] +mod tests { + use chrono::Utc; + use fabro_store::platform_records::{PlatformRecord, StoredPlatformRecord}; + use fabro_types::{FailureCategory, FailureReason, MainEventBody, fixtures, test_support}; + fn native_item(record: PlatformRecord) -> fabro_types::RunStreamItem { + fabro_types::RunStreamItem { + run_id: fixtures::RUN_1, + stream_seq: 42, + kind: fabro_types::RunStreamItemKind::Platform, + id: "7".into(), + recorded_at: 1_800_000_000_000, + item: serde_json::to_value(StoredPlatformRecord { + seq: 7, + recorded_at: 1_800_000_000_000, + record, + position: None, + }) + .unwrap(), + } + } + + #[test] + fn native_creation_uses_the_immutable_record_not_current_projection() { + use fabro_store::platform_records::{PlatformRecord, RunCreatedRecord}; + let spec = test_support::test_run_spec(); + let projection = + fabro_types::RunProjection::new("Renamed later".into(), spec.clone(), Utc::now()); + let item = native_item(PlatformRecord::RunCreated(RunCreatedRecord { + spec, + title: Some("Original title".into()), + parent_id: None, + retried_from: None, + web_url: None, + })); + let Some(MainEventBody::RunCreated(created)) = + super::from_stream_item(&item, &projection).unwrap() + else { + panic!("created"); + }; + assert_eq!(created.title.as_deref(), Some("Original title")); + assert_eq!(created.source.source_event_id, "platform:7"); + assert_eq!( + created.source.source_event_ts.timestamp_millis(), + 1_800_000_000_000 + ); + } + + #[test] + fn native_cancel_request_is_not_a_terminal_event() { + use fabro_store::platform_records::{PlatformRecord, RunLifecycleKind, RunLifecycleRecord}; + let projection = fabro_types::RunProjection::new( + "Test".into(), + test_support::test_run_spec(), + Utc::now(), + ); + let request = native_item(PlatformRecord::RunLifecycle(RunLifecycleRecord::new( + RunLifecycleKind::CancelRequested, + ))); + assert!( + super::from_stream_item(&request, &projection) + .unwrap() + .is_none() + ); + let mut record = RunLifecycleRecord::new(RunLifecycleKind::Failed).with_status( + fabro_types::RunStatus::Failed { + reason: FailureReason::Cancelled, + }, + ); + record.reason = Some("Stopped by user".into()); + let cancelled = native_item(PlatformRecord::RunLifecycle(record)); + let Some(MainEventBody::RunCancelled(cancelled)) = + super::from_stream_item(&cancelled, &projection).unwrap() + else { + panic!("cancelled"); + }; + assert_eq!(cancelled.reason, FailureReason::Cancelled); + assert_eq!(cancelled.category, FailureCategory::Canceled); + assert_eq!(cancelled.message, "Stopped by user"); + assert_eq!(cancelled.usage, None); + } + + #[test] + fn native_lifecycle_maps_success_usage_and_live_transitions() { + use fabro_store::platform_records::{RunLifecycleKind, RunLifecycleRecord}; + use fabro_types::{Conclusion, RunStatus, StageOutcome, SuccessReason}; + use lithos_llm::types::{Cost, CostSource, TokenCounts, Usage}; + + let mut projection = fabro_types::RunProjection::new( + "Test".into(), + test_support::test_run_spec(), + Utc::now(), + ); + let usage = Usage { + tokens: TokenCounts { + input: 10, + cache_read: 4, + ..TokenCounts::default() + }, + cost: Some(Cost { + usd_micros: 42, + source: CostSource::Catalog, + }), + }; + let mut conclusion = Conclusion::outcome_only(Utc::now(), StageOutcome::Succeeded, None); + conclusion.usage = Some(usage); + conclusion.final_git_commit_sha = Some("abc123".into()); + projection.conclusion = Some(conclusion); + let item = native_item(PlatformRecord::RunLifecycle( + RunLifecycleRecord::new(RunLifecycleKind::Succeeded).with_status( + RunStatus::Succeeded { + reason: SuccessReason::Completed, + }, + ), + )); + let Some(MainEventBody::RunCompleted(completed)) = + super::from_stream_item(&item, &projection).unwrap() + else { + panic!("completed"); + }; + assert_eq!(completed.total_usd_micros, Some(42)); + assert_eq!(completed.usage, Some(usage)); + assert_eq!(completed.final_git_commit_sha.as_deref(), Some("abc123")); + + for (transition, name) in [ + (RunLifecycleKind::Running, "fabro.run.running"), + (RunLifecycleKind::Paused, "fabro.run.paused"), + (RunLifecycleKind::Unpaused, "fabro.run.unpaused"), + ] { + let item = native_item(PlatformRecord::RunLifecycle(RunLifecycleRecord::new( + transition, + ))); + assert_eq!( + super::from_stream_item(&item, &projection) + .unwrap() + .unwrap() + .event_name(), + name + ); + } + } + + #[test] + fn native_started_maps_petri_identity_and_ignores_detailed_events() { + use fabro_types::{RunStreamItemKind, StartRecord}; + use serde_json::json; + + let mut projection = fabro_types::RunProjection::new( + "Test".into(), + test_support::test_run_spec(), + Utc::now(), + ); + projection.start = Some(StartRecord { + start_time: Utc::now(), + base_sha: Some("abc123".into()), + run_branch: Some("fabro/run/test".into()), + }); + let mut item = fabro_types::RunStreamItem { + run_id: fixtures::RUN_1, + stream_seq: 3, + kind: RunStreamItemKind::Petri, + id: "coordinator/3/0".into(), + recorded_at: 1_800_000_000_000, + item: json!({"record": {"body": {"event": "run.started"}}}), + }; + let Some(MainEventBody::RunStarted(started)) = + super::from_stream_item(&item, &projection).unwrap() + else { + panic!("started"); + }; + assert_eq!(started.source.source_event_id, "petri:coordinator/3/0"); + assert_eq!(started.base_sha.as_deref(), Some("abc123")); + assert_eq!(started.run_branch.as_deref(), Some("fabro/run/test")); + for event in ["step.started", "step.progress.recorded", "run.finished"] { + item.item["record"]["body"]["event"] = json!(event); + assert!( + super::from_stream_item(&item, &projection) + .unwrap() + .is_none() + ); + } + } + + #[test] + fn native_failed_and_dead_remain_distinct_from_cancellation() { + use fabro_store::platform_records::{RunLifecycleKind, RunLifecycleRecord}; + use fabro_types::RunStatus; + + let projection = fabro_types::RunProjection::new( + "Test".into(), + test_support::test_run_spec(), + Utc::now(), + ); + for transition in [RunLifecycleKind::Failed, RunLifecycleKind::Dead] { + let mut record = RunLifecycleRecord::new(transition).with_status(RunStatus::Failed { + reason: FailureReason::WorkflowError, + }); + record.reason = Some("worker failed".into()); + let item = native_item(PlatformRecord::RunLifecycle(record)); + let Some(MainEventBody::RunFailed(failed)) = + super::from_stream_item(&item, &projection).unwrap() + else { + panic!("failed"); + }; + assert_eq!(failed.message, "worker failed"); + assert_eq!(failed.reason, FailureReason::WorkflowError); + assert_eq!(failed.usage, None); + } + } +}