code-mode: implement standalone host binary

.github/actions/setup-rusty-v8/action.yml

@@ -4,6 +4,10 @@ inputs:
   target:
     description: Rust target triple with Codex-built V8 release artifacts.
     required: true
+  sandbox:
+    description: Use the V8 pointer-compression sandbox artifact profile.
+    required: false
+    default: "false"
 
 runs:
   using: composite
@@ -11,6 +15,7 @@ runs:
     - name: Configure rusty_v8 artifact overrides and verify checksums
       shell: bash
       env:
+        SANDBOX: ${{ inputs.sandbox }}
         TARGET: ${{ inputs.target }}
       run: |
         set -euo pipefail
@@ -19,14 +24,18 @@ runs:
         release_tag="rusty-v8-v${version}"
         base_url="https://github.com/openai/codex/releases/download/${release_tag}"
         binding_dir="${RUNNER_TEMP}/rusty_v8"
-        archive_path="${binding_dir}/librusty_v8_release_${TARGET}.a.gz"
-        binding_path="${binding_dir}/src_binding_release_${TARGET}.rs"
-        checksums_path="${binding_dir}/rusty_v8_release_${TARGET}.sha256"
+        artifact_profile="release"
+        if [[ "${SANDBOX}" == "true" ]]; then
+          artifact_profile="ptrcomp_sandbox_release"
+        fi
+        archive_path="${binding_dir}/librusty_v8_${artifact_profile}_${TARGET}.a.gz"
+        binding_path="${binding_dir}/src_binding_${artifact_profile}_${TARGET}.rs"
+        checksums_path="${binding_dir}/rusty_v8_${artifact_profile}_${TARGET}.sha256"
 
         mkdir -p "${binding_dir}"
-        curl -fsSL "${base_url}/librusty_v8_release_${TARGET}.a.gz" -o "${archive_path}"
-        curl -fsSL "${base_url}/src_binding_release_${TARGET}.rs" -o "${binding_path}"
-        curl -fsSL "${base_url}/rusty_v8_release_${TARGET}.sha256" -o "${checksums_path}"
+        curl -fsSL "${base_url}/$(basename "${archive_path}")" -o "${archive_path}"
+        curl -fsSL "${base_url}/$(basename "${binding_path}")" -o "${binding_path}"
+        curl -fsSL "${base_url}/$(basename "${checksums_path}")" -o "${checksums_path}"
 
         if [[ "$(wc -l < "${checksums_path}")" -ne 2 ]]; then
           echo "Expected exactly two checksums for ${TARGET} in ${checksums_path}" >&2

.github/workflows/rust-ci-full.yml

@@ -382,6 +382,7 @@ jobs:
         name: Configure rusty_v8 artifact overrides and verify checksums
         uses: ./.github/actions/setup-rusty-v8
         with:
+          sandbox: "true"
           target: ${{ matrix.target }}
 
       - name: Install cargo-chef

codex-rs/Cargo.lock

