From fbb04a608d68ad66b74d2033c859b8698b311b53 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Sun, 2 Nov 2025 13:31:05 -0800 Subject: [PATCH] turn start --- .../app-server-protocol/src/protocol/v2.rs | 13 +--- .../app-server/src/codex_message_processor.rs | 7 +-- codex-rs/app-server/tests/common/lib.rs | 1 + .../tests/common/mock_model_server.rs | 21 +++++++ .../app-server/tests/suite/v2/turn_start.rs | 60 +++++++++++-------- 5 files changed, 63 insertions(+), 39 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index b2ff0f2573..d23ef3bf56 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -416,19 +416,12 @@ pub struct TurnInterruptResponse {} // User input types #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type")] #[ts(export_to = "v2/")] pub enum UserInput { - Text(String), - Image(ImageInput), - LocalImage { path: PathBuf }, -} - -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] -#[serde(tag = "type", rename_all = "camelCase")] -#[ts(export_to = "v2/")] -pub enum ImageInput { - #[serde(rename = "image")] + Text { text: String }, Image { url: String }, + LocalImage { path: PathBuf }, } // Thread items diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 4f72df6cf1..fd2f4069cc 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -29,7 +29,6 @@ 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; @@ -981,10 +980,8 @@ impl CodexMessageProcessor { .input .into_iter() .map(|item| match item { - V2UserInput::Text(text) => CoreInputItem::Text { text }, - V2UserInput::Image(V2ImageInput::Image { url }) => { - CoreInputItem::Image { image_url: url } - } + V2UserInput::Text { text } => CoreInputItem::Text { text }, + V2UserInput::Image { url } => CoreInputItem::Image { image_url: url }, V2UserInput::LocalImage { path } => CoreInputItem::LocalImage { path }, }) .collect(); diff --git a/codex-rs/app-server/tests/common/lib.rs b/codex-rs/app-server/tests/common/lib.rs index 71b27190c2..951005b07b 100644 --- a/codex-rs/app-server/tests/common/lib.rs +++ b/codex-rs/app-server/tests/common/lib.rs @@ -10,6 +10,7 @@ pub use auth_fixtures::write_chatgpt_auth; use codex_app_server_protocol::JSONRPCResponse; pub use mcp_process::McpProcess; pub use mock_model_server::create_mock_chat_completions_server; +pub use mock_model_server::create_mock_chat_completions_server_unchecked; pub use responses::create_apply_patch_sse_response; pub use responses::create_final_assistant_message_sse_response; pub use responses::create_shell_sse_response; diff --git a/codex-rs/app-server/tests/common/mock_model_server.rs b/codex-rs/app-server/tests/common/mock_model_server.rs index be7f3eb5b3..1092c80a14 100644 --- a/codex-rs/app-server/tests/common/mock_model_server.rs +++ b/codex-rs/app-server/tests/common/mock_model_server.rs @@ -29,6 +29,27 @@ pub async fn create_mock_chat_completions_server(responses: Vec) -> Mock server } +/// Same as `create_mock_chat_completions_server` but does not set an expectation +/// on the number of requests. Useful for flows where the exact number of +/// model calls may vary (e.g., when the pipeline conditionally compacts or +/// when the test only needs a responsive endpoint). +pub async fn create_mock_chat_completions_server_unchecked(responses: Vec) -> MockServer { + let server = MockServer::start().await; + + let seq_responder = SeqResponder { + num_calls: AtomicUsize::new(0), + responses, + }; + + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(seq_responder) + .mount(&server) + .await; + + server +} + struct SeqResponder { num_calls: AtomicUsize, responses: Vec, diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 59177aab04..55c1537322 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -1,10 +1,8 @@ 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::create_mock_chat_completions_server_unchecked as 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; @@ -22,7 +20,7 @@ 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<()> { +async fn turn_start_emits_notifications_and_accepts_model_override() -> 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![ @@ -60,26 +58,13 @@ async fn turn_start_emits_notification_and_completes_without_overrides() -> Resu .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. + // Start a turn with only input and thread_id set (no overrides). let turn_req = mcp .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), - input: vec![V2UserInput::Text("Hello".to_string())], + input: vec![V2UserInput::Text { + text: "Hello".to_string(), + }], cwd: None, approval_policy: None, sandbox_policy: None, @@ -109,10 +94,35 @@ async fn turn_start_emits_notification_and_completes_without_overrides() -> Resu 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( + // Send a second turn that exercises the overrides path: change the model. + let turn_req2 = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "Second".to_string(), + }], + cwd: None, + approval_policy: None, + sandbox_policy: None, + model: Some("mock-model-override".to_string()), + effort: None, + summary: None, + }) + .await?; + let turn_resp2: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, - mcp.read_stream_until_notification_message("codex/event/task_complete"), + mcp.read_stream_until_response_message(RequestId::Integer(turn_req2)), + ) + .await??; + let TurnStartResponse { turn: turn2 } = to_response::(turn_resp2)?; + assert!(!turn2.id.is_empty()); + // Ensure the second turn has a different id than the first. + assert_ne!(turn.id, turn2.id); + + // Expect a second turn/started notification as well. + let _notif2: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/started"), ) .await??; @@ -180,6 +190,8 @@ async fn turn_start_accepts_local_image_input() -> Result<()> { let TurnStartResponse { turn } = to_response::(turn_resp)?; assert!(!turn.id.is_empty()); + // This test only validates that turn/start responds and returns a turn. + drop(server); Ok(()) }