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 转换只发生在需要该表达的出口。插件未改写的封套继续携带原始文本,观测失败不否定交付。