scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-11 03:40:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

style: format main event stream changes

Browse source
This commit is contained in:
fkukuck 2026-06-21 16:48:41 +02:00
parent fb37518432
commit 4e332ab2c2
7 changed files with 141 additions and 121 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
lib/crates/fabro-server/src/server/handler/runs.rs
View file

@ -35,9 +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,
mirror_persisted_run_created_to_main,
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::{

94
lib/crates/fabro-server/src/server/main_event_mirror.rs
View file

@ -59,9 +59,9 @@ pub(crate) fn main_event_body_from_run_event(event: &RunEvent) -> Option<MainEve
automation: None,
billing: props.billing.clone(),
})),
EventBody::RunPaused(_) => Some(MainEventBody::RunPaused(MainEventLifecycleProps {
source,
})),
EventBody::RunPaused(_) => {
Some(MainEventBody::RunPaused(MainEventLifecycleProps { source }))
}
EventBody::RunUnpaused(_) => Some(MainEventBody::RunUnpaused(MainEventLifecycleProps {
source,
})),
@ -71,33 +71,33 @@ pub(crate) fn main_event_body_from_run_event(event: &RunEvent) -> Option<MainEve
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,
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(),
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 fabro_types::run_event::RunFailedProps;
use serde_json::json;
use super::main_event_body_from_run_event;
fn run_event(event_name: &str, properties: serde_json::Value) -> RunEvent {
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
"properties": properties.clone()
}))
.unwrap()
}
@ -106,7 +106,7 @@ mod tests {
fn maps_created_with_compact_identity_fields() {
let event = run_event(
"run.created",
json!({
&json!({
"title": "Triage bug",
"settings": WorkflowSettings::default(),
"graph": Graph::new("test"),
@ -134,35 +134,35 @@ mod tests {
#[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,
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 {
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,
message: "boom".to_string(),
category: FailureCategory::Deterministic,
causes: Vec::new(),
system_actor: None,
signature: None,
exec_output_tail: None,
},
},
timing: RunTiming::default(),
timing: RunTiming::default(),
final_git_commit_sha: Some("abc123".to_string()),
final_patch: None,
diff_summary: None,
billing: None,
final_patch: None,
diff_summary: None,
billing: None,
}),
};
assert!(matches!(
@ -173,22 +173,22 @@ mod tests {
let mut cancelled = failed.clone();
cancelled.id = "evt-cancelled".to_string();
cancelled.body = EventBody::RunFailed(RunFailedProps {
failure: RunFailure {
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,
detail: FailureDetail {
message: "cancelled".to_string(),
category: FailureCategory::Canceled,
causes: Vec::new(),
system_actor: None,
signature: None,
exec_output_tail: None,
},
},
timing: RunTiming::default(),
timing: RunTiming::default(),
final_git_commit_sha: None,
final_patch: None,
diff_summary: None,
billing: None,
final_patch: None,
diff_summary: None,
billing: None,
});
assert!(matches!(
main_event_body_from_run_event(&cancelled),
@ -200,7 +200,7 @@ mod tests {
fn ignores_non_v1_events_and_cancel_requested() {
let stage = run_event(
"stage.started",
json!({
&json!({
"index": 0,
"handler_type": "command",
"attempt": 1,
@ -209,7 +209,7 @@ mod tests {
);
assert!(main_event_body_from_run_event(&stage).is_none());
let cancel_requested = run_event("run.cancel.requested", json!({ "action": "cancel" }));
let cancel_requested = run_event("run.cancel.requested", &json!({ "action": "cancel" }));
assert!(main_event_body_from_run_event(&cancel_requested).is_none());
}
}

32
lib/crates/fabro-server/src/server/tests.rs
View file

@ -182,14 +182,14 @@ async fn wait_for_main_events(
fn run_event_value(
run_id: RunId,
event_name: &str,
properties: serde_json::Value,
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
"properties": properties.clone()
})
}
@ -197,13 +197,11 @@ async fn append_run_event(
run_store: &fabro_store::RunDatabase,
run_id: RunId,
event_name: &str,
properties: serde_json::Value,
properties: &serde_json::Value,
) {
let payload = fabro_store::EventPayload::new(
run_event_value(run_id, event_name, properties),
&run_id,
)
.unwrap();
let payload =
fabro_store::EventPayload::new(run_event_value(run_id, event_name, properties), &run_id)
.unwrap();
run_store.append_event(&payload).await.unwrap();
}
@ -217,7 +215,7 @@ async fn persisted_run_created_mirrors_to_main_stream_without_forwarder() {
&run_store,
run_id,
"run.created",
json!({
&json!({
"title": "Triage bug",
"settings": WorkflowSettings::default(),
"graph": Graph::new("test"),
@ -254,7 +252,7 @@ async fn forward_run_events_mirrors_live_lifecycle_events_to_main_stream() {
&run_store,
run_id,
"run.created",
json!({
&json!({
"settings": WorkflowSettings::default(),
"graph": Graph::new("test"),
"run_dir": "/tmp/test",
@ -272,7 +270,7 @@ async fn forward_run_events_mirrors_live_lifecycle_events_to_main_stream() {
&run_store,
run_id,
"run.started",
json!({
&json!({
"name": "triage",
"base_branch": "main",
"run_branch": "fabro/run-1",
@ -284,11 +282,11 @@ async fn forward_run_events_mirrors_live_lifecycle_events_to_main_stream() {
&run_store,
run_id,
"run.runnable",
json!({ "source": "start_requested" }),
&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;
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;
@ -309,7 +307,7 @@ async fn forward_run_events_ignores_non_v1_and_maps_cancelled() {
&run_store,
run_id,
"run.created",
json!({
&json!({
"settings": WorkflowSettings::default(),
"graph": Graph::new("test"),
"run_dir": "/tmp/test",
@ -327,7 +325,7 @@ async fn forward_run_events_ignores_non_v1_and_maps_cancelled() {
&run_store,
run_id,
"stage.started",
json!({
&json!({
"index": 0,
"handler_type": "command",
"attempt": 1,
@ -339,7 +337,7 @@ async fn forward_run_events_ignores_non_v1_and_maps_cancelled() {
&run_store,
run_id,
"run.failed",
json!({
&json!({
"failure": {
"reason": "cancelled",
"detail": {

5
lib/crates/fabro-store/src/keys.rs
View file

@ -195,7 +195,10 @@ mod tests {
#[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(main_event_key(7, 123).as_str()),
Some(7)
);
assert_eq!(
parse_main_event_seq(
SlateKey::new("events")

3
lib/crates/fabro-store/src/slate/main_event_store.rs
View file

@ -19,8 +19,7 @@ pub struct MainEventStore {
impl std::fmt::Debug for MainEventStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("MainEventStore")
.finish_non_exhaustive()
f.debug_struct("MainEventStore").finish_non_exhaustive()
}
}

4
lib/crates/fabro-types/src/main_event.rs
View file

@ -22,6 +22,10 @@ pub struct MainEvent {
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 {

118
lib/crates/fabro-types/tests/main_event_serde.rs
View file

@ -1,19 +1,21 @@
use chrono::{TimeZone, Utc};
use fabro_types::{
AuthMethod, FailureCategory, FailureReason, IdpIdentity, MainEvent, MainEventBody,
MainEventCancelledProps, MainEventCompletedProps, MainEventCreatedProps,
MainEventFailedProps, MainEventLifecycleProps, MainEventSource, MainEventStartedProps,
Principal, RunId, SuccessReason,
MainEventCancelledProps, MainEventCompletedProps, MainEventCreatedProps, MainEventFailedProps,
MainEventLifecycleProps, MainEventSource, MainEventStartedProps, Principal, RunId,
SuccessReason,
};
use serde_json::json;
fn run_id() -> RunId {
"01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap()
"01JT56VE4Z5NZ814GZN2JZD65A"
.parse()
.expect("test run id should parse")
}
fn actor() -> Principal {
Principal::user(
IdpIdentity::new("https://github.com", "123").unwrap(),
IdpIdentity::new("https://github.com", "123").expect("test identity should construct"),
"fabian".to_string(),
AuthMethod::Github,
)
@ -21,18 +23,24 @@ fn actor() -> Principal {
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(),
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()),
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(),
ts: Utc
.with_ymd_and_hms(2026, 6, 21, 10, 1, 0)
.single()
.expect("test main event timestamp should be valid"),
body,
}
}
@ -42,25 +50,25 @@ 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,
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()),
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()),
goal: Some("Triage bug".to_string()),
}),
"fabro.run.started",
),
@ -72,37 +80,37 @@ fn v1_event_names_are_stable() {
),
(
MainEventBody::RunCompleted(MainEventCompletedProps {
source: source("run.completed"),
status: "succeeded".to_string(),
reason: SuccessReason::Completed,
total_usd_micros: Some(12),
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,
automation: None,
billing: None,
}),
"fabro.run.completed",
),
(
MainEventBody::RunFailed(MainEventFailedProps {
source: source("run.failed"),
reason: FailureReason::WorkflowError,
category: FailureCategory::Deterministic,
message: "boom".to_string(),
source: source("run.failed"),
reason: FailureReason::WorkflowError,
category: FailureCategory::Deterministic,
message: "boom".to_string(),
final_git_commit_sha: None,
automation: None,
billing: 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(),
source: source("run.failed"),
reason: FailureReason::Cancelled,
category: FailureCategory::Canceled,
message: "cancelled".to_string(),
final_git_commit_sha: None,
automation: None,
billing: None,
automation: None,
billing: None,
}),
"fabro.run.cancelled",
),
@ -129,23 +137,27 @@ fn v1_event_names_are_stable() {
#[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),
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,
automation: None,
billing: None,
}));
let value = serde_json::to_value(&event).unwrap();
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()).unwrap();
assert_eq!(serde_json::to_value(parsed).unwrap(), value);
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]
@ -160,7 +172,8 @@ fn unknown_main_event_preserves_name_and_properties() {
}
});
let parsed: MainEvent = serde_json::from_value(value.clone()).unwrap();
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");
@ -168,5 +181,8 @@ fn unknown_main_event_preserves_name_and_properties() {
}
other => panic!("expected unknown event, got {other:?}"),
}
assert_eq!(serde_json::to_value(parsed).unwrap(), value);
assert_eq!(
serde_json::to_value(parsed).expect("unknown main event should serialize after round trip"),
value
);
}