openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

in-process server

Browse Source
This commit is contained in:
easong-openai
2025-08-13 14:08:44 -07:00
parent d357fb7dba
commit b791c1cc3e
6 changed files with 138 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/login/src/entrypoints.rs
+28 -11
View File
@@ -88,20 +88,37 @@ pub async fn login_with_chatgpt(codex_home: &Path) -> std::io::Result<()> {
let codex_home = codex_home.to_path_buf();
let client_id_cloned = client_id.clone();
tokio::task::spawn_blocking(move || {
let opts = server::LoginServerOptions {
codex_home: codex_home.clone(),
client_id: client_id_cloned,
issuer: server::DEFAULT_ISSUER.to_string(),
port: server::DEFAULT_PORT,
open_browser: true,
expose_state_endpoint: false,
testing_timeout_secs: None,
#[cfg(feature = "http-e2e-tests")]
port_sender: None,
};
let opts = server::LoginServerOptions::for_cli(&codex_home, &client_id_cloned);
server::run_local_login_server_with_options(opts)
})
.await
.map_err(|e| std::io::Error::other(format!("task join error: {e}")))??;
Ok(())
}
pub fn spawn_login_in_process(
codex_home: &Path,
) -> (
std::thread::JoinHandle<std::io::Result<()>>,
std::sync::mpsc::Receiver<crate::server::LoginServerStatus>,
std::sync::mpsc::Sender<()>,
) {
let (status_tx, status_rx) = std::sync::mpsc::channel();
let (cancel_tx, cancel_rx) = std::sync::mpsc::channel();
let client_id = std::env::var("CODEX_CLIENT_ID")
.ok()
.filter(|s| !s.is_empty())
.unwrap_or_else(|| crate::CLIENT_ID.to_string());
let opts = server::LoginServerOptions::for_ui(
codex_home,
&client_id,
status_tx,
cancel_rx,
server::DEFAULT_PORT,
);
let handle = std::thread::spawn(move || server::run_local_login_server_with_options(opts));
(handle, status_rx, cancel_tx)
}
codex-rs/login/src/lib.rs
+2
View File
@@ -18,10 +18,12 @@ pub use auth_store::logout;
pub use auth_store::try_read_auth_json;
pub use entrypoints::SpawnedLogin;
pub use entrypoints::login_with_chatgpt;
pub use entrypoints::spawn_login_in_process;
pub use entrypoints::spawn_login_with_chatgpt;
pub use server::HeadlessOutcome;
pub use server::Http;
pub use server::LoginServerOptions;
pub use server::LoginServerStatus;
pub use server::process_callback_headless;
pub use server::run_local_login_server_with_options;
codex-rs/login/src/server.rs
+70 -20
View File
@@ -34,6 +34,12 @@ fn render_error_html(message: &str) -> String {
}
#[derive(Debug, Clone)]
pub enum LoginServerStatus {
Url(String),
Completed,
}
#[derive(Debug)]
pub struct LoginServerOptions {
pub codex_home: PathBuf,
pub client_id: String,
@@ -41,13 +47,50 @@ pub struct LoginServerOptions {
pub port: u16,
pub open_browser: bool,
pub expose_state_endpoint: bool,
/// timeout after x secs for e2e tests
pub testing_timeout_secs: Option<u64>,
#[cfg(feature = "http-e2e-tests")]
pub port_sender: Option<std::sync::mpsc::Sender<u16>>,
pub status_sender: Option<std::sync::mpsc::Sender<LoginServerStatus>>,
pub cancel_receiver: Option<std::sync::mpsc::Receiver<()>>,
}
impl LoginServerOptions {
pub fn for_cli(codex_home: &Path, client_id: &str) -> Self {
Self {
codex_home: codex_home.to_path_buf(),
client_id: client_id.to_string(),
issuer: DEFAULT_ISSUER.to_string(),
port: DEFAULT_PORT,
open_browser: true,
expose_state_endpoint: false,
testing_timeout_secs: None,
port_sender: None,
status_sender: None,
cancel_receiver: None,
}
}
pub fn for_ui(
codex_home: &Path,
client_id: &str,
status_sender: std::sync::mpsc::Sender<LoginServerStatus>,
cancel_receiver: std::sync::mpsc::Receiver<()>,
port: u16,
) -> Self {
Self {
codex_home: codex_home.to_path_buf(),
client_id: client_id.to_string(),
issuer: DEFAULT_ISSUER.to_string(),
port,
open_browser: true,
expose_state_endpoint: false,
testing_timeout_secs: None,
port_sender: None,
status_sender: Some(status_sender),
cancel_receiver: Some(cancel_receiver),
}
}
}
// Only default issuer supported for platform/api bases
const PLATFORM_BASE: &str = "https://platform.openai.com";
fn default_url_base(port: u16) -> String {
@@ -56,17 +99,7 @@ fn default_url_base(port: u16) -> String {
#[allow(dead_code)]
pub fn run_local_login_server(codex_home: &Path, client_id: &str) -> std::io::Result<()> {
let opts = LoginServerOptions {
codex_home: codex_home.to_path_buf(),
client_id: client_id.to_string(),
issuer: DEFAULT_ISSUER.to_string(),
port: DEFAULT_PORT,
open_browser: true,
expose_state_endpoint: false,
testing_timeout_secs: None,
#[cfg(feature = "http-e2e-tests")]
port_sender: None,
};
let opts = LoginServerOptions::for_cli(codex_home, client_id);
run_local_login_server_with_options(opts)
}
@@ -79,7 +112,6 @@ pub fn run_local_login_server_with_options(mut opts: LoginServerOptions) -> std:
.port();
opts.port = actual_port;
#[cfg(feature = "http-e2e-tests")]
if let Some(tx) = &opts.port_sender {
let _ = tx.send(actual_port);
}
@@ -113,13 +145,20 @@ pub fn run_local_login_server_with_options(mut opts: LoginServerOptions) -> std:
.append_pair("codex_cli_simplified_flow", "true")
.append_pair("state", &state);
eprintln!("Starting local login server on {url_base}");
if opts.status_sender.is_none() {
eprintln!("Starting local login server on {url_base}");
}
if let Some(tx) = &opts.status_sender {
let _ = tx.send(LoginServerStatus::Url(auth_url.as_str().to_string()));
}
if opts.open_browser {
let _ = webbrowser::open(auth_url.as_str());
}
eprintln!(
". If your browser did not open, navigate to this URL to authenticate: \n\n{auth_url}"
);
if opts.status_sender.is_none() {
eprintln!(
". If your browser did not open, navigate to this URL to authenticate: \n\n{auth_url}"
);
}
// If a testing timeout is configured, schedule an internal exit request so tests don't hang CI.
#[cfg(feature = "http-e2e-tests")]
@@ -131,6 +170,15 @@ pub fn run_local_login_server_with_options(mut opts: LoginServerOptions) -> std:
});
}
// If cancellation is requested via cancel_receiver, trigger server shutdown by hitting /success
if let Some(rx) = opts.cancel_receiver.take() {
let port = opts.port;
std::thread::spawn(move || {
let _ = rx.recv();
let _ = reqwest::blocking::get(format!("http://127.0.0.1:{port}/success"));
});
}
// Main request loop
'outer: loop {
let request = match server.recv() {
@@ -263,7 +311,9 @@ pub fn run_local_login_server_with_options(mut opts: LoginServerOptions) -> std:
}
}
}
if let Some(tx) = &opts.status_sender {
let _ = tx.send(LoginServerStatus::Completed);
}
Ok(())
}
codex-rs/login/tests/headless.rs
+2 -1
View File
@@ -59,8 +59,9 @@ fn default_opts(tmp: &TempDir) -> LoginServerOptions {
open_browser: false,
expose_state_endpoint: false,
testing_timeout_secs: None,
#[cfg(feature = "http-e2e-tests")]
port_sender: None,
status_sender: None,
cancel_receiver: None,
}
}
codex-rs/login/tests/server.rs
+2 -1
View File
@@ -208,8 +208,9 @@ fn spawn_login_server_and_wait(
open_browser: false,
expose_state_endpoint: true,
testing_timeout_secs: Some(5),
#[cfg(feature = "http-e2e-tests")]
port_sender: Some(tx),
status_sender: None,
cancel_receiver: None,
};
let handle = thread::spawn(move || run_local_login_server_with_options(opts));
codex-rs/tui/src/onboarding/auth.rs
+34 -53
View File
@@ -38,19 +38,20 @@ pub(crate) enum SignInState {
}
#[derive(Debug)]
/// Used to manage the lifecycle of SpawnedLogin and FrameTicker and ensure they get cleaned up.
/// Used to manage lifecycle of in-process login server and UI ticker.
pub(crate) struct ContinueInBrowserState {
login_child: Option<codex_login::SpawnedLogin>,
join_handle: Option<std::thread::JoinHandle<std::io::Result<()>>>,
cancel_tx: Option<std::sync::mpsc::Sender<()>>,
_frame_ticker: Option<FrameTicker>,
url_shared: std::sync::Arc<std::sync::Mutex<Option<String>>>,
}
impl Drop for ContinueInBrowserState {
fn drop(&mut self) {
if let Some(child) = &self.login_child {
if let Ok(mut locked) = child.child.lock() {
// Best-effort terminate and reap the child to avoid zombies.
let _ = locked.kill();
let _ = locked.wait();
}
if let Some(tx) = self.cancel_tx.take() {
let _ = tx.send(());
}
if let Some(handle) = self.join_handle.take() {
let _ = handle.join();
}
}
}
@@ -186,11 +187,8 @@ impl AuthModeWidget {
let mut lines = vec![Line::from(spans), Line::from("")];
if let SignInState::ChatGptContinueInBrowser(state) = &self.sign_in_state {
if let Some(url) = state
.login_child
.as_ref()
.and_then(|child| child.get_login_url())
{
let url_opt = state.url_shared.lock().ok().and_then(|g| g.clone());
if let Some(url) = url_opt {
lines.push(Line::from(" If the link doesn't open automatically, open the following link to authenticate:"));
lines.push(Line::from(vec![
Span::raw(" "),
@@ -291,22 +289,16 @@ impl AuthModeWidget {
fn start_chatgpt_login(&mut self) {
self.error = None;
match codex_login::spawn_login_with_chatgpt(&self.codex_home) {
Ok(child) => {
self.spawn_completion_poller(child.clone());
self.sign_in_state =
SignInState::ChatGptContinueInBrowser(ContinueInBrowserState {
login_child: Some(child),
_frame_ticker: Some(FrameTicker::new(self.event_tx.clone())),
});
self.event_tx.send(AppEvent::RequestRedraw);
}
Err(e) => {
self.sign_in_state = SignInState::PickMode;
self.error = Some(e.to_string());
self.event_tx.send(AppEvent::RequestRedraw);
}
}
let (handle, status_rx, cancel_tx) = codex_login::spawn_login_in_process(&self.codex_home);
let url_shared = std::sync::Arc::new(std::sync::Mutex::new(None));
self.spawn_status_listener(status_rx, url_shared.clone());
self.sign_in_state = SignInState::ChatGptContinueInBrowser(ContinueInBrowserState {
join_handle: Some(handle),
cancel_tx: Some(cancel_tx),
_frame_ticker: Some(FrameTicker::new(self.event_tx.clone())),
url_shared,
});
self.event_tx.send(AppEvent::RequestRedraw);
}
/// TODO: Read/write from the correct hierarchy config overrides + auth json + OPENAI_API_KEY.
@@ -319,37 +311,26 @@ impl AuthModeWidget {
self.event_tx.send(AppEvent::RequestRedraw);
}
fn spawn_completion_poller(&self, child: codex_login::SpawnedLogin) {
let child_arc = child.child.clone();
let stderr_buf = child.stderr.clone();
fn spawn_status_listener(
&self,
status_rx: std::sync::mpsc::Receiver<codex_login::LoginServerStatus>,
url_shared: std::sync::Arc<std::sync::Mutex<Option<String>>>,
) {
let event_tx = self.event_tx.clone();
std::thread::spawn(move || {
loop {
let done = {
if let Ok(mut locked) = child_arc.lock() {
match locked.try_wait() {
Ok(Some(status)) => Some(status.success()),
Ok(None) => None,
Err(_) => Some(false),
while let Ok(status) = status_rx.recv() {
match status {
codex_login::LoginServerStatus::Url(u) => {
if let Ok(mut g) = url_shared.lock() {
*g = Some(u);
}
} else {
Some(false)
event_tx.send(AppEvent::RequestRedraw);
}
};
if let Some(success) = done {
if success {
codex_login::LoginServerStatus::Completed => {
event_tx.send(AppEvent::OnboardingAuthComplete(Ok(())));
} else {
let err = stderr_buf
.lock()
.ok()
.and_then(|b| String::from_utf8(b.clone()).ok())
.unwrap_or_else(|| "login_with_chatgpt subprocess failed".to_string());
event_tx.send(AppEvent::OnboardingAuthComplete(Err(err)));
break;
}
break;
}
std::thread::sleep(std::time::Duration::from_millis(250));
}
});
}