chore: introduce CodexServer as a clearinghouse for all conversations

codex-rs/core/src/codex_conversation.rs

@@ -0,0 +1,41 @@
+use std::sync::Arc;
+
+use tokio::sync::Notify;
+use tokio::sync::futures::Notified;
+
+use crate::codex::Codex;
+use crate::error::Result as CodexResult;
+use crate::protocol::Event;
+use crate::protocol::Op;
+use crate::protocol::Submission;
+
+pub struct CodexConversation {
+    codex: Codex,
+    cancellation_token: Arc<Notify>,
+}
+
+impl CodexConversation {
+    pub(crate) fn new(codex: Codex, cancellation_token: Arc<Notify>) -> Self {
+        Self {
+            codex,
+            cancellation_token,
+        }
+    }
+
+    pub async fn submit(&self, op: Op) -> CodexResult<String> {
+        self.codex.submit(op).await
+    }
+
+    pub async fn submit_with_id(&self, sub: Submission) -> CodexResult<()> {
+        self.codex.submit_with_id(sub).await
+    }
+
+    pub async fn next_event(&self) -> CodexResult<Event> {
+        self.codex.next_event().await
+    }
+
+    /// await this to be get notified when the user cancels the conversation.
+    pub fn on_cancel(&self) -> Notified {
+        self.cancellation_token.notified()
+    }
+}

codex-rs/core/src/codex_wrapper.rs

@@ -1,59 +0,0 @@
-use std::sync::Arc;
-
-use crate::Codex;
-use crate::CodexSpawnOk;
-use crate::config::Config;
-use crate::protocol::Event;
-use crate::protocol::EventMsg;
-use crate::util::notify_on_sigint;
-use codex_login::CodexAuth;
-use tokio::sync::Notify;
-use uuid::Uuid;
-
-/// Represents an active Codex conversation, including the first event
-/// (which is [`EventMsg::SessionConfigured`]).
-pub struct CodexConversation {
-    pub codex: Codex,
-    pub session_id: Uuid,
-    pub session_configured: Event,
-    pub ctrl_c: Arc<Notify>,
-}
-
-/// Spawn a new [`Codex`] and initialize the session.
-///
-/// Returns the wrapped [`Codex`] **and** the `SessionInitialized` event that
-/// is received as a response to the initial `ConfigureSession` submission so
-/// that callers can surface the information to the UI.
-pub async fn init_codex(config: Config) -> anyhow::Result<CodexConversation> {
-    let ctrl_c = notify_on_sigint();
-    let auth = CodexAuth::from_codex_home(&config.codex_home)?;
-    let CodexSpawnOk {
-        codex,
-        init_id,
-        session_id,
-    } = Codex::spawn(config, auth, ctrl_c.clone()).await?;
-
-    // The first event must be `SessionInitialized`. Validate and forward it to
-    // the caller so that they can display it in the conversation history.
-    let event = codex.next_event().await?;
-    if event.id != init_id
-        || !matches!(
-            &event,
-            Event {
-                id: _id,
-                msg: EventMsg::SessionConfigured(_),
-            }
-        )
-    {
-        return Err(anyhow::anyhow!(
-            "expected SessionInitialized but got {event:?}"
-        ));
-    }
-
-    Ok(CodexConversation {
-        codex,
-        session_id,
-        session_configured: event,
-        ctrl_c,
-    })
-}

codex-rs/core/src/conversation_manager.rs

