Merge remote-tracking branch 'origin/main' into jif/pr-27179-cleanup

# Conflicts:
#	codex-rs/tui/src/app_server_session.rs

codex-rs/app-server/README.md

@@ -151,18 +151,21 @@ Example with notification opt-out:
 - `thread/settings/updated` — experimental notification emitted to subscribed clients when a loaded thread’s effective next-turn settings change; includes `threadId` and the full `threadSettings`.
 - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
 - `thread/archive` — move a thread’s rollout file into the archived directory and attempt to move any spawned descendant thread rollout files; returns `{}` on success and emits `thread/archived` for each archived thread.
+- `thread/delete` — hard-delete an active or archived thread and any spawned descendant threads; returns `{}` on success and emits `thread/deleted` for each deleted thread.
 - `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server keeps the thread loaded and unloads it only after it has had no subscribers and no thread activity for 30 minutes, then emits `thread/closed`.
 - `thread/name/set` — set or update a thread’s user-facing name for either a loaded thread or a persisted rollout; returns `{}` on success and emits `thread/name/updated` to initialized, opted-in clients. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
 - `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
 - `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
 - `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream.
 - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
+- `thread/backgroundTerminals/list` — list running background terminals for a loaded thread (experimental; requires `capabilities.experimentalApi`); returns `data` with the running terminal ids.
+- `thread/backgroundTerminals/terminate` — terminate one running background terminal by app-server `processId` (experimental; requires `capabilities.experimentalApi`); returns whether a process was terminated.
 - `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
 - `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
 - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success.
 - `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Review and manual compaction turns reject `turn/steer`.
 - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`.
-- `thread/realtime/start` — start a thread-scoped realtime session (experimental); pass `outputModality: "text"` or `outputModality: "audio"` to choose model output, returns `{}` and streams `thread/realtime/*` notifications. Omit `transport` for the websocket transport, or pass `{ "type": "webrtc", "sdp": "..." }` to create a WebRTC session from a browser-generated SDP offer; the remote answer SDP is emitted as `thread/realtime/sdp`.
+- `thread/realtime/start` — start a thread-scoped realtime session (experimental); pass `outputModality: "text"` or `outputModality: "audio"` to choose model output, and optionally pass `model` and `version` to override configured realtime selection for this session only. Returns `{}` and streams `thread/realtime/*` notifications. Omit `transport` for the websocket transport, or pass `{ "type": "webrtc", "sdp": "..." }` to create a WebRTC session from a browser-generated SDP offer; the remote answer SDP is emitted as `thread/realtime/sdp`.
 - `thread/realtime/appendAudio` — append an input audio chunk to the active realtime session (experimental); returns `{}`.
 - `thread/realtime/appendText` — append text input to the active realtime session (experimental); returns `{}`.
 - `thread/realtime/stop` — stop the active realtime session for the thread (experimental); returns `{}`.
@@ -203,7 +206,7 @@ Example with notification opt-out:
 - `marketplace/upgrade` — upgrade all configured Git plugin marketplaces, or one named marketplace when `marketplaceName` is provided. Returns selected marketplace names, upgraded roots, and per-marketplace errors.
 - `plugin/list` — list discovered plugin marketplaces and plugin state, including effective marketplace install/auth policy metadata, plugin `availability` (`AVAILABLE` by default or `DISABLED_BY_ADMIN` for remote plugins blocked upstream), fail-open `marketplaceLoadErrors` entries for marketplace files that could not be parsed or loaded, and best-effort `featuredPluginIds` for the official curated marketplace. `interface.category` uses the marketplace category when present; otherwise it falls back to the plugin manifest category (**under development; do not call from production clients yet**).
 - `plugin/installed` — list installed plugin rows plus any explicitly requested local install-suggestion plugin names, without fetching the broader remote catalog. Mention surfaces can use this narrower view when they need plugin mention payloads rather than plugin-page discovery data (**under development; do not call from production clients yet**).
-- `plugin/read` — read one plugin by `marketplacePath` plus `pluginName`, returning marketplace info, a list-style `summary`, manifest descriptions/interface metadata, and bundled skills/hooks/apps/MCP server names. Returned plugin skills include their current `enabled` state after local config filtering; bundled hooks are returned as lightweight declaration summaries keyed for correlation with `hooks/list`. Plugin app summaries also include `needsAuth` when the server can determine connector accessibility (**under development; do not call from production clients yet**).
+- `plugin/read` — read one plugin by `marketplacePath` plus `pluginName`, returning marketplace info, a list-style `summary`, manifest descriptions/interface metadata, and bundled skills/hooks/apps/MCP server names. Returned plugin skills include their current `enabled` state after local config filtering; bundled hooks are returned as lightweight declaration summaries keyed for correlation with `hooks/list`. Use `plugin/install`'s `appsNeedingAuth` to drive post-install authentication and `app/list`'s `isAccessible` to determine current connector accessibility (**under development; do not call from production clients yet**).
 - `plugin/skill/read` — read remote plugin skill markdown on demand by `remoteMarketplaceName`, `remotePluginId`, and `skillName`. This lets clients preview uninstalled remote plugin skills without downloading the plugin bundle.
 - `skills/changed` — notification emitted when watched local skill files change.
 - `app/list` — list available apps.
