openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-17 12:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: introduce CodexServer as a clearinghouse for all conversations

Browse Source
This commit is contained in:
Michael Bolin
2025-08-12 12:36:20 -07:00
parent e8670ad840
commit 0255cb3ad0
15 changed files with 263 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
codex-rs/cli/src/proto.rs
View File

@@ -1,15 +1,13 @@
use std::io::IsTerminal;
use std::sync::Arc;
use clap::Parser;
use codex_common::CliConfigOverrides;
use codex_core::Codex;
use codex_core::CodexSpawnOk;
use codex_core::config::Config;
use codex_core::config::ConfigOverrides;
use codex_core::protocol::Submission;
use codex_core::server::CodexServer;
use codex_core::server::NewConversation;
use codex_core::util::notify_on_sigint;
use codex_login::CodexAuth;
use tokio::io::AsyncBufReadExt;
use tokio::io::BufReader;
use tracing::error;
@@ -36,14 +34,18 @@ pub async fn run_main(opts: ProtoCli) -> anyhow::Result<()> {
.map_err(anyhow::Error::msg)?;
let config = Config::load_with_cli_overrides(overrides_vec, ConfigOverrides::default())?;
let auth = CodexAuth::from_codex_home(&config.codex_home)?;
let ctrl_c = notify_on_sigint();
let CodexSpawnOk { codex, .. } = Codex::spawn(config, auth, ctrl_c.clone()).await?;
let codex = Arc::new(codex);
// Use server API to start a conversation
let server = CodexServer::default();
let NewConversation {
conversation_id: _,
conversation,
session_configured: _,
} = server.new_conversation(config).await?;
// Task that reads JSON lines from stdin and forwards to Submission Queue
let sq_fut = {
let codex = codex.clone();
let conversation = conversation.clone();
let ctrl_c = ctrl_c.clone();
async move {
let stdin = BufReader::new(tokio::io::stdin());
@@ -65,7 +67,7 @@ pub async fn run_main(opts: ProtoCli) -> anyhow::Result<()> {
}
match serde_json::from_str::<Submission>(line) {
Ok(sub) => {
if let Err(e) = codex.submit_with_id(sub).await {
if let Err(e) = conversation.submit_with_id(sub).await {
error!("{e:#}");
break;
}
@@ -88,8 +90,8 @@ pub async fn run_main(opts: ProtoCli) -> anyhow::Result<()> {
let eq_fut = async move {
loop {
let event = tokio::select! {
_ = ctrl_c.notified() => break,
event = codex.next_event() => event,
_ = conversation.on_cancel() => break,
event = conversation.next_event() => event,
};
match event {
Ok(event) => {

59
codex-rs/core/src/codex_wrapper.rs
View File

@@ -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,
})
}

7
codex-rs/core/src/error.rs
View File

@@ -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,

2
codex-rs/core/src/lib.rs
View File

@@ -13,7 +13,6 @@ mod client_common;
pub mod codex;
pub use codex::Codex;
pub use codex::CodexSpawnOk;
pub mod codex_wrapper;
pub mod config;
pub mod config_profile;
pub mod config_types;
@@ -44,6 +43,7 @@ pub mod protocol;
mod rollout;
pub(crate) mod safety;
pub mod seatbelt;
pub mod server;
pub mod shell;
pub mod spawn;
pub mod turn_diff_tracker;

118
codex-rs/core/src/server.rs Normal file
View File

@@ -0,0 +1,118 @@
use std::collections::HashMap;
use std::sync::Arc;
use codex_login::CodexAuth;
use tokio::sync::Notify;
use tokio::sync::RwLock;
use tokio::sync::futures::Notified;
use uuid::Uuid;
use crate::Codex;
use crate::CodexSpawnOk;
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::Op;
use crate::protocol::SessionConfiguredEvent;
use crate::protocol::Submission;
/// Codex "server" that manages multiple conversations.
pub struct CodexServer {
conversations: Arc<RwLock<HashMap<Uuid, Arc<CodexConversation>>>>,
}
/// Represents an active Codex conversation, including the first event
/// (which is [`EventMsg::SessionConfigured`]).
pub struct CodexConversation {
codex: Codex,
cancellation_token: Arc<Notify>,
}
impl CodexConversation {
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()
}
}
pub struct NewConversation {
pub conversation_id: Uuid,
pub conversation: Arc<CodexConversation>,
pub session_configured: SessionConfiguredEvent,
}
impl Default for CodexServer {
fn default() -> Self {
Self {
conversations: Arc::new(RwLock::new(HashMap::new())),
}
}
}
impl CodexServer {
pub async fn new_conversation(&self, config: Config) -> CodexResult<NewConversation> {
// TODO(mbolin): Determine whether this should be wired up to SIGINT.
let cancellation_token = Arc::new(Notify::new());
let auth = CodexAuth::from_codex_home(&config.codex_home)?;
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 {
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))
}
}

32
codex-rs/exec/src/lib.rs
View File

@@ -6,12 +6,9 @@ mod event_processor_with_json_output;
use std::io::IsTerminal;
use std::io::Read;
use std::path::PathBuf;
use std::sync::Arc;
pub use cli::Cli;
use codex_core::BUILT_IN_OSS_MODEL_PROVIDER_ID;
use codex_core::codex_wrapper::CodexConversation;
use codex_core::codex_wrapper::{self};
use codex_core::config::Config;
use codex_core::config::ConfigOverrides;
use codex_core::config_types::SandboxMode;
@@ -21,6 +18,8 @@ use codex_core::protocol::EventMsg;
use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::TaskCompleteEvent;
use codex_core::server::CodexServer;
use codex_core::server::NewConversation;
use codex_core::util::is_inside_git_repo;
use codex_ollama::DEFAULT_OSS_MODEL;
use event_processor_with_human_output::EventProcessorWithHumanOutput;
@@ -185,25 +184,24 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
std::process::exit(1);
}
let CodexConversation {
codex: codex_wrapper,
let server = CodexServer::default();
let NewConversation {
conversation_id: _,
conversation,
session_configured,
ctrl_c,
..
} = codex_wrapper::init_codex(config).await?;
let codex = Arc::new(codex_wrapper);
} = server.new_conversation(config).await?;
info!("Codex initialized with event: {session_configured:?}");
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Event>();
{
let codex = codex.clone();
let conversation = conversation.clone();
tokio::spawn(async move {
loop {
let interrupted = ctrl_c.notified();
let interrupted = conversation.on_cancel();
tokio::select! {
_ = interrupted => {
// Forward an interrupt to the codex so it can abort any in‑flight task.
let _ = codex
let _ = conversation
.submit(
Op::Interrupt,
)
@@ -213,7 +211,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
// will emit a `TurnInterrupted` (Error) event which is drained later.
break;
}
res = codex.next_event() => match res {
res = conversation.next_event() => match res {
Ok(event) => {
debug!("Received event: {event:?}");
@@ -243,9 +241,9 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
.into_iter()
.map(|path| InputItem::LocalImage { path })
.collect();
let initial_images_event_id = codex.submit(Op::UserInput { items }).await?;
let initial_images_event_id = conversation.submit(Op::UserInput { items }).await?;
info!("Sent images with event ID: {initial_images_event_id}");
while let Ok(event) = codex.next_event().await {
while let Ok(event) = conversation.next_event().await {
if event.id == initial_images_event_id
&& matches!(
event.msg,
@@ -261,7 +259,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
// Send the prompt.
let items: Vec<InputItem> = vec![InputItem::Text { text: prompt }];
let initial_prompt_task_id = codex.submit(Op::UserInput { items }).await?;
let initial_prompt_task_id = conversation.submit(Op::UserInput { items }).await?;
info!("Sent prompt with event ID: {initial_prompt_task_id}");
// Run the loop until the task is complete.
@@ -270,7 +268,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
match shutdown {
CodexStatus::Running => continue,
CodexStatus::InitiateShutdown => {
codex.submit(Op::Shutdown).await?;
conversation.submit(Op::Shutdown).await?;
}
CodexStatus::Shutdown => {
break;

51
codex-rs/mcp-server/src/codex_tool_runner.rs
View File

@@ -5,18 +5,19 @@
use std::collections::HashMap;
use std::sync::Arc;
use codex_core::Codex;
use codex_core::codex_wrapper::CodexConversation;
use codex_core::codex_wrapper::init_codex;
use codex_core::config::Config as CodexConfig;
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::ApplyPatchApprovalRequestEvent;
use codex_core::protocol::Event;
use codex_core::protocol::EventMsg;
use codex_core::protocol::ExecApprovalRequestEvent;
use codex_core::protocol::InputItem;
use codex_core::protocol::Op;
use codex_core::protocol::Submission;
use codex_core::protocol::TaskCompleteEvent;
use codex_core::server::CodexConversation;
use codex_core::server::CodexServer;
use codex_core::server::NewConversation;
use mcp_types::CallToolResult;
use mcp_types::ContentBlock;
use mcp_types::RequestId;
@@ -41,15 +42,14 @@ pub async fn run_codex_tool_session(
initial_prompt: String,
config: CodexConfig,
outgoing: Arc<OutgoingMessageSender>,
session_map: Arc<Mutex<HashMap<Uuid, Arc<Codex>>>>,
codex_server: Arc<CodexServer>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
) {
let CodexConversation {
codex,
let NewConversation {
conversation_id,
conversation,
session_configured,
session_id,
..
} = match init_codex(config).await {
} = match codex_server.new_conversation(config).await {
Ok(res) => res,
Err(e) => {
let result = CallToolResult {
@@ -65,16 +65,15 @@ pub async fn run_codex_tool_session(
return;
}
};
let codex = Arc::new(codex);
// update the session map so we can retrieve the session in a reply, and then drop it, since
// we no longer need it for this function
session_map.lock().await.insert(session_id, codex.clone());
drop(session_map);
let session_configured_event = Event {
// Use a fake id value for now.
id: "".to_string(),
msg: EventMsg::SessionConfigured(session_configured.clone()),
};
outgoing
.send_event_as_notification(
&session_configured,
&session_configured_event,
Some(OutgoingNotificationMeta::new(Some(id.clone()))),
)
.await;
@@ -89,7 +88,7 @@ pub async fn run_codex_tool_session(
running_requests_id_to_codex_uuid
.lock()
.await
.insert(id.clone(), session_id);
.insert(id.clone(), conversation_id);
let submission = Submission {
id: sub_id.clone(),
op: Op::UserInput {
@@ -99,18 +98,24 @@ pub async fn run_codex_tool_session(
},
};
if let Err(e) = codex.submit_with_id(submission).await {
if let Err(e) = conversation.submit_with_id(submission).await {
tracing::error!("Failed to submit initial prompt: {e}");
// unregister the id so we don't keep it in the map
running_requests_id_to_codex_uuid.lock().await.remove(&id);
return;
}
run_codex_tool_session_inner(codex, outgoing, id, running_requests_id_to_codex_uuid).await;
run_codex_tool_session_inner(
conversation,
outgoing,
id,
running_requests_id_to_codex_uuid,
)
.await;
}
pub async fn run_codex_tool_session_reply(
codex: Arc<Codex>,
conversation: Arc<CodexConversation>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
prompt: String,
@@ -121,7 +126,7 @@ pub async fn run_codex_tool_session_reply(
.lock()
.await
.insert(request_id.clone(), session_id);
if let Err(e) = codex
if let Err(e) = conversation
.submit(Op::UserInput {
items: vec![InputItem::Text { text: prompt }],
})
@@ -137,7 +142,7 @@ pub async fn run_codex_tool_session_reply(
}
run_codex_tool_session_inner(
codex,
conversation,
outgoing,
request_id,
running_requests_id_to_codex_uuid,
@@ -146,7 +151,7 @@ pub async fn run_codex_tool_session_reply(
}
async fn run_codex_tool_session_inner(
codex: Arc<Codex>,
codex: Arc<CodexConversation>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,

4
codex-rs/mcp-server/src/conversation_loop.rs
View File

@@ -4,16 +4,16 @@ use crate::exec_approval::handle_exec_approval_request;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotificationMeta;
use crate::patch_approval::handle_patch_approval_request;
use codex_core::Codex;
use codex_core::protocol::AgentMessageEvent;
use codex_core::protocol::ApplyPatchApprovalRequestEvent;
use codex_core::protocol::EventMsg;
use codex_core::protocol::ExecApprovalRequestEvent;
use codex_core::server::CodexConversation;
use mcp_types::RequestId;
use tracing::error;
pub async fn run_conversation_loop(
codex: Arc<Codex>,
codex: Arc<CodexConversation>,
outgoing: Arc<OutgoingMessageSender>,
request_id: RequestId,
) {

6
codex-rs/mcp-server/src/exec_approval.rs
View File

@@ -1,9 +1,9 @@
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::Codex;
use codex_core::protocol::Op;
use codex_core::protocol::ReviewDecision;
use codex_core::server::CodexConversation;
use mcp_types::ElicitRequest;
use mcp_types::ElicitRequestParamsRequestedSchema;
use mcp_types::JSONRPCErrorError;
@@ -51,7 +51,7 @@ pub(crate) async fn handle_exec_approval_request(
command: Vec<String>,
cwd: PathBuf,
outgoing: Arc<crate::outgoing_message::OutgoingMessageSender>,
codex: Arc<Codex>,
codex: Arc<CodexConversation>,
request_id: RequestId,
tool_call_id: String,
event_id: String,
@@ -116,7 +116,7 @@ pub(crate) async fn handle_exec_approval_request(
async fn on_exec_approval_response(
event_id: String,
receiver: tokio::sync::oneshot::Receiver<mcp_types::Result>,
codex: Arc<Codex>,
codex: Arc<CodexConversation>,
) {
let response = receiver.await;
let value = match response {

73
codex-rs/mcp-server/src/message_processor.rs
View File

@@ -14,9 +14,9 @@ use crate::outgoing_message::OutgoingMessageSender;
use crate::tool_handlers::create_conversation::handle_create_conversation;
use crate::tool_handlers::send_message::handle_send_message;
use codex_core::Codex;
use codex_core::config::Config as CodexConfig;
use codex_core::protocol::Submission;
use codex_core::server::CodexServer;
use mcp_types::CallToolRequest;
use mcp_types::CallToolRequestParams;
use mcp_types::CallToolResult;
@@ -42,7 +42,7 @@ pub(crate) struct MessageProcessor {
outgoing: Arc<OutgoingMessageSender>,
initialized: bool,
codex_linux_sandbox_exe: Option<PathBuf>,
session_map: Arc<Mutex<HashMap<Uuid, Arc<Codex>>>>,
codex_server: Arc<CodexServer>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
running_session_ids: Arc<Mutex<HashSet<Uuid>>>,
}
@@ -58,14 +58,14 @@ impl MessageProcessor {
outgoing: Arc::new(outgoing),
initialized: false,
codex_linux_sandbox_exe,
session_map: Arc::new(Mutex::new(HashMap::new())),
codex_server: Arc::new(CodexServer::default()),
running_requests_id_to_codex_uuid: Arc::new(Mutex::new(HashMap::new())),
running_session_ids: Arc::new(Mutex::new(HashSet::new())),
}
}
pub(crate) fn session_map(&self) -> Arc<Mutex<HashMap<Uuid, Arc<Codex>>>> {
self.session_map.clone()
pub(crate) fn codex_server(&self) -> Arc<CodexServer> {
self.codex_server.clone()
}
pub(crate) fn outgoing(&self) -> Arc<OutgoingMessageSender> {
@@ -431,9 +431,9 @@ impl MessageProcessor {
}
};
// Clone outgoing and session map to move into async task.
// Clone outgoing and server to move into async task.
let outgoing = self.outgoing.clone();
let session_map = self.session_map.clone();
let codex_server = self.codex_server.clone();
let running_requests_id_to_codex_uuid = self.running_requests_id_to_codex_uuid.clone();
// Spawn an async task to handle the Codex session so that we do not
@@ -445,7 +445,7 @@ impl MessageProcessor {
initial_prompt,
config,
outgoing,
session_map,
codex_server,
running_requests_id_to_codex_uuid,
)
.await;
@@ -516,33 +516,27 @@ impl MessageProcessor {
}
};
// load codex from session map
let session_map_mutex = Arc::clone(&self.session_map);
// Clone outgoing and session map to move into async task.
// Clone outgoing to move into async task.
let outgoing = self.outgoing.clone();
let running_requests_id_to_codex_uuid = self.running_requests_id_to_codex_uuid.clone();
let codex = {
let session_map = session_map_mutex.lock().await;
match session_map.get(&session_id).cloned() {
Some(c) => c,
None => {
tracing::warn!("Session not found for session_id: {session_id}");
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_owned(),
text: format!("Session not found for session_id: {session_id}"),
annotations: None,
})],
is_error: Some(true),
structured_content: None,
};
outgoing
.send_response(request_id, serde_json::to_value(result).unwrap_or_default())
.await;
return;
}
let codex = match self.codex_server.get_conversation(session_id).await {
Ok(c) => c,
Err(_) => {
tracing::warn!("Session not found for session_id: {session_id}");
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_owned(),
text: format!("Session not found for session_id: {session_id}"),
annotations: None,
})],
is_error: Some(true),
structured_content: None,
};
outgoing
.send_response(request_id, serde_json::to_value(result).unwrap_or_default())
.await;
return;
}
};
@@ -609,15 +603,12 @@ impl MessageProcessor {
};
tracing::info!("session_id: {session_id}");
// Obtain the Codex Arc while holding the session_map lock, then release.
let codex_arc = {
let sessions_guard = self.session_map.lock().await;
match sessions_guard.get(&session_id) {
Some(codex) => Arc::clone(codex),
None => {
tracing::warn!("Session not found for session_id: {session_id}");
return;
}
// Obtain the Codex conversation from the server.
let codex_arc = match self.codex_server.get_conversation(session_id).await {
Ok(c) => c,
Err(_) => {
tracing::warn!("Session not found for session_id: {session_id}");
return;
}
};

6
codex-rs/mcp-server/src/patch_approval.rs
View File

@@ -2,10 +2,10 @@ use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::Codex;
use codex_core::protocol::FileChange;
use codex_core::protocol::Op;
use codex_core::protocol::ReviewDecision;
use codex_core::server::CodexConversation;
use mcp_types::ElicitRequest;
use mcp_types::ElicitRequestParamsRequestedSchema;
use mcp_types::JSONRPCErrorError;
@@ -47,7 +47,7 @@ pub(crate) async fn handle_patch_approval_request(
grant_root: Option<PathBuf>,
changes: HashMap<PathBuf, FileChange>,
outgoing: Arc<OutgoingMessageSender>,
codex: Arc<Codex>,
codex: Arc<CodexConversation>,
request_id: RequestId,
tool_call_id: String,
event_id: String,
@@ -111,7 +111,7 @@ pub(crate) async fn handle_patch_approval_request(
pub(crate) async fn on_patch_approval_response(
event_id: String,
receiver: tokio::sync::oneshot::Receiver<mcp_types::Result>,
codex: Arc<Codex>,
codex: Arc<CodexConversation>,
) {
let response = receiver.await;
let value = match response {

58
codex-rs/mcp-server/src/tool_handlers/create_conversation.rs
View File

@@ -1,16 +1,9 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::Codex;
use codex_core::codex_wrapper::init_codex;
use codex_core::config::Config as CodexConfig;
use codex_core::config::ConfigOverrides;
use codex_core::protocol::EventMsg;
use codex_core::protocol::SessionConfiguredEvent;
use codex_core::server::NewConversation;
use mcp_types::RequestId;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::conversation_loop::run_conversation_loop;
use crate::json_to_toml::json_to_toml;
@@ -81,8 +74,12 @@ pub(crate) async fn handle_create_conversation(
}
};
// Initialize Codex session
let codex_conversation = match init_codex(cfg).await {
// Initialize Codex session via server API
let NewConversation {
conversation_id: session_id,
conversation,
session_configured,
} = match message_processor.codex_server().new_conversation(cfg).await {
Ok(conv) => conv,
Err(e) => {
message_processor
@@ -100,41 +97,13 @@ pub(crate) async fn handle_create_conversation(
}
};
// Expect SessionConfigured; if not, return error.
let EventMsg::SessionConfigured(SessionConfiguredEvent { model, .. }) =
&codex_conversation.session_configured.msg
else {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationCreate(
ConversationCreateResult::Error {
message: "Expected SessionConfigured event".to_string(),
},
)),
Some(true),
)
.await;
return;
};
let effective_model = session_configured.model.clone();
let effective_model = model.clone();
let session_id = codex_conversation.session_id;
let codex_arc = Arc::new(codex_conversation.codex);
// Store session for future calls
insert_session(
session_id,
codex_arc.clone(),
message_processor.session_map(),
)
.await;
// Run the conversation loop in the background so this request can return immediately.
let outgoing = message_processor.outgoing();
let spawn_id = id.clone();
tokio::spawn(async move {
run_conversation_loop(codex_arc.clone(), outgoing, spawn_id).await;
run_conversation_loop(conversation.clone(), outgoing, spawn_id).await;
});
// Reply with the new conversation id and effective model
@@ -151,12 +120,3 @@ pub(crate) async fn handle_create_conversation(
)
.await;
}
async fn insert_session(
session_id: Uuid,
codex: Arc<Codex>,
session_map: Arc<Mutex<HashMap<Uuid, Arc<Codex>>>>,
) {
let mut guard = session_map.lock().await;
guard.insert(session_id, codex);
}

21
codex-rs/mcp-server/src/tool_handlers/send_message.rs
View File

@@ -1,12 +1,6 @@
use std::collections::HashMap;
use std::sync::Arc;
use codex_core::Codex;
use codex_core::protocol::Op;
use codex_core::protocol::Submission;
use mcp_types::RequestId;
use tokio::sync::Mutex;
use uuid::Uuid;
use crate::mcp_protocol::ConversationSendMessageArgs;
use crate::mcp_protocol::ConversationSendMessageResult;
@@ -41,7 +35,11 @@ pub(crate) async fn handle_send_message(
}
let session_id = conversation_id.0;
let Some(codex) = get_session(session_id, message_processor.session_map()).await else {
let Ok(codex) = message_processor
.codex_server()
.get_conversation(session_id)
.await
else {
message_processor
.send_response_with_optional_error(
id,
@@ -114,11 +112,4 @@ pub(crate) async fn handle_send_message(
)
.await;
}
pub(crate) async fn get_session(
session_id: Uuid,
session_map: Arc<Mutex<HashMap<Uuid, Arc<Codex>>>>,
) -> Option<Arc<Codex>> {
let guard = session_map.lock().await;
guard.get(&session_id).cloned()
}
// Session lookup is delegated to CodexServer now.

8
codex-rs/tui/src/app.rs
View File

@@ -12,6 +12,7 @@ use crate::tui;
use codex_core::config::Config;
use codex_core::protocol::Event;
use codex_core::protocol::Op;
use codex_core::server::CodexServer;
use color_eyre::eyre::Result;
use crossterm::SynchronizedUpdate;
use crossterm::event::KeyCode;
@@ -48,6 +49,7 @@ enum AppState<'a> {
}
pub(crate) struct App<'a> {
server: Arc<CodexServer>,
app_event_tx: AppEventSender,
app_event_rx: Receiver<AppEvent>,
app_state: AppState<'a>,
@@ -82,6 +84,8 @@ impl App<'_> {
initial_images: Vec<std::path::PathBuf>,
show_trust_screen: bool,
) -> Self {
let server = Arc::new(CodexServer::default());
let (app_event_tx, app_event_rx) = channel();
let app_event_tx = AppEventSender::new(app_event_tx);
let pending_redraw = Arc::new(AtomicBool::new(false));
@@ -152,6 +156,7 @@ impl App<'_> {
} else {
let chat_widget = ChatWidget::new(
config.clone(),
server.clone(),
app_event_tx.clone(),
initial_prompt,
initial_images,
@@ -164,6 +169,7 @@ impl App<'_> {
let file_search = FileSearchManager::new(config.cwd.clone(), app_event_tx.clone());
Self {
server,
app_event_tx,
pending_history_lines: Vec::new(),
app_event_rx,
@@ -316,6 +322,7 @@ impl App<'_> {
// User accepted – switch to chat view.
let new_widget = Box::new(ChatWidget::new(
self.config.clone(),
self.server.clone(),
self.app_event_tx.clone(),
None,
Vec::new(),
@@ -437,6 +444,7 @@ impl App<'_> {
self.app_state = AppState::Chat {
widget: Box::new(ChatWidget::new(
config,
self.server.clone(),
app_event_tx.clone(),
initial_prompt,
initial_images,

32
codex-rs/tui/src/chatwidget.rs
View File

@@ -2,8 +2,6 @@ use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::codex_wrapper::CodexConversation;
use codex_core::codex_wrapper::init_codex;
use codex_core::config::Config;
use codex_core::parse_command::ParsedCommand;
use codex_core::protocol::AgentMessageDeltaEvent;
@@ -28,6 +26,7 @@ use codex_core::protocol::PatchApplyBeginEvent;
use codex_core::protocol::TaskCompleteEvent;
use codex_core::protocol::TokenUsage;
use codex_core::protocol::TurnDiffEvent;
use codex_core::server::NewConversation;
use crossterm::event::KeyEvent;
use crossterm::event::KeyEventKind;
use ratatui::buffer::Buffer;
@@ -53,6 +52,7 @@ use crate::history_cell::HistoryCell;
use crate::history_cell::PatchEventType;
use crate::live_wrap::RowBuilder;
use crate::user_approval_widget::ApprovalRequest;
use codex_core::server::CodexServer;
use codex_file_search::FileMatch;
use ratatui::style::Stylize;
@@ -161,6 +161,7 @@ impl ChatWidget<'_> {
}
pub(crate) fn new(
config: Config,
server: Arc<CodexServer>,
app_event_tx: AppEventSender,
initial_prompt: Option<String>,
initial_images: Vec<PathBuf>,
@@ -172,12 +173,12 @@ impl ChatWidget<'_> {
// Create the Codex asynchronously so the UI loads as quickly as possible.
let config_for_agent_loop = config.clone();
tokio::spawn(async move {
let CodexConversation {
codex,
let NewConversation {
conversation_id: _,
conversation,
session_configured,
..
} = match init_codex(config_for_agent_loop).await {
Ok(vals) => vals,
} = match server.new_conversation(config_for_agent_loop).await {
Ok(v) => v,
Err(e) => {
// TODO: surface this error to the user.
tracing::error!("failed to initialize codex: {e}");
@@ -185,21 +186,24 @@ impl ChatWidget<'_> {
}
};
// Forward the captured `SessionInitialized` event that was consumed
// inside `init_codex()` so it can be rendered in the UI.
app_event_tx_clone.send(AppEvent::CodexEvent(session_configured.clone()));
let codex = Arc::new(codex);
let codex_clone = codex.clone();
// Forward the captured `SessionConfigured` event so it can be rendered in the UI.
let ev = codex_core::protocol::Event {
// The `id` does not matter for rendering, so we can use a fake value.
id: "".to_string(),
msg: codex_core::protocol::EventMsg::SessionConfigured(session_configured),
};
app_event_tx_clone.send(AppEvent::CodexEvent(ev));
let conversation_clone = conversation.clone();
tokio::spawn(async move {
while let Some(op) = codex_op_rx.recv().await {
let id = codex_clone.submit(op).await;
let id = conversation_clone.submit(op).await;
if let Err(e) = id {
tracing::error!("failed to submit op: {e}");
}
}
});
while let Ok(event) = codex.next_event().await {
while let Ok(event) = conversation.next_event().await {
app_event_tx_clone.send(AppEvent::CodexEvent(event));
}
});