fix: Anthropic 流式 Usage Policy SSE 错误与 mapUpstreamError 对齐 (#503)

CHANGELOG.md

@@ -27,6 +27,8 @@
 
 ## 未发布
 
+- Anthropic 流式 `event: error` 经 `mapUpstreamError` 重映射，Usage Policy 与 non-stream 一致为 `policy_blocked` / `refusal_guard`。（#503）
+
 ## 1.3.182 — 2026-10-17
 
 - 代理地理可选 `KIN_PROXY_GEO_V6_LOOKUP_MODE=control-plane`：IPv6 探针仍经代理取真实出口，元数据查询走控制面默认路由；修正默认 IPv6 指定地址 URL（`/json/{ip}`）；仅显式开启且默认主查询传输失败时回退，并校验服务商返回 IP 与已验证 IPv6 一致（含缩写/展开）；默认 `proxy` 两阶段仍经代理。启用方式见 `docs/PANEL_API.md`。（#492、#307）

src/lib/protocol/handle-protocol.mjs

@@ -123,7 +123,12 @@ import {
   isClaudeCodeFamilyRequest,
 } from '../identity/claude-code-only.mjs'
 import { createDownstreamKeepalive } from './stream-keepalive.mjs'
-import { buildAnthropicSseError, createAnthropicStreamTracker, isRecoverableTruncation } from './stream-truncation.mjs'
+import {
+  buildAnthropicSseError,
+  createAnthropicSseErrorRemapper,
+  createAnthropicStreamTracker,
+  isRecoverableTruncation,
+} from './stream-truncation.mjs'
 import {
   hidePersonaUsageInSseLine,
   hidePersonaUsageOnMessage,
@@ -1115,6 +1120,8 @@ export function createHandleProtocol(deps) {
 
     const clientStream = isClientStream(inbound, req.headers)
     const streamTracker = clientStream && protocol === 'anthropic.messages' ? createAnthropicStreamTracker() : null
+    const anthropicSseErrorRemapper =
+      clientStream && protocol === 'anthropic.messages' ? createAnthropicSseErrorRemapper() : null
     // One keepalive for the whole Claude stream, including session/seat queues
     // and the gaps between failover attempts. OpenAI keeps its per-attempt one.
     let activeAttemptCommit = null
@@ -1456,7 +1463,8 @@ export function createHandleProtocol(deps) {
                 if (protocol === 'anthropic.messages') {
                   streamTracker?.observe(line)
                   if (!res.headersSent) writeSSEHeaders(res)
-                  res.write(String(line).endsWith('\n') ? String(line) : String(line) + '\n')
+                  const outbound = anthropicSseErrorRemapper ? anthropicSseErrorRemapper.remapLine(line) : line
+                  res.write(String(outbound).endsWith('\n') ? String(outbound) : String(outbound) + '\n')
                   return
                 }
                 const writeChunks = (chunks) => {

src/lib/protocol/stream-error-status.mjs

@@ -0,0 +1,24 @@
+import { isPlanLimitMessage } from '../pool/quota-window.mjs'
+
+const AUTH_ERROR_TEXT =
+  /authentication_error|token has been revoked|oauth_revoked|invalid_grant|invalid (?:bearer|x-api-key)|OAuth token has expired/i
+const PERMISSION_ERROR_TEXT = /organization does not have access to claude/i
+
+/**
+ * The kernel cli-hop answers 200 and then streams `event: error` (sub2api
+ * sseStreamErrorEventError). Before any downstream byte, restore the HTTP
+ * status the upstream meant so the pool classifies it like a real response.
+ */
+export function semanticStatusForStreamError(errorBody) {
+  const type = String(errorBody?.error?.type || errorBody?.type || '')
+  const message = String(errorBody?.error?.message || errorBody?.message || '')
+  if (isPlanLimitMessage(message)) return 429
+  const explicit = Number(errorBody?.error?.status ?? errorBody?.status)
+  if (explicit >= 400 && explicit < 600) return explicit
+  if (type === 'rate_limit_error') return 429
+  if (type === 'overloaded_error') return 529
+  if (type === 'authentication_error' || AUTH_ERROR_TEXT.test(message)) return 401
+  if (type === 'permission_error' || PERMISSION_ERROR_TEXT.test(message)) return 403
+  if (type === 'invalid_request_error') return 400
+  return 502
+}

src/lib/protocol/stream-truncation.mjs

@@ -8,6 +8,9 @@
  * A thinking block cut mid-way cannot be closed (it needs a signature).
  */
 
+import { mapUpstreamError, rewritePoolErrorForClient } from '../core/errors.mjs'
+import { semanticStatusForStreamError } from './stream-error-status.mjs'
+
 const CLOSABLE_BLOCKS = new Set(['text', 'tool_use'])
 
 function sse(event, data) {
@@ -24,6 +27,54 @@ export function buildAnthropicSseError(error = {}) {
   return sse('error', { type: 'error', error: body })
 }
 
+/**
+ * Remap upstream `event: error` SSE data lines through mapUpstreamError so
+ * streaming Anthropic clients see the same taxonomy as non-streaming responses.
+ */
+export function createAnthropicSseErrorRemapper({ headers = {} } = {}) {
+  let pendingEvent = null
+  return {
+    remapLine(line) {
+      const text = String(line)
+      const trimmed = text.trimEnd()
+      const bare = trimmed.trim()
+      if (bare.startsWith('event:')) {
+        pendingEvent = bare.slice(6).trim()
+        return line
+      }
+      if (bare === '') {
+        pendingEvent = null
+        return line
+      }
+      if (!bare.startsWith('data:') || pendingEvent !== 'error') return line
+      pendingEvent = null
+      let payload
+      try {
+        payload = JSON.parse(bare.slice(5).trim())
+      } catch {
+        return line
+      }
+      if (payload?.type !== 'error' && !payload?.error) return line
+      const hopStatus = semanticStatusForStreamError(payload)
+      const mapped = rewritePoolErrorForClient(mapUpstreamError(hopStatus, payload, headers), payload)
+      const err = mapped.body?.error
+      if (!err) return line
+      const suffix = text.slice(trimmed.length)
+      const data = JSON.stringify({
+        type: 'error',
+        error: {
+          type: err.type,
+          message: err.message,
+          ...(err.code ? { code: err.code } : {}),
+          ...(err.details && Object.keys(err.details).length ? { details: err.details } : {}),
+          ...(err.request_id ? { request_id: err.request_id } : {}),
+        },
+      })
+      return `data: ${data}${suffix || '\n'}`
+    },
+  }
+}
+
 export function createAnthropicStreamTracker() {
   let started = false
   let finished = false

src/lib/transport/go-worker-client.mjs

@@ -28,7 +28,7 @@ import {
   isCompleteAssistantMessage,
   isWrapConnectionError,
 } from '../core/errors.mjs'
-import { extraHeadersFromLimitError, isPlanLimitMessage } from '../pool/quota-window.mjs'
+import { extraHeadersFromLimitError } from '../pool/quota-window.mjs'
 
 const MAX_BODY = 64 * 1024 * 1024
 const credentialRefreshTail = new Map()
@@ -325,32 +325,9 @@ export function usageFromSseEvent(event) {
   return null
 }
 
-const AUTH_ERROR_TEXT =
-  /authentication_error|token has been revoked|oauth_revoked|invalid_grant|invalid (?:bearer|x-api-key)|OAuth token has expired/i
-// cli-hop can flatten an organization permission denial into generic api_error.
-// Preserve the denial instead of replacing it with an empty-hop error.
-const PERMISSION_ERROR_TEXT = /organization does not have access to claude/i
+import { semanticStatusForStreamError } from '../protocol/stream-error-status.mjs'
 
-/**
- * The kernel cli-hop answers 200 and then streams `event: error` (sub2api
- * sseStreamErrorEventError). Before any downstream byte, restore the HTTP
- * status the upstream meant so the pool classifies it like a real response.
- */
-export function semanticStatusForStreamError(errorBody) {
-  const type = String(errorBody?.error?.type || errorBody?.type || '')
-  const message = String(errorBody?.error?.message || errorBody?.message || '')
-  if (isPlanLimitMessage(message)) return 429
-  // The kernel names the status the job failed with; older kernels did not.
-  // Codex Responses Lite error events put it on the outer object (`{type,status,error}`).
-  const explicit = Number(errorBody?.error?.status ?? errorBody?.status)
-  if (explicit >= 400 && explicit < 600) return explicit
-  if (type === 'rate_limit_error') return 429
-  if (type === 'overloaded_error') return 529
-  if (type === 'authentication_error' || AUTH_ERROR_TEXT.test(message)) return 401
-  if (type === 'permission_error' || PERMISSION_ERROR_TEXT.test(message)) return 403
-  if (type === 'invalid_request_error') return 400
-  return 502
-}
+export { semanticStatusForStreamError } from '../protocol/stream-error-status.mjs'
 
 /**
  * Kernel `map_kernel` folds every provider failure into 502 `provider_error`.

test/unit/stream-truncation.test.mjs

@@ -2,6 +2,7 @@ import test from 'node:test'
 import assert from 'node:assert/strict'
 import {
   buildAnthropicSseError,
+  createAnthropicSseErrorRemapper,
   createAnthropicStreamTracker,
   isRecoverableTruncation,
 } from '../../src/lib/protocol/stream-truncation.mjs'
@@ -52,3 +53,30 @@ test('a worker timeout is not closed as max_tokens', () => {
     false,
   )
 })
+
+test('streaming Usage Policy SSE error remaps to policy_blocked / refusal_guard (#503)', () => {
+  const msg =
+    'provider error: provider error: API Error: Claude Code is unable to respond to this request, which appears to violate our Usage Policy (https://www.anthropic.com/legal/aup). Try rephrasing the request'
+  const upstream = JSON.stringify({
+    type: 'error',
+    error: { type: 'invalid_request_error', message: msg },
+  })
+  const remapper = createAnthropicSseErrorRemapper()
+  remapper.remapLine('event: error\n')
+  const remapped = remapper.remapLine(`data: ${upstream}\n`)
+  assert.match(remapped, /^data: /)
+  const payload = JSON.parse(remapped.trim().slice(5))
+  assert.equal(payload.type, 'error')
+  assert.equal(payload.error.type, 'policy_blocked')
+  assert.equal(payload.error.code, 'refusal_guard')
+  assert.notEqual(payload.error.type, 'invalid_request_error')
+  assert.notEqual(payload.error.code, 'upstream_invalid_request')
+  assert.equal(payload.error.details?.blocked_by, 'upstream')
+  assert.equal(payload.error.details?.retryable, false)
+})
+
+test('non-error SSE lines pass through the Anthropic error remapper unchanged', () => {
+  const remapper = createAnthropicSseErrorRemapper()
+  const line = 'data: {"type":"message_start","message":{"id":"msg_1","content":[]}}\n'
+  assert.equal(remapper.remapLine(line), line)
+})