@@ -610,6 +613,16 @@ Use `thread/archive` to move the persisted rollout (stored as a JSONL file on di
 
 An archived thread will not appear in `thread/list` unless `archived` is set to `true`.
 
+### Example: Delete a thread
+
+Use `thread/delete` to hard-delete a thread and its spawned descendant threads. Existing rollout files and associated metadata must be removed before the request succeeds; missing rollout files are treated as already deleted.
+
+```json
+{ "method": "thread/delete", "id": 23, "params": { "threadId": "thr_b" } }
+{ "id": 23, "result": {} }
+{ "method": "thread/deleted", "params": { "threadId": "thr_b" } }
+```
+
 ### Example: Unarchive a thread
 
 Use `thread/unarchive` to move an archived rollout back into the sessions directory.
@@ -837,6 +850,8 @@ Then send `offer.sdp` to app-server. Core uses `experimental_realtime_ws_backend
 
 Omit `prompt` to use Codex's default realtime backend prompt. Send `prompt: null` or
 `prompt: ""` when the session should start without that default backend prompt.
+Clients may also pass `model` and `version` on `thread/realtime/start` to select a
+different realtime session configuration without changing thread or user config.
 
 ```javascript
 await pc.setRemoteDescription({
@@ -870,6 +885,32 @@ Use `thread/backgroundTerminals/clean` to terminate all running background termi
 { "id": 35, "result": {} }
 ```
 
+### Example: List and terminate background terminals
+
+Use `thread/backgroundTerminals/list` to inspect running background terminals associated with a loaded thread. The `backgroundTerminals` segment intentionally follows the existing `thread/backgroundTerminals/clean` method. The returned `processId` is the app-server process id; host OS metadata is nullable. The request accepts the standard `cursor` and `limit` pagination fields. When `nextCursor` is non-null, pass it as `cursor` to fetch the next page.
+
+```json
+{ "method": "thread/backgroundTerminals/list", "id": 36, "params": { "threadId": "thr_123" } }
+{ "id": 36, "result": { "data": [
+    {
+        "itemId": "item_456",
+        "processId": "42",
+        "command": "python3 -m http.server",
+        "cwd": "/workspace",
+        "osPid": null,
+        "cpuPercent": null,
+        "rssKb": null
+    }
+], "nextCursor": null } }
+```
+
+Use `thread/backgroundTerminals/terminate` to terminate one running background terminal by that `processId`.
+
+```json
+{ "method": "thread/backgroundTerminals/terminate", "id": 37, "params": { "threadId": "thr_123", "processId": "42" } }
+{ "id": 37, "result": { "terminated": true } }
+```
+
 ### Example: Steer an active turn
 
 Use `turn/steer` to append additional user input to the currently active regular turn. This does

codex-rs/app-server/src/config/external_agent_config.rs

@@ -1147,7 +1147,7 @@ fn configured_marketplace_plugins(
 ) -> io::Result<BTreeMap<String, HashSet<String>>> {
     let plugins_input = config.plugins_config_input();
     let marketplaces = plugins_manager
-        .list_marketplaces_for_config(&plugins_input, &[])
+        .list_marketplaces_for_config(&plugins_input, &[], /*include_openai_curated*/ true)
         .map_err(|err| {
             invalid_data_error(format!("failed to list configured marketplaces: {err}"))
         })?;

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

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::sync::Weak;
 
+use codex_analytics::AnalyticsEventsClient;
 use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ThreadGoal;
 use codex_app_server_protocol::ThreadGoalUpdatedNotification;
@@ -20,28 +21,47 @@ use codex_protocol::error::CodexErr;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
 use codex_rollout::state_db::StateDbHandle;
+use codex_thread_store::ThreadStore;
 
 use crate::outgoing_message::OutgoingMessageSender;
 use crate::thread_state::ThreadListenerCommand;
 use crate::thread_state::ThreadStateManager;
 
+pub(crate) struct ThreadExtensionDependencies {
+    pub(crate) event_sink: Arc<dyn ExtensionEventSink>,
+    pub(crate) auth_manager: Arc<AuthManager>,
+    pub(crate) state_db: Option<StateDbHandle>,
+    pub(crate) analytics_events_client: AnalyticsEventsClient,
+    pub(crate) thread_manager: Weak<ThreadManager>,
+    pub(crate) goal_service: Arc<GoalService>,
+    pub(crate) executor_skill_provider: Arc<dyn codex_skills_extension::SkillProvider>,
+    /// Process-scoped persistence backend for extensions that need stored thread history.
+    pub(crate) thread_store: Arc<dyn ThreadStore>,
+}
+
 pub(crate) fn thread_extensions<S>(
     guardian_agent_spawner: S,
-    event_sink: Arc<dyn ExtensionEventSink>,
-    auth_manager: Arc<AuthManager>,
-    state_db: Option<StateDbHandle>,
-    thread_manager: Weak<ThreadManager>,
-    goal_service: Arc<GoalService>,
-    executor_skill_provider: Arc<dyn codex_skills_extension::SkillProvider>,
+    dependencies: ThreadExtensionDependencies,
 ) -> Arc<ExtensionRegistry<Config>>
 where
     S: AgentSpawner<StartThreadOptions, Spawned = NewThread, Error = CodexErr> + 'static,
 {
+    let ThreadExtensionDependencies {
+        event_sink,
+        auth_manager,
+        state_db,
+        analytics_events_client,
+        thread_manager,
+        goal_service,
+        executor_skill_provider,
+        thread_store: _thread_store,
+    } = dependencies;
     let mut builder = ExtensionRegistryBuilder::<Config>::with_event_sink(event_sink);
     if let Some(state_db) = state_db {
         codex_goal_extension::install_with_backend(
             &mut builder,
             state_db,
+            analytics_events_client,
             codex_otel::global(),
             thread_manager,
             goal_service,
@@ -56,7 +76,14 @@ where
     codex_skills_extension::install_with_providers(
         &mut builder,
         codex_skills_extension::SkillProviders::new()
-            .with_executor_provider(executor_skill_provider),
+            .with_executor_provider(executor_skill_provider)
+            .with_orchestrator_provider(Arc::new(
+                codex_skills_extension::OrchestratorSkillProvider::new(),
+            )),
+        |config: &Config| codex_skills_extension::SkillsExtensionConfig {
+            include_instructions: config.include_skill_instructions,
+            bundled_skills_enabled: config.bundled_skills_enabled(),
+        },
     );
     Arc::new(builder.build())
 }
@@ -140,7 +167,6 @@ pub(crate) fn guardian_agent_spawner(
 mod tests {
     use std::time::Duration;
 
-    use codex_analytics::AnalyticsEventsClient;
     use codex_protocol::protocol::ThreadGoal as CoreThreadGoal;
     use codex_protocol::protocol::ThreadGoalStatus;
     use codex_protocol::protocol::ThreadGoalUpdatedEvent;

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

@@ -104,7 +104,9 @@ type PendingClientRequestResponse = std::result::Result<Result, JSONRPCErrorErro
 fn server_notification_requires_delivery(notification: &ServerNotification) -> bool {
     matches!(
         notification,
-        ServerNotification::TurnCompleted(_) | ServerNotification::ThreadSettingsUpdated(_)
+        ServerNotification::TurnCompleted(_)
+            | ServerNotification::ThreadSettingsUpdated(_)
+            | ServerNotification::ExternalAgentConfigImportCompleted(_)
     )
 }
 
@@ -729,6 +731,7 @@ mod tests {
     use super::*;
     use codex_app_server_protocol::ClientInfo;
     use codex_app_server_protocol::ConfigRequirementsReadResponse;
+    use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification;
     use codex_app_server_protocol::SessionSource as ApiSessionSource;
     use codex_app_server_protocol::ThreadStartParams;
     use codex_app_server_protocol::ThreadStartResponse;
@@ -893,5 +896,10 @@ mod tests {
                 },
             })
         ));
+        assert!(server_notification_requires_delivery(
+            &ServerNotification::ExternalAgentConfigImportCompleted(
+                ExternalAgentConfigImportCompletedNotification {},
+            )
+        ));
     }
 }

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

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 #![deny(clippy::print_stdout, clippy::print_stderr)]
 
 use codex_arg0::Arg0DispatchPaths;
@@ -74,6 +75,8 @@ use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::registry::Registry;
 use tracing_subscriber::util::SubscriberInitExt;
 
+const SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY: &str = "Codex rebuilt its local database.";
+
 mod analytics_utils;
 mod app_server_tracing;
 mod attestation;
@@ -533,8 +536,8 @@ pub async fn run_main_with_transport_options(
         }
         _ => None,
     };
-    let state_db = match rollout_state_db::try_init(&config).await {
-        Ok(state_db) => Some(state_db),
+    let state_db_init = match init_sqlite_state_db_with_fresh_start_on_corruption(&config).await {
+        Ok(state_db_init) => state_db_init,
         Err(err) => {
             return Err(std::io::Error::other(format!(
                 "failed to initialize sqlite state runtime under {}: {err}",
@@ -542,6 +545,15 @@ pub async fn run_main_with_transport_options(
             )));
         }
     };
+    let state_db = state_db_init.state_db;
+    if let Some(recovery_notice) = state_db_init.recovery_notice {
+        config_warnings.push(ConfigWarningNotification {
+            summary: SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY.to_string(),
+            details: Some(recovery_notice.details),
+            path: None,
+            range: None,
+        });
+    }
 
     if should_run_personality_migration {
         let effective_toml = config.config_layer_stack.effective_config();
@@ -1099,6 +1111,121 @@ pub async fn run_main_with_transport_options(
     Ok(())
 }
 
+struct SqliteRecoveryNotice {
+    details: String,
+}
+
+struct RecoveredSqliteDatabase {
+    database_path: String,
+    backup_folder: String,
+}
+
+struct StateDbInitResult {
+    state_db: Option<rollout_state_db::StateDbHandle>,
+    recovery_notice: Option<SqliteRecoveryNotice>,
+}
+
+async fn init_sqlite_state_db_with_fresh_start_on_corruption(
+    config: &Config,
+) -> anyhow::Result<StateDbInitResult> {
+    let mut attempted_backups = HashSet::new();
+    let mut recovered_databases = Vec::new();
+    loop {
+        let err = match rollout_state_db::try_init(config).await {
+            Ok(state_db) => {
+                let recovery_notice = sqlite_recovery_notice(&recovered_databases);
+                if recovery_notice.is_some() {
+                    emit_state_db_backup_warning(SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY);
+                    for recovered_database in &recovered_databases {
+                        emit_state_db_backup_warning(&format!(
+                            "Database path: {}",
+                            recovered_database.database_path
+                        ));
+                        emit_state_db_backup_warning(&format!(
+                            "Backup folder: {}",
+                            recovered_database.backup_folder
+                        ));
+                    }
+                }
+                return Ok(StateDbInitResult {
+                    state_db: Some(state_db),
+                    recovery_notice,
+                });
+            }
+            Err(err) => err,
+        };
+        if !codex_state::is_sqlite_corruption_error(&err) {
+            return Err(err);
+        }
+
+        let database_path = codex_state::runtime_db_path_for_corruption_error(&err)
+            .unwrap_or_else(|| codex_state::state_db_path(config.sqlite_home.as_path()));
+        if !attempted_backups.insert(database_path.clone()) {
+            return Err(anyhow::anyhow!(
+                "failed to initialize sqlite state runtime after moving damaged database file into a backup folder: {err}"
+            ));
+        }
+
+        let original_error = err.to_string();
+        emit_state_db_backup_warning(&format!(
+            "Codex local database at {} appears damaged. Moving it into a backup folder so the app server can rebuild it from saved data.",
+            database_path.display()
+        ));
+        let backups = codex_state::backup_runtime_db_for_fresh_start(database_path.as_path())
+            .await
+            .map_err(|backup_err| {
+                anyhow::anyhow!(
+                    "failed to move damaged sqlite state database files into a backup folder: {backup_err}; original error: {original_error}"
+                )
+            })?;
+        for backup in &backups {
+            emit_state_db_backup_warning(&format!(
+                "Moved damaged Codex local database file {} to {}",
+                backup.original_path.display(),
+                backup.backup_path.display()
+            ));
+        }
+        if let Some(first_backup) = backups.first()
+            && let Some(backup_folder) = first_backup.backup_path.parent()
+        {
+            recovered_databases.push(RecoveredSqliteDatabase {
+                database_path: first_backup.original_path.display().to_string(),
+                backup_folder: backup_folder.display().to_string(),
+            });
+        }
+    }
+}
+
+fn sqlite_recovery_notice(
+    recovered_databases: &[RecoveredSqliteDatabase],
+) -> Option<SqliteRecoveryNotice> {
+    if recovered_databases.is_empty() {
+        return None;
+    }
+
+    let details = recovered_databases
+        .iter()
+        .map(|recovered_database| {
+            format!(
+                "Database path: {}\nBackup folder: {}",
+                recovered_database.database_path, recovered_database.backup_folder
+            )
+        })
+        .collect::<Vec<_>>()
+        .join("\n\n");
+    Some(SqliteRecoveryNotice { details })
+}
+
+fn emit_state_db_backup_warning(message: &str) {
+    warn!("{message}");
+    if !tracing::dispatcher::has_been_set() {
+        #[allow(clippy::print_stderr)]
+        {
+            eprintln!("{message}");
+        }
+    }
+}
+
 fn analytics_rpc_transport(transport: &AppServerTransport) -> AppServerRpcTransport {
     match transport {
         AppServerTransport::Stdio => AppServerRpcTransport::Stdio,

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

@@ -96,6 +96,7 @@ async fn queue_refresh(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::extensions::ThreadExtensionDependencies;
     use crate::extensions::guardian_agent_spawner;
     use crate::extensions::thread_extensions;
     use async_trait::async_trait;
@@ -193,12 +194,16 @@ mod tests {
                 Arc::clone(&environment_manager),
                 thread_extensions(
                     guardian_agent_spawner(thread_manager.clone()),
-                    Arc::new(NoopExtensionEventSink),
-                    auth_manager.clone(),
-                    Some(state_db.clone()),
-                    thread_manager.clone(),
-                    Arc::new(codex_goal_extension::GoalService::new()),
-                    Arc::clone(&executor_skill_provider),
+                    ThreadExtensionDependencies {
+                        event_sink: Arc::new(NoopExtensionEventSink),
+                        auth_manager: auth_manager.clone(),
+                        state_db: Some(state_db.clone()),
+                        analytics_events_client: codex_analytics::AnalyticsEventsClient::disabled(),
+                        thread_manager: thread_manager.clone(),
+                        goal_service: Arc::new(codex_goal_extension::GoalService::new()),
+                        executor_skill_provider: Arc::clone(&executor_skill_provider),
+                        thread_store: Arc::clone(&thread_store),
+                    },
                 ),
                 /*analytics_events_client*/ None,
                 Arc::clone(&thread_store),

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

@@ -8,6 +8,7 @@ use crate::attestation::app_server_attestation_provider;
 use crate::config_manager::ConfigManager;
 use crate::connection_rpc_gate::ConnectionRpcGate;
 use crate::error_code::invalid_request;
+use crate::extensions::ThreadExtensionDependencies;
 use crate::extensions::app_server_extension_event_sink;
 use crate::extensions::guardian_agent_spawner;
 use crate::extensions::thread_extensions;
@@ -324,12 +325,19 @@ impl MessageProcessor {
                 environment_manager,
                 thread_extensions(
                     guardian_agent_spawner(thread_manager.clone()),
-                    app_server_extension_event_sink(outgoing.clone(), thread_state_manager.clone()),
-                    auth_manager.clone(),
-                    state_db.clone(),
-                    thread_manager.clone(),
-                    Arc::clone(&goal_service),
-                    Arc::clone(&executor_skill_provider),
+                    ThreadExtensionDependencies {
+                        event_sink: app_server_extension_event_sink(
+                            outgoing.clone(),
+                            thread_state_manager.clone(),
+                        ),
+                        auth_manager: auth_manager.clone(),
+                        state_db: state_db.clone(),
+                        analytics_events_client: analytics_events_client.clone(),
+                        thread_manager: thread_manager.clone(),
+                        goal_service: Arc::clone(&goal_service),
+                        executor_skill_provider: Arc::clone(&executor_skill_provider),
+                        thread_store: Arc::clone(&thread_store),
+                    },
                 ),
                 Some(analytics_events_client.clone()),
                 Arc::clone(&thread_store),
@@ -393,7 +401,7 @@ impl MessageProcessor {
             Arc::clone(&thread_manager),
             Arc::clone(&config),
             feedback,
-            log_db,
+            log_db.clone(),
             state_db.clone(),
         );
         let git_processor = GitRequestProcessor::new();
@@ -447,6 +455,7 @@ impl MessageProcessor {
             Arc::clone(&thread_list_state_permit),
             thread_goal_processor.clone(),
             state_db,
+            log_db,
             Arc::clone(&skills_watcher),
         );
         let turn_processor = TurnRequestProcessor::new(
@@ -1064,6 +1073,11 @@ impl MessageProcessor {
                     .thread_archive(request_id.clone(), params)
                     .await
             }
+            ClientRequest::ThreadDelete { params, .. } => {
+                self.thread_processor
+                    .thread_delete(request_id.clone(), params)
+                    .await
+            }
             ClientRequest::ThreadIncrementElicitation { params, .. } => {
                 self.thread_processor
                     .thread_increment_elicitation(params)
@@ -1119,6 +1133,16 @@ impl MessageProcessor {
                     .thread_background_terminals_clean(&request_id, params)
                     .await
             }
+            ClientRequest::ThreadBackgroundTerminalsList { params, .. } => {
+                self.thread_processor
+                    .thread_background_terminals_list(params)
+                    .await
+            }
+            ClientRequest::ThreadBackgroundTerminalsTerminate { params, .. } => {
+                self.thread_processor
+                    .thread_background_terminals_terminate(params)
+                    .await
+            }
             ClientRequest::ThreadRollback { params, .. } => {
                 self.thread_processor
                     .thread_rollback(&request_id, params)

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

@@ -171,13 +171,21 @@ use codex_app_server_protocol::ThreadApproveGuardianDeniedActionResponse;
 use codex_app_server_protocol::ThreadArchiveParams;
 use codex_app_server_protocol::ThreadArchiveResponse;
 use codex_app_server_protocol::ThreadArchivedNotification;
+use codex_app_server_protocol::ThreadBackgroundTerminal;
 use codex_app_server_protocol::ThreadBackgroundTerminalsCleanParams;
 use codex_app_server_protocol::ThreadBackgroundTerminalsCleanResponse;
+use codex_app_server_protocol::ThreadBackgroundTerminalsListParams;
+use codex_app_server_protocol::ThreadBackgroundTerminalsListResponse;
+use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateParams;
+use codex_app_server_protocol::ThreadBackgroundTerminalsTerminateResponse;
 use codex_app_server_protocol::ThreadClosedNotification;
 use codex_app_server_protocol::ThreadCompactStartParams;
 use codex_app_server_protocol::ThreadCompactStartResponse;
 use codex_app_server_protocol::ThreadDecrementElicitationParams;
 use codex_app_server_protocol::ThreadDecrementElicitationResponse;
+use codex_app_server_protocol::ThreadDeleteParams;
+use codex_app_server_protocol::ThreadDeleteResponse;
+use codex_app_server_protocol::ThreadDeletedNotification;
 use codex_app_server_protocol::ThreadForkParams;
 use codex_app_server_protocol::ThreadForkResponse;
 use codex_app_server_protocol::ThreadGoal;
@@ -416,6 +424,7 @@ use codex_rollout::state_db::reconcile_rollout;
 use codex_state::ThreadMetadata;
 use codex_state::log_db::LogDbLayer;
 use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams;
+use codex_thread_store::DeleteThreadParams as StoreDeleteThreadParams;
 use codex_thread_store::GitInfoPatch as StoreGitInfoPatch;
 use codex_thread_store::ListThreadsParams as StoreListThreadsParams;
 use codex_thread_store::LocalThreadStore;
@@ -536,6 +545,7 @@ fn resolve_runtime_workspace_roots(workspace_roots: Vec<AbsolutePathBuf>) -> Vec
 
 mod config_errors;
 mod request_errors;
+mod thread_delete;
 mod thread_goal_processor;
 mod thread_lifecycle;
 mod thread_resume_redaction;

codex-rs/app-server/src/request_processors/account_processor.rs

@@ -158,6 +158,9 @@ impl AccountRequestProcessor {
 
     pub(crate) fn clear_external_auth(&self) {
         self.auth_manager.clear_external_auth();
+        self.thread_manager
+            .plugins_manager()
+            .set_auth_mode(self.auth_manager.get_api_auth_mode());
     }
 
     fn current_account_updated_notification(&self) -> AccountUpdatedNotification {
@@ -173,6 +176,10 @@ impl AccountRequestProcessor {
         thread_manager: &Arc<ThreadManager>,
         auth: Option<CodexAuth>,
     ) {
+        thread_manager
+            .plugins_manager()
+            .set_auth_mode(auth.as_ref().map(CodexAuth::api_auth_mode));
+
         match config_manager
             .load_latest_config(/*fallback_cwd*/ None)
             .await

codex-rs/app-server/src/request_processors/plugins.rs

@@ -573,6 +573,7 @@ impl PluginRequestProcessor {
                     .list_marketplaces_for_config(
                         &config_for_marketplace_listing,
                         &roots_for_marketplace_listing,
+                        /*include_openai_curated*/ true,
                     )?;
                 Ok::<
                     (
@@ -837,8 +838,11 @@ impl PluginRequestProcessor {
         let config_for_marketplace_listing = plugins_input.clone();
         let shared_plugin_ids_by_local_path = load_shared_plugin_ids_by_local_path(config)?;
         match tokio::task::spawn_blocking(move || {
-            let outcome = plugins_manager
-                .list_marketplaces_for_config(&config_for_marketplace_listing, &roots)?;
+            let outcome = plugins_manager.list_marketplaces_for_config(
+                &config_for_marketplace_listing,
+                &roots,
+                /*include_openai_curated*/ true,
+            )?;
             Ok::<
                 (
                     Vec<PluginMarketplaceEntry>,
@@ -1031,14 +1035,7 @@ impl PluginRequestProcessor {
                     }
                     None => None,
                 };
-                let environment_manager = self.thread_manager.environment_manager();
-                let app_summaries = load_plugin_app_summaries(
-                    &config,
-                    &outcome.plugin.apps,
-                    Arc::clone(&environment_manager),
-                    self.thread_manager.mcp_manager(),
-                )
-                .await;
+                let app_summaries = load_plugin_app_summaries(&config, &outcome.plugin.apps).await;
                 let visible_skills = outcome
                     .plugin
                     .skills
@@ -1114,14 +1111,7 @@ impl PluginRequestProcessor {
                     .cloned()
                     .map(codex_plugin::AppConnectorId)
                     .collect::<Vec<_>>();
-                let environment_manager = self.thread_manager.environment_manager();
-                let app_summaries = load_plugin_app_summaries(
-                    &config,
-                    &plugin_apps,
-                    Arc::clone(&environment_manager),
-                    self.thread_manager.mcp_manager(),
-                )
-                .await;
+                let app_summaries = load_plugin_app_summaries(&config, &plugin_apps).await;
                 remote_plugin_detail_to_info(remote_detail, app_summaries)
             }
         };
@@ -1578,7 +1568,6 @@ impl PluginRequestProcessor {
                             name: connector.name,
                             description: connector.description,
                             install_url: connector.install_url,
-                            needs_auth: true,
                         })
                         .collect()
                 }
@@ -1883,8 +1872,6 @@ impl PluginRequestProcessor {
 async fn load_plugin_app_summaries(
     config: &Config,
     plugin_apps: &[codex_plugin::AppConnectorId],
-    environment_manager: Arc<EnvironmentManager>,
-    mcp_manager: Arc<McpManager>,
 ) -> Vec<AppSummary> {
     if plugin_apps.is_empty() {
         return Vec::new();
@@ -1902,49 +1889,9 @@ async fn load_plugin_app_summaries(
         };
 
     let plugin_connectors = connectors::connectors_for_plugin_apps(connectors, plugin_apps);
-
-    let accessible_connectors =
-        match connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager(
-            config,
-            /*force_refetch*/ false,
-            environment_manager,
-            mcp_manager,
-        )
-        .await
-        {
-            Ok(status) if status.codex_apps_ready => status.connectors,
-            Ok(_) => {
-                return plugin_connectors
-                    .into_iter()
-                    .map(AppSummary::from)
-                    .collect();
-            }
-            Err(err) => {
-                warn!("failed to load app auth state for plugin/read: {err:#}");
-                return plugin_connectors
-                    .into_iter()
-                    .map(AppSummary::from)
-                    .collect();
-            }
-        };
-
-    let accessible_ids = accessible_connectors
-        .iter()
-        .map(|connector| connector.id.as_str())
-        .collect::<HashSet<_>>();
-
     plugin_connectors
         .into_iter()
-        .map(|connector| {
-            let needs_auth = !accessible_ids.contains(connector.id.as_str());
-            AppSummary {
-                id: connector.id,
-                name: connector.name,
-                description: connector.description,
-                install_url: connector.install_url,
-                needs_auth,
-            }
-        })
+        .map(AppSummary::from)
         .collect()
 }
 
@@ -1979,7 +1926,6 @@ fn plugin_apps_needing_auth(
             name: connector.name,
             description: connector.description,
             install_url: connector.install_url,
-            needs_auth: true,
         })
         .collect()
 }

codex-rs/app-server/src/request_processors/thread_delete.rs

@@ -0,0 +1,188 @@
+//! `thread/delete` request handling.
+
+use super::thread_processor::core_thread_write_error;
+use super::thread_processor::unsupported_thread_store_operation;
+use super::*;
+
+impl ThreadRequestProcessor {
+    pub(crate) async fn thread_delete(
+        &self,
+        request_id: ConnectionRequestId,
+        params: ThreadDeleteParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        let mut deleted_thread_ids = Vec::new();
+        let result = {
+            let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
+            self.thread_delete_response(params, &mut deleted_thread_ids)
+                .await
+        };
+        match result {
+            Ok(response) => {
+                self.outgoing
+                    .send_response(request_id.clone(), response)
+                    .await;
+                self.send_thread_deleted_notifications(deleted_thread_ids)
+                    .await;
+                Ok(None)
+            }
+            Err(error) => Err(error),
+        }
+    }
+
+    async fn thread_delete_response(
+        &self,
+        params: ThreadDeleteParams,
+        deleted_thread_ids: &mut Vec<String>,
+    ) -> Result<ThreadDeleteResponse, JSONRPCErrorError> {
+        let thread_id = ThreadId::from_string(&params.thread_id)
+            .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?;
+
+        let mut thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?;
+        let mut seen = thread_ids.iter().copied().collect::<HashSet<_>>();
+
+        match self
+            .thread_manager
+            .list_agent_subtree_thread_ids(thread_id)
+            .await
+        {
+            Ok(live_thread_ids) => {
+                for live_thread_id in live_thread_ids {
+                    if seen.insert(live_thread_id) {
+                        thread_ids.push(live_thread_id);
+                    }
+                }
+            }
+            Err(err) => return Err(core_thread_write_error("delete thread", err)),
+        }
+
+        self.validate_root_thread_delete(thread_id, thread_ids.len() > 1)
+            .await?;
+        for thread_id_to_delete in thread_ids.iter().copied() {
+            self.prepare_thread_for_delete(thread_id_to_delete).await;
+        }
+
+        let mut delete_order: Vec<_> = thread_ids.iter().skip(1).rev().copied().collect();
+        delete_order.push(thread_id);
+
+        for thread_id_to_delete in delete_order.iter().copied() {
+            match self
+                .thread_store
+                .delete_thread(StoreDeleteThreadParams {
+                    thread_id: thread_id_to_delete,
+                })
+                .await
+            {
+                Ok(()) => {}
+                Err(ThreadStoreError::ThreadNotFound { .. }) => {
+                    warn!(
+                        "thread {thread_id_to_delete} was already missing while deleting {thread_id}"
+                    );
+                }
+                Err(err) => {
+                    return Err(thread_store_delete_error(err));
+                }
+            }
+        }
+
+        if let Some(state_db) = self.state_db.as_ref() {
+            state_db
+                .delete_threads_strict(thread_ids.as_slice())
+                .await
+                .map_err(|err| {
+                    internal_error(format!(
+                        "failed to delete app-server state for {thread_id}: {err}"
+                    ))
+                })?;
+        }
+
+        deleted_thread_ids.extend(
+            delete_order
+                .into_iter()
+                .map(|thread_id| thread_id.to_string()),
+        );
+        Ok(ThreadDeleteResponse {})
+    }
+
+    async fn send_thread_deleted_notifications(&self, deleted_thread_ids: Vec<String>) {
+        for thread_id in deleted_thread_ids {
+            self.outgoing
+                .send_server_notification(ServerNotification::ThreadDeleted(
+                    ThreadDeletedNotification { thread_id },
+                ))
+                .await;
+        }
+    }
+
+    async fn validate_root_thread_delete(
+        &self,
+        thread_id: ThreadId,
+        has_descendants: bool,
+    ) -> Result<(), JSONRPCErrorError> {
+        if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
+            if !thread.config_snapshot().await.ephemeral {
+                return Ok(());
+            }
+            return Err(invalid_request(format!(
+                "thread is not persisted and cannot be deleted: {thread_id}"
+            )));
+        }
+        match self
+            .thread_store
+            .read_thread(StoreReadThreadParams {
+                thread_id,
+                include_archived: true,
+                include_history: false,
+            })
+            .await
+        {
+            Ok(_) => Ok(()),
+            Err(ThreadStoreError::ThreadNotFound { .. }) => {
+                if has_descendants {
+                    return Ok(());
+                }
+                let Some(state_db) = self.state_db.as_ref() else {
+                    return Err(thread_store_delete_error(
+                        ThreadStoreError::ThreadNotFound { thread_id },
+                    ));
+                };
+                if state_db
+                    .get_thread(thread_id)
+                    .await
+                    .map_err(|err| {
+                        internal_error(format!(
+                            "failed to read app-server state for {thread_id}: {err}"
+                        ))
+                    })?
+                    .is_some()
+                {
+                    Ok(())
+                } else {
+                    Err(thread_store_delete_error(
+                        ThreadStoreError::ThreadNotFound { thread_id },
+                    ))
+                }
+            }
+            Err(err) => Err(thread_store_delete_error(err)),
+        }
+    }
+
+    async fn prepare_thread_for_delete(&self, thread_id: ThreadId) {
+        self.prepare_thread_for_removal(thread_id, "delete").await;
+        if let Some(log_db) = self.log_db.as_ref() {
+            log_db.flush().await;
+        }
+    }
+}
+
+fn thread_store_delete_error(err: ThreadStoreError) -> JSONRPCErrorError {
+    match err {
+        ThreadStoreError::ThreadNotFound { thread_id } => {
+            invalid_request(format!("thread not found: {thread_id}"))
+        }
+        ThreadStoreError::InvalidRequest { message } => invalid_request(message),
+        ThreadStoreError::Unsupported { operation } => {
+            unsupported_thread_store_operation(operation)
+        }
+        err => internal_error(format!("failed to delete thread: {err}")),
+    }
+}

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -325,6 +325,7 @@ pub(crate) struct ThreadRequestProcessor {
     pub(super) thread_list_state_permit: Arc<Semaphore>,
     pub(super) thread_goal_processor: ThreadGoalRequestProcessor,
     pub(super) state_db: Option<StateDbHandle>,
+    pub(super) log_db: Option<LogDbLayer>,
     pub(super) background_tasks: TaskTracker,
     pub(super) skills_watcher: Arc<SkillsWatcher>,
 }
@@ -356,6 +357,7 @@ impl ThreadRequestProcessor {
         thread_list_state_permit: Arc<Semaphore>,
         thread_goal_processor: ThreadGoalRequestProcessor,
         state_db: Option<StateDbHandle>,
+        log_db: Option<LogDbLayer>,
         skills_watcher: Arc<SkillsWatcher>,
     ) -> Self {
         Self {
@@ -372,6 +374,7 @@ impl ThreadRequestProcessor {
             thread_list_state_permit,
             thread_goal_processor,
             state_db,
+            log_db,
             background_tasks: TaskTracker::new(),
             skills_watcher,
         }
@@ -569,6 +572,24 @@ impl ThreadRequestProcessor {
             .map(|response| Some(response.into()))
     }
 
+    pub(crate) async fn thread_background_terminals_list(
+        &self,
+        params: ThreadBackgroundTerminalsListParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.thread_background_terminals_list_inner(params)
+            .await
+            .map(|response| Some(response.into()))
+    }
+
+    pub(crate) async fn thread_background_terminals_terminate(
+        &self,
+        params: ThreadBackgroundTerminalsTerminateParams,
+    ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
+        self.thread_background_terminals_terminate_inner(params)
+            .await
+            .map(|response| Some(response.into()))
+    }
+
     pub(crate) async fn thread_rollback(
         &self,
         request_id: &ConnectionRequestId,
@@ -678,7 +699,7 @@ impl ThreadRequestProcessor {
 
         Ok((thread_id, thread))
     }
-    async fn acquire_thread_list_state_permit(
+    pub(super) async fn acquire_thread_list_state_permit(
         &self,
     ) -> Result<SemaphorePermit<'_>, JSONRPCErrorError> {
         self.thread_list_state_permit
@@ -750,6 +771,10 @@ impl ThreadRequestProcessor {
     }
 
     async fn prepare_thread_for_archive(&self, thread_id: ThreadId) {
+        self.prepare_thread_for_removal(thread_id, "archive").await;
+    }
+
+    pub(super) async fn prepare_thread_for_removal(&self, thread_id: ThreadId, operation: &str) {
         let removed_conversation = self.thread_manager.remove_thread(&thread_id).await;
         if let Some(conversation) = removed_conversation {
             info!("thread {thread_id} was active; shutting down");
@@ -757,11 +782,11 @@ impl ThreadRequestProcessor {
                 ThreadShutdownResult::Complete => {}
                 ThreadShutdownResult::SubmitFailed => {
                     error!(
-                        "failed to submit Shutdown to thread {thread_id}; proceeding with archive"
+                        "failed to submit Shutdown to thread {thread_id}; proceeding with {operation}"
                     );
                 }
                 ThreadShutdownResult::TimedOut => {
-                    warn!("thread {thread_id} shutdown timed out; proceeding with archive");
+                    warn!("thread {thread_id} shutdown timed out; proceeding with {operation}");
                 }
             }
         }
@@ -1294,23 +1319,7 @@ impl ThreadRequestProcessor {
         let thread_id = ThreadId::from_string(&params.thread_id)
             .map_err(|err| invalid_request(format!("invalid session id: {err}")))?;
 
-        let mut thread_ids = vec![thread_id];
-        if let Some(state_db_ctx) = self.state_db.as_ref() {
-            let descendants = state_db_ctx
-                .list_thread_spawn_descendants(thread_id)
-                .await
-                .map_err(|err| {
-                    internal_error(format!(
-                        "failed to list spawned descendants for session {thread_id}: {err}"
-                    ))
-                })?;
-            let mut seen = HashSet::from([thread_id]);
-            for descendant_id in descendants {
-                if seen.insert(descendant_id) {
-                    thread_ids.push(descendant_id);
-                }
-            }
-        }
+        let thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?;
 
         let mut archive_thread_ids = Vec::new();
         match self
@@ -1395,6 +1404,31 @@ impl ThreadRequestProcessor {
         Ok((ThreadArchiveResponse {}, archived_thread_ids))
     }
 
+    pub(super) async fn state_db_spawn_subtree_thread_ids(
+        &self,
+        thread_id: ThreadId,
+    ) -> Result<Vec<ThreadId>, JSONRPCErrorError> {
+        let mut thread_ids = vec![thread_id];
+        let Some(state_db_ctx) = self.state_db.as_ref() else {
+            return Ok(thread_ids);
+        };
+        let mut seen = HashSet::from([thread_id]);
+        let descendants = state_db_ctx
+            .list_thread_spawn_descendants(thread_id)
+            .await
+            .map_err(|err| {
+                internal_error(format!(
+                    "failed to list spawned descendants for thread id {thread_id}: {err}"
+                ))
+            })?;
+        for descendant_id in descendants {
+            if seen.insert(descendant_id) {
+                thread_ids.push(descendant_id);
+            }
+        }
+        Ok(thread_ids)
+    }
+
     async fn thread_increment_elicitation_inner(
         &self,
         params: ThreadIncrementElicitationParams,
@@ -1725,6 +1759,54 @@ impl ThreadRequestProcessor {
         Ok(ThreadBackgroundTerminalsCleanResponse {})
     }
 
+    async fn thread_background_terminals_list_inner(
+        &self,
+        params: ThreadBackgroundTerminalsListParams,
+    ) -> Result<ThreadBackgroundTerminalsListResponse, JSONRPCErrorError> {
+        let ThreadBackgroundTerminalsListParams {
+            thread_id,
+            cursor,
+            limit,
+        } = params;
+
+        let (_, thread) = self.load_thread(&thread_id).await?;
+        let terminals = thread
+            .list_background_terminals()
+            .await
+            .into_iter()
+            .map(|terminal| ThreadBackgroundTerminal {
+                item_id: terminal.item_id,
+                process_id: terminal.process_id,
+                command: terminal.command,
+                cwd: terminal.cwd,
+                os_pid: None,
+                cpu_percent: None,
+                rss_kb: None,
+            })
+            .collect::<Vec<_>>();
+
+        let (data, next_cursor) = paginate_background_terminals(&terminals, cursor, limit)?;
+
+        Ok(ThreadBackgroundTerminalsListResponse { data, next_cursor })
+    }
+
+    async fn thread_background_terminals_terminate_inner(
+        &self,
+        params: ThreadBackgroundTerminalsTerminateParams,
+    ) -> Result<ThreadBackgroundTerminalsTerminateResponse, JSONRPCErrorError> {
+        let ThreadBackgroundTerminalsTerminateParams {
+            thread_id,
+            process_id,
+        } = params;
+        let process_id = process_id.parse::<i32>().map_err(|err| {
+            invalid_request(format!("invalid background terminal process id: {err}"))
+        })?;
+
+        let (_, thread) = self.load_thread(&thread_id).await?;
+        let terminated = thread.terminate_background_terminal(process_id).await;
+        Ok(ThreadBackgroundTerminalsTerminateResponse { terminated })
+    }
+
     async fn thread_shell_command_inner(
         &self,
         request_id: &ConnectionRequestId,
@@ -3851,7 +3933,7 @@ fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError {
     }
 }
 
-fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError {
+pub(super) fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError {
     method_not_found(format!("{operation} is not supported yet"))
 }
 
@@ -3972,7 +4054,7 @@ fn conversation_summary_rollout_path_read_error(
     }
 }
 
-fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPCErrorError {
+pub(super) fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPCErrorError {
     match err {
         CodexErr::ThreadNotFound(thread_id) => {
             invalid_request(format!("thread not found: {thread_id}"))
@@ -4258,6 +4340,34 @@ fn build_thread_from_snapshot(
     }
 }
 
+fn paginate_background_terminals(
+    terminals: &[ThreadBackgroundTerminal],
+    cursor: Option<String>,
+    limit: Option<u32>,
+) -> Result<(Vec<ThreadBackgroundTerminal>, Option<String>), JSONRPCErrorError> {
+    let start = match cursor {
+        Some(cursor) => {
+            let cursor = cursor
+                .parse::<i32>()
+                .map_err(|err| invalid_request(format!("invalid cursor: {err}")))?;
+            terminals
+                .iter()
+                .position(|terminal| {
+                    terminal
+                        .process_id
+                        .parse::<i32>()
+                        .is_ok_and(|process_id| process_id > cursor)
+                })
+                .unwrap_or(terminals.len())
+        }
+        None => 0,
+    };
+    let effective_limit = limit.unwrap_or(terminals.len() as u32).max(1) as usize;
+    let end = start.saturating_add(effective_limit).min(terminals.len());
+    let next_cursor = (end < terminals.len()).then(|| terminals[end - 1].process_id.clone());
+    Ok((terminals[start..end].to_vec(), next_cursor))
+}
+
 fn build_thread_from_loaded_snapshot(
     thread_id: ThreadId,
     config_snapshot: &ThreadConfigSnapshot,

codex-rs/app-server/src/request_processors/thread_processor_tests.rs

@@ -36,6 +36,65 @@ mod thread_list_cwd_filter_tests {
     }
 }
 
+mod background_terminal_pagination_tests {
+    use super::super::paginate_background_terminals;
+    use codex_app_server_protocol::ThreadBackgroundTerminal;
+    use codex_utils_absolute_path::AbsolutePathBuf;
+    use pretty_assertions::assert_eq;
+
+    fn terminal(process_id: &str) -> ThreadBackgroundTerminal {
+        let cwd = if cfg!(windows) { r"C:\tmp" } else { "/tmp" };
+
+        ThreadBackgroundTerminal {
+            item_id: format!("item-{process_id}"),
+            process_id: process_id.to_string(),
+            command: format!("command-{process_id}"),
+            cwd: AbsolutePathBuf::from_absolute_path(cwd).expect("absolute cwd"),
+            os_pid: None,
+            cpu_percent: None,
+            rss_kb: None,
+        }
+    }
+
+    #[test]
+    fn paginates_with_process_id_cursor() {
+        let terminals = vec![
+            terminal("1"),
+            terminal("2"),
+            terminal("3"),
+            terminal("4"),
+            terminal("5"),
+        ];
+
+        let (data, next_cursor) =
+            paginate_background_terminals(&terminals, /*cursor*/ None, Some(2))
+                .expect("valid page");
+
+        assert_eq!(data, vec![terminal("1"), terminal("2")]);
+        assert_eq!(next_cursor, Some("2".to_string()));
+        let first_cursor = next_cursor;
+
+        let terminals_without_anchor = vec![terminal("1"), terminal("3"), terminal("4")];
+        let (data, next_cursor) =
+            paginate_background_terminals(&terminals_without_anchor, first_cursor.clone(), Some(2))
+                .expect("valid page");
+
+        assert_eq!(data, vec![terminal("3"), terminal("4")]);
+        assert_eq!(next_cursor, None);
+
+        let (data, next_cursor) =
+            paginate_background_terminals(&terminals, first_cursor, Some(2)).expect("valid page");
+
+        assert_eq!(data, vec![terminal("3"), terminal("4")]);
+        assert_eq!(next_cursor, Some("4".to_string()));
+
+        assert!(
+            paginate_background_terminals(&terminals, Some("missing".to_string()), Some(1))
+                .is_err()
+        );
+    }
+}
+
 mod thread_processor_behavior_tests {
     async fn forked_from_id_from_rollout(path: &Path) -> Option<String> {
         codex_core::read_session_meta_line(path)

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -934,6 +934,7 @@ impl TurnRequestProcessor {
             request_id,
             thread.as_ref(),
             Op::RealtimeConversationStart(ConversationStartParams {
+                model: params.model,
                 output_modality: params.output_modality,
                 prompt: params.prompt,
                 realtime_session_id: params.realtime_session_id,
@@ -945,6 +946,7 @@ impl TurnRequestProcessor {
                         ConversationStartTransport::Webrtc { sdp }
                     }
                 }),
+                version: params.version,
                 voice: params.voice,
             }),
         )

codex-rs/app-server/tests/common/auth_fixtures.rs

@@ -165,6 +165,7 @@ pub fn write_chatgpt_auth(
         last_refresh,
         agent_identity: None,
         personal_access_token: None,
+        bedrock_api_key: None,
     };
 
     save_auth(codex_home, &auth, cli_auth_credentials_store_mode).context("write auth.json")

codex-rs/app-server/tests/common/models_cache.rs

@@ -47,6 +47,7 @@ fn preset_to_info(preset: &ModelPreset, priority: i32) -> ModelInfo {
         context_window: Some(272_000),
         max_context_window: None,
         auto_compact_token_limit: None,
+        comp_hash: None,
         effective_context_window_percent: 95,
         experimental_supported_tools: Vec::new(),
         input_modalities: default_input_modalities(),

codex-rs/app-server/tests/common/test_app_server.rs

@@ -79,6 +79,7 @@ use codex_app_server_protocol::SkillsExtraRootsSetParams;
 use codex_app_server_protocol::SkillsListParams;
 use codex_app_server_protocol::ThreadArchiveParams;
 use codex_app_server_protocol::ThreadCompactStartParams;
+use codex_app_server_protocol::ThreadDeleteParams;
 use codex_app_server_protocol::ThreadForkParams;
 use codex_app_server_protocol::ThreadInjectItemsParams;
 use codex_app_server_protocol::ThreadListParams;
@@ -456,6 +457,15 @@ impl TestAppServer {
         self.send_request("thread/archive", params).await
     }
 
+    /// Send a `thread/delete` JSON-RPC request.
+    pub async fn send_thread_delete_request(
+        &mut self,
+        params: ThreadDeleteParams,
+    ) -> anyhow::Result<i64> {
+        let params = Some(serde_json::to_value(params)?);
+        self.send_request("thread/delete", params).await
+    }
+
     /// Send a `thread/name/set` JSON-RPC request.
     pub async fn send_thread_set_name_request(
         &mut self,

codex-rs/app-server/tests/suite/v2/analytics.rs

@@ -124,6 +124,31 @@ pub(crate) async fn wait_for_analytics_event(
     server: &MockServer,
     read_timeout: Duration,
     event_type: &str,
+) -> Result<Value> {
+    wait_for_matching_analytics_event(server, read_timeout, |event| {
+        event["event_type"] == event_type
+    })
+    .await
+}
+
+pub(crate) async fn wait_for_goal_event(
+    server: &MockServer,
+    read_timeout: Duration,
+    event_kind: &str,
+    goal_status: &str,
+) -> Result<Value> {
+    wait_for_matching_analytics_event(server, read_timeout, |event| {
+        event["event_type"] == "codex_goal_event"
+            && event["event_params"]["event_kind"] == event_kind
+            && event["event_params"]["goal_status"] == goal_status
+    })
+    .await
+}
+
+async fn wait_for_matching_analytics_event(
+    server: &MockServer,
+    read_timeout: Duration,
+    matches: impl Fn(&Value) -> bool,
 ) -> Result<Value> {
     timeout(read_timeout, async {
         loop {
@@ -142,10 +167,7 @@ pub(crate) async fn wait_for_analytics_event(
                 let Some(events) = payload["events"].as_array() else {
                     continue;
                 };
-                if let Some(event) = events
-                    .iter()
-                    .find(|event| event["event_type"] == event_type)
-                {
+                if let Some(event) = events.iter().find(|event| matches(event)) {
                     return Ok::<Value, anyhow::Error>(event.clone());
                 }
             }

codex-rs/app-server/tests/suite/v2/app_list.rs

@@ -119,6 +119,7 @@ async fn list_apps_returns_empty_with_api_key_auth() -> Result<()> {
             last_refresh: None,
             agent_identity: None,
             personal_access_token: None,
+            bedrock_api_key: None,
         },
         AuthCredentialsStoreMode::File,
     )?;
@@ -1516,7 +1517,7 @@ async fn start_apps_server_with_delays_and_control_inner(
             get(workspace_settings_response),
         )
         .with_state(state)
-        .nest_service("/api/codex/apps", mcp_service);
+        .nest_service("/api/codex/ps/mcp", mcp_service);
 
     let handle = tokio::spawn(async move {
         let _ = axum::serve(listener, router).await;

codex-rs/app-server/tests/suite/v2/compaction.rs

@@ -29,6 +29,7 @@ use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_config::types::AuthCredentialsStoreMode;
+use codex_features::Feature;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseItem;
 use core_test_support::responses;
@@ -150,7 +151,7 @@ async fn auto_compaction_remote_emits_started_and_completed_items() -> Result<()
     write_mock_responses_config_toml(
         codex_home.path(),
         &server.uri(),
-        &BTreeMap::default(),
+        &BTreeMap::from([(Feature::RemoteCompactionV2, false)]),
         REMOTE_AUTO_COMPACT_LIMIT,
         Some(true),
         "mock_provider",

codex-rs/app-server/tests/suite/v2/experimental_api.rs

@@ -80,10 +80,12 @@ async fn realtime_conversation_start_requires_experimental_api_capability() -> R
     let request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: "thr_123".to_string(),
+            model: None,
             output_modality: RealtimeOutputModality::Audio,
             prompt: Some(Some("hello".to_string())),
             realtime_session_id: None,
             transport: None,
+            version: None,
             voice: None,
         })
         .await?;
@@ -186,12 +188,14 @@ async fn realtime_webrtc_start_requires_experimental_api_capability() -> Result<
     let request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: "thr_123".to_string(),
+            model: None,
             output_modality: RealtimeOutputModality::Audio,
             prompt: Some(Some("hello".to_string())),
             realtime_session_id: None,
             transport: Some(ThreadRealtimeStartTransport::Webrtc {
                 sdp: "v=offer\r\n".to_string(),
             }),
+            version: None,
             voice: None,
         })
         .await?;

codex-rs/app-server/tests/suite/v2/mcp_resource.rs

@@ -19,6 +19,8 @@ use codex_app_server_protocol::McpResourceReadResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::UserInput;
 use codex_arg0::Arg0DispatchPaths;
 use codex_config::CloudConfigBundleLoader;
 use codex_config::LoaderOverrides;
@@ -30,9 +32,14 @@ use codex_protocol::protocol::SessionSource;
 use core_test_support::responses;
 use pretty_assertions::assert_eq;
 use rmcp::handler::server::ServerHandler;
+use rmcp::model::ListResourcesResult;
+use rmcp::model::Meta;
+use rmcp::model::PaginatedRequestParams;
 use rmcp::model::ProtocolVersion;
+use rmcp::model::RawResource;
 use rmcp::model::ReadResourceRequestParams;
 use rmcp::model::ReadResourceResult;
+use rmcp::model::Resource;
 use rmcp::model::ResourceContents;
 use rmcp::model::ServerCapabilities;
 use rmcp::model::ServerInfo;
@@ -41,6 +48,7 @@ use rmcp::service::RoleServer;
 use rmcp::transport::StreamableHttpServerConfig;
 use rmcp::transport::StreamableHttpService;
 use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
+use serde_json::json;
 use tempfile::TempDir;
 use tokio::net::TcpListener;
 use tokio::task::JoinHandle;
@@ -51,49 +59,34 @@ const TEST_RESOURCE_URI: &str = "test://codex/resource";
 const TEST_BLOB_RESOURCE_URI: &str = "test://codex/resource.bin";
 const TEST_RESOURCE_BLOB: &str = "YmluYXJ5LXJlc291cmNl";
 const TEST_RESOURCE_TEXT: &str = "Resource body from the MCP server.";
+const SKILL_NAME: &str = "demo-plugin:deploy";
+const RAW_SKILL_DESCRIPTION: &str = "Deploy\nthrough the <hosted> orchestrator.";
+const SKILL_DESCRIPTION: &str = "Deploy through the &lt;hosted&gt; orchestrator.";
+const SKILL_RESOURCE_URI: &str = "skill://plugin_demo/deploy";
+const SKILL_MAIN_PROMPT_URI: &str = "skill://plugin_demo/deploy/SKILL.md";
+const SKILL_REFERENCE_URI: &str = "skill://plugin_demo/deploy/references/deploy.md";
+const SKILL_MARKER: &str = "ORCHESTRATOR_SKILL_BODY_MARKER";
+const SKILL_CONTENTS: &str = concat!(
+    "---\n",
+    "name: deploy\n",
+    "description: Deploy through the orchestrator.\n",
+    "---\n\n",
+    "# Deploy\n\n",
+    "ORCHESTRATOR_SKILL_BODY_MARKER\n\n",
+    "Read the [deployment reference](skill://plugin_demo/deploy/references/deploy.md).\n",
+);
+const SKILL_REFERENCE_CONTENTS: &str =
+    "# Deploy reference\n\nUse the orchestrator deployment API.\n";
+const SKILLS_LIST_CALL_ID: &str = "skills-list";
+const SKILLS_READ_CALL_ID: &str = "skills-read";
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn mcp_resource_read_returns_resource_contents() -> Result<()> {
     let responses_server = responses::start_mock_server().await;
     let (apps_server_url, apps_server_handle) = start_resource_apps_mcp_server().await?;
-
-    let codex_home = TempDir::new()?;
     let responses_server_uri = responses_server.uri();
-    std::fs::write(
-        codex_home.path().join("config.toml"),
-        format!(
-            r#"
-model = "mock-model"
-approval_policy = "untrusted"
-sandbox_mode = "read-only"
-
-model_provider = "mock_provider"
-chatgpt_base_url = "{apps_server_url}"
-mcp_oauth_credentials_store = "file"
-
-[features]
-apps = true
-
-[model_providers.mock_provider]
-name = "Mock provider for test"
-base_url = "{responses_server_uri}/v1"
-wire_api = "responses"
-request_max_retries = 0
-stream_max_retries = 0
-"#
-        ),
-    )?;
-    write_chatgpt_auth(
-        codex_home.path(),
-        ChatGptAuthFixture::new("chatgpt-token")
-            .account_id("account-123")
-            .chatgpt_user_id("user-123")
-            .chatgpt_account_id("account-123"),
-        AuthCredentialsStoreMode::File,
-    )?;
-
-    let mut mcp = TestAppServer::new(codex_home.path()).await?;
-    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+    let (_codex_home, mut mcp) =
+        start_resource_test_app_server(&apps_server_url, &responses_server_uri).await?;
 
     let thread_start_id = mcp
         .send_thread_start_request(ThreadStartParams {
@@ -120,7 +113,6 @@ stream_max_retries = 0
         mcp.read_stream_until_response_message(RequestId::Integer(read_request_id)),
     )
     .await??;
-
     assert_eq!(
         to_response::<McpResourceReadResponse>(read_response)?,
         expected_resource_read_response()
@@ -131,6 +123,164 @@ stream_max_retries = 0
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn orchestrator_skill_can_read_referenced_resource_without_an_executor() -> Result<()> {
+    let responses_server = responses::start_mock_server().await;
+    let (apps_server_url, apps_server_handle) = start_resource_apps_mcp_server().await?;
+    let responses_server_uri = responses_server.uri();
+    let (_codex_home, mut mcp) =
+        start_resource_test_app_server(&apps_server_url, &responses_server_uri).await?;
+
+    let thread_start_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            environments: Some(Vec::new()),
+            ..Default::default()
+        })
+        .await?;
+    let thread_start_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response(thread_start_resp)?;
+
+    let response_mock = responses::mount_sse_sequence(
+        &responses_server,
+        vec![
+            responses::sse(vec![
+                responses::ev_response_created("resp-skills-list"),
+                responses::ev_function_call_with_namespace(
+                    SKILLS_LIST_CALL_ID,
+                    "skills",
+                    "list",
+                    &json!({
+                        "authority": {
+                            "kind": "orchestrator",
+                        },
+                    })
+                    .to_string(),
+                ),
+                responses::ev_completed("resp-skills-list"),
+            ]),
+            responses::sse(vec![
+                responses::ev_response_created("resp-skills-read"),
+                responses::ev_function_call_with_namespace(
+                    SKILLS_READ_CALL_ID,
+                    "skills",
+                    "read",
+                    &json!({
+                        "authority": {
+                            "kind": "orchestrator",
+                        },
+                        "package": SKILL_RESOURCE_URI,
+                        "resource": SKILL_REFERENCE_URI,
+                    })
+                    .to_string(),
+                ),
+                responses::ev_completed("resp-skills-read"),
+            ]),
+            responses::sse(vec![
+                responses::ev_response_created("resp-orchestrator-skill"),
+                responses::ev_assistant_message("msg-orchestrator-skill", "Done"),
+                responses::ev_completed("resp-orchestrator-skill"),
+            ]),
+        ],
+    )
+    .await;
+    let turn_start_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![UserInput::Text {
+                text: format!("Use ${SKILL_NAME}"),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)),
+    )
+    .await??;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    let requests = response_mock.requests();
+    assert_eq!(requests.len(), 3);
+    let first_request = &requests[0];
+    assert!(first_request.tool_by_name("skills", "list").is_some());
+    assert!(first_request.tool_by_name("skills", "read").is_some());
+    assert!(first_request.tool_by_name("skills", "search").is_none());
+
+    let developer_messages = first_request.message_input_texts("developer");
+    let catalog_line = format!(
+        "- {SKILL_NAME}: {SKILL_DESCRIPTION} (orchestrator resource: {SKILL_RESOURCE_URI})"
+    );
+    assert_eq!(
+        1,
+        developer_messages
+            .iter()
+            .filter(|text| text.contains(&catalog_line))
+            .count()
+    );
+    assert!(
+        developer_messages
+            .iter()
+            .all(|text| !text.contains("ignored-plugin:ignored"))
+    );
+    assert!(
+        developer_messages
+            .iter()
+            .any(|text| text.contains("do not treat `skill://` identifiers as filesystem paths"))
+    );
+    let skill_fragments = first_request
+        .message_input_texts("user")
+        .into_iter()
+        .filter(|text| text.starts_with("<skill>"))
+        .collect::<Vec<_>>();
+    assert_eq!(1, skill_fragments.len());
+    assert!(skill_fragments[0].contains(&format!("<name>{SKILL_NAME}</name>")));
+    assert!(skill_fragments[0].contains(SKILL_MARKER));
+    assert!(skill_fragments[0].contains(SKILL_REFERENCE_URI));
+
+    let list_output = requests[1]
+        .function_call_output_text(SKILLS_LIST_CALL_ID)
+        .ok_or_else(|| anyhow::anyhow!("skills.list output should be sent to the model"))?;
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&list_output)?,
+        json!({
+            "skills": [{
+                "authority": {
+                    "kind": "orchestrator",
+                },
+                "package": SKILL_RESOURCE_URI,
+                "name": SKILL_NAME,
+                "description": SKILL_DESCRIPTION,
+                "main_resource": SKILL_MAIN_PROMPT_URI,
+            }],
+            "warnings": ["Orchestrator skill discovery stopped after 2 resource pages: failed to list orchestrator skill resources: resources/list failed for `codex_apps`: Mcp error: -32603: simulated later-page failure"],
+        })
+    );
+
+    let read_output = requests[2]
+        .function_call_output_text(SKILLS_READ_CALL_ID)
+        .ok_or_else(|| anyhow::anyhow!("skills.read output should be sent to the model"))?;
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&read_output)?,
+        json!({
+            "resource": SKILL_REFERENCE_URI,
+            "contents": SKILL_REFERENCE_CONTENTS,
+        })
+    );
+    apps_server_handle.abort();
+    let _ = apps_server_handle.await;
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn mcp_resource_read_returns_resource_contents_without_thread() -> Result<()> {
     let (apps_server_url, apps_server_handle) = start_resource_apps_mcp_server().await?;
@@ -246,6 +396,52 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> {
     Ok(())
 }
 
+async fn start_resource_test_app_server(
+    apps_server_url: &str,
+    responses_server_uri: &str,
+) -> Result<(TempDir, TestAppServer)> {
+    let codex_home = TempDir::new()?;
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "untrusted"
+sandbox_mode = "read-only"
+
+model_provider = "mock_provider"
+chatgpt_base_url = "{apps_server_url}"
+mcp_oauth_credentials_store = "file"
+
+[features]
+apps = true
+
+[skills]
+include_instructions = true
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{responses_server_uri}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+"#
+        ),
+    )?;
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("chatgpt-token")
+            .account_id("account-123")
+            .chatgpt_user_id("user-123")
+            .chatgpt_account_id("account-123"),
+        AuthCredentialsStoreMode::File,
+    )?;
+
+    let mut mcp = TestAppServer::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+    Ok((codex_home, mcp))
+}
+
 async fn start_resource_apps_mcp_server() -> Result<(String, JoinHandle<()>)> {
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let addr = listener.local_addr()?;
@@ -256,7 +452,7 @@ async fn start_resource_apps_mcp_server() -> Result<(String, JoinHandle<()>)> {
         Arc::new(LocalSessionManager::default()),
         StreamableHttpServerConfig::default(),
     );
-    let router = Router::new().nest_service("/api/codex/apps", mcp_service);
+    let router = Router::new().nest_service("/api/codex/ps/mcp", mcp_service);
     let apps_server_handle = tokio::spawn(async move {
         let _ = axum::serve(listener, router).await;
     });
@@ -292,12 +488,79 @@ impl ServerHandler for ResourceAppsMcpServer {
             .with_protocol_version(ProtocolVersion::V_2025_06_18)
     }
 
+    async fn list_resources(
+        &self,
+        request: Option<PaginatedRequestParams>,
+        _context: RequestContext<RoleServer>,
+    ) -> Result<ListResourcesResult, rmcp::ErrorData> {
+        let cursor = request.and_then(|request| request.cursor);
+        if cursor.is_none() {
+            return Ok(ListResourcesResult {
+                resources: vec![skill_resource(
+                    "skill://plugin_ignored/ignored",
+                    "plugin_ignored/ignored",
+                    "Not an MCP skill resource.",
+                    "text/plain",
+                    "ignored-plugin",
+                    "ignored",
+                )],
+                next_cursor: Some("skills-page".to_string()),
+                meta: None,
+            });
+        }
+        if cursor.as_deref() == Some("failing-page") {
+            return Err(rmcp::ErrorData::internal_error(
+                "simulated later-page failure",
+                /*data*/ None,
+            ));
+        }
+        if cursor.as_deref() != Some("skills-page") {
+            return Err(rmcp::ErrorData::invalid_params(
+                "unexpected resources/list cursor",
+                /*data*/ None,
+            ));
+        }
+
+        Ok(ListResourcesResult {
+            resources: vec![skill_resource(
+                SKILL_RESOURCE_URI,
+                "plugin_demo/deploy",
+                RAW_SKILL_DESCRIPTION,
+                "mcp/skill",
+                "demo-plugin",
+                "deploy",
+            )],
+            next_cursor: Some("failing-page".to_string()),
+            meta: None,
+        })
+    }
+
     async fn read_resource(
         &self,
         request: ReadResourceRequestParams,
         _context: RequestContext<RoleServer>,
     ) -> Result<ReadResourceResult, rmcp::ErrorData> {
         let uri = request.uri;
+        if uri == SKILL_MAIN_PROMPT_URI {
+            return Ok(ReadResourceResult::new(vec![
+                ResourceContents::TextResourceContents {
+                    uri: SKILL_MAIN_PROMPT_URI.to_string(),
+                    mime_type: Some("text/markdown".to_string()),
+                    text: SKILL_CONTENTS.to_string(),
+                    meta: None,
+                },
+            ]));
+        }
+        if uri == SKILL_REFERENCE_URI {
+            return Ok(ReadResourceResult::new(vec![
+                ResourceContents::TextResourceContents {
+                    uri: SKILL_REFERENCE_URI.to_string(),
+                    mime_type: Some("text/markdown".to_string()),
+                    text: SKILL_REFERENCE_CONTENTS.to_string(),
+                    meta: None,
+                },
+            ]));
+        }
         if uri != TEST_RESOURCE_URI {
             return Err(rmcp::ErrorData::resource_not_found(
                 format!("resource not found: {uri}"),
@@ -321,3 +584,27 @@ impl ServerHandler for ResourceAppsMcpServer {
         ]))
     }
 }
+
+fn skill_resource(
+    uri: &str,
+    name: &str,
+    description: &str,
+    mime_type: &str,
+    plugin_name: &str,
+    skill_name: &str,
+) -> Resource {
+    Resource::new(
+        RawResource::new(uri, name)
+            .with_description(description)
+            .with_mime_type(mime_type)
+            .with_meta(skill_resource_meta(plugin_name, skill_name)),
+        /*annotations*/ None,
+    )
+}
+
+fn skill_resource_meta(plugin_name: &str, skill_name: &str) -> Meta {
+    Meta(serde_json::Map::from_iter([
+        ("plugin_name".to_string(), json!(plugin_name)),
+        ("skill_name".to_string(), json!(skill_name)),
+    ]))
+}

codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs

@@ -406,7 +406,7 @@ async fn start_apps_server() -> Result<(String, JoinHandle<()>)> {
             get(list_directory_connectors),
         )
         .with_state(state)
-        .nest_service("/api/codex/apps", mcp_service);
+        .nest_service("/api/codex/ps/mcp", mcp_service);
 
     let handle = tokio::spawn(async move {
         let _ = axum::serve(listener, router).await;

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -50,6 +50,7 @@ mod review;
 mod safety_check_downgrade;
 mod skills_list;
 mod thread_archive;
+mod thread_delete;
 mod thread_fork;
 mod thread_inject_items;
 mod thread_list;

codex-rs/app-server/tests/suite/v2/plugin_install.rs

@@ -327,7 +327,6 @@ async fn plugin_install_uses_remote_apps_needing_auth_response() -> Result<()> {
                 name: "alpha".to_string(),
                 description: None,
                 install_url: Some("https://chatgpt.com/apps/alpha/alpha".to_string()),
-                needs_auth: true,
             }],
         }
     );
