From 7a60bc6dedd09ab75d5a3a7505e783d985008d0f Mon Sep 17 00:00:00 2001 From: Zyy Date: Wed, 21 Oct 2026 18:29:52 +0800 Subject: [PATCH] fix(openai): isolate account correlation identities Scope outgoing session, turn, parent, window and cache identifiers to the selected account across HTTP, WebSocket, Desktop context and Search. Strip client trace propagation and forward account-scoped image turn metadata. Co-authored-by: Codex --- backend/crates/providers/openai/Cargo.toml | 2 +- .../openai/src/provider/execution.rs | 42 ++- .../openai/src/transport/client_json.rs | 6 + .../openai/src/transport/client_sse.rs | 8 +- .../openai/src/transport/correlation.rs | 229 ++++++++++++++ .../providers/openai/src/transport/mod.rs | 1 + .../provider/contract/account_isolation.rs | 2 +- .../tests/provider/contract/correlation.rs | 284 ++++++++++++++++++ .../openai/tests/provider/contract/mod.rs | 68 ++++- .../tests/provider/contract/passthrough.rs | 6 +- .../openai/tests/provider/contract/privacy.rs | 10 +- .../tests/transport/downstream/headers.rs | 10 +- .../openai/tests/transport/headers.rs | 23 +- docs/api.md | 12 +- docs/architecture.md | 5 +- 15 files changed, 649 insertions(+), 59 deletions(-) create mode 100644 backend/crates/providers/openai/src/transport/correlation.rs create mode 100644 backend/crates/providers/openai/tests/provider/contract/correlation.rs diff --git a/backend/crates/providers/openai/Cargo.toml b/backend/crates/providers/openai/Cargo.toml index 9c78f1771..1f33c7169 100644 --- a/backend/crates/providers/openai/Cargo.toml +++ b/backend/crates/providers/openai/Cargo.toml @@ -46,7 +46,7 @@ rustls-native-certs = "=0.8.3" rustls-pki-types = "=1.14.0" secrecy = "0.10.3" serde = { version = "1.0.228", features = ["derive"] } -serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] } +serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order", "raw_value"] } sha2 = "0.11.0" subtle = "2.6.1" tar = { version = "0.4.46", default-features = false } diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 547547e34..0697bdbb2 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -43,7 +43,6 @@ impl CodexProvider { endpoint_path, body: image.payload().body().clone(), image_turn_id, - turn_metadata: None, session_affinity, }, ) @@ -119,12 +118,6 @@ impl CodexProvider { UpstreamSendState::NotSent, )); } - let turn_metadata = search - .payload() - .context() - .get("turn_metadata") - .and_then(Value::as_str) - .map(str::to_owned); let session_affinity = derive_codex_endpoint_session_affinity( search.payload(), context.client_api_key_ref(), @@ -139,7 +132,6 @@ impl CodexProvider { endpoint_path: CODEX_ALPHA_SEARCH_PATH, body: search.payload().body().clone(), image_turn_id: None, - turn_metadata, session_affinity, }, ) @@ -256,7 +248,7 @@ impl CodexProvider { .map_err(map_selection_error)?; } let lease = Arc::new(lease); - let passthrough_headers = match operation { + let payload = match &operation { Operation::GenerateImage(image) => { let Operation::GenerateImage(original) = &request.operation else { return Err(provider_error( @@ -270,14 +262,13 @@ impl CodexProvider { UpstreamSendState::NotSent, )); } - request.body = image.payload().body().clone(); request.image_turn_id = image .payload() .context() .get("image_turn_id") .and_then(Value::as_str) .map(str::to_owned); - crate::transport::request::decode_passthrough_headers(image.payload().context()) + image.payload() } Operation::Search(search) => { if !matches!(&request.operation, Operation::Search(_)) @@ -288,14 +279,7 @@ impl CodexProvider { UpstreamSendState::NotSent, )); } - request.body = search.payload().body().clone(); - request.turn_metadata = search - .payload() - .context() - .get("turn_metadata") - .and_then(Value::as_str) - .map(str::to_owned); - crate::transport::request::decode_passthrough_headers(search.payload().context()) + search.payload() } _ => { return Err(provider_error( @@ -304,6 +288,9 @@ impl CodexProvider { )); } }; + request.body = payload.body().clone(); + let passthrough_headers = + crate::transport::request::decode_passthrough_headers(payload.context()); if !context.is_diagnostic_required_account() && let Some(affinity) = affinity.as_ref() && let Some(turn) = affinity.turn_alias() @@ -336,11 +323,19 @@ impl CodexProvider { account_selection_wait_ms, lease.capacity_snapshot(), )); - // Standalone Provider 端点没有可证明的账号 owner；Search metadata 必须按 + // Standalone Provider 端点没有可证明的账号 owner；Images/Search metadata 必须按 // 跨账号输入收敛到当前 lease，不能沿用下游声明的账号或 installation identity - let turn_metadata = request.turn_metadata.as_deref().and_then(|metadata| { - crate::transport::request::scope_turn_metadata(metadata, lease.installation_id(), true) - }); + let turn_metadata = payload + .context() + .get("turn_metadata") + .and_then(Value::as_str) + .and_then(|metadata| { + crate::transport::request::scope_turn_metadata( + metadata, + lease.installation_id(), + true, + ) + }); let events = cold_json_response_stream(ColdJsonResponse { client: self .client_for_request(&context)? @@ -378,7 +373,6 @@ struct RawJsonEndpointRequest { endpoint_path: &'static str, body: Bytes, image_turn_id: Option, - turn_metadata: Option, session_affinity: Option, } diff --git a/backend/crates/providers/openai/src/transport/client_json.rs b/backend/crates/providers/openai/src/transport/client_json.rs index 5ae6a2c8f..2625635cc 100644 --- a/backend/crates/providers/openai/src/transport/client_json.rs +++ b/backend/crates/providers/openai/src/transport/client_json.rs @@ -25,6 +25,11 @@ impl CodexBackendClient { image_turn_id: Option<&str>, context: CodexRequestContext<'_>, ) -> CodexClientResult { + let body = if endpoint_path == super::endpoints::CODEX_ALPHA_SEARCH_PATH { + super::correlation::scope_search_body(body, context.installation_id) + } else { + body + }; // Provider 端点以 Codex 路径标识；API Key 在自己的 API 前缀下使用对应相对路径 let endpoint_path = if self.protocol == super::client::OpenAiUpstreamProtocol::ResponsesApi { @@ -49,6 +54,7 @@ impl CodexBackendClient { context.turn_metadata, ); self.append_middleware_headers(&mut headers)?; + super::correlation::scope_headers(&mut headers, context.installation_id); let body = if self.privacy.is_some() { let mut value: serde_json::Value = serde_json::from_slice(&body).map_err(CodexClientError::RequestBodyEncode)?; diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index deddce11f..77b88576a 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -130,6 +130,11 @@ impl CodexBackendClient { headers.insert(CONTENT_ENCODING, HeaderValue::from_static("zstd")); } profile.apply_context_versions(&mut upstream_body); + super::correlation::scope_request( + &mut upstream_body, + &mut headers, + context.installation_id, + ); self.apply_privacy(&mut upstream_body, &mut headers)?; let body = serde_json::to_vec(&upstream_body).map_err(CodexClientError::RequestBodyEncode)?; @@ -291,8 +296,9 @@ impl CodexBackendClient { let mut headers = self.request_headers_for_websocket_response(&profile, &websocket_request, context)?; let mut body = serde_json::Value::Object(std::mem::take(websocket_request.body_mut())); - let original_headers = self.privacy.as_ref().map(|_| headers.clone()); profile.apply_context_versions(&mut body); + super::correlation::scope_request(&mut body, &mut headers, context.installation_id); + let original_headers = self.privacy.as_ref().map(|_| headers.clone()); self.apply_privacy(&mut body, &mut headers)?; *websocket_request.body_mut() = body.as_object().cloned().ok_or_else(|| { CodexClientError::Privacy(gateway_core::settings::privacy::PrivacyError { diff --git a/backend/crates/providers/openai/src/transport/correlation.rs b/backend/crates/providers/openai/src/transport/correlation.rs new file mode 100644 index 000000000..97700c2fd --- /dev/null +++ b/backend/crates/providers/openai/src/transport/correlation.rs @@ -0,0 +1,229 @@ +//! 在出站副本中隔离账号的追踪标识，不改变本地亲和、续接引用或业务正文 + +use reqwest::header::{HeaderMap, HeaderValue}; +use serde_json::{Map, Value}; +use sha2::{Digest, Sha256}; + +use super::automatic_context::{locate_contexts, rewrite_git_context}; +use super::request::serialize_ascii_turn_metadata; + +const ID_FIELDS: &[&str] = &[ + "session_id", + "thread_id", + "turn_id", + "window_id", + "context_window_id", + "parent_thread_id", + "parent_turn_id", + "root_turn_id", + "forked_from_thread_id", + "conversation_id", + "x-codex-window-id", + "x-codex-parent-thread-id", + "x-client-request-id", + "x-client-turn-id", + "codexWindowId", + "parentThreadId", +]; +const METADATA_FIELDS: &[&str] = &["turnMetadata", "turn_metadata", "x-codex-turn-metadata"]; +const TRACE_FIELDS: &[&str] = &[ + "traceparent", + "tracestate", + "ws_request_header_traceparent", + "ws_request_header_tracestate", +]; +const ID_HEADERS: &[&str] = &[ + "session-id", + "thread-id", + "x-codex-window-id", + "x-codex-parent-thread-id", + "x-client-request-id", + "x-client-turn-id", + "x-codex-image-turn-id", +]; + +fn scoped_id(installation_id: &str, original: &str) -> String { + let mut digest = Sha256::new(); + digest.update(b"codex-account-correlation-v1\0"); + digest.update(installation_id.as_bytes()); + digest.update(b"\0"); + digest.update(original.as_bytes()); + let hash = digest.finalize(); + let mut bytes = [0_u8; 16]; + bytes.copy_from_slice(&hash[..16]); + // 保持官方 UUID 的形状；各类引用共享映射，保留相同原值的等值关系 + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + uuid::Uuid::from_bytes(bytes).to_string() +} + +/// Search 的 id 是会话关联键，仅替换该值的原始字节，保留其他字段、重复键和空白 +pub(super) fn scope_search_body(body: bytes::Bytes, installation_id: Option<&str>) -> bytes::Bytes { + #[derive(serde::Deserialize)] + struct SearchIdentity<'a> { + #[serde(borrow)] + id: Option<&'a serde_json::value::RawValue>, + } + let Some(installation_id) = installation_id else { + return body; + }; + let Ok(identity) = serde_json::from_slice::>(&body) else { + return body; + }; + let Some(raw) = identity.id.map(serde_json::value::RawValue::get) else { + return body; + }; + let Ok(original) = serde_json::from_str::(raw) else { + return body; + }; + if original.is_empty() { + return body; + } + // from_slice 返回的 RawValue 借用原始正文，跨度不包含相邻键或分隔符 + let offset = raw.as_ptr() as usize - body.as_ptr() as usize; + let replacement = format!("\"{}\"", scoped_id(installation_id, &original)); + let mut scoped = Vec::with_capacity(body.len() - raw.len() + replacement.len()); + scoped.extend_from_slice(&body[..offset]); + scoped.extend_from_slice(replacement.as_bytes()); + scoped.extend_from_slice(&body[offset + raw.len()..]); + scoped.into() +} + +fn map_fields(object: &mut Map, installation_id: &str, fields: &[&str]) -> usize { + let mut changed = 0; + for key in fields { + if let Some(Value::String(value)) = object.get_mut(*key) + && !value.is_empty() + { + *value = scoped_id(installation_id, value); + changed += 1; + } + } + changed +} + +fn scope_object(object: &mut Map, 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>) { + match value { + Value::Object(object) => { + scope_object(object, installation_id); + } + Value::String(raw) => { + if let Ok(Value::Object(mut object)) = serde_json::from_str(raw) + && scope_object(&mut object, installation_id) > 0 + && let Some(encoded) = serialize_ascii_turn_metadata(&Value::Object(object)) + { + *raw = encoded; + } + } + _ => {} + } +} + +pub(super) fn scope_headers(headers: &mut HeaderMap, installation_id: Option<&str>) { + for name in TRACE_FIELDS { + headers.remove(*name); + } + 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("x-codex-turn-metadata") + .iter() + .map(|value| { + let Ok(raw) = value.to_str() else { + return value.clone(); + }; + let mut metadata = Value::String(raw.to_owned()); + scope_metadata(&mut metadata, installation_id); + metadata + .as_str() + .and_then(|raw| HeaderValue::from_str(raw).ok()) + .unwrap_or_else(|| value.clone()) + }) + .collect(); + headers.remove("x-codex-turn-metadata"); + for value in values { + headers.append("x-codex-turn-metadata", value); + } +} + +pub(super) fn scope_request( + body: &mut Value, + headers: &mut HeaderMap, + installation_id: Option<&str>, +) { + scope_headers(headers, installation_id); + 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"]); + } + 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); + for key in METADATA_FIELDS { + if let Some(value) = metadata.get_mut(*key) { + scope_metadata(value, installation_id); + } + } + } + } + let Some(installation_id) = installation_id else { + return; + }; + // 只处理官方自动 Git 上下文的关联字段，不扫描普通用户输入和工具输出 + let Ok(targets) = locate_contexts(body, true, || Ok(())) else { + return; + }; + for target in targets { + let Some(Value::String(text)) = body.pointer_mut(&target.path) else { + continue; + }; + if text.len() > 1024 * 1024 || !target.is_complete(text) { + continue; + } + let _ = rewrite_git_context(text, 1024 * 1024, |value| { + Ok(value.as_object_mut().map_or(0, |object| { + map_fields( + object, + installation_id, + &["sessionId", "threadId", "clientUserMessageId"], + ) + })) + }); + } +} diff --git a/backend/crates/providers/openai/src/transport/mod.rs b/backend/crates/providers/openai/src/transport/mod.rs index 520e2666d..eabb3b22c 100644 --- a/backend/crates/providers/openai/src/transport/mod.rs +++ b/backend/crates/providers/openai/src/transport/mod.rs @@ -8,6 +8,7 @@ mod client_json; mod client_live; mod client_sse; pub(crate) mod connection; +mod correlation; pub mod diagnostics; mod downstream; pub mod endpoints; diff --git a/backend/crates/providers/openai/tests/provider/contract/account_isolation.rs b/backend/crates/providers/openai/tests/provider/contract/account_isolation.rs index 852c6c84f..7d92dc4bc 100644 --- a/backend/crates/providers/openai/tests/provider/contract/account_isolation.rs +++ b/backend/crates/providers/openai/tests/provider/contract/account_isolation.rs @@ -4,7 +4,7 @@ use super::*; const SELECTED_ACCOUNT: &str = "acct_scope_same"; -fn isolation_history() -> Value { +pub(super) fn isolation_history() -> Value { json!([ {"role":"user", "content":"hello"}, {"type":"function_call", "id":"fc_history", "call_id":"call_history", "name":"lookup", "arguments":"{}"}, diff --git a/backend/crates/providers/openai/tests/provider/contract/correlation.rs b/backend/crates/providers/openai/tests/provider/contract/correlation.rs new file mode 100644 index 000000000..3e3da3d4c --- /dev/null +++ b/backend/crates/providers/openai/tests/provider/contract/correlation.rs @@ -0,0 +1,284 @@ +//! 账号追踪域的稳定性、HTTP/WS 一致性、父子引用与工具端点归属回归 + +use super::*; + +fn metadata() -> Value { + json!({"session_id":"source-session", "thread_id":"source-thread", "turn_id":"source-turn", + "parent_thread_id":"source-session", "forked_from_thread_id":"source-session", + "parent_turn_id":"source-turn", "root_turn_id":"source-turn", + "window_id":"source-window", "context_window_id":"source-window", "future":{"keep":true}}) +} + +fn payload(websocket: bool) -> ProtocolPayload { + let mut full = metadata(); + full["tool_namespaces_info"] = json!({"inventory":"body-only"}); + let mut input = account_isolation::isolation_history() + .as_array() + .unwrap() + .clone(); + input.insert(0, json!({"role":"user", "content":[{"type":"input_text", "text":concat!( + "", + r#"{"sessionId":"source-session","threadId":"source-thread","clientUserMessageId":"source-turn","path":"/workspace/project"}"#, + "" + )}], "internal_chat_message_metadata_passthrough":{"content_item_kinds":["additional_content.codex_git_state"]}})); + ProtocolPayload::json_object("openai", json!({ + "model":"gpt-5.4", "input":input, + "prompt_cache_key":"custom-cache-key", "client_metadata":{ + "session_id":"source-session", "thread_id":"source-thread", "turn_id":"source-turn", + "x-codex-window-id":"source-window", "x-codex-parent-thread-id":"source-session", + "x-codex-turn-metadata":full.to_string(), + "ws_request_header_traceparent":"00-11111111111111111111111111111111-2222222222222222-01", + "ws_request_header_tracestate":"vendor=source" + } + }).as_object().unwrap().clone()).unwrap().with_context(Map::from_iter([ + ("use_websocket".into(), json!(websocket)), + ("session_id".into(), json!("source-session")), + ("thread_id".into(), json!("source-thread")), + ("turn_metadata".into(), json!(metadata().to_string())), + ("opaque_request_headers".into(), json!([ + ["traceparent",STANDARD.encode("00-11111111111111111111111111111111-2222222222222222-01")], + ["tracestate",STANDARD.encode("vendor=source")] + ])), + ])) +} + +#[tokio::test] +async fn account_correlation_is_stable_across_transports_and_changes_only_with_account() { + let stores = [ + Arc::new(MemoryAccountStore::default()), + Arc::new(MemoryAccountStore::default()), + ]; + create_account(&stores[0], "acct_scope_same").await; + create_account(&stores[1], "acct_scope_new").await; + let mut captures = Vec::new(); + for account in [0, 0, 1, 0] { + for websocket in [false, true] { + let (base_url, capture): ( + String, + BoxFuture<'static, (reqwest::header::HeaderMap, Value)>, + ) = if websocket { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let task = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut headers = None; + let mut socket = + crate::transport::accept_codex_test_websocket_with(socket, |request, _| { + headers = Some(request.headers().clone()); + }) + .await; + let frame = socket.next().await.unwrap().unwrap(); + let body = serde_json::from_str(frame.to_text().unwrap()).unwrap(); + socket.send(Message::Text(json!({"type":"response.completed", "response":{ + "id":"resp_scoped", "model":"gpt-5.4", "status":"completed", "output":[] + }}).to_string().into())).await.unwrap(); + (headers.unwrap(), body) + }); + (url, Box::pin(async { task.await.unwrap() })) + } else { + 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") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(1) + .mount(&server) + .await; + ( + server.uri(), + Box::pin(async move { + let requests = server.received_requests().await.unwrap(); + ( + requests[0].headers.clone(), + captured_request_body(&requests[0]), + ) + }), + ) + }; + let mut stream = provider_with_base_url(&stores[account], base_url) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload(payload( + websocket, + ))), + ), + context("req_correlation", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = stream.next().await { + event.unwrap(); + } + let (headers, body) = capture.await; + assert!(!headers.contains_key("traceparent")); + assert!(!headers.contains_key("tracestate")); + let flat = &body["client_metadata"]; + assert!(flat.get("ws_request_header_traceparent").is_none()); + assert!(flat.get("ws_request_header_tracestate").is_none()); + let header: Value = + serde_json::from_str(headers["x-codex-turn-metadata"].to_str().unwrap()).unwrap(); + let full: Value = + serde_json::from_str(flat["x-codex-turn-metadata"].as_str().unwrap()).unwrap(); + for (field, original) in [ + ("session_id", "source-session"), + ("thread_id", "source-thread"), + ("turn_id", "source-turn"), + ] { + assert_scoped_id(flat[field].as_str().unwrap(), original); + assert_eq!(flat[field], header[field]); + assert_eq!(flat[field], full[field]); + } + assert_eq!(headers["session-id"].to_str().unwrap(), flat["session_id"]); + assert_eq!(headers["thread-id"].to_str().unwrap(), flat["thread_id"]); + assert_eq!(full["parent_thread_id"], flat["session_id"]); + assert_eq!(full["forked_from_thread_id"], flat["session_id"]); + assert_eq!(full["parent_turn_id"], flat["turn_id"]); + assert_eq!(full["root_turn_id"], flat["turn_id"]); + assert_eq!(full["window_id"], full["context_window_id"]); + assert!(header.get("tool_namespaces_info").is_none()); + assert_eq!(full["tool_namespaces_info"]["inventory"], "body-only"); + assert_eq!(full["future"], metadata()["future"]); + let input = body["input"].as_array().unwrap(); + assert_eq!(json!(&input[1..]), account_isolation::isolation_history()); + let document = + roxmltree::Document::parse(input[0]["content"][0]["text"].as_str().unwrap()) + .unwrap(); + let git: Value = serde_json::from_str( + document + .descendants() + .find(|node| node.has_tag_name("git")) + .unwrap() + .text() + .unwrap(), + ) + .unwrap(); + for (desktop, wire) in [ + ("sessionId", "session_id"), + ("threadId", "thread_id"), + ("clientUserMessageId", "turn_id"), + ] { + assert_eq!(git[desktop], flat[wire]); + } + assert_eq!(git["path"], "/workspace/project"); + assert_scoped_id( + body["prompt_cache_key"].as_str().unwrap(), + "custom-cache-key", + ); + captures.push((flat["session_id"].clone(), body["prompt_cache_key"].clone())); + } + } + for index in [1, 2, 3, 6, 7] { + assert_eq!(captures[0], captures[index]); + } + assert_eq!(captures[4], captures[5]); + assert_ne!(captures[0].0, captures[4].0); + assert_ne!(captures[0].1, captures[4].1); +} + +#[tokio::test] +async fn image_generation_and_edit_share_the_selected_account_turn_identity() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"data":[]}))) + .expect(2) + .mount(&server) + .await; + let provider = provider_with_base_url(&store, server.uri()); + let body = Bytes::from_static(br#"{ "prompt":"test", "future":1, "future":2 }"#); + for kind in [ImageRequestKind::Generation, ImageRequestKind::Edit] { + let operation = Operation::GenerateImage(ImageRequest::from_raw_json( + kind, + RawJsonPayload::new("openai", body.clone()) + .unwrap() + .with_context(Map::from_iter([ + ("image_turn_id".into(), json!("source-turn")), + ("turn_metadata".into(), json!(metadata().to_string())), + ])), + )); + let mut stream = provider + .clone() + .execute( + planned_provider_endpoint_request("openai", operation), + context("req_image_correlation", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = stream.next().await { + event.unwrap(); + } + } + let requests = server.received_requests().await.unwrap(); + let first = &requests[0].headers["x-codex-image-turn-id"]; + assert_scoped_id(first.to_str().unwrap(), "source-turn"); + for request in &requests { + assert_eq!(request.body.as_slice(), body.as_ref()); + assert_eq!(&request.headers["x-codex-image-turn-id"], first); + let metadata: Value = + serde_json::from_str(request.headers["x-codex-turn-metadata"].to_str().unwrap()) + .unwrap(); + assert_eq!(metadata["turn_id"], first.to_str().unwrap()); + assert_eq!(metadata["parent_turn_id"], metadata["turn_id"]); + } +} + +#[tokio::test] +async fn cache_key_preserves_explicit_sharing_and_omission_within_an_account() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").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") + .set_body_string(CAPTURE_COMPLETED_SSE), + ) + .expect(4) + .mount(&server) + .await; + let provider = provider_with_base_url(&store, server.uri()); + for (index, cache) in [Some("shared"), Some("shared"), None, Some("different")] + .into_iter() + .enumerate() + { + let mut body = + json!({"model":"gpt-5.4", "input":[], "session_id":format!("session-{index}")}); + if let Some(cache) = cache { + body["prompt_cache_key"] = json!(cache); + } + let payload = ProtocolPayload::json_object("openai", body.as_object().unwrap().clone()) + .unwrap() + .with_context(Map::from_iter([("use_websocket".into(), json!(false))])); + let mut stream = Arc::clone(&provider) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + ), + context("req_cache_scope", CancellationToken::new()), + ) + .await + .unwrap(); + while let Some(event) = stream.next().await { + event.unwrap(); + } + } + let bodies: Vec<_> = server + .received_requests() + .await + .unwrap() + .iter() + .map(captured_request_body) + .collect(); + assert_scoped_id(bodies[0]["prompt_cache_key"].as_str().unwrap(), "shared"); + assert_eq!(bodies[0]["prompt_cache_key"], bodies[1]["prompt_cache_key"]); + assert_ne!(bodies[0]["session_id"], bodies[1]["session_id"]); + assert!(bodies[2].get("prompt_cache_key").is_none()); + assert_ne!(bodies[0]["prompt_cache_key"], bodies[3]["prompt_cache_key"]); +} diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 6d6e70247..8a651babf 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -3,6 +3,7 @@ mod account_isolation; mod capacity; mod cookies; +mod correlation; mod error_details; mod image_account_eligibility; mod image_model_access; @@ -2788,7 +2789,13 @@ async fn image_endpoints_bypass_only_the_text_catalog_and_preserve_the_current_c Mock::given(method("POST")) .and(path(*endpoint)) .and(header("originator", "codex_cli_rs")) - .and(header("x-codex-image-turn-id", "turn_image_contract")) + .and(|request: &wiremock::Request| { + request + .headers + .get("x-codex-image-turn-id") + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| uuid::Uuid::parse_str(value).is_ok()) + }) .and(header("version", "0.144.0")) .and(header("accept", "*/*")) .and(body_bytes(body.to_vec())) @@ -3550,14 +3557,13 @@ async fn standalone_search_preserves_wire_and_scopes_turn_metadata_to_the_select let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; let server = MockServer::start().await; - let request_body = br#"{ "id":"search-session", "model":"gpt-future", "commands":{"search_query":[{"q":"private query","future":9007199254740993}]}, "future":1, "future":2 }"#; + let request_body = br#"{ "id":"search-session", "future_session":"search-session", "model":"gpt-future", "commands":{"search_query":[{"q":"private query","future":9007199254740993}]}, "future":1, "future":2 }"#; let response_body = br#"{ "encrypted_output":"ciphertext", "output":"search result", "results":[{"type":"text_result","ref_id":"turn0search0","future":9007199254740993}] }"#; Mock::given(method("POST")) .and(path("/codex/alpha/search")) .and(header("originator", "codex_cli_rs")) .and(header("version", "0.144.0")) .and(header("accept", "*/*")) - .and(body_bytes(request_body.to_vec())) .respond_with( ResponseTemplate::new(200) .insert_header("content-type", "application/json") @@ -3624,13 +3630,13 @@ async fn standalone_search_preserves_wire_and_scopes_turn_metadata_to_the_select assert_eq!(raw_response.as_deref(), Some(response_body.as_slice())); let requests = server.received_requests().await.expect("received requests"); let request = requests.first().expect("single search request"); - assert_eq!(request.body.as_slice(), request_body); + assert_scoped_search_body(&request.body, request_body); let metadata_values = captured_header_values(request, "x-codex-turn-metadata"); assert_eq!(metadata_values.len(), 1); let metadata: Value = serde_json::from_slice(&metadata_values[0]).expect("turn metadata JSON"); - assert_eq!(metadata.get("session_id"), Some(&json!("session"))); - assert_eq!(metadata.get("thread_id"), Some(&json!("thread"))); - assert_eq!(metadata.get("turn_id"), Some(&json!("turn"))); + assert_scoped_id(metadata["session_id"].as_str().unwrap(), "session"); + assert_scoped_id(metadata["thread_id"].as_str().unwrap(), "thread"); + assert_scoped_id(metadata["turn_id"].as_str().unwrap(), "turn"); assert_eq!(metadata.get("future"), Some(&json!(true))); assert!(metadata.get("account_id").is_none()); assert_ne!( @@ -3650,7 +3656,6 @@ async fn standalone_search_returns_the_exact_upstream_error_response() { let response_body = br#"{ "error":{"message":"future search validation","type":"search_error","code":"future_code"}, "future":9007199254740993 }"#; Mock::given(method("POST")) .and(path("/codex/alpha/search")) - .and(body_bytes(request_body.to_vec())) .respond_with( ResponseTemplate::new(422) .insert_header("content-type", "application/problem+json") @@ -5271,6 +5276,8 @@ async fn oversized_http_threshold_counts_utf8_and_json_escaping() { let requests = server.received_requests().await.unwrap(); let mut frame = captured_request_body(&requests[0]); frame["type"] = json!("response.create"); + // 发送前阈值按本地请求副本计量，账号域映射在实际出站时执行 + frame["session_id"] = body["session_id"].clone(); let baseline_len = serde_json::to_vec(&frame).unwrap().len(); for (payload_len, expected) in [(THRESHOLD - 1, "websocket"), (THRESHOLD, "http_sse")] { @@ -6364,9 +6371,9 @@ async fn cross_account_scope_removes_only_account_bound_body_fields() { assert!(body.get("authorization").is_none()); assert!(body.get("conversation").is_none()); - assert_eq!( - body.get("conversation_id"), - Some(&json!("client-correlation")) + assert_scoped_id( + body["conversation_id"].as_str().unwrap(), + "client-correlation", ); assert_eq!( body.get("client_metadata"), @@ -11201,7 +11208,11 @@ async fn api_key_default_http_uses_own_prefix_plain_json_and_only_own_authentica } assert_eq!(request.headers["x-business-extension"], "keep"); for name in API_KEY_BUSINESS_HEADERS { - assert_eq!(request.headers[*name], "downstream-value", "lost {name}"); + if matches!(*name, "session-id" | "thread-id") { + assert_scoped_id(request.headers[*name].to_str().unwrap(), "downstream-value"); + } else { + assert_eq!(request.headers[*name], "downstream-value", "lost {name}"); + } } for header in ["cookie", "chatgpt-account-id"] { assert!(!request.headers.contains_key(header), "unexpected {header}"); @@ -11430,7 +11441,14 @@ async fn api_key_websocket_uses_api_path_and_bearer_without_oauth_identity() { } assert_eq!(request.headers()["x-business-extension"], "keep"); for name in API_KEY_BUSINESS_HEADERS { - assert_eq!(request.headers()[*name], "downstream-value", "lost {name}"); + if matches!(*name, "session-id" | "thread-id") { + assert_scoped_id( + request.headers()[*name].to_str().unwrap(), + "downstream-value", + ); + } else { + assert_eq!(request.headers()[*name], "downstream-value", "lost {name}"); + } } }) .await; @@ -12844,3 +12862,27 @@ async fn public_catalog_filters_each_api_account_before_union_without_gating_inf ); } } + +fn assert_scoped_id(actual: &str, original: &str) { + assert_ne!(actual, original); + assert!( + uuid::Uuid::parse_str(actual).is_ok(), + "scoped ID must retain UUID format" + ); +} + +fn assert_scoped_search_body(actual: &[u8], original: &[u8]) { + let actual_json: Value = serde_json::from_slice(actual).unwrap(); + let original_json: Value = serde_json::from_slice(original).unwrap(); + let id = actual_json["id"].as_str().unwrap(); + let original_id = original_json["id"].as_str().unwrap(); + assert_scoped_id(id, original_id); + assert_eq!( + std::str::from_utf8(actual).unwrap(), + std::str::from_utf8(original).unwrap().replacen( + &serde_json::to_string(original_id).unwrap(), + &serde_json::to_string(id).unwrap(), + 1 + ) + ); +} diff --git a/backend/crates/providers/openai/tests/provider/contract/passthrough.rs b/backend/crates/providers/openai/tests/provider/contract/passthrough.rs index be89feb9c..5492387ae 100644 --- a/backend/crates/providers/openai/tests/provider/contract/passthrough.rs +++ b/backend/crates/providers/openai/tests/provider/contract/passthrough.rs @@ -134,7 +134,11 @@ 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); - assert_eq!(sent.body, raw); + if endpoint.ends_with("search") { + assert_scoped_search_body(&sent.body, &raw); + } else { + assert_eq!(sent.body, raw); + } assert_eq!( captured_header_values(sent, "x-future-business"), vec![b"first".to_vec(), b"second".to_vec()] diff --git a/backend/crates/providers/openai/tests/provider/contract/privacy.rs b/backend/crates/providers/openai/tests/provider/contract/privacy.rs index cc13984ba..8513a9274 100644 --- a/backend/crates/providers/openai/tests/provider/contract/privacy.rs +++ b/backend/crates/providers/openai/tests/provider/contract/privacy.rs @@ -129,12 +129,12 @@ async fn privacy_json_endpoints_preserve_untouched_bytes_and_rewrite_matched_bod "Bearer configured-synthetic-token" ); if change_body { - assert_eq!( - serde_json::from_slice::(&sent.body).unwrap(), - json!({"id":"search","future":2}) - ); + let mut value: Value = serde_json::from_slice(&sent.body).unwrap(); + assert_scoped_id(value["id"].as_str().unwrap(), "search"); + value.as_object_mut().unwrap().remove("id"); + assert_eq!(value, json!({"future":2})); } else { - assert_eq!(sent.body, raw); + assert_scoped_search_body(&sent.body, &raw); } } } diff --git a/backend/crates/providers/openai/tests/transport/downstream/headers.rs b/backend/crates/providers/openai/tests/transport/downstream/headers.rs index e77ed4dbb..b868767df 100644 --- a/backend/crates/providers/openai/tests/transport/downstream/headers.rs +++ b/backend/crates/providers/openai/tests/transport/downstream/headers.rs @@ -256,7 +256,6 @@ async fn backend_http_should_preserve_business_headers_without_downstream_transp ("accept", b"text/event-stream".as_slice()), ("content-type", b"application/json".as_slice()), ("x-still-valid", b"after-invalid".as_slice()), - ("session-id", b"synthetic-alias".as_slice()), ("authorization", b"Bearer lease-token".as_slice()), ("chatgpt-account-id", b"lease-account".as_slice()), ("chatgpt-organization-id", b"unclassified-org".as_slice()), @@ -266,6 +265,9 @@ 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()); + let session = raw_header_values(&raw, "session-id"); + assert!(uuid::Uuid::parse_str(std::str::from_utf8(&session[0]).unwrap()).is_ok()); // 指纹头由运行时画像生成，不透传客户端值 assert_eq!( raw_header_values(&raw, "user-agent"), @@ -293,7 +295,6 @@ async fn backend_http_should_preserve_business_headers_without_downstream_transp ); } for name in [ - "traceparent", "x-business-origin", "x-cf-business-field", "sec-ch-business", @@ -430,7 +431,6 @@ async fn backend_websocket_should_preserve_business_headers_without_downstream_t vec![b"lease-account".to_vec()] ); for name in [ - "traceparent", "x-business-origin", "x-cf-business-field", "sec-ch-business", @@ -446,6 +446,8 @@ async fn backend_websocket_should_preserve_business_headers_without_downstream_t ] { assert_eq!(values(name), vec![value.to_vec()], "missing {name}"); } - assert_eq!(values("session-id"), vec![b"synthetic-alias".to_vec()]); + assert!(values("traceparent").is_empty()); + 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()]); } diff --git a/backend/crates/providers/openai/tests/transport/headers.rs b/backend/crates/providers/openai/tests/transport/headers.rs index 4f1e2eb5a..bad262ed6 100644 --- a/backend/crates/providers/openai/tests/transport/headers.rs +++ b/backend/crates/providers/openai/tests/transport/headers.rs @@ -197,7 +197,8 @@ async fn backend_websocket_should_forward_context_headers_and_preserve_payload_f let payload = server.await.expect("header server task"); assert!(response.body.contains("resp_ws_security")); - assert_eq!(payload["prompt_cache_key"], "client-thread"); + assert_ne!(payload["prompt_cache_key"], "client-thread"); + assert!(uuid::Uuid::parse_str(payload["prompt_cache_key"].as_str().unwrap()).is_ok()); let metadata = payload["client_metadata"] .as_object() .expect("client metadata"); @@ -226,11 +227,16 @@ async fn backend_websocket_should_forward_context_headers_and_preserve_payload_f ("x-openai-memgen-request", "true"), ("session-id", "cp_derived"), ] { - assert!( - headers - .iter() - .any(|(header, value)| { header == name && value == expected }) - ); + let value = &headers.iter().find(|(header, _)| header == name).unwrap().1; + if matches!( + name, + "session-id" | "x-client-request-id" | "x-codex-window-id" | "x-codex-parent-thread-id" + ) { + assert_ne!(value, expected); + assert!(uuid::Uuid::parse_str(value).is_ok()); + } else { + assert_eq!(value, expected); + } } for forbidden in [ "x-codex-installation-id", @@ -326,7 +332,7 @@ async fn backend_http_should_send_codex_context_without_browser_headers() { Some(expected_user_agent.as_str()), Some("text/event-stream"), None, - Some("session-1"), + read_header_value(&raw_request, "session-id"), Some("turn-state"), ) ); @@ -334,6 +340,9 @@ async fn backend_http_should_send_codex_context_without_browser_headers() { read_header_value(&raw_request, "version"), Some(expected_core_version.as_str()) ); + let scoped_session = read_header_value(&raw_request, "session-id").unwrap(); + assert_ne!(scoped_session, "session-1"); + assert!(uuid::Uuid::parse_str(scoped_session).is_ok()); for required in [ "authorization", "chatgpt-account-id", diff --git a/docs/api.md b/docs/api.md index 5a97ed017..1b5711dac 100644 --- a/docs/api.md +++ b/docs/api.md @@ -293,7 +293,17 @@ Responses 也不透传 `x-stainless-*`、`Origin`、`Referer`、`sec-ch-ua*` 和 两者同时存在时仍优先使用 `session-id`。正文中的 `client_metadata.session_id`、 `prompt_cache_key` 不受这条请求头规则影响。 `thread-id`、turn metadata 等 Codex 协议字段及未知业务扩展不受下游环境头过滤规则影响； -`traceparent`、`tracestate` 不因属于追踪字段而被删除 +出站移除下游的 `traceparent`、`tracestate` 及 WS metadata 中对应的 +`ws_request_header_traceparent`、`ws_request_header_tracestate` + +会话、线程、轮次、窗口及父子引用在出站时按账号安装身份稳定映射为 UUID； +同账号保留引用关系，换账号使用不同标识。HTTP 请求头、正文 metadata、WS 副本及 +已识别的 Desktop 自动 Git 上下文使用一致映射。 +Images 的 `x-codex-image-turn-id`、turn metadata 和 Search 的会话 `id` 使用同一账号域； +Search 只替换该 `id` 的原始字节，其余正文保持不变。 +`prompt_cache_key` 同样按账号映射，保留同账号内的缓存共享关系；未提供时不补造。 +续接 response ID、历史 item/call ID、业务正文和未知扩展不参与映射， +该处理减少显式追踪标识的跨账号复用，不保证会话内容或业务 metadata 无法关联 #### 正文兼容 diff --git a/docs/architecture.md b/docs/architecture.md index fa774e525..d2e26de5e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -451,7 +451,10 @@ OAuth 与 API Key 共用业务请求、响应和能力透传链路，差异限 OpenAI 模型目录用于发现，不因目录缺项拒绝请求；管理员配置的模型权限仍由 Core 与选号链路执行 - OpenAI 是透明边界。Responses 请求保留未知字段和字段顺序；SSE、WebSocket、Images 与 standalone - Search 的业务正文按原始字节转发，原生续写额度恢复遵循下述 continuation 例外。 + Search 的业务正文按原始字节转发，Search 请求的会话 `id` 遵循账号归一规则， + 原生续写额度恢复遵循下述 continuation 例外。 + `transport/correlation` 在出站副本上统一映射账号关联标识并剥离客户端 trace， + 内部选号、亲和、连接池与本地诊断继续读取原始身份；具体字段边界见 [API 协议](api.md#请求头与身份) canonical facts 从同一数据旁路提取，用于路由、恢复判断、观测和计费。 WebSocket transport 按完整文本消息交付，wire 同时保留原文与可选解析视图；同协议出口使用原文， SSE 转换只发生在需要该表达的出口。插件未改写的封套继续携带原始文本，观测失败不否定交付。