scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor: simplify run event representation

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-04 12:59:42 -04:00
parent f6c5823bc5
commit 6c9877cc73
No known key found for this signature in database
8 changed files with 1472 additions and 187 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
docs/plans/2026-04-04-run-event-simplification-plan.md Normal file
View file

@ -0,0 +1,74 @@
# Simplify `RunEvent` While Keeping Wire JSON Stable
## Summary
- Refactor the event model so `RunEvent` stores only envelope metadata plus a typed `EventBody`; the JSON wire format stays `{ ..., "event": "...", "properties": { ... } }`.
- Treat this as an internal Rust API break now: remove public `RunEvent.event` and `RunEvent.properties`, update repo call sites in one pass, and align the code with `docs-internal/events-strategy.md`'s "canonical envelope built once" rule.
- Preserve forward-compatibility for unknown stored events explicitly instead of relying on duplicate cached fields.
- Phase the refactor into three commits so the direct `Event -> RunEvent` mapping can land and be verified before the cached-field removal.
## Implementation Changes
- Sequencing
- Commit 1: add `EventBody::event_name() -> &str`, remove `event_name_from_body()` and `properties_from_body()`, and replace both with explicit implementations that keep cached fields working through commits 1 and 2.
- Commit 2: rework `fabro-workflow` to construct `RunEvent` directly from `Event`; this is the main structural refactor and the primary regression risk.
- Commit 3: remove cached `RunEvent.event` / `RunEvent.properties`, update callers and tests, and replace `EventBody::Unknown` with a raw-preserving variant.
- `lib/crates/fabro-types/src/run_event/mod.rs`
- Redefine `RunEvent` to contain `id`, `ts`, `run_id`, optional envelope metadata, and `body: EventBody` only.
- Keep `RunEvent::from_value`, `from_json_str`, and `to_value`, but make them thin wire-boundary helpers around a private raw wire struct for `{ event, properties }`.
- Replace `EventBody::Unknown` with a raw-preserving variant such as `Unknown { name: String, properties: Value }`.
- Add `EventBody::event_name() -> &str` implemented as an exhaustive `match` returning the serde rename string for each known variant and `name.as_str()` for `Unknown`.
- In commit 1, replace `properties_from_body()` with an explicit property-serialization helper that derives the inner properties payload without the current serialize-and-pluck helper pattern; it may still serialize as an interim step, but it should exist only to support cached fields and wire serialization during the transition.
- Keep JSON property extraction as a serialization helper, not a hot-path public API. Use it only in `RunEvent::to_value` / `Serialize` and in wire-shape tests that need JSON-level assertions.
- Remove `refresh_cache`, `event_name_from_body`, and `properties_from_body`.
- Call out unknown-event fallback explicitly: `Unknown { name, properties }` cannot rely on `#[serde(other)]`, so `RunEvent::from_value` must use a custom fallback path that preserves raw `event` and `properties` when typed `EventBody` deserialization fails.
- `lib/crates/fabro-workflow/src/event.rs`
- Split the current conversion into two explicit pieces: envelope metadata extraction and `Event -> EventBody` construction.
- Rework `to_run_event_at()` to build `RunEvent` directly, not via `json!` plus `RunEvent::from_value`.
- Keep all existing canonicalization rules, but express them as Rust matches: `run_id` stripping, node/session extraction, node-label defaults, failure/error normalization, and agent/sandbox nested event flattening.
- Treat `Event::Agent` and `Event::Sandbox` as the bulk of the work:
- `Event::Agent` must expand each `AgentEvent` sub-variant into the corresponding `EventBody` variant while also lifting `stage -> node_id`, preserving `session_id` / `parent_session_id` in the envelope, and merging `visit` into the inner props where required.
- `Event::Sandbox` must unwrap each `SandboxEvent` sub-variant into the corresponding `EventBody` variant while preserving the current flattened wire shape.
- `Event::WorkflowRunFailed` must continue converting `FabroError` into the stored string form used by `RunFailedProps`.
- stage/parallel/prompt/watchdog variants must continue moving `node_id`/`stage`/`branch`/`node` into the envelope with the same current `node_label` defaults.
- Make the lossy cross-crate conversions explicit in the implementation and guard them with wire-shape characterization tests:
- `fabro_agent::AgentError -> String`
- `fabro_llm::error::SdkError -> string fields in retry props`
- `fabro_llm` usage types -> `fabro_types` usage structs
- `fabro_workflow::error::FabroError -> String`
- Delete `tagged_variant_fields*` once all variant mapping is direct and covered by tests.
- Keep redaction/persistence logic driven by serialized `RunEvent` wire value; no wire-shape change and no redaction contract change.
- Explicitly keep the `build_redacted_event_payload` pipeline out of scope for this pass: no changes to `to_value() -> normalize -> to_string -> redact -> from_str`.
- `lib/crates/fabro-store/src/types.rs`, `lib/crates/fabro-store/src/run_state.rs`, and repo consumers
- Update call sites to stop reading `RunEvent.event` and `RunEvent.properties` directly.
- Default rule: production consumers match on `body`; only serialization/wire tests should rely on JSON property extraction.
- Route store decoding through one helper path (`TryFrom<&EventPayload>` or `RunEvent::from_value`) and keep clone-based payload parsing for now; zero-copy parsing is out of scope for this pass.
- Update strategy/docs terminology only, not code naming: leave `RunEvent` as the code type in this pass and align `docs-internal/events-strategy.md` if needed.
## Test Plan
- `lib/crates/fabro-types/src/run_event/mod.rs`
- known event round-trip preserves the wire JSON shape
- unknown event round-trip preserves raw `event` and `properties`
- known event name with invalid properties still fails deserialization
- absent optional envelope fields serialize as omitted fields, not `null`
- `lib/crates/fabro-workflow/src/event.rs`
- characterization tests for representative variants: stage event, agent event, sandbox event, and run failure
- assert direct construction produces the same wire JSON and envelope fields as today
- assert `build_redacted_event_payload` still returns a valid `EventPayload`
- add focused coverage for agent flattening and sandbox flattening, since those wrappers are the highest-risk conversion paths
- `lib/crates/fabro-store/src/run_state.rs` or adjacent store tests
- replay persisted payloads into `RunProjection` still reconstructs run, status, checkpoint, retro, and pull-request state correctly
- Test migration rules
- behavior tests should prefer matching on typed `body` instead of reintroducing JSON-shaped assertions
- wire-contract tests should assert on `to_value()` / serialized JSON when the exact `properties` shape matters
- do not replace all former `stored.properties["foo"]` assertions with a general-purpose allocating helper in production code
- `fabro-cli/src/commands/run/run_progress/event.rs::from_run_event()` is already aligned with the target design because it matches on `EventBody`; only any remaining CLI tests asserting through `stored.properties` need migration in commit 3
- Verification
- run `cargo nextest run -p fabro-types`
- run `cargo nextest run -p fabro-workflow`
- run `cargo nextest run -p fabro-store`
- run `cargo fmt --check --all` and `cargo clippy --workspace -- -D warnings`
## Assumptions
- The JSON wire protocol stays compatible; only the internal Rust representation and helper APIs change.
- Internal Rust API break is acceptable now; tests and internal consumers will be updated in the same pass.
- Unknown stored events are a supported forward-compatibility case and must survive parse/serialize unchanged.
- This pass optimizes for simplicity and maintainability first; deeper read-side performance work such as borrowed parsing, eliminating `Value` clones in projection, or optimizing the redaction pipeline can follow separately.