@@ -1001,7 +1000,6 @@ async fn plugin_install_returns_apps_needing_auth() -> Result<()> {
                 name: "Alpha".to_string(),
                 description: Some("Alpha connector".to_string()),
                 install_url: Some("https://chatgpt.com/apps/alpha/alpha".to_string()),
-                needs_auth: true,
             }],
         }
     );
@@ -1089,7 +1087,6 @@ async fn plugin_install_filters_disallowed_apps_needing_auth() -> Result<()> {
                 name: "Alpha".to_string(),
                 description: Some("Alpha connector".to_string()),
                 install_url: Some("https://chatgpt.com/apps/alpha/alpha".to_string()),
-                needs_auth: true,
             }],
         }
     );
@@ -1289,7 +1286,7 @@ async fn start_apps_server(
             get(list_directory_connectors),
         )
         .with_state(state)
-        .nest_service("/api/codex/apps", mcp_service);
+        .nest_service("/api/codex/ps/mcp", mcp_service);
 
     let handle = tokio::spawn(async move {
         let _ = axum::serve(listener, router).await;

codex-rs/app-server/tests/suite/v2/plugin_read.rs

@@ -1,6 +1,3 @@
-use std::borrow::Cow;
-use std::sync::Arc;
-use std::sync::Mutex as StdMutex;
 use std::time::Duration;
 
 use anyhow::Result;
@@ -8,15 +5,6 @@ use app_test_support::ChatGptAuthFixture;
 use app_test_support::TestAppServer;
 use app_test_support::to_response;
 use app_test_support::write_chatgpt_auth;
-use axum::Json;
-use axum::Router;
-use axum::extract::State;
-use axum::http::HeaderMap;
-use axum::http::StatusCode;
-use axum::http::Uri;
-use axum::http::header::AUTHORIZATION;
-use axum::routing::get;
-use codex_app_server_protocol::AppInfo;
 use codex_app_server_protocol::AppTemplateSummary;
 use codex_app_server_protocol::AppTemplateUnavailableReason;
 use codex_app_server_protocol::HookEventName;
@@ -37,21 +25,8 @@ use codex_app_server_protocol::RequestId;
 use codex_config::types::AuthCredentialsStoreMode;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use pretty_assertions::assert_eq;
-use rmcp::handler::server::ServerHandler;
-use rmcp::model::JsonObject;
-use rmcp::model::ListToolsResult;
-use rmcp::model::Meta;
-use rmcp::model::ServerCapabilities;
-use rmcp::model::ServerInfo;
-use rmcp::model::Tool;
-use rmcp::model::ToolAnnotations;
-use rmcp::transport::StreamableHttpServerConfig;
-use rmcp::transport::StreamableHttpService;
-use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
 use serde_json::json;
 use tempfile::TempDir;
-use tokio::net::TcpListener;
-use tokio::task::JoinHandle;
 use tokio::time::timeout;
 use wiremock::Mock;
 use wiremock::MockServer;
@@ -1483,104 +1458,11 @@ enabled = false
         response.plugin.apps[0].install_url.as_deref(),
         Some("https://chatgpt.com/apps/gmail/gmail")
     );
-    assert_eq!(response.plugin.apps[0].needs_auth, true);
     assert_eq!(response.plugin.mcp_servers.len(), 1);
     assert_eq!(response.plugin.mcp_servers[0], "demo");
     Ok(())
 }
 
