Move code mode behind an IPC host

Split the code-mode protocol and client from the V8-backed runtime so core and
app-server no longer link codex-code-mode in production. ThreadManager now
provisions durable code-mode sessions through a shared external host process,
while tests can still inject the in-process provider.

The IPC protocol uses a persistent stdin/stdout transport. Each frame is a
4-byte big-endian length followed by JSON, with a 16 MiB frame limit. Client
requests carry u64 request IDs so create, execute, wait, terminate, and shutdown
operations can be multiplexed over one process. Session IDs isolate durable
stored values. Execute returns an ExecutionStarted response immediately and an
asynchronous InitialResponse when the initial yield or completion is available.

Nested tool calls and notifications travel from the host back to the client as
delegate requests with their own IDs. Delegate responses, cancellation, and
cell-closed lifecycle messages use the same framed channel. Wire operations
encode errors as Result values. A dead connection fails pending operations,
cancels outstanding delegates, and lets the provider spawn a new host for later
sessions.

Build codex-code-mode-host with V8 pointer-compression sandbox support and add
it to canonical primary and app-server packages, legacy Linux and Windows
bundles, signing verification, installers, Python runtime packages, and release
CI for macOS, Linux, and Windows. The host is discovered next to the current
executable, through CODEX_CODE_MODE_HOST_PATH, or on PATH. OS-level seccomp or
seatbelt restrictions remain a follow-up to this cross-platform process split.

Benchmarks were run from release builds on Linux x86_64 with the V8 sandbox
profile and a text('ok') workload. Cold measurements used 30 samples, warm
session provisioning used 200, and warm command execution used 500. Values are
mean/p50/p95 in milliseconds:

- session startup: in-process 0.002/0.002/0.005, IPC 2.623/2.599/2.894
- fresh-session command: in-process 1.831/1.758/1.915, IPC 7.428/7.252/8.306
- warm session provisioning: in-process 0.002/0.002/0.003,
  IPC 0.471/0.463/0.581
- warm command: in-process 1.759/1.757/1.940, IPC 2.005/2.001/2.166

The steady-state median command overhead is approximately 0.244 ms. The median
fresh host plus first command cost is 7.252 ms.

Validation:

- 62/62 core code-mode integration tests passed against the external host
- focused protocol, client, runtime, host, tools, and trace tests passed
- Cargo and Bazel real-process host IPC tests passed
- 11/11 package builder tests passed
- Bazel lock verification, scoped Clippy fixes, and repository formatting passed

codex-rs/code-mode-host/Cargo.toml

@@ -10,3 +10,14 @@ path = "src/main.rs"
 
 [lints]
 workspace = true
+
+[dependencies]
+codex-code-mode = { workspace = true, features = ["sandbox"] }
+codex-code-mode-protocol = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true, features = ["io-std", "io-util", "macros", "rt-multi-thread", "sync"] }
+tokio-util = { workspace = true, features = ["rt"] }
+
+[dev-dependencies]
+codex-utils-cargo-bin = { workspace = true }
+pretty_assertions = { workspace = true }

codex-rs/code-mode-host/src/main.rs

