diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 1b02e4bb6c..7290fca908 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -5995,6 +5995,15 @@ impl CodexMessageProcessor { let collaboration_mode = params.collaboration_mode.map(|mode| { self.normalize_turn_start_collaboration_mode(mode, collaboration_modes_config) }); + let config_snapshot = thread.config_snapshot().await; + let collaboration_mode = match collaboration_mode { + Some(mode) => mode, + None => config_snapshot.collaboration_mode.with_updates( + params.model.clone(), + params.effort.map(Some), + /*developer_instructions*/ None, + ), + }; // Map v2 input items to core input items. let mapped_items: Vec = params @@ -6003,50 +6012,34 @@ impl CodexMessageProcessor { .map(V2UserInput::into_core) .collect(); - let has_any_overrides = params.cwd.is_some() - || params.approval_policy.is_some() - || params.approvals_reviewer.is_some() - || params.sandbox_policy.is_some() - || params.model.is_some() - || params.service_tier.is_some() - || params.effort.is_some() - || params.summary.is_some() - || collaboration_mode.is_some() - || params.personality.is_some(); - - // If any overrides are provided, update the session turn context first. - if has_any_overrides { - let _ = self - .submit_core_op( - &request_id, - thread.as_ref(), - Op::OverrideTurnContext { - cwd: params.cwd, - approval_policy: params.approval_policy.map(AskForApproval::to_core), - approvals_reviewer: params - .approvals_reviewer - .map(codex_app_server_protocol::ApprovalsReviewer::to_core), - sandbox_policy: params.sandbox_policy.map(|p| p.to_core()), - windows_sandbox_level: None, - model: params.model, - effort: params.effort.map(Some), - summary: params.summary, - service_tier: params.service_tier, - collaboration_mode, - personality: params.personality, - }, - ) - .await; - } - - // Start the turn by submitting the user input. Return its submission id as turn_id. let turn_id = self .submit_core_op( &request_id, thread.as_ref(), - Op::UserInput { + Op::UserTurn { items: mapped_items, + cwd: params.cwd.unwrap_or(config_snapshot.cwd), + approval_policy: params + .approval_policy + .map(AskForApproval::to_core) + .unwrap_or(config_snapshot.approval_policy), + approvals_reviewer: Some( + params + .approvals_reviewer + .map(codex_app_server_protocol::ApprovalsReviewer::to_core) + .unwrap_or(config_snapshot.approvals_reviewer), + ), + sandbox_policy: params + .sandbox_policy + .map(|policy| policy.to_core()) + .unwrap_or(config_snapshot.sandbox_policy), + model: collaboration_mode.model().to_string(), + effort: collaboration_mode.reasoning_effort(), + summary: params.summary, + service_tier: params.service_tier, final_output_json_schema: params.output_schema, + collaboration_mode: Some(collaboration_mode), + personality: params.personality, }, ) .await; @@ -8353,6 +8346,8 @@ mod tests { use anyhow::Result; use codex_app_server_protocol::ServerRequestPayload; use codex_app_server_protocol::ToolRequestUserInputParams; + use codex_protocol::config_types::ModeKind; + use codex_protocol::config_types::Settings; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -8475,6 +8470,14 @@ mod tests { cwd: PathBuf::from("/tmp"), ephemeral: false, reasoning_effort: None, + collaboration_mode: CollaborationMode { + mode: ModeKind::Default, + settings: Settings { + model: "gpt-5".to_string(), + reasoning_effort: None, + developer_instructions: None, + }, + }, personality: None, session_source: SessionSource::Cli, }; 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 8d7ca02613..49f64631b2 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -77,6 +77,14 @@ fn body_contains(req: &wiremock::Request, text: &str) -> bool { .is_some_and(|body| body.contains(text)) } +fn request_payload_for_prompt(requests: &[wiremock::Request], prompt: &str) -> serde_json::Value { + let request = requests + .iter() + .find(|req| body_contains(req, prompt)) + .unwrap_or_else(|| panic!("missing request containing prompt: {prompt}")); + serde_json::from_slice(&request.body).expect("request body should be valid JSON") +} + #[tokio::test] async fn turn_start_sends_originator_header() -> Result<()> { let responses = vec![create_final_assistant_message_sse_response("Done")?]; @@ -509,6 +517,300 @@ async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<( Ok(()) } +#[tokio::test] +async fn turn_start_omitted_settings_inherit_latest_session_state_v2() -> Result<()> { + skip_if_no_network!(Ok(())); + + let tmp = TempDir::new()?; + let codex_home = tmp.path().to_path_buf(); + let workspace = tmp.path().join("workspace"); + std::fs::create_dir(&workspace)?; + let inherited_cwd = workspace.join("inherited"); + std::fs::create_dir(&inherited_cwd)?; + + let responses = vec![ + create_final_assistant_message_sse_response("thread started")?, + create_final_assistant_message_sse_response("session updated")?, + create_shell_command_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(42)".to_string(), + ], + None, + Some(5000), + "call-inherited", + )?, + create_final_assistant_message_sse_response("inherited done")?, + ]; + let server = create_mock_responses_server_sequence(responses).await; + create_config_toml( + codex_home.as_path(), + &server.uri(), + "untrusted", + &BTreeMap::default(), + )?; + + let mut mcp = McpProcess::new(codex_home.as_path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + + let first_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "set session defaults".to_string(), + text_elements: Vec::new(), + }], + cwd: Some(inherited_cwd.clone()), + approval_policy: Some(codex_app_server_protocol::AskForApproval::Never), + sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess), + model: Some("mock-model-session".to_string()), + effort: Some(ReasoningEffort::Medium), + summary: Some(ReasoningSummary::Auto), + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(first_turn_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + mcp.clear_message_buffer(); + + let second_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "run inherited command".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)), + ) + .await??; + + let command_execution_item = timeout(DEFAULT_READ_TIMEOUT, async { + loop { + match mcp.read_next_message().await? { + JSONRPCMessage::Request(_) => { + anyhow::bail!("unexpected server request for inherited turn") + } + JSONRPCMessage::Notification(notification) + if notification.method == "item/started" => + { + let item_started: ItemStartedNotification = + serde_json::from_value(notification.params.expect("item/started params"))?; + if let ThreadItem::CommandExecution { .. } = item_started.item { + return Ok::(item_started.item); + } + } + _ => {} + } + } + }) + .await??; + let ThreadItem::CommandExecution { cwd, .. } = command_execution_item else { + unreachable!("loop returns only command execution items"); + }; + assert_eq!(cwd, inherited_cwd); + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let requests = server + .received_requests() + .await + .expect("failed to fetch received requests"); + let payload = request_payload_for_prompt(&requests, "run inherited command"); + assert_eq!(payload["model"].as_str(), Some("mock-model-session")); + + Ok(()) +} + +#[tokio::test] +async fn turn_start_explicit_settings_override_latest_session_state_v2() -> Result<()> { + skip_if_no_network!(Ok(())); + + let tmp = TempDir::new()?; + let codex_home = tmp.path().to_path_buf(); + let workspace = tmp.path().join("workspace"); + std::fs::create_dir(&workspace)?; + let session_cwd = workspace.join("session"); + let override_cwd = workspace.join("override"); + std::fs::create_dir(&session_cwd)?; + std::fs::create_dir(&override_cwd)?; + + let responses = vec![ + create_final_assistant_message_sse_response("thread started")?, + create_final_assistant_message_sse_response("session updated")?, + create_shell_command_sse_response( + vec![ + "python3".to_string(), + "-c".to_string(), + "print(99)".to_string(), + ], + None, + Some(5000), + "call-override", + )?, + create_final_assistant_message_sse_response("override done")?, + ]; + let server = create_mock_responses_server_sequence(responses).await; + create_config_toml( + codex_home.as_path(), + &server.uri(), + "untrusted", + &BTreeMap::default(), + )?; + + let mut mcp = McpProcess::new(codex_home.as_path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let start_id = mcp + .send_thread_start_request(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let start_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(start_resp)?; + + let first_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "set session defaults".to_string(), + text_elements: Vec::new(), + }], + cwd: Some(session_cwd.clone()), + approval_policy: Some(codex_app_server_protocol::AskForApproval::Never), + sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess), + model: Some("mock-model-session".to_string()), + effort: Some(ReasoningEffort::Medium), + summary: Some(ReasoningSummary::Auto), + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(first_turn_id)), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + mcp.clear_message_buffer(); + + let second_turn_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![V2UserInput::Text { + text: "run explicit override command".to_string(), + text_elements: Vec::new(), + }], + cwd: Some(override_cwd.clone()), + approval_policy: Some(codex_app_server_protocol::AskForApproval::UnlessTrusted), + sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess), + model: Some("mock-model-override".to_string()), + effort: Some(ReasoningEffort::Medium), + summary: Some(ReasoningSummary::Auto), + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)), + ) + .await??; + + let approval_request = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_request_message(), + ) + .await??; + let ServerRequest::CommandExecutionRequestApproval { request_id, params } = approval_request + else { + panic!("expected command approval request for explicit override turn"); + }; + assert_eq!(params.item_id, "call-override"); + + mcp.send_response( + request_id, + serde_json::to_value(CommandExecutionRequestApprovalResponse { + decision: CommandExecutionApprovalDecision::Accept, + })?, + ) + .await?; + + let command_execution_item = timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let message = mcp.read_next_message().await?; + let JSONRPCMessage::Notification(notification) = message else { + continue; + }; + if notification.method != "item/started" { + continue; + } + let item_started: ItemStartedNotification = + serde_json::from_value(notification.params.expect("item/started params"))?; + if let ThreadItem::CommandExecution { .. } = item_started.item { + return Ok::(item_started.item); + } + } + }) + .await??; + let ThreadItem::CommandExecution { cwd, .. } = command_execution_item else { + unreachable!("loop returns only command execution items"); + }; + assert_eq!(cwd, override_cwd); + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let requests = server + .received_requests() + .await + .expect("failed to fetch received requests"); + let payload = request_payload_for_prompt(&requests, "run explicit override command"); + assert_eq!(payload["model"].as_str(), Some("mock-model-override")); + + Ok(()) +} + #[tokio::test] async fn turn_start_accepts_collaboration_mode_override_v2() -> Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index c17710278f..4f93cee49a 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -1075,6 +1075,7 @@ impl SessionConfiguration { cwd: self.cwd.clone(), ephemeral: self.original_config_do_not_use.ephemeral, reasoning_effort: self.collaboration_mode.reasoning_effort(), + collaboration_mode: self.collaboration_mode.clone(), personality: self.personality, session_source: self.session_source.clone(), } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index e016fec977..a498e42232 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -39,6 +39,7 @@ pub struct ThreadConfigSnapshot { pub cwd: PathBuf, pub ephemeral: bool, pub reasoning_effort: Option, + pub collaboration_mode: codex_protocol::config_types::CollaborationMode, pub personality: Option, pub session_source: SessionSource, }