diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index b3f8d88027..e5c2218fa0 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1672,6 +1672,7 @@ dependencies = [ "codex-core", "codex-exec", "codex-execpolicy", + "codex-executor", "codex-features", "codex-login", "codex-mcp-server", @@ -2068,6 +2069,27 @@ dependencies = [ "tempfile", ] +[[package]] +name = "codex-executor" +version = "0.0.0" +dependencies = [ + "anyhow", + "clap", + "codex-executor-protocol", + "pretty_assertions", + "serde_json", + "tokio", +] + +[[package]] +name = "codex-executor-protocol" +version = "0.0.0" +dependencies = [ + "pretty_assertions", + "serde", + "serde_json", +] + [[package]] name = "codex-experimental-api-macros" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 6d768d6963..a46b7c380e 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -26,6 +26,8 @@ members = [ "hooks", "secrets", "exec", + "executor", + "executor-protocol", "exec-server", "execpolicy", "execpolicy-legacy", @@ -108,6 +110,8 @@ codex-connectors = { path = "connectors" } codex-config = { path = "config" } codex-core = { path = "core" } codex-exec = { path = "exec" } +codex-executor = { path = "executor" } +codex-executor-protocol = { path = "executor-protocol" } codex-exec-server = { path = "exec-server" } codex-execpolicy = { path = "execpolicy" } codex-experimental-api-macros = { path = "codex-experimental-api-macros" } diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index c2fd1300c6..0c272b3b12 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -29,6 +29,7 @@ codex-utils-cli = { workspace = true } codex-config = { workspace = true } codex-core = { workspace = true } codex-exec = { workspace = true } +codex-executor = { workspace = true } codex-execpolicy = { workspace = true } codex-features = { workspace = true } codex-login = { workspace = true } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 8446e457b2..b55f00b521 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -21,6 +21,7 @@ use codex_exec::Cli as ExecCli; use codex_exec::Command as ExecCommand; use codex_exec::ReviewArgs; use codex_execpolicy::ExecPolicyCheckCommand; +use codex_executor::Cli as ExecutorCli; use codex_responses_api_proxy::Args as ResponsesApiProxyArgs; use codex_state::StateRuntime; use codex_state::state_db_path; @@ -109,6 +110,9 @@ enum Subcommand { /// [experimental] Run the app server or related tooling. AppServer(AppServerCommand), + /// [experimental] Run Codex as an executor. + Executor(ExecutorCli), + /// Launch the Codex desktop app (downloads the macOS installer if missing). #[cfg(target_os = "macos")] App(app_cmd::AppCommand), @@ -682,6 +686,10 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> { codex_app_server_protocol::generate_internal_json_schema(&gen_cli.out_dir)?; } }, + Some(Subcommand::Executor(executor_cli)) => { + reject_remote_mode_for_subcommand(root_remote.as_deref(), "executor")?; + codex_executor::run_main(executor_cli).await?; + } #[cfg(target_os = "macos")] Some(Subcommand::App(app_cli)) => { reject_remote_mode_for_subcommand(root_remote.as_deref(), "app")?; @@ -1385,6 +1393,14 @@ mod tests { app_server } + fn executor_from_args(args: &[&str]) -> ExecutorCli { + let cli = MultitoolCli::try_parse_from(args).expect("parse"); + let Subcommand::Executor(executor) = cli.subcommand.expect("executor present") else { + unreachable!() + }; + executor + } + fn sample_exit_info(conversation_id: Option<&str>, thread_name: Option<&str>) -> AppExitInfo { let token_usage = TokenUsage { output_tokens: 2, @@ -1622,6 +1638,11 @@ mod tests { assert!(app_server.analytics_default_enabled); } + #[test] + fn executor_subcommand_parses_without_extra_args() { + let _executor = executor_from_args(["codex", "executor"].as_ref()); + } + #[test] fn remote_flag_parses_for_interactive_root() { let cli = MultitoolCli::try_parse_from(["codex", "--remote", "ws://127.0.0.1:4500"]) diff --git a/codex-rs/executor-protocol/Cargo.toml b/codex-rs/executor-protocol/Cargo.toml new file mode 100644 index 0000000000..47a6e078da --- /dev/null +++ b/codex-rs/executor-protocol/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "codex-executor-protocol" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lib] +name = "codex_executor_protocol" +path = "src/lib.rs" + +[lints] +workspace = true + +[dependencies] +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } + +[dev-dependencies] +pretty_assertions = { workspace = true } diff --git a/codex-rs/executor-protocol/src/lib.rs b/codex-rs/executor-protocol/src/lib.rs new file mode 100644 index 0000000000..d2d6785142 --- /dev/null +++ b/codex-rs/executor-protocol/src/lib.rs @@ -0,0 +1,189 @@ +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value as JsonValue; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ExecutorToolSpec { + pub name: String, + pub description: String, + pub input_schema: JsonValue, +} + +impl ExecutorToolSpec { + pub fn new( + name: impl Into, + description: impl Into, + input_schema: JsonValue, + ) -> Self { + Self { + name: name.into(), + description: description.into(), + input_schema, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ListToolsRequest { + pub request_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct CallToolRequest { + pub request_id: String, + pub tool_name: String, + pub arguments: JsonValue, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ShutdownRequest { + pub request_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum OrchestratorToExecutorMessage { + ListTools(ListToolsRequest), + CallTool(CallToolRequest), + Shutdown(ShutdownRequest), +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ListToolsResponse { + pub request_id: String, + pub tools: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ToolCallContent { + InputText { text: String }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "status", rename_all = "snake_case")] +pub enum ToolCallOutcome { + Success { content: Vec }, + Error { message: String }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct CallToolResponse { + pub request_id: String, + pub tool_name: String, + pub outcome: ToolCallOutcome, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ShutdownResponse { + pub request_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct ErrorResponse { + pub request_id: Option, + pub message: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ExecutorToOrchestratorMessage { + ListToolsResponse(ListToolsResponse), + CallToolResponse(CallToolResponse), + ShutdownResponse(ShutdownResponse), + Error(ErrorResponse), +} + +impl ExecutorToOrchestratorMessage { + pub fn error(request_id: Option, message: impl Into) -> Self { + Self::Error(ErrorResponse { + request_id, + message: message.into(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::CallToolRequest; + use super::ExecutorToOrchestratorMessage; + use super::ExecutorToolSpec; + use super::ListToolsResponse; + use super::OrchestratorToExecutorMessage; + use pretty_assertions::assert_eq; + use serde_json::json; + + #[test] + fn list_tools_response_round_trips_with_camel_case_fields() { + let message = ExecutorToOrchestratorMessage::ListToolsResponse(ListToolsResponse { + request_id: "req-1".to_string(), + tools: vec![ExecutorToolSpec::new( + "exec_command", + "Run a command", + json!({ + "type": "object", + "properties": { + "cmd": {"type": "string"}, + }, + "required": ["cmd"], + }), + )], + }); + + let value = serde_json::to_value(&message).expect("serialize"); + assert_eq!( + value, + json!({ + "type": "list_tools_response", + "requestId": "req-1", + "tools": [{ + "name": "exec_command", + "description": "Run a command", + "inputSchema": { + "type": "object", + "properties": { + "cmd": {"type": "string"}, + }, + "required": ["cmd"], + }, + }], + }) + ); + } + + #[test] + fn call_tool_request_round_trips() { + let value = json!({ + "type": "call_tool", + "requestId": "req-2", + "toolName": "exec_command", + "arguments": { + "cmd": "pwd", + }, + }); + + let actual: OrchestratorToExecutorMessage = + serde_json::from_value(value.clone()).expect("deserialize"); + + assert_eq!( + actual, + OrchestratorToExecutorMessage::CallTool(CallToolRequest { + request_id: "req-2".to_string(), + tool_name: "exec_command".to_string(), + arguments: json!({ + "cmd": "pwd", + }), + }) + ); + + assert_eq!(serde_json::to_value(actual).expect("serialize"), value); + } +} diff --git a/codex-rs/executor/Cargo.toml b/codex-rs/executor/Cargo.toml new file mode 100644 index 0000000000..944836e2f5 --- /dev/null +++ b/codex-rs/executor/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "codex-executor" +version.workspace = true +edition.workspace = true +license.workspace = true + +[[bin]] +name = "codex-executor" +path = "src/main.rs" + +[lib] +name = "codex_executor" +path = "src/lib.rs" + +[lints] +workspace = true + +[dependencies] +anyhow = { workspace = true } +clap = { workspace = true, features = ["derive"] } +codex-executor-protocol = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync"] } + +[dev-dependencies] +pretty_assertions = { workspace = true } +serde_json = { workspace = true } diff --git a/codex-rs/executor/src/cli.rs b/codex-rs/executor/src/cli.rs new file mode 100644 index 0000000000..ece976a326 --- /dev/null +++ b/codex-rs/executor/src/cli.rs @@ -0,0 +1,5 @@ +use clap::Parser; + +#[derive(Debug, Parser, Default)] +#[command(version)] +pub struct Cli {} diff --git a/codex-rs/executor/src/lib.rs b/codex-rs/executor/src/lib.rs new file mode 100644 index 0000000000..ecceeb6910 --- /dev/null +++ b/codex-rs/executor/src/lib.rs @@ -0,0 +1,249 @@ +mod cli; + +pub use cli::Cli; + +use anyhow::Context; +use codex_executor_protocol::CallToolResponse; +use codex_executor_protocol::ExecutorToOrchestratorMessage; +use codex_executor_protocol::ExecutorToolSpec; +use codex_executor_protocol::ListToolsResponse; +use codex_executor_protocol::OrchestratorToExecutorMessage; +use codex_executor_protocol::ShutdownResponse; +use codex_executor_protocol::ToolCallOutcome; +use tokio::sync::mpsc; + +pub async fn run_main(cli: Cli) -> anyhow::Result<()> { + let (outbound, inbound) = establish_connection(&cli).await?; + Executor::new(outbound, inbound).run().await +} + +async fn establish_connection( + _cli: &Cli, +) -> anyhow::Result<( + mpsc::Sender, + mpsc::Receiver, +)> { + unimplemented!("executor transport establishment is not implemented yet"); +} + +pub struct Executor { + outbound: mpsc::Sender, + inbound: mpsc::Receiver, + tools: Vec, +} + +impl Executor { + pub fn new( + outbound: mpsc::Sender, + inbound: mpsc::Receiver, + ) -> Self { + Self::with_tools(outbound, inbound, Vec::new()) + } + + pub fn with_tools( + outbound: mpsc::Sender, + inbound: mpsc::Receiver, + tools: Vec, + ) -> Self { + Self { + outbound, + inbound, + tools, + } + } + + pub async fn run(mut self) -> anyhow::Result<()> { + while let Some(message) = self.inbound.recv().await { + if self.handle_message(message).await? { + break; + } + } + + Ok(()) + } + + async fn handle_message( + &mut self, + message: OrchestratorToExecutorMessage, + ) -> anyhow::Result { + match message { + OrchestratorToExecutorMessage::ListTools(request) => { + self.send(ExecutorToOrchestratorMessage::ListToolsResponse( + ListToolsResponse { + request_id: request.request_id, + tools: self.tools.clone(), + }, + )) + .await?; + Ok(false) + } + OrchestratorToExecutorMessage::CallTool(request) => { + let outcome = if self.tool_exists(request.tool_name.as_str()) { + ToolCallOutcome::Error { + message: format!( + "tool `{}` is registered but execution is not implemented yet", + request.tool_name + ), + } + } else { + ToolCallOutcome::Error { + message: format!("unknown executor tool `{}`", request.tool_name), + } + }; + + self.send(ExecutorToOrchestratorMessage::CallToolResponse( + CallToolResponse { + request_id: request.request_id, + tool_name: request.tool_name, + outcome, + }, + )) + .await?; + Ok(false) + } + OrchestratorToExecutorMessage::Shutdown(request) => { + self.send(ExecutorToOrchestratorMessage::ShutdownResponse( + ShutdownResponse { + request_id: request.request_id, + }, + )) + .await?; + Ok(true) + } + } + } + + async fn send(&self, message: ExecutorToOrchestratorMessage) -> anyhow::Result<()> { + self.outbound + .send(message) + .await + .context("executor outbound channel closed") + } + + fn tool_exists(&self, tool_name: &str) -> bool { + self.tools.iter().any(|tool| tool.name == tool_name) + } +} + +#[cfg(test)] +mod tests { + use super::Executor; + use codex_executor_protocol::CallToolRequest; + use codex_executor_protocol::ExecutorToOrchestratorMessage; + use codex_executor_protocol::ExecutorToolSpec; + use codex_executor_protocol::ListToolsRequest; + use codex_executor_protocol::ListToolsResponse; + use codex_executor_protocol::OrchestratorToExecutorMessage; + use codex_executor_protocol::ShutdownRequest; + use codex_executor_protocol::ShutdownResponse; + use codex_executor_protocol::ToolCallOutcome; + use pretty_assertions::assert_eq; + use serde_json::json; + use tokio::sync::mpsc; + + #[tokio::test] + async fn executor_lists_registered_tools() { + let (outbound_tx, mut outbound_rx) = mpsc::channel(4); + let (inbound_tx, inbound_rx) = mpsc::channel(4); + let executor = Executor::with_tools( + outbound_tx, + inbound_rx, + vec![ExecutorToolSpec::new( + "exec_command", + "Run a command", + json!({"type": "object"}), + )], + ); + + let task = tokio::spawn(async move { executor.run().await }); + inbound_tx + .send(OrchestratorToExecutorMessage::ListTools(ListToolsRequest { + request_id: "req-1".to_string(), + })) + .await + .expect("send request"); + + let response = outbound_rx.recv().await.expect("response"); + assert_eq!( + response, + ExecutorToOrchestratorMessage::ListToolsResponse(ListToolsResponse { + request_id: "req-1".to_string(), + tools: vec![ExecutorToolSpec::new( + "exec_command", + "Run a command", + json!({"type": "object"}), + )], + }) + ); + + drop(inbound_tx); + task.await.expect("task join").expect("task result"); + } + + #[tokio::test] + async fn executor_returns_error_for_call_tool_until_handlers_exist() { + let (outbound_tx, mut outbound_rx) = mpsc::channel(4); + let (inbound_tx, inbound_rx) = mpsc::channel(4); + let executor = Executor::with_tools( + outbound_tx, + inbound_rx, + vec![ExecutorToolSpec::new( + "exec_command", + "Run a command", + json!({"type": "object"}), + )], + ); + + let task = tokio::spawn(async move { executor.run().await }); + inbound_tx + .send(OrchestratorToExecutorMessage::CallTool(CallToolRequest { + request_id: "req-2".to_string(), + tool_name: "exec_command".to_string(), + arguments: json!({"cmd": "pwd"}), + })) + .await + .expect("send request"); + + let response = outbound_rx.recv().await.expect("response"); + let ExecutorToOrchestratorMessage::CallToolResponse(response) = response else { + panic!("expected call tool response"); + }; + assert_eq!(response.request_id, "req-2"); + assert_eq!(response.tool_name, "exec_command"); + assert_eq!( + response.outcome, + ToolCallOutcome::Error { + message: "tool `exec_command` is registered but execution is not implemented yet" + .to_string(), + } + ); + + drop(inbound_tx); + task.await.expect("task join").expect("task result"); + } + + #[tokio::test] + async fn executor_acknowledges_shutdown_and_exits() { + let (outbound_tx, mut outbound_rx) = mpsc::channel(4); + let (inbound_tx, inbound_rx) = mpsc::channel(4); + let executor = Executor::new(outbound_tx, inbound_rx); + + let task = tokio::spawn(async move { executor.run().await }); + inbound_tx + .send(OrchestratorToExecutorMessage::Shutdown(ShutdownRequest { + request_id: "req-3".to_string(), + })) + .await + .expect("send request"); + + let response = outbound_rx.recv().await.expect("response"); + assert_eq!( + response, + ExecutorToOrchestratorMessage::ShutdownResponse(ShutdownResponse { + request_id: "req-3".to_string(), + }) + ); + + task.await.expect("task join").expect("task result"); + } +} diff --git a/codex-rs/executor/src/main.rs b/codex-rs/executor/src/main.rs new file mode 100644 index 0000000000..e89c5c93dc --- /dev/null +++ b/codex-rs/executor/src/main.rs @@ -0,0 +1,7 @@ +use clap::Parser; +use codex_executor::Cli; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + codex_executor::run_main(Cli::parse()).await +}