-#[tokio::test]
-async fn plugin_read_returns_app_needs_auth() -> Result<()> {
-    let connectors = vec![
-        AppInfo {
-            id: "alpha".to_string(),
-            name: "Alpha".to_string(),
-            description: Some("Alpha connector".to_string()),
-            logo_url: Some("https://example.com/alpha.png".to_string()),
-            logo_url_dark: None,
-            distribution_channel: Some("featured".to_string()),
-            branding: None,
-            app_metadata: None,
-            labels: None,
-            install_url: None,
-            is_accessible: false,
-            is_enabled: true,
-            plugin_display_names: Vec::new(),
-        },
-        AppInfo {
-            id: "beta".to_string(),
-            name: "Beta".to_string(),
-            description: Some("Beta connector".to_string()),
-            logo_url: None,
-            logo_url_dark: None,
-            distribution_channel: None,
-            branding: None,
-            app_metadata: None,
-            labels: None,
-            install_url: None,
-            is_accessible: false,
-            is_enabled: true,
-            plugin_display_names: Vec::new(),
-        },
-    ];
-    let tools = vec![connector_tool("beta", "Beta App")?];
-    let (server_url, server_handle) = start_apps_server(connectors, tools).await?;
-
-    let codex_home = TempDir::new()?;
-    write_connectors_config(codex_home.path(), &server_url)?;
-    write_chatgpt_auth(
-        codex_home.path(),
-        ChatGptAuthFixture::new("chatgpt-token")
-            .account_id("account-123")
-            .chatgpt_user_id("user-123")
-            .chatgpt_account_id("account-123"),
-        AuthCredentialsStoreMode::File,
-    )?;
-
-    let repo_root = TempDir::new()?;
-    write_plugin_marketplace(
-        repo_root.path(),
-        "debug",
-        "sample-plugin",
-        "./sample-plugin",
-    )?;
-    write_plugin_source(repo_root.path(), "sample-plugin", &["alpha", "beta"])?;
-    let marketplace_path =
-        AbsolutePathBuf::try_from(repo_root.path().join(".agents/plugins/marketplace.json"))?;
-
-    let mut mcp = TestAppServer::new(codex_home.path()).await?;
-    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
-
-    let request_id = mcp
-        .send_plugin_read_request(PluginReadParams {
-            marketplace_path: Some(marketplace_path),
-            remote_marketplace_name: None,
-            plugin_name: "sample-plugin".to_string(),
-        })
-        .await?;
-
-    let response: JSONRPCResponse = timeout(
-        DEFAULT_TIMEOUT,
-        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
-    )
-    .await??;
-    let response: PluginReadResponse = to_response(response)?;
-
-    assert_eq!(
-        response
-            .plugin
-            .apps
-            .iter()
-            .map(|app| (app.id.as_str(), app.needs_auth))
-            .collect::<Vec<_>>(),
-        vec![("alpha", true), ("beta", false)]
-    );
-
-    server_handle.abort();
-    let _ = server_handle.await;
-    Ok(())
-}
-
 #[tokio::test]
 async fn plugin_read_accepts_legacy_string_default_prompt() -> Result<()> {
     let codex_home = TempDir::new()?;
@@ -1846,151 +1728,6 @@ plugins = true
     Ok(())
 }
 
