Finish moving codex exec to app-server

codex-rs/app-server-client/Cargo.toml

@@ -16,7 +16,6 @@ codex-app-server = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-arg0 = { workspace = true }
 codex-core = { workspace = true }
-codex-features = { workspace = true }
 codex-feedback = { workspace = true }
 codex-protocol = { workspace = true }
 futures = { workspace = true }

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

@@ -35,19 +35,14 @@ use codex_app_server_protocol::ConfigWarningNotification;
 use codex_app_server_protocol::InitializeCapabilities;
 use codex_app_server_protocol::InitializeParams;
 use codex_app_server_protocol::JSONRPCErrorError;
-use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::Result as JsonRpcResult;
 use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ServerRequest;
 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_features::Feature;
 use codex_feedback::CodexFeedback;
 use codex_protocol::protocol::SessionSource;
 use serde::de::DeserializeOwned;
@@ -73,7 +68,6 @@ pub type RequestResult = std::result::Result<JsonRpcResult, JSONRPCErrorError>;
 pub enum AppServerEvent {
     Lagged { skipped: usize },
     ServerNotification(ServerNotification),
-    LegacyNotification(JSONRPCNotification),
     ServerRequest(ServerRequest),
     Disconnected { message: String },
 }
@@ -85,9 +79,6 @@ impl From<InProcessServerEvent> for AppServerEvent {
             InProcessServerEvent::ServerNotification(notification) => {
                 Self::ServerNotification(notification)
             }
-            InProcessServerEvent::LegacyNotification(notification) => {
-                Self::LegacyNotification(notification)
-            }
             InProcessServerEvent::ServerRequest(request) => Self::ServerRequest(request),
         }
     }
@@ -97,19 +88,12 @@ fn event_requires_delivery(event: &InProcessServerEvent) -> bool {
     // These terminal events drive surface shutdown/completion state. Dropping
     // them under backpressure can leave exec/TUI waiting forever even though
     // the underlying turn has already ended.
-    match event {
+    matches!(
+        event,
         InProcessServerEvent::ServerNotification(
             codex_app_server_protocol::ServerNotification::TurnCompleted(_),
-        ) => true,
-        InProcessServerEvent::LegacyNotification(notification) => matches!(
-            notification
-                .method
-                .strip_prefix("codex/event/")
-                .unwrap_or(&notification.method),
-            "task_complete" | "turn_aborted" | "shutdown_complete"
-        ),
-        _ => false,
-    }
+        )
+    )
 }
 
 /// Layered error for [`InProcessAppServerClient::request_typed`].
@@ -159,16 +143,6 @@ impl Error for TypedRequestError {
     }
 }
 
