Streaming /v1/messages and /v1/responses skip TPM/TPD + release concurrency early

[jarvis9443]

What

Streamed /v1/messages and /v1/responses don't reconcile the rate-limit reservation the way streamed /v1/chat/completions does. When usage_handled_by_stream is set, the outer dispatch skips reservation.commit_tokens(...) and the reservation simply drops at handler return. Two consequences:

1. TPM/TPD isn't enforced for streaming. The terminal token counts are only known inside the stream's Drop guard (which emits the UsageEvent), so commit_tokens never runs — the token-rate counter never moves. A caller can exceed TPM/TPD by streaming through these two endpoints.
2. The concurrency permit is released at handler return, not at stream end. Because the reservation drops when dispatch() returns (the streaming body outlives it), a key capped at N concurrent can run more than N simultaneous streams — the streaming analog of the /v1/chat/completions fix in security: make quota, streaming, and telemetry accounting consistent across request modes #450.

RPM (counted at pre_commit) and the $ budget (enforced via the cp-api ledger fed by the post-stream UsageEvent) are unaffected, so the residual gap is token-rate + concurrent-stream count, both bounded by the $ budget.

Where

• crates/aisix-proxy/src/messages.rs — outer dispatch skips commit_tokens on usage_handled_by_stream; the two streaming variants (build_anthropic_passthrough_stream verbatim + the cross-provider bridge) emit their UsageEvent from the Drop guard.
• crates/aisix-proxy/src/responses.rs — same shape for the verbatim /v1/responses stream.

Both carry an inline comment marking this as a tracked follow-up.

Fix direction

Mirror chat.rs: capture reservation.keys(), convert the reservation into into_stream_hold() so the concurrency permit lives for the stream's lifetime, move the hold + keys into the stream-completion closure, and call limiter.add_tokens_post_stream(key, total) with the terminal usage. The wrinkle vs chat.rs is that these two endpoints build the SSE body inside the per-attempt dispatch_to_target (inside the retry loop) rather than after a winner is resolved, so the reservation ownership has to be threaded down to the winning streaming attempt (e.g. &mut Option<MultiReservation> taken only on the streaming-success path, left in place for non-streaming / retry).

Needs a DP standalone E2E that streams enough tokens to trip a low TPM cap and asserts the next request is 429 (fails before the fix, passes after), plus a concurrent-streams test for the permit hold.

Surfaced by CodeRabbit review on #683; split out to keep that security PR focused.

[coderabbitai]

🔗 Related PRs

#363 - fix(proxy): record messages stream telemetry [merged]
#480 - fix(anthropic): capture message_start input tokens in streaming (#450 #1) [merged]
#481 - fix(ratelimit): hold streaming concurrency until stream end (#450 #30) [merged]
#491 - fix(messages): output guardrails on /v1/messages, streaming + non-streaming (#448) [merged]
#613 - fix(telemetry): emit usage event on streaming /v1/responses (#808) [closed]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!