@@ -0,0 +1,100 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use codex_login::CodexAuth;
+use tokio::sync::Notify;
+use tokio::sync::RwLock;
+use uuid::Uuid;
+
+use crate::codex::Codex;
+use crate::codex::CodexSpawnOk;
+use crate::codex_conversation::CodexConversation;
+use crate::config::Config;
+use crate::error::CodexErr;
+use crate::error::Result as CodexResult;
+use crate::protocol::Event;
+use crate::protocol::EventMsg;
+use crate::protocol::SessionConfiguredEvent;
+
+/// Represents an active Codex conversation, including the first event
+/// (which is [`EventMsg::SessionConfigured`]).
+pub struct NewConversation {
+    pub conversation_id: Uuid,
+    pub conversation: Arc<CodexConversation>,
+    pub session_configured: SessionConfiguredEvent,
+}
+
+/// [`ConversationManager`] is responsible for creating conversations and
+/// maintaining them in memory.
+pub struct ConversationManager {
+    conversations: Arc<RwLock<HashMap<Uuid, Arc<CodexConversation>>>>,
+}
+
+impl Default for ConversationManager {
+    fn default() -> Self {
+        Self {
+            conversations: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+}
+
+impl ConversationManager {
+    pub async fn new_conversation(&self, config: Config) -> CodexResult<NewConversation> {
+        let auth = CodexAuth::from_codex_home(&config.codex_home)?;
+        self.new_conversation_with_auth(config, auth).await
+    }
+
+    /// Used for integration tests: should not be used by ordinary business
+    /// logic.
+    pub async fn new_conversation_with_auth(
+        &self,
+        config: Config,
+        auth: Option<CodexAuth>,
+    ) -> CodexResult<NewConversation> {
+        // TODO(mbolin): Determine whether this should be wired up to SIGINT.
+        let cancellation_token = Arc::new(Notify::new());
+
+        let CodexSpawnOk {
+            codex,
+            init_id,
+            session_id: conversation_id,
+        } = Codex::spawn(config, auth, cancellation_token.clone()).await?;
+
+        // The first event must be `SessionInitialized`. Validate and forward it
+        // to the caller so that they can display it in the conversation
+        // history.
+        let event = codex.next_event().await?;
+        let session_configured = match event {
+            Event {
+                id,
+                msg: EventMsg::SessionConfigured(session_configured),
+            } if id == init_id => session_configured,
+            _ => {
+                return Err(CodexErr::SessionConfiguredNotFirstEvent);
+            }
+        };
+
+        let conversation = Arc::new(CodexConversation::new(codex, cancellation_token));
+        self.conversations
+            .write()
+            .await
+            .insert(conversation_id, conversation.clone());
+
+        Ok(NewConversation {
+            conversation_id,
+            conversation,
+            session_configured,
+        })
+    }
+
+    pub async fn get_conversation(
+        &self,
+        conversation_id: Uuid,
+    ) -> CodexResult<Arc<CodexConversation>> {
+        let conversations = self.conversations.read().await;
+        conversations
+            .get(&conversation_id)
+            .cloned()
+            .ok_or_else(|| CodexErr::ConversationNotFound(conversation_id))
+    }
+}

codex-rs/core/src/error.rs

@@ -3,6 +3,7 @@ use serde_json;
 use std::io;
 use thiserror::Error;
 use tokio::task::JoinError;
+use uuid::Uuid;
 
 pub type Result<T> = std::result::Result<T, CodexErr>;
 
@@ -44,6 +45,12 @@ pub enum CodexErr {
     #[error("stream disconnected before completion: {0}")]
     Stream(String),
 
+    #[error("no conversation with id: {0}")]
+    ConversationNotFound(Uuid),
+
+    #[error("session configured event was not the first event in the stream")]
+    SessionConfiguredNotFirstEvent,
+
     /// Returned by run_command_stream when the spawned child process timed out (10s).
     #[error("timeout waiting for child process to exit")]
     Timeout,

codex-rs/core/src/lib.rs

@@ -11,9 +11,8 @@ mod chat_completions;
 mod client;
 mod client_common;
 pub mod codex;
-pub use codex::Codex;
-pub use codex::CodexSpawnOk;
-pub mod codex_wrapper;
+mod codex_conversation;
+pub use codex_conversation::CodexConversation;
 pub mod config;
 pub mod config_profile;
 pub mod config_types;
@@ -34,6 +33,9 @@ pub use model_provider_info::ModelProviderInfo;
 pub use model_provider_info::WireApi;
 pub use model_provider_info::built_in_model_providers;
 pub use model_provider_info::create_oss_provider_with_base_url;
+mod conversation_manager;
+pub use conversation_manager::ConversationManager;
+pub use conversation_manager::NewConversation;
 pub mod model_family;
 mod models;
 mod openai_model_info;

codex-rs/core/tests/client.rs

@@ -1,14 +1,13 @@
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
-use codex_core::Codex;
-use codex_core::CodexSpawnOk;
+use codex_core::ConversationManager;
 use codex_core::ModelProviderInfo;
+use codex_core::NewConversation;
 use codex_core::WireApi;
 use codex_core::built_in_model_providers;
 use codex_core::protocol::EventMsg;
 use codex_core::protocol::InputItem;
 use codex_core::protocol::Op;
-use codex_core::protocol::SessionConfiguredEvent;
 use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
 use codex_login::CodexAuth;
 use core_test_support::load_default_config_for_test;
@@ -90,14 +89,15 @@ async fn includes_session_id_and_model_headers_in_request() {
     let mut config = load_default_config_for_test(&codex_home);
     config.model_provider = model_provider;
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        ctrl_c.clone(),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let NewConversation {
+        conversation: codex,
+        conversation_id,
+        session_configured: _,
+    } = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .expect("create new conversation");
 
     codex
         .submit(Op::UserInput {
@@ -108,13 +108,6 @@ async fn includes_session_id_and_model_headers_in_request() {
         .await
         .unwrap();
 
-    let EventMsg::SessionConfigured(SessionConfiguredEvent { session_id, .. }) =
-        wait_for_event(&codex, |ev| matches!(ev, EventMsg::SessionConfigured(_))).await
-    else {
-        unreachable!()
-    };
-
-    let current_session_id = Some(session_id.to_string());
     wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
 
     // get request from the server
@@ -123,10 +116,9 @@ async fn includes_session_id_and_model_headers_in_request() {
     let request_authorization = request.headers.get("authorization").unwrap();
     let request_originator = request.headers.get("originator").unwrap();
 
-    assert!(current_session_id.is_some());
     assert_eq!(
         request_session_id.to_str().unwrap(),
-        current_session_id.as_ref().unwrap()
+        conversation_id.to_string()
     );
     assert_eq!(request_originator.to_str().unwrap(), "codex_cli_rs");
     assert_eq!(
@@ -164,14 +156,12 @@ async fn includes_base_instructions_override_in_request() {
     config.base_instructions = Some("test instructions".to_string());
     config.model_provider = model_provider;
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        ctrl_c.clone(),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .expect("create new conversation")
+        .conversation;
 
     codex
         .submit(Op::UserInput {
@@ -223,14 +213,12 @@ async fn originator_config_override_is_used() {
     config.model_provider = model_provider;
     config.internal_originator = Some("my_override".to_string());
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        ctrl_c.clone(),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .expect("create new conversation")
+        .conversation;
 
     codex
         .submit(Op::UserInput {
@@ -283,11 +271,15 @@ async fn chatgpt_auth_sends_correct_request() {
     let codex_home = TempDir::new().unwrap();
     let mut config = load_default_config_for_test(&codex_home);
     config.model_provider = model_provider;
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } =
-        Codex::spawn(config, Some(create_dummy_codex_auth()), ctrl_c.clone())
-            .await
-            .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let NewConversation {
+        conversation: codex,
+        conversation_id,
+        session_configured: _,
+    } = conversation_manager
+        .new_conversation_with_auth(config, Some(create_dummy_codex_auth()))
+        .await
+        .expect("create new conversation");
 
     codex
         .submit(Op::UserInput {
@@ -298,13 +290,6 @@ async fn chatgpt_auth_sends_correct_request() {
         .await
         .unwrap();
 
-    let EventMsg::SessionConfigured(SessionConfiguredEvent { session_id, .. }) =
-        wait_for_event(&codex, |ev| matches!(ev, EventMsg::SessionConfigured(_))).await
-    else {
-        unreachable!()
-    };
-
-    let current_session_id = Some(session_id.to_string());
     wait_for_event(&codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await;
 
     // get request from the server
@@ -315,10 +300,9 @@ async fn chatgpt_auth_sends_correct_request() {
     let request_chatgpt_account_id = request.headers.get("chatgpt-account-id").unwrap();
     let request_body = request.body_json::<serde_json::Value>().unwrap();
 
-    assert!(current_session_id.is_some());
     assert_eq!(
         request_session_id.to_str().unwrap(),
-        current_session_id.as_ref().unwrap()
+        conversation_id.to_string()
     );
     assert_eq!(request_originator.to_str().unwrap(), "codex_cli_rs");
     assert_eq!(
@@ -361,14 +345,12 @@ async fn includes_user_instructions_message_in_request() {
     config.model_provider = model_provider;
     config.user_instructions = Some("be nice".to_string());
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        ctrl_c.clone(),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .expect("create new conversation")
+        .conversation;
 
     codex
         .submit(Op::UserInput {
@@ -457,8 +439,12 @@ async fn azure_overrides_assign_properties_used_for_responses_url() {
     let mut config = load_default_config_for_test(&codex_home);
     config.model_provider = provider;
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(config, None, ctrl_c.clone()).await.unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, None)
+        .await
+        .expect("create new conversation")
+        .conversation;
 
     codex
         .submit(Op::UserInput {
@@ -531,11 +517,12 @@ async fn env_var_overrides_loaded_auth() {
     let mut config = load_default_config_for_test(&codex_home);
     config.model_provider = provider;
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } =
-        Codex::spawn(config, Some(create_dummy_codex_auth()), ctrl_c.clone())
-            .await
-            .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(create_dummy_codex_auth()))
+        .await
+        .expect("create new conversation")
+        .conversation;
 
     codex
         .submit(Op::UserInput {

codex-rs/core/tests/common/lib.rs

@@ -2,6 +2,7 @@
 
 use tempfile::TempDir;
 
+use codex_core::CodexConversation;
 use codex_core::config::Config;
 use codex_core::config::ConfigOverrides;
 use codex_core::config::ConfigToml;
@@ -72,7 +73,7 @@ pub fn load_sse_fixture_with_id(path: impl AsRef<std::path::Path>, id: &str) ->
 }
 
 pub async fn wait_for_event<F>(
-    codex: &codex_core::Codex,
+    codex: &CodexConversation,
     predicate: F,
 ) -> codex_core::protocol::EventMsg
 where
@@ -83,7 +84,7 @@ where
 }
 
 pub async fn wait_for_event_with_timeout<F>(
-    codex: &codex_core::Codex,
+    codex: &CodexConversation,
     mut predicate: F,
     wait_time: tokio::time::Duration,
 ) -> codex_core::protocol::EventMsg

codex-rs/core/tests/compact.rs

@@ -1,7 +1,6 @@
 #![expect(clippy::unwrap_used)]
 
-use codex_core::Codex;
-use codex_core::CodexSpawnOk;
+use codex_core::ConversationManager;
 use codex_core::ModelProviderInfo;
 use codex_core::built_in_model_providers;
 use codex_core::protocol::EventMsg;
@@ -142,14 +141,12 @@ async fn summarize_context_three_requests_and_instructions() {
     let home = TempDir::new().unwrap();
     let mut config = load_default_config_for_test(&home);
     config.model_provider = model_provider;
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("dummy")),
-        ctrl_c.clone(),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("dummy")))
+        .await
+        .unwrap()
+        .conversation;
 
     // 1) Normal user input – should hit server once.
     codex

codex-rs/core/tests/live_agent.rs

@@ -1,209 +0,0 @@
-#![expect(clippy::unwrap_used, clippy::expect_used)]
-
-//! Live integration tests that exercise the full [`Agent`] stack **against the real
-//! OpenAI `/v1/responses` API**.  These tests complement the lightweight mock‑based
-//! unit tests by verifying that the agent can drive an end‑to‑end conversation,
-//! stream incremental events, execute function‑call tool invocations and safely
-//! chain multiple turns inside a single session – the exact scenarios that have
-//! historically been brittle.
-//!
-//! The live tests are **ignored by default** so CI remains deterministic and free
-//! of external dependencies.  Developers can opt‑in locally with e.g.
-//!
-//! ```bash
-//! OPENAI_API_KEY=sk‑... cargo test --test live_agent -- --ignored --nocapture
-//! ```
-//!
-//! Make sure your key has access to the experimental *Responses* API and that
-//! any billable usage is acceptable.
-
-use std::time::Duration;
-
-use codex_core::Codex;
-use codex_core::CodexSpawnOk;
-use codex_core::error::CodexErr;
-use codex_core::protocol::AgentMessageEvent;
-use codex_core::protocol::ErrorEvent;
-use codex_core::protocol::EventMsg;
-use codex_core::protocol::InputItem;
-use codex_core::protocol::Op;
-use core_test_support::load_default_config_for_test;
-use tempfile::TempDir;
-use tokio::sync::Notify;
-use tokio::time::timeout;
-
-fn api_key_available() -> bool {
-    std::env::var("OPENAI_API_KEY").is_ok()
-}
-
-/// Helper that spawns a fresh Agent and sends the mandatory *ConfigureSession*
-/// submission.  The caller receives the constructed [`Agent`] plus the unique
-/// submission id used for the initialization message.
-async fn spawn_codex() -> Result<Codex, CodexErr> {
-    assert!(
-        api_key_available(),
-        "OPENAI_API_KEY must be set for live tests"
-    );
-
-    let codex_home = TempDir::new().unwrap();
-    let mut config = load_default_config_for_test(&codex_home);
-    config.model_provider.request_max_retries = Some(2);
-    config.model_provider.stream_max_retries = Some(2);
-    let CodexSpawnOk { codex: agent, .. } =
-        Codex::spawn(config, None, std::sync::Arc::new(Notify::new())).await?;
-
-    Ok(agent)
-}
-
-/// Verifies that the agent streams incremental *AgentMessage* events **before**
-/// emitting `TaskComplete` and that a second task inside the same session does
-/// not get tripped up by a stale `previous_response_id`.
-#[ignore]
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn live_streaming_and_prev_id_reset() {
-    if !api_key_available() {
-        eprintln!("skipping live_streaming_and_prev_id_reset – OPENAI_API_KEY not set");
-        return;
-    }
-
-    let codex = spawn_codex().await.unwrap();
-
-    // ---------- Task 1 ----------
-    codex
-        .submit(Op::UserInput {
-            items: vec![InputItem::Text {
-                text: "Say the words 'stream test'".into(),
-            }],
-        })
-        .await
-        .unwrap();
-
-    let mut saw_message_before_complete = false;
-    loop {
-        let ev = timeout(Duration::from_secs(60), codex.next_event())
-            .await
-            .expect("timeout waiting for task1 events")
-            .expect("agent closed");
-
-        match ev.msg {
-            EventMsg::AgentMessage(_) => saw_message_before_complete = true,
-            EventMsg::TaskComplete(_) => break,
-            EventMsg::Error(ErrorEvent { message }) => {
-                panic!("agent reported error in task1: {message}")
-            }
-            _ => {
-                // Ignore other events.
-            }
-        }
-    }
-
-    assert!(
-        saw_message_before_complete,
-        "Agent did not stream any AgentMessage before TaskComplete"
-    );
-
-    // ---------- Task 2 (same session) ----------
-    codex
-        .submit(Op::UserInput {
-            items: vec![InputItem::Text {
-                text: "Respond with exactly: second turn succeeded".into(),
-            }],
-        })
-        .await
-        .unwrap();
-
-    let mut got_expected = false;
-    loop {
-        let ev = timeout(Duration::from_secs(60), codex.next_event())
-            .await
-            .expect("timeout waiting for task2 events")
-            .expect("agent closed");
-
-        match &ev.msg {
-            EventMsg::AgentMessage(AgentMessageEvent { message })
-                if message.contains("second turn succeeded") =>
-            {
-                got_expected = true;
-            }
-            EventMsg::TaskComplete(_) => break,
-            EventMsg::Error(ErrorEvent { message }) => {
-                panic!("agent reported error in task2: {message}")
-            }
-            _ => {
-                // Ignore other events.
-            }
-        }
-    }
-
-    assert!(got_expected, "second task did not receive expected answer");
-}
-
-/// Exercises a *function‑call → shell execution* round‑trip by instructing the
-/// model to run a harmless `echo` command.  The test asserts that:
-///   1. the function call is executed (we see `ExecCommandBegin`/`End` events)
-///   2. the captured stdout reaches the client unchanged.
-#[ignore]
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn live_shell_function_call() {
-    if !api_key_available() {
-        eprintln!("skipping live_shell_function_call – OPENAI_API_KEY not set");
-        return;
-    }
-
-    let codex = spawn_codex().await.unwrap();
-
-    const MARKER: &str = "codex_live_echo_ok";
-
-    codex
-        .submit(Op::UserInput {
-            items: vec![InputItem::Text {
-                text: format!(
-                    "Use the shell function to run the command `echo {MARKER}` and no other commands."
-                ),
-            }],
-        })
-        .await
-        .unwrap();
-
-    let mut saw_begin = false;
-    let mut saw_end_with_output = false;
-
-    loop {
-        let ev = timeout(Duration::from_secs(60), codex.next_event())
-            .await
-            .expect("timeout waiting for function‑call events")
-            .expect("agent closed");
-
-        match ev.msg {
-            EventMsg::ExecCommandBegin(codex_core::protocol::ExecCommandBeginEvent {
-                command,
-                ..
-            }) => {
-                assert_eq!(command, vec!["echo", MARKER]);
-                saw_begin = true;
-            }
-            EventMsg::ExecCommandEnd(codex_core::protocol::ExecCommandEndEvent {
-                stdout,
-                exit_code,
-                ..
-            }) => {
-                assert_eq!(exit_code, 0, "echo returned non‑zero exit code");
-                assert!(stdout.contains(MARKER));
-                saw_end_with_output = true;
-            }
-            EventMsg::TaskComplete(_) => break,
-            EventMsg::Error(codex_core::protocol::ErrorEvent { message }) => {
-                panic!("agent error during shell test: {message}")
-            }
-            _ => {
-                // Ignore other events.
-            }
-        }
-    }
-
-    assert!(saw_begin, "ExecCommandBegin event missing");
-    assert!(
-        saw_end_with_output,
-        "ExecCommandEnd with expected output missing"
-    );
-}

codex-rs/core/tests/prompt_caching.rs

@@ -1,7 +1,6 @@
 #![allow(clippy::expect_used, clippy::unwrap_used)]
 
-use codex_core::Codex;
-use codex_core::CodexSpawnOk;
+use codex_core::ConversationManager;
 use codex_core::ModelProviderInfo;
 use codex_core::built_in_model_providers;
 use codex_core::protocol::EventMsg;
@@ -55,14 +54,12 @@ async fn prefixes_context_and_instructions_once_and_consistently_across_requests
     config.model_provider = model_provider;
     config.user_instructions = Some("be consistent and helpful".to_string());
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        ctrl_c.clone(),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .expect("create new conversation")
+        .conversation;
 
     codex
         .submit(Op::UserInput {

codex-rs/core/tests/stream_error_allows_next_turn.rs

@@ -1,7 +1,6 @@
 use std::time::Duration;
 
-use codex_core::Codex;
-use codex_core::CodexSpawnOk;
+use codex_core::ConversationManager;
 use codex_core::ModelProviderInfo;
 use codex_core::WireApi;
 use codex_core::protocol::EventMsg;
@@ -90,13 +89,12 @@ async fn continue_after_stream_error() {
     config.base_instructions = Some("You are a helpful assistant".to_string());
     config.model_provider = provider;
 
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        std::sync::Arc::new(tokio::sync::Notify::new()),
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .unwrap()
+        .conversation;
 
     codex
         .submit(Op::UserInput {

codex-rs/core/tests/stream_no_completed.rs

@@ -3,8 +3,7 @@
 
 use std::time::Duration;
 
-use codex_core::Codex;
-use codex_core::CodexSpawnOk;
+use codex_core::ConversationManager;
 use codex_core::ModelProviderInfo;
 use codex_core::protocol::EventMsg;
 use codex_core::protocol::InputItem;
@@ -93,17 +92,15 @@ async fn retries_on_early_close() {
         requires_openai_auth: false,
     };
 
-    let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());
     let codex_home = TempDir::new().unwrap();
     let mut config = load_default_config_for_test(&codex_home);
     config.model_provider = model_provider;
-    let CodexSpawnOk { codex, .. } = Codex::spawn(
-        config,
-        Some(CodexAuth::from_api_key("Test API Key")),
-        ctrl_c,
-    )
-    .await
-    .unwrap();
+    let conversation_manager = ConversationManager::default();
+    let codex = conversation_manager
+        .new_conversation_with_auth(config, Some(CodexAuth::from_api_key("Test API Key")))
+        .await
+        .unwrap()
+        .conversation;
 
     codex
         .submit(Op::UserInput {