diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 55f754e4aa..05e804c148 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -176,9 +176,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.98" +version = "1.0.99" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e16d2d3311acee920a9eb8d33b8cbc1787ce4a264e85f964c2404b969bdcd487" +checksum = "b0674a1ddeecb70197781e945de4b3b8ffb61fa939a5597bcf48503737663100" [[package]] name = "arbitrary" @@ -538,9 +538,9 @@ checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" [[package]] name = "clap" -version = "4.5.43" +version = "4.5.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50fd97c9dc2399518aa331917ac6f274280ec5eb34e555dd291899745c48ec6f" +checksum = "1fc0e74a703892159f5ae7d3aac52c8e6c392f5ae5f359c70b5881d60aaac318" dependencies = [ "clap_builder", "clap_derive", @@ -548,9 +548,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.43" +version = "4.5.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c35b5830294e1fa0462034af85cc95225a4cb07092c088c55bda3147cfcd8f65" +checksum = "b3e7f4214277f3c7aa526a59dd3fbe306a370daee1f8b7b8c987069cd8e888a8" dependencies = [ "anstream", "anstyle", @@ -570,9 +570,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.5.41" +version = "4.5.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef4f52386a59ca4c860f7393bcf8abd8dfd91ecccc0f774635ff68e92eeef491" +checksum = "14cb31bb0a7d536caef2639baa7fad459e15c3144efefa6dbd1c84562c4739f6" dependencies = [ "heck", "proc-macro2", @@ -2639,9 +2639,9 @@ checksum = "03087c2bad5e1034e8cace5926dec053fb3790248370865f5117a7d0213354c8" [[package]] name = "libc" -version = "0.2.174" +version = "0.2.175" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776" +checksum = "6a82ae493e598baaea5209805c49bbf2ea7de956d50d7da0da1164f9c6d28543" [[package]] name = "libfuzzer-sys" diff --git a/codex-rs/cli/src/login.rs b/codex-rs/cli/src/login.rs index 5f9dc5f908..36bbf2208d 100644 --- a/codex-rs/cli/src/login.rs +++ b/codex-rs/cli/src/login.rs @@ -14,14 +14,14 @@ use std::path::PathBuf; pub async fn login_with_chatgpt(codex_home: PathBuf) -> std::io::Result<()> { let opts = ServerOptions::new(codex_home, CLIENT_ID.to_string()); - let server = run_login_server(opts, None)?; + let server = run_login_server(opts)?; eprintln!( "Starting local login server on http://localhost:{}.\nIf your browser did not open, navigate to this URL to authenticate:\n\n{}", server.actual_port, server.auth_url, ); - server.block_until_done() + server.block_until_done().await } pub async fn run_login_with_chatgpt(cli_config_overrides: CliConfigOverrides) -> ! { diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 74eaf6704b..78942ff69f 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -25,7 +25,7 @@ env-flags = "0.1.1" eventsource-stream = "0.2.3" fs2 = "0.4.3" futures = "0.3" -libc = "0.2.174" +libc = "0.2.175" mcp-types = { path = "../mcp-types" } mime_guess = "2.0" os_info = "3.12.0" diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 7319c9a026..86a711e436 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -33,6 +33,7 @@ use crate::error::CodexErr; use crate::error::Result; use crate::error::UsageLimitReachedError; use crate::flags::CODEX_RS_SSE_FIXTURE; +use crate::model_family::ModelFamily; use crate::model_provider_info::ModelProviderInfo; use crate::model_provider_info::WireApi; use crate::models::ResponseItem; @@ -311,6 +312,30 @@ impl ModelClient { pub fn get_provider(&self) -> ModelProviderInfo { self.provider.clone() } + + /// Returns the currently configured model slug. + pub fn get_model(&self) -> String { + self.config.model.clone() + } + + /// Returns the currently configured model family. + pub fn get_model_family(&self) -> ModelFamily { + self.config.model_family.clone() + } + + /// Returns the current reasoning effort setting. + pub fn get_reasoning_effort(&self) -> ReasoningEffortConfig { + self.effort + } + + /// Returns the current reasoning summary setting. + pub fn get_reasoning_summary(&self) -> ReasoningSummaryConfig { + self.summary + } + + pub fn get_auth(&self) -> Option { + self.auth.clone() + } } #[derive(Debug, Deserialize, Serialize)] diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 8670978e86..397246a7de 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -989,7 +989,7 @@ async fn submission_loop( rx_sub: Receiver, ) { // Wrap once to avoid cloning TurnContext for each task. - let turn_context = Arc::new(turn_context); + let mut turn_context = Arc::new(turn_context); // To break out of this loop, send Op::Shutdown. while let Ok(sub) = rx_sub.recv().await { debug!(?sub, "Submission"); @@ -997,6 +997,83 @@ async fn submission_loop( Op::Interrupt => { sess.interrupt_task(); } + Op::OverrideTurnContext { + cwd, + approval_policy, + sandbox_policy, + model, + effort, + summary, + } => { + // Recalculate the persistent turn context with provided overrides. + let prev = Arc::clone(&turn_context); + let provider = prev.client.get_provider(); + + // Effective model + family + let (effective_model, effective_family) = if let Some(m) = model { + let fam = + find_family_for_model(&m).unwrap_or_else(|| config.model_family.clone()); + (m, fam) + } else { + (prev.client.get_model(), prev.client.get_model_family()) + }; + + // Effective reasoning settings + let effective_effort = effort.unwrap_or(prev.client.get_reasoning_effort()); + let effective_summary = summary.unwrap_or(prev.client.get_reasoning_summary()); + + let auth = prev.client.get_auth(); + // Build updated config for the client + let mut updated_config = (*config).clone(); + updated_config.model = effective_model.clone(); + updated_config.model_family = effective_family.clone(); + + let client = ModelClient::new( + Arc::new(updated_config), + auth, + provider, + effective_effort, + effective_summary, + sess.session_id, + ); + + let new_approval_policy = approval_policy.unwrap_or(prev.approval_policy); + let new_sandbox_policy = sandbox_policy + .clone() + .unwrap_or(prev.sandbox_policy.clone()); + let new_cwd = cwd.clone().unwrap_or_else(|| prev.cwd.clone()); + + let tools_config = ToolsConfig::new( + &effective_family, + new_approval_policy, + new_sandbox_policy.clone(), + config.include_plan_tool, + config.include_apply_patch_tool, + ); + + let new_turn_context = TurnContext { + client, + tools_config, + user_instructions: prev.user_instructions.clone(), + base_instructions: prev.base_instructions.clone(), + approval_policy: new_approval_policy, + sandbox_policy: new_sandbox_policy.clone(), + shell_environment_policy: prev.shell_environment_policy.clone(), + cwd: new_cwd.clone(), + disable_response_storage: prev.disable_response_storage, + }; + + // Install the new persistent context for subsequent tasks/turns. + turn_context = Arc::new(new_turn_context); + if cwd.is_some() || approval_policy.is_some() || sandbox_policy.is_some() { + sess.record_conversation_items(&[ResponseItem::from(EnvironmentContext::new( + new_cwd, + new_approval_policy, + new_sandbox_policy, + ))]) + .await; + } + } Op::UserInput { items } => { // attempt to inject input into current task if let Err(items) = sess.inject_input(items) { @@ -1057,7 +1134,7 @@ async fn submission_loop( cwd, disable_response_storage: turn_context.disable_response_storage, }; - + // TODO: record the new environment context in the conversation history // no current task, spawn a new one with the per‑turn context let task = AgentTask::spawn(sess.clone(), Arc::new(fresh_turn_context), sub.id, items); diff --git a/codex-rs/core/tests/prompt_caching.rs b/codex-rs/core/tests/prompt_caching.rs index d637eb674e..9f5829e113 100644 --- a/codex-rs/core/tests/prompt_caching.rs +++ b/codex-rs/core/tests/prompt_caching.rs @@ -1,9 +1,13 @@ use codex_core::ConversationManager; use codex_core::ModelProviderInfo; use codex_core::built_in_model_providers; +use codex_core::protocol::AskForApproval; use codex_core::protocol::EventMsg; use codex_core::protocol::InputItem; use codex_core::protocol::Op; +use codex_core::protocol::SandboxPolicy; +use codex_core::protocol_config_types::ReasoningEffort; +use codex_core::protocol_config_types::ReasoningSummary; use codex_login::CodexAuth; use core_test_support::load_default_config_for_test; use core_test_support::load_sse_fixture_with_id; @@ -129,3 +133,230 @@ async fn prefixes_context_and_instructions_once_and_consistently_across_requests ); assert_eq!(body2["input"], expected_body2); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn overrides_turn_context_but_keeps_cached_prefix_and_key_constant() { + use pretty_assertions::assert_eq; + + let server = MockServer::start().await; + + let sse = sse_completed("resp"); + let template = ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_raw(sse, "text/event-stream"); + + // Expect two POSTs to /v1/responses + Mock::given(method("POST")) + .and(path("/v1/responses")) + .respond_with(template) + .expect(2) + .mount(&server) + .await; + + let model_provider = ModelProviderInfo { + base_url: Some(format!("{}/v1", server.uri())), + ..built_in_model_providers()["openai"].clone() + }; + + let cwd = TempDir::new().unwrap(); + let codex_home = TempDir::new().unwrap(); + let mut config = load_default_config_for_test(&codex_home); + config.cwd = cwd.path().to_path_buf(); + config.model_provider = model_provider; + config.user_instructions = Some("be consistent and helpful".to_string()); + + let conversation_manager = ConversationManager::default(); + let codex = conversation_manager + .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key"))) + .await + .expect("create new conversation") + .conversation; + + // First turn + codex + .submit(Op::UserInput { + items: vec![InputItem::Text { + text: "hello 1".into(), + }], + }) + .await + .unwrap(); + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + + // Change everything about the turn context. + let new_cwd = TempDir::new().unwrap(); + let writable = TempDir::new().unwrap(); + codex + .submit(Op::OverrideTurnContext { + cwd: Some(new_cwd.path().to_path_buf()), + approval_policy: Some(AskForApproval::Never), + sandbox_policy: Some(SandboxPolicy::WorkspaceWrite { + writable_roots: vec![writable.path().to_path_buf()], + network_access: true, + exclude_tmpdir_env_var: true, + exclude_slash_tmp: true, + }), + model: Some("o3".to_string()), + effort: Some(ReasoningEffort::High), + summary: Some(ReasoningSummary::Detailed), + }) + .await + .unwrap(); + + // Second turn after overrides + codex + .submit(Op::UserInput { + items: vec![InputItem::Text { + text: "hello 2".into(), + }], + }) + .await + .unwrap(); + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + + // Verify we issued exactly two requests, and the cached prefix stayed identical. + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 2, "expected two POST requests"); + + let body1 = requests[0].body_json::().unwrap(); + let body2 = requests[1].body_json::().unwrap(); + + // prompt_cache_key should remain constant across overrides + assert_eq!( + body1["prompt_cache_key"], body2["prompt_cache_key"], + "prompt_cache_key should not change across overrides" + ); + + // The entire prefix from the first request should be identical and reused + // as the prefix of the second request, ensuring cache hit potential. + let expected_user_message_2 = serde_json::json!({ + "type": "message", + "id": serde_json::Value::Null, + "role": "user", + "content": [ { "type": "input_text", "text": "hello 2" } ] + }); + // After overriding the turn context, the environment context should be emitted again + // reflecting the new cwd, approval policy and sandbox settings. + let expected_env_text_2 = format!( + "\nCurrent working directory: {}\nApproval policy: never\nSandbox mode: workspace-write\nNetwork access: enabled\n", + new_cwd.path().to_string_lossy() + ); + let expected_env_msg_2 = serde_json::json!({ + "type": "message", + "id": serde_json::Value::Null, + "role": "user", + "content": [ { "type": "input_text", "text": expected_env_text_2 } ] + }); + let expected_body2 = serde_json::json!( + [ + body1["input"].as_array().unwrap().as_slice(), + [expected_env_msg_2, expected_user_message_2].as_slice(), + ] + .concat() + ); + assert_eq!(body2["input"], expected_body2); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn per_turn_overrides_keep_cached_prefix_and_key_constant() { + use pretty_assertions::assert_eq; + + let server = MockServer::start().await; + + let sse = sse_completed("resp"); + let template = ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_raw(sse, "text/event-stream"); + + // Expect two POSTs to /v1/responses + Mock::given(method("POST")) + .and(path("/v1/responses")) + .respond_with(template) + .expect(2) + .mount(&server) + .await; + + let model_provider = ModelProviderInfo { + base_url: Some(format!("{}/v1", server.uri())), + ..built_in_model_providers()["openai"].clone() + }; + + let cwd = TempDir::new().unwrap(); + let codex_home = TempDir::new().unwrap(); + let mut config = load_default_config_for_test(&codex_home); + config.cwd = cwd.path().to_path_buf(); + config.model_provider = model_provider; + config.user_instructions = Some("be consistent and helpful".to_string()); + + let conversation_manager = ConversationManager::default(); + let codex = conversation_manager + .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key"))) + .await + .expect("create new conversation") + .conversation; + + // First turn + codex + .submit(Op::UserInput { + items: vec![InputItem::Text { + text: "hello 1".into(), + }], + }) + .await + .unwrap(); + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + + // Second turn using per-turn overrides via UserTurn + let new_cwd = TempDir::new().unwrap(); + let writable = TempDir::new().unwrap(); + codex + .submit(Op::UserTurn { + items: vec![InputItem::Text { + text: "hello 2".into(), + }], + cwd: new_cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + sandbox_policy: SandboxPolicy::WorkspaceWrite { + writable_roots: vec![writable.path().to_path_buf()], + network_access: true, + exclude_tmpdir_env_var: true, + exclude_slash_tmp: true, + }, + model: "o3".to_string(), + effort: ReasoningEffort::High, + summary: ReasoningSummary::Detailed, + }) + .await + .unwrap(); + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + + // Verify we issued exactly two requests, and the cached prefix stayed identical. + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 2, "expected two POST requests"); + + let body1 = requests[0].body_json::().unwrap(); + let body2 = requests[1].body_json::().unwrap(); + + // prompt_cache_key should remain constant across per-turn overrides + assert_eq!( + body1["prompt_cache_key"], body2["prompt_cache_key"], + "prompt_cache_key should not change across per-turn overrides" + ); + + // The entire prefix from the first request should be identical and reused + // as the prefix of the second request. + let expected_user_message_2 = serde_json::json!({ + "type": "message", + "id": serde_json::Value::Null, + "role": "user", + "content": [ { "type": "input_text", "text": "hello 2" } ] + }); + let expected_body2 = serde_json::json!( + [ + body1["input"].as_array().unwrap().as_slice(), + [expected_user_message_2].as_slice(), + ] + .concat() + ); + assert_eq!(body2["input"], expected_body2); +} diff --git a/codex-rs/linux-sandbox/Cargo.toml b/codex-rs/linux-sandbox/Cargo.toml index ea7052c409..d769fae2c6 100644 --- a/codex-rs/linux-sandbox/Cargo.toml +++ b/codex-rs/linux-sandbox/Cargo.toml @@ -20,7 +20,7 @@ clap = { version = "4", features = ["derive"] } codex-common = { path = "../common", features = ["cli"] } codex-core = { path = "../core" } landlock = "0.4.1" -libc = "0.2.172" +libc = "0.2.175" seccompiler = "0.5.0" [target.'cfg(target_os = "linux")'.dev-dependencies] diff --git a/codex-rs/login/src/server.rs b/codex-rs/login/src/server.rs index ef85df69ea..32229484ff 100644 --- a/codex-rs/login/src/server.rs +++ b/codex-rs/login/src/server.rs @@ -3,11 +3,7 @@ use std::io::{self}; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::Ordering; -use std::sync::mpsc; use std::thread; -use std::time::Duration; use crate::AuthDotJson; use crate::get_auth_file; @@ -32,7 +28,6 @@ pub struct ServerOptions { pub port: u16, pub open_browser: bool, pub force_state: Option, - pub login_timeout: Option, } impl ServerOptions { @@ -44,7 +39,6 @@ impl ServerOptions { port: DEFAULT_PORT, open_browser: true, force_state: None, - login_timeout: None, } } } @@ -52,52 +46,38 @@ impl ServerOptions { pub struct LoginServer { pub auth_url: String, pub actual_port: u16, - pub server_handle: thread::JoinHandle>, - pub shutdown_flag: Arc, - pub server: Arc, + server_handle: tokio::task::JoinHandle>, + shutdown_handle: ShutdownHandle, } impl LoginServer { - pub fn block_until_done(self) -> io::Result<()> { - #[expect(clippy::expect_used)] + pub async fn block_until_done(self) -> io::Result<()> { self.server_handle - .join() - .expect("can't join on the server thread") + .await + .map_err(|err| io::Error::other(format!("login server thread panicked: {err:?}")))? } pub fn cancel(&self) { - shutdown(&self.shutdown_flag, &self.server); + self.shutdown_handle.shutdown(); } pub fn cancel_handle(&self) -> ShutdownHandle { - ShutdownHandle { - shutdown_flag: self.shutdown_flag.clone(), - server: self.server.clone(), - } + self.shutdown_handle.clone() } } -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct ShutdownHandle { - shutdown_flag: Arc, - server: Arc, + shutdown_notify: Arc, } impl ShutdownHandle { - pub fn cancel(&self) { - shutdown(&self.shutdown_flag, &self.server); + pub fn shutdown(&self) { + self.shutdown_notify.notify_waiters(); } } -pub fn shutdown(shutdown_flag: &AtomicBool, server: &Server) { - shutdown_flag.store(true, Ordering::SeqCst); - server.unblock(); -} - -pub fn run_login_server( - opts: ServerOptions, - shutdown_flag: Option>, -) -> io::Result { +pub fn run_login_server(opts: ServerOptions) -> io::Result { let pkce = generate_pkce(); let state = opts.force_state.clone().unwrap_or_else(generate_state); @@ -119,75 +99,71 @@ pub fn run_login_server( if opts.open_browser { let _ = webbrowser::open(&auth_url); } - let shutdown_flag = shutdown_flag.unwrap_or_else(|| Arc::new(AtomicBool::new(false))); - let shutdown_flag_clone = shutdown_flag.clone(); - let timeout_flag = Arc::new(AtomicBool::new(false)); - // Channel used to signal completion to timeout watcher. - let (done_tx, done_rx) = mpsc::channel::<()>(); + // Map blocking reads from server.recv() to an async channel. + let (tx, mut rx) = tokio::sync::mpsc::channel::(16); + let _server_handle = { + let server = server.clone(); + thread::spawn(move || -> io::Result<()> { + while let Ok(request) = server.recv() { + tx.blocking_send(request).map_err(|e| { + eprintln!("Failed to send request to channel: {e}"); + io::Error::other("Failed to send request to channel") + })?; + } + Ok(()) + }) + }; - if let Some(timeout) = opts.login_timeout { - spawn_timeout_watcher( - done_rx, - timeout, - shutdown_flag.clone(), - timeout_flag.clone(), - server.clone(), - ); - } + let shutdown_notify = Arc::new(tokio::sync::Notify::new()); + let server_handle = { + let shutdown_notify = shutdown_notify.clone(); + let server = server.clone(); + tokio::spawn(async move { + let result = loop { + tokio::select! { + _ = shutdown_notify.notified() => { + break Err(io::Error::other("Login was not completed")); + } + maybe_req = rx.recv() => { + let Some(req) = maybe_req else { + break Err(io::Error::other("Login was not completed")); + }; - let server_for_thread = server.clone(); - let server_handle = thread::spawn(move || { - while !shutdown_flag.load(Ordering::SeqCst) { - let req = match server_for_thread.recv() { - Ok(r) => r, - Err(e) => { - // If we've been asked to shut down, break gracefully so that - // we can report timeout or cancellation status uniformly. - if shutdown_flag.load(Ordering::SeqCst) { - break; - } else { - return Err(io::Error::other(e)); + let url_raw = req.url().to_string(); + let response = + process_request(&url_raw, &opts, &redirect_uri, &pkce, actual_port, &state).await; + + let is_login_complete = matches!(response, HandledRequest::ResponseAndExit(_)); + match response { + HandledRequest::Response(r) | HandledRequest::ResponseAndExit(r) => { + let _ = tokio::task::spawn_blocking(move || req.respond(r)).await; + } + HandledRequest::RedirectWithHeader(header) => { + let redirect = Response::empty(302).with_header(header); + let _ = tokio::task::spawn_blocking(move || req.respond(redirect)).await; + } + } + + if is_login_complete { + break Ok(()); + } } } }; - let response = process_request(&req, &opts, &redirect_uri, &pkce, actual_port, &state); - let is_login_complete = matches!(response, HandledRequest::ResponseAndExit(_)); - match response { - HandledRequest::Response(r) | HandledRequest::ResponseAndExit(r) => { - let _ = req.respond(r); - } - HandledRequest::RedirectWithHeader(header) => { - let redirect = Response::empty(302).with_header(header); - let _ = req.respond(redirect); - } - } - - if is_login_complete { - shutdown_flag.store(true, Ordering::SeqCst); - // Login has succeeded, so disarm the timeout watcher. - let _ = done_tx.send(()); - return Ok(()); - } - } - - // Login has failed or timed out, so disarm the timeout watcher. - let _ = done_tx.send(()); - - if timeout_flag.load(Ordering::SeqCst) { - Err(io::Error::other("Login timed out")) - } else { - Err(io::Error::other("Login was not completed")) - } - }); + // Ensure that the server is unblocked so the thread dedicated to + // running `server.recv()` in a loop exits cleanly. + server.unblock(); + result + }) + }; Ok(LoginServer { - auth_url: auth_url.clone(), + auth_url, actual_port, server_handle, - shutdown_flag: shutdown_flag_clone, - server, + shutdown_handle: ShutdownHandle { shutdown_notify }, }) } @@ -197,15 +173,14 @@ enum HandledRequest { ResponseAndExit(Response>>), } -fn process_request( - req: &Request, +async fn process_request( + url_raw: &str, opts: &ServerOptions, redirect_uri: &str, pkce: &PkceCodes, actual_port: u16, state: &str, ) -> HandledRequest { - let url_raw = req.url().to_string(); let parsed_url = match url::Url::parse(&format!("http://localhost{url_raw}")) { Ok(u) => u, Err(e) => { @@ -236,18 +211,22 @@ fn process_request( }; match exchange_code_for_tokens(&opts.issuer, &opts.client_id, redirect_uri, pkce, &code) + .await { Ok(tokens) => { // Obtain API key via token-exchange and persist - let api_key = - obtain_api_key(&opts.issuer, &opts.client_id, &tokens.id_token).ok(); - if let Err(err) = persist_tokens( + let api_key = obtain_api_key(&opts.issuer, &opts.client_id, &tokens.id_token) + .await + .ok(); + if let Err(err) = persist_tokens_async( &opts.codex_home, api_key.clone(), tokens.id_token.clone(), Some(tokens.access_token.clone()), Some(tokens.refresh_token.clone()), - ) { + ) + .await + { eprintln!("Persist error: {err}"); return HandledRequest::Response( Response::from_string(format!("Unable to persist auth file: {err}")) @@ -292,29 +271,6 @@ fn process_request( } } -/// Spawns a detached thread that waits for either a completion signal on `done_rx` -/// or the specified `timeout` to elapse. If the timeout elapses first it marks -/// the `shutdown_flag`, records `timeout_flag`, and unblocks the HTTP server so -/// that the main server loop can exit promptly. -fn spawn_timeout_watcher( - done_rx: mpsc::Receiver<()>, - timeout: Duration, - shutdown_flag: Arc, - timeout_flag: Arc, - server: Arc, -) { - thread::spawn(move || { - if done_rx.recv_timeout(timeout).is_err() - && shutdown_flag - .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) - .is_ok() - { - timeout_flag.store(true, Ordering::SeqCst); - server.unblock(); - } - }); -} - fn build_authorize_url( issuer: &str, client_id: &str, @@ -353,7 +309,7 @@ struct ExchangedTokens { refresh_token: String, } -fn exchange_code_for_tokens( +async fn exchange_code_for_tokens( issuer: &str, client_id: &str, redirect_uri: &str, @@ -367,7 +323,7 @@ fn exchange_code_for_tokens( refresh_token: String, } - let client = reqwest::blocking::Client::new(); + let client = reqwest::Client::new(); let resp = client .post(format!("{issuer}/oauth/token")) .header("Content-Type", "application/x-www-form-urlencoded") @@ -379,6 +335,7 @@ fn exchange_code_for_tokens( urlencoding::encode(&pkce.code_verifier) )) .send() + .await .map_err(io::Error::other)?; if !resp.status().is_success() { @@ -388,7 +345,7 @@ fn exchange_code_for_tokens( ))); } - let tokens: TokenResponse = resp.json().map_err(io::Error::other)?; + let tokens: TokenResponse = resp.json().await.map_err(io::Error::other)?; Ok(ExchangedTokens { id_token: tokens.id_token, access_token: tokens.access_token, @@ -396,43 +353,49 @@ fn exchange_code_for_tokens( }) } -fn persist_tokens( +async fn persist_tokens_async( codex_home: &Path, api_key: Option, id_token: String, access_token: Option, refresh_token: Option, ) -> io::Result<()> { - let auth_file = get_auth_file(codex_home); - if let Some(parent) = auth_file.parent() { - if !parent.exists() { - std::fs::create_dir_all(parent).map_err(io::Error::other)?; + // Reuse existing synchronous logic but run it off the async runtime. + let codex_home = codex_home.to_path_buf(); + tokio::task::spawn_blocking(move || { + let auth_file = get_auth_file(&codex_home); + if let Some(parent) = auth_file.parent() { + if !parent.exists() { + std::fs::create_dir_all(parent).map_err(io::Error::other)?; + } } - } - let mut auth = read_or_default(&auth_file); - if let Some(key) = api_key { - auth.openai_api_key = Some(key); - } - let tokens = auth - .tokens - .get_or_insert_with(crate::token_data::TokenData::default); - tokens.id_token = crate::token_data::parse_id_token(&id_token).map_err(io::Error::other)?; - // Persist chatgpt_account_id if present in claims - if let Some(acc) = jwt_auth_claims(&id_token) - .get("chatgpt_account_id") - .and_then(|v| v.as_str()) - { - tokens.account_id = Some(acc.to_string()); - } - if let Some(at) = access_token { - tokens.access_token = at; - } - if let Some(rt) = refresh_token { - tokens.refresh_token = rt; - } - auth.last_refresh = Some(Utc::now()); - super::write_auth_json(&auth_file, &auth) + let mut auth = read_or_default(&auth_file); + if let Some(key) = api_key { + auth.openai_api_key = Some(key); + } + let tokens = auth + .tokens + .get_or_insert_with(crate::token_data::TokenData::default); + tokens.id_token = crate::token_data::parse_id_token(&id_token).map_err(io::Error::other)?; + // Persist chatgpt_account_id if present in claims + if let Some(acc) = jwt_auth_claims(&id_token) + .get("chatgpt_account_id") + .and_then(|v| v.as_str()) + { + tokens.account_id = Some(acc.to_string()); + } + if let Some(at) = access_token { + tokens.access_token = at; + } + if let Some(rt) = refresh_token { + tokens.refresh_token = rt; + } + auth.last_refresh = Some(Utc::now()); + super::write_auth_json(&auth_file, &auth) + }) + .await + .map_err(|e| io::Error::other(format!("persist task failed: {e}")))? } fn read_or_default(path: &Path) -> AuthDotJson { @@ -525,13 +488,13 @@ fn jwt_auth_claims(jwt: &str) -> serde_json::Map { serde_json::Map::new() } -fn obtain_api_key(issuer: &str, client_id: &str, id_token: &str) -> io::Result { +async fn obtain_api_key(issuer: &str, client_id: &str, id_token: &str) -> io::Result { // Token exchange for an API key access token #[derive(serde::Deserialize)] struct ExchangeResp { access_token: String, } - let client = reqwest::blocking::Client::new(); + let client = reqwest::Client::new(); let resp = client .post(format!("{issuer}/oauth/token")) .header("Content-Type", "application/x-www-form-urlencoded") @@ -544,6 +507,7 @@ fn obtain_api_key(issuer: &str, client_id: &str, id_token: &str) -> io::Result io::Result (SocketAddr, thread::JoinHandle<()>) { (addr, handle) } -#[test] -fn end_to_end_login_flow_persists_auth_json() { +#[tokio::test] +async fn end_to_end_login_flow_persists_auth_json() { if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() { println!( "Skipping test because it cannot execute when network is disabled in a Codex sandbox." @@ -100,22 +100,21 @@ fn end_to_end_login_flow_persists_auth_json() { port: 0, open_browser: false, force_state: Some(state), - login_timeout: None, }; - let server = run_login_server(opts, None).unwrap(); + let server = run_login_server(opts).unwrap(); let login_port = server.actual_port; // Simulate browser callback, and follow redirect to /success - let client = reqwest::blocking::Client::builder() + let client = reqwest::Client::builder() .redirect(reqwest::redirect::Policy::limited(5)) .build() .unwrap(); let url = format!("http://127.0.0.1:{login_port}/auth/callback?code=abc&state=test_state_123"); - let resp = client.get(&url).send().unwrap(); + let resp = client.get(&url).send().await.unwrap(); assert!(resp.status().is_success()); // Wait for server shutdown - server.block_until_done().unwrap(); + server.block_until_done().await.unwrap(); // Validate auth.json let auth_path = codex_home.join("auth.json"); @@ -133,8 +132,8 @@ fn end_to_end_login_flow_persists_auth_json() { drop(issuer_handle); } -#[test] -fn creates_missing_codex_home_dir() { +#[tokio::test] +async fn creates_missing_codex_home_dir() { if std::env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() { println!( "Skipping test because it cannot execute when network is disabled in a Codex sandbox." @@ -159,17 +158,16 @@ fn creates_missing_codex_home_dir() { port: 0, open_browser: false, force_state: Some(state), - login_timeout: None, }; - let server = run_login_server(opts, None).unwrap(); + let server = run_login_server(opts).unwrap(); let login_port = server.actual_port; - let client = reqwest::blocking::Client::new(); + let client = reqwest::Client::new(); let url = format!("http://127.0.0.1:{login_port}/auth/callback?code=abc&state=state2"); - let resp = client.get(&url).send().unwrap(); + let resp = client.get(&url).send().await.unwrap(); assert!(resp.status().is_success()); - server.block_until_done().unwrap(); + server.block_until_done().await.unwrap(); let auth_path = codex_home.join("auth.json"); assert!( diff --git a/codex-rs/mcp-server/src/codex_message_processor.rs b/codex-rs/mcp-server/src/codex_message_processor.rs index d13bdbf346..1decf11da6 100644 --- a/codex-rs/mcp-server/src/codex_message_processor.rs +++ b/codex-rs/mcp-server/src/codex_message_processor.rs @@ -66,7 +66,7 @@ struct ActiveLogin { impl ActiveLogin { fn drop(&self) { - self.shutdown_handle.cancel(); + self.shutdown_handle.shutdown(); } } @@ -146,7 +146,6 @@ impl CodexMessageProcessor { let opts = LoginServerOptions { open_browser: false, - login_timeout: Some(LOGIN_CHATGPT_TIMEOUT), ..LoginServerOptions::new(config.codex_home.clone(), CLIENT_ID.to_string()) }; @@ -155,9 +154,10 @@ impl CodexMessageProcessor { Error(JSONRPCErrorError), } - let reply = match run_login_server(opts, None) { + let reply = match run_login_server(opts) { Ok(server) => { let login_id = Uuid::new_v4(); + let shutdown_handle = server.cancel_handle(); // Replace active login if present. { @@ -166,7 +166,7 @@ impl CodexMessageProcessor { existing.drop(); } *guard = Some(ActiveLogin { - shutdown_handle: server.cancel_handle(), + shutdown_handle: shutdown_handle.clone(), login_id, }); } @@ -180,15 +180,19 @@ impl CodexMessageProcessor { let outgoing_clone = self.outgoing.clone(); let active_login = self.active_login.clone(); tokio::spawn(async move { - let result = - tokio::task::spawn_blocking(move || server.block_until_done()).await; - let (success, error_msg) = match result { + let (success, error_msg) = match tokio::time::timeout( + LOGIN_CHATGPT_TIMEOUT, + server.block_until_done(), + ) + .await + { Ok(Ok(())) => (true, None), Ok(Err(err)) => (false, Some(format!("Login server error: {err}"))), - Err(join_err) => ( - false, - Some(format!("failed to join login server thread: {join_err}")), - ), + Err(_elapsed) => { + // Timeout: cancel server and report + shutdown_handle.shutdown(); + (false, Some("Login timed out".to_string())) + } }; let notification = LoginChatGptCompleteNotification { login_id, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 39e63e1019..23ef4668f1 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -76,6 +76,38 @@ pub enum Op { summary: ReasoningSummaryConfig, }, + /// Override parts of the persistent turn context for subsequent turns. + /// + /// All fields are optional; when omitted, the existing value is preserved. + /// This does not enqueue any input – it only updates defaults used for + /// future `UserInput` turns. + OverrideTurnContext { + /// Updated `cwd` for sandbox/tool calls. + #[serde(skip_serializing_if = "Option::is_none")] + cwd: Option, + + /// Updated command approval policy. + #[serde(skip_serializing_if = "Option::is_none")] + approval_policy: Option, + + /// Updated sandbox policy for tool calls. + #[serde(skip_serializing_if = "Option::is_none")] + sandbox_policy: Option, + + /// Updated model slug. When set, the model family is derived + /// automatically. + #[serde(skip_serializing_if = "Option::is_none")] + model: Option, + + /// Updated reasoning effort (honored only for reasoning-capable models). + #[serde(skip_serializing_if = "Option::is_none")] + effort: Option, + + /// Updated reasoning summary preference (honored only for reasoning-capable models). + #[serde(skip_serializing_if = "Option::is_none")] + summary: Option, + }, + /// Approve a command execution ExecApproval { /// The id of the submission we are approving diff --git a/codex-rs/tui/src/bottom_pane/textarea.rs b/codex-rs/tui/src/bottom_pane/textarea.rs index 33cdbbbc1d..029ffc8590 100644 --- a/codex-rs/tui/src/bottom_pane/textarea.rs +++ b/codex-rs/tui/src/bottom_pane/textarea.rs @@ -230,6 +230,11 @@ impl TextArea { code: KeyCode::Backspace, modifiers: KeyModifiers::NONE, .. + } + | KeyEvent { + code: KeyCode::Char('h'), + modifiers: KeyModifiers::CONTROL, + .. } => self.delete_backward(1), KeyEvent { code: KeyCode::Delete, diff --git a/codex-rs/tui/src/onboarding/auth.rs b/codex-rs/tui/src/onboarding/auth.rs index 8407e84e21..facd06c64a 100644 --- a/codex-rs/tui/src/onboarding/auth.rs +++ b/codex-rs/tui/src/onboarding/auth.rs @@ -1,5 +1,6 @@ use codex_login::CLIENT_ID; use codex_login::ServerOptions; +use codex_login::ShutdownHandle; use codex_login::run_login_server; use crossterm::event::KeyCode; use crossterm::event::KeyEvent; @@ -24,10 +25,6 @@ use crate::onboarding::onboarding_screen::KeyboardHandler; use crate::onboarding::onboarding_screen::StepStateProvider; use crate::shimmer::shimmer_spans; use std::path::PathBuf; -use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::Ordering; -use std::thread::JoinHandle; use super::onboarding_screen::StepState; // no additional imports @@ -46,13 +43,14 @@ pub(crate) enum SignInState { /// Used to manage the lifecycle of SpawnedLogin and ensure it gets cleaned up. pub(crate) struct ContinueInBrowserState { auth_url: String, - shutdown_flag: Option>, - _login_wait_handle: Option>, + shutdown_handle: Option, + _login_wait_handle: Option>, } + impl Drop for ContinueInBrowserState { fn drop(&mut self) { - if let Some(flag) = &self.shutdown_flag { - flag.store(true, Ordering::SeqCst); + if let Some(flag) = &self.shutdown_handle { + flag.shutdown(); } } } @@ -283,16 +281,21 @@ impl AuthModeWidget { fn start_chatgpt_login(&mut self) { self.error = None; let opts = ServerOptions::new(self.codex_home.clone(), CLIENT_ID.to_string()); - let server = run_login_server(opts, None); + let server = run_login_server(opts); match server { Ok(child) => { let auth_url = child.auth_url.clone(); - let shutdown_flag = child.shutdown_flag.clone(); + let shutdown_handle = child.cancel_handle(); + + let event_tx = self.event_tx.clone(); + let join_handle = tokio::spawn(async move { + spawn_completion_poller(child, event_tx).await; + }); self.sign_in_state = SignInState::ChatGptContinueInBrowser(ContinueInBrowserState { auth_url, - shutdown_flag: Some(shutdown_flag), - _login_wait_handle: Some(self.spawn_completion_poller(child)), + shutdown_handle: Some(shutdown_handle), + _login_wait_handle: Some(join_handle), }); self.event_tx.send(AppEvent::RequestRedraw); } @@ -313,19 +316,21 @@ impl AuthModeWidget { } self.event_tx.send(AppEvent::RequestRedraw); } +} - fn spawn_completion_poller(&self, child: codex_login::LoginServer) -> JoinHandle<()> { - let event_tx = self.event_tx.clone(); - std::thread::spawn(move || { - if let Ok(()) = child.block_until_done() { - event_tx.send(AppEvent::OnboardingAuthComplete(Ok(()))); - } else { - event_tx.send(AppEvent::OnboardingAuthComplete(Err( - "login failed".to_string() - ))); - } - }) - } +async fn spawn_completion_poller( + child: codex_login::LoginServer, + event_tx: AppEventSender, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + if let Ok(()) = child.block_until_done().await { + event_tx.send(AppEvent::OnboardingAuthComplete(Ok(()))); + } else { + event_tx.send(AppEvent::OnboardingAuthComplete(Err( + "login failed".to_string() + ))); + } + }) } impl StepStateProvider for AuthModeWidget {