scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor: rename workflow run events

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-04 11:15:32 -04:00
parent cf304664cc
commit ee50eeda79
49 changed files with 516 additions and 552 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
AGENTS.md
View file

@ -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

14
docs-internal/events-strategy.md
View file

@ -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.

2
docs-internal/logging-strategy.md
View file

@ -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

2
docs-internal/plan-events-as-source-of-truth-follow-ups.md
View file

@ -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`

22
lib/crates/fabro-cli/src/commands/run/rewind.rs
View file

@ -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(),

40
lib/crates/fabro-cli/src/commands/run/run_progress/event.rs
View file

@ -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<ProgressEvent> {
pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
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<ProgressEvent> {
}
pub(super) fn from_json_line(line: &str) -> Option<ProgressEvent> {
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<String> {
@ -495,15 +495,15 @@ fn display_value(value: &Value) -> Option<String> {
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 {

126
lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs
View file

@ -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",

10
lib/crates/fabro-cli/src/commands/runs/rm.rs
View file

@ -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,

58
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -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<Utc> {
@ -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(),

6
lib/crates/fabro-cli/tests/it/cmd/pr_view.rs
View file

@ -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(),

4
lib/crates/fabro-server/src/server.rs
View file

@ -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<chrono::Utc>,
// Populated when running:
interviewer: Option<Arc<WebInterviewer>>,
event_tx: Option<broadcast::Sender<StoredEvent>>,
event_tx: Option<broadcast::Sender<RunEvent>>,
context: Option<Context>,
checkpoint: Option<Checkpoint>,
cancel_tx: Option<oneshot::Sender<()>>,

16
lib/crates/fabro-store/src/run_state.rs
View file

@ -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<Utc>,
) -> 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<Value> {
fn provider_used_from_prompt(props: &fabro_types::run_event::StagePromptProps) -> Option<Value> {
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()));

6
lib/crates/fabro-store/src/types.rs
View file

@ -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<Self> {
StoredEvent::from_value(value.as_value().clone())
RunEvent::from_value(value.as_value().clone())
.map_err(|err| StoreError::InvalidEvent(format!("invalid stored event: {err}")))
}
}

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

@ -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;

0
lib/crates/fabro-types/src/stored_event/agent.rs → lib/crates/fabro-types/src/run_event/agent.rs
View file

0
lib/crates/fabro-types/src/stored_event/infra.rs → lib/crates/fabro-types/src/run_event/infra.rs
View file

0
lib/crates/fabro-types/src/stored_event/misc.rs → lib/crates/fabro-types/src/run_event/misc.rs
View file

22
lib/crates/fabro-types/src/stored_event/mod.rs → lib/crates/fabro-types/src/run_event/mod.rs
View file

@ -43,7 +43,7 @@ pub struct TokenUsage {
}
#[derive(Debug, Clone, PartialEq)]
pub struct StoredEvent {
pub struct RunEvent {
pub id: String,
pub ts: DateTime<Utc>,
pub run_id: RunId,
@ -264,7 +264,7 @@ pub enum EventBody {
}
#[derive(Debug, Clone, Deserialize)]
struct StoredEventRaw {
struct RunEventRaw {
id: String,
ts: DateTime<Utc>,
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<Self> {
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<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
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<D>(deserializer: D) -> Result<Self, D::Error>
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(_)));
}
}

2
lib/crates/fabro-types/src/stored_event/run.rs → lib/crates/fabro-types/src/run_event/run.rs
View file

@ -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<chrono::Utc>,
pub run_id: RunId,

0
lib/crates/fabro-types/src/stored_event/stage.rs → lib/crates/fabro-types/src/run_event/stage.rs
View file

22
lib/crates/fabro-workflow/src/devcontainer_bridge.rs
View file

@ -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::<fabro_types::StoredEvent>::new()));
let events = Arc::new(Mutex::new(Vec::<fabro_types::RunEvent>::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::<fabro_types::StoredEvent>::new()));
let events = Arc::new(Mutex::new(Vec::<fabro_types::RunEvent>::new()));
let events_clone = Arc::clone(&events);
emitter.on_event(move |event| {
events_clone.lock().unwrap().push(event.clone());

6
lib/crates/fabro-workflow/src/error.rs
View file

@ -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"),

301
lib/crates/fabro-workflow/src/event.rs
View file

@ -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<String>) -> 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<Utc>,
) -> 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<Utc>) -> 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<EventPayload> {
pub fn build_redacted_event_payload(event: &RunEvent, run_id: &RunId) -> Result<EventPayload> {
let line = redacted_event_json(event)?;
event_payload_from_redacted_json(&line, run_id)
}
pub fn redacted_event_json(event: &StoredEvent) -> Result<String> {
pub fn redacted_event_json(event: &RunEvent) -> Result<String> {
let line = serde_json::to_string(&normalized_event_value(event)?)?;
Ok(redact_jsonl_line(&line))
}
fn normalized_event_value(event: &StoredEvent) -> Result<Value> {
fn normalized_event_value(event: &RunEvent) -> Result<Value> {
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<Ev
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
}
pub async fn append_workflow_event(
run_store: &SlateRunStore,
run_id: &RunId,
event: &WorkflowRunEvent,
) -> 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<dyn Fn(&StoredEvent) + Send + Sync>;
type EventListener = Arc<dyn Fn(&RunEvent) + Send + Sync>;
/// 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<Self>) -> 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 {

30
lib/crates/fabro-workflow/src/git.rs
View file

@ -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(),

8
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -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<dyn fabro_agent::Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, FabroError> {
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),

6
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -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(),

8
lib/crates/fabro-workflow/src/handler/fan_in.rs
View file

@ -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(),

10
lib/crates/fabro-workflow/src/handler/human.rs
View file

@ -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()),

6
lib/crates/fabro-workflow/src/handler/llm/api.rs
View file

@ -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(),

14
lib/crates/fabro-workflow/src/handler/llm/cli.rs
View file

@ -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(),

16
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -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()),

6
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -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,

10
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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}"),

35
lib/crates/fabro-workflow/src/lifecycle/event.rs
View file

@ -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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> for EventLifecycle {
state: &WfRunState,
) -> CoreResult<NodeDecision<Option<StageUsage>>> {
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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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),

12
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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}"),

12
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -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)

22
lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
View file

@ -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(),

6
lib/crates/fabro-workflow/src/operations/resume.rs
View file

@ -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<Started,
.ok_or_else(|| FabroError::Precondition("no checkpoint to resume from".to_string()))?;
cleanup_resume_artifacts(run_dir);
append_workflow_event(
append_event(
&services.run_store,
&services.run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
&Event::RunSubmitted { reason: None },
)
.await
.map_err(|err| FabroError::engine(err.to_string()))?;

42
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -15,8 +15,8 @@ use fabro_types::{RunId, Settings};
use crate::context::Context;
use crate::error::FabroError;
use crate::event::{
EventBody, EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent,
append_workflow_event, event_payload_from_redacted_json, redacted_event_json, to_stored_event,
Event, EventBody, EventEmitter, RunNoticeLevel, StoreProgressLogger, append_event,
event_payload_from_redacted_json, redacted_event_json, to_run_event,
};
use crate::git::MetadataStore;
use crate::handler::HandlerRegistry;
@ -125,10 +125,10 @@ pub(super) async fn execute_persisted_run(
.await;
return Err(error);
}
if let Err(err) = append_workflow_event(
if let Err(err) = append_event(
&run_store,
&run_id,
&WorkflowRunEvent::RunStarting {
&Event::RunStarting {
reason: Some(StatusReason::SandboxInitializing),
},
)
@ -222,10 +222,10 @@ async fn persist_terminal_engine_failure(
None,
)
.await;
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: u64::try_from(duration.as_millis()).unwrap(),
reason: status_reason,
@ -618,10 +618,10 @@ impl Drop for DetachedRunBootstrapGuard {
let run_store = self.run_store.clone();
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
let _ = append_workflow_event(
let _ = append_event(
&run_store,
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
&Event::WorkflowRunFailed {
error: FabroError::engine(format!("{reason:?}")),
duration_ms: 0,
reason: Some(reason),
@ -687,9 +687,9 @@ impl Drop for DetachedRunCompletionGuard {
};
let serialized_notice = {
let stored = to_stored_event(
let stored = to_run_event(
&self.run_id,
&WorkflowRunEvent::RunNotice {
&Event::RunNotice {
level: RunNoticeLevel::Error,
code: code.to_string(),
message: message.to_string(),
@ -712,10 +712,10 @@ impl Drop for DetachedRunCompletionGuard {
let run_id = self.run_id;
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
let _ = append_workflow_event(
let _ = append_event(
&run_store,
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
&Event::WorkflowRunFailed {
error: FabroError::engine(message.to_string()),
duration_ms: 0,
reason: Some(reason),
@ -724,9 +724,9 @@ impl Drop for DetachedRunCompletionGuard {
)
.await;
if let Some((run_id, line)) = serialized_notice.or_else(|| {
let stored = to_stored_event(
let stored = to_run_event(
&run_id,
&WorkflowRunEvent::RunNotice {
&Event::RunNotice {
level: RunNoticeLevel::Error,
code: code.to_string(),
message: message.to_string(),
@ -761,10 +761,10 @@ async fn persist_detached_failure(
) -> 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

4
lib/crates/fabro-workflow/src/pipeline/execute.rs
View file

@ -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,
});

4
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -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::<fabro_types::StoredEvent>::new()));
let events = Arc::new(std::sync::Mutex::new(Vec::<fabro_types::RunEvent>::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::<fabro_types::StoredEvent>::new()));
let events = Arc::new(std::sync::Mutex::new(Vec::<fabro_types::RunEvent>::new()));
let events_clone = Arc::clone(&events);
let emitter = test_emitter("git-cp-test");
emitter.on_event(move |event| {

4
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -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<String>,
message: impl Into<String>,
) {
emitter.emit(&WorkflowRunEvent::RunNotice {
emitter.emit(&Event::RunNotice {
level,
code: code.into(),
message: message.into(),

52
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -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<String>,
message: impl Into<String>,
) {
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()),
});
}

6
lib/crates/fabro-workflow/src/pipeline/persist.rs
View file

@ -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(),

34
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -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<String>,
message: impl Into<String>,
) {
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(),

24
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -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<Retro> {
let state = match options.run_store.state().await {
@ -15,7 +15,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
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<Retro> {
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<Retro> {
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<Retro> {
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<Retro> {
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<Retro> {
}
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(),

10
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -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();

8
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -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<EventEmitter>) -> Arc<EventEmitter> {
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"),

14
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -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<std::sync::Mutex<Vec<StoredEvent>>> {
fn collect_events(emitter: &EventEmitter) -> Arc<std::sync::Mutex<Vec<RunEvent>>> {
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<EventEmitter>, Arc<std::sync::Mutex<Vec<StoredEvent>>>) {
fn emitter_with_events() -> (Arc<EventEmitter>, Arc<std::sync::Mutex<Vec<RunEvent>>>) {
let emitter = EventEmitter::default();
let events = collect_events(&emitter);
(Arc::new(emitter), events)
@ -7331,7 +7331,7 @@ fn engine_with_hooks(hooks: Vec<fabro_hooks::HookDefinition>) -> HookTestRunner
fn engine_with_hooks_and_events(
hooks: Vec<fabro_hooks::HookDefinition>,
) -> (HookTestRunner, Arc<std::sync::Mutex<Vec<StoredEvent>>>) {
) -> (HookTestRunner, Arc<std::sync::Mutex<Vec<RunEvent>>>) {
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();