[Task][AI] Content moderation plugin for AI requests and responses #7623

Description

ShenYu can already reject a prompt containing a forbidden word: ai-sensitive-word (#7153, registered
in #7155) compiles a rule dictionary into an Aho–Corasick automaton and inspects the request body. That
covers deterministic keyword compliance. It cannot cover what the AI gateway space has converged on: a
judgement about the meaning of a prompt or of a model response.

Commercial gateways ship exactly that:

• Apache APISIX — ai-aliyun-content-moderation / ai-aws-content-moderation: check_request /
  check_response, stream_check_mode: realtime | final_packet, risk_level_bar, deny_code /
  deny_message, fail_mode.
• Higress — ai-security-guard: checkRequest / checkResponse, JSONPath extraction per direction,
  denyCode, ai_sec_request_deny / ai_sec_response_deny metrics.

All of them hard-code one moderation vendor (one plugin per provider). The gap this issue proposes to
fill is a judge-agnostic moderation plugin: the decision comes from an extension point, so a
self-hosted judge model and a commercial moderation API are the same thing to the gateway.

That is practical now: Qwen3Guard-Gen / Qwen3Guard-Stream, Llama Guard and ShieldGemma can all be
served behind an OpenAI-compatible /v1/chat/completions endpoint (vLLM, SGLang, Ollama), while a
vendor API is a thin adapter away. One OpenAI-compatible judge implementation therefore covers both,
with no vendor SDK inside the gateway.

Scope of this issue: design agreement first. I deliberately do not include the admin registration
rows or the console form here — I would like the behaviour and the extension point agreed before
generating tens of rows per dialect for six tables.

Design

1. Judge extension point

ContentModerationJudge is the only contract: given text plus the rule handle, return a verdict. Two
implementations ship with the plugin:

• openAiCompatibleJudge — any OpenAI-compatible chat endpoint (self-hosted guard model, or a hosted
  service speaking the same shape);
• httpApiJudge — a vendor moderation HTTP API driven by path/template fields, no vendor SDK.

Model, prompt template, threshold, timeout and action live in the rule handle, so the policy is
configuration rather than code (precedent: LiteLLM's decision_model guardrail, where the operator
writes natural-language questions, each with its own threshold and action).

2. Where the checks run, and plugin order

Two sides, independently configurable: checkRequest (before the prompt leaves) and checkResponse
(response content before the client sees it).

The response side requires an order lower than aiProxy (199), because a response decorator only
wraps responses produced by plugins ordered after it:

agentGateway 169 · aiPrompt 170 · aiTokenLimiter 171 · … · piiDetection 196 · sensitiveWord 197 · aiProxy 199

196 is already PII detection. The moderator order requires review against current master: define relative PII/dictionary/judge semantics and validate effective Admin overrides. Do not assume an unused slot or move Agent Gateway back after governance.

3. Verdict model — four states, per-rule action

Not two booleans. Following LiteLLM's guardrail_status:

This also borrows Bedrock Guardrails' per-rule action: BLOCK | NONE: "detect without blocking" is a
first-class setting, which is what makes an observe-then-enforce rollout possible. Keeping UNAVAILABLE
distinct matters — otherwise "the judge was down" and "the content was clean" are indistinguishable in
the metrics.

4. Streaming semantics

Three modes. The plugin must never claim more than it can do:

The rule that a previous prototype of mine got wrong, and which I want stated explicitly in the review:
a batch must never be released without its own verdict. Judging an accumulated window and then
releasing the current batch unchecked is retrospective detection, not pre-release moderation. In
realtime mode the judge receives the window (unreleased content plus the current batch) as context,
but the gate is always the current batch.

realtime / final_packet follow APISIX's stream_check_mode names and caveats; buffered is
LiteLLM's shipped default for its built-in guardrails ("buffers every chunk until the assembled
response passes moderation"), which is why I propose it as ours. The README and the rule form will
state plainly that in realtime / final_packet bytes already sent cannot be retracted.

One caution this design must not repeat: #6795 reports that collectList on streaming responses in the
AI transformer / MCP decorator paths breaks backpressure and buffers without bound. buffered mode must
therefore hold a bounded window (holdBackChars, with a policy when a single event exceeds it) rather
than aggregate a whole response, and final_packet should reuse the existing incremental SSE parsing
rather than introduce another full-body join.

5. Content extraction is configuration, not code

The prototype hard-codes choices[0].delta.content. The plugin should default to a protocol profile and
allow JSONPath overrides per direction:

Higress uses JSONPath for the same reason (requestContentJsonPath / responseContentJsonPath /
responseStreamContentJsonPath), and Envoy's guardrail adapter has an MCP-specific parser so the
provider only ever sees inspectable text.

Independently: matched content is never echoed to the caller (the same rule as ai-sensitive-word) —
only the verdict.

6. Failure policy — the two sides are not symmetric

7. Rejection response must be protocol-conformant

• Non-streaming: built through ResponsiveException + WebFluxResultUtils, never hand-assembled JSON.
• Streaming: a provider-conformant refusal event and a clean end of stream — never a bare JSON object
  injected into an SSE body, which no OpenAI client can parse.
• Never echo the matched text; a generic reason only.

8. Observability

Counters ai_sec_request_deny, ai_sec_response_deny, ai_sec_judge_failure plus a judge-latency
timer; exchange attributes (ai_sec_risklabel, ai_sec_deny_phase) and one mirrored risk-level
attribute so the logging and metrics plugins can consume the verdict instead of this plugin
re-implementing them (APISIX publishes llm_content_risk_level for the same reason).

9. Non-blocking guarantee

No Hystrix: upstream externalized it in #7499, and the enum only remains for compatibility. An HTTP
judge needs no thread hand-off at all — WebClient with .timeout(), onErrorResume and cancellation
propagation. If a blocking vendor SDK is ever wrapped it must be
subscribeOn(Schedulers.boundedElastic()) behind a bounded bulkhead, not a per-request thread pool.
Circuit breaking reuses the existing Resilience4j dependency (shenyu-plugin-resilience4j already
depends on resilience4j-reactor) with state shared per rule key, since per-request breaker state is
useless.

10. Dependencies and boundaries — ai-proxy is under active refactor

This plugin consumes wire format and plugin order only:

• It must not import org.apache.shenyu.plugin.ai.proxy.*. [Task] Refactor AI Proxy into protocol, provider, and transport layers #7351 is replacing those internal models
  (its own background: "tightly coupled to OpenAI Chat Completions and Spring AI types"), and [ISSUE #7381] Define internal models and extension contracts. #7397
  defines the new internal models and extension contracts.
• It must override both writeWith and writeAndFlushWith: AiProxyPlugin.writeResponse() writes
  through writeAndFlushWith, and this area is still moving (fix(ai): transform flushed response bodies fix(ai): transform flushed response bodies #7123, fix(ai-proxy): propagate stream cancellation Fix(ai-proxy): propagate stream cancellation #7338, avoid replaying partial streams [type:fix] avoid retrying partial AI streams #7205).
• It must not assume the response keeps being written through exchange.getResponse() ([ISSUE #7533] Migrate AI proxy plugin to engine execution #7543 already
  migrated ai-proxy to engine execution once). A fixture-based test with fixed SSE samples and no live
  upstream will fail if the response stops flowing through the decorator.
• No general AI protocol layer here — that is [Task][AI Proxy] Complete and extend native AI Protocol support #7382. If the Protocol contract later lands in
  shenyu-plugin-ai-common with a usable extraction hook, this plugin can consume it as a follow-up.
• The order coupling (response side must precede aiProxy) is documented, not encoded.
• The plugin also works on a plain HTTP route (request side only), like ai-sensitive-word, so it does
  not require ai-proxy to be configured.

11. Non-goals

• Admin plugin / plugin_handle / resource / permission / shenyu_dict /
  namespace_plugin_rel rows and the console form — a follow-up issue, per the scope note above
  ([Task] register the sensitive word plugin in the admin (db rows + rule form) and bound the inspected body #7154 is the precedent: the registration of ai-sensitive-word was tracked, and closed, separately
  from the plugin itself).
• The watermark plugin from the same prototype branch (unrelated concern; a separate proposal if wanted).
• Changes to shenyu-plugin-base — the prototype's shared response decorator belongs inside this module.
• Hosting or managing judge models; word-list dictionaries (that is ai-sensitive-word); a sidecar /
  ext_proc deployment of the judge as Envoy AI Gateway does (noted as a possible future architecture,
  not part of this).

Task List

1. Plugin module shenyu-plugin-ai/shenyu-plugin-ai-content-moderator, its starter, a PluginEnum
   entry and the rule-handle DTO in shenyu-common (plus its test).
2. ContentModerationJudge SPI + openAiCompatibleJudge + httpApiJudge.
3. Request side: size bound checked against the declared Content-Length before the body is read, read
   through ServerWebExchangeUtils.rewriteRequestBody.
4. Response side: non-streaming first, then streaming on both writeWith and writeAndFlushWith, with
   the three modes of §4.
5. Verdict model, failure policy, metrics, README (streaming caveats, order requirement, judge
   throughput).
6. Tests: StepVerifier for both write paths; per-batch gating; end-of-stream flush; short-stream tail
   (a stream shorter than the window must still be judged); no judge call for content-less frames;
   judge unavailable → configured policy; a test asserting the scan does not run on the event loop.
7. Follow-up issue: admin registration rows and the console form.

Open questions for maintainers (with my proposals)

I have written a proposal after each question so a maintainer can simply agree or redirect; nothing here
is decided on my own.

1. Plugin name and placement. Proposal: retain shenyu-plugin-ai/shenyu-plugin-ai-content-moderator / aiContentModerator in the AI family. Its final order and Admin registration need reviewed allocation; 196 is already PII's order.

2. Effective order and inspection semantics. Proposal: reconcile the complete path and RFC against master before allocation. Agent Gateway is 169. Decide whether judgement sees original or redacted content, whether dictionary rejection precedes the expensive judge, and how Admin overrides are validated. Do not reuse the historical unused-196 claim or renumber another plugin merely to fit it. Response decorators must precede the actual producer.

3. Default policy when the judge is unreachable. Proposal: fail open by default
   (failClosed=false), per-rule override to fail closed. Rationale: it matches the merged sibling
   ai-sensitive-word, whose README states that a broken dictionary must not take the traffic down;
   the moderation service is an external dependency shared by every rule that uses it, and the plugin is
   opt-in per rule, so a strict deployment can flip one flag. The request side and the response side are
   documented separately, because on the response side nothing can be retracted after the first byte is
   released.

4. Denial status and body for a recognized AI protocol. Proposal: denyStatusCode default 403
   with a provider-shaped body when a protocol is recognized. Rationale: 403 follows the sibling
   WafPlugin and keeps the rejection visible to infrastructure and monitoring, and the body is still
   built in the client's protocol so an AI SDK surfaces a parseable refusal rather than an opaque error.
   I am aware APISIX and Higress default to 200 specifically so the AI SDK treats the refusal as a
   normal completion; if maintainers prefer that behaviour for AI routes, I will make 200 the default
   there and keep 403 for plain HTTP. On the response side this only applies while nothing has been
   committed (i.e. before the first flush); afterwards the refusal can only be carried inside the stream.

5. Circuit breaking in v1, and the dependency. Proposal: v1 ships timeout + failClosed and no
   new dependency. If maintainers want circuit breaking in v1, implement it on the Resilience4j
   dependency the fault-tolerance plugins already bring (resilience4j-circuitbreaker / -reactor via
   shenyu-plugin-resilience4j) with breaker state shared per rule key. Rationale: a breaker is only
   meaningful with shared state, and that is already the pattern in Resilience4JRegistryFactory; adding
   a second resilience library, or a Hystrix-style per-request thread pool, would be scope this issue does
   not need.

6. Where the protocol extraction profiles live. Proposal: in this module for v1, with a follow-up
   to consume a shared hook if one appears. Rationale: [Task][AI Proxy] Complete and extend native AI Protocol support #7382 is still completing the Protocol layer's
   response and streaming semantics, and those contracts are for the proxy's own encode/decode path
   rather than an extraction API for other plugins; depending on them now would tie this plugin's PR to
   an in-flight refactor. If the extraction contract later lands in shenyu-plugin-ai-common, moving
   these profiles there is a small, contained change.

References

• Merged sibling: [type:feat] add the sensitive word plugin #7153 (ai-sensitive-word), [type:feat] register the sensitive word plugin in the admin #7155 (its admin registration).
• Related existing defect to avoid repeating: [BUG] — AI request/response transformer + MCP decorator: collectList on streaming responses breaks backpressure / unbounded in-memory buffering #6795 (unbounded buffering on streaming responses in the
  AI transformer / MCP decorator paths).
• ai-proxy layering: [Task] Refactor AI Proxy into protocol, provider, and transport layers #7351, [Task][AI Proxy] Establish initial layered contracts with a runnable OpenAI path #7381, [Task][AI Proxy] Complete and extend native AI Protocol support #7382, [Task][AI Proxy] Extend and harden the AI Provider layer #7383, [Task][AI Proxy] Harden the reactive AI Transport for production #7384.
• APISIX: ai-aliyun-content-moderation / ai-aws-content-moderation docs, and feat(ai-aws-content-moderation): moderate LLM responses, including streams apisix#13735
  for the response/streaming work.
• Higress: ai-security-guard.
• LiteLLM guardrails: modes pre_call | during_call | post_call | logging_only,
  streaming_buffer_until_moderated, guardrail_status, decision_model.
• Amazon Bedrock Guardrails: content filters (categories including PROMPT_ATTACK,
  inputStrength / outputStrength), and the standalone ApplyGuardrail API.
• Envoy AI Gateway / guardrail adapter over ext_proc.

Current ordering reconciliation and staged design (2026-10-30)

At master 8046953ca, PluginEnum has AG 169, Prompt 170, Token Limiter 171, PII 196, Sensitive Word 197 and Proxy 199. The historical AG-198/unused-196 assumptions are superseded. The RFC is still a proposal with the older ordering snapshot; reconcile it before allocating an order. This update does not allocate an enum/database ID or approve an API.

Publish effective-order/coverage rules including Admin overrides: identity/context, post-template content, PII/redaction, cheap dictionary, judge, budget admission and terminal response producer. Explain inspection semantics and judge latency/cost; report/reject response coverage when effective order makes it impossible.

Separate design acceptance from runnable delivery: bounded request judgement and buffered non-streaming responses first; then explicit streaming release modes with stated leakage/latency limits. Preserve unavailable versus safe, #8121 remaining budget, cancellation and idempotent cleanup. JSON schema compliance/repair (#8257) is not a safety verdict and cannot evade refusals or replay tools. Reuse compatible PII/protocol/audit mechanisms.

Safety metrics and inspection coverage (2026-10-30)

Own semantic safety judgement measurements under #8055 while reusing PII, trusted authorization and #7950 audit for their actual observations. Define policy-violation, prompt-injection-detected, PII-detected, unauthorized-tool-blocked and unsafe-output-blocked rates, plus data-exfiltration-risk alerts, by detector/policy coverage and declared units. Not every detector exists in this task; publish supported, externally supplied and unsupported sources instead of claiming universal implementation.

Expose inspected/evaluated denominators and skipped, unavailable, fail-open and unmatched coverage. Judge/regex unavailability is not safe, an uninspected request is not a negative detection, and a detection rate is not ground-truth attack prevalence. Distinguish observed violations, audit-only flags and enforced blocks; post-release detection cannot be counted as prevented leakage. Multiple policy detections may overlap and their rates must not be added as disjoint totals.

Keep tool authorization block events with the existing enforcement/#8047 boundary. Exfiltration-risk alerts need an explicit signal/policy/evidence source and are suspected risk rather than proof of a completed or prevented incident; absent coverage is unsupported, not zero. Test judge outage, skipped representations/streams, PII audit/redact/block, rejected tools and pre/post-release unsafe output with privacy-preserving bounded reasons. Sampling, redaction, access and retention remain required; do not collect raw content merely to populate a dashboard.