use super::{LineRead, LineReader, ShellState};
use crate::config::AuthState;
use std::{
io::Write,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
mpsc,
},
thread,
time::{Duration, Instant},
};
const LOGIN_WORKER_RECV_POLL_INTERVAL: Duration = Duration::from_millis(25);
const LOGIN_WORKER_JOIN_TIMEOUT: Duration = Duration::from_millis(500);
const LOGIN_WORKER_JOIN_POLL_INTERVAL: Duration = Duration::from_millis(10);
struct LoginWorker {
cancel: Arc<AtomicBool>,
manual_tx: Option<mpsc::Sender<String>>,
progress_rx: mpsc::Receiver<crate::login::LoginInstructions>,
result_rx: mpsc::Receiver<Result<String, String>>,
handle: Option<thread::JoinHandle<()>>,
}
pub(super) fn handle_login_builtin_interactive(
arg: Option<&str>,
state: &mut ShellState,
line_reader: &mut dyn LineReader,
writer: &mut dyn Write,
) -> anyhow::Result<String> {
let config = state
.config
.as_ref()
.ok_or_else(|| anyhow::anyhow!("/login requires loaded runtime config"))?;
let Some(provider) = arg else {
return Ok(crate::login::provider_list_text(&config.paths));
};
if provider == crate::login::CUSTOM_PROVIDER_LOGIN_ID || provider == "custom" {
return prompt_custom_provider_login(state, line_reader, writer);
}
if provider == crate::providers::CLAUDE_CODE_PROVIDER {
return Ok(crate::login::claude_code_login_instructions().to_string());
}
if provider != crate::providers::OPENAI_CODEX_PROVIDER {
anyhow::bail!(
"unsupported login provider '{provider}'; supported providers: openai-codex, claude-code, custom-provider"
);
}
let worker = spawn_login_worker(config.paths.clone());
let instructions = match receive_login_instructions(&worker) {
Ok(instructions) => instructions,
Err(error) => return cleanup_login_worker_after_error(worker, error),
};
writeln!(writer, "{}", instructions.message)?;
writeln!(
writer,
"Paste redirect URL/code and press Enter, or press Enter to continue waiting for loopback callback. Input is not added to shell history."
)?;
writer.flush()?;
match line_reader.read_sensitive_line("oauth> ", writer) {
Ok(LineRead::Line(input)) if !input.trim().is_empty() => {
if let Some(manual_tx) = &worker.manual_tx {
let _ = manual_tx.send(input);
}
}
Ok(LineRead::Line(_)) => {}
Ok(LineRead::Eof | LineRead::Interrupted) => {
worker.cancel.store(true, Ordering::Relaxed);
}
Err(error) => return cleanup_login_worker_after_error(worker, error),
}
let message = finish_login_worker(worker)?;
refresh_codex_auth_state(state);
Ok(message)
}
fn spawn_login_worker(paths: crate::config::McPaths) -> LoginWorker {
let cancel = Arc::new(AtomicBool::new(false));
let worker_cancel = Arc::clone(&cancel);
let (manual_tx, manual_rx) = mpsc::channel();
let (progress_tx, progress_rx) = mpsc::channel();
let (result_tx, result_rx) = mpsc::channel();
let handle = thread::spawn(move || {
let result = crate::login::login_openai_codex_with_controls(
&paths,
worker_cancel,
Some(manual_rx),
move |instructions| {
let _ = progress_tx.send(instructions);
},
)
.map(|result| result.message)
.map_err(|error| error.to_string());
let _ = result_tx.send(result);
});
LoginWorker {
cancel,
manual_tx: Some(manual_tx),
progress_rx,
result_rx,
handle: Some(handle),
}
}
fn receive_login_instructions(
worker: &LoginWorker,
) -> anyhow::Result<crate::login::LoginInstructions> {
loop {
if worker.cancel.load(Ordering::Relaxed) {
anyhow::bail!("login cancelled before OAuth instructions were available");
}
match worker
.progress_rx
.recv_timeout(LOGIN_WORKER_RECV_POLL_INTERVAL)
{
Ok(instructions) => return Ok(instructions),
Err(mpsc::RecvTimeoutError::Timeout) if worker_is_finished(worker) => {
anyhow::bail!("login worker exited before producing OAuth instructions");
}
Err(mpsc::RecvTimeoutError::Timeout) => continue,
Err(mpsc::RecvTimeoutError::Disconnected) => {
anyhow::bail!("login worker exited before producing OAuth instructions");
}
}
}
}
fn receive_login_result(worker: &LoginWorker) -> anyhow::Result<Result<String, String>> {
loop {
if worker.cancel.load(Ordering::Relaxed) {
anyhow::bail!("login cancelled before worker reported a result");
}
match worker
.result_rx
.recv_timeout(LOGIN_WORKER_RECV_POLL_INTERVAL)
{
Ok(result) => return Ok(result),
Err(mpsc::RecvTimeoutError::Timeout) if worker_is_finished(worker) => {
anyhow::bail!("login worker exited without reporting a result");
}
Err(mpsc::RecvTimeoutError::Timeout) => continue,
Err(mpsc::RecvTimeoutError::Disconnected) => {
anyhow::bail!("login worker exited without reporting a result");
}
}
}
}
fn worker_is_finished(worker: &LoginWorker) -> bool {
worker
.handle
.as_ref()
.is_none_or(thread::JoinHandle::is_finished)
}
fn join_login_worker_with_timeout(worker: &mut LoginWorker) -> anyhow::Result<bool> {
let Some(handle) = worker.handle.take() else {
return Ok(true);
};
let deadline = Instant::now() + LOGIN_WORKER_JOIN_TIMEOUT;
while !handle.is_finished() && Instant::now() < deadline {
thread::sleep(LOGIN_WORKER_JOIN_POLL_INTERVAL);
}
if handle.is_finished() {
handle
.join()
.map_err(|_| anyhow::anyhow!("login worker panicked"))?;
Ok(true)
} else {
eprintln!("login warning=worker_join_timeout action=detach");
Ok(false)
}
}
fn cleanup_login_worker_after_error(
mut worker: LoginWorker,
error: anyhow::Error,
) -> anyhow::Result<String> {
worker.cancel.store(true, Ordering::Relaxed);
drop(worker.manual_tx.take());
let _ = receive_login_result(&worker);
join_login_worker_with_timeout(&mut worker)?;
Err(error)
}
fn finish_login_worker(mut worker: LoginWorker) -> anyhow::Result<String> {
drop(worker.manual_tx.take());
let result = match receive_login_result(&worker) {
Ok(result) => result,
Err(error) => {
worker.cancel.store(true, Ordering::Relaxed);
join_login_worker_with_timeout(&mut worker)?;
return Err(error);
}
};
join_login_worker_with_timeout(&mut worker)?;
result.map_err(|error| {
worker.cancel.store(true, Ordering::Relaxed);
anyhow::anyhow!(error)
})
}
fn prompt_required_line(
line_reader: &mut dyn LineReader,
writer: &mut dyn Write,
prompt: &str,
) -> anyhow::Result<String> {
match line_reader.read_line(prompt, writer)? {
LineRead::Line(value) if !value.trim().is_empty() => Ok(value.trim().to_string()),
LineRead::Line(_) => {
anyhow::bail!("custom provider setup cancelled: required field was empty")
}
LineRead::Eof | LineRead::Interrupted => anyhow::bail!("custom provider setup cancelled"),
}
}
fn prompt_custom_provider_login(
state: &mut ShellState,
line_reader: &mut dyn LineReader,
writer: &mut dyn Write,
) -> anyhow::Result<String> {
writeln!(
writer,
"Configure a custom OpenAI-compatible provider. API key values are never stored; only the env var name is saved."
)?;
let id = prompt_required_line(line_reader, writer, "provider id> ")?;
let label = prompt_required_line(line_reader, writer, "display label> ")?;
let base_url = prompt_required_line(line_reader, writer, "base URL (e.g. https://host/v1)> ")?;
let env_var = match line_reader.read_line(
"API key env var name (optional; blank for no auth)> ",
writer,
)? {
LineRead::Line(value) => value.trim().to_string(),
LineRead::Eof | LineRead::Interrupted => anyhow::bail!("custom provider setup cancelled"),
};
let paths = state
.config
.as_ref()
.ok_or_else(|| anyhow::anyhow!("custom provider setup requires loaded runtime config"))?
.paths
.clone();
let settings = crate::config::read_settings(&paths)?;
if settings.custom_providers.contains_key(id.trim()) {
writeln!(
writer,
"Custom provider '{}' already exists. Type yes to replace: ",
id.trim()
)?;
match line_reader.read_sensitive_line("replace> ", writer)? {
LineRead::Line(input) if matches!(input.trim(), "yes" | "y") => {}
_ => {
return Ok(format!(
"custom provider setup cancelled; '{}' unchanged",
id.trim()
));
}
}
}
let result = crate::login::configure_custom_provider(&paths, &id, &label, &base_url, &env_var)?;
let provider = state
.config
.as_ref()
.ok_or_else(|| anyhow::anyhow!("custom provider setup requires loaded runtime config"))?
.provider_id()
.to_string();
let model = state
.config
.as_ref()
.and_then(|config| config.model.clone())
.unwrap_or_else(|| state.model.clone());
crate::commands::runtime::refresh_runtime_provider_selection(state, &provider, &model)?;
Ok(result.message)
}
pub(super) fn refresh_codex_auth_state(state: &mut ShellState) {
let Some(config) = &mut state.config else {
return;
};
if let Ok(credential) = crate::login::codex_credential_from_store(&config.paths) {
state.auth_state = AuthState::Ready {
provider: crate::providers::OPENAI_CODEX_PROVIDER.to_string(),
credential: credential.clone(),
};
config.auth = Some(credential);
}
}
pub(super) fn handle_logout_builtin_interactive(
arg: Option<&str>,
state: &mut ShellState,
line_reader: &mut dyn LineReader,
writer: &mut dyn Write,
) -> anyhow::Result<String> {
let config = state
.config
.as_ref()
.ok_or_else(|| anyhow::anyhow!("/logout requires loaded runtime config"))?;
let Some(provider_id) = arg else {
return crate::login::logout_provider_list_text(&config.paths);
};
if crate::config::read_settings(&config.paths)?
.custom_providers
.contains_key(provider_id)
{
writeln!(
writer,
"Remove custom provider metadata for {provider_id}? Type yes to confirm: "
)?;
match line_reader.read_sensitive_line("logout> ", writer)? {
LineRead::Line(input) if matches!(input.trim(), "yes" | "y") => {}
_ => {
return Ok(format!(
"logout cancelled; custom provider {provider_id} unchanged"
));
}
}
if crate::config::remove_custom_provider(&config.paths, provider_id)? {
crate::commands::runtime::clear_runtime_auth_for_provider(state, provider_id);
return Ok(format!("removed custom provider {provider_id}"));
}
return Ok(format!("custom provider {provider_id} is not configured"));
}
let provider = crate::login::validate_logout_provider(provider_id)?;
let status = crate::login::logout_provider_status(&config.paths, provider.id)?;
if status == crate::login::LogoutProviderStatus::Missing {
return Ok(format!(
"{} ({}) is not configured; credentials unchanged",
provider.label, provider.id
));
}
writeln!(
writer,
"Remove local auth for {} ({})? Type yes to confirm: ",
provider.label, provider.id
)?;
match line_reader.read_sensitive_line("logout> ", writer)? {
LineRead::Line(input) if matches!(input.trim(), "yes" | "y") => {}
LineRead::Line(_) | LineRead::Eof | LineRead::Interrupted => {
return Ok(format!(
"logout cancelled; {} ({}) credentials unchanged",
provider.label, provider.id
));
}
}
let removal = crate::config::remove_provider_auth(&config.paths, provider.id)?;
if removal.removed {
crate::commands::runtime::clear_runtime_auth_for_provider(state, provider.id);
Ok(format!(
"removed local auth for {} ({})",
provider.label, provider.id
))
} else {
Ok(format!(
"{} ({}) is not configured; credentials unchanged",
provider.label, provider.id
))
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::{
sync::atomic::AtomicBool,
time::{Duration, Instant},
};
fn fake_worker(
body: impl FnOnce(
mpsc::Sender<crate::login::LoginInstructions>,
mpsc::Sender<Result<String, String>>,
Arc<AtomicBool>,
) + Send
+ 'static,
) -> (LoginWorker, Arc<AtomicBool>, Arc<AtomicBool>) {
let cancel = Arc::new(AtomicBool::new(false));
let joined = Arc::new(AtomicBool::new(false));
let (manual_tx, _manual_rx) = mpsc::channel();
let (progress_tx, progress_rx) = mpsc::channel();
let (result_tx, result_rx) = mpsc::channel();
let worker_cancel = Arc::clone(&cancel);
let worker_joined = Arc::clone(&joined);
let handle = thread::spawn(move || {
body(progress_tx, result_tx, worker_cancel);
worker_joined.store(true, Ordering::Relaxed);
});
(
LoginWorker {
cancel: Arc::clone(&cancel),
manual_tx: Some(manual_tx),
progress_rx,
result_rx,
handle: Some(handle),
},
cancel,
joined,
)
}
#[test]
fn interactive_login_worker_joins_on_input_error() {
let (worker, cancel, joined) = fake_worker(|progress_tx, result_tx, cancel| {
let _ = progress_tx.send(crate::login::LoginInstructions {
url: "http://127.0.0.1/".to_string(),
message: "instructions".to_string(),
});
while !cancel.load(Ordering::Relaxed) {
thread::sleep(Duration::from_millis(1));
}
let _ = result_tx.send(Ok("cancelled".to_string()));
});
let error = cleanup_login_worker_after_error(worker, anyhow::anyhow!("input failed"))
.unwrap_err()
.to_string();
assert_eq!(error, "input failed");
assert!(cancel.load(Ordering::Relaxed));
assert!(joined.load(Ordering::Relaxed));
}
#[test]
fn interactive_login_cleanup_detaches_stalled_worker() {
let (worker, cancel, joined) = fake_worker(|progress_tx, _result_tx, _cancel| {
let _ = progress_tx.send(crate::login::LoginInstructions {
url: "http://127.0.0.1/".to_string(),
message: "instructions".to_string(),
});
loop {
thread::sleep(Duration::from_millis(50));
}
});
let started = Instant::now();
let error = cleanup_login_worker_after_error(worker, anyhow::anyhow!("input failed"))
.unwrap_err()
.to_string();
assert_eq!(error, "input failed");
assert!(cancel.load(Ordering::Relaxed));
assert!(!joined.load(Ordering::Relaxed));
assert!(started.elapsed() < Duration::from_secs(2));
}
#[test]
fn interactive_login_worker_joins_on_progress_failure() {
let (worker, cancel, joined) = fake_worker(|_progress_tx, result_tx, _cancel| {
let _ = result_tx.send(Err("progress failed".to_string()));
});
let _ = worker.progress_rx.recv().unwrap_err();
let error = cleanup_login_worker_after_error(
worker,
anyhow::anyhow!("login worker exited before producing OAuth instructions"),
)
.unwrap_err()
.to_string();
assert!(error.contains("before producing OAuth instructions"));
assert!(cancel.load(Ordering::Relaxed));
assert!(joined.load(Ordering::Relaxed));
}
#[test]
fn interactive_login_worker_joins_on_result_failure() {
let (worker, cancel, joined) = fake_worker(|_progress_tx, _result_tx, _cancel| {});
let error = finish_login_worker(worker).unwrap_err().to_string();
assert!(error.contains("without reporting a result"));
assert!(cancel.load(Ordering::Relaxed));
assert!(joined.load(Ordering::Relaxed));
}
#[test]
fn interactive_login_worker_joins_before_worker_error() {
let (worker, cancel, joined) = fake_worker(|_progress_tx, result_tx, _cancel| {
let _ = result_tx.send(Err("worker failed".to_string()));
thread::sleep(Duration::from_millis(20));
});
let error = finish_login_worker(worker).unwrap_err().to_string();
assert_eq!(error, "worker failed");
assert!(cancel.load(Ordering::Relaxed));
assert!(joined.load(Ordering::Relaxed));
}
#[test]
fn interactive_login_worker_joins_on_panic() {
let (worker, cancel, _joined) = fake_worker(|_progress_tx, _result_tx, _cancel| {
panic!("boom");
});
let error = finish_login_worker(worker).unwrap_err().to_string();
assert!(error.contains("login worker panicked"));
assert!(cancel.load(Ordering::Relaxed));
}
}
#[cfg(test)]
mod custom_provider_regression_tests {
use super::*;
use crate::config::{
CustomProviderConfig, CustomReasoningProtocol, EffectiveConfig, McPaths, ProviderCredential,
};
use std::io::Cursor;
#[test]
fn custom_provider_login_refreshes_active_runtime_config() {
let temp = tempfile::TempDir::new().unwrap();
let paths = McPaths::from_root(temp.path().join("mc"));
let custom = CustomProviderConfig {
label: "Old label".to_string(),
base_url: "http://old.example/v1".to_string(),
api_key_env_var: None,
models_dev_provider: Some("old-models".to_string()),
use_responses_endpoint: true,
supports_text_verbosity: false,
reasoning_protocol: CustomReasoningProtocol::AnthropicLike,
extra_models: vec!["old-model".to_string()],
};
crate::config::write_settings(
&paths,
&crate::config::Settings {
custom_providers: std::collections::BTreeMap::from([(
"local-ai".to_string(),
custom.clone(),
)]),
selected_model: crate::config::SelectedModelSettings {
provider: Some("local-ai".to_string()),
model: Some("model-a".to_string()),
..Default::default()
},
..Default::default()
},
)
.unwrap();
let config = EffectiveConfig {
provider: Some("local-ai".to_string()),
model: Some("model-a".to_string()),
no_color: false,
file_autocomplete_respects_gitignore: true,
custom_providers: std::collections::BTreeMap::from([("local-ai".to_string(), custom)]),
thinking_level: crate::thinking::ThinkingLevel::Default,
api_key: None,
auth: Some(ProviderCredential::NoAuth),
paths,
};
let manager = crate::sessions::SessionManager::new(temp.path().join("sessions"));
let mut state = ShellState::new(
manager,
None,
temp.path().to_path_buf(),
"model-a".to_string(),
config.auth_state(),
)
.with_config(config);
let mut input = Cursor::new("local-ai\nNew label\nhttp://new.example/v1\n\nyes\n");
let mut reader = crate::shell::BufReadLineReader::new(&mut input);
let mut output = Vec::new();
handle_login_builtin_interactive(
Some(crate::login::CUSTOM_PROVIDER_LOGIN_ID),
&mut state,
&mut reader,
&mut output,
)
.unwrap();
let refreshed = state.config.as_ref().unwrap();
let provider = refreshed.custom_providers.get("local-ai").unwrap();
assert_eq!(provider.label, "New label");
assert_eq!(provider.base_url, "http://new.example/v1");
assert_eq!(provider.models_dev_provider.as_deref(), Some("old-models"));
assert!(provider.use_responses_endpoint);
assert_eq!(
provider.reasoning_protocol,
CustomReasoningProtocol::AnthropicLike
);
assert_eq!(provider.extra_models, vec!["old-model"]);
assert!(matches!(
state.auth_state,
AuthState::Ready {
credential: ProviderCredential::NoAuth,
..
}
));
}
}