Move TUI on top of app server (in place)

codex-rs/app-server-client/src/lib.rs

@@ -26,6 +26,7 @@ use std::time::Duration;
 pub use codex_app_server::in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY;
 pub use codex_app_server::in_process::InProcessServerEvent;
 use codex_app_server::in_process::InProcessStartArgs;
+pub use codex_app_server::shared_cloud_requirements_loader;
 use codex_app_server_protocol::ClientInfo;
 use codex_app_server_protocol::ClientNotification;
 use codex_app_server_protocol::ClientRequest;
@@ -39,7 +40,6 @@ use codex_arg0::Arg0DispatchPaths;
 use codex_core::AuthManager;
 use codex_core::ThreadManager;
 use codex_core::config::Config;
-use codex_core::config_loader::CloudRequirementsLoader;
 use codex_core::config_loader::LoaderOverrides;
 use codex_core::models_manager::collaboration_mode_presets::CollaborationModesConfig;
 use codex_feedback::CodexFeedback;
@@ -146,8 +146,6 @@ pub struct InProcessClientStartArgs {
     pub cli_overrides: Vec<(String, TomlValue)>,
     /// Loader override knobs used by config API paths.
     pub loader_overrides: LoaderOverrides,
-    /// Preloaded cloud requirements provider.
-    pub cloud_requirements: CloudRequirementsLoader,
     /// Feedback sink used by app-server/core telemetry and logs.
     pub feedback: CodexFeedback,
     /// Startup warnings emitted after initialize succeeds.
@@ -221,7 +219,6 @@ impl InProcessClientStartArgs {
             config: self.config,
             cli_overrides: self.cli_overrides,
             loader_overrides: self.loader_overrides,
-            cloud_requirements: self.cloud_requirements,
             auth_manager: Some(shared_core.auth_manager.clone()),
             thread_manager: Some(shared_core.thread_manager.clone()),
             feedback: self.feedback,
@@ -277,8 +274,7 @@ pub struct InProcessAppServerClient {
     command_tx: mpsc::Sender<ClientCommand>,
     event_rx: mpsc::Receiver<InProcessServerEvent>,
     worker_handle: tokio::task::JoinHandle<()>,
-    auth_manager: Arc<AuthManager>,
-    thread_manager: Arc<ThreadManager>,
+    shared_core: SharedCoreManagers,
 }
 
 impl InProcessAppServerClient {
@@ -442,21 +438,10 @@ impl InProcessAppServerClient {
             command_tx,
             event_rx,
             worker_handle,
-            auth_manager: shared_core.auth_manager,
-            thread_manager: shared_core.thread_manager,
+            shared_core,
         })
     }
 
-    /// Temporary bootstrap escape hatch for embedders migrating toward RPC-only usage.
-    pub fn auth_manager(&self) -> Arc<AuthManager> {
-        self.auth_manager.clone()
-    }
-
-    /// Temporary bootstrap escape hatch for embedders migrating toward RPC-only usage.
-    pub fn thread_manager(&self) -> Arc<ThreadManager> {
-        self.thread_manager.clone()
-    }
-
     /// Sends a typed client request and returns raw JSON-RPC result.
     ///
     /// Callers that expect a concrete response type should usually prefer
@@ -600,6 +585,45 @@ impl InProcessAppServerClient {
         self.event_rx.recv().await
     }
 
+    /// Receives the next non-legacy server event from the in-process runtime.
+    ///
+    /// Legacy bridge notifications are skipped so fully migrated callers can
+    /// consume only typed app-server notifications and requests.
+    pub async fn next_typed_event(&mut self) -> Option<InProcessServerEvent> {
+        loop {
+            match self.next_event().await {
+                Some(InProcessServerEvent::LegacyNotification(notification)) => {
+                    warn!(
+                        notification.method = %notification.method,
+                        "dropping legacy in-process app-server notification"
+                    );
+                }
+                Some(event) => return Some(event),
+                None => return None,
+            }
+        }
+    }
+
+    /// Temporary escape hatch for legacy TUI operations that still have no
+    /// app-server RPC equivalent.
+    pub async fn submit_legacy_thread_op(
+        &self,
+        thread_id: codex_protocol::ThreadId,
+        op: codex_protocol::protocol::Op,
+    ) -> IoResult<()> {
+        let thread = self
+            .shared_core
+            .thread_manager
+            .get_thread(thread_id)
+            .await
+            .map_err(|err| IoError::other(format!("failed to get thread: {err}")))?;
+        thread
+            .submit(op)
+            .await
+            .map(|_| ())
+            .map_err(|err| IoError::other(format!("failed to submit legacy thread op: {err}")))
+    }
+
     /// Shuts down worker and in-process runtime with bounded wait.
     ///
     /// If graceful shutdown exceeds timeout, the worker task is aborted to
@@ -609,8 +633,7 @@ impl InProcessAppServerClient {
             command_tx,
             event_rx,
             worker_handle,
-            auth_manager: _,
-            thread_manager: _,
+            ..
         } = self;
         let mut worker_handle = worker_handle;
         // Drop the caller-facing receiver before asking the worker to shut
@@ -662,8 +685,6 @@ mod tests {
     use codex_app_server_protocol::SessionSource as ApiSessionSource;
     use codex_app_server_protocol::ThreadStartParams;
     use codex_app_server_protocol::ThreadStartResponse;
-    use codex_core::AuthManager;
-    use codex_core::ThreadManager;
     use codex_core::config::ConfigBuilder;
     use pretty_assertions::assert_eq;
     use tokio::time::Duration;
@@ -686,7 +707,6 @@ mod tests {
             config: Arc::new(build_test_config().await),
             cli_overrides: Vec::new(),
             loader_overrides: LoaderOverrides::default(),
-            cloud_requirements: CloudRequirementsLoader::default(),
             feedback: CodexFeedback::new(),
             config_warnings: Vec::new(),
             session_source,
@@ -761,7 +781,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn shared_thread_manager_tracks_threads_started_via_app_server() {
+    async fn thread_read_loads_threads_started_via_app_server() {
         let client = start_test_client(SessionSource::Cli).await;
 
         let response: ThreadStartResponse = client
@@ -774,17 +794,18 @@ mod tests {
             })
             .await
             .expect("thread/start should succeed");
-        let created_thread_id = codex_protocol::ThreadId::from_string(&response.thread.id)
-            .expect("thread id should parse");
-        timeout(
-            Duration::from_secs(2),
-            client.thread_manager().get_thread(created_thread_id),
-        )
-        .await
-        .expect("timed out waiting for retained thread manager to observe started thread")
-        .expect("started thread should be visible through the shared thread manager");
-        let thread_ids = client.thread_manager().list_thread_ids().await;
-        assert!(thread_ids.contains(&created_thread_id));
+        let thread_id = response.thread.id;
+        let read: codex_app_server_protocol::ThreadReadResponse = client
+            .request_typed(ClientRequest::ThreadRead {
+                request_id: RequestId::Integer(4),
+                params: codex_app_server_protocol::ThreadReadParams {
+                    thread_id: thread_id.clone(),
+                    include_turns: false,
+                },
+            })
+            .await
+            .expect("thread/read should succeed");
+        assert_eq!(read.thread.id, thread_id);
 
         client.shutdown().await.expect("shutdown should complete");
     }
@@ -833,22 +854,6 @@ mod tests {
         let (command_tx, _command_rx) = mpsc::channel(1);
         let (event_tx, event_rx) = mpsc::channel(1);
         let worker_handle = tokio::spawn(async {});
-        let config = build_test_config().await;
-        let auth_manager = AuthManager::shared(
-            config.codex_home.clone(),
-            false,
-            config.cli_auth_credentials_store_mode,
-        );
-        let thread_manager = Arc::new(ThreadManager::new(
-            &config,
-            auth_manager.clone(),
-            SessionSource::Exec,
-            CollaborationModesConfig {
-                default_mode_request_user_input: config
-                    .features
-                    .enabled(codex_core::features::Feature::DefaultModeRequestUserInput),
-            },
-        ));
         event_tx
             .send(InProcessServerEvent::Lagged { skipped: 3 })
             .await
@@ -859,8 +864,22 @@ mod tests {
             command_tx,
             event_rx,
             worker_handle,
-            auth_manager,
-            thread_manager,
+            shared_core: InProcessClientStartArgs {
+                arg0_paths: Arg0DispatchPaths::default(),
+                config: Arc::new(build_test_config().await),
+                cli_overrides: Vec::new(),
+                loader_overrides: LoaderOverrides::default(),
+                feedback: CodexFeedback::new(),
+                config_warnings: Vec::new(),
+                session_source: SessionSource::Cli,
+                enable_codex_api_key_env: false,
+                client_name: "test-client".to_string(),
+                client_version: "0.0.0".to_string(),
+                experimental_api: false,
+                opt_out_notification_methods: Vec::new(),
+                channel_capacity: 1,
+            }
+            .shared_core_managers(),
         };
 
         let event = timeout(Duration::from_secs(2), client.next_event())
@@ -905,17 +924,54 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn accessors_expose_retained_shared_managers() {
-        let client = start_test_client(SessionSource::Cli).await;
+    async fn next_typed_event_skips_legacy_notifications() {
+        let (command_tx, _command_rx) = mpsc::channel(1);
+        let (event_tx, event_rx) = mpsc::channel(2);
+        let worker_handle = tokio::spawn(async {});
+        event_tx
+            .send(InProcessServerEvent::LegacyNotification(
+                codex_app_server_protocol::JSONRPCNotification {
+                    method: "codex/event/task_complete".to_string(),
+                    params: None,
+                },
+            ))
+            .await
+            .expect("legacy notification should enqueue");
+        event_tx
+            .send(InProcessServerEvent::Lagged { skipped: 2 })
+            .await
+            .expect("lagged marker should enqueue");
+        drop(event_tx);
 
-        assert!(
-            Arc::ptr_eq(&client.auth_manager(), &client.auth_manager()),
-            "auth_manager accessor should clone the retained shared manager"
-        );
-        assert!(
-            Arc::ptr_eq(&client.thread_manager(), &client.thread_manager()),
-            "thread_manager accessor should clone the retained shared manager"
-        );
+        let mut client = InProcessAppServerClient {
+            command_tx,
+            event_rx,
+            worker_handle,
+            shared_core: InProcessClientStartArgs {
+                arg0_paths: Arg0DispatchPaths::default(),
+                config: Arc::new(build_test_config().await),
+                cli_overrides: Vec::new(),
+                loader_overrides: LoaderOverrides::default(),
+                feedback: CodexFeedback::new(),
+                config_warnings: Vec::new(),
+                session_source: SessionSource::Cli,
+                enable_codex_api_key_env: false,
+                client_name: "test-client".to_string(),
+                client_version: "0.0.0".to_string(),
+                experimental_api: false,
+                opt_out_notification_methods: Vec::new(),
+                channel_capacity: 1,
+            }
+            .shared_core_managers(),
+        };
+
+        let event = timeout(Duration::from_secs(2), client.next_typed_event())
+            .await
+            .expect("typed event should arrive before timeout");
+        assert!(matches!(
+            event,
+            Some(InProcessServerEvent::Lagged { skipped: 2 })
+        ));
 
         client.shutdown().await.expect("shutdown should complete");
     }

codex-rs/app-server/src/in_process.rs

@@ -77,7 +77,6 @@ use codex_arg0::Arg0DispatchPaths;
 use codex_core::AuthManager;
 use codex_core::ThreadManager;
 use codex_core::config::Config;
-use codex_core::config_loader::CloudRequirementsLoader;
 use codex_core::config_loader::LoaderOverrides;
 use codex_feedback::CodexFeedback;
 use codex_protocol::protocol::SessionSource;
@@ -122,8 +121,6 @@ pub struct InProcessStartArgs {
     pub cli_overrides: Vec<(String, TomlValue)>,
     /// Loader override knobs used by config API paths.
     pub loader_overrides: LoaderOverrides,
-    /// Preloaded cloud requirements provider.
-    pub cloud_requirements: CloudRequirementsLoader,
     /// Optional prebuilt auth manager reused by an embedding caller.
     pub auth_manager: Option<Arc<AuthManager>>,
     /// Optional prebuilt thread manager reused by an embedding caller.
@@ -409,7 +406,6 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
                 config: args.config,
                 cli_overrides: args.cli_overrides,
                 loader_overrides: args.loader_overrides,
-                cloud_requirements: args.cloud_requirements,
                 auth_manager: args.auth_manager,
                 thread_manager: args.thread_manager,
                 feedback: args.feedback,
@@ -757,7 +753,6 @@ mod tests {
             config: Arc::new(build_test_config().await),
             cli_overrides: Vec::new(),
             loader_overrides: LoaderOverrides::default(),
-            cloud_requirements: CloudRequirementsLoader::default(),
             auth_manager: None,
             thread_manager: None,
             feedback: CodexFeedback::new(),

codex-rs/app-server/src/lib.rs

@@ -1,8 +1,6 @@
 #![deny(clippy::print_stdout, clippy::print_stderr)]
 
 use codex_arg0::Arg0DispatchPaths;
-use codex_cloud_requirements::cloud_requirements_loader;
-use codex_core::AuthManager;
 use codex_core::config::Config;
 use codex_core::config::ConfigBuilder;
 use codex_core::config_loader::CloudRequirementsLoader;
@@ -71,13 +69,15 @@ pub mod in_process;
 mod message_processor;
 mod models;
 mod outgoing_message;
+mod runtime_bootstrap;
 mod server_request_error;
 mod thread_state;
 mod thread_status;
 mod transport;
 
 pub use crate::error_code::INPUT_TOO_LARGE_ERROR_CODE;
 pub use crate::error_code::INVALID_PARAMS_ERROR_CODE;
+pub use crate::runtime_bootstrap::shared_cloud_requirements_loader;
 pub use crate::transport::AppServerTransport;
 
 const LOG_FORMAT_ENV_VAR: &str = "LOG_FORMAT";
@@ -416,16 +416,7 @@ pub async fn run_main_with_transport(
                 }
             }
 
-            let auth_manager = AuthManager::shared(
-                config.codex_home.clone(),
-                false,
-                config.cli_auth_credentials_store_mode,
-            );
-            cloud_requirements_loader(
-                auth_manager,
-                config.chatgpt_base_url,
-                config.codex_home.clone(),
-            )
+            runtime_bootstrap::shared_cloud_requirements_loader_for_config(&config, false)
         }
         Err(err) => {
             warn!(error = %err, "Failed to preload config for cloud requirements");
@@ -607,7 +598,6 @@ pub async fn run_main_with_transport(
             config: Arc::new(config),
             cli_overrides,
             loader_overrides,
-            cloud_requirements: cloud_requirements.clone(),
             auth_manager: None,
             thread_manager: None,
             feedback: feedback.clone(),