-#[derive(Clone)]
-struct AppsServerState {
-    response: Arc<StdMutex<serde_json::Value>>,
-}
-
-#[derive(Clone)]
-struct PluginReadMcpServer {
-    tools: Arc<StdMutex<Vec<Tool>>>,
-}
-
-impl ServerHandler for PluginReadMcpServer {
-    fn get_info(&self) -> ServerInfo {
-        ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
-    }
-
-    fn list_tools(
-        &self,
-        _request: Option<rmcp::model::PaginatedRequestParams>,
-        _context: rmcp::service::RequestContext<rmcp::service::RoleServer>,
-    ) -> impl std::future::Future<Output = Result<ListToolsResult, rmcp::ErrorData>> + Send + '_
-    {
-        let tools = self.tools.clone();
-        async move {
-            let tools = tools
-                .lock()
-                .unwrap_or_else(std::sync::PoisonError::into_inner)
-                .clone();
-            Ok(ListToolsResult {
-                tools,
-                next_cursor: None,
-                meta: None,
-            })
-        }
-    }
-}
-
-async fn start_apps_server(
-    connectors: Vec<AppInfo>,
-    tools: Vec<Tool>,
-) -> Result<(String, JoinHandle<()>)> {
-    let state = Arc::new(AppsServerState {
-        response: Arc::new(StdMutex::new(
-            json!({ "apps": connectors, "next_token": null }),
-        )),
-    });
-    let tools = Arc::new(StdMutex::new(tools));
-
-    let listener = TcpListener::bind("127.0.0.1:0").await?;
-    let addr = listener.local_addr()?;
-    let mcp_service = StreamableHttpService::new(
-        {
-            let tools = tools.clone();
-            move || {
-                Ok(PluginReadMcpServer {
-                    tools: tools.clone(),
-                })
-            }
-        },
-        Arc::new(LocalSessionManager::default()),
-        StreamableHttpServerConfig::default(),
-    );
-    let router = Router::new()
-        .route("/connectors/directory/list", get(list_directory_connectors))
-        .route(
-            "/connectors/directory/list_workspace",
-            get(list_directory_connectors),
-        )
-        .with_state(state)
-        .nest_service("/api/codex/apps", mcp_service);
-
-    let handle = tokio::spawn(async move {
-        let _ = axum::serve(listener, router).await;
-    });
-
-    Ok((format!("http://{addr}"), handle))
-}
-
-async fn list_directory_connectors(
-    State(state): State<Arc<AppsServerState>>,
-    headers: HeaderMap,
-    uri: Uri,
-) -> Result<impl axum::response::IntoResponse, StatusCode> {
-    let bearer_ok = headers
-        .get(AUTHORIZATION)
-        .and_then(|value| value.to_str().ok())
-        .is_some_and(|value| value == "Bearer chatgpt-token");
-    let account_ok = headers
-        .get("chatgpt-account-id")
-        .and_then(|value| value.to_str().ok())
-        .is_some_and(|value| value == "account-123");
-    let external_logos_ok = uri
-        .query()
-        .is_some_and(|query| query.split('&').any(|pair| pair == "external_logos=true"));
-
-    if !bearer_ok || !account_ok {
-        Err(StatusCode::UNAUTHORIZED)
-    } else if !external_logos_ok {
-        Err(StatusCode::BAD_REQUEST)
-    } else {
-        let response = state
-            .response
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .clone();
-        Ok(Json(response))
-    }
-}
-
-fn connector_tool(connector_id: &str, connector_name: &str) -> Result<Tool> {
-    let schema: JsonObject = serde_json::from_value(json!({
-        "type": "object",
-        "additionalProperties": false
-    }))?;
-    let mut tool = Tool::new(
-        Cow::Owned(format!("connector_{connector_id}")),
-        Cow::Borrowed("Connector test tool"),
-        Arc::new(schema),
-    );
-    tool.annotations = Some(ToolAnnotations::new().read_only(true));
-
-    let mut meta = Meta::new();
-    meta.0
-        .insert("connector_id".to_string(), json!(connector_id));
-    meta.0
-        .insert("connector_name".to_string(), json!(connector_name));
-    tool.meta = Some(meta);
-    Ok(tool)
-}
-
-fn write_connectors_config(codex_home: &std::path::Path, base_url: &str) -> std::io::Result<()> {
-    std::fs::write(
-        codex_home.join("config.toml"),
-        format!(
-            r#"
-chatgpt_base_url = "{base_url}"
-mcp_oauth_credentials_store = "file"
-
-[features]
-plugins = true
-connectors = true
-"#
-        ),
-    )
-}
-
 fn write_remote_plugin_catalog_config(
     codex_home: &std::path::Path,
     base_url: &str,

codex-rs/app-server/tests/suite/v2/realtime_conversation.rs

@@ -36,7 +36,10 @@ use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::TurnCompletedNotification;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::TurnStartedNotification;
+use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_features::FEATURES;
 use codex_features::Feature;
 use codex_protocol::protocol::RealtimeConversationVersion;
@@ -311,12 +314,14 @@ impl RealtimeE2eHarness {
             .mcp
             .send_thread_realtime_start_request(ThreadRealtimeStartParams {
                 thread_id: self.thread_id.clone(),
+                model: None,
                 output_modality: RealtimeOutputModality::Audio,
                 prompt: Some(Some("backend prompt".to_string())),
                 realtime_session_id: None,
                 transport: Some(ThreadRealtimeStartTransport::Webrtc {
                     sdp: offer_sdp.to_string(),
                 }),
+                version: None,
                 voice: None,
             })
             .await?;
@@ -556,10 +561,12 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: thread_start.thread.id.clone(),
+            model: Some("realtime-treatment-model".to_string()),
             output_modality: RealtimeOutputModality::Audio,
             prompt: None,
             realtime_session_id: None,
             transport: None,
+            version: None,
             voice: Some(RealtimeVoice::Cedar),
         })
         .await?;
