diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index d9043f1de3..7ddd776fb8 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -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": [ { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 2bd1ece011..84ab0860d6 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 13da6880b2..e104117e33 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json index 96327a5cf6..1dfb7ed708 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartParams.json @@ -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", diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 33d6d70de8..e1dfe6eaa4 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 0df4766976..11383fee3d 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/CyberAccessProgram.ts b/codex-rs/app-server-protocol/schema/typescript/v2/CyberAccessProgram.ts new file mode 100644 index 0000000000..ccd4118a83 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/CyberAccessProgram.ts @@ -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"; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 9c6f665f0e..dbfa4aa4c9 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -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"; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 491b882941..4e8b55fca7 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -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!({ diff --git a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs index 6778748814..9c3b588d23 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/turn.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/turn.rs @@ -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 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, + + /// 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, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index d07a6ca62a..741f52659f 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -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. diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index a452929e0a..2ffd5e76eb 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -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), diff --git a/codex-rs/app-server/src/request_processors/feedback_processor.rs b/codex-rs/app-server/src/request_processors/feedback_processor.rs index 8de82abdb0..e6b8eef3a2 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/codex-rs/app-server/src/request_processors/feedback_processor.rs @@ -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, }), diff --git a/codex-rs/app-server/src/request_processors/persisted_resume_settings_tests.rs b/codex-rs/app-server/src/request_processors/persisted_resume_settings_tests.rs index ed96867eb1..98ab3f59a2 100644 --- a/codex-rs/app-server/src/request_processors/persisted_resume_settings_tests.rs +++ b/codex-rs/app-server/src/request_processors/persisted_resume_settings_tests.rs @@ -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, }) diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index 6ece29cc0b..948e2b3f68 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/codex-rs/app-server/src/request_processors/turn_processor.rs @@ -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) diff --git a/codex-rs/app-server/tests/suite/v2/cyber_access_program.rs b/codex-rs/app-server/tests/suite/v2/cyber_access_program.rs new file mode 100644 index 0000000000..a894b1ad2e --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/cyber_access_program.rs @@ -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::>(), + [ + 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(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 07426518fb..d654961905 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -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; diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index 33f22aa063..4d705dc70b 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -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?; diff --git a/codex-rs/codex-api/src/common.rs b/codex-rs/codex-api/src/common.rs index ef28c6ad74..937e6d83d0 100644 --- a/codex-rs/codex-api/src/common.rs +++ b/codex-rs/codex-api/src/common.rs @@ -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 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, + #[serde(skip_serializing_if = "Option::is_none")] + pub access_programs: Option, } /// Canonical input payload for the memory summarize endpoint. @@ -272,6 +293,8 @@ pub struct ResponsesApiRequest { pub text: Option, #[serde(skip_serializing_if = "Option::is_none")] pub client_metadata: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub access_programs: Option, } 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, #[serde(skip_serializing_if = "Option::is_none")] pub client_metadata: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub access_programs: Option, } pub fn response_create_client_metadata( diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 96c2e04a3e..a85c993d03 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -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(), diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 83af9f0dc0..a31a0e2fba 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -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; diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index 4a5f70473e..1ad1161848 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -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(); diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index f32efca616..afba0247a0 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -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, pub(crate) environments: Option>, pub(crate) multi_agent_v2_usage_hints: Option, + pub(crate) cyber_access_program: Option, } #[derive(Clone, Debug)] @@ -183,19 +185,12 @@ impl AgentControl { &self, agent_id: ThreadId, input: Vec, - parent_turn_id: Option, - root_turn_id: Option, + start_options: TurnStartOptions, ) -> CodexResult { 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, - root_turn_id: Option, + start_options: TurnStartOptions, ) -> CodexResult { 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, communication: InterAgentCommunication, context: AgentCommunicationContext, - parent_turn_id: Option, - root_turn_id: Option, + start_options: TurnStartOptions, ) -> CodexResult { 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, communication: InterAgentCommunication, context: AgentCommunicationContext, - parent_turn_id: Option, - root_turn_id: Option, + start_options: TurnStartOptions, ) -> CodexResult { 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; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 3a9e2533bb..9e06fd4e74 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -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?; } diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index b4abc81177..1ee3b009a1 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -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(), }, ); diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index ed0f10d3d4..6fd3f1ff57 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -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(), diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs index 1e98b25d11..f2d6be2a23 100644 --- a/codex-rs/core/src/client_common.rs +++ b/codex-rs/core/src/client_common.rs @@ -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, } 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, } } } diff --git a/codex-rs/core/src/client_common_tests.rs b/codex-rs/core/src/client_common_tests.rs index 6f0dac2696..1c8625af56 100644 --- a/codex-rs/core/src/client_common_tests.rs +++ b/codex-rs/core/src/client_common_tests.rs @@ -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"); diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 75866e17c3..41e1f9b69b 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -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 } => { diff --git a/codex-rs/core/src/compact_remote_request.rs b/codex-rs/core/src/compact_remote_request.rs index 9ca0b88392..2f4b489f96 100644 --- a/codex-rs/core/src/compact_remote_request.rs +++ b/codex-rs/core/src/compact_remote_request.rs @@ -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( diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs index 029b821901..9569649f8b 100644 --- a/codex-rs/core/src/compact_remote_v2_attempt.rs +++ b/codex-rs/core/src/compact_remote_v2_attempt.rs @@ -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 diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index a81085e2e4..190f617bda 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -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, } diff --git a/codex-rs/core/src/cyber_access_program.rs b/codex-rs/core/src/cyber_access_program.rs new file mode 100644 index 0000000000..4a508715fd --- /dev/null +++ b/codex-rs/core/src/cyber_access_program.rs @@ -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, +) -> Option { + program + .filter(|_| auth.is_some_and(CodexAuth::is_chatgpt_auth)) + .map(AccessPrograms::from) +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index f7c636a5b8..8463ae9e16 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -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; diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 98d02580a4..615ea7b8fb 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -82,16 +82,11 @@ pub async fn inter_agent_communication( sess: &Arc, sub_id: String, communication: InterAgentCommunication, - parent_turn_id: Option, - root_turn_id: Option, + 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, 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) { } pub async fn compact(sess: &Arc, 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, 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 { diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 1238cab10d..4fa79b9e2a 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -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, - root_turn_id: Option, + 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, - root_turn_id: Option, + 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, Option, Option) { + pub(crate) async fn drain_mailbox_input_items(&self) -> (Vec, TurnStartOptions) { let pending_mails = self .mailbox_pending_mails .lock() .await .drain(..) .collect::>(); - 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>, - ) -> (Vec, Option, Option) { + ) -> (Vec, 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); } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index bf07f114d9..6f00a1cb63 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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, turn_id: String, ) -> CodexResult { @@ -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 { diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index b634e7510e..0cff774179 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -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. diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 5447999e15..abc875c3ba 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -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, }; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 53c35644e0..c4a4078809 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -537,7 +537,6 @@ pub(crate) struct SessionSettingsUpdate { pub(crate) active_permission_profile: Option, pub(crate) windows_sandbox_level: Option, pub(crate) service_tier_for_turn: Option, - pub(crate) final_output_json_schema: Option>, pub(crate) app_server_client_name: Option, pub(crate) app_server_client_version: Option, } diff --git a/codex-rs/core/src/session/step_activation_tests.rs b/codex-rs/core/src/session/step_activation_tests.rs index 3fc026b3ca..0e0bcd8c52 100644 --- a/codex-rs/core/src/session/step_activation_tests.rs +++ b/codex-rs/core/src/session/step_activation_tests.rs @@ -223,7 +223,7 @@ async fn activation_fixture(models: Vec) -> 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), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index b988a09c89..211cd48c90 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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 { diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 1371f55915..eb9d4e9881 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -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; diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 05b67cbe86..27304344be 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -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, } } diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 9cffab37bf..bab8d92c51 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -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, + pub(crate) cyber_access_program: Option, +} + /// 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>>, 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, } 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, 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, 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>, + options: NewTurnContextOptions, ) -> Arc { 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>, + options: NewTurnContextOptions, multi_agent_runtime: TurnMultiAgentRuntime, git_enrichment_policy: GitEnrichmentPolicy, ) -> Arc { @@ -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 { - 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 { - 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 { + 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, diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 7eb922ba42..bbc37657d6 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -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, thread_settings: ThreadSettingsOverrides, + start_options: TurnStartOptions, submission_id: String, ) -> CodexResult { 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, diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index da374e176d..fc31d04407 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/codex-rs/core/src/session/turn_input_tests.rs @@ -175,11 +175,12 @@ async fn start_only_rejects_active_turn_without_injecting() { submission ); assert_eq!( - (Vec::::new(), None, None), + Vec::::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::::new(), None, None) + .await + .0, + Vec::::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::::new(), None, None), + Vec::::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::::new(), None, None) + .await + .0, + Vec::::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::::new(), None, None) + .await + .0, + Vec::::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 diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 87ee5a12f1..24bbae05a8 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -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>; -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( @@ -287,7 +282,6 @@ impl Session { turn_context: Arc, input: Vec, task: T, - mailbox_parent_provenance: MailboxParentProvenance, ) { let task: Arc = 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, reason: TurnAbortReason) { diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 20f134e31a..333e59d2d0 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -118,8 +118,12 @@ pub(crate) type ThreadIdGenerator = Arc ThreadId + Send + Sync>; fn capture_test_op(op: &Op) -> Option { 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, diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 87878eab23..730ce25dd3 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs index b4ea2e2c5e..e35233d131 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/send_input.rs @@ -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)); diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index d7a69f5a89..0a5325c195 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -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 diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index d364236b07..e4a1174281 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -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; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index 8f514be495..9f0ac3cb13 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -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)); diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index f783f23405..61a6240f9a 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -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, }, ), ) diff --git a/codex-rs/core/tests/suite/abort_tasks.rs b/codex-rs/core/tests/suite/abort_tasks.rs index 8d8bf98ca4..4b995595d5 100644 --- a/codex-rs/core/tests/suite/abort_tasks.rs +++ b/codex-rs/core/tests/suite/abort_tasks.rs @@ -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"), diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index 2a5c3f3aed..f0f921cf06 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -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; diff --git a/codex-rs/core/tests/suite/cyber_access_program.rs b/codex-rs/core/tests/suite/cyber_access_program.rs new file mode 100644 index 0000000000..8c12c95e9b --- /dev/null +++ b/codex-rs/core/tests/suite/cyber_access_program.rs @@ -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![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![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::>() + }; + 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::>(), + [ + 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::>(), + response_ids[..response_ids.len() - 1] + .iter() + .map(|id| Some(json!(id))) + .collect::>() + ); + test.codex.shutdown_and_wait().await?; + server.shutdown().await; + Ok(()) +} + +async fn submit(test: &TestCodex, program: Option) -> 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(()) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 584890e37a..8437a41d5d 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -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; diff --git a/codex-rs/core/tests/suite/pending_input.rs b/codex-rs/core/tests/suite/pending_input.rs index 135f245231..ef4b6b98c9 100644 --- a/codex-rs/core/tests/suite/pending_input.rs +++ b/codex-rs/core/tests/suite/pending_input.rs @@ -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 } #[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::::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![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?; diff --git a/codex-rs/core/tests/suite/resume_warning.rs b/codex-rs/core/tests/suite/resume_warning.rs index a025192822..1be835b6fe 100644 --- a/codex-rs/core/tests/suite/resume_warning.rs +++ b/codex-rs/core/tests/suite/resume_warning.rs @@ -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 diff --git a/codex-rs/core/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index 475b6a0584..837d7b2add 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/codex-rs/core/tests/suite/turn_input_submission.rs @@ -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(_)) diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 3d049ab907..5f2984e4c8 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -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", diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs index cd13f43183..7db9ca5e05 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs @@ -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)); diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 29f54c42f9..f76267a233 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -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>, }, @@ -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, #[serde(default, skip_serializing_if = "Option::is_none")] pub realtime_active: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cyber_access_program: Option, #[serde(skip_serializing_if = "Option::is_none")] pub effort: Option, // 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, }; diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index 209de883ca..4819d0fb9b 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -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, + /// Program recorded in the interrupted turn's persisted context. + pub cyber_access_program: Option, } 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, /// Causal root turn lineage recorded if this request starts a new turn. pub root_turn_id: Option, + /// Explicit cyber treatment for this turn. Omission preserves the backend's + /// automatic behavior. + pub cyber_access_program: Option, } /// What Core did with input submitted through `start_or_steer_turn`. diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 0c4e663aa3..22cbcf5dee 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -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, }), diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 9b0c7f3375..b5cbd93de6 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -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, }), diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index aa88ea0280..de9f72e618 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -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, }) diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 7e83d8a74f..e86f040704 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -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, }) diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index 9e7a380ac1..8de938b6af 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -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, }), diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 186b5393af..ca0f99aa2a 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -1222,6 +1222,7 @@ impl AppServerSession { output_schema, collaboration_mode, multi_agent_mode: None, + cyber_access_program: None, }, }) .await