From 7a9a6e60e90419b80e6bd85236fe5b832f7f6053 Mon Sep 17 00:00:00 2001 From: lojanica Date: Mon, 31 Aug 2026 07:48:18 -0300 Subject: [PATCH 1/4] feat: suporte nativo a API Anthropic Messages (/v1/messages) e correcoes na UI do Qwen --- src/api/server.ts | 26 +- src/routes/anthropic.ts | 516 +++++++++++++++++++++++++++++ src/services/browser-manager.ts | 20 +- src/services/header-interceptor.ts | 14 +- src/services/stream-creator.ts | 4 +- src/services/warm-pool.ts | 12 +- 6 files changed, 568 insertions(+), 24 deletions(-) create mode 100644 src/routes/anthropic.ts diff --git a/src/api/server.ts b/src/api/server.ts index f02e7d99..c4880955 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -9,6 +9,7 @@ import { cache } from '../cache/memory-cache.js' import { Watchdog } from '../core/watchdog.js' import { app as modelsApp } from './models.js' import { chatCompletions, chatCompletionsStop } from '../routes/chat.js' +import { anthropicMessages } from '../routes/anthropic.js' import { uploadFile } from '../routes/upload.js' import { adminApp } from './admin.js' import { getBaseAccountId, makeAccountLaneId } from '../core/account-lanes.js' @@ -63,10 +64,11 @@ app.use('/v1/*', async (c, next) => { if (!apiKey) { return c.json({ error: 'AUTH_REQUIRED=true but no API_KEY is configured' }, 500) } - const auth = c.req.header('Authorization') - if (!auth?.startsWith('Bearer ')) { + const rawAuth = c.req.header('Authorization') || c.req.header('x-api-key') + if (!rawAuth) { return c.json({ error: 'Missing or invalid Authorization header' }, 401) } + const auth = rawAuth.startsWith('Bearer ') ? rawAuth : `Bearer ${rawAuth}` const { resolveUserFromAuthHeader } = await import('../core/user-manager.js') const identity = resolveUserFromAuthHeader(auth) if (!identity) { @@ -84,6 +86,26 @@ app.post('/v1/chat/completions', bodyLimit({ }), chatCompletions) app.post('/v1/chat/completions/stop', chatCompletionsStop) app.post('/v1/upload', uploadFile) +app.post('/v1/messages', bodyLimit({ + maxSize: 52 * 1024 * 1024, + onError: (c: Context) => c.json({ error: { message: 'Request body too large' } }, 413), +}), anthropicMessages) +app.post('/v1/messages/count_tokens', async (c) => { + const body = await c.req.json() + let chars = 0 + if (typeof body.system === 'string') chars += body.system.length + if (Array.isArray(body.messages)) { + for (const m of body.messages) { + if (typeof m.content === 'string') chars += m.content.length + else if (Array.isArray(m.content)) { + for (const b of m.content) { + if (b.text) chars += b.text.length + } + } + } + } + return c.json({ input_tokens: Math.max(1, Math.ceil(chars / 4)) }) +}) // Admin dashboard (served at /admin). app.route('/admin', adminApp) diff --git a/src/routes/anthropic.ts b/src/routes/anthropic.ts new file mode 100644 index 00000000..40aae84b --- /dev/null +++ b/src/routes/anthropic.ts @@ -0,0 +1,516 @@ +import type { Context } from 'hono'; +import { stream as honoStream } from 'hono/streaming'; +import crypto from 'crypto'; +import type { OpenAIRequest } from '../utils/types.js'; +import { createQwenStream, RetryableQwenStreamError } from '../services/qwen.js'; +import { getNextAccount, getNextAvailableAccount, getAccountById, markAccountRateLimited, onAccountFreed, getAccountCooldownInfo, markAccountInUse, releaseAccountInUse } from '../core/account-manager.js'; +import { loadAccounts } from '../core/accounts.js'; +import { registerStream, removeStream } from '../core/stream-registry.js'; +import { metrics } from '../core/metrics.js'; +import { config } from '../core/config.js'; +import { checkUserRateLimit, tryAcquireUserSlot, releaseUserSlot, getUserActiveStreams } from '../core/user-manager.js'; +import type { UserIdentity } from '../core/user-manager.js'; +import { countTokens } from '../core/tokenizer.js'; +import { QwenStreamParser } from '../utils/qwen-stream-parser.js'; +import { collectNonStreamingResult } from './stream-handler.js'; +import { trackUsage, trackModelUsage } from '../core/usage-tracker.js'; + +function resolveModelName(model?: string): string { + if (!model) return "qwen3.7-plus"; + const m = model.toLowerCase(); + if (m === "qwen-plus" || m.startsWith("qwen-plus")) return "qwen3.7-plus"; + if (m === "qwen-max" || m.startsWith("qwen-max")) return "qwen3.8-max"; + if (m.includes("max") || m.includes("opus")) { + return m.includes("thinking") ? "qwen3.8-max-thinking" : "qwen3.8-max"; + } + if (m.includes("thinking")) { + return "qwen3.7-plus-thinking"; + } + if (m.startsWith("claude") || m === "sonnet" || m === "haiku") { + return "qwen3.7-plus"; + } + return model; +} + +export async function anthropicMessages(c: Context) { + const user = (c as any).get?.('user') as UserIdentity | undefined; + let userSlotHeld = false; + let userSlotReleased = false; + const releaseUserSlotOnce = () => { + if (!userSlotHeld || userSlotReleased || !user) return; + userSlotReleased = true; + releaseUserSlot(user.id); + }; + + const startTime = Date.now(); + + try { + const body = await c.req.json(); + const isStream = body.stream ?? false; + metrics.increment('requests.completions'); + + if (user) { + if (!checkUserRateLimit(user.id, user.rateLimitRpm)) { + return c.json({ type: 'error', error: { type: 'rate_limit_error', message: `Rate limit exceeded for user ${user.id}` } }, 429); + } + if (!tryAcquireUserSlot(user.id, user.maxConcurrency)) { + return c.json({ type: 'error', error: { type: 'rate_limit_error', message: `Concurrency limit exceeded for user ${user.id} (max ${user.maxConcurrency})` } }, 429); + } + userSlotHeld = true; + } + + const rawModel = body.model || 'qwen-plus'; + const targetModel = resolveModelName(rawModel); + const isThinkingModel = targetModel.endsWith('-thinking'); + + // 1. Build system prompt + let systemPrompt = ''; + if (typeof body.system === 'string') { + systemPrompt = body.system; + } else if (Array.isArray(body.system)) { + systemPrompt = body.system.map((s: any) => (typeof s === 'string' ? s : s.text || '')).join('\n'); + } + + // 2. Build messages and prompt string + const promptParts: string[] = []; + if (systemPrompt.trim()) { + promptParts.push(`System: ${systemPrompt.trim()}\n`); + } + + const messages = Array.isArray(body.messages) ? body.messages : []; + for (const msg of messages) { + if (typeof msg.content === 'string') { + if (msg.role === 'assistant') { + promptParts.push(`Assistant: ${msg.content}\n`); + } else { + promptParts.push(`User: ${msg.content}\n`); + } + } else if (Array.isArray(msg.content)) { + let textPart = ''; + const toolCalls: any[] = []; + const toolResults: any[] = []; + + for (const block of msg.content) { + if (block.type === 'text') { + textPart += (textPart ? '\n' : '') + (block.text || ''); + } else if (block.type === 'tool_use') { + const args = typeof block.input === 'string' ? block.input : JSON.stringify(block.input || {}); + toolCalls.push({ name: block.name, arguments: args }); + } else if (block.type === 'tool_result') { + const content = typeof block.content === 'string' ? block.content : JSON.stringify(block.content || ''); + toolResults.push({ id: block.tool_use_id, content }); + } + } + + if (msg.role === 'assistant') { + let assistantContent = textPart; + for (const tc of toolCalls) { + let parsedArgs = tc.arguments; + try { parsedArgs = JSON.parse(tc.arguments); } catch {} + const callStr = `\n\n${JSON.stringify({ name: tc.name, arguments: parsedArgs })}\n`; + assistantContent = assistantContent ? assistantContent + callStr : callStr.trim(); + } + if (assistantContent) { + promptParts.push(`Assistant: ${assistantContent}\n`); + } + } else { + if (textPart) { + promptParts.push(`User: ${textPart}\n`); + } + for (const tr of toolResults) { + promptParts.push(`Tool Response: ${tr.content}\n`); + } + } + } + } + + const hasTools = Array.isArray(body.tools) && body.tools.length > 0; + if (hasTools) { + const formattedTools = body.tools.map((t: any) => ({ + name: t.name, + description: t.description || '', + parameters: t.input_schema || {} + })); + const toolsJson = JSON.stringify(formattedTools); + const toolDirective = `\n\n# TOOLS AVAILABLE\nYou have access to the following tools:\n${toolsJson}\n\n# TOOL CALLING FORMAT (MANDATORY)\nTo use a tool, you MUST output a JSON object wrapped EXACTLY in tags:\n\n\n{"name": "tool_name", "arguments": {"param_name": "value"}}\n\n\nCRITICAL RULES:\n1. ONLY use the tags above for tool calling.\n2. Output tool call immediately without preamble when needed.\n\n`; + promptParts.unshift(toolDirective); + } + + const finalPrompt = promptParts.join('\n'); + const inputTokens = countTokens(finalPrompt); + const completionId = `comp_${crypto.randomUUID().replace(/-/g, '')}`; + const stopToken = crypto.randomUUID(); + + // Stream retrieval logic matching qwenproxy core + const isGuestModeOnly = config.qwen.guestModeOnly; + const baseStreamOptions = { + streamOptions: undefined, + completionId, + temperature: body.temperature, + top_p: body.top_p, + max_tokens: body.max_tokens, + authUserId: user?.id, + }; + + let retries = 3; + let streamResult: { stream: ReadableStream; uiSessionId: string } | null = null; + + if (isGuestModeOnly) { + const result = await createQwenStream( + finalPrompt, + isThinkingModel, + targetModel, + null, + 'guest', + undefined, + undefined, + { ...baseStreamOptions, forceBootstrap: true } + ); + registerStream(completionId, { + abortController: result.controller, + accountId: 'guest', + uiSessionId: result.uiSessionId, + targetResponseId: '', + headers: result.headers, + stopToken, + }); + streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; + } else { + const accounts = loadAccounts(); + const account = getNextAccount(); + const accountId = account?.id; + + if (!accountId) { + // Fallback to guest + const result = await createQwenStream( + finalPrompt, + isThinkingModel, + targetModel, + null, + 'guest', + undefined, + undefined, + { ...baseStreamOptions, forceBootstrap: true } + ); + registerStream(completionId, { + abortController: result.controller, + accountId: 'guest', + uiSessionId: result.uiSessionId, + targetResponseId: '', + headers: result.headers, + stopToken, + }); + streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; + } else { + markAccountInUse(accountId); + try { + const result = await createQwenStream( + finalPrompt, + isThinkingModel, + targetModel, + null, + accountId === 'global' ? undefined : accountId, + undefined, + undefined, + { ...baseStreamOptions, forceBootstrap: false } + ); + registerStream(completionId, { + abortController: result.controller, + accountId: result.accountId, + uiSessionId: result.uiSessionId, + targetResponseId: '', + headers: result.headers, + stopToken, + }); + releaseAccountInUse(accountId); + streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; + } catch (err: any) { + releaseAccountInUse(accountId); + // Guest fallback on failure + const result = await createQwenStream( + finalPrompt, + isThinkingModel, + targetModel, + null, + 'guest', + undefined, + undefined, + { ...baseStreamOptions, forceBootstrap: true } + ); + registerStream(completionId, { + abortController: result.controller, + accountId: 'guest', + uiSessionId: result.uiSessionId, + targetResponseId: '', + headers: result.headers, + stopToken, + }); + streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; + } + } + } + + const onComplete = () => { + removeStream(completionId); + releaseUserSlotOnce(); + }; + + if (isStream) { + return handleAnthropicStream( + c, + streamResult.stream, + rawModel, + completionId, + streamResult.uiSessionId, + inputTokens, + hasTools, + body.tools || [], + onComplete + ); + } else { + return handleAnthropicNonStreaming( + c, + streamResult.stream, + rawModel, + streamResult.uiSessionId, + inputTokens, + hasTools, + body.tools || [], + onComplete + ); + } + + } catch (err: any) { + releaseUserSlotOnce(); + console.error('[Anthropic API Error]:', err); + return c.json({ type: 'error', error: { type: 'api_error', message: err.message || 'Internal Server Error' } }, 500); + } +} + +function handleAnthropicStream( + c: Context, + stream: ReadableStream, + model: string, + completionId: string, + uiSessionId: string, + inputTokens: number, + hasTools: boolean, + tools: any[], + onComplete?: () => void, +) { + const socket = (c.env as any)?.incoming?.socket || (c.req.raw as any).socket; + if (socket && typeof socket.setNoDelay === 'function') { + socket.setNoDelay(true); + } + + c.header('Content-Type', 'text/event-stream'); + c.header('Cache-Control', 'no-cache, no-transform'); + c.header('Connection', 'keep-alive'); + c.header('X-Accel-Buffering', 'no'); + + return honoStream(c, async (streamWriter: any) => { + let heartbeatInterval: any; + let blockIndex = 0; + let textBlockOpen = false; + let totalOutputTokens = 0; + let stopReason = 'end_turn'; + const msgId = `msg_${crypto.randomUUID().replace(/-/g, '').slice(0, 24)}`; + + const sendEvent = (event: string, data: any) => { + streamWriter.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + }; + + try { + sendEvent('message_start', { + type: 'message_start', + message: { + id: msgId, + type: 'message', + role: 'assistant', + model, + content: [], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: inputTokens, output_tokens: 1 }, + }, + }); + + heartbeatInterval = setInterval(async () => { + try { + await streamWriter.write(': keep-alive\n\n'); + } catch { + clearInterval(heartbeatInterval); + } + }, 15000); + + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let streamEnded = false; + let rawBuffer = ''; + + const formattedTools = tools.map((t: any) => ({ + name: t.name, + description: t.description || '', + parameters: t.input_schema || {} + })); + + const qwenParser = new QwenStreamParser(uiSessionId, { + tools: hasTools ? formattedTools : [], + onAnswer: (deltaText: string) => { + if (!deltaText) return; + if (!textBlockOpen) { + textBlockOpen = true; + sendEvent('content_block_start', { + type: 'content_block_start', + index: blockIndex, + content_block: { type: 'text', text: '' }, + }); + } + sendEvent('content_block_delta', { + type: 'content_block_delta', + index: blockIndex, + delta: { type: 'text_delta', text: deltaText }, + }); + totalOutputTokens += Math.ceil(deltaText.length / 4); + }, + onToolCall: (tc) => { + stopReason = 'tool_use'; + if (textBlockOpen) { + sendEvent('content_block_stop', { type: 'content_block_stop', index: blockIndex }); + textBlockOpen = false; + blockIndex++; + } + const toolId = tc.id || `toolu_${crypto.randomUUID().replace(/-/g, '').slice(0, 16)}`; + const argsStr = typeof tc.arguments === 'string' ? tc.arguments : JSON.stringify(tc.arguments || {}); + + sendEvent('content_block_start', { + type: 'content_block_start', + index: blockIndex, + content_block: { + type: 'tool_use', + id: toolId, + name: tc.name, + input: {}, + }, + }); + sendEvent('content_block_delta', { + type: 'content_block_delta', + index: blockIndex, + delta: { + type: 'input_json_delta', + partial_json: argsStr, + }, + }); + sendEvent('content_block_stop', { type: 'content_block_stop', index: blockIndex }); + blockIndex++; + }, + }); + + while (!streamEnded) { + const { done, value } = await reader.read(); + if (done) break; + + rawBuffer += decoder.decode(value, { stream: true }); + const lines = rawBuffer.split('\n'); + rawBuffer = lines.pop() || ''; + + for (const line of lines) { + const trimmed = line.trim(); + if (trimmed === 'data: [DONE]') { + streamEnded = true; + break; + } + if (trimmed.startsWith('data: ')) { + qwenParser.parseLine(trimmed.slice(6)); + } + } + } + + if (rawBuffer.trim() && rawBuffer.trim().startsWith('data: ') && rawBuffer.trim() !== 'data: [DONE]') { + qwenParser.parseLine(rawBuffer.trim().slice(6)); + } + + if (textBlockOpen) { + sendEvent('content_block_stop', { type: 'content_block_stop', index: blockIndex }); + } + + sendEvent('message_delta', { + type: 'message_delta', + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { output_tokens: Math.max(1, totalOutputTokens) }, + }); + sendEvent('message_stop', { type: 'message_stop' }); + + } catch (err: any) { + console.error('[Anthropic Stream Error]:', err); + } finally { + if (heartbeatInterval) clearInterval(heartbeatInterval); + onComplete?.(); + } + }); +} + +async function handleAnthropicNonStreaming( + c: Context, + stream: ReadableStream, + model: string, + uiSessionId: string, + inputTokens: number, + hasTools: boolean, + tools: any[], + onComplete?: () => void, +) { + const formattedTools = tools.map((t: any) => ({ + name: t.name, + description: t.description || '', + parameters: t.input_schema || {} + })); + + const result = await collectNonStreamingResult( + c, + stream, + `comp_${crypto.randomUUID().replace(/-/g, '')}`, + model, + uiSessionId, + hasTools, + formattedTools, + onComplete, + ); + + const contentBlocks: any[] = []; + if (result.content) { + contentBlocks.push({ type: 'text', text: result.content }); + } + if (result.tool_calls && Array.isArray(result.tool_calls)) { + for (const tc of result.tool_calls) { + let inputObj = {}; + try { + inputObj = typeof tc.function?.arguments === 'string' ? JSON.parse(tc.function.arguments) : tc.function?.arguments || {}; + } catch { + inputObj = { raw: tc.function?.arguments }; + } + contentBlocks.push({ + type: 'tool_use', + id: tc.id || `toolu_${crypto.randomUUID().replace(/-/g, '').slice(0, 16)}`, + name: tc.function?.name, + input: inputObj, + }); + } + } + + const outTokens = result.usage?.completion_tokens || Math.ceil((result.content || '').length / 4); + + return c.json({ + id: `msg_${crypto.randomUUID().replace(/-/g, '').slice(0, 24)}`, + type: 'message', + role: 'assistant', + model, + content: contentBlocks, + stop_reason: result.tool_calls && result.tool_calls.length > 0 ? 'tool_use' : 'end_turn', + stop_sequence: null, + usage: { + input_tokens: inputTokens, + output_tokens: Math.max(1, outTokens), + }, + }); +} diff --git a/src/services/browser-manager.ts b/src/services/browser-manager.ts index 4668e4f2..b8db874b 100644 --- a/src/services/browser-manager.ts +++ b/src/services/browser-manager.ts @@ -501,10 +501,15 @@ export async function resetBrowserProfile(cacheKey: string, accountId?: string): } markAccountNotReady(accountId || cacheKey); markAccountNotReady(profileId); - fs.rmSync(profilePath, { recursive: true, force: true }); - fs.rmSync(storageStatePath(profileId), { force: true }); - - console.warn(`[Playwright] Cleared browser profile for ${cacheKey}: ${profilePath}`); + const { getAccountCredentials } = await import("../core/accounts.js"); + const hasCreds = accountId ? !!getAccountCredentials(getBaseAccountId(accountId))?.password : false; + if (accountId === "guest" || hasCreds) { + fs.rmSync(profilePath, { recursive: true, force: true }); + fs.rmSync(storageStatePath(profileId), { force: true }); + console.warn(`[Playwright] Cleared browser profile for ${cacheKey}: ${profilePath}`); + } else { + console.warn(`[Playwright] Preserving cookies/storage for manual login account: ${cacheKey}`); + } } catch (err: any) { console.warn(`[Playwright] Failed to clear browser profile for ${cacheKey}: ${err.message}`); } @@ -620,8 +625,11 @@ export async function initPlaywrightForAccount(account: QwenAccount, _headless = } if (await hasValidAuthCookie(acctPage)) { - await saveStorageState(acctContext, baseAccountId); - } + await saveStorageState(acctContext, baseAccountId); + const { markAccountReady } = await import("../core/account-manager.js"); + markAccountReady(account.id); + markAccountReady(baseAccountId); + } } export async function launchManualLoginAccount(accountId: string, browserType: BrowserType = 'chromium'): Promise<{ context: BrowserContext, page: Page }> { diff --git a/src/services/header-interceptor.ts b/src/services/header-interceptor.ts index bbf296d9..095ef98b 100644 --- a/src/services/header-interceptor.ts +++ b/src/services/header-interceptor.ts @@ -78,8 +78,8 @@ export async function getBasicHeaders(accountId?: string): Promise<{ cookie: str let bxUmidtoken = cache.currentHeaders['bx-umidtoken']; const bxV = cache.currentHeaders['bx-v'] || '2.5.36'; - if (!bxUa || !bxUmidtoken) { - console.log(`[Playwright] Missing bx-ua/bx-umidtoken for ${cacheKey}, triggering header interception...`); + if (!cache.cachedQwenHeaders && config.directFetch.enabled && (!bxUa || !bxUmidtoken)) { + console.log(`[Playwright] Capturing initial headers for ${cacheKey}...`); try { const result = await getQwenHeaders(true, accountId); bxUa = result.headers['bx-ua']; @@ -97,9 +97,7 @@ export async function getBasicHeaders(accountId?: string): Promise<{ cookie: str } } - if (bxUa && bxUmidtoken) { - markAccountReady(cacheKey); - } + markAccountReady(cacheKey); return { cookie, userAgent, bxV, bxUa, bxUmidtoken }; } @@ -184,7 +182,7 @@ export async function getGuestHeaders(): Promise> { await humanType(guestPage!, inputSelector, 'Hello'); await sleep(humanDelay(800, 1500)); - const selectors = ['.message-input-right-button-send .send-button', '.chat-prompt-send-button', 'button.send-button']; + const selectors = ['.message-input-right-button-send .send-button', '.chat-prompt-send-button', 'button.send-button', 'button[type="submit"]', 'button:has(svg)']; let clicked = false; for (const selector of selectors) { const btn = await guestPage!.$(selector); @@ -463,8 +461,8 @@ async function _getQwenHeadersInternalOnce(forceNew = false, accountId?: string) 'user-agent': reqHeaders['user-agent'] || '' }; - if (!extractedHeaders.cookie || !extractedHeaders['bx-ua']) { - console.log(`[Playwright] Intercepted request missing critical headers for ${cacheKey}, skipping...`); + if (!extractedHeaders.cookie) { + console.log(`[Playwright] Intercepted request missing cookie for ${cacheKey}, skipping...`); await route.continue(); return; } diff --git a/src/services/stream-creator.ts b/src/services/stream-creator.ts index 1052751e..75c7b37d 100644 --- a/src/services/stream-creator.ts +++ b/src/services/stream-creator.ts @@ -22,8 +22,8 @@ const BASE_TIMEOUT_MS = 120000; const TIMEOUT_PER_MB = 30000; function assertAntiBotHeaders(headers: Record, label: string): void { - if (!headers['cookie'] || !headers['user-agent'] || !headers['bx-ua'] || !headers['bx-umidtoken'] || !headers['bx-v']) { - throw new Error(`${label} missing required browser anti-bot headers`); + if (!headers["cookie"] || !headers["user-agent"]) { + throw new Error(`${label} missing required cookie or user-agent`); } } diff --git a/src/services/warm-pool.ts b/src/services/warm-pool.ts index b88fe0a0..a9cfbf72 100644 --- a/src/services/warm-pool.ts +++ b/src/services/warm-pool.ts @@ -64,15 +64,15 @@ function isWarmChatInFlight(accountId: string, chatId: string) { async function getBasicQwenHeaders(accountId?: string): Promise> { const { cookie, userAgent, bxV, bxUa, bxUmidtoken } = await getBasicHeaders(accountId); - if (!cookie || !userAgent || !bxV || !bxUa || !bxUmidtoken) { - throw new Error('Missing required browser anti-bot headers for warm pool'); + if (!cookie || !userAgent) { + throw new Error("Missing required cookie or user-agent for warm pool"); } return { cookie, - 'user-agent': userAgent, - 'bx-v': bxV, - 'bx-ua': bxUa, - 'bx-umidtoken': bxUmidtoken, + "user-agent": userAgent, + "bx-v": bxV || "2.5.36", + "bx-ua": bxUa || "", + "bx-umidtoken": bxUmidtoken || "", }; } From c37cd3252eaafdcffa732be2bb1bc6f247ba5a23 Mon Sep 17 00:00:00 2001 From: lojanica Date: Mon, 31 Aug 2026 08:43:26 -0300 Subject: [PATCH 2/4] fix(review): address CodeRabbit feedback on Anthropic API, token counter, session validation and stream lifecycle --- src/api/server.ts | 24 +- src/routes/anthropic.ts | 385 ++++++++++++++--------------- src/services/browser-manager.ts | 21 +- src/services/header-interceptor.ts | 2 +- 4 files changed, 220 insertions(+), 212 deletions(-) diff --git a/src/api/server.ts b/src/api/server.ts index c4880955..97f41d03 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -90,21 +90,31 @@ app.post('/v1/messages', bodyLimit({ maxSize: 52 * 1024 * 1024, onError: (c: Context) => c.json({ error: { message: 'Request body too large' } }, 413), }), anthropicMessages) -app.post('/v1/messages/count_tokens', async (c) => { - const body = await c.req.json() - let chars = 0 - if (typeof body.system === 'string') chars += body.system.length +app.post('/v1/messages/count_tokens', bodyLimit({ + maxSize: 52 * 1024 * 1024, + onError: (c: Context) => c.json({ error: { message: 'Request body too large' } }, 413), +}), async (c) => { + let body: any + try { + body = await c.req.json() + } catch { + return c.json({ type: 'error', error: { type: 'invalid_request_error', message: 'Invalid JSON body' } }, 400) + } + const promptParts: string[] = [] + if (typeof body.system === 'string') promptParts.push(body.system) if (Array.isArray(body.messages)) { for (const m of body.messages) { - if (typeof m.content === 'string') chars += m.content.length + if (typeof m.content === 'string') promptParts.push(m.content) else if (Array.isArray(m.content)) { for (const b of m.content) { - if (b.text) chars += b.text.length + if (b.text) promptParts.push(b.text) } } } } - return c.json({ input_tokens: Math.max(1, Math.ceil(chars / 4)) }) + const { countTokens } = await import('../core/tokenizer.js') + const fullText = promptParts.join('\n') + return c.json({ input_tokens: Math.max(1, countTokens(fullText)) }) }) // Admin dashboard (served at /admin). diff --git a/src/routes/anthropic.ts b/src/routes/anthropic.ts index 40aae84b..3b956b74 100644 --- a/src/routes/anthropic.ts +++ b/src/routes/anthropic.ts @@ -1,19 +1,17 @@ -import type { Context } from 'hono'; -import { stream as honoStream } from 'hono/streaming'; -import crypto from 'crypto'; -import type { OpenAIRequest } from '../utils/types.js'; -import { createQwenStream, RetryableQwenStreamError } from '../services/qwen.js'; -import { getNextAccount, getNextAvailableAccount, getAccountById, markAccountRateLimited, onAccountFreed, getAccountCooldownInfo, markAccountInUse, releaseAccountInUse } from '../core/account-manager.js'; -import { loadAccounts } from '../core/accounts.js'; -import { registerStream, removeStream } from '../core/stream-registry.js'; -import { metrics } from '../core/metrics.js'; -import { config } from '../core/config.js'; -import { checkUserRateLimit, tryAcquireUserSlot, releaseUserSlot, getUserActiveStreams } from '../core/user-manager.js'; -import type { UserIdentity } from '../core/user-manager.js'; -import { countTokens } from '../core/tokenizer.js'; -import { QwenStreamParser } from '../utils/qwen-stream-parser.js'; -import { collectNonStreamingResult } from './stream-handler.js'; -import { trackUsage, trackModelUsage } from '../core/usage-tracker.js'; +import type { Context } from "hono"; +import { stream as honoStream } from "hono/streaming"; +import crypto from "crypto"; +import type { OpenAIRequest } from "../utils/types.js"; +import { createQwenStream } from "../services/qwen.js"; +import { getNextAccount, getAccountById, markAccountRateLimited, releaseAccountInUse } from "../core/account-manager.js"; +import { loadAccounts } from "../core/accounts.js"; +import { registerStream, removeStream } from "../core/stream-registry.js"; +import { checkUserRateLimit, tryAcquireUserSlot, releaseUserSlot, getUserActiveStreams } from "../core/user-manager.js"; +import type { UserIdentity } from "../core/user-manager.js"; +import { countTokens } from "../core/tokenizer.js"; +import { QwenStreamParser } from "../utils/qwen-stream-parser.js"; +import { collectNonStreamingResult } from "./stream-handler.js"; +import { trackUsage, trackModelUsage } from "../core/usage-tracker.js"; function resolveModelName(model?: string): string { if (!model) return "qwen3.7-plus"; @@ -33,127 +31,101 @@ function resolveModelName(model?: string): string { } export async function anthropicMessages(c: Context) { - const user = (c as any).get?.('user') as UserIdentity | undefined; + const user = (c as any).get?.("user") as UserIdentity | undefined; let userSlotHeld = false; let userSlotReleased = false; + let completionId = `comp_${crypto.randomUUID().replace(/-/g, "")}`; + const releaseUserSlotOnce = () => { if (!userSlotHeld || userSlotReleased || !user) return; - userSlotReleased = true; releaseUserSlot(user.id); + userSlotReleased = true; }; - const startTime = Date.now(); - try { const body = await c.req.json(); - const isStream = body.stream ?? false; - metrics.increment('requests.completions'); + if (!body || typeof body !== "object") { + return c.json({ type: "error", error: { type: "invalid_request_error", message: "Invalid JSON body" } }, 400); + } + + const isStream = Boolean(body.stream); + const rawModel = body.model || "qwen3.7-plus"; + const targetModel = resolveModelName(rawModel); + const isThinkingModel = targetModel.includes("thinking"); if (user) { if (!checkUserRateLimit(user.id, user.rateLimitRpm)) { - return c.json({ type: 'error', error: { type: 'rate_limit_error', message: `Rate limit exceeded for user ${user.id}` } }, 429); + return c.json({ + type: "error", + error: { + type: "rate_limit_error", + message: `Rate limit exceeded for user ${user.id}`, + }, + }, 429); } + if (!tryAcquireUserSlot(user.id, user.maxConcurrency)) { - return c.json({ type: 'error', error: { type: 'rate_limit_error', message: `Concurrency limit exceeded for user ${user.id} (max ${user.maxConcurrency})` } }, 429); + return c.json({ + type: "error", + error: { + type: "rate_limit_error", + message: `Concurrency limit exceeded for user ${user.id} (max ${user.maxConcurrency})`, + }, + }, 429); } userSlotHeld = true; } - const rawModel = body.model || 'qwen-plus'; - const targetModel = resolveModelName(rawModel); - const isThinkingModel = targetModel.endsWith('-thinking'); - - // 1. Build system prompt - let systemPrompt = ''; - if (typeof body.system === 'string') { - systemPrompt = body.system; - } else if (Array.isArray(body.system)) { - systemPrompt = body.system.map((s: any) => (typeof s === 'string' ? s : s.text || '')).join('\n'); - } - - // 2. Build messages and prompt string - const promptParts: string[] = []; - if (systemPrompt.trim()) { - promptParts.push(`System: ${systemPrompt.trim()}\n`); + const messages = Array.isArray(body.messages) ? body.messages : []; + const openAIMessages: OpenAIRequest["messages"] = []; + + if (body.system) { + if (typeof body.system === "string") { + openAIMessages.push({ role: "system", content: body.system }); + } else if (Array.isArray(body.system)) { + const sysText = body.system.map((s: any) => s.text || "").join("\n"); + openAIMessages.push({ role: "system", content: sysText }); + } } - const messages = Array.isArray(body.messages) ? body.messages : []; for (const msg of messages) { - if (typeof msg.content === 'string') { - if (msg.role === 'assistant') { - promptParts.push(`Assistant: ${msg.content}\n`); - } else { - promptParts.push(`User: ${msg.content}\n`); - } + const role = msg.role === "assistant" ? "assistant" : "user"; + if (typeof msg.content === "string") { + openAIMessages.push({ role, content: msg.content }); } else if (Array.isArray(msg.content)) { - let textPart = ''; - const toolCalls: any[] = []; - const toolResults: any[] = []; - + let textParts = ""; for (const block of msg.content) { - if (block.type === 'text') { - textPart += (textPart ? '\n' : '') + (block.text || ''); - } else if (block.type === 'tool_use') { - const args = typeof block.input === 'string' ? block.input : JSON.stringify(block.input || {}); - toolCalls.push({ name: block.name, arguments: args }); - } else if (block.type === 'tool_result') { - const content = typeof block.content === 'string' ? block.content : JSON.stringify(block.content || ''); - toolResults.push({ id: block.tool_use_id, content }); - } - } - - if (msg.role === 'assistant') { - let assistantContent = textPart; - for (const tc of toolCalls) { - let parsedArgs = tc.arguments; - try { parsedArgs = JSON.parse(tc.arguments); } catch {} - const callStr = `\n\n${JSON.stringify({ name: tc.name, arguments: parsedArgs })}\n`; - assistantContent = assistantContent ? assistantContent + callStr : callStr.trim(); - } - if (assistantContent) { - promptParts.push(`Assistant: ${assistantContent}\n`); - } - } else { - if (textPart) { - promptParts.push(`User: ${textPart}\n`); - } - for (const tr of toolResults) { - promptParts.push(`Tool Response: ${tr.content}\n`); + if (block.type === "text") { + textParts += (block.text || "") + "\n"; + } else if (block.type === "tool_result") { + const contentStr = typeof block.content === "string" ? block.content : JSON.stringify(block.content || ""); + textParts += `[Tool Result for ${block.tool_use_id}]: ${contentStr}\n`; + } else if (block.type === "tool_use") { + textParts += `[Tool Use: ${block.name} (${block.id})]: ${JSON.stringify(block.input || {})}\n`; } } + openAIMessages.push({ role, content: textParts.trim() }); } } - const hasTools = Array.isArray(body.tools) && body.tools.length > 0; - if (hasTools) { - const formattedTools = body.tools.map((t: any) => ({ - name: t.name, - description: t.description || '', - parameters: t.input_schema || {} - })); - const toolsJson = JSON.stringify(formattedTools); - const toolDirective = `\n\n# TOOLS AVAILABLE\nYou have access to the following tools:\n${toolsJson}\n\n# TOOL CALLING FORMAT (MANDATORY)\nTo use a tool, you MUST output a JSON object wrapped EXACTLY in tags:\n\n\n{"name": "tool_name", "arguments": {"param_name": "value"}}\n\n\nCRITICAL RULES:\n1. ONLY use the tags above for tool calling.\n2. Output tool call immediately without preamble when needed.\n\n`; - promptParts.unshift(toolDirective); - } + const rawPromptText = openAIMessages.map(m => `${m.role}: ${m.content}`).join("\n"); + const inputTokens = Math.max(1, countTokens(rawPromptText)); - const finalPrompt = promptParts.join('\n'); - const inputTokens = countTokens(finalPrompt); - const completionId = `comp_${crypto.randomUUID().replace(/-/g, '')}`; - const stopToken = crypto.randomUUID(); + const finalPrompt = openAIMessages.map(m => { + const role = m.role === "assistant" ? "Assistant" : (m.role === "system" ? "System" : "User"); + return `${role}: ${m.content}`; + }).join("\n\n") + "\n\nAssistant:"; - // Stream retrieval logic matching qwenproxy core - const isGuestModeOnly = config.qwen.guestModeOnly; const baseStreamOptions = { - streamOptions: undefined, - completionId, - temperature: body.temperature, - top_p: body.top_p, - max_tokens: body.max_tokens, - authUserId: user?.id, + forceBootstrap: false, }; - let retries = 3; - let streamResult: { stream: ReadableStream; uiSessionId: string } | null = null; + const stopToken = crypto.randomUUID(); + const hasTools = Array.isArray(body.tools) && body.tools.length > 0; + + let streamResult: { stream: ReadableStream; uiSessionId: string }; + const accounts = loadAccounts(); + const isGuestModeOnly = accounts.length === 0; if (isGuestModeOnly) { const result = await createQwenStream( @@ -161,64 +133,60 @@ export async function anthropicMessages(c: Context) { isThinkingModel, targetModel, null, - 'guest', + "guest", undefined, undefined, { ...baseStreamOptions, forceBootstrap: true } ); registerStream(completionId, { abortController: result.controller, - accountId: 'guest', + accountId: "guest", uiSessionId: result.uiSessionId, - targetResponseId: '', + targetResponseId: "", headers: result.headers, stopToken, }); streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; } else { - const accounts = loadAccounts(); - const account = getNextAccount(); - const accountId = account?.id; - - if (!accountId) { - // Fallback to guest + const selectedAccount = getNextAccount(); + if (!selectedAccount) { const result = await createQwenStream( finalPrompt, isThinkingModel, targetModel, null, - 'guest', + "guest", undefined, undefined, { ...baseStreamOptions, forceBootstrap: true } ); registerStream(completionId, { abortController: result.controller, - accountId: 'guest', + accountId: "guest", uiSessionId: result.uiSessionId, - targetResponseId: '', + targetResponseId: "", headers: result.headers, stopToken, }); streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; } else { - markAccountInUse(accountId); + const accountId = selectedAccount.id; try { const result = await createQwenStream( finalPrompt, isThinkingModel, targetModel, null, - accountId === 'global' ? undefined : accountId, + accountId, undefined, undefined, - { ...baseStreamOptions, forceBootstrap: false } + baseStreamOptions ); registerStream(completionId, { abortController: result.controller, accountId: result.accountId, uiSessionId: result.uiSessionId, - targetResponseId: '', + targetResponseId: "", headers: result.headers, stopToken, }); @@ -226,22 +194,25 @@ export async function anthropicMessages(c: Context) { streamResult = { stream: result.stream, uiSessionId: result.uiSessionId }; } catch (err: any) { releaseAccountInUse(accountId); - // Guest fallback on failure + console.warn(`[Anthropic] Account ${accountId} stream failed, falling back to guest: ${err?.message}`); + if (/rate limit|429/i.test(err?.message || "")) { + markAccountRateLimited(accountId); + } const result = await createQwenStream( finalPrompt, isThinkingModel, targetModel, null, - 'guest', + "guest", undefined, undefined, { ...baseStreamOptions, forceBootstrap: true } ); registerStream(completionId, { abortController: result.controller, - accountId: 'guest', + accountId: "guest", uiSessionId: result.uiSessionId, - targetResponseId: '', + targetResponseId: "", headers: result.headers, stopToken, }); @@ -250,9 +221,13 @@ export async function anthropicMessages(c: Context) { } } - const onComplete = () => { + const onComplete = (outputTokens = 1) => { removeStream(completionId); releaseUserSlotOnce(); + if (user?.id) { + trackUsage(user.id, rawPromptText, false); + } + trackModelUsage(targetModel); }; if (isStream) { @@ -268,10 +243,11 @@ export async function anthropicMessages(c: Context) { onComplete ); } else { - return handleAnthropicNonStreaming( + return await handleAnthropicNonStreaming( c, streamResult.stream, rawModel, + completionId, streamResult.uiSessionId, inputTokens, hasTools, @@ -281,9 +257,10 @@ export async function anthropicMessages(c: Context) { } } catch (err: any) { + if (completionId) removeStream(completionId); releaseUserSlotOnce(); - console.error('[Anthropic API Error]:', err); - return c.json({ type: 'error', error: { type: 'api_error', message: err.message || 'Internal Server Error' } }, 500); + console.error("[Anthropic API Error]:", err); + return c.json({ type: "error", error: { type: "api_error", message: err.message || "Internal Server Error" } }, 500); } } @@ -296,37 +273,42 @@ function handleAnthropicStream( inputTokens: number, hasTools: boolean, tools: any[], - onComplete?: () => void, + onComplete?: (outTokens: number) => void, ) { const socket = (c.env as any)?.incoming?.socket || (c.req.raw as any).socket; - if (socket && typeof socket.setNoDelay === 'function') { + if (socket && typeof socket.setNoDelay === "function") { socket.setNoDelay(true); } - c.header('Content-Type', 'text/event-stream'); - c.header('Cache-Control', 'no-cache, no-transform'); - c.header('Connection', 'keep-alive'); - c.header('X-Accel-Buffering', 'no'); + c.header("Content-Type", "text/event-stream"); + c.header("Cache-Control", "no-cache, no-transform"); + c.header("Connection", "keep-alive"); + c.header("X-Accel-Buffering", "no"); return honoStream(c, async (streamWriter: any) => { + streamWriter.onAbort?.(() => { + removeStream(completionId); + }); + let heartbeatInterval: any; let blockIndex = 0; let textBlockOpen = false; let totalOutputTokens = 0; - let stopReason = 'end_turn'; - const msgId = `msg_${crypto.randomUUID().replace(/-/g, '').slice(0, 24)}`; + let stopReason = "end_turn"; + let reader: any; + const msgId = `msg_${crypto.randomUUID().replace(/-/g, "").slice(0, 24)}`; const sendEvent = (event: string, data: any) => { streamWriter.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); }; try { - sendEvent('message_start', { - type: 'message_start', + sendEvent("message_start", { + type: "message_start", message: { id: msgId, - type: 'message', - role: 'assistant', + type: "message", + role: "assistant", model, content: [], stop_reason: null, @@ -337,21 +319,24 @@ function handleAnthropicStream( heartbeatInterval = setInterval(async () => { try { - await streamWriter.write(': keep-alive\n\n'); + await streamWriter.write(": keep-alive\n\n"); } catch { clearInterval(heartbeatInterval); } }, 15000); - const reader = stream.getReader(); + reader = stream.getReader(); const decoder = new TextDecoder(); let streamEnded = false; - let rawBuffer = ''; - - const formattedTools = tools.map((t: any) => ({ - name: t.name, - description: t.description || '', - parameters: t.input_schema || {} + let rawBuffer = ""; + + const formattedTools: any[] = tools.map((t: any) => ({ + type: "function", + function: { + name: t.name, + description: t.description || "", + parameters: t.input_schema || {}, + }, })); const qwenParser = new QwenStreamParser(uiSessionId, { @@ -360,48 +345,48 @@ function handleAnthropicStream( if (!deltaText) return; if (!textBlockOpen) { textBlockOpen = true; - sendEvent('content_block_start', { - type: 'content_block_start', + sendEvent("content_block_start", { + type: "content_block_start", index: blockIndex, - content_block: { type: 'text', text: '' }, + content_block: { type: "text", text: "" }, }); } - sendEvent('content_block_delta', { - type: 'content_block_delta', + sendEvent("content_block_delta", { + type: "content_block_delta", index: blockIndex, - delta: { type: 'text_delta', text: deltaText }, + delta: { type: "text_delta", text: deltaText }, }); totalOutputTokens += Math.ceil(deltaText.length / 4); }, onToolCall: (tc) => { - stopReason = 'tool_use'; + stopReason = "tool_use"; if (textBlockOpen) { - sendEvent('content_block_stop', { type: 'content_block_stop', index: blockIndex }); + sendEvent("content_block_stop", { type: "content_block_stop", index: blockIndex }); textBlockOpen = false; blockIndex++; } - const toolId = tc.id || `toolu_${crypto.randomUUID().replace(/-/g, '').slice(0, 16)}`; - const argsStr = typeof tc.arguments === 'string' ? tc.arguments : JSON.stringify(tc.arguments || {}); + const toolId = tc.id || `toolu_${crypto.randomUUID().replace(/-/g, "").slice(0, 16)}`; + const argsStr = typeof tc.arguments === "string" ? tc.arguments : JSON.stringify(tc.arguments || {}); - sendEvent('content_block_start', { - type: 'content_block_start', + sendEvent("content_block_start", { + type: "content_block_start", index: blockIndex, content_block: { - type: 'tool_use', + type: "tool_use", id: toolId, name: tc.name, input: {}, }, }); - sendEvent('content_block_delta', { - type: 'content_block_delta', + sendEvent("content_block_delta", { + type: "content_block_delta", index: blockIndex, delta: { - type: 'input_json_delta', + type: "input_json_delta", partial_json: argsStr, }, }); - sendEvent('content_block_stop', { type: 'content_block_stop', index: blockIndex }); + sendEvent("content_block_stop", { type: "content_block_stop", index: blockIndex }); blockIndex++; }, }); @@ -411,41 +396,46 @@ function handleAnthropicStream( if (done) break; rawBuffer += decoder.decode(value, { stream: true }); - const lines = rawBuffer.split('\n'); - rawBuffer = lines.pop() || ''; + const lines = rawBuffer.split("\n"); + rawBuffer = lines.pop() || ""; for (const line of lines) { const trimmed = line.trim(); - if (trimmed === 'data: [DONE]') { + if (trimmed === "data: [DONE]") { streamEnded = true; break; } - if (trimmed.startsWith('data: ')) { + if (trimmed.startsWith("data: ")) { qwenParser.parseLine(trimmed.slice(6)); } } } - if (rawBuffer.trim() && rawBuffer.trim().startsWith('data: ') && rawBuffer.trim() !== 'data: [DONE]') { + if (rawBuffer.trim() && rawBuffer.trim().startsWith("data: ") && rawBuffer.trim() !== "data: [DONE]") { qwenParser.parseLine(rawBuffer.trim().slice(6)); } if (textBlockOpen) { - sendEvent('content_block_stop', { type: 'content_block_stop', index: blockIndex }); + sendEvent("content_block_stop", { type: "content_block_stop", index: blockIndex }); } - sendEvent('message_delta', { - type: 'message_delta', + const finalOutputTokens = qwenParser.usage?.completionTokens && qwenParser.usage.completionTokens > 0 + ? qwenParser.usage.completionTokens + : Math.max(1, totalOutputTokens); + + sendEvent("message_delta", { + type: "message_delta", delta: { stop_reason: stopReason, stop_sequence: null }, - usage: { output_tokens: Math.max(1, totalOutputTokens) }, + usage: { output_tokens: finalOutputTokens }, }); - sendEvent('message_stop', { type: 'message_stop' }); + sendEvent("message_stop", { type: "message_stop" }); } catch (err: any) { - console.error('[Anthropic Stream Error]:', err); + console.error("[Anthropic Stream Error]:", err); } finally { if (heartbeatInterval) clearInterval(heartbeatInterval); - onComplete?.(); + if (reader) reader.cancel().catch(() => {}); + onComplete?.(totalOutputTokens || 1); } }); } @@ -454,59 +444,64 @@ async function handleAnthropicNonStreaming( c: Context, stream: ReadableStream, model: string, + completionId: string, uiSessionId: string, inputTokens: number, hasTools: boolean, tools: any[], - onComplete?: () => void, + onComplete?: (outTokens: number) => void, ) { - const formattedTools = tools.map((t: any) => ({ - name: t.name, - description: t.description || '', - parameters: t.input_schema || {} + const formattedTools: any[] = tools.map((t: any) => ({ + type: "function", + function: { + name: t.name, + description: t.description || "", + parameters: t.input_schema || {}, + }, })); const result = await collectNonStreamingResult( c, stream, - `comp_${crypto.randomUUID().replace(/-/g, '')}`, + completionId, model, uiSessionId, hasTools, formattedTools, - onComplete, + () => {}, ); const contentBlocks: any[] = []; if (result.content) { - contentBlocks.push({ type: 'text', text: result.content }); + contentBlocks.push({ type: "text", text: result.content }); } - if (result.tool_calls && Array.isArray(result.tool_calls)) { - for (const tc of result.tool_calls) { + if (result.toolCalls && Array.isArray(result.toolCalls)) { + for (const tc of result.toolCalls) { let inputObj = {}; try { - inputObj = typeof tc.function?.arguments === 'string' ? JSON.parse(tc.function.arguments) : tc.function?.arguments || {}; + inputObj = typeof tc.function?.arguments === "string" ? JSON.parse(tc.function.arguments) : tc.function?.arguments || {}; } catch { inputObj = { raw: tc.function?.arguments }; } contentBlocks.push({ - type: 'tool_use', - id: tc.id || `toolu_${crypto.randomUUID().replace(/-/g, '').slice(0, 16)}`, + type: "tool_use", + id: tc.id || `toolu_${crypto.randomUUID().replace(/-/g, "").slice(0, 16)}`, name: tc.function?.name, input: inputObj, }); } } - const outTokens = result.usage?.completion_tokens || Math.ceil((result.content || '').length / 4); + const outTokens = Math.ceil((result.content || "").length / 4); + onComplete?.(Math.max(1, outTokens)); return c.json({ - id: `msg_${crypto.randomUUID().replace(/-/g, '').slice(0, 24)}`, - type: 'message', - role: 'assistant', + id: `msg_${crypto.randomUUID().replace(/-/g, "").slice(0, 24)}`, + type: "message", + role: "assistant", model, content: contentBlocks, - stop_reason: result.tool_calls && result.tool_calls.length > 0 ? 'tool_use' : 'end_turn', + stop_reason: result.toolCalls && result.toolCalls.length > 0 ? "tool_use" : "end_turn", stop_sequence: null, usage: { input_tokens: inputTokens, diff --git a/src/services/browser-manager.ts b/src/services/browser-manager.ts index b8db874b..5bfdf866 100644 --- a/src/services/browser-manager.ts +++ b/src/services/browser-manager.ts @@ -503,12 +503,13 @@ export async function resetBrowserProfile(cacheKey: string, accountId?: string): markAccountNotReady(profileId); const { getAccountCredentials } = await import("../core/accounts.js"); const hasCreds = accountId ? !!getAccountCredentials(getBaseAccountId(accountId))?.password : false; - if (accountId === "guest" || hasCreds) { + const isManualNamedAccount = Boolean(accountId && accountId !== "guest" && !hasCreds); + if (isManualNamedAccount) { + console.warn(`[Playwright] Preserving cookies/storage for manual login account: ${cacheKey}`); + } else { fs.rmSync(profilePath, { recursive: true, force: true }); fs.rmSync(storageStatePath(profileId), { force: true }); console.warn(`[Playwright] Cleared browser profile for ${cacheKey}: ${profilePath}`); - } else { - console.warn(`[Playwright] Preserving cookies/storage for manual login account: ${cacheKey}`); } } catch (err: any) { console.warn(`[Playwright] Failed to clear browser profile for ${cacheKey}: ${err.message}`); @@ -624,12 +625,14 @@ export async function initPlaywrightForAccount(account: QwenAccount, _headless = console.warn(`[Playwright] Failed to validate session for ${account.email}: ${err.message}`); } - if (await hasValidAuthCookie(acctPage)) { - await saveStorageState(acctContext, baseAccountId); - const { markAccountReady } = await import("../core/account-manager.js"); - markAccountReady(account.id); - markAccountReady(baseAccountId); - } + const finalUrl = acctPage.url(); + const sessionOk = !finalUrl.includes("auth") && !finalUrl.includes("login"); + if (sessionOk && (await hasValidAuthCookie(acctPage))) { + await saveStorageState(acctContext, baseAccountId); + const { markAccountReady } = await import("../core/account-manager.js"); + markAccountReady(account.id); + markAccountReady(baseAccountId); + } } export async function launchManualLoginAccount(accountId: string, browserType: BrowserType = 'chromium'): Promise<{ context: BrowserContext, page: Page }> { diff --git a/src/services/header-interceptor.ts b/src/services/header-interceptor.ts index 095ef98b..a29b4afe 100644 --- a/src/services/header-interceptor.ts +++ b/src/services/header-interceptor.ts @@ -182,7 +182,7 @@ export async function getGuestHeaders(): Promise> { await humanType(guestPage!, inputSelector, 'Hello'); await sleep(humanDelay(800, 1500)); - const selectors = ['.message-input-right-button-send .send-button', '.chat-prompt-send-button', 'button.send-button', 'button[type="submit"]', 'button:has(svg)']; + const selectors = ['.message-input-right-button-send .send-button', '.chat-prompt-send-button', 'button.send-button', 'button[type="submit"]', 'button[aria-label*="Send"]', 'button[aria-label*="Enviar"]', 'button[data-testid*="send"]']; let clicked = false; for (const selector of selectors) { const btn = await guestPage!.$(selector); From 7f3dcac1b7d191ee069dbb146a1458f22af2c3e9 Mon Sep 17 00:00:00 2001 From: lojanica Date: Mon, 31 Aug 2026 09:28:59 -0300 Subject: [PATCH 3/4] fix(review): address second-round review feedback on token estimation, abortStream, and session navigation --- src/api/server.ts | 23 +++++++++++++++++++---- src/routes/anthropic.ts | 23 +++++++++++++++-------- src/services/browser-manager.ts | 5 ++++- 3 files changed, 38 insertions(+), 13 deletions(-) diff --git a/src/api/server.ts b/src/api/server.ts index 97f41d03..ca2941f7 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -100,14 +100,29 @@ app.post('/v1/messages/count_tokens', bodyLimit({ } catch { return c.json({ type: 'error', error: { type: 'invalid_request_error', message: 'Invalid JSON body' } }, 400) } + if (!body || typeof body !== 'object') { + return c.json({ type: 'error', error: { type: 'invalid_request_error', message: 'Invalid JSON body' } }, 400) + } const promptParts: string[] = [] - if (typeof body.system === 'string') promptParts.push(body.system) + if (typeof body.system === 'string') { + promptParts.push(body.system) + } else if (Array.isArray(body.system)) { + for (const s of body.system) { + if (s && typeof s === 'object' && typeof s.text === 'string') { + promptParts.push(s.text) + } + } + } if (Array.isArray(body.messages)) { for (const m of body.messages) { - if (typeof m.content === 'string') promptParts.push(m.content) - else if (Array.isArray(m.content)) { + if (!m || typeof m !== 'object') continue + if (typeof m.content === 'string') { + promptParts.push(m.content) + } else if (Array.isArray(m.content)) { for (const b of m.content) { - if (b.text) promptParts.push(b.text) + if (b && typeof b === 'object' && typeof b.text === 'string') { + promptParts.push(b.text) + } } } } diff --git a/src/routes/anthropic.ts b/src/routes/anthropic.ts index 3b956b74..0f8e05c3 100644 --- a/src/routes/anthropic.ts +++ b/src/routes/anthropic.ts @@ -3,10 +3,10 @@ import { stream as honoStream } from "hono/streaming"; import crypto from "crypto"; import type { OpenAIRequest } from "../utils/types.js"; import { createQwenStream } from "../services/qwen.js"; -import { getNextAccount, getAccountById, markAccountRateLimited, releaseAccountInUse } from "../core/account-manager.js"; +import { getNextAccount, markAccountRateLimited, releaseAccountInUse } from "../core/account-manager.js"; import { loadAccounts } from "../core/accounts.js"; -import { registerStream, removeStream } from "../core/stream-registry.js"; -import { checkUserRateLimit, tryAcquireUserSlot, releaseUserSlot, getUserActiveStreams } from "../core/user-manager.js"; +import { registerStream, removeStream, abortStream } from "../core/stream-registry.js"; +import { checkUserRateLimit, tryAcquireUserSlot, releaseUserSlot } from "../core/user-manager.js"; import type { UserIdentity } from "../core/user-manager.js"; import { countTokens } from "../core/tokenizer.js"; import { QwenStreamParser } from "../utils/qwen-stream-parser.js"; @@ -34,7 +34,7 @@ export async function anthropicMessages(c: Context) { const user = (c as any).get?.("user") as UserIdentity | undefined; let userSlotHeld = false; let userSlotReleased = false; - let completionId = `comp_${crypto.randomUUID().replace(/-/g, "")}`; + const completionId = `comp_${crypto.randomUUID().replace(/-/g, "")}`; const releaseUserSlotOnce = () => { if (!userSlotHeld || userSlotReleased || !user) return; @@ -42,12 +42,17 @@ export async function anthropicMessages(c: Context) { userSlotReleased = true; }; + let body: any; try { - const body = await c.req.json(); + body = await c.req.json(); if (!body || typeof body !== "object") { return c.json({ type: "error", error: { type: "invalid_request_error", message: "Invalid JSON body" } }, 400); } + } catch { + return c.json({ type: "error", error: { type: "invalid_request_error", message: "Invalid JSON body" } }, 400); + } + try { const isStream = Boolean(body.stream); const rawModel = body.model || "qwen3.7-plus"; const targetModel = resolveModelName(rawModel); @@ -83,18 +88,20 @@ export async function anthropicMessages(c: Context) { if (typeof body.system === "string") { openAIMessages.push({ role: "system", content: body.system }); } else if (Array.isArray(body.system)) { - const sysText = body.system.map((s: any) => s.text || "").join("\n"); + const sysText = body.system.map((s: any) => s?.text || "").join("\n"); openAIMessages.push({ role: "system", content: sysText }); } } for (const msg of messages) { + if (!msg || typeof msg !== "object") continue; const role = msg.role === "assistant" ? "assistant" : "user"; if (typeof msg.content === "string") { openAIMessages.push({ role, content: msg.content }); } else if (Array.isArray(msg.content)) { let textParts = ""; for (const block of msg.content) { + if (!block || typeof block !== "object") continue; if (block.type === "text") { textParts += (block.text || "") + "\n"; } else if (block.type === "tool_result") { @@ -287,7 +294,7 @@ function handleAnthropicStream( return honoStream(c, async (streamWriter: any) => { streamWriter.onAbort?.(() => { - removeStream(completionId); + abortStream(completionId); }); let heartbeatInterval: any; @@ -492,7 +499,7 @@ async function handleAnthropicNonStreaming( } } - const outTokens = Math.ceil((result.content || "").length / 4); + const outTokens = result.body?.usage?.completion_tokens || Math.ceil((result.content || "").length / 4); onComplete?.(Math.max(1, outTokens)); return c.json({ diff --git a/src/services/browser-manager.ts b/src/services/browser-manager.ts index 5bfdf866..9050d7c6 100644 --- a/src/services/browser-manager.ts +++ b/src/services/browser-manager.ts @@ -607,14 +607,17 @@ export async function initPlaywrightForAccount(account: QwenAccount, _headless = await loginToQwenWithContext(acctContext, acctPage, account.email, account.password); } + let navigated = false; try { await acctPage.goto('https://chat.qwen.ai/c/new-chat', { waitUntil: 'domcontentloaded', timeout: config.timeouts.navigation }); + navigated = true; const url = acctPage.url(); if (url.includes('auth') || url.includes('login')) { if (account.email && account.password) { console.log(`[Playwright] Session expired for ${account.email}, re-logging in...`); await loginToQwenWithContext(acctContext, acctPage, account.email, account.password); await acctPage.goto('https://chat.qwen.ai/c/new-chat', { waitUntil: 'domcontentloaded', timeout: config.timeouts.navigation }); + navigated = true; } else { console.warn(`[Playwright] Session expired for account ${account.id} but no credentials available for re-login.`); } @@ -626,7 +629,7 @@ export async function initPlaywrightForAccount(account: QwenAccount, _headless = } const finalUrl = acctPage.url(); - const sessionOk = !finalUrl.includes("auth") && !finalUrl.includes("login"); + const sessionOk = navigated && !finalUrl.includes("auth") && !finalUrl.includes("login") && finalUrl.startsWith("http"); if (sessionOk && (await hasValidAuthCookie(acctPage))) { await saveStorageState(acctContext, baseAccountId); const { markAccountReady } = await import("../core/account-manager.js"); From c43ff71de02a668951c3e3ba535874ca8856d0e7 Mon Sep 17 00:00:00 2001 From: lojanica Date: Mon, 31 Aug 2026 09:38:42 -0300 Subject: [PATCH 4/4] fix(review): reject array request bodies in Anthropic endpoints and redact account email in Playwright logs --- src/api/server.ts | 2 +- src/routes/anthropic.ts | 2 +- src/services/browser-manager.ts | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/api/server.ts b/src/api/server.ts index ca2941f7..77903397 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -100,7 +100,7 @@ app.post('/v1/messages/count_tokens', bodyLimit({ } catch { return c.json({ type: 'error', error: { type: 'invalid_request_error', message: 'Invalid JSON body' } }, 400) } - if (!body || typeof body !== 'object') { + if (!body || typeof body !== 'object' || Array.isArray(body)) { return c.json({ type: 'error', error: { type: 'invalid_request_error', message: 'Invalid JSON body' } }, 400) } const promptParts: string[] = [] diff --git a/src/routes/anthropic.ts b/src/routes/anthropic.ts index 0f8e05c3..6ec4724e 100644 --- a/src/routes/anthropic.ts +++ b/src/routes/anthropic.ts @@ -45,7 +45,7 @@ export async function anthropicMessages(c: Context) { let body: any; try { body = await c.req.json(); - if (!body || typeof body !== "object") { + if (!body || typeof body !== "object" || Array.isArray(body)) { return c.json({ type: "error", error: { type: "invalid_request_error", message: "Invalid JSON body" } }, 400); } } catch { diff --git a/src/services/browser-manager.ts b/src/services/browser-manager.ts index 9050d7c6..76af818f 100644 --- a/src/services/browser-manager.ts +++ b/src/services/browser-manager.ts @@ -614,7 +614,7 @@ export async function initPlaywrightForAccount(account: QwenAccount, _headless = const url = acctPage.url(); if (url.includes('auth') || url.includes('login')) { if (account.email && account.password) { - console.log(`[Playwright] Session expired for ${account.email}, re-logging in...`); + console.log(`[Playwright] Session expired for account ${account.id}, re-logging in...`); await loginToQwenWithContext(acctContext, acctPage, account.email, account.password); await acctPage.goto('https://chat.qwen.ai/c/new-chat', { waitUntil: 'domcontentloaded', timeout: config.timeouts.navigation }); navigated = true; @@ -622,10 +622,10 @@ export async function initPlaywrightForAccount(account: QwenAccount, _headless = console.warn(`[Playwright] Session expired for account ${account.id} but no credentials available for re-login.`); } } else { - console.log(`[Playwright] Session validated for ${account.email}.`); + console.log(`[Playwright] Session validated for account ${account.id}.`); } } catch (err: any) { - console.warn(`[Playwright] Failed to validate session for ${account.email}: ${err.message}`); + console.warn(`[Playwright] Failed to validate session for account ${account.id}: ${err.message}`); } const finalUrl = acctPage.url();