From b37d7901441e123cb9126a6f9af4ff3321548d24 Mon Sep 17 00:00:00 2001 From: Zyy Date: Tue, 29 Sep 2026 22:15:19 +0800 Subject: [PATCH] fix(api): default HTTP responses to JSON when stream is omitted Preserve explicit stream preferences and the WebSocket streaming default without rewriting request bodies or changing upstream execution. Add decoder and HTTP delivery regression coverage. Co-authored-by: Codex --- .../src/openai/responses/request.rs | 9 +-- .../tests/openai/responses/http.rs | 66 +++++++++++++++++++ .../gateway-api/tests/openai/responses/mod.rs | 21 ++++++ 3 files changed, 92 insertions(+), 4 deletions(-) diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index 5aa5998ee..46e235f70 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -469,10 +469,11 @@ pub(super) fn decode_request_object( }); } let model = model.to_owned(); - let stream = object - .get("stream") - .and_then(Value::as_bool) - .unwrap_or(true); + // 缺省值只决定下游交付方式，不补写正文或改变 Provider 的上游流式执行。 + let stream = match object.get("stream") { + Some(value) => value.as_bool().unwrap_or(true), + None => matches!(source, RequestDecodeSource::WebSocketFrame), + }; let store = object .get("store") .and_then(Value::as_bool) diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 8853e3375..c2d3f14ad 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -523,6 +523,72 @@ async fn http_request_with_body( (response, observed) } +#[tokio::test] +async fn http_requests_should_honor_stream_preference() { + let admin = crate::admin::AdminTestFixture::new().await.services; + for stream in [None, Some(false), Some(true)] { + let trace = Arc::new(Trace::default()); + let streaming = stream == Some(true); + let session = if streaming { + FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::Event(delivery(started(), CommitRequirement::CommitBeforeDelivery)), + NextStep::Event(delivery(completed(), CommitRequirement::AlreadyCommitted)), + NextStep::FinalizeSuccess, + ], + ) + } else { + FakeSession::buffered(Arc::clone(&trace), vec![started(), completed()]) + }; + let execution = Arc::new(SessionExecution { + client: authenticated_client_for_provider("sk_stream_test", "openai"), + session: Mutex::new(Some(Box::new(session))), + middleware: None, + }); + let mut body = json!({"model": "model-a", "input": "hello"}); + if let Some(stream) = stream { + body["stream"] = json!(stream); + } + let response = crate::openai::api_router_with_admin_and_execution(admin.clone(), execution) + .oneshot( + Request::post("/v1/responses") + .header(AUTHORIZATION, "Bearer sk_stream_test") + .header(CONTENT_TYPE, "application/json") + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK, "stream={stream:?}"); + assert_eq!( + response.headers()[CONTENT_TYPE], + if streaming { + "text/event-stream" + } else { + "application/json" + }, + "stream={stream:?}", + ); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + if streaming { + let body = String::from_utf8(body.to_vec()).unwrap(); + assert!(body.contains("event: response.completed\n")); + assert!(body.ends_with("data: [DONE]\n\n")); + } else { + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + json!({ + "id": "resp_test", "model": "public-model", "status": "completed", "output": [] + }), + ); + assert_eq!(trace.snapshot(), vec!["collect", "commit"]); + } + assert!(!trace.is_cancelled()); + } +} + #[tokio::test] async fn compressed_http_requests_should_preserve_execution_context_without_transport_headers() { let body = json!({ diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index b06202ebe..18f96b79b 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/mod.rs @@ -119,6 +119,26 @@ fn decoder_should_preserve_the_openai_body_and_only_derive_stable_routing_facts( assert!(decoded.metadata().store()); } +#[test] +fn decoder_should_default_omitted_http_stream_to_json_without_rewriting_body() { + let body = json!({"model": "smart-code", "input": "hello"}); + let decoded = generate_request(body.clone()); + + assert!(!decoded.metadata().stream()); + assert_eq!(openai_wire_body(&decoded), body.as_object().unwrap()); +} + +#[test] +fn decoder_should_preserve_explicit_http_stream_values() { + for stream in [false, true] { + let body = json!({"model": "smart-code", "input": "hello", "stream": stream}); + let decoded = generate_request(body.clone()); + + assert_eq!(decoded.metadata().stream(), stream); + assert_eq!(openai_wire_body(&decoded), body.as_object().unwrap()); + } +} + #[test] fn decoder_should_preserve_opaque_client_model_values() { for model in [ @@ -631,6 +651,7 @@ fn decoder_should_leave_openai_semantic_validation_to_the_upstream() { "future_official_field": [1, 2, 3] })); + assert!(decoded.metadata().stream()); assert_eq!( Value::Object(openai_wire_body(&decoded).clone()), json!({