Enforce host-required cyber access programs per thread (#59754)

## What changed

- Expose `RequiredCyberAccessPrograms` so hosts can require an explicit allowed program on a thread. Snapshot the requirement before extension callbacks; hosts must reapply it on cold resume.
- Validate resolved `access_programs` before HTTP and WebSocket model requests, rejecting omitted or disallowed programs. Explicit `Standard` remains distinct from omission. Reject memory summarization and realtime requests when a requirement is present.
- Keep required programs scoped to the parent thread by preventing Guardian reviewers from inheriting its program, including Guardian v2 reviews.

## Testing

Add HTTP and WebSocket coverage for allowed programs, omission, disallowed programs, unsupported providers, and connection reuse. Extend Guardian tests to verify parent-only program scope and that removing the public attachment does not erase the startup snapshot.

GitOrigin-RevId: 9306a9af0e03e7a1728dfeb610cd7ec7517367fd

codex-rs/core/src/client.rs

@@ -310,6 +310,7 @@ pub struct ModelClient {
     executed_tool_calls: Option<ExecutedToolCalls>,
     // Resolved once when the session is created, like other session feature flags.
     api_key_cyber_access_programs: cyber_access_program::ApiKeyCyberAccessPrograms,
+    required_cyber_access_programs: Option<Arc<cyber_access_program::RequiredCyberAccessPrograms>>,
 }
 
 /// A turn-scoped streaming session created from a [`ModelClient`].
@@ -656,6 +657,7 @@ impl ModelClient {
             executed_tool_calls: None,
             api_key_cyber_access_programs:
                 cyber_access_program::ApiKeyCyberAccessPrograms::UnsupportedProvider,
+            required_cyber_access_programs: None,
         }
     }
 
@@ -682,15 +684,36 @@ impl ModelClient {
         event_sender: Sender<ProtocolEvent>,
         codex_responses_headers: Option<Arc<CodexResponsesHeaders>>,
         api_key_cyber_access_programs: cyber_access_program::ApiKeyCyberAccessPrograms,
+        required_cyber_access_programs: Option<
+            Arc<cyber_access_program::RequiredCyberAccessPrograms>,
+        >,
     ) -> Self {
         self.prompt_cache_key_policy = prompt_cache_key;
         self.session_id_override = session_id_override;
         self.event_sender = Some(event_sender);
         self.codex_responses_headers = codex_responses_headers;
         self.api_key_cyber_access_programs = api_key_cyber_access_programs;
+        self.required_cyber_access_programs = required_cyber_access_programs;
         self
     }
 
+    pub(crate) fn validate_required_cyber_access_program(
+        &self,
+        programs: Option<codex_api::AccessPrograms>,
+    ) -> Result<()> {
+        if let Some(required) = &self.required_cyber_access_programs
+            && !required
+                .allowed
+                .iter()
+                .any(|program| programs == Some((*program).into()))
+        {
+            return Err(CodexErr::InvalidRequest(
+                "this thread requires an allowed explicit access program".into(),
+            ));
+        }
+        Ok(())
+    }
+
     pub(crate) fn prompt_cache_key_policy(&self) -> PromptCacheKey {
         self.prompt_cache_key_policy.clone()
     }