@@ -1 +1,332 @@
-fn main() {}
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
+
+use codex_code_mode::CodeModeService;
+use codex_code_mode_protocol::CellId;
+use codex_code_mode_protocol::CodeModeNestedToolCall;
+use codex_code_mode_protocol::CodeModeSessionDelegate;
+use codex_code_mode_protocol::NotificationFuture;
+use codex_code_mode_protocol::ToolInvocationFuture;
+use codex_code_mode_protocol::wire::ClientMessage;
+use codex_code_mode_protocol::wire::DelegateRequest;
+use codex_code_mode_protocol::wire::DelegateRequestId;
+use codex_code_mode_protocol::wire::DelegateResponse;
+use codex_code_mode_protocol::wire::HostMessage;
+use codex_code_mode_protocol::wire::HostRequest;
+use codex_code_mode_protocol::wire::HostResponse;
+use codex_code_mode_protocol::wire::SessionId;
+use codex_code_mode_protocol::wire::read_frame;
+use codex_code_mode_protocol::wire::write_frame;
+use tokio::sync::Mutex;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
+
+const IPC_CHANNEL_CAPACITY: usize = 128;
+
+fn main() {
+    let runtime = build_runtime()
+        .unwrap_or_else(|err| panic!("failed to build code-mode host runtime: {err}"));
+    if let Err(err) = runtime.block_on(run()) {
+        eprintln!("codex-code-mode-host failed: {err}");
+        std::process::exit(1);
+    }
+}
+
+fn build_runtime() -> std::io::Result<tokio::runtime::Runtime> {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_io()
+        .enable_time()
+        .build()
+}
+
+async fn run() -> Result<(), String> {
+    let (outgoing_tx, mut outgoing_rx) = mpsc::channel(IPC_CHANNEL_CAPACITY);
+    let peer = Arc::new(HostPeer::new(outgoing_tx));
+    let state = Arc::new(HostState {
+        sessions: Mutex::new(HashMap::new()),
+        next_session_id: AtomicU64::new(1),
+        peer,
+    });
+
+    let writer = tokio::spawn(async move {
+        let mut stdout = tokio::io::stdout();
+        while let Some(message) = outgoing_rx.recv().await {
+            write_frame(&mut stdout, &message)
+                .await
+                .map_err(|err| err.to_string())?;
+        }
+        Ok::<(), String>(())
+    });
+
+    let mut stdin = tokio::io::stdin();
+    while let Some(message) = read_frame::<_, ClientMessage>(&mut stdin)
+        .await
+        .map_err(|err| err.to_string())?
+    {
+        match message {
+            ClientMessage::Request { id, request } => {
+                let state = Arc::clone(&state);
+                tokio::spawn(async move {
+                    state.handle_request(id, request).await;
+                });
+            }
+            ClientMessage::DelegateResponse { id, response } => {
+                state.peer.complete(id, response).await;
+            }
+        }
+    }
+
+    let sessions = state
+        .sessions
+        .lock()
+        .await
+        .drain()
+        .map(|(_, session)| session)
+        .collect::<Vec<_>>();
+    for session in sessions {
+        let _ = session.shutdown().await;
+    }
+    drop(state);
+    writer.await.map_err(|err| err.to_string())?
+}
+
+struct HostState {
+    sessions: Mutex<HashMap<SessionId, Arc<CodeModeService>>>,
+    next_session_id: AtomicU64,
+    peer: Arc<HostPeer>,
+}
+
+impl HostState {
+    async fn handle_request(self: Arc<Self>, request_id: u64, request: HostRequest) {
+        match request {
+            HostRequest::CreateSession => {
+                let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed);
+                let delegate = Arc::new(RemoteDelegate {
+                    session_id,
+                    peer: Arc::clone(&self.peer),
+                });
+                self.sessions.lock().await.insert(
+                    session_id,
+                    Arc::new(CodeModeService::with_delegate(delegate)),
+                );
+                self.respond(request_id, Ok(HostResponse::SessionCreated { session_id }))
+                    .await;
+            }
+            HostRequest::Execute {
+                session_id,
+                request,
+            } => {
+                let error = match self.session(session_id).await {
+                    Ok(session) => match session.execute(request).await {
+                        Ok(started) => {
+                            let cell_id = started.cell_id.clone();
+                            self.respond(
+                                request_id,
+                                Ok(HostResponse::ExecutionStarted { cell_id }),
+                            )
+                            .await;
+                            let peer = Arc::clone(&self.peer);
+                            tokio::spawn(async move {
+                                let response = started.initial_response().await;
+                                peer.send(HostMessage::InitialResponse {
+                                    id: request_id,
+                                    response,
+                                })
+                                .await;
+                            });
+                            return;
+                        }
+                        Err(err) => err,
+                    },
+                    Err(err) => err,
+                };
+                self.respond(request_id, Err(error)).await;
+            }
+            HostRequest::Wait {
+                session_id,
+                request,
+            } => {
+                let result = match self.session(session_id).await {
+                    Ok(session) => session
+                        .wait(request)
+                        .await
+                        .map(|outcome| HostResponse::WaitCompleted { outcome }),
+                    Err(err) => Err(err),
+                };
+                self.respond(request_id, result).await;
+            }
+            HostRequest::Terminate {
+                session_id,
+                cell_id,
+            } => {
+                let result = match self.session(session_id).await {
+                    Ok(session) => session
+                        .terminate(cell_id)
+                        .await
+                        .map(|outcome| HostResponse::WaitCompleted { outcome }),
+                    Err(err) => Err(err),
+                };
+                self.respond(request_id, result).await;
+            }
+            HostRequest::ShutdownSession { session_id } => {
+                let session = self.sessions.lock().await.remove(&session_id);
+                let result = match session {
+                    Some(session) => session
+                        .shutdown()
+                        .await
+                        .map(|()| HostResponse::SessionShutdown),
+                    None => Ok(HostResponse::SessionShutdown),
+                };
+                self.respond(request_id, result).await;
+            }
+        }
+    }
+
+    async fn session(&self, session_id: SessionId) -> Result<Arc<CodeModeService>, String> {
+        self.sessions
+            .lock()
+            .await
+            .get(&session_id)
+            .cloned()
+            .ok_or_else(|| format!("unknown code-mode session {session_id}"))
+    }
+
+    async fn respond(&self, id: u64, response: Result<HostResponse, String>) {
+        self.peer.send(HostMessage::Response { id, response }).await;
+    }
+}
+
+struct HostPeer {
+    outgoing_tx: mpsc::Sender<HostMessage>,
+    pending: Mutex<HashMap<DelegateRequestId, oneshot::Sender<Result<DelegateResponse, String>>>>,
+    next_request_id: AtomicU64,
+}
+
+impl HostPeer {
+    fn new(outgoing_tx: mpsc::Sender<HostMessage>) -> Self {
+        Self {
+            outgoing_tx,
+            pending: Mutex::new(HashMap::new()),
+            next_request_id: AtomicU64::new(1),
+        }
+    }
+
+    async fn send(&self, message: HostMessage) {
+        let _ = self.outgoing_tx.send(message).await;
+    }
+
+    async fn call(
+        &self,
+        session_id: SessionId,
+        request: DelegateRequest,
+        cancellation_token: CancellationToken,
+    ) -> Result<DelegateResponse, String> {
+        let id = self.next_request_id.fetch_add(1, Ordering::Relaxed);
+        let (response_tx, response_rx) = oneshot::channel();
+        self.pending.lock().await.insert(id, response_tx);
+        if self
+            .outgoing_tx
+            .send(HostMessage::DelegateRequest {
+                id,
+                session_id,
+                request,
+            })
+            .await
+            .is_err()
+        {
+            self.pending.lock().await.remove(&id);
+            return Err("code-mode client connection closed".to_string());
+        }
+        tokio::select! {
+            response = response_rx => {
+                response.map_err(|_| "code-mode client closed before returning delegate output".to_string())?
+            }
+            _ = cancellation_token.cancelled() => {
+                self.pending.lock().await.remove(&id);
+                self.send(HostMessage::CancelDelegateRequest { id }).await;
+                Err("code mode delegate request cancelled".to_string())
+            }
+        }
+    }
+
+    async fn complete(&self, id: DelegateRequestId, response: Result<DelegateResponse, String>) {
+        if let Some(sender) = self.pending.lock().await.remove(&id) {
+            let _ = sender.send(response);
+        }
+    }
+}
+
+struct RemoteDelegate {
+    session_id: SessionId,
+    peer: Arc<HostPeer>,
+}
+
+impl CodeModeSessionDelegate for RemoteDelegate {
+    fn invoke_tool<'a>(
+        &'a self,
+        invocation: CodeModeNestedToolCall,
+        cancellation_token: CancellationToken,
+    ) -> ToolInvocationFuture<'a> {
+        Box::pin(async move {
+            match self
+                .peer
+                .call(
+                    self.session_id,
+                    DelegateRequest::InvokeTool(invocation),
+                    cancellation_token,
+                )
+                .await?
+            {
+                DelegateResponse::ToolResult(result) => Ok(result),
+                DelegateResponse::NotificationDelivered => {
+                    Err("code-mode client returned an invalid tool result".to_string())
+                }
+            }
+        })
+    }
+
+    fn notify<'a>(
+        &'a self,
+        call_id: String,
+        cell_id: CellId,
+        text: String,
+        cancellation_token: CancellationToken,
+    ) -> NotificationFuture<'a> {
+        Box::pin(async move {
+            match self
+                .peer
+                .call(
+                    self.session_id,
+                    DelegateRequest::Notify {
+                        call_id,
+                        cell_id,
+                        text,
+                    },
+                    cancellation_token,
+                )
+                .await?
+            {
+                DelegateResponse::NotificationDelivered => Ok(()),
+                DelegateResponse::ToolResult(_) => {
+                    Err("code-mode client returned an invalid notification result".to_string())
+                }
+            }
+        })
+    }
+
+    fn cell_closed(&self, cell_id: &CellId) {
+        let peer = Arc::clone(&self.peer);
+        let cell_id = cell_id.clone();
+        let session_id = self.session_id;
+        tokio::spawn(async move {
+            peer.send(HostMessage::CellClosed {
+                session_id,
+                cell_id,
+            })
+            .await;
+        });
+    }
+}