@@ -588,6 +595,10 @@ async fn realtime_conversation_streams_v2_notifications() -> Result<()> {
         startup_context_request.body_json()["session"]["audio"]["output"]["voice"],
         "cedar"
     );
+    assert_eq!(
+        realtime_server.single_handshake().uri(),
+        "/v1/realtime?model=realtime-treatment-model"
+    );
     assert_eq!(
         startup_context_request.body_json()["session"]["output_modalities"],
         json!(["audio"])
@@ -805,10 +816,12 @@ async fn realtime_text_output_modality_requests_text_output_and_final_transcript
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: thread_start.thread.id.clone(),
+            model: None,
             output_modality: RealtimeOutputModality::Text,
             prompt: None,
             realtime_session_id: None,
             transport: None,
+            version: None,
             voice: None,
         })
         .await?;
@@ -979,10 +992,12 @@ async fn realtime_conversation_stop_emits_closed_notification() -> Result<()> {
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: thread_start.thread.id.clone(),
+            model: None,
             output_modality: RealtimeOutputModality::Audio,
             prompt: Some(Some("backend prompt".to_string())),
             realtime_session_id: None,
             transport: None,
+            version: None,
             voice: None,
         })
         .await?;
@@ -1076,12 +1091,14 @@ async fn realtime_webrtc_start_emits_sdp_notification() -> Result<()> {
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: thread_id.clone(),
+            model: None,
             output_modality: RealtimeOutputModality::Audio,
             prompt: Some(Some("backend prompt".to_string())),
             realtime_session_id: None,
             transport: Some(ThreadRealtimeStartTransport::Webrtc {
                 sdp: "v=offer\r\n".to_string(),
             }),
+            version: None,
             voice: None,
         })
         .await?;
