From cffb00bcdbbd5b2f404c0c6052714533afe95603 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 2 Mar 2026 21:53:23 -0500 Subject: [PATCH] Carry typed errors through the entire stack instead of stringifying early - AgentError gains Clone + Serialize/Deserialize with tagged serde format - ArcError::Handler/Engine become struct variants with eager FailureClass classification via smart constructors handler()/engine() - Outcome replaces failure_reason: Option with failure: Option carrying message, failure_class, and failure_signature together - AgentEvent::Error/LlmRetry/SubAgentFailed carry typed AgentError/SdkError instead of pre-stringified error messages - WorkflowRunEvent::WorkflowRunFailed carries ArcError; StageFailed/StageCompleted carry FailureDetail instead of separate string fields - classify_outcome() trivially reads from FailureDetail; circuit breaker reads failure_signature from FailureDetail instead of context_updates hack - flatten_event() decomposes structured errors into flat fields for progress.jsonl backward compatibility Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/arc-agent/src/error.rs | 128 ++++++- crates/arc-agent/src/event.rs | 6 +- crates/arc-agent/src/session.rs | 11 +- crates/arc-agent/src/subagent.rs | 8 +- crates/arc-agent/src/types.rs | 110 +++++- crates/arc-workflows/src/artifact.rs | 14 +- crates/arc-workflows/src/cli/backend.rs | 6 +- crates/arc-workflows/src/cli/cli_backend.rs | 8 +- crates/arc-workflows/src/cli/mod.rs | 29 +- crates/arc-workflows/src/cli/run.rs | 8 +- crates/arc-workflows/src/condition.rs | 2 +- crates/arc-workflows/src/engine.rs | 136 +++----- crates/arc-workflows/src/error.rs | 328 +++++++++++++++--- crates/arc-workflows/src/event.rs | 104 ++++-- crates/arc-workflows/src/git.rs | 2 +- crates/arc-workflows/src/handler/codergen.rs | 4 +- crates/arc-workflows/src/handler/fan_in.rs | 7 +- .../arc-workflows/src/handler/manager_loop.rs | 25 +- crates/arc-workflows/src/handler/mod.rs | 4 +- crates/arc-workflows/src/handler/parallel.rs | 23 +- crates/arc-workflows/src/handler/script.rs | 19 +- .../arc-workflows/src/handler/wait_human.rs | 8 +- crates/arc-workflows/src/outcome.rs | 156 ++++++++- crates/arc-workflows/src/preamble.rs | 16 +- crates/arc-workflows/src/retro.rs | 2 +- .../tests/daytona_integration.rs | 4 +- crates/arc-workflows/tests/integration.rs | 71 ++-- 27 files changed, 899 insertions(+), 340 deletions(-) diff --git a/crates/arc-agent/src/error.rs b/crates/arc-agent/src/error.rs index dd118042f..b03db2181 100644 --- a/crates/arc-agent/src/error.rs +++ b/crates/arc-agent/src/error.rs @@ -1,6 +1,7 @@ use arc_llm::error::SdkError; -#[derive(Debug, thiserror::Error)] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, thiserror::Error)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum AgentError { #[error("LLM error: {0}")] Llm(#[from] SdkError), @@ -21,6 +22,7 @@ pub enum AgentError { #[cfg(test)] mod tests { use super::*; + use arc_llm::error::{ProviderErrorDetail, ProviderErrorKind}; #[test] fn agent_error_from_sdk_error() { @@ -55,4 +57,128 @@ mod tests { let err = AgentError::Aborted; assert_eq!(err.to_string(), "Aborted"); } + + // --- Serde roundtrip tests --- + + #[test] + fn serde_roundtrip_llm_network() { + let err = AgentError::Llm(SdkError::Network { + message: "connection refused".into(), + }); + let json = serde_json::to_string(&err).unwrap(); + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + #[test] + fn serde_roundtrip_llm_provider() { + let err = AgentError::Llm(SdkError::Provider { + kind: ProviderErrorKind::RateLimit, + detail: Box::new(ProviderErrorDetail { + message: "too fast".into(), + provider: "openai".into(), + status_code: Some(429), + error_code: None, + retry_after: Some(2.0), + raw: None, + }), + }); + let json = serde_json::to_string(&err).unwrap(); + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + #[test] + fn serde_roundtrip_session_closed() { + let err = AgentError::SessionClosed; + let json = serde_json::to_string(&err).unwrap(); + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + #[test] + fn serde_roundtrip_invalid_state() { + let err = AgentError::InvalidState("bad".into()); + let json = serde_json::to_string(&err).unwrap(); + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + #[test] + fn serde_roundtrip_tool_execution() { + let err = AgentError::ToolExecution("cmd failed".into()); + let json = serde_json::to_string(&err).unwrap(); + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + #[test] + fn serde_roundtrip_aborted() { + let err = AgentError::Aborted; + let json = serde_json::to_string(&err).unwrap(); + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + // --- Clone tests --- + + #[test] + fn clone_all_variants() { + let errors: Vec = vec![ + AgentError::Llm(SdkError::Network { + message: "refused".into(), + }), + AgentError::SessionClosed, + AgentError::InvalidState("reason".into()), + AgentError::ToolExecution("reason".into()), + AgentError::Aborted, + ]; + for err in &errors { + assert_eq!(err.to_string(), err.clone().to_string()); + } + } + + // --- Serde tag format tests --- + + #[test] + fn serde_tag_format_llm() { + let err = AgentError::Llm(SdkError::Network { + message: "refused".into(), + }); + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "llm"); + } + + #[test] + fn serde_tag_format_session_closed() { + let err = AgentError::SessionClosed; + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "session_closed"); + } + + #[test] + fn serde_tag_format_invalid_state() { + let err = AgentError::InvalidState("x".into()); + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "invalid_state"); + } + + #[test] + fn serde_tag_format_tool_execution() { + let err = AgentError::ToolExecution("x".into()); + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "tool_execution"); + } + + #[test] + fn serde_tag_format_aborted() { + let err = AgentError::Aborted; + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "aborted"); + } } diff --git a/crates/arc-agent/src/event.rs b/crates/arc-agent/src/event.rs index d6320cefe..348df9b67 100644 --- a/crates/arc-agent/src/event.rs +++ b/crates/arc-agent/src/event.rs @@ -61,13 +61,13 @@ mod tests { emitter.emit( "sess-2".into(), AgentEvent::Error { - error: "something went wrong".into(), + error: crate::error::AgentError::ToolExecution("something went wrong".into()), }, ); let event = receiver.recv().await.unwrap(); assert!( - matches!(&event.event, AgentEvent::Error { error } if error == "something went wrong") + matches!(&event.event, AgentEvent::Error { error } if error.to_string().contains("something went wrong")) ); } @@ -93,7 +93,7 @@ mod tests { emitter.emit( "sess-4".into(), AgentEvent::Error { - error: "test".into(), + error: crate::error::AgentError::ToolExecution("test".into()), }, ); } diff --git a/crates/arc-agent/src/session.rs b/crates/arc-agent/src/session.rs index 4ea16137f..3e0462512 100644 --- a/crates/arc-agent/src/session.rs +++ b/crates/arc-agent/src/session.rs @@ -429,7 +429,7 @@ impl Session { model: retry_model.clone(), attempt: attempt as usize, delay_secs: delay, - error: err.to_string(), + error: err.clone(), }, ); })), @@ -448,7 +448,7 @@ impl Session { self.event_emitter.emit( self.id.clone(), AgentEvent::Error { - error: err.to_string(), + error: AgentError::Llm(err.clone()), }, ); if is_auth_error(&err) { @@ -477,7 +477,7 @@ impl Session { self.event_emitter.emit( self.id.clone(), AgentEvent::Error { - error: err.to_string(), + error: AgentError::Llm(err.clone()), }, ); return Err(AgentError::Llm(err)); @@ -567,7 +567,7 @@ impl Session { self.event_emitter.emit( self.id.clone(), AgentEvent::Error { - error: format!("Context compaction failed: {e}"), + error: AgentError::InvalidState(format!("Context compaction failed: {e}")), }, ); } @@ -1793,7 +1793,8 @@ mod tests { let mut found_error = false; while let Ok(event) = rx.try_recv() { if let AgentEvent::Error { error } = &event.event { - if error.contains("compaction") || error.contains("summarization") { + let msg = error.to_string(); + if msg.contains("compaction") || msg.contains("summarization") { found_error = true; } } diff --git a/crates/arc-agent/src/subagent.rs b/crates/arc-agent/src/subagent.rs index 5526beeb3..8aaceb989 100644 --- a/crates/arc-agent/src/subagent.rs +++ b/crates/arc-agent/src/subagent.rs @@ -167,18 +167,18 @@ impl SubAgentManager { self.emit_event(AgentEvent::SubAgentFailed { agent_id: agent_id.to_string(), depth, - error: e.to_string(), + error: e.clone(), }); Err(e) } Err(e) => { - let error = format!("Agent task panicked: {e}"); + let error_msg = format!("Agent task panicked: {e}"); self.emit_event(AgentEvent::SubAgentFailed { agent_id: agent_id.to_string(), depth, - error: error.clone(), + error: AgentError::InvalidState(error_msg.clone()), }); - Err(AgentError::InvalidState(error)) + Err(AgentError::InvalidState(error_msg)) } }, None => Err(AgentError::InvalidState(format!( diff --git a/crates/arc-agent/src/types.rs b/crates/arc-agent/src/types.rs index 73a4ca5d3..dcb27f393 100644 --- a/crates/arc-agent/src/types.rs +++ b/crates/arc-agent/src/types.rs @@ -118,7 +118,7 @@ pub enum AgentEvent { is_error: bool, }, Error { - error: String, + error: crate::error::AgentError, }, ContextWindowWarning { estimated_tokens: usize, @@ -150,7 +150,7 @@ pub enum AgentEvent { model: String, attempt: usize, delay_secs: f64, - error: String, + error: arc_llm::error::SdkError, }, SubAgentSpawned { agent_id: String, @@ -166,7 +166,7 @@ pub enum AgentEvent { SubAgentFailed { agent_id: String, depth: usize, - error: String, + error: crate::error::AgentError, }, SubAgentClosed { agent_id: String, @@ -247,7 +247,7 @@ impl AgentEvent { ); } Self::Error { error } => { - error!(session_id, error, "Agent error"); + error!(session_id, error = %error, "Agent error"); } Self::ContextWindowWarning { estimated_tokens, @@ -313,7 +313,7 @@ impl AgentEvent { model, attempt, delay_secs, - error, + error = %error, "LLM request failed, retrying" ); } @@ -354,7 +354,7 @@ impl AgentEvent { session_id, agent_id, depth, - error, + error = %error, "Sub-agent failed" ); } @@ -483,7 +483,7 @@ mod tests { let event = AgentEvent::SubAgentFailed { agent_id: "sa-1".into(), depth: 0, - error: "timeout".into(), + error: crate::error::AgentError::ToolExecution("timeout".into()), }; assert!(matches!(event, AgentEvent::SubAgentFailed { depth: 0, .. })); } @@ -529,7 +529,7 @@ mod tests { AgentEvent::SubAgentFailed { agent_id: "sa-1".into(), depth: 0, - error: "oops".into(), + error: crate::error::AgentError::ToolExecution("oops".into()), }, AgentEvent::SubAgentClosed { agent_id: "sa-1".into(), @@ -647,4 +647,98 @@ mod tests { _ => panic!("expected AssistantMessage"), } } + + // --- Phase 4: Typed error event tests --- + + #[test] + fn error_event_serde_roundtrip_with_agent_error() { + let event = AgentEvent::Error { + error: crate::error::AgentError::Llm(arc_llm::error::SdkError::Network { + message: "refused".into(), + }), + }; + let json = serde_json::to_string(&event).unwrap(); + let deserialized: AgentEvent = serde_json::from_str(&json).unwrap(); + match deserialized { + AgentEvent::Error { error } => { + assert!(error.to_string().contains("refused")); + } + _ => panic!("expected Error variant"), + } + } + + #[test] + fn llm_retry_event_carries_sdk_error() { + use arc_llm::error::{ProviderErrorDetail, ProviderErrorKind}; + let event = AgentEvent::LlmRetry { + provider: "openai".into(), + model: "gpt-4".into(), + attempt: 1, + delay_secs: 2.0, + error: arc_llm::error::SdkError::Provider { + kind: ProviderErrorKind::RateLimit, + detail: Box::new(ProviderErrorDetail { + message: "too fast".into(), + provider: "openai".into(), + status_code: Some(429), + error_code: None, + retry_after: Some(2.0), + raw: None, + }), + }, + }; + let json = serde_json::to_string(&event).unwrap(); + let deserialized: AgentEvent = serde_json::from_str(&json).unwrap(); + match deserialized { + AgentEvent::LlmRetry { error, .. } => { + assert!(error.retryable()); + assert_eq!(error.retry_after(), Some(2.0)); + } + _ => panic!("expected LlmRetry variant"), + } + } + + #[test] + fn subagent_failed_carries_agent_error() { + let event = AgentEvent::SubAgentFailed { + agent_id: "sa-1".into(), + depth: 0, + error: crate::error::AgentError::ToolExecution("cmd failed".into()), + }; + let json = serde_json::to_string(&event).unwrap(); + let deserialized: AgentEvent = serde_json::from_str(&json).unwrap(); + match deserialized { + AgentEvent::SubAgentFailed { error, .. } => { + assert!(error.to_string().contains("cmd failed")); + } + _ => panic!("expected SubAgentFailed variant"), + } + } + + #[test] + fn error_event_preserves_error_type_through_json() { + let event = AgentEvent::Error { + error: crate::error::AgentError::ToolExecution("cmd failed".into()), + }; + let json = serde_json::to_string(&event).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + // The error field should contain the AgentError's tagged type + assert_eq!(v["Error"]["error"]["type"], "tool_execution"); + } + + #[test] + fn mcp_server_failed_still_string() { + let event = AgentEvent::McpServerFailed { + server_name: "broken".into(), + error: "connection refused".into(), + }; + let json = serde_json::to_string(&event).unwrap(); + let deserialized: AgentEvent = serde_json::from_str(&json).unwrap(); + match deserialized { + AgentEvent::McpServerFailed { error, .. } => { + assert_eq!(error, "connection refused"); + } + _ => panic!("expected McpServerFailed variant"), + } + } } diff --git a/crates/arc-workflows/src/artifact.rs b/crates/arc-workflows/src/artifact.rs index 08200d493..0d9f2200a 100644 --- a/crates/arc-workflows/src/artifact.rs +++ b/crates/arc-workflows/src/artifact.rs @@ -71,7 +71,7 @@ impl ArtifactStore { let id = id.into(); let name = name.into(); let serialized = serde_json::to_string(&data) - .map_err(|e| ArcError::Engine(format!("artifact serialize failed: {e}")))?; + .map_err(|e| ArcError::engine(format!("artifact serialize failed: {e}")))?; let size_bytes = serialized.len(); let is_file_backed = size_bytes > FILE_BACKING_THRESHOLD && self.base_dir.is_some(); @@ -117,7 +117,7 @@ impl ArtifactStore { let guard = self.artifacts.read().expect("artifact lock poisoned"); let (_, stored) = guard .get(id) - .ok_or_else(|| ArcError::Engine(format!("artifact not found: {id}")))?; + .ok_or_else(|| ArcError::engine(format!("artifact not found: {id}")))?; match stored { StoredData::InMemory(v) => Ok(v.clone()), @@ -125,10 +125,10 @@ impl ArtifactStore { let path = path.clone(); drop(guard); let data = std::fs::read_to_string(&path).map_err(|e| { - ArcError::Engine(format!("failed to read file-backed artifact {id}: {e}")) + ArcError::engine(format!("failed to read file-backed artifact {id}: {e}")) })?; serde_json::from_str(&data).map_err(|e| { - ArcError::Engine(format!( + ArcError::engine(format!( "failed to deserialize file-backed artifact {id}: {e}" )) }) @@ -275,14 +275,14 @@ pub async fn sync_artifacts_to_env( Ok(true) => continue, Ok(false) => {} Err(e) => { - return Err(ArcError::Engine(format!( + return Err(ArcError::engine(format!( "failed to check artifact existence: {e}" ))); } } let content = std::fs::read_to_string(&local_path).map_err(|e| { - ArcError::Engine(format!("failed to read local artifact {local_path}: {e}")) + ArcError::engine(format!("failed to read local artifact {local_path}: {e}")) })?; let filename = std::path::Path::new(&local_path) @@ -293,7 +293,7 @@ pub async fn sync_artifacts_to_env( let remote_path = format!("{}/.arc/artifacts/{filename}", env.working_directory()); env.write_file(&remote_path, &content).await.map_err(|e| { - ArcError::Engine(format!("failed to write artifact to remote env: {e}")) + ArcError::engine(format!("failed to write artifact to remote env: {e}")) })?; *value = Value::String(format!("{ARTIFACT_POINTER_PREFIX}{remote_path}")); diff --git a/crates/arc-workflows/src/cli/backend.rs b/crates/arc-workflows/src/cli/backend.rs index ad8b2e2ae..a4902a5f4 100644 --- a/crates/arc-workflows/src/cli/backend.rs +++ b/crates/arc-workflows/src/cli/backend.rs @@ -49,7 +49,7 @@ impl AgentApiBackend { ) -> Result { let client = Client::from_env() .await - .map_err(|e| ArcError::Handler(format!("Failed to create LLM client: {e}")))?; + .map_err(|e| ArcError::handler(format!("Failed to create LLM client: {e}")))?; let mut profile = self.build_profile(); @@ -122,7 +122,7 @@ impl CodergenBackend for AgentApiBackend { ) -> Result { let client = Client::from_env() .await - .map_err(|e| ArcError::Handler(format!("Failed to create LLM client: {e}")))?; + .map_err(|e| ArcError::handler(format!("Failed to create LLM client: {e}")))?; let model = node.llm_model().unwrap_or(&self.model); let provider = node @@ -331,7 +331,7 @@ impl CodergenBackend for AgentApiBackend { match e { AgentError::Llm(sdk_err) => ArcError::Llm(sdk_err), AgentError::Aborted => ArcError::Cancelled, - other => ArcError::Handler(format!("Agent session failed: {other}")), + other => ArcError::handler(format!("Agent session failed: {other}")), } }); diff --git a/crates/arc-workflows/src/cli/cli_backend.rs b/crates/arc-workflows/src/cli/cli_backend.rs index 6325bc241..b09e6cd48 100644 --- a/crates/arc-workflows/src/cli/cli_backend.rs +++ b/crates/arc-workflows/src/cli/cli_backend.rs @@ -299,7 +299,7 @@ impl CodergenBackend for AgentCliBackend { sandbox .write_file(prompt_path, prompt) .await - .map_err(|e| ArcError::Handler(format!("Failed to write prompt file: {e}")))?; + .map_err(|e| ArcError::handler(format!("Failed to write prompt file: {e}")))?; // 3. Build and execute CLI command let model = node.llm_model().unwrap_or(&self.model); @@ -323,7 +323,7 @@ impl CodergenBackend for AgentCliBackend { let result = sandbox .exec_command(&command, 600_000, None, None, None) .await - .map_err(|e| ArcError::Handler(format!("CLI command failed: {e}")))?; + .map_err(|e| ArcError::handler(format!("CLI command failed: {e}")))?; if let Ok(json) = serde_json::to_string_pretty(&serde_json::json!({ "exit_code": result.exit_code, @@ -335,7 +335,7 @@ impl CodergenBackend for AgentCliBackend { } if result.exit_code != 0 { - return Err(ArcError::Handler(format!( + return Err(ArcError::handler(format!( "CLI command exited with code {}: {}", result.exit_code, result.stderr.chars().take(500).collect::() @@ -344,7 +344,7 @@ impl CodergenBackend for AgentCliBackend { // 4. Parse the CLI output let parsed = parse_cli_response(provider, &result.stdout) - .ok_or_else(|| ArcError::Handler("Failed to parse CLI output".to_string()))?; + .ok_or_else(|| ArcError::handler("Failed to parse CLI output".to_string()))?; // 5. Detect changed files let files_after = self.detect_changed_files(sandbox).await; diff --git a/crates/arc-workflows/src/cli/mod.rs b/crates/arc-workflows/src/cli/mod.rs index 046f029a5..3f48c8a76 100644 --- a/crates/arc-workflows/src/cli/mod.rs +++ b/crates/arc-workflows/src/cli/mod.rs @@ -240,12 +240,11 @@ pub fn format_event_summary(event: &WorkflowRunEvent, styles: &Styles) -> String preferred_label, suggested_next_ids, usage, - failure_reason, + failure, notes, files_touched, attempt, max_attempts, - failure_class, } => { let mut s = format!("[STAGE_COMPLETED] node_id={node_id} name={name} index={index} duration={duration_ms}ms status={status}"); if let Some(label) = preferred_label { @@ -266,8 +265,9 @@ pub fn format_event_summary(event: &WorkflowRunEvent, styles: &Styles) -> String s.push_str(&format!(" tokens={tokens_str}")); } } - if let Some(reason) = failure_reason { - s.push_str(&format!(" failure_reason=\"{reason}\"")); + if let Some(ref f) = failure { + s.push_str(&format!(" failure_reason=\"{}\"", f.message)); + s.push_str(&format!(" failure_class={}", f.failure_class)); } if let Some(n) = notes { s.push_str(&format!(" notes=\"{n}\"")); @@ -276,30 +276,19 @@ pub fn format_event_summary(event: &WorkflowRunEvent, styles: &Styles) -> String s.push_str(&format!(" files_touched={}", files_touched.len())); } s.push_str(&format!(" attempt={attempt}/{max_attempts}")); - if let Some(fc) = failure_class { - s.push_str(&format!(" failure_class={fc}")); - } s } WorkflowRunEvent::StageFailed { node_id, name, index, - error, + failure, will_retry, - failure_reason, - failure_class, } => { - let mut s = format!( - "[STAGE_FAILED] node_id={node_id} name={name} index={index} error=\"{error}\" will_retry={will_retry}" - ); - if let Some(reason) = failure_reason { - s.push_str(&format!(" failure_reason=\"{reason}\"")); - } - if let Some(fc) = failure_class { - s.push_str(&format!(" failure_class={fc}")); - } - s + format!( + "[STAGE_FAILED] node_id={node_id} name={name} index={index} error=\"{}\" will_retry={will_retry} failure_class={}", + failure.message, failure.failure_class + ) } WorkflowRunEvent::StageRetrying { node_id, diff --git a/crates/arc-workflows/src/cli/run.rs b/crates/arc-workflows/src/cli/run.rs index c47bc8f5f..323b63da0 100644 --- a/crates/arc-workflows/src/cli/run.rs +++ b/crates/arc-workflows/src/cli/run.rs @@ -603,7 +603,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu { let (status, failure_reason) = match &engine_result { - Ok(o) => (o.status.to_string(), o.failure_reason.clone()), + Ok(o) => (o.status.to_string(), o.failure_reason().map(String::from)), Err(e) => ("fail".to_string(), Some(e.to_string())), }; let mut final_json = serde_json::json!({ @@ -623,7 +623,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu // Auto-derive retro (always, cheap) and optionally run retro agent { let (failed, failure_reason) = match &engine_result { - Ok(o) => (o.status == StageStatus::Fail, o.failure_reason.clone()), + Ok(o) => (o.status == StageStatus::Fail, o.failure_reason().map(String::from)), Err(e) => (true, Some(e.to_string())), }; generate_retro( @@ -699,7 +699,7 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu if let Some(notes) = &outcome.notes { eprintln!("Notes: {notes}"); } - if let Some(failure) = &outcome.failure_reason { + if let Some(failure) = outcome.failure_reason() { eprintln!( "{red}Failure: {failure}{reset}", red = styles.red, @@ -944,7 +944,7 @@ async fn run_from_branch( // Auto-derive retro { let (failed, failure_reason) = match &engine_result { - Ok(o) => (o.status == StageStatus::Fail, o.failure_reason.clone()), + Ok(o) => (o.status == StageStatus::Fail, o.failure_reason().map(String::from)), Err(e) => (true, Some(e.to_string())), }; diff --git a/crates/arc-workflows/src/condition.rs b/crates/arc-workflows/src/condition.rs index eba61f69c..5d994f91d 100644 --- a/crates/arc-workflows/src/condition.rs +++ b/crates/arc-workflows/src/condition.rs @@ -149,7 +149,7 @@ mod tests { suggested_next_ids: Vec::new(), context_updates: std::collections::HashMap::new(), notes: None, - failure_reason: None, + failure: None, usage: None, files_touched: Vec::new(), } diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs index 34faef828..489d0cfd5 100644 --- a/crates/arc-workflows/src/engine.rs +++ b/crates/arc-workflows/src/engine.rs @@ -18,7 +18,7 @@ use crate::asset_snapshot; use crate::checkpoint::Checkpoint; use crate::condition::evaluate_condition; use crate::context::Context; -use crate::error::{classify_failure_reason, ArcError, FailureClass, FailureSignature, Result}; +use crate::error::{ArcError, FailureClass, FailureSignature, Result}; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::graph::{Edge, Graph, Node}; use crate::handler::{EngineServices, HandlerRegistry}; @@ -43,20 +43,7 @@ fn classify_outcome(outcome: &Outcome) -> Option { match outcome.status { StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped => None, StageStatus::Fail | StageStatus::Retry => { - // Check handler hint in context_updates - if let Some(hint) = outcome.context_updates.get("failure_class") { - if let Some(s) = hint.as_str() { - let fc: FailureClass = s.parse().unwrap(); - return Some(fc); - } - } - - // Fall back to string heuristics on failure_reason - if let Some(ref reason) = outcome.failure_reason { - return Some(classify_failure_reason(reason)); - } - - Some(FailureClass::Deterministic) + outcome.failure_class().or(Some(FailureClass::Deterministic)) } } } @@ -344,7 +331,7 @@ fn write_node_status(logs_root: &Path, node_id: &str, visit: usize, outcome: &Ou let status = serde_json::json!({ "status": outcome.status.to_string(), "notes": outcome.notes, - "failure_reason": outcome.failure_reason, + "failure_reason": outcome.failure_reason(), "timestamp": Utc::now().to_rfc3339(), }); if let Ok(json) = serde_json::to_string_pretty(&status) { @@ -868,7 +855,7 @@ impl WorkflowRunEngine { let timed_result = if let Some(duration) = node_timeout { match tokio::time::timeout(duration, panic_safe).await { Ok(inner) => inner, - Err(_elapsed) => Ok(Ok(Outcome::fail(format!( + Err(_elapsed) => Ok(Ok(Outcome::fail_classify(format!( "handler timed out after {}ms", duration.as_millis() )))), @@ -889,7 +876,7 @@ impl WorkflowRunEngine { let panic_dir = node_dir(logs_root, &node.id, visit); let _ = std::fs::create_dir_all(&panic_dir); let _ = std::fs::write(panic_dir.join("panic.txt"), &msg); - Err(ArcError::Handler(msg)) + Err(ArcError::handler(msg)) } } }; @@ -945,10 +932,12 @@ impl WorkflowRunEngine { node_id: node.id.clone(), name: node.label().to_string(), index: stage_index, - error: e.to_string(), + failure: crate::outcome::FailureDetail { + message: e.to_string(), + failure_class: e.failure_class(), + failure_signature: e.failure_signature_hint(), + }, will_retry: true, - failure_reason: None, - failure_class: Some(e.failure_class().to_string()), }); self.services.emitter.emit(&WorkflowRunEvent::StageRetrying { node_id: node.id.clone(), @@ -998,12 +987,12 @@ impl WorkflowRunEngine { attempt, )); } - return Ok((Outcome::fail("max retries exceeded"), attempt)); + return Ok((Outcome::fail_classify("max retries exceeded"), attempt)); } } } - Ok((Outcome::fail("max retries exceeded"), policy.max_attempts)) + Ok((Outcome::fail_classify("max retries exceeded"), policy.max_attempts)) } /// Run the workflow. Returns the final outcome. @@ -1194,7 +1183,7 @@ impl WorkflowRunEngine { let start_node = graph .find_start_node() - .ok_or_else(|| ArcError::Engine("no start node found".to_string()))?; + .ok_or_else(|| ArcError::engine("no start node found".to_string()))?; current_node_id = start_node.id.clone(); } @@ -1256,7 +1245,7 @@ impl WorkflowRunEngine { let node = graph .nodes .get(¤t_node_id) - .ok_or_else(|| ArcError::Engine(format!("node not found: {current_node_id}")))?; + .ok_or_else(|| ArcError::engine(format!("node not found: {current_node_id}")))?; // Always track visit count (used for stage directory naming) let count = loop_state @@ -1266,7 +1255,7 @@ impl WorkflowRunEngine { *count += 1; if max_node_visits > 0 && *count >= max_node_visits { tracing::warn!(node = %current_node_id, visits = *count, limit = max_node_visits, "Node visit limit exceeded, run stuck in cycle"); - return Err(ArcError::Engine(format!( + return Err(ArcError::engine(format!( "node \"{}\" visited {count} times (limit {max_node_visits}); run is stuck in a cycle", current_node_id ))); @@ -1282,15 +1271,15 @@ impl WorkflowRunEngine { continue; } let duration_ms = millis_u64(run_start.elapsed()); - let error_msg = format!( - "goal gate unsatisfied for node {failed_node_id} and no retry target" + let error = ArcError::engine( + format!("goal gate unsatisfied for node {failed_node_id} and no retry target") ); self.services.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed { - error: error_msg.clone(), + error: error.clone(), duration_ms, git_commit_sha: last_git_sha.clone(), }); - return Ok((Outcome::fail(error_msg), context)); + return Ok((error.to_fail_outcome(), context)); } } } @@ -1357,7 +1346,7 @@ impl WorkflowRunEngine { node: node.id.clone(), idle_seconds: idle_secs, }); - return Err(ArcError::Engine(format!( + return Err(ArcError::engine(format!( "stall watchdog: node \"{}\" had no activity for {}s", node.id, idle_secs, ))); @@ -1394,14 +1383,14 @@ impl WorkflowRunEngine { // Circuit breaker: track deterministic/structural failure signatures let failure_sig = if let Some(fc) = outcome_failure_class { let sig_hint = outcome - .context_updates - .get("failure_signature") - .and_then(|v| v.as_str()); + .failure + .as_ref() + .and_then(|f| f.failure_signature.as_deref()); let sig = FailureSignature::new( &node.id, fc, sig_hint, - outcome.failure_reason.as_deref(), + outcome.failure_reason(), ); if fc.is_signature_tracked() { let count = loop_state @@ -1411,7 +1400,7 @@ impl WorkflowRunEngine { *count += 1; let limit = graph.loop_restart_signature_limit(); if *count >= limit { - return Err(ArcError::Engine(format!( + return Err(ArcError::engine(format!( "deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})" ))); } @@ -1426,14 +1415,10 @@ impl WorkflowRunEngine { node_id: node.id.clone(), name: node.label().to_string(), index: stage_index, - error: outcome - .failure_reason - .as_deref() - .unwrap_or("unknown") - .to_string(), + failure: outcome.failure.clone().unwrap_or_else(|| { + crate::outcome::FailureDetail::new("unknown", FailureClass::Deterministic) + }), will_retry: false, - failure_reason: outcome.failure_reason.clone(), - failure_class: outcome_failure_class.map(|fc| fc.to_string()), }); } else { self.services.emitter.emit(&WorkflowRunEvent::StageCompleted { @@ -1445,12 +1430,11 @@ impl WorkflowRunEngine { preferred_label: outcome.preferred_label.clone(), suggested_next_ids: outcome.suggested_next_ids.clone(), usage: outcome.usage.clone(), - failure_reason: outcome.failure_reason.clone(), + failure: outcome.failure.clone(), notes: outcome.notes.clone(), files_touched: outcome.files_touched.clone(), attempt: usize::try_from(attempts_used).unwrap_or(usize::MAX), max_attempts: usize::try_from(retry_policy.max_attempts).unwrap_or(usize::MAX), - failure_class: outcome_failure_class.map(|fc| fc.to_string()), }); self.inform(&format!("Stage completed: {}", node.label()), &node.id); } @@ -1648,14 +1632,15 @@ impl WorkflowRunEngine { continue; } let duration_ms = millis_u64(run_start.elapsed()); - let error_msg = - format!("stage {} failed with no outgoing fail edge", node.id); + let error = ArcError::engine( + format!("stage {} failed with no outgoing fail edge", node.id) + ); self.services.emitter.emit(&WorkflowRunEvent::WorkflowRunFailed { - error: error_msg.clone(), + error: error.clone(), duration_ms, git_commit_sha: last_git_sha.clone(), }); - return Err(ArcError::Engine(error_msg)); + return Err(error); } break; } @@ -1667,10 +1652,10 @@ impl WorkflowRunEngine { // Guard: only transient_infra failures may loop_restart (matches Kilroy) if let Some(fc) = outcome_failure_class { if fc != FailureClass::TransientInfra { - return Err(ArcError::Engine(format!( + return Err(ArcError::engine(format!( "loop_restart blocked: failure_class={fc} (requires transient_infra), node={}, failure_reason={}", node.id, - outcome.failure_reason.as_deref().unwrap_or("none"), + outcome.failure_reason().unwrap_or("none"), ))); } } @@ -1683,7 +1668,7 @@ impl WorkflowRunEngine { *count += 1; let limit = graph.loop_restart_signature_limit(); if *count >= limit { - return Err(ArcError::Engine(format!( + return Err(ArcError::engine(format!( "loop_restart circuit breaker: signature {sig} repeated {count} times (limit {limit})" ))); } @@ -1787,7 +1772,7 @@ mod tests { _logs_root: &Path, _services: &crate::handler::EngineServices, ) -> std::result::Result { - Ok(Outcome::fail("always fails")) + Ok(Outcome::fail_classify("always fails")) } } @@ -2258,7 +2243,7 @@ mod tests { g.nodes.insert("work".to_string(), n); let mut outcomes = HashMap::new(); - outcomes.insert("work".to_string(), Outcome::fail("test")); + outcomes.insert("work".to_string(), Outcome::fail_classify("test")); assert_eq!(check_goal_gates(&g, &outcomes), Err("work".to_string())); } @@ -2269,7 +2254,7 @@ mod tests { g.nodes.insert("work".to_string(), Node::new("work")); let mut outcomes = HashMap::new(); - outcomes.insert("work".to_string(), Outcome::fail("test")); + outcomes.insert("work".to_string(), Outcome::fail_classify("test")); assert!(check_goal_gates(&g, &outcomes).is_ok()); } @@ -3732,35 +3717,19 @@ mod tests { } #[test] - fn classify_outcome_respects_handler_hint() { - let mut outcome = Outcome::fail("some error"); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!("budget_exhausted"), - ); + fn classify_outcome_reads_failure_detail() { + let mut outcome = Outcome::fail_classify("some error"); + // Override the FailureDetail's class directly + outcome.failure.as_mut().unwrap().failure_class = FailureClass::BudgetExhausted; assert_eq!( classify_outcome(&outcome), Some(FailureClass::BudgetExhausted) ); } - #[test] - fn classify_outcome_unknown_hint_defaults_to_deterministic() { - let mut outcome = Outcome::fail("timeout occurred"); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!("not_a_valid_class"), - ); - // Unknown hint normalizes to Deterministic (fail-closed), taking priority over heuristics - assert_eq!( - classify_outcome(&outcome), - Some(FailureClass::Deterministic) - ); - } - #[test] fn classify_outcome_uses_failure_reason_heuristics() { - let outcome = Outcome::fail("rate limited by provider"); + let outcome = Outcome::fail_classify("rate limited by provider"); assert_eq!( classify_outcome(&outcome), Some(FailureClass::TransientInfra) @@ -3769,7 +3738,7 @@ mod tests { #[test] fn classify_outcome_defaults_to_deterministic() { - let outcome = Outcome::fail("something went wrong"); + let outcome = Outcome::fail_classify("something went wrong"); assert_eq!( classify_outcome(&outcome), Some(FailureClass::Deterministic) @@ -3780,7 +3749,7 @@ mod tests { fn classify_outcome_fail_no_reason_is_deterministic() { let outcome = Outcome { status: StageStatus::Fail, - failure_reason: None, + failure: None, ..Outcome::success() }; assert_eq!( @@ -3791,7 +3760,7 @@ mod tests { #[test] fn classify_outcome_retry_status_uses_heuristics() { - let outcome = Outcome::retry("connection refused"); + let outcome = Outcome::retry_classify("connection refused"); assert_eq!( classify_outcome(&outcome), Some(FailureClass::TransientInfra) @@ -3863,12 +3832,7 @@ mod tests { _logs_root: &Path, _services: &crate::handler::EngineServices, ) -> std::result::Result { - let mut outcome = Outcome::fail("connection refused"); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!("transient_infra"), - ); - Ok(outcome) + Ok(Outcome::fail_classify("connection refused")) } } @@ -3898,7 +3862,7 @@ mod tests { ) -> std::result::Result { let n = self.counter.fetch_add(1, Ordering::Relaxed); let reason = VARYING_REASONS[n % VARYING_REASONS.len()]; - Ok(Outcome::fail(reason)) + Ok(Outcome::fail_classify(reason)) } } diff --git a/crates/arc-workflows/src/error.rs b/crates/arc-workflows/src/error.rs index db089eccc..f4164f129 100644 --- a/crates/arc-workflows/src/error.rs +++ b/crates/arc-workflows/src/error.rs @@ -288,7 +288,8 @@ impl FailureClass { } } -#[derive(Error, Debug, Clone)] +#[derive(Error, Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum ArcError { #[error("Parse error: {0}")] Parse(String), @@ -296,11 +297,17 @@ pub enum ArcError { #[error("Validation error: {0}")] Validation(String), - #[error("Engine error: {0}")] - Engine(String), + #[error("Engine error: {message}")] + Engine { + message: String, + failure_class: FailureClass, + }, - #[error("Handler error: {0}")] - Handler(String), + #[error("Handler error: {message}")] + Handler { + message: String, + failure_class: FailureClass, + }, #[error("LLM error: {0}")] Llm(SdkError), @@ -319,6 +326,26 @@ pub enum ArcError { } impl ArcError { + /// Smart constructor for Handler errors. Classifies the failure reason eagerly. + pub fn handler(message: impl Into) -> Self { + let message = message.into(); + let failure_class = classify_failure_reason(&message); + Self::Handler { + message, + failure_class, + } + } + + /// Smart constructor for Engine errors. Classifies the failure reason eagerly. + pub fn engine(message: impl Into) -> Self { + let message = message.into(); + let failure_class = classify_failure_reason(&message); + Self::Engine { + message, + failure_class, + } + } + /// Whether this error category is retryable (transient) or terminal. /// /// Retryable: Handler (transient handler failures), Engine (could be transient), @@ -328,7 +355,7 @@ impl ArcError { #[must_use] pub fn is_retryable(&self) -> bool { match self { - Self::Handler(_) | Self::Engine(_) | Self::Io(_) => true, + Self::Handler { .. } | Self::Engine { .. } | Self::Io(_) => true, Self::Llm(sdk_err) => sdk_err.retryable(), Self::Parse(_) | Self::Validation(_) @@ -348,7 +375,9 @@ impl ArcError { Self::Parse(_) | Self::Validation(_) | Self::Stylesheet(_) | Self::Checkpoint(_) => { FailureClass::Deterministic } - Self::Handler(msg) | Self::Engine(msg) => classify_failure_reason(msg), + Self::Handler { failure_class, .. } | Self::Engine { failure_class, .. } => { + *failure_class + } } } @@ -361,21 +390,23 @@ impl ArcError { } } - /// Build an `Outcome::fail` with `failure_class` and optional `failure_signature` - /// populated in `context_updates`. + /// Build a fail `Outcome` with structured `FailureDetail`. pub fn to_fail_outcome(&self) -> crate::outcome::Outcome { - let mut outcome = crate::outcome::Outcome::fail(self.to_string()); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!(self.failure_class().to_string()), - ); - if let Some(sig) = self.failure_signature_hint() { - outcome.context_updates.insert( - "failure_signature".to_string(), - serde_json::json!(sig), - ); + let failure = crate::outcome::FailureDetail { + message: self.to_string(), + failure_class: self.failure_class(), + failure_signature: self.failure_signature_hint(), + }; + crate::outcome::Outcome { + status: crate::outcome::StageStatus::Fail, + preferred_label: None, + suggested_next_ids: Vec::new(), + context_updates: std::collections::HashMap::new(), + notes: None, + failure: Some(failure), + usage: None, + files_touched: Vec::new(), } - outcome } } @@ -412,13 +443,13 @@ mod tests { #[test] fn engine_error_display() { - let err = ArcError::Engine("no outgoing edge".to_string()); + let err = ArcError::engine("no outgoing edge"); assert_eq!(err.to_string(), "Engine error: no outgoing edge"); } #[test] fn handler_error_display() { - let err = ArcError::Handler("LLM call failed".to_string()); + let err = ArcError::handler("LLM call failed"); assert_eq!(err.to_string(), "Handler error: LLM call failed"); } @@ -472,8 +503,8 @@ mod tests { #[test] fn is_retryable_transient_errors() { - assert!(ArcError::Handler("timeout".to_string()).is_retryable()); - assert!(ArcError::Engine("transient".to_string()).is_retryable()); + assert!(ArcError::handler("timeout").is_retryable()); + assert!(ArcError::engine("transient").is_retryable()); assert!(ArcError::Io("connection reset".to_string()).is_retryable()); } @@ -698,7 +729,7 @@ mod tests { #[test] fn failure_class_handler_with_timeout() { assert_eq!( - ArcError::Handler("request timed out".into()).failure_class(), + ArcError::handler("request timed out").failure_class(), FailureClass::TransientInfra ); } @@ -706,7 +737,7 @@ mod tests { #[test] fn failure_class_handler_deterministic() { assert_eq!( - ArcError::Handler("invalid configuration".into()).failure_class(), + ArcError::handler("invalid configuration").failure_class(), FailureClass::Deterministic ); } @@ -1460,13 +1491,13 @@ mod tests { #[test] fn failure_signature_hint_handler_returns_none() { - let err = ArcError::Handler("something failed".to_string()); + let err = ArcError::handler("something failed"); assert_eq!(err.failure_signature_hint(), None); } #[test] fn failure_signature_hint_engine_returns_none() { - let err = ArcError::Engine("engine error".to_string()); + let err = ArcError::engine("engine error"); assert_eq!(err.failure_signature_hint(), None); } @@ -1480,26 +1511,22 @@ mod tests { }); let outcome = err.to_fail_outcome(); assert_eq!(outcome.status, crate::outcome::StageStatus::Fail); + let failure = outcome.failure.as_ref().unwrap(); + assert_eq!(failure.failure_class, FailureClass::Deterministic); assert_eq!( - outcome.context_updates.get("failure_class"), - Some(&serde_json::json!("deterministic")) - ); - assert_eq!( - outcome.context_updates.get("failure_signature"), - Some(&serde_json::json!("api_deterministic|openai|authentication")) + failure.failure_signature.as_deref(), + Some("api_deterministic|openai|authentication") ); } #[test] fn to_fail_outcome_handler_has_class_but_no_signature() { - let err = ArcError::Handler("connection refused".to_string()); + let err = ArcError::handler("connection refused"); let outcome = err.to_fail_outcome(); assert_eq!(outcome.status, crate::outcome::StageStatus::Fail); - assert_eq!( - outcome.context_updates.get("failure_class"), - Some(&serde_json::json!("transient_infra")) - ); - assert!(!outcome.context_updates.contains_key("failure_signature")); + let failure = outcome.failure.as_ref().unwrap(); + assert_eq!(failure.failure_class, FailureClass::TransientInfra); + assert!(failure.failure_signature.is_none()); } #[test] @@ -1508,10 +1535,221 @@ mod tests { message: "connection refused".into(), }); let outcome = err.to_fail_outcome(); - assert!(outcome - .failure_reason - .as_ref() - .unwrap() - .contains("connection refused")); + assert!(outcome.failure_reason().unwrap().contains("connection refused")); + } + + #[test] + fn to_fail_outcome_no_context_updates() { + let err = ArcError::Llm(SdkError::Network { + message: "refused".into(), + }); + let outcome = err.to_fail_outcome(); + assert!(outcome.context_updates.is_empty()); + } + + // --- Phase 2: Eager classification tests --- + + #[test] + fn handler_eager_classification() { + let err = ArcError::handler("connection refused"); + assert_eq!(err.failure_class(), FailureClass::TransientInfra); + } + + #[test] + fn handler_eager_classification_roundtrip() { + let err = ArcError::handler("connection refused"); + let json = serde_json::to_string(&err).unwrap(); + let deserialized: ArcError = serde_json::from_str(&json).unwrap(); + assert_eq!(deserialized.failure_class(), FailureClass::TransientInfra); + } + + #[test] + fn handler_smart_constructor_preserves_message() { + let err = ArcError::handler("some error"); + assert!(err.to_string().contains("some error")); + } + + #[test] + fn engine_eager_classification() { + let err = ArcError::engine("rate limit exceeded"); + assert_eq!(err.failure_class(), FailureClass::TransientInfra); + } + + #[test] + fn arc_error_serde_roundtrip_all_variants() { + let errors: Vec = vec![ + ArcError::Parse("bad".into()), + ArcError::Validation("bad".into()), + ArcError::engine("engine err"), + ArcError::handler("handler err"), + ArcError::Llm(SdkError::Network { + message: "refused".into(), + }), + ArcError::Checkpoint("cp err".into()), + ArcError::Stylesheet("style err".into()), + ArcError::Io("io err".into()), + ArcError::Cancelled, + ]; + for err in &errors { + let json = serde_json::to_string(err).unwrap(); + let deserialized: ArcError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + } + + #[test] + fn handler_display_unchanged() { + assert_eq!( + ArcError::handler("LLM call failed").to_string(), + "Handler error: LLM call failed" + ); + } + + #[test] + fn engine_display_unchanged() { + assert_eq!( + ArcError::engine("no outgoing edge").to_string(), + "Engine error: no outgoing edge" + ); + } + + #[test] + fn failure_class_stability() { + let messages = [ + "connection refused", + "timeout", + "rate limit", + "context length exceeded", + "cancel", + "invalid configuration", + "write_scope_violation", + ]; + for msg in messages { + assert_eq!( + ArcError::handler(msg).failure_class(), + classify_failure_reason(msg), + "mismatch for message: {msg}" + ); + } + } + + #[test] + fn to_fail_outcome_preserves_class() { + let err = ArcError::handler("timeout"); + let outcome = err.to_fail_outcome(); + assert_eq!( + outcome.failure_class(), + Some(FailureClass::TransientInfra) + ); + } + + // --- E2E error pipeline tests --- + + #[test] + fn e2e_llm_error_to_outcome_to_event_preserves_classification() { + use crate::event::WorkflowRunEvent; + use crate::outcome::FailureDetail; + + // 1. Create SdkError → ArcError + let sdk_err = SdkError::Provider { + kind: ProviderErrorKind::RateLimit, + detail: Box::new(ProviderErrorDetail::new("too fast", "openai")), + }; + let arc_err = ArcError::Llm(sdk_err); + assert_eq!(arc_err.failure_class(), FailureClass::TransientInfra); + + // 2. ArcError → Outcome + let outcome = arc_err.to_fail_outcome(); + assert_eq!(outcome.failure_class(), Some(FailureClass::TransientInfra)); + + // 3. Outcome → StageFailed event + let failure = outcome.failure.clone().unwrap(); + let event = WorkflowRunEvent::StageFailed { + node_id: "code".into(), + name: "code".into(), + index: 0, + failure: failure.clone(), + will_retry: false, + }; + + // 4. Verify classification survived all the way through + match &event { + WorkflowRunEvent::StageFailed { failure, .. } => { + assert_eq!(failure.failure_class, FailureClass::TransientInfra); + } + _ => panic!("expected StageFailed"), + } + } + + #[test] + fn e2e_handler_error_classified_at_edge() { + // handler smart constructor classifies eagerly + let err = ArcError::handler("connection refused"); + assert_eq!(err.failure_class(), FailureClass::TransientInfra); + + // to_fail_outcome preserves + let outcome = err.to_fail_outcome(); + assert_eq!(outcome.failure_class(), Some(FailureClass::TransientInfra)); + + // event preserves + let failure = outcome.failure.unwrap(); + assert_eq!(failure.failure_class, FailureClass::TransientInfra); + } + + #[test] + fn e2e_handler_retryable_checks() { + assert!(ArcError::handler("timeout").is_retryable()); + assert!(ArcError::handler("auth error").is_retryable()); + } + + #[test] + fn e2e_serde_stability_arc_error() { + let err = ArcError::handler("connection refused"); + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + + // Verify wire format + assert_eq!(v["type"], "handler"); + assert!(v["data"]["message"].as_str().unwrap().contains("connection refused")); + assert_eq!(v["data"]["failure_class"], "transient_infra"); + + // Round-trip + let deserialized: ArcError = serde_json::from_str(&json).unwrap(); + assert_eq!(deserialized.failure_class(), FailureClass::TransientInfra); + } + + #[test] + fn e2e_serde_stability_agent_error() { + use arc_agent::error::AgentError; + + let err = AgentError::Llm(SdkError::Provider { + kind: ProviderErrorKind::RateLimit, + detail: Box::new(ProviderErrorDetail::new("too fast", "openai")), + }); + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "llm"); + + let deserialized: AgentError = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + + #[test] + fn e2e_failure_detail_in_outcome_serde_roundtrip() { + use crate::outcome::{FailureDetail, Outcome}; + + let outcome = Outcome::fail_classify("rate limit exceeded") + .with_signature(Some("api_transient|openai|rate_limited")); + + let json = serde_json::to_string(&outcome).unwrap(); + let deserialized: Outcome = serde_json::from_str(&json).unwrap(); + + let failure = deserialized.failure.unwrap(); + assert_eq!(failure.message, "rate limit exceeded"); + assert_eq!(failure.failure_class, FailureClass::TransientInfra); + assert_eq!( + failure.failure_signature.as_deref(), + Some("api_transient|openai|rate_limited") + ); } } diff --git a/crates/arc-workflows/src/event.rs b/crates/arc-workflows/src/event.rs index f7bf5eb81..178efaf6a 100644 --- a/crates/arc-workflows/src/event.rs +++ b/crates/arc-workflows/src/event.rs @@ -27,7 +27,7 @@ pub enum WorkflowRunEvent { final_git_commit_sha: Option, }, WorkflowRunFailed { - error: String, + error: crate::error::ArcError, duration_ms: u64, #[serde(default, skip_serializing_if = "Option::is_none")] git_commit_sha: Option, @@ -49,21 +49,19 @@ pub enum WorkflowRunEvent { preferred_label: Option, suggested_next_ids: Vec, usage: Option, - failure_reason: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + failure: Option, notes: Option, files_touched: Vec, attempt: usize, max_attempts: usize, - failure_class: Option, }, StageFailed { node_id: String, name: String, index: usize, - error: String, + failure: crate::outcome::FailureDetail, will_retry: bool, - failure_reason: Option, - failure_class: Option, }, StageRetrying { node_id: String, @@ -206,7 +204,7 @@ impl WorkflowRunEvent { Self::WorkflowRunFailed { error, duration_ms, .. } => { - error!(error, duration_ms, "Workflow run failed"); + error!(error = %error, duration_ms, "Workflow run failed"); } Self::StageStarted { node_id, @@ -251,16 +249,16 @@ impl WorkflowRunEvent { node_id, name, index, - error, + failure, will_retry, - .. } => { + let error_msg = &failure.message; if *will_retry { warn!( node_id, stage = name.as_str(), index, - error, + error = error_msg.as_str(), will_retry, "Stage failed" ); @@ -269,7 +267,7 @@ impl WorkflowRunEvent { node_id, stage = name.as_str(), index, - error, + error = error_msg.as_str(), will_retry, "Stage failed" ); @@ -640,7 +638,41 @@ fn rename_fields(event_name: &str, fields: &mut serde_json::Map) -> ArcError { - ArcError::Engine(msg.into()) + ArcError::engine(msg.into()) } /// Return a pre-configured `git` command with auto-maintenance disabled. diff --git a/crates/arc-workflows/src/handler/codergen.rs b/crates/arc-workflows/src/handler/codergen.rs index 89e75cee0..65f054fee 100644 --- a/crates/arc-workflows/src/handler/codergen.rs +++ b/crates/arc-workflows/src/handler/codergen.rs @@ -681,7 +681,7 @@ mod tests { _stage_dir: &Path, _sandbox: &Arc, ) -> Result { - Err(ArcError::Handler("Request timed out".to_string())) + Err(ArcError::handler("Request timed out".to_string())) } } @@ -927,7 +927,7 @@ Some text in between. .await .unwrap(); assert_eq!(outcome.status, crate::outcome::StageStatus::Fail); - assert!(outcome.failure_reason.unwrap().contains("bad config")); + assert!(outcome.failure_reason().unwrap().contains("bad config")); } #[tokio::test] diff --git a/crates/arc-workflows/src/handler/fan_in.rs b/crates/arc-workflows/src/handler/fan_in.rs index 8c5b79296..ae8665ddc 100644 --- a/crates/arc-workflows/src/handler/fan_in.rs +++ b/crates/arc-workflows/src/handler/fan_in.rs @@ -37,7 +37,7 @@ impl Handler for FanInHandler { ) -> Result { let results = context.get("parallel.results"); let Some(results) = results else { - return Ok(Outcome::fail("No parallel results to evaluate")); + return Ok(Outcome::fail_deterministic("No parallel results to evaluate")); }; let prompt = node.prompt().filter(|p| !p.is_empty()); @@ -69,7 +69,7 @@ impl Handler for FanInHandler { }; if all_failed { - return Ok(Outcome::fail("all candidates failed")); + return Ok(Outcome::fail_deterministic("all candidates failed")); } // --- Fast-forward to winner's HEAD when git isolation is active --- @@ -502,8 +502,7 @@ mod tests { .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); assert!(outcome - .failure_reason - .as_deref() + .failure_reason() .unwrap() .contains("all candidates failed")); } diff --git a/crates/arc-workflows/src/handler/manager_loop.rs b/crates/arc-workflows/src/handler/manager_loop.rs index 44e2c5926..5f7aa245d 100644 --- a/crates/arc-workflows/src/handler/manager_loop.rs +++ b/crates/arc-workflows/src/handler/manager_loop.rs @@ -49,9 +49,9 @@ fn read_child_dot(node: &Node) -> Result { } if let Some(path) = node.attrs.get("stack.child_dotfile").and_then(|v| v.as_str()) { return std::fs::read_to_string(path) - .map_err(|e| ArcError::Handler(format!("Failed to read child dotfile {path}: {e}"))); + .map_err(|e| ArcError::handler(format!("Failed to read child dotfile {path}: {e}"))); } - Err(ArcError::Handler("No child DOT source".to_string())) + Err(ArcError::handler("No child DOT source".to_string())) } /// Compute the context diff: keys that changed or were added relative to `before`. @@ -107,13 +107,13 @@ impl Handler for SubWorkflowHandler { // Read and parse child DOT source let dot_source = match read_child_dot(node) { Ok(s) => s, - Err(e) => return Ok(Outcome::fail(e.to_string())), + Err(e) => return Ok(Outcome::fail_classify(e.to_string())), }; let child_graph = match prepare_workflow(&dot_source) { Ok(g) => g, Err(e) => { - return Ok(Outcome::fail(format!( + return Ok(Outcome::fail_classify(format!( "Failed to parse child pipeline: {e}" ))) } @@ -164,8 +164,8 @@ impl Handler for SubWorkflowHandler { // Child finished let (child_outcome, child_final_context) = match result { Ok(Ok(pair)) => pair, - Ok(Err(e)) => return Ok(Outcome::fail(format!("Child engine error: {e}"))), - Err(e) => return Ok(Outcome::fail(format!("Child task panicked: {e}"))), + Ok(Err(e)) => return Ok(Outcome::fail_classify(format!("Child engine error: {e}"))), + Err(e) => return Ok(Outcome::fail_classify(format!("Child task panicked: {e}"))), }; // Compute context diff @@ -180,7 +180,7 @@ impl Handler for SubWorkflowHandler { }; if child_outcome.status == StageStatus::Fail { - outcome.failure_reason = child_outcome.failure_reason; + outcome.failure = child_outcome.failure.clone(); } return Ok(outcome); @@ -211,7 +211,7 @@ impl Handler for SubWorkflowHandler { child_cancel.store(true, Ordering::Relaxed); let _ = tokio::time::timeout(Duration::from_millis(100), &mut child_handle).await; - Ok(Outcome::fail(format!( + Ok(Outcome::fail_classify(format!( "Max cycles ({max_cycles}) exceeded for manager loop node: {}", node.id ))) @@ -293,8 +293,7 @@ mod tests { .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); assert!(outcome - .failure_reason - .as_deref() + .failure_reason() .unwrap() .contains("No child DOT source")); } @@ -324,8 +323,7 @@ mod tests { .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); assert!(outcome - .failure_reason - .as_deref() + .failure_reason() .unwrap() .contains("Failed to parse child pipeline")); } @@ -503,8 +501,7 @@ mod tests { .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); assert!(outcome - .failure_reason - .as_deref() + .failure_reason() .unwrap() .contains("Max cycles")); } diff --git a/crates/arc-workflows/src/handler/mod.rs b/crates/arc-workflows/src/handler/mod.rs index 468205e51..dadbe3715 100644 --- a/crates/arc-workflows/src/handler/mod.rs +++ b/crates/arc-workflows/src/handler/mod.rs @@ -225,7 +225,7 @@ mod tests { let handler = TestHandler { _name: "test".to_string(), }; - assert!(handler.should_retry(&ArcError::Handler("timeout".to_string()))); + assert!(handler.should_retry(&ArcError::handler("timeout".to_string()))); assert!(!handler.should_retry(&ArcError::Parse("bad".to_string()))); } @@ -252,7 +252,7 @@ mod tests { #[test] fn custom_should_retry_override() { let handler = NeverRetryHandler; - assert!(!handler.should_retry(&ArcError::Handler("timeout".to_string()))); + assert!(!handler.should_retry(&ArcError::handler("timeout".to_string()))); assert!(!handler.should_retry(&ArcError::Io("connection reset".to_string()))); } diff --git a/crates/arc-workflows/src/handler/parallel.rs b/crates/arc-workflows/src/handler/parallel.rs index d9c595b81..a648db3a6 100644 --- a/crates/arc-workflows/src/handler/parallel.rs +++ b/crates/arc-workflows/src/handler/parallel.rs @@ -199,7 +199,7 @@ impl Handler for ParallelHandler { let parallel_start = Instant::now(); let branches = graph.outgoing_edges(&node.id); if branches.is_empty() { - return Ok(Outcome::fail("No branches for parallel node")); + return Ok(Outcome::fail_classify("No branches for parallel node")); } let join_policy = parse_join_policy( @@ -306,7 +306,7 @@ impl Handler for ParallelHandler { }) .await .map_err(|e| { - ArcError::Handler(format!("worktree setup join error: {e}")) + ArcError::handler(format!("worktree setup join error: {e}")) })??; branch_context.set( "internal.work_dir", @@ -332,7 +332,7 @@ impl Handler for ParallelHandler { ) .await; if !ok { - return Err(ArcError::Handler(format!( + return Err(ArcError::handler(format!( "failed to create remote branch {branch_name}" ))); } @@ -343,7 +343,7 @@ impl Handler for ParallelHandler { ) .await; if !ok { - return Err(ArcError::Handler(format!( + return Err(ArcError::handler(format!( "failed to add remote worktree {wt_path_str}" ))); } @@ -355,7 +355,7 @@ impl Handler for ParallelHandler { .exec_command(&reset_cmd, 30_000, Some(&wt_path_str), None, None) .await; if !matches!(reset_result, Ok(ref r) if r.exit_code == 0) { - return Err(ArcError::Handler(format!( + return Err(ArcError::handler(format!( "failed to reset remote worktree {wt_path_str}" ))); } @@ -396,7 +396,7 @@ impl Handler for ParallelHandler { let _permit = sem .acquire() .await - .map_err(|e| ArcError::Handler(format!("semaphore error: {e}")))?; + .map_err(|e| ArcError::handler(format!("semaphore error: {e}")))?; emitter.emit(&WorkflowRunEvent::ParallelBranchStarted { branch: setup.target_id.clone(), @@ -406,7 +406,7 @@ impl Handler for ParallelHandler { let Some(target_node) = graph.nodes.get(&setup.target_id) else { let outcome = - Outcome::fail(format!("branch target node not found: {}", setup.target_id)); + Outcome::fail_classify(format!("branch target node not found: {}", setup.target_id)); emitter.emit(&WorkflowRunEvent::ParallelBranchCompleted { branch: setup.target_id.clone(), index: setup.branch_index, @@ -534,7 +534,7 @@ impl Handler for ParallelHandler { Err(join_err) => { let result = BranchResult { id: String::new(), - outcome: Outcome::fail(format!("task join error: {join_err}")), + outcome: Outcome::fail_classify(format!("task join error: {join_err}")), head_sha: None, worktree_path: None, }; @@ -694,8 +694,11 @@ impl Handler for ParallelHandler { notes: Some(format!( "Parallel node dispatched {total} branches ({success_count} succeeded, {fail_count} failed)" )), - failure_reason: if is_fail { - Some(format!("Join policy not satisfied: {success_count}/{total} succeeded")) + failure: if is_fail { + Some(crate::outcome::FailureDetail::new( + format!("Join policy not satisfied: {success_count}/{total} succeeded"), + crate::error::FailureClass::Deterministic, + )) } else { None }, diff --git a/crates/arc-workflows/src/handler/script.rs b/crates/arc-workflows/src/handler/script.rs index 28a82accf..1c4b34112 100644 --- a/crates/arc-workflows/src/handler/script.rs +++ b/crates/arc-workflows/src/handler/script.rs @@ -34,7 +34,7 @@ impl Handler for ScriptHandler { .unwrap_or(""); if script.is_empty() { - return Ok(Outcome::fail("No script specified")); + return Ok(Outcome::fail_classify("No script specified")); } let language = node @@ -44,7 +44,7 @@ impl Handler for ScriptHandler { .unwrap_or("shell"); if language != "shell" && language != "python" { - return Ok(Outcome::fail(format!( + return Ok(Outcome::fail_classify(format!( "Invalid language: {language:?} (expected \"shell\" or \"python\")" ))); } @@ -94,7 +94,7 @@ impl Handler for ScriptHandler { ) .await?; - return Err(ArcError::Handler(format!( + return Err(ArcError::handler(format!( "Script timed out after {}ms: {script}", timeout_dur.as_millis() ))); @@ -148,7 +148,7 @@ impl Handler for ScriptHandler { reason.push_str("\n\n## stderr\n"); reason.push_str(&stderr); } - let mut outcome = Outcome::fail(reason); + let mut outcome = Outcome::fail_classify(reason); outcome .context_updates .insert("script.output".to_string(), serde_json::json!(stdout)); @@ -158,7 +158,7 @@ impl Handler for ScriptHandler { Ok(outcome) } } - Err(e) => Err(ArcError::Handler(format!("Failed to spawn script: {e}"))), + Err(e) => Err(ArcError::handler(format!("Failed to spawn script: {e}"))), } } } @@ -198,7 +198,7 @@ mod tests { .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); assert_eq!( - outcome.failure_reason.as_deref(), + outcome.failure_reason(), Some("No script specified") ); } @@ -540,8 +540,7 @@ mod tests { .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); assert!(outcome - .failure_reason - .as_deref() + .failure_reason() .unwrap() .contains("Invalid language")); } @@ -633,7 +632,7 @@ mod tests { .await .unwrap(); assert_eq!(outcome.status, StageStatus::Fail); - let reason = outcome.failure_reason.as_deref().unwrap(); + let reason = outcome.failure_reason().unwrap(); assert!( reason.contains("build output"), "failure_reason should contain stdout, got: {reason}" @@ -659,7 +658,7 @@ mod tests { // // Pragmatic approach: verify the error construction matches what the // handler produces. The timeout test covers the other Err path. - let err = ArcError::Handler(format!("Failed to spawn script: {}", "No such file")); + let err = ArcError::handler(format!("Failed to spawn script: {}", "No such file")); assert!(err.to_string().contains("Failed to spawn script")); } diff --git a/crates/arc-workflows/src/handler/wait_human.rs b/crates/arc-workflows/src/handler/wait_human.rs index 0e693be2a..f2c3cda5b 100644 --- a/crates/arc-workflows/src/handler/wait_human.rs +++ b/crates/arc-workflows/src/handler/wait_human.rs @@ -127,7 +127,7 @@ impl Handler for WaitHumanHandler { } if choices.is_empty() && freeform_target.is_none() { - return Ok(Outcome::fail("No outgoing edges for human gate")); + return Ok(Outcome::fail_deterministic("No outgoing edges for human gate")); } // 2. Build question @@ -172,12 +172,12 @@ impl Handler for WaitHumanHandler { default_target, )); } - return Ok(Outcome::retry("human gate timeout, no default")); + return Ok(Outcome::retry_classify("human gate timeout, no default")); } // 5. Handle skipped if answer.value == AnswerValue::Skipped { - return Ok(Outcome::fail("human skipped interaction")); + return Ok(Outcome::fail_deterministic("human skipped interaction")); } // Emit interview completed for successful interactions @@ -219,7 +219,7 @@ impl Handler for WaitHumanHandler { return Ok(make_choice_outcome(&first.key, &first.label, &first.to)); } - Ok(Outcome::fail("No matching choice")) + Ok(Outcome::fail_deterministic("No matching choice")) } } diff --git a/crates/arc-workflows/src/outcome.rs b/crates/arc-workflows/src/outcome.rs index 003421915..ec2b10fd1 100644 --- a/crates/arc-workflows/src/outcome.rs +++ b/crates/arc-workflows/src/outcome.rs @@ -4,6 +4,8 @@ use std::str::FromStr; use serde::{Deserialize, Serialize}; +use crate::error::{classify_failure_reason, FailureClass}; + /// Status of a pipeline stage execution. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -59,6 +61,25 @@ pub struct StageUsage { pub cost: Option, } +/// Structured failure information carried through the pipeline. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct FailureDetail { + pub message: String, + pub failure_class: FailureClass, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure_signature: Option, +} + +impl FailureDetail { + pub fn new(message: impl Into, failure_class: FailureClass) -> Self { + Self { + message: message.into(), + failure_class, + failure_signature: None, + } + } +} + /// The result of executing a node handler. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Outcome { @@ -72,7 +93,7 @@ pub struct Outcome { #[serde(default, skip_serializing_if = "Option::is_none")] pub notes: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub failure_reason: Option, + pub failure: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub usage: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] @@ -88,38 +109,67 @@ impl Outcome { suggested_next_ids: Vec::new(), context_updates: HashMap::new(), notes: None, - failure_reason: None, + failure: None, usage: None, files_touched: Vec::new(), } } - pub fn fail(reason: impl Into) -> Self { + /// Create a failed outcome with a deterministic failure class. + pub fn fail_deterministic(reason: impl Into) -> Self { Self { status: StageStatus::Fail, preferred_label: None, suggested_next_ids: Vec::new(), context_updates: HashMap::new(), notes: None, - failure_reason: Some(reason.into()), + failure: Some(FailureDetail::new(reason, FailureClass::Deterministic)), usage: None, files_touched: Vec::new(), } } - pub fn retry(reason: impl Into) -> Self { + /// Create a failed outcome with the failure class inferred from the message via heuristics. + pub fn fail_classify(reason: impl Into) -> Self { + let reason = reason.into(); + let failure_class = classify_failure_reason(&reason); + Self { + status: StageStatus::Fail, + preferred_label: None, + suggested_next_ids: Vec::new(), + context_updates: HashMap::new(), + notes: None, + failure: Some(FailureDetail::new(reason, failure_class)), + usage: None, + files_touched: Vec::new(), + } + } + + /// Create a retry outcome with the failure class inferred from the message via heuristics. + pub fn retry_classify(reason: impl Into) -> Self { + let reason = reason.into(); + let failure_class = classify_failure_reason(&reason); Self { status: StageStatus::Retry, preferred_label: None, suggested_next_ids: Vec::new(), context_updates: HashMap::new(), notes: None, - failure_reason: Some(reason.into()), + failure: Some(FailureDetail::new(reason, failure_class)), usage: None, files_touched: Vec::new(), } } + /// Set the failure signature on this outcome. Returns self for chaining. + #[must_use] + pub fn with_signature(mut self, sig: Option>) -> Self { + if let Some(ref mut f) = self.failure { + f.failure_signature = sig.map(Into::into); + } + self + } + #[must_use] pub fn skipped() -> Self { Self { @@ -128,11 +178,21 @@ impl Outcome { suggested_next_ids: Vec::new(), context_updates: HashMap::new(), notes: None, - failure_reason: None, + failure: None, usage: None, files_touched: Vec::new(), } } + + /// Get the failure reason message, if any. + pub fn failure_reason(&self) -> Option<&str> { + self.failure.as_ref().map(|f| f.message.as_str()) + } + + /// Get the failure class, if this is a failed outcome. + pub fn failure_class(&self) -> Option { + self.failure.as_ref().map(|f| f.failure_class) + } } #[cfg(test)] @@ -179,28 +239,94 @@ mod tests { assert!(o.suggested_next_ids.is_empty()); assert!(o.context_updates.is_empty()); assert!(o.notes.is_none()); - assert!(o.failure_reason.is_none()); + assert!(o.failure.is_none()); } #[test] - fn outcome_fail_factory() { - let o = Outcome::fail("something broke"); + fn outcome_fail_deterministic_factory() { + let o = Outcome::fail_deterministic("something broke"); assert_eq!(o.status, StageStatus::Fail); - assert_eq!(o.failure_reason.as_deref(), Some("something broke")); + assert_eq!(o.failure_reason(), Some("something broke")); + assert_eq!(o.failure_class(), Some(FailureClass::Deterministic)); } #[test] - fn outcome_retry_factory() { - let o = Outcome::retry("try again"); + fn outcome_fail_classify_factory() { + let o = Outcome::fail_classify("connection refused"); + assert_eq!(o.status, StageStatus::Fail); + assert_eq!(o.failure_reason(), Some("connection refused")); + assert_eq!(o.failure_class(), Some(FailureClass::TransientInfra)); + } + + #[test] + fn outcome_retry_classify_factory() { + let o = Outcome::retry_classify("try again"); assert_eq!(o.status, StageStatus::Retry); - assert_eq!(o.failure_reason.as_deref(), Some("try again")); + assert_eq!(o.failure_reason(), Some("try again")); } #[test] fn outcome_skipped_factory() { let o = Outcome::skipped(); assert_eq!(o.status, StageStatus::Skipped); - assert!(o.failure_reason.is_none()); + assert!(o.failure.is_none()); + } + + #[test] + fn failure_detail_construction() { + let fd = FailureDetail::new("timeout", FailureClass::TransientInfra); + assert_eq!(fd.message, "timeout"); + assert_eq!(fd.failure_class, FailureClass::TransientInfra); + assert!(fd.failure_signature.is_none()); + } + + #[test] + fn failure_detail_serde_roundtrip() { + let fd = FailureDetail { + message: "timeout".into(), + failure_class: FailureClass::TransientInfra, + failure_signature: Some("sig".into()), + }; + let json = serde_json::to_string(&fd).unwrap(); + let deserialized: FailureDetail = serde_json::from_str(&json).unwrap(); + assert_eq!(deserialized.message, "timeout"); + assert_eq!(deserialized.failure_class, FailureClass::TransientInfra); + assert_eq!(deserialized.failure_signature.as_deref(), Some("sig")); + } + + #[test] + fn fail_classify_known_patterns() { + assert_eq!( + Outcome::fail_classify("timeout").failure_class(), + Some(FailureClass::TransientInfra) + ); + assert_eq!( + Outcome::fail_classify("context length exceeded").failure_class(), + Some(FailureClass::BudgetExhausted) + ); + assert_eq!( + Outcome::fail_classify("cancel").failure_class(), + Some(FailureClass::Canceled) + ); + } + + #[test] + fn failure_field_is_some_for_failures() { + assert!(Outcome::fail_deterministic("x").failure.is_some()); + } + + #[test] + fn failure_field_is_none_for_success() { + assert!(Outcome::success().failure.is_none()); + } + + #[test] + fn with_signature_builder() { + let o = Outcome::fail_deterministic("x").with_signature(Some("sig")); + assert_eq!( + o.failure.as_ref().unwrap().failure_signature.as_deref(), + Some("sig") + ); } #[test] diff --git a/crates/arc-workflows/src/preamble.rs b/crates/arc-workflows/src/preamble.rs index baa961621..c10f170c1 100644 --- a/crates/arc-workflows/src/preamble.rs +++ b/crates/arc-workflows/src/preamble.rs @@ -270,7 +270,7 @@ fn render_summary_high_stage_section( if let Some(notes) = outcome.notes.as_deref() { lines.push(format!("- Notes: {notes}")); } - if let Some(reason) = outcome.failure_reason.as_deref() { + if let Some(reason) = outcome.failure_reason() { lines.push(format!("- Failure reason: {reason}")); } } @@ -435,7 +435,7 @@ fn build_summary_preamble( if let Some(notes) = outcome.notes.as_deref() { line.push_str(&format!(" ({notes})")); } - if let Some(reason) = outcome.failure_reason.as_deref() { + if let Some(reason) = outcome.failure_reason() { line.push_str(&format!(" [reason: {reason}]")); } parts.push(line); @@ -475,7 +475,7 @@ fn build_summary_preamble( if let Some(notes) = outcome.notes.as_deref() { line.push_str(&format!(" ({notes})")); } - if let Some(reason) = outcome.failure_reason.as_deref() { + if let Some(reason) = outcome.failure_reason() { line.push_str(&format!(" [reason: {reason}]")); } parts.push(line); @@ -601,7 +601,7 @@ mod tests { let completed_nodes = vec!["plan".to_string(), "code".to_string()]; let mut node_outcomes: HashMap = HashMap::new(); node_outcomes.insert("plan".to_string(), Outcome::success()); - node_outcomes.insert("code".to_string(), Outcome::fail("compilation error")); + node_outcomes.insert("code".to_string(), Outcome::fail_classify("compilation error")); let preamble = build_preamble( "compact", @@ -952,7 +952,7 @@ mod tests { let mut node_outcomes: HashMap = HashMap::new(); node_outcomes.insert("plan".to_string(), Outcome::success()); node_outcomes.insert("code".to_string(), Outcome::success()); - node_outcomes.insert("test".to_string(), Outcome::fail("test failure")); + node_outcomes.insert("test".to_string(), Outcome::fail_classify("test failure")); let preamble = build_preamble( "summary:low", @@ -990,7 +990,7 @@ mod tests { node_outcomes.insert("step1".to_string(), Outcome::success()); node_outcomes.insert("step2".to_string(), Outcome::success()); node_outcomes.insert("step3".to_string(), Outcome::success()); - node_outcomes.insert("step4".to_string(), Outcome::fail("error")); + node_outcomes.insert("step4".to_string(), Outcome::fail_classify("error")); let preamble = build_preamble( "summary:low", @@ -1033,7 +1033,7 @@ mod tests { let context = Context::new(); let completed_nodes = vec!["run_tests".to_string()]; let mut node_outcomes: HashMap = HashMap::new(); - let mut outcome = Outcome::fail("exit code 1"); + let mut outcome = Outcome::fail_classify("exit code 1"); outcome.context_updates.insert( "script.output".to_string(), serde_json::json!("test failed"), @@ -1330,7 +1330,7 @@ mod tests { let context = Context::new(); let completed_nodes = vec!["work".to_string()]; let mut node_outcomes: HashMap = HashMap::new(); - node_outcomes.insert("work".to_string(), Outcome::fail("connection timeout")); + node_outcomes.insert("work".to_string(), Outcome::fail_classify("connection timeout")); let preamble = build_preamble( "summary:high", diff --git a/crates/arc-workflows/src/retro.rs b/crates/arc-workflows/src/retro.rs index dc0f4441d..0fb3b20c4 100644 --- a/crates/arc-workflows/src/retro.rs +++ b/crates/arc-workflows/src/retro.rs @@ -254,7 +254,7 @@ pub fn derive_retro( retries, cost, notes: outcome.and_then(|o| o.notes.clone()), - failure_reason: outcome.and_then(|o| o.failure_reason.clone()), + failure_reason: outcome.and_then(|o| o.failure_reason().map(String::from)), files_touched: files, }); } diff --git a/crates/arc-workflows/tests/daytona_integration.rs b/crates/arc-workflows/tests/daytona_integration.rs index 57f51f3fd..76b20f829 100644 --- a/crates/arc-workflows/tests/daytona_integration.rs +++ b/crates/arc-workflows/tests/daytona_integration.rs @@ -664,7 +664,7 @@ async fn daytona_parallel_git_branching_e2e() { outcome.status, StageStatus::Success, "pipeline failed: {:?}", - outcome.failure_reason + outcome.failure_reason() ); // Verify parallel.results has head_sha for each branch @@ -1052,7 +1052,7 @@ impl Handler for AssetCreatorHandler { .sandbox .exec_command(script, 30_000, None, None, None) .await - .map_err(|e| ArcError::Handler(format!("exec failed: {e}")))?; + .map_err(|e| ArcError::handler(format!("exec failed: {e}")))?; Ok(Outcome::success()) } } diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs index a97f23e1e..59a48029c 100644 --- a/crates/arc-workflows/tests/integration.rs +++ b/crates/arc-workflows/tests/integration.rs @@ -492,7 +492,7 @@ impl Handler for AlwaysFailHandler { _logs_root: &Path, _services: &arc_workflows::handler::EngineServices, ) -> Result { - Ok(Outcome::fail(format!("forced failure for {}", node.id))) + Ok(Outcome::fail_classify(format!("forced failure for {}", node.id))) } } @@ -575,7 +575,7 @@ async fn goal_gate_routes_to_retry_target_on_failure() { StageStatus::Fail, "pipeline outcome should be 'fail' when goal gate unsatisfied" ); - let failure_reason = outcome.failure_reason.unwrap_or_default(); + let failure_reason = outcome.failure_reason().unwrap_or_default(); assert!( failure_reason.contains("goal gate unsatisfied"), "failure_reason should mention goal gate, got: {failure_reason}" @@ -616,7 +616,7 @@ async fn goal_gate_routes_to_retry_target_when_present() { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if count == 0 { - Ok(Outcome::fail("first attempt fails")) + Ok(Outcome::fail_classify("first attempt fails")) } else { Ok(Outcome::success()) } @@ -929,7 +929,7 @@ async fn retry_on_failure_then_succeed() { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if count == 0 { - Ok(Outcome::retry("transient failure")) + Ok(Outcome::retry_classify("transient failure")) } else { Ok(Outcome::success()) } @@ -1179,12 +1179,8 @@ impl Handler for CounterHandler { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if count == 0 { - let mut outcome = Outcome::fail("first call fails"); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!("transient_infra"), - ); - Ok(outcome) + // Use a message that heuristics classify as transient_infra + Ok(Outcome::fail_classify("connection refused")) } else { Ok(Outcome::success()) } @@ -2037,7 +2033,7 @@ async fn branching_loop_back_on_failure() { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if count == 0 { - Ok(Outcome::fail("first attempt fails")) + Ok(Outcome::fail_classify("first attempt fails")) } else { Ok(Outcome::success()) } @@ -2354,7 +2350,7 @@ async fn scenario_node_retries_on_retry_status() { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if count == 0 { - Ok(Outcome::retry("transient failure")) + Ok(Outcome::retry_classify("transient failure")) } else { Ok(Outcome::success()) } @@ -2761,8 +2757,7 @@ async fn manager_loop_max_cycles_exceeded_e2e() { let manager_outcome = cp.node_outcomes.get("manager").expect("manager outcome"); assert_eq!(manager_outcome.status, StageStatus::Fail); assert!(manager_outcome - .failure_reason - .as_deref() + .failure_reason() .unwrap() .contains("Max cycles")); // Overall pipeline outcome is from last completed node (manager) = Fail @@ -5518,7 +5513,7 @@ mod real_llm { .client .complete(&request) .await - .map_err(|e| ArcError::Handler(e.to_string()))?; + .map_err(|e| ArcError::handler(e.to_string()))?; Ok(CodergenResult::Text { text: response.text(), usage: None, @@ -7813,7 +7808,7 @@ async fn node_dir_uses_visit_count_on_revisit() { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if n == 0 { - Ok(Outcome::fail("first attempt fails")) + Ok(Outcome::fail_classify("first attempt fails")) } else { Ok(Outcome::success()) } @@ -8998,7 +8993,7 @@ impl Handler for FileWriterHandler { .sandbox .write_file(&file_path, &format!("written by {}", node.id)) .await - .map_err(|e| ArcError::Handler(format!("write_file failed: {e}")))?; + .map_err(|e| ArcError::handler(format!("write_file failed: {e}")))?; Ok(Outcome::success()) } } @@ -9488,7 +9483,7 @@ async fn parallel_git_branching_host_e2e() { outcome.status, StageStatus::Success, "pipeline failed: {:?}", - outcome.failure_reason + outcome.failure_reason() ); // 6. Verify parallel.results has head_sha for each branch @@ -9671,11 +9666,11 @@ impl Handler for DeterministicFailHandler { _logs_root: &Path, _services: &arc_workflows::handler::EngineServices, ) -> Result { - Ok(Outcome::fail(&self.reason)) + Ok(Outcome::fail_classify(&self.reason)) } } -/// Handler that always fails with a transient_infra classification hint. +/// Handler that always fails with a transient_infra classification. struct TransientInfraFailHandler; #[async_trait::async_trait] @@ -9688,16 +9683,11 @@ impl Handler for TransientInfraFailHandler { _logs_root: &Path, _services: &arc_workflows::handler::EngineServices, ) -> Result { - let mut outcome = Outcome::fail("connection refused"); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!("transient_infra"), - ); - Ok(outcome) + Ok(Outcome::fail_classify("connection refused")) } } -/// Handler that provides an explicit `failure_signature` hint in context_updates. +/// Handler that provides an explicit `failure_signature` hint via FailureDetail. struct SignatureHintHandler; #[async_trait::async_trait] @@ -9710,12 +9700,8 @@ impl Handler for SignatureHintHandler { _logs_root: &Path, _services: &arc_workflows::handler::EngineServices, ) -> Result { - let mut outcome = Outcome::fail("error at line 42 in commit abc123def0"); - outcome.context_updates.insert( - "failure_signature".to_string(), - serde_json::json!("custom-grouping-key"), - ); - Ok(outcome) + Ok(Outcome::fail_classify("error at line 42 in commit abc123def0") + .with_signature(Some("custom-grouping-key"))) } } @@ -9750,7 +9736,7 @@ impl Handler for VaryingReasonFailHandler { let n = self .counter .fetch_add(1, std::sync::atomic::Ordering::SeqCst) as usize; - Ok(Outcome::fail( + Ok(Outcome::fail_classify( E2E_VARYING_REASONS[n % E2E_VARYING_REASONS.len()], )) } @@ -9778,7 +9764,7 @@ impl Handler for SucceedOnNthHandler { if n >= self.succeed_on { Ok(Outcome::success()) } else { - Ok(Outcome::fail("not yet ready")) + Ok(Outcome::fail_classify("not yet ready")) } } } @@ -10683,11 +10669,12 @@ impl Handler for ClassifiedFailHandler { if n >= self.succeed_on { return Ok(Outcome::success()); } - let mut outcome = Outcome::fail("classified failure"); - outcome.context_updates.insert( - "failure_class".to_string(), - serde_json::json!(self.failure_class), - ); + let failure_class: arc_workflows::error::FailureClass = + self.failure_class.parse().unwrap(); + let mut outcome = Outcome::fail_classify("classified failure"); + if let Some(ref mut f) = outcome.failure { + f.failure_class = failure_class; + } Ok(outcome) } } @@ -11205,10 +11192,10 @@ impl Handler for AssetCreatorHandler { .sandbox .exec_command(script, 30_000, None, None, None) .await - .map_err(|e| ArcError::Handler(format!("exec failed: {e}")))?; + .map_err(|e| ArcError::handler(format!("exec failed: {e}")))?; if self.should_fail { - Ok(Outcome::fail("intentional failure")) + Ok(Outcome::fail_classify("intentional failure")) } else { Ok(Outcome::success()) }