diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index b93b9f923..5c3573d6c 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -338,11 +338,10 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option { EventBody::StageFailed(props) => Some(ProgressEvent::StageFailed { node_id, name: node_label, - error: props - .failure - .as_ref() - .map(|failure| failure.message.clone()) - .unwrap_or_else(|| "unknown error".to_string()), + error: props.failure.as_ref().map_or_else( + || "unknown error".to_string(), + |failure| failure.message.clone(), + ), }), EventBody::StageRetrying(props) => Some(ProgressEvent::StageRetrying { name: node_label, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 145e9d32d..eae6a5ab1 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -6,7 +6,10 @@ use chrono::{DateTime, Utc}; use serde_json::Value; use crate::{EventEnvelope, Result, RunSummary, StageId, StoreError}; -use fabro_types::run_event::{RunCompletedProps, RunFailedProps, StageCompletedProps}; +use fabro_types::run_event::{ + AgentCliStartedProps, AgentSessionStartedProps, CheckpointCompletedProps, RunCompletedProps, + RunFailedProps, StageCompletedProps, StagePromptProps, +}; use fabro_types::{ Checkpoint, Conclusion, EventBody, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro, RunEvent, RunId, RunRecord, RunStatus, RunStatusRecord, @@ -94,7 +97,7 @@ impl RunProjection { base_branch: props.base_branch.clone(), labels, }); - self.graph_source = props.workflow_source.clone(); + self.graph_source.clone_from(&props.workflow_source); } EventBody::RunStarted(props) => { self.start = Some(StartRecord { @@ -105,48 +108,24 @@ impl RunProjection { }); } EventBody::RunSubmitted(props) => { - self.status = Some(run_status_record( - RunStatus::Submitted, - props.reason.clone(), - ts, - )); + self.status = Some(run_status_record(RunStatus::Submitted, props.reason, ts)); } EventBody::RunStarting(props) => { - self.status = Some(run_status_record( - RunStatus::Starting, - props.reason.clone(), - ts, - )); + self.status = Some(run_status_record(RunStatus::Starting, props.reason, ts)); } EventBody::RunRunning(props) => { - self.status = Some(run_status_record( - RunStatus::Running, - props.reason.clone(), - ts, - )); + self.status = Some(run_status_record(RunStatus::Running, props.reason, ts)); } EventBody::RunRemoving(props) => { - self.status = Some(run_status_record( - RunStatus::Removing, - props.reason.clone(), - ts, - )); + self.status = Some(run_status_record(RunStatus::Removing, props.reason, ts)); } EventBody::RunCompleted(props) => { - self.status = Some(run_status_record( - RunStatus::Succeeded, - props.reason.clone(), - ts, - )); + self.status = Some(run_status_record(RunStatus::Succeeded, props.reason, ts)); self.conclusion = Some(conclusion_from_completed(props, ts)?); - self.final_patch = props.final_patch.clone(); + self.final_patch.clone_from(&props.final_patch); } EventBody::RunFailed(props) => { - self.status = Some(run_status_record( - RunStatus::Failed, - props.reason.clone(), - ts, - )); + self.status = Some(run_status_record(RunStatus::Failed, props.reason, ts)); self.conclusion = Some(conclusion_from_failed(props, ts)); } EventBody::RunRewound(_) => { @@ -177,10 +156,10 @@ impl RunProjection { }); } EventBody::RetroStarted(props) => { - self.retro_prompt = props.prompt.clone(); + self.retro_prompt.clone_from(&props.prompt); } EventBody::RetroCompleted(props) => { - self.retro_response = props.response.clone(); + self.retro_response.clone_from(&props.response); self.retro = props .retro .clone() @@ -400,10 +379,7 @@ fn run_status_record( } } -fn checkpoint_from_props( - props: &fabro_types::run_event::CheckpointCompletedProps, - timestamp: DateTime, -) -> Checkpoint { +fn checkpoint_from_props(props: &CheckpointCompletedProps, timestamp: DateTime) -> Checkpoint { let loop_failure_signatures = props .loop_failure_signatures .clone() @@ -531,7 +507,7 @@ fn node_status_from_outcome( } } -fn provider_used_from_prompt(props: &fabro_types::run_event::StagePromptProps) -> Option { +fn provider_used_from_prompt(props: &StagePromptProps) -> Option { let mut provider_used = serde_json::Map::new(); if let Some(mode) = props.mode.clone() { provider_used.insert("mode".to_string(), Value::String(mode)); @@ -545,9 +521,7 @@ fn provider_used_from_prompt(props: &fabro_types::run_event::StagePromptProps) - (!provider_used.is_empty()).then_some(Value::Object(provider_used)) } -fn provider_used_from_agent_session_started( - props: &fabro_types::run_event::AgentSessionStartedProps, -) -> Value { +fn provider_used_from_agent_session_started(props: &AgentSessionStartedProps) -> Value { let mut provider_used = serde_json::Map::new(); provider_used.insert("mode".to_string(), Value::String("agent".to_string())); if let Some(provider) = props.provider.clone() { @@ -559,9 +533,7 @@ fn provider_used_from_agent_session_started( Value::Object(provider_used) } -fn provider_used_from_agent_cli_started( - props: &fabro_types::run_event::AgentCliStartedProps, -) -> Value { +fn provider_used_from_agent_cli_started(props: &AgentCliStartedProps) -> Value { let mut provider_used = serde_json::Map::new(); provider_used.insert("mode".to_string(), Value::String("cli".to_string())); provider_used.insert( diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index a859ccdba..d9afd8841 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -74,7 +74,7 @@ impl TryFrom<&EventPayload> for RunEvent { type Error = StoreError; fn try_from(value: &EventPayload) -> Result { - RunEvent::from_ref(value.as_value()) + Self::from_ref(value.as_value()) .map_err(|err| StoreError::InvalidEvent(format!("invalid stored event: {err}"))) } } diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index 216bbe788..91d35e355 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -413,109 +413,109 @@ impl EventBody { } fn is_known_event_name(event: &str) -> bool { - match event { + matches!( + 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, - } + | "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" + ) } impl RunEvent { @@ -529,8 +529,8 @@ impl RunEvent { raw.node_label, raw.session_id, raw.parent_session_id, - raw.event, - raw.properties, + &raw.event, + &raw.properties, ) } @@ -572,8 +572,8 @@ impl RunEvent { obj.get("parent_session_id") .and_then(Value::as_str) .map(str::to_string), - event.to_string(), - properties, + event, + &properties, ) } @@ -585,8 +585,8 @@ impl RunEvent { node_label: Option, session_id: Option, parent_session_id: Option, - event: String, - properties: Value, + event: &str, + properties: &Value, ) -> serde_json::Result { let body_payload = json!({ "event": event, @@ -594,9 +594,9 @@ impl RunEvent { }); let body: EventBody = match serde_json::from_value(body_payload) { Ok(body) => body, - Err(err) if is_known_event_name(&event) => return Err(err), + Err(err) if is_known_event_name(event) => return Err(err), Err(_) => EventBody::Unknown { - name: event.clone(), + name: event.to_string(), properties: properties.clone(), }, }; diff --git a/lib/crates/fabro-util/src/warnings.rs b/lib/crates/fabro-util/src/warnings.rs index 95ec24f07..72ff7a0c4 100644 --- a/lib/crates/fabro-util/src/warnings.rs +++ b/lib/crates/fabro-util/src/warnings.rs @@ -34,24 +34,32 @@ mod tests { #[test] fn warn_user_once_deduplicates() { - let before = WARNINGS.lock().unwrap().len(); - warn_user_once!("dup-test-{}", "alpha"); - let after_first = WARNINGS.lock().unwrap().len(); - warn_user_once!("dup-test-{}", "alpha"); - let after_second = WARNINGS.lock().unwrap().len(); - assert_eq!(after_first, before + 1); - assert_eq!( - after_second, after_first, - "duplicate should not grow the set" + let message = "dup-test-alpha"; + WARNINGS.lock().unwrap().remove(message); + + warn_user_once!("{message}"); + warn_user_once!("{message}"); + + assert!( + WARNINGS.lock().unwrap().contains(message), + "warning should be recorded once in the set" ); } #[test] fn warn_user_once_different_messages() { - let before = WARNINGS.lock().unwrap().len(); - warn_user_once!("unique-msg-beta-1"); - warn_user_once!("unique-msg-beta-2"); - let after = WARNINGS.lock().unwrap().len(); - assert_eq!(after, before + 2); + let first = "unique-msg-beta-1"; + let second = "unique-msg-beta-2"; + let mut warnings = WARNINGS.lock().unwrap(); + warnings.remove(first); + warnings.remove(second); + drop(warnings); + + warn_user_once!("{first}"); + warn_user_once!("{second}"); + + let warnings = WARNINGS.lock().unwrap(); + assert!(warnings.contains(first)); + assert!(warnings.contains(second)); } } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 2e8543689..3428afdd8 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1209,17 +1209,10 @@ fn stage_status_from_string(status: &str) -> StageStatus { fn stored_event_fields(event: &Event) -> StoredEventFields { match event { - Event::StageCompleted { node_id, name, .. } | Event::StageFailed { node_id, name, .. } => { - let node_id = Some(node_id.clone()); - let node_label = default_node_label(node_id.as_ref(), Some(name.clone())); - StoredEventFields { - session_id: None, - parent_session_id: None, - node_id, - node_label, - } - } - Event::StageStarted { node_id, name, .. } | Event::StageRetrying { node_id, name, .. } => { + Event::StageCompleted { node_id, name, .. } + | Event::StageFailed { node_id, name, .. } + | Event::StageStarted { node_id, name, .. } + | Event::StageRetrying { node_id, name, .. } => { let node_id = Some(node_id.clone()); let node_label = default_node_label(node_id.as_ref(), Some(name.clone())); StoredEventFields { @@ -1362,24 +1355,16 @@ fn event_body_from_event(event: &Event) -> EventBody { goal: goal.clone(), }), Event::RunSubmitted { reason } => { - EventBody::RunSubmitted(fabro_types::RunStatusTransitionProps { - reason: reason.clone(), - }) + EventBody::RunSubmitted(fabro_types::RunStatusTransitionProps { reason: *reason }) } Event::RunStarting { reason } => { - EventBody::RunStarting(fabro_types::RunStatusTransitionProps { - reason: reason.clone(), - }) + EventBody::RunStarting(fabro_types::RunStatusTransitionProps { reason: *reason }) } Event::RunRunning { reason } => { - EventBody::RunRunning(fabro_types::RunStatusTransitionProps { - reason: reason.clone(), - }) + EventBody::RunRunning(fabro_types::RunStatusTransitionProps { reason: *reason }) } Event::RunRemoving { reason } => { - EventBody::RunRemoving(fabro_types::RunStatusTransitionProps { - reason: reason.clone(), - }) + EventBody::RunRemoving(fabro_types::RunStatusTransitionProps { reason: *reason }) } Event::RunRewound { target_checkpoint_ordinal, @@ -1407,7 +1392,7 @@ fn event_body_from_event(event: &Event) -> EventBody { duration_ms: *duration_ms, artifact_count: *artifact_count, status: status.clone(), - reason: reason.clone(), + reason: *reason, total_cost: *total_cost, final_git_commit_sha: final_git_commit_sha.clone(), final_patch: final_patch.clone(), @@ -1421,7 +1406,7 @@ fn event_body_from_event(event: &Event) -> EventBody { } => EventBody::RunFailed(fabro_types::RunFailedProps { error: error.to_string(), duration_ms: *duration_ms, - reason: reason.clone(), + reason: *reason, git_commit_sha: git_commit_sha.clone(), }), Event::RunNotice { diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index 5d87b5809..bead23a7d 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -81,7 +81,7 @@ impl WorkflowLifecycle { sandbox: &Arc, graph: Arc, run_dir: &PathBuf, - run_store: SlateRunStore, + run_store: &SlateRunStore, run_options: &Arc, is_resume: bool, on_node: crate::OnNodeCallback, diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index dc3cabe87..aa0ecddc3 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -469,7 +469,7 @@ impl RunSession { } event if matches!(&event.body, EventBody::GitCommit(_)) => { if let EventBody::GitCommit(props) = &event.body { - *sha_clone.lock().unwrap() = Some(props.sha.to_string()); + *sha_clone.lock().unwrap() = Some(props.sha.clone()); } } _ => {} diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index eb37f30c0..771cb5bd6 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -93,7 +93,7 @@ pub async fn execute(init: Initialized) -> Executed { &sandbox, graph_arc, &run_options.run_dir, - run_store.clone(), + &run_store, &settings_arc, checkpoint.is_some(), on_node,