openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-13 11:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use UserTurn for app-server turn start

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Dylan Hurd
2026-03-21 14:21:33 -07:00
parent 25c5ccc751
commit 4648ec4e9b
4 changed files with 345 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
codex-rs/app-server/src/codex_message_processor.rs
View File

@@ -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<CoreInputItem> = 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,
};

302
codex-rs/app-server/tests/suite/v2/turn_start.rs
View File

@@ -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::<ThreadStartResponse>(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::<ThreadItem, anyhow::Error>(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::<ThreadStartResponse>(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::<ThreadItem, anyhow::Error>(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(()));

1
codex-rs/core/src/codex.rs
View File

@@ -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(),
}

1
codex-rs/core/src/codex_thread.rs
View File

@@ -39,6 +39,7 @@ pub struct ThreadConfigSnapshot {
pub cwd: PathBuf,
pub ephemeral: bool,
pub reasoning_effort: Option<ReasoningEffort>,
pub collaboration_mode: codex_protocol::config_types::CollaborationMode,
pub personality: Option<Personality>,
pub session_source: SessionSource,
}