Codex/increase claude timeout

docs/en/user-guide/ai-devices/local-device-support.md

@@ -212,6 +212,15 @@ wegent-executor
 
 The installer and first startup create `~/.wegent-executor/device-config.json`. Configuration priority is environment variables, device config, then defaults. If `WEGENT_EXECUTOR_HOME` is not set, the executor uses `~/.wegent-executor`. `EXECUTOR_MODE=remote` starts the local socket and, after `WEGENT_BACKEND_URL` or `connection.backend_url` is set, connects to Backend as a remote device. `EXECUTOR_STARTUP_MODE=socket` remains compatible for old scripts, but new startup commands no longer need it. Wework App manages executors it starts itself; if you start an executor manually outside the App, the App attaches to the existing socket but does not terminate that external process on exit. Do not run multiple manual executors with the same executor home or socket path. Logs are written to `~/.wegent-executor/logs/executor.log`.
 
+#### Claude Code Execution Timeout
+
+When the local executor starts a Claude Code child process, it waits up to 1 hour by default. Long-running code generation, dependency installation, or file processing tasks can continue within that window. To tune the limit for a specific environment, set `WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS` before starting the executor. This setting only affects Claude Code child processes, not the native Codex app-server path; Codex RPC timeouts are controlled by `WEGENT_CODEX_RPC_TIMEOUT_SECONDS`.
+
+```bash
+export WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS=7200
+wegent-executor
+```
+
 ### Getting JWT Token
 
 1. Log in to Wegent Web UI

docs/zh/user-guide/ai-devices/local-device-support.md

@@ -214,6 +214,15 @@ wegent-executor
 
 安装脚本和首次启动会创建 `~/.wegent-executor/device-config.json`。配置优先级是环境变量、device config、默认值；未设置 `WEGENT_EXECUTOR_HOME` 时默认使用 `~/.wegent-executor`。`EXECUTOR_MODE=remote` 会启动本机 socket，并在设置 `WEGENT_BACKEND_URL` 或配置文件中的 `connection.backend_url` 后以远程设备模式连接 Backend。`EXECUTOR_STARTUP_MODE=socket` 仍兼容旧脚本，但新启动命令不再需要设置它。Wework App 会管理自己启动的 executor；如果你手动在 App 外启动 executor，App 会连接已有 socket，但退出 App 时不会终止这个外部进程。不要让多个手动 executor 复用同一个 executor home 或 socket 路径。日志写入 `~/.wegent-executor/logs/executor.log`。
 
+#### Claude Code 执行超时
+
+本地 executor 启动 Claude Code 子进程时，默认最长等待 1 小时。长时间代码生成、依赖安装或文件处理任务可以在这个时间内继续运行。若需要为特定环境调整该限制，可在启动 executor 前设置 `WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS`。该配置只影响 Claude Code 子进程，不影响 native Codex app-server；Codex RPC 超时由 `WEGENT_CODEX_RPC_TIMEOUT_SECONDS` 控制。
+
+```bash
+export WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS=7200
+wegent-executor
+```
+
 ### 获取 JWT Token
 
 1. 登录 Wegent Web 界面

executor/src/agents/mod.rs

