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
151 lines
4.6 KiB
Rust
151 lines
4.6 KiB
Rust
use std::collections::HashMap;
|
|
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
|
|
use codex_core::CodexConversation;
|
|
use codex_core::protocol::FileChange;
|
|
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;
|
|
use crate::outgoing_message::OutgoingMessageSender;
|
|
|
|
#[derive(Debug, Serialize)]
|
|
pub struct PatchApprovalElicitRequestParams {
|
|
pub message: String,
|
|
#[serde(rename = "requestedSchema")]
|
|
pub requested_schema: ElicitRequestParamsRequestedSchema,
|
|
pub codex_elicitation: String,
|
|
pub codex_mcp_tool_call_id: String,
|
|
pub codex_event_id: String,
|
|
pub codex_call_id: String,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub codex_reason: Option<String>,
|
|
#[serde(skip_serializing_if = "Option::is_none")]
|
|
pub codex_grant_root: Option<PathBuf>,
|
|
pub codex_changes: HashMap<PathBuf, FileChange>,
|
|
}
|
|
|
|
#[derive(Debug, Deserialize, Serialize)]
|
|
pub struct PatchApprovalResponse {
|
|
pub decision: ReviewDecision,
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub(crate) async fn handle_patch_approval_request(
|
|
call_id: String,
|
|
reason: Option<String>,
|
|
grant_root: Option<PathBuf>,
|
|
changes: HashMap<PathBuf, FileChange>,
|
|
outgoing: Arc<OutgoingMessageSender>,
|
|
codex: Arc<CodexConversation>,
|
|
request_id: RequestId,
|
|
tool_call_id: String,
|
|
event_id: String,
|
|
) {
|
|
let mut message_lines = Vec::new();
|
|
if let Some(r) = &reason {
|
|
message_lines.push(r.clone());
|
|
}
|
|
message_lines.push("Allow Codex to apply proposed code changes?".to_string());
|
|
|
|
let params = PatchApprovalElicitRequestParams {
|
|
message: message_lines.join("\n"),
|
|
requested_schema: ElicitRequestParamsRequestedSchema {
|
|
r#type: "object".to_string(),
|
|
properties: json!({}),
|
|
required: None,
|
|
},
|
|
codex_elicitation: "patch-approval".to_string(),
|
|
codex_mcp_tool_call_id: tool_call_id.clone(),
|
|
codex_event_id: event_id.clone(),
|
|
codex_call_id: call_id,
|
|
codex_reason: reason,
|
|
codex_grant_root: grant_root,
|
|
codex_changes: changes,
|
|
};
|
|
let params_json = match serde_json::to_value(¶ms) {
|
|
Ok(value) => value,
|
|
Err(err) => {
|
|
let message = format!("Failed to serialize PatchApprovalElicitRequestParams: {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_patch_approval_response(event_id, on_response, codex).await;
|
|
});
|
|
}
|
|
}
|
|
|
|
pub(crate) async fn on_patch_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:?}");
|
|
if let Err(submit_err) = codex
|
|
.submit(Op::PatchApproval {
|
|
id: event_id.clone(),
|
|
decision: ReviewDecision::Denied,
|
|
})
|
|
.await
|
|
{
|
|
error!("failed to submit denied PatchApproval after request failure: {submit_err}");
|
|
}
|
|
return;
|
|
}
|
|
};
|
|
|
|
let response = serde_json::from_value::<PatchApprovalResponse>(value).unwrap_or_else(|err| {
|
|
error!("failed to deserialize PatchApprovalResponse: {err}");
|
|
PatchApprovalResponse {
|
|
decision: ReviewDecision::Denied,
|
|
}
|
|
});
|
|
|
|
if let Err(err) = codex
|
|
.submit(Op::PatchApproval {
|
|
id: event_id,
|
|
decision: response.decision,
|
|
})
|
|
.await
|
|
{
|
|
error!("failed to submit PatchApproval: {err}");
|
|
}
|
|
}
|