diff --git a/CHANGELOG.md b/CHANGELOG.md index e582645f2..822718387 100644 --- a/CHANGELOG.md +++ b/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） diff --git a/src/lib/protocol/handle-protocol.mjs b/src/lib/protocol/handle-protocol.mjs index abda32e83..67815fdcb 100644 --- a/src/lib/protocol/handle-protocol.mjs +++ b/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) => { diff --git a/src/lib/protocol/stream-error-status.mjs b/src/lib/protocol/stream-error-status.mjs new file mode 100644 index 000000000..ee1c9867d --- /dev/null +++ b/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 +} diff --git a/src/lib/protocol/stream-truncation.mjs b/src/lib/protocol/stream-truncation.mjs index 21caa7393..1b24d6029 100644 --- a/src/lib/protocol/stream-truncation.mjs +++ b/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 diff --git a/src/lib/transport/go-worker-client.mjs b/src/lib/transport/go-worker-client.mjs index 2b6780f3d..9fbdf7bab 100644 --- a/src/lib/transport/go-worker-client.mjs +++ b/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`. diff --git a/test/unit/stream-truncation.test.mjs b/test/unit/stream-truncation.test.mjs index e4b7022da..1384a093e 100644 --- a/test/unit/stream-truncation.test.mjs +++ b/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) +})