@@ -2493,6 +2493,15 @@ dependencies = [
 [[package]]
 name = "codex-code-mode-host"
 version = "0.0.0"
+dependencies = [
+ "codex-code-mode",
+ "codex-code-mode-protocol",
+ "codex-utils-cargo-bin",
+ "pretty_assertions",
+ "tokio",
+ "tokio-util",
+ "v8",
+]
 
 [[package]]
 name = "codex-code-mode-protocol"

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

@@ -10,3 +10,21 @@ path = "src/main.rs"
 
 [lints]
 workspace = true
+
+[dependencies]
+codex-code-mode = { workspace = true }
+codex-code-mode-protocol = { workspace = true }
+tokio = { workspace = true, features = ["io-std", "io-util", "macros", "rt-multi-thread", "sync"] }
+tokio-util = { workspace = true, features = ["rt"] }
+# The host owns V8 process configuration, so enable the sandbox without an
+# internal workspace-crate feature gate.
+v8 = { workspace = true, features = ["v8_enable_sandbox"] }
+
+[dev-dependencies]
+codex-utils-cargo-bin = { workspace = true }
+pretty_assertions = { workspace = true }
+
+[package.metadata.cargo-shear]
+# This direct dependency intentionally selects the V8 sandbox feature for the
+# host without activating an internal workspace-crate feature.
+ignored = ["v8"]

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

@@ -1 +1,353 @@
-fn main() {}
+use std::collections::HashMap;
+use std::future::Future;
+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),
+        host_id: std::process::id().to_string(),
+        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);
+                spawn_critical_request_task("request handler", 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())?
+}
+
+fn spawn_critical_request_task(
+    description: &'static str,
+    future: impl Future<Output = ()> + Send + 'static,
+) {
+    let task = tokio::spawn(future);
+    tokio::spawn(async move {
+        if let Err(err) = task.await
+            && err.is_panic()
+        {
+            eprintln!("code-mode host {description} panicked: {err}");
+            std::process::exit(1);
+        }
+    });
+}
+
+struct HostState {
+    sessions: Mutex<HashMap<SessionId, Arc<CodeModeService>>>,
+    next_session_id: AtomicU64,
+    host_id: String,
+    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_and_cell_id_prefix(
+                        delegate,
+                        self.host_id.clone(),
+                    )),
+                );
+                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);
+                            spawn_critical_request_task("initial response handler", 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,125 @@
+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 host_id = child.id().expect("host process id");
+    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:?}"),
+    };
+    assert_eq!(cell_id.as_str(), format!("{host_id}_1"));
+    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}");
+}

codex-rs/code-mode-protocol/src/lib.rs

@@ -2,6 +2,7 @@ mod description;
 mod response;
 mod runtime;
 mod session;
+pub mod wire;
 
 pub use description::CODE_MODE_PRAGMA_PREFIX;
 pub use description::CodeModeToolKind;

codex-rs/code-mode-protocol/src/wire.rs

