fix: preserve upstream stream accept negotiation

This commit is contained in:
stabey
2026-05-09 12:29:23 +08:00
parent 7e5a08e09d
commit 5f4fa4ce1f
4 changed files with 143 additions and 14 deletions

View File

@@ -1,8 +1,11 @@
use axum::body::Bytes;
use crate::ai_serving::{
endpoint_config_forces_upstream_stream_policy as endpoint_config_forces_upstream_stream_policy_impl,
enforce_request_body_stream_field as enforce_request_body_stream_field_impl,
force_upstream_streaming_for_provider as force_upstream_streaming_for_provider_impl,
is_json_request, parse_direct_request_body as parse_direct_request_body_impl,
resolve_upstream_is_stream_from_endpoint_config as resolve_upstream_is_stream_from_endpoint_config_impl,
};
pub(crate) use crate::ai_serving::{
CLAUDE_CHAT_STREAM_PLAN_KIND, CLAUDE_CHAT_SYNC_PLAN_KIND, CLAUDE_CLI_STREAM_PLAN_KIND,
@@ -46,7 +49,7 @@ pub(crate) fn resolve_upstream_is_stream_for_provider(
) -> bool {
let hard_requires_streaming = hard_requires_streaming
|| force_upstream_streaming_for_provider(provider_type, provider_api_format);
aether_ai_formats::resolve_upstream_is_stream_from_endpoint_config(
resolve_upstream_is_stream_from_endpoint_config_impl(
endpoint_config,
client_is_stream,
hard_requires_streaming,
@@ -56,7 +59,7 @@ pub(crate) fn resolve_upstream_is_stream_for_provider(
pub(crate) fn endpoint_config_forces_body_stream_field(
endpoint_config: Option<&serde_json::Value>,
) -> bool {
aether_ai_formats::endpoint_config_forces_upstream_stream_policy(endpoint_config)
endpoint_config_forces_upstream_stream_policy_impl(endpoint_config)
}
pub(crate) fn request_requires_body_stream_field(
@@ -75,7 +78,7 @@ pub(crate) fn enforce_provider_body_stream_policy(
upstream_is_stream: bool,
require_body_stream_field: bool,
) {
aether_ai_formats::enforce_request_body_stream_field(
enforce_request_body_stream_field_impl(
provider_request_body,
provider_api_format,
upstream_is_stream,

View File

@@ -18,6 +18,18 @@ use crate::ai_serving::transport::{
};
use crate::{AiExecutionDecision, GatewayError};
fn effective_stream_accept_mode(
payload_upstream_is_stream: bool,
provider_request_body: &serde_json::Value,
) -> bool {
payload_upstream_is_stream
|| provider_request_body
.as_object()
.and_then(|body| body.get("stream"))
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
}
pub(crate) fn build_openai_chat_stream_plan_from_decision(
parts: &http::request::Parts,
body_json: &serde_json::Value,
@@ -80,6 +92,8 @@ pub(crate) fn build_openai_chat_stream_plan_from_decision(
}
provider_request_body
};
let effective_upstream_is_stream =
effective_stream_accept_mode(payload.upstream_is_stream, &provider_request_body_value);
let extra_headers = std::mem::take(&mut payload.extra_headers);
let mut provider_request_headers =
build_standard_plan_fallback_headers(StandardPlanFallbackHeadersInput {
@@ -93,9 +107,9 @@ pub(crate) fn build_openai_chat_stream_plan_from_decision(
content_type: payload.content_type.as_deref(),
provider_api_format: core.provider_api_format.as_str(),
client_api_format: core.client_api_format.as_str(),
upstream_is_stream: payload.upstream_is_stream,
upstream_is_stream: effective_upstream_is_stream,
build_from_request_when_empty: true,
accept_policy: StandardPlanFallbackAcceptPolicy::TextEventStreamIfStreaming,
accept_policy: StandardPlanFallbackAcceptPolicy::TextEventStreamIfStreamingOrWildcard,
});
let content_type = payload
.content_type
@@ -168,14 +182,18 @@ pub(crate) fn build_openai_responses_stream_plan_from_decision(
.as_ref()
.and_then(|context| context.get("envelope_name"))
.and_then(serde_json::Value::as_str);
let accept_policy = if payload.upstream_is_stream
let effective_upstream_is_stream =
effective_stream_accept_mode(payload.upstream_is_stream, &provider_request_body_value);
let accept_policy = if effective_upstream_is_stream
&& provider_adaptation_requires_eventstream_accept(
envelope_name,
core.provider_api_format.as_str(),
) {
StandardPlanFallbackAcceptPolicy::ProviderEventStreamIfMissing
} else {
} else if envelope_name.is_some() {
StandardPlanFallbackAcceptPolicy::TextEventStreamIfStreaming
} else {
StandardPlanFallbackAcceptPolicy::TextEventStreamIfStreamingOrWildcard
};
let mut provider_request_headers =
build_standard_plan_fallback_headers(StandardPlanFallbackHeadersInput {
@@ -189,7 +207,7 @@ pub(crate) fn build_openai_responses_stream_plan_from_decision(
content_type: payload.content_type.as_deref(),
provider_api_format: core.provider_api_format.as_str(),
client_api_format: core.client_api_format.as_str(),
upstream_is_stream: payload.upstream_is_stream,
upstream_is_stream: effective_upstream_is_stream,
build_from_request_when_empty: false,
accept_policy,
});
@@ -231,7 +249,7 @@ pub(crate) fn build_openai_responses_stream_plan_from_decision(
plan_url = %plan.url,
client_api_format = %plan.client_api_format,
provider_api_format = %plan.provider_api_format,
upstream_is_stream = payload.upstream_is_stream,
upstream_is_stream = effective_upstream_is_stream,
compact,
"gateway built local openai responses stream execution plan"
);