This is an automated email from the ASF dual-hosted git repository.
nic-6443 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new 77086b0db fix(ai-proxy): do not turn a streaming read error after
partial output into a 5xx (#13876)
77086b0db is described below
commit 77086b0db0dc91a120d0aa21834e6e41e4823d93
Author: Nic <[email protected]>
AuthorDate: Wed Aug 26 14:35:54 2026 +0800
fix(ai-proxy): do not turn a streaming read error after partial output into
a 5xx (#13876)
---
apisix/plugins/ai-aliyun-content-moderation.lua | 7 +-
apisix/plugins/ai-aws-content-moderation.lua | 7 +-
apisix/plugins/ai-providers/base.lua | 41 ++-
docs/en/latest/plugins/ai-proxy-multi.md | 2 +
docs/en/latest/plugins/ai-proxy.md | 2 +
t/lib/truncated_sse.lua | 114 +++++++
t/plugin/ai-proxy-stream-truncated.t | 375 ++++++++++++++++++++++++
7 files changed, 545 insertions(+), 3 deletions(-)
diff --git a/apisix/plugins/ai-aliyun-content-moderation.lua
b/apisix/plugins/ai-aliyun-content-moderation.lua
index 8778b754d..fe7748cb2 100644
--- a/apisix/plugins/ai-aliyun-content-moderation.lua
+++ b/apisix/plugins/ai-aliyun-content-moderation.lua
@@ -530,7 +530,12 @@ function _M.lua_body_filter(conf, ctx, headers, body)
end
table.insert(raw_events, sse.encode(event))
end
- if not contains_done_event and proto and ctx.var.llm_request_done then
+ -- llm_request_done only means "no more content is coming", which is
also
+ -- set when a stream is cut short (upstream read error, stream limit).
+ -- ctx.ai_stream_aborted marks those cases: synthesizing a terminator
there
+ -- would tell the client a truncated response completed successfully.
+ if not contains_done_event and proto and ctx.var.llm_request_done
+ and not ctx.ai_stream_aborted then
table.insert(raw_events, proto.build_done_event())
end
return nil, table.concat(raw_events, "\n")
diff --git a/apisix/plugins/ai-aws-content-moderation.lua
b/apisix/plugins/ai-aws-content-moderation.lua
index cc23e627a..56b985465 100644
--- a/apisix/plugins/ai-aws-content-moderation.lua
+++ b/apisix/plugins/ai-aws-content-moderation.lua
@@ -446,7 +446,12 @@ local function annotate_stream(ctx, body)
table.insert(raw_events, sse.encode(event))
end
- if not contains_done_event and proto.build_done_event and
ctx.var.llm_request_done then
+ -- llm_request_done only means "no more content is coming", which is also
+ -- set when a stream is cut short (upstream read error, stream limit).
+ -- ctx.ai_stream_aborted marks those cases: synthesizing a terminator there
+ -- would tell the client a truncated response completed successfully.
+ if not contains_done_event and proto.build_done_event
+ and ctx.var.llm_request_done and not ctx.ai_stream_aborted then
table.insert(raw_events, proto.build_done_event())
end
return table.concat(raw_events)
diff --git a/apisix/plugins/ai-providers/base.lua
b/apisix/plugins/ai-providers/base.lua
index c8a5d6a71..41e3bffaa 100644
--- a/apisix/plugins/ai-providers/base.lua
+++ b/apisix/plugins/ai-providers/base.lua
@@ -460,6 +460,12 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
-- attempt emitted no output (with headers sent, the retry dies earlier in
-- core.response.set_header), so the flag is stale, not protective
ctx.ai_stream_aborted = nil
+ -- same for the completion flag. An attempt can set it and still produce no
+ -- downstream output -- a converter fed a [DONE]-only stream emits nothing
--
+ -- which returns 502 and lets ai-proxy-multi fall back inside this same
+ -- request context. Left set, it would make the next attempt look finished
+ -- before it has parsed a completion event of its own.
+ ctx.var.llm_request_done = nil
ngx.status = res.status
local body_reader = res.body_reader
local contents = {}
@@ -475,6 +481,10 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
-- all events may be skipped and no output produced, leaving the response
-- uncommitted and causing nginx to fall through to the balancer phase.
local output_sent = false
+ -- Set when THIS attempt parses the protocol's completion event. The read
+ -- error path must not consult ctx.var.llm_request_done for that: it is
+ -- shared across fallback attempts and is also set on abort finalization.
+ local protocol_completed = false
-- Runaway-upstream safeguards. Both are opt-in; unset means no cap.
local max_duration_ms = conf and conf.max_stream_duration_ms
@@ -555,7 +565,12 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
local chunk, err = body_reader()
if err then
- ctx.ai_stream_aborted = "read_error"
+ -- A read error that arrives after the protocol's completion event
+ -- means the stream itself finished and only the transport died
late,
+ -- so the response is complete rather than aborted.
+ if not protocol_completed then
+ ctx.ai_stream_aborted = "read_error"
+ end
ctx.var.apisix_upstream_response_time = math.floor(
(ngx_now() - ctx.llm_request_start_time) * 1000)
core.log.warn("failed to read response chunk: ", err)
@@ -564,6 +579,29 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
ngx.thread.kill(flush_thread)
flush_thread = nil
end
+ -- The connection broke mid-body, so it must never go back into the
+ -- keepalive pool: drop it before ai-proxy's set_keepalive() runs.
+ if res._httpc then
+ res._httpc:close()
+ res._httpc = nil
+ end
+ if output_sent then
+ -- The downstream response is already committed as 200 and
part of
+ -- the stream has reached the client, so the status can no
longer
+ -- be changed and ai-proxy-multi must not bill another
instance for
+ -- a retry. Mirror the max_stream_duration_ms path: one last
+ -- body_filter pass with llm_request_done set, so plugins that
+ -- buffer the whole stream flush what they hold instead of
+ -- stranding it. No terminator is synthesized -- the client
detects
+ -- the truncation from the missing protocol completion event
(e.g.
+ -- OpenAI [DONE], Anthropic message_stop, Responses
+ -- response.completed).
+ if not protocol_completed then
+ ctx.var.llm_request_done = true
+ plugin.lua_response_filter(ctx, res.headers, "", nil, true)
+ end
+ return
+ end
return transport_http.handle_error(err)
end
if not chunk then
@@ -685,6 +723,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
if parsed.type == "done" or parsed.type == "usage_and_done" then
ctx.var.llm_request_done = true
+ protocol_completed = true
end
::CONTINUE::
diff --git a/docs/en/latest/plugins/ai-proxy-multi.md
b/docs/en/latest/plugins/ai-proxy-multi.md
index e04b327b4..4f0dc647a 100644
--- a/docs/en/latest/plugins/ai-proxy-multi.md
+++ b/docs/en/latest/plugins/ai-proxy-multi.md
@@ -177,6 +177,8 @@ When the selected LLM upstream returns a `429` or `5xx`
status, `ai-proxy-multi`
- If the request is retried on another instance (per `fallback_strategy`,
`fallback_http_statuses`, `max_retries`, and `retry_on_failure_within_ms`), the
failed instance's error body is recorded in the error log for diagnostics,
since a later attempt's response is sent to the client instead.
- If the request is not retried (no matching `fallback_strategy` or
`fallback_http_statuses`, retries exhausted, or the failure took longer than
`retry_on_failure_within_ms`), the upstream status code and error body are
returned to the client, preserving the upstream `Content-Type`.
+A streaming response is different once part of it has been delivered. The
downstream response is committed as `200` with the first SSE event, so a later
failure to read from the upstream (a connection reset or a read timeout) can no
longer change the status, and falling back would bill another instance for a
response the client can never receive. In that case `ai-proxy-multi` stops
reading, closes the upstream connection, and ends the downstream stream where
it is, without a protocol-spe [...]
+
## Examples
The examples below demonstrate how you can configure `ai-proxy-multi` for
different scenarios.
diff --git a/docs/en/latest/plugins/ai-proxy.md
b/docs/en/latest/plugins/ai-proxy.md
index c9f132bbb..005f3108e 100644
--- a/docs/en/latest/plugins/ai-proxy.md
+++ b/docs/en/latest/plugins/ai-proxy.md
@@ -166,6 +166,8 @@ The setting covers `ai-proxy`, `ai-proxy-multi`, and
`ai-request-rewrite`, which
When the LLM upstream returns a `429` or `5xx` status, `ai-proxy` reads the
upstream error body and returns it to the client together with the upstream
status code and `Content-Type`, so provider-side error details (such as
rate-limit information or validation errors) are not discarded.
+A streaming response is different once part of it has been delivered. The
downstream response is committed as `200` with the first SSE event, so a later
failure to read from the upstream (a connection reset or a read timeout) can no
longer change the status. In that case `ai-proxy` stops reading, closes the
upstream connection, and ends the downstream stream where it is, without a
protocol-specific terminator such as `[DONE]`, `message_stop`, or
`response.completed`; well-behaved clients [...]
+
## Examples
The examples below demonstrate how you can configure `ai-proxy` for different
scenarios.
diff --git a/t/lib/truncated_sse.lua b/t/lib/truncated_sse.lua
new file mode 100644
index 000000000..3ceec4a67
--- /dev/null
+++ b/t/lib/truncated_sse.lua
@@ -0,0 +1,114 @@
+--
+-- 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.
+--
+
+-- Mock LLM upstream that commits a 200 SSE response and then drops the
+-- connection mid-body. Writing the response on the raw downstream socket is
+-- what makes the truncation possible: through the normal nginx output chain
the
+-- terminating chunk is always appended, which is a clean EOF rather than the
+-- transport error this mock has to produce.
+local ngx = ngx
+
+local _M = {}
+
+local CRLF = string.char(13, 10)
+
+local DONE_STREAM = table.concat({
+ 'data: {"id":"chatcmpl-1","object":"chat.completion.chunk",'
+ ..
'"choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}',
+ "",
+ "data: [DONE]",
+ "",
+ "",
+}, "\n")
+
+
+local function take_over()
+ ngx.req.read_body()
+ local sock, err = ngx.req.socket(true)
+ if not sock then
+ ngx.log(ngx.ERR, "failed to take over the downstream socket: ", err)
+ return nil
+ end
+ sock:send("HTTP/1.1 200 OK" .. CRLF
+ .. "Content-Type: text/event-stream" .. CRLF
+ .. "Transfer-Encoding: chunked" .. CRLF
+ .. "Connection: keep-alive" .. CRLF .. CRLF)
+ return sock
+end
+
+
+-- Sends every event in `events` as its own chunk, then closes without the
+-- terminating zero-length chunk, so the client's body reader fails with
+-- "closed" after the events have already been delivered.
+function _M.serve(events)
+ local sock = take_over()
+ if not sock then
+ return
+ end
+ for _, event in ipairs(events) do
+ sock:send(string.format("%x", #event) .. CRLF .. event .. CRLF)
+ end
+ return ngx.exit(444)
+end
+
+
+local done_then_truncate_hits = 0
+
+-- First call answers with a [DONE]-only stream. Behind a protocol converter
+-- that yields no downstream event for it, the attempt sets llm_request_done
+-- while output_sent stays false, so EOF becomes the 502 that ai-proxy-multi
+-- falls back on. Every later call emits one real event and then truncates,
+-- which is the attempt that must not inherit the first one's completion state.
+function _M.serve_done_then_truncate()
+ done_then_truncate_hits = done_then_truncate_hits + 1
+ if done_then_truncate_hits == 1 then
+ ngx.header["Content-Type"] = "text/event-stream"
+ ngx.print("data: [DONE]\n\n")
+ return ngx.flush(true)
+ end
+ return _M.serve({
+ 'data: {"id":"chatcmpl-1","object":"chat.completion.chunk",'
+ .. '"choices":[{"index":0,"delta":{"content":"hello"},'
+ .. '"finish_reason":null}]}\n\n',
+ -- a usage event, so the content-moderation final_packet path has an
+ -- assembled completion to work with before the transport dies
+ 'data: {"id":"chatcmpl-1","object":"chat.completion.chunk",'
+ .. '"choices":[],"usage":{"prompt_tokens":1,'
+ .. '"completion_tokens":1,"total_tokens":2}}\n\n',
+ })
+end
+
+
+local aborts = 0
+
+-- First call closes right after the headers, before any body byte; every later
+-- call streams a complete response.
+function _M.serve_abort_once()
+ aborts = aborts + 1
+ if aborts > 1 then
+ ngx.header["Content-Type"] = "text/event-stream"
+ ngx.print(DONE_STREAM)
+ return ngx.flush(true)
+ end
+ if not take_over() then
+ return
+ end
+ return ngx.exit(444)
+end
+
+
+return _M
diff --git a/t/plugin/ai-proxy-stream-truncated.t
b/t/plugin/ai-proxy-stream-truncated.t
new file mode 100644
index 000000000..b75934626
--- /dev/null
+++ b/t/plugin/ai-proxy-stream-truncated.t
@@ -0,0 +1,375 @@
+#
+# 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.
+#
+
+# The control API v1 mounts a "/v1/" prefix route that would shadow the
+# "/v1/messages" route TEST 9 needs to activate the Anthropic client protocol.
+BEGIN {
+ $ENV{TEST_ENABLE_CONTROL_API_V1} = "0";
+}
+
+use t::APISIX 'no_plan';
+
+log_level("info");
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+
+ if (!defined $block->request) {
+ $block->set_value("request", "GET /t");
+ }
+
+ # The behaviour under test lives in ai-providers/base.lua and only depends
on
+ # body_reader() reporting a transport error, so it is the same for either
HTTP
+ # client. Pinning one keeps the error string this file asserts on stable.
+ my $user_yaml_config = <<_EOC_;
+plugins:
+ - ai-proxy
+ - ai-proxy-multi
+ - ai-aliyun-content-moderation
+plugin_attr:
+ ai-proxy:
+ http_client: lua-resty-http
+_EOC_
+ if (!defined $block->extra_yaml_config) {
+ $block->set_value("extra_yaml_config", $user_yaml_config);
+ }
+
+ my $http_config = $block->http_config // <<_EOC_;
+ server {
+ listen 6724;
+
+ # Commits a 200 SSE response, flushes one valid event, then drops the
+ # connection without the terminating chunk and without [DONE]. Takes
over
+ # the raw socket because nginx would otherwise append a well-formed
+ # end-of-chunked-body on its own.
+ location /v1/chat/completions-truncate {
+ content_by_lua_block {
+ require("lib.truncated_sse").serve({
+ 'data:
{"id":"chatcmpl-1","object":"chat.completion.chunk",'
+ .. '"choices":[{"index":0,"delta":{"content":"hello"},'
+ .. '"finish_reason":null}]}\\n\\n',
+ })
+ }
+ }
+
+ # Same, but a usage event lands before the truncation, so
+ # ctx.var.llm_response_text is set and the content-moderation
+ # final_packet path runs on the last body_filter pass.
+ location /v1/chat/completions-usage-then-truncate {
+ content_by_lua_block {
+ require("lib.truncated_sse").serve({
+ 'data:
{"id":"chatcmpl-1","object":"chat.completion.chunk",'
+ .. '"choices":[{"index":0,"delta":{"content":"hello"},'
+ .. '"finish_reason":null}]}\\n\\n',
+ 'data:
{"id":"chatcmpl-1","object":"chat.completion.chunk",'
+ .. '"choices":[],"usage":{"prompt_tokens":1,'
+ .. '"completion_tokens":1,"total_tokens":2}}\\n\\n',
+ })
+ }
+ }
+
+ # First hit commits the 200 SSE headers and drops the connection before
+ # any body byte; later hits stream a complete response. Reproduces a
read
+ # error that happens before anything reaches the client.
+ location /v1/chat/completions-abort-once {
+ content_by_lua_block {
+ require("lib.truncated_sse").serve_abort_once()
+ }
+ }
+
+ # First hit answers with a [DONE]-only stream (no downstream output
once
+ # the anthropic converter has skipped it, so the attempt 502s and is
+ # retryable); the retry emits one event and then truncates.
+ location /v1/chat/completions-done-then-truncate {
+ content_by_lua_block {
+ require("lib.truncated_sse").serve_done_then_truncate()
+ }
+ }
+
+ # Aliyun content-moderation endpoint.
+ location / {
+ content_by_lua_block {
+ local content =
require("lib.fixture_loader").load("aliyun/moderation-safe.json")
+ ngx.status = 200
+ ngx.header["Content-Type"] = "application/json"
+ ngx.print(content)
+ }
+ }
+ }
+_EOC_
+ $block->set_value("http_config", $http_config);
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: ai-proxy route on the truncating upstream
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/truncated",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": {"header": {"Authorization": "Bearer test"}},
+ "options": {"model": "gpt-4", "stream": true},
+ "override": {
+ "endpoint":
"http://127.0.0.1:6724/v1/chat/completions-truncate";
+ },
+ "ssl_verify": false
+ }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 2: a read error after partial output leaves the committed 200 alone
+--- request
+POST /truncated
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_body eval
+qq{data:
{"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}\n\n}
+--- error_log
+failed to read response chunk: closed
+--- no_error_log
+exits with http status code
+attempt to set ngx.status
+failed to keepalive connection
+--- timeout: 10
+
+
+
+=== TEST 3: ai-proxy-multi route on the truncating upstream, with http_5xx
fallback
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/truncated",
+ "plugins": {
+ "ai-proxy-multi": {
+ "fallback_strategy": ["http_5xx"],
+ "ssl_verify": false,
+ "instances": [
+ {"name":"first","provider":"openai","weight":1,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-truncate"}},
+ {"name":"second","provider":"openai","weight":1,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-truncate"}}
+ ]
+ }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 4: no fallback after partial output -- exactly one upstream request
is billed
+--- request
+POST /truncated
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_body eval
+qq{data:
{"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}\n\n}
+--- grep_error_log eval
+qr/sending request to LLM server/
+--- grep_error_log_out
+sending request to LLM server
+--- no_error_log
+falling back to
+attempt to set ngx.status
+--- timeout: 10
+
+
+
+=== TEST 5: ai-proxy-multi route on the abort-before-body upstream
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/truncated",
+ "plugins": {
+ "ai-proxy-multi": {
+ "fallback_strategy": ["http_5xx"],
+ "ssl_verify": false,
+ "instances": [
+ {"name":"first","provider":"openai","weight":1,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-abort-once"}},
+ {"name":"spare","provider":"openai","weight":0,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-abort-once"}}
+ ]
+ }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 6: a read error before the first downstream byte still falls back
+--- request
+POST /truncated
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_body_like eval
+qr/data: \[DONE\]/
+--- error_log
+failed to read response chunk: closed
+falling back to
+--- timeout: 10
+
+
+
+=== TEST 7: ai-proxy + ai-aliyun-content-moderation on the usage-then-truncate
upstream
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/truncated",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": {"header": {"Authorization": "Bearer test"}},
+ "options": {"model": "gpt-4", "stream": true},
+ "override": {
+ "endpoint":
"http://127.0.0.1:6724/v1/chat/completions-usage-then-truncate";
+ },
+ "ssl_verify": false
+ },
+ "ai-aliyun-content-moderation": {
+ "endpoint": "http://127.0.0.1:6724";,
+ "region_id": "cn-shanghai",
+ "access_key_id": "fake-key-id",
+ "access_key_secret": "fake-key-secret",
+ "risk_level_bar": "high",
+ "check_request": false,
+ "check_response": true,
+ "stream_check_mode": "final_packet"
+ }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 8: a truncated stream is never terminated with a synthesized [DONE]
+--- request
+POST /truncated
+{"messages":[{"role":"user","content":"hi"}],"model":"gpt-4","stream":true}
+--- response_body_like eval
+# The moderation plugin re-encodes every event, so the key order is not stable
+# enough to assert the body verbatim: require the delivered content and the
+# risk_level annotation that proves the final_packet branch ran, and reject a
+# [DONE] anywhere in the response.
+qr/^(?!.*\[DONE\])(?=.*"content":"hello")(?=.*"risk_level":"none")/s
+--- error_log
+failed to read response chunk: closed
+--- timeout: 10
+
+
+
+=== TEST 9: ai-proxy-multi + a converter, where attempt 1 is a [DONE]-only
stream
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1', ngx.HTTP_PUT, [[{
+ "uri": "/v1/messages",
+ "plugins": {
+ "ai-proxy-multi": {
+ "fallback_strategy": ["http_5xx"],
+ "ssl_verify": false,
+ "instances": [
+ {"name":"primary","provider":"openai","weight":1,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-done-then-truncate"}},
+ {"name":"spare","provider":"openai","weight":0,
+ "auth":{"header":{"Authorization":"Bearer test"}},
+ "options":{"model":"gpt-4","stream":true},
+
"override":{"endpoint":"http://127.0.0.1:6724/v1/chat/completions-done-then-truncate"}}
+ ]
+ },
+ "ai-aliyun-content-moderation": {
+ "endpoint": "http://127.0.0.1:6724";,
+ "region_id": "cn-shanghai",
+ "access_key_id": "fake-key-id",
+ "access_key_secret": "fake-key-secret",
+ "risk_level_bar": "high",
+ "check_request": false,
+ "check_response": true,
+ "stream_check_mode": "final_packet"
+ }
+ }
+ }]])
+ if code >= 300 then ngx.status = code end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 10: the retried attempt does not inherit the first attempt's
completion state
+--- request
+POST /v1/messages
+{"model":"claude-3-5-sonnet","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}
+--- more_headers
+Content-Type: application/json
+--- response_body_like eval
+# Inheriting the first attempt's completion state makes the retry look finished
+# from its very first chunk: the abort marker is never set, so the moderation
+# plugin appends a message_stop and reports the truncated stream as complete.
+qr/^(?!.*message_stop)(?=.*"text":"hello")/s
+--- error_log
+failed to read response chunk: closed
+falling back to
+--- timeout: 10
