diff --git a/crates/arc-llm/src/error.rs b/crates/arc-llm/src/error.rs index f3240cd73..02e42c995 100644 --- a/crates/arc-llm/src/error.rs +++ b/crates/arc-llm/src/error.rs @@ -142,22 +142,21 @@ impl SdkError { /// Whether this error is eligible for provider-level failover. /// - /// Transient provider issues (rate limits, server errors, quota, timeouts, - /// network, stream) can be retried on a different provider. Deterministic - /// errors (auth, invalid request, context length, content filter) and - /// non-provider errors (abort, configuration) cannot. + /// Includes everything that is `retryable()` (transient errors good for + /// same-provider retry) plus `QuotaExceeded` — a different provider won't + /// share the same quota. #[must_use] pub const fn failover_eligible(&self) -> bool { - match self { - Self::Provider { kind, .. } => matches!( - kind, - ProviderErrorKind::RateLimit - | ProviderErrorKind::Server - | ProviderErrorKind::QuotaExceeded - ), - Self::RequestTimeout { .. } | Self::Network { .. } | Self::Stream { .. } => true, - _ => false, + if self.retryable() { + return true; } + matches!( + self, + Self::Provider { + kind: ProviderErrorKind::QuotaExceeded, + .. + } + ) } #[must_use] diff --git a/crates/arc-workflows/src/cli/backend.rs b/crates/arc-workflows/src/cli/backend.rs index 4de32dd42..5ce209301 100644 --- a/crates/arc-workflows/src/cli/backend.rs +++ b/crates/arc-workflows/src/cli/backend.rs @@ -19,6 +19,81 @@ use crate::graph::Node; use crate::handler::codergen::{CodergenBackend, CodergenResult}; use crate::outcome::StageUsage; +fn build_profile(model: &str, provider: Provider) -> Box { + match provider { + Provider::OpenAi => Box::new(OpenAiProfile::new(model)), + Provider::Kimi | Provider::Zai | Provider::Minimax | Provider::Inception => { + Box::new(OpenAiProfile::new(model).with_provider(provider)) + } + Provider::Gemini => Box::new(GeminiProfile::new(model)), + Provider::Anthropic => Box::new(AnthropicProfile::new(model)), + } +} + +/// Spawn a task that subscribes to session events and: +/// 1. Tracks file changes (write_file/edit_file tool calls) into shared state. +/// 2. Forwards non-streaming agent events to the pipeline emitter. +fn spawn_event_forwarder( + session: &Session, + node_id: String, + emitter: Arc, + pending_tool_calls: Arc>>, + files_touched: Arc>>, +) { + let mut rx = session.subscribe(); + tokio::spawn(async move { + while let Ok(event) = rx.recv().await { + // Track file changes from tool calls + match &event.event { + AgentEvent::ToolCallStarted { + tool_name, + tool_call_id, + arguments, + } => { + if tool_name == "write_file" || tool_name == "edit_file" { + if let Some(path) = arguments.get("file_path").and_then(|v| v.as_str()) { + pending_tool_calls + .lock() + .unwrap() + .insert(tool_call_id.clone(), path.to_string()); + } + } + } + AgentEvent::ToolCallCompleted { + tool_call_id, + is_error, + .. + } => { + if !*is_error { + if let Some(path) = pending_tool_calls.lock().unwrap().remove(tool_call_id) { + files_touched.lock().unwrap().insert(path); + } + } else { + pending_tool_calls.lock().unwrap().remove(tool_call_id); + } + } + _ => {} + } + + // Forward non-streaming agent events to pipeline + if !matches!( + &event.event, + AgentEvent::SessionStarted + | AgentEvent::SessionEnded + | AgentEvent::AssistantTextStart + | AgentEvent::TextDelta { .. } + | AgentEvent::ToolCallOutputDelta { .. } + | AgentEvent::SkillExpanded { .. } + ) { + emitter.emit(&WorkflowRunEvent::Agent { + stage: node_id.clone(), + event: event.event.clone(), + }); + } + } + }); +} + /// LLM backend that delegates to an `agent` Session per invocation. /// /// For `full` fidelity nodes sharing a thread key, sessions are cached @@ -45,12 +120,27 @@ impl AgentApiBackend { &self, node: &Node, sandbox: &Arc, + ) -> Result { + Self::create_session_for( + &self.model, + self.provider, + node, + sandbox, + ) + .await + } + + async fn create_session_for( + model: &str, + provider: Provider, + node: &Node, + sandbox: &Arc, ) -> Result { let client = Client::from_env() .await .map_err(|e| ArcError::handler(format!("Failed to create LLM client: {e}")))?; - let mut profile = self.build_profile(); + let mut profile = build_profile(model, provider); let config = SessionConfig { max_tokens: node.max_tokens(), @@ -65,14 +155,13 @@ impl AgentApiBackend { // Build factory that creates child sessions WITHOUT subagent tools let factory_client = client.clone(); - let factory_provider = self.provider; - let factory_model = self.model.clone(); + let factory_model = model.to_string(); let factory_env = Arc::clone(sandbox); let factory: SessionFactory = Arc::new(move || { - let child_profile: Arc = match factory_provider { + let child_profile: Arc = match provider { Provider::OpenAi => Arc::new(OpenAiProfile::new(&factory_model)), Provider::Kimi | Provider::Zai | Provider::Minimax | Provider::Inception => { - Arc::new(OpenAiProfile::new(&factory_model).with_provider(factory_provider)) + Arc::new(OpenAiProfile::new(&factory_model).with_provider(provider)) } Provider::Gemini => Arc::new(GeminiProfile::new(&factory_model)), Provider::Anthropic => Arc::new(AnthropicProfile::new(&factory_model)), @@ -98,17 +187,6 @@ impl AgentApiBackend { Ok(session) } - - fn build_profile(&self) -> Box { - match self.provider { - Provider::OpenAi => Box::new(OpenAiProfile::new(&self.model)), - Provider::Kimi | Provider::Zai | Provider::Minimax | Provider::Inception => { - Box::new(OpenAiProfile::new(&self.model).with_provider(self.provider)) - } - Provider::Gemini => Box::new(GeminiProfile::new(&self.model)), - Provider::Anthropic => Box::new(AnthropicProfile::new(&self.model)), - } - } } #[async_trait] @@ -155,32 +233,32 @@ impl CodergenBackend for AgentApiBackend { } // Build per-request fallback chain: if the node overrides the provider, - // compute a fresh chain for that provider; otherwise use the backend's. - let fallback_chain = if node.llm_provider().is_some() { - // Node-level override: build fallback chain if we have fallback config - // For node overrides without explicit fallbacks, no failover is available. - Vec::new() + // no failover is available; otherwise use the backend's. + let fallback_chain: &[FallbackTarget] = if node.llm_provider().is_some() { + &[] } else { - self.fallback_chain.clone() + &self.fallback_chain }; let result = client.complete(&request).await; + let default_provider = self.provider.as_str().to_string(); + let (response, actual_model, actual_provider) = match result { Ok(resp) => ( resp, request.model.clone(), - request.provider.clone().unwrap_or_else(|| "anthropic".to_string()), + request.provider.clone().unwrap_or_else(|| default_provider.clone()), ), Err(sdk_err) if sdk_err.failover_eligible() && !fallback_chain.is_empty() => { let error_msg = sdk_err.to_string(); - let from_provider = request.provider.clone().unwrap_or_else(|| "anthropic".to_string()); + let from_provider = request.provider.clone().unwrap_or_else(|| default_provider.clone()); let from_model = request.model.clone(); let mut last_err = sdk_err; let mut found = None; - for target in &fallback_chain { + for target in fallback_chain { tracing::warn!( stage = node.id.as_str(), from_provider = from_provider.as_str(), @@ -287,67 +365,15 @@ impl CodergenBackend for AgentApiBackend { let pending_tool_calls: Arc>> = Arc::new(Mutex::new(HashMap::new())); let files_touched: Arc>> = Arc::new(Mutex::new(HashSet::new())); - let pending_clone = Arc::clone(&pending_tool_calls); - let files_clone = Arc::clone(&files_touched); - // Subscribe to session events: forward to pipeline emitter. - let node_id = node.id.clone(); - let pipeline_emitter = Arc::clone(emitter); - let mut rx = session.subscribe(); - tokio::spawn(async move { - use crate::event::WorkflowRunEvent; - while let Ok(event) = rx.recv().await { - // Track file changes from tool calls - match &event.event { - AgentEvent::ToolCallStarted { - tool_name, - tool_call_id, - arguments, - } => { - if tool_name == "write_file" || tool_name == "edit_file" { - if let Some(path) = arguments.get("file_path").and_then(|v| v.as_str()) - { - pending_clone - .lock() - .unwrap() - .insert(tool_call_id.clone(), path.to_string()); - } - } - } - AgentEvent::ToolCallCompleted { - tool_call_id, - is_error, - .. - } => { - if !*is_error { - if let Some(path) = pending_clone.lock().unwrap().remove(tool_call_id) { - files_clone.lock().unwrap().insert(path); - } - } else { - pending_clone.lock().unwrap().remove(tool_call_id); - } - } - _ => {} - } - - // Forward non-streaming agent events to pipeline - if !matches!( - &event.event, - AgentEvent::SessionStarted - | AgentEvent::SessionEnded - | AgentEvent::AssistantTextStart - | AgentEvent::TextDelta { .. } - | AgentEvent::ToolCallOutputDelta { .. } - | AgentEvent::SkillExpanded { .. } - ) { - pipeline_emitter.emit(&WorkflowRunEvent::Agent { - stage: node_id.clone(), - event: event.event.clone(), - }); - } - - } - }); + // Subscribe to session events: forward to pipeline emitter + track files. + spawn_event_forwarder( + &session, + node.id.clone(), + Arc::clone(emitter), + Arc::clone(&pending_tool_calls), + Arc::clone(&files_touched), + ); // Emit Prompt event before processing emitter.emit(&crate::event::WorkflowRunEvent::Prompt { @@ -392,13 +418,14 @@ impl CodergenBackend for AgentApiBackend { Err(_) => continue, }; - // Create a temporary backend with fallback provider/model for session creation - let fallback_backend = AgentApiBackend::new( - target.model.clone(), + let new_session = match Self::create_session_for( + &target.model, target_provider, - Vec::new(), - ); - let new_session = match fallback_backend.create_session(node, sandbox).await { + node, + sandbox, + ) + .await + { Ok(s) => s, Err(e) => { last_err = e; @@ -407,28 +434,14 @@ impl CodergenBackend for AgentApiBackend { }; session = new_session; - // Re-subscribe to forward events from the new session - let node_id2 = node.id.clone(); - let emitter2 = Arc::clone(emitter); - let mut rx2 = session.subscribe(); - tokio::spawn(async move { - while let Ok(event) = rx2.recv().await { - if !matches!( - &event.event, - AgentEvent::SessionStarted - | AgentEvent::SessionEnded - | AgentEvent::AssistantTextStart - | AgentEvent::TextDelta { .. } - | AgentEvent::ToolCallOutputDelta { .. } - | AgentEvent::SkillExpanded { .. } - ) { - emitter2.emit(&WorkflowRunEvent::Agent { - stage: node_id2.clone(), - event: event.event.clone(), - }); - } - } - }); + // Re-subscribe to forward events + track files from the new session + spawn_event_forwarder( + &session, + node.id.clone(), + Arc::clone(emitter), + Arc::clone(&pending_tool_calls), + Arc::clone(&files_touched), + ); session.initialize().await; match session.process_input(prompt).await { @@ -548,8 +561,7 @@ mod tests { #[test] fn build_profile_can_register_subagent_tools() { - let backend = AgentApiBackend::new("claude-opus-4-6".to_string(), Provider::Anthropic, Vec::new()); - let mut profile = backend.build_profile(); + let mut profile = build_profile("claude-opus-4-6", Provider::Anthropic); let manager = Arc::new(tokio::sync::Mutex::new(SubAgentManager::new(1))); let factory: SessionFactory = Arc::new(|| { panic!("factory should not be called in this test"); diff --git a/crates/arc-workflows/src/cli/run.rs b/crates/arc-workflows/src/cli/run.rs index 3d22b0b3a..fda27bae8 100644 --- a/crates/arc-workflows/src/cli/run.rs +++ b/crates/arc-workflows/src/cli/run.rs @@ -151,22 +151,18 @@ fn resolve_daytona_config( }) } -/// Resolve the fallback chain from config: task config wins, then run defaults. +/// Resolve the fallback chain from config. +/// +/// `apply_defaults` must be called on `run_cfg` before this — it merges +/// `run_defaults.llm.fallbacks` into `run_cfg.llm.fallbacks` already. fn resolve_fallback_chain( provider: Provider, model: &str, run_cfg: Option<&WorkflowRunConfig>, - run_defaults: &RunDefaults, ) -> Vec { let fallbacks = run_cfg .and_then(|c| c.llm.as_ref()) - .and_then(|l| l.fallbacks.as_ref()) - .or_else(|| { - run_defaults - .llm - .as_ref() - .and_then(|l| l.fallbacks.as_ref()) - }); + .and_then(|l| l.fallbacks.as_ref()); match fallbacks { Some(map) => arc_llm::catalog::build_fallback_chain(provider.as_str(), model, map), @@ -628,7 +624,6 @@ pub async fn run_command( provider_enum, &model, run_cfg.as_ref(), - &run_defaults, ); // 7. Build engine