Add app-server background terminal process APIs (#26041)

## Summary

Codex Apps needs app-server as the source of truth for chat-started
background terminals instead of guessing from local process trees.

This PR adds experimental v2 APIs to list and terminate background
terminals for a loaded thread using app-server process ids, so clients
can manage background terminals without local PID discovery.

## Changes

- `thread/backgroundTerminals/list` returns paginated background
terminal records with `itemId`, app-server `processId`, `command`,
`cwd`, nullable `osPid`, nullable `cpuPercent`, and nullable `rssKb`.
- `thread/backgroundTerminals/terminate` terminates one running
background terminal by app-server `processId` and returns whether a
process was terminated.
- Background terminal list and terminate operations use unified-exec
process manager state as their source of truth.

codex-rs/app-server-protocol/src/export.rs

@@ -39,8 +39,11 @@ use ts_rs::TS;
 pub(crate) const GENERATED_TS_HEADER: &str = "// GENERATED CODE! DO NOT MODIFY BY HAND!\n\n";
 const IGNORED_DEFINITIONS: &[&str] = &["Option<()>"];
 const JSON_V1_ALLOWLIST: &[&str] = &["InitializeParams", "InitializeResponse"];
-const EXPERIMENTAL_CLIENT_METHOD_DEPENDENCY_TYPES: &[&str] =
-    &["RemoteControlClient", "RemoteControlClientsListOrder"];
+const EXPERIMENTAL_CLIENT_METHOD_DEPENDENCY_TYPES: &[&str] = &[
+    "RemoteControlClient",
+    "RemoteControlClientsListOrder",
+    "ThreadBackgroundTerminal",
+];
 const SPECIAL_DEFINITIONS: &[&str] = &[
     "ClientNotification",
     "ClientRequest",

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -574,6 +574,18 @@ client_request_definitions! {
         serialization: thread_id(params.thread_id),
         response: v2::ThreadBackgroundTerminalsCleanResponse,
     },
+    #[experimental("thread/backgroundTerminals/list")]
+    ThreadBackgroundTerminalsList => "thread/backgroundTerminals/list" {
+        params: v2::ThreadBackgroundTerminalsListParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadBackgroundTerminalsListResponse,
+    },
+    #[experimental("thread/backgroundTerminals/terminate")]
+    ThreadBackgroundTerminalsTerminate => "thread/backgroundTerminals/terminate" {
+        params: v2::ThreadBackgroundTerminalsTerminateParams,
+        serialization: thread_id(params.thread_id),
+        response: v2::ThreadBackgroundTerminalsTerminateResponse,
+    },
     ThreadRollback => "thread/rollback" {
         params: v2::ThreadRollbackParams,
         serialization: thread_id(params.thread_id),
@@ -2936,6 +2948,54 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn serialize_thread_background_terminals_list() -> Result<()> {
+        let request = ClientRequest::ThreadBackgroundTerminalsList {
+            request_id: RequestId::Integer(8),
+            params: v2::ThreadBackgroundTerminalsListParams {
+                thread_id: "thr_123".to_string(),
+                cursor: None,
+                limit: None,
+            },
+        };
+        assert_eq!(
+            json!({
+                "method": "thread/backgroundTerminals/list",
+                "id": 8,
+                "params": {
+                    "threadId": "thr_123",
+                    "cursor": null,
+                    "limit": null
+                }
+            }),
+            serde_json::to_value(&request)?,
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn serialize_thread_background_terminals_terminate() -> Result<()> {
+        let request = ClientRequest::ThreadBackgroundTerminalsTerminate {
+            request_id: RequestId::Integer(8),
+            params: v2::ThreadBackgroundTerminalsTerminateParams {
+                thread_id: "thr_123".to_string(),
+                process_id: "42".to_string(),
+            },
+        };
+        assert_eq!(
+            json!({
+                "method": "thread/backgroundTerminals/terminate",
+                "id": 8,
+                "params": {
+                    "threadId": "thr_123",
+                    "processId": "42"
+                }
+            }),
+            serde_json::to_value(&request)?,
+        );
+        Ok(())
+    }
+
     #[test]
     fn serialize_thread_realtime_start() -> Result<()> {
         let request = ClientRequest::ThreadRealtimeStart {

codex-rs/app-server-protocol/src/protocol/v2/thread.rs

@@ -935,6 +935,57 @@ pub struct ThreadBackgroundTerminalsCleanParams {
 #[ts(export_to = "v2/")]
 pub struct ThreadBackgroundTerminalsCleanResponse {}
 
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadBackgroundTerminalsListParams {
+    pub thread_id: String,
+    /// Opaque pagination cursor returned by a previous call.
+    #[ts(optional = nullable)]
+    pub cursor: Option<String>,
+    /// Optional page size.
+    #[ts(optional = nullable)]
+    pub limit: Option<u32>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadBackgroundTerminal {
+    pub item_id: String,
+    pub process_id: String,
+    pub command: String,
+    pub cwd: AbsolutePathBuf,
+    pub os_pid: Option<u32>,
+    pub cpu_percent: Option<f64>,
+    pub rss_kb: Option<u64>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadBackgroundTerminalsListResponse {
+    pub data: Vec<ThreadBackgroundTerminal>,
+    /// Opaque cursor to pass to the next call to continue after the last item.
+    /// If None, there are no more items to return.
+    pub next_cursor: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadBackgroundTerminalsTerminateParams {
+    pub thread_id: String,
+    pub process_id: String,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct ThreadBackgroundTerminalsTerminateResponse {
+    pub terminated: bool,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]

codex-rs/app-server/README.md

@@ -157,6 +157,8 @@ Example with notification opt-out:
 - `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
 - `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream.
 - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
+- `thread/backgroundTerminals/list` — list running background terminals for a loaded thread (experimental; requires `capabilities.experimentalApi`); returns `data` with the running terminal ids.
+- `thread/backgroundTerminals/terminate` — terminate one running background terminal by app-server `processId` (experimental; requires `capabilities.experimentalApi`); returns whether a process was terminated.
 - `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
 - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
 - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success.
@@ -872,6 +874,32 @@ Use `thread/backgroundTerminals/clean` to terminate all running background termi
 { "id": 35, "result": {} }
 ```
 
+### Example: List and terminate background terminals
+
+Use `thread/backgroundTerminals/list` to inspect running background terminals associated with a loaded thread. The `backgroundTerminals` segment intentionally follows the existing `thread/backgroundTerminals/clean` method. The returned `processId` is the app-server process id; host OS metadata is nullable. The request accepts the standard `cursor` and `limit` pagination fields. When `nextCursor` is non-null, pass it as `cursor` to fetch the next page.
+
+```json
+{ "method": "thread/backgroundTerminals/list", "id": 36, "params": { "threadId": "thr_123" } }
+{ "id": 36, "result": { "data": [
+    {
+        "itemId": "item_456",
+        "processId": "42",
+        "command": "python3 -m http.server",
+        "cwd": "/workspace",
+        "osPid": null,
+        "cpuPercent": null,
+        "rssKb": null
+    }
+], "nextCursor": null } }
+```
+
+Use `thread/backgroundTerminals/terminate` to terminate one running background terminal by that `processId`.
+
+```json
+{ "method": "thread/backgroundTerminals/terminate", "id": 37, "params": { "threadId": "thr_123", "processId": "42" } }
+{ "id": 37, "result": { "terminated": true } }
+```
+
 ### Example: Steer an active turn
 
 Use `turn/steer` to append additional user input to the currently active regular turn. This does

codex-rs/app-server/src/message_processor.rs

@@ -1126,6 +1126,16 @@ impl MessageProcessor {
                     .thread_background_terminals_clean(&request_id, params)
                     .await
             }
+            ClientRequest::ThreadBackgroundTerminalsList { params, .. } => {
+                self.thread_processor
+                    .thread_background_terminals_list(params)
+                    .await
+            }
+            ClientRequest::ThreadBackgroundTerminalsTerminate { params, .. } => {
+                self.thread_processor
+                    .thread_background_terminals_terminate(params)
+                    .await
+            }
             ClientRequest::ThreadRollback { params, .. } => {
                 self.thread_processor
                     .thread_rollback(&request_id, params)

codex-rs/app-server/src/request_processors.rs

@@ -171,8 +171,13 @@ use codex_app_server_protocol::ThreadApproveGuardianDeniedActionResponse;
 use codex_app_server_protocol::ThreadArchiveParams;
 use codex_app_server_protocol::ThreadArchiveResponse;
 use codex_app_server_protocol::ThreadArchivedNotification;
+use codex_app_server_protocol::ThreadBackgroundTerminal;
 use codex_app_server_protocol::ThreadBackgroundTerminalsCleanParams;
 use codex_app_server_protocol::ThreadBackgroundTerminalsCleanResponse;
+use codex_app_server_protocol::ThreadBackgroundTerminalsListParams;
+use codex_app_server_protocol::ThreadBackgroundTerminalsListResponse;
+use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateParams;
+use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateResponse;
 use codex_app_server_protocol::ThreadClosedNotification;
 use codex_app_server_protocol::ThreadCompactStartParams;
 use codex_app_server_protocol::ThreadCompactStartResponse;

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -569,6 +569,24 @@ impl ThreadRequestProcessor {
             .map(|response| Some(response.into()))
     }
 
+    pub(crate) async fn thread_background_terminals_list(
+        &self,
+        params: ThreadBackgroundTerminalsListParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.thread_background_terminals_list_inner(params)
+            .await
+            .map(|response| Some(response.into()))
+    }
+
+    pub(crate) async fn thread_background_terminals_terminate(
+        &self,
+        params: ThreadBackgroundTerminalsTerminateParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.thread_background_terminals_terminate_inner(params)
+            .await
+            .map(|response| Some(response.into()))
+    }
+
     pub(crate) async fn thread_rollback(
         &self,
         request_id: &ConnectionRequestId,
@@ -1725,6 +1743,54 @@ impl ThreadRequestProcessor {
         Ok(ThreadBackgroundTerminalsCleanResponse {})
     }
 
+    async fn thread_background_terminals_list_inner(
+        &self,
+        params: ThreadBackgroundTerminalsListParams,
+    ) -> Result<ThreadBackgroundTerminalsListResponse, JSONRPCErrorError> {
+        let ThreadBackgroundTerminalsListParams {
+            thread_id,
+            cursor,
+            limit,
+        } = params;
+
+        let (_, thread) = self.load_thread(&thread_id).await?;
+        let terminals = thread
+            .list_background_terminals()
+            .await
+            .into_iter()
+            .map(|terminal| ThreadBackgroundTerminal {
+                item_id: terminal.item_id,
+                process_id: terminal.process_id,
+                command: terminal.command,
+                cwd: terminal.cwd,
+                os_pid: None,
+                cpu_percent: None,
+                rss_kb: None,
+            })
+            .collect::<Vec<_>>();
+
+        let (data, next_cursor) = paginate_background_terminals(&terminals, cursor, limit)?;
+
+        Ok(ThreadBackgroundTerminalsListResponse { data, next_cursor })
+    }
+
+    async fn thread_background_terminals_terminate_inner(
+        &self,
+        params: ThreadBackgroundTerminalsTerminateParams,
+    ) -> Result<ThreadBackgroundTerminalsTerminateResponse, JSONRPCErrorError> {
+        let ThreadBackgroundTerminalsTerminateParams {
+            thread_id,
+            process_id,
+        } = params;
+        let process_id = process_id.parse::<i32>().map_err(|err| {
+            invalid_request(format!("invalid background terminal process id: {err}"))
+        })?;
+
+        let (_, thread) = self.load_thread(&thread_id).await?;
+        let terminated = thread.terminate_background_terminal(process_id).await;
+        Ok(ThreadBackgroundTerminalsTerminateResponse { terminated })
+    }
+
     async fn thread_shell_command_inner(
         &self,
         request_id: &ConnectionRequestId,
@@ -4250,6 +4316,34 @@ fn build_thread_from_snapshot(
     }
 }
 
+fn paginate_background_terminals(
+    terminals: &[ThreadBackgroundTerminal],
+    cursor: Option<String>,
+    limit: Option<u32>,
+) -> Result<(Vec<ThreadBackgroundTerminal>, Option<String>), JSONRPCErrorError> {
+    let start = match cursor {
+        Some(cursor) => {
+            let cursor = cursor
+                .parse::<i32>()
+                .map_err(|err| invalid_request(format!("invalid cursor: {err}")))?;
+            terminals
+                .iter()
+                .position(|terminal| {
+                    terminal
+                        .process_id
+                        .parse::<i32>()
+                        .is_ok_and(|process_id| process_id > cursor)
+                })
+                .unwrap_or(terminals.len())
+        }
+        None => 0,
+    };
+    let effective_limit = limit.unwrap_or(terminals.len() as u32).max(1) as usize;
+    let end = start.saturating_add(effective_limit).min(terminals.len());
+    let next_cursor = (end < terminals.len()).then(|| terminals[end - 1].process_id.clone());
+    Ok((terminals[start..end].to_vec(), next_cursor))
+}
+
 fn build_thread_from_loaded_snapshot(
     thread_id: ThreadId,
     config_snapshot: &ThreadConfigSnapshot,

codex-rs/app-server/src/request_processors/thread_processor_tests.rs

@@ -36,6 +36,65 @@ mod thread_list_cwd_filter_tests {
     }
 }
 
+mod background_terminal_pagination_tests {
+    use super::super::paginate_background_terminals;
+    use codex_app_server_protocol::ThreadBackgroundTerminal;
+    use codex_utils_absolute_path::AbsolutePathBuf;
+    use pretty_assertions::assert_eq;
+
+    fn terminal(process_id: &str) -> ThreadBackgroundTerminal {
+        let cwd = if cfg!(windows) { r"C:\tmp" } else { "/tmp" };
+
+        ThreadBackgroundTerminal {
+            item_id: format!("item-{process_id}"),
+            process_id: process_id.to_string(),
+            command: format!("command-{process_id}"),
+            cwd: AbsolutePathBuf::from_absolute_path(cwd).expect("absolute cwd"),
+            os_pid: None,
+            cpu_percent: None,
+            rss_kb: None,
+        }
+    }
+
+    #[test]
+    fn paginates_with_process_id_cursor() {
+        let terminals = vec![
+            terminal("1"),
+            terminal("2"),
+            terminal("3"),
+            terminal("4"),
+            terminal("5"),
+        ];
+
+        let (data, next_cursor) =
+            paginate_background_terminals(&terminals, /*cursor*/ None, Some(2))
+                .expect("valid page");
+
+        assert_eq!(data, vec![terminal("1"), terminal("2")]);
+        assert_eq!(next_cursor, Some("2".to_string()));
+        let first_cursor = next_cursor;
+
+        let terminals_without_anchor = vec![terminal("1"), terminal("3"), terminal("4")];
+        let (data, next_cursor) =
+            paginate_background_terminals(&terminals_without_anchor, first_cursor.clone(), Some(2))
+                .expect("valid page");
+
+        assert_eq!(data, vec![terminal("3"), terminal("4")]);
+        assert_eq!(next_cursor, None);
+
+        let (data, next_cursor) =
+            paginate_background_terminals(&terminals, first_cursor, Some(2)).expect("valid page");
+
+        assert_eq!(data, vec![terminal("3"), terminal("4")]);
+        assert_eq!(next_cursor, Some("4".to_string()));
+
+        assert!(
+            paginate_background_terminals(&terminals, Some("missing".to_string()), Some(1))
+                .is_err()
+        );
+    }
+}
+
 mod thread_processor_behavior_tests {
     async fn forked_from_id_from_rollout(path: &Path) -> Option<String> {
         codex_core::read_session_meta_line(path)

codex-rs/core/src/codex_thread.rs

@@ -159,6 +159,14 @@ pub struct CodexThread {
     out_of_band_elicitation_count: Mutex<u64>,
 }
 
+#[derive(Debug, Eq, PartialEq)]
+pub struct BackgroundTerminalInfo {
+    pub item_id: String,
+    pub process_id: String,
+    pub command: String,
+    pub cwd: AbsolutePathBuf,
+}
+
 /// Conduit for the bidirectional stream of messages that compose a thread
 /// (formerly called a conversation) in Codex.
 impl CodexThread {
@@ -396,6 +404,17 @@ impl CodexThread {
         self.codex.agent_status().await
     }
 
+    pub async fn list_background_terminals(&self) -> Vec<BackgroundTerminalInfo> {
+        self.codex.session.list_background_terminals().await
+    }
+
+    pub async fn terminate_background_terminal(&self, process_id: i32) -> bool {
+        self.codex
+            .session
+            .terminate_background_terminal(process_id)
+            .await
+    }
+
     pub(crate) fn subscribe_status(&self) -> watch::Receiver<AgentStatus> {
         self.codex.agent_status.clone()
     }

codex-rs/core/src/lib.rs

@@ -19,6 +19,7 @@ mod codex_thread;
 mod compact_remote;
 mod compact_remote_v2;
 mod config_lock;
+pub use codex_thread::BackgroundTerminalInfo;
 pub use codex_thread::CodexThread;
 pub use codex_thread::CodexThreadSettingsOverrides;
 pub use codex_thread::ThreadConfigSnapshot;

codex-rs/core/src/tasks/mod.rs

@@ -21,6 +21,7 @@ use tracing::info_span;
 use tracing::trace;
 use tracing::warn;
 
+use crate::codex_thread::BackgroundTerminalInfo;
 use crate::config::Config;
 use crate::context::ContextualUserFragment;
 use crate::hook_runtime::inspect_pending_input;
@@ -782,6 +783,17 @@ impl Session {
             .await;
     }
 
+    pub(crate) async fn list_background_terminals(&self) -> Vec<BackgroundTerminalInfo> {
+        self.services.unified_exec_manager.list_processes().await
+    }
+
+    pub(crate) async fn terminate_background_terminal(&self, process_id: i32) -> bool {
+        self.services
+            .unified_exec_manager
+            .terminate_process(process_id)
+            .await
+    }
+
     async fn handle_task_abort(self: &Arc<Self>, task: RunningTask, reason: TurnAbortReason) {
         let sub_id = task.turn_context.sub_id.clone();
         if task.cancellation_token.is_cancelled() {

codex-rs/core/src/unified_exec/mod.rs

@@ -155,6 +155,8 @@ struct ProcessEntry {
     process: Arc<UnifiedExecProcess>,
     call_id: String,
     process_id: i32,
+    cwd: AbsolutePathBuf,
+    initial_exec_command_active: Arc<std::sync::atomic::AtomicBool>,
     hook_command: String,
     tty: bool,
     network_approval: Option<DeferredNetworkApproval>,

codex-rs/core/src/unified_exec/mod_tests.rs

@@ -1,5 +1,6 @@
 use super::head_tail_buffer::HeadTailBuffer;
 use super::*;
+use crate::codex_thread::BackgroundTerminalInfo;
 use crate::exec::ExecCapturePolicy;
 use crate::exec::ExecExpiration;
 use crate::sandboxing::ExecRequest;
@@ -9,6 +10,16 @@ use crate::session::turn_context::TurnContext;
 use crate::tools::context::ExecCommandToolOutput;
 use crate::unified_exec::WriteStdinRequest;
 use crate::unified_exec::process::OutputHandles;
+use async_trait::async_trait;
+use codex_exec_server::ExecProcess;
+use codex_exec_server::ExecProcessEventReceiver;
+use codex_exec_server::ExecServerError;
+use codex_exec_server::ProcessId;
+use codex_exec_server::ProcessSignal;
+use codex_exec_server::ReadResponse;
+use codex_exec_server::StartedExecProcess;
+use codex_exec_server::WriteResponse;
+use codex_exec_server::WriteStatus;
 use codex_sandboxing::SandboxType;
 use codex_utils_output_truncation::TruncationPolicy;
 use codex_utils_output_truncation::approx_token_count;
@@ -19,6 +30,8 @@ use pretty_assertions::assert_eq;
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
+use tokio::sync::Notify;
+use tokio::sync::watch;
 use tokio::time::Duration;
 use tokio::time::Instant;
 
@@ -116,6 +129,8 @@ async fn exec_command_with_tty(
             process: Arc::clone(&process),
             call_id: context.call_id.clone(),
             process_id,
+            cwd: cwd.clone(),
+            initial_exec_command_active: Arc::new(std::sync::atomic::AtomicBool::new(true)),
             hook_command: cmd.to_string(),
             tty,
             network_approval: None,
@@ -158,6 +173,18 @@ async fn exec_command_with_tty(
         manager.release_process_id(process_id).await;
         None
     };
+    if response_process_id.is_some()
+        && let Some(entry) = manager
+            .process_store
+            .lock()
+            .await
+            .processes
+            .get_mut(&process_id)
+    {
+        entry
+            .initial_exec_command_active
+            .store(false, std::sync::atomic::Ordering::Release);
+    }
 
     Ok(ExecCommandToolOutput {
         event_call_id: context.call_id,
@@ -184,6 +211,82 @@ impl SpawnLifecycle for TestSpawnLifecycle {
     }
 }
 
+struct BlockingTerminateExecProcess {
+    process_id: ProcessId,
+    terminate_started: watch::Sender<bool>,
+    allow_terminate: Arc<Notify>,
+    wake_tx: watch::Sender<u64>,
+}
+
+#[async_trait]
+impl ExecProcess for BlockingTerminateExecProcess {
+    fn process_id(&self) -> &ProcessId {
+        &self.process_id
+    }
+
+    fn subscribe_wake(&self) -> watch::Receiver<u64> {
+        self.wake_tx.subscribe()
+    }
+
+    fn subscribe_events(&self) -> ExecProcessEventReceiver {
+        ExecProcessEventReceiver::empty()
+    }
+
+    async fn read(
+        &self,
+        _after_seq: Option<u64>,
+        _max_bytes: Option<usize>,
+        _wait_ms: Option<u64>,
+    ) -> Result<ReadResponse, ExecServerError> {
+        Ok(ReadResponse {
+            chunks: Vec::new(),
+            next_seq: 1,
+            exited: false,
+            exit_code: None,
+            closed: false,
+            failure: None,
+        })
+    }
+
+    async fn write(&self, _chunk: Vec<u8>) -> Result<WriteResponse, ExecServerError> {
+        Ok(WriteResponse {
+            status: WriteStatus::Accepted,
+        })
+    }
+
+    async fn signal(&self, _signal: ProcessSignal) -> Result<(), ExecServerError> {
+        Ok(())
+    }
+
+    async fn terminate(&self) -> Result<(), ExecServerError> {
+        let _ = self.terminate_started.send(true);
+        self.allow_terminate.notified().await;
+        Ok(())
+    }
+}
+
+async fn blocking_terminate_unified_process(
+    process_id: i32,
+    terminate_started: watch::Sender<bool>,
+    allow_terminate: Arc<Notify>,
+) -> anyhow::Result<Arc<UnifiedExecProcess>> {
+    let (wake_tx, _wake_rx) = watch::channel(0);
+    Ok(Arc::new(
+        UnifiedExecProcess::from_exec_server_started(
+            StartedExecProcess {
+                process: Arc::new(BlockingTerminateExecProcess {
+                    process_id: process_id.to_string().into(),
+                    terminate_started,
+                    allow_terminate,
+                    wake_tx,
+                }),
+            },
+            SandboxType::None,
+        )
+        .await?,
+    ))
+}
+
 async fn write_stdin(
     session: &Arc<Session>,
     process_id: i32,
@@ -241,12 +344,23 @@ async fn unified_exec_persists_across_requests() -> anyhow::Result<()> {
     skip_if_sandbox!(Ok(()));
 
     let (session, turn) = test_session_and_turn().await;
+    #[allow(deprecated)]
+    let cwd = turn.cwd.clone();
 
     let open_shell = exec_command(
         &session, &turn, "bash -i", /*yield_time_ms*/ 2_500, /*workdir*/ None,
     )
     .await?;
     let process_id = open_shell.process_id.expect("expected process_id");
+    assert_eq!(
+        session.list_background_terminals().await,
+        vec![BackgroundTerminalInfo {
+            item_id: "call".to_string(),
+            process_id: process_id.to_string(),
+            command: "bash -i".to_string(),
+            cwd,
+        }]
+    );
 
     write_stdin(
         &session,
@@ -270,6 +384,10 @@ async fn unified_exec_persists_across_requests() -> anyhow::Result<()> {
         "expected environment variable output"
     );
 
+    assert!(session.terminate_background_terminal(process_id).await);
+    assert!(!session.terminate_background_terminal(process_id).await);
+    assert!(session.list_background_terminals().await.is_empty());
+
     Ok(())
 }
 
@@ -523,6 +641,148 @@ async fn reusing_completed_process_returns_unknown_process() -> anyhow::Result<(
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn terminating_initial_exec_command_rechecks_initial_response_state() -> anyhow::Result<()> {
+    let (session, turn) = test_session_and_turn().await;
+    let manager = &session.services.unified_exec_manager;
+    let process_id = manager.allocate_process_id().await;
+    let (terminate_started_tx, mut terminate_started_rx) = watch::channel(false);
+    let allow_terminate = Arc::new(Notify::new());
+    let process = blocking_terminate_unified_process(
+        process_id,
+        terminate_started_tx,
+        Arc::clone(&allow_terminate),
+    )
+    .await?;
+    #[allow(deprecated)]
+    let cwd = turn.cwd.clone();
+    manager.process_store.lock().await.processes.insert(
+        process_id,
+        ProcessEntry {
+            process,
+            call_id: "call".to_string(),
+            process_id,
+            cwd,
+            initial_exec_command_active: Arc::new(std::sync::atomic::AtomicBool::new(true)),
+            hook_command: "sleep 60".to_string(),
+            tty: true,
+            network_approval: None,
+            session: Arc::downgrade(&session),
+            last_used: Instant::now(),
+        },
+    );
+
+    let terminate_task = tokio::spawn({
+        let session = Arc::clone(&session);
+        async move { session.terminate_background_terminal(process_id).await }
+    });
+    tokio::time::timeout(
+        Duration::from_secs(2),
+        terminate_started_rx.wait_for(|started| *started),
+    )
+    .await
+    .expect("terminate should start")
+    .expect("terminate signal sender should stay open");
+
+    {
+        let mut store = manager.process_store.lock().await;
+        let entry = store
+            .processes
+            .get_mut(&process_id)
+            .expect("process should remain stored until initial response returns");
+        entry
+            .initial_exec_command_active
+            .store(false, std::sync::atomic::Ordering::Release);
+    }
+
+    allow_terminate.notify_waiters();
+    let terminated = tokio::time::timeout(Duration::from_secs(2), terminate_task)
+        .await
+        .expect("terminate should finish")
+        .expect("terminate task should not panic");
+    assert!(terminated);
+    assert!(
+        !manager
+            .process_store
+            .lock()
+            .await
+            .processes
+            .contains_key(&process_id)
+    );
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn terminating_during_stdin_poll_returns_exited_response() -> anyhow::Result<()> {
+    let (session, turn) = test_session_and_turn().await;
+    let manager = &session.services.unified_exec_manager;
+    let process_id = manager.allocate_process_id().await;
+    let (terminate_started_tx, _terminate_started_rx) = watch::channel(false);
+    let allow_terminate = Arc::new(Notify::new());
+    let process = blocking_terminate_unified_process(
+        process_id,
+        terminate_started_tx,
+        Arc::clone(&allow_terminate),
+    )
+    .await?;
+    #[allow(deprecated)]
+    let cwd = turn.cwd.clone();
+    let last_used = Instant::now() - Duration::from_secs(1);
+    manager.process_store.lock().await.processes.insert(
+        process_id,
+        ProcessEntry {
+            process: Arc::clone(&process),
+            call_id: "call".to_string(),
+            process_id,
+            cwd,
+            initial_exec_command_active: Arc::new(std::sync::atomic::AtomicBool::new(false)),
+            hook_command: "sleep 60".to_string(),
+            tty: true,
+            network_approval: None,
+            session: Arc::downgrade(&session),
+            last_used,
+        },
+    );
+
+    let poll_task = tokio::spawn({
+        let session = Arc::clone(&session);
+        async move {
+            write_stdin(&session, process_id, "", /*yield_time_ms*/ 60_000).await
+        }
+    });
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let poll_started = manager
+                .process_store
+                .lock()
+                .await
+                .processes
+                .get(&process_id)
+                .is_some_and(|entry| entry.last_used != last_used);
+            if poll_started {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("poll should clone process handles");
+
+    manager.release_process_id(process_id).await;
+    allow_terminate.notify_one();
+    process.terminate_confirmed().await?;
+
+    let output = tokio::time::timeout(Duration::from_secs(2), poll_task)
+        .await
+        .expect("poll should finish")
+        .expect("poll task should not panic")?;
+    assert_eq!(output.process_id, None);
+    assert!(manager.process_store.lock().await.processes.is_empty());
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn completed_pipe_commands_preserve_exit_code() -> anyhow::Result<()> {
     let (_, turn) = make_session_and_context().await;

codex-rs/core/src/unified_exec/process.rs

@@ -195,9 +195,16 @@ impl UnifiedExecProcess {
         }
     }
 
-    pub(super) fn terminate(&self) {
+    fn finish_termination(&self) {
         self.output_closed.store(true, Ordering::Release);
         self.output_closed_notify.notify_waiters();
+        self.cancellation_token.cancel();
+        if let Some(output_task) = &self.output_task {
+            output_task.abort();
+        }
+    }
+
+    pub(super) fn terminate(&self) {
         match &self.process_handle {
             ProcessHandle::Local(process_handle) => process_handle.terminate(),
             ProcessHandle::ExecServer(process_handle) => {
@@ -207,10 +214,22 @@ impl UnifiedExecProcess {
                 });
             }
         }
-        self.cancellation_token.cancel();
-        if let Some(output_task) = &self.output_task {
-            output_task.abort();
+        self.finish_termination();
+    }
+
+    pub(super) async fn terminate_confirmed(&self) -> Result<(), UnifiedExecError> {
+        match &self.process_handle {
+            ProcessHandle::Local(process_handle) => process_handle.terminate(),
+            ProcessHandle::ExecServer(process_handle) => {
+                process_handle
+                    .terminate()
+                    .await
+                    .map_err(|err| UnifiedExecError::process_failed(err.to_string()))?;
+            }
         }
+        self.signal_exit(self.exit_code());
+        self.finish_termination();
+        Ok(())
     }
 
     pub(super) async fn interrupt(&self) -> Result<(), UnifiedExecError> {

codex-rs/core/src/unified_exec/process_manager.rs

@@ -11,6 +11,7 @@ use tokio::time::Duration;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 
+use crate::codex_thread::BackgroundTerminalInfo;
 use crate::exec_env::CODEX_THREAD_ID_ENV_VAR;
 use crate::exec_env::create_env;
 use crate::exec_policy::ExecApprovalRequest;
@@ -175,11 +176,22 @@ struct PreparedProcessHandles {
     pause_state: Option<watch::Receiver<bool>>,
     session: Option<Arc<crate::session::session::Session>>,
     network_approval: Option<DeferredNetworkApproval>,
+    call_id: String,
     hook_command: String,
     process_id: i32,
     tty: bool,
 }
 
+struct InitialExecCommandGuard {
+    active: Arc<AtomicBool>,
+}
+
+impl Drop for InitialExecCommandGuard {
+    fn drop(&mut self) {
+        self.active.store(false, Ordering::Release);
+    }
+}
+
 fn exec_server_process_id(process_id: i32) -> String {
     process_id.to_string()
 }
@@ -414,7 +426,8 @@ impl UnifiedExecProcessManager {
         // Persist live sessions before the initial yield wait so interrupting the
         // turn cannot drop the last Arc and terminate the background process.
         let process_started_alive = !process.has_exited() && process.exit_code().is_none();
-        if process_started_alive {
+        let _initial_exec_command_guard = if process_started_alive {
+            let initial_exec_command_active = Arc::new(AtomicBool::new(true));
             self.store_process(
                 Arc::clone(&process),
                 context,
@@ -426,9 +439,15 @@ impl UnifiedExecProcessManager {
                 request.tty,
                 deferred_network_approval.clone(),
                 Arc::clone(&transcript),
+                Arc::clone(&initial_exec_command_active),
             )
             .await;
-        }
+            Some(InitialExecCommandGuard {
+                active: initial_exec_command_active,
+            })
+        } else {
+            None
+        };
 
         let yield_time_ms = clamp_yield_time(request.yield_time_ms);
         // For the initial exec_command call, we both stream output to events
@@ -609,6 +628,7 @@ impl UnifiedExecProcessManager {
             pause_state,
             session,
             network_approval,
+            call_id,
             hook_command,
             process_id,
             tty,
@@ -721,9 +741,13 @@ impl UnifiedExecProcessManager {
                 (None, exit_code, call_id)
             }
             ProcessStatus::Unknown => {
-                return Err(UnifiedExecError::UnknownProcessId {
-                    process_id: request.process_id,
-                });
+                if process.has_exited() {
+                    (None, process.exit_code(), call_id)
+                } else {
+                    return Err(UnifiedExecError::UnknownProcessId {
+                        process_id: request.process_id,
+                    });
+                }
             }
         };
 
@@ -744,29 +768,27 @@ impl UnifiedExecProcessManager {
     }
 
     async fn refresh_process_state(&self, process_id: i32) -> ProcessStatus {
-        {
-            let mut store = self.process_store.lock().await;
-            let Some(entry) = store.processes.get(&process_id) else {
+        let mut store = self.process_store.lock().await;
+        let Some(entry) = store.processes.get_mut(&process_id) else {
+            return ProcessStatus::Unknown;
+        };
+
+        let exit_code = entry.process.exit_code();
+        let process_id = entry.process_id;
+
+        if entry.process.has_exited() {
+            let Some(entry) = store.remove(process_id) else {
                 return ProcessStatus::Unknown;
             };
-
-            let exit_code = entry.process.exit_code();
-            let process_id = entry.process_id;
-
-            if entry.process.has_exited() {
-                let Some(entry) = store.remove(process_id) else {
-                    return ProcessStatus::Unknown;
-                };
-                ProcessStatus::Exited {
-                    exit_code,
-                    entry: Box::new(entry),
-                }
-            } else {
-                ProcessStatus::Alive {
-                    exit_code,
-                    call_id: entry.call_id.clone(),
-                    process_id,
-                }
+            ProcessStatus::Exited {
+                exit_code,
+                entry: Box::new(entry),
+            }
+        } else {
+            ProcessStatus::Alive {
+                exit_code,
+                call_id: entry.call_id.clone(),
+                process_id,
             }
         }
     }
@@ -804,6 +826,7 @@ impl UnifiedExecProcessManager {
             pause_state,
             session,
             network_approval: entry.network_approval.clone(),
+            call_id: entry.call_id.clone(),
             hook_command: entry.hook_command.clone(),
             process_id: entry.process_id,
             tty: entry.tty,
@@ -823,11 +846,14 @@ impl UnifiedExecProcessManager {
         tty: bool,
         network_approval: Option<DeferredNetworkApproval>,
         transcript: Arc<tokio::sync::Mutex<HeadTailBuffer>>,
+        initial_exec_command_active: Arc<AtomicBool>,
     ) {
         let entry = ProcessEntry {
             process: Arc::clone(&process),
             call_id: context.call_id.clone(),
             process_id,
+            cwd: cwd.clone(),
+            initial_exec_command_active,
             hook_command,
             tty,
             network_approval,
@@ -1274,6 +1300,59 @@ impl UnifiedExecProcessManager {
             entry.process.terminate();
         }
     }
+
+    pub(crate) async fn list_processes(&self) -> Vec<BackgroundTerminalInfo> {
+        let store = self.process_store.lock().await;
+        let mut entries = store
+            .processes
+            .values()
+            .filter(|entry| !entry.process.has_exited())
+            .collect::<Vec<_>>();
+        entries.sort_by_key(|entry| entry.process_id);
+        entries
+            .into_iter()
+            .map(|entry| BackgroundTerminalInfo {
+                item_id: entry.call_id.clone(),
+                process_id: entry.process_id.to_string(),
+                command: entry.hook_command.clone(),
+                cwd: entry.cwd.clone(),
+            })
+            .collect()
+    }
+
+    pub(crate) async fn terminate_process(&self, process_id: i32) -> bool {
+        let (process, already_exited) = {
+            let store = self.process_store.lock().await;
+            let Some(entry) = store.processes.get(&process_id) else {
+                return false;
+            };
+            (Arc::clone(&entry.process), entry.process.has_exited())
+        };
+
+        if !already_exited && process.terminate_confirmed().await.is_err() {
+            return false;
+        }
+
+        let entry = {
+            let mut store = self.process_store.lock().await;
+            let Some(entry) = store.processes.get(&process_id) else {
+                return true;
+            };
+            if !Arc::ptr_eq(&entry.process, &process) {
+                return true;
+            }
+            if entry.initial_exec_command_active.load(Ordering::Acquire) {
+                return true;
+            }
+            let Some(entry) = store.remove(process_id) else {
+                return false;
+            };
+            entry
+        };
+
+        unregister_network_approval_for_entry(&entry).await;
+        true
+    }
 }
 
 enum ProcessStatus {

codex-rs/core/src/unified_exec/process_tests.rs

@@ -22,6 +22,7 @@ struct MockExecProcess {
     process_id: ProcessId,
     write_response: WriteResponse,
     read_responses: Mutex<VecDeque<ReadResponse>>,
+    terminate_error: Option<String>,
     wake_tx: watch::Sender<u64>,
 }
 
@@ -69,11 +70,17 @@ impl ExecProcess for MockExecProcess {
     }
 
     async fn terminate(&self) -> Result<(), ExecServerError> {
+        if let Some(message) = &self.terminate_error {
+            return Err(ExecServerError::Protocol(message.clone()));
+        }
         Ok(())
     }
 }
 
-async fn remote_process(write_status: WriteStatus) -> UnifiedExecProcess {
+async fn remote_process(
+    write_status: WriteStatus,
+    terminate_error: Option<String>,
+) -> UnifiedExecProcess {
     let (wake_tx, _wake_rx) = watch::channel(0);
     let started = StartedExecProcess {
         process: Arc::new(MockExecProcess {
@@ -82,6 +89,7 @@ async fn remote_process(write_status: WriteStatus) -> UnifiedExecProcess {
                 status: write_status,
             },
             read_responses: Mutex::new(VecDeque::new()),
+            terminate_error,
             wake_tx,
         }),
     };
@@ -93,7 +101,7 @@ async fn remote_process(write_status: WriteStatus) -> UnifiedExecProcess {
 
 #[tokio::test]
 async fn remote_write_unknown_process_marks_process_exited() {
-    let process = remote_process(WriteStatus::UnknownProcess).await;
+    let process = remote_process(WriteStatus::UnknownProcess, /*terminate_error*/ None).await;
 
     let err = process
         .write(b"hello")
@@ -106,7 +114,7 @@ async fn remote_write_unknown_process_marks_process_exited() {
 
 #[tokio::test]
 async fn remote_write_closed_stdin_marks_process_exited() {
-    let process = remote_process(WriteStatus::StdinClosed).await;
+    let process = remote_process(WriteStatus::StdinClosed, /*terminate_error*/ None).await;
 
     let err = process
         .write(b"hello")
@@ -119,7 +127,7 @@ async fn remote_write_closed_stdin_marks_process_exited() {
 
 #[tokio::test]
 async fn fail_and_terminate_preserves_failure_message() {
-    let process = remote_process(WriteStatus::Accepted).await;
+    let process = remote_process(WriteStatus::Accepted, /*terminate_error*/ None).await;
 
     process.fail_and_terminate("network denied".to_string());
     process.fail_and_terminate("second failure".to_string());
@@ -131,6 +139,32 @@ async fn fail_and_terminate_preserves_failure_message() {
     );
 }
 
+#[tokio::test]
+async fn remote_terminate_confirmed_updates_state_on_success_only() {
+    let process = remote_process(
+        WriteStatus::Accepted,
+        Some("terminate unavailable".to_string()),
+    )
+    .await;
+
+    let err = process
+        .terminate_confirmed()
+        .await
+        .expect_err("expected terminate failure");
+
+    assert!(matches!(err, UnifiedExecError::ProcessFailed { .. }));
+    assert!(!process.has_exited());
+
+    let process = remote_process(WriteStatus::Accepted, /*terminate_error*/ None).await;
+
+    process
+        .terminate_confirmed()
+        .await
+        .expect("terminate should succeed");
+
+    assert!(process.has_exited());
+}
+
 #[tokio::test]
 async fn remote_process_waits_for_early_exit_event() {
     let (wake_tx, _wake_rx) = watch::channel(0);
@@ -148,6 +182,7 @@ async fn remote_process_waits_for_early_exit_event() {
                 closed: true,
                 failure: None,
             }])),
+            terminate_error: None,
             wake_tx: wake_tx.clone(),
         }),
     };