diff --git a/AGENTS.md b/AGENTS.md index f28f0ae63..fe5cfe033 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -79,7 +79,7 @@ Fabro is an AI-powered workflow orchestration platform. Workflows are defined as When working on Rust crates, read the relevant strategy doc **before** making changes: - **`docs-internal/logging-strategy.md`** — read when adding `tracing` calls (`info!`, `debug!`, `warn!`, `error!`), working on error handling paths, or adding new operations that should be observable -- **`docs-internal/events-strategy.md`** — read when adding or modifying `WorkflowRunEvent` variants, touching `EventEmitter`/`emit()`, changing `progress.jsonl` output, or adding new workflow stage types +- **`docs-internal/events-strategy.md`** — read when adding or modifying `Event` variants, touching `EventEmitter`/`emit()`, changing `progress.jsonl` output, or adding new workflow stage types - **`files-internal/testing-strategy.md`** — read when adding or reorganizing tests, choosing between unit vs `tests/it`, deciding whether a test belongs in `cmd` vs `workflow` vs `scenario`, or deciding how to structure snapshots and fixtures ## Shell quoting in sandbox code diff --git a/docs-internal/events-strategy.md b/docs-internal/events-strategy.md index 88a3e40b9..ee69a070f 100644 --- a/docs-internal/events-strategy.md +++ b/docs-internal/events-strategy.md @@ -4,12 +4,12 @@ Fabro emits structured **workflow run events** during execution for observabilit Events are distinct from tracing logs. Tracing is developer diagnostics; events are product-facing state transitions and activity records that other systems consume. -Detached runs rely on this distinction. If something needs to be visible after reattach, emit a `WorkflowRunEvent` rather than only logging to stderr or `detach.log`. +Detached runs rely on this distinction. If something needs to be visible after reattach, emit a `Event` rather than only logging to stderr or `detach.log`. ## Architecture ```text -Engine/Handler -> WorkflowRunEvent -> EventEmitter::emit() +Engine/Handler -> Event -> EventEmitter::emit() |- trace(raw event) |- canonicalize -> RunEventEnvelope `- on_event(&RunEventEnvelope) @@ -21,9 +21,9 @@ Engine/Handler -> WorkflowRunEvent -> EventEmitter::emit() The canonical envelope is built exactly once in `fabro-workflow/src/event.rs`. -- `WorkflowRunEvent` remains the internal typed source of truth. +- `Event` remains the internal typed source of truth. - `EventEmitter` owns an immutable `run_id` and converts typed events into `RunEventEnvelope`. -- Every listener receives `&RunEventEnvelope`, not `&WorkflowRunEvent`. +- Every listener receives `&RunEventEnvelope`, not `&Event`. - Bypass paths that cannot go through the emitter must call `canonicalize_event()` once and reuse the same envelope for every sink. ## Canonical Envelope @@ -112,11 +112,11 @@ Never canonicalize the same logical event twice if multiple sinks receive it. ### 1. Add the typed event -Add a variant to `WorkflowRunEvent`, `AgentEvent`, or `SandboxEvent` as appropriate. +Add a variant to `Event`, `AgentEvent`, or `SandboxEvent` as appropriate. ### 2. Add tracing -Extend `WorkflowRunEvent::trace()` so the raw event is observable in tracing output. +Extend `Event::trace()` so the raw event is observable in tracing output. ### 3. Add an external name @@ -132,7 +132,7 @@ Update `extract_envelope_fields()`: ### 5. Emit it -Prefer `EventEmitter::emit(&WorkflowRunEvent::...)`. +Prefer `EventEmitter::emit(&Event::...)`. Use `canonicalize_event()` only for true bypass paths. diff --git a/docs-internal/logging-strategy.md b/docs-internal/logging-strategy.md index 0b6b0d4b1..bca730944 100644 --- a/docs-internal/logging-strategy.md +++ b/docs-internal/logging-strategy.md @@ -23,7 +23,7 @@ Production runs at INFO level. INFO should be low-volume and high-signal — the - Hot loops or per-token streaming events (use DEBUG only if truly needed for diagnosis) - Data that belongs in user-facing output (`eprintln!` for interactive CLI feedback, not tracing) -- Detached user-visible warnings or errors that need to survive `attach`/`logs` (`detach.log` is debug-only; emit a `WorkflowRunEvent` into `progress.jsonl` instead) +- Detached user-visible warnings or errors that need to survive `attach`/`logs` (`detach.log` is debug-only; emit a `Event` into `progress.jsonl` instead) - Redundant information already captured by a parent event (if you logged "starting X", you don't need to log every sub-step at the same level) - Events that are already traced via `EventEnum::trace()` — the event enums (`AgentEvent`, `PipelineEvent`, `ExecutionEnvEvent`) each have a `trace()` method called automatically at their emit site; do not add manual `info!`/`debug!` calls that duplicate what `trace()` already emits - Wrapper/forwarding variants that re-emit an inner event — `PipelineEvent::Agent`, `PipelineEvent::ExecutionEnv`, and `AgentEvent::SubAgentEvent` are no-ops in `trace()` because the inner event is already traced at its origin diff --git a/docs-internal/plan-events-as-source-of-truth-follow-ups.md b/docs-internal/plan-events-as-source-of-truth-follow-ups.md index 59ccde68f..65dd707ca 100644 --- a/docs-internal/plan-events-as-source-of-truth-follow-ups.md +++ b/docs-internal/plan-events-as-source-of-truth-follow-ups.md @@ -149,7 +149,7 @@ This matrix is the contract for this plan. Each row must be green before `memoiz ### 1. Add semantic run lifecycle events -Add new `WorkflowRunEvent` variants for: +Add new `Event` variants for: - `RunSubmitted` - `RunStarting` diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index d3413f444..9791aa392 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -4,7 +4,7 @@ use cli_table::format::{Border, Separator}; use cli_table::{Cell, CellStruct, Color, Style, Table}; use fabro_checkpoint::git::Store; use fabro_util::terminal::Styles; -use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event}; +use fabro_workflow::event::{Event, append_event}; use fabro_workflow::git::MetadataStore; use fabro_workflow::operations::{ RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, @@ -133,10 +133,10 @@ async fn reset_rewound_run_state( let run_store = durable_store.open_run(run_id).await.map_err(|err| { anyhow::anyhow!("failed to open durable store run for rewind reset: {err}") })?; - append_workflow_event( + append_event( &run_store, run_id, - &WorkflowRunEvent::RunRewound { + &Event::RunRewound { target_checkpoint_ordinal: entry.ordinal, target_node_id: entry.node_name.clone(), target_visit: entry.visit, @@ -146,20 +146,16 @@ async fn reset_rewound_run_state( ) .await .map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?; - append_workflow_event(&run_store, run_id, &restored_checkpoint_event(&checkpoint)) + append_event(&run_store, run_id, &restored_checkpoint_event(&checkpoint)) .await .map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?; - append_workflow_event( - &run_store, - run_id, - &WorkflowRunEvent::RunSubmitted { reason: None }, - ) - .await - .map_err(|err| anyhow::anyhow!("failed to append restored run status event: {err}"))?; + append_event(&run_store, run_id, &Event::RunSubmitted { reason: None }) + .await + .map_err(|err| anyhow::anyhow!("failed to append restored run status event: {err}"))?; Ok(()) } -fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> WorkflowRunEvent { +fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> Event { let current_status = checkpoint .node_outcomes .get(&checkpoint.current_node) @@ -167,7 +163,7 @@ fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> WorkflowRu || "success".to_string(), |outcome| outcome.status.to_string(), ); - WorkflowRunEvent::CheckpointCompleted { + Event::CheckpointCompleted { node_id: checkpoint.current_node.clone(), status: current_status, current_node: checkpoint.current_node.clone(), diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index 95d54be96..67e8a8932 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -1,7 +1,7 @@ use std::convert::TryFrom; use chrono::{DateTime, Utc}; -use fabro_types::{EventBody, StageUsage, StoredEvent}; +use fabro_types::{EventBody, RunEvent, StageUsage}; use fabro_workflow::event::RunNoticeLevel; use fabro_workflow::outcome::compute_stage_cost; use serde_json::Value; @@ -236,7 +236,7 @@ pub(super) enum ProgressEvent { }, } -pub(super) fn from_stored_event(stored: &StoredEvent) -> Option { +pub(super) fn from_run_event(stored: &RunEvent) -> Option { let node_id = stored.node_id.clone().unwrap_or_else(|| "?".to_string()); let node_label = stored.node_label.clone().unwrap_or_else(|| node_id.clone()); @@ -459,8 +459,8 @@ pub(super) fn from_stored_event(stored: &StoredEvent) -> Option { } pub(super) fn from_json_line(line: &str) -> Option { - let stored = StoredEvent::from_json_str(line).ok()?; - from_stored_event(&stored) + let stored = RunEvent::from_json_str(line).ok()?; + from_run_event(&stored) } fn display_value(value: &Value) -> Option { @@ -495,15 +495,15 @@ fn display_value(value: &Value) -> Option { mod tests { use fabro_agent::AgentEvent; use fabro_types::fixtures; - use fabro_workflow::event::{WorkflowRunEvent, to_stored_event}; + use fabro_workflow::event::{Event, to_run_event}; use super::*; #[test] fn parse_edge_selected() { - let stored = to_stored_event( + let stored = to_run_event( &fixtures::RUN_1, - &WorkflowRunEvent::EdgeSelected { + &Event::EdgeSelected { from_node: "a".into(), to_node: "b".into(), label: Some("yes".into()), @@ -516,7 +516,7 @@ mod tests { }, ); - let event = from_stored_event(&stored).unwrap(); + let event = from_run_event(&stored).unwrap(); assert!(matches!( event, ProgressEvent::EdgeSelected { @@ -530,7 +530,7 @@ mod tests { #[test] fn round_trip_stage_completed() { - let event = WorkflowRunEvent::StageCompleted { + let event = Event::StageCompleted { node_id: "plan".into(), name: "Plan".into(), index: 0, @@ -553,8 +553,8 @@ mod tests { max_attempts: 1, }; - let stored = to_stored_event(&fixtures::RUN_1, &event); - let parsed = from_stored_event(&stored).unwrap(); + let stored = to_run_event(&fixtures::RUN_1, &event); + let parsed = from_run_event(&stored).unwrap(); assert!(matches!( parsed, ProgressEvent::StageCompleted { @@ -568,7 +568,7 @@ mod tests { #[test] fn round_trip_agent_tool_call() { - let event = WorkflowRunEvent::Agent { + let event = Event::Agent { stage: "code".into(), visit: 1, event: AgentEvent::ToolCallStarted { @@ -580,8 +580,8 @@ mod tests { parent_session_id: None, }; - let stored = to_stored_event(&fixtures::RUN_1, &event); - let parsed = from_stored_event(&stored).unwrap(); + let stored = to_run_event(&fixtures::RUN_1, &event); + let parsed = from_run_event(&stored).unwrap(); assert!(matches!( parsed, ProgressEvent::ToolCallStarted { @@ -656,7 +656,7 @@ mod tests { #[test] fn round_trip_sandbox_ready() { - let event = WorkflowRunEvent::Sandbox { + let event = Event::Sandbox { event: fabro_agent::SandboxEvent::Ready { provider: "daytona".into(), duration_ms: 2500, @@ -667,8 +667,8 @@ mod tests { }, }; - let stored = to_stored_event(&fixtures::RUN_1, &event); - let parsed = from_stored_event(&stored).unwrap(); + let stored = to_run_event(&fixtures::RUN_1, &event); + let parsed = from_run_event(&stored).unwrap(); assert!(matches!( parsed, ProgressEvent::SandboxReady { @@ -682,14 +682,14 @@ mod tests { #[test] fn round_trip_run_notice() { - let event = WorkflowRunEvent::RunNotice { + let event = Event::RunNotice { level: RunNoticeLevel::Warn, code: "sandbox_cleanup_failed".into(), message: "sandbox cleanup failed".into(), }; - let stored = to_stored_event(&fixtures::RUN_1, &event); - let parsed = from_stored_event(&stored).unwrap(); + let stored = to_run_event(&fixtures::RUN_1, &event); + let parsed = from_run_event(&stored).unwrap(); assert!(matches!( parsed, ProgressEvent::RunNotice { diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index dface5807..bcd56f8e5 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -1,4 +1,4 @@ -use fabro_types::StoredEvent; +use fabro_types::RunEvent; mod event; mod info_display; @@ -7,7 +7,7 @@ mod setup_display; mod stage_display; mod styles; -use event::{ProgressEvent, from_json_line, from_stored_event}; +use event::{ProgressEvent, from_json_line, from_run_event}; use info_display::InfoDisplay; use renderer::ProgressRenderer; use setup_display::SetupDisplay; @@ -66,8 +66,8 @@ impl ProgressUI { } #[cfg_attr(not(test), allow(dead_code))] - pub(crate) fn handle_event(&mut self, event: &StoredEvent) { - if let Some(progress_event) = from_stored_event(event) { + pub(crate) fn handle_event(&mut self, event: &RunEvent) { + if let Some(progress_event) = from_run_event(event) { self.dispatch(progress_event); } } @@ -418,9 +418,7 @@ mod tests { use fabro_agent::{AgentEvent, SandboxEvent}; use fabro_llm::types::Usage; use fabro_types::fixtures; - use fabro_workflow::event::{ - RunNoticeLevel, WorkflowRunEvent, to_stored_event, to_stored_event_at, - }; + use fabro_workflow::event::{Event, RunNoticeLevel, to_run_event, to_run_event_at}; use fabro_workflow::outcome::StageUsage; use super::*; @@ -461,18 +459,18 @@ mod tests { .expect("valid utf-8") } - fn emit(ui: &mut ProgressUI, event: WorkflowRunEvent) { - let stored = to_stored_event(&fixtures::RUN_1, &event); + fn emit(ui: &mut ProgressUI, event: Event) { + let stored = to_run_event(&fixtures::RUN_1, &event); ui.handle_event(&stored); } - fn emit_ref(ui: &mut ProgressUI, event: &WorkflowRunEvent) { - let stored = to_stored_event(&fixtures::RUN_1, event); + fn emit_ref(ui: &mut ProgressUI, event: &Event) { + let stored = to_run_event(&fixtures::RUN_1, event); ui.handle_event(&stored); } - fn agent_event(stage: &str, event: AgentEvent) -> WorkflowRunEvent { - WorkflowRunEvent::Agent { + fn agent_event(stage: &str, event: AgentEvent) -> Event { + Event::Agent { stage: stage.into(), visit: 1, event, @@ -481,8 +479,8 @@ mod tests { } } - fn stage_started(node_id: &str, name: &str) -> WorkflowRunEvent { - WorkflowRunEvent::StageStarted { + fn stage_started(node_id: &str, name: &str) -> Event { + Event::StageStarted { node_id: node_id.into(), name: name.into(), index: 0, @@ -492,7 +490,7 @@ mod tests { } } - fn assistant_message(stage: &str, model: &str) -> WorkflowRunEvent { + fn assistant_message(stage: &str, model: &str) -> Event { agent_event( stage, AgentEvent::AssistantMessage { @@ -504,8 +502,8 @@ mod tests { ) } - fn stage_completed(node_id: &str, name: &str) -> WorkflowRunEvent { - WorkflowRunEvent::StageCompleted { + fn stage_completed(node_id: &str, name: &str) -> Event { + Event::StageCompleted { node_id: node_id.into(), name: name.into(), index: 0, @@ -548,7 +546,7 @@ mod tests { emit( &mut ui, - WorkflowRunEvent::ParallelStarted { + Event::ParallelStarted { node_id: "fork1".into(), visit: 1, branch_count: 2, @@ -559,7 +557,7 @@ mod tests { emit( &mut ui, - WorkflowRunEvent::ParallelBranchStarted { + Event::ParallelBranchStarted { branch: "security".into(), index: 0, }, @@ -574,7 +572,7 @@ mod tests { emit( &mut ui, - WorkflowRunEvent::ParallelBranchCompleted { + Event::ParallelBranchCompleted { branch: "security".into(), index: 0, duration_ms: 2000, @@ -596,7 +594,7 @@ mod tests { emit(&mut ui, stage_started("fork1", "Fork")); emit( &mut ui, - WorkflowRunEvent::ParallelStarted { + Event::ParallelStarted { node_id: "fork1".into(), visit: 1, branch_count: 1, @@ -605,7 +603,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::ParallelBranchStarted { + Event::ParallelBranchStarted { branch: "security".into(), index: 0, }, @@ -666,7 +664,7 @@ mod tests { fn handle_json_line_matches_handle_event_for_verbose_events() { let events = vec![ stage_started("code", "Code"), - WorkflowRunEvent::SandboxInitialized { + Event::SandboxInitialized { working_directory: "/home/daytona/workspace".into(), provider: "daytona".into(), identifier: None, @@ -684,7 +682,7 @@ mod tests { }, ), assistant_message("code", "gpt-5-mini"), - WorkflowRunEvent::EdgeSelected { + Event::EdgeSelected { from_node: "code".into(), to_node: "review".into(), label: Some("ship".into()), @@ -695,7 +693,7 @@ mod tests { stage_status: "success".into(), is_jump: false, }, - WorkflowRunEvent::StageRetrying { + Event::StageRetrying { node_id: "code".into(), name: "Code".into(), index: 0, @@ -741,26 +739,26 @@ mod tests { turns_used: 3, }, ), - WorkflowRunEvent::SetupStarted { command_count: 1 }, - WorkflowRunEvent::SetupCommandCompleted { + Event::SetupStarted { command_count: 1 }, + Event::SetupCommandCompleted { command: "bun install".into(), index: 0, exit_code: 0, duration_ms: 2200, }, - WorkflowRunEvent::SetupCompleted { duration_ms: 2200 }, - WorkflowRunEvent::DevcontainerLifecycleStarted { + Event::SetupCompleted { duration_ms: 2200 }, + Event::DevcontainerLifecycleStarted { phase: "postCreate".into(), command_count: 1, }, - WorkflowRunEvent::DevcontainerLifecycleCommandCompleted { + Event::DevcontainerLifecycleCommandCompleted { phase: "postCreate".into(), command: "npm run setup".into(), index: 0, exit_code: 0, duration_ms: 1400, }, - WorkflowRunEvent::DevcontainerLifecycleCompleted { + Event::DevcontainerLifecycleCompleted { phase: "postCreate".into(), duration_ms: 1400, }, @@ -773,7 +771,7 @@ mod tests { let (mut json_ui, json_buffer) = capture_ui(true); for event in &events { - let line = serde_json::to_string(&to_stored_event(&fixtures::RUN_1, event)).unwrap(); + let line = serde_json::to_string(&to_run_event(&fixtures::RUN_1, event)).unwrap(); json_ui.handle_json_line(&line); } @@ -822,7 +820,7 @@ mod tests { emit( &mut ui, - WorkflowRunEvent::Sandbox { + Event::Sandbox { event: SandboxEvent::Initializing { provider: "daytona".into(), }, @@ -830,7 +828,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::Sandbox { + Event::Sandbox { event: SandboxEvent::Ready { provider: "daytona".into(), duration_ms: 2500, @@ -843,18 +841,15 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::SshAccessReady { + Event::SshAccessReady { ssh_command: "ssh daytona@example".into(), }, ); - emit(&mut ui, WorkflowRunEvent::SetupStarted { command_count: 2 }); + emit(&mut ui, Event::SetupStarted { command_count: 2 }); + emit(&mut ui, Event::SetupCompleted { duration_ms: 8200 }); emit( &mut ui, - WorkflowRunEvent::SetupCompleted { duration_ms: 8200 }, - ); - emit( - &mut ui, - WorkflowRunEvent::CliEnsureCompleted { + Event::CliEnsureCompleted { cli_name: "gh".into(), provider: "github".into(), already_installed: false, @@ -864,7 +859,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::DevcontainerResolved { + Event::DevcontainerResolved { dockerfile_lines: 24, environment_count: 3, lifecycle_command_count: 2, @@ -873,14 +868,14 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::DevcontainerLifecycleStarted { + Event::DevcontainerLifecycleStarted { phase: "postCreate".into(), command_count: 2, }, ); emit( &mut ui, - WorkflowRunEvent::DevcontainerLifecycleCompleted { + Event::DevcontainerLifecycleCompleted { phase: "postCreate".into(), duration_ms: 1800, }, @@ -906,7 +901,7 @@ mod tests { emit(&mut ui, stage_started("code", "Code")); emit( &mut ui, - WorkflowRunEvent::SandboxInitialized { + Event::SandboxInitialized { working_directory: "/home/daytona/workspace".into(), provider: "daytona".into(), identifier: None, @@ -930,7 +925,7 @@ mod tests { emit(&mut ui, assistant_message("code", "gpt-5-mini")); emit( &mut ui, - WorkflowRunEvent::EdgeSelected { + Event::EdgeSelected { from_node: "code".into(), to_node: "review".into(), label: Some("ship".into()), @@ -944,7 +939,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::StageRetrying { + Event::StageRetrying { node_id: "code".into(), name: "Code".into(), index: 0, @@ -1003,30 +998,27 @@ mod tests { }, ), ); - emit(&mut ui, WorkflowRunEvent::SetupStarted { command_count: 1 }); + emit(&mut ui, Event::SetupStarted { command_count: 1 }); emit( &mut ui, - WorkflowRunEvent::SetupCommandCompleted { + Event::SetupCommandCompleted { command: "bun install".into(), index: 0, exit_code: 0, duration_ms: 2200, }, ); + emit(&mut ui, Event::SetupCompleted { duration_ms: 2200 }); emit( &mut ui, - WorkflowRunEvent::SetupCompleted { duration_ms: 2200 }, - ); - emit( - &mut ui, - WorkflowRunEvent::DevcontainerLifecycleStarted { + Event::DevcontainerLifecycleStarted { phase: "postCreate".into(), command_count: 1, }, ); emit( &mut ui, - WorkflowRunEvent::DevcontainerLifecycleCommandCompleted { + Event::DevcontainerLifecycleCommandCompleted { phase: "postCreate".into(), command: "npm run setup".into(), index: 0, @@ -1036,7 +1028,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::DevcontainerLifecycleCompleted { + Event::DevcontainerLifecycleCompleted { phase: "postCreate".into(), duration_ms: 1400, }, @@ -1065,7 +1057,7 @@ mod tests { emit( &mut ui, - WorkflowRunEvent::RunNotice { + Event::RunNotice { level: RunNoticeLevel::Warn, code: "sandbox_cleanup_failed".into(), message: "sandbox cleanup failed".into(), @@ -1073,7 +1065,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::PullRequestCreated { + Event::PullRequestCreated { pr_url: "https://github.com/fabro-sh/fabro/pull/42".into(), pr_number: 42, owner: "fabro-sh".into(), @@ -1086,7 +1078,7 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::PullRequestFailed { + Event::PullRequestFailed { error: "auth token expired".into(), }, ); @@ -1105,7 +1097,7 @@ mod tests { emit(&mut ui, stage_started("fork1", "Fork")); emit( &mut ui, - WorkflowRunEvent::ParallelStarted { + Event::ParallelStarted { node_id: "fork1".into(), visit: 1, branch_count: 1, @@ -1114,14 +1106,14 @@ mod tests { ); emit( &mut ui, - WorkflowRunEvent::ParallelBranchStarted { + Event::ParallelBranchStarted { branch: "security".into(), index: 0, }, ); emit( &mut ui, - WorkflowRunEvent::ParallelBranchCompleted { + Event::ParallelBranchCompleted { branch: "security".into(), index: 0, duration_ms: 500, @@ -1145,9 +1137,9 @@ mod tests { .unwrap() .with_timezone(&Utc); - let stage_started = serde_json::to_string(&to_stored_event_at( + let stage_started = serde_json::to_string(&to_run_event_at( &fixtures::RUN_1, - &WorkflowRunEvent::StageStarted { + &Event::StageStarted { node_id: "code".into(), name: "Code".into(), index: 0, @@ -1158,7 +1150,7 @@ mod tests { started_ts, )) .unwrap(); - let tool_started = serde_json::to_string(&to_stored_event_at( + let tool_started = serde_json::to_string(&to_run_event_at( &fixtures::RUN_1, &agent_event( "code", @@ -1171,7 +1163,7 @@ mod tests { started_ts, )) .unwrap(); - let tool_completed = serde_json::to_string(&to_stored_event_at( + let tool_completed = serde_json::to_string(&to_run_event_at( &fixtures::RUN_1, &agent_event( "code", diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index cab426cac..a25071356 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -9,7 +9,7 @@ use crate::shared::print_json_pretty; use crate::store; use crate::user_config::load_user_settings_with_globals; use fabro_sandbox::reconnect::reconnect as reconnect_sandbox; -use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event}; +use fabro_workflow::event::{Event, append_event}; use fabro_workflow::run_lookup::RunInfo; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; @@ -126,12 +126,8 @@ async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Resul } }; if let Some(run_store) = run_store.as_ref() { - if let Err(err) = append_workflow_event( - run_store, - &run_id, - &WorkflowRunEvent::RunRemoving { reason: None }, - ) - .await + if let Err(err) = + append_event(run_store, &run_id, &Event::RunRemoving { reason: None }).await { warn!( run_id = %run_id, diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index f4cc1bda8..81c66e2d4 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -143,7 +143,7 @@ mod tests { RunRecord, RunStatus, RunStatusRecord, SandboxRecord, Settings, StageStatus, StartRecord, StatusReason, fixtures, }; - use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event}; + use fabro_workflow::event::{Event, append_event}; use object_store::memory::InMemory; fn dt(rfc3339: &str) -> DateTime { @@ -294,10 +294,10 @@ mod tests { let sandbox = sample_sandbox(); let node = StageId::new("code", 2); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -314,10 +314,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::WorkflowRunStarted { + &Event::WorkflowRunStarted { name: "night-sky".to_string(), run_id, base_branch: run_record.base_branch.clone(), @@ -329,20 +329,20 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::RunRunning { + &Event::RunRunning { reason: status_record.reason, }, ) .await .unwrap(); for checkpoint in [&first_checkpoint, &second_checkpoint] { - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::CheckpointCompleted { + &Event::CheckpointCompleted { node_id: checkpoint.current_node.clone(), status: "success".to_string(), current_node: checkpoint.current_node.clone(), @@ -371,10 +371,10 @@ mod tests { .await .unwrap(); } - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::SandboxInitialized { + &Event::SandboxInitialized { working_directory: sandbox.working_directory.clone(), provider: sandbox.provider.clone(), identifier: sandbox.identifier.clone(), @@ -384,10 +384,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::Prompt { + &Event::Prompt { stage: "code".to_string(), visit: 2, text: "Plan the fix".to_string(), @@ -398,10 +398,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::PromptCompleted { + &Event::PromptCompleted { node_id: "code".to_string(), response: "Implemented".to_string(), model: "gpt-5".to_string(), @@ -411,10 +411,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::StageCompleted { + &Event::StageCompleted { node_id: "code".to_string(), name: "Code".to_string(), index: 1, @@ -442,10 +442,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::CommandStarted { + &Event::CommandStarted { node_id: "code".to_string(), script: "echo hi".to_string(), command: "echo hi".to_string(), @@ -455,10 +455,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::CommandCompleted { + &Event::CommandCompleted { node_id: "code".to_string(), stdout: "stdout line".to_string(), stderr: String::new(), @@ -469,10 +469,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::RetroStarted { + &Event::RetroStarted { prompt: Some("How did it go?".to_string()), provider: None, model: None, @@ -480,10 +480,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::RetroCompleted { + &Event::RetroCompleted { duration_ms: 50, response: Some("Smooth enough".to_string()), retro: Some(serde_json::to_value(&retro).unwrap()), @@ -491,10 +491,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::WorkflowRunCompleted { + &Event::WorkflowRunCompleted { duration_ms: conclusion.duration_ms, artifact_count: 0, status: "success".to_string(), @@ -645,10 +645,10 @@ mod tests { let run_id = test_run_id(); let run = store.create_run(&run_id).await.unwrap(); let run_record = sample_run_record(run_id, created_at); - append_workflow_event( + append_event( &run, &run_id, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), diff --git a/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs b/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs index db2415057..311206fa7 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/pr_view.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use fabro_test::{fabro_snapshot, test_context}; use fabro_types::RunId; -use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event}; +use fabro_workflow::event::{Event, append_event}; use object_store::local::LocalFileSystem; use super::support::setup_completed_dry_run; @@ -78,10 +78,10 @@ fn pr_view_reads_pull_request_from_store_without_pull_request_json() { runtime.block_on(async { let store = build_store(&context.storage_dir); let run_store = store.open_run(&run_id).await.unwrap(); - append_workflow_event( + append_event( &run_store, &run_id, - &WorkflowRunEvent::PullRequestCreated { + &Event::PullRequestCreated { pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), pr_number: 123, owner: "fabro-sh".to_string(), diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index edc9cdf21..640948c71 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -20,7 +20,7 @@ use fabro_llm::types::{ Response as LlmResponse, Role, StreamEvent, ToolChoice, ToolDefinition, Usage, }; use fabro_store::StoreHandle; -use fabro_types::{RunId, Settings, StoredEvent}; +use fabro_types::{RunEvent, RunId, Settings}; use fabro_util::redact::redact_jsonl_line; use fabro_workflow::error::FabroError; use fabro_workflow::handler::HandlerRegistry; @@ -97,7 +97,7 @@ struct ManagedRun { created_at: chrono::DateTime, // Populated when running: interviewer: Option>, - event_tx: Option>, + event_tx: Option>, context: Option, checkpoint: Option, cancel_tx: Option>, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index ff513b60f..2f26f4304 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -6,11 +6,11 @@ use chrono::{DateTime, Utc}; use serde_json::Value; use crate::{EventEnvelope, Result, RunSummary, StageId, StoreError}; -use fabro_types::stored_event::{RunCompletedProps, RunFailedProps, StageCompletedProps}; +use fabro_types::run_event::{RunCompletedProps, RunFailedProps, StageCompletedProps}; use fabro_types::{ Checkpoint, Conclusion, EventBody, FailureSignature, NodeStatusRecord, Outcome, - PullRequestRecord, Retro, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, - StageStatus, StageUsage, StartRecord, StatusReason, StoredEvent, TokenUsage, + PullRequestRecord, Retro, RunEvent, RunId, RunRecord, RunStatus, RunStatusRecord, + SandboxRecord, StageStatus, StageUsage, StartRecord, StatusReason, TokenUsage, }; #[derive(Debug, Clone, Default)] @@ -75,7 +75,7 @@ impl RunProjection { } pub(crate) fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> { - let stored = StoredEvent::from_value(event.payload.as_value().clone()) + let stored = RunEvent::from_value(event.payload.as_value().clone()) .map_err(|err| StoreError::InvalidEvent(format!("invalid stored event: {err}")))?; let ts = stored.ts; let run_id = stored.run_id; @@ -401,7 +401,7 @@ fn run_status_record( } fn checkpoint_from_props( - props: &fabro_types::stored_event::CheckpointCompletedProps, + props: &fabro_types::run_event::CheckpointCompletedProps, timestamp: DateTime, ) -> Checkpoint { let loop_failure_signatures = props @@ -531,7 +531,7 @@ fn node_status_from_outcome( } } -fn provider_used_from_prompt(props: &fabro_types::stored_event::StagePromptProps) -> Option { +fn provider_used_from_prompt(props: &fabro_types::run_event::StagePromptProps) -> Option { let mut provider_used = serde_json::Map::new(); if let Some(mode) = props.mode.clone() { provider_used.insert("mode".to_string(), Value::String(mode)); @@ -546,7 +546,7 @@ fn provider_used_from_prompt(props: &fabro_types::stored_event::StagePromptProps } fn provider_used_from_agent_session_started( - props: &fabro_types::stored_event::AgentSessionStartedProps, + props: &fabro_types::run_event::AgentSessionStartedProps, ) -> Value { let mut provider_used = serde_json::Map::new(); provider_used.insert("mode".to_string(), Value::String("agent".to_string())); @@ -560,7 +560,7 @@ fn provider_used_from_agent_session_started( } fn provider_used_from_agent_cli_started( - props: &fabro_types::stored_event::AgentCliStartedProps, + props: &fabro_types::run_event::AgentCliStartedProps, ) -> Value { let mut provider_used = serde_json::Map::new(); provider_used.insert("mode".to_string(), Value::String("cli".to_string())); diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index d43c63e1d..49fc3f900 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -4,7 +4,7 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use crate::{Result, StoreError}; -use fabro_types::{RunId, RunStatus, StatusReason, StoredEvent}; +use fabro_types::{RunEvent, RunId, RunStatus, StatusReason}; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct RunSummary { @@ -70,11 +70,11 @@ impl EventPayload { } } -impl TryFrom<&EventPayload> for StoredEvent { +impl TryFrom<&EventPayload> for RunEvent { type Error = StoreError; fn try_from(value: &EventPayload) -> Result { - StoredEvent::from_value(value.as_value().clone()) + RunEvent::from_value(value.as_value().clone()) .map_err(|err| StoreError::InvalidEvent(format!("invalid stored event: {err}"))) } } diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index c3ba9a854..f6954b279 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -11,13 +11,13 @@ pub mod pull_request; pub mod retro; pub mod run; pub mod run_blob_id; +pub mod run_event; pub mod run_id; pub mod sandbox_record; pub mod settings; pub mod stage_id; pub mod start; pub mod status; -pub mod stored_event; pub mod usage; pub use checkpoint::Checkpoint; @@ -33,6 +33,7 @@ pub use retro::{ }; pub use run::RunRecord; pub use run_blob_id::RunBlobId; +pub use run_event::{EventBody, RunEvent, RunNoticeLevel, TokenUsage}; pub use run_id::RunId; pub use run_id::fixtures; pub use sandbox_record::SandboxRecord; @@ -42,7 +43,6 @@ pub use start::StartRecord; pub use status::{ InvalidTransition, ParseRunStatusError, RunStatus, RunStatusRecord, StatusReason, }; -pub use stored_event::{EventBody, RunNoticeLevel, StoredEvent, TokenUsage}; pub use usage::StageUsage; pub use fabro_macros::Combine; diff --git a/lib/crates/fabro-types/src/stored_event/agent.rs b/lib/crates/fabro-types/src/run_event/agent.rs similarity index 100% rename from lib/crates/fabro-types/src/stored_event/agent.rs rename to lib/crates/fabro-types/src/run_event/agent.rs diff --git a/lib/crates/fabro-types/src/stored_event/infra.rs b/lib/crates/fabro-types/src/run_event/infra.rs similarity index 100% rename from lib/crates/fabro-types/src/stored_event/infra.rs rename to lib/crates/fabro-types/src/run_event/infra.rs diff --git a/lib/crates/fabro-types/src/stored_event/misc.rs b/lib/crates/fabro-types/src/run_event/misc.rs similarity index 100% rename from lib/crates/fabro-types/src/stored_event/misc.rs rename to lib/crates/fabro-types/src/run_event/misc.rs diff --git a/lib/crates/fabro-types/src/stored_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs similarity index 97% rename from lib/crates/fabro-types/src/stored_event/mod.rs rename to lib/crates/fabro-types/src/run_event/mod.rs index 9ca4ea640..1c9b6cd50 100644 --- a/lib/crates/fabro-types/src/stored_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -43,7 +43,7 @@ pub struct TokenUsage { } #[derive(Debug, Clone, PartialEq)] -pub struct StoredEvent { +pub struct RunEvent { pub id: String, pub ts: DateTime, pub run_id: RunId, @@ -264,7 +264,7 @@ pub enum EventBody { } #[derive(Debug, Clone, Deserialize)] -struct StoredEventRaw { +struct RunEventRaw { id: String, ts: DateTime, run_id: RunId, @@ -285,9 +285,9 @@ fn default_properties() -> Value { Value::Object(Map::new()) } -impl StoredEvent { +impl RunEvent { pub fn from_value(value: Value) -> serde_json::Result { - let raw: StoredEventRaw = serde_json::from_value(value)?; + let raw: RunEventRaw = serde_json::from_value(value)?; let body = serde_json::from_value(json!({ "event": raw.event, "properties": raw.properties, @@ -371,7 +371,7 @@ fn properties_from_body(body: &EventBody) -> Value { .unwrap_or_else(default_properties) } -impl Serialize for StoredEvent { +impl Serialize for RunEvent { fn serialize(&self, serializer: S) -> Result where S: Serializer, @@ -382,7 +382,7 @@ impl Serialize for StoredEvent { } } -impl<'de> Deserialize<'de> for StoredEvent { +impl<'de> Deserialize<'de> for RunEvent { fn deserialize(deserializer: D) -> Result where D: Deserializer<'de>, @@ -403,8 +403,8 @@ mod tests { use super::*; #[test] - fn stored_event_round_trips_json() { - let event = StoredEvent { + fn run_event_round_trips_json() { + let event = RunEvent { id: "evt_1".to_string(), ts: DateTime::parse_from_rfc3339("2026-04-04T12:00:00.000Z") .unwrap() @@ -448,13 +448,13 @@ mod tests { }; let value = event.to_value().unwrap(); - let parsed = StoredEvent::from_value(value).unwrap(); + let parsed = RunEvent::from_value(value).unwrap(); assert_eq!(parsed, event); } #[test] - fn stored_event_deserializes_adjacent_layout() { + fn run_event_deserializes_adjacent_layout() { let settings = Settings::default(); let graph = Graph { name: "test".to_string(), @@ -488,7 +488,7 @@ mod tests { } }); - let parsed = StoredEvent::from_value(line).unwrap(); + let parsed = RunEvent::from_value(line).unwrap(); assert!(matches!(parsed.body, EventBody::RunCreated(_))); } } diff --git a/lib/crates/fabro-types/src/stored_event/run.rs b/lib/crates/fabro-types/src/run_event/run.rs similarity index 99% rename from lib/crates/fabro-types/src/stored_event/run.rs rename to lib/crates/fabro-types/src/run_event/run.rs index 509f1e9d8..8937ae9f0 100644 --- a/lib/crates/fabro-types/src/stored_event/run.rs +++ b/lib/crates/fabro-types/src/run_event/run.rs @@ -95,7 +95,7 @@ pub struct RunNoticeProps { } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct StoredEventHeader { +pub struct RunEventHeader { pub id: String, pub ts: chrono::DateTime, pub run_id: RunId, diff --git a/lib/crates/fabro-types/src/stored_event/stage.rs b/lib/crates/fabro-types/src/run_event/stage.rs similarity index 100% rename from lib/crates/fabro-types/src/stored_event/stage.rs rename to lib/crates/fabro-types/src/run_event/stage.rs diff --git a/lib/crates/fabro-workflow/src/devcontainer_bridge.rs b/lib/crates/fabro-workflow/src/devcontainer_bridge.rs index 2b41334ce..b7558da27 100644 --- a/lib/crates/fabro-workflow/src/devcontainer_bridge.rs +++ b/lib/crates/fabro-workflow/src/devcontainer_bridge.rs @@ -4,7 +4,7 @@ use sha2::{Digest, Sha256}; use fabro_devcontainer::DevcontainerSpec; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use fabro_agent::sandbox::Sandbox; use fabro_sandbox::daytona::{DaytonaSnapshotConfig, DockerfileSource}; use futures::future::try_join_all; @@ -41,7 +41,7 @@ pub async fn run_devcontainer_lifecycle( return Ok(()); } - emitter.emit(&WorkflowRunEvent::DevcontainerLifecycleStarted { + emitter.emit(&Event::DevcontainerLifecycleStarted { phase: phase.to_string(), command_count: commands.len(), }); @@ -81,7 +81,7 @@ pub async fn run_devcontainer_lifecycle( let name = name.clone(); async move { let cmd_start = Instant::now(); - emitter.emit(&WorkflowRunEvent::DevcontainerLifecycleCommandStarted { + emitter.emit(&Event::DevcontainerLifecycleCommandStarted { phase: phase.clone(), command: name.clone(), index, @@ -97,7 +97,7 @@ pub async fn run_devcontainer_lifecycle( let cmd_duration = crate::millis_u64(cmd_start.elapsed()); if result.exit_code != 0 { emitter.emit( - &WorkflowRunEvent::DevcontainerLifecycleFailed { + &Event::DevcontainerLifecycleFailed { phase: phase.clone(), command: name.clone(), index, @@ -112,7 +112,7 @@ pub async fn run_devcontainer_lifecycle( ); } emitter.emit( - &WorkflowRunEvent::DevcontainerLifecycleCommandCompleted { + &Event::DevcontainerLifecycleCommandCompleted { phase: phase.clone(), command: name.clone(), index, @@ -130,7 +130,7 @@ pub async fn run_devcontainer_lifecycle( } let phase_duration = crate::millis_u64(phase_start.elapsed()); - emitter.emit(&WorkflowRunEvent::DevcontainerLifecycleCompleted { + emitter.emit(&Event::DevcontainerLifecycleCompleted { phase: phase.to_string(), duration_ms: phase_duration, }); @@ -145,7 +145,7 @@ async fn run_single_lifecycle_command( index: usize, timeout_ms: u64, ) -> anyhow::Result<()> { - emitter.emit(&WorkflowRunEvent::DevcontainerLifecycleCommandStarted { + emitter.emit(&Event::DevcontainerLifecycleCommandStarted { phase: phase.to_string(), command: command.to_string(), index, @@ -157,7 +157,7 @@ async fn run_single_lifecycle_command( .map_err(|e| anyhow::anyhow!("Devcontainer {phase} command failed: {e}"))?; let cmd_duration = crate::millis_u64(cmd_start.elapsed()); if result.exit_code != 0 { - emitter.emit(&WorkflowRunEvent::DevcontainerLifecycleFailed { + emitter.emit(&Event::DevcontainerLifecycleFailed { phase: phase.to_string(), command: command.to_string(), index, @@ -170,7 +170,7 @@ async fn run_single_lifecycle_command( result.stderr, ); } - emitter.emit(&WorkflowRunEvent::DevcontainerLifecycleCommandCompleted { + emitter.emit(&Event::DevcontainerLifecycleCommandCompleted { phase: phase.to_string(), command: command.to_string(), index, @@ -381,7 +381,7 @@ mod tests { #[tokio::test] async fn emits_started_and_completed_events() { let emitter = EventEmitter::default(); - let events = Arc::new(Mutex::new(Vec::::new())); + let events = Arc::new(Mutex::new(Vec::::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { events_clone.lock().unwrap().push(event.clone()); @@ -418,7 +418,7 @@ mod tests { #[tokio::test] async fn failed_command_emits_failed_and_returns_error() { let emitter = EventEmitter::default(); - let events = Arc::new(Mutex::new(Vec::::new())); + let events = Arc::new(Mutex::new(Vec::::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { events_clone.lock().unwrap().push(event.clone()); diff --git a/lib/crates/fabro-workflow/src/error.rs b/lib/crates/fabro-workflow/src/error.rs index f4b51e2e8..9dad4226f 100644 --- a/lib/crates/fabro-workflow/src/error.rs +++ b/lib/crates/fabro-workflow/src/error.rs @@ -1669,7 +1669,7 @@ mod tests { #[test] fn e2e_llm_error_to_outcome_to_event_preserves_classification() { - use crate::event::WorkflowRunEvent; + use crate::event::Event; // 1. Create SdkError → FabroError let sdk_err = SdkError::Provider { @@ -1688,7 +1688,7 @@ mod tests { // 3. Outcome → StageFailed event let failure = outcome.failure.clone().unwrap(); - let event = WorkflowRunEvent::StageFailed { + let event = Event::StageFailed { node_id: "code".into(), name: "code".into(), index: 0, @@ -1698,7 +1698,7 @@ mod tests { // 4. Verify classification survived all the way through match &event { - WorkflowRunEvent::StageFailed { failure, .. } => { + Event::StageFailed { failure, .. } => { assert_eq!(failure.category, FailureCategory::TransientInfra); } _ => panic!("expected StageFailed"), diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 20d143386..5b63a4561 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicI64, Ordering}; use anyhow::{Context, Result}; use chrono::{SecondsFormat, Utc}; use fabro_store::{EventPayload, SlateRunStore}; -use fabro_types::{RunId, StoredEvent}; +use fabro_types::{RunEvent, RunId}; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value, json}; use std::collections::BTreeMap; @@ -23,7 +23,7 @@ pub use fabro_types::{EventBody, RunNoticeLevel}; /// Events emitted during workflow run execution for observability. #[derive(Debug, Clone, Serialize, Deserialize)] #[allow(clippy::large_enum_variant)] -pub enum WorkflowRunEvent { +pub enum Event { RunCreated { run_id: RunId, settings: serde_json::Value, @@ -502,7 +502,7 @@ pub enum WorkflowRunEvent { }, } -impl WorkflowRunEvent { +impl Event { pub fn trace(&self) { use tracing::{debug, error, info, warn}; match self { @@ -1055,43 +1055,43 @@ impl WorkflowRunEvent { } } -pub fn event_name(event: &WorkflowRunEvent) -> &'static str { +pub fn event_name(event: &Event) -> &'static str { match event { - WorkflowRunEvent::RunCreated { .. } => "run.created", - WorkflowRunEvent::WorkflowRunStarted { .. } => "run.started", - WorkflowRunEvent::RunSubmitted { .. } => "run.submitted", - WorkflowRunEvent::RunStarting { .. } => "run.starting", - WorkflowRunEvent::RunRunning { .. } => "run.running", - WorkflowRunEvent::RunRemoving { .. } => "run.removing", - WorkflowRunEvent::RunRewound { .. } => "run.rewound", - WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed", - WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed", - WorkflowRunEvent::RunNotice { .. } => "run.notice", - WorkflowRunEvent::StageStarted { .. } => "stage.started", - WorkflowRunEvent::StageCompleted { .. } => "stage.completed", - WorkflowRunEvent::StageFailed { .. } => "stage.failed", - WorkflowRunEvent::StageRetrying { .. } => "stage.retrying", - WorkflowRunEvent::ParallelStarted { .. } => "parallel.started", - WorkflowRunEvent::ParallelBranchStarted { .. } => "parallel.branch.started", - WorkflowRunEvent::ParallelBranchCompleted { .. } => "parallel.branch.completed", - WorkflowRunEvent::ParallelCompleted { .. } => "parallel.completed", - WorkflowRunEvent::InterviewStarted { .. } => "interview.started", - WorkflowRunEvent::InterviewCompleted { .. } => "interview.completed", - WorkflowRunEvent::InterviewTimeout { .. } => "interview.timeout", - WorkflowRunEvent::CheckpointCompleted { .. } => "checkpoint.completed", - WorkflowRunEvent::CheckpointFailed { .. } => "checkpoint.failed", - WorkflowRunEvent::GitCommit { .. } => "git.commit", - WorkflowRunEvent::GitPush { .. } => "git.push", - WorkflowRunEvent::GitBranch { .. } => "git.branch", - WorkflowRunEvent::GitWorktreeAdd { .. } => "git.worktree.added", - WorkflowRunEvent::GitWorktreeRemove { .. } => "git.worktree.removed", - WorkflowRunEvent::GitFetch { .. } => "git.fetch", - WorkflowRunEvent::GitReset { .. } => "git.reset", - WorkflowRunEvent::EdgeSelected { .. } => "edge.selected", - WorkflowRunEvent::LoopRestart { .. } => "loop.restart", - WorkflowRunEvent::Prompt { .. } => "stage.prompt", - WorkflowRunEvent::PromptCompleted { .. } => "prompt.completed", - WorkflowRunEvent::Agent { event, .. } => match event { + Event::RunCreated { .. } => "run.created", + Event::WorkflowRunStarted { .. } => "run.started", + Event::RunSubmitted { .. } => "run.submitted", + Event::RunStarting { .. } => "run.starting", + Event::RunRunning { .. } => "run.running", + Event::RunRemoving { .. } => "run.removing", + Event::RunRewound { .. } => "run.rewound", + Event::WorkflowRunCompleted { .. } => "run.completed", + Event::WorkflowRunFailed { .. } => "run.failed", + Event::RunNotice { .. } => "run.notice", + Event::StageStarted { .. } => "stage.started", + Event::StageCompleted { .. } => "stage.completed", + Event::StageFailed { .. } => "stage.failed", + Event::StageRetrying { .. } => "stage.retrying", + Event::ParallelStarted { .. } => "parallel.started", + Event::ParallelBranchStarted { .. } => "parallel.branch.started", + Event::ParallelBranchCompleted { .. } => "parallel.branch.completed", + Event::ParallelCompleted { .. } => "parallel.completed", + Event::InterviewStarted { .. } => "interview.started", + Event::InterviewCompleted { .. } => "interview.completed", + Event::InterviewTimeout { .. } => "interview.timeout", + Event::CheckpointCompleted { .. } => "checkpoint.completed", + Event::CheckpointFailed { .. } => "checkpoint.failed", + Event::GitCommit { .. } => "git.commit", + Event::GitPush { .. } => "git.push", + Event::GitBranch { .. } => "git.branch", + Event::GitWorktreeAdd { .. } => "git.worktree.added", + Event::GitWorktreeRemove { .. } => "git.worktree.removed", + Event::GitFetch { .. } => "git.fetch", + Event::GitReset { .. } => "git.reset", + Event::EdgeSelected { .. } => "edge.selected", + Event::LoopRestart { .. } => "loop.restart", + Event::Prompt { .. } => "stage.prompt", + Event::PromptCompleted { .. } => "prompt.completed", + Event::Agent { event, .. } => match event { AgentEvent::SessionStarted { .. } => "agent.session.started", AgentEvent::SessionEnded => "agent.session.ended", AgentEvent::ProcessingEnd => "agent.processing.end", @@ -1120,9 +1120,9 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str { AgentEvent::McpServerReady { .. } => "agent.mcp.ready", AgentEvent::McpServerFailed { .. } => "agent.mcp.failed", }, - WorkflowRunEvent::SubgraphStarted { .. } => "subgraph.started", - WorkflowRunEvent::SubgraphCompleted { .. } => "subgraph.completed", - WorkflowRunEvent::Sandbox { event } => match event { + Event::SubgraphStarted { .. } => "subgraph.started", + Event::SubgraphCompleted { .. } => "subgraph.completed", + Event::Sandbox { event } => match event { SandboxEvent::Initializing { .. } => "sandbox.initializing", SandboxEvent::Ready { .. } => "sandbox.ready", SandboxEvent::InitializeFailed { .. } => "sandbox.failed", @@ -1139,40 +1139,38 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str { SandboxEvent::GitCloneCompleted { .. } => "sandbox.git.completed", SandboxEvent::GitCloneFailed { .. } => "sandbox.git.failed", }, - WorkflowRunEvent::SandboxInitialized { .. } => "sandbox.initialized", - WorkflowRunEvent::SetupStarted { .. } => "setup.started", - WorkflowRunEvent::SetupCommandStarted { .. } => "setup.command.started", - WorkflowRunEvent::SetupCommandCompleted { .. } => "setup.command.completed", - WorkflowRunEvent::SetupCompleted { .. } => "setup.completed", - WorkflowRunEvent::SetupFailed { .. } => "setup.failed", - WorkflowRunEvent::StallWatchdogTimeout { .. } => "watchdog.timeout", - WorkflowRunEvent::AssetCaptured { .. } => "asset.captured", - WorkflowRunEvent::SshAccessReady { .. } => "ssh.ready", - WorkflowRunEvent::Failover { .. } => "agent.failover", - WorkflowRunEvent::CliEnsureStarted { .. } => "cli.ensure.started", - WorkflowRunEvent::CliEnsureCompleted { .. } => "cli.ensure.completed", - WorkflowRunEvent::CliEnsureFailed { .. } => "cli.ensure.failed", - WorkflowRunEvent::CommandStarted { .. } => "command.started", - WorkflowRunEvent::CommandCompleted { .. } => "command.completed", - WorkflowRunEvent::AgentCliStarted { .. } => "agent.cli.started", - WorkflowRunEvent::AgentCliCompleted { .. } => "agent.cli.completed", - WorkflowRunEvent::PullRequestCreated { .. } => "pull_request.created", - WorkflowRunEvent::PullRequestFailed { .. } => "pull_request.failed", - WorkflowRunEvent::DevcontainerResolved { .. } => "devcontainer.resolved", - WorkflowRunEvent::DevcontainerLifecycleStarted { .. } => "devcontainer.lifecycle.started", - WorkflowRunEvent::DevcontainerLifecycleCommandStarted { .. } => { + Event::SandboxInitialized { .. } => "sandbox.initialized", + Event::SetupStarted { .. } => "setup.started", + Event::SetupCommandStarted { .. } => "setup.command.started", + Event::SetupCommandCompleted { .. } => "setup.command.completed", + Event::SetupCompleted { .. } => "setup.completed", + Event::SetupFailed { .. } => "setup.failed", + Event::StallWatchdogTimeout { .. } => "watchdog.timeout", + Event::AssetCaptured { .. } => "asset.captured", + Event::SshAccessReady { .. } => "ssh.ready", + Event::Failover { .. } => "agent.failover", + Event::CliEnsureStarted { .. } => "cli.ensure.started", + Event::CliEnsureCompleted { .. } => "cli.ensure.completed", + Event::CliEnsureFailed { .. } => "cli.ensure.failed", + Event::CommandStarted { .. } => "command.started", + Event::CommandCompleted { .. } => "command.completed", + Event::AgentCliStarted { .. } => "agent.cli.started", + Event::AgentCliCompleted { .. } => "agent.cli.completed", + Event::PullRequestCreated { .. } => "pull_request.created", + Event::PullRequestFailed { .. } => "pull_request.failed", + Event::DevcontainerResolved { .. } => "devcontainer.resolved", + Event::DevcontainerLifecycleStarted { .. } => "devcontainer.lifecycle.started", + Event::DevcontainerLifecycleCommandStarted { .. } => { "devcontainer.lifecycle.command.started" } - WorkflowRunEvent::DevcontainerLifecycleCommandCompleted { .. } => { + Event::DevcontainerLifecycleCommandCompleted { .. } => { "devcontainer.lifecycle.command.completed" } - WorkflowRunEvent::DevcontainerLifecycleCompleted { .. } => { - "devcontainer.lifecycle.completed" - } - WorkflowRunEvent::DevcontainerLifecycleFailed { .. } => "devcontainer.lifecycle.failed", - WorkflowRunEvent::RetroStarted { .. } => "retro.started", - WorkflowRunEvent::RetroCompleted { .. } => "retro.completed", - WorkflowRunEvent::RetroFailed { .. } => "retro.failed", + Event::DevcontainerLifecycleCompleted { .. } => "devcontainer.lifecycle.completed", + Event::DevcontainerLifecycleFailed { .. } => "devcontainer.lifecycle.failed", + Event::RetroStarted { .. } => "retro.started", + Event::RetroCompleted { .. } => "retro.completed", + Event::RetroFailed { .. } => "retro.failed", } } @@ -1223,9 +1221,9 @@ fn default_node_label(node_id: Option<&String>, node_label: Option) -> O node_label.or_else(|| node_id.cloned()) } -fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { +fn extract_run_event_fields(event: &Event) -> StoredEventFields { match event { - WorkflowRunEvent::RunCreated { .. } | WorkflowRunEvent::WorkflowRunStarted { .. } => { + Event::RunCreated { .. } | Event::WorkflowRunStarted { .. } => { let mut fields = tagged_variant_fields(event); fields.remove("run_id"); StoredEventFields { @@ -1236,7 +1234,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::WorkflowRunFailed { error, .. } => { + Event::WorkflowRunFailed { error, .. } => { let mut fields = tagged_variant_fields(event); fields.insert("error".to_string(), Value::String(error.to_string())); StoredEventFields { @@ -1247,7 +1245,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::StageCompleted { .. } | WorkflowRunEvent::StageFailed { .. } => { + Event::StageCompleted { .. } | Event::StageFailed { .. } => { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "node_id"); let node_label = @@ -1260,20 +1258,20 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::StageStarted { .. } - | WorkflowRunEvent::StageRetrying { .. } - | WorkflowRunEvent::CheckpointCompleted { .. } - | WorkflowRunEvent::CheckpointFailed { .. } - | WorkflowRunEvent::SubgraphStarted { .. } - | WorkflowRunEvent::SubgraphCompleted { .. } - | WorkflowRunEvent::AssetCaptured { .. } - | WorkflowRunEvent::PromptCompleted { .. } - | WorkflowRunEvent::ParallelStarted { .. } - | WorkflowRunEvent::ParallelCompleted { .. } - | WorkflowRunEvent::CommandStarted { .. } - | WorkflowRunEvent::CommandCompleted { .. } - | WorkflowRunEvent::AgentCliStarted { .. } - | WorkflowRunEvent::AgentCliCompleted { .. } => { + Event::StageStarted { .. } + | Event::StageRetrying { .. } + | Event::CheckpointCompleted { .. } + | Event::CheckpointFailed { .. } + | Event::SubgraphStarted { .. } + | Event::SubgraphCompleted { .. } + | Event::AssetCaptured { .. } + | Event::PromptCompleted { .. } + | Event::ParallelStarted { .. } + | Event::ParallelCompleted { .. } + | Event::CommandStarted { .. } + | Event::CommandCompleted { .. } + | Event::AgentCliStarted { .. } + | Event::AgentCliCompleted { .. } => { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "node_id"); let node_label = @@ -1286,7 +1284,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::Agent { + Event::Agent { session_id, parent_session_id, .. @@ -1312,7 +1310,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties, } } - WorkflowRunEvent::Sandbox { .. } => { + Event::Sandbox { .. } => { let mut fields = tagged_variant_fields(event); let properties = fields.remove("event").map_or_else( || Value::Object(Map::new()), @@ -1326,7 +1324,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties, } } - WorkflowRunEvent::GitCommit { .. } => { + Event::GitCommit { .. } => { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "node_id"); let node_label = default_node_label(node_id.as_ref(), None); @@ -1338,8 +1336,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::ParallelBranchStarted { .. } - | WorkflowRunEvent::ParallelBranchCompleted { .. } => { + Event::ParallelBranchStarted { .. } | Event::ParallelBranchCompleted { .. } => { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "branch"); let node_label = default_node_label(node_id.as_ref(), None); @@ -1351,10 +1348,10 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::Prompt { .. } - | WorkflowRunEvent::InterviewStarted { .. } - | WorkflowRunEvent::InterviewTimeout { .. } - | WorkflowRunEvent::Failover { .. } => { + Event::Prompt { .. } + | Event::InterviewStarted { .. } + | Event::InterviewTimeout { .. } + | Event::Failover { .. } => { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "stage"); let node_label = default_node_label(node_id.as_ref(), None); @@ -1366,7 +1363,7 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { properties: Value::Object(fields), } } - WorkflowRunEvent::StallWatchdogTimeout { .. } => { + Event::StallWatchdogTimeout { .. } => { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "node"); let node_label = default_node_label(node_id.as_ref(), None); @@ -1388,17 +1385,13 @@ fn extract_stored_event_fields(event: &WorkflowRunEvent) -> StoredEventFields { } } -pub fn to_stored_event(run_id: &RunId, event: &WorkflowRunEvent) -> StoredEvent { - to_stored_event_at(run_id, event, Utc::now()) +pub fn to_run_event(run_id: &RunId, event: &Event) -> RunEvent { + to_run_event_at(run_id, event, Utc::now()) } -pub fn to_stored_event_at( - run_id: &RunId, - event: &WorkflowRunEvent, - ts: chrono::DateTime, -) -> StoredEvent { - let fields = extract_stored_event_fields(event); - StoredEvent::from_value(json!({ +pub fn to_run_event_at(run_id: &RunId, event: &Event, ts: chrono::DateTime) -> RunEvent { + let fields = extract_run_event_fields(event); + RunEvent::from_value(json!({ "id": Uuid::now_v7().to_string(), "ts": ts.to_rfc3339_opts(SecondsFormat::Millis, true), "run_id": run_id.to_string(), @@ -1412,17 +1405,17 @@ pub fn to_stored_event_at( .expect("workflow event converts to stored event") } -pub fn build_redacted_event_payload(event: &StoredEvent, run_id: &RunId) -> Result { +pub fn build_redacted_event_payload(event: &RunEvent, run_id: &RunId) -> Result { let line = redacted_event_json(event)?; event_payload_from_redacted_json(&line, run_id) } -pub fn redacted_event_json(event: &StoredEvent) -> Result { +pub fn redacted_event_json(event: &RunEvent) -> Result { let line = serde_json::to_string(&normalized_event_value(event)?)?; Ok(redact_jsonl_line(&line)) } -fn normalized_event_value(event: &StoredEvent) -> Result { +fn normalized_event_value(event: &RunEvent) -> Result { let value = event.to_value()?; Ok(normalize_json_value(value)) } @@ -1450,12 +1443,8 @@ pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result Result<()> { - let stored = to_stored_event(run_id, event); +pub async fn append_event(run_store: &SlateRunStore, run_id: &RunId, event: &Event) -> Result<()> { + let stored = to_run_event(run_id, event); let payload = build_redacted_event_payload(&stored, run_id)?; run_store .append_event(&payload) @@ -1537,7 +1526,7 @@ fn epoch_millis() -> i64 { } /// Listener callback type for workflow run events. -type EventListener = Arc; +type EventListener = Arc; /// Callback-based event emitter for workflow run events. pub struct EventEmitter { @@ -1579,27 +1568,27 @@ impl EventEmitter { self.run_id } - pub fn on_event(&self, listener: impl Fn(&StoredEvent) + Send + Sync + 'static) { + pub fn on_event(&self, listener: impl Fn(&RunEvent) + Send + Sync + 'static) { self.listeners .lock() .expect("listeners lock poisoned") .push(Arc::new(listener)); } - pub fn emit(&self, event: &WorkflowRunEvent) { + pub fn emit(&self, event: &Event) { self.last_event_at.store(epoch_millis(), Ordering::Relaxed); event.trace(); - if let WorkflowRunEvent::WorkflowRunStarted { run_id, .. } = event { + if let Event::WorkflowRunStarted { run_id, .. } = event { debug_assert_eq!( *run_id, self.run_id, "workflow run started event must match emitter run_id" ); } - let stored = to_stored_event(&self.run_id, event); - self.dispatch_stored_event(&stored); + let stored = to_run_event(&self.run_id, event); + self.dispatch_run_event(&stored); } - pub(crate) fn dispatch_stored_event(&self, event: &StoredEvent) { + pub(crate) fn dispatch_run_event(&self, event: &RunEvent) { self.last_event_at.store(epoch_millis(), Ordering::Relaxed); // Clone the listener list so we don't hold the lock during dispatch. // This prevents deadlocks if a listener calls emit() reentrantly. @@ -1626,17 +1615,17 @@ impl EventEmitter { } /// Build a [`WorktreeEventCallback`] that forwards worktree lifecycle events as - /// [`WorkflowRunEvent`]s on this emitter. + /// [`Event`]s on this emitter. pub fn worktree_callback(self: Arc) -> WorktreeEventCallback { Arc::new(move |event| match event { WorktreeEvent::BranchCreated { branch, sha } => { - self.emit(&WorkflowRunEvent::GitBranch { branch, sha }); + self.emit(&Event::GitBranch { branch, sha }); } WorktreeEvent::WorktreeAdded { path, branch } => { - self.emit(&WorkflowRunEvent::GitWorktreeAdd { path, branch }); + self.emit(&Event::GitWorktreeAdd { path, branch }); } WorktreeEvent::WorktreeRemoved { path } => { - self.emit(&WorkflowRunEvent::GitWorktreeRemove { path }); + self.emit(&Event::GitWorktreeRemove { path }); } }) } @@ -1662,7 +1651,7 @@ mod tests { emitter.on_event(move |event| { received_clone.lock().unwrap().push(event.clone()); }); - emitter.emit(&WorkflowRunEvent::WorkflowRunStarted { + emitter.emit(&Event::WorkflowRunStarted { name: "test".to_string(), run_id: fixtures::RUN_1, base_branch: None, @@ -1685,10 +1674,10 @@ mod tests { } #[test] - fn stored_stage_completed_places_node_fields_in_header() { - let stored = to_stored_event( + fn run_event_stage_completed_places_node_fields_in_header() { + let stored = to_run_event( &fixtures::RUN_2, - &WorkflowRunEvent::StageCompleted { + &Event::StageCompleted { node_id: "plan".to_string(), name: "Plan".to_string(), index: 0, @@ -1722,10 +1711,10 @@ mod tests { } #[test] - fn stored_stage_completed_keeps_response_and_signature_snapshots() { - let stored = to_stored_event( + fn run_event_stage_completed_keeps_response_and_signature_snapshots() { + let stored = to_run_event( &fixtures::RUN_2, - &WorkflowRunEvent::StageCompleted { + &Event::StageCompleted { node_id: "plan".to_string(), name: "Plan".to_string(), index: 0, @@ -1755,10 +1744,10 @@ mod tests { } #[test] - fn stored_stage_failure_keeps_failure_detail() { - let stored = to_stored_event( + fn run_event_stage_failure_keeps_failure_detail() { + let stored = to_run_event( &fixtures::RUN_3, - &WorkflowRunEvent::StageFailed { + &Event::StageFailed { node_id: "code".to_string(), name: "Code".to_string(), index: 1, @@ -1780,10 +1769,10 @@ mod tests { } #[test] - fn stored_agent_tool_started_moves_session_metadata_to_header() { - let stored = to_stored_event( + fn run_event_agent_tool_started_moves_session_metadata_to_header() { + let stored = to_run_event( &fixtures::RUN_4, - &WorkflowRunEvent::Agent { + &Event::Agent { stage: "code".to_string(), visit: 2, event: AgentEvent::ToolCallStarted { @@ -1807,10 +1796,10 @@ mod tests { } #[test] - fn stored_sandbox_event_keeps_properties_nested() { - let stored = to_stored_event( + fn run_event_sandbox_event_keeps_properties_nested() { + let stored = to_run_event( &fixtures::RUN_5, - &WorkflowRunEvent::Sandbox { + &Event::Sandbox { event: SandboxEvent::Ready { provider: "daytona".to_string(), duration_ms: 2500, @@ -1829,10 +1818,10 @@ mod tests { } #[test] - fn stored_workflow_failure_uses_display_error() { - let stored = to_stored_event( + fn run_event_workflow_failure_uses_display_error() { + let stored = to_run_event( &fixtures::RUN_6, - &WorkflowRunEvent::WorkflowRunFailed { + &Event::WorkflowRunFailed { error: FabroError::handler("boom"), duration_ms: 900, reason: Some(StatusReason::WorkflowError), @@ -1846,16 +1835,16 @@ mod tests { } #[tokio::test] - async fn append_workflow_event_writes_store_event_shape() { + async fn append_event_writes_store_event_shape() { let store = fabro_store::SlateStore::new( std::sync::Arc::new(object_store::memory::InMemory::new()), "", std::time::Duration::from_millis(1), ); let run_store = store.create_run(&fixtures::RUN_7).await.unwrap(); - let stored = to_stored_event( + let stored = to_run_event( &fixtures::RUN_7, - &WorkflowRunEvent::RunNotice { + &Event::RunNotice { level: RunNoticeLevel::Warn, code: "example".to_string(), message: "notice".to_string(), @@ -1877,9 +1866,9 @@ mod tests { #[test] fn build_redacted_event_payload_requires_id() { - let stored = to_stored_event( + let stored = to_run_event( &fixtures::RUN_8, - &WorkflowRunEvent::RetroStarted { + &Event::RetroStarted { prompt: Some("Analyze the run".to_string()), provider: None, model: None, @@ -1897,7 +1886,7 @@ mod tests { #[test] fn event_name_matches_new_dot_notation() { assert_eq!( - event_name(&WorkflowRunEvent::RetroStarted { + event_name(&Event::RetroStarted { prompt: None, provider: None, model: None, @@ -1905,14 +1894,14 @@ mod tests { "retro.started" ); assert_eq!( - event_name(&WorkflowRunEvent::ParallelBranchStarted { + event_name(&Event::ParallelBranchStarted { branch: "fork".to_string(), index: 0, }), "parallel.branch.started" ); assert_eq!( - event_name(&WorkflowRunEvent::Agent { + event_name(&Event::Agent { stage: "code".to_string(), visit: 1, event: AgentEvent::SubAgentSpawned { diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index f0569e7ce..e14d1c8ca 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -411,14 +411,14 @@ mod tests { #[tokio::test] async fn scan_node_files_from_state_reconstructs_allowlisted_entries() { - use crate::event::{WorkflowRunEvent, append_workflow_event}; + use crate::event::{Event, append_event}; let store = test_store(); let run = store.create_run(&fixtures::RUN_1).await.unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::Prompt { + &Event::Prompt { stage: "work".into(), visit: 2, text: "hello".into(), @@ -429,10 +429,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::PromptCompleted { + &Event::PromptCompleted { node_id: "work".into(), response: "world".into(), model: "gpt-5.4".into(), @@ -442,10 +442,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::StageCompleted { + &Event::StageCompleted { node_id: "work".into(), name: "Work".into(), index: 2, @@ -470,10 +470,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::CommandStarted { + &Event::CommandStarted { node_id: "work".into(), script: "echo hi".into(), command: "echo hi".into(), @@ -483,10 +483,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::CommandCompleted { + &Event::CommandCompleted { node_id: "work".into(), stdout: "hi\n".into(), stderr: String::new(), @@ -497,10 +497,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::ParallelCompleted { + &Event::ParallelCompleted { node_id: "work".into(), visit: 2, duration_ms: 100, @@ -511,10 +511,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run, &fixtures::RUN_1, - &WorkflowRunEvent::CheckpointCompleted { + &Event::CheckpointCompleted { node_id: "work".into(), status: "success".into(), current_node: "work".into(), diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 7680d290a..03f601db4 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -10,7 +10,7 @@ use fabro_types::RunId; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use crate::outcome::{ FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus, StageUsage, }; @@ -256,7 +256,7 @@ impl Handler for AgentHandler { .map(String::from) .or_else(|| Some(Provider::default_from_env().as_str().to_string())); let prompt_model = node.model().map(String::from); - services.emitter.emit(&WorkflowRunEvent::Prompt { + services.emitter.emit(&Event::Prompt { stage: node.id.clone(), visit, text: prompt.clone(), @@ -329,7 +329,7 @@ impl Handler for AgentHandler { .map(String::from) .or_else(|| Some(Provider::default_from_env().as_str().to_string())) .unwrap_or_default(); - services.emitter.emit(&WorkflowRunEvent::PromptCompleted { + services.emitter.emit(&Event::PromptCompleted { node_id: node.id.clone(), response: response_text.clone(), model: response_model, @@ -709,7 +709,7 @@ mod tests { _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { - emitter.emit(&crate::event::WorkflowRunEvent::Agent { + emitter.emit(&crate::event::Event::Agent { stage: node.id.clone(), visit: u32::try_from(crate::run_dir::visit_from_context(context)) .unwrap_or(u32::MAX), diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 8eb437f24..af65d97bf 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -3,7 +3,7 @@ use std::path::Path; use crate::context::Context; use crate::context::keys; use crate::error::FabroError; -use crate::event::WorkflowRunEvent; +use crate::event::Event; use crate::outcome::{Outcome, OutcomeExt}; use async_trait::async_trait; use fabro_graphviz::graph::{Graph, Node}; @@ -90,7 +90,7 @@ impl Handler for CommandHandler { } else { script.to_string() }; - services.emitter.emit(&WorkflowRunEvent::CommandStarted { + services.emitter.emit(&Event::CommandStarted { node_id: node.id.clone(), script: script.to_string(), command: command.clone(), @@ -113,7 +113,7 @@ impl Handler for CommandHandler { .await .map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?; - services.emitter.emit(&WorkflowRunEvent::CommandCompleted { + services.emitter.emit(&Event::CommandCompleted { node_id: node.id.clone(), stdout: result.stdout.clone(), stderr: result.stderr.clone(), diff --git a/lib/crates/fabro-workflow/src/handler/fan_in.rs b/lib/crates/fabro-workflow/src/handler/fan_in.rs index 52e1d3ebc..3989aaa4b 100644 --- a/lib/crates/fabro-workflow/src/handler/fan_in.rs +++ b/lib/crates/fabro-workflow/src/handler/fan_in.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use crate::context::Context; use crate::context::keys; use crate::error::FabroError; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use crate::outcome::{Outcome, OutcomeExt}; use crate::run_dir::visit_from_context; use crate::sandbox_git::git_merge_ff_only; @@ -233,7 +233,7 @@ async fn llm_evaluate( let visit_u32 = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); - emitter.emit(&WorkflowRunEvent::Prompt { + emitter.emit(&Event::Prompt { stage: node_id.to_string(), visit: visit_u32, text: full_prompt.clone(), @@ -269,7 +269,7 @@ async fn llm_evaluate( .unwrap_or_else(|| "unknown".to_string()); let response_text = serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); - emitter.emit(&WorkflowRunEvent::PromptCompleted { + emitter.emit(&Event::PromptCompleted { node_id: node_id.to_string(), response: response_text.clone(), model: String::new(), @@ -283,7 +283,7 @@ async fn llm_evaluate( }) } Ok(CodergenResult::Text { text, .. }) => { - emitter.emit(&WorkflowRunEvent::PromptCompleted { + emitter.emit(&Event::PromptCompleted { node_id: node_id.to_string(), response: text.clone(), model: String::new(), diff --git a/lib/crates/fabro-workflow/src/handler/human.rs b/lib/crates/fabro-workflow/src/handler/human.rs index 8f07adb56..1cdb3eec8 100644 --- a/lib/crates/fabro-workflow/src/handler/human.rs +++ b/lib/crates/fabro-workflow/src/handler/human.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use crate::context::Context; use crate::context::keys; use crate::error::FabroError; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use crate::millis_u64; use crate::outcome::{Outcome, OutcomeExt}; use fabro_graphviz::graph::{Graph, Node}; @@ -85,7 +85,7 @@ impl HumanHandler { self } - fn emit(&self, event: &WorkflowRunEvent) { + fn emit(&self, event: &Event) { if let Some(emitter) = &self.emitter { emitter.emit(event); } @@ -203,7 +203,7 @@ impl Handler for HumanHandler { // 3. Present to interviewer let question_text = node.label().to_string(); - self.emit(&WorkflowRunEvent::InterviewStarted { + self.emit(&Event::InterviewStarted { question: question_text.clone(), stage: node.id.clone(), question_type: question.question_type.to_string(), @@ -213,7 +213,7 @@ impl Handler for HumanHandler { // 4. Handle timeout if answer.value == AnswerValue::Timeout { - self.emit(&WorkflowRunEvent::InterviewTimeout { + self.emit(&Event::InterviewTimeout { question: question_text, stage: node.id.clone(), duration_ms: millis_u64(interview_start.elapsed()), @@ -243,7 +243,7 @@ impl Handler for HumanHandler { } // Emit interview completed for successful interactions - self.emit(&WorkflowRunEvent::InterviewCompleted { + self.emit(&Event::InterviewCompleted { question: question_text, answer: answer_text(&answer), duration_ms: millis_u64(interview_start.elapsed()), diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 70dd2318b..4eae3e871 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -19,7 +19,7 @@ use super::super::agent::{CodergenBackend, CodergenResult}; use crate::context::keys::Fidelity; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use crate::outcome::StageUsage; use crate::outcome::compute_stage_cost; use crate::run_dir::visit_from_context; @@ -104,7 +104,7 @@ fn spawn_event_forwarder( if !event.event.is_streaming_noise() && !matches!(&event.event, AgentEvent::ProcessingEnd) { - emitter.emit(&WorkflowRunEvent::Agent { + emitter.emit(&Event::Agent { stage: node_id.clone(), visit, event: event.event.clone(), @@ -493,7 +493,7 @@ impl CodergenBackend for AgentApiBackend { let mut succeeded = false; for target in &self.fallback_chain { - emitter.emit(&WorkflowRunEvent::Failover { + emitter.emit(&Event::Failover { stage: node.id.clone(), from_provider: from_provider.clone(), from_model: from_model.clone(), diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 25b58a589..ead561430 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -10,7 +10,7 @@ use tokio::time::sleep; use super::super::agent::{CodergenBackend, CodergenResult}; use crate::context::Context; use crate::error::FabroError; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use crate::outcome::StageUsage; use crate::outcome::compute_stage_cost; use crate::run_dir::visit_from_context; @@ -73,7 +73,7 @@ async fn ensure_cli( let cli_name = cli.name(); let provider_str = provider.as_str(); - emitter.emit(&WorkflowRunEvent::CliEnsureStarted { + emitter.emit(&Event::CliEnsureStarted { cli_name: cli_name.to_string(), provider: provider_str.to_string(), }); @@ -92,7 +92,7 @@ async fn ensure_cli( if version_check.exit_code == 0 { let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX); - emitter.emit(&WorkflowRunEvent::CliEnsureCompleted { + emitter.emit(&Event::CliEnsureCompleted { cli_name: cli_name.to_string(), provider: provider_str.to_string(), already_installed: true, @@ -135,7 +135,7 @@ async fn ensure_cli( "{cli_name} install exited with code {}: {detail}", install_result.exit_code ); - emitter.emit(&WorkflowRunEvent::CliEnsureFailed { + emitter.emit(&Event::CliEnsureFailed { cli_name: cli_name.to_string(), provider: provider_str.to_string(), error: error_msg.clone(), @@ -145,7 +145,7 @@ async fn ensure_cli( } let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX); - emitter.emit(&WorkflowRunEvent::CliEnsureCompleted { + emitter.emit(&Event::CliEnsureCompleted { cli_name: cli_name.to_string(), provider: provider_str.to_string(), already_installed: false, @@ -496,7 +496,7 @@ impl CodergenBackend for AgentCliBackend { ensure_cli(cli, provider, sandbox, emitter).await?; let command = cli_command_for_provider(provider, model, &prompt_path); - emitter.emit(&WorkflowRunEvent::AgentCliStarted { + emitter.emit(&Event::AgentCliStarted { node_id: node.id.clone(), visit: current_visit(_context), mode: "cli".to_string(), @@ -620,7 +620,7 @@ impl CodergenBackend for AgentCliBackend { timed_out: false, duration_ms, }; - emitter.emit(&WorkflowRunEvent::AgentCliCompleted { + emitter.emit(&Event::AgentCliCompleted { node_id: node.id.clone(), stdout: result.stdout.clone(), stderr: result.stderr.clone(), diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 8994348dc..9c410bd91 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -10,7 +10,7 @@ use tokio::sync::Semaphore; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::WorkflowRunEvent; +use crate::event::Event; use crate::git::sanitize_ref_component; use crate::hook_context::set_hook_node; use crate::millis_u64; @@ -150,7 +150,7 @@ impl Handler for ParallelHandler { .unwrap_or("wait_all"), ); - services.emitter.emit(&WorkflowRunEvent::ParallelStarted { + services.emitter.emit(&Event::ParallelStarted { node_id: node.id.clone(), visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), branch_count: branches.len(), @@ -287,7 +287,7 @@ impl Handler for ParallelHandler { .await .map_err(|e| FabroError::handler(format!("semaphore error: {e}")))?; - emitter.emit(&WorkflowRunEvent::ParallelBranchStarted { + emitter.emit(&Event::ParallelBranchStarted { branch: setup.target_id.clone(), index: setup.branch_index, }); @@ -298,7 +298,7 @@ impl Handler for ParallelHandler { "branch target node not found: {}", setup.target_id )); - emitter.emit(&WorkflowRunEvent::ParallelBranchCompleted { + emitter.emit(&Event::ParallelBranchCompleted { branch: setup.target_id.clone(), index: setup.branch_index, duration_ms: millis_u64(branch_start.elapsed()), @@ -367,7 +367,7 @@ impl Handler for ParallelHandler { match sha_result { Ok(r) if r.exit_code == 0 => { let sha = r.stdout.trim().to_string(); - emitter.emit(&WorkflowRunEvent::GitCommit { + emitter.emit(&Event::GitCommit { node_id: Some(setup.target_id.clone()), sha: sha.clone(), }); @@ -379,7 +379,7 @@ impl Handler for ParallelHandler { None }; - emitter.emit(&WorkflowRunEvent::ParallelBranchCompleted { + emitter.emit(&Event::ParallelBranchCompleted { branch: setup.target_id.clone(), index: setup.branch_index, duration_ms: millis_u64(branch_start.elapsed()), @@ -432,7 +432,7 @@ impl Handler for ParallelHandler { git_remove_worktree(&*services.sandbox, &wt_str).await; services .emitter - .emit(&WorkflowRunEvent::GitWorktreeRemove { path: wt_str }); + .emit(&Event::GitWorktreeRemove { path: wt_str }); } } @@ -478,7 +478,7 @@ impl Handler for ParallelHandler { context.set(keys::PARALLEL_RESULTS, serde_json::json!(results_json)); context.set(keys::PARALLEL_BRANCH_COUNT, serde_json::json!(total)); - services.emitter.emit(&WorkflowRunEvent::ParallelCompleted { + services.emitter.emit(&Event::ParallelCompleted { node_id: node.id.clone(), visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), duration_ms: millis_u64(parallel_start.elapsed()), diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 8aa8ab7e9..9db3e5dac 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -3,7 +3,7 @@ use std::path::Path; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::WorkflowRunEvent; +use crate::event::Event; use crate::outcome::Outcome; use crate::run_dir::visit_from_context; use async_trait::async_trait; @@ -91,7 +91,7 @@ impl Handler for PromptHandler { .map(String::from) .or_else(|| Some(Provider::default_from_env().as_str().to_string())); let prompt_model = node.model().map(String::from); - services.emitter.emit(&WorkflowRunEvent::Prompt { + services.emitter.emit(&Event::Prompt { stage: node.id.clone(), visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), text: prompt.clone(), @@ -140,7 +140,7 @@ impl Handler for PromptHandler { .or_else(|| Some(Provider::default_from_env().as_str().to_string())) .unwrap_or_default(); - services.emitter.emit(&WorkflowRunEvent::PromptCompleted { + services.emitter.emit(&Event::PromptCompleted { node_id: node.id.clone(), response: response_text.clone(), model: response_model, diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index 39348c861..199b29a7b 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -10,7 +10,7 @@ use fabro_core::state::ExecutionState; use crate::artifact::{ArtifactStore, offload_large_values, sync_artifacts_to_env}; use crate::asset_snapshot::collect_assets; -use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter, RunNoticeLevel}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::StageUsage; @@ -104,7 +104,7 @@ impl RunLifecycle for ArtifactLifecycle { match collect_assets(&*self.sandbox, &asset_capture_dir, &self.asset_globs, epoch).await { Ok(summary) if summary.files_copied > 0 => { for asset in &summary.captured_assets { - self.emitter.emit(&WorkflowRunEvent::AssetCaptured { + self.emitter.emit(&Event::AssetCaptured { node_id: node_id.to_string(), attempt: ctx.attempt, node_slug: node_slug.clone(), @@ -118,7 +118,7 @@ impl RunLifecycle for ArtifactLifecycle { } Ok(_) => {} // no files collected Err(e) => { - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "asset_collection_failed".to_string(), message: format!("[node: {node_id}] asset collection failed: {e}"), @@ -141,7 +141,7 @@ impl RunLifecycle for ArtifactLifecycle { { let store = self.artifact_store.lock().unwrap(); if let Err(e) = offload_large_values(&mut result.outcome.context_updates, &store) { - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "artifact_offload_failed".to_string(), message: format!("[node: {node_id}] artifact offload failed: {e}"), @@ -153,7 +153,7 @@ impl RunLifecycle for ArtifactLifecycle { if let Err(e) = sync_artifacts_to_env(&mut result.outcome.context_updates, &*self.sandbox).await { - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "artifact_sync_failed".to_string(), message: format!("[node: {node_id}] artifact sync failed: {e}"), diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 27ba53629..38e679e04 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -17,7 +17,7 @@ use super::git::GitCheckpointResult; use crate::artifact::ArtifactStore; use crate::context; use crate::error::FabroError; -use crate::event::{EventEmitter, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::{ @@ -89,7 +89,7 @@ impl RunLifecycle for EventLifecycle { let mut restarted = self.restarted_from.lock().unwrap(); if let Some((from_node, to_node)) = restarted.take() { self.emitter - .emit(&WorkflowRunEvent::LoopRestart { from_node, to_node }); + .emit(&Event::LoopRestart { from_node, to_node }); } } @@ -97,7 +97,7 @@ impl RunLifecycle for EventLifecycle { *self.run_start.lock().unwrap() = Instant::now(); // Emit WorkflowRunStarted - self.emitter.emit(&WorkflowRunEvent::WorkflowRunStarted { + self.emitter.emit(&Event::WorkflowRunStarted { name: self.graph_name.clone(), run_id: self.run_id, base_branch: self.base_branch.clone(), @@ -106,8 +106,7 @@ impl RunLifecycle for EventLifecycle { worktree_dir: self.worktree_dir.clone(), goal: self.goal.clone(), }); - self.emitter - .emit(&WorkflowRunEvent::RunRunning { reason: None }); + self.emitter.emit(&Event::RunRunning { reason: None }); Ok(()) } @@ -125,7 +124,7 @@ impl RunLifecycle for EventLifecycle { let stage_index = state.stage_index; let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); - self.emitter.emit(&WorkflowRunEvent::StageStarted { + self.emitter.emit(&Event::StageStarted { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, @@ -133,7 +132,7 @@ impl RunLifecycle for EventLifecycle { attempt: 1, max_attempts: 1, }); - self.emitter.emit(&WorkflowRunEvent::StageCompleted { + self.emitter.emit(&Event::StageCompleted { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, @@ -166,7 +165,7 @@ impl RunLifecycle for EventLifecycle { state: &WfRunState, ) -> CoreResult>> { let gv = ctx.node.inner(); - self.emitter.emit(&WorkflowRunEvent::StageStarted { + self.emitter.emit(&Event::StageStarted { node_id: gv.id.clone(), name: gv.label().to_string(), index: state.stage_index, @@ -187,7 +186,7 @@ impl RunLifecycle for EventLifecycle { let outcome = &ctx.result.outcome; let stage_index = state.stage_index; - self.emitter.emit(&WorkflowRunEvent::StageFailed { + self.emitter.emit(&Event::StageFailed { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, @@ -197,7 +196,7 @@ impl RunLifecycle for EventLifecycle { will_retry: true, }); - self.emitter.emit(&WorkflowRunEvent::StageRetrying { + self.emitter.emit(&Event::StageRetrying { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, @@ -229,7 +228,7 @@ impl RunLifecycle for EventLifecycle { snapshot_failure_signatures(&self.circuit_breaker); if outcome.status == StageStatus::Fail { - self.emitter.emit(&WorkflowRunEvent::StageFailed { + self.emitter.emit(&Event::StageFailed { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, @@ -239,7 +238,7 @@ impl RunLifecycle for EventLifecycle { will_retry: false, }); } else { - self.emitter.emit(&WorkflowRunEvent::StageCompleted { + self.emitter.emit(&Event::StageCompleted { node_id: gv.id.clone(), name: gv.label().to_string(), index: stage_index, @@ -294,7 +293,7 @@ impl RunLifecycle for EventLifecycle { .edge .as_ref() .and_then(|e| e.inner().condition().map(String::from)); - self.emitter.emit(&WorkflowRunEvent::EdgeSelected { + self.emitter.emit(&Event::EdgeSelected { from_node: ctx.from.to_string(), to_node: ctx.to.to_string(), label, @@ -325,7 +324,7 @@ impl RunLifecycle for EventLifecycle { let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); - self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted { + self.emitter.emit(&Event::CheckpointCompleted { node_id: node.id().to_string(), status, current_node: node.id().to_string(), @@ -364,13 +363,13 @@ impl RunLifecycle for EventLifecycle { // Emit GitCommit + GitPush events if git produced results if let Some(ref result) = git_result { if let Some(ref sha) = result.commit_sha { - self.emitter.emit(&WorkflowRunEvent::GitCommit { + self.emitter.emit(&Event::GitCommit { node_id: Some(node.id().to_string()), sha: sha.clone(), }); } for (branch, success) in &result.push_results { - self.emitter.emit(&WorkflowRunEvent::GitPush { + self.emitter.emit(&Event::GitPush { branch: branch.clone(), success: *success, }); @@ -404,7 +403,7 @@ impl RunLifecycle for EventLifecycle { .reduce(|a, b| a + b); if outcome.status == StageStatus::Success || outcome.status == StageStatus::PartialSuccess { - self.emitter.emit(&WorkflowRunEvent::WorkflowRunCompleted { + self.emitter.emit(&Event::WorkflowRunCompleted { duration_ms, artifact_count, status: outcome.status.to_string(), @@ -422,7 +421,7 @@ impl RunLifecycle for EventLifecycle { .failure .as_ref() .map_or_else(|| "run failed".to_string(), |f| f.message.clone()); - self.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed { + self.emitter.emit(&Event::WorkflowRunFailed { error: FabroError::engine(error_msg), duration_ms, reason: Some(StatusReason::WorkflowError), diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 9f00c8ecc..022463915 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -14,7 +14,7 @@ use fabro_core::outcome::NodeResult; use fabro_core::state::ExecutionState; use crate::artifact::ArtifactStore; -use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter, RunNoticeLevel}; use crate::git::MetadataStore; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; @@ -180,7 +180,7 @@ impl RunLifecycle for GitLifecycle { match store.write_checkpoint(&self.run_id.to_string(), &cp_json, &extra_refs) { Ok(sha) => Some(sha), Err(e) => { - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "checkpoint_metadata_write_failed".to_string(), message: format!( @@ -285,7 +285,7 @@ impl RunLifecycle for GitLifecycle { } Ok(_) => {} Err(err) => { - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "git_diff_failed".to_string(), message: format!("[node: {node_id}] git diff failed: {err}"), @@ -299,7 +299,7 @@ impl RunLifecycle for GitLifecycle { } Err(e) => { // Emit CheckpointFailed and return error - self.emitter.emit(&WorkflowRunEvent::CheckpointFailed { + self.emitter.emit(&Event::CheckpointFailed { node_id: node_id.to_string(), error: e.clone(), }); @@ -327,7 +327,7 @@ impl RunLifecycle for GitLifecycle { Ok(patch) if !patch.is_empty() => { *self.final_patch.lock().unwrap() = Some(patch.clone()); if let Err(err) = fs::write(self.run_dir.join("final.patch"), patch).await { - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "final_patch_write_failed".to_string(), message: format!("failed to write final.patch: {err}"), @@ -339,7 +339,7 @@ impl RunLifecycle for GitLifecycle { } Err(err) => { *self.final_patch.lock().unwrap() = None; - self.emitter.emit(&WorkflowRunEvent::RunNotice { + self.emitter.emit(&Event::RunNotice { level: RunNoticeLevel::Warn, code: "git_diff_failed".to_string(), message: format!("final diff failed: {err}"), diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index a8f13514d..d5b13b1ab 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -17,9 +17,7 @@ use crate::transforms::{Transform, expand_vars}; use fabro_sandbox::daytona::detect_repo_info; use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow}; -use crate::event::{ - WorkflowRunEvent, append_workflow_event, normalize_json_value, to_stored_event_at, -}; +use crate::event::{Event, append_event, normalize_json_value, to_run_event_at}; #[derive(Clone, Debug)] pub struct CreateRunInput { @@ -136,9 +134,9 @@ async fn persist_created_run( .map_err(|_| FabroError::engine(err.to_string()))?, }; - let stored = to_stored_event_at( + let stored = to_run_event_at( &record.run_id, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: record.run_id, settings: normalize_json_value( serde_json::to_value(&record.settings) @@ -174,10 +172,10 @@ async fn persist_created_run( .await .map(|_| ()) .map_err(store_error)?; - append_workflow_event( + append_event( &run_store, &record.run_id, - &WorkflowRunEvent::RunSubmitted { reason: None }, + &Event::RunSubmitted { reason: None }, ) .await .map_err(store_error) diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index fa21cc5d2..a9db0b46b 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -343,7 +343,7 @@ mod tests { use std::time::Duration; use super::*; - use crate::event::{WorkflowRunEvent, append_workflow_event}; + use crate::event::{Event, append_event}; use crate::operations::test_support::{make_checkpoint_json, temp_repo, test_sig}; use crate::records::Checkpoint; @@ -433,10 +433,10 @@ mod tests { ) -> DurableRunStore { let run_store = store.create_run(&run_id).await.unwrap(); let run_record = sample_run_record(run_id, host_repo_path); - append_workflow_event( + append_event( &run_store, &run_id, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -458,10 +458,10 @@ mod tests { async fn append_start_event(run_store: &DurableRunStore, run_id: RunId) { let start = sample_start_record(run_id); - append_workflow_event( + append_event( run_store, &run_id, - &WorkflowRunEvent::WorkflowRunStarted { + &Event::WorkflowRunStarted { name: "test".to_string(), run_id, base_branch: None, @@ -477,10 +477,10 @@ mod tests { async fn append_sandbox_event(run_store: &DurableRunStore, run_id: RunId) { let sandbox = sample_sandbox_record(); - append_workflow_event( + append_event( run_store, &run_id, - &WorkflowRunEvent::SandboxInitialized { + &Event::SandboxInitialized { provider: sandbox.provider, working_directory: sandbox.working_directory, identifier: sandbox.identifier, @@ -497,10 +497,10 @@ mod tests { run_id: RunId, checkpoint: Checkpoint, ) { - append_workflow_event( + append_event( run_store, &run_id, - &WorkflowRunEvent::CheckpointCompleted { + &Event::CheckpointCompleted { node_id: checkpoint.current_node.clone(), status: "success".to_string(), current_node: checkpoint.current_node.clone(), @@ -536,10 +536,10 @@ mod tests { node: &StageId, text: &str, ) { - append_workflow_event( + append_event( run_store, &run_id, - &WorkflowRunEvent::Prompt { + &Event::Prompt { stage: node.node_id().to_string(), visit: node.visit(), text: text.to_string(), diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs index 966a1edda..99ffdf30b 100644 --- a/lib/crates/fabro-workflow/src/operations/resume.rs +++ b/lib/crates/fabro-workflow/src/operations/resume.rs @@ -3,7 +3,7 @@ use std::path::Path; use fabro_store::RuntimeState; use crate::error::FabroError; -use crate::event::{WorkflowRunEvent, append_workflow_event}; +use crate::event::{Event, append_event}; use crate::outcome::StageStatus; use crate::run_status::RunStatus; @@ -40,10 +40,10 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result Result<(), FabroError> { let message = error.to_string(); - if let Err(err) = append_workflow_event( + if let Err(err) = append_event( run_store, &run_id, - &WorkflowRunEvent::WorkflowRunFailed { + &Event::WorkflowRunFailed { error: error.clone(), duration_ms: 0, reason: Some(reason), @@ -776,12 +776,12 @@ async fn persist_detached_failure( tracing::warn!(error = %err, "Failed to append detached failure event"); } - let event = WorkflowRunEvent::RunNotice { + let event = Event::RunNotice { level: RunNoticeLevel::Error, code: format!("{phase}_failed"), message: message.clone(), }; - let stored = to_stored_event(&run_id, &event); + let stored = to_run_event(&run_id, &event); let line = redacted_event_json(&stored).map_err(|err| FabroError::Io(err.to_string()))?; match event_payload_from_redacted_json(&line, &run_id) { Ok(payload) => { @@ -905,7 +905,7 @@ mod tests { && event.node_id.as_deref() == Some("start") { injected.store(true, Ordering::SeqCst); - emitter_for_injection.emit(&WorkflowRunEvent::CheckpointCompleted { + emitter_for_injection.emit(&Event::CheckpointCompleted { node_id: "start".to_string(), status: "success".to_string(), current_node: "start".to_string(), @@ -1014,10 +1014,10 @@ mod tests { restart_failure_signatures: HashMap::new(), node_visits: HashMap::new(), }; - append_workflow_event( + append_event( &services.run_store, &services.run_id, - &WorkflowRunEvent::CheckpointCompleted { + &Event::CheckpointCompleted { node_id: checkpoint.current_node.clone(), status: checkpoint .node_outcomes diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index b4083b33f..eb37f30c0 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -9,7 +9,7 @@ use tokio_util::sync::CancellationToken; use crate::context::{self, Context}; use crate::error::FabroError; -use crate::event::WorkflowRunEvent; +use crate::event::Event; use crate::graph::WorkflowGraph; use crate::handler::EngineServices; use crate::lifecycle::WorkflowLifecycle; @@ -295,7 +295,7 @@ pub async fn execute(init: Initialized) -> Executed { Err(fabro_core::CoreError::StallTimeout { node_id }) => { let stall_timeout = graph.stall_timeout().unwrap_or_default(); let idle_secs = stall_timeout.as_secs(); - emitter.emit(&WorkflowRunEvent::StallWatchdogTimeout { + emitter.emit(&Event::StallWatchdogTimeout { node: node_id.clone(), idle_seconds: idle_secs, }); diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index b65154d28..6344492ca 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -894,7 +894,7 @@ async fn retry_emits_stage_started_per_attempt() { g.edges.push(Edge::new("start", "work")); g.edges.push(Edge::new("work", "exit")); - let events = Arc::new(std::sync::Mutex::new(Vec::::new())); + let events = Arc::new(std::sync::Mutex::new(Vec::::new())); let events_clone = Arc::clone(&events); let emitter = test_emitter("retry-events-test"); emitter.on_event(move |event| { @@ -1019,7 +1019,7 @@ async fn git_checkpoint_skips_start_node() { g.edges.push(Edge::new("start", "work")); g.edges.push(Edge::new("work", "exit")); - let events = Arc::new(std::sync::Mutex::new(Vec::::new())); + let events = Arc::new(std::sync::Mutex::new(Vec::::new())); let events_clone = Arc::clone(&events); let emitter = test_emitter("git-cp-test"); emitter.on_event(move |event| { diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index d5757c757..498cb2ac7 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use crate::error::FabroError; -use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter, RunNoticeLevel}; use crate::git::MetadataStore; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; use crate::records::{Checkpoint, Conclusion, StageSummary}; @@ -20,7 +20,7 @@ fn emit_run_notice( code: impl Into, message: impl Into, ) { - emitter.emit(&WorkflowRunEvent::RunNotice { + emitter.emit(&Event::RunNotice { level, code: code.into(), message: message.into(), diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 513c5b085..ef48e8486 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -15,7 +15,7 @@ use shlex::try_quote; use crate::devcontainer_bridge::{devcontainer_to_snapshot_config, run_devcontainer_lifecycle}; use crate::error::FabroError; -use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter, RunNoticeLevel}; use crate::git::{self, GitSyncStatus, MetadataStore}; use crate::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter}; use crate::handler::{HandlerRegistry, default_registry}; @@ -52,7 +52,7 @@ fn emit_run_notice( code: impl Into, message: impl Into, ) { - emitter.emit(&WorkflowRunEvent::RunNotice { + emitter.emit(&Event::RunNotice { level, code: code.into(), message: message.into(), @@ -323,14 +323,12 @@ async fn resolve_devcontainer(options: &mut InitOptions) -> Result<(), FabroErro let lifecycle_command_count = config.on_create_commands.len() + config.post_create_commands.len() + config.post_start_commands.len(); - options - .emitter - .emit(&WorkflowRunEvent::DevcontainerResolved { - dockerfile_lines: config.dockerfile.lines().count(), - environment_count: config.environment.len(), - lifecycle_command_count, - workspace_folder: config.workspace_folder.clone(), - }); + options.emitter.emit(&Event::DevcontainerResolved { + dockerfile_lines: config.dockerfile.lines().count(), + environment_count: config.environment.len(), + lifecycle_command_count, + workspace_folder: config.workspace_folder.clone(), + }); options .sandbox @@ -429,7 +427,7 @@ pub async fn initialize( let sandbox_event_callback: SandboxEventCallback = { let emitter = Arc::clone(&options.emitter); Arc::new(move |event| { - emitter.emit(&WorkflowRunEvent::Sandbox { event }); + emitter.emit(&Event::Sandbox { event }); }) }; let mut worktree_created = false; @@ -513,7 +511,7 @@ pub async fn initialize( } let sandbox_record = options.sandbox.to_sandbox_record(&*sandbox); - options.emitter.emit(&WorkflowRunEvent::SandboxInitialized { + options.emitter.emit(&Event::SandboxInitialized { working_directory: sandbox_record.working_directory.clone(), provider: sandbox_record.provider.clone(), identifier: sandbox_record.identifier.clone(), @@ -580,17 +578,15 @@ pub async fn initialize( } if !options.lifecycle.setup_commands.is_empty() { - options.emitter.emit(&WorkflowRunEvent::SetupStarted { + options.emitter.emit(&Event::SetupStarted { command_count: options.lifecycle.setup_commands.len(), }); let setup_start = Instant::now(); for (index, command) in options.lifecycle.setup_commands.iter().enumerate() { - options - .emitter - .emit(&WorkflowRunEvent::SetupCommandStarted { - command: command.clone(), - index, - }); + options.emitter.emit(&Event::SetupCommandStarted { + command: command.clone(), + index, + }); let cmd_start = Instant::now(); let result = sandbox .exec_command( @@ -604,7 +600,7 @@ pub async fn initialize( .map_err(|e| FabroError::engine(format!("Setup command failed: {e}")))?; let duration_ms = crate::millis_u64(cmd_start.elapsed()); if result.exit_code != 0 { - options.emitter.emit(&WorkflowRunEvent::SetupFailed { + options.emitter.emit(&Event::SetupFailed { command: command.clone(), index, exit_code: result.exit_code, @@ -615,16 +611,14 @@ pub async fn initialize( result.exit_code, result.stderr, ))); } - options - .emitter - .emit(&WorkflowRunEvent::SetupCommandCompleted { - command: command.clone(), - index, - exit_code: result.exit_code, - duration_ms, - }); + options.emitter.emit(&Event::SetupCommandCompleted { + command: command.clone(), + index, + exit_code: result.exit_code, + duration_ms, + }); } - options.emitter.emit(&WorkflowRunEvent::SetupCompleted { + options.emitter.emit(&Event::SetupCompleted { duration_ms: crate::millis_u64(setup_start.elapsed()), }); } diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index 1245ed52e..4cc816d7a 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -61,7 +61,7 @@ mod tests { use std::time::Duration; use super::*; - use crate::event::{WorkflowRunEvent, append_workflow_event}; + use crate::event::{Event, append_event}; use crate::records::RunRecord; fn memory_store() -> StoreHandle { @@ -143,10 +143,10 @@ mod tests { ) -> SlateRunStore { let store = memory_store(); let run_store = store.create_run(&record.run_id).await.unwrap(); - append_workflow_event( + append_event( &run_store, &record.run_id, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: record.run_id, settings: serde_json::to_value(&record.settings).unwrap(), graph: serde_json::to_value(&record.graph).unwrap(), diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 976944c59..0a2108759 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -9,7 +9,7 @@ use fabro_llm::generate::{GenerateParams, generate}; use fabro_util::text::strip_goal_decoration; use super::types::{Concluded, Finalized, PullRequestOptions}; -use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; +use crate::event::{Event, EventEmitter, RunNoticeLevel}; use crate::outcome::{StageStatus, format_cost as outcome_format_cost}; use crate::records::{Conclusion, RunRecord}; use fabro_retro::retro::Retro; @@ -273,7 +273,7 @@ fn emit_run_notice( code: impl Into, message: impl Into, ) { - emitter.emit(&WorkflowRunEvent::RunNotice { + emitter.emit(&Event::RunNotice { level, code: code.into(), message: message.into(), @@ -535,7 +535,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> .await { Ok(Some(record)) => { - emitter.emit(&WorkflowRunEvent::PullRequestCreated { + emitter.emit(&Event::PullRequestCreated { pr_url: record.html_url.clone(), pr_number: record.number, owner: record.owner.clone(), @@ -549,7 +549,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> } Ok(None) => {} Err(e) => { - emitter.emit(&WorkflowRunEvent::PullRequestFailed { error: e.clone() }); + emitter.emit(&Event::PullRequestFailed { error: e.clone() }); emit_run_notice( &emitter, RunNoticeLevel::Warn, @@ -579,7 +579,7 @@ mod tests { use std::sync::{Arc, Once}; use super::*; - use crate::event::{WorkflowRunEvent, append_workflow_event}; + use crate::event::{Event, append_event}; use crate::records::StageSummary; use chrono::Utc; use fabro_graphviz::graph::Graph; @@ -1093,10 +1093,10 @@ mod tests { base_branch: Some("main".to_string()), labels: HashMap::new(), }; - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: fixtures::RUN_1, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -1113,10 +1113,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::RetroCompleted { + &Event::RetroCompleted { duration_ms: 1, response: Some(String::new()), retro: Some(serde_json::to_value(make_test_retro()).unwrap()), @@ -1159,10 +1159,10 @@ mod tests { base_branch: Some("main".to_string()), labels: HashMap::new(), }; - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: fixtures::RUN_1, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -1179,10 +1179,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::StageCompleted { + &Event::StageCompleted { node_id: "plan".to_string(), name: "plan".to_string(), index: 0, @@ -1378,10 +1378,10 @@ mod tests { base_branch: None, labels: std::collections::HashMap::new(), }; - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: fixtures::RUN_1, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -1398,10 +1398,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::WorkflowRunCompleted { + &Event::WorkflowRunCompleted { duration_ms: 1, artifact_count: 0, status: "success".to_string(), diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 3127035e4..4b1f08162 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -7,7 +7,7 @@ use fabro_retro::retro_agent::{ }; use super::types::{Executed, RetroOptions, Retroed}; -use crate::event::WorkflowRunEvent; +use crate::event::Event; pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { let state = match options.run_store.state().await { @@ -15,7 +15,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { Err(e) => { tracing::warn!(error = %e, "Could not load run state, skipping retro"); if let Some(ref emitter) = options.emitter { - emitter.emit(&WorkflowRunEvent::RetroFailed { + emitter.emit(&Event::RetroFailed { error: e.to_string(), duration_ms: 0, }); @@ -26,7 +26,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { let Some(cp) = state.checkpoint else { tracing::warn!("Could not load checkpoint, skipping retro"); if let Some(ref emitter) = options.emitter { - emitter.emit(&WorkflowRunEvent::RetroFailed { + emitter.emit(&Event::RetroFailed { error: "checkpoint not found".to_string(), duration_ms: 0, }); @@ -54,7 +54,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { let retro_start = std::time::Instant::now(); let retro_prompt = build_retro_prompt(RETRO_DATA_DIR); if let Some(ref emitter) = options.emitter { - emitter.emit(&WorkflowRunEvent::RetroStarted { + emitter.emit(&Event::RetroStarted { prompt: Some(retro_prompt), provider: Some(options.provider.as_str().to_string()), model: Some(options.model.clone()), @@ -70,7 +70,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { Arc::new(move |event: SessionEvent| { emitter.touch(); if !event.event.is_streaming_noise() { - emitter.emit(&WorkflowRunEvent::Agent { + emitter.emit(&Event::Agent { stage: "retro".to_string(), visit: 1, event: event.event.clone(), @@ -100,7 +100,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { Ok((narrative, response)) => { retro.apply_narrative(narrative); if let Some(ref emitter) = options.emitter { - emitter.emit(&WorkflowRunEvent::RetroCompleted { + emitter.emit(&Event::RetroCompleted { duration_ms, response: Some(response), retro: serde_json::to_value(&retro).ok(), @@ -109,7 +109,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { } Err(e) => { if let Some(ref emitter) = options.emitter { - emitter.emit(&WorkflowRunEvent::RetroFailed { + emitter.emit(&Event::RetroFailed { error: e.to_string(), duration_ms, }); @@ -176,7 +176,7 @@ mod tests { use super::*; use crate::context::Context; use crate::event::EventEmitter; - use crate::event::{StoreProgressLogger, WorkflowRunEvent, append_workflow_event}; + use crate::event::{Event, StoreProgressLogger, append_event}; use crate::pipeline::types::Executed; use crate::records::{Checkpoint, CheckpointExt, RunRecord}; use crate::run_options::RunOptions; @@ -227,10 +227,10 @@ mod tests { base_branch: None, labels: std::collections::HashMap::new(), }; - append_workflow_event( + append_event( &run_store, &test_run_id(), - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: test_run_id(), settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -247,10 +247,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run_store, &test_run_id(), - &WorkflowRunEvent::CheckpointCompleted { + &Event::CheckpointCompleted { node_id: checkpoint.current_node.clone(), status: "success".to_string(), current_node: checkpoint.current_node.clone(), diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index a7049248b..648af7fd6 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -384,7 +384,7 @@ mod tests { use object_store::memory::InMemory; use super::scan_runs_combined; - use crate::event::{WorkflowRunEvent, append_workflow_event}; + use crate::event::{Event, append_event}; use crate::operations::make_run_dir; use crate::records::RunRecord; @@ -419,10 +419,10 @@ mod tests { let store = memory_store(); let run_record = sample_run_record(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: fixtures::RUN_1, settings: serde_json::to_value(&run_record.settings).unwrap(), graph: serde_json::to_value(&run_record.graph).unwrap(), @@ -439,10 +439,10 @@ mod tests { ) .await .unwrap(); - append_workflow_event( + append_event( &run_store, &fixtures::RUN_1, - &WorkflowRunEvent::RunSubmitted { reason: None }, + &Event::RunSubmitted { reason: None }, ) .await .unwrap(); diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index 6c720cf2c..2be28e3ba 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -9,7 +9,7 @@ use fabro_store::{RunProjection, SlateStore}; use object_store::local::LocalFileSystem; use crate::error::{FabroError, Result}; -use crate::event::{EventEmitter, StoreProgressLogger, WorkflowRunEvent, append_workflow_event}; +use crate::event::{Event, EventEmitter, StoreProgressLogger, append_event}; use crate::handler::HandlerRegistry; use crate::outcome::Outcome; use crate::pipeline; @@ -31,7 +31,7 @@ struct InitializedState { fn bound_emitter(run_id: fabro_types::RunId, observer: &Arc) -> Arc { let emitter = Arc::new(EventEmitter::new(run_id)); let observer_clone = Arc::clone(observer); - emitter.on_event(move |event| observer_clone.dispatch_stored_event(event)); + emitter.on_event(move |event| observer_clone.dispatch_run_event(event)); emitter } @@ -64,10 +64,10 @@ async fn initialized( .await .expect("failed to create slate-backed test run store"); let run_store = inner_store; - append_workflow_event( + append_event( &run_store, &run_options.run_id, - &WorkflowRunEvent::RunCreated { + &Event::RunCreated { run_id: run_options.run_id, settings: serde_json::to_value(&run_options.settings) .expect("failed to serialize settings"), diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 392e3ac05..feb0c7c3f 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -25,11 +25,11 @@ use fabro_interview::{ }; use fabro_llm::provider::Provider; use fabro_store::{RuntimeState, SlateStore}; -use fabro_types::{RunId, Settings, StoredEvent}; +use fabro_types::{RunEvent, RunId, Settings}; use fabro_validate::{Severity, validate, validate_or_raise}; use fabro_workflow::context::Context; use fabro_workflow::error::{FabroError, FailureSignatureExt}; -use fabro_workflow::event::{EventEmitter, WorkflowRunEvent}; +use fabro_workflow::event::{Event, EventEmitter}; use fabro_workflow::handler::agent::{AgentHandler, CodergenBackend, CodergenResult}; use fabro_workflow::handler::command::CommandHandler; use fabro_workflow::handler::conditional::ConditionalHandler; @@ -1545,7 +1545,7 @@ impl Handler for ContextSetterHandler { } } -fn collect_events(emitter: &EventEmitter) -> Arc>> { +fn collect_events(emitter: &EventEmitter) -> Arc>> { let events = Arc::new(std::sync::Mutex::new(Vec::new())); let events_clone = Arc::clone(&events); emitter.on_event(move |event| { @@ -7316,7 +7316,7 @@ impl HookTestRunner { } } -fn emitter_with_events() -> (Arc, Arc>>) { +fn emitter_with_events() -> (Arc, Arc>>) { let emitter = EventEmitter::default(); let events = collect_events(&emitter); (Arc::new(emitter), events) @@ -7331,7 +7331,7 @@ fn engine_with_hooks(hooks: Vec) -> HookTestRunner fn engine_with_hooks_and_events( hooks: Vec, -) -> (HookTestRunner, Arc>>) { +) -> (HookTestRunner, Arc>>) { let (emitter, events) = emitter_with_events(); ( HookTestRunner { @@ -12055,7 +12055,7 @@ impl Handler for KeepaliveHandler { let start = std::time::Instant::now(); while start.elapsed() < std::time::Duration::from_millis(self.total_ms) { tokio::time::sleep(std::time::Duration::from_millis(self.interval_ms)).await; - services.emitter.emit(&WorkflowRunEvent::Prompt { + services.emitter.emit(&Event::Prompt { stage: node.id.clone(), visit: 1, text: "keepalive".to_string(), @@ -12443,7 +12443,7 @@ async fn asset_collection_local_sandbox_success() { // Check that AssetCaptured events were emitted let captured_events = events.lock().unwrap(); - let asset_events: Vec<&StoredEvent> = captured_events + let asset_events: Vec<&RunEvent> = captured_events .iter() .filter(|e| e.event == "asset.captured") .collect();