Add per-turn cyber access program selection (#40687)

## What changed

- Add the experimental `turn/start.cyberAccessProgram` option with `standard`,
  `daybreakBlue`, and `daybreakRed` values.
- Forward the selection as `access_programs.cyber` on Responses, WebSocket,
  and remote-compaction requests made with ChatGPT authentication, while
  omitting it for API-key and custom-provider requests.
- Preserve the per-turn selection across recovery, compaction, and child-agent
  turns without making it a persistent thread setting.

## Testing

- Cover app-server forwarding, authentication boundaries, WebSocket reuse,
  turn recovery, compaction, and child-agent inheritance.

GitOrigin-RevId: d2eb468f365b2214c5099bc21741cf30f8bd2eb5
This commit is contained in:
faizan-oai
2026-08-25 19:54:52 +00:00
committed by copyberry
parent a6e63f9f32
commit 94967e03e5
72 changed files with 1323 additions and 273 deletions
@@ -727,6 +727,15 @@
],
"type": "string"
},
"CyberAccessProgram": {
"description": "Requested cyber treatment for a ChatGPT-authenticated Codex turn. Authorization and model-tier restrictions remain server-owned.",
"enum": [
"standard",
"daybreakBlue",
"daybreakRed"
],
"type": "string"
},
"DynamicToolNamespaceTool": {
"oneOf": [
{
@@ -10256,6 +10256,15 @@
],
"type": "object"
},
"CyberAccessProgram": {
"description": "Requested cyber treatment for a ChatGPT-authenticated Codex turn. Authorization and model-tier restrictions remain server-owned.",
"enum": [
"standard",
"daybreakBlue",
"daybreakRed"
],
"type": "string"
},
"DeprecationNoticeNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -6217,6 +6217,15 @@
],
"type": "object"
},
"CyberAccessProgram": {
"description": "Requested cyber treatment for a ChatGPT-authenticated Codex turn. Authorization and model-tier restrictions remain server-owned.",
"enum": [
"standard",
"daybreakBlue",
"daybreakRed"
],
"type": "string"
},
"DeprecationNoticeNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
@@ -120,6 +120,15 @@
],
"type": "object"
},
"CyberAccessProgram": {
"description": "Requested cyber treatment for a ChatGPT-authenticated Codex turn. Authorization and model-tier restrictions remain server-owned.",
"enum": [
"standard",
"daybreakBlue",
"daybreakRed"
],
"type": "string"
},
"ImageDetail": {
"enum": [
"auto",
@@ -0,0 +1,9 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
/**
* Requested cyber treatment for a ChatGPT-authenticated Codex turn.
* Authorization and model-tier restrictions remain server-owned.
*/
export type CyberAccessProgram = "standard" | "daybreakBlue" | "daybreakRed";
@@ -113,6 +113,7 @@ export type { ConsumeAccountRateLimitResetCreditParams } from "./ConsumeAccountR
export type { ConsumeAccountRateLimitResetCreditResponse } from "./ConsumeAccountRateLimitResetCreditResponse";
export type { ContextCompactedNotification } from "./ContextCompactedNotification";
export type { CreditsSnapshot } from "./CreditsSnapshot";
export type { CyberAccessProgram } from "./CyberAccessProgram";
export type { DeprecationNoticeNotification } from "./DeprecationNoticeNotification";
export type { DesktopOnboardingEntrypoint } from "./DesktopOnboardingEntrypoint";
export type { DynamicToolCallOutputContentItem } from "./DynamicToolCallOutputContentItem";
@@ -4733,12 +4733,42 @@ fn turn_start_params_preserve_explicit_null_service_tier() {
collaboration_mode: None,
multi_agent_mode: None,
personality: None,
cyber_access_program: None,
};
let serialized_without_override =
serde_json::to_value(&without_override).expect("params should serialize");
assert_eq!(serialized_without_override.get("serviceTier"), None);
}
#[test]
fn turn_start_cyber_access_program_uses_separate_wire_formats() {
for (app_server_value, core_value) in [
("standard", "standard"),
("daybreakBlue", "daybreak_blue"),
("daybreakRed", "daybreak_red"),
] {
let params: TurnStartParams = serde_json::from_value(json!({
"threadId": "thread_123",
"input": [],
"cyberAccessProgram": app_server_value,
}))
.expect("params should deserialize");
let core_program: codex_protocol::turn_input::CyberAccessProgram = params
.cyber_access_program
.expect("explicit program")
.into();
assert_eq!(
serde_json::to_value(params).expect("params should serialize")["cyberAccessProgram"],
app_server_value,
);
assert_eq!(
serde_json::to_value(core_program).expect("core program should serialize"),
json!(core_value),
);
}
}
#[test]
fn turn_start_params_round_trip_multi_agent_mode() {
let params: TurnStartParams = serde_json::from_value(json!({
@@ -13,6 +13,7 @@ use codex_protocol::models::ImageDetail;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::plan_tool::PlanItemArg as CorePlanItemArg;
use codex_protocol::plan_tool::StepStatus as CorePlanStepStatus;
use codex_protocol::turn_input::CyberAccessProgram as CoreCyberAccessProgram;
use codex_protocol::user_input::ByteRange as CoreByteRange;
use codex_protocol::user_input::TextElement as CoreTextElement;
use codex_protocol::user_input::UserInput as CoreUserInput;
@@ -111,6 +112,28 @@ pub struct AdditionalContextEntry {
pub kind: AdditionalContextKind,
}
/// Requested cyber treatment for a ChatGPT-authenticated Codex turn.
/// Authorization and model-tier restrictions remain server-owned.
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub enum CyberAccessProgram {
Standard,
DaybreakBlue,
DaybreakRed,
}
impl From<CyberAccessProgram> for CoreCyberAccessProgram {
fn from(value: CyberAccessProgram) -> Self {
match value {
CyberAccessProgram::Standard => Self::Standard,
CyberAccessProgram::DaybreakBlue => Self::DaybreakBlue,
CyberAccessProgram::DaybreakRed => Self::DaybreakRed,
}
}
}
#[derive(
Serialize, Deserialize, Debug, Default, Clone, PartialEq, JsonSchema, TS, ExperimentalApi,
)]
@@ -215,6 +238,12 @@ pub struct TurnStartParams {
#[experimental("turn/start.multiAgentMode")]
#[ts(optional = nullable)]
pub multi_agent_mode: Option<MultiAgentMode>,
/// EXPERIMENTAL - Request a workspace-authorized cyber program for this
/// turn. Omission preserves automatic behavior. This does not grant access.
#[experimental("turn/start.cyberAccessProgram")]
#[ts(optional = nullable)]
pub cyber_access_program: Option<CyberAccessProgram>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+4
View File
@@ -914,6 +914,10 @@ You can optionally specify config overrides on the new turn. If specified, these
`serviceTierForTurn` overrides the tier only when the request starts a new turn, without changing the thread's saved tier. Use `"default"` for standard speed, or omit it (or pass `null`) to inherit the thread's tier. It is ignored when the request steers an active turn. The existing `serviceTier` field still changes the tier for subsequent turns, including when both fields are supplied.
Experimental `cyberAccessProgram` also applies only to the new turn. It accepts `standard`, `daybreakBlue`, or `daybreakRed`; omission preserves automatic backend behavior. For ChatGPT-authenticated requests through the built-in OpenAI provider, Codex sends the corresponding `standard`, `daybreak_blue`, or `daybreak_red` value in `access_programs.cyber` on Responses and remote-compaction requests. WebSocket `response.create` messages carry the choice per request, so changing it does not require reconnecting. The server still enforces workspace authorization and model restrictions. API-key and custom-provider requests omit this field. This field does not change the saved model or grant access.
Child agents use the invoking turn's choice when spawned or started on a new follow-up, including after a reload. Input delivered into an already-running child turn does not change that turn's choice.
`approvalsReviewer` accepts:
- `"user"` — default. Review approval requests directly in the client.
@@ -681,6 +681,7 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
output_schema: None,
collaboration_mode: None,
multi_agent_mode: None,
cyber_access_program: None,
},
},
Some(remote_trace),
@@ -720,6 +720,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: effort.clone(),
summary: ReasoningSummary::Auto,
}),
@@ -79,6 +79,7 @@ fn turn_context_item(
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: codex_protocol::config_types::ReasoningSummary::Auto,
})
@@ -583,6 +583,7 @@ impl TurnRequestProcessor {
turn_trigger: params.turn_trigger,
final_output_json_schema: params.output_schema,
service_tier: params.service_tier_for_turn,
cyber_access_program: params.cyber_access_program.map(Into::into),
..Default::default()
})
.with_additional_context(additional_context)
@@ -0,0 +1,169 @@
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::TestAppServer;
use app_test_support::write_chatgpt_auth;
use codex_app_server_protocol::CyberAccessProgram;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_login::AuthCredentialsStoreMode;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use wiremock::Mock;
use wiremock::ResponseTemplate;
use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[tokio::test]
async fn turn_start_forwards_explicit_cyber_access_program() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
Mock::given(method("GET"))
.and(path("/v1/responses"))
.respond_with(ResponseTemplate::new(426))
.mount(&server)
.await;
let programs = [
None,
Some(CyberAccessProgram::DaybreakBlue),
Some(CyberAccessProgram::DaybreakRed),
None,
Some(CyberAccessProgram::Standard),
None,
];
let requests = responses::mount_sse_sequence(
&server,
programs
.iter()
.enumerate()
.map(|(index, _)| {
responses::sse(vec![responses::ev_completed(&format!("resp-{index}"))])
})
.collect(),
)
.await;
let home = TempDir::new()?;
std::fs::write(
home.path().join("config.toml"),
format!(
"model = \"gpt-5.5\"\napproval_policy = \"never\"\nopenai_base_url = \"{}/v1\"\ncli_auth_credentials_store = \"file\"\n",
server.uri()
),
)?;
write_chatgpt_auth(
home.path(),
ChatGptAuthFixture::new("chatgpt-test-token").plan_type("pro"),
AuthCredentialsStoreMode::File,
)?;
let mut app = TestAppServer::builder()
.with_codex_home(home.path())
.without_managed_config()
.build_initialized()
.await?;
let thread = app.start_thread(ThreadStartParams::default()).await?.thread;
for program in programs {
let completed = app
.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id.clone(),
cyber_access_program: program,
input: vec![UserInput::Text {
text: "hello".to_owned(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
assert_eq!(completed.turn.status, TurnStatus::Completed);
}
let requests = requests.requests();
assert_eq!(
requests
.iter()
.map(|request| request.body_json().get("access_programs").cloned())
.collect::<Vec<_>>(),
[
None,
Some(json!({"cyber": "daybreak_blue"})),
Some(json!({"cyber": "daybreak_red"})),
None,
Some(json!({"cyber": "standard"})),
None,
]
);
Ok(())
}
#[tokio::test]
async fn turn_start_forwards_cyber_access_program_with_personal_access_token() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
Mock::given(method("GET"))
.and(path("/v1/responses"))
.respond_with(ResponseTemplate::new(426))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/v1/user-auth-credential/whoami"))
.and(header("Authorization", "Bearer at-test-token"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"email": null,
"chatgpt_user_id": "user-123",
"chatgpt_account_id": "account-123",
"chatgpt_plan_type": "enterprise_cbp_automation",
"chatgpt_account_is_fedramp": false,
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/backend-api/wham/config/bundle"))
.and(header("Authorization", "Bearer at-test-token"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
.mount(&server)
.await;
let request = responses::mount_sse_once(
&server,
responses::sse(vec![responses::ev_completed("resp-1")]),
)
.await;
let home = TempDir::new()?;
std::fs::write(
home.path().join("config.toml"),
format!(
"model = \"gpt-5.5\"\napproval_policy = \"never\"\nopenai_base_url = \"{0}/v1\"\nchatgpt_base_url = \"{0}/backend-api\"\n",
server.uri(),
),
)?;
let authapi_base_url = server.uri();
let mut app = TestAppServer::builder()
.with_codex_home(home.path())
.without_managed_config()
.with_env_overrides(&[
("OPENAI_API_KEY", None),
("CODEX_ACCESS_TOKEN", Some("at-test-token")),
("CODEX_AUTHAPI_BASE_URL", Some(authapi_base_url.as_str())),
])
.build_initialized()
.await?;
let thread = app.start_thread(ThreadStartParams::default()).await?.thread;
app.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id,
cyber_access_program: Some(CyberAccessProgram::DaybreakBlue),
input: vec![UserInput::Text {
text: "hello".to_owned(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
assert_eq!(
request.single_request().body_json()["access_programs"],
json!({"cyber": "daybreak_blue"})
);
Ok(())
}
@@ -21,6 +21,7 @@ mod connection_handling_websocket_unix;
#[cfg(unix)]
mod curated_mcp_sync;
mod current_time;
mod cyber_access_program;
mod dynamic_tools;
mod environment_add;
mod environment_info;
@@ -2804,6 +2804,7 @@ async fn turn_start_explicit_local_environment_updates_legacy_cwd_between_turns(
output_schema: None,
collaboration_mode: None,
multi_agent_mode: None,
cyber_access_program: None,
},
})
.await?;
@@ -2849,6 +2850,7 @@ async fn turn_start_explicit_local_environment_updates_legacy_cwd_between_turns(
output_schema: None,
collaboration_mode: None,
multi_agent_mode: None,
cyber_access_program: None,
},
})
.await?;
+26
View File
@@ -8,6 +8,7 @@ use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TurnModerationMetadataEvent;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::turn_input::CyberAccessProgram;
use futures::Stream;
use serde::Deserialize;
use serde::Serialize;
@@ -23,6 +24,24 @@ use tokio::sync::mpsc;
pub const WS_REQUEST_HEADER_TRACEPARENT_CLIENT_METADATA_KEY: &str = "ws_request_header_traceparent";
pub const WS_REQUEST_HEADER_TRACESTATE_CLIENT_METADATA_KEY: &str = "ws_request_header_tracestate";
/// Explicit per-request access selection using the Responses API wire values.
#[derive(Debug, Clone, Copy, PartialEq, Serialize)]
pub struct AccessPrograms {
cyber: &'static str,
}
impl From<CyberAccessProgram> for AccessPrograms {
fn from(program: CyberAccessProgram) -> Self {
Self {
cyber: match program {
CyberAccessProgram::Standard => "standard",
CyberAccessProgram::DaybreakBlue => "daybreak_blue",
CyberAccessProgram::DaybreakRed => "daybreak_red",
},
}
}
}
/// Canonical input payload for the compaction endpoint.
#[derive(Debug, Clone, Serialize)]
pub struct CompactionInput<'a> {
@@ -41,6 +60,8 @@ pub struct CompactionInput<'a> {
pub prompt_cache_key: Option<&'a str>,
#[serde(skip_serializing_if = "Option::is_none")]
pub text: Option<TextControls>,
#[serde(skip_serializing_if = "Option::is_none")]
pub access_programs: Option<AccessPrograms>,
}
/// Canonical input payload for the memory summarize endpoint.
@@ -272,6 +293,8 @@ pub struct ResponsesApiRequest {
pub text: Option<TextControls>,
#[serde(skip_serializing_if = "Option::is_none")]
pub client_metadata: Option<HashMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub access_programs: Option<AccessPrograms>,
}
impl<'a> From<&'a ResponsesApiRequest> for ResponseCreateWsRequest<'a> {
@@ -294,6 +317,7 @@ impl<'a> From<&'a ResponsesApiRequest> for ResponseCreateWsRequest<'a> {
text: request.text.as_ref(),
generate: None,
client_metadata: request.client_metadata.clone(),
access_programs: request.access_programs,
}
}
}
@@ -326,6 +350,8 @@ pub struct ResponseCreateWsRequest<'a> {
pub generate: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub client_metadata: Option<HashMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub access_programs: Option<AccessPrograms>,
}
pub fn response_create_client_metadata(
@@ -955,6 +955,9 @@ mod tests {
service_tier: Some("priority".to_string()),
prompt_cache_key: Some("cache-key".to_string()),
text: None,
access_programs: Some(
codex_protocol::turn_input::CyberAccessProgram::DaybreakBlue.into(),
),
client_metadata: Some(HashMap::from([(
"traceparent".to_string(),
"00-0123456789abcdef0123456789abcdef-0123456789abcdef-01".to_string(),
+1
View File
@@ -27,6 +27,7 @@ pub use crate::auth::AuthProvider;
pub use crate::auth::AuthProviderFuture;
pub use crate::auth::SharedAuthProvider;
pub use crate::auth::auth_header_telemetry;
pub use crate::common::AccessPrograms;
pub use crate::common::CompactionInput;
pub use crate::common::MemorySummarizeInput;
pub use crate::common::MemorySummarizeOutput;
+3
View File
@@ -333,6 +333,7 @@ async fn responses_client_stream_request_preserves_item_ids() -> Result<()> {
prompt_cache_key: None,
text: None,
client_metadata: None,
access_programs: None,
};
let expected = serde_json::to_value(&request)?;
@@ -420,6 +421,7 @@ async fn streaming_client_retries_on_transport_error() -> Result<()> {
prompt_cache_key: None,
text: None,
client_metadata: None,
access_programs: None,
};
let client = ResponsesClient::new(transport.clone(), provider, Arc::new(NoAuth));
@@ -540,6 +542,7 @@ async fn azure_store_sends_ids_and_headers() -> Result<()> {
prompt_cache_key: None,
text: None,
client_metadata: None,
access_programs: None,
};
let mut extra_headers = HeaderMap::new();
+22 -24
View File
@@ -52,6 +52,7 @@ use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::turn_input::CyberAccessProgram;
use codex_protocol::user_input::UserInput;
use codex_thread_store::LoadThreadHistoryParams;
use codex_thread_store::ReadThreadParams;
@@ -89,6 +90,7 @@ pub(crate) struct SpawnAgentOptions {
pub(crate) root_turn_id: Option<String>,
pub(crate) environments: Option<Vec<TurnEnvironmentSelection>>,
pub(crate) multi_agent_v2_usage_hints: Option<ResolvedMultiAgentV2UsageHints>,
pub(crate) cyber_access_program: Option<CyberAccessProgram>,
}
#[derive(Clone, Debug)]
@@ -183,19 +185,12 @@ impl AgentControl {
&self,
agent_id: ThreadId,
input: Vec<UserInput>,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: TurnStartOptions,
) -> CodexResult<String> {
let state = self.upgrade()?;
let thread = state.get_thread(agent_id).await?;
let result = match thread
.start_or_steer_turn(
TurnInputRequest::user_input(input).on_start(TurnStartOptions {
parent_turn_id,
root_turn_id,
..Default::default()
}),
)
.start_or_steer_turn(TurnInputRequest::user_input(input).on_start(start_options))
.await
{
Ok(TurnInputSubmission::Started { turn_id }) => Ok(turn_id),
@@ -220,8 +215,7 @@ impl AgentControl {
agent_id: ThreadId,
communication: InterAgentCommunication,
agent_communication_context: AgentCommunicationContext,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: TurnStartOptions,
) -> CodexResult<String> {
let state = self.upgrade()?;
if communication.trigger_turn {
@@ -234,8 +228,7 @@ impl AgentControl {
&state,
communication,
agent_communication_context,
parent_turn_id,
root_turn_id,
start_options,
)
.await
}
@@ -295,16 +288,14 @@ impl AgentControl {
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: TurnStartOptions,
) -> CodexResult<String> {
self.submit_inter_agent_communication(
agent_id,
state,
communication,
context,
parent_turn_id,
root_turn_id,
start_options,
)
.await
}
@@ -315,13 +306,18 @@ impl AgentControl {
state: &Arc<ThreadManagerState>,
communication: InterAgentCommunication,
context: AgentCommunicationContext,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: TurnStartOptions,
) -> CodexResult<String> {
let communication_for_log =
crate::agent_communication::logging_enabled().then(|| communication.clone());
let parent_turn_id = parent_turn_id.filter(|_| communication.trigger_turn);
let root_turn_id = root_turn_id.filter(|_| communication.trigger_turn);
let (parent_turn_id, root_turn_id) = if communication.trigger_turn {
(
start_options.parent_turn_id.clone(),
start_options.root_turn_id.clone(),
)
} else {
(None, None)
};
let result = self
.handle_thread_request_result(
agent_id,
@@ -329,7 +325,10 @@ impl AgentControl {
state
.send_op(
agent_id,
Op::InterAgentCommunication { communication },
Op::InterAgentCommunication {
communication,
start_options,
},
parent_turn_id,
root_turn_id,
)
@@ -642,8 +641,7 @@ impl AgentControl {
parent_thread_id,
communication,
context,
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
TurnStartOptions::default(),
)
.await;
return;
+9 -9
View File
@@ -742,15 +742,16 @@ impl AgentControl {
)
.await;
let start_options = TurnStartOptions {
parent_turn_id: options.parent_turn_id,
root_turn_id: options.root_turn_id,
cyber_access_program: options.cyber_access_program,
..Default::default()
};
match initial_input {
SpawnInitialInput::UserInput(input) => {
self.send_input(
new_thread.thread_id,
input,
options.parent_turn_id,
options.root_turn_id,
)
.await?;
self.send_input(new_thread.thread_id, input, start_options)
.await?;
}
SpawnInitialInput::InterAgentCommunication(communication, context) => {
self.send_inter_agent_communication_after_capacity_check(
@@ -758,8 +759,7 @@ impl AgentControl {
&state,
communication,
context,
options.parent_turn_id,
options.root_turn_id,
start_options,
)
.await?;
}
+14 -12
View File
@@ -111,9 +111,11 @@ fn captured_op_matches(actual: &(ThreadId, Op), expected: &(ThreadId, Op)) -> bo
(
Op::InterAgentCommunication {
communication: actual,
..
},
Op::InterAgentCommunication {
communication: expected,
..
},
) => actual == expected,
_ => false,
@@ -450,8 +452,7 @@ async fn send_input_errors_when_manager_dropped() {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await
.expect_err("send_input should fail without a manager");
@@ -576,8 +577,7 @@ async fn send_input_errors_when_thread_missing() {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await
.expect_err("send_input should fail for missing thread");
@@ -650,8 +650,7 @@ async fn send_input_submits_user_message() {
text: "hello from tests".to_string(),
text_elements: Vec::new(),
}],
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await
.expect("send_input should succeed");
@@ -677,8 +676,7 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await
.expect("send_inter_agent_communication should succeed");
@@ -688,6 +686,7 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig
thread_id,
Op::InterAgentCommunication {
communication: communication.clone(),
start_options: Default::default(),
},
);
let captured = harness
@@ -959,14 +958,16 @@ async fn check_v2_agent_reload(route: V2ReloadRoute) {
spawned_agent.thread_id,
communication.clone(),
AgentCommunicationContext::new(AgentCommunicationKind::Message, ThreadId::new()),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await
.expect("send_inter_agent_communication should succeed after reload");
let expected = (
spawned_agent.thread_id,
Op::InterAgentCommunication { communication },
Op::InterAgentCommunication {
communication,
start_options: Default::default(),
},
);
let captured = harness
.manager
@@ -3129,7 +3130,7 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
thread_id == worker_thread_id
&& matches!(
op,
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == tester_path
&& communication.recipient == worker_path
&& communication.content == "done"
@@ -3216,6 +3217,7 @@ async fn multi_agent_v2_completion_queues_message_for_direct_parent() {
expected_message.clone(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
},
);
+17
View File
@@ -116,6 +116,7 @@ use crate::client_common::ResponseEvent;
use crate::client_common::ResponseStream;
use crate::context::BaseInstructionsFragment;
use crate::context::ContextualUserFragment;
use crate::cyber_access_program;
use crate::feedback_tags;
use crate::responses_metadata::CodexResponsesMetadata;
use crate::responses_metadata::subagent_header_value;
@@ -307,6 +308,7 @@ struct WebsocketSession {
// This is intentionally not a `PartialEq` implementation: request equality includes `input` and
// `client_metadata`, while websocket reuse compares the input separately and ignores metadata.
// Access programs are authorized per response, including continuations, without replaying input.
// Keep the destructuring exhaustive so new request fields require an explicit reuse decision.
fn responses_request_properties_match(
previous: &ResponsesApiRequest,
@@ -328,6 +330,7 @@ fn responses_request_properties_match(
prompt_cache_key: previous_prompt_cache_key,
text: previous_text,
client_metadata: _,
access_programs: _,
} = previous;
let ResponsesApiRequest {
model: current_model,
@@ -345,6 +348,7 @@ fn responses_request_properties_match(
prompt_cache_key: current_prompt_cache_key,
text: current_text,
client_metadata: _,
access_programs: _,
} = current;
previous_model == current_model
@@ -620,6 +624,10 @@ impl ModelClient {
service_tier: service_tier.as_deref(),
prompt_cache_key: prompt_cache_key.as_deref(),
text,
access_programs: cyber_access_program::for_auth(
client_setup.auth.as_ref(),
prompt.cyber_access_program,
),
};
let mut extra_headers = ApiHeaderMap::new();
@@ -957,6 +965,7 @@ impl ModelClient {
prompt_cache_key,
text,
client_metadata: Some(responses_metadata.client_metadata()),
access_programs: None,
};
Ok(request)
}
@@ -1520,6 +1529,10 @@ impl ModelClientSession {
.extra_headers
.insert(X_CODEX_ROUTING_HINT_HEADER, header_value);
}
request.access_programs = cyber_access_program::for_auth(
client_setup.auth.as_ref(),
prompt.cyber_access_program,
);
self.client
.prepare_response_items_for_request(&mut request.input);
let request_session_telemetry =
@@ -1638,6 +1651,10 @@ impl ModelClientSession {
service_tier.clone(),
responses_metadata,
)?;
request.access_programs = cyber_access_program::for_auth(
client_setup.auth.as_ref(),
prompt.cyber_access_program,
);
let mut websocket_metadata = responses_metadata.clone();
websocket_metadata.routing_hint = self.client.build_routing_hint_header(
client_setup.auth.as_ref(),
+3
View File
@@ -34,6 +34,8 @@ pub struct Prompt {
/// Whether the Responses API should strictly validate `output_schema`.
pub output_schema_strict: bool,
pub(crate) cyber_access_program: Option<codex_protocol::turn_input::CyberAccessProgram>,
}
impl Default for Prompt {
@@ -45,6 +47,7 @@ impl Default for Prompt {
base_instructions: BaseInstructions::default(),
output_schema: None,
output_schema_strict: true,
cyber_access_program: None,
}
}
}
+4
View File
@@ -134,6 +134,7 @@ fn serializes_text_verbosity_when_set() {
format: None,
}),
client_metadata: None,
access_programs: None,
};
let v = serde_json::to_value(&req).expect("json");
@@ -178,6 +179,7 @@ fn serializes_text_schema_with_strict_format() {
service_tier: None,
text: Some(text_controls),
client_metadata: None,
access_programs: None,
};
let v = serde_json::to_value(&req).expect("json");
@@ -239,6 +241,7 @@ fn omits_text_when_not_set() {
service_tier: None,
text: None,
client_metadata: None,
access_programs: None,
};
let v = serde_json::to_value(&req).expect("json");
@@ -263,6 +266,7 @@ fn serializes_flex_service_tier_when_set() {
service_tier: Some(ServiceTier::Flex.to_string()),
text: None,
client_metadata: None,
access_programs: None,
};
let v = serde_json::to_value(&req).expect("json");
+7 -1
View File
@@ -55,6 +55,7 @@ use codex_protocol::turn_input::SuspendTurnOutcome;
use codex_protocol::turn_input::TurnInputMode;
use codex_protocol::turn_input::TurnInputRequest;
use codex_protocol::turn_input::TurnInputSubmission;
use codex_protocol::turn_input::TurnStartOptions;
use codex_thread_store::PersistContext;
use codex_thread_store::StoredThread;
use codex_thread_store::StoredThreadHistory;
@@ -391,10 +392,15 @@ impl CodexThread {
turn_id,
thread_settings,
trace,
cyber_access_program,
} = request;
let start_options = TurnStartOptions {
cyber_access_program,
..Default::default()
};
match self
.io
.submit_recover_turn(thread_settings, trace, turn_id)
.submit_recover_turn(thread_settings, start_options, trace, turn_id)
.await?
{
TurnInputSubmission::Started { turn_id } => {
@@ -67,6 +67,7 @@ pub(super) async fn run_remote_compact_attempt(
base_instructions,
output_schema: None,
output_schema_strict: true,
cyber_access_program: turn_context.cyber_access_program,
};
let responses_metadata = sess
.responses_metadata(
@@ -82,6 +82,7 @@ pub(super) async fn run_remote_compact_v2_attempt(
base_instructions,
output_schema: None,
output_schema_strict: true,
cyber_access_program: turn_context.cyber_access_program,
};
let responses_metadata = sess
@@ -336,6 +336,7 @@ fn reference_context_item() -> TurnContextItem {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(false),
cyber_access_program: None,
effort: None,
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}
+12
View File
@@ -0,0 +1,12 @@
use codex_api::AccessPrograms;
use codex_login::CodexAuth;
use codex_protocol::turn_input::CyberAccessProgram;
pub(crate) fn for_auth(
auth: Option<&CodexAuth>,
program: Option<CyberAccessProgram>,
) -> Option<AccessPrograms> {
program
.filter(|_| auth.is_some_and(CodexAuth::is_chatgpt_auth))
.map(AccessPrograms::from)
}
+1
View File
@@ -58,6 +58,7 @@ pub mod connectors;
pub mod context;
mod context_manager;
mod current_time;
mod cyber_access_program;
mod elicitation;
mod environment_selection;
pub mod exec;
+28 -22
View File
@@ -82,16 +82,11 @@ pub async fn inter_agent_communication(
sess: &Arc<Session>,
sub_id: String,
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: codex_protocol::turn_input::TurnStartOptions,
) {
let trigger_turn = communication.trigger_turn;
sess.input_queue
.enqueue_mailbox_communication(
communication,
parent_turn_id.filter(|_| trigger_turn),
root_turn_id.filter(|_| trigger_turn),
)
.enqueue_mailbox_communication(communication, start_options)
.await;
crate::agent_communication::emit_agent_communication_receive(&sub_id);
if trigger_turn || sess.has_outstanding_durable_sleep() {
@@ -118,7 +113,9 @@ pub async fn run_user_shell_command(sess: &Arc<Session>, sub_id: String, command
return;
}
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
let turn_context = sess
.new_turn_with_default_settings(sub_id, Default::default())
.await;
sess.spawn_task(turn_context, Vec::new(), UserShellCommandTask::new(command))
.await;
}
@@ -237,7 +234,9 @@ pub async fn reload_user_config(sess: &Arc<Session>) {
}
pub async fn compact(sess: &Arc<Session>, sub_id: String) {
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
let turn_context = sess
.new_turn_with_default_settings(sub_id, Default::default())
.await;
sess.spawn_task(turn_context, Vec::new(), CompactTask).await;
}
@@ -268,7 +267,9 @@ pub async fn thread_rollback(sess: &Arc<Session>, sub_id: String, num_turns: u32
return;
}
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
let turn_context = sess
.new_turn_with_default_settings(sub_id, Default::default())
.await;
let live_thread = match sess.live_thread_for_persistence("rollback thread") {
Ok(live_thread) => live_thread,
Err(_) => {
@@ -478,7 +479,9 @@ pub async fn review(
sub_id: String,
review_request: ReviewRequest,
) {
let turn_context = sess.new_default_turn_with_sub_id(sub_id.clone()).await;
let turn_context = sess
.new_turn_with_default_settings(sub_id.clone(), Default::default())
.await;
sess.maybe_emit_model_warnings_for_turn(turn_context.as_ref())
.await;
#[allow(deprecated)]
@@ -572,10 +575,16 @@ pub(super) async fn submission_loop(
}
Op::RecoverTurn {
thread_settings,
start_options,
reply,
} => {
let result =
turn_input::handle_recovery(&sess, thread_settings, sub.id.clone()).await;
let result = turn_input::handle_recovery(
&sess,
thread_settings,
start_options,
sub.id.clone(),
)
.await;
let _ = reply.send(result);
false
}
@@ -605,15 +614,12 @@ pub(super) async fn submission_loop(
let _ = reply.send(outcome);
false
}
Op::InterAgentCommunication { communication } => {
inter_agent_communication(
&sess,
sub.id.clone(),
communication,
sub.parent_turn_id,
sub.root_turn_id,
)
.await;
Op::InterAgentCommunication {
communication,
start_options,
} => {
inter_agent_communication(&sess, sub.id.clone(), communication, start_options)
.await;
false
}
Op::ExecApproval {
+69 -52
View File
@@ -6,6 +6,7 @@ use codex_diagnostics::GaugeGuard;
use codex_history::ResponseItemEnvelope;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::turn_input::TurnStartOptions;
use codex_protocol::user_input::UserInput;
use serde::Deserialize;
use serde::Serialize;
@@ -81,8 +82,7 @@ pub(crate) struct InputQueue {
struct PendingMailboxCommunication {
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: TurnStartOptions,
_diagnostics_guard: GaugeGuard,
}
@@ -121,16 +121,14 @@ impl InputQueue {
pub(crate) async fn enqueue_mailbox_communication(
&self,
communication: InterAgentCommunication,
parent_turn_id: Option<String>,
root_turn_id: Option<String>,
start_options: TurnStartOptions,
) {
self.mailbox_pending_mails
.lock()
.await
.push_back(PendingMailboxCommunication {
communication,
parent_turn_id,
root_turn_id,
start_options,
_diagnostics_guard: PENDING_MAILBOX_MESSAGES.track(),
});
self.activity_tx.send_replace(InputQueueActivity::Mailbox);
@@ -148,29 +146,35 @@ impl InputQueue {
.any(|mail| mail.communication.trigger_turn)
}
pub(crate) async fn drain_mailbox_input_items(
&self,
) -> (Vec<TurnInput>, Option<String>, Option<String>) {
pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec<TurnInput>, TurnStartOptions) {
let pending_mails = self
.mailbox_pending_mails
.lock()
.await
.drain(..)
.collect::<Vec<_>>();
let parent_turn_id = pending_mails
// A later follow-up supersedes the earlier choice, including an omitted choice.
let mut start_options = pending_mails
.iter()
.rev()
.find(|mail| mail.communication.trigger_turn)
.map(|mail| mail.start_options.clone())
.unwrap_or_default();
start_options.parent_turn_id = pending_mails
.iter()
.filter(|mail| mail.communication.trigger_turn)
.map(|mail| mail.parent_turn_id.as_deref())
.map(|mail| mail.start_options.parent_turn_id.as_deref())
.reduce(|expected, candidate| expected.filter(|id| candidate == Some(*id)))
.and_then(|id| id.filter(|id| !id.trim().is_empty()).map(str::to_string));
let root_turn_id = pending_mails
start_options.root_turn_id = pending_mails
.iter()
.filter(|mail| mail.communication.trigger_turn)
.map(|mail| {
mail.parent_turn_id
mail.start_options
.parent_turn_id
.as_deref()
.filter(|id| !id.trim().is_empty())
.and(mail.root_turn_id.as_deref())
.and(mail.start_options.root_turn_id.as_deref())
})
.reduce(|expected, candidate| expected.filter(|id| candidate == Some(*id)))
.and_then(|id| id.filter(|id| !id.trim().is_empty()).map(str::to_string));
@@ -178,7 +182,7 @@ impl InputQueue {
.into_iter()
.map(|mail| TurnInput::InterAgentCommunication(mail.communication))
.collect();
(items, parent_turn_id, root_turn_id)
(items, start_options)
}
pub(crate) async fn turn_state_for_sub_id(
@@ -284,7 +288,7 @@ impl InputQueue {
pub(crate) async fn get_pending_input(
&self,
active_turn: &Mutex<Option<ActiveTurn>>,
) -> (Vec<TurnInput>, Option<String>, Option<String>) {
) -> (Vec<TurnInput>, TurnStartOptions) {
let (pending_input, accepts_mailbox_delivery, active_turn_metadata) = {
let mut active = active_turn.lock().await;
match active.as_mut() {
@@ -311,9 +315,9 @@ impl InputQueue {
}
};
if !accepts_mailbox_delivery {
return (pending_input, None, None);
return (pending_input, TurnStartOptions::default());
}
let (mailbox_items, parent_turn_id, root_turn_id) = self.drain_mailbox_input_items().await;
let (mailbox_items, start_options) = self.drain_mailbox_input_items().await;
if let Some(active_turn_metadata) = active_turn_metadata
&& mailbox_items.iter().any(|item| {
matches!(
@@ -322,16 +326,17 @@ impl InputQueue {
if communication.trigger_turn
)
})
&& (root_turn_id.is_none() || active_turn_metadata.root_turn_id() != root_turn_id)
&& (start_options.root_turn_id.is_none()
|| active_turn_metadata.root_turn_id() != start_options.root_turn_id)
{
active_turn_metadata.mark_root_turn_ambiguous();
}
if pending_input.is_empty() {
(mailbox_items, parent_turn_id, root_turn_id)
(mailbox_items, start_options)
} else {
let mut pending_input = pending_input;
pending_input.extend(mailbox_items);
(pending_input, parent_turn_id, root_turn_id)
(pending_input, start_options)
}
}
@@ -439,9 +444,7 @@ mod tests {
/*trigger_turn*/ false,
);
input_queue
.enqueue_mailbox_communication(
mail_one, /*parent_turn_id*/ None, /*root_turn_id*/ None,
)
.enqueue_mailbox_communication(mail_one, Default::default())
.await;
let mail_two = make_mail(
AgentPath::root(),
@@ -450,9 +453,7 @@ mod tests {
/*trigger_turn*/ false,
);
input_queue
.enqueue_mailbox_communication(
mail_two, /*parent_turn_id*/ None, /*root_turn_id*/ None,
)
.enqueue_mailbox_communication(mail_two, Default::default())
.await;
activity_rx.changed().await.expect("mailbox update");
@@ -527,18 +528,10 @@ mod tests {
);
input_queue
.enqueue_mailbox_communication(
mail_one.clone(),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(mail_one.clone(), Default::default())
.await;
input_queue
.enqueue_mailbox_communication(
mail_two.clone(),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(mail_two.clone(), Default::default())
.await;
assert_eq!(
@@ -581,14 +574,46 @@ mod tests {
input_queue
.enqueue_mailbox_communication(
make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
parent_turn_id.map(str::to_string),
root_turn_id.map(str::to_string),
TurnStartOptions {
parent_turn_id: parent_turn_id.map(str::to_string),
root_turn_id: root_turn_id.map(str::to_string),
..Default::default()
},
)
.await;
}
let (_, parent_turn_id, root_turn_id) = input_queue.drain_mailbox_input_items().await;
assert_eq!(parent_turn_id.as_deref(), expected_parent_turn_id);
assert_eq!(root_turn_id.as_deref(), expected_root_turn_id);
let (_, start_options) = input_queue.drain_mailbox_input_items().await;
assert_eq!(
start_options.parent_turn_id.as_deref(),
expected_parent_turn_id
);
assert_eq!(start_options.root_turn_id.as_deref(), expected_root_turn_id);
}
}
#[tokio::test]
async fn input_queue_uses_latest_followup_choice_and_ignores_queue_only_mail() {
use codex_protocol::turn_input::CyberAccessProgram;
for latest in [Some(CyberAccessProgram::Standard), None] {
let input_queue = InputQueue::new();
for (trigger_turn, program) in [
(true, Some(CyberAccessProgram::DaybreakBlue)),
(true, latest),
(false, Some(CyberAccessProgram::DaybreakRed)),
] {
input_queue
.enqueue_mailbox_communication(
make_mail(AgentPath::root(), AgentPath::root(), "task", trigger_turn),
TurnStartOptions {
cyber_access_program: program,
..Default::default()
},
)
.await;
}
let (_, start_options) = input_queue.drain_mailbox_input_items().await;
assert_eq!(start_options.cyber_access_program, latest);
}
}
@@ -603,11 +628,7 @@ mod tests {
/*trigger_turn*/ false,
);
input_queue
.enqueue_mailbox_communication(
queued_mail,
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(queued_mail, Default::default())
.await;
assert!(!input_queue.has_trigger_turn_mailbox_items().await);
@@ -618,11 +639,7 @@ mod tests {
/*trigger_turn*/ true,
);
input_queue
.enqueue_mailbox_communication(
trigger_mail,
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(trigger_mail, Default::default())
.await;
assert!(input_queue.has_trigger_turn_mailbox_items().await);
}
+4 -2
View File
@@ -363,6 +363,7 @@ use codex_protocol::protocol::WarningEvent;
use codex_protocol::turn_input::TurnInputMode;
use codex_protocol::turn_input::TurnInputRequest;
use codex_protocol::turn_input::TurnInputSubmission;
use codex_protocol::turn_input::TurnStartOptions;
use codex_protocol::user_input::UserInput;
use codex_skills_extension::HostSkillsService;
use codex_tools::ToolName;
@@ -887,6 +888,7 @@ impl SessionIo {
pub(crate) async fn submit_recover_turn(
&self,
thread_settings: ThreadSettingsOverrides,
start_options: TurnStartOptions,
trace: Option<W3cTraceContext>,
turn_id: String,
) -> CodexResult<TurnInputSubmission> {
@@ -895,6 +897,7 @@ impl SessionIo {
id: turn_id,
op: Op::RecoverTurn {
thread_settings,
start_options,
reply: reply_tx,
},
trace,
@@ -2151,8 +2154,7 @@ impl Session {
parent_thread_id,
communication,
context,
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
TurnStartOptions::default(),
)
.await
{
+1
View File
@@ -174,6 +174,7 @@ pub(super) async fn spawn_review_thread(
terminal_error: Arc::new(Mutex::new(None)),
server_model_warning_emitted: AtomicBool::new(false),
model_verification_emitted: AtomicBool::new(false),
cyber_access_program: None,
};
// Seed the child task with the review prompt as the initial user message.
@@ -246,6 +246,7 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previ
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -294,6 +295,7 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -1365,6 +1367,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -1456,6 +1459,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}))
@@ -1489,6 +1493,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -1621,6 +1626,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -1749,6 +1755,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -1921,6 +1928,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
-1
View File
@@ -537,7 +537,6 @@ pub(crate) struct SessionSettingsUpdate {
pub(crate) active_permission_profile: Option<ActivePermissionProfile>,
pub(crate) windows_sandbox_level: Option<WindowsSandboxLevel>,
pub(crate) service_tier_for_turn: Option<String>,
pub(crate) final_output_json_schema: Option<Option<Value>>,
pub(crate) app_server_client_name: Option<String>,
pub(crate) app_server_client_version: Option<String>,
}
@@ -223,7 +223,7 @@ async fn activation_fixture(models: Vec<ModelInfo>) -> ActivationFixture {
settings.reasoning_summary = Some(ReasoningSummary::Concise);
settings.service_tier = None;
let prepared = session
.new_default_turn_with_sub_id("step-activation-turn".to_string())
.new_turn_with_default_settings("step-activation-turn".to_string(), Default::default())
.await;
let turn = Arc::clone(&prepared);
let finish = Arc::new(Notify::new());
@@ -538,7 +538,10 @@ async fn delayed_activation_does_not_retarget_a_task(change: TaskChangeDuringLoo
let replacement = match change {
TaskChangeDuringLookup::FinishedAndReplaced => {
session
.new_default_turn_with_sub_id("replacement-turn".to_string())
.new_turn_with_default_settings(
"replacement-turn".to_string(),
Default::default(),
)
.await
}
TaskChangeDuringLookup::FinishedAndReusedContext => Arc::clone(&turn),
+23 -28
View File
@@ -1234,6 +1234,7 @@ async fn new_turn_refreshes_managed_network_proxy_for_sandbox_change() -> anyhow
sandbox_policy: Some(SandboxPolicy::DangerFullAccess),
..Default::default()
},
Default::default(),
)
.await?;
@@ -3679,6 +3680,7 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: Some(turn_context.realtime_active),
cyber_access_program: None,
effort: turn_context.reasoning_effort().cloned(),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
};
@@ -5594,7 +5596,7 @@ enabled = false
}
let child_turn = session
.new_default_turn_with_sub_id("role-skill-turn".to_string())
.new_turn_with_default_settings("role-skill-turn".to_string(), Default::default())
.await;
let skills_snapshot = child_turn.skills_snapshot();
let child_skill = skills_snapshot
@@ -5791,7 +5793,11 @@ async fn permission_profile_updates_apply_to_next_turn_environment() {
let next_turn = if apply_on_turn_start {
let (next_turn, _) = session
.new_turn_with_sub_id("permission-profile-update".to_string(), updates)
.new_turn_with_sub_id(
"permission-profile-update".to_string(),
updates,
Default::default(),
)
.await
.expect("turn permission profile update should succeed");
next_turn
@@ -5906,6 +5912,7 @@ async fn absolute_cwd_update_with_turn_environment_is_allowed() {
)),
..Default::default()
},
Default::default(),
)
.await
.expect("absolute cwd with explicit environments should succeed");
@@ -7473,6 +7480,7 @@ async fn turn_environments_set_primary_environment() {
)),
..Default::default()
},
Default::default(),
)
.await
.expect("turn should start");
@@ -7667,6 +7675,7 @@ async fn empty_turn_environments_clear_primary_environment() {
)),
..Default::default()
},
Default::default(),
)
.await
.expect("turn should start");
@@ -8931,6 +8940,7 @@ async fn mcp_policy_changes_schedule_runtime_refresh() {
},
..Default::default()
},
Default::default(),
)
.await
.expect("approval policy update should succeed");
@@ -11036,8 +11046,7 @@ async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() {
"pending trigger".to_string(),
/*trigger_turn*/ true,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -11116,11 +11125,7 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(
communication.clone(),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(communication.clone(), Default::default())
.await;
assert!(
@@ -11128,8 +11133,11 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
"queue-only mailbox mail should stay buffered once the current turn emitted its answer"
);
assert_eq!(
sess.input_queue.get_pending_input(&sess.active_turn).await,
(Vec::new(), None, None)
sess.input_queue
.get_pending_input(&sess.active_turn)
.await
.0,
Vec::new()
);
sess.abort_all_tasks(TurnAbortReason::Replaced).await;
@@ -11165,8 +11173,7 @@ async fn trigger_turn_mailbox_mail_waits_for_next_turn_after_answer_boundary() {
"late trigger update".to_string(),
/*trigger_turn*/ true,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -11204,11 +11211,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(
communication.clone(),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(communication.clone(), Default::default())
.await;
let submission = submit_steer_only(
&sess,
@@ -11260,11 +11263,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(
communication.clone(),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(communication.clone(), Default::default())
.await;
let submission = submit_steer_only(
&sess,
@@ -11320,11 +11319,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() {
.defer_mailbox_delivery_to_next_turn(&sess.active_turn, &tc.sub_id)
.await;
sess.input_queue
.enqueue_mailbox_communication(
communication.clone(),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
)
.enqueue_mailbox_communication(communication.clone(), Default::default())
.await;
let item = ResponseItem::FunctionCall {
@@ -499,7 +499,6 @@ async fn strict_auto_review_turn_grant_forces_guardian_for_exec_command_policy_s
kind: crate::state::TaskKind::Regular,
listen_to_cancellation_token: true,
},
crate::tasks::MailboxParentProvenance::Ignore,
)
.await;
+1
View File
@@ -1337,6 +1337,7 @@ pub(crate) fn build_prompt(
output_schema_strict: !crate::guardian::is_basic_session_source(
&turn_context.session_source,
),
cyber_access_program: turn_context.cyber_access_program,
}
}
+37 -19
View File
@@ -25,6 +25,7 @@ use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::TurnEnvironmentSelection;
use codex_protocol::turn_input::CyberAccessProgram;
use codex_sandboxing::policy_transforms::effective_permission_profile;
use codex_skills_extension::HostSkillsSnapshot;
use codex_skills_extension::SkillLoadOutcome;
@@ -173,6 +174,13 @@ impl std::fmt::Debug for TurnEnvironment {
}
}
/// Request options for one turn, separate from persistent thread settings.
#[derive(Default)]
pub(crate) struct NewTurnContextOptions {
pub(crate) final_output_json_schema: Option<Value>,
pub(crate) cyber_access_program: Option<CyberAccessProgram>,
}
/// The context needed for a single turn of the thread.
#[derive(Debug)]
pub struct TurnContext {
@@ -221,6 +229,8 @@ pub struct TurnContext {
pub(crate) terminal_error: Arc<Mutex<Option<ErrorEvent>>>,
pub(crate) server_model_warning_emitted: AtomicBool,
pub(crate) model_verification_emitted: AtomicBool,
/// Effective cyber treatment for this turn, including any child-agent inheritance.
pub(crate) cyber_access_program: Option<CyberAccessProgram>,
}
enum TurnMultiAgentRuntime {
@@ -532,6 +542,7 @@ impl TurnContext {
model_verification_emitted: AtomicBool::new(
self.model_verification_emitted.load(Ordering::Relaxed),
),
cyber_access_program: self.cyber_access_program,
}
}
@@ -581,6 +592,7 @@ impl TurnContext {
multi_agent_version: Some(self.multi_agent_version),
multi_agent_mode: None,
realtime_active: Some(self.realtime_active),
cyber_access_program: self.cyber_access_program,
effort: self.reasoning_effort().cloned(),
summary: ReasoningSummaryConfig::Auto,
}
@@ -784,6 +796,7 @@ impl Session {
terminal_error: Arc::new(Mutex::new(None)),
server_model_warning_emitted: AtomicBool::new(false),
model_verification_emitted: AtomicBool::new(false),
cyber_access_program: None,
}
}
@@ -791,9 +804,10 @@ impl Session {
&self,
sub_id: String,
updates: SessionSettingsUpdate,
options: NewTurnContextOptions,
) -> CodexResult<(Arc<TurnContext>, ThreadSettingsSnapshot)> {
let Some((turn_context, snapshot)) = self
.new_turn_with_sub_id_if(sub_id, updates, |_, _| true)
.new_turn_with_sub_id_if(sub_id, updates, options, |_, _| true)
.await?
else {
unreachable!("unconditional turn construction must accept valid settings");
@@ -811,9 +825,9 @@ impl Session {
&self,
sub_id: String,
updates: SessionSettingsUpdate,
options: NewTurnContextOptions,
should_start: impl FnOnce(&SessionConfiguration, &SessionConfiguration) -> bool + Send,
) -> CodexResult<Option<(Arc<TurnContext>, ThreadSettingsSnapshot)>> {
let final_output_json_schema = updates.final_output_json_schema.clone();
let service_tier_for_turn = updates.service_tier_for_turn.clone();
let commit = match self.update_settings_if(updates, should_start).await {
Ok(Some(commit)) => commit,
@@ -837,7 +851,7 @@ impl Session {
Arc::make_mut(&mut configuration.step_settings).service_tier = Some(service_tier);
}
let turn_context = self
.new_turn_from_configuration(sub_id, configuration, final_output_json_schema)
.new_turn_from_configuration(sub_id, configuration, options)
.await;
Ok(Some((turn_context, commit.snapshot)))
}
@@ -847,12 +861,12 @@ impl Session {
&self,
sub_id: String,
session_configuration: SessionConfiguration,
final_output_json_schema: Option<Option<Value>>,
options: NewTurnContextOptions,
) -> Arc<TurnContext> {
self.new_turn_context_from_configuration(
sub_id,
session_configuration,
final_output_json_schema,
options,
TurnMultiAgentRuntime::ResolveAndStore,
self.git_enrichment_policy,
)
@@ -867,7 +881,7 @@ impl Session {
self.new_turn_context_from_configuration(
sub_id,
session_configuration,
/*final_output_json_schema*/ None,
NewTurnContextOptions::default(),
TurnMultiAgentRuntime::Preview,
GitEnrichmentPolicy::Skip,
)
@@ -879,7 +893,7 @@ impl Session {
&self,
sub_id: String,
session_configuration: SessionConfiguration,
final_output_json_schema: Option<Option<Value>>,
options: NewTurnContextOptions,
multi_agent_runtime: TurnMultiAgentRuntime,
git_enrichment_policy: GitEnrichmentPolicy,
) -> Arc<TurnContext> {
@@ -988,8 +1002,9 @@ impl Session {
turn_context.extension_data.insert(trusted_plugin_roots);
turn_context.realtime_active = self.conversation.running_state().await.is_some();
if let Some(final_schema) = final_output_json_schema {
turn_context.final_output_json_schema = final_schema;
turn_context.final_output_json_schema = options.final_output_json_schema;
if turn_context.config.model_provider_id == codex_model_provider_info::OPENAI_PROVIDER_ID {
turn_context.cyber_access_program = options.cyber_access_program;
}
let turn_context = Arc::new(turn_context);
if git_enrichment_policy == GitEnrichmentPolicy::Fresh
@@ -1041,20 +1056,23 @@ impl Session {
}
pub(crate) async fn new_default_turn(&self) -> Arc<TurnContext> {
self.new_default_turn_with_sub_id(self.next_internal_sub_id())
.await
}
pub(crate) async fn new_default_turn_with_sub_id(&self, sub_id: String) -> Arc<TurnContext> {
let session_configuration = self.default_turn_configuration().await;
self.new_turn_from_configuration(
sub_id,
session_configuration,
/*final_output_json_schema*/ None,
self.new_turn_with_default_settings(
self.next_internal_sub_id(),
NewTurnContextOptions::default(),
)
.await
}
pub(crate) async fn new_turn_with_default_settings(
&self,
sub_id: String,
options: NewTurnContextOptions,
) -> Arc<TurnContext> {
let session_configuration = self.default_turn_configuration().await;
self.new_turn_from_configuration(sub_id, session_configuration, options)
.await
}
pub(crate) async fn new_startup_prewarm_turn_with_sub_id(
&self,
sub_id: String,
+16 -13
View File
@@ -13,10 +13,10 @@ use super::session::Session;
use super::session::SessionConfiguration;
use super::session::SessionSettingsUpdate;
use super::thread_settings;
use super::turn_context::NewTurnContextOptions;
use super::turn_context::TurnContext;
use crate::state::ActiveTurn;
use crate::state::TurnState;
use crate::tasks::MailboxParentProvenance;
use crate::tasks::RegularTask;
use codex_protocol::config_types::ModeKind;
use codex_protocol::error::CodexErr;
@@ -125,23 +125,30 @@ impl PreparedTurnInputSettings {
service_tier,
parent_turn_id,
root_turn_id,
cyber_access_program,
} = self.start_options;
let emit_thread_settings_applied = self.thread_settings_update.is_some();
let mut updates = self.thread_settings_update.unwrap_or_default();
updates.final_output_json_schema = Some(final_output_json_schema);
updates.service_tier_for_turn = service_tier;
let options = NewTurnContextOptions {
final_output_json_schema,
cyber_access_program,
};
let turn_context = match kind {
TurnStartKind::User | TurnStartKind::Recovery => Some(
session
.new_turn_with_sub_id(submission_id.clone(), updates)
.new_turn_with_sub_id(submission_id.clone(), updates, options)
.await?,
),
TurnStartKind::Automatic => {
session
.new_turn_with_sub_id_if(submission_id.clone(), updates, |current, proposed| {
kind.permits_settings(current, proposed)
})
.new_turn_with_sub_id_if(
submission_id.clone(),
updates,
options,
|current, proposed| kind.permits_settings(current, proposed),
)
.await?
}
};
@@ -209,13 +216,14 @@ pub(super) async fn handle(
pub(super) async fn handle_recovery(
session: &Arc<Session>,
thread_settings: ThreadSettingsOverrides,
start_options: TurnStartOptions,
submission_id: String,
) -> CodexResult<TurnInputSubmission> {
let request = TurnInputRequest::user_input(Vec::new())
.with_thread_settings(thread_settings)
.on_start(TurnStartOptions {
turn_trigger: Some("retry".to_string()),
..Default::default()
..start_options
});
start_if_idle(session, request, submission_id, TurnStartKind::Recovery).await
}
@@ -425,12 +433,7 @@ async fn start_if_idle(
}
}
session
.start_task(
turn_context,
task_input,
RegularTask::new(),
MailboxParentProvenance::Ignore,
)
.start_task(turn_context, task_input, RegularTask::new())
.await;
Ok(TurnInputSubmission::Started {
turn_id: submission_id,
+16 -11
View File
@@ -175,11 +175,12 @@ async fn start_only_rejects_active_turn_without_injecting() {
submission
);
assert_eq!(
(Vec::<TurnInput>::new(), None, None),
Vec::<TurnInput>::new(),
session
.input_queue
.get_pending_input(&session.active_turn)
.await
.0
);
session.abort_all_tasks(TurnAbortReason::Interrupted).await;
@@ -211,6 +212,7 @@ async fn recovery_rejects_active_turn_without_injecting_or_applying_settings() {
approval_policy: Some(AskForApproval::Never),
..Default::default()
},
TurnStartOptions::default(),
"recovered-turn".to_string(),
)
.await
@@ -235,8 +237,9 @@ async fn recovery_rejects_active_turn_without_injecting_or_applying_settings() {
session
.input_queue
.get_pending_input(&session.active_turn)
.await,
(Vec::<TurnInput>::new(), None, None)
.await
.0,
Vec::<TurnInput>::new()
);
session.abort_all_tasks(TurnAbortReason::Interrupted).await;
@@ -310,11 +313,12 @@ async fn start_only_rejects_current_plan_before_validating_settings() {
assert_eq!(session.thread_settings_snapshot().await, desired_settings);
assert!(session.active_turn.lock().await.is_none());
assert_eq!(
(Vec::<TurnInput>::new(), None, None),
Vec::<TurnInput>::new(),
session
.input_queue
.get_pending_input(&session.active_turn)
.await
.0
);
}
@@ -569,8 +573,9 @@ async fn automatic_admission_rechecks_plan_mode_without_committing_sparse_settin
session
.input_queue
.get_pending_input(&session.active_turn)
.await,
(Vec::<TurnInput>::new(), None, None)
.await
.0,
Vec::<TurnInput>::new()
);
// The rejected candidate is valid and would have real runtime effects if
@@ -657,8 +662,9 @@ async fn admission_revalidates_constraints_before_committing(kind: TurnStartKind
session
.input_queue
.get_pending_input(&session.active_turn)
.await,
(Vec::<TurnInput>::new(), None, None)
.await
.0,
Vec::<TurnInput>::new()
);
}
@@ -746,8 +752,7 @@ async fn start_only_rejects_pending_trigger_turn_without_injecting() {
"pending trigger".to_string(),
/*trigger_turn*/ true,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -840,7 +845,7 @@ async fn rejects_non_regular_turns() {
.turn_metadata_state
.set_root_turn_id("incoming-root".to_string());
let turn_context = session
.new_default_turn_with_sub_id("turn".to_string())
.new_turn_with_default_settings("turn".to_string(), Default::default())
.await;
turn_context
.turn_metadata_state
+54 -42
View File
@@ -30,6 +30,7 @@ use crate::hook_runtime::run_turn_interrupt_hooks;
use crate::session::TurnInput;
use crate::session::session::Session;
use crate::session::turn::run_hooks_and_record_inputs;
use crate::session::turn_context::NewTurnContextOptions;
use crate::session::turn_context::TurnContext;
use crate::state::ActiveTurn;
use crate::state::RunningTask;
@@ -70,11 +71,6 @@ static ACTIVE_TURNS: Gauge = Gauge::new("core.turns.active");
pub(crate) type SessionTaskResult = CodexResult<Option<String>>;
pub(crate) enum MailboxParentProvenance {
Ignore,
Attribute,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InterruptedTurnHistoryMarker {
Disabled,
@@ -278,8 +274,7 @@ impl Session {
) {
self.abort_all_tasks(TurnAbortReason::Replaced).await;
self.clear_connector_selection().await;
self.start_task(turn_context, input, task, MailboxParentProvenance::Ignore)
.await;
self.start_task(turn_context, input, task).await;
}
pub(crate) async fn start_task<T: SessionTask>(
@@ -287,7 +282,6 @@ impl Session {
turn_context: Arc<TurnContext>,
input: Vec<TurnInput>,
task: T,
mailbox_parent_provenance: MailboxParentProvenance,
) {
let task: Arc<dyn AnySessionTask> = Arc::new(task);
let task_kind = task.kind();
@@ -311,33 +305,14 @@ impl Session {
.await
.clear_turn(&turn_context.sub_id);
let (pending_items, parent_turn_id, root_turn_id) =
self.input_queue.get_pending_input(&self.active_turn).await;
if let MailboxParentProvenance::Attribute = mailbox_parent_provenance {
if let Some(id) = parent_turn_id {
if let Some(initiating_agent_path) = pending_items.iter().find_map(|item| {
let TurnInput::InterAgentCommunication(communication) = item else {
return None;
};
communication
.trigger_turn
.then(|| communication.author.clone())
}) {
turn_context
.turn_metadata_state
.set_initiating_agent_path(initiating_agent_path);
}
turn_context.turn_metadata_state.set_parent_turn_id(id);
}
if let Some(id) = root_turn_id {
turn_context.turn_metadata_state.set_root_turn_id(id);
}
} else if pending_items.iter().any(|item| {
// Reserved turn input already has its context; only newly arriving mail can change lineage.
let (pending_items, start_options) = self.input_queue.drain_mailbox_input_items().await;
if pending_items.iter().any(|item| {
matches!(
item,
TurnInput::InterAgentCommunication(communication) if communication.trigger_turn
)
}) && turn_context.turn_metadata_state.root_turn_id() != root_turn_id
}) && turn_context.turn_metadata_state.root_turn_id() != start_options.root_turn_id
{
turn_context.turn_metadata_state.mark_root_turn_ambiguous();
}
@@ -477,24 +452,61 @@ impl Session {
return;
}
{
let turn_state = {
let mut active_turn = self.active_turn.lock().await;
if active_turn.is_some() {
return;
}
*active_turn = Some(ActiveTurn::default());
}
let active_turn = active_turn.get_or_insert_with(ActiveTurn::default);
Arc::clone(&active_turn.turn_state)
};
let turn_context = self.new_default_turn_with_sub_id(sub_id).await;
let (input, mut start_options) =
self.input_queue.get_pending_input(&self.active_turn).await;
if !input.iter().any(
|item| matches!(item, TurnInput::InterAgentCommunication(mail) if mail.trigger_turn),
) {
// Queue-only mail wakes durable sleep without selecting a new task's settings.
start_options.cyber_access_program = self
.reference_context_item()
.await
.and_then(|context| context.cyber_access_program);
}
let turn_context = self
.new_turn_with_default_settings(
sub_id,
NewTurnContextOptions {
final_output_json_schema: start_options.final_output_json_schema,
cyber_access_program: start_options.cyber_access_program,
},
)
.await;
if let Some(id) = start_options.parent_turn_id {
if let Some(initiating_agent_path) = input.iter().find_map(|item| {
let TurnInput::InterAgentCommunication(communication) = item else {
return None;
};
communication
.trigger_turn
.then(|| communication.author.clone())
}) {
turn_context
.turn_metadata_state
.set_initiating_agent_path(initiating_agent_path);
}
turn_context.turn_metadata_state.set_parent_turn_id(id);
}
if let Some(id) = start_options.root_turn_id {
turn_context.turn_metadata_state.set_root_turn_id(id);
}
self.maybe_emit_model_warnings_for_turn(turn_context.as_ref())
.await;
self.start_task(
turn_context,
Vec::new(),
RegularTask::new(),
MailboxParentProvenance::Attribute,
)
.await;
// Task completion must still save this mail if pre-turn compaction fails.
self.input_queue
.extend_pending_input_for_turn_state(turn_state.as_ref(), input)
.await;
self.start_task(turn_context, Vec::new(), RegularTask::new())
.await;
}
pub async fn abort_all_tasks(self: &Arc<Self>, reason: TurnAbortReason) {
+5 -1
View File
@@ -118,8 +118,12 @@ pub(crate) type ThreadIdGenerator = Arc<dyn Fn() -> ThreadId + Send + Sync>;
fn capture_test_op(op: &Op) -> Option<Op> {
match op {
Op::Interrupt => Some(Op::Interrupt),
Op::InterAgentCommunication { communication } => Some(Op::InterAgentCommunication {
Op::InterAgentCommunication {
communication,
start_options,
} => Some(Op::InterAgentCommunication {
communication: communication.clone(),
start_options: start_options.clone(),
}),
Op::Shutdown => Some(Op::Shutdown),
_ => None,
+10 -2
View File
@@ -1564,7 +1564,11 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
let (prepared_turn, _) = resumed
.thread
.session
.new_turn_with_sub_id("resume-turn".to_string(), SessionSettingsUpdate::default())
.new_turn_with_sub_id(
"resume-turn".to_string(),
SessionSettingsUpdate::default(),
Default::default(),
)
.await
.expect("build resumed turn context");
let resumed_turn = prepared_turn;
@@ -1599,7 +1603,11 @@ async fn resume_and_fork_do_not_restore_thread_environments_from_rollout() {
let (prepared_turn, _) = forked
.thread
.session
.new_turn_with_sub_id("fork-turn".to_string(), SessionSettingsUpdate::default())
.new_turn_with_sub_id(
"fork-turn".to_string(),
SessionSettingsUpdate::default(),
Default::default(),
)
.await
.expect("build forked turn context");
let forked_turn = prepared_turn;
@@ -87,8 +87,12 @@ impl Handler {
.send_input(
receiver_thread_id,
input_items,
Some(turn.sub_id.clone()),
turn.turn_metadata_state.root_turn_id(),
crate::TurnStartOptions {
parent_turn_id: Some(turn.sub_id.clone()),
root_turn_id: turn.turn_metadata_state.root_turn_id(),
cyber_access_program: turn.cyber_access_program,
..Default::default()
},
)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
@@ -132,6 +132,7 @@ async fn handle_spawn_agent(
root_turn_id: turn.turn_metadata_state.root_turn_id(),
environments: Some(step_context.environments.to_selections()),
multi_agent_v2_usage_hints: None,
cyber_access_program: turn.cyber_access_program,
},
))
.await
@@ -1132,7 +1132,7 @@ async fn multi_agent_v2_spawn_returns_path_and_send_message_accepts_relative_pat
*id == child_thread_id
&& matches!(
op,
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == AgentPath::root()
&& communication.recipient.as_str() == "/root/test_process"
&& communication.other_recipients.is_empty()
@@ -1159,7 +1159,7 @@ async fn multi_agent_v2_spawn_returns_path_and_send_message_accepts_relative_pat
*id == child_thread_id
&& matches!(
op,
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == AgentPath::root()
&& communication.recipient.as_str() == "/root/test_process"
&& communication.other_recipients.is_empty()
@@ -1355,7 +1355,7 @@ async fn multi_agent_v2_send_message_accepts_root_target_from_child() {
*id == root.thread_id
&& matches!(
op,
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == child_path
&& communication.recipient == AgentPath::root()
&& communication.other_recipients.is_empty()
@@ -1869,7 +1869,7 @@ async fn multi_agent_v2_send_message_rejects_interrupt_parameter() {
assert!(!ops_for_agent.iter().any(|op| matches!(op, Op::Interrupt)));
assert!(!ops_for_agent.iter().any(|op| matches!(
op,
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == AgentPath::root()
&& communication.recipient.as_str() == "/root/worker"
&& communication.other_recipients.is_empty()
@@ -1956,7 +1956,7 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
*id == agent_id
&& matches!(
op,
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == AgentPath::root()
&& communication.recipient == worker_path
&& communication.encrypted_content.as_deref() == Some("continue")
@@ -2003,7 +2003,7 @@ async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn()
(id == root.thread_id)
.then_some(op)
.and_then(|op| match op {
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author == worker_path
&& communication.recipient == AgentPath::root()
&& communication.other_recipients.is_empty()
@@ -2149,7 +2149,7 @@ async fn multi_agent_v2_interrupted_turn_does_not_notify_parent() {
(id == root.thread_id)
.then_some(op)
.and_then(|op| match op {
Op::InterAgentCommunication { communication }
Op::InterAgentCommunication { communication, .. }
if communication.author.as_str() == "/root/worker"
&& communication.recipient == AgentPath::root()
&& communication.other_recipients.is_empty()
@@ -3019,8 +3019,7 @@ async fn multi_agent_v2_wait_agent_accepts_timeout_only_argument() {
"hello from worker".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -3524,8 +3523,7 @@ async fn multi_agent_v2_wait_agent_returns_summary_for_mailbox_activity() {
"completed".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -3601,8 +3599,7 @@ async fn multi_agent_v2_wait_agent_returns_for_already_queued_mail() {
"already queued".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -3704,8 +3701,7 @@ async fn multi_agent_v2_wait_agent_wakes_on_any_mailbox_notification() {
"from worker b".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -3796,8 +3792,7 @@ async fn multi_agent_v2_wait_agent_does_not_return_completed_content() {
"sensitive child output".to_string(),
/*trigger_turn*/ false,
),
/*parent_turn_id*/ None,
/*root_turn_id*/ None,
Default::default(),
)
.await;
@@ -117,8 +117,12 @@ pub(super) async fn handle_message_string_tool(
receiver_thread_id,
communication,
context,
parent_turn_id,
turn.turn_metadata_state.root_turn_id(),
crate::TurnStartOptions {
parent_turn_id,
root_turn_id: turn.turn_metadata_state.root_turn_id(),
cyber_access_program: turn.cyber_access_program,
..Default::default()
},
)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
@@ -223,6 +223,7 @@ async fn handle_spawn_agent(
root_turn_id: turn.turn_metadata_state.root_turn_id(),
environments: Some(step_context.environments.to_selections()),
multi_agent_v2_usage_hints,
cyber_access_program: turn.cyber_access_program,
},
),
)
+1
View File
@@ -187,6 +187,7 @@ async fn root_turn_suspension_preserves_unfinished_turn_history() {
turn_id: turn_id.clone(),
thread_settings: Default::default(),
trace: None,
cyber_access_program: None,
})
.await
.expect("recover the unfinished turn"),
@@ -1586,6 +1586,7 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
.to_string(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
})
.await?;
codex
@@ -1597,6 +1598,7 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
"Message Type: FINAL_ANSWER\nTask name: /root\nSender: /root/child\nPayload:\nchild completion".to_string(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
})
.await?;
let delegated_task_ciphertext = format!("delegated compact task{}", "x".repeat(40_000));
@@ -1609,6 +1611,7 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
delegated_task_ciphertext.clone(),
/*trigger_turn*/ true,
),
start_options: Default::default(),
})
.await?;
wait_for_turn_complete(&codex).await;
@@ -1624,6 +1627,7 @@ async fn remote_compact_v2_reuses_compaction_trigger_for_followups() -> Result<(
descendant_followup_ciphertext.to_string(),
/*trigger_turn*/ true,
),
start_options: Default::default(),
})
.await?;
wait_for_turn_complete(&codex).await;
@@ -0,0 +1,522 @@
use anyhow::Result;
use codex_core::RecoverTurnRequest;
use codex_core::StartIfIdleSubmission;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_core::TurnStartOptions;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::turn_input::CyberAccessProgram;
use codex_protocol::user_input::UserInput;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutRecorder;
use core_test_support::responses;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use core_test_support::test_codex::TestCodex;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use serde_json::json;
use tokio::sync::oneshot;
use wiremock::Mock;
use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn recover_turn_restores_cyber_access_program_without_making_it_sticky() -> Result<()> {
// Keep sampling pending until interruption so restart must reload the unfinished turn.
let (release_response, response_gate) = oneshot::channel();
let (initial_server, _completions) =
start_streaming_sse_server(vec![vec![StreamingSseChunk {
gate: Some(response_gate),
body: responses::sse(vec![responses::ev_completed("resp-initial")]),
}]])
.await;
let mut builder = test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing());
let initial = builder.build_with_streaming_server(&initial_server).await?;
let TurnInputSubmission::Started { turn_id } = initial
.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "recover this turn".to_owned(),
text_elements: Vec::new(),
}])
.on_start(TurnStartOptions {
cyber_access_program: Some(CyberAccessProgram::DaybreakBlue),
..Default::default()
}),
)
.await?
else {
panic!("expected a new turn");
};
initial_server.wait_for_request_count(/*count*/ 1).await;
initial.codex.submit(Op::Interrupt).await?;
wait_for_event(&initial.codex, |event| {
matches!(event, EventMsg::TurnAborted(_))
})
.await;
drop(release_response);
initial_server.shutdown().await;
initial.codex.flush_rollout().await?;
let server = responses::start_mock_server().await;
let test = builder.restart(&server, &initial).await?;
let (rollout, _, _) = RolloutRecorder::load_rollout_items(
&test
.codex
.rollout_path()
.expect("recovered turn rollout path"),
)
.await?;
let persisted_context = rollout
.iter()
.rev()
.find_map(|item| match item {
RolloutItem::TurnContext(context)
if context.turn_id.as_deref() == Some(turn_id.as_str()) =>
{
Some(context)
}
_ => None,
})
.expect("persisted context for the interrupted turn");
assert_eq!(
persisted_context.cyber_access_program,
Some(CyberAccessProgram::DaybreakBlue)
);
let response_mock =
responses::mount_sse_once(&server, responses::sse_completed("resp-1")).await;
let submission = test
.codex
.recover_turn_if_idle(RecoverTurnRequest {
turn_id: turn_id.clone(),
thread_settings: Default::default(),
trace: None,
cyber_access_program: persisted_context.cyber_access_program,
})
.await?;
assert_eq!(
submission,
StartIfIdleSubmission::Started {
turn_id: turn_id.clone(),
}
);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(
response_mock.single_request().body_json()["access_programs"],
json!({"cyber": "daybreak_blue"})
);
assert_eq!(
response_mock
.single_request()
.message_input_texts("user")
.into_iter()
.filter(|text| text == "recover this turn")
.count(),
1
);
let next_response = responses::mount_sse_once(
&server,
responses::sse(vec![responses::ev_completed("resp-next")]),
)
.await;
let TurnInputSubmission::Started {
turn_id: next_turn_id,
} = test
.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "start a new turn".to_owned(),
text_elements: Vec::new(),
}]))
.await?
else {
panic!("expected a new turn");
};
assert_ne!(next_turn_id, turn_id);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(
next_response
.single_request()
.body_json()
.get("access_programs"),
None
);
test.codex.shutdown_and_wait().await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cyber_access_program_omits_api_key_and_spoofed_custom_provider() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
for (auth, provider_id) in [
(CodexAuth::from_api_key("test-key"), "openai"),
(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "custom"),
] {
let server = responses::start_mock_server().await;
let request = responses::mount_sse_once(
&server,
responses::sse(vec![responses::ev_completed("resp-1")]),
)
.await;
let test = test_codex()
.with_auth(auth)
.with_config(move |config| {
// Keep the display name "OpenAI": provider identity must not use it.
config.model_provider_id = provider_id.to_owned();
})
.build_with_auto_env(&server)
.await?;
submit(&test, Some(CyberAccessProgram::DaybreakRed)).await?;
assert_eq!(
request.single_request().body_json().get("access_programs"),
None
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cyber_access_program_survives_mid_turn_remote_compaction() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let requests = responses::mount_sse_sequence(
&server,
vec![
responses::sse(vec![
// A dummy tool response forces a follow-up without shell approvals.
responses::ev_function_call("tool-1", "test_tool", "{}"),
responses::ev_completed_with_tokens("resp-1", /*total_tokens*/ 100000000),
]),
responses::sse(vec![responses::ev_completed("resp-2")]),
],
)
.await;
let compact =
responses::mount_compact_user_history_with_summary_once(&server, "compacted history").await;
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(|config| {
let _ = config.features.disable(Feature::RemoteCompactionV2);
})
.build_with_auto_env(&server)
.await?;
submit(&test, Some(CyberAccessProgram::DaybreakBlue)).await?;
assert_eq!(
compact.single_request().body_json()["access_programs"],
json!({"cyber": "daybreak_blue"})
);
assert_eq!(
requests
.requests()
.iter()
.map(|request| request.body_json()["access_programs"].clone())
.collect::<Vec<_>>(),
vec![json!({"cyber": "daybreak_blue"}); 2]
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cyber_access_program_survives_mid_turn_remote_compaction_v2() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let requests = responses::mount_sse_sequence(
&server,
vec![
responses::sse(vec![
responses::ev_function_call("call-before-compact", "test_tool", "{}"),
responses::ev_completed_with_tokens("resp-1", /*total_tokens*/ 500),
]),
responses::sse(vec![
serde_json::json!({
"type": "response.output_item.done",
"item": {
"type": "compaction",
"encrypted_content": "V2_COMPACT_SUMMARY",
}
}),
responses::ev_completed("resp-compact"),
]),
responses::sse(vec![responses::ev_completed_with_tokens(
"resp-2", /*total_tokens*/ 80,
)]),
],
)
.await;
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(|config| {
let _ = config.features.enable(Feature::RemoteCompactionV2);
config.model_auto_compact_token_limit = Some(200);
})
.build_with_auto_env(&server)
.await?;
submit(&test, Some(CyberAccessProgram::DaybreakBlue)).await?;
let requests = requests.requests();
assert_eq!(
requests
.iter()
.map(|request| request.body_json()["access_programs"].clone())
.collect::<Vec<_>>(),
vec![json!({"cyber": "daybreak_blue"}); 3]
);
assert_eq!(requests[1].inputs_of_type("compaction_trigger").len(), 1);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cyber_access_program_is_inherited_by_child_turns() -> Result<()> {
// Final answers defer late child completion mail to the next parent turn.
let final_response = |id: &str| {
responses::sse(vec![
responses::ev_response_created(id),
responses::ev_assistant_message(&format!("message-{id}"), "Done."),
responses::ev_completed(id),
])
};
for (namespace, fork_turns) in [
("collaboration", "none"),
("collaboration", "all"),
("multi_agent_v1", "none"),
] {
let is_v2 = namespace == "collaboration";
let spawn_arguments = if is_v2 {
json!({"message": "inspect the repository", "task_name": "worker", "fork_turns": fork_turns})
} else {
json!({"message": "inspect the repository"})
};
let server = responses::start_mock_server().await;
// V1 completion notifications can require another parent response.
Mock::given(method("POST"))
.and(path("/v1/responses"))
.and(|request: &wiremock::Request| !request.headers.contains_key("x-openai-subagent"))
.respond_with(responses::sse_response(final_response("resp-parent-done")))
.with_priority(/*p*/ 10)
.mount(&server)
.await;
responses::mount_sse_once(
&server,
responses::sse(vec![
responses::ev_function_call_with_namespace(
"spawn-worker",
namespace,
"spawn_agent",
&spawn_arguments.to_string(),
),
responses::ev_completed("resp-spawn"),
]),
)
.await;
let initial_child_request = responses::mount_sse_once_match(
&server,
header("x-openai-subagent", "collab_spawn"),
final_response("resp-child-initial"),
)
.await;
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_model(if is_v2 { "gpt-5.6-sol" } else { "gpt-5.1" })
.with_config(move |config| {
config
.features
.enable(Feature::Collab)
.expect("enable multi-agent tools");
if is_v2 {
config
.features
.enable(Feature::MultiAgentV2)
.expect("enable v2 tools");
}
})
.build_with_auto_env(&server)
.await?;
let mut created_threads = test.thread_manager.subscribe_thread_created();
submit(&test, Some(CyberAccessProgram::DaybreakRed)).await?;
let child_id = created_threads.recv().await?;
let child = test.thread_manager.get_thread(child_id).await?;
wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let child_programs = |requests: &responses::ResponseMock| {
requests
.requests()
.iter()
.filter(|request| {
request.body_json()["client_metadata"]["thread_id"] == json!(child_id)
})
.map(|request| request.body_json()["access_programs"].clone())
.collect::<Vec<_>>()
};
assert_eq!(
child_programs(&initial_child_request),
vec![json!({"cyber": "daybreak_red"})],
"namespace={namespace}, fork_turns={fork_turns}"
);
for (program, expected, reload) in [
(
Some(CyberAccessProgram::Standard),
json!({"cyber": "standard"}),
false,
),
(
Some(CyberAccessProgram::DaybreakBlue),
json!({"cyber": "daybreak_blue"}),
true,
),
(None, json!(null), false),
(
Some(CyberAccessProgram::Standard),
json!({"cyber": "standard"}),
true,
),
] {
if reload {
let child = test.thread_manager.get_thread(child_id).await?;
child.shutdown_and_wait().await?;
test.thread_manager.remove_thread(&child_id).await;
}
let mut reply_sequence = Vec::new();
if reload && !is_v2 {
reply_sequence.push(responses::sse(vec![
responses::ev_function_call_with_namespace(
"resume-worker",
namespace,
"resume_agent",
&json!({"id": child_id}).to_string(),
),
responses::ev_completed("resp-resume"),
]));
}
reply_sequence.push(responses::sse(vec![
responses::ev_function_call_with_namespace(
"followup-worker",
namespace,
if is_v2 { "followup_task" } else { "send_input" },
&json!({
"target": if is_v2 { "worker".to_string() } else { child_id.to_string() },
"message": "inspect the tests too",
})
.to_string(),
),
responses::ev_completed("resp-followup"),
]));
responses::mount_sse_sequence(&server, reply_sequence).await;
let followup_child_request = responses::mount_sse_once_match(
&server,
header("x-openai-subagent", "collab_spawn"),
final_response("resp-child-next"),
)
.await;
submit(&test, program).await?;
let child = test.thread_manager.get_thread(child_id).await?;
wait_for_event(&child, |event| matches!(event, EventMsg::TurnComplete(_))).await;
assert_eq!(
child_programs(&followup_child_request),
vec![expected],
"namespace={namespace}, fork_turns={fork_turns}, reload={reload}"
);
}
}
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cyber_access_program_changes_on_one_websocket_with_response_reuse() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let response_ids = [
"prewarm", "first", "second", "blue", "red", "off", "omitted",
];
let server = responses::start_websocket_server(vec![
response_ids
.iter()
.map(|id| vec![responses::ev_completed(id)])
.collect(),
])
.await;
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.build_with_websocket_server(&server)
.await?;
for program in [
Some(CyberAccessProgram::Standard),
Some(CyberAccessProgram::Standard),
Some(CyberAccessProgram::DaybreakBlue),
Some(CyberAccessProgram::DaybreakRed),
Some(CyberAccessProgram::Standard),
None,
] {
submit(&test, program).await?;
}
let requests = server.single_connection();
assert_eq!(
requests
.iter()
.map(|request| request.body_json().get("access_programs").cloned())
.collect::<Vec<_>>(),
[
None,
Some(json!({"cyber": "standard"})),
Some(json!({"cyber": "standard"})),
Some(json!({"cyber": "daybreak_blue"})),
Some(json!({"cyber": "daybreak_red"})),
Some(json!({"cyber": "standard"})),
None,
]
);
// Each choice is sent alongside the previous response id, not a full replay.
assert_eq!(
requests
.iter()
.skip(1)
.map(|request| request.body_json().get("previous_response_id").cloned())
.collect::<Vec<_>>(),
response_ids[..response_ids.len() - 1]
.iter()
.map(|id| Some(json!(id)))
.collect::<Vec<_>>()
);
test.codex.shutdown_and_wait().await?;
server.shutdown().await;
Ok(())
}
async fn submit(test: &TestCodex, program: Option<CyberAccessProgram>) -> Result<()> {
test.codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "hello".to_owned(),
text_elements: Vec::new(),
}])
.on_start(TurnStartOptions {
cyber_access_program: program,
..Default::default()
}),
)
.await?;
let event = wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_) | EventMsg::Error(_))
})
.await;
if let EventMsg::Error(error) = event {
anyhow::bail!("cyber program {program:?}: {error:?}");
}
Ok(())
}
+1
View File
@@ -60,6 +60,7 @@ mod compact_remote_parity;
mod compact_resume_fork;
mod context_annotations;
mod current_time_reminder;
mod cyber_access_program;
mod cyber_exec_policy;
mod deprecation_notice;
mod exec;
+38 -5
View File
@@ -6,6 +6,7 @@ use codex_core::StartIfIdleSubmission;
use codex_core::TurnInput;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_core::TurnStartOptions;
use codex_core::config::CurrentTimeReminderConfig;
use codex_extension_items::ExtensionItem;
use codex_extension_items::sleep::SleepItem;
@@ -28,6 +29,7 @@ use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InterAgentCommunication;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::turn_input::CyberAccessProgram;
use codex_protocol::user_input::UserInput;
use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions;
@@ -343,6 +345,7 @@ async fn enqueue_queue_only_agent_mail(codex: &CodexThread, text: &str) {
text.to_string(),
/*trigger_turn*/ false,
),
start_options: Default::default(),
})
.await
.expect("submit queue-only agent mail");
@@ -457,26 +460,55 @@ impl codex_extension_api::ThreadLifecycleContributor<codex_core::config::Config>
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_only_agent_mail_wakes_sleeping_root_and_persists_message() {
async fn queue_only_agent_mail_wakes_sleeping_root_with_previous_turn_context() {
const CHILD_MESSAGE: &str = "worker completed";
let (server, _completions) =
start_streaming_sse_server(vec![response_completed_chunks("resp-1")]).await;
let server = responses::start_mock_server().await;
let requests = responses::mount_sse_sequence(
&server,
vec![
responses::sse_completed("resp-initial"),
responses::sse_completed("resp-wake"),
],
)
.await;
let mut extensions =
codex_extension_api::ExtensionRegistryBuilder::<codex_core::config::Config>::new();
extensions.thread_lifecycle_contributor(Arc::new(SleepingRootExtension));
let codex = test_codex()
.with_model("gpt-5.4")
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_extensions(Arc::new(extensions.build()))
.build_with_streaming_server(&server)
.build_with_auto_env(&server)
.await
.expect("build Codex test session")
.codex;
codex
.start_or_steer_turn(
TurnInputRequest::user_input(vec![UserInput::Text {
text: "wait for the worker".to_string(),
text_elements: Vec::new(),
}])
.on_start(TurnStartOptions {
cyber_access_program: Some(CyberAccessProgram::Standard),
..Default::default()
}),
)
.await
.expect("start initial turn");
wait_for_turn_complete(&codex).await;
enqueue_queue_only_agent_mail(&codex, CHILD_MESSAGE).await;
wait_for_turn_complete(&codex).await;
assert_eq!(server.requests().await.len(), 1);
assert_eq!(
requests
.requests()
.iter()
.map(|request| request.body_json()["access_programs"].clone())
.collect::<Vec<_>>(),
vec![json!({"cyber": "standard"}); 2],
);
let history = codex
.load_history(/*include_archived*/ true)
.await
@@ -1199,6 +1231,7 @@ async fn terminal_compaction_error_does_not_retry_pending_input(
PENDING_MESSAGE.to_string(),
/*trigger_turn*/ true,
),
start_options: Default::default(),
})
.await?;
codex.submit(Op::RealtimeConversationListVoices).await?;
@@ -48,6 +48,7 @@ fn resume_history(
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: config.model_reasoning_effort.clone(),
summary: config
.model_reasoning_summary
@@ -128,11 +128,8 @@ async fn start_turn_if_idle_keeps_automatic_plan_rejections_atomic(
#[tokio::test]
async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() {
let server = responses::start_mock_server().await;
let response_mock = responses::mount_sse_once(
&server,
responses::sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
)
.await;
let response_mock =
responses::mount_sse_once(&server, responses::sse_completed("resp-1")).await;
let test = test_codex()
.build_with_auto_env(&server)
.await
@@ -155,6 +152,7 @@ async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() {
..Default::default()
},
trace: None,
cyber_access_program: None,
})
.await
.expect("recovered turn should start");
@@ -164,6 +162,10 @@ async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() {
turn_id: turn_id.to_string(),
}
);
assert_eq!(
test.codex.config_snapshot().await.collaboration_mode.mode,
ModeKind::Plan
);
let started = wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnStarted(_))
+1
View File
@@ -994,6 +994,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
output_schema,
collaboration_mode: None,
multi_agent_mode: None,
cyber_access_program: None,
},
},
"turn/start",
@@ -489,6 +489,7 @@ impl LunaSampler {
prompt_cache_key: Some(format!("guardian-v2:{}", self.config.thread_id)),
text: None,
client_metadata: None,
access_programs: None,
};
let (supersede, mut superseded) = oneshot::channel();
let scored = Arc::new(AtomicBool::new(false));
+7
View File
@@ -56,10 +56,12 @@ use crate::plan_tool::UpdatePlanArgs;
use crate::request_permissions::RequestPermissionsEvent;
use crate::request_permissions::RequestPermissionsResponse;
use crate::request_user_input::RequestUserInputResponse;
use crate::turn_input::CyberAccessProgram;
use crate::turn_input::SuspendTurnOutcome;
use crate::turn_input::TurnInputMode;
use crate::turn_input::TurnInputRequest;
use crate::turn_input::TurnInputSubmission;
use crate::turn_input::TurnStartOptions;
use codex_extension_items::image_generation::ImageGenerationFailure;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path_uri::PathUri;
@@ -604,6 +606,7 @@ pub enum Op {
/// Resume an interrupted regular turn.
RecoverTurn {
thread_settings: ThreadSettingsOverrides,
start_options: TurnStartOptions,
reply: oneshot::Sender<CodexResult<TurnInputSubmission>>,
},
@@ -633,6 +636,7 @@ pub enum Op {
/// while still using the normal thread submission lifecycle.
InterAgentCommunication {
communication: InterAgentCommunication,
start_options: TurnStartOptions,
},
/// Approve a command execution
@@ -3112,6 +3116,8 @@ pub struct TurnContextItem {
pub multi_agent_mode: Option<MultiAgentMode>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub realtime_active: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cyber_access_program: Option<CyberAccessProgram>,
#[serde(skip_serializing_if = "Option::is_none")]
pub effort: Option<ReasoningEffortConfig>,
// Compatibility-only field written with a default value so older Codex
@@ -5862,6 +5868,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: ReasoningSummaryConfig::Auto,
};
+18
View File
@@ -7,11 +7,13 @@ use crate::protocol::NonSteerableTurnKind;
use crate::protocol::ThreadSettingsOverrides;
use crate::protocol::W3cTraceContext;
use crate::user_input::UserInput;
use schemars::JsonSchema;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
use std::collections::BTreeMap;
use std::collections::HashMap;
use ts_rs::TS;
/// Result of stopping an unfinished root turn so another worker can recover it.
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -59,6 +61,8 @@ pub struct RecoverTurnRequest {
pub turn_id: String,
pub thread_settings: ThreadSettingsOverrides,
pub trace: Option<W3cTraceContext>,
/// Program recorded in the interrupted turn's persisted context.
pub cyber_access_program: Option<CyberAccessProgram>,
}
impl TurnInputRequest {
@@ -135,6 +139,17 @@ pub enum TurnInputMode {
Steer { expected_turn_id: String },
}
/// Requested cyber treatment for a ChatGPT-authenticated Codex turn.
/// Authorization and model-tier restrictions remain server-owned.
#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub enum CyberAccessProgram {
Standard,
DaybreakBlue,
DaybreakRed,
}
/// Options for the new-turn branch of a submission.
///
/// Core only records these options when input starts a turn. When input steers,
@@ -155,6 +170,9 @@ pub struct TurnStartOptions {
pub parent_turn_id: Option<String>,
/// Causal root turn lineage recorded if this request starts a new turn.
pub root_turn_id: Option<String>,
/// Explicit cyber treatment for this turn. Omission preserves the backend's
/// automatic behavior.
pub cyber_access_program: Option<CyberAccessProgram>,
}
/// What Core did with input submitted through `start_or_steer_turn`.
+1
View File
@@ -1718,6 +1718,7 @@ async fn resume_candidate_matches_cwd_reads_latest_turn_context() -> std::io::Re
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}),
+4
View File
@@ -446,6 +446,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}),
@@ -493,6 +494,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}),
@@ -536,6 +538,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: Some(ReasoningEffort::High),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}),
@@ -576,6 +579,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: Some(ReasoningEffort::High),
summary: codex_protocol::config_types::ReasoningSummary::Auto,
}),
+1
View File
@@ -951,6 +951,7 @@ mod tests {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: ReasoningSummary::Auto,
})
@@ -613,6 +613,7 @@ fn turn_context(root: &Path, turn_id: &str) -> RolloutItem {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: ReasoningSummary::Auto,
})
@@ -1452,6 +1452,7 @@ async fn migration_compacts_subagent_prefix_and_does_not_project_it() {
multi_agent_version: None,
multi_agent_mode: None,
realtime_active: None,
cyber_access_program: None,
effort: None,
summary: ReasoningSummary::Auto,
}),
+1
View File
@@ -1222,6 +1222,7 @@ impl AppServerSession {
output_schema,
collaboration_mode,
multi_agent_mode: None,
cyber_access_program: None,
},
})
.await