Merge b792c4d0bb into sapling-pr-archive-bolinfest

This commit is contained in:
Michael Bolin
2025-08-13 15:50:12 -07:00
committed by GitHub
13 changed files with 906 additions and 18 deletions
+36 -3
View File
@@ -43,7 +43,7 @@ switch (platform) {
targetTriple = "x86_64-pc-windows-msvc.exe";
break;
case "arm64":
// We do not build this today, fall through...
// We do not build this today, fall through...
default:
break;
}
@@ -65,9 +65,43 @@ const binaryPath = path.join(__dirname, "..", "bin", `codex-${targetTriple}`);
// receives a fatal signal, both processes exit in a predictable manner.
const { spawn } = await import("child_process");
async function tryImport(moduleName) {
try {
// eslint-disable-next-line node/no-unsupported-features/es-syntax
return await import(moduleName);
} catch (err) {
return null;
}
}
async function resolveRgDir() {
const ripgrep = await tryImport("@vscode/ripgrep");
if (!ripgrep?.rgPath) {
return null;
}
return path.dirname(ripgrep.rgPath);
}
function getUpdatedPath(newDirs) {
const pathSep = process.platform === "win32" ? ";" : ":";
const existingPath = process.env.PATH || "";
const updatedPath = [
...newDirs,
...existingPath.split(pathSep).filter(Boolean),
].join(pathSep);
return updatedPath;
}
const additionalDirs = [];
const rgDir = await resolveRgDir();
if (rgDir) {
additionalDirs.push(rgDir);
}
const updatedPath = getUpdatedPath(additionalDirs);
const child = spawn(binaryPath, process.argv.slice(2), {
stdio: "inherit",
env: { ...process.env, CODEX_MANAGED_BY_NPM: "1" },
env: { ...process.env, PATH: updatedPath, CODEX_MANAGED_BY_NPM: "1" },
});
child.on("error", (err) => {
@@ -120,4 +154,3 @@ if (childResult.type === "signal") {
} else {
process.exit(childResult.exitCode);
}
+119
View File
@@ -0,0 +1,119 @@
{
"name": "@openai/codex",
"version": "0.0.0-dev",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "@openai/codex",
"version": "0.0.0-dev",
"license": "Apache-2.0",
"dependencies": {
"@vscode/ripgrep": "^1.15.14"
},
"bin": {
"codex": "bin/codex.js"
},
"engines": {
"node": ">=20"
}
},
"node_modules/@vscode/ripgrep": {
"version": "1.15.14",
"resolved": "https://registry.npmjs.org/@vscode/ripgrep/-/ripgrep-1.15.14.tgz",
"integrity": "sha512-/G1UJPYlm+trBWQ6cMO3sv6b8D1+G16WaJH1/DSqw32JOVlzgZbLkDxRyzIpTpv30AcYGMkCf5tUqGlW6HbDWw==",
"hasInstallScript": true,
"license": "MIT",
"dependencies": {
"https-proxy-agent": "^7.0.2",
"proxy-from-env": "^1.1.0",
"yauzl": "^2.9.2"
}
},
"node_modules/agent-base": {
"version": "7.1.4",
"resolved": "https://registry.npmjs.org/agent-base/-/agent-base-7.1.4.tgz",
"integrity": "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ==",
"license": "MIT",
"engines": {
"node": ">= 14"
}
},
"node_modules/buffer-crc32": {
"version": "0.2.13",
"resolved": "https://registry.npmjs.org/buffer-crc32/-/buffer-crc32-0.2.13.tgz",
"integrity": "sha512-VO9Ht/+p3SN7SKWqcrgEzjGbRSJYTx+Q1pTQC0wrWqHx0vpJraQ6GtHx8tvcg1rlK1byhU5gccxgOgj7B0TDkQ==",
"license": "MIT",
"engines": {
"node": "*"
}
},
"node_modules/debug": {
"version": "4.4.1",
"resolved": "https://registry.npmjs.org/debug/-/debug-4.4.1.tgz",
"integrity": "sha512-KcKCqiftBJcZr++7ykoDIEwSa3XWowTfNPo92BYxjXiyYEVrUQh2aLyhxBCwww+heortUFxEJYcRzosstTEBYQ==",
"license": "MIT",
"dependencies": {
"ms": "^2.1.3"
},
"engines": {
"node": ">=6.0"
},
"peerDependenciesMeta": {
"supports-color": {
"optional": true
}
}
},
"node_modules/fd-slicer": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/fd-slicer/-/fd-slicer-1.1.0.tgz",
"integrity": "sha512-cE1qsB/VwyQozZ+q1dGxR8LBYNZeofhEdUNGSMbQD3Gw2lAzX9Zb3uIU6Ebc/Fmyjo9AWWfnn0AUCHqtevs/8g==",
"license": "MIT",
"dependencies": {
"pend": "~1.2.0"
}
},
"node_modules/https-proxy-agent": {
"version": "7.0.6",
"resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-7.0.6.tgz",
"integrity": "sha512-vK9P5/iUfdl95AI+JVyUuIcVtd4ofvtrOr3HNtM2yxC9bnMbEdp3x01OhQNnjb8IJYi38VlTE3mBXwcfvywuSw==",
"license": "MIT",
"dependencies": {
"agent-base": "^7.1.2",
"debug": "4"
},
"engines": {
"node": ">= 14"
}
},
"node_modules/ms": {
"version": "2.1.3",
"resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz",
"integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==",
"license": "MIT"
},
"node_modules/pend": {
"version": "1.2.0",
"resolved": "https://registry.npmjs.org/pend/-/pend-1.2.0.tgz",
"integrity": "sha512-F3asv42UuXchdzt+xXqfW1OGlVBe+mxa2mqI0pg5yAHZPvFmY3Y6drSf/GQ1A86WgWEN9Kzh/WrgKa6iGcHXLg==",
"license": "MIT"
},
"node_modules/proxy-from-env": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/proxy-from-env/-/proxy-from-env-1.1.0.tgz",
"integrity": "sha512-D+zkORCbA9f1tdWRK0RaCR3GPv50cMxcrz4X8k5LTSUD1Dkw47mKJEZQNunItRTkWwgtaUSo1RVFRIG9ZXiFYg==",
"license": "MIT"
},
"node_modules/yauzl": {
"version": "2.10.0",
"resolved": "https://registry.npmjs.org/yauzl/-/yauzl-2.10.0.tgz",
"integrity": "sha512-p4a9I6X6nu6IhoGmBqAcbJy1mlC4j27vEPZX9F4L4/vZT3Lyq1VkFHw/V/PUcB9Buo+DG3iHkT0x3Qya58zc3g==",
"license": "MIT",
"dependencies": {
"buffer-crc32": "~0.2.3",
"fd-slicer": "~1.1.0"
}
}
}
}
+6
View File
@@ -16,5 +16,11 @@
"repository": {
"type": "git",
"url": "git+https://github.com/openai/codex.git"
},
"dependencies": {
"@vscode/ripgrep": "^1.15.14"
},
"devDependencies": {
"prettier": "^3.3.3"
}
}
@@ -0,0 +1,268 @@
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use codex_core::ConversationManager;
use codex_core::NewConversation;
use codex_core::config::Config;
use codex_core::config::ConfigOverrides;
use mcp_types::JSONRPCErrorError;
use mcp_types::RequestId;
use tokio::sync::oneshot;
use uuid::Uuid;
use crate::error_code::INTERNAL_ERROR_CODE;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use crate::json_to_toml::json_to_toml;
use crate::outgoing_message::OutgoingMessageSender;
use crate::outgoing_message::OutgoingNotificationMeta;
use crate::wire_format::AddConversationListenerParams;
use crate::wire_format::AddConversationSubscriptionResponse;
use crate::wire_format::CodexRequest;
use crate::wire_format::ConversationId;
use crate::wire_format::InputItem as WireInputItem;
use crate::wire_format::NewConversationParams;
use crate::wire_format::NewConversationResponse;
use crate::wire_format::RemoveConversationListenerParams;
use crate::wire_format::RemoveConversationSubscriptionResponse;
use crate::wire_format::SendUserMessageParams;
use crate::wire_format::SendUserMessageResponse;
use codex_core::protocol::InputItem as CoreInputItem;
use codex_core::protocol::Op;
/// Handles JSON-RPC messages for Codex conversations.
pub(crate) struct CodexMessageProcessor {
conversation_manager: Arc<ConversationManager>,
outgoing: Arc<OutgoingMessageSender>,
codex_linux_sandbox_exe: Option<PathBuf>,
conversation_listeners: HashMap<Uuid, oneshot::Sender<()>>,
}
impl CodexMessageProcessor {
pub fn new(
conversation_manager: Arc<ConversationManager>,
outgoing: Arc<OutgoingMessageSender>,
codex_linux_sandbox_exe: Option<PathBuf>,
) -> Self {
Self {
conversation_manager,
outgoing,
codex_linux_sandbox_exe,
conversation_listeners: HashMap::new(),
}
}
pub async fn process_request(&mut self, request: CodexRequest) {
match request {
CodexRequest::NewConversation { request_id, params } => {
// Do not tokio::spawn() to process new_conversation()
// asynchronously because we need to ensure the conversation is
// created before processing any subsequent messages.
self.process_new_conversation(request_id, params).await;
}
CodexRequest::SendUserMessage { request_id, params } => {
self.send_user_message(request_id, params).await;
}
CodexRequest::AddConversationListener { request_id, params } => {
self.add_conversation_listener(request_id, params).await;
}
CodexRequest::RemoveConversationListener { request_id, params } => {
self.remove_conversation_listener(request_id, params).await;
}
}
}
async fn process_new_conversation(&self, request_id: RequestId, params: NewConversationParams) {
let config = match derive_config_from_params(params, self.codex_linux_sandbox_exe.clone()) {
Ok(config) => config,
Err(err) => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("error deriving config: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
};
match self.conversation_manager.new_conversation(config).await {
Ok(conversation_id) => {
let NewConversation {
conversation_id,
session_configured,
..
} = conversation_id;
let response = NewConversationResponse {
conversation_id: ConversationId(conversation_id),
model: session_configured.model,
};
self.outgoing.send_response(request_id, response).await;
}
Err(err) => {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("error creating conversation: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn send_user_message(&self, request_id: RequestId, params: SendUserMessageParams) {
let SendUserMessageParams {
conversation_id,
items,
} = params;
let Ok(conversation) = self
.conversation_manager
.get_conversation(conversation_id.0)
.await
else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("conversation not found: {conversation_id}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
};
let mapped_items: Vec<CoreInputItem> = items
.into_iter()
.map(|item| match item {
WireInputItem::Text { text } => CoreInputItem::Text { text },
WireInputItem::Image { image_url } => CoreInputItem::Image { image_url },
WireInputItem::LocalImage { path } => CoreInputItem::LocalImage { path },
})
.collect();
// Submit user input to the conversation.
let _ = conversation
.submit(Op::UserInput {
items: mapped_items,
})
.await;
// Acknowledge with an empty result.
self.outgoing
.send_response(request_id, SendUserMessageResponse {})
.await;
}
async fn add_conversation_listener(
&mut self,
request_id: RequestId,
params: AddConversationListenerParams,
) {
let AddConversationListenerParams { conversation_id } = params;
let Ok(conversation) = self
.conversation_manager
.get_conversation(conversation_id.0)
.await
else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("conversation not found: {}", conversation_id.0),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
};
let subscription_id = Uuid::new_v4();
let (sender, mut receiver) = oneshot::channel();
self.conversation_listeners.insert(subscription_id, sender);
let outgoing_for_task = self.outgoing.clone();
let add_listener_request_id = request_id.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = &mut receiver => {
// User has unsubscribed, so exit this task.
break;
}
event = conversation.next_event() => {
let event = match event {
Ok(event) => event,
Err(err) => {
tracing::warn!("conversation.next_event() failed with: {err}");
break;
}
};
outgoing_for_task.send_event_as_notification(
&event,
Some(OutgoingNotificationMeta::new(Some(add_listener_request_id.clone()))),
)
.await;
}
}
}
});
let response = AddConversationSubscriptionResponse { subscription_id };
self.outgoing.send_response(request_id, response).await;
}
async fn remove_conversation_listener(
&mut self,
request_id: RequestId,
params: RemoveConversationListenerParams,
) {
let RemoveConversationListenerParams { subscription_id } = params;
match self.conversation_listeners.remove(&subscription_id) {
Some(sender) => {
// Signal the spawned task to exit and acknowledge.
let _ = sender.send(());
let response = RemoveConversationSubscriptionResponse {};
self.outgoing.send_response(request_id, response).await;
}
None => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("subscription not found: {subscription_id}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
}
}
}
fn derive_config_from_params(
params: NewConversationParams,
codex_linux_sandbox_exe: Option<PathBuf>,
) -> std::io::Result<Config> {
let NewConversationParams {
model,
profile,
cwd,
approval_policy,
sandbox,
config: cli_overrides,
base_instructions,
include_plan_tool,
} = params;
let overrides = ConfigOverrides {
model,
config_profile: profile,
cwd: cwd.map(PathBuf::from),
approval_policy: approval_policy.map(Into::into),
sandbox_mode: sandbox.map(Into::into),
model_provider: None,
codex_linux_sandbox_exe,
base_instructions,
include_plan_tool,
disable_response_storage: None,
show_raw_agent_reasoning: None,
};
let cli_overrides = cli_overrides
.unwrap_or_default()
.into_iter()
.map(|(k, v)| (k, json_to_toml(v)))
.collect();
Config::load_with_cli_overrides(cli_overrides, overrides)
}
+2 -2
View File
@@ -58,7 +58,7 @@ pub struct CodexToolCallParam {
/// Custom enum mirroring [`AskForApproval`], but has an extra dependency on
/// [`JsonSchema`].
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub enum CodexToolCallApprovalPolicy {
Untrusted,
@@ -80,7 +80,7 @@ impl From<CodexToolCallApprovalPolicy> for AskForApproval {
/// Custom enum mirroring [`SandboxMode`] from config_types.rs, but with
/// `JsonSchema` support.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
#[serde(rename_all = "kebab-case")]
pub enum CodexToolCallSandboxMode {
ReadOnly,
+2
View File
@@ -15,6 +15,7 @@ use tracing::error;
use tracing::info;
use tracing_subscriber::EnvFilter;
mod codex_message_processor;
mod codex_tool_config;
mod codex_tool_runner;
mod conversation_loop;
@@ -26,6 +27,7 @@ pub(crate) mod message_processor;
mod outgoing_message;
mod patch_approval;
pub(crate) mod tool_handlers;
pub mod wire_format;
use crate::message_processor::MessageProcessor;
use crate::outgoing_message::OutgoingMessage;
+24 -2
View File
@@ -3,6 +3,7 @@ use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use crate::codex_message_processor::CodexMessageProcessor;
use crate::codex_tool_config::CodexToolCallParam;
use crate::codex_tool_config::CodexToolCallReplyParam;
use crate::codex_tool_config::create_tool_for_codex_tool_call_param;
@@ -14,6 +15,7 @@ use crate::mcp_protocol::ToolCallResponseResult;
use crate::outgoing_message::OutgoingMessageSender;
use crate::tool_handlers::create_conversation::handle_create_conversation;
use crate::tool_handlers::send_message::handle_send_message;
use crate::wire_format::CodexRequest;
use codex_core::ConversationManager;
use codex_core::config::Config as CodexConfig;
@@ -40,6 +42,7 @@ use tokio::task;
use uuid::Uuid;
pub(crate) struct MessageProcessor {
codex_message_processor: CodexMessageProcessor,
outgoing: Arc<OutgoingMessageSender>,
initialized: bool,
codex_linux_sandbox_exe: Option<PathBuf>,
@@ -55,11 +58,19 @@ impl MessageProcessor {
outgoing: OutgoingMessageSender,
codex_linux_sandbox_exe: Option<PathBuf>,
) -> Self {
let outgoing = Arc::new(outgoing);
let conversation_manager = Arc::new(ConversationManager::default());
let codex_message_processor = CodexMessageProcessor::new(
conversation_manager.clone(),
outgoing.clone(),
codex_linux_sandbox_exe.clone(),
);
Self {
outgoing: Arc::new(outgoing),
codex_message_processor,
outgoing,
initialized: false,
codex_linux_sandbox_exe,
conversation_manager: Arc::new(ConversationManager::default()),
conversation_manager,
running_requests_id_to_codex_uuid: Arc::new(Mutex::new(HashMap::new())),
running_session_ids: Arc::new(Mutex::new(HashSet::new())),
}
@@ -78,6 +89,17 @@ impl MessageProcessor {
}
pub(crate) async fn process_request(&mut self, request: JSONRPCRequest) {
if let Ok(request_json) = serde_json::to_value(request.clone())
&& let Ok(codex_request) = serde_json::from_value::<CodexRequest>(request_json)
{
// If the request is a Codex request, handle it with the Codex
// message processor.
self.codex_message_processor
.process_request(codex_request)
.await;
return;
}
// Hold on to the ID so we can respond.
let request_id = request.id.clone();
+180
View File
@@ -0,0 +1,180 @@
use std::collections::HashMap;
use std::fmt::Display;
use mcp_types::RequestId;
use serde::Deserialize;
use serde::Serialize;
use crate::codex_tool_config::CodexToolCallApprovalPolicy;
use crate::codex_tool_config::CodexToolCallSandboxMode;
use uuid::Uuid;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct ConversationId(pub Uuid);
impl Display for ConversationId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
/// Request from the client to the server.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(tag = "method", rename_all = "camelCase")]
pub enum CodexRequest {
NewConversation {
#[serde(rename = "id")]
request_id: RequestId,
params: NewConversationParams,
},
SendUserMessage {
#[serde(rename = "id")]
request_id: RequestId,
params: SendUserMessageParams,
},
AddConversationListener {
#[serde(rename = "id")]
request_id: RequestId,
params: AddConversationListenerParams,
},
RemoveConversationListener {
#[serde(rename = "id")]
request_id: RequestId,
params: RemoveConversationListenerParams,
},
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
#[serde(rename_all = "camelCase")]
pub struct NewConversationParams {
/// Optional override for the model name (e.g. "o3", "o4-mini").
#[serde(skip_serializing_if = "Option::is_none")]
pub model: Option<String>,
/// Configuration profile from config.toml to specify default options.
#[serde(skip_serializing_if = "Option::is_none")]
pub profile: Option<String>,
/// Working directory for the session. If relative, it is resolved against
/// the server process's current working directory.
#[serde(skip_serializing_if = "Option::is_none")]
pub cwd: Option<String>,
/// Approval policy for shell commands generated by the model:
/// `untrusted`, `on-failure`, `on-request`, `never`.
#[serde(skip_serializing_if = "Option::is_none")]
pub approval_policy: Option<CodexToolCallApprovalPolicy>,
/// Sandbox mode: `read-only`, `workspace-write`, or `danger-full-access`.
#[serde(skip_serializing_if = "Option::is_none")]
pub sandbox: Option<CodexToolCallSandboxMode>,
/// Individual config settings that will override what is in
/// CODEX_HOME/config.toml.
#[serde(skip_serializing_if = "Option::is_none")]
pub config: Option<HashMap<String, serde_json::Value>>,
/// The set of instructions to use instead of the default ones.
#[serde(skip_serializing_if = "Option::is_none")]
pub base_instructions: Option<String>,
/// Whether to include the plan tool in the conversation.
#[serde(skip_serializing_if = "Option::is_none")]
pub include_plan_tool: Option<bool>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct NewConversationResponse {
pub conversation_id: ConversationId,
pub model: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct AddConversationSubscriptionResponse {
pub subscription_id: Uuid,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct RemoveConversationSubscriptionResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct SendUserMessageParams {
pub conversation_id: ConversationId,
pub items: Vec<InputItem>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct SendUserMessageResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct AddConversationListenerParams {
pub conversation_id: ConversationId,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct RemoveConversationListenerParams {
pub subscription_id: Uuid,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
#[serde(rename_all = "camelCase")]
pub enum InputItem {
Text {
text: String,
},
/// Pre‑encoded data: URI image.
Image {
image_url: String,
},
/// Local image path provided by the user. This will be converted to an
/// `Image` variant (base64 data URL) during request serialization.
LocalImage {
path: std::path::PathBuf,
},
}
#[allow(clippy::unwrap_used)]
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
use serde_json::json;
#[test]
fn serialize_new_conversation() {
let request = CodexRequest::NewConversation {
request_id: RequestId::Integer(42),
params: NewConversationParams {
model: Some("gpt-5".to_string()),
profile: None,
cwd: None,
approval_policy: Some(CodexToolCallApprovalPolicy::OnRequest),
sandbox: None,
config: None,
base_instructions: None,
include_plan_tool: None,
},
};
assert_eq!(
json!({
"method": "newConversation",
"id": 42,
"params": {
"model": "gpt-5",
"approvalPolicy": "on-request"
}
}),
serde_json::to_value(&request).unwrap(),
);
}
}
@@ -0,0 +1,171 @@
#![allow(clippy::expect_used, clippy::unwrap_used)]
use std::path::Path;
use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use codex_mcp_server::wire_format::AddConversationListenerParams;
use codex_mcp_server::wire_format::AddConversationSubscriptionResponse;
use codex_mcp_server::wire_format::NewConversationParams;
use codex_mcp_server::wire_format::NewConversationResponse;
use codex_mcp_server::wire_format::RemoveConversationListenerParams;
use codex_mcp_server::wire_format::RemoveConversationSubscriptionResponse;
use codex_mcp_server::wire_format::SendUserMessageParams;
use codex_mcp_server::wire_format::SendUserMessageResponse;
use mcp_test_support::McpProcess;
use mcp_test_support::create_mock_chat_completions_server;
use mcp_test_support::create_shell_sse_response;
use mcp_types::JSONRPCResponse;
use mcp_types::RequestId;
use pretty_assertions::assert_eq;
use serde::de::DeserializeOwned;
use std::env;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_codex_jsonrpc_conversation_flow() {
if env::var(CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR).is_ok() {
println!(
"Skipping test because it cannot execute when network is disabled in a Codex sandbox."
);
return;
}
let tmp = TempDir::new().expect("tmp dir");
// Temporary Codex home with config pointing at the mock server.
let codex_home = tmp.path().join("codex_home");
std::fs::create_dir(&codex_home).expect("create codex home dir");
let working_directory = tmp.path().join("workdir");
std::fs::create_dir(&working_directory).expect("create working directory");
// Create a mock model server that immediately ends each turn.
// Two turns are expected: initial session configure + one user message.
let responses = vec![
create_shell_sse_response(
vec!["ls".to_string()],
Some(&working_directory),
Some(5000),
"call1234",
)
.expect("create shell sse response"),
];
let server = create_mock_chat_completions_server(responses).await;
create_config_toml(&codex_home, &server.uri()).expect("write config");
// Start MCP server and initialize.
let mut mcp = McpProcess::new(&codex_home).await.expect("spawn mcp");
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize())
.await
.expect("init timeout")
.expect("init error");
// 1) newConversation
let new_conv_id = mcp
.send_new_conversation_request(NewConversationParams {
cwd: Some(working_directory.to_string_lossy().into_owned()),
..Default::default()
})
.await
.expect("send newConversation");
let new_conv_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(new_conv_id)),
)
.await
.expect("newConversation timeout")
.expect("newConversation resp");
let new_conv_resp = to_response::<NewConversationResponse>(new_conv_resp)
.expect("deserialize newConversation response");
let NewConversationResponse {
conversation_id,
model,
} = new_conv_resp;
assert_eq!(model, "mock-model");
// 2) addConversationListener
let add_listener_id = mcp
.send_add_conversation_listener_request(AddConversationListenerParams { conversation_id })
.await
.expect("send addConversationListener");
let add_listener_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(add_listener_id)),
)
.await
.expect("addConversationListener timeout")
.expect("addConversationListener resp");
let AddConversationSubscriptionResponse { subscription_id } =
to_response::<AddConversationSubscriptionResponse>(add_listener_resp)
.expect("deserialize addConversationListener response");
// 3) sendUserMessage (should trigger notifications; we only validate an OK response)
let send_user_id = mcp
.send_send_user_message_request(SendUserMessageParams {
conversation_id,
items: vec![codex_mcp_server::wire_format::InputItem::Text {
text: "text".to_string(),
}],
})
.await
.expect("send sendUserMessage");
let send_user_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(send_user_id)),
)
.await
.expect("sendUserMessage timeout")
.expect("sendUserMessage resp");
let SendUserMessageResponse {} = to_response::<SendUserMessageResponse>(send_user_resp)
.expect("deserialize sendUserMessage response");
// 4) removeConversationListener
let remove_listener_id = mcp
.send_remove_conversation_listener_request(RemoveConversationListenerParams {
subscription_id,
})
.await
.expect("send removeConversationListener");
let remove_listener_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(remove_listener_id)),
)
.await
.expect("removeConversationListener timeout")
.expect("removeConversationListener resp");
let RemoveConversationSubscriptionResponse {} =
to_response(remove_listener_resp).expect("deserialize removeConversationListener response");
// TODO(mbolin): This should not be necessary.
std::mem::forget(server);
}
fn to_response<T: DeserializeOwned>(response: JSONRPCResponse) -> anyhow::Result<T> {
let value = serde_json::to_value(response.result)?;
let codex_response = serde_json::from_value(value)?;
Ok(codex_response)
}
// Helper: minimal config.toml pointing at mock provider.
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "never"
model_provider = "mock_provider"
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "chat"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}
@@ -18,6 +18,10 @@ use codex_mcp_server::mcp_protocol::ConversationCreateArgs;
use codex_mcp_server::mcp_protocol::ConversationId;
use codex_mcp_server::mcp_protocol::ConversationSendMessageArgs;
use codex_mcp_server::mcp_protocol::ToolCallRequestParams;
use codex_mcp_server::wire_format::AddConversationListenerParams;
use codex_mcp_server::wire_format::NewConversationParams;
use codex_mcp_server::wire_format::RemoveConversationListenerParams;
use codex_mcp_server::wire_format::SendUserMessageParams;
use mcp_types::CallToolRequestParams;
use mcp_types::ClientCapabilities;
@@ -236,6 +240,47 @@ impl McpProcess {
.await
}
// ---------------------------------------------------------------------
// Codex JSON-RPC (non-tool) helpers
// ---------------------------------------------------------------------
/// Send a `newConversation` JSON-RPC request.
pub async fn send_new_conversation_request(
&mut self,
params: NewConversationParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("newConversation", params).await
}
/// Send an `addConversationListener` JSON-RPC request.
pub async fn send_add_conversation_listener_request(
&mut self,
params: AddConversationListenerParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("addConversationListener", params).await
}
/// Send a `sendUserMessage` JSON-RPC request with a single text item.
pub async fn send_send_user_message_request(
&mut self,
params: SendUserMessageParams,
) -> anyhow::Result<i64> {
// Wire format expects variants in camelCase; text item uses external tagging.
let params = Some(serde_json::to_value(params)?);
self.send_request("sendUserMessage", params).await
}
/// Send a `removeConversationListener` JSON-RPC request.
pub async fn send_remove_conversation_listener_request(
&mut self,
params: RemoveConversationListenerParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("removeConversationListener", params)
.await
}
async fn send_request(
&mut self,
method: &str,
+18 -9
View File
@@ -270,13 +270,14 @@ fn push_wrapped_diff_line(
// Reserve a fixed number of spaces after the line number so that content starts
// at a consistent column. The sign ("+"/"-") is rendered as part of the content
// with the same background as the edit, not as a separate dimmed column.
// and colored with the same foreground as the edited text, not as a separate
// dimmed column.
let gap_after_ln = SPACES_AFTER_LINE_NUMBER.saturating_sub(ln_str.len());
let first_prefix_cols = indent.len() + ln_str.len() + gap_after_ln;
let cont_prefix_cols = indent.len() + ln_str.len() + gap_after_ln;
let mut first = true;
let (sign_opt, bg_style) = match kind {
let (sign_opt, line_style) = match kind {
DiffLineType::Insert => (Some('+'), Some(style_add())),
DiffLineType::Delete => (Some('-'), Some(style_del())),
DiffLineType::Context => (None, None),
@@ -307,18 +308,22 @@ fn push_wrapped_diff_line(
spans.push(RtSpan::raw(" ".repeat(gap_after_ln)));
// Prefix the content with the sign if it is an insertion or deletion, and color
// the sign with the same background as the edited text.
// the sign and content with the same foreground as the edited text.
let display_chunk = match sign_opt {
Some(sign_char) => format!("{sign_char}{chunk}"),
None => chunk.to_string(),
};
let content_span = match bg_style {
let content_span = match line_style {
Some(style) => RtSpan::styled(display_chunk, style),
None => RtSpan::raw(display_chunk),
};
spans.push(content_span);
lines.push(RtLine::from(spans));
let mut line = RtLine::from(spans);
if let Some(style) = line_style {
line.style = line.style.patch(style);
}
lines.push(line);
first = false;
} else {
let hang_prefix = format!(
@@ -326,11 +331,15 @@ fn push_wrapped_diff_line(
" ".repeat(ln_str.len()),
" ".repeat(gap_after_ln)
);
let content_span = match bg_style {
let content_span = match line_style {
Some(style) => RtSpan::styled(chunk.to_string(), style),
None => RtSpan::raw(chunk.to_string()),
};
lines.push(RtLine::from(vec![RtSpan::raw(hang_prefix), content_span]));
let mut line = RtLine::from(vec![RtSpan::raw(hang_prefix), content_span]);
if let Some(style) = line_style {
line.style = line.style.patch(style);
}
lines.push(line);
}
}
lines
@@ -341,11 +350,11 @@ fn style_dim() -> Style {
}
fn style_add() -> Style {
Style::default().bg(Color::Green)
Style::default().fg(Color::Green)
}
fn style_del() -> Style {
Style::default().bg(Color::Red)
Style::default().fg(Color::Red)
}
#[allow(clippy::expect_used)]
+33
View File
@@ -0,0 +1,33 @@
{
"name": "codex-monorepo",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "codex-monorepo",
"devDependencies": {
"prettier": "^3.5.3"
},
"engines": {
"node": ">=22",
"pnpm": ">=9.0.0"
}
},
"node_modules/prettier": {
"version": "3.6.2",
"resolved": "https://registry.npmjs.org/prettier/-/prettier-3.6.2.tgz",
"integrity": "sha512-I7AIg5boAr5R0FFtJ6rCfD+LFsWHp81dolrFD8S79U9tb8Az2nGrJncnMSnys+bpQJfRUzqs9hnA81OAA3hCuQ==",
"dev": true,
"license": "MIT",
"bin": {
"prettier": "bin/prettier.cjs"
},
"engines": {
"node": ">=14"
},
"funding": {
"url": "https://github.com/prettier/prettier?sponsor=1"
}
}
}
}
+2 -2
View File
@@ -3,8 +3,8 @@
"private": true,
"description": "Tools for repo-wide maintenance.",
"scripts": {
"format": "prettier --check *.json *.md .github/workflows/*.yml",
"format:fix": "prettier --write *.json *.md .github/workflows/*.yml"
"format": "prettier --check *.json *.md .github/workflows/*.yml **/*.js",
"format:fix": "prettier --write *.json *.md .github/workflows/*.yml **/*.js"
},
"devDependencies": {
"prettier": "^3.5.3"