qdaxb · Jun 30, 2026 · Jun 30, 2026 · Jun 30, 2026 · coderabbitai · Jun 30, 2026
diff --git a/docs/en/developer-guide/task-runtime-state-machine.md b/docs/en/developer-guide/task-runtime-state-machine.md
@@ -97,6 +97,21 @@ When a task becomes terminal:
 
 `chat:done` only means the current message stream ended. It does not prove the task lifecycle is terminal. If queued messages are still blocked after `chat:done`, `useChatStreamHandlers` calls the current task machine with `requestRuntimeCheck('queued-message-blocked')` so the frontend converges back to the server state.
 
+### Cancellation And State Convergence
+
+Runtime task cancellation must converge two state surfaces:
+
+- Runtime task state exposed by the executor, such as `running` and `status` from `runtime.tasks.list`. Sidebars, close guards, and task lists consume this state.
+- Message stream state inside the current pane, such as assistant messages that are still `assistant:streaming`. The composer stop button, send lock, and local streaming UI consume this state.
+
+The cancellation flow must follow these rules:
+
+- After receiving `runtime.tasks.cancel`, the executor must interrupt the active turn and wait for the app-server or child process to actually exit. If it does not exit before the timeout, return `cancel_timeout`; the frontend must not treat that as stopped.
+- Runners that can spawn child processes, such as the Codex app-server, must be managed and terminated as a process group so background workers cannot keep running after the parent process exits.
+- After the frontend receives an accepted cancellation result, it must refresh the runtime work list so sidebars and close guards observe `running: false`.
+- The pane session must locally finalize any still-streaming assistant messages as cancelled because cancellation may not emit a normal `chat:done` or response completed event.
+- Updating only runtime work leaves the composer looking busy. Updating only message streams leaves the sidebar looking busy. Both state surfaces must be updated in the same stop path.
+
 ## Change Guidelines
 
 When adding or changing task runtime UI:

diff --git a/docs/zh/developer-guide/task-runtime-state-machine.md b/docs/zh/developer-guide/task-runtime-state-machine.md
@@ -97,6 +97,21 @@ WebSocket 连接恢复也由状态机按这个顺序统一处理：外部触发
 
 `chat:done` 只表示当前消息流结束，不等价于任务生命周期终态。若 `chat:done` 后运行态仍阻塞 queued message，`useChatStreamHandlers` 会触发当前任务状态机的 `requestRuntimeCheck('queued-message-blocked')`，让前端重新对齐服务端状态。
 
+### 取消与状态收敛
+
+运行时任务取消必须同时收敛两类状态：
+
+- executor 暴露的运行时任务状态，例如 `runtime.tasks.list` 返回的 `running` 和 `status`，用于侧边栏运行标记、关闭保护和任务列表。
+- 当前 pane 内的消息流状态，例如仍处于 `assistant:streaming` 的消息，用于输入框停止按钮、发送锁和本地流式 UI。
+
+取消流程必须遵循以下规则：
+
+- executor 收到 `runtime.tasks.cancel` 后必须中断当前 turn，并等待 app-server 或子进程实际退出；如果超过超时时间仍未退出，应返回 `cancel_timeout`，前端不能把它当作已停止。
+- Codex app-server 这类会派生子进程的运行器必须按进程组管理和终止，避免只停止父进程而后台 worker 继续执行。
+- 前端收到已接受的取消结果后，必须刷新 runtime work 列表，让侧边栏和关闭保护看到 `running: false`。
+- pane session 必须把本地仍在 streaming 的 assistant 消息固化为 cancelled，因为取消不一定会产生正常的 `chat:done` 或 response completed 事件。
+- 只更新 runtime work 会导致输入框仍显示运行中；只更新消息流会导致侧边栏仍显示运行中。两类状态必须在同一个停止路径里同时更新。
+
 ## 修改指南
 
 新增或修改任务运行态 UI 时：

diff --git a/executor/src/agents/codex.rs b/executor/src/agents/codex.rs
@@ -19,7 +19,7 @@ use serde_json::{json, Value};
 use tokio::{
     io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
     process::{Child, ChildStdin, ChildStdout, Command},
-    sync::{mpsc, Mutex},
+    sync::{mpsc, oneshot, Mutex},
     time::timeout,
 };
 
@@ -38,6 +38,7 @@ use super::{model_id, prompt_text};
 
 const DEFAULT_CODEX_RPC_TIMEOUT_SECONDS: u64 = 300;
 const DEFAULT_PROVIDER_ID: &str = "wecode-openai";