@@ -1328,6 +1345,136 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
     Ok(())
 }
 
+#[tokio::test]
+async fn webrtc_assistant_output_without_handoff_reaches_realtime() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let final_answer = "long output ".repeat(1_000);
+    for (version, expected_version, preamble) in [
+        (
+            RealtimeTestVersion::V1,
+            RealtimeConversationVersion::V1,
+            "direct preamble from v1",
+        ),
+        (
+            RealtimeTestVersion::V2,
+            RealtimeConversationVersion::V2,
+            "direct preamble from v2",
+        ),
+    ] {
+        let mut harness = RealtimeE2eHarness::new(
+            version,
+            main_loop_responses(vec![responses::sse(vec![
+                responses::ev_response_created("resp-1"),
+                json!({
+                    "type": "response.output_item.done",
+                    "item": {
+                        "type": "message",
+                        "role": "assistant",
+                        "id": "msg-preamble",
+                        "phase": "commentary",
+                        "content": [{"type": "output_text", "text": preamble}]
+                    }
+                }),
+                responses::ev_assistant_message("msg-final", &final_answer),
+                responses::ev_completed("resp-1"),
+            ])]),
+            realtime_sideband(vec![realtime_sideband_connection(vec![
+                vec![session_updated("sess_standalone_output")],
+                vec![],
+                match version {
+                    RealtimeTestVersion::V1 => vec![],
+                    RealtimeTestVersion::V2 => vec![
+                        json!({
+                            "type": "response.created",
+                            "response": { "id": "resp_preamble" }
+                        }),
+                        json!({
+                            "type": "response.done",
+                            "response": { "id": "resp_preamble" }
+                        }),
+                    ],
+                },
+                vec![],
+                vec![],
+            ])]),
+        )
+        .await?;
+
+        let started = harness.start_webrtc_realtime("v=offer\r\n").await?;
+        assert_eq!(started.started.version, expected_version);
+
+        let request_id = harness
+            .mcp
+            .send_turn_start_request(TurnStartParams {
+                thread_id: harness.thread_id.clone(),
+                input: vec![V2UserInput::Text {
+                    text: "direct text turn".to_string(),
+                    text_elements: Vec::new(),
+                }],
+                ..Default::default()
+            })
+            .await?;
+        let response: JSONRPCResponse = timeout(
+            DEFAULT_TIMEOUT,
+            harness
+                .mcp
+                .read_stream_until_response_message(RequestId::Integer(request_id)),
+        )
+        .await??;
+        let _: TurnStartResponse = to_response(response)?;
+        let _ = harness
+            .read_notification::<TurnCompletedNotification>("turn/completed")
+            .await?;
+
+        let preamble_request = harness.sideband_outbound_request(/*request_index*/ 1).await;
+        let output_text = match version {
+            RealtimeTestVersion::V1 => {
+                let final_request = harness.sideband_outbound_request(/*request_index*/ 2).await;
+                assert_eq!(
+                    preamble_request,
+                    json!({
+                        "type": "conversation.handoff.append",
+                        "handoff_id": "codex",
+                        "output_text": preamble,
+                    })
+                );
+                assert_eq!(final_request["type"], "conversation.handoff.append");
+                assert_eq!(final_request["handoff_id"], "codex");
+                final_request["output_text"]
+                    .as_str()
+                    .expect("output text")
+                    .to_string()
+            }
+            RealtimeTestVersion::V2 => {
+                assert_v2_progress_update(&preamble_request, preamble);
+                assert_v2_response_create(
+                    &harness.sideband_outbound_request(/*request_index*/ 2).await,
+                );
+                let final_request = harness.sideband_outbound_request(/*request_index*/ 3).await;
+                assert_eq!(final_request["type"], "conversation.item.create");
+                assert_eq!(final_request["item"]["type"], "message");
+                assert_eq!(final_request["item"]["role"], "user");
+                assert_eq!(final_request["item"]["content"][0]["type"], "input_text");
+                let output_text = final_request["item"]["content"][0]["text"]
+                    .as_str()
+                    .expect("output text");
+                assert!(output_text.starts_with("[BACKEND] "));
+                assert_v2_response_create(
+                    &harness.sideband_outbound_request(/*request_index*/ 4).await,
+                );
+                output_text.to_string()
+            }
+        };
+        assert!(output_text.contains("tokens truncated"));
+        assert!(output_text.len() <= 4_000);
+
+        harness.shutdown().await;
+    }
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn webrtc_v2_forwards_audio_and_text_between_client_and_sideband() -> Result<()> {
     skip_if_no_network!(Ok(()));
@@ -1991,12 +2138,14 @@ async fn realtime_webrtc_start_surfaces_backend_error() -> Result<()> {
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: thread_start.thread.id,
+            model: None,
             output_modality: RealtimeOutputModality::Audio,
             prompt: Some(Some("backend prompt".to_string())),
             realtime_session_id: None,
             transport: Some(ThreadRealtimeStartTransport::Webrtc {
                 sdp: "v=offer\r\n".to_string(),
             }),
+            version: None,
             voice: None,
         })
         .await?;
@@ -2050,10 +2199,12 @@ async fn realtime_conversation_requires_feature_flag() -> Result<()> {
     let start_request_id = mcp
         .send_thread_realtime_start_request(ThreadRealtimeStartParams {
             thread_id: thread_start.thread.id.clone(),
+            model: None,
             output_modality: RealtimeOutputModality::Audio,
             prompt: Some(Some("backend prompt".to_string())),
             realtime_session_id: None,
             transport: None,
+            version: None,
             voice: None,
         })
         .await?;

codex-rs/app-server/tests/suite/v2/remote_thread_store.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 use anyhow::Result;
 use app_test_support::create_mock_responses_server_repeating_assistant;
 use codex_app_server::in_process;
+use codex_app_server::in_process::InProcessClientHandle;
 use codex_app_server::in_process::InProcessServerEvent;
 use codex_app_server::in_process::InProcessStartArgs;
 use codex_app_server_protocol::ClientInfo;
@@ -27,6 +28,8 @@ use codex_app_server_protocol::ClientRequest;
 use codex_app_server_protocol::InitializeParams;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::ThreadDeleteParams;
+use codex_app_server_protocol::ThreadDeleteResponse;
 use codex_app_server_protocol::ThreadListParams;
 use codex_app_server_protocol::ThreadListResponse;
 use codex_app_server_protocol::ThreadResumeParams;
@@ -42,8 +45,14 @@ use codex_core::config::Config;
 use codex_core::config::ConfigBuilder;
 use codex_exec_server::EnvironmentManager;
 use codex_feedback::CodexFeedback;
+use codex_protocol::ThreadId;
+use codex_protocol::models::BaseInstructions;
 use codex_protocol::protocol::SessionSource;
+use codex_protocol::protocol::ThreadMemoryMode;
+use codex_thread_store::CreateThreadParams as StoreCreateThreadParams;
 use codex_thread_store::InMemoryThreadStore;
+use codex_thread_store::ThreadPersistenceMetadata;
+use codex_thread_store::ThreadStore;
 use pretty_assertions::assert_eq;
 use tempfile::TempDir;
 use tokio::time::timeout;
@@ -52,7 +61,7 @@ use uuid::Uuid;
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
 
 #[tokio::test]
-async fn thread_start_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()>
+async fn thread_delete_with_non_local_thread_store_does_not_create_local_persistence() -> Result<()>
 {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
     let codex_home = TempDir::new()?;
@@ -61,43 +70,10 @@ async fn thread_start_with_non_local_thread_store_does_not_create_local_persiste
     // here so this regression stays focused on thread persistence artifacts.
     create_config_toml_with_thread_store(codex_home.path(), &server.uri(), &store_id)?;
 
-    let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
-    let config = ConfigBuilder::default()
-        .codex_home(codex_home.path().to_path_buf())
-        .fallback_cwd(Some(codex_home.path().to_path_buf()))
-        .loader_overrides(loader_overrides.clone())
-        .build()
-        .await?;
-
     let thread_store = InMemoryThreadStore::for_id(store_id.clone());
     let _in_memory_store = InMemoryThreadStoreId { store_id };
 
-    let mut client = in_process::start(InProcessStartArgs {
-        arg0_paths: Arg0DispatchPaths::default(),
-        config: Arc::new(config),
-        cli_overrides: Vec::new(),
-        loader_overrides,
-        strict_config: false,
-        cloud_config_bundle: CloudConfigBundleLoader::default(),
-        thread_config_loader: Arc::new(NoopThreadConfigLoader),
-        feedback: CodexFeedback::new(),
-        log_db: None,
-        state_db: None,
-        environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
-        config_warnings: Vec::new(),
-        session_source: SessionSource::Cli,
-        enable_codex_api_key_env: false,
-        initialize: InitializeParams {
-            client_info: ClientInfo {
-                name: "codex-app-server-tests".to_string(),
-                title: None,
-                version: "0.1.0".to_string(),
-            },
-            capabilities: None,
-        },
-        channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
-    })
-    .await?;
+    let mut client = start_in_process_server(codex_home.path()).await?;
 
     let response = client
         .request(ClientRequest::ThreadStart {
@@ -166,11 +142,39 @@ async fn thread_start_with_non_local_thread_store_does_not_create_local_persiste
     assert_eq!(data[0].id, thread.id);
     assert_eq!(data[0].path, None);
 
+    delete_thread(&client, /*request_id*/ 4, thread.id.clone()).await?;
+    let unloaded_thread_id = ThreadId::from_string(&Uuid::new_v4().to_string())?;
+    thread_store
+        .create_thread(StoreCreateThreadParams {
+            thread_id: unloaded_thread_id,
+            extra_config: None,
+            forked_from_id: None,
+            parent_thread_id: None,
+            source: SessionSource::Cli,
+            thread_source: None,
+            base_instructions: BaseInstructions::default(),
+            dynamic_tools: Vec::new(),
+            multi_agent_version: None,
+            metadata: ThreadPersistenceMetadata {
+                cwd: Some(codex_home.path().to_path_buf()),
+                model_provider: "mock_provider".to_string(),
+                memory_mode: ThreadMemoryMode::Enabled,
+            },
+        })
+        .await?;
+    delete_thread(
+        &client,
+        /*request_id*/ 5,
+        unloaded_thread_id.to_string(),
+    )
+    .await?;
+
     client.shutdown().await?;
 
     let calls = thread_store.calls().await;
-    assert_eq!(calls.create_thread, 1);
+    assert_eq!(calls.create_thread, 2);
     assert_eq!(calls.list_threads, 1);
+    assert_eq!(calls.delete_thread, 2);
     assert!(
         calls.append_items > 0,
         "turn/start should append rollout items through the injected store"
@@ -269,10 +273,24 @@ async fn cold_thread_resume_reuses_non_local_history_probe() -> Result<()> {
     Ok(())
 }
 
+async fn start_in_process_server(codex_home: &Path) -> Result<InProcessClientHandle> {
+    let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
+    let config = Arc::new(
+        ConfigBuilder::default()
+            .codex_home(codex_home.to_path_buf())
+            .fallback_cwd(Some(codex_home.to_path_buf()))
+            .loader_overrides(loader_overrides.clone())
+            .build()
+            .await?,
+    );
+
+    Ok(start_in_process_client(config, loader_overrides).await?)
+}
+
 async fn start_in_process_client(
     config: Arc<Config>,
     loader_overrides: LoaderOverrides,
-) -> std::io::Result<in_process::InProcessClientHandle> {
+) -> std::io::Result<InProcessClientHandle> {
     in_process::start(InProcessStartArgs {
         arg0_paths: Arg0DispatchPaths::default(),
         config,
@@ -301,6 +319,22 @@ async fn start_in_process_client(
     .await
 }
 
+async fn delete_thread(
+    client: &InProcessClientHandle,
+    request_id: i64,
+    thread_id: String,
+) -> Result<()> {
+    let response = client
+        .request(ClientRequest::ThreadDelete {
+            request_id: RequestId::Integer(request_id),
+            params: ThreadDeleteParams { thread_id },
+        })
+        .await?
+        .map_err(|error| anyhow::anyhow!("thread/delete failed: {}", error.message))?;
+    let _: ThreadDeleteResponse = serde_json::from_value(response)?;
+    Ok(())
+}
+
 fn assert_no_local_persistence_artifacts(codex_home: &Path) -> Result<()> {
     // These are the observable tripwires for accidental local persistence. If a
     // future code path constructs a local rollout/session store or opens the

codex-rs/app-server/tests/suite/v2/thread_delete.rs

@@ -0,0 +1,200 @@
+use anyhow::Result;
+use app_test_support::TestAppServer;
+use app_test_support::create_fake_rollout;
+use app_test_support::to_response;
+use codex_app_server_protocol::JSONRPCError;
+use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ThreadDeleteParams;
+use codex_app_server_protocol::ThreadDeleteResponse;
+use codex_app_server_protocol::ThreadDeletedNotification;
+use codex_app_server_protocol::ThreadLoadedListParams;
+use codex_app_server_protocol::ThreadLoadedListResponse;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_core::find_thread_path_by_id_str;
+use codex_protocol::ThreadId;
+use codex_state::DirectionalThreadSpawnEdgeStatus;
+use codex_state::StateRuntime;
+use pretty_assertions::assert_eq;
+use std::path::Path;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test]
+async fn thread_delete_deletes_spawned_descendants() -> Result<()> {
+    let codex_home = TempDir::new()?;
+
+    let parent_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 0, "parent")?;
+    let child_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 1, "child")?;
+    let grandchild_id =
+        create_delete_test_rollout(codex_home.path(), /*minute*/ 2, "grandchild")?;
+
+    let state_db =
+        StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()).await?;
+    let parent_thread_id = ThreadId::from_string(&parent_id)?;
+    let child_thread_id = ThreadId::from_string(&child_id)?;
+    let grandchild_thread_id = ThreadId::from_string(&grandchild_id)?;
+
+    for (parent, child, status) in [
+        (
+            parent_thread_id,
+            child_thread_id,
+            DirectionalThreadSpawnEdgeStatus::Closed,
+        ),
+        (
+            child_thread_id,
+            grandchild_thread_id,
+            DirectionalThreadSpawnEdgeStatus::Open,
+        ),
+    ] {
+        state_db
+            .upsert_thread_spawn_edge(parent, child, status)
+            .await?;
+    }
+
+    let mut mcp = TestAppServer::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let delete_id = mcp
+        .send_thread_delete_request(ThreadDeleteParams {
+            thread_id: parent_id.clone(),
+        })
+        .await?;
+    let delete_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(delete_id)),
+    )
+    .await??;
+    let _: ThreadDeleteResponse = to_response::<ThreadDeleteResponse>(delete_resp)?;
+
+    let mut deleted_ids = Vec::new();
+    for _ in 0..3 {
+        let notification = timeout(
+            DEFAULT_READ_TIMEOUT,
+            mcp.read_stream_until_notification_message("thread/deleted"),
+        )
+        .await??;
+        let deleted_notification: ThreadDeletedNotification = serde_json::from_value(
+            notification
+                .params
+                .expect("thread/deleted notification params"),
+        )?;
+        deleted_ids.push(deleted_notification.thread_id);
+    }
+    assert_eq!(deleted_ids, vec![grandchild_id, child_id, parent_id]);
+
+    for thread_id in [parent_thread_id, child_thread_id, grandchild_thread_id] {
+        let rollout_path = find_thread_path_by_id_str(
+            codex_home.path(),
+            &thread_id.to_string(),
+            /*state_db_ctx*/ None,
+        )
+        .await?;
+        assert!(
+            rollout_path.is_none(),
+            "expected active rollout for {thread_id} to be deleted"
+        );
+    }
+    assert_eq!(
+        state_db
+            .list_thread_spawn_descendants(parent_thread_id)
+            .await?,
+        Vec::<ThreadId>::new()
+    );
+    Ok(())
+}
+
+fn create_delete_test_rollout(codex_home: &Path, minute: u8, preview: &str) -> Result<String> {
+    create_fake_rollout(
+        codex_home,
+        &format!("2025-01-01T00-{minute:02}-00"),
+        &format!("2025-01-01T00:{minute:02}:00Z"),
+        preview,
+        Some("mock_provider"),
+        /*git_info*/ None,
+    )
+}
+
+#[tokio::test]
+async fn thread_delete_handles_live_threads_before_rollout_exists() -> Result<()> {
+    let codex_home = TempDir::new()?;
+
+    let mut mcp = TestAppServer::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let start_id = mcp
+        .send_thread_start_request(ThreadStartParams::default())
+        .await?;
+    let start_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
+    )
+    .await??;
+    let persisted_thread = to_response::<ThreadStartResponse>(start_resp)?.thread;
+    let rollout_path = find_thread_path_by_id_str(
+        codex_home.path(),
+        &persisted_thread.id,
+        /*state_db_ctx*/ None,
+    )
+    .await?;
+    assert_eq!(rollout_path, None);
+
+    let delete_id = mcp
+        .send_thread_delete_request(ThreadDeleteParams {
+            thread_id: persisted_thread.id,
+        })
+        .await?;
+    let delete_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(delete_id)),
+    )
+    .await??;
+    let _: ThreadDeleteResponse = to_response::<ThreadDeleteResponse>(delete_resp)?;
+
+    let start_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            ephemeral: Some(true),
+            ..Default::default()
+        })
+        .await?;
+    let start_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
+
+    let delete_id = mcp
+        .send_thread_delete_request(ThreadDeleteParams {
+            thread_id: thread.id.clone(),
+        })
+        .await?;
+    let delete_err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(delete_id)),
+    )
+    .await??;
+    let expected_message = format!(
+        "thread is not persisted and cannot be deleted: {}",
+        thread.id
+    );
+    assert_eq!(delete_err.error.message, expected_message);
+
+    let list_id = mcp
+        .send_thread_loaded_list_request(ThreadLoadedListParams::default())
+        .await?;
+    let list_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
+    )
+    .await??;
+    let ThreadLoadedListResponse { mut data, .. } =
+        to_response::<ThreadLoadedListResponse>(list_resp)?;
+    data.sort();
+    assert_eq!(data, vec![thread.id]);
+
+    Ok(())
+}

