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 1f6120d19 fix(ai-aliyun-content-moderation): report final results
without usage (#13922)
1f6120d19 is described below
commit 1f6120d19bf4935df1cf9a5647fa7f8cb40f28e5
Author: Nic <[email protected]>
AuthorDate: Wed Sep 9 13:27:32 2026 +0800
fix(ai-aliyun-content-moderation): report final results without usage
(#13922)
---
apisix/plugins/ai-aliyun-content-moderation.lua | 58 ++--
apisix/plugins/ai-lakera-guard.lua | 4 +-
apisix/plugins/ai-protocols/anthropic-messages.lua | 20 ++
apisix/plugins/ai-protocols/openai-chat.lua | 23 ++
apisix/plugins/ai-protocols/openai-responses.lua | 6 +
apisix/plugins/ai-providers/base.lua | 20 +-
.../latest/plugins/ai-aliyun-content-moderation.md | 12 +
.../latest/plugins/ai-aliyun-content-moderation.md | 12 +
t/fixtures/aliyun/result-anthropic-deny.sse | 18 ++
t/fixtures/aliyun/result-anthropic-error.sse | 18 ++
t/fixtures/aliyun/result-anthropic-null-delta.sse | 18 ++
t/fixtures/aliyun/result-anthropic-safe.sse | 18 ++
.../aliyun/result-anthropic-stop-sequence.sse | 18 ++
t/fixtures/aliyun/result-anthropic-usage.sse | 18 ++
t/fixtures/aliyun/result-chat-clean-eof.sse | 8 +
t/fixtures/aliyun/result-chat-deny.sse | 8 +
t/fixtures/aliyun/result-chat-error.sse | 8 +
t/fixtures/aliyun/result-chat-failure.sse | 8 +
t/fixtures/aliyun/result-chat-length.sse | 10 +
t/fixtures/aliyun/result-chat-multiple-choices.sse | 10 +
t/fixtures/aliyun/result-chat-multiple-reasons.sse | 10 +
t/fixtures/aliyun/result-chat-safe.sse | 8 +
t/fixtures/aliyun/result-chat-tool_calls.sse | 10 +
t/fixtures/aliyun/result-chat-usage.sse | 10 +
t/fixtures/aliyun/result-responses-deny.sse | 24 ++
t/fixtures/aliyun/result-responses-eof.sse | 21 ++
t/fixtures/aliyun/result-responses-error.sse | 24 ++
t/fixtures/aliyun/result-responses-failed.sse | 24 ++
t/fixtures/aliyun/result-responses-incomplete.sse | 24 ++
t/fixtures/aliyun/result-responses-safe.sse | 24 ++
t/fixtures/aliyun/result-responses-usage.sse | 24 ++
.../openai/chat-streaming-incomplete-no-usage.sse | 9 +
.../openai/chat-streaming-injection-no-usage.sse | 10 +
t/fixtures/openai/moderation-eof-no-usage.sse | 6 +
t/fixtures/openai/moderation-incomplete-event.sse | 7 +
t/fixtures/openai/moderation-no-usage.sse | 8 +
t/fixtures/openai/moderation-with-usage.sse | 10 +
t/lib/ai_moderation_result.lua | 255 ++++++++++++++++
t/plugin/ai-aliyun-content-moderation-result.t | 322 +++++++++++++++++++++
.../ai-aliyun-content-moderation-stream-usage.t | 189 ++++++++++++
t/plugin/ai-lakera-guard.t | 108 ++++++-
t/plugin/ai-proxy-stream-truncated.t | 8 +-
42 files changed, 1414 insertions(+), 36 deletions(-)
diff --git a/apisix/plugins/ai-aliyun-content-moderation.lua
b/apisix/plugins/ai-aliyun-content-moderation.lua
index fe7748cb2..a0f866bda 100644
--- a/apisix/plugins/ai-aliyun-content-moderation.lua
+++ b/apisix/plugins/ai-aliyun-content-moderation.lua
@@ -172,7 +172,6 @@ local function calculate_sign(params, secret)
table.sort(params_arr)
local canonical_str = table.concat(params_arr, "&")
local str_to_sign = "POST&%2F&" .. ngx.escape_uri(canonical_str)
- core.log.debug("string to calculate signature: ", str_to_sign)
return ngx.encode_base64(ngx.hmac_sha1(secret, str_to_sign))
end
@@ -284,12 +283,13 @@ local function deny_message(ctx, message)
local usage = ctx.llm_raw_usage
or (proto.empty_usage and proto.empty_usage())
or { prompt_tokens = 0, completion_tokens = 0, total_tokens = 0 }
+ message = message or "Your request violate our content policy."
return proto.build_deny_response({
- text = message or "Your request violate our content policy.",
+ text = message,
model = model,
usage = usage,
stream = stream,
- })
+ }), message
end
@@ -500,45 +500,67 @@ function _M.lua_body_filter(conf, ctx, headers, body)
local proto = protocols.get(ctx.ai_client_protocol)
if conf.stream_check_mode == "final_packet" then
- if not ctx.var.llm_response_text then
+ local content = ctx.var.llm_response_text
+ if not content or content == "" or ctx.ai_stream_aborted or
ctx.ai_stream_failed
+ or ctx.aliyun_cm_done_sent then
return
end
if not ctx.ai_aliyun_response_moderated then
- response_content_moderation(ctx, conf, ctx.var.llm_response_text)
+ ctx.var.llm_content_risk_level = nil
+ local _, _, message = response_content_moderation(ctx, conf,
content)
+ ctx.aliyun_cm_deny_message = message or ""
release_cm_httpc(ctx, conf)
ctx.ai_aliyun_response_moderated = true
end
- local events = sse.decode(body)
+ if not ctx.var.llm_content_risk_level then
+ return
+ end
+ local events = sse.decode(body or "")
for _, event in ipairs(events) do
- if proto and proto.is_data_event(event) then
+ if proto.is_data_event(event) then
local data, err = core.json.decode(event.data)
if not data then
core.log.warn("failed to decode SSE data: ", err)
goto CONTINUE
end
data.risk_level = ctx.var.llm_content_risk_level
+ data.deny_message = ctx.aliyun_cm_deny_message
event.data = core.json.encode(data)
+ if not ctx.ai_stream_has_usage and event.type ==
"message_delta"
+ and type(data.delta) == "table" then
+ -- Anthropic SDKs overwrite stop fields on every
message_delta.
+ ctx.aliyun_cm_message_delta = data.delta
+ end
end
::CONTINUE::
end
local raw_events = {}
- local contains_done_event = false
+ local done_index
for _, event in ipairs(events) do
- if proto and proto.is_done_event(event) then
- contains_done_event = true
+ if proto.is_done_event(event) then
+ done_index = #raw_events + 1
end
table.insert(raw_events, sse.encode(event))
end
- -- 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())
+ -- Converters dispatch several events after the source has completed.
+ -- Only a client terminator or the empty EOF flush can finish its
output.
+ local eof = ctx.var.llm_request_done and (not body or body == "")
+ if not ctx.ai_stream_has_usage and proto.build_moderation_event
+ and (done_index or eof) then
+ table.insert(raw_events, done_index or #raw_events + 1,
+ sse.encode(proto.build_moderation_event({
+ deny_message = ctx.aliyun_cm_deny_message,
+ risk_level = ctx.var.llm_content_risk_level,
+ model = ctx.var.request_llm_model,
+ delta = ctx.aliyun_cm_message_delta or {},
+ })))
+ end
+ if not done_index and eof and proto.build_moderation_event then
+ table.insert(raw_events, proto.build_done_event() .. "\n\n")
end
- return nil, table.concat(raw_events, "\n")
+ ctx.aliyun_cm_done_sent = done_index ~= nil or eof
+ return nil, table.concat(raw_events)
end
if conf.stream_check_mode == "realtime" then
diff --git a/apisix/plugins/ai-lakera-guard.lua
b/apisix/plugins/ai-lakera-guard.lua
index 0f159625f..444e83fcb 100644
--- a/apisix/plugins/ai-lakera-guard.lua
+++ b/apisix/plugins/ai-lakera-guard.lua
@@ -268,13 +268,13 @@ function _M.lua_body_filter(conf, ctx, headers, body)
if not text then
if conf.fail_open then
core.log.warn("ai-lakera-guard: streamed response ended
without ",
- "an assembled completion (no upstream usage
event?); ",
+ "an assembled completion; ",
"fail_open=true, releasing unscanned")
ctx.lakera_response_decided = "clean"
return nil, concat(buffer)
end
core.log.error("ai-lakera-guard: streamed response ended without ",
- "an assembled completion (no upstream usage
event?); ",
+ "an assembled completion; ",
"fail_open=false, blocking response")
ctx.lakera_response_decided = "blocked"
return ngx.OK, deny_message(ctx, conf,
conf.response_failure_message)
diff --git a/apisix/plugins/ai-protocols/anthropic-messages.lua
b/apisix/plugins/ai-protocols/anthropic-messages.lua
index 7cef9a473..75971e172 100644
--- a/apisix/plugins/ai-protocols/anthropic-messages.lua
+++ b/apisix/plugins/ai-protocols/anthropic-messages.lua
@@ -424,12 +424,32 @@ function _M.empty_usage()
end
+--- Message metadata can be appended after content blocks have closed.
+function _M.build_moderation_event(opts)
+ return {
+ type = "message_delta",
+ data = core.json.encode({
+ type = "message_delta",
+ delta = opts.delta,
+ usage = _M.empty_usage(),
+ risk_level = opts.risk_level,
+ deny_message = opts.deny_message or "",
+ }),
+ }
+end
+
+
--- Check if an SSE event is a data event (contains parseable content).
function _M.is_data_event(event)
return event.type == "content_block_delta" or event.type == "message_delta"
end
+function _M.is_error_event(event)
+ return event.type == "error"
+end
+
+
--- Check if an SSE event is the terminal/done event.
function _M.is_done_event(event)
return event.type == "message_stop"
diff --git a/apisix/plugins/ai-protocols/openai-chat.lua
b/apisix/plugins/ai-protocols/openai-chat.lua
index 4ac1cf0c7..a92b5164f 100644
--- a/apisix/plugins/ai-protocols/openai-chat.lua
+++ b/apisix/plugins/ai-protocols/openai-chat.lua
@@ -22,8 +22,10 @@
local core = require("apisix.core")
local uuid = require("resty.jit-uuid")
local table = table
+local setmetatable = setmetatable
local type = type
local ipairs = ipairs
+local ngx_time = ngx.time
local _M = {}
@@ -410,12 +412,33 @@ function _M.empty_usage()
end
+--- Build a final moderation chunk without ending or replacing the original
stream.
+function _M.build_moderation_event(opts)
+ local data = {
+ id = uuid.generate_v4(),
+ object = "chat.completion.chunk",
+ created = ngx_time(),
+ model = opts.model,
+ choices = setmetatable({}, core.json.array_mt),
+ usage = _M.empty_usage(),
+ risk_level = opts.risk_level,
+ deny_message = opts.deny_message or "",
+ }
+ return { type = "message", data = core.json.encode(data) }
+end
+
+
--- Check if an SSE event is a data event (contains parseable content).
function _M.is_data_event(event)
return event.type == "message" and event.data ~= "[DONE]"
end
+function _M.is_error_event(event, data)
+ return type(data) == "table" and type(data.error) == "table"
+end
+
+
--- Check if an SSE event is the terminal/done event.
function _M.is_done_event(event)
return event.data == "[DONE]"
diff --git a/apisix/plugins/ai-protocols/openai-responses.lua
b/apisix/plugins/ai-protocols/openai-responses.lua
index cd38b2296..485f8be4e 100644
--- a/apisix/plugins/ai-protocols/openai-responses.lua
+++ b/apisix/plugins/ai-protocols/openai-responses.lua
@@ -455,6 +455,12 @@ function _M.is_data_event(event)
end
+function _M.is_error_event(event)
+ return event.type == "error" or event.type == "response.failed"
+ or event.type == "response.incomplete"
+end
+
+
--- Check if an SSE event is the terminal/done event.
function _M.is_done_event(event)
return event.type == "response.completed"
diff --git a/apisix/plugins/ai-providers/base.lua
b/apisix/plugins/ai-providers/base.lua
index 41e3bffaa..8612f5dfd 100644
--- a/apisix/plugins/ai-providers/base.lua
+++ b/apisix/plugins/ai-providers/base.lua
@@ -460,6 +460,8 @@ 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
+ ctx.ai_stream_has_usage = nil
+ ctx.ai_stream_failed = 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
@@ -607,6 +609,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
if not chunk then
local sse_rem = table.concat(sse_parts)
if #sse_rem > 0 then
+ ctx.ai_stream_aborted = "incomplete_frame"
core.log.warn("dropping incomplete stream frame at EOF, size:
",
#sse_rem)
end
@@ -633,6 +636,9 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
flush_thread = nil
end
if output_sent and not ctx.var.llm_request_done then
+ if #sse_rem == 0 then
+ ctx.var.llm_response_text = table.concat(contents, "")
+ end
ctx.var.llm_request_done = true
plugin.lua_response_filter(ctx, res.headers, "", nil, true)
end
@@ -665,6 +671,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
-- One-pass split + decode: finds all complete SSE events and the
-- trailing remainder in a single forward scan (no PCRE, no double
scan).
local events, remainder = framing.decode_buf(candidate)
+ local complete = candidate:sub(1, #candidate - #remainder)
local max_remainder = framing.max_remainder or 1024 * 1024
if #remainder > max_remainder then
core.log.warn("stream remainder exceeded ", max_remainder, "
bytes, resetting")
@@ -682,6 +689,10 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
for _, event in ipairs(events) do
-- Target protocol parses the provider's SSE format
local parsed = target_proto.parse_sse_event(event, ctx, sse_state)
+ if target_proto.is_error_event
+ and target_proto.is_error_event(event, parsed and parsed.data)
then
+ ctx.ai_stream_failed = true
+ end
if parsed and parsed.has_tool_call then
ctx.var.llm_has_tool_calls = "true"
end
@@ -707,6 +718,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
end
if parsed.usage then
+ ctx.ai_stream_has_usage = true
core.log.info("got token usage from ai service: ",
core.json.delay_encode(parsed.raw_usage or
parsed.usage))
merge_usage(ctx, parsed)
@@ -722,6 +734,7 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
end
if parsed.type == "done" or parsed.type == "usage_and_done" then
+ ctx.var.llm_response_text = table.concat(contents, "")
ctx.var.llm_request_done = true
protocol_completed = true
end
@@ -754,9 +767,12 @@ function _M.parse_streaming_response(self, ctx, res,
target_proto, converter, co
return
end
end
- else
+ elseif ctx.ai_stream_framing ~= "sse" or complete ~= "" then
+ -- Native SSE filters need complete frames just like converters do.
+ -- Keep the original bytes, including comments and blank lines.
+ local downstream = ctx.ai_stream_framing == "sse" and complete or
chunk
local ok, flush_err = plugin.lua_response_filter(
- ctx, res.headers, chunk, no_flush, true)
+ ctx, res.headers, downstream, no_flush, true)
if not ok then
abort_on_disconnect(flush_err)
return
diff --git a/docs/en/latest/plugins/ai-aliyun-content-moderation.md
b/docs/en/latest/plugins/ai-aliyun-content-moderation.md
index 7019149ed..b2e548311 100644
--- a/docs/en/latest/plugins/ai-aliyun-content-moderation.md
+++ b/docs/en/latest/plugins/ai-aliyun-content-moderation.md
@@ -74,6 +74,18 @@ The `ai-aliyun-content-moderation` Plugin should be used
with either [`ai-proxy`
| ssl_verify | boolean | False | `true` | | If `true`, enable SSL certificate
verification. |
| fail_mode | string | False | `"skip"` | `skip`, `warn`, `error` | Behavior
when the request is not a recognized AI request that this plugin can inspect
(for example, plain HTTP traffic on a Consumer-bound plugin, or a request that
did not pass through `ai-proxy`). `skip`: let the request pass through
unchecked; `warn`: pass through and log a warning; `error`: reject the request.
|
+In `final_packet` mode, `risk_level` and `deny_message` are added together to
existing data events after the assembled response text is available. Rejected
responses use the actual denial message; allowed responses use an empty string.
Existing response content and token usage are preserved. The result is
informational and cannot retract content already sent to the client.
+
+When the upstream omits `usage`, moderation runs at the end of the stream:
+
+- **OpenAI Chat Completions:** an additional `chat.completion.chunk` with
`risk_level`, `deny_message`, and zero-valued `usage` is inserted before
`[DONE]`. This is an empty usage chunk with `choices: []`: the denial message
appears only in the top-level `deny_message` field, without appending text or
changing the upstream finish reason.
+- **Anthropic Messages:** an additional `message_delta` carries `risk_level`,
`deny_message`, zero-valued `usage`, and the original stop information before
`message_stop`. Content blocks and `message_start` are not replayed.
+- **OpenAI Responses:** the existing `response.completed` event carries
`risk_level` and `deny_message`; its output and usage are preserved. EOF
without `response.completed` does not produce a synthetic completed response.
+
+Clients must consume through the stream terminator to receive an injected
result. Usage-bearing streams receive in-place fields without an additional
result event, so their reported usage is unchanged. New result events report
zero usage; gateway accounting continues to use upstream usage. If moderation
fails without returning a risk level, no result is fabricated. Error events are
preserved, and aborted streams do not receive a synthesized terminator.
+
+Read moderation extensions from the SSE events. For example, the OpenAI Python
SDK exposes Responses extensions through `responses.create(stream=True)`, while
its higher-level `responses.stream()` wrapper can discard unknown top-level
fields.
+
## Examples
The following examples use OpenAI as the Upstream service provider. Before
proceeding, create an [OpenAI account](https://openai.com) and obtain an [API
key](https://openai.com/blog/openai-api). If you are working with other LLM
providers, please refer to the provider's documentation to obtain an API key.
diff --git a/docs/zh/latest/plugins/ai-aliyun-content-moderation.md
b/docs/zh/latest/plugins/ai-aliyun-content-moderation.md
index 5b68b94be..64c7297ec 100644
--- a/docs/zh/latest/plugins/ai-aliyun-content-moderation.md
+++ b/docs/zh/latest/plugins/ai-aliyun-content-moderation.md
@@ -74,6 +74,18 @@ import TabItem from '@theme/TabItem';
| ssl_verify | boolean | 否 | `true` | | 如果为 `true`，启用 SSL 证书验证。 |
| fail_mode | string | 否 | `"skip"` | `skip`、`warn`、`error` | 当请求不是该插件可识别的 AI
请求时的处理行为（例如 Consumer 级别绑定时的普通 HTTP 流量，或未经过 `ai-proxy`
的请求）。`skip`：放行请求且不做检查；`warn`：放行并记录 warning 日志；`error`：拒绝请求。 |
+`final_packet` 模式在已汇总的响应正文可用后，在现有数据事件中同时追加 `risk_level` 和
`deny_message`。拒绝时使用最终实际拒绝文案，通过时为空字符串，保留事件原有正文和 token 用量。结果仅用于告知客户端，无法撤回已经发送的内容。
+
+上游未提供 `usage` 时，在流结束时执行审核：
+
+- **OpenAI Chat Completions：**在 `[DONE]` 前插入一个携带 `risk_level`、`deny_message`
和全零 `usage` 的 `chat.completion.chunk`。这是 `choices: []` 的空用量包，拒绝文案仅通过顶层
`deny_message` 字段返回，不追加正文，也不改变上游结束原因。
+- **Anthropic Messages：**在 `message_stop` 前补一个携带
`risk_level`、`deny_message`、全零 `usage` 和原始结束信息的 `message_delta`，不重放内容块或
`message_start`。
+- **OpenAI Responses：**在现有 `response.completed` 事件中附加 `risk_level` 和
`deny_message`，保留原始输出和用量。EOF 时若没有 `response.completed`，不合成完成响应。
+
+客户端需读取至流终止符才能获取注入结果。有用量的流只在原事件追加字段，不额外补结果包，因此客户端收到的用量不变。新结果事件使用零用量；网关计费仍使用上游用量。审核失败且没有返回风险等级时不伪造结果；上游错误事件保留，中断的流不合成终止符。
+
+请从 SSE 事件读取审核扩展字段。例如，OpenAI Python SDK 的 `responses.create(stream=True)` 会保留
Responses 扩展字段，但更高层的 `responses.stream()` 封装可能丢弃未知的顶层字段。
+
## 示例
以下示例使用 OpenAI 作为上游服务提供商。在开始之前，请创建一个 [OpenAI 账号](https://openai.com) 并获取 [API
密钥](https://openai.com/blog/openai-api)。如果你使用其他 LLM 提供商，请参考相应提供商的文档获取 API 密钥。
diff --git a/t/fixtures/aliyun/result-anthropic-deny.sse
b/t/fixtures/aliyun/result-anthropic-deny.sse
new file mode 100644
index 000000000..5093a7f55
--- /dev/null
+++ b/t/fixtures/aliyun/result-anthropic-deny.sse
@@ -0,0 +1,18 @@
+event: message_start
+data:
{"type":"message_start","message":{"id":"msg_moderation","type":"message","role":"assistant","model":"test-model","content":[],"stop_reason":null,"stop_sequence":null}}
+
+event: content_block_start
+data:
{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data:
{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"kill
you"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data:
{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null}}
+
+event: message_stop
+data: {"type":"message_stop"}
+
diff --git a/t/fixtures/aliyun/result-anthropic-error.sse
b/t/fixtures/aliyun/result-anthropic-error.sse
new file mode 100644
index 000000000..b72daa182
--- /dev/null
+++ b/t/fixtures/aliyun/result-anthropic-error.sse
@@ -0,0 +1,18 @@
+event: message_start
+data:
{"type":"message_start","message":{"id":"msg_moderation","type":"message","role":"assistant","model":"test-model","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}
+
+event: content_block_start
+data:
{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data:
{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"kill
you"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data:
{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":8}}
+
+event: error
+data: {"type":"error","error":{"type":"overloaded_error","message":"stream
failed"}}
+
diff --git a/t/fixtures/aliyun/result-anthropic-null-delta.sse
b/t/fixtures/aliyun/result-anthropic-null-delta.sse
new file mode 100644
index 000000000..5adbf3cd3
--- /dev/null
+++ b/t/fixtures/aliyun/result-anthropic-null-delta.sse
@@ -0,0 +1,18 @@
+event: message_start
+data:
{"type":"message_start","message":{"id":"msg_moderation","type":"message","role":"assistant","model":"test-model","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}
+
+event: content_block_start
+data:
{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data:
{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"kill
you"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data: {"type":"message_delta","delta":null,"usage":{"output_tokens":8}}
+
+event: message_stop
+data: {"type":"message_stop"}
+
diff --git a/t/fixtures/aliyun/result-anthropic-safe.sse
b/t/fixtures/aliyun/result-anthropic-safe.sse
new file mode 100644
index 000000000..f9782f99f
--- /dev/null
+++ b/t/fixtures/aliyun/result-anthropic-safe.sse
@@ -0,0 +1,18 @@
+event: message_start
+data:
{"type":"message_start","message":{"id":"msg_moderation","type":"message","role":"assistant","model":"test-model","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}
+
+event: content_block_start
+data:
{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data:
{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"safe
output"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data:
{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":8}}
+
+event: message_stop
+data: {"type":"message_stop"}
+
diff --git a/t/fixtures/aliyun/result-anthropic-stop-sequence.sse
b/t/fixtures/aliyun/result-anthropic-stop-sequence.sse
new file mode 100644
index 000000000..fd4250735
--- /dev/null
+++ b/t/fixtures/aliyun/result-anthropic-stop-sequence.sse
@@ -0,0 +1,18 @@
+event: message_start
+data:
{"type":"message_start","message":{"id":"msg_moderation","type":"message","role":"assistant","model":"test-model","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}
+
+event: content_block_start
+data:
{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data:
{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"kill
you"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data:
{"type":"message_delta","delta":{"stop_reason":"stop_sequence","stop_sequence":"END"},"usage":{"output_tokens":8}}
+
+event: message_stop
+data: {"type":"message_stop"}
+
diff --git a/t/fixtures/aliyun/result-anthropic-usage.sse
b/t/fixtures/aliyun/result-anthropic-usage.sse
new file mode 100644
index 000000000..22b5a5d2b
--- /dev/null
+++ b/t/fixtures/aliyun/result-anthropic-usage.sse
@@ -0,0 +1,18 @@
+event: message_start
+data:
{"type":"message_start","message":{"id":"msg_moderation","type":"message","role":"assistant","model":"test-model","content":[],"stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":0}}}
+
+event: content_block_start
+data:
{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data:
{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"kill
you"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data:
{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":8}}
+
+event: message_stop
+data: {"type":"message_stop"}
+
diff --git a/t/fixtures/aliyun/result-chat-clean-eof.sse
b/t/fixtures/aliyun/result-chat-clean-eof.sse
new file mode 100644
index 000000000..9296e8ffe
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-clean-eof.sse
@@ -0,0 +1,8 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"you"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
diff --git a/t/fixtures/aliyun/result-chat-deny.sse
b/t/fixtures/aliyun/result-chat-deny.sse
new file mode 100644
index 000000000..2be78f0ed
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-deny.sse
@@ -0,0 +1,8 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-error.sse
b/t/fixtures/aliyun/result-chat-error.sse
new file mode 100644
index 000000000..de60bc004
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-error.sse
@@ -0,0 +1,8 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data: {"error":{"type":"server_error","message":"stream failed"}}
+
diff --git a/t/fixtures/aliyun/result-chat-failure.sse
b/t/fixtures/aliyun/result-chat-failure.sse
new file mode 100644
index 000000000..bed0323d5
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-failure.sse
@@ -0,0 +1,8 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"scan-"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"error"},"finish_reason":"stop"}]}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-length.sse
b/t/fixtures/aliyun/result-chat-length.sse
new file mode 100644
index 000000000..6da1ae625
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-length.sse
@@ -0,0 +1,10 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"length"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-multiple-choices.sse
b/t/fixtures/aliyun/result-chat-multiple-choices.sse
new file mode 100644
index 000000000..a2eddfd37
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-multiple-choices.sse
@@ -0,0 +1,10 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":"stop"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":1,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-multiple-reasons.sse
b/t/fixtures/aliyun/result-chat-multiple-reasons.sse
new file mode 100644
index 000000000..d854a37d2
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-multiple-reasons.sse
@@ -0,0 +1,10 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":"length"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":1,"delta":{"content":"you","tool_calls":[{"index":0,"id":"call_test","type":"function","function":{"name":"example","arguments":"{}"}}]},"finish_reason":"tool_calls"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-safe.sse
b/t/fixtures/aliyun/result-chat-safe.sse
new file mode 100644
index 000000000..7686320b7
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-safe.sse
@@ -0,0 +1,8 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"safe
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"output"},"finish_reason":"stop"}]}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-tool_calls.sse
b/t/fixtures/aliyun/result-chat-tool_calls.sse
new file mode 100644
index 000000000..b522b0b74
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-tool_calls.sse
@@ -0,0 +1,10 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"you","tool_calls":[{"index":0,"id":"call_test","type":"function","function":{"name":"example","arguments":"{}"}}]},"finish_reason":"tool_calls"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-chat-usage.sse
b/t/fixtures/aliyun/result-chat-usage.sse
new file mode 100644
index 000000000..7dbad95f7
--- /dev/null
+++ b/t/fixtures/aliyun/result-chat-usage.sse
@@ -0,0 +1,10 @@
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data:
{"id":"chatcmpl-moderation","object":"chat.completion.chunk","created":1700000000,"model":"test-model","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
+data: [DONE]
+
diff --git a/t/fixtures/aliyun/result-responses-deny.sse
b/t/fixtures/aliyun/result-responses-deny.sse
new file mode 100644
index 000000000..0a680b138
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-deny.sse
@@ -0,0 +1,24 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"kill
you","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"kill
you","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
+event: response.completed
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"completed","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]}],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store
[...]
+
diff --git a/t/fixtures/aliyun/result-responses-eof.sse
b/t/fixtures/aliyun/result-responses-eof.sse
new file mode 100644
index 000000000..ecf572c81
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-eof.sse
@@ -0,0 +1,21 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"kill
you","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"kill
you","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
diff --git a/t/fixtures/aliyun/result-responses-error.sse
b/t/fixtures/aliyun/result-responses-error.sse
new file mode 100644
index 000000000..149d37e19
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-error.sse
@@ -0,0 +1,24 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"kill
you","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"kill
you","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
+event: error
+data:
{"type":"error","sequence_number":7,"code":"server_error","message":"stream
failed"}
+
diff --git a/t/fixtures/aliyun/result-responses-failed.sse
b/t/fixtures/aliyun/result-responses-failed.sse
new file mode 100644
index 000000000..59b94db8b
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-failed.sse
@@ -0,0 +1,24 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"kill
you","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"kill
you","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
+event: response.failed
+data:
{"type":"response.failed","sequence_number":7,"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"failed","error":{"code":"server_error","message":"stream
failed"},"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]}],"parallel_tool_ca [...]
+
diff --git a/t/fixtures/aliyun/result-responses-incomplete.sse
b/t/fixtures/aliyun/result-responses-incomplete.sse
new file mode 100644
index 000000000..1fab1e145
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-incomplete.sse
@@ -0,0 +1,24 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"kill
you","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"kill
you","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
+event: response.incomplete
+data:
{"type":"response.incomplete","sequence_number":7,"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"incomplete","error":null,"incomplete_details":{"reason":"max_output_tokens"},"instructions":null,"max_output_tokens":null,"model":"test-model","output":[{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]}],"parallel_tool_calls":true," [...]
+
diff --git a/t/fixtures/aliyun/result-responses-safe.sse
b/t/fixtures/aliyun/result-responses-safe.sse
new file mode 100644
index 000000000..e2259392c
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-safe.sse
@@ -0,0 +1,24 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"safe
output","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"safe
output","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"safe
output","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"safe
output","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
+event: response.completed
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"completed","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"safe
output","annotations":[],"logprobs":[]}]}],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"st
[...]
+
diff --git a/t/fixtures/aliyun/result-responses-usage.sse
b/t/fixtures/aliyun/result-responses-usage.sse
new file mode 100644
index 000000000..d901da1e3
--- /dev/null
+++ b/t/fixtures/aliyun/result-responses-usage.sse
@@ -0,0 +1,24 @@
+event: response.created
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"in_progress","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":false,"temperature":1,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1,"truncation":"disabled","usage":null,"user":null,"metadata":{}
[...]
+
+event: response.output_item.added
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"in_progress","content":[]},"type":"response.output_item.added","sequence_number":1}
+
+event: response.content_part.added
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"","annotations":[],"logprobs":[]},"type":"response.content_part.added","sequence_number":2}
+
+event: response.output_text.delta
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"delta":"kill
you","logprobs":[],"type":"response.output_text.delta","sequence_number":3}
+
+event: response.output_text.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"text":"kill
you","logprobs":[],"type":"response.output_text.done","sequence_number":4}
+
+event: response.content_part.done
+data:
{"item_id":"msg_moderation","output_index":0,"content_index":0,"part":{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]},"type":"response.content_part.done","sequence_number":5}
+
+event: response.output_item.done
+data:
{"output_index":0,"item":{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]},"type":"response.output_item.done","sequence_number":6}
+
+event: response.completed
+data:
{"response":{"id":"resp_moderation","object":"response","created_at":1700000000,"status":"completed","error":null,"incomplete_details":null,"instructions":null,"max_output_tokens":null,"model":"test-model","output":[{"id":"msg_moderation","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"kill
you","annotations":[],"logprobs":[]}]}],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store
[...]
+
diff --git a/t/fixtures/openai/chat-streaming-incomplete-no-usage.sse
b/t/fixtures/openai/chat-streaming-incomplete-no-usage.sse
new file mode 100644
index 000000000..bdb8a3df1
--- /dev/null
+++ b/t/fixtures/openai/chat-streaming-incomplete-no-usage.sse
@@ -0,0 +1,9 @@
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{"content":"!"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
+
+data: {"choices":
diff --git a/t/fixtures/openai/chat-streaming-injection-no-usage.sse
b/t/fixtures/openai/chat-streaming-injection-no-usage.sse
new file mode 100644
index 000000000..b2c430046
--- /dev/null
+++ b/t/fixtures/openai/chat-streaming-injection-no-usage.sse
@@ -0,0 +1,10 @@
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{"content":"injection
payload"},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{"content":""},"finish_reason":null}]}
+
+data:
{"id":"chatcmpl-nousage","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-2024-05-13","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
+
+data: [DONE]
+
diff --git a/t/fixtures/openai/moderation-eof-no-usage.sse
b/t/fixtures/openai/moderation-eof-no-usage.sse
new file mode 100644
index 000000000..9bba72258
--- /dev/null
+++ b/t/fixtures/openai/moderation-eof-no-usage.sse
@@ -0,0 +1,6 @@
+data:
{"choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data: {"choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
diff --git a/t/fixtures/openai/moderation-incomplete-event.sse
b/t/fixtures/openai/moderation-incomplete-event.sse
new file mode 100644
index 000000000..2f90bb69c
--- /dev/null
+++ b/t/fixtures/openai/moderation-incomplete-event.sse
@@ -0,0 +1,7 @@
+data:
{"choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data: {"choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data: {"choices":
diff --git a/t/fixtures/openai/moderation-no-usage.sse
b/t/fixtures/openai/moderation-no-usage.sse
new file mode 100644
index 000000000..1f3337827
--- /dev/null
+++ b/t/fixtures/openai/moderation-no-usage.sse
@@ -0,0 +1,8 @@
+data:
{"choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data: {"choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data: [DONE]
+
diff --git a/t/fixtures/openai/moderation-with-usage.sse
b/t/fixtures/openai/moderation-with-usage.sse
new file mode 100644
index 000000000..ef5ab28ab
--- /dev/null
+++ b/t/fixtures/openai/moderation-with-usage.sse
@@ -0,0 +1,10 @@
+data:
{"choices":[{"index":0,"delta":{"role":"assistant","content":null},"finish_reason":null}]}
+
+data: {"choices":[{"index":0,"delta":{"content":"kill
"},"finish_reason":null}]}
+
+data:
{"choices":[{"index":0,"delta":{"content":"you"},"finish_reason":"stop"}]}
+
+data:
{"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":8,"total_tokens":18}}
+
+data: [DONE]
+
diff --git a/t/lib/ai_moderation_result.lua b/t/lib/ai_moderation_result.lua
new file mode 100644
index 000000000..9bc25153d
--- /dev/null
+++ b/t/lib/ai_moderation_result.lua
@@ -0,0 +1,255 @@
+--
+-- 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.
+--
+
+local core = require("apisix.core")
+local http = require("resty.http")
+local sse = require("apisix.plugins.ai-transport.sse")
+local _M = {}
+
+function _M.setup()
+ local test = require("lib.test_admin").test
+ local core = require("apisix.core")
+ local code, body = test("/apisix/admin/consumers", ngx.HTTP_PUT, [[{
+ "username":"moderation-result",
+ "plugins":{
+ "key-auth":{"key":"moderation-result"},
+ "ai-aliyun-content-moderation":{
+ "endpoint":"http://127.0.0.1:6724","region_id":"cn-beijing";,
+ "access_key_id":"test-key","access_key_secret":"test-secret",
+ "check_request":true,"check_response":true,
+ "request_check_service":"query_security_check",
+ "response_check_service":"response_security_check",
+ "stream_check_mode":"final_packet","deny_message":"response
rejected"
+ }
+ }
+ }]])
+ assert(code < 300, body)
+ for id, route in ipairs({
+ {"/chat", "openai"}, {"/v1/messages", "anthropic"},
+ {"/v1/responses", "openai"}, {"/converted/v1/messages", "openai"}
+ }) do
+ local conf = {
+ uri = route[1],
+ plugins = {
+ ["key-auth"] = {},
+ ["ai-proxy"] = {
+ provider = route[2],
+ auth = {header = {Authorization = "Bearer test"}},
+ override = {endpoint = "http://127.0.0.1:6725"},
+ streaming_flush_interval_ms = 0,
+ },
+ ["serverless-post-function"] = {
+ phase = "log",
+ functions = {"return function(conf, ctx) "
+ ..
"require('lib.ai_moderation_result').record_usage(ctx) end"},
+ },
+ },
+ }
+ code, body = test("/apisix/admin/routes/" .. id, ngx.HTTP_PUT,
+ core.json.encode(conf))
+ assert(code < 300, body)
+ end
+end
+
+
+function _M.record_usage(ctx)
+ local request_id = ctx.var.http_x_result_request_id
+ assert(ngx.shared.test:set("accounting:" .. request_id, core.json.encode({
+ tokens = tonumber(ctx.var.llm_total_tokens) or 0,
+ raw_usage = ctx.llm_raw_usage or {},
+ })))
+end
+
+
+function _M.check(case)
+ local request_id = require("resty.jit-uuid").generate_v4()
+ local accounting_key = "accounting:" .. request_id
+ ngx.shared.test:delete(accounting_key)
+ for _, service in ipairs({"query_security_check",
"response_security_check"}) do
+ ngx.shared.test:delete(service .. "_calls")
+ ngx.shared.test:delete(service .. "_content")
+ end
+ local path = "/chat"
+ local body = {model = "test-model", stream = true,
+ messages = {{role = "user", content = "hello"}}}
+ if case.protocol == "anthropic" then
+ path = "/v1/messages"
+ body.max_tokens = 100
+ elseif case.protocol == "responses" then
+ path = "/v1/responses"
+ body.messages = nil
+ body.input = "hello"
+ end
+ if case.converted then
+ path = "/converted" .. path
+ end
+ local res = assert(http.new():request_uri("http://127.0.0.1:"; ..
ngx.var.server_port .. path, {
+ method = "POST",
+ headers = {
+ ["Content-Type"] = "application/json",
+ apikey = "moderation-result",
+ ["X-Result-Request-ID"] = request_id,
+ ["X-AI-Fixture"] = "aliyun/result-" .. case.fixture .. ".sse",
+ ["X-Buffered"] = case.buffered and "true" or nil,
+ ["X-Split-Done"] = case.split_done and "true" or nil,
+ },
+ body = core.json.encode(body),
+ }))
+ assert(res.status == 200, res.body)
+ if case.split_done then
+ assert(res.body:sub(1, 17) == ": keep-alive\r\n\r\n\n",
+ "complete-frame dispatch changed comments or blank lines")
+ end
+ local expected_text = case.text or "kill you"
+ for _, service in ipairs({"query_security_check",
"response_security_check"}) do
+ local expected_calls = case.error and not case.scanned
+ and service == "response_security_check" and 0
or 1
+ assert((ngx.shared.test:get(service .. "_calls") or 0) ==
expected_calls,
+ service .. ": wrong scan count")
+ end
+ assert(ngx.shared.test:get("response_security_check_content") ==
+ ((not case.error or case.scanned) and expected_text or nil),
+ "response scan did not receive all content")
+
+ if case.error then
+ assert(res.body:find(case.error_text or "stream failed", 1, true),
+ "upstream error was lost")
+ end
+ local events, remainder = sse.decode_buf(res.body)
+ assert(remainder == "", "partial downstream event")
+ local risk_count, done_count, start_count, injected_count = 0, 0, 0, 0
+ local original = {}
+ local finish_reason
+ for i, event in ipairs(events) do
+ if event.data == "[DONE]" or event.type == "message_stop"
+ or event.type == "response.completed" then
+ done_count = done_count + 1
+ assert(i == #events, "event emitted after completion")
+ end
+ if event.type == "message_start" or event.type == "response.created"
then
+ start_count = start_count + 1
+ end
+ if event.data == "[DONE]" then
+ goto CONTINUE
+ end
+ local data = assert(core.json.decode(event.data))
+ local injected = case.protocol == "chat" and data.risk_level and
data.usage
+ and data.usage.total_tokens == 0
+ if injected then
+ injected_count = injected_count + 1
+ end
+ if case.protocol == "chat" then
+ for _, choice in ipairs(data.choices or {}) do
+ if choice.finish_reason ~= core.json.null then
+ finish_reason = choice.finish_reason or finish_reason
+ end
+ if type(choice.delta.content) == "string" then
+ original[#original + 1] = choice.delta.content
+ end
+ end
+ elseif event.type == "content_block_delta" then
+ original[#original + 1] = data.delta.text
+ elseif event.type == "response.output_text.delta" then
+ original[#original + 1] = data.delta
+ end
+
+ if data.risk_level then
+ risk_count = risk_count + 1
+ assert(data.risk_level == (case.safe and "none" or "high"), "wrong
risk")
+ assert(data.deny_message == (case.safe and "" or "response
rejected"),
+ "missing top-level denial message")
+ if case.protocol == "chat" and injected then
+ assert(i == #events - 1, "moderation result must precede DONE")
+ assert(type(data.id) == "string" and type(data.created) ==
"number",
+ "invalid result metadata")
+ assert(data.model == "test-model", "invalid result model")
+ assert(type(data.choices) == "table" and #data.choices == 0
+ and event.data:find('"choices":[]', 1, true),
+ "moderation usage chunk must have empty choices")
+ assert(data.usage.prompt_tokens == 0 and
data.usage.completion_tokens == 0
+ and data.usage.total_tokens == 0, "result usage must be
zero")
+ elseif case.protocol == "anthropic" then
+ assert(event.type == "message_delta" or event.type ==
"content_block_delta",
+ "wrong Anthropic event")
+ if event.type ~= "message_delta" then
+ goto CONTINUE
+ end
+ if case.missing_stop then
+ assert(data.delta == core.json.null, "original delta was
replaced")
+ goto CONTINUE
+ end
+ assert(type(data.delta) == "table", "invalid Anthropic result
delta")
+ if case.stop_sequence then
+ assert(data.delta.stop_reason == "stop_sequence"
+ and data.delta.stop_sequence == case.stop_sequence,
+ "original Anthropic stop information was changed")
+ end
+ if data.usage then
+ assert(data.usage.output_tokens == (case.tokens and 8 or
0), "wrong usage")
+ if not case.tokens then
+ injected_count = injected_count + 1
+ end
+ end
+ elseif case.protocol == "responses" then
+ assert(event.type == "response.completed", "wrong Responses
event")
+ assert(data.response.id == "resp_moderation", "changed
response identity")
+ assert(data.response.output[1].content[1].text ==
expected_text,
+ "completed output was replaced")
+ if case.tokens then
+ assert(data.response.usage.total_tokens == case.tokens,
"usage was replaced")
+ end
+ end
+ end
+ ::CONTINUE::
+ end
+ assert(table.concat(original) == expected_text, "original response text
was changed")
+ if case.finish_reason then
+ assert(finish_reason == case.finish_reason, "original finish reason
was changed")
+ end
+ if case.failure or case.eof or (case.error and not case.scanned) then
+ assert(risk_count == 0, "unexpected moderation result")
+ else
+ assert(risk_count > 0, "missing moderation result")
+ end
+ if not case.error and not case.failure and case.protocol ~= "responses"
then
+ assert(injected_count == (case.tokens and 0 or 1), "wrong injected
result count")
+ end
+ assert(done_count == ((case.eof or case.error) and 0 or 1), "wrong
terminator count")
+ if case.protocol ~= "chat" then
+ assert(start_count == 1, "message/response start was replayed")
+ end
+ local deadline = ngx.now() + 1
+ local accounting
+ repeat
+ accounting = ngx.shared.test:get(accounting_key)
+ if accounting then
+ break
+ end
+ ngx.sleep(0.01)
+ until ngx.now() >= deadline
+ assert(accounting, "log phase did not publish this request's accounting")
+ accounting = assert(core.json.decode(accounting))
+ ngx.shared.test:delete(accounting_key)
+ assert(accounting.tokens == (case.tokens or 0), "token accounting changed")
+ if case.tokens then
+ local raw = accounting.raw_usage
+ assert((raw.prompt_tokens or raw.input_tokens) == 10, "raw input usage
changed")
+ assert((raw.completion_tokens or raw.output_tokens) == 8, "raw output
usage changed")
+ end
+end
+
+return _M
diff --git a/t/plugin/ai-aliyun-content-moderation-result.t
b/t/plugin/ai-aliyun-content-moderation-result.t
new file mode 100644
index 000000000..ab4d89cb4
--- /dev/null
+++ b/t/plugin/ai-aliyun-content-moderation-result.t
@@ -0,0 +1,322 @@
+#
+# 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.
+#
+
+BEGIN { $ENV{TEST_ENABLE_CONTROL_API_V1} = "0"; }
+use t::APISIX 'no_plan';
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+ $block->set_value("request", "GET /t");
+ $block->set_value("extra_yaml_config", <<_EOC_);
+plugins:
+ - ai-proxy
+ - ai-aliyun-content-moderation
+ - key-auth
+ - serverless-post-function
+_EOC_
+ $block->set_value("http_config", <<'_EOC_');
+ server {
+ listen 6724;
+ location / {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ ngx.req.read_body()
+ local args = ngx.req.get_post_args()
+ local params = assert(core.json.decode(args.ServiceParameters))
+ assert(ngx.shared.test:incr(args.Service .. "_calls", 1, 0))
+ assert(ngx.shared.test:set(args.Service .. "_content",
params.content))
+ if params.content == "scan-error" then
+ ngx.status = 500
+ ngx.say("scan failed")
+ return
+ end
+ require("lib.server").aliyun_moderation()
+ }
+ }
+ }
+ server {
+ listen 6725;
+ location / {
+ content_by_lua_block {
+ local fixture = ngx.req.get_headers()["X-AI-Fixture"]
+ local content =
assert(require("lib.fixture_loader").load(fixture))
+ ngx.header.content_type = "text/event-stream"
+ if ngx.req.get_headers()["X-Split-Done"] then
+ ngx.print(": keep-alive\r\n\r\n\n" .. content:sub(1, -6))
+ ngx.flush(true)
+ ngx.sleep(0.01)
+ ngx.print(content:sub(-5))
+ elseif ngx.req.get_headers()["X-Buffered"] then
+ ngx.print(content)
+ else
+ for event in content:gmatch("(.-\n\n)") do
+ ngx.print(event)
+ ngx.flush(true)
+ ngx.sleep(0.01)
+ end
+ end
+ }
+ }
+ }
+_EOC_
+ if ($block->case) {
+ my $case = $block->case;
+ $block->set_value("config", <<_EOC_);
+ location /t {
+ content_by_lua_block {
+ require("lib.ai_moderation_result").check($case)
+ ngx.say("passed")
+ }
+ }
+_EOC_
+ }
+});
+run_tests();
+__DATA__
+
+=== TEST 1: configure native protocol routes and Consumer moderation
+--- config
+ location /t {
+ content_by_lua_block {
+ require("lib.ai_moderation_result").setup()
+ ngx.say("passed")
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 2: Chat standalone DONE exposes denial and zero usage
+--- case: {fixture="chat-deny", protocol="chat"}
+--- response_body
+passed
+
+
+
+=== TEST 3: Chat coalesced JSON and DONE has the same result
+--- case: {fixture="chat-deny", protocol="chat", buffered=true}
+--- response_body
+passed
+
+
+
+=== TEST 4: Chat split DONE never exposes a partial terminator
+--- case: {fixture="chat-deny", protocol="chat", split_done=true}
+--- response_body
+passed
+
+
+
+=== TEST 5: Chat existing usage carries risk and denial without a new chunk
+--- case: {fixture="chat-usage", protocol="chat", tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 6: Chat safe result has no denial text
+--- case: {fixture="chat-safe", protocol="chat", text="safe output", safe=true}
+--- response_body
+passed
+
+
+
+=== TEST 7: Failed response scan cannot reuse the request risk level
+--- case: {fixture="chat-failure", protocol="chat", text="scan-error",
failure=true}
+--- response_body
+passed
+--- error_log
+failed to check content:
+
+
+
+=== TEST 8: All choices finish before the single moderation result
+--- case: {fixture="chat-multiple-choices", protocol="chat", tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 9: Anthropic without usage gets a result without new content blocks
+--- case: {fixture="anthropic-deny", protocol="anthropic"}
+--- response_body
+passed
+
+
+
+=== TEST 10: Anthropic split usage is decorated in place
+--- case: {fixture="anthropic-usage", protocol="anthropic", tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 11: Anthropic coalesced message_delta is decorated in place
+--- case: {fixture="anthropic-usage", protocol="anthropic", tokens=18,
buffered=true}
+--- response_body
+passed
+
+
+
+=== TEST 12: Responses completed retains output and denial metadata
+--- case: {fixture="responses-deny", protocol="responses"}
+--- response_body
+passed
+
+
+
+=== TEST 13: Responses completed preserves actual usage
+--- case: {fixture="responses-usage", protocol="responses", tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 14: Responses EOF does not fabricate a completed response
+--- case: {fixture="responses-eof", protocol="responses", eof=true}
+--- response_body
+passed
+
+
+
+=== TEST 15: Chat provider emits the final result in the Anthropic client
protocol
+--- case: {fixture="chat-usage", protocol="anthropic", converted=true,
tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 16: chat-error is preserved without a final moderation result
+--- case: {fixture="chat-error", protocol="chat", error=true}
+--- response_body
+passed
+
+
+
+=== TEST 17: Anthropic error after a scanned usage event is preserved without
a terminator
+--- case: {fixture="anthropic-error", protocol="anthropic", error=true,
tokens=18, scanned=true}
+--- response_body
+passed
+
+
+
+=== TEST 18: responses-error is preserved without a final moderation result
+--- case: {fixture="responses-error", protocol="responses", error=true}
+--- response_body
+passed
+
+
+
+=== TEST 19: responses-failed is preserved without a final moderation result
+--- case: {fixture="responses-failed", protocol="responses", error=true}
+--- response_body
+passed
+
+
+
+=== TEST 20: responses-incomplete is preserved without a final moderation
result
+--- case: {fixture="responses-incomplete", protocol="responses", error=true,
error_text="max_output_tokens"}
+--- response_body
+passed
+
+
+
+=== TEST 21: Chat result preserves the upstream length finish reason
+--- case: {fixture="chat-length", protocol="chat", tokens=18,
finish_reason="length"}
+--- response_body
+passed
+
+
+
+=== TEST 22: Chat result preserves the upstream tool_calls finish reason
+--- case: {fixture="chat-tool_calls", protocol="chat", tokens=18,
finish_reason="tool_calls"}
+--- response_body
+passed
+
+
+
+=== TEST 23: Anthropic result does not invent missing stop information
+--- case: {fixture="anthropic-null-delta", protocol="anthropic", tokens=18,
missing_stop=true}
+--- response_body
+passed
+
+
+
+=== TEST 24: Chat clean EOF does not invent a finish reason
+--- case: {fixture="chat-clean-eof", protocol="chat", tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 25: multiple choices retain their independent finish reasons
+--- case: {fixture="chat-multiple-reasons", protocol="chat", tokens=18,
finish_reason="tool_calls"}
+--- response_body
+passed
+
+
+
+=== TEST 26: anthropic safe result has an empty top-level denial message
+--- case: {fixture="anthropic-safe", protocol="anthropic", text="safe output",
safe=true, tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 27: anthropic safe result has an empty top-level denial message in an
existing event
+--- case: {fixture="anthropic-safe", protocol="anthropic", text="safe output",
safe=true, tokens=18, buffered=true}
+--- response_body
+passed
+
+
+
+=== TEST 28: responses safe result has an empty top-level denial message
+--- case: {fixture="responses-safe", protocol="responses", text="safe output",
safe=true, tokens=18}
+--- response_body
+passed
+
+
+
+=== TEST 29: Anthropic separate delta retains stop sequence
+--- case: {fixture="anthropic-stop-sequence", protocol="anthropic", tokens=18,
stop_sequence="END"}
+--- response_body
+passed
+
+
+
+=== TEST 30: Anthropic result preserves stop sequence in the original event
+--- case: {fixture="anthropic-stop-sequence", protocol="anthropic", tokens=18,
stop_sequence="END", buffered=true}
+--- response_body
+passed
+
+
+
+=== TEST 31: Chat coalesced usage retains the original chunks and billing
+--- case: {fixture="chat-usage", protocol="chat", tokens=18, buffered=true}
+--- response_body
+passed
+
+
+
+=== TEST 32: Chat without usage injects a result in the Anthropic client
protocol
+--- case: {fixture="chat-deny", protocol="anthropic", converted=true}
+--- response_body
+passed
diff --git a/t/plugin/ai-aliyun-content-moderation-stream-usage.t
b/t/plugin/ai-aliyun-content-moderation-stream-usage.t
new file mode 100644
index 000000000..e9816d6cd
--- /dev/null
+++ b/t/plugin/ai-aliyun-content-moderation-stream-usage.t
@@ -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.
+#
+
+use t::APISIX 'no_plan';
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+
+add_block_preprocessor(sub {
+ my ($block) = @_;
+ $block->set_value("request", "GET /t");
+ $block->set_value("extra_yaml_config", <<_EOC_);
+plugins:
+ - ai-proxy
+ - ai-aliyun-content-moderation
+ - key-auth
+_EOC_
+ $block->set_value("http_config", <<'_EOC_');
+ server {
+ listen 6724;
+ location / {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ ngx.req.read_body()
+ local args = ngx.req.get_post_args()
+ local params = assert(core.json.decode(args.ServiceParameters))
+ assert(ngx.shared.test:incr(args.Service .. "_calls", 1, 0))
+ assert(ngx.shared.test:set(args.Service .. "_content",
params.content))
+ local fixture = args.Service == "response_security_check"
+ and "aliyun/moderation-risk.json" or
"aliyun/moderation-safe.json"
+ ngx.header.content_type = "application/json"
+ ngx.print(assert(require("lib.fixture_loader").load(fixture)))
+ }
+ }
+ }
+_EOC_
+ if ($block->fixture) {
+ my $fixture = $block->fixture;
+ my $flush_events = $block->buffered ? "nil" : '"true"';
+ $block->set_value("config", <<_EOC_);
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local httpc = require("resty.http").new()
+ local res = assert(httpc:request_uri("http://127.0.0.1:"; ..
ngx.var.server_port
+ .. "/chat", {
+ method = "POST",
+ headers = {
+ ["Content-Type"] = "application/json",
+ ["apikey"] = "stream-usage-key",
+ ["X-AI-Fixture"] = "$fixture",
+ ["X-AI-Fixture-Flush-Events"] = $flush_events,
+ },
+ body =
[[{"messages":[{"role":"user","content":"hello"}],"stream":true}]],
+ }))
+ ngx.say("status: ", res.status)
+ for _, service in ipairs({"query_security_check",
"response_security_check"}) do
+ ngx.say(service, ": ", ngx.shared.test:get(service ..
"_calls"),
+ " / ", ngx.shared.test:get(service .. "_content"))
+ end
+ local text = {}
+ local events, remainder =
require("apisix.plugins.ai-transport.sse").decode_buf(res.body)
+ assert(remainder == "", "incomplete frame reached the client")
+ for _, event in ipairs(events) do
+ if event.data ~= "[DONE]" then
+ local data = assert(core.json.decode(event.data))
+ local content = core.table.try_read_attr(data, "choices",
1,
+ "delta",
"content")
+ if type(content) == "string" then
+ text[#text + 1] = content
+ end
+ end
+ end
+ ngx.say("text: ", table.concat(text))
+ local _, done_count = res.body:gsub("data: %[DONE%]", "")
+ ngx.say("done events: ", done_count)
+ }
+ }
+_EOC_
+ }
+});
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: configure consumer request and final-packet response moderation
+--- config
+ location /t {
+ content_by_lua_block {
+ local test = require("lib.test_admin").test
+ local code, body = test("/apisix/admin/consumers", ngx.HTTP_PUT,
[[{
+ "username": "stream-usage",
+ "plugins": {
+ "key-auth": {"key": "stream-usage-key"},
+ "ai-aliyun-content-moderation": {
+ "endpoint": "http://127.0.0.1:6724";,
+ "region_id": "cn-beijing",
+ "access_key_id": "fake-key-id",
+ "access_key_secret": "fake-key-secret",
+ "check_request": true,
+ "check_response": true,
+ "request_check_service": "query_security_check",
+ "response_check_service": "response_security_check",
+ "stream_check_mode": "final_packet"
+ }
+ }
+ }]])
+ assert(code < 300, body)
+ code, body = test("/apisix/admin/routes/1", ngx.HTTP_PUT, [[{
+ "uri": "/chat",
+ "plugins": {
+ "key-auth": {},
+ "ai-proxy": {
+ "provider": "openai",
+ "auth": {"header": {"Authorization": "Bearer test"}},
+ "override": {"endpoint": "http://127.0.0.1:1980"},
+ "streaming_flush_interval_ms": 0
+ }
+ }
+ }]])
+ assert(code < 300, body)
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 2: moderate complete text when the stream ends without usage
+--- fixture: openai/moderation-no-usage.sse
+--- response_body
+status: 200
+query_security_check: 1 / hello
+response_security_check: 1 / kill you
+text: kill you
+done events: 1
+
+
+
+=== TEST 3: preserve moderation when the stream includes usage
+--- fixture: openai/moderation-with-usage.sse
+--- response_body
+status: 200
+query_security_check: 1 / hello
+response_security_check: 1 / kill you
+text: kill you
+done events: 1
+
+
+
+=== TEST 4: moderate complete text at clean EOF without usage or a done event
+--- fixture: openai/moderation-eof-no-usage.sse
+--- response_body
+status: 200
+query_security_check: 1 / hello
+response_security_check: 1 / kill you
+text: kill you
+done events: 1
+
+
+
+=== TEST 5: do not finalize a stream with an incomplete trailing SSE event
+--- fixture: openai/moderation-incomplete-event.sse
+--- buffered: 1
+--- response_body
+status: 200
+query_security_check: 1 / hello
+response_security_check: nil / nil
+text: kill you
+done events: 0
+--- error_log
+dropping incomplete stream frame at EOF
diff --git a/t/plugin/ai-lakera-guard.t b/t/plugin/ai-lakera-guard.t
index 8ae98b449..7ae844538 100644
--- a/t/plugin/ai-lakera-guard.t
+++ b/t/plugin/ai-lakera-guard.t
@@ -917,22 +917,24 @@ passed
-=== TEST 37: a streamed response with no usage event cannot be scanned, so
fail-closed blocks it
+=== TEST 37: an incomplete streamed response cannot be assembled, so
fail-closed blocks it
--- request
POST /anything
{ "messages": [ { "role": "user", "content": "say hello" } ], "stream": true }
--- more_headers
-X-AI-Fixture: openai/chat-streaming-no-usage.sse
+X-AI-Fixture: openai/chat-streaming-incomplete-no-usage.sse
--- error_code: 200
--- response_body_like eval
qr/\A(?!.*Hello).*"content":"Response blocked by Lakera Guard".*\[DONE\]/s
--- error_log
streamed response ended without an assembled completion
fail_open=false, blocking response
+--- no_error_log
+ai-lakera-guard mock: forwarded body=
-=== TEST 38: create a direction=output route with fail_open for the no-usage
stream
+=== TEST 38: create a direction=output route with fail_open for an incomplete
stream
--- config
location /t {
content_by_lua_block {
@@ -970,18 +972,20 @@ passed
-=== TEST 39: with fail_open, an unscannable (no-usage) stream is released to
the client unscanned
+=== TEST 39: with fail_open, an incomplete stream is released to the client
unscanned
--- request
POST /anything
{ "messages": [ { "role": "user", "content": "say hello" } ], "stream": true }
--- more_headers
-X-AI-Fixture: openai/chat-streaming-no-usage.sse
+X-AI-Fixture: openai/chat-streaming-incomplete-no-usage.sse
--- error_code: 200
--- response_body_like eval
-qr/\A(?!.*Response blocked by Lakera Guard).*Hello.*\[DONE\]/s
+qr/\A(?!.*Response blocked by Lakera Guard).*Hello/s
--- error_log
streamed response ended without an assembled completion
fail_open=true, releasing unscanned
+--- no_error_log
+ai-lakera-guard mock: forwarded body=
@@ -1240,15 +1244,15 @@ passed
-=== TEST 50: a stream that ends at EOF with no terminal event is finalized
(fail-closed block), not stranded as keep-alive heartbeats
+=== TEST 50: an incomplete stream at EOF is finalized (fail-closed block), not
stranded as keep-alive heartbeats
--- request
POST /anything
{ "messages": [ { "role": "user", "content": "say hello" } ], "stream": true }
--- more_headers
-X-AI-Fixture: openai/chat-streaming-many-chunks-no-usage.sse
+X-AI-Fixture: openai/chat-streaming-incomplete-no-usage.sse
--- error_code: 200
--- response_body_like eval
-qr/\A(?!.*chunk-00).*"content":"Response blocked by Lakera Guard"/s
+qr/\A(?!.*Hello).*"content":"Response blocked by Lakera Guard"/s
--- error_log
streamed response ended without an assembled completion
fail_open=false, blocking response
@@ -1422,3 +1426,89 @@ POST /anything
qr/\Adata:.*injection payload.*\[DONE\]/s
--- error_log
ai-lakera-guard: response flagged by Lakera Guard
+
+
+
+=== TEST 58: create a direction=output block route for complete streams
without usage
+--- 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": "/anything",
+ "plugins": {
+ "ai-proxy": {
+ "provider": "openai-compatible",
+ "auth": { "header": { "Authorization": "Bearer
token" } },
+ "options": { "model": "gpt-4" },
+ "override": { "endpoint":
"http://127.0.0.1:1980/v1/chat/completions"; },
+ "ssl_verify": false
+ },
+ "ai-lakera-guard": {
+ "api_key": "test-key",
+ "lakera_endpoint": "http://127.0.0.1:6724/v2/guard";,
+ "direction": "output"
+ }
+ }
+ }]]
+ )
+
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 59: scan and release a complete no-usage stream
+--- request
+POST /anything
+{ "messages": [ { "role": "user", "content": "say hello" } ], "stream": true }
+--- more_headers
+X-AI-Fixture: openai/chat-streaming-no-usage.sse
+--- error_code: 200
+--- response_body_like eval
+qr/\A(?!.*Response blocked by Lakera Guard).*Hello.*\[DONE\]/s
+--- error_log
+"content":"Hello!"
+--- no_error_log
+streamed response ended without an assembled completion
+
+
+
+=== TEST 60: a flagged no-usage stream is scanned and blocked before its
content is released
+--- request
+POST /anything
+{ "messages": [ { "role": "user", "content": "say hello" } ], "stream": true }
+--- more_headers
+X-AI-Fixture: openai/chat-streaming-injection-no-usage.sse
+--- error_code: 200
+--- response_body_like eval
+qr/\A(?!.*injection payload).*Response blocked by Lakera Guard.*\[DONE\]/s
+--- error_log
+ai-lakera-guard mock: forwarded body=
+"injection payload"
+--- no_error_log
+streamed response ended without an assembled completion
+
+
+
+=== TEST 61: clean EOF without usage is scanned and released
+--- request
+POST /anything
+{ "messages": [ { "role": "user", "content": "say hello" } ], "stream": true }
+--- more_headers
+X-AI-Fixture: openai/chat-streaming-many-chunks-no-usage.sse
+--- error_code: 200
+--- response_body_like eval
+qr/\A(?!.*Response blocked by Lakera Guard).*chunk-00/s
+--- error_log
+ai-lakera-guard mock: forwarded body=
+--- no_error_log
+streamed response ended without an assembled completion
diff --git a/t/plugin/ai-proxy-stream-truncated.t
b/t/plugin/ai-proxy-stream-truncated.t
index 5e92f4b50..93e1b6d2f 100644
--- a/t/plugin/ai-proxy-stream-truncated.t
+++ b/t/plugin/ai-proxy-stream-truncated.t
@@ -319,11 +319,9 @@ passed
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
+# Final-packet moderation requires completion; a transport failure must not
+# fabricate a final risk result or a successful protocol terminator.
+qr/^(?!.*\[DONE\])(?=.*"risk_level":"none")(?=.*"content":"hello")/s
--- error_log
failed to read response chunk: closed
--- timeout: 10
