mirror of
https://github.com/openai/codex.git
synced 2026-10-21 17:48:44 +00:00
feat: code mode in dedicated process
This commit is contained in:
Generated
+2
@@ -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",
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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<Arg0PathEntryGuard> {
|
||||
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 {
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<usize>,
|
||||
}
|
||||
|
||||
pub(crate) fn spawn_runtime_with_options(
|
||||
stored_values: HashMap<String, JsonValue>,
|
||||
request: ExecuteRequest,
|
||||
event_tx: mpsc::UnboundedSender<RuntimeEvent>,
|
||||
pending_mode: PendingRuntimeMode,
|
||||
options: RuntimeOptions,
|
||||
) -> Result<
|
||||
(
|
||||
std_mpsc::Sender<RuntimeCommand>,
|
||||
std_mpsc::Sender<RuntimeControlCommand>,
|
||||
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<RuntimeCommand>,
|
||||
control_rx: std_mpsc::Receiver<RuntimeControlCommand>,
|
||||
pending_mode: PendingRuntimeMode,
|
||||
options: RuntimeOptions,
|
||||
isolate_handle_tx: std_mpsc::SyncSender<v8::IsolateHandle>,
|
||||
runtime_command_tx: std_mpsc::Sender<RuntimeCommand>,
|
||||
) {
|
||||
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;
|
||||
|
||||
@@ -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<PathBuf>,
|
||||
}
|
||||
|
||||
impl SubprocessCodeModeSessionProvider {
|
||||
pub fn new(executable: Option<PathBuf>) -> Self {
|
||||
Self { executable }
|
||||
}
|
||||
}
|
||||
|
||||
impl CodeModeSessionProvider for SubprocessCodeModeSessionProvider {
|
||||
fn create_session<'a>(
|
||||
&'a self,
|
||||
delegate: Arc<dyn CodeModeSessionDelegate>,
|
||||
) -> 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<dyn CodeModeSession> = 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<CellControlCommand>,
|
||||
@@ -64,6 +108,7 @@ struct Inner {
|
||||
stored_values: Mutex<HashMap<String, JsonValue>>,
|
||||
cells: Mutex<HashMap<CellId, CellHandle>>,
|
||||
delegate: Arc<dyn CodeModeSessionDelegate>,
|
||||
runtime_backend: RuntimeBackend,
|
||||
shutting_down: AtomicBool,
|
||||
next_cell_id: AtomicU64,
|
||||
}
|
||||
@@ -78,11 +123,26 @@ impl CodeModeService {
|
||||
}
|
||||
|
||||
pub fn with_delegate(delegate: Arc<dyn CodeModeSessionDelegate>) -> Self {
|
||||
Self::with_runtime_backend(delegate, RuntimeBackend::InProcess)
|
||||
}
|
||||
|
||||
fn with_subprocess_delegate(
|
||||
delegate: Arc<dyn CodeModeSessionDelegate>,
|
||||
executable: PathBuf,
|
||||
) -> Self {
|
||||
Self::with_runtime_backend(delegate, RuntimeBackend::Subprocess { executable })
|
||||
}
|
||||
|
||||
fn with_runtime_backend(
|
||||
delegate: Arc<dyn CodeModeSessionDelegate>,
|
||||
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<RuntimeCommand>,
|
||||
runtime_control_tx: std::sync::mpsc::Sender<RuntimeControlCommand>,
|
||||
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,
|
||||
|
||||
@@ -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<String, JsonValue>,
|
||||
request: ExecuteRequest,
|
||||
pending_mode: PendingRuntimeMode,
|
||||
options: RuntimeOptions,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize)]
|
||||
enum WorkerCommand {
|
||||
Runtime(RuntimeCommand),
|
||||
Control(RuntimeControlCommand),
|
||||
}
|
||||
|
||||
pub(crate) struct SubprocessRuntimeHandle {
|
||||
child: Arc<Mutex<Child>>,
|
||||
}
|
||||
|
||||
struct ChildKillGuard(Option<Child>);
|
||||
|
||||
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<String, JsonValue>,
|
||||
request: ExecuteRequest,
|
||||
event_tx: mpsc::UnboundedSender<RuntimeEvent>,
|
||||
pending_mode: PendingRuntimeMode,
|
||||
) -> Result<
|
||||
(
|
||||
std_mpsc::Sender<RuntimeCommand>,
|
||||
std_mpsc::Sender<RuntimeControlCommand>,
|
||||
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::<usize>() {
|
||||
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<T>(
|
||||
stdin: Arc<Mutex<ChildStdin>>,
|
||||
receiver: std_mpsc::Receiver<T>,
|
||||
wrap: fn(T) -> WorkerCommand,
|
||||
child: Arc<Mutex<Child>>,
|
||||
) 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<RuntimeEvent>,
|
||||
child: Arc<Mutex<Child>>,
|
||||
) {
|
||||
thread::spawn(move || {
|
||||
for line in std::io::BufReader::new(stdout).lines() {
|
||||
let event = match line {
|
||||
Ok(line) => match serde_json::from_str::<RuntimeEvent>(&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<Mutex<Child>>) {
|
||||
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<Mutex<Child>>) {
|
||||
if let Ok(mut child) = child.lock() {
|
||||
let _ = child.kill();
|
||||
}
|
||||
}
|
||||
|
||||
fn write_worker_message<T>(stdin: &Arc<Mutex<ChildStdin>>, 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::<WorkerCommand>(&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(())
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ToolDefinition>,
|
||||
|
||||
@@ -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()),
|
||||
|
||||
Reference in New Issue
Block a user