This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 99caf99584 refactor(ai-proxy): integrate Connect protocol conversion
(#7575)
99caf99584 is described below
commit 99caf99584e0bbc85e1948cd28c48f997be4cd5e
Author: shown <[email protected]>
AuthorDate: Tue Oct 13 09:55:28 2026 +0800
refactor(ai-proxy): integrate Connect protocol conversion (#7575)
* refactor(ai-proxy): integrate Connect protocol conversion
* docs(ai-proxy): document Connect profile and protocol entry points
---
.../src/main/release-docs/LICENSE | 1 +
.../main/release-docs/licenses/LICENSE-connect.txt | 21 ++
.../shenyu-plugin-ai-proxy/README.md | 124 ++++++++++
.../shenyu-plugin-ai-proxy/pom.xml | 7 +
.../ai/proxy/enhanced/model/AiUpstreamRequest.java | 22 ++
.../enhanced/protocol/ConnectProtocolAdapter.java | 189 +++++++++++++++
.../proxy/enhanced/protocol/OpenAiResponses.java | 75 ++++++
.../plugin/ai/proxy/enhanced/provider/OpenAi.java | 46 +++-
.../proxy/enhanced/provider/OpenAiCompatible.java | 5 +-
.../ai/proxy/enhanced/service/AiProxyEngine.java | 4 +-
.../enhanced/AiProxyHttpCompatibilityTest.java | 107 ++++++++-
.../protocol/ConnectProtocolAdapterTest.java | 258 +++++++++++++++++++++
.../enhanced/protocol/OpenAiResponsesTest.java | 82 +++++++
.../service/AiProxyEngineOrchestrationTest.java | 48 +++-
.../ai/proxy/AiProxyPluginConfiguration.java | 11 +
.../ai/proxy/AiProxyPluginConfigurationTest.java | 4 +
16 files changed, 990 insertions(+), 14 deletions(-)
diff --git a/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/LICENSE
b/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/LICENSE
index 6c018b594f..6e66f46a10 100644
--- a/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/LICENSE
+++ b/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/LICENSE
@@ -726,6 +726,7 @@ MIT licenses
The following components are provided under the MIT License. See project link
for details.
The text of each license is also included at licenses/LICENSE-[project].txt.
+ connect 0.1.0: https://github.com/agentic-ai-java/connect, MIT (see
licenses/LICENSE-connect.txt)
jtokkit 1.1.0: https://github.com/knuddelsgmbh/jtokkit, MIT
animal-sniffer-annotations 1.23:
https://github.com/mojohaus/animal-sniffer, MIT
bcpkix-jdk15on 1.68: http://www.bouncycastle.org/licence.html , MIT
diff --git
a/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/licenses/LICENSE-connect.txt
b/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/licenses/LICENSE-connect.txt
new file mode 100644
index 0000000000..301bf95e13
--- /dev/null
+++
b/shenyu-dist/shenyu-bootstrap-dist/src/main/release-docs/licenses/LICENSE-connect.txt
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 connect contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/README.md
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/README.md
new file mode 100644
index 0000000000..eda17a845e
--- /dev/null
+++ b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/README.md
@@ -0,0 +1,124 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+# AI Proxy protocol conversion with Connect
+
+The AI Proxy uses `io.github.agentic-ai-java:connect:0.1.0` for JSON protocol
+conversion and explicitly selected provider dialects. ShenYu remains the
+reactive gateway, not an agent runtime. This integration supports OpenAI Chat
+Completions and OpenAI Responses clients against existing Chat Completions
+upstreams. It does not add Claude, Gemini, or native Responses upstream
transports.
+
+## Responsibility boundary
+
+| Concern | Connect | ShenYu |
+| --- | --- | --- |
+| Protocol semantics | Convert request/response JSON, messages, tool
definitions/calls, reasoning, finish reasons and usage | Validate the HTTP
request envelope and select the client protocol |
+| Provider dialect | Apply an explicitly selected profile, such as DeepSeek
Chat | Resolve provider, routed model and gateway parameter overrides |
+| Conversion losses | Produce structured diagnostics and enforce the `SAFE`
loss policy | Classify failures and log diagnostic codes without request
content or secrets |
+| Request context | Retain tool-name/type mappings needed for response
restoration | Allocate one context per request and bind it to the matching
response |
+| Streaming | Convert JSON chunks into protocol events and finalize conversion
state | Parse/frame SSE bytes, validate content type, preserve backpressure and
cancellation, enforce conversion limits |
+| Gateway execution | No networking or gateway policy | Configuration
synchronization, selector/rule matching, credentials, endpoints, WebClient
transport, HTTP status/headers, retries and fallback |
+| Billing and agents | Normalize usage; no charging or tool execution |
Billing/quota policy remains a gateway concern; this change adds no billing
system or agent/tool execution |
+
+Protocol field mappings belong in Connect, not duplicated in provider classes.
+A new upstream protocol still requires a ShenYu transport/provider integration:
+a library codec alone does not define its endpoint or authentication contract.
+
+## Exchange lifecycle
+
+1. ShenYu resolves global/selector configuration, proxy credentials and
fallback.
+2. The protocol decoder validates the envelope and retains the original
payload.
+3. The provider applies gateway overrides, then `ConnectProtocolAdapter`
converts
+ to Chat using a request-scoped `ConversionContext`. The configured upstream
+ model is authoritative. The provider profile runs once after conversion.
+4. ShenYu constructs the HTTP endpoint and authorization headers and sends the
+ request through its existing reactive transport.
+5. The request-bound response adapter uses the same context to convert
successful
+ Chat responses back to Responses, including custom-tool restoration. HTTP
+ errors and Chat client responses remain passthrough.
+6. For streams, ShenYu incrementally frames upstream SSE; Connect consumes JSON
+ chunks; ShenYu emits `event: <type>` and `data: <json>` Responses SSE
frames.
+ Stream state is allocated per subscription, including retries. Finalization
+ runs only on normal completion, never cancellation or transport failure.
+
+## Configuration and use
+
+Add the existing AI Proxy starter to the gateway. Its configuration registers
+both `openai-chat` and `openai-responses` protocol implementations.
+Set the selector handle's existing `protocol` field explicitly for a Responses
+route; the default remains `openai-chat` for compatibility. For example:
+
+```json
+{
+ "provider": "OpenAI",
+ "baseUrl": "https://api.openai.com";,
+ "apiKey": "<upstream-api-key>",
+ "model": "gpt-4o-mini",
+ "protocol": "openai-responses",
+ "stream": false
+}
+```
+
+Route a client request such as the following to that selector:
+
+```json
+{
+ "model": "client-model-alias",
+ "input": "Explain reactive backpressure briefly.",
+ "max_output_tokens": 128,
+ "stream": true
+}
+```
+
+The upstream receives Chat `messages`, the configured routed model and
+`max_completion_tokens` at `/v1/chat/completions`. The client receives
Responses
+events, including usage when supplied upstream. An explicit client `stream`
+setting takes precedence over the configured default. Existing `maxTokens` and
+`temperature` gateway overrides still apply before conversion.
+
+The existing `DeepSeek` provider identifier selects Connect's DeepSeek Chat
+profile. Other compatible identifiers (`ALiYun`, `OpenAPI`, `Moonshot`) remain
+generic; neither URLs nor model names are used to guess a dialect. DeepSeek's
+profile maps token-limit and reasoning/thinking fields and rejects incompatible
+thinking/tool configurations rather than silently changing their meaning.
+
+## Compatibility and limits
+
+- Existing Chat routes retain their protocol identifier, upstream endpoint,
+ credential handling, response bytes and SSE framing. Requests remain
byte-for-byte
+ unchanged when no gateway override, stream-usage addition or provider
adaptation
+ changes the payload; generic extension fields are retained on identity
conversion.
+- Conversion uses Connect's `SAFE` policy. Error-level losses, including
Responses
+ conversation state that cannot be represented by Chat, fail with HTTP 400
before
+ sending a request. Warning-level conversions are permitted and logged by
code.
+ This is not a stateful Responses API service or hosted-tool implementation.
+- Complete Responses conversion buffers at most 5 MiB. Streaming Responses
+ conversion accepts at most 5 MiB of cumulative framed upstream data because
the
+ codec retains output/tool state. Existing Chat passthrough has no new
cumulative
+ response limit. Oversized conversions fail; after commitment the stream
aborts.
+- Successful conversion removes stale length/digest/ETag headers. Upstream HTTP
+ errors retain their original status/body. Conversion errors are not retried;
+ ShenYu's existing transport/status retry and fallback rules remain in force.
+ No retry or fallback occurs after a converted response event reaches the
client.
+- Usage conversion is not billing. Provider charges and gateway quotas must not
+ be inferred from synthetic or missing usage values.
+
+No database/configuration migration is required. Existing Chat routes need no
+change. Responses routes opt in through the selector handle; DeepSeek routes
+should review the stricter dialect validation. Roll back by reverting the
+integration commit and restoring any opt-in selectors to `openai-chat`.
diff --git a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/pom.xml
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/pom.xml
index 3adef8c041..74ac1bc6e8 100644
--- a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/pom.xml
+++ b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/pom.xml
@@ -26,6 +26,13 @@
<artifactId>shenyu-plugin-ai-proxy</artifactId>
<dependencies>
+ <!-- Protocol conversion only; ShenYu retains transport and gateway
policy. -->
+ <dependency>
+ <groupId>io.github.agentic-ai-java</groupId>
+ <artifactId>connect</artifactId>
+ <version>0.1.0</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.shenyu</groupId>
<artifactId>shenyu-plugin-base</artifactId>
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/model/AiUpstreamRequest.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/model/AiUpstreamRequest.java
index 8be38557b2..11ea070761 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/model/AiUpstreamRequest.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/model/AiUpstreamRequest.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.plugin.ai.proxy.enhanced.model;
import java.net.URI;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
/**
* Protocol-independent request sent to an AI upstream.
@@ -34,6 +35,27 @@ public final class AiUpstreamRequest {
private byte[] body;
+ private Function<AiUpstreamResponse, AiUpstreamResponse> responseAdapter =
Function.identity();
+
+ /**
+ * Bind request-scoped protocol conversion to its corresponding response.
+ *
+ * @param responseAdapter response conversion retaining this request's
tool context
+ */
+ public void setResponseAdapter(final Function<AiUpstreamResponse,
AiUpstreamResponse> responseAdapter) {
+ this.responseAdapter =
java.util.Objects.requireNonNull(responseAdapter);
+ }
+
+ /**
+ * Adapt the upstream response without subscribing to its body.
+ *
+ * @param response provider response
+ * @return client-protocol response
+ */
+ public AiUpstreamResponse adaptResponse(final AiUpstreamResponse response)
{
+ return responseAdapter.apply(response);
+ }
+
/**
* Get request URI.
*
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/ConnectProtocolAdapter.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/ConnectProtocolAdapter.java
new file mode 100644
index 0000000000..ef257f6999
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/ConnectProtocolAdapter.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.ai.proxy.enhanced.protocol;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.github.agenticai.connect.Connect;
+import io.github.agenticai.connect.ConversionContext;
+import io.github.agenticai.connect.ConversionException;
+import io.github.agenticai.connect.ConversionResult;
+import io.github.agenticai.connect.Diagnostic;
+import io.github.agenticai.connect.Format;
+import io.github.agenticai.connect.Json;
+import io.github.agenticai.connect.Options;
+import io.github.agenticai.connect.ProviderProfile;
+import io.github.agenticai.connect.ResponseStreamState;
+import io.github.agenticai.connect.StreamOptions;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiProxyException;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiUpstreamResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Request-scoped bridge between ShenYu's reactive HTTP boundary and Connect's
JSON codecs.
+ * No transport, routing, credentials, retry or billing policy is delegated to
Connect.
+ */
+public final class ConnectProtocolAdapter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ConnectProtocolAdapter.class);
+
+ private static final ObjectMapper MAPPER = new
ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+
+ private static final Connect CONNECT = new Connect();
+
+ private static final int MAX_CONVERSION_BYTES = 5 * 1024 * 1024;
+
+ private final Format clientFormat;
+
+ private final ConversionContext context;
+
+ /**
+ * Create an isolated context for a request and its response, never a
singleton context.
+ *
+ * @param protocol client protocol identifier
+ * @param upstreamModel authoritative routed model, or null to retain the
client model
+ * @param profile explicitly selected provider dialect, or null for
generic Chat
+ */
+ public ConnectProtocolAdapter(final String protocol, final String
upstreamModel, final ProviderProfile profile) {
+ clientFormat = OpenAiResponses.NAME.equals(protocol) ?
Format.OPENAI_RESPONSES : Format.OPENAI_CHAT;
+ Options options = new Options();
+ options.toolLossPolicy = Options.LossPolicy.SAFE;
+ options.providerProfile = profile;
+ context = new ConversionContext(options);
+ context.upstreamModel = Objects.isNull(upstreamModel) ? "" :
upstreamModel;
+ }
+
+ /**
+ * Convert the final gateway-configured payload to Chat and apply the
provider profile once.
+ *
+ * @param payload copied client payload with gateway overrides applied
+ * @return converted payload
+ */
+ public ObjectNode convertRequest(final ObjectNode payload) {
+ try {
+ ConversionResult result = CONNECT.convertRequest(clientFormat,
Format.OPENAI_CHAT, Json.object(payload.toString()), context);
+ reportDiagnostics(result.diagnostics());
+ return (ObjectNode) MAPPER.readTree(result.json());
+ } catch (Exception error) {
+ if (error instanceof ConversionException) {
+ reportDiagnostics(((ConversionException) error).diagnostics());
+ }
+ throw new AiProxyException(AiProxyException.Kind.REQUEST, "AI
request protocol conversion failed", error);
+ }
+ }
+
+ /**
+ * Convert successful Responses bodies; preserve Chat and upstream HTTP
errors unchanged.
+ *
+ * @param response upstream response with an unconsumed body
+ * @param stream effective streaming mode
+ * @return adapted response without subscribing or blocking
+ */
+ public AiUpstreamResponse convertResponse(final AiUpstreamResponse
response, final boolean stream) {
+ if (clientFormat == Format.OPENAI_CHAT || response.getStatusCode() <
200 || response.getStatusCode() >= 300) {
+ return response;
+ }
+ if (stream) {
+ // Reuse ShenYu's incremental SSE framing and content-type
validation, not a library HTTP client.
+ AiUpstreamResponse framed = new
OpenAiChat().encodeResponse(response, true);
+ return new AiUpstreamResponse(response.getStatusCode(),
responseHeaders(response, "text/event-stream"),
convertStream(framed.getBody()));
+ }
+ Flux<byte[]> body =
response.getBody().reduceWith(ByteArrayOutputStream::new, (buffer, bytes) -> {
+ if (bytes.length > MAX_CONVERSION_BYTES - buffer.size()) {
+ throw new AiProxyException(AiProxyException.Kind.PROTOCOL, "AI
response exceeds the conversion size limit", null);
+ }
+ buffer.writeBytes(bytes);
+ return buffer;
+ }).map(buffer ->
convertCompleteResponse(buffer.toString(StandardCharsets.UTF_8))).flux();
+ return new AiUpstreamResponse(response.getStatusCode(),
responseHeaders(response, "application/json"), body);
+ }
+
+ private byte[] convertCompleteResponse(final String body) {
+ try {
+ JsonNode payload = MAPPER.readTree(body);
+ if (Objects.isNull(payload) || !payload.isObject()) {
+ throw new IllegalArgumentException("Expected an AI response
object");
+ }
+ ConversionResult result =
CONNECT.convertResponse(Format.OPENAI_CHAT, clientFormat,
Json.object(payload.toString()), context);
+ reportDiagnostics(result.diagnostics());
+ return result.json().getBytes(StandardCharsets.UTF_8);
+ } catch (Exception error) {
+ throw new AiProxyException(AiProxyException.Kind.PROTOCOL, "AI
response protocol conversion failed", error);
+ }
+ }
+
+ private Flux<byte[]> convertStream(final Flux<byte[]> framed) {
+ return Flux.defer(() -> {
+ StreamOptions options = new StreamOptions();
+ options.emitSequenceNumber = true;
+ ResponseStreamState state =
CONNECT.newResponseStream(Format.OPENAI_CHAT, clientFormat, options, context);
+ int[] size = {0};
+ return framed.concatMapIterable(bytes -> {
+ // The framer produces exactly one validated event, retaining
multiline JSON.
+ String event = new String(bytes, StandardCharsets.UTF_8);
+ String payload = event.substring(6, event.length() -
2).replace("\ndata: ", "\n");
+ if ("[DONE]".equals(payload)) {
+ return List.<ConversionResult>of();
+ }
+ if (bytes.length > MAX_CONVERSION_BYTES - size[0]) {
+ throw new AiProxyException(AiProxyException.Kind.PROTOCOL,
"AI stream exceeds the conversion size limit", null);
+ }
+ size[0] += bytes.length;
+ return state.accept(payload);
+ }, 1).concatWith(Flux.defer(() -> {
+ List<ConversionResult> terminal = state.finish();
+ reportDiagnostics(state.diagnostics());
+ return Flux.fromIterable(terminal);
+ })).map(this::frameEvent)
+ .onErrorMap(error -> !(error instanceof AiProxyException),
+ error -> new
AiProxyException(AiProxyException.Kind.PROTOCOL, "AI stream protocol conversion
failed", error));
+ });
+ }
+
+ private byte[] frameEvent(final ConversionResult result) {
+ String type = result.value().get("type").getAsString();
+ return ("event: " + type + "\ndata: " + result.json() +
"\n\n").getBytes(StandardCharsets.UTF_8);
+ }
+
+ private Map<String, List<String>> responseHeaders(final AiUpstreamResponse
response, final String contentType) {
+ Map<String, List<String>> headers = new
LinkedHashMap<>(response.getHeaders());
+ headers.keySet().removeIf(name ->
"Content-Length".equalsIgnoreCase(name) || "Content-Type".equalsIgnoreCase(name)
+ || "ETag".equalsIgnoreCase(name) ||
"Content-MD5".equalsIgnoreCase(name) || "Digest".equalsIgnoreCase(name));
+ headers.put("Content-Type", List.of(contentType));
+ return headers;
+ }
+
+ private void reportDiagnostics(final List<Diagnostic> diagnostics) {
+ for (Diagnostic diagnostic : diagnostics) {
+ // Never log payloads, credentials, private reasoning or library
diagnostic messages.
+ LOG.warn("AI conversion diagnostic: code={}, severity={}, from={},
to={}",
+ diagnostic.code(), diagnostic.severity(),
diagnostic.from(), diagnostic.to());
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/OpenAiResponses.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/OpenAiResponses.java
new file mode 100644
index 0000000000..d578125e7f
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/OpenAiResponses.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.ai.proxy.enhanced.protocol;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.shenyu.common.exception.ShenyuException;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.model.ShenyuAiRequest;
+
+import java.util.Objects;
+
+/**
+ * Responses request boundary; Connect owns conversion of its protocol fields.
+ */
+public final class OpenAiResponses implements AiProxyProtocol {
+
+ /**
+ * Selector handle protocol identifier for Responses clients.
+ */
+ public static final String NAME = "openai-responses";
+
+ @Override
+ public String name() {
+ return NAME;
+ }
+
+ @Override
+ public boolean matches(final JsonNode payload) {
+ return Objects.nonNull(payload) && payload.isObject() &&
payload.has("input") && !payload.has("messages");
+ }
+
+ @Override
+ public ShenyuAiRequest decodeRequest(final String rawBody, final
ObjectNode payload) {
+ JsonNode input = payload.get("input");
+ if (Objects.isNull(input) || !(input.isTextual() || input.isArray())) {
+ throw new ShenyuException("OpenAI Responses input must be a string
or array");
+ }
+ JsonNode stream = payload.get("stream");
+ if (Objects.nonNull(stream) && !stream.isNull() &&
!stream.isBoolean()) {
+ throw new ShenyuException("OpenAI Responses stream must be
boolean");
+ }
+ JsonNode model = payload.get("model");
+ if (Objects.nonNull(model) && !model.isNull() && !model.isTextual()) {
+ throw new ShenyuException("OpenAI Responses model must be a
string");
+ }
+ JsonNode maxTokens = payload.get("max_output_tokens");
+ if (Objects.nonNull(maxTokens) && !maxTokens.isNull()
+ && (!maxTokens.isIntegralNumber() ||
!maxTokens.canConvertToInt() || maxTokens.intValue() <= 0)) {
+ throw new ShenyuException("OpenAI Responses max_output_tokens must
be a positive integer");
+ }
+ ShenyuAiRequest request = new ShenyuAiRequest();
+ request.setProtocol(NAME);
+ request.setRawBody(rawBody);
+ request.setRawPayload(payload.deepCopy());
+ request.setModel(Objects.isNull(model) || model.isNull() ? null :
model.textValue());
+ request.setStream(Objects.isNull(stream) || stream.isNull() ? null :
stream.booleanValue());
+ request.setMaxTokens(Objects.isNull(maxTokens) || maxTokens.isNull() ?
null : maxTokens.intValue());
+ return request;
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAi.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAi.java
index 5ced8e7870..c41f491a56 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAi.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAi.java
@@ -19,11 +19,14 @@ package org.apache.shenyu.plugin.ai.proxy.enhanced.provider;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.github.agenticai.connect.ProviderProfile;
import org.apache.shenyu.common.exception.ShenyuException;
import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig;
import org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiUpstreamRequest;
import org.apache.shenyu.plugin.ai.proxy.enhanced.model.ShenyuAiRequest;
import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiChat;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiResponses;
+import
org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.ConnectProtocolAdapter;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
@@ -44,6 +47,24 @@ public final class OpenAi implements AiProxyProvider {
private static final String CHAT_COMPLETIONS_PATH = "/v1/chat/completions";
+ private final ProviderProfile profile;
+
+ /**
+ * Create a generic OpenAI Chat upstream without provider-specific
adaptation.
+ */
+ public OpenAi() {
+ this(null);
+ }
+
+ /**
+ * Create a Chat upstream with an explicitly selected provider dialect.
+ *
+ * @param profile provider profile, or null for generic OpenAI
compatibility
+ */
+ public OpenAi(final ProviderProfile profile) {
+ this.profile = profile;
+ }
+
@Override
public String name() {
return NAME;
@@ -52,7 +73,7 @@ public final class OpenAi implements AiProxyProvider {
@Override
public AiUpstreamRequest buildRequest(final ShenyuAiRequest request, final
AiCommonConfig config) {
validateConfig(config);
- if (!OpenAiChat.NAME.equals(request.getProtocol())) {
+ if (!supports(request.getProtocol())) {
throw new IllegalArgumentException(
"Unsupported OpenAI protocol: " + request.getProtocol());
}
@@ -61,17 +82,17 @@ public final class OpenAi implements AiProxyProvider {
boolean changed = payload.has("fallbackConfig");
payload.remove("fallbackConfig");
- if (Objects.nonNull(config.getModel())) {
- payload.put("model", config.getModel());
- changed = true;
- }
if (Objects.nonNull(config.getTemperature())) {
payload.put("temperature", config.getTemperature());
changed = true;
}
if (Objects.nonNull(config.getMaxTokens())) {
- payload.put("max_tokens", config.getMaxTokens());
- payload.put("max_completion_tokens", config.getMaxTokens());
+ if (OpenAiResponses.NAME.equals(request.getProtocol())) {
+ payload.put("max_output_tokens", config.getMaxTokens());
+ } else {
+ payload.put("max_tokens", config.getMaxTokens());
+ payload.put("max_completion_tokens", config.getMaxTokens());
+ }
changed = true;
}
@@ -82,8 +103,12 @@ public final class OpenAi implements AiProxyProvider {
payload.put("stream", stream);
changed = true;
}
+ // Gateway overrides precede Connect's final protocol/provider
adaptation.
+ final ConnectProtocolAdapter adapter = new
ConnectProtocolAdapter(request.getProtocol(), config.getModel(), profile);
+ final ObjectNode converted = adapter.convertRequest(payload);
+ changed = changed || !payload.equals(converted);
if (stream) {
- changed = enableStreamUsage(payload) || changed;
+ changed = enableStreamUsage(converted) || changed;
}
final Map<String, List<String>> headers =
@@ -107,8 +132,9 @@ public final class OpenAi implements AiProxyProvider {
config.getBaseUrl(),
CHAT_COMPLETIONS_PATH));
upstreamRequest.setHeaders(headers);
- final String body = changed ? payload.toString() :
request.getRawBody();
+ final String body = changed ? converted.toString() :
request.getRawBody();
upstreamRequest.setBody(body.getBytes(StandardCharsets.UTF_8));
+ upstreamRequest.setResponseAdapter(response ->
adapter.convertResponse(response, stream));
return upstreamRequest;
}
@@ -158,6 +184,6 @@ public final class OpenAi implements AiProxyProvider {
@Override
public boolean supports(final String protocol) {
- return OpenAiChat.NAME.equals(protocol);
+ return OpenAiChat.NAME.equals(protocol) ||
OpenAiResponses.NAME.equals(protocol);
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAiCompatible.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAiCompatible.java
index 5761ac63f5..06daf73886 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAiCompatible.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/provider/OpenAiCompatible.java
@@ -17,6 +17,7 @@
package org.apache.shenyu.plugin.ai.proxy.enhanced.provider;
+import io.github.agenticai.connect.ProviderProfile;
import org.apache.shenyu.plugin.ai.common.config.AiCommonConfig;
import org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiUpstreamRequest;
import org.apache.shenyu.plugin.ai.proxy.enhanced.model.ShenyuAiRequest;
@@ -28,10 +29,12 @@ public final class OpenAiCompatible implements
AiProxyProvider {
private final String name;
- private final OpenAi delegate = new OpenAi();
+ private final OpenAi delegate;
public OpenAiCompatible(final String name) {
this.name = name;
+ // Only an explicitly selected DeepSeek provider enables its dialect
rules.
+ this.delegate = new OpenAi("DeepSeek".equalsIgnoreCase(name) ?
ProviderProfile.deepSeekChat() : null);
}
@Override
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngine.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngine.java
index 90da0a59b5..b25664b097 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngine.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/main/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngine.java
@@ -87,6 +87,8 @@ public final class AiProxyEngine {
Optional<AiCommonConfig> fallback = fallbackConfig.map(config -> {
AiCommonConfig copy = new AiCommonConfig(config);
copy.setStream(primary.stream());
+ // Fallback changes the upstream, never the already selected
client protocol.
+ copy.setProtocol(primary.protocol().name());
return copy;
});
return attempt(requestBody, primary, fallback, primary.stream() ?
1 : 3, responseConsumer, new AtomicBoolean());
@@ -114,7 +116,7 @@ public final class AiProxyEngine {
// Release this response without accumulating the
error body before another attempt.
return response.getBody().then(Mono.error(new
UpstreamStatusFailure(status)));
}
- AiUpstreamResponse adapted =
prepared.provider().adaptResponse(response);
+ AiUpstreamResponse adapted =
prepared.request().adaptResponse(prepared.provider().adaptResponse(response));
AiUpstreamResponse encoded =
prepared.protocol().encodeResponse(adapted, prepared.stream());
return encoded.getBody().switchOnFirst((signal, body) -> {
if (signal.isOnError()) {
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyHttpCompatibilityTest.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyHttpCompatibilityTest.java
index 5c82bc4fc4..f3e2616b76 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyHttpCompatibilityTest.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/AiProxyHttpCompatibilityTest.java
@@ -17,6 +17,8 @@
package org.apache.shenyu.plugin.ai.proxy.enhanced;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.dto.RuleData;
import org.apache.shenyu.common.dto.SelectorData;
@@ -24,6 +26,7 @@ import
org.apache.shenyu.common.dto.convert.rule.AiProxyHandle;
import org.apache.shenyu.plugin.ai.proxy.enhanced.handler.AiProxyPluginHandler;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.AiProxyProtocolFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiChat;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiResponses;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.provider.AiProxyProviderFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.provider.OpenAi;
import org.apache.shenyu.plugin.ai.proxy.enhanced.service.AiProxyConfigService;
@@ -47,6 +50,7 @@ import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -59,6 +63,76 @@ class AiProxyHttpCompatibilityTest {
private static final Duration TIMEOUT = Duration.ofSeconds(5);
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String CHAT_REPLY =
"{\"id\":\"chatcmpl-http\",\"model\":\"model\",\"choices\":[{\"index\":0,"
+ +
"\"message\":{\"role\":\"assistant\",\"content\":\"hello\"},\"finish_reason\":\"stop\"}],"
+ +
"\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"total_tokens\":5}}";
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void convertsResponsesThroughThePluginAndRealHttpTransport(final boolean
stream) throws Exception {
+ AtomicReference<String> forwarded = new AtomicReference<>();
+ String reply = stream ? "data:
{\"id\":\"chatcmpl-http\",\"model\":\"model\",\"choices\":[{\"index\":0,"
+ +
"\"delta\":{\"role\":\"assistant\",\"content\":\"hello\"},\"finish_reason\":\"stop\"}]}\n\n"
+ + "data:
{\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2}}\n\ndata:
[DONE]\n\n" : CHAT_REPLY;
+ DisposableServer server =
HttpServer.create().host("127.0.0.1").port(0).handle((request, response) -> {
+ assertEquals("/v1/chat/completions", request.uri());
+ assertEquals("Bearer real-key",
request.requestHeaders().get("Authorization"));
+ return request.receive().aggregate().asString().flatMap(body -> {
+ forwarded.set(body);
+ return response.header("Content-Type", stream ?
"text/event-stream" : "application/json")
+ .header("ETag",
"old-chat-representation").header("X-Upstream",
"retained").sendString(Flux.just(reply)).then();
+ });
+ }).bindNow(TIMEOUT);
+ try {
+ MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.post("/v1/responses")
+
.body("{\"model\":\"client-alias\",\"input\":\"hi\",\"max_output_tokens\":12,\"stream\":"
+ stream + "}"));
+ StepVerifier.create(plugin(server, stream,
OpenAiResponses.NAME).doExecute(exchange, mock(ShenyuPluginChain.class),
selector(), new RuleData())).verifyComplete();
+ JsonNode upstream = MAPPER.readTree(forwarded.get());
+ assertEquals("model", upstream.get("model").asText());
+ assertEquals("hi", upstream.at("/messages/0/content").asText());
+ assertEquals(12, upstream.get("max_completion_tokens").asInt());
+ assertTrue(!upstream.has("input"));
+ String actual =
exchange.getResponse().getBodyAsString().block(TIMEOUT);
+ assertEquals(200, exchange.getResponse().getStatusCode().value());
+ assertEquals("retained",
exchange.getResponse().getHeaders().getFirst("X-Upstream"));
+
assertTrue(!exchange.getResponse().getHeaders().containsKey("ETag"));
+ if (stream) {
+ assertTrue(actual.contains("event:
response.output_text.delta\n"));
+ assertEquals(1, actual.split("event: response.completed\n",
-1).length - 1);
+ assertTrue(actual.contains("\"input_tokens\":3"));
+ assertTrue(!actual.contains("[DONE]"));
+
assertTrue(upstream.at("/stream_options/include_usage").asBoolean());
+ } else {
+ JsonNode result = MAPPER.readTree(actual);
+ assertEquals("response", result.get("object").asText());
+ assertEquals("hello",
result.at("/output/0/content/0/text").asText());
+ assertEquals(5, result.at("/usage/total_tokens").asInt());
+ }
+ } finally {
+ server.disposeNow(TIMEOUT);
+ }
+ }
+
+ @Test
+ void rejectsResponsesStateBeforeOpeningAnUpstreamConnection() {
+ AtomicBoolean called = new AtomicBoolean();
+ DisposableServer server =
HttpServer.create().host("127.0.0.1").port(0).handle((request, response) -> {
+ called.set(true);
+ return response.send();
+ }).bindNow(TIMEOUT);
+ try {
+ MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.post("/v1/responses")
+
.body("{\"input\":\"hi\",\"previous_response_id\":\"resp_previous\"}"));
+ StepVerifier.create(plugin(server, false,
OpenAiResponses.NAME).doExecute(exchange, mock(ShenyuPluginChain.class),
selector(), new RuleData())).verifyComplete();
+ assertEquals(400, exchange.getResponse().getStatusCode().value());
+ assertTrue(!called.get());
+ } finally {
+ server.disposeNow(TIMEOUT);
+ }
+ }
+
@ParameterizedTest
@ValueSource(booleans = {false, true})
void
forwardsTheFirstChunkWithoutWaitingForCompletionAndCancelsTheSocket(final
boolean stream) {
@@ -87,6 +161,32 @@ class AiProxyHttpCompatibilityTest {
}
}
+ @Test
+ void cancelsTheResponsesUpstreamSocketWithoutCompletingTheStream() {
+ AtomicBoolean cancelled = new AtomicBoolean();
+ String first = "data:
{\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hello\"}}]}\n\n";
+ DisposableServer server =
HttpServer.create().host("127.0.0.1").port(0).handle((request, response) ->
+ response.header("Content-Type", "text/event-stream")
+
.sendString(Flux.just(first).concatWith(Flux.never()).doOnCancel(() ->
cancelled.set(true)))).bindNow(TIMEOUT);
+ try {
+ MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.post("/v1/responses")
+ .body("{\"input\":\"hi\",\"stream\":true}"));
+ List<String> received = new
java.util.concurrent.CopyOnWriteArrayList<>();
+ exchange.getResponse().setWriteHandler(body ->
body.doOnNext(buffer -> {
+ received.add(buffer.toString(StandardCharsets.UTF_8));
+ DataBufferUtils.release(buffer);
+ }).then());
+ StepVerifier.create(plugin(server, true,
OpenAiResponses.NAME).doExecute(exchange, mock(ShenyuPluginChain.class),
selector(), new RuleData()))
+ .then(() ->
org.awaitility.Awaitility.await().atMost(TIMEOUT)
+ .until(() -> received.stream().anyMatch(event ->
event.startsWith("event: response.output_text.delta\n"))))
+ .thenCancel().verify(TIMEOUT);
+
org.awaitility.Awaitility.await().atMost(TIMEOUT).untilTrue(cancelled);
+ assertTrue(received.stream().noneMatch(event ->
event.startsWith("event: response.completed\n")));
+ } finally {
+ server.disposeNow(TIMEOUT);
+ }
+ }
+
@Test
void forwardsNormalBodiesBeyondTheDefaultWebClientCodecLimit() {
String responseBody = "{\"unknown\":\"" + "x".repeat(512 * 1024) +
"\"}";
@@ -105,15 +205,20 @@ class AiProxyHttpCompatibilityTest {
}
private AiProxyPlugin plugin(final DisposableServer server, final boolean
stream) {
+ return plugin(server, stream, OpenAiChat.NAME);
+ }
+
+ private AiProxyPlugin plugin(final DisposableServer server, final boolean
stream, final String protocol) {
final AiProxyPluginHandler handler = new AiProxyPluginHandler();
AiProxyHandle handle = new AiProxyHandle();
handle.setProvider("OpenAI");
handle.setBaseUrl("http://127.0.0.1:"; + server.port());
handle.setApiKey("real-key");
handle.setStream(stream);
+ handle.setProtocol(protocol);
handle.setModel("model");
handler.getSelectorCachedHandle().cachedHandle(CacheKeyUtils.INST.getKey("http-selector",
Constants.DEFAULT_RULE), handle);
- AiProxyEngine engine = new AiProxyEngine(new
AiProxyProtocolFactory(List.of(new OpenAiChat())), new
AiProxyProviderFactory(List.of(new OpenAi())),
+ AiProxyEngine engine = new AiProxyEngine(new
AiProxyProtocolFactory(List.of(new OpenAiChat(), new OpenAiResponses())), new
AiProxyProviderFactory(List.of(new OpenAi())),
new WebClientAiProxyTransport(WebClient.builder().build()),
Duration.ZERO);
return new AiProxyPlugin(new AiProxyConfigService(), engine, handler);
}
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/ConnectProtocolAdapterTest.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/ConnectProtocolAdapterTest.java
new file mode 100644
index 0000000000..dab3b60d1f
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/ConnectProtocolAdapterTest.java
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.ai.proxy.enhanced.protocol;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.github.agenticai.connect.ProviderProfile;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiProxyException;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiUpstreamResponse;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Library contracts at ShenYu's JSON/reactive boundary, without live-provider
calls.
+ */
+class ConnectProtocolAdapterTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String RESPONSE =
"{\"id\":\"chatcmpl-test\",\"model\":\"upstream\",\"choices\":[{\"index\":0,"
+ +
"\"message\":{\"role\":\"assistant\",\"content\":\"hello\"},\"finish_reason\":\"stop\"}],"
+ +
"\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"total_tokens\":5}}";
+
+ private static final String CHUNK =
"{\"id\":\"chatcmpl-test\",\"model\":\"upstream\",\"choices\":[{\"index\":0,"
+ +
"\"delta\":{\"role\":\"assistant\",\"content\":\"hello\"},\"finish_reason\":null}]}";
+
+ @Test
+ void delegatesRequestConversionAndProviderAdaptationExactlyOnce() throws
Exception {
+ AtomicInteger calls = new AtomicInteger();
+ ProviderProfile profile = (payload, adaptation) -> {
+ calls.incrementAndGet();
+ assertEquals("routed", payload.get("model").getAsString());
+ payload.addProperty("custom", true);
+ adaptation.warning("custom_profile", "custom", "Test profile");
+ };
+ ObjectNode input = object("{\"model\":\"alias\",\"instructions\":\"be
helpful\",\"input\":\"hello\",\"max_output_tokens\":12}");
+ ObjectNode before = input.deepCopy();
+ ObjectNode result = new ConnectProtocolAdapter(OpenAiResponses.NAME,
"routed", profile).convertRequest(input);
+ assertEquals(1, calls.get());
+ assertEquals(before, input);
+ assertEquals("system", result.at("/messages/0/role").asText());
+ assertEquals("hello", result.at("/messages/1/content").asText());
+ assertEquals(12, result.get("max_completion_tokens").asInt());
+ assertTrue(result.get("custom").asBoolean());
+ }
+
+ @Test
+ void keepsGenericIdentityFieldsAndRejectsUnrepresentableResponsesState()
throws Exception {
+ ObjectNode input =
object("{\"messages\":[{\"role\":\"user\"}],\"vendor\":true,\"content\":null}");
+ assertEquals(input, new ConnectProtocolAdapter(OpenAiChat.NAME, null,
null).convertRequest(input));
+ ConnectProtocolAdapter adapter = adapter();
+ for (String body :
List.of("{\"model\":\"x\",\"input\":\"hi\",\"previous_response_id\":\"resp_old\"}",
"{\"input\":\"hi\"}")) {
+ AiProxyException error = assertThrows(AiProxyException.class, ()
-> adapter.convertRequest(object(body)));
+ assertEquals(AiProxyException.Kind.REQUEST, error.getKind());
+ }
+ }
+
+ @Test
+ void adaptsDeepSeekTokensAndReasoningWithoutGuessingFromAModel() throws
Exception {
+ ObjectNode input =
object("{\"model\":\"deepseek-model\",\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],"
+ +
"\"max_completion_tokens\":20,\"max_tokens\":10,\"reasoning_effort\":\"medium\"}");
+ assertEquals(input, new ConnectProtocolAdapter(OpenAiChat.NAME, null,
null).convertRequest(input));
+ ObjectNode result = new ConnectProtocolAdapter(OpenAiChat.NAME, null,
ProviderProfile.deepSeekChat()).convertRequest(input);
+ assertEquals(20, result.get("max_tokens").asInt());
+ assertFalse(result.has("max_completion_tokens"));
+ assertEquals("high", result.get("reasoning_effort").asText());
+ assertEquals("enabled", result.at("/thinking/type").asText());
+ input.put("tool_choice", "required");
+ assertThrows(AiProxyException.class,
+ () -> new ConnectProtocolAdapter(OpenAiChat.NAME, null,
ProviderProfile.deepSeekChat()).convertRequest(input));
+ }
+
+ @Test
+ void convertsCompleteResponsesAndRemovesStaleRepresentationHeaders()
throws Exception {
+ AiUpstreamResponse upstream = new AiUpstreamResponse(200,
Map.of("content-length", List.of("123"), "ETag", List.of("old"),
+ "Content-MD5", List.of("old"), "Digest", List.of("old"),
"X-Request-ID", List.of("kept")),
+ Flux.just(bytes(RESPONSE.substring(0, 20)),
bytes(RESPONSE.substring(20))));
+ AiUpstreamResponse converted = adapter().convertResponse(upstream,
false);
+ assertEquals(Map.of("Content-Type", List.of("application/json"),
"X-Request-ID", List.of("kept")), converted.getHeaders());
+ JsonNode result =
MAPPER.readTree(converted.getBody().single().block(Duration.ofSeconds(3)));
+ assertEquals("response", result.get("object").asText());
+ assertEquals("hello", result.at("/output/0/content/0/text").asText());
+ assertEquals(3, result.at("/usage/input_tokens").asInt());
+ assertEquals(2, result.at("/usage/output_tokens").asInt());
+ }
+
+ @Test
+ void restoresCustomToolsUsingTheOriginalRequestContext() throws Exception {
+ ConnectProtocolAdapter adapter = adapter();
+ ObjectNode converted =
adapter.convertRequest(object("{\"model\":\"x\",\"input\":\"hi\",\"tools\":[{\"type\":\"custom\",\"name\":\"execute\"}]}"));
+ String name = converted.at("/tools/0/function/name").asText();
+ String body =
"{\"model\":\"x\",\"choices\":[{\"index\":0,\"finish_reason\":\"tool_calls\",\"message\":{\"role\":\"assistant\","
+ +
"\"tool_calls\":[{\"id\":\"call_one\",\"type\":\"function\",\"function\":{\"name\":\""
+ name + "\",\"arguments\":\"{\\\"input\\\":\\\"print(1)\\\"}\"}}]}}]}";
+ JsonNode result =
MAPPER.readTree(adapter.convertResponse(response(200, Flux.just(bytes(body))),
false).getBody().single().block());
+ assertEquals("custom_tool_call", result.at("/output/0/type").asText());
+ assertEquals("execute", result.at("/output/0/name").asText());
+ assertEquals("print(1)", result.at("/output/0/input").asText());
+ JsonNode unrelated =
MAPPER.readTree(adapter().convertResponse(response(200,
Flux.just(bytes(body))), false).getBody().single().block());
+ assertEquals("function_call", unrelated.at("/output/0/type").asText());
+ }
+
+ @Test
+ void restoresFragmentedCustomToolStreamsWithoutCrossRequestState() throws
Exception {
+ ConnectProtocolAdapter custom = adapter();
+ ObjectNode request =
custom.convertRequest(object("{\"model\":\"x\",\"input\":\"hi\",\"tools\":[{\"type\":\"custom\",\"name\":\"execute\"}]}"));
+ String name = request.at("/tools/0/function/name").asText();
+ ObjectNode start =
object("{\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_one\",\"type\":\"function\",\"function\":{}}]}}]}");
+ ObjectNode function = (ObjectNode)
start.at("/choices/0/delta/tool_calls/0/function");
+ function.put("name", name);
+ function.put("arguments", "{\"input\":\"print(");
+ ObjectNode next = start.deepCopy();
+ ObjectNode continuation = (ObjectNode)
next.at("/choices/0/delta/tool_calls/0");
+ continuation.remove(List.of("id", "type"));
+ ((ObjectNode) continuation.get("function")).remove("name");
+ ((ObjectNode) continuation.get("function")).put("arguments", "1)\"}");
+ Flux<byte[]> chunks = Flux.just(bytes("data: " + start + "\n\n"),
bytes("data: " + next + "\n\n"),
+ bytes("data:
{\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n"));
+ Flux<String> restored = custom.convertResponse(response(200, chunks),
true).getBody().map(bytes -> new String(bytes, StandardCharsets.UTF_8));
+ Flux<String> isolated = adapter().convertResponse(response(200,
chunks), true).getBody().map(bytes -> new String(bytes,
StandardCharsets.UTF_8));
+ List<List<String>> results =
reactor.core.publisher.Mono.zip(restored.collectList(), isolated.collectList(),
List::of).block(Duration.ofSeconds(3));
+ JsonNode customResult = terminalResponse(results.get(0));
+ JsonNode genericResult = terminalResponse(results.get(1));
+ assertEquals("custom_tool_call",
customResult.at("/response/output/0/type").asText());
+ assertEquals("print(1)",
customResult.at("/response/output/0/input").asText());
+ assertEquals("function_call",
genericResult.at("/response/output/0/type").asText());
+ assertEquals("{\"input\":\"print(1)\"}",
genericResult.at("/response/output/0/arguments").asText());
+ }
+
+ private JsonNode terminalResponse(final List<String> events) throws
Exception {
+ String terminal = events.get(events.size() - 1);
+ return MAPPER.readTree(terminal.substring(terminal.indexOf("data: ") +
6).trim());
+ }
+
+ @Test
+ void preservesHttpErrorsAndChatResponsesWithoutConsumingThem() {
+ for (int status : List.of(199, 400, 500)) {
+ AiUpstreamResponse upstream = response(status, Flux.error(new
AssertionError("Body must not be consumed")));
+ assertSame(upstream, adapter().convertResponse(upstream, false));
+ assertSame(upstream, adapter().convertResponse(upstream, true));
+ }
+ AiUpstreamResponse upstream = response(200, Flux.empty());
+ assertSame(upstream, new ConnectProtocolAdapter(OpenAiChat.NAME, null,
null).convertResponse(upstream, true));
+ }
+
+ @Test
+ void rejectsMalformedEmptyTrailingAndOversizedCompleteResponses() {
+ for (String body : List.of("invalid", "[]", "", "null", RESPONSE + "
{}", "x".repeat(5 * 1024 * 1024 + 1))) {
+ StepVerifier.create(adapter().convertResponse(response(200,
Flux.just(bytes(body))), false).getBody())
+ .expectErrorMatches(error -> error instanceof
AiProxyException && ((AiProxyException) error).getKind() ==
AiProxyException.Kind.PROTOCOL).verify();
+ }
+ StepVerifier.create(adapter().convertResponse(response(200,
Flux.empty()), false).getBody()).expectError(AiProxyException.class).verify();
+ }
+
+ @Test
+ void convertsFragmentedMultilineSseAndProducesOneTerminalEvent() {
+ String wire = ": heartbeat\r\ndata: {\r\ndata: " + CHUNK.substring(1)
+ "\r\n\r\n"
+ + "data:
{\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2}}\r\n\r\ndata:
[DONE]\r\n\r\n";
+ List<byte[]> fragments = new ArrayList<>();
+ for (byte value : bytes(wire)) {
+ fragments.add(new byte[] {value});
+ }
+ List<String> events =
stream(Flux.fromIterable(fragments)).collectList().block(Duration.ofSeconds(3));
+ assertTrue(events.stream().anyMatch(event -> event.startsWith("event:
response.output_text.delta\n") && event.contains("hello")));
+ assertEquals(1, events.stream().filter(event ->
event.startsWith("event: response.completed\n")).count());
+ assertTrue(events.get(events.size() -
1).contains("\"input_tokens\":3"));
+ assertTrue(events.stream().allMatch(event ->
event.contains("\"sequence_number\":") && !event.contains("[DONE]")));
+ }
+
+ @Test
+ void isolatesStreamStateOnResubscriptionAndCancelsWithoutATerminalEvent() {
+ Flux<String> reusable = stream(Flux.just(bytes("data: " + CHUNK +
"\n\n")));
+ for (int attempt = 0; attempt < 2; attempt++) {
+ List<String> events = reusable.collectList().block();
+ assertEquals(1, events.stream().filter(event ->
event.startsWith("event: response.completed\n")).count());
+ }
+ AtomicBoolean cancelled = new AtomicBoolean();
+ Flux<String> live = stream(Flux.just(bytes("data: " + CHUNK +
"\n\n")).concatWith(Flux.never()).doOnCancel(() -> cancelled.set(true)));
+ StepVerifier.create(live, 0).thenRequest(1).expectNextMatches(event ->
event.startsWith("event: response.created\n"))
+ .thenCancel().verify(Duration.ofSeconds(3));
+ assertTrue(cancelled.get());
+ }
+
+ @Test
+ void doesNotSynthesizeSuccessAfterMalformedEventsOrTransportFailure() {
+ for (Flux<byte[]> body : List.of(Flux.just(bytes("data: " + CHUNK +
"\n\ndata: invalid\n\n")),
+ Flux.just(bytes("data: " + CHUNK +
"\n\n")).concatWith(Flux.error(new IllegalStateException("socket closed"))))) {
+ List<String> events = new ArrayList<>();
+
StepVerifier.create(stream(body).doOnNext(events::add)).thenConsumeWhile(event
-> !event.startsWith("event: response.completed\n"))
+ .expectError(AiProxyException.class).verify();
+ assertTrue(events.stream().noneMatch(event ->
event.startsWith("event: response.completed\n")));
+ }
+ AiUpstreamResponse json = new AiUpstreamResponse(200,
Map.of("Content-Type", List.of("application/json")),
Flux.just(bytes(RESPONSE)));
+ StepVerifier.create(adapter().convertResponse(json,
true).getBody()).expectError(AiProxyException.class).verify();
+ }
+
+ @Test
+ void boundsCumulativeConversionStateAndCancelsAtDone() {
+ String large = "data: " + CHUNK.replace("hello", "x".repeat(1024 *
1024)) + "\n\n";
+ StepVerifier.create(stream(Flux.range(0, 6).map(index ->
bytes(large))))
+ .thenConsumeWhile(event -> !event.startsWith("event:
response.completed\n")).expectError(AiProxyException.class).verify();
+ AtomicBoolean cancelled = new AtomicBoolean();
+ List<String> events = stream(Flux.just(bytes("data:
[DONE]\n\n")).concatWith(Flux.never()).doOnCancel(() -> cancelled.set(true)))
+ .collectList().block(Duration.ofSeconds(3));
+ assertTrue(cancelled.get());
+ assertEquals(1, events.stream().filter(event ->
event.startsWith("event: response.completed\n")).count());
+ }
+
+ private ConnectProtocolAdapter adapter() {
+ return new ConnectProtocolAdapter(OpenAiResponses.NAME, null, null);
+ }
+
+ private ObjectNode object(final String json) throws Exception {
+ return (ObjectNode) MAPPER.readTree(json);
+ }
+
+ private AiUpstreamResponse response(final int status, final Flux<byte[]>
body) {
+ return new AiUpstreamResponse(status, Map.of(), body);
+ }
+
+ private Flux<String> stream(final Flux<byte[]> body) {
+ return adapter().convertResponse(response(200, body),
true).getBody().map(bytes -> new String(bytes, StandardCharsets.UTF_8));
+ }
+
+ private byte[] bytes(final String value) {
+ return value.getBytes(StandardCharsets.UTF_8);
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/OpenAiResponsesTest.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/OpenAiResponsesTest.java
new file mode 100644
index 0000000000..776f4b9e2b
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/protocol/OpenAiResponsesTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.ai.proxy.enhanced.protocol;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.shenyu.common.exception.ShenyuException;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.model.ShenyuAiRequest;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Responses shape detection and gateway envelope validation.
+ */
+class OpenAiResponsesTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private final OpenAiResponses protocol = new OpenAiResponses();
+
+ @Test
+ void detectsResponsesWithoutAmbiguousChatRequests() throws Exception {
+ assertEquals(OpenAiResponses.NAME, protocol.name());
+ assertFalse(protocol.matches(null));
+ assertFalse(protocol.matches(MAPPER.readTree("[]")));
+ assertFalse(protocol.matches(MAPPER.readTree("{}")));
+
assertFalse(protocol.matches(MAPPER.readTree("{\"input\":\"hi\",\"messages\":[]}")));
+ assertTrue(protocol.matches(MAPPER.readTree("{\"input\":\"hi\"}")));
+ assertEquals(protocol, new AiProxyProtocolFactory(List.of(new
OpenAiChat(), protocol)).detect(MAPPER.readTree("{\"input\":\"hi\"}")));
+ }
+
+ @Test
+ void retainsTheOriginalPayloadAndOptionalFields() throws Exception {
+ String body =
"{\"input\":[],\"model\":\"alias\",\"stream\":true,\"max_output_tokens\":100,\"unknown\":true}";
+ ObjectNode payload = (ObjectNode) MAPPER.readTree(body);
+ ShenyuAiRequest request = protocol.decodeRequest(body, payload);
+ payload.remove("unknown");
+ assertTrue(request.getRawPayload().has("unknown"));
+ assertEquals(body, request.getRawBody());
+ assertEquals("alias", request.getModel());
+ assertEquals(100, request.getMaxTokens());
+ assertTrue(request.isStream());
+ assertTrue(request.hasStream());
+ for (String minimal : List.of("{\"input\":\"hi\"}",
"{\"input\":\"hi\",\"model\":null,\"stream\":null,\"max_output_tokens\":null}"))
{
+ ShenyuAiRequest optional = protocol.decodeRequest(minimal,
(ObjectNode) MAPPER.readTree(minimal));
+ assertNull(optional.getModel());
+ assertNull(optional.getMaxTokens());
+ assertFalse(optional.hasStream());
+ }
+ }
+
+ @Test
+ void rejectsInvalidEnvelopeFields() {
+ for (String body : List.of("{}", "{\"input\":null}", "{\"input\":1}",
"{\"input\":{}}", "{\"input\":\"hi\",\"stream\":\"yes\"}",
+ "{\"input\":\"hi\",\"model\":42}",
"{\"input\":\"hi\",\"max_output_tokens\":0}",
"{\"input\":\"hi\",\"max_output_tokens\":1.5}",
+ "{\"input\":\"hi\",\"max_output_tokens\":99999999999}")) {
+ assertThrows(ShenyuException.class, () ->
protocol.decodeRequest(body, (ObjectNode) MAPPER.readTree(body)));
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngineOrchestrationTest.java
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngineOrchestrationTest.java
index 041236636c..b31d6d5116 100644
---
a/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngineOrchestrationTest.java
+++
b/shenyu-plugin/shenyu-plugin-ai/shenyu-plugin-ai-proxy/src/test/java/org/apache/shenyu/plugin/ai/proxy/enhanced/service/AiProxyEngineOrchestrationTest.java
@@ -25,6 +25,7 @@ import
org.apache.shenyu.plugin.ai.proxy.enhanced.model.AiUpstreamResponse;
import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.AiProxyProtocol;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.AiProxyProtocolFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiChat;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiResponses;
import org.apache.shenyu.plugin.ai.proxy.enhanced.provider.AiProxyProvider;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.provider.AiProxyProviderFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.provider.OpenAi;
@@ -45,6 +46,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -57,6 +59,50 @@ class AiProxyEngineOrchestrationTest {
private static final String BODY =
"{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"unknown\":42}";
+ @Test
+ void convertsResponsesAfterRetryAndFallbackWithoutSharingStreamState() {
+ List<AiUpstreamRequest> requests = new ArrayList<>();
+ AiCommonConfig fallback = config(false, "fallback");
+ fallback.setProvider("DeepSeek");
+ fallback.setMaxTokens(23);
+ AiProxyEngine engine = engine(request -> {
+ requests.add(request);
+ return Mono.just(requests.size() <= 2 ? response(503,
Flux.just(bytes("busy"))) : response(200,
+ Flux.just(bytes("data:
{\"model\":\"fallback\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n"))));
+ });
+ AiCommonConfig primary = config(false, "main");
+ primary.setProtocol(OpenAiResponses.NAME);
+ List<String> received = new ArrayList<>();
+
StepVerifier.create(engine.execute("{\"input\":\"hi\",\"stream\":true}",
primary, Optional.of(fallback),
+ response -> consume(response, received))).verifyComplete();
+ assertEquals(3, requests.size());
+ assertArrayEquals(requests.get(0).getBody(),
requests.get(1).getBody());
+ String body = new String(requests.get(2).getBody(),
StandardCharsets.UTF_8);
+ assertTrue(body.contains("\"model\":\"fallback\""));
+ assertTrue(body.contains("\"max_tokens\":23"));
+ assertFalse(body.contains("max_completion_tokens"));
+ assertEquals(1, received.stream().filter(event ->
event.startsWith("event: response.created\n")).count());
+ assertEquals(1, received.stream().filter(event ->
event.startsWith("event: response.completed\n")).count());
+ }
+
+ @Test
+ void doesNotRetryResponsesAfterAConvertedEvent() {
+ AtomicInteger calls = new AtomicInteger();
+ List<String> received = new ArrayList<>();
+ AiProxyEngine engine = engine(request -> {
+ calls.incrementAndGet();
+ return Mono.just(response(200, Flux.just(bytes("data:
{\"choices\":[{\"index\":0,\"delta\":{\"content\":\"first\"}}]}\n\n"))
+ .concatWith(Flux.error(networkFailure()))));
+ });
+ AiCommonConfig primary = config(true, "main");
+ primary.setProtocol(OpenAiResponses.NAME);
+
StepVerifier.create(engine.execute("{\"input\":\"hi\",\"stream\":true}",
primary, Optional.of(config(true, "fallback")),
+ response -> consume(response,
received))).expectError(AiProxyException.class).verify();
+ assertEquals(1, calls.get());
+ assertTrue(received.stream().anyMatch(event ->
event.startsWith("event: response.created\n")));
+ assertTrue(received.stream().noneMatch(event ->
event.startsWith("event: response.completed\n")));
+ }
+
@Test
void normalRequestsRetryThreeTimesAndPreserveTheFinalErrorBody() {
AtomicInteger calls = new AtomicInteger();
@@ -231,7 +277,7 @@ class AiProxyEngineOrchestrationTest {
}
private AiProxyEngine engine(final AiProxyTransport transport) {
- return new AiProxyEngine(new AiProxyProtocolFactory(List.of(new
OpenAiChat())),
+ return new AiProxyEngine(new AiProxyProtocolFactory(List.of(new
OpenAiChat(), new OpenAiResponses())),
new AiProxyProviderFactory(List.of(new OpenAi(), new
OpenAiCompatible("DeepSeek"))), transport, Duration.ZERO);
}
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/main/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfiguration.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/main/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfiguration.java
index 0678e727b0..1708b69a2e 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/main/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfiguration.java
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/main/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfiguration.java
@@ -22,6 +22,7 @@ import
org.apache.shenyu.plugin.ai.proxy.enhanced.handler.AiProxyPluginHandler;
import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.AiProxyProtocol;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.AiProxyProtocolFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiChat;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiResponses;
import org.apache.shenyu.plugin.ai.proxy.enhanced.provider.AiProxyProvider;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.provider.AiProxyProviderFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.provider.OpenAi;
@@ -119,6 +120,16 @@ public class AiProxyPluginConfiguration {
return new OpenAiChat();
}
+ /**
+ * OpenAI Responses protocol backed by Connect conversion to Chat
upstreams.
+ *
+ * @return the OpenAI Responses protocol
+ */
+ @Bean
+ public OpenAiResponses openAiResponses() {
+ return new OpenAiResponses();
+ }
+
/**
* Ai proxy protocol factory.
*
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/test/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfigurationTest.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/test/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfigurationTest.java
index 5ea1bf813e..80aad054f2 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/test/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfigurationTest.java
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-ai-proxy/src/test/java/org/apache/shenyu/springboot/starter/plugin/ai/proxy/AiProxyPluginConfigurationTest.java
@@ -19,6 +19,8 @@ package org.apache.shenyu.springboot.starter.plugin.ai.proxy;
import
org.apache.shenyu.plugin.ai.common.spring.ai.registry.AiModelFactoryRegistry;
import org.apache.shenyu.plugin.ai.proxy.enhanced.AiProxyPlugin;
+import
org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.AiProxyProtocolFactory;
+import org.apache.shenyu.plugin.ai.proxy.enhanced.protocol.OpenAiResponses;
import
org.apache.shenyu.plugin.ai.proxy.enhanced.provider.AiProxyProviderFactory;
import org.apache.shenyu.plugin.ai.proxy.enhanced.service.AiProxyEngine;
import org.apache.shenyu.plugin.ai.proxy.enhanced.transport.AiProxyTransport;
@@ -42,8 +44,10 @@ class AiProxyPluginConfigurationTest {
org.assertj.core.api.Assertions.assertThat(context).hasNotFailed().hasSingleBean(AiProxyPlugin.class).hasSingleBean(AiProxyEngine.class)
.doesNotHaveBean(AiModelFactoryRegistry.class);
AiProxyProviderFactory factory =
context.getBean(AiProxyProviderFactory.class);
+ assertSame(context.getBean(OpenAiResponses.class),
context.getBean(AiProxyProtocolFactory.class).getProtocol(OpenAiResponses.NAME));
for (String provider : List.of("OpenAI", "DeepSeek", "ALiYun",
"OpenAPI", "Moonshot")) {
assertEquals(provider.toLowerCase(java.util.Locale.ROOT),
factory.getProvider(provider).name().toLowerCase(java.util.Locale.ROOT));
+
org.junit.jupiter.api.Assertions.assertTrue(factory.getProvider(provider).supports(OpenAiResponses.NAME));
}
});
}
