openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge cbec700603 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-03-17 14:27:42 -07:00
committed by GitHub
parent f3d18f7a4d cbec700603
commit 82794e512c
71 changed files with 3107 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/Cargo.toml
+1
View File
@@ -37,6 +37,7 @@ codex-config = { workspace = true }
codex-shell-command = { workspace = true }
codex-skills = { workspace = true }
codex-execpolicy = { workspace = true }
codex-fs-ops = { workspace = true }
codex-file-search = { workspace = true }
codex-git = { workspace = true }
codex-hooks = { workspace = true }
codex-rs/core/config.schema.json
+7 -1
View File
@@ -877,6 +877,12 @@
"description": "Whether this provider supports the Responses API WebSocket transport.",
"type": "boolean"
},
"websocket_connect_timeout_ms": {
"description": "Maximum time (in milliseconds) to wait for a websocket connection attempt before treating it as failed.",
"format": "uint64",
"minimum": 0.0,
"type": "integer"
},
"wire_api": {
"allOf": [
{
@@ -2473,4 +2479,4 @@
},
"title": "ConfigToml",
"type": "object"
}
}
codex-rs/core/src/agent/control.rs
+14
View File
@@ -3,6 +3,7 @@ use crate::agent::guards::Guards;
use crate::agent::role::DEFAULT_ROLE_NAME;
use crate::agent::role::resolve_role_config;
use crate::agent::status::is_final;
use crate::codex_thread::ThreadConfigSnapshot;
use crate::error::CodexErr;
use crate::error::Result as CodexResult;
use crate::find_thread_path_by_id_str;
@@ -360,6 +361,19 @@ impl AgentControl {
))
}
pub(crate) async fn get_agent_config_snapshot(
&self,
agent_id: ThreadId,
) -> Option<ThreadConfigSnapshot> {
let Ok(state) = self.upgrade() else {
return None;
};
let Ok(thread) = state.get_thread(agent_id).await else {
return None;
};
Some(thread.config_snapshot().await)
}
/// Subscribe to status updates for `agent_id`, yielding the latest value and changes.
pub(crate) async fn subscribe_status(
&self,
codex-rs/core/src/client.rs
+62 -38
View File
@@ -117,6 +117,9 @@ const RESPONSES_WEBSOCKETS_V2_BETA_HEADER_VALUE: &str = "responses_websockets=20
const RESPONSES_ENDPOINT: &str = "/responses";
const RESPONSES_COMPACT_ENDPOINT: &str = "/responses/compact";
const MEMORIES_SUMMARIZE_ENDPOINT: &str = "/memories/trace_summarize";
#[cfg(test)]
pub(crate) const WEBSOCKET_CONNECT_TIMEOUT: Duration =
Duration::from_millis(crate::model_provider_info::DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS);
pub fn ws_version_from_features(config: &Config) -> bool {
config
.features
@@ -310,6 +313,27 @@ impl ModelClient {
.unwrap_or_else(std::sync::PoisonError::into_inner) = websocket_session;
}
pub(crate) fn force_http_fallback(
&self,
session_telemetry: &SessionTelemetry,
model_info: &ModelInfo,
) -> bool {
let websocket_enabled = self.responses_websocket_enabled(model_info);
let activated =
websocket_enabled && !self.state.disable_websockets.swap(true, Ordering::Relaxed);
if activated {
warn!("falling back to HTTP");
session_telemetry.counter(
"codex.transport.fallback_to_http",
/*inc*/ 1,
&[("from_wire_api", "responses_websocket")],
);
}
self.store_cached_websocket_session(WebsocketSession::default());
activated
}
/// Compacts the current conversation history using the Compact endpoint.
///
/// This is a unary call (no streaming) that returns a new list of
@@ -538,15 +562,22 @@ impl ModelClient {
auth_context,
request_route_telemetry,
);
let websocket_connect_timeout = self.state.provider.websocket_connect_timeout();
let start = Instant::now();
let result = ApiWebSocketResponsesClient::new(api_provider, api_auth)
.connect(
let result = match tokio::time::timeout(
websocket_connect_timeout,
ApiWebSocketResponsesClient::new(api_provider, api_auth).connect(
headers,
crate::default_client::default_headers(),
turn_state,
Some(websocket_telemetry),
)
.await;
),
)
.await
{
Ok(result) => result,
Err(_) => Err(ApiError::Transport(TransportError::Timeout)),
};
let error_message = result.as_ref().err().map(telemetry_api_error_message);
let response_debug = result
.as_ref()
@@ -637,13 +668,12 @@ impl Drop for ModelClientSession {
}
impl ModelClientSession {
fn activate_http_fallback(&self, websocket_enabled: bool) -> bool {
websocket_enabled
&& !self
.client
.state
.disable_websockets
.swap(true, Ordering::Relaxed)
fn reset_websocket_session(&mut self) {
self.websocket_session.connection = None;
self.websocket_session.last_request = None;
self.websocket_session.last_response_rx = None;
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
}
fn build_responses_request(
@@ -896,7 +926,7 @@ impl ModelClientSession {
.turn_state
.clone()
.unwrap_or_else(|| Arc::clone(&self.turn_state));
let new_conn = self
let new_conn = match self
.client
.connect_websocket(
session_telemetry,
@@ -907,7 +937,16 @@ impl ModelClientSession {
auth_context,
request_route_telemetry,
)
.await?;
.await
{
Ok(new_conn) => new_conn,
Err(err) => {
if matches!(err, ApiError::Transport(TransportError::Timeout)) {
self.reset_websocket_session();
}
return Err(err);
}
};
self.websocket_session.connection = Some(new_conn);
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
@@ -1130,15 +1169,12 @@ impl ModelClientSession {
let ws_request = self.prepare_websocket_request(ws_payload, &request);
self.websocket_session.last_request = Some(request);
let stream_result = self
.websocket_session
.connection
.as_ref()
.ok_or_else(|| {
map_api_error(ApiError::Stream(
"websocket connection is unavailable".to_string(),
))
})?
let stream_result = self.websocket_session.connection.as_ref().ok_or_else(|| {
map_api_error(ApiError::Stream(
"websocket connection is unavailable".to_string(),
))
})?;
let stream_result = stream_result
.stream_request(ws_request, self.websocket_session.connection_reused())
.await
.map_err(map_api_error)?;
@@ -1296,22 +1332,10 @@ impl ModelClientSession {
session_telemetry: &SessionTelemetry,
model_info: &ModelInfo,
) -> bool {
let websocket_enabled = self.client.responses_websocket_enabled(model_info);
let activated = self.activate_http_fallback(websocket_enabled);
if activated {
warn!("falling back to HTTP");
session_telemetry.counter(
"codex.transport.fallback_to_http",
/*inc*/ 1,
&[("from_wire_api", "responses_websocket")],
);
self.websocket_session.connection = None;
self.websocket_session.last_request = None;
self.websocket_session.last_response_rx = None;
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
}
let activated = self
.client
.force_http_fallback(session_telemetry, model_info);
self.websocket_session = WebsocketSession::default();
activated
}
}
codex-rs/core/src/codex.rs
+16 -74
View File
@@ -105,7 +105,6 @@ use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::TurnContextNetworkItem;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::request_permissions::PermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionProfile;
@@ -267,6 +266,7 @@ use crate::rollout::RolloutRecorderParams;
use crate::rollout::map_session_init_error;
use crate::rollout::metadata;
use crate::rollout::policy::EventPersistenceMode;
use crate::session_startup_prewarm::SessionStartupPrewarmHandle;
use crate::shell;
use crate::shell_snapshot::ShellSnapshot;
use crate::skills::SkillError;
@@ -286,7 +286,6 @@ use crate::state::SessionServices;
use crate::state::SessionState;
use crate::state_db;
use crate::tasks::GhostSnapshotTask;
use crate::tasks::RegularTask;
use crate::tasks::ReviewTask;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
@@ -2411,70 +2410,17 @@ impl Session {
.await
}
pub(crate) async fn take_startup_regular_task(&self) -> Option<RegularTask> {
let startup_regular_task = {
let mut state = self.state.lock().await;
state.take_startup_regular_task()
};
let startup_regular_task = startup_regular_task?;
match startup_regular_task.await {
Ok(Ok(regular_task)) => Some(regular_task),
Ok(Err(err)) => {
warn!("startup websocket prewarm setup failed: {err:#}");
None
}
Err(err) => {
warn!("startup websocket prewarm setup join failed: {err}");
None
}
}
}
async fn schedule_startup_prewarm(self: &Arc<Self>, base_instructions: String) {
let sess = Arc::clone(self);
let startup_regular_task: JoinHandle<CodexResult<RegularTask>> =
tokio::spawn(
async move { sess.schedule_startup_prewarm_inner(base_instructions).await },
);
pub(crate) async fn set_session_startup_prewarm(
&self,
startup_prewarm: SessionStartupPrewarmHandle,
) {
let mut state = self.state.lock().await;
state.set_startup_regular_task(startup_regular_task);
state.set_session_startup_prewarm(startup_prewarm);
}
async fn schedule_startup_prewarm_inner(
self: &Arc<Self>,
base_instructions: String,
) -> CodexResult<RegularTask> {
let startup_turn_context = self
.new_default_turn_with_sub_id(INITIAL_SUBMIT_ID.to_owned())
.await;
let startup_cancellation_token = CancellationToken::new();
let startup_router = built_tools(
self,
startup_turn_context.as_ref(),
&[],
&HashSet::new(),
/*skills_outcome*/ None,
&startup_cancellation_token,
)
.await?;
let startup_prompt = build_prompt(
Vec::new(),
startup_router.as_ref(),
startup_turn_context.as_ref(),
BaseInstructions {
text: base_instructions,
},
);
let startup_turn_metadata_header = startup_turn_context
.turn_metadata_state
.current_header_value();
RegularTask::with_startup_prewarm(
self.services.model_client.clone(),
startup_prompt,
startup_turn_context,
startup_turn_metadata_header,
)
.await
pub(crate) async fn take_session_startup_prewarm(&self) -> Option<SessionStartupPrewarmHandle> {
let mut state = self.state.lock().await;
state.take_session_startup_prewarm()
}
pub(crate) async fn get_config(&self) -> std::sync::Arc<Config> {
@@ -4553,9 +4499,12 @@ mod handlers {
{
sess.refresh_mcp_servers_if_requested(&current_context)
.await;
let regular_task = sess.take_startup_regular_task().await.unwrap_or_default();
sess.spawn_task(Arc::clone(&current_context), items, regular_task)
.await;
sess.spawn_task(
Arc::clone(&current_context),
items,
crate::tasks::RegularTask::new(),
)
.await;
}
}
@@ -5485,13 +5434,6 @@ pub(crate) async fn run_turn(
let model_info = turn_context.model_info.clone();
let auto_compact_limit = model_info.auto_compact_token_limit().unwrap_or(i64::MAX);
let event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: turn_context.sub_id.clone(),
model_context_window: turn_context.model_context_window(),
collaboration_mode_kind: turn_context.collaboration_mode.mode,
});
sess.send_event(&turn_context, event).await;
// TODO(ccunningham): Pre-turn compaction runs before context updates and the
// new user message are recorded. Estimate pending incoming items (context
// diffs/full reinjection + user input) and trigger compaction preemptively
@@ -6236,7 +6178,7 @@ fn codex_apps_connector_id(tool: &crate::mcp_connection_manager::ToolInfo) -> Op
tool.connector_id.as_deref()
}
fn build_prompt(
pub(crate) fn build_prompt(
input: Vec<ResponseItem>,
router: &ToolRouter,
turn_context: &TurnContext,
codex-rs/core/src/codex_tests.rs
+103
View File
@@ -39,6 +39,7 @@ use crate::protocol::TokenCountEvent;
use crate::protocol::TokenUsage;
use crate::protocol::TokenUsageInfo;
use crate::protocol::TurnCompleteEvent;
use crate::protocol::TurnStartedEvent;
use crate::protocol::UserMessageEvent;
use crate::rollout::policy::EventPersistenceMode;
use crate::rollout::recorder::RolloutRecorder;
@@ -142,6 +143,108 @@ fn skill_message(text: &str) -> ResponseItem {
}
}
#[tokio::test]
async fn regular_turn_emits_turn_started_without_waiting_for_startup_prewarm() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let (_tx, startup_prewarm_rx) = tokio::sync::oneshot::channel::<()>();
let handle = tokio::spawn(async move {
let _ = startup_prewarm_rx.await;
Ok(test_model_client_session())
});
sess.set_session_startup_prewarm(
crate::session_startup_prewarm::SessionStartupPrewarmHandle::new(
handle,
std::time::Instant::now(),
crate::client::WEBSOCKET_CONNECT_TIMEOUT,
),
)
.await;
sess.spawn_task(
Arc::clone(&tc),
Vec::new(),
crate::tasks::RegularTask::new(),
)
.await;
let first = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
.await
.expect("expected turn started event without waiting for startup prewarm")
.expect("channel open");
assert!(matches!(
first.msg,
EventMsg::TurnStarted(TurnStartedEvent { turn_id, .. }) if turn_id == tc.sub_id
));
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
}
#[tokio::test]
async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let (_tx, startup_prewarm_rx) = tokio::sync::oneshot::channel::<()>();
let handle = tokio::spawn(async move {
let _ = startup_prewarm_rx.await;
Ok(test_model_client_session())
});
sess.set_session_startup_prewarm(
crate::session_startup_prewarm::SessionStartupPrewarmHandle::new(
handle,
std::time::Instant::now(),
crate::client::WEBSOCKET_CONNECT_TIMEOUT,
),
)
.await;
sess.spawn_task(
Arc::clone(&tc),
Vec::new(),
crate::tasks::RegularTask::new(),
)
.await;
let first = tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv())
.await
.expect("expected turn started event without waiting for startup prewarm")
.expect("channel open");
assert!(matches!(
first.msg,
EventMsg::TurnStarted(TurnStartedEvent { turn_id, .. }) if turn_id == tc.sub_id
));
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("expected turn aborted event")
.expect("channel open");
assert!(matches!(
second.msg,
EventMsg::TurnAborted(crate::protocol::TurnAbortedEvent {
turn_id: Some(turn_id),
reason: TurnAbortReason::Interrupted,
}) if turn_id == tc.sub_id
));
}
fn test_model_client_session() -> crate::client::ModelClientSession {
crate::client::ModelClient::new(
None,
ThreadId::try_from("00000000-0000-4000-8000-000000000001")
.expect("test thread id should be valid"),
crate::model_provider_info::ModelProviderInfo::create_openai_provider(
/* base_url */ None,
),
codex_protocol::protocol::SessionSource::Exec,
None,
false,
false,
false,
None,
)
.new_session()
}
fn developer_input_texts(items: &[ResponseItem]) -> Vec<&str> {
items
.iter()
codex-rs/core/src/config/config_tests.rs
+2
View File
@@ -4076,6 +4076,7 @@ wire_api = "responses"
request_max_retries = 4 # retry failed HTTP requests
stream_max_retries = 10 # retry dropped SSE streams
stream_idle_timeout_ms = 300000 # 5m idle timeout
websocket_connect_timeout_ms = 15000
[profiles.o3]
model = "o3"
@@ -4130,6 +4131,7 @@ model_verbosity = "high"
request_max_retries: Some(4),
stream_max_retries: Some(10),
stream_idle_timeout_ms: Some(300_000),
websocket_connect_timeout_ms: Some(15_000),
requires_openai_auth: false,
supports_websockets: false,
};
codex-rs/core/src/config/schema_tests.rs
+8 -1
View File
@@ -6,6 +6,10 @@ use pretty_assertions::assert_eq;
use similar::TextDiff;
use tempfile::TempDir;
fn trim_single_trailing_newline(contents: &str) -> &str {
contents.strip_suffix('\n').unwrap_or(contents)
}
#[test]
fn config_schema_matches_fixture() {
let fixture_path = codex_utils_cargo_bin::find_resource!("config.schema.json")
@@ -40,9 +44,12 @@ Run `just write-config-schema` to overwrite with your changes.\n\n{diff}"
std::fs::read_to_string(&tmp_path).expect("read back config schema from temp path");
#[cfg(windows)]
let fixture = fixture.replace("\r\n", "\n");
#[cfg(windows)]
let tmp_contents = tmp_contents.replace("\r\n", "\n");
assert_eq!(
fixture, tmp_contents,
trim_single_trailing_newline(&fixture),
trim_single_trailing_newline(&tmp_contents),
"fixture should match exactly with generated schema"
);
}
codex-rs/core/src/lib.rs
+2
View File
@@ -69,6 +69,7 @@ pub mod plugins;
mod sandbox_tags;
pub mod sandboxing;
mod session_prefix;
mod session_startup_prewarm;
mod shell_detect;
mod stream_events_utils;
pub mod test_support;
@@ -111,6 +112,7 @@ pub mod default_client;
pub mod project_doc;
mod rollout;
pub(crate) mod safety;
mod sandboxed_fs;
pub mod seatbelt;
pub mod shell;
pub mod shell_snapshot;
codex-rs/core/src/model_provider_info.rs
+14
View File
@@ -22,6 +22,7 @@ use std::time::Duration;
const DEFAULT_STREAM_IDLE_TIMEOUT_MS: u64 = 300_000;
const DEFAULT_STREAM_MAX_RETRIES: u64 = 5;
const DEFAULT_REQUEST_MAX_RETRIES: u64 = 4;
pub(crate) const DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS: u64 = 15_000;
/// Hard cap for user-configured `stream_max_retries`.
const MAX_STREAM_MAX_RETRIES: u64 = 100;
/// Hard cap for user-configured `request_max_retries`.
@@ -112,6 +113,10 @@ pub struct ModelProviderInfo {
/// the connection as lost.
pub stream_idle_timeout_ms: Option<u64>,
/// Maximum time (in milliseconds) to wait for a websocket connection attempt before treating
/// it as failed.
pub websocket_connect_timeout_ms: Option<u64>,
/// Does this provider require an OpenAI API Key or ChatGPT login token? If true,
/// user is presented with login screen on first run, and login preference and token/key
/// are stored in auth.json. If false (which is the default), login screen is skipped,
@@ -227,6 +232,13 @@ impl ModelProviderInfo {
.unwrap_or(Duration::from_millis(DEFAULT_STREAM_IDLE_TIMEOUT_MS))
}
/// Effective timeout for websocket connect attempts.
pub fn websocket_connect_timeout(&self) -> Duration {
self.websocket_connect_timeout_ms
.map(Duration::from_millis)
.unwrap_or(Duration::from_millis(DEFAULT_WEBSOCKET_CONNECT_TIMEOUT_MS))
}
pub fn create_openai_provider(base_url: Option<String>) -> ModelProviderInfo {
ModelProviderInfo {
name: OPENAI_PROVIDER_NAME.into(),
@@ -256,6 +268,7 @@ impl ModelProviderInfo {
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: true,
supports_websockets: true,
}
@@ -332,6 +345,7 @@ pub fn create_oss_provider_with_base_url(base_url: &str, wire_api: WireApi) -> M
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
}
codex-rs/core/src/model_provider_info_tests.rs
+16
View File
@@ -20,6 +20,7 @@ base_url = "http://localhost:11434/v1"
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -51,6 +52,7 @@ query_params = { api-version = "2025-04-01-preview" }
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -85,6 +87,7 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" }
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -105,3 +108,16 @@ wire_api = "chat"
let err = toml::from_str::<ModelProviderInfo>(provider_toml).unwrap_err();
assert!(err.to_string().contains(CHAT_WIRE_API_REMOVED_ERROR));
}
#[test]
fn test_deserialize_websocket_connect_timeout() {
let provider_toml = r#"
name = "OpenAI"
base_url = "https://api.openai.com/v1"
websocket_connect_timeout_ms = 15000
supports_websockets = true
"#;
let provider: ModelProviderInfo = toml::from_str(provider_toml).unwrap();
assert_eq!(provider.websocket_connect_timeout_ms, Some(15_000));
}
codex-rs/core/src/models_manager/manager_tests.rs
+1
View File
@@ -71,6 +71,7 @@ fn provider_for(base_url: String) -> ModelProviderInfo {
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
}
codex-rs/core/src/realtime_context_tests.rs
+2
View File
@@ -26,6 +26,8 @@ fn thread_metadata(cwd: &str, title: &str, first_user_message: &str) -> ThreadMe
agent_nickname: None,
agent_role: None,
model_provider: "test-provider".to_string(),
model: Some("gpt-5".to_string()),
reasoning_effort: None,
cwd: PathBuf::from(cwd),
cli_version: "test".to_string(),
title: title.to_string(),
codex-rs/core/src/sandboxed_fs.rs
+234
View File
@@ -0,0 +1,234 @@
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::exec::ExecExpiration;
use crate::sandboxing::CommandSpec;
use crate::sandboxing::SandboxPermissions;
use crate::sandboxing::execute_env;
use crate::sandboxing::merge_permission_profiles;
use crate::tools::sandboxing::SandboxAttempt;
use crate::tools::sandboxing::SandboxablePreference;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_fs_ops::CODEX_CORE_FS_OPS_ARG1;
use codex_fs_ops::FsError;
use codex_fs_ops::FsErrorKind;
use codex_fs_ops::FsPayload;
use codex_fs_ops::FsResponse;
use codex_protocol::models::PermissionProfile;
use std::collections::HashMap;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
const SANDBOXED_FS_TIMEOUT: Duration = Duration::from_secs(30);
#[derive(Debug, thiserror::Error)]
pub(crate) enum SandboxedFsError {
#[error("failed to determine codex executable: {message}")]
ResolveExe { message: String },
#[error("sandboxed fs helper timed out while accessing `{path}`")]
TimedOut { path: PathBuf },
#[error("sandboxed fs helper exited with code {exit_code} while accessing `{path}`: {message}")]
ProcessFailed {
path: PathBuf,
exit_code: i32,
message: String,
},
#[error("sandboxed fs helper returned invalid output for `{path}`: {message}")]
InvalidResponse { path: PathBuf, message: String },
#[error("sandboxed fs helper could not access `{path}`: {error}")]
Operation { path: PathBuf, error: FsError },
}
impl SandboxedFsError {
pub(crate) fn operation_error_kind(&self) -> Option<&FsErrorKind> {
match self {
Self::Operation { error, .. } => Some(&error.kind),
_ => None,
}
}
pub(crate) fn operation_error_message(&self) -> Option<&str> {
match self {
Self::Operation { error, .. } => Some(error.message.as_str()),
_ => None,
}
}
#[allow(dead_code)]
pub(crate) fn to_io_error(&self) -> std::io::Error {
match self {
Self::Operation { error, .. } => error.to_io_error(),
Self::TimedOut { .. } => {
std::io::Error::new(std::io::ErrorKind::TimedOut, self.to_string())
}
Self::ResolveExe { .. } | Self::ProcessFailed { .. } | Self::InvalidResponse { .. } => {
std::io::Error::other(self.to_string())
}
}
}
}
pub(crate) async fn read_bytes(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
path: &Path,
) -> Result<Vec<u8>, SandboxedFsError> {
let path_buf = path.to_path_buf();
let payload = run_request(session, turn, "read_bytes", &path_buf).await?;
let FsPayload::Bytes { base64 } = payload else {
return Err(SandboxedFsError::InvalidResponse {
path: path_buf,
message: "expected bytes payload".to_string(),
});
};
BASE64_STANDARD
.decode(base64)
.map_err(|error| SandboxedFsError::InvalidResponse {
path: path.to_path_buf(),
message: format!("failed to decode base64 payload: {error}"),
})
}
#[allow(dead_code)]
pub(crate) async fn read_text(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
path: &Path,
) -> Result<String, SandboxedFsError> {
let path_buf = path.to_path_buf();
let payload = run_request(session, turn, "read_text", &path_buf).await?;
let FsPayload::Text { text } = payload else {
return Err(SandboxedFsError::InvalidResponse {
path: path_buf,
message: "expected text payload".to_string(),
});
};
Ok(text)
}
async fn run_request(
session: &Arc<Session>,
turn: &Arc<TurnContext>,
operation: &str,
path: &Path,
) -> Result<FsPayload, SandboxedFsError> {
let exe = resolve_codex_exe(&turn.config.codex_home)?;
let additional_permissions = effective_granted_permissions(session).await;
let sandbox_manager = crate::sandboxing::SandboxManager::new();
let attempt = SandboxAttempt {
sandbox: sandbox_manager.select_initial(
&turn.file_system_sandbox_policy,
turn.network_sandbox_policy,
SandboxablePreference::Auto,
turn.windows_sandbox_level,
/*has_managed_network_requirements*/ false,
),
policy: &turn.sandbox_policy,
file_system_policy: &turn.file_system_sandbox_policy,
network_policy: turn.network_sandbox_policy,
enforce_managed_network: false,
manager: &sandbox_manager,
sandbox_cwd: &turn.cwd,
codex_linux_sandbox_exe: turn.codex_linux_sandbox_exe.as_ref(),
use_legacy_landlock: turn.features.use_legacy_landlock(),
windows_sandbox_level: turn.windows_sandbox_level,
windows_sandbox_private_desktop: turn.config.permissions.windows_sandbox_private_desktop,
};
let exec_request = attempt
.env_for(
CommandSpec {
program: exe.to_string_lossy().to_string(),
args: vec![
CODEX_CORE_FS_OPS_ARG1.to_string(),
operation.to_string(),
path.to_string_lossy().to_string(),
],
cwd: turn.cwd.clone(),
env: HashMap::new(),
expiration: ExecExpiration::Timeout(SANDBOXED_FS_TIMEOUT),
sandbox_permissions: SandboxPermissions::UseDefault,
additional_permissions,
justification: None,
},
None,
)
.map_err(|error| SandboxedFsError::ProcessFailed {
path: path.to_path_buf(),
exit_code: -1,
message: error.to_string(),
})?;
let output =
execute_env(exec_request, None)
.await
.map_err(|error| SandboxedFsError::ProcessFailed {
path: path.to_path_buf(),
exit_code: -1,
message: error.to_string(),
})?;
if output.timed_out {
return Err(SandboxedFsError::TimedOut {
path: path.to_path_buf(),
});
}
if output.exit_code != 0 {
let stderr = output.stderr.text.trim();
let stdout = output.stdout.text.trim();
let message = if !stderr.is_empty() {
stderr.to_string()
} else if !stdout.is_empty() {
stdout.to_string()
} else {
"no error details emitted".to_string()
};
return Err(SandboxedFsError::ProcessFailed {
path: path.to_path_buf(),
exit_code: output.exit_code,
message,
});
}
let response: FsResponse =
serde_json::from_str(output.stdout.text.trim()).map_err(|error| {
SandboxedFsError::InvalidResponse {
path: path.to_path_buf(),
message: format!("failed to parse helper response: {error}"),
}
})?;
match response {
FsResponse::Success { payload } => Ok(payload),
FsResponse::Error { error } => Err(SandboxedFsError::Operation {
path: path.to_path_buf(),
error,
}),
}
}
async fn effective_granted_permissions(session: &Session) -> Option<PermissionProfile> {
let granted_session_permissions = session.granted_session_permissions().await;
let granted_turn_permissions = session.granted_turn_permissions().await;
merge_permission_profiles(
granted_session_permissions.as_ref(),
granted_turn_permissions.as_ref(),
)
}
fn resolve_codex_exe(codex_home: &Path) -> Result<PathBuf, SandboxedFsError> {
#[cfg(target_os = "windows")]
{
Ok(codex_windows_sandbox::resolve_current_exe_for_launch(
codex_home,
"codex.exe",
))
}
#[cfg(not(target_os = "windows"))]
{
let _ = codex_home;
std::env::current_exe().map_err(|error| SandboxedFsError::ResolveExe {
message: error.to_string(),
})
}
}
codex-rs/core/src/session_startup_prewarm.rs
+241
View File
@@ -0,0 +1,241 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::info;
use tracing::warn;
use crate::client::ModelClientSession;
use crate::codex::INITIAL_SUBMIT_ID;
use crate::codex::Session;
use crate::codex::build_prompt;
use crate::codex::built_tools;
use crate::error::Result as CodexResult;
use codex_otel::SessionTelemetry;
use codex_otel::metrics::names::STARTUP_PREWARM_AGE_AT_FIRST_TURN_METRIC;
use codex_otel::metrics::names::STARTUP_PREWARM_DURATION_METRIC;
use codex_protocol::models::BaseInstructions;
pub(crate) struct SessionStartupPrewarmHandle {
task: JoinHandle<CodexResult<ModelClientSession>>,
started_at: Instant,
timeout: Duration,
}
pub(crate) enum SessionStartupPrewarmResolution {
Cancelled,
Ready(Box<ModelClientSession>),
Unavailable {
status: &'static str,
prewarm_duration: Option<Duration>,
},
}
impl SessionStartupPrewarmHandle {
pub(crate) fn new(
task: JoinHandle<CodexResult<ModelClientSession>>,
started_at: Instant,
timeout: Duration,
) -> Self {
Self {
task,
started_at,
timeout,
}
}
async fn resolve(
self,
session_telemetry: &SessionTelemetry,
cancellation_token: &CancellationToken,
) -> SessionStartupPrewarmResolution {
let Self {
mut task,
started_at,
timeout,
} = self;
let age_at_first_turn = started_at.elapsed();
let remaining = timeout.saturating_sub(age_at_first_turn);
let resolution = if task.is_finished() {
Self::resolution_from_join_result(task.await, started_at)
} else {
match tokio::select! {
_ = cancellation_token.cancelled() => None,
result = tokio::time::timeout(remaining, &mut task) => Some(result),
} {
Some(Ok(result)) => Self::resolution_from_join_result(result, started_at),
Some(Err(_elapsed)) => {
task.abort();
info!("startup websocket prewarm timed out before the first turn could use it");
SessionStartupPrewarmResolution::Unavailable {
status: "timed_out",
prewarm_duration: Some(started_at.elapsed()),
}
}
None => {
task.abort();
session_telemetry.record_duration(
STARTUP_PREWARM_AGE_AT_FIRST_TURN_METRIC,
age_at_first_turn,
&[("status", "cancelled")],
);
session_telemetry.record_duration(
STARTUP_PREWARM_DURATION_METRIC,
started_at.elapsed(),
&[("status", "cancelled")],
);
return SessionStartupPrewarmResolution::Cancelled;
}
}
};
match resolution {
SessionStartupPrewarmResolution::Cancelled => {
SessionStartupPrewarmResolution::Cancelled
}
SessionStartupPrewarmResolution::Ready(prewarmed_session) => {
session_telemetry.record_duration(
STARTUP_PREWARM_AGE_AT_FIRST_TURN_METRIC,
age_at_first_turn,
&[("status", "consumed")],
);
SessionStartupPrewarmResolution::Ready(prewarmed_session)
}
SessionStartupPrewarmResolution::Unavailable {
status,
prewarm_duration,
} => {
session_telemetry.record_duration(
STARTUP_PREWARM_AGE_AT_FIRST_TURN_METRIC,
age_at_first_turn,
&[("status", status)],
);
if let Some(prewarm_duration) = prewarm_duration {
session_telemetry.record_duration(
STARTUP_PREWARM_DURATION_METRIC,
prewarm_duration,
&[("status", status)],
);
}
SessionStartupPrewarmResolution::Unavailable {
status,
prewarm_duration,
}
}
}
}
fn resolution_from_join_result(
result: std::result::Result<CodexResult<ModelClientSession>, tokio::task::JoinError>,
started_at: Instant,
) -> SessionStartupPrewarmResolution {
match result {
Ok(Ok(prewarmed_session)) => {
SessionStartupPrewarmResolution::Ready(Box::new(prewarmed_session))
}
Ok(Err(err)) => {
warn!("startup websocket prewarm setup failed: {err:#}");
SessionStartupPrewarmResolution::Unavailable {
status: "failed",
prewarm_duration: None,
}
}
Err(err) => {
warn!("startup websocket prewarm setup join failed: {err}");
SessionStartupPrewarmResolution::Unavailable {
status: "join_failed",
prewarm_duration: Some(started_at.elapsed()),
}
}
}
}
}
impl Session {
pub(crate) async fn schedule_startup_prewarm(self: &Arc<Self>, base_instructions: String) {
let session_telemetry = self.services.session_telemetry.clone();
let websocket_connect_timeout = self.provider().await.websocket_connect_timeout();
let started_at = Instant::now();
let startup_prewarm_session = Arc::clone(self);
let startup_prewarm = tokio::spawn(async move {
let result =
schedule_startup_prewarm_inner(startup_prewarm_session, base_instructions).await;
let status = if result.is_ok() { "ready" } else { "failed" };
session_telemetry.record_duration(
STARTUP_PREWARM_DURATION_METRIC,
started_at.elapsed(),
&[("status", status)],
);
result
});
self.set_session_startup_prewarm(SessionStartupPrewarmHandle::new(
startup_prewarm,
started_at,
websocket_connect_timeout,
))
.await;
}
pub(crate) async fn consume_startup_prewarm_for_regular_turn(
&self,
cancellation_token: &CancellationToken,
) -> SessionStartupPrewarmResolution {
let Some(startup_prewarm) = self.take_session_startup_prewarm().await else {
return SessionStartupPrewarmResolution::Unavailable {
status: "not_scheduled",
prewarm_duration: None,
};
};
startup_prewarm
.resolve(&self.services.session_telemetry, cancellation_token)
.await
}
}
async fn schedule_startup_prewarm_inner(
session: Arc<Session>,
base_instructions: String,
) -> CodexResult<ModelClientSession> {
let startup_turn_context = session
.new_default_turn_with_sub_id(INITIAL_SUBMIT_ID.to_owned())
.await;
let startup_cancellation_token = CancellationToken::new();
let startup_router = built_tools(
session.as_ref(),
startup_turn_context.as_ref(),
&[],
&HashSet::new(),
/*skills_outcome*/ None,
&startup_cancellation_token,
)
.await?;
let startup_prompt = build_prompt(
Vec::new(),
startup_router.as_ref(),
startup_turn_context.as_ref(),
BaseInstructions {
text: base_instructions,
},
);
let startup_turn_metadata_header = startup_turn_context
.turn_metadata_state
.current_header_value();
let mut client_session = session.services.model_client.new_session();
client_session
.prewarm_websocket(
&startup_prompt,
&startup_turn_context.model_info,
&startup_turn_context.session_telemetry,
startup_turn_context.reasoning_effort,
startup_turn_context.reasoning_summary,
startup_turn_context.config.service_tier,
startup_turn_metadata_header.as_deref(),
)
.await?;
Ok(client_session)
}
codex-rs/core/src/shell_snapshot.rs
+18 -9
View File
@@ -120,13 +120,13 @@ impl ShellSnapshot {
ShellType::PowerShell => "ps1",
_ => "sh",
};
let path = codex_home
.join(SNAPSHOT_DIR)
.join(format!("{session_id}.{extension}"));
let nonce = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map(|duration| duration.as_nanos())
.unwrap_or(0);
let path = codex_home
.join(SNAPSHOT_DIR)
.join(format!("{session_id}.{nonce}.{extension}"));
let temp_path = codex_home
.join(SNAPSHOT_DIR)
.join(format!("{session_id}.tmp-{nonce}"));
@@ -510,12 +510,9 @@ pub async fn cleanup_stale_snapshots(codex_home: &Path, active_session_id: Threa
let file_name = entry.file_name();
let file_name = file_name.to_string_lossy();
let (session_id, _) = match file_name.rsplit_once('.') {
Some((stem, ext)) => (stem, ext),
None => {
remove_snapshot_file(&path).await;
continue;
}
let Some(session_id) = snapshot_session_id_from_file_name(&file_name) else {
remove_snapshot_file(&path).await;
continue;
};
if session_id == active_session_id {
continue;
@@ -556,6 +553,18 @@ async fn remove_snapshot_file(path: &Path) {
}
}
fn snapshot_session_id_from_file_name(file_name: &str) -> Option<&str> {
let (stem, extension) = file_name.rsplit_once('.')?;
match extension {
"sh" | "ps1" => Some(
stem.split_once('.')
.map_or(stem, |(session_id, _generation)| session_id),
),
_ if extension.starts_with("tmp-") => Some(stem),
_ => None,
}
}
#[cfg(test)]
#[path = "shell_snapshot_tests.rs"]
mod tests;
codex-rs/core/src/shell_snapshot_tests.rs
+62 -4
View File
@@ -98,6 +98,28 @@ fn strip_snapshot_preamble_requires_marker() {
assert!(result.is_err());
}
#[test]
fn snapshot_file_name_parser_supports_legacy_and_suffixed_names() {
let session_id = "019cf82b-6a62-7700-bbbd-46909794ef89";
assert_eq!(
snapshot_session_id_from_file_name(&format!("{session_id}.sh")),
Some(session_id)
);
assert_eq!(
snapshot_session_id_from_file_name(&format!("{session_id}.123.sh")),
Some(session_id)
);
assert_eq!(
snapshot_session_id_from_file_name(&format!("{session_id}.tmp-123")),
Some(session_id)
);
assert_eq!(
snapshot_session_id_from_file_name("not-a-snapshot.txt"),
None
);
}
#[cfg(unix)]
#[test]
fn bash_snapshot_filters_invalid_exports() -> Result<()> {
@@ -186,6 +208,42 @@ async fn try_new_creates_and_deletes_snapshot_file() -> Result<()> {
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn try_new_uses_distinct_generation_paths() -> Result<()> {
let dir = tempdir()?;
let session_id = ThreadId::new();
let shell = Shell {
shell_type: ShellType::Bash,
shell_path: PathBuf::from("/bin/bash"),
shell_snapshot: crate::shell::empty_shell_snapshot_receiver(),
};
let initial_snapshot = ShellSnapshot::try_new(dir.path(), session_id, dir.path(), &shell)
.await
.expect("initial snapshot should be created");
let refreshed_snapshot = ShellSnapshot::try_new(dir.path(), session_id, dir.path(), &shell)
.await
.expect("refreshed snapshot should be created");
let initial_path = initial_snapshot.path.clone();
let refreshed_path = refreshed_snapshot.path.clone();
assert_ne!(initial_path, refreshed_path);
assert_eq!(initial_path.exists(), true);
assert_eq!(refreshed_path.exists(), true);
drop(initial_snapshot);
assert_eq!(initial_path.exists(), false);
assert_eq!(refreshed_path.exists(), true);
drop(refreshed_snapshot);
assert_eq!(refreshed_path.exists(), false);
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn snapshot_shell_does_not_inherit_stdin() -> Result<()> {
@@ -339,8 +397,8 @@ async fn cleanup_stale_snapshots_removes_orphans_and_keeps_live() -> Result<()>
let live_session = ThreadId::new();
let orphan_session = ThreadId::new();
let live_snapshot = snapshot_dir.join(format!("{live_session}.sh"));
let orphan_snapshot = snapshot_dir.join(format!("{orphan_session}.sh"));
let live_snapshot = snapshot_dir.join(format!("{live_session}.123.sh"));
let orphan_snapshot = snapshot_dir.join(format!("{orphan_session}.456.sh"));
let invalid_snapshot = snapshot_dir.join("not-a-snapshot.txt");
write_rollout_stub(codex_home, live_session).await?;
@@ -365,7 +423,7 @@ async fn cleanup_stale_snapshots_removes_stale_rollouts() -> Result<()> {
fs::create_dir_all(&snapshot_dir).await?;
let stale_session = ThreadId::new();
let stale_snapshot = snapshot_dir.join(format!("{stale_session}.sh"));
let stale_snapshot = snapshot_dir.join(format!("{stale_session}.123.sh"));
let rollout_path = write_rollout_stub(codex_home, stale_session).await?;
fs::write(&stale_snapshot, "stale").await?;
@@ -386,7 +444,7 @@ async fn cleanup_stale_snapshots_skips_active_session() -> Result<()> {
fs::create_dir_all(&snapshot_dir).await?;
let active_session = ThreadId::new();
let active_snapshot = snapshot_dir.join(format!("{active_session}.sh"));
let active_snapshot = snapshot_dir.join(format!("{active_session}.123.sh"));
let rollout_path = write_rollout_stub(codex_home, active_session).await?;
fs::write(&active_snapshot, "active").await?;
codex-rs/core/src/state/session.rs
+11 -12
View File
@@ -4,17 +4,15 @@ use codex_protocol::models::PermissionProfile;
use codex_protocol::models::ResponseItem;
use std::collections::HashMap;
use std::collections::HashSet;
use tokio::task::JoinHandle;
use crate::codex::PreviousTurnSettings;
use crate::codex::SessionConfiguration;
use crate::context_manager::ContextManager;
use crate::error::Result as CodexResult;
use crate::protocol::RateLimitSnapshot;
use crate::protocol::TokenUsage;
use crate::protocol::TokenUsageInfo;
use crate::sandboxing::merge_permission_profiles;
use crate::tasks::RegularTask;
use crate::session_startup_prewarm::SessionStartupPrewarmHandle;
use crate::truncate::TruncationPolicy;
use codex_protocol::protocol::TurnContextItem;
@@ -30,8 +28,8 @@ pub(crate) struct SessionState {
/// model/realtime handling on subsequent regular turns (including full-context
/// reinjection after resume or `/compact`).
previous_turn_settings: Option<PreviousTurnSettings>,
/// Startup regular task pre-created during session initialization.
pub(crate) startup_regular_task: Option<JoinHandle<CodexResult<RegularTask>>>,
/// Startup prewarmed session prepared during session initialization.
pub(crate) startup_prewarm: Option<SessionStartupPrewarmHandle>,
pub(crate) active_connector_selection: HashSet<String>,
pub(crate) pending_session_start_source: Option<codex_hooks::SessionStartSource>,
granted_permissions: Option<PermissionProfile>,
@@ -49,7 +47,7 @@ impl SessionState {
dependency_env: HashMap::new(),
mcp_dependency_prompted: HashSet::new(),
previous_turn_settings: None,
startup_regular_task: None,
startup_prewarm: None,
active_connector_selection: HashSet::new(),
pending_session_start_source: None,
granted_permissions: None,
@@ -165,14 +163,15 @@ impl SessionState {
self.dependency_env.clone()
}
pub(crate) fn set_startup_regular_task(&mut self, task: JoinHandle<CodexResult<RegularTask>>) {
self.startup_regular_task = Some(task);
pub(crate) fn set_session_startup_prewarm(
&mut self,
startup_prewarm: SessionStartupPrewarmHandle,
) {
self.startup_prewarm = Some(startup_prewarm);
}
pub(crate) fn take_startup_regular_task(
&mut self,
) -> Option<JoinHandle<CodexResult<RegularTask>>> {
self.startup_regular_task.take()
pub(crate) fn take_session_startup_prewarm(&mut self) -> Option<SessionStartupPrewarmHandle> {
self.startup_prewarm.take()
}
// Adds connector IDs to the active set and returns the merged selection.
codex-rs/core/src/tasks/mod.rs
-1
View File
@@ -384,7 +384,6 @@ impl Session {
turn.add_task(task);
*active = Some(turn);
}
async fn take_active_turn(&self) -> Option<ActiveTurn> {
let mut active = self.active_turn.lock().await;
active.take()
codex-rs/core/src/tasks/regular.rs
+28 -48
View File
@@ -1,64 +1,27 @@
use std::sync::Arc;
use std::sync::Mutex;
use crate::client::ModelClient;
use crate::client::ModelClientSession;
use crate::client_common::Prompt;
use async_trait::async_trait;
use tokio_util::sync::CancellationToken;
use crate::codex::TurnContext;
use crate::codex::run_turn;
use crate::error::Result as CodexResult;
use crate::protocol::EventMsg;
use crate::protocol::TurnStartedEvent;
use crate::session_startup_prewarm::SessionStartupPrewarmResolution;
use crate::state::TaskKind;
use async_trait::async_trait;
use codex_protocol::user_input::UserInput;
use tokio_util::sync::CancellationToken;
use tracing::Instrument;
use tracing::trace_span;
use super::SessionTask;
use super::SessionTaskContext;
pub(crate) struct RegularTask {
prewarmed_session: Mutex<Option<ModelClientSession>>,
}
impl Default for RegularTask {
fn default() -> Self {
Self {
prewarmed_session: Mutex::new(None),
}
}
}
#[derive(Default)]
pub(crate) struct RegularTask;
impl RegularTask {
pub(crate) async fn with_startup_prewarm(
model_client: ModelClient,
prompt: Prompt,
turn_context: Arc<TurnContext>,
turn_metadata_header: Option<String>,
) -> CodexResult<Self> {
let mut client_session = model_client.new_session();
client_session
.prewarm_websocket(
&prompt,
&turn_context.model_info,
&turn_context.session_telemetry,
turn_context.reasoning_effort,
turn_context.reasoning_summary,
turn_context.config.service_tier,
turn_metadata_header.as_deref(),
)
.await?;
Ok(Self {
prewarmed_session: Mutex::new(Some(client_session)),
})
}
async fn take_prewarmed_session(&self) -> Option<ModelClientSession> {
self.prewarmed_session
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take()
pub(crate) fn new() -> Self {
Self
}
}
@@ -81,8 +44,25 @@ impl SessionTask for RegularTask {
) -> Option<String> {
let sess = session.clone_session();
let run_turn_span = trace_span!("run_turn");
// Regular turns emit `TurnStarted` inline so first-turn lifecycle does
// not wait on startup prewarm resolution.
let event = EventMsg::TurnStarted(TurnStartedEvent {
turn_id: ctx.sub_id.clone(),
model_context_window: ctx.model_context_window(),
collaboration_mode_kind: ctx.collaboration_mode.mode,
});
sess.send_event(ctx.as_ref(), event).await;
sess.set_server_reasoning_included(/*included*/ false).await;
let prewarmed_client_session = self.take_prewarmed_session().await;
let prewarmed_client_session = match sess
.consume_startup_prewarm_for_regular_turn(&cancellation_token)
.await
{
SessionStartupPrewarmResolution::Cancelled => return None,
SessionStartupPrewarmResolution::Unavailable { .. } => None,
SessionStartupPrewarmResolution::Ready(prewarmed_client_session) => {
Some(*prewarmed_client_session)
}
};
run_turn(
sess,
ctx,
codex-rs/core/src/tools/code_mode/runner.cjs
+10 -5
View File
@@ -465,6 +465,7 @@ function codeModeWorkerMain() {
writable: false,
});
parentPort.postMessage({ type: 'started' });
try {
await runModule(context, start, callTool, helpers);
parentPort.postMessage({
@@ -639,6 +640,10 @@ function startSession(protocol, sessions, start) {
content_items: [],
default_yield_time_ms: normalizeYieldTime(start.default_yield_time_ms),
id: start.cell_id,
initial_yield_time_ms:
start.yield_time_ms == null
? normalizeYieldTime(start.default_yield_time_ms)
: normalizeYieldTime(start.yield_time_ms),
initial_yield_timer: null,
initial_yield_triggered: false,
max_output_tokens_per_exec_call: maxOutputTokensPerExecCall,
@@ -651,11 +656,6 @@ function startSession(protocol, sessions, start) {
}),
};
sessions.set(session.id, session);
const initialYieldTime =
start.yield_time_ms == null
? session.default_yield_time_ms
: normalizeYieldTime(start.yield_time_ms);
scheduleInitialYield(protocol, session, initialYieldTime);
session.worker.on('message', (message) => {
void handleWorkerMessage(protocol, sessions, session, message).catch((error) => {
@@ -694,6 +694,11 @@ async function handleWorkerMessage(protocol, sessions, session, message) {
return;
}
if (message.type === 'started') {
scheduleInitialYield(protocol, session, session.initial_yield_time_ms);
return;
}
if (message.type === 'yield') {
void sendYielded(protocol, session);
return;
codex-rs/core/src/tools/handlers/agent_jobs.rs
+63 -27
View File
@@ -15,9 +15,12 @@ use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
use async_trait::async_trait;
use codex_protocol::ThreadId;
use codex_protocol::protocol::AgentStatus;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
use futures::StreamExt;
use futures::stream::FuturesUnordered;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
@@ -26,8 +29,10 @@ use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::watch::Receiver;
use tokio::time::Duration;
use tokio::time::Instant;
use tokio::time::timeout;
use uuid::Uuid;
pub struct BatchJobHandler;
@@ -103,6 +108,7 @@ struct JobRunnerOptions {
struct ActiveJobItem {
item_id: String,
started_at: Instant,
status_rx: Option<Receiver<AgentStatus>>,
}
struct JobProgressEmitter {
@@ -676,6 +682,12 @@ async fn run_agent_job_loop(
ActiveJobItem {
item_id: item.item_id.clone(),
started_at: Instant::now(),
status_rx: session
.services
.agent_control
.subscribe_status(thread_id)
.await
.ok(),
},
);
progressed = true;
@@ -708,7 +720,7 @@ async fn run_agent_job_loop(
break;
}
if !progressed {
tokio::time::sleep(STATUS_POLL_INTERVAL).await;
wait_for_status_change(&active_items).await;
}
continue;
}
@@ -863,6 +875,12 @@ async fn recover_running_items(
ActiveJobItem {
item_id: item.item_id.clone(),
started_at: started_at_from_item(&item),
status_rx: session
.services
.agent_control
.subscribe_status(thread_id)
.await
.ok(),
},
);
}
@@ -876,13 +894,44 @@ async fn find_finished_threads(
) -> Vec<(ThreadId, String)> {
let mut finished = Vec::new();
for (thread_id, item) in active_items {
if is_final(&session.services.agent_control.get_status(*thread_id).await) {
let status = active_item_status(session.as_ref(), *thread_id, item).await;
if is_final(&status) {
finished.push((*thread_id, item.item_id.clone()));
}
}
finished
}
async fn active_item_status(
session: &Session,
thread_id: ThreadId,
item: &ActiveJobItem,
) -> AgentStatus {
if let Some(status_rx) = item.status_rx.as_ref()
&& status_rx.has_changed().is_ok()
{
return status_rx.borrow().clone();
}
session.services.agent_control.get_status(thread_id).await
}
async fn wait_for_status_change(active_items: &HashMap<ThreadId, ActiveJobItem>) {
let mut waiters = FuturesUnordered::new();
for item in active_items.values() {
if let Some(status_rx) = item.status_rx.as_ref() {
let mut status_rx = status_rx.clone();
waiters.push(async move {
let _ = status_rx.changed().await;
});
}
}
if waiters.is_empty() {
tokio::time::sleep(STATUS_POLL_INTERVAL).await;
return;
}
let _ = timeout(STATUS_POLL_INTERVAL, waiters.next()).await;
}
async fn reap_stale_active_items(
session: Arc<Session>,
db: Arc<codex_state::StateRuntime>,
@@ -920,37 +969,24 @@ async fn finalize_finished_item(
item_id: &str,
thread_id: ThreadId,
) -> anyhow::Result<()> {
let mut item = db
let item = db
.get_agent_job_item(job_id, item_id)
.await?
.ok_or_else(|| {
anyhow::anyhow!("job item not found for finalization: {job_id}/{item_id}")
})?;
if item.result_json.is_none() {
tokio::time::sleep(Duration::from_millis(250)).await;
item = db
.get_agent_job_item(job_id, item_id)
.await?
.ok_or_else(|| {
anyhow::anyhow!("job item not found after grace period: {job_id}/{item_id}")
})?;
}
if item.result_json.is_some() {
if !db.mark_agent_job_item_completed(job_id, item_id).await? {
db.mark_agent_job_item_failed(
job_id,
item_id,
"worker reported result but item could not transition to completed",
)
.await?;
if matches!(item.status, codex_state::AgentJobItemStatus::Running) {
if item.result_json.is_some() {
let _ = db.mark_agent_job_item_completed(job_id, item_id).await?;
} else {
let _ = db
.mark_agent_job_item_failed(
job_id,
item_id,
"worker finished without calling report_agent_job_result",
)
.await?;
}
} else {
db.mark_agent_job_item_failed(
job_id,
item_id,
"worker finished without calling report_agent_job_result",
)
.await?;
}
let _ = session
.services
codex-rs/core/src/tools/handlers/multi_agents/close_agent.rs
+4 -2
View File
@@ -95,13 +95,15 @@ impl ToolHandler for Handler {
.await;
result?;
Ok(CloseAgentResult { status })
Ok(CloseAgentResult {
previous_status: status,
})
}
}
#[derive(Debug, Deserialize, Serialize)]
pub(crate) struct CloseAgentResult {
pub(crate) status: AgentStatus,
pub(crate) previous_status: AgentStatus,
}
impl ToolOutput for CloseAgentResult {
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+27 -5
View File
@@ -98,15 +98,37 @@ impl ToolHandler for Handler {
),
Err(_) => (None, AgentStatus::NotFound),
};
let (new_agent_nickname, new_agent_role) = match new_thread_id {
Some(thread_id) => session
let agent_snapshot = match new_thread_id {
Some(thread_id) => {
session
.services
.agent_control
.get_agent_config_snapshot(thread_id)
.await
}
None => None,
};
let (new_agent_nickname, new_agent_role) = match (&agent_snapshot, new_thread_id) {
(Some(snapshot), _) => (
snapshot.session_source.get_nickname(),
snapshot.session_source.get_agent_role(),
),
(None, Some(thread_id)) => session
.services
.agent_control
.get_agent_nickname_and_role(thread_id)
.await
.unwrap_or((None, None)),
None => (None, None),
(None, None) => (None, None),
};
let effective_model = agent_snapshot
.as_ref()
.map(|snapshot| snapshot.model.clone())
.unwrap_or_else(|| args.model.clone().unwrap_or_default());
let effective_reasoning_effort = agent_snapshot
.as_ref()
.and_then(|snapshot| snapshot.reasoning_effort)
.unwrap_or(args.reasoning_effort.unwrap_or_default());
let nickname = new_agent_nickname.clone();
session
.send_event(
@@ -118,8 +140,8 @@ impl ToolHandler for Handler {
new_agent_nickname,
new_agent_role,
prompt,
model: args.model.clone().unwrap_or_default(),
reasoning_effort: args.reasoning_effort.unwrap_or_default(),
model: effective_model,
reasoning_effort: effective_reasoning_effort,
status,
}
.into(),
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+2 -2
View File
@@ -971,7 +971,7 @@ async fn wait_agent_returns_final_status_without_timeout() {
}
#[tokio::test]
async fn close_agent_submits_shutdown_and_returns_status() {
async fn close_agent_submits_shutdown_and_returns_previous_status() {
let (mut session, turn) = make_session_and_context().await;
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
@@ -993,7 +993,7 @@ async fn close_agent_submits_shutdown_and_returns_status() {
let (content, success) = expect_text_output(output);
let result: close_agent::CloseAgentResult =
serde_json::from_str(&content).expect("close_agent result should be json");
assert_eq!(result.status, status_before);
assert_eq!(result.previous_status, status_before);
assert_eq!(success, Some(true));
let ops = manager.captured_ops();
codex-rs/core/src/tools/handlers/view_image.rs
+39 -18
View File
@@ -2,16 +2,16 @@ use async_trait::async_trait;
use codex_protocol::models::ContentItem;
use codex_protocol::models::FunctionCallOutputContentItem;
use codex_protocol::models::ImageDetail;
use codex_protocol::models::local_image_content_items_with_label_number;
use codex_protocol::models::local_image_content_items_from_bytes_with_label_number;
use codex_protocol::openai_models::InputModality;
use codex_utils_image::PromptImageMode;
use serde::Deserialize;
use tokio::fs;
use crate::function_tool::FunctionCallError;
use crate::original_image_detail::can_request_original_image_detail;
use crate::protocol::EventMsg;
use crate::protocol::ViewImageToolCallEvent;
use crate::sandboxed_fs;
use crate::tools::context::FunctionToolOutput;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolPayload;
@@ -88,20 +88,6 @@ impl ToolHandler for ViewImageHandler {
};
let abs_path = turn.resolve_path(Some(args.path));
let metadata = fs::metadata(&abs_path).await.map_err(|error| {
FunctionCallError::RespondToModel(format!(
"unable to locate image at `{}`: {error}",
abs_path.display()
))
})?;
if !metadata.is_file() {
return Err(FunctionCallError::RespondToModel(format!(
"image path `{}` is not a file",
abs_path.display()
)));
}
let event_path = abs_path.clone();
let can_request_original_detail =
@@ -114,9 +100,17 @@ impl ToolHandler for ViewImageHandler {
PromptImageMode::ResizeToFit
};
let image_detail = use_original_detail.then_some(ImageDetail::Original);
let image_bytes = sandboxed_fs::read_bytes(&session, &turn, &abs_path)
.await
.map_err(|error| {
FunctionCallError::RespondToModel(render_view_image_read_error(&abs_path, &error))
})?;
let content = local_image_content_items_with_label_number(
&abs_path, /*label_number*/ None, image_mode,
let content = local_image_content_items_from_bytes_with_label_number(
&abs_path,
image_bytes,
/*label_number*/ None,
image_mode,
)
.into_iter()
.map(|item| match item {
@@ -142,3 +136,30 @@ impl ToolHandler for ViewImageHandler {
Ok(FunctionToolOutput::from_content(content, Some(true)))
}
}
fn render_view_image_read_error(
path: &std::path::Path,
error: &sandboxed_fs::SandboxedFsError,
) -> String {
let operation_message = error
.operation_error_message()
.map(str::to_owned)
.unwrap_or_else(|| error.to_string());
match error.operation_error_kind() {
Some(codex_fs_ops::FsErrorKind::IsADirectory) => {
format!("image path `{}` is not a file", path.display())
}
Some(codex_fs_ops::FsErrorKind::NotFound) => {
format!(
"unable to locate image at `{}`: {operation_message}",
path.display()
)
}
Some(_) | None => {
format!(
"unable to read image at `{}`: {operation_message}",
path.display()
)
}
}
}
codex-rs/core/src/tools/spec.rs
+6 -3
View File
@@ -195,9 +195,12 @@ fn close_agent_output_schema() -> JsonValue {
json!({
"type": "object",
"properties": {
"status": agent_status_output_schema()
"previous_status": {
"description": "The agent status observed before shutdown was requested.",
"allOf": [agent_status_output_schema()]
}
},
"required": ["status"],
"required": ["previous_status"],
"additionalProperties": false
})
}
@@ -1528,7 +1531,7 @@ fn create_close_agent_tool() -> ToolSpec {
ToolSpec::Function(ResponsesApiTool {
name: "close_agent".to_string(),
description: "Close an agent when it is no longer needed and return its last known status. Don't keep agents open for too long if they are not needed anymore.".to_string(),
description: "Close an agent when it is no longer needed and return its previous status before shutdown was requested. Don't keep agents open for too long if they are not needed anymore.".to_string(),
strict: false,
defer_loading: None,
parameters: JsonSchema::Object {
codex-rs/core/tests/responses_headers.rs
+3
View File
@@ -53,6 +53,7 @@ async fn responses_stream_includes_subagent_header_on_review() {
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -165,6 +166,7 @@ async fn responses_stream_includes_subagent_header_on_other() {
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -272,6 +274,7 @@ async fn responses_respects_model_info_overrides_from_config() {
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
codex-rs/core/tests/suite/client.rs
+3
View File
@@ -1792,6 +1792,7 @@ async fn azure_responses_request_includes_store_and_reasoning_ids() {
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -2393,6 +2394,7 @@ async fn azure_overrides_assign_properties_used_for_responses_url() {
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
@@ -2477,6 +2479,7 @@ async fn env_var_overrides_loaded_auth() {
request_max_retries: None,
stream_max_retries: None,
stream_idle_timeout_ms: None,
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
codex-rs/core/tests/suite/client_websockets.rs
+25 -1
View File
@@ -1500,6 +1500,13 @@ fn prompt_with_input_and_instructions(input: Vec<ResponseItem>, instructions: &s
}
fn websocket_provider(server: &WebSocketTestServer) -> ModelProviderInfo {
websocket_provider_with_connect_timeout(server, None)
}
fn websocket_provider_with_connect_timeout(
server: &WebSocketTestServer,
websocket_connect_timeout_ms: Option<u64>,
) -> ModelProviderInfo {
ModelProviderInfo {
name: "mock-ws".into(),
base_url: Some(format!("{}/v1", server.uri())),
@@ -1513,6 +1520,7 @@ fn websocket_provider(server: &WebSocketTestServer) -> ModelProviderInfo {
request_max_retries: Some(0),
stream_max_retries: Some(0),
stream_idle_timeout_ms: Some(5_000),
websocket_connect_timeout_ms,
requires_openai_auth: false,
supports_websockets: true,
}
@@ -1543,7 +1551,23 @@ async fn websocket_harness_with_options(
websocket_v2_enabled: bool,
prefer_websockets: bool,
) -> WebsocketTestHarness {
let provider = websocket_provider(server);
websocket_harness_with_provider_options(
websocket_provider(server),
runtime_metrics_enabled,
websocket_enabled,
websocket_v2_enabled,
prefer_websockets,
)
.await
}
async fn websocket_harness_with_provider_options(
provider: ModelProviderInfo,
runtime_metrics_enabled: bool,
websocket_enabled: bool,
websocket_v2_enabled: bool,
prefer_websockets: bool,
) -> WebsocketTestHarness {
let codex_home = TempDir::new().unwrap();
let mut config = load_default_config_for_test(&codex_home).await;
config.model = Some(MODEL.to_string());
codex-rs/core/tests/suite/stream_error_allows_next_turn.rs
+1
View File
@@ -76,6 +76,7 @@ async fn continue_after_stream_error() {
request_max_retries: Some(1),
stream_max_retries: Some(1),
stream_idle_timeout_ms: Some(2_000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
codex-rs/core/tests/suite/stream_no_completed.rs
+1
View File
@@ -61,6 +61,7 @@ async fn retries_on_early_close() {
request_max_retries: Some(0),
stream_max_retries: Some(1),
stream_idle_timeout_ms: Some(2000),
websocket_connect_timeout_ms: None,
requires_openai_auth: false,
supports_websockets: false,
};
codex-rs/core/tests/suite/view_image.rs
+111
View File
@@ -3,6 +3,7 @@
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use codex_core::CodexAuth;
use codex_core::config::Constrained;
use codex_core::features::Feature;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::openai_models::ConfigShellToolType;
@@ -13,9 +14,15 @@ use codex_protocol::openai_models::ModelsResponse;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::openai_models::ReasoningEffortPreset;
use codex_protocol::openai_models::TruncationPolicyConfig;
use codex_protocol::permissions::FileSystemAccessMode;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSandboxEntry;
use codex_protocol::permissions::FileSystemSandboxPolicy;
use codex_protocol::permissions::FileSystemSpecialPath;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ReadOnlyAccess;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::user_input::UserInput;
use core_test_support::responses;
@@ -1244,6 +1251,110 @@ async fn view_image_tool_errors_when_file_missing() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn view_image_tool_respects_filesystem_sandbox() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let sandbox_policy_for_config = SandboxPolicy::ReadOnly {
access: ReadOnlyAccess::Restricted {
include_platform_defaults: true,
readable_roots: Vec::new(),
},
network_access: false,
};
let mut builder = test_codex().with_config({
let sandbox_policy_for_config = sandbox_policy_for_config.clone();
move |config| {
config.permissions.sandbox_policy =
Constrained::allow_any(sandbox_policy_for_config);
config.permissions.file_system_sandbox_policy =
FileSystemSandboxPolicy::restricted(vec![
FileSystemSandboxEntry {
path: FileSystemPath::Special {
value: FileSystemSpecialPath::Minimal,
},
access: FileSystemAccessMode::Read,
},
FileSystemSandboxEntry {
path: FileSystemPath::Special {
value: FileSystemSpecialPath::CurrentWorkingDirectory,
},
access: FileSystemAccessMode::Read,
},
]);
}
});
let TestCodex {
codex,
config,
cwd,
session_configured,
..
} = builder.build(&server).await?;
let outside_dir = tempfile::tempdir()?;
let abs_path = outside_dir.path().join("blocked.png");
let image = ImageBuffer::from_pixel(256, 128, Rgba([10u8, 20, 30, 255]));
image.save(&abs_path)?;
let call_id = "view-image-sandbox-denied";
let arguments = serde_json::json!({ "path": abs_path }).to_string();
let first_response = sse(vec![
ev_response_created("resp-1"),
ev_function_call(call_id, "view_image", &arguments),
ev_completed("resp-1"),
]);
responses::mount_sse_once(&server, first_response).await;
let second_response = sse(vec![
ev_assistant_message("msg-1", "done"),
ev_completed("resp-2"),
]);
let mock = responses::mount_sse_once(&server, second_response).await;
let session_model = session_configured.model.clone();
codex
.submit(Op::UserTurn {
items: vec![UserInput::Text {
text: "please attach the outside image".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
cwd: cwd.path().to_path_buf(),
approval_policy: AskForApproval::Never,
sandbox_policy: config.permissions.sandbox_policy.get().clone(),
model: session_model,
effort: None,
summary: None,
service_tier: None,
collaboration_mode: None,
personality: None,
})
.await?;
wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let request = mock.single_request();
assert!(
request.inputs_of_type("input_image").is_empty(),
"sandbox-denied image should not produce an input_image message"
);
let output_text = request
.function_call_output_content_and_success(call_id)
.and_then(|(content, _)| content)
.expect("output text present");
let expected_prefix = format!("unable to read image at `{}`:", abs_path.display());
assert!(
output_text.starts_with(&expected_prefix),
"expected sandbox denial prefix `{expected_prefix}` but got `{output_text}`"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn view_image_tool_returns_unsupported_message_for_text_only_model() -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));