chore(release): v3.39.1

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-api/src/openai/responses/request.rs

@@ -618,8 +618,6 @@ fn passthrough_header_name(name: &str, connection_headers: &[String]) -> bool {
             | "chatgpt-project-id"
             | "openai-organization"
             | "openai-project"
-            | "x-openai-account-routing-override"
-            | "x-openai-fedramp"
             // 上游指纹必须由运行时画像统一生成，客户端 originator/User-Agent/version
             // 不能作为不透明头透传覆盖，避免不同下游客户端暴露不一致的设备指纹
             | "originator"

backend/crates/gateway-api/tests/openai/responses/mod.rs

@@ -431,6 +431,9 @@ fn decoder_should_preserve_ordinary_request_headers_as_opaque_multivalues() {
         values("x-codex-installation-id"),
         vec![b"client-installation".to_vec()]
     );
+    for name in ["x-openai-account-routing-override", "x-openai-fedramp"] {
+        assert_eq!(values(name), vec![b"first".to_vec(), b"second".to_vec()]);
+    }
     for preserved in [
         "accept",
         "content-type",
@@ -450,8 +453,6 @@ fn decoder_should_preserve_ordinary_request_headers_as_opaque_multivalues() {
         "cookie",
         "chatgpt-account-id",
         "chatgpt-project-id",
-        "x-openai-account-routing-override",
-        "x-openai-fedramp",
         // 上游指纹由运行时画像统一生成，客户端不得覆盖
         "user-agent",
         "originator",

backend/crates/gateway-core/src/routing/snapshot.rs

@@ -23,7 +23,7 @@ use super::{
     RuntimeAccountDirectory, UpstreamModelId,
 };
 
-const MAXIMUM_CATALOG_STABILITY_ATTEMPTS: usize = 4;
+const MAXIMUM_SNAPSHOT_STABILITY_ATTEMPTS: usize = 4;
 
 type ModelCatalogAccounts = BTreeMap<ProviderKind, BTreeMap<String, BTreeSet<ProviderAccountId>>>;
 
@@ -328,14 +328,17 @@ impl RuntimeSnapshotCompiler {
             ConfigRevision,
             crate::routing::extensions::ExtensionSetReference,
         )> = None;
-        for _ in 0..MAXIMUM_CATALOG_STABILITY_ATTEMPTS {
+        let mut instability = RuntimeSnapshotCompileError::CatalogChanged;
+        for _ in 0..MAXIMUM_SNAPSHOT_STABILITY_ATTEMPTS {
             let facts = self
                 .store
                 .load_snapshot_facts()
                 .await
                 .map_err(|_| RuntimeSnapshotCompileError::StoreUnavailable)?;
             if facts.config_revision != facts.observed_current_revision {
-                return Err(RuntimeSnapshotCompileError::RevisionChanged);
+                // 并发提交使本轮事实过期时重读完整快照，不能混用不同版本的权限
+                instability = RuntimeSnapshotCompileError::RevisionChanged;
+                continue;
             }
             let revision = facts.config_revision;
             let extensions = match &self.extensions {
@@ -347,7 +350,7 @@ impl RuntimeSnapshotCompiler {
                                 *prepared_revision != revision || !prepared.is_ready()
                             });
                     if must_prepare {
-                        let prepared = preparer
+                        let prepared = match preparer
                             .prepare_with_request_body_limit(
                                 revision,
                                 usize::try_from(
@@ -356,7 +359,21 @@ impl RuntimeSnapshotCompiler {
                                 .unwrap_or(usize::MAX),
                             )
                             .await
-                            .map_err(|_| RuntimeSnapshotCompileError::ExtensionsUnavailable)?;
+                        {
+                            Ok(prepared) => prepared,
+                            Err(_) => {
+                                // 扩展读取也可能遇到并发提交；只有版本已变才重读事实
+                                let observed =
+                                    self.store.current_config_revision().await.map_err(|_| {
+                                        RuntimeSnapshotCompileError::StoreUnavailable
+                                    })?;
+                                if observed == revision {
+                                    return Err(RuntimeSnapshotCompileError::ExtensionsUnavailable);
+                                }
+                                instability = RuntimeSnapshotCompileError::RevisionChanged;
+                                continue;
+                            }
+                        };
                         prepared_extensions = Some((revision, prepared));
                     }
                     prepared_extensions
@@ -390,7 +407,8 @@ impl RuntimeSnapshotCompiler {
                         .map_err(|_| RuntimeSnapshotCompileError::StoreUnavailable)?
                         != revision
                 {
-                    return Err(RuntimeSnapshotCompileError::RevisionChanged);
+                    instability = RuntimeSnapshotCompileError::RevisionChanged;
+                    continue;
                 }
                 let snapshot = snapshot
                     .with_provider_catalog_generations(if cached.is_some() && !refresh_catalog {
@@ -401,8 +419,9 @@ impl RuntimeSnapshotCompiler {
                     .with_extensions(extensions);
                 return Ok(snapshot);
             }
+            instability = RuntimeSnapshotCompileError::CatalogChanged;
         }
-        Err(RuntimeSnapshotCompileError::CatalogChanged)
+        Err(instability)
     }
 }
 

backend/crates/gateway-core/tests/routing/snapshot.rs

@@ -730,12 +730,116 @@ impl ProviderCatalogPort for PublishingCatalog {
 
 #[test]
 fn compiler_should_reject_revision_changed_during_consistent_read() {
-    let facts = facts(1, 2);
-    let compiler = compiler(Arc::new(TestSnapshotStore::new(Ok(facts))));
+    let store = Arc::new(SequencedSnapshotStore {
+        facts: vec![facts(1, 2)],
+        loads: AtomicUsize::new(0),
+        current_revision: revision(2),
+    });
+    let compiler = compiler(store.clone());
 
     let error = block_on(compiler.compile()).expect_err("revision drift must fail closed");
 
     assert_eq!(error, RuntimeSnapshotCompileError::RevisionChanged);
+    assert_eq!(store.loads.load(Ordering::SeqCst), 4);
+}
+
+#[test]
+fn compiler_reloads_complete_facts_after_revision_changes_during_read() {
+    let store = Arc::new(SequencedSnapshotStore {
+        facts: vec![facts(1, 2), facts(2, 2)],
+        loads: AtomicUsize::new(0),
+        current_revision: revision(2),
+    });
+
+    let snapshot = block_on(compiler(store.clone()).compile()).expect("latest consistent facts");
+
+    assert_eq!(snapshot.revision(), revision(2));
+    assert_eq!(store.loads.load(Ordering::SeqCst), 2);
+}
+
+#[test]
+fn compiler_reloads_revision_changed_after_extension_preparation() {
+    let drops = Arc::new(AtomicUsize::new(0));
+    let extensions = Arc::new(CountingExtensionPreparation {
+        prepares: AtomicUsize::new(0),
+        drops: Arc::clone(&drops),
+    });
+    let compiler = RuntimeSnapshotCompiler::new(
+        Arc::new(SequencedSnapshotStore {
+            facts: vec![facts(1, 1), facts(2, 2)],
+            loads: AtomicUsize::new(0),
+            current_revision: revision(2),
+        }),
+        Arc::new(TestCatalog::Empty),
+    )
+    .with_extensions(extensions.clone());
+
+    let snapshot = block_on(compiler.compile()).expect("latest prepared snapshot");
+
+    assert_eq!(snapshot.revision(), revision(2));
+    assert_eq!(snapshot.extensions().unwrap().id().as_str(), "prepared-2");
+    assert_eq!(extensions.prepares.load(Ordering::SeqCst), 2);
+    assert_eq!(drops.load(Ordering::SeqCst), 1);
+    drop(snapshot);
+    assert_eq!(drops.load(Ordering::SeqCst), 2);
+}
+
+#[test]
+fn compiler_rechecks_revision_when_extension_preparation_rejects_stale_facts() {
+    struct RacingPreparation {
+        store: Arc<TestSnapshotStore>,
+        changes: usize,
+        calls: AtomicUsize,
+    }
+
+    impl ExtensionPreparationPort for RacingPreparation {
+        fn prepare(
+            &self,
+            _: ConfigRevision,
+        ) -> BoxFuture<'_, Result<ExtensionSetReference, ExtensionPreparationError>> {
+            Box::pin(async move {
+                let call = self.calls.fetch_add(1, Ordering::SeqCst);
+                if self.changes == 0 || call < self.changes {
+                    if self.changes != 0 {
+                        let next = u64::try_from(call).unwrap() + 2;
+                        *self.store.facts.lock().unwrap() = Ok(facts(next, next));
+                        *self.store.current_revision.lock().unwrap() = Ok(revision(next));
+                    }
+                    return Err(ExtensionPreparationError);
+                }
+                Ok(ExtensionSetReference::new(
+                    ExtensionSetId::new("ready".to_owned()).unwrap(),
+                    Arc::new(AliasLease(vec![])),
+                ))
+            })
+        }
+    }
+
+    for (changes, expected_calls) in [(0, 1), (1, 2), (4, 4)] {
+        let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1))));
+        let extensions = Arc::new(RacingPreparation {
+            store: store.clone(),
+            changes,
+            calls: AtomicUsize::new(0),
+        });
+        let compiler = RuntimeSnapshotCompiler::new(store, Arc::new(TestCatalog::Empty))
+            .with_extensions(extensions.clone());
+
+        let result = block_on(compiler.compile());
+
+        match changes {
+            0 => assert_eq!(
+                result.unwrap_err(),
+                RuntimeSnapshotCompileError::ExtensionsUnavailable
+            ),
+            1 => assert_eq!(result.unwrap().revision(), revision(2)),
+            _ => assert_eq!(
+                result.unwrap_err(),
+                RuntimeSnapshotCompileError::RevisionChanged
+            ),
+        }
+        assert_eq!(extensions.calls.load(Ordering::SeqCst), expected_calls);
+    }
 }
 
 #[test]

backend/crates/gateway-core/tests/runtime/mod.rs

@@ -326,6 +326,39 @@ fn publisher_should_suspend_but_still_notify_after_committed_refresh_failure() {
     );
 }
 
+#[test]
+fn committed_refresh_reloads_raced_facts_without_losing_revoked_scope() {
+    block_on(async {
+        let store = Arc::new(TestSnapshotStore::new(Ok(scoped_facts(10, true))));
+        let initial = compiler(store.clone())
+            .compile()
+            .await
+            .expect("initial scope");
+        let handle = RuntimeSnapshotHandle::new(initial);
+        let frozen = handle.acquire().expect("initial request");
+        assert!(removed_account_allowed(&frozen));
+        let publisher = RuntimeSnapshotPublisher::new(
+            Arc::new(compiler(store.clone())),
+            handle.clone(),
+            Arc::new(TestSnapshotSubscriptions::default()),
+        );
+        let (release, gate) = oneshot::channel();
+        *store.next_load.lock().expect("load gate lock") = Some(gate);
+        let mut committed = publisher.publish_committed(revision(11));
+        assert!(futures::poll!(committed.as_mut()).is_pending());
+        *store.facts.lock().expect("facts lock") = Ok(scoped_facts(11, false));
+        *store.current_revision.lock().expect("revision lock") = Ok(revision(11));
+        release.send(Ok(facts(10, 11))).expect("release raced read");
+
+        committed.await;
+
+        let current = handle.acquire().expect("latest snapshot remains available");
+        assert_eq!(current.revision(), revision(11));
+        assert!(!removed_account_allowed(&current));
+        assert!(removed_account_allowed(&frozen));
+    });
+}
+
 #[test]
 fn publisher_should_contribute_reconciliation_and_subscription_workers() {
     let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1))));

