From 8878b15eb691c493a983dbc568d84ee2547e31ff Mon Sep 17 00:00:00 2001 From: jif-oai Date: Wed, 3 Jun 2026 21:36:27 +0100 Subject: [PATCH] feat: code mode in dedicated process --- codex-rs/Cargo.lock | 2 + codex-rs/app-server/src/message_processor.rs | 7 +- codex-rs/arg0/Cargo.toml | 1 + codex-rs/arg0/src/lib.rs | 4 + codex-rs/code-mode-runtime/Cargo.toml | 1 + codex-rs/code-mode-runtime/src/lib.rs | 4 + codex-rs/code-mode-runtime/src/runtime/mod.rs | 46 ++- codex-rs/code-mode-runtime/src/service.rs | 115 +++++- codex-rs/code-mode-runtime/src/worker.rs | 339 ++++++++++++++++++ codex-rs/code-mode/src/description.rs | 2 +- codex-rs/code-mode/src/runtime.rs | 3 +- codex-rs/mcp-server/src/message_processor.rs | 7 +- 12 files changed, 518 insertions(+), 13 deletions(-) create mode 100644 codex-rs/code-mode-runtime/src/worker.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index a8a5948da8..6938c320d6 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2145,6 +2145,7 @@ version = "0.0.0" dependencies = [ "anyhow", "codex-apply-patch", + "codex-code-mode-runtime", "codex-exec-server", "codex-linux-sandbox", "codex-sandboxing", @@ -2444,6 +2445,7 @@ dependencies = [ "codex-protocol", "deno_core_icudata", "pretty_assertions", + "serde", "serde_json", "tokio", "tokio-util", diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 1e45624d2c..ae22efbdfd 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -305,13 +305,18 @@ impl MessageProcessor { // resumed, or forked threads to a different persistence backend/root. let thread_store = codex_core::thread_store_from_config(config.as_ref(), state_db.clone()); let environment_manager_for_requests = Arc::clone(&environment_manager); + let code_mode_session_provider = Arc::new( + codex_code_mode_runtime::SubprocessCodeModeSessionProvider::new( + arg0_paths.codex_self_exe.clone(), + ), + ); let thread_manager = Arc::new_cyclic(|thread_manager| { ThreadManager::new( config.as_ref(), auth_manager.clone(), session_source, environment_manager, - Arc::new(codex_code_mode_runtime::InProcessCodeModeSessionProvider), + code_mode_session_provider.clone(), thread_extensions( guardian_agent_spawner(thread_manager.clone()), app_server_extension_event_sink(outgoing.clone()), diff --git a/codex-rs/arg0/Cargo.toml b/codex-rs/arg0/Cargo.toml index 7ee21a770e..2a6f6a26a1 100644 --- a/codex-rs/arg0/Cargo.toml +++ b/codex-rs/arg0/Cargo.toml @@ -15,6 +15,7 @@ workspace = true [dependencies] anyhow = { workspace = true } codex-apply-patch = { workspace = true } +codex-code-mode-runtime = { workspace = true } codex-exec-server = { workspace = true } codex-linux-sandbox = { workspace = true } codex-sandboxing = { workspace = true } diff --git a/codex-rs/arg0/src/lib.rs b/codex-rs/arg0/src/lib.rs index 87940f1182..f0ab76114a 100644 --- a/codex-rs/arg0/src/lib.rs +++ b/codex-rs/arg0/src/lib.rs @@ -4,6 +4,7 @@ use std::path::Path; use std::path::PathBuf; use codex_apply_patch::CODEX_CORE_APPLY_PATCH_ARG1; +use codex_code_mode_runtime::CODEX_CODE_MODE_WORKER_ARG1; use codex_exec_server::CODEX_FS_HELPER_ARG1; use codex_sandboxing::landlock::CODEX_LINUX_SANDBOX_ARG0; use codex_utils_home_dir::find_codex_home; @@ -97,6 +98,9 @@ pub fn arg0_dispatch() -> Option { if argv1 == CODEX_FS_HELPER_ARG1 { codex_exec_server::run_fs_helper_main(); } + if argv1 == CODEX_CODE_MODE_WORKER_ARG1 { + codex_code_mode_runtime::run_worker_main(); + } if argv1 == CODEX_CORE_APPLY_PATCH_ARG1 { let patch_arg = args.next().and_then(|s| s.to_str().map(str::to_owned)); let exit_code = match patch_arg { diff --git a/codex-rs/code-mode-runtime/Cargo.toml b/codex-rs/code-mode-runtime/Cargo.toml index b5985577b6..dbb7284935 100644 --- a/codex-rs/code-mode-runtime/Cargo.toml +++ b/codex-rs/code-mode-runtime/Cargo.toml @@ -19,6 +19,7 @@ workspace = true codex-code-mode = { workspace = true } codex-protocol = { workspace = true } deno_core_icudata = { workspace = true } +serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } tokio-util = { workspace = true, features = ["rt"] } diff --git a/codex-rs/code-mode-runtime/src/lib.rs b/codex-rs/code-mode-runtime/src/lib.rs index 2cd9f3518c..94322083ec 100644 --- a/codex-rs/code-mode-runtime/src/lib.rs +++ b/codex-rs/code-mode-runtime/src/lib.rs @@ -1,6 +1,10 @@ mod runtime; mod service; +mod worker; pub(crate) use codex_code_mode::*; pub use service::CodeModeService; pub use service::InProcessCodeModeSessionProvider; +pub use service::SubprocessCodeModeSessionProvider; +pub use worker::CODEX_CODE_MODE_WORKER_ARG1; +pub use worker::run_worker_main; diff --git a/codex-rs/code-mode-runtime/src/runtime/mod.rs b/codex-rs/code-mode-runtime/src/runtime/mod.rs index cd678b494f..2be6a7f655 100644 --- a/codex-rs/code-mode-runtime/src/runtime/mod.rs +++ b/codex-rs/code-mode-runtime/src/runtime/mod.rs @@ -10,6 +10,8 @@ use std::sync::mpsc as std_mpsc; use std::thread; use codex_protocol::ToolName; +use serde::Deserialize; +use serde::Serialize; use serde_json::Value as JsonValue; use tokio::sync::mpsc; @@ -20,7 +22,7 @@ pub(crate) use codex_code_mode::ExecuteRequest; pub(crate) use codex_code_mode::FunctionCallOutputContentItem; const EXIT_SENTINEL: &str = "__codex_code_mode_exit__"; -#[derive(Debug)] +#[derive(Debug, Deserialize, Serialize)] pub(crate) enum RuntimeCommand { ToolResponse { id: String, result: JsonValue }, ToolError { id: String, error_text: String }, @@ -28,19 +30,19 @@ pub(crate) enum RuntimeCommand { Terminate, } -#[derive(Clone, Copy, Debug, PartialEq)] +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)] pub(crate) enum PendingRuntimeMode { Continue, PauseUntilResumed, } -#[derive(Debug)] +#[derive(Debug, Deserialize, Serialize)] pub(crate) enum RuntimeControlCommand { Resume, Terminate, } -#[derive(Debug)] +#[derive(Debug, Deserialize, Serialize)] pub(crate) enum RuntimeEvent { Started, Pending, @@ -74,6 +76,34 @@ pub(crate) fn spawn_runtime( v8::IsolateHandle, ), String, +> { + spawn_runtime_with_options( + stored_values, + request, + event_tx, + pending_mode, + RuntimeOptions::default(), + ) +} + +#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)] +pub(crate) struct RuntimeOptions { + pub max_heap_size_bytes: Option, +} + +pub(crate) fn spawn_runtime_with_options( + stored_values: HashMap, + request: ExecuteRequest, + event_tx: mpsc::UnboundedSender, + pending_mode: PendingRuntimeMode, + options: RuntimeOptions, +) -> Result< + ( + std_mpsc::Sender, + std_mpsc::Sender, + v8::IsolateHandle, + ), + String, > { initialize_v8()?; @@ -100,6 +130,7 @@ pub(crate) fn spawn_runtime( command_rx, control_rx, pending_mode, + options, isolate_handle_tx, runtime_command_tx, ); @@ -163,10 +194,15 @@ fn run_runtime( command_rx: std_mpsc::Receiver, control_rx: std_mpsc::Receiver, pending_mode: PendingRuntimeMode, + options: RuntimeOptions, isolate_handle_tx: std_mpsc::SyncSender, runtime_command_tx: std_mpsc::Sender, ) { - let isolate = &mut v8::Isolate::new(v8::CreateParams::default()); + let mut create_params = v8::CreateParams::default(); + if let Some(max_heap_size_bytes) = options.max_heap_size_bytes { + create_params = create_params.heap_limits(0, max_heap_size_bytes); + } + let isolate = &mut v8::Isolate::new(create_params); let isolate_handle = isolate.thread_safe_handle(); if isolate_handle_tx.send(isolate_handle).is_err() { return; diff --git a/codex-rs/code-mode-runtime/src/service.rs b/codex-rs/code-mode-runtime/src/service.rs index 31fc76d6cd..0093938cd5 100644 --- a/codex-rs/code-mode-runtime/src/service.rs +++ b/codex-rs/code-mode-runtime/src/service.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicU64; @@ -36,6 +37,8 @@ use crate::runtime::RuntimeCommand; use crate::runtime::RuntimeControlCommand; use crate::runtime::RuntimeEvent; use crate::runtime::spawn_runtime; +use crate::worker::SubprocessRuntimeHandle; +use crate::worker::spawn_subprocess_runtime; #[derive(Default)] pub struct InProcessCodeModeSessionProvider; @@ -53,6 +56,47 @@ impl CodeModeSessionProvider for InProcessCodeModeSessionProvider { } } +/// Runs each code-mode cell in a dedicated subprocess. +/// +/// A fatal V8 failure terminates only the active cell process. Session state, +/// nested-tool dispatch, and subsequent code-mode calls remain in the host. +#[derive(Clone, Default)] +pub struct SubprocessCodeModeSessionProvider { + executable: Option, +} + +impl SubprocessCodeModeSessionProvider { + pub fn new(executable: Option) -> Self { + Self { executable } + } +} + +impl CodeModeSessionProvider for SubprocessCodeModeSessionProvider { + fn create_session<'a>( + &'a self, + delegate: Arc, + ) -> CodeModeSessionProviderFuture<'a> { + Box::pin(async move { + let executable = match self.executable.clone() { + Some(executable) => executable, + None => std::env::current_exe().map_err(|err| { + format!("failed to resolve code mode worker executable: {err}") + })?, + }; + let session: Arc = Arc::new( + CodeModeService::with_subprocess_delegate(delegate, executable), + ); + Ok(session) + }) + } +} + +#[derive(Clone)] +enum RuntimeBackend { + InProcess, + Subprocess { executable: PathBuf }, +} + #[derive(Clone)] struct CellHandle { control_tx: mpsc::UnboundedSender, @@ -64,6 +108,7 @@ struct Inner { stored_values: Mutex>, cells: Mutex>, delegate: Arc, + runtime_backend: RuntimeBackend, shutting_down: AtomicBool, next_cell_id: AtomicU64, } @@ -78,11 +123,26 @@ impl CodeModeService { } pub fn with_delegate(delegate: Arc) -> Self { + Self::with_runtime_backend(delegate, RuntimeBackend::InProcess) + } + + fn with_subprocess_delegate( + delegate: Arc, + executable: PathBuf, + ) -> Self { + Self::with_runtime_backend(delegate, RuntimeBackend::Subprocess { executable }) + } + + fn with_runtime_backend( + delegate: Arc, + runtime_backend: RuntimeBackend, + ) -> Self { Self { inner: Arc::new(Inner { stored_values: Mutex::new(HashMap::new()), cells: Mutex::new(HashMap::new()), delegate, + runtime_backend, shutting_down: AtomicBool::new(false), next_cell_id: AtomicU64::new(1), }), @@ -159,7 +219,32 @@ impl CodeModeService { } let (runtime_tx, runtime_control_tx, runtime_terminate_handle) = - spawn_runtime(stored_values, request, event_tx, pending_mode)?; + match &self.inner.runtime_backend { + RuntimeBackend::InProcess => { + let (runtime_tx, runtime_control_tx, runtime_terminate_handle) = + spawn_runtime(stored_values, request, event_tx, pending_mode)?; + ( + runtime_tx, + runtime_control_tx, + RuntimeTerminateHandle::InProcess(runtime_terminate_handle), + ) + } + RuntimeBackend::Subprocess { executable } => { + let (runtime_tx, runtime_control_tx, runtime_terminate_handle) = + spawn_subprocess_runtime( + executable, + stored_values, + request, + event_tx, + pending_mode, + )?; + ( + runtime_tx, + runtime_control_tx, + RuntimeTerminateHandle::Subprocess(runtime_terminate_handle), + ) + } + }; cells.insert( cell_id.clone(), @@ -358,10 +443,24 @@ struct CellControlContext { runtime_tx: std::sync::mpsc::Sender, runtime_control_tx: std::sync::mpsc::Sender, pending_mode: PendingRuntimeMode, - runtime_terminate_handle: v8::IsolateHandle, + runtime_terminate_handle: RuntimeTerminateHandle, cancellation_token: CancellationToken, } +enum RuntimeTerminateHandle { + InProcess(v8::IsolateHandle), + Subprocess(SubprocessRuntimeHandle), +} + +impl RuntimeTerminateHandle { + fn terminate_execution(&self) -> bool { + match self { + Self::InProcess(handle) => handle.terminate_execution(), + Self::Subprocess(handle) => handle.terminate_execution(), + } + } +} + fn missing_cell_response(cell_id: CellId) -> RuntimeResponse { RuntimeResponse::Result { error_text: Some(format!("exec cell {cell_id} not found")), @@ -475,7 +574,10 @@ async fn run_cell_control( if pending_result.is_none() { let result = PendingResult { content_items: std::mem::take(&mut content_items), - error_text: Some("exec runtime ended unexpectedly".to_string()), + error_text: Some( + "Code mode runtime crashed. No stored values from this cell were committed." + .to_string(), + ), }; if send_or_buffer_result( &cell_id, @@ -746,8 +848,10 @@ mod tests { use super::Inner; use super::NoopCodeModeSessionDelegate; use super::PendingRuntimeMode; + use super::RuntimeBackend; use super::RuntimeCommand; use super::RuntimeResponse; + use super::RuntimeTerminateHandle; use super::WaitOutcome; use super::WaitRequest; use super::WaitToPendingOutcome; @@ -790,6 +894,7 @@ mod tests { stored_values: Mutex::new(HashMap::new()), cells: Mutex::new(HashMap::new()), delegate: Arc::new(NoopCodeModeSessionDelegate), + runtime_backend: RuntimeBackend::InProcess, shutting_down: std::sync::atomic::AtomicBool::new(false), next_cell_id: AtomicU64::new(1), }) @@ -1667,7 +1772,9 @@ image({ runtime_tx: runtime_tx.clone(), runtime_control_tx, pending_mode: PendingRuntimeMode::Continue, - runtime_terminate_handle, + runtime_terminate_handle: RuntimeTerminateHandle::InProcess( + runtime_terminate_handle, + ), cancellation_token: tokio_util::sync::CancellationToken::new(), }, event_rx, diff --git a/codex-rs/code-mode-runtime/src/worker.rs b/codex-rs/code-mode-runtime/src/worker.rs new file mode 100644 index 0000000000..8c4722b0bc --- /dev/null +++ b/codex-rs/code-mode-runtime/src/worker.rs @@ -0,0 +1,339 @@ +use std::collections::HashMap; +use std::io::BufRead; +use std::io::Write; +use std::path::Path; +use std::process::Child; +use std::process::ChildStdin; +use std::process::Stdio; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::mpsc as std_mpsc; +use std::thread; +use std::time::Duration; + +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value as JsonValue; +use tokio::sync::mpsc; +use tracing::warn; + +use crate::ExecuteRequest; +use crate::runtime::PendingRuntimeMode; +use crate::runtime::RuntimeCommand; +use crate::runtime::RuntimeControlCommand; +use crate::runtime::RuntimeEvent; +use crate::runtime::RuntimeOptions; +use crate::runtime::spawn_runtime_with_options; + +pub const CODEX_CODE_MODE_WORKER_ARG1: &str = "--codex-run-as-code-mode-worker"; +pub const CODEX_CODE_MODE_V8_MAX_HEAP_BYTES_ENV_VAR: &str = "CODEX_CODE_MODE_V8_MAX_HEAP_BYTES"; + +#[derive(Deserialize, Serialize)] +struct WorkerStart { + stored_values: HashMap, + request: ExecuteRequest, + pending_mode: PendingRuntimeMode, + options: RuntimeOptions, +} + +#[derive(Deserialize, Serialize)] +enum WorkerCommand { + Runtime(RuntimeCommand), + Control(RuntimeControlCommand), +} + +pub(crate) struct SubprocessRuntimeHandle { + child: Arc>, +} + +struct ChildKillGuard(Option); + +impl Drop for ChildKillGuard { + fn drop(&mut self) { + if let Some(mut child) = self.0.take() { + let _ = child.kill(); + } + } +} + +impl SubprocessRuntimeHandle { + pub(crate) fn terminate_execution(&self) -> bool { + let Ok(mut child) = self.child.lock() else { + return false; + }; + child.kill().is_ok() + } +} + +impl Drop for SubprocessRuntimeHandle { + fn drop(&mut self) { + self.terminate_execution(); + } +} + +pub(crate) fn spawn_subprocess_runtime( + executable: &Path, + stored_values: HashMap, + request: ExecuteRequest, + event_tx: mpsc::UnboundedSender, + pending_mode: PendingRuntimeMode, +) -> Result< + ( + std_mpsc::Sender, + std_mpsc::Sender, + SubprocessRuntimeHandle, + ), + String, +> { + let child = std::process::Command::new(executable) + .arg(CODEX_CODE_MODE_WORKER_ARG1) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|err| format!("failed to spawn code mode runtime worker: {err}"))?; + let mut child_guard = ChildKillGuard(Some(child)); + let child = child_guard + .0 + .as_mut() + .expect("child exists until guard is disarmed"); + let stdin = child + .stdin + .take() + .ok_or_else(|| "code mode runtime worker has no stdin".to_string())?; + let stdout = child + .stdout + .take() + .ok_or_else(|| "code mode runtime worker has no stdout".to_string())?; + let stderr = child + .stderr + .take() + .ok_or_else(|| "code mode runtime worker has no stderr".to_string())?; + let stdin = Arc::new(Mutex::new(stdin)); + + write_worker_message( + &stdin, + &WorkerStart { + stored_values, + request, + pending_mode, + options: runtime_options_from_env(), + }, + )?; + + let child = child_guard + .0 + .take() + .expect("child exists until guard is disarmed"); + let (runtime_tx, runtime_rx) = std_mpsc::channel(); + let (control_tx, control_rx) = std_mpsc::channel(); + let child = Arc::new(Mutex::new(child)); + spawn_command_writer( + Arc::clone(&stdin), + runtime_rx, + WorkerCommand::Runtime, + Arc::clone(&child), + ); + spawn_command_writer( + stdin, + control_rx, + WorkerCommand::Control, + Arc::clone(&child), + ); + spawn_event_reader(stdout, event_tx, Arc::clone(&child)); + spawn_stderr_reader(stderr); + spawn_child_reaper(Arc::clone(&child)); + + Ok((runtime_tx, control_tx, SubprocessRuntimeHandle { child })) +} + +fn runtime_options_from_env() -> RuntimeOptions { + let max_heap_size_bytes = std::env::var(CODEX_CODE_MODE_V8_MAX_HEAP_BYTES_ENV_VAR) + .ok() + .and_then(|value| match value.parse::() { + Ok(value) => Some(value), + Err(err) => { + warn!("ignoring invalid {CODEX_CODE_MODE_V8_MAX_HEAP_BYTES_ENV_VAR} value: {err}"); + None + } + }); + RuntimeOptions { + max_heap_size_bytes, + } +} + +fn spawn_command_writer( + stdin: Arc>, + receiver: std_mpsc::Receiver, + wrap: fn(T) -> WorkerCommand, + child: Arc>, +) where + T: Send + 'static, +{ + thread::spawn(move || { + for message in receiver { + if let Err(err) = write_worker_message(&stdin, &wrap(message)) { + warn!("failed to write to code mode runtime worker: {err}"); + kill_child(&child); + break; + } + } + }); +} + +fn spawn_event_reader( + stdout: std::process::ChildStdout, + event_tx: mpsc::UnboundedSender, + child: Arc>, +) { + thread::spawn(move || { + for line in std::io::BufReader::new(stdout).lines() { + let event = match line { + Ok(line) => match serde_json::from_str::(&line) { + Ok(event) => event, + Err(err) => { + warn!("invalid event from code mode runtime worker: {err}"); + kill_child(&child); + break; + } + }, + Err(err) => { + warn!("failed to read from code mode runtime worker: {err}"); + kill_child(&child); + break; + } + }; + if event_tx.send(event).is_err() { + kill_child(&child); + break; + } + } + }); +} + +fn spawn_stderr_reader(stderr: std::process::ChildStderr) { + thread::spawn(move || { + for line in std::io::BufReader::new(stderr).lines() { + match line { + Ok(line) => warn!("code mode runtime worker stderr: {line}"), + Err(err) => { + warn!("failed to read code mode runtime worker stderr: {err}"); + break; + } + } + } + }); +} + +fn spawn_child_reaper(child: Arc>) { + thread::spawn(move || { + loop { + let exited = match child.lock() { + Ok(mut child) => match child.try_wait() { + Ok(Some(_)) => true, + Ok(None) => false, + Err(err) => { + warn!("failed to wait for code mode runtime worker: {err}"); + true + } + }, + Err(_) => true, + }; + if exited { + break; + } + thread::sleep(Duration::from_millis(50)); + } + }); +} + +fn kill_child(child: &Arc>) { + if let Ok(mut child) = child.lock() { + let _ = child.kill(); + } +} + +fn write_worker_message(stdin: &Arc>, message: &T) -> Result<(), String> +where + T: Serialize, +{ + let mut stdin = stdin + .lock() + .map_err(|_| "code mode runtime worker stdin lock poisoned".to_string())?; + serde_json::to_writer(&mut *stdin, message) + .map_err(|err| format!("failed to serialize code mode runtime worker message: {err}"))?; + stdin + .write_all(b"\n") + .and_then(|()| stdin.flush()) + .map_err(|err| format!("failed to write code mode runtime worker message: {err}")) +} + +pub fn run_worker_main() -> ! { + let exit_code = match run_worker() { + Ok(()) => 0, + Err(err) => { + eprintln!("code mode runtime worker failed: {err}"); + 1 + } + }; + std::process::exit(exit_code); +} + +fn run_worker() -> Result<(), String> { + let mut stdin = std::io::BufReader::new(std::io::stdin()); + let mut start = String::new(); + let bytes_read = stdin + .read_line(&mut start) + .map_err(|err| format!("failed to read code mode runtime worker start message: {err}"))?; + if bytes_read == 0 { + return Err("missing code mode runtime worker start message".to_string()); + } + let start: WorkerStart = serde_json::from_str(&start) + .map_err(|err| format!("invalid code mode runtime worker start message: {err}"))?; + let (event_tx, mut event_rx) = mpsc::unbounded_channel(); + let (runtime_tx, control_tx, _terminate_handle) = spawn_runtime_with_options( + start.stored_values, + start.request, + event_tx, + start.pending_mode, + start.options, + )?; + + thread::spawn(move || { + for line in stdin.lines() { + let command = match line { + Ok(line) => match serde_json::from_str::(&line) { + Ok(command) => command, + Err(_) => break, + }, + Err(_) => break, + }; + match command { + WorkerCommand::Runtime(command) => { + if runtime_tx.send(command).is_err() { + return; + } + } + WorkerCommand::Control(command) => { + if control_tx.send(command).is_err() { + return; + } + } + } + } + let _ = runtime_tx.send(RuntimeCommand::Terminate); + let _ = control_tx.send(RuntimeControlCommand::Terminate); + }); + + let stdout = std::io::stdout(); + let mut stdout = stdout.lock(); + while let Some(event) = event_rx.blocking_recv() { + serde_json::to_writer(&mut stdout, &event) + .map_err(|err| format!("failed to serialize code mode runtime event: {err}"))?; + stdout + .write_all(b"\n") + .and_then(|()| stdout.flush()) + .map_err(|err| format!("failed to write code mode runtime event: {err}"))?; + } + Ok(()) +} diff --git a/codex-rs/code-mode/src/description.rs b/codex-rs/code-mode/src/description.rs index 50fcc034ba..e53308dbd8 100644 --- a/codex-rs/code-mode/src/description.rs +++ b/codex-rs/code-mode/src/description.rs @@ -127,7 +127,7 @@ pub enum CodeModeToolKind { Freeform, } -#[derive(Clone, Debug, PartialEq)] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] pub struct ToolDefinition { pub name: String, pub tool_name: ToolName, diff --git a/codex-rs/code-mode/src/runtime.rs b/codex-rs/code-mode/src/runtime.rs index 5cdbcac32b..0817ce3352 100644 --- a/codex-rs/code-mode/src/runtime.rs +++ b/codex-rs/code-mode/src/runtime.rs @@ -1,4 +1,5 @@ use codex_protocol::ToolName; +use serde::Deserialize; use serde::Serialize; use serde_json::Value as JsonValue; @@ -11,7 +12,7 @@ pub const DEFAULT_EXEC_YIELD_TIME_MS: u64 = 10_000; pub const DEFAULT_WAIT_YIELD_TIME_MS: u64 = 10_000; pub const DEFAULT_MAX_OUTPUT_TOKENS_PER_EXEC_CALL: usize = 10_000; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Deserialize, Serialize)] pub struct ExecuteRequest { pub tool_call_id: String, pub enabled_tools: Vec, diff --git a/codex-rs/mcp-server/src/message_processor.rs b/codex-rs/mcp-server/src/message_processor.rs index bb0e6753af..14b300a152 100644 --- a/codex-rs/mcp-server/src/message_processor.rs +++ b/codex-rs/mcp-server/src/message_processor.rs @@ -62,12 +62,17 @@ impl MessageProcessor { /*enable_codex_api_key_env*/ false, ) .await; + let code_mode_session_provider = Arc::new( + codex_code_mode_runtime::SubprocessCodeModeSessionProvider::new( + arg0_paths.codex_self_exe.clone(), + ), + ); let thread_manager = Arc::new(ThreadManager::new( config.as_ref(), auth_manager, SessionSource::Mcp, environment_manager, - Arc::new(codex_code_mode_runtime::InProcessCodeModeSessionProvider), + code_mode_session_provider, empty_extension_registry(), /*analytics_events_client*/ None, codex_core::thread_store_from_config(config.as_ref(), state_db.clone()),