346
lib/crates/fabro-types/src/run_event/mod.rs
View file

@ -48,12 +48,10 @@ pub struct RunEvent {
pub id: String,
pub ts: DateTime<Utc>,
pub run_id: RunId,
pub event: String,
pub node_id: Option<String>,
pub node_label: Option<String>,
pub session_id: Option<String>,
pub parent_session_id: Option<String>,
pub properties: Value,
pub body: EventBody,
}
@ -261,8 +259,10 @@ pub enum EventBody {
RetroCompleted(RetroCompletedProps),
#[serde(rename = "retro.failed")]
RetroFailed(RetroFailedProps),
#[serde(other)]
Unknown,
Unknown {
name: String,
properties: Value,
},
}
#[derive(Debug, Clone, Deserialize)]
@ -287,23 +287,260 @@ fn default_properties() -> Value {
Value::Object(Map::new())
}
impl EventBody {
pub fn event_name(&self) -> &str {
match self {
Self::RunCreated(_) => "run.created",
Self::RunStarted(_) => "run.started",
Self::RunSubmitted(_) => "run.submitted",
Self::RunStarting(_) => "run.starting",
Self::RunRunning(_) => "run.running",
Self::RunRemoving(_) => "run.removing",
Self::RunRewound(_) => "run.rewound",
Self::RunCompleted(_) => "run.completed",
Self::RunFailed(_) => "run.failed",
Self::RunNotice(_) => "run.notice",
Self::StageStarted(_) => "stage.started",
Self::StageCompleted(_) => "stage.completed",
Self::StageFailed(_) => "stage.failed",
Self::StageRetrying(_) => "stage.retrying",
Self::ParallelStarted(_) => "parallel.started",
Self::ParallelBranchStarted(_) => "parallel.branch.started",
Self::ParallelBranchCompleted(_) => "parallel.branch.completed",
Self::ParallelCompleted(_) => "parallel.completed",
Self::InterviewStarted(_) => "interview.started",
Self::InterviewCompleted(_) => "interview.completed",
Self::InterviewTimeout(_) => "interview.timeout",
Self::CheckpointCompleted(_) => "checkpoint.completed",
Self::CheckpointFailed(_) => "checkpoint.failed",
Self::GitCommit(_) => "git.commit",
Self::GitPush(_) => "git.push",
Self::GitBranch(_) => "git.branch",
Self::GitWorktreeAdd(_) => "git.worktree.added",
Self::GitWorktreeRemove(_) => "git.worktree.removed",
Self::GitFetch(_) => "git.fetch",
Self::GitReset(_) => "git.reset",
Self::EdgeSelected(_) => "edge.selected",
Self::LoopRestart(_) => "loop.restart",
Self::StagePrompt(_) => "stage.prompt",
Self::PromptCompleted(_) => "prompt.completed",
Self::AgentSessionStarted(_) => "agent.session.started",
Self::AgentSessionEnded(_) => "agent.session.ended",
Self::AgentProcessingEnd(_) => "agent.processing.end",
Self::AgentInput(_) => "agent.input",
Self::AgentMessage(_) => "agent.message",
Self::AgentToolStarted(_) => "agent.tool.started",
Self::AgentToolCompleted(_) => "agent.tool.completed",
Self::AgentError(_) => "agent.error",
Self::AgentWarning(_) => "agent.warning",
Self::AgentLoopDetected(_) => "agent.loop.detected",
Self::AgentTurnLimitReached(_) => "agent.turn.limit",
Self::AgentSteeringInjected(_) => "agent.steering.injected",
Self::AgentCompactionStarted(_) => "agent.compaction.started",
Self::AgentCompactionCompleted(_) => "agent.compaction.completed",
Self::AgentLlmRetry(_) => "agent.llm.retry",
Self::AgentSubSpawned(_) => "agent.sub.spawned",
Self::AgentSubCompleted(_) => "agent.sub.completed",
Self::AgentSubFailed(_) => "agent.sub.failed",
Self::AgentSubClosed(_) => "agent.sub.closed",
Self::AgentMcpReady(_) => "agent.mcp.ready",
Self::AgentMcpFailed(_) => "agent.mcp.failed",
Self::SubgraphStarted(_) => "subgraph.started",
Self::SubgraphCompleted(_) => "subgraph.completed",
Self::SandboxInitializing(_) => "sandbox.initializing",
Self::SandboxReady(_) => "sandbox.ready",
Self::SandboxFailed(_) => "sandbox.failed",
Self::SandboxCleanupStarted(_) => "sandbox.cleanup.started",
Self::SandboxCleanupCompleted(_) => "sandbox.cleanup.completed",
Self::SandboxCleanupFailed(_) => "sandbox.cleanup.failed",
Self::SnapshotPulling(_) => "sandbox.snapshot.pulling",
Self::SnapshotPulled(_) => "sandbox.snapshot.pulled",
Self::SnapshotEnsuring(_) => "sandbox.snapshot.ensuring",
Self::SnapshotCreating(_) => "sandbox.snapshot.creating",
Self::SnapshotReady(_) => "sandbox.snapshot.ready",
Self::SnapshotFailed(_) => "sandbox.snapshot.failed",
Self::GitCloneStarted(_) => "sandbox.git.started",
Self::GitCloneCompleted(_) => "sandbox.git.completed",
Self::GitCloneFailed(_) => "sandbox.git.failed",
Self::SandboxInitialized(_) => "sandbox.initialized",
Self::SetupStarted(_) => "setup.started",
Self::SetupCommandStarted(_) => "setup.command.started",
Self::SetupCommandCompleted(_) => "setup.command.completed",
Self::SetupCompleted(_) => "setup.completed",
Self::SetupFailed(_) => "setup.failed",
Self::StallWatchdogTimeout(_) => "watchdog.timeout",
Self::ArtifactCaptured(_) => "artifact.captured",
Self::SshAccessReady(_) => "ssh.ready",
Self::Failover(_) => "agent.failover",
Self::CliEnsureStarted(_) => "cli.ensure.started",
Self::CliEnsureCompleted(_) => "cli.ensure.completed",
Self::CliEnsureFailed(_) => "cli.ensure.failed",
Self::CommandStarted(_) => "command.started",
Self::CommandCompleted(_) => "command.completed",
Self::AgentCliStarted(_) => "agent.cli.started",
Self::AgentCliCompleted(_) => "agent.cli.completed",
Self::PullRequestCreated(_) => "pull_request.created",
Self::PullRequestFailed(_) => "pull_request.failed",
Self::DevcontainerResolved(_) => "devcontainer.resolved",
Self::DevcontainerLifecycleStarted(_) => "devcontainer.lifecycle.started",
Self::DevcontainerLifecycleCommandStarted(_) => {
"devcontainer.lifecycle.command.started"
}
Self::DevcontainerLifecycleCommandCompleted(_) => {
"devcontainer.lifecycle.command.completed"
}
Self::DevcontainerLifecycleCompleted(_) => "devcontainer.lifecycle.completed",
Self::DevcontainerLifecycleFailed(_) => "devcontainer.lifecycle.failed",
Self::RetroStarted(_) => "retro.started",
Self::RetroCompleted(_) => "retro.completed",
Self::RetroFailed(_) => "retro.failed",
Self::Unknown { name, .. } => name.as_str(),
}
}
fn properties_value(&self) -> serde_json::Result<Value> {
if let Self::Unknown { properties, .. } = self {
return Ok(properties.clone());
}
match serde_json::to_value(self)? {
Value::Object(mut map) => {
Ok(map.remove("properties").unwrap_or_else(default_properties))
}
_ => Ok(default_properties()),
}
}
}
fn is_known_event_name(event: &str) -> bool {
match event {
"run.created"
| "run.started"
| "run.submitted"
| "run.starting"
| "run.running"
| "run.removing"
| "run.rewound"
| "run.completed"
| "run.failed"
| "run.notice"
| "stage.started"
| "stage.completed"
| "stage.failed"
| "stage.retrying"
| "parallel.started"
| "parallel.branch.started"
| "parallel.branch.completed"
| "parallel.completed"
| "interview.started"
| "interview.completed"
| "interview.timeout"
| "checkpoint.completed"
| "checkpoint.failed"
| "git.commit"
| "git.push"
| "git.branch"
| "git.worktree.added"
| "git.worktree.removed"
| "git.fetch"
| "git.reset"
| "edge.selected"
| "loop.restart"
| "stage.prompt"
| "prompt.completed"
| "agent.session.started"
| "agent.session.ended"
| "agent.processing.end"
| "agent.input"
| "agent.message"
| "agent.tool.started"
| "agent.tool.completed"
| "agent.error"
| "agent.warning"
| "agent.loop.detected"
| "agent.turn.limit"
| "agent.steering.injected"
| "agent.compaction.started"
| "agent.compaction.completed"
| "agent.llm.retry"
| "agent.sub.spawned"
| "agent.sub.completed"
| "agent.sub.failed"
| "agent.sub.closed"
| "agent.mcp.ready"
| "agent.mcp.failed"
| "subgraph.started"
| "subgraph.completed"
| "sandbox.initializing"
| "sandbox.ready"
| "sandbox.failed"
| "sandbox.cleanup.started"
| "sandbox.cleanup.completed"
| "sandbox.cleanup.failed"
| "sandbox.snapshot.pulling"
| "sandbox.snapshot.pulled"
| "sandbox.snapshot.ensuring"
| "sandbox.snapshot.creating"
| "sandbox.snapshot.ready"
| "sandbox.snapshot.failed"
| "sandbox.git.started"
| "sandbox.git.completed"
| "sandbox.git.failed"
| "sandbox.initialized"
| "setup.started"
| "setup.command.started"
| "setup.command.completed"
| "setup.completed"
| "setup.failed"
| "watchdog.timeout"
| "artifact.captured"
| "ssh.ready"
| "agent.failover"
| "cli.ensure.started"
| "cli.ensure.completed"
| "cli.ensure.failed"
| "command.started"
| "command.completed"
| "agent.cli.started"
| "agent.cli.completed"
| "pull_request.created"
| "pull_request.failed"
| "devcontainer.resolved"
| "devcontainer.lifecycle.started"
| "devcontainer.lifecycle.command.started"
| "devcontainer.lifecycle.command.completed"
| "devcontainer.lifecycle.completed"
| "devcontainer.lifecycle.failed"
| "retro.started"
| "retro.completed"
| "retro.failed" => true,
_ => false,
}
}
impl RunEvent {
pub fn from_value(value: Value) -> serde_json::Result<Self> {
let raw: RunEventRaw = serde_json::from_value(value)?;
let body = serde_json::from_value(json!({
let body_payload = json!({
"event": raw.event,
"properties": raw.properties,
}))?;
});
let body: EventBody = match serde_json::from_value(body_payload) {
Ok(body) => body,
Err(err) if is_known_event_name(&raw.event) => return Err(err),
Err(_) => EventBody::Unknown {
name: raw.event.clone(),
properties: raw.properties.clone(),
},
};
Ok(Self {
id: raw.id,
ts: raw.ts,
run_id: raw.run_id,
event: event_name_from_body(&body),
node_id: raw.node_id,
node_label: raw.node_label,
session_id: raw.session_id,
parent_session_id: raw.parent_session_id,
properties: raw.properties,
body,
})
}
@ -319,7 +556,7 @@ impl RunEvent {
map.insert("run_id".to_string(), serde_json::to_value(self.run_id)?);
map.insert(
"event".to_string(),
Value::String(event_name_from_body(&self.body)),
Value::String(self.body.event_name().to_string()),
);
if let Some(value) = &self.session_id {
map.insert("session_id".to_string(), Value::String(value.clone()));
@ -336,41 +573,17 @@ impl RunEvent {
if let Some(value) = &self.node_label {
map.insert("node_label".to_string(), Value::String(value.clone()));
}
map.insert("properties".to_string(), properties_from_body(&self.body));
map.insert("properties".to_string(), self.body.properties_value()?);
Ok(Value::Object(map))
}
pub fn event_name(&self) -> &str {
&self.event
self.body.event_name()
}
pub fn properties(&self) -> &Value {
&self.properties
pub fn properties(&self) -> serde_json::Result<Value> {
self.body.properties_value()
}
pub fn refresh_cache(&mut self) {
self.event = event_name_from_body(&self.body);
self.properties = properties_from_body(&self.body);
}
}
fn event_name_from_body(body: &EventBody) -> String {
serde_json::to_value(body)
.ok()
.and_then(|value| {
value
.get("event")
.and_then(Value::as_str)
.map(str::to_owned)
})
.unwrap_or_else(|| "unknown".to_string())
}
fn properties_from_body(body: &EventBody) -> Value {
serde_json::to_value(body)
.ok()
.and_then(|value| value.get("properties").cloned())
.unwrap_or_else(default_properties)
}
impl Serialize for RunEvent {
@ -412,21 +625,10 @@ mod tests {
.unwrap()
.with_timezone(&Utc),
run_id: fixtures::RUN_1,
event: "stage.completed".to_string(),
node_id: Some("build".to_string()),
node_label: Some("Build".to_string()),
session_id: None,
parent_session_id: None,
properties: json!({
"index": 1,
"duration_ms": 1234,
"status": "success",
"suggested_next_ids": ["next"],
"notes": "done",
"files_touched": ["src/main.rs"],
"attempt": 1,
"max_attempts": 1
}),
body: EventBody::StageCompleted(StageCompletedProps {
index: 1,
duration_ms: 1234,
@ -493,4 +695,52 @@ mod tests {
let parsed = RunEvent::from_value(line).unwrap();
assert!(matches!(parsed.body, EventBody::RunCreated(_)));
}
#[test]
fn event_body_event_name_matches_wire_name() {
let body = EventBody::StageCompleted(StageCompletedProps {
index: 1,
duration_ms: 1234,
status: crate::StageStatus::Success,
preferred_label: None,
suggested_next_ids: vec!["next".to_string()],
usage: None,
failure: None,
notes: Some("done".to_string()),
files_touched: vec!["src/main.rs".to_string()],
context_updates: None,
jump_to_node: None,
context_values: None,
node_visits: None,
loop_failure_signatures: None,
restart_failure_signatures: None,
response: None,
attempt: 1,
max_attempts: 1,
});
assert_eq!(body.event_name(), "stage.completed");
}
#[test]
fn run_event_preserves_unknown_event_name_and_properties() {
let value = json!({
"id": "evt_unknown",
"ts": "2026-04-04T12:00:00.000Z",
"run_id": fixtures::RUN_1,
"event": "vendor.custom.event",
"properties": {
"answer": 42,
"nested": { "ok": true }
}
});
let parsed = RunEvent::from_value(value.clone()).unwrap();
let serialized = parsed.to_value().unwrap();
assert_eq!(parsed.event_name(), "vendor.custom.event");
assert_eq!(parsed.properties().unwrap(), value["properties"]);
assert_eq!(serialized["event"], value["event"]);
assert_eq!(serialized["properties"], value["properties"]);
}
}

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

@ -392,27 +392,30 @@ mod tests {
.await
.unwrap();
let events = events.lock().unwrap();
let started = events[0].properties().unwrap();
assert_eq!(events[0].event_name(), "devcontainer.lifecycle.started");
assert_eq!(events[0].properties()["phase"], "on_create");
assert_eq!(events[0].properties()["command_count"], 1);
assert_eq!(started["phase"], "on_create");
assert_eq!(started["command_count"], 1);
assert_eq!(
events[1].event_name(),
"devcontainer.lifecycle.command.started"
);
assert_eq!(events[1].properties()["phase"], "on_create");
assert_eq!(events[1].properties()["index"], 0);
let command_started = events[1].properties().unwrap();
assert_eq!(command_started["phase"], "on_create");
assert_eq!(command_started["index"], 0);
assert_eq!(
events[2].event_name(),
"devcontainer.lifecycle.command.completed"
);
assert_eq!(events[2].properties()["phase"], "on_create");
assert_eq!(events[2].properties()["index"], 0);
assert_eq!(events[2].properties()["exit_code"], 0);
let command_completed = events[2].properties().unwrap();
assert_eq!(command_completed["phase"], "on_create");
assert_eq!(command_completed["index"], 0);
assert_eq!(command_completed["exit_code"], 0);
assert_eq!(events[3].event_name(), "devcontainer.lifecycle.completed");
assert_eq!(events[3].properties()["phase"], "on_create");
assert_eq!(events[3].properties().unwrap()["phase"], "on_create");
}
#[tokio::test]
@ -431,8 +434,9 @@ mod tests {
let events = events.lock().unwrap();
assert!(events.iter().any(|event| {
event.event_name() == "devcontainer.lifecycle.failed"
&& event.properties()["phase"] == "on_create"
&& event.properties()["exit_code"] == 1
&& event.properties().is_ok_and(|properties| {
properties["phase"] == "on_create" && properties["exit_code"] == 1
})
}));
}

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

File diff suppressed because it is too large Load diff

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

@ -926,7 +926,7 @@ async fn retry_emits_stage_started_per_attempt() {
.filter(|event| {
event.event_name() == "stage.started" && event.node_id.as_deref() == Some("work")
})
.map(|event| event.properties()["attempt"].as_u64().unwrap())
.map(|event| event.properties().unwrap()["attempt"].as_u64().unwrap())
.collect();
assert_eq!(work_started, vec![1, 2]);
}
@ -1052,11 +1052,12 @@ async fn git_checkpoint_skips_start_node() {
.iter()
.filter(|event| {
event.event_name() == "checkpoint.completed"
&& event
.properties()
.get("git_commit_sha")
.and_then(|value| value.as_str())
.is_some()
&& event.properties().is_ok_and(|properties| {
properties
.get("git_commit_sha")
.and_then(|value| value.as_str())
.is_some()
})
})
.filter_map(|event| event.node_id.as_deref())
.collect();

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

@ -392,10 +392,11 @@ mod tests {
.iter()
.find(|event| event.event_name() == "retro.started")
.unwrap();
assert_eq!(retro_started.properties()["provider"], "anthropic");
assert_eq!(retro_started.properties()["model"], "test-model");
let retro_started_properties = retro_started.properties().unwrap();
assert_eq!(retro_started_properties["provider"], "anthropic");
assert_eq!(retro_started_properties["model"], "test-model");
assert!(
retro_started.properties()["prompt"]
retro_started_properties["prompt"]
.as_str()
.is_some_and(|prompt| prompt.contains("/tmp/retro_data/progress.jsonl"))
);
@ -404,11 +405,9 @@ mod tests {
.iter()
.find(|event| event.event_name() == "retro.completed")
.unwrap();
assert_eq!(retro_completed.properties()["response"], "");
assert!(retro_completed.properties().get("retro").is_some());
assert_eq!(
retro_completed.properties()["retro"]["smoothness"],
"smooth"
);
let retro_completed_properties = retro_completed.properties().unwrap();
assert_eq!(retro_completed_properties["response"], "");
assert!(retro_completed_properties.get("retro").is_some());
assert_eq!(retro_completed_properties["retro"]["smoothness"], "smooth");
}
}

9
lib/crates/fabro-workflow/tests/it/daytona_integration.rs
View file

@ -687,12 +687,13 @@ async fn daytona_git_checkpoint_remote_emits_events() {
let git_events: Vec<_> = events
.iter()
.filter_map(|e| {
if e.event != "checkpoint.completed" {
if e.event_name() != "checkpoint.completed" {
return None;
}
let properties = e.properties().ok()?;
Some((
e.node_id.clone()?,
e.properties.get("git_commit_sha")?.as_str()?.to_string(),
properties.get("git_commit_sha")?.as_str()?.to_string(),
))
})
.collect();
@ -931,7 +932,7 @@ async fn daytona_parallel_git_branching_e2e() {
let events = events.lock().unwrap();
let parallel_started: Vec<_> = events
.iter()
.filter(|e| e.event == "parallel.started")
.filter(|e| e.event_name() == "parallel.started")
.collect();
assert_eq!(
parallel_started.len(),
@ -940,7 +941,7 @@ async fn daytona_parallel_git_branching_e2e() {
);
let parallel_completed: Vec<_> = events
.iter()
.filter(|e| e.event == "parallel.completed")
.filter(|e| e.event_name() == "parallel.completed")
.collect();
assert_eq!(
parallel_completed.len(),

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

@ -2091,32 +2091,36 @@ async fn event_streaming_lifecycle() {
engine.run(&graph, &run_options).await.expect("run");
let collected = events.lock().unwrap();
assert!(collected.iter().any(|e| e.event == "run.started"));
assert!(collected.iter().any(|e| e.event_name() == "run.started"));
assert!(
collected
.iter()
.any(|e| e.event == "stage.started" && e.node_id.as_deref() == Some("start"))
.any(|e| e.event_name() == "stage.started" && e.node_id.as_deref() == Some("start"))
);
assert!(
collected
.iter()
.any(|e| e.event == "stage.completed" && e.node_id.as_deref() == Some("start"))
.any(|e| e.event_name() == "stage.completed" && e.node_id.as_deref() == Some("start"))
);
assert!(
collected
.iter()
.any(|e| e.event == "stage.started" && e.node_id.as_deref() == Some("task"))
.any(|e| e.event_name() == "stage.started" && e.node_id.as_deref() == Some("task"))
);
assert!(
collected
.iter()
.any(|e| e.event == "stage.completed" && e.node_id.as_deref() == Some("task"))
.any(|e| e.event_name() == "stage.completed" && e.node_id.as_deref() == Some("task"))
);
assert!(collected.iter().any(|e| e.event == "checkpoint.completed"));
assert!(collected.iter().any(|e| e.event == "run.completed"));
assert!(
collected
.iter()
.any(|e| e.event_name() == "checkpoint.completed")
);
assert!(collected.iter().any(|e| e.event_name() == "run.completed"));
// WorkflowRunStarted first, WorkflowRunCompleted last
assert_eq!(collected.first().unwrap().event, "run.started");
assert_eq!(collected.last().unwrap().event, "run.completed");
assert_eq!(collected.first().unwrap().event_name(), "run.started");
assert_eq!(collected.last().unwrap().event_name(), "run.completed");
}
#[tokio::test]
@ -2598,8 +2602,8 @@ async fn scenario_ship_a_feature() {
assert!(cp.completed_nodes.contains(&"review".to_string()));
let collected = events.lock().unwrap();
assert!(collected.iter().any(|e| e.event == "run.started"));
assert!(collected.iter().any(|e| e.event == "run.completed"));
assert!(collected.iter().any(|e| e.event_name() == "run.started"));
assert!(collected.iter().any(|e| e.event_name() == "run.completed"));
}
#[tokio::test]
@ -3667,8 +3671,8 @@ async fn integration_smoke_plan_implement_review_done() {
// Verify events
let collected = events.lock().unwrap();
assert!(collected.iter().any(|e| e.event == "run.started"));
assert!(collected.iter().any(|e| e.event == "run.completed"));
assert!(collected.iter().any(|e| e.event_name() == "run.started"));
assert!(collected.iter().any(|e| e.event_name() == "run.completed"));
}
// ===========================================================================
@ -7439,13 +7443,13 @@ async fn hook_run_start_block_prevents_run() {
// WorkflowRunStarted should still have been emitted (it fires before the hook)
let captured = events.lock().unwrap();
assert!(
captured.iter().any(|e| e.event == "run.started"),
captured.iter().any(|e| e.event_name() == "run.started"),
"WorkflowRunStarted should be emitted before hook blocks"
);
// But no StageStarted — the run never reached node execution
assert!(
!captured.iter().any(|e| e.event == "stage.started"),
!captured.iter().any(|e| e.event_name() == "stage.started"),
"No stage should start when RunStart hook blocks"
);
}
@ -7522,13 +7526,15 @@ async fn hook_stage_start_skip_bypasses_node() {
let stage_starts: Vec<_> = captured
.iter()
.filter(|e| {
e.event == "stage.started"
&& !matches!(
e.properties
.get("handler_type")
.and_then(|value| value.as_str()),
Some("start" | "exit")
)
e.event_name() == "stage.started"
&& e.properties().is_ok_and(|properties| {
!matches!(
properties
.get("handler_type")
.and_then(|value| value.as_str()),
Some("start" | "exit")
)
})
})
.collect();
assert!(
@ -7847,7 +7853,7 @@ async fn hook_edge_selected_override_redirects_routing() {
let completed_nodes: Vec<String> = captured
.iter()
.filter_map(|e| {
(e.event == "stage.completed")
(e.event_name() == "stage.completed")
.then(|| e.node_id.clone())
.flatten()
})
@ -8247,13 +8253,16 @@ async fn hooks_do_not_duplicate_workflow_events() {
let captured = events.lock().unwrap();
// Count WorkflowRunStarted — should be exactly 1
let run_started = captured.iter().filter(|e| e.event == "run.started").count();
let run_started = captured
.iter()
.filter(|e| e.event_name() == "run.started")
.count();
assert_eq!(run_started, 1, "Should have exactly 1 WorkflowRunStarted");
// Count WorkflowRunCompleted — should be exactly 1
let run_completed = captured
.iter()
.filter(|e| e.event == "run.completed")
.filter(|e| e.event_name() == "run.completed")
.count();
assert_eq!(
run_completed, 1,
@ -8261,7 +8270,10 @@ async fn hooks_do_not_duplicate_workflow_events() {
);
// No WorkflowRunFailed
let run_failed = captured.iter().filter(|e| e.event == "run.failed").count();
let run_failed = captured
.iter()
.filter(|e| e.event_name() == "run.failed")
.count();
assert_eq!(run_failed, 0, "Should have 0 WorkflowRunFailed");
}
@ -8598,9 +8610,9 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
let evts = events.lock().unwrap();
let completed_event = evts
.iter()
.find(|e| e.event == "run.completed")
.find(|e| e.event_name() == "run.completed")
.expect("should have WorkflowRunCompleted event");
let artifact_count = completed_event.properties["artifact_count"]
let artifact_count = completed_event.properties().unwrap()["artifact_count"]
.as_u64()
.expect("run.completed should include artifact_count");
assert_eq!(
@ -10092,12 +10104,13 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() {
let git_events: Vec<_> = events
.iter()
.filter_map(|e| {
if e.event != "checkpoint.completed" {
if e.event_name() != "checkpoint.completed" {
return None;
}
let properties = e.properties().ok()?;
Some((
e.node_id.clone()?,
e.properties.get("git_commit_sha")?.as_str()?.to_string(),
properties.get("git_commit_sha")?.as_str()?.to_string(),
))
})
.collect();
@ -10588,7 +10601,7 @@ async fn parallel_git_branching_host_e2e() {
let events = events.lock().unwrap();
let parallel_started: Vec<_> = events
.iter()
.filter(|e| e.event == "parallel.started")
.filter(|e| e.event_name() == "parallel.started")
.collect();
assert_eq!(
parallel_started.len(),
@ -10598,7 +10611,7 @@ async fn parallel_git_branching_host_e2e() {
let parallel_completed: Vec<_> = events
.iter()
.filter(|e| e.event == "parallel.completed")
.filter(|e| e.event_name() == "parallel.completed")
.collect();
assert_eq!(
parallel_completed.len(),
@ -11565,7 +11578,7 @@ async fn e2e_circuit_breaker_emits_events_before_abort() {
let events = events.lock().unwrap();
// Should have at least WorkflowRunStarted and some StageFailed/StageCompleted events
let has_pipeline_started = events.iter().any(|e| e.event == "run.started");
let has_pipeline_started = events.iter().any(|e| e.event_name() == "run.started");
assert!(
has_pipeline_started,
"WorkflowRunStarted event should be emitted"
@ -11576,11 +11589,11 @@ async fn e2e_circuit_breaker_emits_events_before_abort() {
// the stage event for that iteration is emitted, so we see limit-1 events.
let stage_failed_count = events
.iter()
.filter(|e| e.event == "stage.failed" && e.node_id.as_deref() == Some("work"))
.filter(|e| e.event_name() == "stage.failed" && e.node_id.as_deref() == Some("work"))
.count();
let stage_completed_count = events
.iter()
.filter(|e| e.event == "stage.completed" && e.node_id.as_deref() == Some("work"))
.filter(|e| e.event_name() == "stage.completed" && e.node_id.as_deref() == Some("work"))
.count();
let total_work_events = stage_completed_count + stage_failed_count;
// With limit=3, the breaker fires on the 3rd failure before its event is emitted.
@ -12447,31 +12460,23 @@ async fn asset_collection_local_sandbox_success() {
let captured_events = events.lock().unwrap();
let asset_events: Vec<&RunEvent> = captured_events
.iter()
.filter(|e| e.event == "artifact.captured")
.filter(|e| e.event_name() == "artifact.captured")
.collect();
assert!(
!asset_events.is_empty(),
"should emit at least one ArtifactCaptured event"
);
let asset_event = asset_events[0];
assert!(!asset_event.properties["path"].as_str().unwrap().is_empty());
assert!(!asset_event.properties["mime"].as_str().unwrap().is_empty());
let asset_properties = asset_event.properties().unwrap();
assert!(!asset_properties["path"].as_str().unwrap().is_empty());
assert!(!asset_properties["mime"].as_str().unwrap().is_empty());
assert_eq!(asset_properties["content_md5"].as_str().unwrap().len(), 32);
assert_eq!(
asset_event.properties["content_md5"]
.as_str()
.unwrap()
.len(),
32
);
assert_eq!(
asset_event.properties["content_sha256"]
.as_str()
.unwrap()
.len(),
asset_properties["content_sha256"].as_str().unwrap().len(),
64
);
assert!(asset_event.properties["bytes"].as_u64().unwrap() > 0);
assert_eq!(asset_event.properties["attempt"].as_u64().unwrap(), 1);
assert!(asset_properties["bytes"].as_u64().unwrap() > 0);
assert_eq!(asset_properties["attempt"].as_u64().unwrap(), 1);
}
/// Local sandbox: assets are still collected even when the handler fails.