-#[derive(Clone)]
-struct SharedCoreManagers {
-    // Temporary bootstrap escape hatch for embedders that still need direct
-    // core handles during the in-process app-server migration. Once TUI/exec
-    // stop depending on direct manager access, remove this wrapper and keep
-    // manager ownership entirely inside the app-server runtime.
-    auth_manager: Arc<AuthManager>,
-    thread_manager: Arc<ThreadManager>,
-}
-
 #[derive(Clone)]
 pub struct InProcessClientStartArgs {
     /// Resolved argv0 dispatch paths used by command execution internals.
@@ -202,30 +176,6 @@ pub struct InProcessClientStartArgs {
 }
 
 impl InProcessClientStartArgs {
-    fn shared_core_managers(&self) -> SharedCoreManagers {
-        let auth_manager = AuthManager::shared(
-            self.config.codex_home.clone(),
-            self.enable_codex_api_key_env,
-            self.config.cli_auth_credentials_store_mode,
-        );
-        let thread_manager = Arc::new(ThreadManager::new(
-            self.config.as_ref(),
-            auth_manager.clone(),
-            self.session_source.clone(),
-            CollaborationModesConfig {
-                default_mode_request_user_input: self
-                    .config
-                    .features
-                    .enabled(Feature::DefaultModeRequestUserInput),
-            },
-        ));
-
-        SharedCoreManagers {
-            auth_manager,
-            thread_manager,
-        }
-    }
-
     /// Builds initialize params from caller-provided metadata.
     pub fn initialize_params(&self) -> InitializeParams {
         let capabilities = InitializeCapabilities {
@@ -247,16 +197,14 @@ impl InProcessClientStartArgs {
         }
     }
 
-    fn into_runtime_start_args(self, shared_core: &SharedCoreManagers) -> InProcessStartArgs {
+    fn into_runtime_start_args(self) -> InProcessStartArgs {
         let initialize = self.initialize_params();
         InProcessStartArgs {
             arg0_paths: self.arg0_paths,
             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,
             config_warnings: self.config_warnings,
             session_source: self.session_source,
@@ -310,8 +258,6 @@ 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>,
 }
 
 #[derive(Clone)]
@@ -338,9 +284,8 @@ impl InProcessAppServerClient {
     /// with overload error instead of being silently dropped.
     pub async fn start(args: InProcessClientStartArgs) -> IoResult<Self> {
         let channel_capacity = args.channel_capacity.max(1);
-        let shared_core = args.shared_core_managers();
         let mut handle =
-            codex_app_server::in_process::start(args.into_runtime_start_args(&shared_core)).await?;
+            codex_app_server::in_process::start(args.into_runtime_start_args()).await?;
         let request_sender = handle.sender();
         let (command_tx, mut command_rx) = mpsc::channel::<ClientCommand>(channel_capacity);
         let (event_tx, event_rx) = mpsc::channel::<InProcessServerEvent>(channel_capacity);
@@ -401,6 +346,25 @@ impl InProcessAppServerClient {
                         let Some(event) = event else {
                             break;
                         };
+                        if let InProcessServerEvent::ServerRequest(
+                            ServerRequest::ChatgptAuthTokensRefresh { request_id, .. }
+                        ) = &event
+                        {
+                            let send_result = request_sender.fail_server_request(
+                                request_id.clone(),
+                                JSONRPCErrorError {
+                                    code: -32000,
+                                    message: "chatgpt auth token refresh is not supported for in-process app-server clients".to_string(),
+                                    data: None,
+                                },
+                            );
+                            if let Err(err) = send_result {
+                                warn!(
+                                    "failed to reject unsupported chatgpt auth token refresh request: {err}"
+                                );
+                            }
+                            continue;
+                        }
 
                         if skipped_events > 0 {
                             if event_requires_delivery(&event) {
@@ -491,21 +455,9 @@ impl InProcessAppServerClient {
             command_tx,
             event_rx,
             worker_handle,
-            auth_manager: shared_core.auth_manager,
-            thread_manager: shared_core.thread_manager,
         })
     }
 
-    /// 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()
-    }
-
     pub fn request_handle(&self) -> InProcessAppServerRequestHandle {
         InProcessAppServerRequestHandle {
             command_tx: self.command_tx.clone(),
@@ -664,8 +616,6 @@ 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
@@ -857,8 +807,6 @@ mod tests {
     use codex_app_server_protocol::ThreadStartResponse;
     use codex_app_server_protocol::ToolRequestUserInputParams;
     use codex_app_server_protocol::ToolRequestUserInputQuestion;
-    use codex_core::AuthManager;
-    use codex_core::ThreadManager;
     use codex_core::config::ConfigBuilder;
     use futures::SinkExt;
     use futures::StreamExt;
@@ -1052,7 +1000,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn shared_thread_manager_tracks_threads_started_via_app_server() {
+    async fn threads_started_via_app_server_are_visible_through_typed_requests() {
         let client = start_test_client(SessionSource::Cli).await;
 
         let response: ThreadStartResponse = client
@@ -1065,17 +1013,19 @@ 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 read = client
+            .request_typed::<codex_app_server_protocol::ThreadReadResponse>(
+                ClientRequest::ThreadRead {
+                    request_id: RequestId::Integer(4),
+                    params: codex_app_server_protocol::ThreadReadParams {
+                        thread_id: response.thread.id.clone(),
+                        include_turns: false,
+                    },
+                },
+            )
+            .await
+            .expect("thread/read should return the newly started thread");
+        assert_eq!(read.thread.id, response.thread.id);
 
         client.shutdown().await.expect("shutdown should complete");
     }
@@ -1472,22 +1422,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(Feature::DefaultModeRequestUserInput),
-            },
-        ));
         event_tx
             .send(InProcessServerEvent::Lagged { skipped: 3 })
             .await
@@ -1498,8 +1432,6 @@ mod tests {
             command_tx,
             event_rx,
             worker_handle,
-            auth_manager,
-            thread_manager,
         };
 
         let event = timeout(Duration::from_secs(2), client.next_event())
@@ -1530,37 +1462,38 @@ mod tests {
                 )
             )
         ));
-        assert!(event_requires_delivery(
-            &InProcessServerEvent::LegacyNotification(
-                codex_app_server_protocol::JSONRPCNotification {
-                    method: "codex/event/turn_aborted".to_string(),
-                    params: None,
-                }
-            )
-        ));
         assert!(!event_requires_delivery(&InProcessServerEvent::Lagged {
             skipped: 1
         }));
     }
 
     #[tokio::test]
-    async fn accessors_expose_retained_shared_managers() {
-        let client = start_test_client(SessionSource::Cli).await;
+    async fn runtime_start_args_leave_manager_bootstrap_to_app_server() {
+        let config = Arc::new(build_test_config().await);
 
-        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 runtime_args = InProcessClientStartArgs {
+            arg0_paths: Arg0DispatchPaths::default(),
+            config: config.clone(),
+            cli_overrides: Vec::new(),
+            loader_overrides: LoaderOverrides::default(),
+            cloud_requirements: CloudRequirementsLoader::default(),
+            feedback: CodexFeedback::new(),
+            config_warnings: Vec::new(),
+            session_source: SessionSource::Exec,
+            enable_codex_api_key_env: false,
+            client_name: "codex-app-server-client-test".to_string(),
+            client_version: "0.0.0-test".to_string(),
+            experimental_api: true,
+            opt_out_notification_methods: Vec::new(),
+            channel_capacity: DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
+        }
+        .into_runtime_start_args();
 
-        client.shutdown().await.expect("shutdown should complete");
+        assert_eq!(runtime_args.config, config);
     }
 
     #[tokio::test]
-    async fn shutdown_completes_promptly_with_retained_shared_managers() {
+    async fn shutdown_completes_promptly_without_retained_managers() {
         let client = start_test_client(SessionSource::Cli).await;
 
         timeout(Duration::from_secs(1), client.shutdown())