backend/crates/gateway-host/src/proxy_probe.rs

@@ -125,6 +125,17 @@ impl HttpProxyProbe {
 }
 
 impl HttpProxyProbe {
+    async fn timed_exit_ip_at(
+        &self,
+        proxy: Option<&OutboundProxy>,
+        endpoint: &str,
+    ) -> (Result<IpAddr, &'static str>, u64) {
+        let started = Instant::now();
+        let result = self.exit_ip_at(proxy, endpoint).await;
+        let latency_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
+        (result, latency_ms)
+    }
+
     async fn test_connection(&self, proxy: Option<&OutboundProxy>) -> ProxyTestResult {
         let started = Instant::now();
         let timeout_limit = Duration::from_secs(15);
@@ -169,15 +180,15 @@ impl HttpProxyProbe {
             } => {
                 let probe_dual = async {
                     tokio::join!(
-                        self.exit_ip_at(proxy, ipv4_endpoint),
-                        self.exit_ip_at(proxy, ipv6_endpoint),
+                        self.timed_exit_ip_at(proxy, ipv4_endpoint),
+                        self.timed_exit_ip_at(proxy, ipv6_endpoint),
                     )
                 };
                 let result = tokio::time::timeout(timeout_limit, probe_dual).await;
                 let latency_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
 
                 match result {
-                    Ok((res_v4, res_v6)) => {
+                    Ok(((res_v4, latency_v4), (res_v6, latency_v6))) => {
                         let exit_ipv4: Option<Ipv4Addr> = match res_v4 {
                             Ok(IpAddr::V4(v4)) => Some(v4),
                             _ => None,
@@ -186,6 +197,14 @@ impl HttpProxyProbe {
                             Ok(IpAddr::V6(v6)) => Some(v6),
                             _ => None,
                         };
+                        // 延迟对应首选的可用出口，另一地址族失败等待不代表该出口延迟
+                        let latency_ms = if exit_ipv4.is_some() {
+                            latency_v4
+                        } else if exit_ipv6.is_some() {
+                            latency_v6
+                        } else {
+                            latency_ms
+                        };
 
                         if exit_ipv4.is_some() && exit_ipv6.is_some() {
                             ProxyTestResult {

backend/crates/gateway-host/tests/proxy_probe.rs

@@ -155,6 +155,39 @@ async fn dual_stack_proxy_probe_reports_single_stack_when_only_one_succeeds() {
     assert!(result.message.contains("仅 IPv4"));
 }
 
+#[tokio::test]
+async fn dual_stack_latency_measures_the_selected_exit_without_the_other_family_wait() {
+    use std::time::{Duration, Instant};
+
+    for working_family in ["v4", "v6"] {
+        let server = MockServer::start().await;
+        for (family, ip) in [("v4", "203.0.113.8"), ("v6", "2001:db8::8")] {
+            let response = if family == working_family {
+                ResponseTemplate::new(200).set_body_json(json!({"ip": ip}))
+            } else {
+                ResponseTemplate::new(503).set_delay(Duration::from_millis(500))
+            };
+            Mock::given(path(format!("/{family}")))
+                .respond_with(response)
+                .expect(1)
+                .mount(&server)
+                .await;
+        }
+        let started = Instant::now();
+        let result = HttpProxyProbe::new_dual(
+            format!("{}/v4", server.uri()),
+            format!("{}/v6", server.uri()),
+        )
+        .test(None, false)
+        .await;
+        assert!(result.success);
+        assert_eq!(result.exit_ipv4.is_some(), working_family == "v4");
+        assert_eq!(result.exit_ipv6.is_some(), working_family == "v6");
+        assert!(started.elapsed() >= Duration::from_millis(500));
+        assert!(u128::from(result.latency_ms) + 400 < started.elapsed().as_millis());
+    }
+}
+
 fn location_response(ip: &str, timezone: &str) -> serde_json::Value {
     json!({"success": true, "ip": ip, "country_code": "JP", "region": "Tokyo", "city": "Tokyo", "timezone": {"id": timezone}})
 }

backend/crates/gateway-store/src/postgres/proxies.rs

@@ -857,7 +857,12 @@ impl ProxyStore for PgProxyRepository {
         .await
         .map_err(|source| store_error(unavailable(source)))?;
         if locked.is_none() {
-            return Err(store_error(conflict(id)));
+            // 返回版本冲突前释放事务锁，避免误挡紧接着的删除或编辑
+            let error = match transaction.rollback().await {
+                Ok(()) => conflict(id),
+                Err(cleanup) => conflict(id).with_cleanup(cleanup),
+            };
+            return Err(store_error(error));
         }
         let account_ids: Vec<String> = sqlx::query_scalar(
             "update provider_accounts set outbound_proxy_id = null, outbound_proxy_url = null, provider_quota_json = null, quota_observed_at = null,

backend/crates/gateway-store/tests/postgres/proxies.rs

@@ -735,8 +735,8 @@ async fn rejected_import_reservations_release_proxy_lock_before_returning() {
 }
 
 #[tokio::test]
-async fn stale_proxy_test_releases_lock_before_next_update() {
-    let Some(database) = TestDatabase::create("stale_proxy_test_lock").await else {
+async fn stale_proxy_mutations_release_locks_before_next_update() {
+    let Some(database) = TestDatabase::create("stale_proxy_mutation_lock").await else {
         return;
     };
     let store = PgProxyRepository::new(database.pool.clone());
@@ -762,52 +762,63 @@ async fn stale_proxy_test_releases_lock_before_next_update() {
         .unwrap()
         .record;
 
-    // 暂停连接归还时的后台清理，确保锁由返回错误前的显式回滚释放
-    let release_gate = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
-    let rejected_pool = database
-        .pool
-        .options()
-        .clone()
-        .max_connections(1)
-        .after_release({
-            let release_gate = release_gate.clone();
-            move |_, _| {
+    for mutation in ["record_test", "update", "delete"] {
+        // 暂停连接归还时的后台清理，确保锁由返回错误前的显式回滚释放
+        let release_gate = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
+        let rejected_pool = database
+            .pool
+            .options()
+            .clone()
+            .max_connections(1)
+            .after_release({
                 let release_gate = release_gate.clone();
-                Box::pin(async move {
-                    let _permit = release_gate.acquire().await.unwrap();
-                    Ok(true)
-                })
-            }
-        })
-        .connect_with((*database.pool.connect_options()).clone())
-        .await
-        .unwrap();
-    let stale_writer = PgProxyRepository::new(rejected_pool.clone());
-    let error = stale_writer
-        .record_test(&saved.id, stale_revision, success(), &context)
-        .await
-        .unwrap_err();
-    let update = store
-        .update(
-            UpdateProxy {
-                max_websocket_connections: None,
-                enabled: None,
-                id: saved.id.clone(),
-                revision: saved.revision,
-                name: saved.name.clone(),
-                proxy: None,
-                auto_location: None,
-                location: None,
-                test: None,
-            },
-            &context,
-        )
-        .await;
-    release_gate.add_permits(1);
-    rejected_pool.close().await;
+                move |_, _| {
+                    let release_gate = release_gate.clone();
+                    Box::pin(async move {
+                        let _permit = release_gate.acquire().await.unwrap();
+                        Ok(true)
+                    })
+                }
+            })
+            .connect_with((*database.pool.connect_options()).clone())
+            .await
+            .unwrap();
+        let stale_writer = PgProxyRepository::new(rejected_pool.clone());
+        let command = |revision| UpdateProxy {
+            max_websocket_connections: None,
+            enabled: None,
+            id: saved.id.clone(),
+            revision,
+            name: saved.name.clone(),
+            proxy: None,
+            auto_location: None,
+            location: None,
+            test: None,
+        };
+        let error = match mutation {
+            "record_test" => stale_writer
+                .record_test(&saved.id, stale_revision, success(), &context)
+                .await
+                .unwrap_err(),
+            "update" => stale_writer
+                .update(command(stale_revision), &context)
+                .await
+                .unwrap_err(),
+            "delete" => stale_writer
+                .delete(&saved.id, stale_revision, &context)
+                .await
+                .unwrap_err(),
+            _ => unreachable!(),
+        };
+        let update = store.update(command(saved.revision), &context).await;
+        release_gate.add_permits(1);
+        rejected_pool.close().await;
+        assert_eq!(error.kind(), AdminStoreErrorKind::Conflict);
+        saved = update
+            .unwrap_or_else(|error| panic!("过期 {mutation} 不应阻挡后续编辑: {error:?}"))
+            .record;
+    }
     database.close().await;
-    assert_eq!(error.kind(), AdminStoreErrorKind::Conflict);
-    assert!(update.is_ok(), "过期检测结果不应阻挡后续编辑: {update:?}");
 }
 
 #[tokio::test]

backend/crates/providers/openai/src/provider/execution.rs

@@ -397,6 +397,7 @@ pub(super) struct ColdResponse {
     pub(super) session_transport_key_hash: Option<String>,
     pub(super) session_transport_recovery: CodexSessionTransportRecovery,
     pub(super) guardian_routing: guardian::GuardianRouting,
+    pub(super) turn_state_scope: Option<turn_state::TurnStateScope>,
     pub(super) websocket_retry_count: u32,
     pub(super) stream_max_retries: u32,
     pub(super) session_capture: Option<OpenAiSessionCapture>,
@@ -784,6 +785,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
         session_transport_key_hash,
         session_transport_recovery,
         guardian_routing,
+        turn_state_scope,
         websocket_retry_count,
         stream_max_retries,
         mut session_capture,
@@ -873,6 +875,17 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         },
                     );
                 }
+                if let Some(scope) = &turn_state_scope
+                    && let Some(response) = failure.error.client_visible_upstream_response()
+                {
+                    for header in response.headers() {
+                        if header.name().eq_ignore_ascii_case("x-codex-turn-state")
+                            && let Ok(token) = std::str::from_utf8(header.value().as_ref())
+                        {
+                            scope.remember(token);
+                        }
+                    }
+                }
                 if let Some(observation) = failure.observation.take() {
                     yield ProviderEvent::observation(observation);
                 }
@@ -897,6 +910,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             Err(failure.error)?;
             return;
         }
+        if let Some(scope) = &turn_state_scope
+            && let Some(token) = response.turn_state.as_deref()
+        {
+            scope.remember(token);
+        }
         if let Some(capture) = session_capture.as_mut() {
             capture.continuation_scope = Some(if capture.response_store {
                 OpenAiContinuationScope::Persisted
@@ -1040,6 +1058,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         response_metadata_updates.as_ref(),
                         &mut session_capture,
                         &mut observation_state,
+                        turn_state_scope.as_ref(),
                     )
                     .await;
                     let observation_event = if rate_limits_changed || metadata_merge.is_some()
@@ -1117,6 +1136,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                 response_metadata_updates.as_ref(),
                 &mut session_capture,
                 &mut observation_state,
+                turn_state_scope.as_ref(),
             )
             .await;
             let metadata_changed = metadata_merge.unwrap_or(false);
@@ -1317,6 +1337,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             response_metadata_updates.as_ref(),
             &mut session_capture,
             &mut observation_state,
+            turn_state_scope.as_ref(),
         )
         .await
         .unwrap_or(false);
@@ -1384,8 +1405,12 @@ async fn merge_response_metadata_updates(
     updates: Option<&CodexResponseMetadataUpdates>,
     session_capture: &mut Option<OpenAiSessionCapture>,
     observation_state: &mut OpenAiResponseObservationState,
+    turn_state_scope: Option<&turn_state::TurnStateScope>,
 ) -> Option<bool> {
     let turn_state = updates?.lock().await.turn_state.take()?;
+    if let Some(scope) = turn_state_scope {
+        scope.remember(&turn_state);
+    }
     if let Some(capture) = session_capture.as_mut() {
         capture.turn_state = Some(turn_state.clone());
     }

backend/crates/providers/openai/src/provider/mod.rs

@@ -107,6 +107,7 @@ mod failure;
 mod guardian;
 mod live;
 mod observation;
+mod turn_state;
 mod upstream_adapter;
 mod workers;
 pub(crate) use live::{CodexLiveGateway, CodexLiveRegistry};
@@ -156,6 +157,7 @@ pub struct CodexProvider {
     session_identity: Option<CodexSessionIdentity>,
     session_transport_recovery: CodexSessionTransportRecovery,
     guardian_routing: guardian::GuardianRouting,
+    turn_state_ownership: turn_state::TurnStateOwnership,
     stream_max_retries: u32,
     live_registry: Arc<CodexLiveRegistry>,
     live_gateway: Option<Arc<CodexLiveGateway>>,
@@ -361,6 +363,7 @@ impl CodexProvider {
             session_identity: None,
             session_transport_recovery: CodexSessionTransportRecovery::default(),
             guardian_routing: guardian::GuardianRouting::default(),
+            turn_state_ownership: turn_state::TurnStateOwnership::default(),
             stream_max_retries,
             live_registry: Arc::new(CodexLiveRegistry::default()),
             live_gateway: None,
@@ -963,10 +966,25 @@ impl CodexProvider {
                 )
                 .map(|affinity| (response_id.to_owned(), affinity))
         });
+        let turn_state_scope = self
+            .turn_state_ownership
+            .scope(&context, &lease, &upstream_request);
+        // 全量请求的 token 证明不能用于放行 response ID 或其他账号状态
+        let verified_turn_state = upstream_request
+            .turn_state
+            .as_ref()
+            .filter(|token| {
+                matches!(account_scope, RequestAccountScope::Unknown)
+                    && turn_state_scope
+                        .as_ref()
+                        .is_some_and(|scope| scope.owns(token))
+            })
+            .cloned();
         scope_request_to_account(
             &mut upstream_request,
             lease.installation_id(),
             account_scope,
+            verified_turn_state.as_deref(),
         );
         // 父响应的可信关联只恢复此字段，不把首次 Guardian 请求的其他账号状态视为可信
         if let Some((response_id, _)) = &guardian_parent
@@ -989,6 +1007,7 @@ impl CodexProvider {
                 "ownerCrossAccount": state_owner_cross_account,
                 "accountSwitch": lease.account_switch(),
                 "accountStateReusable": account_scope.can_reuse_account_state(),
+                "verifiedTurnStateEcho": verified_turn_state.is_some(),
                 "before": before,
                 "after": crate::transport::diagnostics::evidence::request_state(&upstream_request),
             }));
@@ -1162,6 +1181,7 @@ impl CodexProvider {
             session_transport_key_hash,
             session_transport_recovery: self.session_transport_recovery.clone(),
             guardian_routing: self.guardian_routing.clone(),
+            turn_state_scope,
             websocket_retry_count,
             stream_max_retries: self.stream_max_retries,
             session_capture,

backend/crates/providers/openai/src/provider/turn_state.rs

@@ -0,0 +1,124 @@
+//! 只为上游已返回的 turn state 保存归属证明，不恢复正文或参与账号选择
+
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+
+use gateway_core::engine::AttemptContext;
+use sha2::{Digest, Sha256};
+
+use crate::{
+    credential::{CodexCredentialLease, CodexRuntimeAuthentication},
+    transport::protocol::responses::CodexResponsesRequest,
+};
+
+// 仅保存 32 字节摘要与过期时间，4096 项的条目数据约 192 KiB，不随 token 长度增长
+const MAX_PROOFS: usize = 4096;
+
+#[derive(Clone, Default)]
+pub(super) struct TurnStateOwnership {
+    proofs: Arc<Mutex<HashMap<[u8; 32], Instant>>>,
+}
+
+pub(super) struct TurnStateScope {
+    ownership: TurnStateOwnership,
+    identity: Sha256,
+    retention: Duration,
+}
+
+impl TurnStateOwnership {
+    pub(super) fn scope(
+        &self,
+        context: &AttemptContext,
+        lease: &CodexCredentialLease,
+        request: &CodexResponsesRequest,
+    ) -> Option<TurnStateScope> {
+        let turn_id = request
+            .client_turn_id
+            .as_deref()
+            .filter(|id| !id.is_empty())?;
+        let mut identity = Sha256::new();
+        for part in [
+            context.client_api_key_ref().as_str(),
+            lease.account_id().as_str(),
+            request.client_session_id.as_deref().unwrap_or_default(),
+            request.client_thread_id.as_deref().unwrap_or_default(),
+            turn_id,
+        ] {
+            identity.update(part.len().to_le_bytes());
+            identity.update(part.as_bytes());
+        }
+        // OAuth 刷新不改变账号归属；API Key 替换则不能继承原认证的路由状态
+        if matches!(
+            lease.authentication(),
+            CodexRuntimeAuthentication::ApiKey(_)
+        ) {
+            identity.update([1]);
+            identity.update(lease.account().revision().get().to_le_bytes());
+        } else {
+            identity.update([0]);
+        }
+        Some(TurnStateScope {
+            ownership: self.clone(),
+            identity,
+            retention: context
+                .account_selection_policy()
+                .openai_session_affinity_ttl(),
+        })
+    }
+}
+
+impl TurnStateScope {
+    pub(super) fn owns(&self, token: &str) -> bool {
+        let key = self.key(token);
+        let mut proofs = self
+            .ownership
+            .proofs
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let Some(expires_at) = proofs.get_mut(&key) else {
+            return false;
+        };
+        let now = Instant::now();
+        if *expires_at <= now {
+            proofs.remove(&key);
+            return false;
+        }
+        *expires_at = now + self.retention;
+        true
+    }
+
+    pub(super) fn remember(&self, token: &str) {
+        if token.is_empty() {
+            return;
+        }
+        let key = self.key(token);
+        let now = Instant::now();
+        let mut proofs = self
+            .ownership
+            .proofs
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if proofs.len() >= MAX_PROOFS && !proofs.contains_key(&key) {
+            proofs.retain(|_, expires_at| *expires_at > now);
+            if proofs.len() >= MAX_PROOFS
+                && let Some(oldest) = proofs
+                    .iter()
+                    .min_by_key(|(_, expires_at)| **expires_at)
+                    .map(|(key, _)| *key)
+            {
+                proofs.remove(&oldest);
+            }
+        }
+        proofs.insert(key, now + self.retention);
+    }
+
+    fn key(&self, token: &str) -> [u8; 32] {
+        let mut hash = self.identity.clone();
+        hash.update(token.len().to_le_bytes());
+        hash.update(token.as_bytes());
+        hash.finalize().into()
+    }
+}

backend/crates/providers/openai/src/transport/correlation.rs

@@ -23,16 +23,6 @@ const ID_FIELDS: &[&str] = &[
     "parentThreadId",
 ];
 const METADATA_FIELDS: &[&str] = &["turnMetadata", "turn_metadata", "x-codex-turn-metadata"];
-const TRACE_FIELDS: &[&str] = &[
-    "traceparent",
-    "tracestate",
-    "sentry-trace",
-    "baggage",
-    "ws_request_header_traceparent",
-    "ws_request_header_tracestate",
-    "ws_request_header_sentry-trace",
-    "ws_request_header_baggage",
-];
 // Live 的 x-session-id 与 voice-session-id 是业务会话引用，必须原样传递
 const ID_HEADERS: &[&str] = &[
     "session-id",
@@ -102,25 +92,14 @@ fn map_fields(object: &mut Map<String, Value>, installation_id: &str, fields: &[
     changed
 }
 
-fn scope_object(object: &mut Map<String, Value>, installation_id: Option<&str>) -> usize {
-    let mut changed = 0;
-    for key in TRACE_FIELDS {
-        changed += usize::from(object.remove(*key).is_some());
-    }
-    if let Some(installation_id) = installation_id {
-        changed += map_fields(object, installation_id, ID_FIELDS);
-    }
-    changed
-}
-
-fn scope_metadata(value: &mut Value, installation_id: Option<&str>) {
+fn scope_metadata(value: &mut Value, installation_id: &str) {
     match value {
         Value::Object(object) => {
-            scope_object(object, installation_id);
+            map_fields(object, installation_id, ID_FIELDS);
         }
         Value::String(raw) => {
             if let Ok(Value::Object(mut object)) = serde_json::from_str(raw)
-                && scope_object(&mut object, installation_id) > 0
+                && map_fields(&mut object, installation_id, ID_FIELDS) > 0
                 && let Some(encoded) = serialize_ascii_turn_metadata(&Value::Object(object))
             {
                 *raw = encoded;
@@ -131,29 +110,27 @@ fn scope_metadata(value: &mut Value, installation_id: Option<&str>) {
 }
 
 pub(super) fn scope_headers(headers: &mut HeaderMap, installation_id: Option<&str>) {
-    for name in TRACE_FIELDS {
-        headers.remove(*name);
-    }
+    let Some(installation_id) = installation_id else {
+        return;
+    };
     for name in ID_HEADERS {
-        if let Some(installation_id) = installation_id {
-            let values: Vec<_> = headers
-                .get_all(*name)
-                .iter()
-                .map(|value| {
-                    value
-                        .to_str()
-                        .ok()
-                        .filter(|value| !value.is_empty())
-                        .and_then(|value| {
-                            HeaderValue::from_str(&scoped_id(installation_id, value)).ok()
-                        })
-                        .unwrap_or_else(|| value.clone())
-                })
-                .collect();
-            headers.remove(*name);
-            for value in values {
-                headers.append(*name, value);
-            }
+        let values: Vec<_> = headers
+            .get_all(*name)
+            .iter()
+            .map(|value| {
+                value
+                    .to_str()
+                    .ok()
+                    .filter(|value| !value.is_empty())
+                    .and_then(|value| {
+                        HeaderValue::from_str(&scoped_id(installation_id, value)).ok()
+                    })
+                    .unwrap_or_else(|| value.clone())
+            })
+            .collect();
+        headers.remove(*name);
+        for value in values {
+            headers.append(*name, value);
         }
     }
     let values: Vec<_> = headers
@@ -183,18 +160,19 @@ pub(super) fn scope_request(
     installation_id: Option<&str>,
 ) {
     scope_headers(headers, installation_id);
+    let Some(installation_id) = installation_id else {
+        return;
+    };
     if let Some(object) = body.as_object_mut() {
-        scope_object(object, installation_id);
-        if let Some(installation_id) = installation_id {
-            map_fields(object, installation_id, &["prompt_cache_key"]);
-        }
+        map_fields(object, installation_id, ID_FIELDS);
+        map_fields(object, installation_id, &["prompt_cache_key"]);
         for key in METADATA_FIELDS {
             if let Some(value) = object.get_mut(*key) {
                 scope_metadata(value, installation_id);
             }
         }
         if let Some(Value::Object(metadata)) = object.get_mut("client_metadata") {
-            scope_object(metadata, installation_id);
+            map_fields(metadata, installation_id, ID_FIELDS);
             for key in METADATA_FIELDS {
                 if let Some(value) = metadata.get_mut(*key) {
                     scope_metadata(value, installation_id);
@@ -202,9 +180,6 @@ pub(super) fn scope_request(
             }
         }
     }
-    let Some(installation_id) = installation_id else {
-        return;
-    };
     // 只处理官方自动 Git 上下文的关联字段，不扫描普通用户输入和工具输出
     let Ok(targets) = locate_contexts(body, true, || Ok(())) else {
         return;

backend/crates/providers/openai/src/transport/headers.rs

@@ -28,16 +28,15 @@ pub(super) fn is_managed_identity_header(name: &str) -> bool {
             | "chatgpt-project-id"
             | "openai-organization"
             | "openai-project"
-            // 工作区路由和合规属性属于认证账号，不能继承下游账号的值
-            | "x-openai-account-routing-override"
-            | "x-openai-fedramp"
-            // 安装身份由当前账号写入 client_metadata，不继承下游安装头
+            // 安装头与 client_metadata 共用当前账号的安装身份
             | "x-codex-installation-id"
             // 画像和设备证明须属于选定账号，不能由下游覆盖
             | "originator"
             | "user-agent"
             | "version"
+            // 地域和账号路由统一按部署配置生成
             | "x-openai-internal-codex-residency"
+            | "x-openai-account-routing-override"
             | "x-oai-attestation"
             | "x-oai-is"
             | "x-oai-is-update"
@@ -127,6 +126,11 @@ impl CodexBackendClient {
         };
         let mut headers = build_codex_model_headers(profile, context.authorization, account_id)?;
         insert_optional_header(&mut headers, "cookie", cookie_header)?;
+        insert_optional_header(
+            &mut headers,
+            "x-codex-installation-id",
+            context.installation_id,
+        )?;
         Ok(headers)
     }
 

backend/crates/providers/openai/src/transport/request.rs

@@ -469,19 +469,17 @@ pub(crate) fn scope_request_to_account(
     request: &mut CodexResponsesRequest,
     installation_id: &str,
     account_scope: RequestAccountScope,
+    verified_turn_state: Option<&str>,
 ) {
     let reset_account_state = !account_scope.can_reuse_account_state();
     let client_metadata_turn_state = metadata_string(request, "x-codex-turn-state");
-    let preserve_turn_state = !reset_account_state
-        && (request.turn_state.is_some() || client_metadata_turn_state.is_some());
-    let turn_state = preserve_turn_state
-        .then(|| request.turn_state.clone())
-        .flatten();
-    let client_metadata_turn_state = if preserve_turn_state {
-        client_metadata_turn_state
+    let turn_state = if reset_account_state {
+        verified_turn_state.map(ToOwned::to_owned)
     } else {
-        None
+        request.turn_state.clone()
     };
+    let client_metadata_turn_state = client_metadata_turn_state
+        .filter(|token| !reset_account_state || Some(token.as_str()) == verified_turn_state);
     let turn_metadata = request
         .turn_metadata
         .as_deref()

backend/crates/providers/openai/tests/compat/headers.rs

@@ -45,6 +45,7 @@ const DOWNSTREAM_CLIENT_HEADERS: &[(&str, &str)] = &[
 ];
 
 const BUSINESS_EXTENSION_HEADERS: &[(&str, &str)] = &[
+    ("x-openai-fedramp", "true"),
     ("x-grok-turn-idx", "7"),
     ("X-Grok-Model-Override", "synthetic-model"),
     ("x-grok-session-id", "synthetic-grok-session"),
@@ -77,6 +78,10 @@ fn request_with_opaque_headers(use_websocket: bool) -> CodexResponsesRequest {
                     "x-openai-internal-codex-residency",
                     STANDARD.encode(b"future-region")
                 ],
+                [
+                    "X-OpenAI-Account-Routing-Override",
+                    STANDARD.encode(b"client-route")
+                ],
                 ["x-codex-turn-state", STANDARD.encode(b"turn-ascii")],
                 ["x-codex-turn-state", STANDARD.encode(b"turn-\x80")],
                 ["bad header name", STANDARD.encode(b"ignored")],
@@ -202,7 +207,10 @@ async fn backend_http_should_preserve_business_headers_without_downstream_transp
     });
     // 不经 API 解码，直接构造协议上下文，验证 Provider 自身的过滤边界
     let request = request_with_opaque_headers(false);
-    let profile = test_wire_profile();
+    let mut configured_profile = test_wire_profile().snapshot();
+    configured_profile.account_routing_override = Some("configured-route".into());
+    let profile =
+        provider_openai::transport::profile::CodexWireProfileState::new(configured_profile);
     let profile_snapshot = profile.snapshot();
     let expected_user_agent = profile_snapshot.user_agent();
     let expected_core_version = profile_snapshot.codex_version;
@@ -277,7 +285,18 @@ async fn backend_http_should_preserve_business_headers_without_downstream_transp
     ] {
         assert_eq!(raw_header_values(&raw, name), vec![value.to_vec()]);
     }
-    assert!(raw_header_values(&raw, "traceparent").is_empty());
+    assert_eq!(
+        raw_header_values(&raw, "traceparent"),
+        vec![b"synthetic-trace".to_vec()]
+    );
+    assert_eq!(
+        raw_header_values(&raw, "x-openai-account-routing-override"),
+        vec![b"configured-route".to_vec()]
+    );
+    assert_eq!(
+        raw_header_values(&raw, "x-codex-installation-id"),
+        vec![b"lease-installation".to_vec()]
+    );
     let session = raw_header_values(&raw, "session-id");
     assert!(uuid::Uuid::parse_str(std::str::from_utf8(&session[0]).unwrap()).is_ok());
     // 指纹头由运行时画像生成，不透传客户端值
@@ -295,7 +314,6 @@ async fn backend_http_should_preserve_business_headers_without_downstream_transp
     );
     for dropped in [
         "openai-beta",
-        "x-codex-installation-id",
         "x-openai-actor-authorization",
         "x-oai-attestation",
         "x-oai-is",
@@ -358,13 +376,15 @@ async fn backend_websocket_should_preserve_business_headers_without_downstream_t
             .expect("send opaque terminal event");
     });
     let request = request_with_opaque_headers(true);
+    let mut profile = test_wire_profile().snapshot();
+    profile.account_routing_override = Some("configured-route".into());
     let client = CodexBackendClient::new(
         reqwest::Client::builder()
             .no_proxy()
             .build()
             .expect("HTTP client"),
         format!("http://{address}"),
-        test_wire_profile(),
+        provider_openai::transport::profile::CodexWireProfileState::new(profile),
     )
     .with_websocket_pool(Arc::new(CodexWebSocketPool::new(Duration::from_mins(1))));
 
@@ -430,7 +450,6 @@ async fn backend_websocket_should_preserve_business_headers_without_downstream_t
     assert_eq!(values("version"), vec![b"1.2.3".to_vec()]);
     assert!(values("x-openai-internal-codex-residency").is_empty());
     for dropped in [
-        "x-codex-installation-id",
         "x-openai-actor-authorization",
         "x-oai-attestation",
         "x-oai-is",
@@ -462,7 +481,15 @@ async fn backend_websocket_should_preserve_business_headers_without_downstream_t
     ] {
         assert_eq!(values(name), vec![value.to_vec()], "missing {name}");
     }
-    assert!(values("traceparent").is_empty());
+    assert_eq!(values("traceparent"), vec![b"synthetic-trace".to_vec()]);
+    assert_eq!(
+        values("x-openai-account-routing-override"),
+        vec![b"configured-route".to_vec()]
+    );
+    assert_eq!(
+        values("x-codex-installation-id"),
+        vec![b"lease-installation".to_vec()]
+    );
     let session = values("session-id");
     assert!(uuid::Uuid::parse_str(std::str::from_utf8(&session[0]).unwrap()).is_ok());
     assert_eq!(values("x-codex-turn-state"), vec![b"turn-ascii".to_vec()]);

backend/crates/providers/openai/tests/provider/contract/account_isolation.rs

@@ -71,12 +71,15 @@ fn assert_isolated_request(
     body: &Value,
     owner: Option<&str>,
 ) {
-    for name in ["x-openai-account-routing-override", "x-openai-fedramp"] {
-        assert!(
-            !headers.contains_key(name),
-            "leaked {name} with owner {owner:?}"
-        );
-    }
+    assert!(!headers.contains_key("x-openai-account-routing-override"));
+    assert_eq!(
+        headers
+            .get_all("x-openai-fedramp")
+            .iter()
+            .map(|value| value.to_str().unwrap())
+            .collect::<Vec<_>>(),
+        ["true", "false"]
+    );
     assert_eq!(headers["x-codex-guardian"], "true");
     assert_eq!(
         headers
@@ -112,7 +115,7 @@ fn assert_isolated_request(
 }
 
 #[tokio::test]
-async fn http_isolates_account_headers_and_only_scopes_guardian_parent_reference() {
+async fn http_applies_deployment_routing_and_only_scopes_guardian_parent_reference() {
     for owner in [Some(SELECTED_ACCOUNT), Some("acct_scope_old"), None] {
         let store = Arc::new(MemoryAccountStore::default());
         create_account(&store, SELECTED_ACCOUNT).await;
@@ -155,7 +158,7 @@ async fn http_isolates_account_headers_and_only_scopes_guardian_parent_reference
 }
 
 #[tokio::test]
-async fn websocket_isolates_account_headers_and_only_scopes_guardian_parent_reference() {
+async fn websocket_applies_deployment_routing_and_only_scopes_guardian_parent_reference() {
     for owner in [Some(SELECTED_ACCOUNT), Some("acct_scope_old"), None] {
         let store = Arc::new(MemoryAccountStore::default());
         create_account(&store, SELECTED_ACCOUNT).await;

backend/crates/providers/openai/tests/provider/contract/correlation.rs

@@ -136,15 +136,35 @@ async fn account_correlation_preserves_turns_across_transports_and_account_switc
                 event.unwrap();
             }
             let (headers, body) = capture.await;
-            assert!(!headers.contains_key("traceparent"));
-            assert!(!headers.contains_key("tracestate"));
-            assert!(!headers.contains_key("sentry-trace"));
-            assert!(!headers.contains_key("baggage"));
+            assert_eq!(
+                headers["traceparent"],
+                "00-11111111111111111111111111111111-2222222222222222-01"
+            );
+            assert_eq!(headers["tracestate"], "vendor=source");
+            assert_eq!(headers["sentry-trace"], "source-trace");
+            assert_eq!(
+                headers
+                    .get_all("baggage")
+                    .iter()
+                    .map(|value| value.to_str().unwrap())
+                    .collect::<Vec<_>>(),
+                ["sentry-environment=synthetic", "sentry-release=synthetic"]
+            );
             let flat = &body["client_metadata"];
-            assert!(flat.get("ws_request_header_traceparent").is_none());
-            assert!(flat.get("ws_request_header_tracestate").is_none());
-            assert!(flat.get("ws_request_header_sentry-trace").is_none());
-            assert!(flat.get("ws_request_header_baggage").is_none());
+            assert_eq!(
+                flat["ws_request_header_traceparent"],
+                headers["traceparent"].to_str().unwrap()
+            );
+            assert_eq!(flat["ws_request_header_tracestate"], "vendor=source");
+            assert_eq!(flat["ws_request_header_sentry-trace"], "source-trace");
+            assert_eq!(
+                flat["ws_request_header_baggage"],
+                "sentry-environment=synthetic"
+            );
+            assert_eq!(
+                headers["x-codex-installation-id"].to_str().unwrap(),
+                flat["x-codex-installation-id"]
+            );
             let header: Value =
                 serde_json::from_str(headers["x-codex-turn-metadata"].to_str().unwrap()).unwrap();
             let full: Value =
@@ -163,8 +183,8 @@ async fn account_correlation_preserves_turns_across_transports_and_account_switc
             assert_eq!(flat["x-client-turn-id"], "source-turn");
             assert_eq!(flat["turn_id"], "source-turn");
             for metadata in [&header, &full] {
-                assert!(metadata.get("sentry-trace").is_none());
-                assert!(metadata.get("baggage").is_none());
+                assert_eq!(metadata["sentry-trace"], "source-trace");
+                assert_eq!(metadata["baggage"], "sentry-environment=synthetic");
                 assert_eq!(metadata["turn_id"], "source-turn");
                 assert_eq!(metadata["parent_turn_id"], "source-turn");
                 assert_eq!(metadata["root_turn_id"], "source-turn");

backend/crates/providers/openai/tests/provider/contract/mod.rs

@@ -20,6 +20,7 @@ mod search_model_access;
 mod selection_retry;
 mod session_binding;
 mod timing;
+mod turn_state;
 mod upstream_adapter;
 
 use std::collections::{BTreeMap, BTreeSet};
@@ -6009,7 +6010,7 @@ async fn opaque_provider_options_do_not_change_openai_account_selection() {
 }
 
 #[tokio::test]
-async fn http_account_scoping_drops_downstream_installation_header() {
+async fn http_account_scoping_replaces_downstream_installation_header() {
     for owner in ["acct_scope_same", "acct_scope_old"] {
         let raw = r#"{"installation_id":"client-installation","future":true}"#;
         let request = capture_scoped_http_request(
@@ -6050,7 +6051,10 @@ async fn http_account_scoping_drops_downstream_installation_header() {
         assert_ne!(installation_id, "client-installation");
         assert!(uuid::Uuid::parse_str(installation_id).is_ok());
         assert_eq!(body["installation_id"], installation_id);
-        assert!(captured_header_values(&request, "x-codex-installation-id").is_empty());
+        assert_eq!(
+            captured_header_values(&request, "x-codex-installation-id"),
+            vec![installation_id.as_bytes().to_vec()]
+        );
         let headers = captured_header_values(&request, "x-codex-turn-metadata");
         assert_eq!(headers.len(), 1);
         for encoded in [
@@ -6068,7 +6072,7 @@ async fn http_account_scoping_drops_downstream_installation_header() {
 }
 
 #[tokio::test]
-async fn websocket_account_scoping_drops_downstream_installation_header() {
+async fn websocket_account_scoping_replaces_downstream_installation_header() {
     for owner in ["acct_scope_same", "acct_scope_old"] {
         let raw = r#"{"installation_id":"client-installation","future":true}"#;
         let store = Arc::new(MemoryAccountStore::default());
@@ -6147,7 +6151,7 @@ async fn websocket_account_scoping_drops_downstream_installation_header() {
         assert_ne!(installation_id, "client-installation");
         assert!(uuid::Uuid::parse_str(installation_id).is_ok());
         assert_eq!(body["installation_id"], installation_id);
-        assert!(!header.contains_key("x-codex-installation-id"));
+        assert_eq!(header["x-codex-installation-id"], installation_id);
         for encoded in [
             header["x-codex-turn-metadata"]
                 .to_str()
@@ -6219,7 +6223,10 @@ async fn same_account_scope_preserves_future_protocol_shapes() {
         body.pointer("/client_metadata/x-codex-installation-id")
     );
     assert!(body.pointer("/client_metadata/installation_id").is_none());
-    assert!(captured_header_values(&request, "x-codex-installation-id").is_empty());
+    assert_eq!(
+        request.headers["x-codex-installation-id"].to_str().unwrap(),
+        body["client_metadata"]["x-codex-installation-id"]
+    );
 }
 
 #[tokio::test]

backend/crates/providers/openai/tests/provider/contract/passthrough.rs

@@ -85,6 +85,13 @@ async fn independent_endpoints_send_opaque_headers_without_client_identity() {
     let entries: Vec<_> = [
         ("x-future-business", "first"),
         ("x-future-business", "second"),
+        ("x-codex-installation-id", "client-installation"),
+        ("x-openai-account-routing-override", "client-route"),
+        ("x-openai-fedramp", "true"),
+        ("traceparent", "client-trace"),
+        ("tracestate", "vendor=client"),
+        ("sentry-trace", "sentry-client-trace"),
+        ("baggage", "environment=synthetic"),
         ("authorization", "Bearer client-secret"),
         ("chatgpt-account-id", "client-account"),
         ("x-stainless-runtime", "client-runtime"),
@@ -134,6 +141,21 @@ async fn independent_endpoints_send_opaque_headers_without_client_identity() {
         let requests = server.received_requests().await.unwrap();
         let sent = requests.last().unwrap();
         assert_eq!(sent.url.path(), endpoint);
+        uuid::Uuid::parse_str(sent.headers["x-codex-installation-id"].to_str().unwrap()).unwrap();
+        assert!(
+            !sent
+                .headers
+                .contains_key("x-openai-account-routing-override")
+        );
+        for (name, expected) in [
+            ("x-openai-fedramp", "true"),
+            ("traceparent", "client-trace"),
+            ("tracestate", "vendor=client"),
+            ("sentry-trace", "sentry-client-trace"),
+            ("baggage", "environment=synthetic"),
+        ] {
+            assert_eq!(sent.headers[name], expected);
+        }
         if endpoint.ends_with("search") {
             assert_scoped_search_body(&sent.body, &raw);
         } else {

backend/crates/providers/openai/tests/provider/contract/privacy.rs

@@ -291,9 +291,9 @@ async fn privacy_header_change_isolates_websocket_then_reuses_stable_headers() {
         let mut sockets = Vec::new();
         let mut cache_keys = Vec::new();
         let mut session_headers = Vec::new();
-        for (expected, requests) in [
-            ("Bearer first-synthetic-token", 1),
-            ("Bearer second-synthetic-token", 2),
+        for (expected, trace, requests) in [
+            ("Bearer first-synthetic-token", "first-trace", 1),
+            ("Bearer second-synthetic-token", "second-trace", 2),
         ] {
             let (stream, _) = timeout(Duration::from_secs(5), listener.accept())
                 .await
@@ -302,8 +302,11 @@ async fn privacy_header_change_isolates_websocket_then_reuses_stable_headers() {
             let mut socket =
                 crate::transport::accept_codex_test_websocket_with(stream, |request, response| {
                     assert_eq!(request.headers()["authorization"], expected);
-                    assert!(!request.headers().contains_key("sentry-trace"));
-                    assert!(!request.headers().contains_key("baggage"));
+                    assert_eq!(request.headers()["sentry-trace"], trace);
+                    assert_eq!(
+                        request.headers()["baggage"],
+                        format!("sentry-trace_id={trace}")
+                    );
                     session_headers.push(request.headers()["session-id"].clone());
                     response.headers_mut().insert(
                         "sec-websocket-extensions",

backend/crates/providers/openai/tests/provider/contract/turn_state.rs

@@ -0,0 +1,415 @@
+//! 验证同轮全量请求回传的 token 归属、失败路径与账号状态隔离
+
+use super::*;
+
+const TOKEN: &str = "upstream-turn-state";
+
+fn full_body() -> Value {
+    json!({
+        "model": "gpt-5.4", "store": false,
+        "input": [{"role": "user", "content": "hello"}],
+        "client_metadata": {
+            "session_id": "turn-state-session",
+            "thread_id": "turn-state-thread",
+            "turn_id": "turn-state-turn"
+        }
+    })
+}
+
+fn full_request(body: Value, token: Option<&str>, websocket: bool) -> Operation {
+    let mut protocol_context = Map::from_iter([
+        ("use_websocket".to_owned(), json!(websocket)),
+        (
+            "session_id".to_owned(),
+            body["client_metadata"]["session_id"].clone(),
+        ),
+        (
+            "thread_id".to_owned(),
+            body["client_metadata"]["thread_id"].clone(),
+        ),
+        (
+            "turn_metadata".to_owned(),
+            json!(body["client_metadata"].to_string()),
+        ),
+    ]);
+    if let Some(token) = token {
+        protocol_context.insert("turn_state".to_owned(), json!(token));
+        protocol_context.insert(
+            "opaque_request_headers".to_owned(),
+            json!([["x-codex-turn-state", STANDARD.encode(token)]]),
+        );
+    }
+    Operation::Generate(GenerateRequest::from_protocol_payload(
+        ProtocolPayload::json_object("openai", body.as_object().unwrap().clone())
+            .unwrap()
+            .with_context(protocol_context),
+    ))
+}
+
+async fn collect_response(
+    provider: &Arc<CodexProvider>,
+    operation: Operation,
+    context: AttemptContext,
+) -> Result<bool, gateway_core::error::ProviderError> {
+    let mut stream = provider
+        .clone()
+        .execute(planned_request("openai", operation), context)
+        .await?;
+    let mut observed_token = false;
+    while let Some(event) = stream.next().await {
+        let event = event?;
+        if let Some(observation) = event.response_observation() {
+            observed_token |= observation.client_headers().iter().any(|header| {
+                header.name() == "x-codex-turn-state" && header.value().as_ref() == TOKEN.as_bytes()
+            });
+        }
+    }
+    Ok(observed_token)
+}
+
+async fn http_fixture() -> (Arc<MemoryAccountStore>, MockServer, Arc<CodexProvider>) {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_session_affinity").await;
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .insert_header("x-codex-turn-state", TOKEN)
+                .set_body_string(CAPTURE_COMPLETED_SSE),
+        )
+        .mount(&server)
+        .await;
+    let provider = provider_with_base_url(&store, server.uri());
+    (store, server, provider)
+}
+
+#[tokio::test]
+async fn same_turn_full_http_request_preserves_observed_token_without_restoring_other_account_state()
+ {
+    let (_store, server, provider) = http_fixture().await;
+    assert!(
+        collect_response(
+            &provider,
+            full_request(full_body(), None, false),
+            context("req_turn_first", CancellationToken::new())
+        )
+        .await
+        .unwrap()
+    );
+    let mut body = full_body();
+    body["client_metadata"]["x-codex-turn-state"] = json!(TOKEN);
+    body["client_metadata"]["response_id"] = json!("unowned-response");
+    body["client_metadata"]["parent_response_id"] = json!("unowned-parent");
+    body["client_metadata"]["account_id"] = json!("unowned-account");
+    collect_response(
+        &provider,
+        full_request(body, Some(TOKEN), false),
+        context("req_turn_echo", CancellationToken::new()),
+    )
+    .await
+    .unwrap();
+    let requests = server.received_requests().await.unwrap();
+    assert_eq!(requests.len(), 2);
+    assert_eq!(requests[1].headers["x-codex-turn-state"], TOKEN);
+    let body = captured_request_body(&requests[1]);
+    assert_eq!(body["client_metadata"]["x-codex-turn-state"], TOKEN);
+    assert!(body.get("previous_response_id").is_none());
+    for field in ["response_id", "parent_response_id", "account_id"] {
+        assert!(
+            body["client_metadata"].get(field).is_none(),
+            "{field} must remain scoped"
+        );
+    }
+}
+
+#[tokio::test]
+async fn verified_header_echo_does_not_authorize_a_different_metadata_token() {
+    let (_store, server, provider) = http_fixture().await;
+    collect_response(
+        &provider,
+        full_request(full_body(), None, false),
+        context("req_turn_first", CancellationToken::new()),
+    )
+    .await
+    .unwrap();
+    let mut body = full_body();
+    body["client_metadata"]["x-codex-turn-state"] = json!("unobserved-token");
+    collect_response(
+        &provider,
+        full_request(body, Some(TOKEN), false),
+        context("req_turn_echo", CancellationToken::new()),
+    )
+    .await
+    .unwrap();
+    let requests = server.received_requests().await.unwrap();
+    assert_eq!(requests[1].headers["x-codex-turn-state"], TOKEN);
+    assert!(
+        captured_request_body(&requests[1])["client_metadata"]
+            .get("x-codex-turn-state")
+            .is_none()
+    );
+}
+
+#[tokio::test]
+async fn turn_state_from_failed_http_response_is_owned_before_a_terminal_success() {
+    let (_store, server, provider) = http_fixture().await;
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(ResponseTemplate::new(400)
+            .insert_header("x-codex-turn-state", TOKEN)
+            .set_body_json(json!({"error": {"type": "invalid_request_error", "message": "Invalid request."}})))
+        .up_to_n_times(1)
+        .with_priority(1)
+        .mount(&server).await;
+    assert!(
+        collect_response(
+            &provider,
+            full_request(full_body(), None, false),
+            context("req_turn_first", CancellationToken::new())
+        )
+        .await
+        .is_err()
+    );
+    collect_response(
+        &provider,
+        full_request(full_body(), Some(TOKEN), false),
+        context("req_turn_echo", CancellationToken::new()),
+    )
+    .await
+    .unwrap();
+    let requests = server.received_requests().await.unwrap();
+    assert_eq!(requests.len(), 2);
+    assert_eq!(requests[1].headers["x-codex-turn-state"], TOKEN);
+}
+
+#[tokio::test]
+async fn turn_state_proof_does_not_authorize_an_unowned_previous_response_id() {
+    let (_store, server, provider) = http_fixture().await;
+    collect_response(
+        &provider,
+        full_request(full_body(), None, false),
+        context("req_turn_first", CancellationToken::new()),
+    )
+    .await
+    .unwrap();
+    let mut body = full_body();
+    body["previous_response_id"] = json!("unowned-response");
+    let error = collect_response(
+        &provider,
+        full_request(body, Some(TOKEN), false),
+        context("req_turn_echo", CancellationToken::new()),
+    )
+    .await
+    .unwrap_err();
+    assert_eq!(
+        error
+            .client_visible_upstream_error()
+            .and_then(|detail| detail.code()),
+        Some("previous_response_not_found")
+    );
+    assert_eq!(server.received_requests().await.unwrap().len(), 1);
+}
+
+#[tokio::test]
+async fn full_request_turn_state_requires_matching_client_account_thread_turn_and_token() {
+    for case in [
+        "new_turn",
+        "missing_turn",
+        "new_thread",
+        "new_session",
+        "new_key",
+        "new_account",
+        "unknown_token",
+        "conflicting_owner",
+        "expired",
+        "new_provider",
+    ] {
+        let (store, server, provider) = http_fixture().await;
+        let first_context = if case == "expired" {
+            AttemptContext::new(
+                RequestAttemptContext::new(
+                    ModelRequestId::new("req_turn_first").unwrap(),
+                    ClientApiKeyId::new("key_openai_contract").unwrap(),
+                ),
+                NonZeroU32::new(1).unwrap(),
+                SystemTime::now() + Duration::from_secs(30),
+                account_policy().with_openai_session_affinity_ttl(Duration::ZERO),
+                AccountAttemptContext::new(BTreeSet::new(), None, None)
+                    .with_account_scope(contract_account_scope()),
+                None,
+                CancellationToken::new(),
+            )
+        } else if case == "new_account" {
+            diagnostic_context("req_turn_first", "acct_session_affinity")
+        } else {
+            context("req_turn_first", CancellationToken::new())
+        };
+        collect_response(
+            &provider,
+            full_request(full_body(), None, false),
+            first_context,
+        )
+        .await
+        .unwrap();
+        let mut body = full_body();
+        let mut token = TOKEN;
+        let mut echo_context = context("req_turn_echo", CancellationToken::new());
+        let provider = if case == "new_provider" {
+            provider_with_base_url(&store, server.uri())
+        } else {
+            provider
+        };
+        match case {
+            "new_turn" => body["client_metadata"]["turn_id"] = json!("another-turn"),
+            "missing_turn" => {
+                body["client_metadata"]
+                    .as_object_mut()
+                    .unwrap()
+                    .remove("turn_id");
+            }
+            "new_thread" => body["client_metadata"]["thread_id"] = json!("another-thread"),
+            "new_session" => {
+                body["client_metadata"]["session_id"] = json!("another-session");
+                echo_context = diagnostic_context("req_turn_echo", "acct_session_affinity");
+            }
+            "unknown_token" => token = "unobserved-token",
+            "new_account" => {
+                create_account(&store, "acct_scope_new").await;
+                echo_context = diagnostic_context("req_turn_echo", "acct_scope_new");
+            }
+            "conflicting_owner" => {
+                echo_context = context_with_state_owner("req_turn_echo", "acct_scope_old")
+            }
+            "new_key" => {
+                echo_context = AttemptContext::new(
+                    RequestAttemptContext::new(
+                        ModelRequestId::new("req_turn_echo").unwrap(),
+                        ClientApiKeyId::new("another-key").unwrap(),
+                    ),
+                    NonZeroU32::new(1).unwrap(),
+                    SystemTime::now() + Duration::from_secs(30),
+                    account_policy(),
+                    AccountAttemptContext::diagnostic(
+                        BTreeSet::new(),
+                        ProviderAccountId::new("acct_session_affinity").unwrap(),
+                        None,
+                    ),
+                    None,
+                    CancellationToken::new(),
+                );
+            }
+            _ => {}
+        }
+        body["client_metadata"]["x-codex-turn-state"] = json!(token);
+        collect_response(
+            &provider,
+            full_request(body, Some(token), false),
+            echo_context,
+        )
+        .await
+        .unwrap_or_else(|error| panic!("{case}: {error:?}"));
+        let requests = server.received_requests().await.unwrap();
+        assert_eq!(requests.len(), 2, "{case}");
+        assert!(
+            !requests[1].headers.contains_key("x-codex-turn-state"),
+            "{case}"
+        );
+        assert!(
+            captured_request_body(&requests[1])["client_metadata"]
+                .get("x-codex-turn-state")
+                .is_none(),
+            "{case}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn websocket_metadata_turn_state_survives_full_replay_after_completion_or_failure() {
+    for fail_first in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_session_affinity").await;
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base_url = format!("http://{}", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut websocket = accept_codex_test_websocket(stream).await;
+            websocket.next().await.unwrap().unwrap();
+            websocket
+                .send(Message::Text(
+                    json!({
+                        "type": "response.metadata", "headers": {"x-codex-turn-state": [TOKEN]}
+                    })
+                    .to_string()
+                    .into(),
+                ))
+                .await
+                .unwrap();
+            if fail_first {
+                websocket.close(None).await.unwrap();
+                let (stream, _) = listener.accept().await.unwrap();
+                websocket = accept_codex_test_websocket(stream).await;
+            } else {
+                websocket
+                    .send(Message::Text(
+                        CAPTURE_COMPLETED_SSE
+                            .lines()
+                            .find_map(|line| line.strip_prefix("data: "))
+                            .unwrap()
+                            .to_owned()
+                            .into(),
+                    ))
+                    .await
+                    .unwrap();
+            }
+            let echo: Value = loop {
+                match websocket.next().await.unwrap().unwrap() {
+                    Message::Text(text) => break serde_json::from_str(&text).unwrap(),
+                    Message::Ping(payload) => websocket.send(Message::Pong(payload)).await.unwrap(),
+                    Message::Close(_) => {
+                        let (stream, _) = listener.accept().await.unwrap();
+                        websocket = accept_codex_test_websocket(stream).await;
+                    }
+                    other => panic!("expected response.create, got {other:?}"),
+                }
+            };
+            websocket
+                .send(Message::Text(
+                    CAPTURE_COMPLETED_SSE
+                        .lines()
+                        .find_map(|line| line.strip_prefix("data: "))
+                        .unwrap()
+                        .to_owned()
+                        .into(),
+                ))
+                .await
+                .unwrap();
+            echo
+        });
+        let provider = provider_with_base_url(&store, base_url);
+        let first = collect_response(
+            &provider,
+            full_request(full_body(), None, true),
+            context("req_turn_first", CancellationToken::new()),
+        )
+        .await;
+        assert_eq!(first.is_err(), fail_first);
+        collect_response(
+            &provider,
+            full_request(full_body(), Some(TOKEN), true),
+            context("req_turn_echo", CancellationToken::new()),
+        )
+        .await
+        .unwrap();
+        let echo = timeout(Duration::from_secs(5), server)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            echo["client_metadata"]["x-codex-turn-state"], TOKEN,
+            "failed first response: {fail_first}"
+        );
+        assert!(echo.get("previous_response_id").is_none());
+    }
+}

backend/crates/providers/openai/tests/provider/live.rs

@@ -428,6 +428,10 @@ fn live_protocol_headers() -> Vec<gateway_core::operation::ProviderHttpHeader> {
         ("traceparent", "client-trace"),
         ("tracestate", "client-state"),
         ("ws_request_header_traceparent", "client-ws-trace"),
+        ("x-codex-installation-id", "foreign-installation"),
+        ("X-OpenAI-Account-Routing-Override", "first-route"),
+        ("x-openai-account-routing-override", "second-route"),
+        ("x-openai-fedramp", "true"),
         ("x-codex-turn-state", "foreign-turn-state"),
         ("x-codex-routing-hint", "foreign-route"),
         ("openai-alpha", "live-test"),
@@ -452,6 +456,7 @@ fn assert_live_correlation(headers: &reqwest::header::HeaderMap) {
     uuid::Uuid::parse_str(session).unwrap();
     assert_eq!(headers["thread-id"], session);
     assert_eq!(headers["openai-alpha"], "live-test");
+    assert_eq!(headers["x-openai-fedramp"], "true");
     assert_eq!(
         headers
             .get_all("x-test-business")
@@ -465,9 +470,7 @@ fn assert_live_correlation(headers: &reqwest::header::HeaderMap) {
         "x-oai-attestation",
         "x-forwarded-for",
         "x-stainless-runtime",
-        "traceparent",
-        "tracestate",
-        "ws_request_header_traceparent",
+        "x-openai-account-routing-override",
         "x-codex-turn-state",
         "x-codex-routing-hint",
     ] {
@@ -479,7 +482,14 @@ fn assert_live_correlation(headers: &reqwest::header::HeaderMap) {
     assert_eq!(metadata["future"], true);
     assert_ne!(metadata["installation_id"], "foreign-installation");
     assert!(metadata.get("account_id").is_none());
-    assert!(metadata.get("traceparent").is_none());
+    assert_eq!(metadata["traceparent"], "client-trace");
+    assert_eq!(headers["traceparent"], "client-trace");
+    assert_eq!(headers["tracestate"], "client-state");
+    assert_eq!(headers["ws_request_header_traceparent"], "client-ws-trace");
+    assert_eq!(
+        headers["x-codex-installation-id"].to_str().unwrap(),
+        metadata["installation_id"]
+    );
 }
 
 #[tokio::test]

backend/crates/providers/openai/tests/transport/headers.rs

@@ -226,6 +226,7 @@ async fn backend_websocket_should_forward_context_headers_and_preserve_payload_f
         ("x-codex-parent-thread-id", "parent-456"),
         ("x-openai-subagent", "future_codex_mode"),
         ("x-openai-memgen-request", "true"),
+        ("x-codex-installation-id", "install-123"),
         ("session-id", "cp_derived"),
     ] {
         let value = &headers.iter().find(|(header, _)| header == name).unwrap().1;
@@ -240,7 +241,6 @@ async fn backend_websocket_should_forward_context_headers_and_preserve_payload_f
         }
     }
     for forbidden in [
-        "x-codex-installation-id",
         "x-codex-turn-id",
         "x-openai-internal-codex-residency",
         "accept",
@@ -365,8 +365,11 @@ async fn backend_http_should_send_codex_context_without_browser_headers() {
     ] {
         assert!(header_names.iter().any(|name| name == required));
     }
+    assert_eq!(
+        read_header_value(&raw_request, "x-codex-installation-id"),
+        Some("install-1")
+    );
     for forbidden in [
-        "x-codex-installation-id",
         "x-codex-turn-id",
         "x-openai-internal-codex-residency",
         "sec-ch-ua",

backend/crates/providers/openai/tests/transport/profile_contract.rs

@@ -143,7 +143,7 @@ fn deployment_routing_survives_artifact_updates_and_profile_selection() {
 }
 
 #[tokio::test]
-async fn model_transports_apply_configured_routing_and_discard_downstream_values() {
+async fn model_transports_apply_deployment_routing_with_or_without_client_headers() {
     use base64::{Engine as _, engine::general_purpose::STANDARD};
     use gateway_core::operation::{GenerateRequest, ProtocolPayload};
 
@@ -178,7 +178,9 @@ async fn model_transports_apply_configured_routing_and_discard_downstream_values
             [None, Some("NO_CONSTRAINT")],
         ),
     ] {
-        for websocket in [false, true] {
+        for (websocket, has_downstream_headers) in
+            [(false, false), (false, true), (true, false), (true, true)]
+        {
             let (base_url, capture): (
                 String,
                 futures::future::BoxFuture<'static, reqwest::header::HeaderMap>,
@@ -237,6 +239,7 @@ async fn model_transports_apply_configured_routing_and_discard_downstream_values
             );
             let downstream_headers = HEADERS
                 .into_iter()
+                .filter(|_| has_downstream_headers)
                 .flat_map(|name| {
                     ["downstream-first", "downstream-second"]
                         .map(|value| json!([name, STANDARD.encode(value)]))
@@ -270,7 +273,7 @@ async fn model_transports_apply_configured_routing_and_discard_downstream_values
                         .map(|value| value.to_str().unwrap())
                         .collect::<Vec<_>>(),
                     value.into_iter().collect::<Vec<_>>(),
-                    "{document}, websocket={websocket}, {name}",
+                    "{document}, websocket={websocket}, downstream={has_downstream_headers}, {name}",
                 );
             }
         }

deploy/README.md

@@ -392,8 +392,9 @@ PostgreSQL/Redis 启动密码。日常校验使用 `config --quiet`
 位置覆盖只影响 OpenAI 请求中受支持的位置与时区信息，不改变上下文日期、真实出口 IP、epoch 时间戳或系统时区。
 `openai.residency` 和 `openai.account_routing_override` 是独立于客户端身份的启动配置，
 分别指定 `x-openai-internal-codex-residency` 和 `x-openai-account-routing-override` 请求头。
-两项均为可选字符串，填写值原样发送，不校验地域或路由枚举；省略或设为 `null` 时删除对应的下游请求头且不补发。
-示例配置均填写 `us`，修改后重启应用生效。身份与位置字段见 [运行设置 API](../docs/api.md#8-运行设置)
+两项均为可选字符串，配置值原样发送并覆盖客户端同名头，支持自定义地域或路由值。
+省略或设为 `null` 时删除对应请求头。
+修改后重启应用生效。身份与位置字段见 [运行设置 API](../docs/api.md#8-运行设置)
 后台账号操作使用 Provider 自己的官方画像；发布资料刷新不改写用户选择或 `config.yaml`
 
 ### CPR 互联

docs/api.md

@@ -395,10 +395,11 @@ API Key 与 OAuth 共用模拟客户端画像（`User-Agent`、`originator`、`v
 包括会话、线程、Lite 和其他业务扩展头。
 上游认证只来自选中的账号；API Key 不携带 OAuth Cookie、ChatGPT 账号身份或下游的
 `X-OpenAI-Actor-Authorization` 托管认证声明。
-下游的 `x-openai-internal-codex-residency`、`x-openai-account-routing-override`、`x-openai-fedramp`
-也不透传，地域、工作区路由与合规属性不能从原账号继承。
-前两项仅按部署配置 `openai.residency`、`openai.account_routing_override` 的字符串值生成，
-省略或 `null` 时不发送，配置说明见 [部署与运维](../deploy/README.md#启动后的运行设置)
+`x-codex-installation-id` 使用选中账号已生成并持久化的安装 ID，与正文 `client_metadata` 中的安装身份一致。
+`x-openai-internal-codex-residency` 和 `x-openai-account-routing-override` 分别由部署配置
+`openai.residency`、`openai.account_routing_override` 控制，配置值原样发送并覆盖客户端同名头，
+省略或 `null` 时删除对应请求头。配置说明见 [部署与运维](../deploy/README.md#启动后的运行设置)
+`x-openai-fedramp` 按业务头透传，保留原始多值。
 上述过滤描述原生转发；已信任插件的显式 header 改写可以覆盖账号与画像生成的默认头，
 不按认证或会话字段名称拦截，边界见 [SDK 请求中间件](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#模型请求与-attempt)
 
@@ -409,8 +410,8 @@ Responses 也不透传 `x-stainless-*`、`Origin`、`Referer`、`sec-ch-ua*` 和
 两者同时存在时仍优先使用 `session-id`。正文中的 `client_metadata.session_id`、
 `prompt_cache_key` 不受这条请求头规则影响。
 `thread-id`、turn metadata 等 Codex 协议字段及未知业务扩展不受下游环境头过滤规则影响；
-出站移除下游的 `traceparent`、`tracestate`、`sentry-trace`、`baggage`，
-以及 WS metadata 中对应的 `ws_request_header_*` 副本
+`traceparent`、`tracestate`、`sentry-trace`、`baggage` 及 WS metadata 中对应的
+`ws_request_header_*` 副本保持原值
 
 会话、线程、窗口及父子线程引用在出站时按本次安装身份稳定映射为 UUID；
 默认使用账号安装身份，同账号保留引用关系，换账号使用不同标识。HTTP 请求头、正文 metadata、WS 副本及
@@ -425,6 +426,11 @@ Images 的 `x-codex-image-turn-id` 和 Desktop 自动 Git 上下文的 `clientUs
 续接 response ID、历史 item/call ID、业务正文和未知扩展不参与映射，
 该处理减少显式追踪标识的跨账号复用，不保证会话内容或业务 metadata 无法关联
 
+`x-codex-turn-state` 是上游返回的轮次路由 token，同一轮内应原样回传，开始新轮次时清除。
+全量请求没有 `previous_response_id` 时，代理也会保留已确认属于当前 Client Key、账号、会话、线程和轮次的回传值；
+未知归属、跨账号或跨轮次的值仍会清除。该证明只用于 turn state，不使其他账号状态字段成为可信续接上下文。
+归属证明为进程内有界缓存，实例切换、重启、过期或淘汰后，缺少其他可信会话状态的请求按未知归属处理
+
 #### 正文兼容
 
 OpenAI Provider 的 Responses 上游编码会移除 Codex 不接受的顶层 `temperature`、`max_output_tokens` 和
@@ -911,7 +917,7 @@ CPR 默认路径为 `/v1/cpr/link`，使用同一数据面入口认证，不接
 引导、sideband 与 hangup 的业务扩展头默认透传；认证、账号身份、下游环境与传输头遵守
 [请求头与身份](#请求头与身份)边界。`X-Session-Id`、`Voice-Session-Id` 是 Live 业务会话标识，原样透传。
 `Session-Id`、`Thread-Id` 等已识别追踪标识按选中账号的安装身份稳定映射，同账号各阶段保持一致，
-换账号使用不同标识；客户端 trace 被剥离。创建请求可由 attempt 插件改写本次构造身份，后续 sideband / hangup 仍使用钉住账号的身份。
+换账号使用不同标识，客户端 trace 原样透传。创建请求可由 attempt 插件改写本次构造身份，后续 sideband / hangup 仍使用钉住账号的身份。
 Live 不继承 Responses 的路由提示与续接状态；turn metadata 中的原账号绑定信息清理后保留其他字段。
 真实 call ID 保留用于钉住账号访问通话。HTTP 业务头保留原始多值字节，sideband 握手沿用 WebSocket 可编码的头值
 

docs/architecture.md

@@ -503,6 +503,11 @@ OpenAI Provider 按 Client Key、所选 OAuth 账号和父响应 ID 关联 WS
 缺失或失效时按普通上游路由执行。transport 负责 Cookie 有效期和同 origin、同端点路径约束，
 连接池按快照身份隔离审查连接；复用父连接不会重新计算 Cookie 有效期，Cookie 到期不强制关闭已建立的审查连接
 
+OpenAI Provider 在交付响应头或 WS metadata 前记录 turn state 的归属证明，包括失败响应。
+证明按 Client Key、账号、客户端会话、线程和轮次隔离，API Key 额外绑定凭据版本；只保存摘要与过期时间，
+不保存 token 原值或正文。进程内最多 4096 项，保留时长复用请求的会话亲和策略，已验证回传与上游观测续期；
+满额时清理过期项并淘汰最早过期项。证明缺失时保留既有未知归属处理，不参与选号或授权其他账号状态复用
+
 OpenAI 的 OAuth 与 API Key 共用现有账号和事务。API Key 的 Base URL、密钥和传输策略属于 Provider 凭据 JSON，
 随 credential revision 更新；普通详情只投影非敏感连接设置。API Key 客户端目录按账号、凭据版本和客户端版本隔离，
 标准 `data` 模型列表通过通用画像输出，完整 Codex `models` 目录保留原生对象。通用账号层按 Provider 提交的
@@ -515,7 +520,7 @@ OpenAI 模型目录用于发现与原生 Responses 账号资格筛选，OAuth
 
 - OpenAI 是透明边界。正文兼容、身份归一和显式请求策略之外的业务字段保留原始语义；
   业务响应按原始字节转发，客户端错误兼容与原生续写额度恢复按明确例外处理，见 [OpenAI 透传](api.md#openai-透传)。
-  `transport/correlation` 在出站副本上按本次安装身份统一映射关联标识并剥离客户端 trace；
+  `transport/correlation` 在出站副本上按本次安装身份统一映射关联标识，客户端 trace 原样保留。
   原生安装身份取自账号；插件的身份生成、持久化及关联字段改写通过既有中间件链路自行完成。
   内部选号、亲和、连接池与本地诊断继续读取原始身份；具体字段边界见 [API 协议](api.md#请求头与身份)
   canonical facts 从同一数据旁路提取，用于路由、恢复判断、观测和计费。
@@ -584,6 +589,8 @@ client，OIDC 的 JWKS 缓存与单飞归属对应出口状态。自动刷新提
 
 代理独立保存和测试，通过 `outboundProxyId` 绑定账号；账号保留解析后的 URL 供 Provider 使用。
 出口测试结果属于诊断信息，不作为选择、授权或导入绑定的准入条件。
+双栈出口并发检测，延迟对应首选可用出口的 HTTP 探测耗时（IPv4 优先），不包含另一地址族失败的等待；
+检测仍等待两个地址族完成或超时，以保留双栈可用性结果。两者均失败时，延迟为整体检测耗时。
 代理启停由代理记录独立持有；关闭时保留绑定、连接配置和位置规则，账号执行投影改为直连，
 不使用该代理的位置覆盖，重新启用后恢复原出口。
 连接配置改变时在同一事务内同步关联账号并清除旧测试结果。

release/notes.md

@@ -1,23 +1,16 @@
-# v3.39.0
+# v3.39.1
 
-## Windows 支持
+## OpenAI 兼容性
 
-- 新增 Windows x64 与 ARM64 发行包，静态链接 MSVC 运行库，解压后无需额外安装 VC++ Redistributable
-- 支持 Windows 管理端在线更新、重启与回滚，处理运行中程序替换、重启交接及只读发行文件清理，并校验 Windows 特有的不安全归档路径
-- 补充 Windows 二进制部署说明；运行前需准备 PostgreSQL 和 Redis，使用数据库备份时需提供匹配服务端版本的 `pg_dump`
+- 修复同轮全量请求丢失 `x-codex-turn-state` 的问题：HTTP 与 WebSocket 请求可回传已确认归属的上游轮次 token，包括失败响应中收到的值；继续隔离不同 Client Key、账号、会话、线程和轮次，不放宽其他续接状态的校验
+- 保留客户端的 `traceparent`、`tracestate`、`sentry-trace`、`baggage` 及对应 WebSocket metadata，`x-openai-fedramp` 按业务头原样透传并保留多值
+- 出站 `x-codex-installation-id` 使用所选账号的持久安装身份，与正文中的安装身份一致；地域与账号路由头继续由部署配置控制，配置值覆盖客户端同名头，省略或 `null` 时不发送
 
-## 插件设置
+## 运行稳定性
 
-- 敏感配置支持逐字段修改，补填一项不会清除其他已保存密钥；留空保留现有值，单项清除可在保存前撤销
-- 设置弹窗直接展示各敏感字段的保存状态，支持清理目标版本不再使用的字段，并调整表单与 JSON 切换按钮的尺寸
+- 配置提交发生版本竞争时，在既有有界重试预算内重新读取完整快照并准备扩展，减少创建或撤销 Key 等并发操作造成的短暂不可用；持续竞争或真实存储、扩展故障仍按原有规则暂停快照
+- 修复删除代理返回版本冲突后事务锁未及时释放的问题，避免阻塞紧接着的代理编辑或删除
 
-## 互联与部署
+## 代理检测
 
-- 请求记录、Dashboard 最近请求和账号维度诊断补充 CPR 互联徽标，按账号当前互联摘要展示，不改写请求历史快照
-- 补充小内存部署的 PostgreSQL 预算、应用连接池和 Redis 持久化调优说明，包含生效、验证与回退步骤；默认配置保持不变
-
-## 升级事项
-
-- 插件管理 API 的 `secrets` 改为逐字段更新：字符串设置对应值，`null` 清除对应字段，未出现的字段保留；省略 `secrets` 或传 `{}` 均不修改密钥。依赖整份替换或空对象清空语义的调用方需调整
-- 跨插件版本编辑优先使用目标版本保存的密钥快照，没有快照时沿用当前值，再应用本次字段变更
-- Windows 首次部署请下载对应架构的完整发行包；插件安装包须匹配宿主系统与架构，部署前需验证实际插件运行与上游调用
+- 双栈代理的延迟按首选可用出口的 HTTP 探测耗时显示，IPv4 优先，其次 IPv6，不再把另一地址族失败的等待计入成功出口延迟；检测仍等待两个地址族完成或超时，以保留完整可用性结果

release/version.yaml

@@ -1 +1 @@
-version: "3.39.0"
+version: "3.39.1"