+pub const CODEX_APP_SERVER_TURN_CANCELLED: &str = "codex app-server turn cancelled";
 const DEFAULT_PROVIDER_NAME: &str = "wecode openai";
 const DEFAULT_REASONING_EFFORT: &str = "medium";
 const DEFAULT_NO_PROXY: &str = "localhost,127.0.0.1,::1,host.docker.internal";
@@ -210,14 +211,13 @@ struct CodexAppServerProcess {
 
 impl CodexAppServerProcess {
     async fn shutdown(mut self) {
-        let _ = self.child.start_kill();
-        let _ = self.child.wait().await;
+        terminate_codex_app_server_child(&mut self.child).await;
     }
 }
 
 impl Drop for CodexAppServerProcess {
     fn drop(&mut self) {
-        let _ = self.child.start_kill();
+        signal_codex_app_server_child(&mut self.child);
     }
 }
 
@@ -254,8 +254,7 @@ async fn start_persistent_codex_app_server(binary: &str) -> Result<CodexAppServe
     match result {
         Ok(rpc) => Ok(CodexAppServerProcess { child, rpc }),
         Err(error) => {
-            let _ = child.start_kill();
-            let _ = child.wait().await;
+            terminate_codex_app_server_child(&mut child).await;
             Err(error)
         }
     }
@@ -267,6 +266,25 @@ pub async fn run_codex_app_server_turn(
     resume_thread_id: Option<String>,
     initial_thread_name: Option<String>,
     notifications: Option<CodexNotificationSender>,
+) -> Result<CodexAppServerTurn, String> {
+    run_codex_app_server_turn_with_cancel(
+        binary,
+        request,
+        resume_thread_id,
+        initial_thread_name,
+        notifications,
+        None,
+    )
+    .await
+}
+
+pub async fn run_codex_app_server_turn_with_cancel(
+    binary: &str,
+    request: ExecutionRequest,
+    resume_thread_id: Option<String>,
+    initial_thread_name: Option<String>,
+    notifications: Option<CodexNotificationSender>,
+    mut cancellation: Option<oneshot::Receiver<()>>,
 ) -> Result<CodexAppServerTurn, String> {
     let prepared = prepare_codex_execution_request(request);
     let launch_config = build_codex_launch_config(&prepared.request);
@@ -390,9 +408,15 @@ pub async fn run_codex_app_server_turn(
             ),
         )
         .await?;
-        let outcome = rpc
-            .read_turn(turn_request_id, &mut state, notifications)
-            .await?;
+        let outcome = if let Some(cancellation) = cancellation.as_mut() {
+            tokio::select! {
+                outcome = rpc.read_turn(turn_request_id, &mut state, notifications) => outcome?,
+                _ = cancellation => return Err(CODEX_APP_SERVER_TURN_CANCELLED.to_owned()),
+            }
+        } else {
+            rpc.read_turn(turn_request_id, &mut state, notifications)
+                .await?
+        };
         turn_fields.push(("outcome", codex_outcome_name(&outcome).to_owned()));
         if let ExecutionOutcome::Failed { message } = &outcome {
             turn_fields.push(("error", message.clone()));
@@ -403,8 +427,7 @@ pub async fn run_codex_app_server_turn(
     }
     .await;
 
-    let _ = child.start_kill();
-    let _ = child.wait().await;
+    terminate_codex_app_server_child(&mut child).await;
     if let Err(error) = &result {
         let mut failed_fields = fields.clone();
         failed_fields.push(("error", error.clone()));
@@ -445,14 +468,54 @@ fn spawn_codex_app_server(
             env::var("PATH").ok().as_deref().unwrap_or_default(),
         ),
     );
+    configure_codex_app_server_process_group(&mut command);
     command
+        .kill_on_drop(true)
         .stdin(Stdio::piped())
         .stdout(Stdio::piped())
         .stderr(Stdio::inherit())
         .spawn()
         .map_err(|error| format!("failed to start codex app-server: {error}"))
 }
 
+#[cfg(unix)]
+fn configure_codex_app_server_process_group(command: &mut Command) {
+    unsafe {
+        command.pre_exec(|| {
+            if libc::setpgid(0, 0) == 0 {
+                Ok(())
+            } else {
+                Err(std::io::Error::last_os_error())
+            }
+        });
+    }
+}
+
+#[cfg(not(unix))]
+fn configure_codex_app_server_process_group(_command: &mut Command) {}
+
+async fn terminate_codex_app_server_child(child: &mut Child) {
+    signal_codex_app_server_child(child);
+    let _ = child.wait().await;
+}
+
+fn signal_codex_app_server_child(child: &mut Child) {
+    if matches!(child.try_wait(), Ok(Some(_))) {
+        return;
+    }
+
+    #[cfg(unix)]
+    if let Some(process_group_id) = child.id() {
+        unsafe {
+            let _ = libc::kill(-(process_group_id as libc::pid_t), libc::SIGTERM);
+            let _ = libc::kill(-(process_group_id as libc::pid_t), libc::SIGKILL);
+        }
+        return;
+    }
+
+    let _ = child.start_kill();
+}
+
 async fn with_rpc_timeout<T>(
     operation: &str,
     timeout_seconds: u64,

diff --git a/executor/src/agents/mod.rs b/executor/src/agents/mod.rs
@@ -33,8 +33,9 @@ use claude_code::{
 };
 pub use claude_options::{extract_claude_options, ClaudeOptions};
 pub use codex::{
-    run_codex_app_server_turn, CodexAppServerClient, CodexAppServerEngine, CodexAppServerTurn,
-    CodexCancellationState, CodexNotificationSender, CodexTurnInterrupter,
+    run_codex_app_server_turn, run_codex_app_server_turn_with_cancel, CodexAppServerClient,
+    CodexAppServerEngine, CodexAppServerTurn, CodexCancellationState, CodexNotificationSender,
+    CodexTurnInterrupter, CODEX_APP_SERVER_TURN_CANCELLED,
 };
 pub use dify::{build_dify_config, saved_dify_task_id, DifyEngine};
 pub use image_validator::ImageValidatorEngine;

diff --git a/executor/src/runtime_work/handler.rs b/executor/src/runtime_work/handler.rs
@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::{
-    collections::HashSet,
+    collections::{HashMap, HashSet},
     future::Future,
     path::Path,
     pin::Pin,
@@ -12,10 +12,13 @@ use std::{
 };
 
 use serde_json::{json, Map, Value};
-use tokio::sync::{broadcast, mpsc};
+use tokio::sync::{broadcast, mpsc, oneshot};
 
 use crate::{
-    agents::{run_codex_app_server_turn, CodexAppServerClient, CodexNotificationSender},
+    agents::{
+        run_codex_app_server_turn_with_cancel, CodexAppServerClient, CodexNotificationSender,
+        CODEX_APP_SERVER_TURN_CANCELLED,
+    },
     local::app_ipc::{AppIpcError, RuntimeWorkHandler},
     logging::log_executor_event,
     protocol::ExecutionRequest,
@@ -65,11 +68,17 @@ pub struct RuntimeWorkRpcHandler {
     codex_app_server: CodexAppServerClient,
     event_tx: Option<broadcast::Sender<Value>>,
     active_local_tasks: Arc<Mutex<HashSet<String>>>,
+    active_turn_cancellations: Arc<Mutex<HashMap<String, ActiveTurnCancellation>>>,
     store: RuntimeWorkStore,
     transcript_cache: TranscriptCache,
     thread_list_cache: CodexThreadListCache,
 }
 
+struct ActiveTurnCancellation {
+    cancel: oneshot::Sender<()>,
+    stopped: oneshot::Receiver<()>,
+}
+
 impl RuntimeWorkRpcHandler {
     pub fn new(device_id: impl Into<String>, codex_binary: impl Into<String>) -> Self {
         let codex_binary = codex_binary.into();
@@ -79,6 +88,7 @@ impl RuntimeWorkRpcHandler {
             codex_app_server: CodexAppServerClient::new(codex_binary),
             event_tx: None,
             active_local_tasks: Arc::new(Mutex::new(HashSet::new())),
+            active_turn_cancellations: Arc::new(Mutex::new(HashMap::new())),
             store: RuntimeWorkStore::from_env(),
             transcript_cache: TranscriptCache::default(),
             thread_list_cache: CodexThreadListCache::default(),
@@ -684,6 +694,16 @@ impl RuntimeWorkRpcHandler {
                 link.updated_at = now_ms();
             })
             .or_else(|| self.local_task_link(&local_task_id));
+        if !self.abort_active_turn(&local_task_id).await {
+            return Ok(json!({
+                "success": false,
+                "accepted": false,
+                "localTaskId": local_task_id,
+                "runtime": "codex",
+                "error": "runtime task did not stop within timeout",
+                "code": "cancel_timeout",
+            }));
+        }
 
         Ok(match link {
             Some(link) => task_action_success(&link),
@@ -896,39 +916,59 @@ impl RuntimeWorkRpcHandler {
         log_executor_event("runtime work turn spawning", &fields);
 
         self.mark_active_local_task(&local_task_id);
+        let (cancel_tx, cancel_rx) = oneshot::channel();
+        let (stopped_tx, stopped_rx) = oneshot::channel();
+        self.set_active_turn_cancellation(
+            local_task_id.clone(),
+            ActiveTurnCancellation {
+                cancel: cancel_tx,
+                stopped: stopped_rx,
+            },
+        );
         let handler = self.clone();
-        tokio::spawn(async move {
+        let turn_local_task_id = local_task_id.clone();
+        let turn_handle = tokio::spawn(async move {
             emit_response_event(
                 &handler.event_tx,
                 &handler.device_id,
                 "response.created",
-                &local_task_id,
+                &turn_local_task_id,
                 &request,
                 json!({"response": {"status": "in_progress"}}),
             );
 
             let (notifications, mapper_task) =
-                handler.spawn_notification_mapper(local_task_id.clone(), request.clone());
-            let result = run_codex_app_server_turn(
+                handler.spawn_notification_mapper(turn_local_task_id.clone(), request.clone());
+            let result = run_codex_app_server_turn_with_cancel(
                 &handler.codex_binary,
                 request.clone(),
                 resume_thread_id,
                 initial_thread_name,
                 notifications,
+                Some(cancel_rx),
             )
             .await;
             if let Err(error) = mapper_task.await {
                 log_executor_event(
                     "runtime work notification mapper failed",
                     &[
-                        ("local_task_id", local_task_id.clone()),
+                        ("local_task_id", turn_local_task_id.clone()),
                         ("error", error.to_string()),
                     ],
                 );
             }
 
-            handler.handle_turn_result(&local_task_id, &request, result);
+            if matches!(result.as_ref(), Err(error) if error == CODEX_APP_SERVER_TURN_CANCELLED) {
+                handler.clear_active_turn_cancellation(&turn_local_task_id);
+                handler.unmark_active_local_task(&turn_local_task_id);
+                let _ = stopped_tx.send(());
+                return;
+            }
+
+            handler.handle_turn_result(&turn_local_task_id, &request, result);
+            let _ = stopped_tx.send(());
         });
+        drop(turn_handle);
     }
 
     fn handle_turn_result(
@@ -1391,6 +1431,43 @@ impl RuntimeWorkRpcHandler {
             .remove(local_task_id);
     }
 
+    fn set_active_turn_cancellation(&self, local_task_id: String, control: ActiveTurnCancellation) {
+        if let Some(previous) = self
+            .active_turn_cancellations
+            .lock()
+            .expect("active turn cancellation map lock should not be poisoned")
+            .insert(local_task_id, control)
+        {
+            let _ = previous.cancel.send(());
+        }
+    }
+
+    fn clear_active_turn_cancellation(&self, local_task_id: &str) {
+        self.active_turn_cancellations
+            .lock()
+            .expect("active turn cancellation map lock should not be poisoned")
+            .remove(local_task_id);
+    }
+
+    async fn abort_active_turn(&self, local_task_id: &str) -> bool {
+        let control = {
+            self.active_turn_cancellations
+                .lock()
+                .expect("active turn cancellation map lock should not be poisoned")
+                .remove(local_task_id)
+        };
+        if let Some(control) = control {
+            let _ = control.cancel.send(());
+            let stopped =
+                tokio::time::timeout(std::time::Duration::from_secs(10), control.stopped).await;
+            if stopped.is_err() {
+                return false;
+            }
+        }
+        self.unmark_active_local_task(local_task_id);
+        true
+    }
+
     fn is_active_local_task(&self, local_task_id: &str) -> bool {
         self.active_local_tasks
             .lock()
@@ -1399,6 +1476,7 @@ impl RuntimeWorkRpcHandler {
     }
 
     fn finish_local_task(&self, local_task_id: &str, thread_id: Option<String>, status: &str) {
+        self.clear_active_turn_cancellation(local_task_id);
         let invalidate_thread_id = thread_id.clone();
         self.store.update_task(local_task_id, |link| {
             if thread_id.is_some() {