mirror of
https://github.com/openai/codex.git
synced 2026-11-11 06:31:37 +00:00
code-mode: fall back to using in process v8 if we fail to resolve external process (#31899)
## Why Not every Codex distribution currently includes the `codex-code-mode-host` companion binary. Enabling the process-host feature should not make code mode unavailable on those surfaces while packaging support is being completed. ## What changed - Fall back to an in-process code-mode session only when spawning the companion binary returns `io::ErrorKind::NotFound`. - Keep permission, handshake, timeout, and other host failures visible instead of silently falling back. - Store the provider's owned-process/in-process choice as one enum-backed state so later sessions reuse the fallback decision. - Preserve the underlying spawn `io::Error` while retaining the host path in the displayed error. - Update provider, `CodeModeService`, and end-to-end coverage to verify successful fallback execution. ## Test plan - `just test -p codex-code-mode` - `just test -p codex-core missing_process_host`
This commit is contained in:
1 parent
8347b8de21
commit
8cf9a1b1f8
5 files changed
+149
-55
No files matched your search
@@ -22,6 +22,7 @@ use tokio::sync::Semaphore;
|
|||||||
use tokio::sync::watch;
|
use tokio::sync::watch;
|
||||||
|
|
||||||
use self::connection::Connection;
|
use self::connection::Connection;
|
||||||
|
use self::connection::ConnectionError;
|
||||||
use self::connection::RemoteSession;
|
use self::connection::RemoteSession;
|
||||||
use self::connection::SessionCleanup;
|
use self::connection::SessionCleanup;
|
||||||
use crate::NoopCodeModeSessionDelegate;
|
use crate::NoopCodeModeSessionDelegate;
|
||||||
@@ -34,30 +35,32 @@ type ShutdownResultReceiver = watch::Receiver<Option<Result<(), String>>>;
|
|||||||
|
|
||||||
/// Creates code-mode sessions backed by one lazily spawned process host.
|
/// Creates code-mode sessions backed by one lazily spawned process host.
|
||||||
pub struct ProcessOwnedCodeModeSessionProvider {
|
pub struct ProcessOwnedCodeModeSessionProvider {
|
||||||
host_program: PathBuf,
|
state: StdMutex<ProviderState>,
|
||||||
process_host: StdMutex<Option<Arc<OwnedProcessHost>>>,
|
}
|
||||||
|
|
||||||
|
enum ProviderState {
|
||||||
|
OwnedProcess(Arc<OwnedProcessHost>),
|
||||||
|
InProcess,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ProcessOwnedCodeModeSessionProvider {
|
impl ProcessOwnedCodeModeSessionProvider {
|
||||||
pub fn with_host_program(host_program: PathBuf) -> Self {
|
pub fn with_host_program(host_program: PathBuf) -> Self {
|
||||||
Self {
|
Self {
|
||||||
host_program,
|
state: StdMutex::new(ProviderState::OwnedProcess(Arc::new(
|
||||||
process_host: StdMutex::new(None),
|
OwnedProcessHost::new(host_program),
|
||||||
|
))),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn process_host(&self) -> Arc<OwnedProcessHost> {
|
fn process_host(&self) -> Option<Arc<OwnedProcessHost>> {
|
||||||
let mut process_host = self
|
match &*self
|
||||||
.process_host
|
.state
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||||
if let Some(process_host) = process_host.as_ref() {
|
{
|
||||||
return Arc::clone(process_host);
|
ProviderState::OwnedProcess(process_host) => Some(Arc::clone(process_host)),
|
||||||
|
ProviderState::InProcess => None,
|
||||||
}
|
}
|
||||||
|
|
||||||
let new_process_host = Arc::new(OwnedProcessHost::new(self.host_program.clone()));
|
|
||||||
*process_host = Some(Arc::clone(&new_process_host));
|
|
||||||
new_process_host
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -72,8 +75,28 @@ impl CodeModeSessionProvider for ProcessOwnedCodeModeSessionProvider {
|
|||||||
&'a self,
|
&'a self,
|
||||||
delegate: Arc<dyn CodeModeSessionDelegate>,
|
delegate: Arc<dyn CodeModeSessionDelegate>,
|
||||||
) -> CodeModeSessionProviderFuture<'a> {
|
) -> CodeModeSessionProviderFuture<'a> {
|
||||||
let session = ProcessOwnedCodeModeSession::with_process_host(delegate, self.process_host());
|
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
|
let Some(process_host) = self.process_host() else {
|
||||||
|
let session: Arc<dyn CodeModeSession> =
|
||||||
|
Arc::new(crate::InProcessCodeModeSession::with_delegate(delegate));
|
||||||
|
return Ok(session);
|
||||||
|
};
|
||||||
|
|
||||||
|
match process_host.connection().await {
|
||||||
|
Ok(_) => {}
|
||||||
|
Err(error) if error.host_program_not_found() => {
|
||||||
|
*self
|
||||||
|
.state
|
||||||
|
.lock()
|
||||||
|
.unwrap_or_else(std::sync::PoisonError::into_inner) =
|
||||||
|
ProviderState::InProcess;
|
||||||
|
let session: Arc<dyn CodeModeSession> =
|
||||||
|
Arc::new(crate::InProcessCodeModeSession::with_delegate(delegate));
|
||||||
|
return Ok(session);
|
||||||
|
}
|
||||||
|
Err(error) => return Err(error.to_string()),
|
||||||
|
}
|
||||||
|
let session = ProcessOwnedCodeModeSession::with_process_host(delegate, process_host);
|
||||||
session.connection().await?;
|
session.connection().await?;
|
||||||
let session: Arc<dyn CodeModeSession> = Arc::new(session);
|
let session: Arc<dyn CodeModeSession> = Arc::new(session);
|
||||||
Ok(session)
|
Ok(session)
|
||||||
@@ -98,16 +121,14 @@ impl OwnedProcessHost {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn connection(&self) -> Result<Arc<Connection>, String> {
|
async fn connection(&self) -> Result<Arc<Connection>, ConnectionError> {
|
||||||
if let Some(connection) = self.live_connection() {
|
if let Some(connection) = self.live_connection() {
|
||||||
return Ok(connection);
|
return Ok(connection);
|
||||||
}
|
}
|
||||||
|
|
||||||
let _spawn_permit = self
|
let _spawn_permit = self.spawn_permit.acquire().await.map_err(|_| {
|
||||||
.spawn_permit
|
ConnectionError::Other("code-mode host spawn coordinator closed".into())
|
||||||
.acquire()
|
})?;
|
||||||
.await
|
|
||||||
.map_err(|_| "code-mode host spawn coordinator closed".to_string())?;
|
|
||||||
if let Some(connection) = self.live_connection() {
|
if let Some(connection) = self.live_connection() {
|
||||||
return Ok(connection);
|
return Ok(connection);
|
||||||
}
|
}
|
||||||
@@ -284,7 +305,7 @@ impl SessionInner {
|
|||||||
cleanup,
|
cleanup,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
Err(err) => Err(err),
|
Err(err) => Err(err.to_string()),
|
||||||
};
|
};
|
||||||
{
|
{
|
||||||
let mut state = self
|
let mut state = self
|
||||||
|
|||||||
@@ -1,4 +1,7 @@
|
|||||||
|
use std::fmt;
|
||||||
|
use std::io;
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
|
use std::path::PathBuf;
|
||||||
use std::process::Stdio;
|
use std::process::Stdio;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::sync::atomic::AtomicBool;
|
use std::sync::atomic::AtomicBool;
|
||||||
@@ -46,6 +49,39 @@ mod reader;
|
|||||||
const IPC_CHANNEL_CAPACITY: usize = 128;
|
const IPC_CHANNEL_CAPACITY: usize = 128;
|
||||||
const HOST_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
|
const HOST_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
|
pub(super) enum ConnectionError {
|
||||||
|
Spawn {
|
||||||
|
host_program: PathBuf,
|
||||||
|
error: io::Error,
|
||||||
|
},
|
||||||
|
Other(String),
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ConnectionError {
|
||||||
|
pub(super) fn host_program_not_found(&self) -> bool {
|
||||||
|
matches!(
|
||||||
|
self,
|
||||||
|
Self::Spawn { error, .. } if error.kind() == io::ErrorKind::NotFound
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Display for ConnectionError {
|
||||||
|
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
|
match self {
|
||||||
|
Self::Spawn {
|
||||||
|
host_program,
|
||||||
|
error,
|
||||||
|
} => write!(
|
||||||
|
formatter,
|
||||||
|
"failed to spawn code-mode host {}: {error}",
|
||||||
|
host_program.display()
|
||||||
|
),
|
||||||
|
Self::Other(message) => formatter.write_str(message),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub(super) struct Connection {
|
pub(super) struct Connection {
|
||||||
command_tx: mpsc::Sender<DriverCommand>,
|
command_tx: mpsc::Sender<DriverCommand>,
|
||||||
execute_claim_tx: mpsc::UnboundedSender<RequestId>,
|
execute_claim_tx: mpsc::UnboundedSender<RequestId>,
|
||||||
@@ -96,7 +132,7 @@ impl Drop for CallerCancellation {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl Connection {
|
impl Connection {
|
||||||
pub(super) async fn spawn(host_program: &Path) -> Result<Self, String> {
|
pub(super) async fn spawn(host_program: &Path) -> Result<Self, ConnectionError> {
|
||||||
let mut command = Command::new(host_program);
|
let mut command = Command::new(host_program);
|
||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
command.process_group(0);
|
command.process_group(0);
|
||||||
@@ -106,11 +142,9 @@ impl Connection {
|
|||||||
.stderr(Stdio::piped())
|
.stderr(Stdio::piped())
|
||||||
.kill_on_drop(true)
|
.kill_on_drop(true)
|
||||||
.spawn()
|
.spawn()
|
||||||
.map_err(|err| {
|
.map_err(|error| ConnectionError::Spawn {
|
||||||
format!(
|
host_program: host_program.to_path_buf(),
|
||||||
"failed to spawn code-mode host {}: {err}",
|
error,
|
||||||
host_program.display()
|
|
||||||
)
|
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
if let Some(stderr) = child.stderr.take() {
|
if let Some(stderr) = child.stderr.take() {
|
||||||
@@ -132,11 +166,11 @@ impl Connection {
|
|||||||
let stdin = child
|
let stdin = child
|
||||||
.stdin
|
.stdin
|
||||||
.take()
|
.take()
|
||||||
.ok_or_else(|| "spawned code-mode host has no stdin".to_string())?;
|
.ok_or_else(|| ConnectionError::Other("spawned code-mode host has no stdin".into()))?;
|
||||||
let stdout = child
|
let stdout = child
|
||||||
.stdout
|
.stdout
|
||||||
.take()
|
.take()
|
||||||
.ok_or_else(|| "spawned code-mode host has no stdout".to_string())?;
|
.ok_or_else(|| ConnectionError::Other("spawned code-mode host has no stdout".into()))?;
|
||||||
let mut reader = FramedReader::new(stdout);
|
let mut reader = FramedReader::new(stdout);
|
||||||
let mut writer = FramedWriter::new(stdin);
|
let mut writer = FramedWriter::new(stdin);
|
||||||
let handshake = async {
|
let handshake = async {
|
||||||
@@ -174,12 +208,14 @@ impl Connection {
|
|||||||
Ok(result) => result,
|
Ok(result) => result,
|
||||||
Err(_) => {
|
Err(_) => {
|
||||||
kill_and_reap(&mut child).await;
|
kill_and_reap(&mut child).await;
|
||||||
return Err("timed out negotiating with the code-mode host".to_string());
|
return Err(ConnectionError::Other(
|
||||||
|
"timed out negotiating with the code-mode host".into(),
|
||||||
|
));
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if let Err(err) = handshake_result {
|
if let Err(err) = handshake_result {
|
||||||
kill_and_reap(&mut child).await;
|
kill_and_reap(&mut child).await;
|
||||||
return Err(err);
|
return Err(ConnectionError::Other(err));
|
||||||
}
|
}
|
||||||
|
|
||||||
let (command_tx, command_rx) = mpsc::channel(IPC_CHANNEL_CAPACITY);
|
let (command_tx, command_rx) = mpsc::channel(IPC_CHANNEL_CAPACITY);
|
||||||
|
|||||||
@@ -3,6 +3,10 @@ use std::path::PathBuf;
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use codex_code_mode_protocol::CodeModeSessionProvider;
|
use codex_code_mode_protocol::CodeModeSessionProvider;
|
||||||
|
use codex_code_mode_protocol::ExecuteRequest;
|
||||||
|
use codex_code_mode_protocol::FunctionCallOutputContentItem;
|
||||||
|
use codex_code_mode_protocol::RuntimeResponse;
|
||||||
|
use pretty_assertions::assert_eq;
|
||||||
|
|
||||||
use super::ProcessOwnedCodeModeSession;
|
use super::ProcessOwnedCodeModeSession;
|
||||||
use super::ProcessOwnedCodeModeSessionProvider;
|
use super::ProcessOwnedCodeModeSessionProvider;
|
||||||
@@ -13,8 +17,8 @@ use crate::NoopCodeModeSessionDelegate;
|
|||||||
fn provider_reuses_its_live_process_host() {
|
fn provider_reuses_its_live_process_host() {
|
||||||
let provider = ProcessOwnedCodeModeSessionProvider::default();
|
let provider = ProcessOwnedCodeModeSessionProvider::default();
|
||||||
|
|
||||||
let first = provider.process_host();
|
let first = provider.process_host().expect("owned process host");
|
||||||
let second = provider.process_host();
|
let second = provider.process_host().expect("owned process host");
|
||||||
|
|
||||||
assert!(Arc::ptr_eq(&first, &second));
|
assert!(Arc::ptr_eq(&first, &second));
|
||||||
}
|
}
|
||||||
@@ -68,18 +72,39 @@ fn host_program_falls_back_to_its_name_when_main_executable_is_unknown() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn provider_reports_host_spawn_failure() {
|
async fn provider_falls_back_to_in_process_session_when_host_is_missing() {
|
||||||
let provider = ProcessOwnedCodeModeSessionProvider::with_host_program(
|
let provider = ProcessOwnedCodeModeSessionProvider::with_host_program(
|
||||||
"codex-code-mode-host-does-not-exist".into(),
|
"codex-code-mode-host-does-not-exist".into(),
|
||||||
);
|
);
|
||||||
|
|
||||||
let error = provider
|
let session = provider
|
||||||
.create_session(Arc::new(NoopCodeModeSessionDelegate))
|
.create_session(Arc::new(NoopCodeModeSessionDelegate))
|
||||||
.await
|
.await
|
||||||
.err()
|
.expect("missing host should fall back to an in-process session");
|
||||||
.expect("session creation should fail");
|
let response = session
|
||||||
|
.execute(ExecuteRequest {
|
||||||
|
tool_call_id: "call-1".to_string(),
|
||||||
|
enabled_tools: Vec::new(),
|
||||||
|
source: "text('fallback')".to_string(),
|
||||||
|
yield_time_ms: None,
|
||||||
|
max_output_tokens: None,
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("execute fallback session")
|
||||||
|
.initial_response()
|
||||||
|
.await
|
||||||
|
.expect("read fallback response");
|
||||||
|
|
||||||
assert!(error.contains("failed to spawn code-mode host"));
|
assert_eq!(
|
||||||
|
response,
|
||||||
|
RuntimeResponse::Result {
|
||||||
|
cell_id: codex_code_mode_protocol::CellId::new("1".to_string()),
|
||||||
|
content_items: vec![FunctionCallOutputContentItem::InputText {
|
||||||
|
text: "fallback".to_string(),
|
||||||
|
}],
|
||||||
|
error_text: None,
|
||||||
|
}
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
|||||||
@@ -367,7 +367,9 @@ mod tests {
|
|||||||
use crate::tools::context::ToolPayload;
|
use crate::tools::context::ToolPayload;
|
||||||
use codex_code_mode::CodeModeToolKind;
|
use codex_code_mode::CodeModeToolKind;
|
||||||
use codex_code_mode::ExecuteRequest;
|
use codex_code_mode::ExecuteRequest;
|
||||||
|
use codex_code_mode::FunctionCallOutputContentItem as CodeModeOutputContentItem;
|
||||||
use codex_code_mode::ProcessOwnedCodeModeSessionProvider;
|
use codex_code_mode::ProcessOwnedCodeModeSessionProvider;
|
||||||
|
use codex_code_mode::RuntimeResponse;
|
||||||
use codex_protocol::models::FunctionCallOutputContentItem;
|
use codex_protocol::models::FunctionCallOutputContentItem;
|
||||||
use codex_tools::ToolName;
|
use codex_tools::ToolName;
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
@@ -426,26 +428,37 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn missing_process_host_is_reported_without_failing_service_creation() {
|
async fn missing_process_host_falls_back_to_in_process_session() {
|
||||||
let service = CodeModeService::new(Arc::new(
|
let service = CodeModeService::new(Arc::new(
|
||||||
ProcessOwnedCodeModeSessionProvider::with_host_program(
|
ProcessOwnedCodeModeSessionProvider::with_host_program(
|
||||||
"codex-code-mode-host-does-not-exist".into(),
|
"codex-code-mode-host-does-not-exist".into(),
|
||||||
),
|
),
|
||||||
));
|
));
|
||||||
|
|
||||||
let error = service
|
let response = service
|
||||||
.execute(ExecuteRequest {
|
.execute(ExecuteRequest {
|
||||||
tool_call_id: "call-1".to_string(),
|
tool_call_id: "call-1".to_string(),
|
||||||
enabled_tools: Vec::new(),
|
enabled_tools: Vec::new(),
|
||||||
source: "text('unreachable')".to_string(),
|
source: "text('fallback')".to_string(),
|
||||||
yield_time_ms: None,
|
yield_time_ms: None,
|
||||||
max_output_tokens: None,
|
max_output_tokens: None,
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
.err()
|
.expect("missing host should fall back to an in-process session")
|
||||||
.expect("missing host should reject execution");
|
.initial_response()
|
||||||
|
.await
|
||||||
|
.expect("read fallback response");
|
||||||
|
|
||||||
assert!(error.contains("failed to spawn code-mode host"));
|
assert_eq!(
|
||||||
service.shutdown().await.expect("shutdown unused service");
|
response,
|
||||||
|
RuntimeResponse::Result {
|
||||||
|
cell_id: codex_code_mode::CellId::new("1".to_string()),
|
||||||
|
content_items: vec![CodeModeOutputContentItem::InputText {
|
||||||
|
text: "fallback".to_string(),
|
||||||
|
}],
|
||||||
|
error_text: None,
|
||||||
|
}
|
||||||
|
);
|
||||||
|
service.shutdown().await.expect("shutdown service");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -230,7 +230,7 @@ async fn run_code_mode_turn_with_builder(
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||||
async fn missing_process_host_returns_a_tool_error() -> Result<()> {
|
async fn missing_process_host_falls_back_to_in_process_code_mode() -> Result<()> {
|
||||||
skip_if_no_network!(Ok(()));
|
skip_if_no_network!(Ok(()));
|
||||||
|
|
||||||
let server = responses::start_mock_server().await;
|
let server = responses::start_mock_server().await;
|
||||||
@@ -244,16 +244,15 @@ async fn missing_process_host_returns_a_tool_error() -> Result<()> {
|
|||||||
.expect("code mode should be enabled");
|
.expect("code mode should be enabled");
|
||||||
});
|
});
|
||||||
let (_test, follow_up_mock) =
|
let (_test, follow_up_mock) =
|
||||||
run_code_mode_turn_with_builder(&server, "Run code mode", "text('unreachable')", builder)
|
run_code_mode_turn_with_builder(&server, "Run code mode", "text('fallback')", builder)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
let output = follow_up_mock
|
assert_eq!(
|
||||||
.single_request()
|
text_item(
|
||||||
.custom_tool_call_output("call-1");
|
&custom_tool_output_items(&follow_up_mock.single_request(), "call-1"),
|
||||||
assert!(
|
/*index*/ 1,
|
||||||
output["output"]
|
),
|
||||||
.as_str()
|
"fallback"
|
||||||
.is_some_and(|output| output.contains("failed to spawn code-mode host"))
|
|
||||||
);
|
);
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
Reference in new issue
Block a user