starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): scope conversation compaction to xai

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-25 00:33:11 +08:00
co-authored by Codex
parent 1f3bcf68c7
commit be60e2d763
6 changed files with 22 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/providers/openai/src/lib.rs
+1 -1
View File
@@ -41,7 +41,7 @@ pub mod transport;
pub use transport::{
CodexCanonicalDecoder, CodexCanonicalError, CodexRequestEncodeError, codex_request_semantics,
encode_compact_conversation_request, encode_generate_request, openai_billing_breakdown,
encode_generate_request, openai_billing_breakdown,
};
/// OpenAI 初始化后交给组装根的最小能力集。
backend/crates/providers/openai/src/provider.rs
+17 -38
View File
@@ -58,8 +58,8 @@ use crate::transport::profile::{
};
use crate::transport::protocol::responses::{CodexResponsesRequest, PreviousResponseScope};
use crate::transport::request::{
CodexRequestEncodeError, encode_compact_conversation_request, encode_generate_request,
sanitize_cross_account_item, scope_request_to_account,
CodexRequestEncodeError, encode_generate_request, sanitize_cross_account_item,
scope_request_to_account,
};
use crate::transport::websocket::{CodexWebSocketExchangeError, PreviousResponseUnavailableReason};
use crate::transport::{
@@ -265,17 +265,10 @@ impl Provider for CodexProvider {
}
fn request_observation(&self, operation: &Operation) -> ProviderRequestObservation {
let (request, compact) = match operation {
Operation::Generate(request) => (request, false),
Operation::CompactConversation(request) => (request.generation(), true),
_ => return ProviderRequestObservation::default(),
let Operation::Generate(request) = operation else {
return ProviderRequestObservation::default();
};
let encoded = if compact {
encode_compact_conversation_request(request, "observability")
} else {
encode_generate_request(request, "observability")
};
let semantics = encoded
let semantics = encode_generate_request(request, "observability")
.map(|mut encoded| {
if let Ok(Some(previous)) = decode_openai_session_state(request) {
let mut input = previous
@@ -348,32 +341,18 @@ impl Provider for CodexProvider {
UpstreamSendState::NotSent,
));
}
let (generate, compact) = match request.operation() {
Operation::Generate(generate) => (generate, false),
Operation::CompactConversation(compact) => (compact.generation(), true),
_ => {
return Err(provider_error(
ProviderErrorKind::Unsupported,
UpstreamSendState::NotSent,
));
}
let Operation::Generate(generate) = request.operation() else {
return Err(provider_error(
ProviderErrorKind::Unsupported,
UpstreamSendState::NotSent,
));
};
let previous_session = decode_openai_session_state(generate)?;
let continuation_requested = generate.continuation().is_some();
let mut upstream_request = if compact {
encode_compact_conversation_request(generate, candidate.upstream_model().as_str())
} else {
let mut upstream_request =
encode_generate_request(generate, candidate.upstream_model().as_str())
}
.map_err(map_request_error)?;
.map_err(map_request_error)?;
let request_input = upstream_request.input().to_vec();
let session_request_input = if compact {
request_input
.split_last()
.map_or_else(Vec::new, |(_, history)| history.to_vec())
} else {
request_input.clone()
};
let transport = selected_transport(&request, &instance.config)?;
apply_transport(&mut upstream_request, transport);
@@ -485,7 +464,7 @@ impl Provider for CodexProvider {
let session_capture =
(!continuation_requested || previous_session.is_some()).then(|| OpenAiSessionCapture {
previous: previous_session,
request_input: session_request_input,
request_input,
account_id: lease.account_id().as_str().to_owned(),
conversation_id: upstream_request.local_conversation_id.clone(),
response_store,
@@ -1047,7 +1026,6 @@ fn compile_model_capabilities(model: &CodexCatalogModel) -> ProviderModelCapabil
let mut operations = BTreeSet::new();
if evidence.responses_api() == CodexCatalogCapabilityEvidence::DeclaredNative {
operations.insert(OperationKind::Generate);
operations.insert(OperationKind::CompactConversation);
}
let context_window = model
.limits()
@@ -1456,10 +1434,11 @@ fn map_upstream_failure(
if let Some(retry_after) = failure.retry_after_seconds.map(Duration::from_secs) {
error = error.with_retry_after(retry_after);
}
if let Some(code) = failure.persistable_code()
&& let Ok(safe_code) = SafeUpstreamValue::new(code.to_owned())
if let Some(code) = failure
.persistable_code()
.and_then(|code| SafeUpstreamValue::new(code.to_owned()).ok())
{
error = error.with_upstream_code(safe_code);
error = error.with_upstream_code(code);
}
if let Some(request_id) = failure
.request_id
backend/crates/providers/openai/src/transport/mod.rs
+1 -4
View File
@@ -35,10 +35,7 @@ pub use self::{
endpoint_request_path, endpoint_url, usage_endpoint_urls,
},
headers::{build_codex_base_headers, build_codex_headers},
request::{
CodexRequestEncodeError, codex_request_semantics, encode_compact_conversation_request,
encode_generate_request,
},
request::{CodexRequestEncodeError, codex_request_semantics, encode_generate_request},
response_meta::CodexResponseMetadata,
usage::{MAX_CODEX_USAGE_BODY_BYTES, openai_billing_breakdown},
websocket::{
backend/crates/providers/openai/src/transport/request.rs
-13
View File
@@ -191,19 +191,6 @@ pub fn encode_generate_request(
Ok(encoded)
}
/// 将类型化压缩 operation 编码为 OpenAI 原生 Responses compaction 请求。
pub fn encode_compact_conversation_request(
request: &GenerateRequest,
upstream_model: &str,
) -> Result<CodexResponsesRequest, CodexRequestEncodeError> {
let mut encoded = encode_generate_request(request, upstream_model)?;
let mut input = encoded.input().to_vec();
input.push(json!({"type": "compaction_trigger"}));
encoded.set_input(input);
encoded.set_stream(true);
Ok(encoded)
}
/// 复用正式 Codex 编码规则提取请求诊断语义；编码失败时不伪造 Provider 事实。
#[must_use]
pub fn codex_request_semantics(request: &GenerateRequest) -> CodexRequestSemantics {
backend/crates/providers/openai/tests/provider/contract.rs
+2 -125
View File
@@ -454,35 +454,6 @@ fn http_operation() -> Operation {
)
}
fn compact_http_operation() -> Operation {
let payload = ProtocolPayload::json_object(
"openai",
json!({
"model": "gpt-5.4",
"input": [{"type": "message", "role": "user", "content": "history"}],
"stream": false,
"store": false
})
.as_object()
.cloned()
.expect("compact request body"),
)
.expect("OpenAI payload");
let mut options = ProviderOptions::new();
options
.insert(
"openai",
Map::from_iter([
("schema_version".to_owned(), Value::from(1)),
("transport".to_owned(), Value::String("http_sse".to_owned())),
]),
)
.expect("provider options");
Operation::CompactConversation(CompactConversationRequest::new(
GenerateRequest::from_protocol_payload(Vec::new(), payload).with_provider_options(options),
))
}
fn preferred_operation() -> Operation {
let message = Message::new(
MessageRole::User,
@@ -771,7 +742,7 @@ async fn rejected_websocket_opening(
fn planned_request(base_url: &str, operation: &Operation) -> ProviderRequest {
let public_model = PublicModelId::new("gpt-5.4").expect("public model");
let capabilities = ModelCapabilities::new(
BTreeSet::from([OperationKind::Generate, OperationKind::CompactConversation]),
BTreeSet::from([OperationKind::Generate]),
128_000,
Some(32_000),
);
@@ -794,100 +765,6 @@ fn planned_request(base_url: &str, operation: &Operation) -> ProviderRequest {
ProviderRequest::new(operation.clone(), plan.candidates()[0].clone())
}
#[tokio::test]
async fn compact_operation_preserves_native_openai_wire_and_excludes_trigger_from_session() {
let server = MockServer::start().await;
let compaction_item = json!({
"type": "compaction",
"encrypted_content": "opaque-native-ciphertext"
});
let output_done = json!({
"type": "response.output_item.done",
"output_index": 0,
"item": compaction_item
});
let completed_event = json!({
"type": "response.completed",
"response": {
"id": "resp_compact_native",
"model": "gpt-5.4",
"status": "completed",
"output": [compaction_item],
"usage": {"input_tokens": 10, "output_tokens": 1, "total_tokens": 11}
}
});
let body = format!(
"event: response.created\ndata: {}\n\nevent: response.output_item.done\ndata: {output_done}\n\nevent: response.completed\ndata: {completed_event}\n\n",
created("resp_compact_native")
);
Mock::given(method("POST"))
.and(path("/backend-api/codex/responses"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_string(body),
)
.expect(1)
.mount(&server)
.await;
let store = Arc::new(MemoryAccountStore::default());
create_account(&store).await;
let provider = provider(&store);
let operation = compact_http_operation();
assert!(provider.request_observation(&operation).compact);
let mut stream = provider
.execute(
planned_request(&format!("{}/backend-api", server.uri()), &operation),
context("req_native_compaction", None, CancellationToken::new()),
)
.await
.expect("provider stream");
let mut wire_events = Vec::new();
let mut session_update = None;
while let Some(event) = stream.next().await {
let mut event = event.expect("native compaction event");
if let Some(wire) = event.wire_event() {
wire_events.push((wire.event_type().map(str::to_owned), wire.data().clone()));
}
session_update = event.take_session_update().or(session_update);
}
assert_eq!(
wire_events[1].0.as_deref(),
Some("response.output_item.done")
);
assert_eq!(wire_events[1].1, output_done);
assert_eq!(wire_events[2].0.as_deref(), Some("response.completed"));
assert_eq!(wire_events[2].1, completed_event);
let session_update = session_update.expect("compaction session update");
let transcript = session_update
.payload()
.get("transcript")
.and_then(Value::as_array)
.expect("session transcript");
assert!(!transcript.iter().any(|item| {
item.pointer("/client_input/type").and_then(Value::as_str) == Some("compaction_trigger")
}));
let requests = server
.received_requests()
.await
.expect("received compact request");
let upstream_body: Value =
serde_json::from_slice(&requests[0].body).expect("upstream compact JSON");
let input = upstream_body["input"].as_array().expect("upstream input");
assert_eq!(upstream_body["stream"], true);
assert_eq!(input.last(), Some(&json!({"type": "compaction_trigger"})));
assert_eq!(
input
.iter()
.filter(|item| item.get("type").and_then(Value::as_str) == Some("compaction_trigger"))
.count(),
1
);
}
async fn next_provider_error(
stream: &mut gateway_core::engine::provider::ProviderStream,
) -> ProviderError {
@@ -2697,6 +2574,6 @@ async fn capability_query_compiles_only_explicit_catalog_evidence() {
capabilities[0]
.capabilities()
.match_requirements(&compact.capability_requirements())
.is_some()
.is_none()
);
}
backend/crates/providers/openai/tests/transport/request.rs
+1 -39
View File
@@ -4,45 +4,7 @@ use gateway_core::operation::{
};
use serde_json::{Map, Value, json};
use provider_openai::{
CodexRequestEncodeError, codex_request_semantics, encode_compact_conversation_request,
encode_generate_request,
};
#[test]
fn compact_encoder_should_append_one_terminal_trigger_and_force_streaming() {
let payload = ProtocolPayload::json_object(
"openai",
json!({
"model": "client-model",
"input": [
{"role": "user", "content": "first turn"},
{"role": "assistant", "content": "first answer"}
],
"stream": false
})
.as_object()
.cloned()
.expect("request object"),
)
.expect("OpenAI payload");
let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
let encoded =
encode_compact_conversation_request(&request, "gpt-routed").expect("compact request");
let triggers = encoded
.input()
.iter()
.filter(|item| item.get("type").and_then(Value::as_str) == Some("compaction_trigger"))
.count();
assert!(encoded.stream());
assert_eq!(triggers, 1);
assert_eq!(
encoded.input().last(),
Some(&json!({"type": "compaction_trigger"}))
);
}
use provider_openai::{CodexRequestEncodeError, codex_request_semantics, encode_generate_request};
#[test]
fn encoder_should_preserve_openai_wire_fields_without_deriving_accountless_pool_identity() {