From b815f072117ac0e75c4ae09dc632bacc8fb562c2 Mon Sep 17 00:00:00 2001 From: fkukuck Date: Sun, 21 Jun 2026 16:00:50 +0200 Subject: [PATCH] feat: add main event types --- lib/crates/fabro-types/src/lib.rs | 6 + lib/crates/fabro-types/src/main_event.rs | 307 ++++++++++++++++++ .../fabro-types/tests/main_event_serde.rs | 172 ++++++++++ 3 files changed, 485 insertions(+) create mode 100644 lib/crates/fabro-types/src/main_event.rs create mode 100644 lib/crates/fabro-types/tests/main_event_serde.rs diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index ce31e5139..7c67b51f3 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 000000000..f0619b17e --- /dev/null +++ b/lib/crates/fabro-types/src/main_event.rs @@ -0,0 +1,307 @@ +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()) +} 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 000000000..a543ce969 --- /dev/null +++ b/lib/crates/fabro-types/tests/main_event_serde.rs @@ -0,0 +1,172 @@ +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::new("https://github.com", "123").unwrap(), + "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); +}