openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-17 12:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 074ef26b96 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2025-08-18 17:35:45 -07:00
committed by GitHub
parent 72ef24a736 074ef26b96
commit 9182522b45
13 changed files with 561 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
codex-rs/Cargo.lock generated
View File

@@ -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"

4
codex-rs/cli/src/login.rs
View File

@@ -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) -> ! {

2
codex-rs/core/Cargo.toml
View File

@@ -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"

25
codex-rs/core/src/client.rs
View File

@@ -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<CodexAuth> {
self.auth.clone()
}
}
#[derive(Debug, Deserialize, Serialize)]

81
codex-rs/core/src/codex.rs
View File

@@ -989,7 +989,7 @@ async fn submission_loop(
rx_sub: Receiver<Submission>,
) {
// 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);

231
codex-rs/core/tests/prompt_caching.rs
View File

@@ -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::<serde_json::Value>().unwrap();
let body2 = requests[1].body_json::<serde_json::Value>().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!(
"<environment_context>\nCurrent working directory: {}\nApproval policy: never\nSandbox mode: workspace-write\nNetwork access: enabled\n</environment_context>",
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::<serde_json::Value>().unwrap();
let body2 = requests[1].body_json::<serde_json::Value>().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);
}

2
codex-rs/linux-sandbox/Cargo.toml
View File

@@ -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]

274
codex-rs/login/src/server.rs
View File

@@ -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<String>,
pub login_timeout: Option<Duration>,
}
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<io::Result<()>>,
pub shutdown_flag: Arc<AtomicBool>,
pub server: Arc<Server>,
server_handle: tokio::task::JoinHandle<io::Result<()>>,
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<AtomicBool>,
server: Arc<Server>,
shutdown_notify: Arc<tokio::sync::Notify>,
}
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<Arc<AtomicBool>>,
) -> io::Result<LoginServer> {
pub fn run_login_server(opts: ServerOptions) -> io::Result<LoginServer> {
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::<Request>(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<Cursor<Vec<u8>>>),
}
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<AtomicBool>,
timeout_flag: Arc<AtomicBool>,
server: Arc<Server>,
) {
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<String>,
id_token: String,
access_token: Option<String>,
refresh_token: Option<String>,
) -> 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<String, serde_json::Value> {
serde_json::Map::new()
}
fn obtain_api_key(issuer: &str, client_id: &str, id_token: &str) -> io::Result<String> {
async fn obtain_api_key(issuer: &str, client_id: &str, id_token: &str) -> io::Result<String> {
// 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<S
urlencoding::encode("urn:ietf:params:oauth:token-type:id_token")
))
.send()
.await
.map_err(io::Error::other)?;
if !resp.status().is_success() {
return Err(io::Error::other(format!(
@@ -551,6 +515,6 @@ fn obtain_api_key(issuer: &str, client_id: &str, id_token: &str) -> io::Result<S
resp.status()
)));
}
let body: ExchangeResp = resp.json().map_err(io::Error::other)?;
let body: ExchangeResp = resp.json().await.map_err(io::Error::other)?;
Ok(body.access_token)
}

26
codex-rs/login/tests/login_server_e2e.rs
View File

@@ -73,8 +73,8 @@ fn start_mock_issuer() -> (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!(

26
codex-rs/mcp-server/src/codex_message_processor.rs
View File

@@ -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,

32
codex-rs/protocol/src/protocol.rs
View File

@@ -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<PathBuf>,
/// Updated command approval policy.
#[serde(skip_serializing_if = "Option::is_none")]
approval_policy: Option<AskForApproval>,
/// Updated sandbox policy for tool calls.
#[serde(skip_serializing_if = "Option::is_none")]
sandbox_policy: Option<SandboxPolicy>,
/// Updated model slug. When set, the model family is derived
/// automatically.
#[serde(skip_serializing_if = "Option::is_none")]
model: Option<String>,
/// Updated reasoning effort (honored only for reasoning-capable models).
#[serde(skip_serializing_if = "Option::is_none")]
effort: Option<ReasoningEffortConfig>,
/// Updated reasoning summary preference (honored only for reasoning-capable models).
#[serde(skip_serializing_if = "Option::is_none")]
summary: Option<ReasoningSummaryConfig>,
},
/// Approve a command execution
ExecApproval {
/// The id of the submission we are approving

5
codex-rs/tui/src/bottom_pane/textarea.rs
View File

@@ -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,

53
codex-rs/tui/src/onboarding/auth.rs
View File

@@ -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<Arc<AtomicBool>>,
_login_wait_handle: Option<JoinHandle<()>>,
shutdown_handle: Option<ShutdownHandle>,
_login_wait_handle: Option<tokio::task::JoinHandle<()>>,
}
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 {