diff --git a/assets/agents/coder/graph.yaml b/assets/agents/coder/graph.yaml index 06e5697..a70d317 100644 --- a/assets/agents/coder/graph.yaml +++ b/assets/agents/coder/graph.yaml @@ -31,7 +31,7 @@ settings: max_loop_iterations: 20 log_state_snapshots: true validate_before_run: true - timeout: 1800 + timeout: 14400 initial_state: project_dir: '' @@ -90,6 +90,7 @@ nodes: Project directory: {{project_dir}} prompt: '{{initial_prompt}}' tools: [] + timeout: 300 output_schema: type: object properties: @@ -254,6 +255,7 @@ nodes: - fs_patch - execute_command max_iterations: 100 + timeout: 1800 state_updates: last_node_output: '{{output}}' fallback: end_failure @@ -327,6 +329,7 @@ nodes: - fs_ls - execute_command max_iterations: 15 + timeout: 600 output_schema: type: object properties: @@ -384,4 +387,4 @@ nodes: {{build_output}} Last tests output: - {{tests_output}} + {{tests_output}} \ No newline at end of file diff --git a/config.example.yaml b/config.example.yaml index 6f6ef9e..c93a509 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -292,6 +292,7 @@ clients: # extra: # proxy: socks5://127.0.0.1:1080 # Set proxy # connect_timeout: 10 # Set timeout in seconds for connect to api + # read_timeout: 300 # Set timeout in seconds for a read stall (no bytes received); 0 disables (default: 300) # See https://platform.openai.com/docs/quickstart - type: openai @@ -525,4 +526,4 @@ clients: - type: openai-compatible name: voyageai api_base: https://api.voyageai.com/v1 - api_key: '{{VOYAGEAI_API_KEY}}' # You can either hard-code or inject secrets from the Coyote vault + api_key: '{{VOYAGEAI_API_KEY}}' # You can either hard-code or inject secrets from the Coyote vault \ No newline at end of file diff --git a/src/client/common.rs b/src/client/common.rs index be2dae0..3eb1892 100644 --- a/src/client/common.rs +++ b/src/client/common.rs @@ -57,12 +57,16 @@ pub trait Client: Sync + Send { let mut builder = ReqwestClient::builder(); let extra = self.extra_config(); let timeout = extra.and_then(|v| v.connect_timeout).unwrap_or(10); + let read_timeout = extra.and_then(|v| v.read_timeout).unwrap_or(300); if let Some(proxy) = extra.and_then(|v| v.proxy.as_deref()) { builder = set_proxy(builder, proxy)?; } if let Some(user_agent) = self.app_config().user_agent.as_ref() { builder = builder.user_agent(user_agent); } + if read_timeout > 0 { + builder = builder.read_timeout(Duration::from_secs(read_timeout)); + } let client = builder .connect_timeout(Duration::from_secs(timeout)) .build() @@ -261,6 +265,7 @@ impl Default for ClientConfig { pub struct ExtraConfig { pub proxy: Option, pub connect_timeout: Option, + pub read_timeout: Option, } #[derive(Debug, Clone, Deserialize, Default)] @@ -1005,4 +1010,4 @@ mod tests { assert!(!should_retry_auth(&api_status_error(401), client)); assert!(!is_rejected(client, "at-1")); } -} \ No newline at end of file +} diff --git a/src/client/stream.rs b/src/client/stream.rs index 3b176f4..79a2d87 100644 --- a/src/client/stream.rs +++ b/src/client/stream.rs @@ -430,10 +430,7 @@ mod tests { assert!(error_message.contains("test_function_loop")); } - fn new_handler() -> ( - SseHandler, - tokio::sync::mpsc::UnboundedReceiver, - ) { + fn new_handler() -> (SseHandler, tokio::sync::mpsc::UnboundedReceiver) { let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); let abort_signal = crate::utils::create_abort_signal(); (SseHandler::new(sender, abort_signal), receiver) @@ -517,4 +514,4 @@ mod tests { {"key": "value3"}"#; assert_json_stream!(input, output); } -} \ No newline at end of file +} diff --git a/src/config/mcp_factory.rs b/src/config/mcp_factory.rs index 0afc715..961847d 100644 --- a/src/config/mcp_factory.rs +++ b/src/config/mcp_factory.rs @@ -103,18 +103,16 @@ impl McpFactory { } let (auth, auth_reason) = resolve_http_auth(name, spec).await; - let handle = spawn_mcp_server(spec, log_path, auth) - .await - .map_err(|e| { - if is_auth_required_error(&e) { - e.context(McpAuthRequired { - server: name.to_string(), - reason: auth_reason, - }) - } else { - e - } - })?; + let handle = spawn_mcp_server(spec, log_path, auth).await.map_err(|e| { + if is_auth_required_error(&e) { + e.context(McpAuthRequired { + server: name.to_string(), + reason: auth_reason, + }) + } else { + e + } + })?; self.insert_active(key, &handle); Ok(handle) } diff --git a/src/function/mod.rs b/src/function/mod.rs index 2545fb0..b7742fa 100644 --- a/src/function/mod.rs +++ b/src/function/mod.rs @@ -29,16 +29,17 @@ use rust_embed::Embed; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use skill::SKILL_FUNCTION_PREFIX; -use std::collections::VecDeque; use std::ffi::OsStr; use std::fs::File; use std::io::{Read, Write}; use std::sync::atomic::Ordering; +use std::{collections::VecDeque, thread}; use std::{ collections::{HashMap, HashSet}, env, fs, io, path::{Path, PathBuf}, process::{Command, Stdio}, + time::{Duration, Instant}, }; use strum_macros::AsRefStr; use supervisor::SUPERVISOR_FUNCTION_PREFIX; @@ -1456,6 +1457,7 @@ pub fn run_llm_function( let mut child = Command::new(&cmd_name) .args(&cmd_args) .envs(envs) + .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn() @@ -1464,7 +1466,7 @@ pub fn run_llm_function( let stdout = child.stdout.take().expect("Failed to capture stdout"); let stderr = child.stderr.take().expect("Failed to capture stderr"); - let stdout_thread = std::thread::spawn(move || { + let stdout_thread = thread::spawn(move || { let mut buffer = [0; 1024]; let mut reader = stdout; let mut out = io::stdout(); @@ -1491,7 +1493,7 @@ pub fn run_llm_function( buf }); - let stderr_thread = std::thread::spawn(move || { + let stderr_thread = thread::spawn(move || { let mut buffer = [0; 1024]; let mut reader = stderr; let mut err = io::stderr(); @@ -1518,9 +1520,39 @@ pub fn run_llm_function( buf }); - let status = child - .wait() - .map_err(|err| anyhow!("Unable to run {command_name}, {err}"))?; + let timeout_secs = env::var("COYOTE_TOOL_TIMEOUT") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(1800); + let deadline = (timeout_secs > 0).then(|| Instant::now() + Duration::from_secs(timeout_secs)); + let status = loop { + match child.try_wait() { + Ok(Some(status)) => break status, + Ok(None) => {} + Err(err) => bail!("Unable to run {command_name}, {err}"), + } + if let Some(deadline) = deadline + && Instant::now() >= deadline + { + let _ = child.kill(); + let _ = child.wait(); + drop(stdout_thread); + drop(stderr_thread); + let tool_error_message = format!( + "Tool call '{command_name}' timed out after {timeout_secs}s and was killed (set COYOTE_TOOL_TIMEOUT to adjust; 0 = unlimited)" + ); + eprintln!( + "{}", + muted_warning_text(&format!("⚠️ {tool_error_message} ⚠️")) + ); + let error_json = json!({"tool_call_error": tool_error_message}); + + debug!("Tool call error: {error_json:?}"); + + return Ok(Some(error_json.to_string())); + } + thread::sleep(Duration::from_millis(100)); + }; let stdout_bytes = stdout_thread.join().unwrap_or_default(); let stderr_bytes = stderr_thread.join().unwrap_or_default(); diff --git a/src/graph/executor.rs b/src/graph/executor.rs index 46398c6..a53441d 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -250,7 +250,30 @@ impl GraphExecutor { branch_tasks.push(task); } - let joined = join_all(branch_tasks).await; + let joined = match graph_timeout { + Some(t) => { + let remaining = t.saturating_sub(start.elapsed()); + let abort_handles: Vec<_> = branch_tasks + .iter() + .map(|task| task.abort_handle()) + .collect(); + match tokio::time::timeout(remaining, join_all(branch_tasks)).await { + Ok(joined) => joined, + Err(_) => { + for handle in abort_handles { + handle.abort(); + } + bail!( + "Graph '{}' timed out after {}s during super-step with frontier {:?}", + graph.name, + t.as_secs(), + sorted_frontier(&frontier) + ); + } + } + } + None => join_all(branch_tasks).await, + }; let mut branch_writes: Vec = Vec::new(); let mut next_frontier: HashSet = HashSet::new(); @@ -793,4 +816,44 @@ nodes: "error should list both End nodes: {err}" ); } + + #[tokio::test] + async fn graph_timeout_interrupts_in_flight_super_step() { + if !cmd_available("bash") { + eprintln!("skipping: bash not available"); + return; + } + let ws = TestWorkspace::new(); + ws.write_script("sleeper.sh", "#!/bin/bash\nsleep 10\necho '{}'\n"); + + let yaml = r#" +name: inflight_timeout_test +start: sleeper +settings: + timeout: 1 +nodes: + sleeper: + type: script + script: sleeper.sh + state_updates: {} + next: done + done: + type: end + output: "done" +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + let mut ctx = make_ctx(); + let abort = create_abort_signal(); + let result = GraphExecutor::new(graph, &ws.dir) + .execute(&mut ctx, abort) + .await; + + assert!(result.is_err(), "expected in-flight timeout to error"); + let err = format!("{:#}", result.unwrap_err()); + assert!( + err.contains("timed out after 1s during super-step"), + "error should report during-super-step timeout: {err}" + ); + assert!(err.contains("sleeper"), "error should name frontier: {err}"); + } } diff --git a/src/graph/script.rs b/src/graph/script.rs index fd30d1e..9d41c4c 100644 --- a/src/graph/script.rs +++ b/src/graph/script.rs @@ -54,6 +54,7 @@ impl ScriptExecutor { let mut cmd = build_command(language, &script_path)?; cmd.stdout(Stdio::piped()); cmd.stderr(Stdio::piped()); + cmd.kill_on_drop(true); cmd.envs(&self.extra_envs); cmd.env("AUTO_CONFIRM", "true"); match &state_repr {