From b791c1cc3e9aaa3d55e8e3454424b09c0bc80b4c Mon Sep 17 00:00:00 2001 From: easong-openai Date: Wed, 13 Aug 2025 14:08:44 -0700 Subject: [PATCH] in-process server --- codex-rs/login/src/entrypoints.rs | 39 +++++++++---- codex-rs/login/src/lib.rs | 2 + codex-rs/login/src/server.rs | 90 ++++++++++++++++++++++------- codex-rs/login/tests/headless.rs | 3 +- codex-rs/login/tests/server.rs | 3 +- codex-rs/tui/src/onboarding/auth.rs | 87 +++++++++++----------------- 6 files changed, 138 insertions(+), 86 deletions(-) diff --git a/codex-rs/login/src/entrypoints.rs b/codex-rs/login/src/entrypoints.rs index 544a227d19..7e1f1624df 100644 --- a/codex-rs/login/src/entrypoints.rs +++ b/codex-rs/login/src/entrypoints.rs @@ -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::sync::mpsc::Receiver, + 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) +} diff --git a/codex-rs/login/src/lib.rs b/codex-rs/login/src/lib.rs index 1d5a289259..0298720856 100644 --- a/codex-rs/login/src/lib.rs +++ b/codex-rs/login/src/lib.rs @@ -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; diff --git a/codex-rs/login/src/server.rs b/codex-rs/login/src/server.rs index b992f3b36e..667aa58858 100644 --- a/codex-rs/login/src/server.rs +++ b/codex-rs/login/src/server.rs @@ -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, - #[cfg(feature = "http-e2e-tests")] pub port_sender: Option>, + pub status_sender: Option>, + pub cancel_receiver: Option>, +} + +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, + 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(()) } diff --git a/codex-rs/login/tests/headless.rs b/codex-rs/login/tests/headless.rs index 783e7b6c73..bab9b6b973 100644 --- a/codex-rs/login/tests/headless.rs +++ b/codex-rs/login/tests/headless.rs @@ -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, } } diff --git a/codex-rs/login/tests/server.rs b/codex-rs/login/tests/server.rs index 0ed1216a3c..846bb28446 100644 --- a/codex-rs/login/tests/server.rs +++ b/codex-rs/login/tests/server.rs @@ -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)); diff --git a/codex-rs/tui/src/onboarding/auth.rs b/codex-rs/tui/src/onboarding/auth.rs index 6276552849..1faa1fe202 100644 --- a/codex-rs/tui/src/onboarding/auth.rs +++ b/codex-rs/tui/src/onboarding/auth.rs @@ -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, + join_handle: Option>>, + cancel_tx: Option>, _frame_ticker: Option, + url_shared: std::sync::Arc>>, } 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, + url_shared: std::sync::Arc>>, + ) { 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)); } }); }