diff --git a/docs/plans/2026-04-04-run-event-simplification-plan.md b/docs/plans/2026-04-04-run-event-simplification-plan.md new file mode 100644 index 000000000..0f7777a9c --- /dev/null +++ b/docs/plans/2026-04-04-run-event-simplification-plan.md @@ -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. diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index ec7d76f46..d3ac0b661 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -48,12 +48,10 @@ pub struct RunEvent { pub id: String, pub ts: DateTime, pub run_id: RunId, - pub event: String, pub node_id: Option, pub node_label: Option, pub session_id: Option, pub parent_session_id: Option, - 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 { + 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 { 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 { + 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"]); + } } diff --git a/lib/crates/fabro-workflow/src/devcontainer_bridge.rs b/lib/crates/fabro-workflow/src/devcontainer_bridge.rs index 6ed9ee497..3112e8615 100644 --- a/lib/crates/fabro-workflow/src/devcontainer_bridge.rs +++ b/lib/crates/fabro-workflow/src/devcontainer_bridge.rs @@ -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 + }) })); } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index bbb253670..49477226c 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1,12 +1,13 @@ use std::sync::Arc; use std::sync::atomic::{AtomicI64, Ordering}; +use ::fabro_types::run_event as fabro_types; +use ::fabro_types::{RunEvent, RunId, StageStatus, StatusReason}; use anyhow::{Context, Result}; -use chrono::{SecondsFormat, Utc}; +use chrono::Utc; use fabro_store::{EventPayload, SlateRunStore}; -use fabro_types::{RunEvent, RunId}; use serde::{Deserialize, Serialize}; -use serde_json::{Map, Value, json}; +use serde_json::{Map, Value}; use std::collections::BTreeMap; use tokio::sync::{mpsc, oneshot}; use uuid::Uuid; @@ -15,7 +16,6 @@ use crate::error::FabroError; use crate::outcome::{FailureDetail, Outcome, StageUsage}; use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback}; use fabro_llm::types::Usage as LlmUsage; -use fabro_types::StatusReason; use fabro_util::redact::redact_jsonl_line; pub use fabro_types::{EventBody, RunNoticeLevel}; @@ -1183,7 +1183,6 @@ struct StoredEventFields { parent_session_id: Option, node_id: Option, node_label: Option, - properties: Value, } fn tagged_variant_fields(value: &T) -> Map { @@ -1224,6 +1223,976 @@ fn default_node_label(node_id: Option<&String>, node_label: Option) -> O node_label.or_else(|| node_id.cloned()) } +fn token_usage_from_llm(usage: &LlmUsage) -> fabro_types::TokenUsage { + fabro_types::TokenUsage { + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + total_tokens: usage.total_tokens, + reasoning_tokens: usage.reasoning_tokens, + cache_read_tokens: usage.cache_read_tokens, + cache_write_tokens: usage.cache_write_tokens, + speed: usage.speed.clone(), + raw: usage.raw.clone(), + } +} + +fn stage_status_from_string(status: &str) -> StageStatus { + serde_json::from_value(Value::String(status.to_string())).expect("valid stage status") +} + +fn event_body_from_event(event: &Event) -> EventBody { + match event { + Event::RunCreated { + settings, + graph, + workflow_source, + workflow_config, + labels, + run_dir, + working_directory, + host_repo_path, + base_branch, + workflow_slug, + db_prefix, + .. + } => EventBody::RunCreated(fabro_types::RunCreatedProps { + settings: serde_json::from_value(settings.clone()).expect("run.created settings"), + graph: serde_json::from_value(graph.clone()).expect("run.created graph"), + workflow_source: workflow_source.clone(), + workflow_config: workflow_config.clone(), + labels: labels.clone(), + run_dir: run_dir.clone(), + working_directory: working_directory.clone(), + host_repo_path: host_repo_path.clone(), + base_branch: base_branch.clone(), + workflow_slug: workflow_slug.clone(), + db_prefix: db_prefix.clone(), + }), + Event::WorkflowRunStarted { + name, + base_branch, + base_sha, + run_branch, + worktree_dir, + goal, + .. + } => EventBody::RunStarted(fabro_types::RunStartedProps { + name: name.clone(), + base_branch: base_branch.clone(), + base_sha: base_sha.clone(), + run_branch: run_branch.clone(), + worktree_dir: worktree_dir.clone(), + goal: goal.clone(), + }), + Event::RunSubmitted { reason } => { + EventBody::RunSubmitted(fabro_types::RunStatusTransitionProps { + reason: reason.clone(), + }) + } + Event::RunStarting { reason } => { + EventBody::RunStarting(fabro_types::RunStatusTransitionProps { + reason: reason.clone(), + }) + } + Event::RunRunning { reason } => { + EventBody::RunRunning(fabro_types::RunStatusTransitionProps { + reason: reason.clone(), + }) + } + Event::RunRemoving { reason } => { + EventBody::RunRemoving(fabro_types::RunStatusTransitionProps { + reason: reason.clone(), + }) + } + Event::RunRewound { + target_checkpoint_ordinal, + target_node_id, + target_visit, + previous_status, + run_commit_sha, + } => EventBody::RunRewound(fabro_types::RunRewoundProps { + target_checkpoint_ordinal: *target_checkpoint_ordinal, + target_node_id: target_node_id.clone(), + target_visit: *target_visit, + previous_status: previous_status.clone(), + run_commit_sha: run_commit_sha.clone(), + }), + Event::WorkflowRunCompleted { + duration_ms, + artifact_count, + status, + reason, + total_cost, + final_git_commit_sha, + final_patch, + usage, + } => EventBody::RunCompleted(fabro_types::RunCompletedProps { + duration_ms: *duration_ms, + artifact_count: *artifact_count, + status: status.clone(), + reason: reason.clone(), + total_cost: *total_cost, + final_git_commit_sha: final_git_commit_sha.clone(), + final_patch: final_patch.clone(), + usage: usage.as_ref().map(token_usage_from_llm), + }), + Event::WorkflowRunFailed { + error, + duration_ms, + reason, + git_commit_sha, + } => EventBody::RunFailed(fabro_types::RunFailedProps { + error: error.to_string(), + duration_ms: *duration_ms, + reason: reason.clone(), + git_commit_sha: git_commit_sha.clone(), + }), + Event::RunNotice { + level, + code, + message, + } => EventBody::RunNotice(fabro_types::RunNoticeProps { + level: *level, + code: code.clone(), + message: message.clone(), + }), + Event::StageStarted { + index, + handler_type, + attempt, + max_attempts, + .. + } => EventBody::StageStarted(fabro_types::StageStartedProps { + index: *index, + handler_type: handler_type.clone(), + attempt: *attempt, + max_attempts: *max_attempts, + }), + Event::StageCompleted { + index, + duration_ms, + status, + preferred_label, + suggested_next_ids, + usage, + failure, + notes, + files_touched, + context_updates, + jump_to_node, + context_values, + node_visits, + loop_failure_signatures, + restart_failure_signatures, + response, + attempt, + max_attempts, + .. + } => EventBody::StageCompleted(fabro_types::StageCompletedProps { + index: *index, + duration_ms: *duration_ms, + status: stage_status_from_string(status), + preferred_label: preferred_label.clone(), + suggested_next_ids: suggested_next_ids.clone(), + usage: usage.clone(), + failure: failure.clone(), + notes: notes.clone(), + files_touched: files_touched.clone(), + context_updates: context_updates.clone(), + jump_to_node: jump_to_node.clone(), + context_values: context_values.clone(), + node_visits: node_visits.clone(), + loop_failure_signatures: loop_failure_signatures.clone(), + restart_failure_signatures: restart_failure_signatures.clone(), + response: response.clone(), + attempt: *attempt, + max_attempts: *max_attempts, + }), + Event::StageFailed { + index, + failure, + will_retry, + .. + } => EventBody::StageFailed(fabro_types::StageFailedProps { + index: *index, + failure: Some(failure.clone()), + will_retry: *will_retry, + }), + Event::StageRetrying { + index, + attempt, + max_attempts, + delay_ms, + .. + } => EventBody::StageRetrying(fabro_types::StageRetryingProps { + index: *index, + attempt: *attempt, + max_attempts: *max_attempts, + delay_ms: *delay_ms, + }), + Event::ParallelStarted { + visit, + branch_count, + join_policy, + .. + } => EventBody::ParallelStarted(fabro_types::ParallelStartedProps { + visit: *visit, + branch_count: *branch_count, + join_policy: join_policy.clone(), + }), + Event::ParallelBranchStarted { index, .. } => { + EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps { + index: *index, + }) + } + Event::ParallelBranchCompleted { + index, + duration_ms, + status, + head_sha, + .. + } => EventBody::ParallelBranchCompleted(fabro_types::ParallelBranchCompletedProps { + index: *index, + duration_ms: *duration_ms, + status: status.clone(), + head_sha: head_sha.clone(), + }), + Event::ParallelCompleted { + visit, + duration_ms, + success_count, + failure_count, + results, + .. + } => EventBody::ParallelCompleted(fabro_types::ParallelCompletedProps { + visit: *visit, + duration_ms: *duration_ms, + success_count: *success_count, + failure_count: *failure_count, + results: results.clone(), + }), + Event::InterviewStarted { + question, + question_type, + .. + } => EventBody::InterviewStarted(fabro_types::InterviewStartedProps { + question: question.clone(), + question_type: question_type.clone(), + }), + Event::InterviewCompleted { + question, + answer, + duration_ms, + } => EventBody::InterviewCompleted(fabro_types::InterviewCompletedProps { + question: question.clone(), + answer: answer.clone(), + duration_ms: *duration_ms, + }), + Event::InterviewTimeout { + question, + duration_ms, + .. + } => EventBody::InterviewTimeout(fabro_types::InterviewTimeoutProps { + question: question.clone(), + duration_ms: *duration_ms, + }), + Event::CheckpointCompleted { + status, + current_node, + completed_nodes, + node_retries, + context_values, + node_outcomes, + next_node_id, + git_commit_sha, + loop_failure_signatures, + restart_failure_signatures, + node_visits, + diff, + .. + } => EventBody::CheckpointCompleted(fabro_types::CheckpointCompletedProps { + status: status.clone(), + current_node: current_node.clone(), + completed_nodes: completed_nodes.clone(), + node_retries: node_retries.clone(), + context_values: context_values.clone(), + node_outcomes: node_outcomes.clone(), + next_node_id: next_node_id.clone(), + git_commit_sha: git_commit_sha.clone(), + loop_failure_signatures: loop_failure_signatures.clone(), + restart_failure_signatures: restart_failure_signatures.clone(), + node_visits: node_visits.clone(), + diff: diff.clone(), + }), + Event::CheckpointFailed { error, .. } => { + EventBody::CheckpointFailed(fabro_types::CheckpointFailedProps { + error: error.clone(), + }) + } + Event::GitCommit { sha, .. } => { + EventBody::GitCommit(fabro_types::GitCommitProps { sha: sha.clone() }) + } + Event::GitPush { branch, success } => EventBody::GitPush(fabro_types::GitPushProps { + branch: branch.clone(), + success: *success, + }), + Event::GitBranch { branch, sha } => EventBody::GitBranch(fabro_types::GitBranchProps { + branch: branch.clone(), + sha: sha.clone(), + }), + Event::GitWorktreeAdd { path, branch } => { + EventBody::GitWorktreeAdd(fabro_types::GitWorktreeAddProps { + path: path.clone(), + branch: branch.clone(), + }) + } + Event::GitWorktreeRemove { path } => { + EventBody::GitWorktreeRemove(fabro_types::GitWorktreeRemoveProps { path: path.clone() }) + } + Event::GitFetch { branch, success } => EventBody::GitFetch(fabro_types::GitFetchProps { + branch: branch.clone(), + success: *success, + }), + Event::GitReset { sha } => { + EventBody::GitReset(fabro_types::GitResetProps { sha: sha.clone() }) + } + Event::EdgeSelected { + from_node, + to_node, + label, + condition, + reason, + preferred_label, + suggested_next_ids, + stage_status, + is_jump, + } => EventBody::EdgeSelected(fabro_types::EdgeSelectedProps { + from_node: from_node.clone(), + to_node: to_node.clone(), + label: label.clone(), + condition: condition.clone(), + reason: reason.clone(), + preferred_label: preferred_label.clone(), + suggested_next_ids: suggested_next_ids.clone(), + stage_status: stage_status.clone(), + is_jump: *is_jump, + }), + Event::LoopRestart { from_node, to_node } => { + EventBody::LoopRestart(fabro_types::LoopRestartProps { + from_node: from_node.clone(), + to_node: to_node.clone(), + }) + } + Event::Prompt { + visit, + text, + mode, + provider, + model, + .. + } => EventBody::StagePrompt(fabro_types::StagePromptProps { + visit: *visit, + text: text.clone(), + mode: mode.clone(), + provider: provider.clone(), + model: model.clone(), + }), + Event::PromptCompleted { + response, + model, + provider, + usage, + .. + } => EventBody::PromptCompleted(fabro_types::PromptCompletedProps { + response: response.clone(), + model: model.clone(), + provider: provider.clone(), + usage: usage.clone(), + }), + Event::Agent { visit, event, .. } => match event { + AgentEvent::SessionStarted { provider, model } => { + EventBody::AgentSessionStarted(fabro_types::AgentSessionStartedProps { + provider: provider.clone(), + model: model.clone(), + visit: *visit, + }) + } + AgentEvent::SessionEnded => { + EventBody::AgentSessionEnded(fabro_types::AgentSessionEndedProps { visit: *visit }) + } + AgentEvent::ProcessingEnd => { + EventBody::AgentProcessingEnd(fabro_types::AgentProcessingEndProps { + visit: *visit, + }) + } + AgentEvent::UserInput { text } => EventBody::AgentInput(fabro_types::AgentInputProps { + text: text.clone(), + visit: *visit, + }), + AgentEvent::AssistantMessage { + text, + model, + usage, + tool_call_count, + } => EventBody::AgentMessage(fabro_types::AgentMessageProps { + text: text.clone(), + model: model.clone(), + usage: token_usage_from_llm(usage), + tool_call_count: *tool_call_count, + visit: *visit, + }), + AgentEvent::ToolCallStarted { + tool_name, + tool_call_id, + arguments, + } => EventBody::AgentToolStarted(fabro_types::AgentToolStartedProps { + tool_name: tool_name.clone(), + tool_call_id: tool_call_id.clone(), + arguments: arguments.clone(), + visit: *visit, + }), + AgentEvent::ToolCallCompleted { + tool_name, + tool_call_id, + output, + is_error, + } => EventBody::AgentToolCompleted(fabro_types::AgentToolCompletedProps { + tool_name: tool_name.clone(), + tool_call_id: tool_call_id.clone(), + output: output.clone(), + is_error: *is_error, + visit: *visit, + }), + AgentEvent::Error { error } => EventBody::AgentError(fabro_types::AgentErrorProps { + error: serde_json::to_value(error).expect("serializable agent error"), + visit: *visit, + }), + AgentEvent::Warning { + kind, + message, + details, + } => EventBody::AgentWarning(fabro_types::AgentWarningProps { + kind: kind.clone(), + message: message.clone(), + details: details.clone(), + visit: *visit, + }), + AgentEvent::LoopDetected => { + EventBody::AgentLoopDetected(fabro_types::AgentLoopDetectedProps { visit: *visit }) + } + AgentEvent::TurnLimitReached { max_turns } => { + EventBody::AgentTurnLimitReached(fabro_types::AgentTurnLimitReachedProps { + max_turns: *max_turns, + visit: *visit, + }) + } + AgentEvent::SteeringInjected { text } => { + EventBody::AgentSteeringInjected(fabro_types::AgentSteeringInjectedProps { + text: text.clone(), + visit: *visit, + }) + } + AgentEvent::CompactionStarted { + estimated_tokens, + context_window_size, + } => EventBody::AgentCompactionStarted(fabro_types::AgentCompactionStartedProps { + estimated_tokens: *estimated_tokens, + context_window_size: *context_window_size, + visit: *visit, + }), + AgentEvent::CompactionCompleted { + original_turn_count, + preserved_turn_count, + summary_token_estimate, + tracked_file_count, + } => EventBody::AgentCompactionCompleted(fabro_types::AgentCompactionCompletedProps { + original_turn_count: *original_turn_count, + preserved_turn_count: *preserved_turn_count, + summary_token_estimate: *summary_token_estimate, + tracked_file_count: *tracked_file_count, + visit: *visit, + }), + AgentEvent::LlmRetry { + provider, + model, + attempt, + delay_secs, + error, + } => EventBody::AgentLlmRetry(fabro_types::AgentLlmRetryProps { + provider: provider.clone(), + model: model.clone(), + attempt: *attempt, + delay_secs: *delay_secs, + error: serde_json::to_value(error).expect("serializable sdk error"), + visit: *visit, + }), + AgentEvent::SubAgentSpawned { + agent_id, + depth, + task, + } => EventBody::AgentSubSpawned(fabro_types::AgentSubSpawnedProps { + agent_id: agent_id.clone(), + depth: *depth, + task: task.clone(), + visit: *visit, + }), + AgentEvent::SubAgentCompleted { + agent_id, + depth, + success, + turns_used, + } => EventBody::AgentSubCompleted(fabro_types::AgentSubCompletedProps { + agent_id: agent_id.clone(), + depth: *depth, + success: *success, + turns_used: *turns_used, + visit: *visit, + }), + AgentEvent::SubAgentFailed { + agent_id, + depth, + error, + } => EventBody::AgentSubFailed(fabro_types::AgentSubFailedProps { + agent_id: agent_id.clone(), + depth: *depth, + error: serde_json::to_value(error).expect("serializable agent error"), + visit: *visit, + }), + AgentEvent::SubAgentClosed { agent_id, depth } => { + EventBody::AgentSubClosed(fabro_types::AgentSubClosedProps { + agent_id: agent_id.clone(), + depth: *depth, + visit: *visit, + }) + } + AgentEvent::McpServerReady { + server_name, + tool_count, + } => EventBody::AgentMcpReady(fabro_types::AgentMcpReadyProps { + server_name: server_name.clone(), + tool_count: *tool_count, + visit: *visit, + }), + AgentEvent::McpServerFailed { server_name, error } => { + EventBody::AgentMcpFailed(fabro_types::AgentMcpFailedProps { + server_name: server_name.clone(), + error: error.clone(), + visit: *visit, + }) + } + AgentEvent::AssistantTextStart + | AgentEvent::AssistantOutputReplace { .. } + | AgentEvent::TextDelta { .. } + | AgentEvent::ReasoningDelta { .. } + | AgentEvent::ToolCallOutputDelta { .. } + | AgentEvent::SkillExpanded { .. } => { + panic!("streaming-noise agent event should not be converted to RunEvent") + } + }, + Event::SubgraphStarted { start_node, .. } => { + EventBody::SubgraphStarted(fabro_types::SubgraphStartedProps { + start_node: start_node.clone(), + }) + } + Event::SubgraphCompleted { + steps_executed, + status, + duration_ms, + .. + } => EventBody::SubgraphCompleted(fabro_types::SubgraphCompletedProps { + steps_executed: *steps_executed, + status: status.clone(), + duration_ms: *duration_ms, + }), + Event::Sandbox { event } => match event { + SandboxEvent::Initializing { provider } => { + EventBody::SandboxInitializing(fabro_types::SandboxInitializingProps { + provider: provider.clone(), + }) + } + SandboxEvent::Ready { + provider, + duration_ms, + name, + cpu, + memory, + url, + } => EventBody::SandboxReady(fabro_types::SandboxReadyProps { + provider: provider.clone(), + duration_ms: *duration_ms, + name: name.clone(), + cpu: *cpu, + memory: *memory, + url: url.clone(), + }), + SandboxEvent::InitializeFailed { + provider, + error, + duration_ms, + } => EventBody::SandboxFailed(fabro_types::SandboxFailedProps { + provider: provider.clone(), + error: error.clone(), + duration_ms: *duration_ms, + }), + SandboxEvent::CleanupStarted { provider } => { + EventBody::SandboxCleanupStarted(fabro_types::SandboxCleanupStartedProps { + provider: provider.clone(), + }) + } + SandboxEvent::CleanupCompleted { + provider, + duration_ms, + } => EventBody::SandboxCleanupCompleted(fabro_types::SandboxCleanupCompletedProps { + provider: provider.clone(), + duration_ms: *duration_ms, + }), + SandboxEvent::CleanupFailed { provider, error } => { + EventBody::SandboxCleanupFailed(fabro_types::SandboxCleanupFailedProps { + provider: provider.clone(), + error: error.clone(), + }) + } + SandboxEvent::SnapshotPulling { name } => { + EventBody::SnapshotPulling(fabro_types::SnapshotNameProps { name: name.clone() }) + } + SandboxEvent::SnapshotPulled { name, duration_ms } => { + EventBody::SnapshotPulled(fabro_types::SnapshotCompletedProps { + name: name.clone(), + duration_ms: *duration_ms, + }) + } + SandboxEvent::SnapshotEnsuring { name } => { + EventBody::SnapshotEnsuring(fabro_types::SnapshotNameProps { name: name.clone() }) + } + SandboxEvent::SnapshotCreating { name } => { + EventBody::SnapshotCreating(fabro_types::SnapshotNameProps { name: name.clone() }) + } + SandboxEvent::SnapshotReady { name, duration_ms } => { + EventBody::SnapshotReady(fabro_types::SnapshotCompletedProps { + name: name.clone(), + duration_ms: *duration_ms, + }) + } + SandboxEvent::SnapshotFailed { name, error } => { + EventBody::SnapshotFailed(fabro_types::SnapshotFailedProps { + name: name.clone(), + error: error.clone(), + }) + } + SandboxEvent::GitCloneStarted { url, branch } => { + EventBody::GitCloneStarted(fabro_types::GitCloneStartedProps { + url: url.clone(), + branch: branch.clone(), + }) + } + SandboxEvent::GitCloneCompleted { url, duration_ms } => { + EventBody::GitCloneCompleted(fabro_types::GitCloneCompletedProps { + url: url.clone(), + duration_ms: *duration_ms, + }) + } + SandboxEvent::GitCloneFailed { url, error } => { + EventBody::GitCloneFailed(fabro_types::GitCloneFailedProps { + url: url.clone(), + error: error.clone(), + }) + } + }, + Event::SandboxInitialized { + working_directory, + provider, + identifier, + host_working_directory, + container_mount_point, + } => EventBody::SandboxInitialized(fabro_types::SandboxInitializedProps { + working_directory: working_directory.clone(), + provider: provider.clone(), + identifier: identifier.clone(), + host_working_directory: host_working_directory.clone(), + container_mount_point: container_mount_point.clone(), + }), + Event::SetupStarted { command_count } => { + EventBody::SetupStarted(fabro_types::SetupStartedProps { + command_count: *command_count, + }) + } + Event::SetupCommandStarted { command, index } => { + EventBody::SetupCommandStarted(fabro_types::SetupCommandStartedProps { + command: command.clone(), + index: *index, + }) + } + Event::SetupCommandCompleted { + command, + index, + exit_code, + duration_ms, + } => EventBody::SetupCommandCompleted(fabro_types::SetupCommandCompletedProps { + command: command.clone(), + index: *index, + exit_code: *exit_code, + duration_ms: *duration_ms, + }), + Event::SetupCompleted { duration_ms } => { + EventBody::SetupCompleted(fabro_types::SetupCompletedProps { + duration_ms: *duration_ms, + }) + } + Event::SetupFailed { + command, + index, + exit_code, + stderr, + } => EventBody::SetupFailed(fabro_types::SetupFailedProps { + command: command.clone(), + index: *index, + exit_code: *exit_code, + stderr: stderr.clone(), + }), + Event::StallWatchdogTimeout { idle_seconds, .. } => { + EventBody::StallWatchdogTimeout(fabro_types::StallWatchdogTimeoutProps { + idle_seconds: *idle_seconds, + }) + } + Event::ArtifactCaptured { + attempt, + node_slug, + path, + mime, + content_md5, + content_sha256, + bytes, + .. + } => EventBody::ArtifactCaptured(fabro_types::ArtifactCapturedProps { + attempt: *attempt, + node_slug: node_slug.clone(), + path: path.clone(), + mime: mime.clone(), + content_md5: content_md5.clone(), + content_sha256: content_sha256.clone(), + bytes: *bytes, + }), + Event::SshAccessReady { ssh_command } => { + EventBody::SshAccessReady(fabro_types::SshAccessReadyProps { + ssh_command: ssh_command.clone(), + }) + } + Event::Failover { + from_provider, + from_model, + to_provider, + to_model, + error, + .. + } => EventBody::Failover(fabro_types::FailoverProps { + from_provider: from_provider.clone(), + from_model: from_model.clone(), + to_provider: to_provider.clone(), + to_model: to_model.clone(), + error: error.clone(), + }), + Event::CliEnsureStarted { cli_name, provider } => { + EventBody::CliEnsureStarted(fabro_types::CliEnsureStartedProps { + cli_name: cli_name.clone(), + provider: provider.clone(), + }) + } + Event::CliEnsureCompleted { + cli_name, + provider, + already_installed, + node_installed, + duration_ms, + } => EventBody::CliEnsureCompleted(fabro_types::CliEnsureCompletedProps { + cli_name: cli_name.clone(), + provider: provider.clone(), + already_installed: *already_installed, + node_installed: *node_installed, + duration_ms: *duration_ms, + }), + Event::CliEnsureFailed { + cli_name, + provider, + error, + duration_ms, + } => EventBody::CliEnsureFailed(fabro_types::CliEnsureFailedProps { + cli_name: cli_name.clone(), + provider: provider.clone(), + error: error.clone(), + duration_ms: *duration_ms, + }), + Event::CommandStarted { + script, + command, + language, + timeout_ms, + .. + } => EventBody::CommandStarted(fabro_types::CommandStartedProps { + script: script.clone(), + command: command.clone(), + language: language.clone(), + timeout_ms: *timeout_ms, + }), + Event::CommandCompleted { + stdout, + stderr, + exit_code, + duration_ms, + timed_out, + .. + } => EventBody::CommandCompleted(fabro_types::CommandCompletedProps { + stdout: stdout.clone(), + stderr: stderr.clone(), + exit_code: *exit_code, + duration_ms: *duration_ms, + timed_out: *timed_out, + }), + Event::AgentCliStarted { + visit, + mode, + provider, + model, + command, + .. + } => EventBody::AgentCliStarted(fabro_types::AgentCliStartedProps { + visit: *visit, + mode: mode.clone(), + provider: provider.clone(), + model: model.clone(), + command: command.clone(), + }), + Event::AgentCliCompleted { + stdout, + stderr, + exit_code, + duration_ms, + .. + } => EventBody::AgentCliCompleted(fabro_types::AgentCliCompletedProps { + stdout: stdout.clone(), + stderr: stderr.clone(), + exit_code: *exit_code, + duration_ms: *duration_ms, + }), + Event::PullRequestCreated { + pr_url, + pr_number, + owner, + repo, + base_branch, + head_branch, + title, + draft, + } => EventBody::PullRequestCreated(fabro_types::PullRequestCreatedProps { + pr_url: pr_url.clone(), + pr_number: *pr_number, + owner: owner.clone(), + repo: repo.clone(), + base_branch: base_branch.clone(), + head_branch: head_branch.clone(), + title: title.clone(), + draft: *draft, + }), + Event::PullRequestFailed { error } => { + EventBody::PullRequestFailed(fabro_types::PullRequestFailedProps { + error: error.clone(), + }) + } + Event::DevcontainerResolved { + dockerfile_lines, + environment_count, + lifecycle_command_count, + workspace_folder, + } => EventBody::DevcontainerResolved(fabro_types::DevcontainerResolvedProps { + dockerfile_lines: *dockerfile_lines, + environment_count: *environment_count, + lifecycle_command_count: *lifecycle_command_count, + workspace_folder: workspace_folder.clone(), + }), + Event::DevcontainerLifecycleStarted { + phase, + command_count, + } => EventBody::DevcontainerLifecycleStarted( + fabro_types::DevcontainerLifecycleStartedProps { + phase: phase.clone(), + command_count: *command_count, + }, + ), + Event::DevcontainerLifecycleCommandStarted { + phase, + command, + index, + } => EventBody::DevcontainerLifecycleCommandStarted( + fabro_types::DevcontainerLifecycleCommandStartedProps { + phase: phase.clone(), + command: command.clone(), + index: *index, + }, + ), + Event::DevcontainerLifecycleCommandCompleted { + phase, + command, + index, + exit_code, + duration_ms, + } => EventBody::DevcontainerLifecycleCommandCompleted( + fabro_types::DevcontainerLifecycleCommandCompletedProps { + phase: phase.clone(), + command: command.clone(), + index: *index, + exit_code: *exit_code, + duration_ms: *duration_ms, + }, + ), + Event::DevcontainerLifecycleCompleted { phase, duration_ms } => { + EventBody::DevcontainerLifecycleCompleted( + fabro_types::DevcontainerLifecycleCompletedProps { + phase: phase.clone(), + duration_ms: *duration_ms, + }, + ) + } + Event::DevcontainerLifecycleFailed { + phase, + command, + index, + exit_code, + stderr, + } => { + EventBody::DevcontainerLifecycleFailed(fabro_types::DevcontainerLifecycleFailedProps { + phase: phase.clone(), + command: command.clone(), + index: *index, + exit_code: *exit_code, + stderr: stderr.clone(), + }) + } + Event::RetroStarted { + prompt, + provider, + model, + } => EventBody::RetroStarted(fabro_types::RetroStartedProps { + prompt: prompt.clone(), + provider: provider.clone(), + model: model.clone(), + }), + Event::RetroCompleted { + duration_ms, + response, + retro, + } => EventBody::RetroCompleted(fabro_types::RetroCompletedProps { + duration_ms: *duration_ms, + response: response.clone(), + retro: retro.clone(), + }), + Event::RetroFailed { error, duration_ms } => { + EventBody::RetroFailed(fabro_types::RetroFailedProps { + error: error.clone(), + duration_ms: *duration_ms, + }) + } + } +} + fn extract_run_event_fields(event: &Event) -> StoredEventFields { match event { Event::RunCreated { .. } | Event::WorkflowRunStarted { .. } => { @@ -1234,7 +2203,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id: None, node_label: None, - properties: Value::Object(fields), } } Event::WorkflowRunFailed { error, .. } => { @@ -1245,7 +2213,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id: None, node_label: None, - properties: Value::Object(fields), } } Event::StageCompleted { .. } | Event::StageFailed { .. } => { @@ -1258,7 +2225,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id, node_label, - properties: Value::Object(fields), } } Event::StageStarted { .. } @@ -1284,7 +2250,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id, node_label, - properties: Value::Object(fields), } } Event::Agent { @@ -1295,36 +2260,25 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { 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); - let visit = fields.remove("visit"); + fields.remove("visit"); fields.remove("session_id"); fields.remove("parent_session_id"); - let mut properties = fields.remove("event").map_or_else( - || Value::Object(Map::new()), - |value| Value::Object(tagged_variant_fields_from_value(value)), - ); - if let (Some(visit), Value::Object(map)) = (visit, &mut properties) { - map.insert("visit".to_string(), visit); - } + fields.remove("event"); StoredEventFields { session_id: session_id.clone(), parent_session_id: parent_session_id.clone(), node_id, node_label, - properties, } } Event::Sandbox { .. } => { let mut fields = tagged_variant_fields(event); - let properties = fields.remove("event").map_or_else( - || Value::Object(Map::new()), - |value| Value::Object(tagged_variant_fields_from_value(value)), - ); + fields.remove("event"); StoredEventFields { session_id: None, parent_session_id: None, node_id: None, node_label: None, - properties, } } Event::GitCommit { .. } => { @@ -1336,7 +2290,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id, node_label, - properties: Value::Object(fields), } } Event::ParallelBranchStarted { .. } | Event::ParallelBranchCompleted { .. } => { @@ -1348,7 +2301,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id, node_label, - properties: Value::Object(fields), } } Event::Prompt { .. } @@ -1363,7 +2315,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id, node_label, - properties: Value::Object(fields), } } Event::StallWatchdogTimeout { .. } => { @@ -1375,7 +2326,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id, node_label, - properties: Value::Object(fields), } } _ => StoredEventFields { @@ -1383,7 +2333,6 @@ fn extract_run_event_fields(event: &Event) -> StoredEventFields { parent_session_id: None, node_id: None, node_label: None, - properties: Value::Object(tagged_variant_fields(event)), }, } } @@ -1394,18 +2343,17 @@ pub fn to_run_event(run_id: &RunId, event: &Event) -> RunEvent { pub fn to_run_event_at(run_id: &RunId, event: &Event, ts: chrono::DateTime) -> RunEvent { let fields = extract_run_event_fields(event); - RunEvent::from_value(json!({ - "id": Uuid::now_v7().to_string(), - "ts": ts.to_rfc3339_opts(SecondsFormat::Millis, true), - "run_id": run_id.to_string(), - "event": event_name(event), - "session_id": fields.session_id, - "parent_session_id": fields.parent_session_id, - "node_id": fields.node_id, - "node_label": fields.node_label, - "properties": fields.properties, - })) - .expect("workflow event converts to stored event") + let body = event_body_from_event(event); + RunEvent { + id: Uuid::now_v7().to_string(), + ts, + run_id: *run_id, + node_id: fields.node_id, + node_label: fields.node_label, + session_id: fields.session_id, + parent_session_id: fields.parent_session_id, + body, + } } pub fn build_redacted_event_payload(event: &RunEvent, run_id: &RunId) -> Result { @@ -1637,7 +2585,7 @@ impl Emitter { #[cfg(test)] mod tests { use super::*; - use fabro_types::fixtures; + use ::fabro_types::fixtures; use std::sync::{Arc, Mutex}; #[test] @@ -1708,8 +2656,9 @@ mod tests { assert_eq!(stored.run_id, fixtures::RUN_2); assert_eq!(stored.node_id.as_deref(), Some("plan")); assert_eq!(stored.node_label.as_deref(), Some("Plan")); - assert_eq!(stored.properties["duration_ms"], 5000); - assert_eq!(stored.properties["status"], "success"); + let properties = stored.properties().unwrap(); + assert_eq!(properties["duration_ms"], 5000); + assert_eq!(properties["status"], "success"); assert!(stored.session_id.is_none()); } @@ -1741,9 +2690,10 @@ mod tests { }, ); - assert_eq!(stored.properties["response"], "done"); - assert_eq!(stored.properties["loop_failure_signatures"]["sig-a"], 2); - assert_eq!(stored.properties["restart_failure_signatures"]["sig-b"], 1); + let properties = stored.properties().unwrap(); + assert_eq!(properties["response"], "done"); + assert_eq!(properties["loop_failure_signatures"]["sig-a"], 2); + assert_eq!(properties["restart_failure_signatures"]["sig-b"], 1); } #[test] @@ -1763,12 +2713,10 @@ mod tests { ); assert_eq!(stored.event_name(), "stage.failed"); - assert_eq!(stored.properties["failure"]["message"], "lint failed"); - assert_eq!( - stored.properties["failure"]["failure_class"], - "deterministic" - ); - assert_eq!(stored.properties["will_retry"], true); + let properties = stored.properties().unwrap(); + assert_eq!(properties["failure"]["message"], "lint failed"); + assert_eq!(properties["failure"]["failure_class"], "deterministic"); + assert_eq!(properties["will_retry"], true); } #[test] @@ -1793,9 +2741,10 @@ mod tests { assert_eq!(stored.node_label.as_deref(), Some("code")); assert_eq!(stored.session_id.as_deref(), Some("ses_child")); assert_eq!(stored.parent_session_id.as_deref(), Some("ses_parent")); - assert_eq!(stored.properties["tool_name"], "read_file"); - assert_eq!(stored.properties["tool_call_id"], "call_1"); - assert_eq!(stored.properties["visit"], 2); + let properties = stored.properties().unwrap(); + assert_eq!(properties["tool_name"], "read_file"); + assert_eq!(properties["tool_call_id"], "call_1"); + assert_eq!(properties["visit"], 2); } #[test] @@ -1816,8 +2765,9 @@ mod tests { assert_eq!(stored.event_name(), "sandbox.ready"); assert!(stored.node_id.is_none()); - assert_eq!(stored.properties["provider"], "daytona"); - assert_eq!(stored.properties["duration_ms"], 2500); + let properties = stored.properties().unwrap(); + assert_eq!(properties["provider"], "daytona"); + assert_eq!(properties["duration_ms"], 2500); } #[test] @@ -1833,8 +2783,9 @@ mod tests { ); assert_eq!(stored.event_name(), "run.failed"); - assert_eq!(stored.properties["error"], "Handler error: boom"); - assert_eq!(stored.properties["duration_ms"], 900); + let properties = stored.properties().unwrap(); + assert_eq!(properties["error"], "Handler error: boom"); + assert_eq!(properties["duration_ms"], 900); } #[tokio::test] diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index 9240b9d29..d0ed3d401 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -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(); diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index d61ee3024..684ddf26a 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -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"); } } diff --git a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs index a319c88d9..62cc08b9b 100644 --- a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs @@ -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(), diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index f38e57d84..e009ef292 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -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 = 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.