feat(executor): add executor protocol and runtime crates

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"

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" }

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 }

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"])

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 }

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<String>,
+        description: impl Into<String>,
+        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<ExecutorToolSpec>,
+}
+
+#[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<ToolCallContent> },
+    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<String>,
+    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<String>, message: impl Into<String>) -> 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);
+    }
+}

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 }

codex-rs/executor/src/cli.rs

@@ -0,0 +1,5 @@
+use clap::Parser;
+
+#[derive(Debug, Parser, Default)]
+#[command(version)]
+pub struct Cli {}

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<ExecutorToOrchestratorMessage>,
+    mpsc::Receiver<OrchestratorToExecutorMessage>,
+)> {
+    unimplemented!("executor transport establishment is not implemented yet");
+}
+
+pub struct Executor {
+    outbound: mpsc::Sender<ExecutorToOrchestratorMessage>,
+    inbound: mpsc::Receiver<OrchestratorToExecutorMessage>,
+    tools: Vec<ExecutorToolSpec>,
+}
+
+impl Executor {
+    pub fn new(
+        outbound: mpsc::Sender<ExecutorToOrchestratorMessage>,
+        inbound: mpsc::Receiver<OrchestratorToExecutorMessage>,
+    ) -> Self {
+        Self::with_tools(outbound, inbound, Vec::new())
+    }
+
+    pub fn with_tools(
+        outbound: mpsc::Sender<ExecutorToOrchestratorMessage>,
+        inbound: mpsc::Receiver<OrchestratorToExecutorMessage>,
+        tools: Vec<ExecutorToolSpec>,
+    ) -> 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<bool> {
+        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");
+    }
+}

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
+}