From ae7608440e693b45b5d7c7e4d7b6861afdacec8f Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 31 Jul 2026 16:07:32 +0800 Subject: [PATCH] fix(openai): replay client-owned conversation history Co-authored-by: Codex --- .../crates/providers/openai/src/provider.rs | 114 ++---------------- .../openai/tests/provider/contract.rs | 103 ++++++++++++++++ 2 files changed, 111 insertions(+), 106 deletions(-) diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index 858fdc0aa..d0f9ac7d0 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/backend/crates/providers/openai/src/provider.rs @@ -64,8 +64,7 @@ use crate::transport::protocol::responses::{ CodexResponsesRequest, PreviousResponseScope, ResponseEventSignals, }; use crate::transport::request::{ - CodexRequestEncodeError, encode_generate_request, sanitize_cross_account_item, - scope_request_to_account, + CodexRequestEncodeError, encode_generate_request, scope_request_to_account, }; use crate::transport::session::CodexSessionIdentity; use crate::transport::usage::normalize_service_tier; @@ -177,17 +176,7 @@ impl Provider for CodexProvider { return ProviderRequestObservation::default(); }; let (semantics, reasoning_effort) = encode_generate_request(request, "observability") - .map(|mut encoded| { - if let Some(previous) = decode_openai_session_state(request) { - let mut input = previous - .transcript - .iter() - .map(OpenAiReplayItem::value) - .cloned() - .collect::>(); - input.extend(encoded.input().iter().cloned()); - encoded.set_input(input); - } + .map(|encoded| { let semantics = encoded.semantics(); let reasoning_effort = semantics.reasoning_effort.clone(); (semantics, reasoning_effort) @@ -267,7 +256,6 @@ impl Provider for CodexProvider { let session_affinity_key = derive_codex_session_affinity_key(&upstream_request); let cyber_policy_session_key = derive_codex_cyber_policy_session_key(&upstream_request, context.client_api_key_ref()); - let request_input = upstream_request.input().to_vec(); let transport = selected_transport(&upstream_request); apply_transport(&mut upstream_request, transport); @@ -299,11 +287,8 @@ impl Provider for CodexProvider { ) || previous_session.as_ref().is_some_and(|state| { state.continuation_scope == OpenAiContinuationScope::ReplayRequired }); - if replay_previous_response && let Some(state) = previous_session.as_ref() { - let mut input = replay_input_for_account(state, lease.account_id().as_str()); - input.reserve(request_input.len()); - input.extend(request_input.iter().cloned()); - upstream_request.set_input(input); + if replay_previous_response && previous_session.is_some() { + // Codex 客户端 input 是完整历史；恢复时只解除上游续链绑定，避免重复叠加代理副本。 upstream_request.set_previous_response_id(None); upstream_request.previous_response_scope = None; upstream_request.turn_state = None; @@ -369,8 +354,6 @@ impl Provider for CodexProvider { let response_store = upstream_request.store(); let session_capture = (!continuation_requested || previous_session.is_some()).then(|| OpenAiSessionCapture { - previous: previous_session, - request_input, account_id: lease.account_id().as_str().to_owned(), conversation_id: upstream_request.local_conversation_id.clone(), turn_state: upstream_request.turn_state.clone(), @@ -425,7 +408,6 @@ struct OpenAiSessionState { #[serde(default)] turn_state: Option, continuation_scope: OpenAiContinuationScope, - transcript: Vec, } #[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -436,27 +418,7 @@ enum OpenAiContinuationScope { ReplayRequired, } -#[derive(Clone, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -enum OpenAiReplayItem { - ClientInput(Value), - SanitizedOutput(Value), - AccountOutput { account_id: String, item: Value }, -} - -impl OpenAiReplayItem { - fn value(&self) -> &Value { - match self { - Self::ClientInput(value) - | Self::SanitizedOutput(value) - | Self::AccountOutput { item: value, .. } => value, - } - } -} - struct OpenAiSessionCapture { - previous: Option, - request_input: Vec, account_id: String, conversation_id: Option, turn_state: Option, @@ -489,33 +451,15 @@ fn attach_openai_session_update( events: &mut [ProviderEvent], capture: &mut Option, ) { - let Some((terminal_index, output)) = events.iter().enumerate().find_map(|(index, event)| { - terminal_response_output(event).map(|output| (index, output.to_vec())) - }) else { + let Some(terminal_index) = events + .iter() + .position(|event| terminal_response_output(event).is_some()) + else { return; }; let Some(capture) = capture.take() else { return; }; - let mut transcript = capture - .previous - .map(|state| state.transcript) - .unwrap_or_default(); - project_transcript_to_account(&mut transcript, &capture.account_id); - transcript.extend( - capture - .request_input - .into_iter() - .map(OpenAiReplayItem::ClientInput), - ); - transcript.extend( - output - .into_iter() - .map(|item| OpenAiReplayItem::AccountOutput { - account_id: capture.account_id.clone(), - item, - }), - ); let Some(continuation_scope) = capture.continuation_scope else { return; }; @@ -524,7 +468,6 @@ fn attach_openai_session_update( conversation_id: capture.conversation_id, turn_state: capture.turn_state, continuation_scope, - transcript, }) else { return; }; @@ -552,47 +495,6 @@ fn terminal_response_output(event: &ProviderEvent) -> Option<&[Value]> { .flatten() } -fn project_transcript_to_account(transcript: &mut Vec, account_id: &str) { - *transcript = transcript - .drain(..) - .filter_map(|item| match item { - OpenAiReplayItem::AccountOutput { - account_id: owner, - item, - } if owner != account_id => { - sanitize_cross_account_item(item).map(OpenAiReplayItem::SanitizedOutput) - } - item => Some(item), - }) - .collect(); -} - -fn replay_input_for_account(state: &OpenAiSessionState, account_id: &str) -> Vec { - state - .transcript - .iter() - .filter_map(|item| match item { - OpenAiReplayItem::ClientInput(value) | OpenAiReplayItem::SanitizedOutput(value) => { - Some(value.clone()) - } - OpenAiReplayItem::AccountOutput { - account_id: owner, - item, - } if owner == account_id => Some(without_output_id(item.clone())), - OpenAiReplayItem::AccountOutput { item, .. } => { - sanitize_cross_account_item(item.clone()) - } - }) - .collect() -} - -fn without_output_id(mut item: Value) -> Value { - if let Value::Object(object) = &mut item { - object.remove("id"); - } - item -} - enum CodexHandshakeAttemptError { Client(CodexClientError), Cancelled, diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index e745a6fc8..45e2e84c5 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -701,6 +701,109 @@ async fn cross_account_scope_sanitizes_only_known_turn_metadata_fields() { ); } +#[tokio::test] +async fn continuation_replay_should_use_the_client_full_input_without_legacy_transcript() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_client_history").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(1) + .mount(&server) + .await; + let client_input = json!([ + {"role": "user", "content": "complete client history"}, + {"role": "assistant", "content": "previous answer"}, + {"role": "user", "content": "current turn"} + ]); + let generation = GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), client_input.clone()), + ]), + ) + .expect("OpenAI payload") + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + ) + .with_provider_session_state( + ProviderSessionState::new( + "openai", + Map::from_iter([ + ("account_id".to_owned(), json!("acct_client_history")), + ("conversation_id".to_owned(), json!("conversation")), + ("continuation_scope".to_owned(), json!("replay_required")), + ( + "transcript".to_owned(), + json!([{"client_input": {"role": "user", "content": "stale proxy copy"}}]), + ), + ]), + ) + .expect("provider session state"), + ); + let mut stream = provider_with_base_url(&store, server.uri()) + .execute( + planned_request("openai", Operation::Generate(generation)), + context("req_client_history", CancellationToken::new()) + .with_continuation_attempt(ContinuationAttempt::ReplayAny), + ) + .await + .expect("prepare replay stream"); + while let Some(event) = stream.next().await { + event.expect("replay response"); + } + let requests = server + .received_requests() + .await + .expect("captured replay request"); + let body: Value = serde_json::from_slice(&requests[0].body).expect("captured JSON body"); + + assert_eq!(body.get("input"), Some(&client_input)); +} + +#[tokio::test] +async fn completed_response_session_state_should_not_copy_the_conversation_transcript() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_bounded_session_state").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(1) + .mount(&server) + .await; + let mut stream = provider_with_base_url(&store, server.uri()) + .execute( + planned_request("openai", http_generate_operation()), + context("req_bounded_session_state", CancellationToken::new()), + ) + .await + .expect("prepare provider stream"); + let mut session_payload = None; + while let Some(event) = stream.next().await { + let event = event.expect("provider response"); + if let Some(update) = event.session_update() { + session_payload = Some(update.payload().clone()); + } + } + + assert!( + !session_payload + .expect("completed response session update") + .contains_key("transcript") + ); +} + #[tokio::test] async fn account_change_drops_only_account_bound_opaque_headers() { let protocol_context = Map::from_iter([(