codex-rs/code-mode-host/tests/host.rs

@@ -0,0 +1,123 @@
+use std::process::Stdio;
+use std::time::Duration;
+
+use codex_code_mode_protocol::ExecuteRequest;
+use codex_code_mode_protocol::FunctionCallOutputContentItem;
+use codex_code_mode_protocol::RuntimeResponse;
+use codex_code_mode_protocol::wire::ClientMessage;
+use codex_code_mode_protocol::wire::HostMessage;
+use codex_code_mode_protocol::wire::HostRequest;
+use codex_code_mode_protocol::wire::HostResponse;
+use codex_code_mode_protocol::wire::read_frame;
+use codex_code_mode_protocol::wire::write_frame;
+use pretty_assertions::assert_eq;
+use tokio::process::Command;
+
+#[tokio::test]
+async fn serves_code_mode_sessions_over_stdio() {
+    let mut child = Command::new(
+        codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")
+            .expect("resolve codex-code-mode-host binary"),
+    )
+    .stdin(Stdio::piped())
+    .stdout(Stdio::piped())
+    .stderr(Stdio::inherit())
+    .kill_on_drop(true)
+    .spawn()
+    .expect("spawn codex-code-mode-host");
+    let mut stdin = child.stdin.take().expect("host stdin");
+    let mut stdout = child.stdout.take().expect("host stdout");
+
+    write_frame(
+        &mut stdin,
+        &ClientMessage::Request {
+            id: 1,
+            request: HostRequest::CreateSession,
+        },
+    )
+    .await
+    .expect("create session request");
+    let session_id = match read_frame(&mut stdout).await.expect("create response") {
+        Some(HostMessage::Response {
+            id: 1,
+            response: Ok(HostResponse::SessionCreated { session_id }),
+        }) => session_id,
+        message => panic!("unexpected create-session response: {message:?}"),
+    };
+
+    write_frame(
+        &mut stdin,
+        &ClientMessage::Request {
+            id: 2,
+            request: HostRequest::Execute {
+                session_id,
+                request: ExecuteRequest {
+                    tool_call_id: "call-1".to_string(),
+                    enabled_tools: Vec::new(),
+                    source: "text('hello')".to_string(),
+                    yield_time_ms: None,
+                    max_output_tokens: None,
+                },
+            },
+        },
+    )
+    .await
+    .expect("execute request");
+    let cell_id = match read_frame(&mut stdout).await.expect("execute response") {
+        Some(HostMessage::Response {
+            id: 2,
+            response: Ok(HostResponse::ExecutionStarted { cell_id }),
+        }) => cell_id,
+        message => panic!("unexpected execute response: {message:?}"),
+    };
+    let response = match read_frame(&mut stdout)
+        .await
+        .expect("initial response frame")
+    {
+        Some(HostMessage::InitialResponse {
+            id: 2,
+            response: Ok(response),
+        }) => response,
+        message => panic!("unexpected initial response: {message:?}"),
+    };
+    assert_eq!(
+        response,
+        RuntimeResponse::Result {
+            cell_id,
+            content_items: vec![FunctionCallOutputContentItem::InputText {
+                text: "hello".to_string(),
+            }],
+            error_text: None,
+        }
+    );
+
+    write_frame(
+        &mut stdin,
+        &ClientMessage::Request {
+            id: 3,
+            request: HostRequest::ShutdownSession { session_id },
+        },
+    )
+    .await
+    .expect("shutdown request");
+    loop {
+        match read_frame(&mut stdout).await.expect("shutdown response") {
+            Some(HostMessage::CellClosed {
+                session_id: closed_session_id,
+                ..
+            }) if closed_session_id == session_id => {}
+            Some(HostMessage::Response {
+                id: 3,
+                response: Ok(HostResponse::SessionShutdown),
+            }) => break,
+            message => panic!("unexpected shutdown response: {message:?}"),
+        }
+    }
+
+    drop(stdin);
+    let status = tokio::time::timeout(Duration::from_secs(5), child.wait())
+        .await
+        .expect("host exit timeout")
+        .expect("wait for host");
+    assert!(status.success(), "host exited with {status}");
+}