codex-rs/app-server/tests/suite/v2/thread_resume.rs

@@ -103,6 +103,7 @@ use super::analytics::assert_basic_thread_initialized_event;
 use super::analytics::mount_analytics_capture;
 use super::analytics::thread_initialized_event;
 use super::analytics::wait_for_analytics_payload;
+use super::analytics::wait_for_goal_event;
 
 #[cfg(windows)]
 const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
@@ -1316,19 +1317,30 @@ async fn thread_goal_set_edits_objective_without_resetting_usage() -> Result<()>
 }
 
 #[tokio::test]
-async fn thread_goal_clear_deletes_goal_and_notifies() -> Result<()> {
-    let server = create_mock_responses_server_repeating_assistant("Done").await;
+async fn thread_goal_lifecycle_emits_analytics_and_clear_deletes_goal() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(vec![
+        responses::sse(vec![
+            responses::ev_response_created("materialize-thread"),
+            responses::ev_completed("materialize-thread"),
+        ]),
+        responses::sse(vec![
+            responses::ev_response_created("goal-continuation"),
+            responses::ev_completed_with_tokens("goal-continuation", /*total_tokens*/ 200),
+        ]),
+    ])
+    .await;
     let codex_home = TempDir::new()?;
-    create_config_toml(codex_home.path(), &server.uri())?;
+    create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
     let config_path = codex_home.path().join("config.toml");
     let config = std::fs::read_to_string(&config_path)?;
     std::fs::write(
         &config_path,
         config.replace("personality = true\n", "personality = true\ngoals = true\n"),
     )?;
+    mount_analytics_capture(&server, codex_home.path()).await?;
 
     let mut mcp = TestAppServer::new_without_managed_config(codex_home.path()).await?;
-    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+    timeout(DEFAULT_READ_TIMEOUT.saturating_mul(2), mcp.initialize()).await??;
 
     let start_id = mcp
         .send_thread_start_request(ThreadStartParams {
@@ -1370,7 +1382,8 @@ async fn thread_goal_clear_deletes_goal_and_notifies() -> Result<()> {
             "thread/goal/set",
             Some(json!({
                 "threadId": thread.id,
-                "objective": "keep polishing",
+                "objective": "do not serialize this objective",
+                "tokenBudget": 100,
             })),
         )
         .await?;
@@ -1386,6 +1399,55 @@ async fn thread_goal_clear_deletes_goal_and_notifies() -> Result<()> {
     )
     .await??;
 
+    let created = wait_for_goal_event(&server, DEFAULT_READ_TIMEOUT, "created", "active").await?;
+    let persisted_goal_id = created["event_params"]["goal_id"]
+        .as_str()
+        .expect("created goal id");
+    assert_eq!(created["event_params"]["thread_id"], thread.id);
+    assert_eq!(created["event_params"]["turn_id"], serde_json::Value::Null);
+    assert_eq!(created["event_params"]["has_token_budget"], true);
+    assert!(created["event_params"]["session_id"].is_string());
+    assert!(created["event_params"]["app_server_client"].is_object());
+    assert!(created["event_params"]["runtime"].is_object());
+    assert!(created["event_params"].get("objective").is_none());
+    assert!(created["event_params"].get("token_budget").is_none());
+
+    let usage = wait_for_goal_event(
+        &server,
+        DEFAULT_READ_TIMEOUT,
+        "usage_accounted",
+        "budget_limited",
+    )
+    .await?;
+    let causal_turn_id = usage["event_params"]["turn_id"]
+        .as_str()
+        .expect("accounted usage turn id");
+    assert_eq!(usage["event_params"]["goal_id"], persisted_goal_id);
+    assert_eq!(usage["event_params"]["cumulative_tokens_accounted"], 200);
+    assert!(
+        usage["event_params"]["cumulative_time_accounted_seconds"]
+            .as_i64()
+            .is_some()
+    );
+
+    let status = wait_for_goal_event(
+        &server,
+        DEFAULT_READ_TIMEOUT,
+        "status_changed",
+        "budget_limited",
+    )
+    .await?;
+    assert_eq!(status["event_params"]["goal_id"], persisted_goal_id);
+    assert_eq!(status["event_params"]["turn_id"], causal_turn_id);
+    assert_eq!(
+        status["event_params"]["cumulative_tokens_accounted"],
+        serde_json::Value::Null
+    );
+    assert_eq!(
+        status["event_params"]["cumulative_time_accounted_seconds"],
+        serde_json::Value::Null
+    );
+
     let clear_id = mcp
         .send_raw_request(
             "thread/goal/clear",
@@ -1408,6 +1470,11 @@ async fn thread_goal_clear_deletes_goal_and_notifies() -> Result<()> {
     )
     .await??;
 
+    let cleared =
+        wait_for_goal_event(&server, DEFAULT_READ_TIMEOUT, "cleared", "budget_limited").await?;
+    assert_eq!(cleared["event_params"]["goal_id"], persisted_goal_id);
+    assert_eq!(cleared["event_params"]["turn_id"], serde_json::Value::Null);
+
     let get_id = mcp
         .send_raw_request(
             "thread/goal/get",

codex-rs/app-server/tests/suite/v2/turn_start.rs

@@ -43,7 +43,12 @@ use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::ServerRequestResolvedNotification;
 use codex_app_server_protocol::SubAgentActivityKind;
 use codex_app_server_protocol::TextElement;
+use codex_app_server_protocol::ThreadDeleteParams;
+use codex_app_server_protocol::ThreadDeleteResponse;
+use codex_app_server_protocol::ThreadDeletedNotification;
 use codex_app_server_protocol::ThreadItem;
+use codex_app_server_protocol::ThreadLoadedListParams;
+use codex_app_server_protocol::ThreadLoadedListResponse;
 use codex_app_server_protocol::ThreadSource;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
@@ -591,7 +596,15 @@ async fn turn_start_emits_thread_scoped_warning_notification_for_trimmed_skills(
     write_test_skill(codex_home.path(), "alpha-skill")?;
     write_test_skill(codex_home.path(), "beta-skill")?;
 
-    let mut mcp = TestAppServer::new(codex_home.path()).await?;
+    let isolated_home = codex_home.path().to_string_lossy();
+    let mut mcp = TestAppServer::new_with_env(
+        codex_home.path(),
+        &[
+            ("HOME", Some(isolated_home.as_ref())),
+            ("USERPROFILE", Some(isolated_home.as_ref())),
+        ],
+    )
+    .await?;
     timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
 
     let thread_req = mcp
@@ -3362,6 +3375,49 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<()
     assert_eq!(turn_completed.thread_id, thread.id);
     assert_eq!(turn_completed.turn.id, turn.turn.id);
 
+    // Reuse this live spawn setup to cover thread/delete's ThreadManager descendant path.
+    let delete_req = mcp
+        .send_thread_delete_request(ThreadDeleteParams {
+            thread_id: thread.id.clone(),
+        })
+        .await?;
+    let delete_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(delete_req)),
+    )
+    .await??;
+    let _: ThreadDeleteResponse = to_response::<ThreadDeleteResponse>(delete_resp)?;
+
+    let mut deleted_thread_ids = Vec::new();
+    for _ in 0..2 {
+        let deleted_notif = timeout(
+            DEFAULT_READ_TIMEOUT,
+            mcp.read_stream_until_notification_message("thread/deleted"),
+        )
+        .await??;
+        let deleted: ThreadDeletedNotification = serde_json::from_value(
+            deleted_notif
+                .params
+                .expect("thread/deleted notification params"),
+        )?;
+        deleted_thread_ids.push(deleted.thread_id);
+    }
+    assert_eq!(
+        deleted_thread_ids,
+        vec![receiver_thread_id, thread.id.clone()]
+    );
+
+    let list_req = mcp
+        .send_thread_loaded_list_request(ThreadLoadedListParams::default())
+        .await?;
+    let list_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(list_req)),
+    )
+    .await??;
+    let ThreadLoadedListResponse { data, .. } = to_response::<ThreadLoadedListResponse>(list_resp)?;
+    assert_eq!(data, Vec::<String>::new());
+
     Ok(())
 }