@@ -0,0 +1,156 @@
+use std::io;
+
+use serde::Deserialize;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use serde_json::Value as JsonValue;
+use tokio::io::AsyncRead;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWrite;
+use tokio::io::AsyncWriteExt;
+
+use crate::CellId;
+use crate::CodeModeNestedToolCall;
+use crate::ExecuteRequest;
+use crate::RuntimeResponse;
+use crate::WaitOutcome;
+use crate::WaitRequest;
+
+// Keep allocations from untrusted frame lengths bounded while leaving enough
+// room for large inline image results. These are base64 encoded and can easily
+// exceed the previous 16 MiB limit before downstream output truncation runs.
+const MAX_FRAME_BYTES: usize = 256 * 1024 * 1024;
+
+pub type RequestId = u64;
+pub type SessionId = u64;
+pub type DelegateRequestId = u64;
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum ClientMessage {
+    Request {
+        id: RequestId,
+        request: HostRequest,
+    },
+    DelegateResponse {
+        id: DelegateRequestId,
+        response: Result<DelegateResponse, String>,
+    },
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum HostMessage {
+    Response {
+        id: RequestId,
+        response: Result<HostResponse, String>,
+    },
+    InitialResponse {
+        id: RequestId,
+        response: Result<RuntimeResponse, String>,
+    },
+    DelegateRequest {
+        id: DelegateRequestId,
+        session_id: SessionId,
+        request: DelegateRequest,
+    },
+    CancelDelegateRequest {
+        id: DelegateRequestId,
+    },
+    CellClosed {
+        session_id: SessionId,
+        cell_id: CellId,
+    },
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum HostRequest {
+    CreateSession,
+    Execute {
+        session_id: SessionId,
+        request: ExecuteRequest,
+    },
+    Wait {
+        session_id: SessionId,
+        request: WaitRequest,
+    },
+    Terminate {
+        session_id: SessionId,
+        cell_id: CellId,
+    },
+    ShutdownSession {
+        session_id: SessionId,
+    },
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum HostResponse {
+    SessionCreated { session_id: SessionId },
+    ExecutionStarted { cell_id: CellId },
+    WaitCompleted { outcome: WaitOutcome },
+    SessionShutdown,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum DelegateRequest {
+    InvokeTool(CodeModeNestedToolCall),
+    Notify {
+        call_id: String,
+        cell_id: CellId,
+        text: String,
+    },
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum DelegateResponse {
+    ToolResult(JsonValue),
+    NotificationDelivered,
+}
+
+pub async fn read_frame<R, T>(reader: &mut R) -> io::Result<Option<T>>
+where
+    R: AsyncRead + Unpin,
+    T: DeserializeOwned,
+{
+    let length = match reader.read_u32().await {
+        Ok(length) => length as usize,
+        Err(err) if err.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
+        Err(err) => return Err(err),
+    };
+    if length > MAX_FRAME_BYTES {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            format!("code-mode IPC frame exceeds {MAX_FRAME_BYTES} bytes"),
+        ));
+    }
+    let mut payload = vec![0; length];
+    reader.read_exact(&mut payload).await?;
+    serde_json::from_slice(&payload)
+        .map(Some)
+        .map_err(io::Error::other)
+}
+
+pub async fn write_frame<W, T>(writer: &mut W, message: &T) -> io::Result<()>
+where
+    W: AsyncWrite + Unpin,
+    T: Serialize,
+{
+    let payload = serde_json::to_vec(message).map_err(io::Error::other)?;
+    let length = u32::try_from(payload.len()).map_err(|_| {
+        io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "code-mode IPC frame length exceeds u32",
+        )
+    })?;
+    if payload.len() > MAX_FRAME_BYTES {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("code-mode IPC frame exceeds {MAX_FRAME_BYTES} bytes"),
+        ));
+    }
+    writer.write_u32(length).await?;
+    writer.write_all(&payload).await?;
+    writer.flush().await
+}
+
+#[cfg(test)]
+#[path = "wire_tests.rs"]
+mod tests;

codex-rs/code-mode-protocol/src/wire_tests.rs

@@ -0,0 +1,59 @@
+use tokio::io::AsyncWriteExt;
+use tokio::io::duplex;
+
+use super::ClientMessage;
+use super::HostRequest;
+use super::MAX_FRAME_BYTES;
+use super::read_frame;
+use super::write_frame;
+
+#[tokio::test]
+async fn frame_round_trip_preserves_message() {
+    let (mut client, mut server) = duplex(1024);
+    let message = ClientMessage::Request {
+        id: 7,
+        request: HostRequest::CreateSession,
+    };
+
+    write_frame(&mut client, &message)
+        .await
+        .expect("write frame");
+    let decoded = read_frame(&mut server).await.expect("read frame");
+
+    assert!(matches!(
+        decoded,
+        Some(ClientMessage::Request {
+            id: 7,
+            request: HostRequest::CreateSession,
+        })
+    ));
+}
+
+#[tokio::test]
+async fn frame_round_trip_supports_payloads_larger_than_sixteen_mib() {
+    let (mut client, mut server) = duplex(64 * 1024);
+    let payload = "x".repeat(17 * 1024 * 1024);
+
+    let (write_result, read_result) = tokio::join!(
+        write_frame(&mut client, &payload),
+        read_frame::<_, String>(&mut server),
+    );
+
+    write_result.expect("write frame");
+    assert_eq!(read_result.expect("read frame"), Some(payload));
+}
+
+#[tokio::test]
+async fn read_frame_rejects_lengths_above_allocation_limit() {
+    let (mut client, mut server) = duplex(16);
+    client
+        .write_u32(u32::try_from(MAX_FRAME_BYTES + 1).expect("frame limit fits in u32"))
+        .await
+        .expect("write frame length");
+
+    let error = read_frame::<_, String>(&mut server)
+        .await
+        .expect_err("reject oversized frame");
+
+    assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+}