@@ -863,6 +886,7 @@ impl ModelClient {
         if raw_memories.is_empty() {
             return Ok(Vec::new());
         }
+        self.validate_required_cyber_access_program(/*programs*/ None)?;
 
         let client_setup = self
             .current_client_setup(ClientRouting::ConfiguredProvider)
@@ -1200,6 +1224,7 @@ impl ModelClient {
 
     // Realtime keeps provider auth until its runtime integration lands later in the stack.
     async fn realtime_client_setup(&self) -> Result<CurrentClientSetup> {
+        self.validate_required_cyber_access_program(/*programs*/ None)?;
         let provider = if self.state.provider.runtime_auth().is_some() {
             create_model_provider(
                 self.state.provider.info().clone(),
@@ -1916,6 +1941,8 @@ impl ResponsesHttpExecutor<'_> {
                 prompt.cyber_access_program,
                 self.client.api_key_cyber_access_programs,
             )?;
+            self.client
+                .validate_required_cyber_access_program(request.access_programs)?;
             self.client
                 .prepare_response_items_for_request(&mut request.input);
             let request_session_telemetry =
@@ -2076,6 +2103,8 @@ impl ModelClientSession {
                 prompt.cyber_access_program,
                 self.client.api_key_cyber_access_programs,
             )?;
+            self.client
+                .validate_required_cyber_access_program(request.access_programs)?;
             let mut websocket_metadata = responses_metadata.clone();
             websocket_metadata.routing_hint = self.client.build_routing_hint_header(
                 client_setup.auth.as_ref(),

codex-rs/core/src/cyber_access_program.rs

@@ -6,6 +6,27 @@ use codex_login::CodexAuth;
 use codex_protocol::error::CodexErr;
 use codex_protocol::error::Result;
 use codex_protocol::turn_input::CyberAccessProgram;
+use std::sync::Arc;
+
+/// Host requirement for every model request issued by this thread.
+///
+/// Install in thread extension data before startup and reapply on cold resume.
+/// Core snapshots it before extension callbacks; it is not inherited or persisted.
+/// Every request must resolve to one listed program; backend authorization remains authoritative.
+#[derive(Clone, Debug)]
+pub struct RequiredCyberAccessPrograms {
+    pub allowed: Vec<CyberAccessProgram>,
+}
+
+pub(crate) struct RequiredCyberAccessProgramsSnapshot(pub Arc<RequiredCyberAccessPrograms>);
+
+impl RequiredCyberAccessPrograms {
+    /// Read Core's startup snapshot, independent of later edits to the public attachment.
+    pub fn for_thread(data: &codex_extension_api::ExtensionData) -> Option<Arc<Self>> {
+        data.get::<RequiredCyberAccessProgramsSnapshot>()
+            .map(|snapshot| Arc::clone(&snapshot.0))
+    }
+}
 
 /// Configured support for explicit Cyber programs on API-key requests.
 #[derive(Clone, Copy, Debug)]

codex-rs/core/src/guardian/review_session.rs

@@ -588,7 +588,12 @@ async fn run_review_on_session(
         personality: params.personality,
         model: review_model.model.clone(),
         reasoning_effort: review_model.reasoning_effort.clone(),
-        cyber_access_program: parent_turn.cyber_access_program,
+        cyber_access_program: parent_turn.cyber_access_program.filter(|_| {
+            crate::RequiredCyberAccessPrograms::for_thread(
+                &params.parent_session.services.thread_extension_data,
+            )
+            .is_none()
+        }),
         parent_response_id: params
             .parent_context
             .parent_response

codex-rs/core/src/lib.rs

@@ -93,6 +93,7 @@ pub mod context;
 mod context_manager;
 mod current_time;
 pub mod cyber_access_program;
+pub use cyber_access_program::RequiredCyberAccessPrograms;
 mod elicitation;
 mod environment_selection;
 pub mod exec;

codex-rs/core/src/realtime_conversation.rs

@@ -1396,6 +1396,9 @@ async fn prepare_realtime_start(
     sess: &Arc<Session>,
     params: ConversationStartParams,
 ) -> CodexResult<PreparedRealtimeConversationStart> {
+    sess.services
+        .model_client
+        .validate_required_cyber_access_program(/*programs*/ None)?;
     let provider = sess.provider().await;
     let auth_manager = sess
         .services

codex-rs/core/src/session/session.rs

@@ -1012,6 +1012,8 @@ impl Session {
                 ));
             }
         };
+        let required_cyber_access_programs =
+            thread_extension_init.get::<crate::cyber_access_program::RequiredCyberAccessPrograms>();
         let inherited_session_id = thread_extension_init
             .remove::<InheritedSessionId>()
             .map(|inherited| inherited.0);
@@ -1150,6 +1152,13 @@ impl Session {
             thread_id.to_string(),
             thread_extension_init,
         );
+        if let Some(required) = &required_cyber_access_programs {
+            thread_extension_data.insert(
+                crate::cyber_access_program::RequiredCyberAccessProgramsSnapshot(Arc::clone(
+                    required,
+                )),
+            );
+        }
         thread_extension_data.insert(crate::context::GuardianReviewEvidence::default());
         crate::model_search::SearchOwner::restore(&thread_extension_data, &initial_history);
         // Kick off independent async setup tasks in parallel to reduce startup latency.
@@ -1924,6 +1933,7 @@ impl Session {
                 tx_event.clone(),
                 codex_responses_headers,
                 crate::cyber_access_program::ApiKeyCyberAccessPrograms::from_config(&config),
+                required_cyber_access_programs,
             );
             // Once tracing starts, startup cleanup must own the session before the next await.
             let rollout_thread_trace = if matches!(