Bind runtime auth to the provider's selected credential source (#59823)

## Why

Runtime authentication must use the credential selected by the model provider. Explicit API keys and command credentials need independent ownership so cached connections and in-flight work cannot cross credential sources.

## What changed

- Register and reuse runtimes per credential source, supporting explicit provider keys and command credentials without falling back to saved login credentials.
- Track `AuthOwner` by credential manager and owner generation across WebSocket state, compaction, and scoring.
- Preserve cached identity and runtime registration state during credential-source outages or failed storage reloads. Require a successful auth reload before logout revokes credentials.
- Reject requests whose production or staging environment differs from runtime registration, including model catalog requests.

## Testing

Add regression coverage for credential selection and runtime reuse, owner isolation, source outages, failed reloads and logout recovery, and environment mismatches.

GitOrigin-RevId: 7bab55e0ea502593393df10046309fd8502edbd0

codex-rs/core/src/client.rs

@@ -25,6 +25,7 @@
 //! WebSocket prewarm is treated as the first websocket connection attempt for a turn. If it
 //! fails, normal stream retry/fallback logic handles recovery on the same turn.
 
+use codex_login::AuthOwner;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
@@ -368,7 +369,7 @@ struct WebsocketSession {
     parent_routing_affinity: Option<codex_http_client::ChatGptRoutingAffinity>,
     runtime_id: Option<String>,
     /// Owner of the cached state, including before a connection is opened.
-    auth_owner_generation: Option<u64>,
+    auth_owner: Option<AuthOwner>,
     connection_key: Option<ResponsesConnectionKey>,
     last_request: Option<ResponsesApiRequest>,
     last_response_rx: Option<oneshot::Receiver<LastResponse>>,
@@ -524,7 +525,7 @@ impl WebsocketSession {
             .continuation_reset_reason
             .or_else(|| self.last_request.as_ref().and(reason));
         *self = Self {
-            auth_owner_generation: self.auth_owner_generation,
+            auth_owner: self.auth_owner.clone(),
             continuation_reset_reason,
             last_connected_auth_revision: self.last_connected_auth_revision,
             ..Default::default()
@@ -754,12 +755,12 @@ impl ModelClient {
     /// This constructor does not perform network I/O itself; the session opens a websocket lazily
     /// when the first stream request is issued.
     pub fn new_session(&self) -> ModelClientSession {
-        let auth_owner_generation = self.auth_owner_generation();
+        let auth_owner = self.auth_owner();
         let mut websocket_session = self.take_cached_websocket_session();
-        if websocket_session.auth_owner_generation != auth_owner_generation {
+        if websocket_session.auth_owner != auth_owner {
             // Drop the old owner's cache before this turn can establish fresh routing state.
             websocket_session.reset(Some("other"));
-            websocket_session.auth_owner_generation = auth_owner_generation;
+            websocket_session.auth_owner = auth_owner;
         }
         ModelClientSession {
             client: self.clone(),
@@ -772,13 +773,8 @@ impl ModelClient {
         self.state.provider.auth_manager()
     }
 
-    pub(crate) fn auth_owner_generation(&self) -> Option<u64> {
-        self.auth_manager().map(|manager| {
-            manager
-                .auth_change_state_receiver()
-                .borrow()
-                .owner_generation
-        })
+    pub(crate) fn auth_owner(&self) -> Option<AuthOwner> {
+        codex_model_provider::request_auth_owner(self.state.provider.as_ref())
     }
 
     fn take_cached_websocket_session(&self) -> WebsocketSession {
@@ -1545,9 +1541,11 @@ impl<'a> ResponsesHttpExecutor<'a> {
 }
 
 impl ModelClientSession {
-    pub(crate) fn turn_state_with_auth_owner(&self) -> (Option<String>, Option<u64>) {
-        let owner = self.client.auth_owner_generation();
-        let turn_state = (self.websocket_session.auth_owner_generation == owner)
+    pub(crate) fn turn_state_with_auth_owner(&self) -> (Option<String>, Option<AuthOwner>) {
+        let owner = self.websocket_session.auth_owner.clone();
+        let turn_state = owner
+            .as_ref()
+            .is_none_or(AuthOwner::is_current)
             .then(|| self.turn_state.get().cloned())
             .flatten();
         (turn_state, owner)
@@ -1671,7 +1669,7 @@ impl ModelClientSession {
                     auth_revision: client_setup.auth_revision,
                     api_auth: client_setup.api_auth,
                     http_client_factory: client_setup.http_client_factory,
-                    auth_owner_generation: client_setup.auth_owner_generation,
+                    auth_owner: client_setup.auth_owner,
                     responses_metadata: &websocket_metadata,
                     auth_context,
                     request_route_telemetry: RequestRouteTelemetry::for_endpoint("/responses"),
@@ -1731,7 +1729,7 @@ impl ModelClientSession {
             auth_revision,
             api_auth,
             http_client_factory,
-            auth_owner_generation,
+            auth_owner,
             responses_metadata,
             auth_context,
             request_route_telemetry,
@@ -1759,8 +1757,8 @@ impl ModelClientSession {
         };
         let needs_new = self.websocket_session.connection.is_none() || reset_reason.is_some();
         // Resolving an external auth provider can change ownership during client setup.
-        let owner_changed = self.websocket_session.auth_owner_generation != auth_owner_generation
-            || self.client.auth_owner_generation() != auth_owner_generation;
+        let owner_changed = self.websocket_session.auth_owner != auth_owner
+            || auth_owner.as_ref().is_some_and(|owner| !owner.is_current());
         if owner_changed {
             self.turn_state = Arc::new(OnceLock::new());
         }
@@ -1795,7 +1793,7 @@ impl ModelClientSession {
             self.websocket_session.parent_routing_affinity = parent_routing_affinity;
             self.websocket_session.responses_headers = responses_headers.clone();
             self.websocket_session.runtime_id = runtime_id;
-            self.websocket_session.auth_owner_generation = auth_owner_generation;
+            self.websocket_session.auth_owner = auth_owner;
             self.websocket_session.connection_key = Some(connection_key);
             self.websocket_session
                 .set_connection_reused(/*connection_reused*/ false);
@@ -1845,7 +1843,7 @@ impl ResponsesHttpExecutor<'_> {
     pub(crate) async fn stream_responses_http(
         &self,
         turn_state: &Arc<OnceLock<String>>,
-        expected_auth_owner: Option<u64>,
+        expected_auth_owner: Option<AuthOwner>,
         event_sender: Option<&Sender<ProtocolEvent>>,
         prompt: &Prompt,
         model_info: &ModelInfo,
@@ -1867,9 +1865,8 @@ impl ResponsesHttpExecutor<'_> {
                 .client
                 .current_client_setup(ClientRouting::Workspace)
                 .await?;
-            if expected_auth_owner.is_some_and(|owner| {
-                Some(owner) != client_setup.auth_owner_generation
-                    || Some(owner) != self.client.auth_owner_generation()
+            if expected_auth_owner.as_ref().is_some_and(|owner| {
+                Some(owner) != client_setup.auth_owner.as_ref() || !owner.is_current()
             }) {
                 return Err(CodexErr::TurnAborted);
             }
@@ -2141,7 +2138,7 @@ impl ModelClientSession {
                     auth_revision: client_setup.auth_revision,
                     api_auth: client_setup.api_auth,
                     http_client_factory: client_setup.http_client_factory,
-                    auth_owner_generation: client_setup.auth_owner_generation,
+                    auth_owner: client_setup.auth_owner,
                     runtime_id: client_setup.runtime_id,
                     responses_metadata: &websocket_metadata,
                     auth_context: request_auth_context,
@@ -2889,7 +2886,7 @@ struct WebsocketConnectParams<'a> {
     auth_revision: Option<u64>,
     api_auth: SharedAuthProvider,
     http_client_factory: HttpClientFactory,
-    auth_owner_generation: Option<u64>,
+    auth_owner: Option<AuthOwner>,
     responses_metadata: &'a CodexResponsesMetadata,
     auth_context: AuthRequestTelemetryContext,
     request_route_telemetry: RequestRouteTelemetry,

codex-rs/core/src/client_runtime_auth_tests.rs

@@ -218,3 +218,82 @@ async fn realtime_keeps_provider_auth_until_runtime_integration(mode: RuntimeAut
     assert!(!requests[0].headers.contains_key("x-openai-agent-assertion"));
     assert!(registration.received_requests().await.unwrap().is_empty());
 }
+
+#[test_case::test_case(RuntimeAuthMode::DualHeader; "dual_header")]
+#[test_case::test_case(RuntimeAuthMode::AssertionOnly; "assertion_only")]
+#[tokio::test]
+async fn request_owner_tracks_the_selected_explicit_key(mode: RuntimeAuthMode) {
+    let server = MockServer::start().await;
+    Mock::given(path("/register"))
+        .respond_with(|request: &wiremock::Request| ResponseTemplate::new(200).set_body_json(json!({
+            "runtime_agent_instance_id": request.headers["idempotency-key"].to_str().unwrap(),
+            "agent_host_id": "host-a",
+            "actor": {"kind": "platform", "user_id": "user-a", "organization_id": "org-a", "project_id": null},
+        })))
+        .expect(2)
+        .mount(&server).await;
+    let credentials = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("saved-key"));
+    let context = codex_model_provider::ProviderAuthContext {
+        runtime_auth: Some((
+            Arc::new(RuntimeAuthManager::new(
+                credentials.clone(),
+                AgentHostAuthorization::InstallationId {
+                    installation_id: uuid::Uuid::nil(),
+                },
+                RuntimeRegistrationMetadata {
+                    harness: None,
+                    harness_version: None,
+                    role: RuntimeRole::Orchestrator,
+                },
+                format!("{}/register", server.uri()).parse().unwrap(),
+            )),
+            mode,
+        )),
+        credentials,
+    };
+    let mut owners = Vec::new();
+    for key in ["key-a", "key-b", "key-a"] {
+        let mut info = ModelProviderInfo::create_openai_provider(/*base_url*/ None);
+        info.experimental_bearer_token = Some(key.into());
+        let mut client = test_model_client(SessionSource::Exec);
+        Arc::get_mut(&mut client.state).unwrap().provider = context.model_provider(info);
+        let owner = client.auth_owner();
+        let setup = client
+            .current_client_setup(ClientRouting::ConfiguredProvider)
+            .await
+            .unwrap();
+        assert!(owner.is_some());
+        assert_eq!(setup.auth_owner, owner);
+        assert_eq!(setup.auth.unwrap().api_key(), Some(key));
+        let mut headers = http::HeaderMap::new();
+        setup.api_auth.add_auth_headers(&mut headers).unwrap();
+        match mode {
+            RuntimeAuthMode::DualHeader => {
+                assert_eq!(
+                    headers[http::header::AUTHORIZATION],
+                    format!("Bearer {key}")
+                );
+                assert!(headers.contains_key("x-openai-agent-assertion"));
+            }
+            RuntimeAuthMode::AssertionOnly => {
+                assert!(
+                    headers[http::header::AUTHORIZATION]
+                        .to_str()
+                        .unwrap()
+                        .starts_with("AgentAssertion ")
+                );
+                assert!(!headers.contains_key("x-openai-agent-assertion"));
+            }
+        }
+        owners.push(owner);
+    }
+    assert_ne!(
+        owners[0], owners[1],
+        "different keys cannot share turn or connection ownership"
+    );
+    assert_eq!(
+        owners[0], owners[2],
+        "reselecting a key preserves its owner"
+    );
+    server.verify().await;
+}

codex-rs/core/src/client_tests.rs

@@ -400,7 +400,7 @@ async fn client_setup_accepts_command_credential_refresh() {
                 refreshed_revision
             ),
         );
-        assert_ne!(setup.auth_owner_generation, client.auth_owner_generation());
+        assert_ne!(setup.auth_owner, client.auth_owner());
     }
 }
 
@@ -454,8 +454,11 @@ async fn client_setup_rebuilds_chatgpt_refresh_but_rejects_account_switches() {
             );
             assert_eq!(setup.auth.unwrap().get_token().unwrap(), refreshed_token);
             assert_eq!(
-                (setup.auth_revision, setup.auth_owner_generation),
-                (Some(*manager.auth_change_receiver().borrow()), Some(0))
+                (setup.auth_revision, setup.auth_owner),
+                (
+                    Some(*manager.auth_change_receiver().borrow()),
+                    Some(manager.auth_owner())
+                )
             );
         } else {
             assert_eq!(

codex-rs/core/src/compact_remote_v2.rs

@@ -1,3 +1,4 @@
+use codex_login::AuthOwner;
 use std::sync::Arc;
 use std::sync::OnceLock;
 
@@ -102,7 +103,7 @@ enum CompactionRequestKind<'a> {
 
 enum CompactionRequestTransport<'a> {
     Foreground(&'a mut ModelClientSession),
-    Background(Option<String>, Option<u64>),
+    Background(Option<String>, Option<AuthOwner>),
 }
 
 /// The caller decides whether a declined prepared result needs a foreground request.
@@ -297,7 +298,7 @@ async fn run_remote_compaction_request_v2(
 ) -> CodexResult<RemoteCompactionV2Output> {
     let turn_context = &step_context.turn;
     let client_session = match transport {
-        CompactionRequestTransport::Background(turn_state, auth_owner_generation) => {
+        CompactionRequestTransport::Background(turn_state, auth_owner) => {
             let settings = &step_context.settings;
             let cancellation = crate::session::async_compaction::shutdown_cancellation(sess).await;
             let effort = sess
@@ -311,7 +312,7 @@ async fn run_remote_compaction_request_v2(
                 biased;
                 _ = cancellation.cancelled() => return Err(CodexErr::TurnAborted),
                 stream = http_executor.stream_responses_http(
-                    &turn_state, *auth_owner_generation, /*event_sender*/ None, prompt, &settings.model_info, &step_context.session_telemetry, effort,
+                    &turn_state, auth_owner.clone(), /*event_sender*/ None, prompt, &settings.model_info, &step_context.session_telemetry, effort,
                     settings.reasoning_summary, settings.service_tier.clone(), responses_metadata,
                     &trace,
                 ) => stream?,

codex-rs/core/src/compact_remote_v2_attempt.rs

@@ -1,6 +1,7 @@
 //! Builds and sends foreground and speculative remote V2 compaction attempts.
 
 use crate::context::UserGoalUpdate;
+use codex_login::AuthOwner;
 use codex_protocol::ResponseItemId;
 use codex_protocol::openai_models::CatalogModel;
 use std::collections::BTreeSet;
@@ -165,10 +166,7 @@ pub(super) async fn run_remote_compact_v2_attempt(
             input.base_instructions,
             input.input,
             input.metadata,
-            CompactionRequestTransport::Background(
-                input.turn_state,
-                input.prefix.auth_owner_generation,
-            ),
+            CompactionRequestTransport::Background(input.turn_state, input.prefix.auth_owner),
             Some(input.window),
         ),
     };
@@ -283,7 +281,7 @@ pub(crate) async fn capture_background_input(
     mut snapshot: ContextManager,
     base_instructions: BaseInstructions,
     turn_state: Option<String>,
-    auth_owner_generation: Option<u64>,
+    auth_owner: Option<AuthOwner>,
     window: (String, u64, codex_protocol::ContextWindowId),
 ) -> BackgroundCompactionInput {
     let base_instructions = if step_context.settings.model_info.use_responses_lite
@@ -301,7 +299,7 @@ pub(crate) async fn capture_background_input(
     let prefix = CompactionPrefix {
         generation: snapshot.history_version(),
         len: snapshot.annotated_items().len(),
-        auth_owner_generation,
+        auth_owner,
     };
     trim_function_call_history_to_fit_context_window(
         &mut snapshot,
@@ -344,7 +342,7 @@ pub(crate) async fn run_background_compaction(
             sess.services.image_store.as_ref(),
         ) => {},
     }
-    let prefix = prepared_input.prefix;
+    let prefix = prepared_input.prefix.clone();
     let metadata = CompactionTurnMetadata::new(
         CompactionTrigger::Auto,
         CompactionReason::ContextLimit,

codex-rs/core/src/compact_remote_v2_installation_tests.rs

@@ -96,13 +96,19 @@ async fn prepared_prefix_rejection_respects_caller_intent(
     sess.replace_history(vec![original.clone()], /*reference_context_item*/ None)
         .await;
     let history = sess.clone_history().await;
-    let auth = sess.services.model_client.auth_owner_generation();
+    let auth = sess.services.model_client.auth_owner();
     let prefix = CompactionPrefix {
         generation: history.history_version(),
         len: history.annotated_items().len(),
-        auth_owner_generation: match change {
+        auth_owner: match change {
             HistoryChange::Replacement => auth,
-            HistoryChange::Account => Some(auth.unwrap_or_default().wrapping_add(1)),
+            HistoryChange::Account => Some(
+                codex_login::AuthManager::from_auth_for_testing_with_home(
+                    CodexAuth::from_api_key("other-account"),
+                    std::path::PathBuf::from("unused"),
+                )
+                .auth_owner(),
+            ),
         },
     };
     let roster = Arc::new(FailingRosterRead::default());
@@ -119,7 +125,7 @@ async fn prepared_prefix_rejection_respects_caller_intent(
         cutoff_world_state.host_roster_reader = Some(roster.clone());
     }
     let candidate = BackgroundCompactionCandidate {
-        prefix,
+        prefix: prefix.clone(),
         model_info: Arc::clone(&step.settings.model_info),
         cutoff_world_state: Arc::new(cutoff_world_state),
         attempt: RemoteCompactV2Attempt {

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

@@ -2,6 +2,7 @@
 //! Discarded requests still charge completed usage; session shutdown cancels remaining network work.
 //! Snapshot capture and launch registration hold session state; prompt preparation does not.
 
+use codex_login::AuthOwner;
 use std::sync::Arc;
 
 use crate::compact_remote_v2::BackgroundCompactionCandidate;
@@ -26,7 +27,7 @@ struct CompactionAttempt {
     history_version: u64,
     model_slug: String,
     compaction_hash: Option<String>,
-    auth_owner_generation: Option<u64>,
+    auth_owner: Option<AuthOwner>,
 }
 
 #[derive(Default)]
@@ -57,7 +58,7 @@ pub(super) async fn maybe_start(
     phase: CompactionPhase,
     turn_cancellation: &CancellationToken,
     token_status: &ContextWindowTokenStatus,
-    (turn_state, auth_owner_generation): (Option<String>, Option<u64>),
+    (turn_state, auth_owner): (Option<String>, Option<AuthOwner>),
 ) {
     let turn_context = &step_context.turn;
     // Let the foreground establish the backend's model pin before starting a parallel request.
@@ -89,14 +90,14 @@ pub(super) async fn maybe_start(
             history_version: state.history.history_version(),
             model_slug: model_info.slug.clone(),
             compaction_hash: model_info.comp_hash.clone(),
-            auth_owner_generation,
+            auth_owner: auth_owner.clone(),
         };
         // Stop blocks new launches; already-running compaction remains session-owned.
         if turn_cancellation.is_cancelled()
             || state.shutting_down
             || state.async_compaction.last_attempt.as_ref() == Some(&attempt)
             || state.history.annotated_items().is_empty()
-            || sess.services.model_client.auth_owner_generation() != auth_owner_generation
+            || sess.services.model_client.auth_owner() != auth_owner
         {
             return;
         }
@@ -108,7 +109,7 @@ pub(super) async fn maybe_start(
             state.history.clone(),
             base_instructions,
             turn_state,
-            auth_owner_generation,
+            auth_owner,
             sess.current_window_from_state(&state),
         );
         let tasks = state.async_compaction.tasks.clone();
@@ -187,7 +188,7 @@ pub(crate) async fn wait_for_pending(
         if attempt.history_version != state.history.history_version()
             || attempt.model_slug != model.slug
             || attempt.compaction_hash != model.comp_hash
-            || attempt.auth_owner_generation != sess.services.model_client.auth_owner_generation()
+            || attempt.auth_owner != sess.services.model_client.auth_owner()
         {
             return None;
         }

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

@@ -120,7 +120,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac
     let world_state = Arc::new(WorldState::default());
     let cancelled = CancellationToken::new();
     cancelled.cancel();
-    let auth = sess.services.model_client.auth_owner_generation();
+    let auth = sess.services.model_client.auth_owner();
     maybe_start(
         &sess,
         &step,
@@ -128,7 +128,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac
         phase,
         &cancelled,
         &status,
-        (Some("routing".to_string()), auth),
+        (Some("routing".to_string()), auth.clone()),
     )
     .await;
     assert!(
@@ -149,7 +149,7 @@ async fn cancelled_turn_skips_launch_but_preserves_existing_work(trigger: Compac
         phase,
         &turn_cancellation,
         &status,
-        (Some("routing".to_string()), auth),
+        (Some("routing".to_string()), auth.clone()),
     )
     .await;
     turn_cancellation.cancel();

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

@@ -3,6 +3,7 @@ pub(crate) mod startup;
 use crate::PromptCacheKey;
 use crate::WithTurnExtensionData;
 use crate::runtime_auth::RuntimeAuthSlot;
+use codex_login::AuthOwner;
 use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::collections::HashSet;
@@ -477,11 +478,11 @@ pub(crate) enum ForkPersistence {
 }
 
 /// Normal appends keep this generation; compaction and rollback start a new one.
-#[derive(Clone, Copy)]
+#[derive(Clone)]
 pub(crate) struct CompactionPrefix {
     pub(crate) generation: u64,
     pub(crate) len: usize,
-    pub(crate) auth_owner_generation: Option<u64>,
+    pub(crate) auth_owner: Option<AuthOwner>,
 }
 
 pub(crate) enum CompactedHistoryInstallation<'a> {
@@ -4778,8 +4779,7 @@ impl Session {
                     // Validate immediately before slicing and replacing the captured prefix.
                     if state.history.history_version() != prefix.generation
                         || state.history.annotated_items().len() < prefix.len
-                        || self.services.model_client.auth_owner_generation()
-                            != prefix.auth_owner_generation
+                        || self.services.model_client.auth_owner() != prefix.auth_owner
                     {
                         return false;
                     }