codex-rs/code-mode/src/service.rs

@@ -93,6 +93,7 @@ struct Inner {
     cells: Mutex<HashMap<CellId, CellHandle>>,
     delegate: Arc<dyn CodeModeSessionDelegate>,
     shutting_down: AtomicBool,
+    cell_id_prefix: Option<String>,
     next_cell_id: AtomicU64,
 }
 
@@ -106,24 +107,38 @@ impl CodeModeService {
     }
 
     pub fn with_delegate(delegate: Arc<dyn CodeModeSessionDelegate>) -> Self {
+        Self::with_inner(delegate, /*cell_id_prefix*/ None)
+    }
+
+    pub fn with_delegate_and_cell_id_prefix(
+        delegate: Arc<dyn CodeModeSessionDelegate>,
+        cell_id_prefix: String,
+    ) -> Self {
+        Self::with_inner(delegate, Some(cell_id_prefix))
+    }
+
+    fn with_inner(
+        delegate: Arc<dyn CodeModeSessionDelegate>,
+        cell_id_prefix: Option<String>,
+    ) -> Self {
         Self {
             inner: Arc::new(Inner {
                 stored_values: Mutex::new(HashMap::new()),
                 cells: Mutex::new(HashMap::new()),
                 delegate,
                 shutting_down: AtomicBool::new(false),
+                cell_id_prefix,
                 next_cell_id: AtomicU64::new(1),
             }),
         }
     }
 
     fn allocate_cell_id(&self) -> CellId {
-        CellId::new(
-            self.inner
-                .next_cell_id
-                .fetch_add(1, Ordering::Relaxed)
-                .to_string(),
-        )
+        let cell_id = self.inner.next_cell_id.fetch_add(1, Ordering::Relaxed);
+        CellId::new(match &self.inner.cell_id_prefix {
+            Some(prefix) => format!("{prefix}_{cell_id}"),
+            None => cell_id.to_string(),
+        })
     }
 
     pub async fn execute(&self, request: ExecuteRequest) -> Result<StartedCell, String> {
@@ -823,6 +838,7 @@ mod tests {
             cells: Mutex::new(HashMap::new()),
             delegate: Arc::new(NoopCodeModeSessionDelegate),
             shutting_down: std::sync::atomic::AtomicBool::new(false),
+            cell_id_prefix: None,
             next_cell_id: AtomicU64::new(1),
         })
     }
@@ -853,6 +869,27 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn cell_ids_include_the_configured_host_prefix() {
+        let service = CodeModeService::with_delegate_and_cell_id_prefix(
+            Arc::new(NoopCodeModeSessionDelegate),
+            "host7".to_string(),
+        );
+
+        let response = execute(&service, execute_request("text('done');")).await;
+
+        assert_eq!(
+            response,
+            RuntimeResponse::Result {
+                cell_id: cell_id("host7_1"),
+                content_items: vec![FunctionCallOutputContentItem::InputText {
+                    text: "done".to_string(),
+                }],
+                error_text: None,
+            }
+        );
+    }
+
     #[tokio::test]
     async fn stored_values_are_shared_between_cells_but_not_sessions() {
         let first_session = CodeModeService::new();

defs.bzl

@@ -365,8 +365,9 @@ def codex_rust_crate(
     for binary_label in extra_binaries:
         sanitized_binaries.append(binary_label)
         binary = Label(binary_label).name
-        cargo_env_runfiles[binary_label] = "CARGO_BIN_EXE_" + binary
-        cargo_env["CARGO_BIN_EXE_" + binary] = "$(rlocationpath %s)" % binary_label
+        cargo_binary = binary.replace("-", "_")
+        cargo_env_runfiles[binary_label] = "CARGO_BIN_EXE_" + cargo_binary
+        cargo_env["CARGO_BIN_EXE_" + cargo_binary] = "$(rlocationpath %s)" % binary_label
 
     integration_test_kwargs = {}
     if integration_test_args: