From d8075b923c9e93c5b744a862fbb6df30678e2378 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sun, 2 Nov 2025 13:15:53 -0800 Subject: [PATCH] wip --- .../app-server-protocol/src/protocol/v2.rs | 84 ++++++- .../app-server/src/codex_message_processor.rs | 121 +++++++++- .../app-server/tests/common/mcp_process.rs | 10 + codex-rs/app-server/tests/suite/v2/mod.rs | 2 + .../tests/suite/v2/thread_archive.rs | 11 +- .../app-server/tests/suite/v2/turn_start.rs | 209 ++++++++++++++++++ 6 files changed, 424 insertions(+), 13 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/turn_start.rs diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 8b489b219c..b2ff0f2573 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1,8 +1,10 @@ use std::collections::HashMap; +use std::path::PathBuf; use codex_protocol::ConversationId; use codex_protocol::account::PlanType; use codex_protocol::config_types::ReasoningEffort; +use codex_protocol::config_types::ReasoningSummary; use mcp_types::ContentBlock as McpContentBlock; use schemars::JsonSchema; use serde::Deserialize; @@ -48,6 +50,69 @@ v2_enum_from_core!( } ); +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(tag = "mode", rename_all = "camelCase")] +#[ts(tag = "mode")] +#[ts(export_to = "v2/")] +pub enum SandboxPolicy { + DangerFullAccess, + ReadOnly, + WorkspaceWrite { + #[serde(default, skip_serializing_if = "Vec::is_empty")] + writable_roots: Vec, + #[serde(default)] + network_access: bool, + #[serde(default)] + exclude_tmpdir_env_var: bool, + #[serde(default)] + exclude_slash_tmp: bool, + }, +} + +impl SandboxPolicy { + pub fn to_core(&self) -> codex_protocol::protocol::SandboxPolicy { + match self { + SandboxPolicy::DangerFullAccess => { + codex_protocol::protocol::SandboxPolicy::DangerFullAccess + } + SandboxPolicy::ReadOnly => codex_protocol::protocol::SandboxPolicy::ReadOnly, + SandboxPolicy::WorkspaceWrite { + writable_roots, + network_access, + exclude_tmpdir_env_var, + exclude_slash_tmp, + } => codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { + writable_roots: writable_roots.clone(), + network_access: *network_access, + exclude_tmpdir_env_var: *exclude_tmpdir_env_var, + exclude_slash_tmp: *exclude_slash_tmp, + }, + } + } +} + +impl From for SandboxPolicy { + fn from(value: codex_protocol::protocol::SandboxPolicy) -> Self { + match value { + codex_protocol::protocol::SandboxPolicy::DangerFullAccess => { + SandboxPolicy::DangerFullAccess + } + codex_protocol::protocol::SandboxPolicy::ReadOnly => SandboxPolicy::ReadOnly, + codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { + writable_roots, + network_access, + exclude_tmpdir_env_var, + exclude_slash_tmp, + } => SandboxPolicy::WorkspaceWrite { + writable_roots, + network_access, + exclude_tmpdir_env_var, + exclude_slash_tmp, + }, + } + } +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -284,6 +349,7 @@ pub struct Thread { #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct Turn { + pub id: String, pub items: Vec, pub status: TurnStatus, pub error: Option, @@ -314,16 +380,25 @@ pub enum TurnStatus { pub struct TurnStartParams { pub thread_id: String, pub input: Vec, - pub model: String, - pub effort: ReasoningEffort, - pub summary: String, + // Override the working directory for this turn and subsequent turns. + pub cwd: Option, + // Override the approval policy for this turn and subsequent turns. + pub approval_policy: Option, + // Override the sandbox policy for this turn and subsequent turns. + pub sandbox_policy: Option, + // Override the model for this turn and subsequent turns. + pub model: Option, + // Override the reasoning effort for this turn and subsequent turns. + pub effort: Option, + // Override the reasoning summary for this turn and subsequent turns. + pub summary: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct TurnStartResponse { - pub turn_id: String, + pub turn: Turn, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] @@ -345,6 +420,7 @@ pub struct TurnInterruptResponse {} pub enum UserInput { Text(String), Image(ImageInput), + LocalImage { path: PathBuf }, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 3603f823b0..4f72df6cf1 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -29,6 +29,7 @@ use codex_app_server_protocol::GetConversationSummaryResponse; use codex_app_server_protocol::GetUserAgentResponse; use codex_app_server_protocol::GetUserSavedConfigResponse; use codex_app_server_protocol::GitDiffToRemoteResponse; +use codex_app_server_protocol::ImageInput as V2ImageInput; use codex_app_server_protocol::InputItem as WireInputItem; use codex_app_server_protocol::InterruptConversationParams; use codex_app_server_protocol::InterruptConversationResponse; @@ -71,6 +72,7 @@ use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; use codex_app_server_protocol::Turn; use codex_app_server_protocol::UserInfoResponse; +use codex_app_server_protocol::UserInput as V2UserInput; use codex_app_server_protocol::UserSavedConfig; use codex_backend_client::Client as BackendClient; use codex_core::AuthManager; @@ -209,12 +211,8 @@ impl CodexMessageProcessor { self.send_unimplemented_error(request_id, "thread/compact") .await; } - ClientRequest::TurnStart { - request_id, - params: _, - } => { - self.send_unimplemented_error(request_id, "turn/start") - .await; + ClientRequest::TurnStart { request_id, params } => { + self.turn_start(request_id, params).await; } ClientRequest::TurnInterrupt { request_id, @@ -942,6 +940,117 @@ impl CodexMessageProcessor { } } + async fn turn_start( + &self, + request_id: RequestId, + params: codex_app_server_protocol::TurnStartParams, + ) { + // Resolve conversation id from v2 thread id string. + let conversation_id = match ConversationId::from_string(¶ms.thread_id) { + Ok(id) => id, + Err(err) => { + let error = JSONRPCErrorError { + code: INVALID_REQUEST_ERROR_CODE, + message: format!("invalid thread id: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + let Ok(conversation) = self + .conversation_manager + .get_conversation(conversation_id) + .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; + }; + + // Keep a copy of v2 inputs for the notification payload. + let v2_inputs_for_notif = params.input.clone(); + + // Map v2 input items to core input items. + let mapped_items: Vec = params + .input + .into_iter() + .map(|item| match item { + V2UserInput::Text(text) => CoreInputItem::Text { text }, + V2UserInput::Image(V2ImageInput::Image { url }) => { + CoreInputItem::Image { image_url: url } + } + V2UserInput::LocalImage { path } => CoreInputItem::LocalImage { path }, + }) + .collect(); + + let has_any_overrides = params.cwd.is_some() + || params.approval_policy.is_some() + || params.sandbox_policy.is_some() + || params.model.is_some() + || params.effort.is_some() + || params.summary.is_some(); + + // If any overrides are provided, update the session turn context first. + if has_any_overrides { + let _ = conversation + .submit(Op::OverrideTurnContext { + cwd: params.cwd, + approval_policy: params + .approval_policy + .map(codex_app_server_protocol::AskForApproval::to_core), + sandbox_policy: params.sandbox_policy.map(|p| p.to_core()), + model: params.model, + effort: params.effort.map(Some), + summary: params.summary, + }) + .await; + } + + // Start the turn by submitting the user input. Return its submission id as turn_id. + let turn_id = conversation + .submit(Op::UserInput { + items: mapped_items, + }) + .await; + + match turn_id { + Ok(turn_id) => { + let turn = codex_app_server_protocol::Turn { + id: turn_id.clone(), + items: vec![codex_app_server_protocol::ThreadItem::UserMessage { + id: turn_id, + content: v2_inputs_for_notif, + }], + status: codex_app_server_protocol::TurnStatus::InProgress, + error: None, + }; + + let response = codex_app_server_protocol::TurnStartResponse { turn: turn.clone() }; + self.outgoing.send_response(request_id, response).await; + + // Emit v2 turn/started notification. + let notif = codex_app_server_protocol::TurnStartedNotification { turn }; + self.outgoing + .send_server_notification(ServerNotification::TurnStarted(notif)) + .await; + } + Err(err) => { + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to start turn: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + } + } + } + async fn thread_archive(&self, request_id: RequestId, params: ThreadArchiveParams) { // Resolve conversation id from v2 thread id string. let conversation_id = match ConversationId::from_string(¶ms.thread.id) { diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index 7743034a5f..1622d89606 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -35,6 +35,7 @@ use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::TurnStartParams as V2TurnStartParams; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; @@ -347,6 +348,15 @@ impl McpProcess { self.send_request("thread/resume", params).await } + /// Send a `turn/start` JSON-RPC request (v2). + pub async fn send_turn_start_request( + &mut self, + params: V2TurnStartParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("turn/start", params).await + } + /// Send a `cancelLoginChatGpt` JSON-RPC request. pub async fn send_cancel_login_chat_gpt_request( &mut self, diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 18b54c743d..87af7aaf3b 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -1,3 +1,5 @@ +mod thread_archive; mod thread_list; mod thread_resume; mod thread_start; +mod turn_start; diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index 8fad8f47fa..3c830ed3e0 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -52,11 +52,17 @@ async fn thread_archive_moves_rollout_into_archived_directory() -> Result<()> { .path() .join(SESSIONS_SUBDIR) .join(format!("{}.jsonl", thread.id)); - assert!(rollout_path.exists(), "expected {} to exist", rollout_path.display()); + assert!( + rollout_path.exists(), + "expected {} to exist", + rollout_path.display() + ); // Archive the thread. let archive_id = mcp - .send_thread_archive_request(ThreadArchiveParams { thread: thread.clone() }) + .send_thread_archive_request(ThreadArchiveParams { + thread: thread.clone(), + }) .await?; let archive_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, @@ -93,4 +99,3 @@ approval_policy = "never" sandbox_mode = "read-only" "# } - diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs new file mode 100644 index 0000000000..59177aab04 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -0,0 +1,209 @@ +use anyhow::Result; +use app_test_support::McpProcess; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_chat_completions_server; +use app_test_support::to_response; +use codex_app_server_protocol::AddConversationListenerParams; +use codex_app_server_protocol::AddConversationSubscriptionResponse; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::UserInput as V2UserInput; +use pretty_assertions::assert_eq; +use std::path::Path; +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 turn_start_emits_notification_and_completes_without_overrides() -> Result<()> { + // Provide a mock server and config so model wiring is valid. + // Two Codex turns hit the mock model (session start + turn/start). + let responses = vec![ + create_final_assistant_message_sse_response("Done")?, + create_final_assistant_message_sse_response("Done")?, + ]; + let server = create_mock_chat_completions_server(responses).await; + + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + // Start a thread (v2) and capture its id. + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + model_provider: None, + profile: None, + cwd: None, + approval_policy: None, + sandbox: None, + config: None, + base_instructions: None, + developer_instructions: None, + compact_prompt: None, + include_apply_patch_tool: None, + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(thread_resp)?; + + // Add legacy listener to get task_complete signal. + let conversation_id = codex_protocol::ConversationId::from_string(&thread.id)?; + let add_listener_id = mcp + .send_add_conversation_listener_request(AddConversationListenerParams { + conversation_id, + experimental_raw_events: false, + }) + .await?; + let add_listener_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(add_listener_id)), + ) + .await??; + let AddConversationSubscriptionResponse { .. } = to_response::<_>(add_listener_resp)?; + + // Start a turn with only input and thread_id set. + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text("Hello".to_string())], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: None, + effort: None, + summary: None, + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + assert!(!turn.id.is_empty()); + + // Expect a turn/started notification. + let notif: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), + ) + .await??; + let started: TurnStartedNotification = + serde_json::from_value(notif.params.expect("params must be present"))?; + assert_eq!( + started.turn.status, + codex_app_server_protocol::TurnStatus::InProgress + ); + + // And we should see task_complete for the legacy stream as the turn finishes. + let _task_done: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("codex/event/task_complete"), + ) + .await??; + + drop(server); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn turn_start_accepts_local_image_input() -> Result<()> { + // Two Codex turns hit the mock model (session start + turn/start). + let responses = vec![ + create_final_assistant_message_sse_response("Done")?, + create_final_assistant_message_sse_response("Done")?, + ]; + let server = create_mock_chat_completions_server(responses).await; + + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let thread_req = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + model_provider: None, + profile: None, + cwd: None, + approval_policy: None, + sandbox: None, + config: None, + base_instructions: None, + developer_instructions: None, + compact_prompt: None, + include_apply_patch_tool: None, + }) + .await?; + let thread_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_req)), + ) + .await??; + let ThreadStartResponse { thread } = to_response::(thread_resp)?; + + let image_path = codex_home.path().join("image.png"); + // No need to actually write the file; we just exercise the input path. + + let turn_req = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::LocalImage { path: image_path }], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: None, + effort: None, + summary: None, + }) + .await?; + let turn_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_req)), + ) + .await??; + let TurnStartResponse { turn } = to_response::(turn_resp)?; + assert!(!turn.id.is_empty()); + + drop(server); + Ok(()) +} + +// Helper to create a config.toml pointing at the mock model server. +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" +sandbox_mode = "danger-full-access" + +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 +"# + ), + ) +}