@@ -38,6 +38,8 @@ pub use codex::{
 pub use dify::{build_dify_config, saved_dify_task_id, DifyEngine};
 pub use image_validator::ImageValidatorEngine;
 
+const DEFAULT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS: u64 = 3600;
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct AgentCommandPlanner {
     claude_binary: String,
@@ -86,6 +88,23 @@ fn read_binary(primary: &str, secondary: &str, default: &str) -> String {
         .unwrap_or_else(|| default.to_owned())
 }
 
+fn claude_code_process_timeout_seconds() -> u64 {
+    env::var("WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS")
+        .ok()
+        .and_then(|value| value.trim().parse::<u64>().ok())
+        .filter(|value| *value > 0)
+        .unwrap_or(DEFAULT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS)
+}
+
+fn stream_process_engine_for(agent_kind: &AgentKind, spec: CommandSpec) -> StreamProcessEngine {
+    match agent_kind {
+        AgentKind::ClaudeCode => {
+            StreamProcessEngine::new(spec, claude_code_process_timeout_seconds())
+        }
+        agent_kind => unreachable!("unsupported process agent kind: {agent_kind:?}"),
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct AgentProcessEngine {
     planner: AgentCommandPlanner,
@@ -164,7 +183,9 @@ impl AgentEngine for AgentProcessEngine {
                                 git_auth::setup_git_authentication(&request).await;
                                 run_pre_execute_hook(&request, &spec).await;
                             }
-                            StreamProcessEngine::new(spec).run(request).await
+                            stream_process_engine_for(&agent_kind, spec)
+                                .run(request)
+                                .await
                         }
                         Err(message) => {
                             let mut failed_fields = fields;
@@ -250,7 +271,7 @@ impl AgentEngine for AgentProcessEngine {
                                 git_auth::setup_git_authentication(&request).await;
                                 run_pre_execute_hook(&request, &spec).await;
                             }
-                            StreamProcessEngine::new(spec)
+                            stream_process_engine_for(&agent_kind, spec)
                                 .run_with_events(request, sink, builder)
                                 .await
                         }
@@ -266,3 +287,64 @@ impl AgentEngine for AgentProcessEngine {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Mutex, MutexGuard, OnceLock};
+
+    use super::*;
+
+    fn env_lock() -> MutexGuard<'static, ()> {
+        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+        LOCK.get_or_init(|| Mutex::new(()))
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    struct EnvGuard {
+        key: &'static str,
+        previous: Option<String>,
+    }
+
+    impl EnvGuard {
+        fn set(key: &'static str, value: &str) -> Self {
+            let previous = std::env::var(key).ok();
+            std::env::set_var(key, value);
+            Self { key, previous }
+        }
+
+        fn remove(key: &'static str) -> Self {
+            let previous = std::env::var(key).ok();
+            std::env::remove_var(key);
+            Self { key, previous }
+        }
+    }
+
+    impl Drop for EnvGuard {
+        fn drop(&mut self) {
+            if let Some(previous) = &self.previous {
+                std::env::set_var(self.key, previous);
+            } else {
+                std::env::remove_var(self.key);
+            }
+        }
+    }
+
+    #[test]
+    fn claude_code_process_timeout_uses_claude_specific_env_var() {
+        let _lock = env_lock();
+        let _old_timeout = EnvGuard::remove("WEGENT_EXECUTOR_PROCESS_TIMEOUT_SECONDS");
+        let _timeout = EnvGuard::set("WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS", "42");
+
+        assert_eq!(claude_code_process_timeout_seconds(), 42);
+    }
+
+    #[test]
+    fn claude_code_process_timeout_ignores_generic_process_env_var() {
+        let _lock = env_lock();
+        let _old_timeout = EnvGuard::set("WEGENT_EXECUTOR_PROCESS_TIMEOUT_SECONDS", "1");
+        let _timeout = EnvGuard::remove("WEGENT_CLAUDE_CODE_PROCESS_TIMEOUT_SECONDS");
+
+        assert_eq!(claude_code_process_timeout_seconds(), 3600);
+    }
+}

executor/src/process/mod.rs

@@ -36,7 +36,6 @@ use crate::{
     },
 };
 
-const DEFAULT_PROCESS_TIMEOUT_SECONDS: u64 = 300;
 const DEFAULT_STREAM_CHUNK_CHARS: usize = 20;
 const DEFAULT_STREAM_CHUNK_DELAY_MS: u64 = 0;
 const MAX_DEFERRED_MCP_RETRIES: usize = 2;
@@ -106,11 +105,16 @@ impl CommandSpec {
 #[derive(Debug, Clone)]
 pub struct ProcessEngine {
     spec: CommandSpec,
+    timeout_seconds: u64,
 }
 
 impl ProcessEngine {
-    pub fn new(spec: CommandSpec) -> Self {
-        Self { spec }
+    pub fn new(spec: CommandSpec, timeout_seconds: u64) -> Self {
+        assert!(timeout_seconds > 0, "timeout_seconds must be positive");
+        Self {
+            spec,
+            timeout_seconds,
+        }
     }
 }
 
@@ -119,18 +123,24 @@ impl AgentEngine for ProcessEngine {
 
     fn run(&self, _request: ExecutionRequest) -> Self::RunFuture {
         let spec = self.spec.clone();
-        Box::pin(async move { run_command(spec).await })
+        let timeout_seconds = self.timeout_seconds;
+        Box::pin(async move { run_command(spec, timeout_seconds).await })
     }
 }
 
 #[derive(Debug, Clone)]
 pub struct StreamProcessEngine {
     spec: CommandSpec,
+    timeout_seconds: u64,
 }
 
 impl StreamProcessEngine {
-    pub fn new(spec: CommandSpec) -> Self {
-        Self { spec }
+    pub fn new(spec: CommandSpec, timeout_seconds: u64) -> Self {
+        assert!(timeout_seconds > 0, "timeout_seconds must be positive");
+        Self {
+            spec,
+            timeout_seconds,
+        }
     }
 }
 
@@ -139,17 +149,23 @@ impl AgentEngine for StreamProcessEngine {
 
     fn run(&self, request: ExecutionRequest) -> Self::RunFuture {
         let spec = self.spec.clone();
+        let timeout_seconds = self.timeout_seconds;
         Box::pin(async move {
-            match run_command_output(spec.clone()).await {
+            match run_command_output(spec.clone(), timeout_seconds).await {
                 CommandOutcome::Success { stdout } => {
                     let summary = collect_claude_stream_summary(&stdout);
                     if let Some(session_id) = &summary.session_id {
                         claude_session::save_session_id(&request, session_id);
                     }
-                    let summary =
-                        handle_retryable_api_errors(spec.clone(), &request, summary).await;
+                    let summary = handle_retryable_api_errors(
+                        spec.clone(),
+                        &request,
+                        summary,
+                        timeout_seconds,
+                    )
+                    .await;
                     if summary.deferred_tool_use.is_some() {
-                        handle_deferred_mcp_loop(spec, request, summary).await
+                        handle_deferred_mcp_loop(spec, request, summary, timeout_seconds).await
                     } else {
                         summary.outcome
                     }
@@ -171,9 +187,11 @@ impl AgentEngine for StreamProcessEngine {
         S: EventSink,
     {
         let spec = self.spec.clone();
+        let timeout_seconds = self.timeout_seconds;
         Box::pin(async move {
             match run_streaming_command_output(
                 spec.clone(),
+                timeout_seconds,
                 sink,
                 builder,
                 request.task_id,
@@ -186,10 +204,15 @@ impl AgentEngine for StreamProcessEngine {
                     if let Some(session_id) = &summary.session_id {
                         claude_session::save_session_id(&request, session_id);
                     }
-                    let summary =
-                        handle_retryable_api_errors(spec.clone(), &request, summary).await;
+                    let summary = handle_retryable_api_errors(
+                        spec.clone(),
+                        &request,
+                        summary,
+                        timeout_seconds,
+                    )
+                    .await;
                     if summary.deferred_tool_use.is_some() {
-                        handle_deferred_mcp_loop(spec, request, summary).await
+                        handle_deferred_mcp_loop(spec, request, summary, timeout_seconds).await
                     } else {
                         summary.outcome
                     }
@@ -206,6 +229,7 @@ async fn handle_retryable_api_errors(
     base_spec: CommandSpec,
     request: &ExecutionRequest,
     mut summary: crate::stream::ClaudeStreamSummary,
+    timeout_seconds: u64,
 ) -> crate::stream::ClaudeStreamSummary {
     let fields = task_fields(request.task_id, request.subtask_id);
     let mut retry_count = 0;
@@ -222,7 +246,7 @@ async fn handle_retryable_api_errors(
             &session_id,
             ClaudeFollowUpQuery::Prompt("Retry to proceed".to_owned()),
         );
-        match run_command_output(retry_spec).await {
+        match run_command_output(retry_spec, timeout_seconds).await {
             CommandOutcome::Success { stdout } => {
                 summary = collect_claude_stream_summary(&stdout);
                 if let Some(session_id) = &summary.session_id {
@@ -250,6 +274,7 @@ async fn handle_deferred_mcp_loop(
     base_spec: CommandSpec,
     request: ExecutionRequest,
     mut summary: crate::stream::ClaudeStreamSummary,
+    timeout_seconds: u64,
 ) -> ExecutionOutcome {
     let mcp_servers = mcp_servers_from_spec(&base_spec).unwrap_or(Value::Null);
     let mut retry_count = 0;
@@ -266,7 +291,7 @@ async fn handle_deferred_mcp_loop(
         {
             stale_answer_defer_drained = true;
             log_executor_event("draining stale answered interactive form defer", &fields);
-            match run_command_output(base_spec.clone()).await {
+            match run_command_output(base_spec.clone(), timeout_seconds).await {
                 CommandOutcome::Success { stdout } => {
                     summary = collect_claude_stream_summary(&stdout);
                     if let Some(session_id) = &summary.session_id {
@@ -340,7 +365,7 @@ async fn handle_deferred_mcp_loop(
                     &session_id,
                     ClaudeFollowUpQuery::ToolResult(retry_query),
                 );
-                match run_command_output(retry_spec).await {
+                match run_command_output(retry_spec, timeout_seconds).await {
                     CommandOutcome::Success { stdout } => {
                         summary = collect_claude_stream_summary(&stdout);
                         if let Some(session_id) = &summary.session_id {
@@ -424,8 +449,8 @@ fn read_json_file(path: &str) -> Option<Value> {
         .and_then(|content| serde_json::from_str::<Value>(&content).ok())
 }
 
-async fn run_command(spec: CommandSpec) -> ExecutionOutcome {
-    match run_command_output(spec).await {
+async fn run_command(spec: CommandSpec, timeout_seconds: u64) -> ExecutionOutcome {
+    match run_command_output(spec, timeout_seconds).await {
         CommandOutcome::Success { stdout } => ExecutionOutcome::Completed { content: stdout },
         CommandOutcome::Failure { stderr, stdout, .. } => ExecutionOutcome::Failed {
             message: failure_message(stderr.into_bytes(), stdout.into_bytes()),
@@ -444,7 +469,7 @@ enum CommandOutcome {
     },
 }
 
-async fn run_command_output(spec: CommandSpec) -> CommandOutcome {
+async fn run_command_output(spec: CommandSpec, timeout_seconds: u64) -> CommandOutcome {
     let mut command = Command::new(&spec.program);
     command.args(&spec.args).envs(&spec.env);
     command.kill_on_drop(true);
@@ -459,7 +484,6 @@ async fn run_command_output(spec: CommandSpec) -> CommandOutcome {
         command.current_dir(cwd);
     }
 
-    let timeout_seconds = process_timeout_seconds();
     let mut fields = command_log_fields(&spec);
     fields.push(("timeout_seconds", timeout_seconds.to_string()));
     log_executor_event("process started", &fields);
@@ -485,6 +509,7 @@ async fn run_command_output(spec: CommandSpec) -> CommandOutcome {
 
 async fn run_streaming_command_output<S>(
     spec: CommandSpec,
+    timeout_seconds: u64,
     sink: S,
     builder: ResponsesEventBuilder,
     task_id: i64,
@@ -512,7 +537,6 @@ where
         command.current_dir(cwd);
     }
 
-    let timeout_seconds = process_timeout_seconds();
     let mut fields = command_log_fields(&spec);
     fields.push(("timeout_seconds", timeout_seconds.to_string()));
     log_executor_event("process started", &fields);
@@ -923,14 +947,6 @@ fn command_outcome_fields(outcome: &CommandOutcome) -> Vec<(&'static str, String
     }
 }
 
-fn process_timeout_seconds() -> u64 {
-    env::var("WEGENT_EXECUTOR_PROCESS_TIMEOUT_SECONDS")
-        .ok()
-        .and_then(|value| value.trim().parse::<u64>().ok())
-        .filter(|value| *value > 0)
-        .unwrap_or(DEFAULT_PROCESS_TIMEOUT_SECONDS)
-}
-
 fn failure_message(stderr: Vec<u8>, stdout: Vec<u8>) -> String {
     let stderr = decode_output(stderr);
     if !stderr.is_empty() {