This PR does two things because after I got deep into the first one I started pulling on the thread to the second: - Makes `ConversationManager` the place where all in-memory conversations are created and stored. Previously, `MessageProcessor` in the `codex-mcp-server` crate was doing this via its `session_map`, but this is something that should be done in `codex-core`. - It unwinds the `ctrl_c: tokio::sync::Notify` that was threaded throughout our code. I think this made sense at one time, but now that we handle Ctrl-C within the TUI and have a proper `Op::Interrupt` event, I don't think this was quite right, so I removed it. For `codex exec` and `codex proto`, we now use `tokio::signal::ctrl_c()` directly, but we no longer make `Notify` a field of `Codex` or `CodexConversation`. Changes of note: - Adds the files `conversation_manager.rs` and `codex_conversation.rs` to `codex-core`. - `Codex` and `CodexSpawnOk` are no longer exported from `codex-core`: other crates must use `CodexConversation` instead (which is created via `ConversationManager`). - `core/src/codex_wrapper.rs` has been deleted in favor of `ConversationManager`. - `ConversationManager::new_conversation()` returns `NewConversation`, which is in line with the `new_conversation` tool we want to add to the MCP server. Note `NewConversation` includes `SessionConfiguredEvent`, so we eliminate checks in cases like `codex-rs/core/tests/client.rs` to verify `SessionConfiguredEvent` is the first event because that is now internal to `ConversationManager`. - Quite a bit of code was deleted from `codex-rs/mcp-server/src/message_processor.rs` since it no longer has to manage multiple conversations itself: it goes through `ConversationManager` instead. - `core/tests/live_agent.rs` has been deleted because I had to update a bunch of tests and all the tests in here were ignored, and I don't think anyone ever ran them, so this was just technical debt, at this point. - Removed `notify_on_sigint()` from `util.rs` (and in a follow-up, I hope to refactor the blandly-named `util.rs` into more descriptive files). - In general, I started replacing local variables named `codex` as `conversation`, where appropriate, though admittedly I didn't do it through all the integration tests because that would have added a lot of noise to this PR. --- [//]: # (BEGIN SAPLING FOOTER) Stack created with [Sapling](https://sapling-scm.com). Best reviewed with [ReviewStack](https://reviewstack.dev/openai/codex/pull/2240). * #2264 * #2263 * __->__ #2240
150 lines
4.7 KiB
Rust
150 lines
4.7 KiB
Rust
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
|
|
use codex_core::CodexConversation;
|
|
use codex_core::protocol::Op;
|
|
use codex_core::protocol::ReviewDecision;
|
|
use mcp_types::ElicitRequest;
|
|
use mcp_types::ElicitRequestParamsRequestedSchema;
|
|
use mcp_types::JSONRPCErrorError;
|
|
use mcp_types::ModelContextProtocolRequest;
|
|
use mcp_types::RequestId;
|
|
use serde::Deserialize;
|
|
use serde::Serialize;
|
|
use serde_json::json;
|
|
use tracing::error;
|
|
|
|
use crate::codex_tool_runner::INVALID_PARAMS_ERROR_CODE;
|
|
|
|
/// Conforms to [`mcp_types::ElicitRequestParams`] so that it can be used as the
|
|
/// `params` field of an [`ElicitRequest`].
|
|
#[derive(Debug, Deserialize, Serialize)]
|
|
pub struct ExecApprovalElicitRequestParams {
|
|
// These fields are required so that `params`
|
|
// conforms to ElicitRequestParams.
|
|
pub message: String,
|
|
|
|
#[serde(rename = "requestedSchema")]
|
|
pub requested_schema: ElicitRequestParamsRequestedSchema,
|
|
|
|
// These are additional fields the client can use to
|
|
// correlate the request with the codex tool call.
|
|
pub codex_elicitation: String,
|
|
pub codex_mcp_tool_call_id: String,
|
|
pub codex_event_id: String,
|
|
pub codex_call_id: String,
|
|
pub codex_command: Vec<String>,
|
|
pub codex_cwd: PathBuf,
|
|
}
|
|
|
|
// TODO(mbolin): ExecApprovalResponse does not conform to ElicitResult. See:
|
|
// - https://github.com/modelcontextprotocol/modelcontextprotocol/blob/f962dc1780fa5eed7fb7c8a0232f1fc83ef220cd/schema/2025-06-18/schema.json#L617-L636
|
|
// - https://modelcontextprotocol.io/specification/draft/client/elicitation#protocol-messages
|
|
// It should have "action" and "content" fields.
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct ExecApprovalResponse {
|
|
pub decision: ReviewDecision,
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(crate) async fn handle_exec_approval_request(
|
|
command: Vec<String>,
|
|
cwd: PathBuf,
|
|
outgoing: Arc<crate::outgoing_message::OutgoingMessageSender>,
|
|
codex: Arc<CodexConversation>,
|
|
request_id: RequestId,
|
|
tool_call_id: String,
|
|
event_id: String,
|
|
call_id: String,
|
|
) {
|
|
let escaped_command =
|
|
shlex::try_join(command.iter().map(|s| s.as_str())).unwrap_or_else(|_| command.join(" "));
|
|
let message = format!(
|
|
"Allow Codex to run `{escaped_command}` in `{cwd}`?",
|
|
cwd = cwd.to_string_lossy()
|
|
);
|
|
|
|
let params = ExecApprovalElicitRequestParams {
|
|
message,
|
|
requested_schema: ElicitRequestParamsRequestedSchema {
|
|
r#type: "object".to_string(),
|
|
properties: json!({}),
|
|
required: None,
|
|
},
|
|
codex_elicitation: "exec-approval".to_string(),
|
|
codex_mcp_tool_call_id: tool_call_id.clone(),
|
|
codex_event_id: event_id.clone(),
|
|
codex_call_id: call_id,
|
|
codex_command: command,
|
|
codex_cwd: cwd,
|
|
};
|
|
let params_json = match serde_json::to_value(¶ms) {
|
|
Ok(value) => value,
|
|
Err(err) => {
|
|
let message = format!("Failed to serialize ExecApprovalElicitRequestParams: {err}");
|
|
error!("{message}");
|
|
|
|
outgoing
|
|
.send_error(
|
|
request_id.clone(),
|
|
JSONRPCErrorError {
|
|
code: INVALID_PARAMS_ERROR_CODE,
|
|
message,
|
|
data: None,
|
|
},
|
|
)
|
|
.await;
|
|
|
|
return;
|
|
}
|
|
};
|
|
|
|
let on_response = outgoing
|
|
.send_request(ElicitRequest::METHOD, Some(params_json))
|
|
.await;
|
|
|
|
// Listen for the response on a separate task so we don't block the main agent loop.
|
|
{
|
|
let codex = codex.clone();
|
|
let event_id = event_id.clone();
|
|
tokio::spawn(async move {
|
|
on_exec_approval_response(event_id, on_response, codex).await;
|
|
});
|
|
}
|
|
}
|
|
|
|
async fn on_exec_approval_response(
|
|
event_id: String,
|
|
receiver: tokio::sync::oneshot::Receiver<mcp_types::Result>,
|
|
codex: Arc<CodexConversation>,
|
|
) {
|
|
let response = receiver.await;
|
|
let value = match response {
|
|
Ok(value) => value,
|
|
Err(err) => {
|
|
error!("request failed: {err:?}");
|
|
return;
|
|
}
|
|
};
|
|
|
|
// Try to deserialize `value` and then make the appropriate call to `codex`.
|
|
let response = serde_json::from_value::<ExecApprovalResponse>(value).unwrap_or_else(|err| {
|
|
error!("failed to deserialize ExecApprovalResponse: {err}");
|
|
// If we cannot deserialize the response, we deny the request to be
|
|
// conservative.
|
|
ExecApprovalResponse {
|
|
decision: ReviewDecision::Denied,
|
|
}
|
|
});
|
|
|
|
if let Err(err) = codex
|
|
.submit(Op::ExecApproval {
|
|
id: event_id,
|
|
decision: response.decision,
|
|
})
|
|
.await
|
|
{
|
|
error!("failed to submit ExecApproval: {err}");
|
|
}
|
|
}
|