diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index ba946c67b6..80422b792a 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -582,7 +582,10 @@ pub async fn run_main_with_transport( let (remote_control_accept_handle, remote_control_handle) = start_remote_control( config.chatgpt_base_url.clone(), - state_db.clone(), + transport::RemoteControlStateRuntimeConfig { + sqlite_home: config.sqlite_home.clone(), + model_provider_id: config.model_provider_id.clone(), + }, auth_manager.clone(), transport_event_tx.clone(), transport_shutdown_token.clone(), diff --git a/codex-rs/app-server/src/transport/mod.rs b/codex-rs/app-server/src/transport/mod.rs index 75a4971905..81eee1d521 100644 --- a/codex-rs/app-server/src/transport/mod.rs +++ b/codex-rs/app-server/src/transport/mod.rs @@ -34,6 +34,7 @@ mod stdio; mod websocket; pub(crate) use remote_control::RemoteControlHandle; +pub(crate) use remote_control::RemoteControlStateRuntimeConfig; pub(crate) use remote_control::start_remote_control; pub(crate) use stdio::start_stdio_connection; pub(crate) use websocket::start_websocket_acceptor; diff --git a/codex-rs/app-server/src/transport/remote_control/mod.rs b/codex-rs/app-server/src/transport/remote_control/mod.rs index c014c7a2c9..e9cecc9968 100644 --- a/codex-rs/app-server/src/transport/remote_control/mod.rs +++ b/codex-rs/app-server/src/transport/remote_control/mod.rs @@ -13,8 +13,8 @@ use super::CHANNEL_CAPACITY; use super::TransportEvent; use super::next_connection_id; use codex_login::AuthManager; -use codex_state::StateRuntime; use std::io; +use std::path::PathBuf; use std::sync::Arc; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -34,6 +34,12 @@ pub(crate) struct RemoteControlHandle { enabled_tx: Arc>, } +#[derive(Clone)] +pub(crate) struct RemoteControlStateRuntimeConfig { + pub(crate) sqlite_home: PathBuf, + pub(crate) model_provider_id: String, +} + impl RemoteControlHandle { pub(crate) fn set_enabled(&self, enabled: bool) { self.enabled_tx.send_if_modified(|state| { @@ -46,7 +52,7 @@ impl RemoteControlHandle { pub(crate) async fn start_remote_control( remote_control_url: String, - state_db: Option>, + state_runtime_config: RemoteControlStateRuntimeConfig, auth_manager: Arc, transport_event_tx: mpsc::Sender, shutdown_token: CancellationToken, @@ -64,7 +70,7 @@ pub(crate) async fn start_remote_control( RemoteControlWebsocket::new( remote_control_url, remote_control_target, - state_db, + state_runtime_config, auth_manager, transport_event_tx, shutdown_token, diff --git a/codex-rs/app-server/src/transport/remote_control/tests.rs b/codex-rs/app-server/src/transport/remote_control/tests.rs index d702a99c2c..857d4cd33e 100644 --- a/codex-rs/app-server/src/transport/remote_control/tests.rs +++ b/codex-rs/app-server/src/transport/remote_control/tests.rs @@ -107,6 +107,13 @@ async fn remote_control_state_runtime(codex_home: &TempDir) -> Arc .expect("state runtime should initialize") } +fn remote_control_state_runtime_config(codex_home: &TempDir) -> RemoteControlStateRuntimeConfig { + RemoteControlStateRuntimeConfig { + sqlite_home: codex_home.path().to_path_buf(), + model_provider_id: "test-provider".to_string(), + } +} + fn remote_control_url_for_listener(listener: &TcpListener) -> String { let addr = listener .local_addr() @@ -126,7 +133,7 @@ async fn remote_control_transport_manages_virtual_clients_and_routes_messages() let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(remote_control_state_runtime(&codex_home).await), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token.clone(), @@ -391,7 +398,7 @@ async fn remote_control_transport_reconnects_after_disconnect() { let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(remote_control_state_runtime(&codex_home).await), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token.clone(), @@ -458,12 +465,13 @@ async fn remote_control_transport_reconnects_after_disconnect() { #[tokio::test] async fn remote_control_start_allows_remote_control_invalid_url_when_disabled() { + let codex_home = TempDir::new().expect("temp dir should create"); let (transport_event_tx, _transport_event_rx) = mpsc::channel::(CHANNEL_CAPACITY); let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( "https://internal.example.com/backend-api/".to_string(), - /*state_db*/ None, + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token.clone(), @@ -497,7 +505,7 @@ async fn remote_control_start_allows_missing_auth_when_enabled() { let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - /*state_db*/ None, + remote_control_state_runtime_config(&codex_home), auth_manager, transport_event_tx, shutdown_token.clone(), @@ -530,7 +538,7 @@ async fn remote_control_handle_set_enabled_stops_and_restarts_connections() { let shutdown_token = CancellationToken::new(); let (remote_task, remote_handle) = start_remote_control( remote_control_url, - Some(remote_control_state_runtime(&codex_home).await), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token.clone(), @@ -583,7 +591,7 @@ async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() { let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(remote_control_state_runtime(&codex_home).await), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token.clone(), @@ -751,7 +759,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() { let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(remote_control_state_runtime(&codex_home).await), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token.clone(), @@ -968,7 +976,7 @@ async fn remote_control_http_mode_reuses_persisted_enrollment_before_reenrolling let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(state_db.clone()), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager_with_home(&codex_home), transport_event_tx, shutdown_token.clone(), @@ -1035,7 +1043,7 @@ async fn remote_control_stdio_mode_waits_for_client_name_before_connecting() { let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(state_db.clone()), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager_with_home(&codex_home), transport_event_tx, shutdown_token.clone(), @@ -1073,7 +1081,6 @@ async fn remote_control_waits_for_account_id_before_enrolling() { AuthCredentialsStoreMode::File, ) .expect("auth without account id should save"); - let state_db = remote_control_state_runtime(&codex_home).await; let auth_manager = AuthManager::shared( codex_home.path().to_path_buf(), /*enable_codex_api_key_env*/ false, @@ -1092,7 +1099,7 @@ async fn remote_control_waits_for_account_id_before_enrolling() { let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(state_db.clone()), + remote_control_state_runtime_config(&codex_home), auth_manager, transport_event_tx, shutdown_token.clone(), @@ -1175,7 +1182,7 @@ async fn remote_control_http_mode_clears_stale_persisted_enrollment_after_404() let shutdown_token = CancellationToken::new(); let (remote_task, _remote_handle) = start_remote_control( remote_control_url, - Some(state_db.clone()), + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager_with_home(&codex_home), transport_event_tx, shutdown_token.clone(), diff --git a/codex-rs/app-server/src/transport/remote_control/websocket.rs b/codex-rs/app-server/src/transport/remote_control/websocket.rs index d813be1f6a..06a0f30880 100644 --- a/codex-rs/app-server/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server/src/transport/remote_control/websocket.rs @@ -1,4 +1,5 @@ use crate::transport::TransportEvent; +use crate::transport::remote_control::RemoteControlStateRuntimeConfig; use crate::transport::remote_control::client_tracker::ClientTracker; use crate::transport::remote_control::client_tracker::REMOTE_CONTROL_IDLE_SWEEP_INTERVAL; use crate::transport::remote_control::enroll::RemoteControlConnectionAuth; @@ -115,7 +116,7 @@ struct WebsocketState { pub(crate) struct RemoteControlWebsocket { remote_control_url: String, remote_control_target: Option, - state_db: Option>, + state_runtime_config: RemoteControlStateRuntimeConfig, auth_manager: Arc, shutdown_token: CancellationToken, reconnect_attempt: u64, @@ -138,7 +139,7 @@ impl RemoteControlWebsocket { pub(crate) fn new( remote_control_url: String, remote_control_target: Option, - state_db: Option>, + state_runtime_config: RemoteControlStateRuntimeConfig, auth_manager: Arc, transport_event_tx: mpsc::Sender, shutdown_token: CancellationToken, @@ -154,7 +155,7 @@ impl RemoteControlWebsocket { Self { remote_control_url, remote_control_target, - state_db, + state_runtime_config, auth_manager, shutdown_token, reconnect_attempt: 0, @@ -273,7 +274,7 @@ impl RemoteControlWebsocket { } connect_result = connect_remote_control_websocket( &remote_control_target, - self.state_db.as_deref(), + &self.state_runtime_config, &self.auth_manager, &mut self.auth_recovery, &mut self.enrollment, @@ -723,7 +724,7 @@ pub(crate) async fn load_remote_control_auth( pub(super) async fn connect_remote_control_websocket( remote_control_target: &RemoteControlTarget, - state_db: Option<&StateRuntime>, + state_runtime_config: &RemoteControlStateRuntimeConfig, auth_manager: &Arc, auth_recovery: &mut UnauthorizedRecovery, enrollment: &mut Option, @@ -734,6 +735,17 @@ pub(super) async fn connect_remote_control_websocket( tungstenite::http::Response<()>, )> { ensure_rustls_crypto_provider(); + let state_db = StateRuntime::init( + state_runtime_config.sqlite_home.clone(), + state_runtime_config.model_provider_id.clone(), + ) + .await + .map_err(|err| { + io::Error::other(format!( + "failed to open sqlite state db for remote control websocket: {err}" + )) + })?; + let state_db = state_db.as_ref(); let auth = load_remote_control_auth(auth_manager).await?; let enrollment_account_id = enrollment.as_ref().map(|enrollment| &enrollment.account_id); @@ -751,7 +763,7 @@ pub(super) async fn connect_remote_control_websocket( if enrollment.is_none() { *enrollment = load_persisted_remote_control_enrollment( - state_db, + Some(state_db), remote_control_target, &auth.account_id, app_server_client_name, @@ -778,7 +790,7 @@ pub(super) async fn connect_remote_control_websocket( Err(err) => return Err(err), }; if let Err(err) = update_persisted_remote_control_enrollment( - state_db, + Some(state_db), remote_control_target, &auth.account_id, app_server_client_name, @@ -821,7 +833,7 @@ pub(super) async fn connect_remote_control_websocket( enrollment_ref.environment_id ); if let Err(clear_err) = update_persisted_remote_control_enrollment( - state_db, + Some(state_db), remote_control_target, &auth.account_id, app_server_client_name, @@ -918,7 +930,6 @@ mod tests { use codex_login::save_auth; use codex_login::token_data::TokenData; use codex_login::token_data::parse_chatgpt_jwt_claims; - use codex_state::StateRuntime; use futures::StreamExt; use pretty_assertions::assert_eq; use std::sync::Arc; @@ -940,10 +951,13 @@ mod tests { #[cfg(not(windows))] const TEST_HTTP_ACCEPT_TIMEOUT: Duration = Duration::from_secs(5); - async fn remote_control_state_runtime(codex_home: &TempDir) -> Arc { - StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()) - .await - .expect("state runtime should initialize") + fn remote_control_state_runtime_config( + codex_home: &TempDir, + ) -> RemoteControlStateRuntimeConfig { + RemoteControlStateRuntimeConfig { + sqlite_home: codex_home.path().to_path_buf(), + model_provider_id: "test-provider".to_string(), + } } fn remote_control_auth_manager() -> Arc { @@ -1022,7 +1036,6 @@ mod tests { .await; }); let codex_home = TempDir::new().expect("temp dir should create"); - let state_db = remote_control_state_runtime(&codex_home).await; let auth_manager = remote_control_auth_manager(); let mut auth_recovery = auth_manager.unauthorized_recovery(); let mut enrollment = Some(RemoteControlEnrollment { @@ -1034,7 +1047,7 @@ mod tests { let err = match connect_remote_control_websocket( &remote_control_target, - Some(state_db.as_ref()), + &remote_control_state_runtime_config(&codex_home), &auth_manager, &mut auth_recovery, &mut enrollment, @@ -1051,6 +1064,45 @@ mod tests { assert_eq!(err.to_string(), expected_error); } + #[tokio::test] + async fn connect_remote_control_websocket_fails_when_state_runtime_cannot_open() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let remote_control_url = remote_control_url_for_listener(&listener); + let remote_control_target = + normalize_remote_control_url(&remote_control_url).expect("target should parse"); + let codex_home = TempDir::new().expect("temp dir should create"); + let sqlite_home = codex_home.path().join("not-a-directory"); + std::fs::write(&sqlite_home, "file blocks state runtime directory") + .expect("sqlite_home blocker file should write"); + let state_runtime_config = RemoteControlStateRuntimeConfig { + sqlite_home, + model_provider_id: "test-provider".to_string(), + }; + let auth_manager = remote_control_auth_manager(); + let mut auth_recovery = auth_manager.unauthorized_recovery(); + let mut enrollment = None; + + let err = connect_remote_control_websocket( + &remote_control_target, + &state_runtime_config, + &auth_manager, + &mut auth_recovery, + &mut enrollment, + /*subscribe_cursor*/ None, + /*app_server_client_name*/ None, + ) + .await + .expect_err("state runtime open failure should fail the connect attempt"); + + assert!( + err.to_string() + .starts_with("failed to open sqlite state db for remote control websocket:"), + "unexpected error: {err}" + ); + } + #[tokio::test] async fn connect_remote_control_websocket_recovers_after_unauthorized_reload() { let listener = TcpListener::bind("127.0.0.1:0") @@ -1066,7 +1118,6 @@ mod tests { AuthCredentialsStoreMode::File, ) .expect("stale auth should save"); - let state_db = remote_control_state_runtime(&codex_home).await; let auth_manager = AuthManager::shared( codex_home.path().to_path_buf(), /*enable_codex_api_key_env*/ false, @@ -1097,7 +1148,7 @@ mod tests { let err = connect_remote_control_websocket( &remote_control_target, - Some(state_db.as_ref()), + &remote_control_state_runtime_config(&codex_home), &auth_manager, &mut auth_recovery, &mut enrollment, @@ -1147,7 +1198,6 @@ mod tests { AuthCredentialsStoreMode::File, ) .expect("stale auth should save"); - let state_db = remote_control_state_runtime(&codex_home).await; let auth_manager = AuthManager::shared( codex_home.path().to_path_buf(), /*enable_codex_api_key_env*/ false, @@ -1164,7 +1214,7 @@ mod tests { let err = connect_remote_control_websocket( &remote_control_target, - Some(state_db.as_ref()), + &remote_control_state_runtime_config(&codex_home), &auth_manager, &mut auth_recovery, &mut enrollment, @@ -1199,6 +1249,7 @@ mod tests { .expect("listener should bind"); let remote_control_url = remote_control_url_for_listener(&listener); drop(listener); + let codex_home = TempDir::new().expect("temp dir should create"); let remote_control_target = normalize_remote_control_url(&remote_control_url).expect("target should parse"); @@ -1212,7 +1263,7 @@ mod tests { RemoteControlWebsocket::new( remote_control_url, Some(remote_control_target), - /*state_db*/ None, + remote_control_state_runtime_config(&codex_home), remote_control_auth_manager(), transport_event_tx, shutdown_token,