diff --git a/src/screens/REPL.tsx b/src/screens/REPL.tsx index 2000c762e4..4975edc8d0 100644 --- a/src/screens/REPL.tsx +++ b/src/screens/REPL.tsx @@ -1540,6 +1540,11 @@ export function REPL({ '🎯 Verboo Intelligence is locking onto the target', '🎬 Verboo Intelligence is setting the scene', '🎪 Verboo Intelligence is rolling out the red carpet', + '🔥 Getting ready to VerbooCodar!!', + '🚀 Warming up to VerbooCodar!', + '⚡ Firing up the engines to VerbooCodar!', + '🧠 Almost ready to VerbooCodar!', + '✨ Loading up to VerbooCodar!', ]; let cancelled = false; void import('../services/api/openaiShim.js').then(mod => { diff --git a/src/services/api/openaiShim.ts b/src/services/api/openaiShim.ts index 9417f26eb2..af85371aec 100644 --- a/src/services/api/openaiShim.ts +++ b/src/services/api/openaiShim.ts @@ -1046,17 +1046,115 @@ function repairPossiblyTruncatedObjectJson(raw: string): string | null { // e limpo no fim. Module-level (não per-instance) porque o cliente é criado // uma vez por sessão mas setSDKStatus muda a cada query. let routerStatusHandler: ((status: 'warming-up' | null) => void) | null = null +let routerStatusHandlerGeneration = 0 +const activeWarmingHints = new Set() export function setOpenAIShimRouterStatusHandler( fn: ((status: 'warming-up' | null) => void) | null, ): void { + if (routerStatusHandler && activeWarmingHints.size > 0) { + routerStatusHandler(null) + } + activeWarmingHints.clear() + routerStatusHandlerGeneration++ routerStatusHandler = fn } +// ----------------------------------------------------------------------------- +// Aviso "preparando o modelo" (warming hint) por TEMPO. +// Cobre a latência de cauda (cold start / preempção do backend) que NÃO vem +// acompanhada do sinal router_status:"warming". O cronômetro começa no ENVIO +// da requisição (em `create`), então pega tanto o caso do corpo lento quanto o +// da requisição inteira pendurada antes de qualquer resposta. É só comunicação +// visual — nunca aborta nem interfere no resultado. Override: VERBOO_SLOW_HINT_MS. + +function getSlowHintMs(): number { + const raw = process.env.VERBOO_SLOW_HINT_MS + const v = raw ? parseInt(raw, 10) : NaN + return Number.isFinite(v) && v > 0 ? v : 12_000 +} + +type WarmingHintController = { + schedule: (delayMs: number) => void + showNow: () => void + resolve: () => void +} + +/** + * One controller per request. The module-level set only aggregates visible + * hints so one completed request cannot clear another concurrent slow request. + */ +function createWarmingHintController( + signal?: AbortSignal, +): WarmingHintController { + const token = Symbol('warming-hint') + const handlerGeneration = routerStatusHandlerGeneration + let timer: ReturnType | null = null + let shown = false + let resolved = false + + const clearTimer = () => { + if (timer !== null) { + clearTimeout(timer) + timer = null + } + } + + const showNow = () => { + clearTimer() + if ( + resolved || + shown || + handlerGeneration !== routerStatusHandlerGeneration || + !routerStatusHandler + ) { + return + } + shown = true + const wasEmpty = activeWarmingHints.size === 0 + activeWarmingHints.add(token) + if (wasEmpty) routerStatusHandler('warming-up') + } + + const resolve = () => { + if (resolved) return + resolved = true + clearTimer() + signal?.removeEventListener('abort', resolve) + if (!shown) return + + activeWarmingHints.delete(token) + if ( + activeWarmingHints.size === 0 && + handlerGeneration === routerStatusHandlerGeneration + ) { + routerStatusHandler?.(null) + } + } + + const schedule = (delayMs: number) => { + if (resolved) return + clearTimer() + timer = setTimeout(() => { + timer = null + showNow() + }, delayMs) + } + + if (signal?.aborted) { + resolved = true + } else { + signal?.addEventListener('abort', resolve, { once: true }) + } + + return { schedule, showNow, resolve } +} + async function* openaiStreamToAnthropic( response: Response, model: string, signal?: AbortSignal, + warmingHint?: WarmingHintController, ): AsyncGenerator { const messageId = makeMessageId() let contentBlockIndex = 0 @@ -1078,7 +1176,6 @@ async function* openaiStreamToAnthropic( let lastStopReason: 'tool_use' | 'max_tokens' | 'end_turn' | null = null let hasEmittedFinalUsage = false let hasProcessedFinishReason = false - let routerWarmingActive = false const streamState = createStreamState() // Emit message_start @@ -1117,6 +1214,7 @@ async function* openaiStreamToAnthropic( let lastDataTime = Date.now() const streamStartedAt = Date.now() + /** * Read from the stream with an idle timeout. If no data arrives within * STREAM_IDLE_TIMEOUT_MS, assume the connection is dead and throw so @@ -1249,9 +1347,10 @@ async function* openaiStreamToAnthropic( const routerStatus = (chunk as unknown as { router_status?: string }) .router_status if (typeof routerStatus === 'string') { - if (routerStatus === 'warming' && !routerWarmingActive) { - routerWarmingActive = true - routerStatusHandler?.('warming-up') + if (routerStatus === 'warming') { + // Servidor sinalizou cold start: mostra o aviso já (não espera o + // cronômetro por tempo). + warmingHint?.showNow() } continue } @@ -1304,16 +1403,6 @@ async function* openaiStreamToAnthropic( const chunkUsage = convertChunkUsage(chunk.usage) - // Chunk com content real chegou — limpa warming-up se estava ativo. - if ( - routerWarmingActive && - Array.isArray(chunk.choices) && - chunk.choices.length > 0 - ) { - routerWarmingActive = false - routerStatusHandler?.(null) - } - for (const choice of chunk.choices ?? []) { const delta = choice.delta @@ -1706,13 +1795,64 @@ class OpenAIShimStream { private generator: AsyncGenerator // The controller property is checked by claude.ts to distinguish streams from error messages controller = new AbortController() + private warmingHint?: WarmingHintController + private unconsumedCleanupTimer: ReturnType | null - constructor(generator: AsyncGenerator) { + constructor( + generator: AsyncGenerator, + warmingHint?: WarmingHintController, + ) { this.generator = generator + this.warmingHint = warmingHint + // The normal caller starts iterating immediately after awaiting + // withResponse(). If it abandons the returned stream, dispose the hint on + // the next event-loop turn so no status can leak into a later query. + this.unconsumedCleanupTimer = warmingHint + ? setTimeout(() => { + this.unconsumedCleanupTimer = null + warmingHint.resolve() + }, 0) + : null + } + + private markConsumed(): void { + if (this.unconsumedCleanupTimer !== null) { + clearTimeout(this.unconsumedCleanupTimer) + this.unconsumedCleanupTimer = null + } + } + + private hasMeaningfulOutput(event: AnthropicStreamEvent): boolean { + if ( + event.type === 'content_block_start' && + event.content_block?.type === 'tool_use' + ) { + return true + } + if (event.type !== 'content_block_delta') return false + + const delta = event.delta + if (!delta) return false + return ( + (typeof delta.text === 'string' && delta.text.length > 0) || + (typeof delta.thinking === 'string' && delta.thinking.length > 0) || + (typeof delta.partial_json === 'string' && + delta.partial_json.length > 0) + ) } async *[Symbol.asyncIterator]() { - yield* this.generator + this.markConsumed() + try { + for await (const event of this.generator) { + if (this.hasMeaningfulOutput(event)) { + this.warmingHint?.resolve() + } + yield event + } + } finally { + this.warmingHint?.resolve() + } } } @@ -1845,7 +1985,20 @@ class OpenAIShimMessages { reasoningEffortOverride: self.reasoningEffort, suppressReasoningEffort: self.suppressReasoningEffort, }) - const response = await self._doRequest(request, params, options) + // Cronômetro do aviso "preparando" começa AQUI (no envio da requisição), + // então cobre tanto a resposta lenta quanto a requisição pendurada antes + // de qualquer resposta. É resolvido: no gerador de streaming (quando o + // conteúdo chega ou o stream termina), logo abaixo nos caminhos + // não-stream, ou aqui no catch se a própria requisição falhar. + const warmingHint = createWarmingHintController(options?.signal) + warmingHint.schedule(getSlowHintMs()) + let response: Response + try { + response = await self._doRequest(request, params, options) + } catch (e) { + warmingHint.resolve() + throw e + } httpResponse = response if (params.stream) { @@ -1863,10 +2016,16 @@ class OpenAIShimMessages { response, request.resolvedModel, options?.signal, + warmingHint, ), + warmingHint, ) } + // Caminhos não-stream: a resposta HTTP já chegou, então o aviso cumpriu + // seu papel — limpa antes de coletar/converter o corpo. + warmingHint.resolve() + if (request.transport === 'codex_responses') { const data = await collectCodexCompletedResponse( response, diff --git a/src/services/api/openaiShim.warmingHint.test.ts b/src/services/api/openaiShim.warmingHint.test.ts new file mode 100644 index 0000000000..1d31189ca3 --- /dev/null +++ b/src/services/api/openaiShim.warmingHint.test.ts @@ -0,0 +1,267 @@ +import { afterEach, beforeEach, expect, test } from 'bun:test' +import { + createOpenAIShimClient, + setOpenAIShimRouterStatusHandler, +} from './openaiShim.ts' + +type FetchType = typeof globalThis.fetch + +const originalFetch = globalThis.fetch +const originalEnv = { + OPENAI_BASE_URL: process.env.OPENAI_BASE_URL, + OPENAI_API_KEY: process.env.OPENAI_API_KEY, + OPENAI_API_FORMAT: process.env.OPENAI_API_FORMAT, + VERBOO_SLOW_HINT_MS: process.env.VERBOO_SLOW_HINT_MS, +} + +beforeEach(() => { + process.env.OPENAI_BASE_URL = 'http://example.test/v1' + process.env.OPENAI_API_KEY = 'test-key' +}) + +afterEach(() => { + globalThis.fetch = originalFetch + setOpenAIShimRouterStatusHandler(null) + for (const [k, v] of Object.entries(originalEnv)) { + if (v === undefined) delete process.env[k] + else process.env[k] = v + } +}) + +const encoder = new TextEncoder() + +// SSE response que ATRASA a emissão do primeiro chunk por `delayMs`, simulando +// um backend lento (cold start / preempção) que fica em silêncio antes de +// começar a responder — sem emitir o sinal router_status:"warming". +function makeDelayedSseResponse(lines: string[], delayMs: number): Response { + let i = 0 + return new Response( + new ReadableStream({ + async pull(controller) { + if (i === 0 && delayMs > 0) { + await new Promise((r) => setTimeout(r, delayMs)) + } + if (i < lines.length) { + controller.enqueue(encoder.encode(lines[i]!)) + i++ + } else { + controller.close() + } + }, + }), + { headers: { 'Content-Type': 'text/event-stream' } }, + ) +} + +function makeSequencedSseResponse( + entries: Array<{ line: string; delayMs?: number }>, +): Response { + let i = 0 + return new Response( + new ReadableStream({ + async pull(controller) { + const entry = entries[i] + if (!entry) { + controller.close() + return + } + i++ + if (entry.delayMs) { + await new Promise(resolve => setTimeout(resolve, entry.delayMs)) + } + controller.enqueue(encoder.encode(entry.line)) + }, + }), + { headers: { 'Content-Type': 'text/event-stream' } }, + ) +} + +function contentChunks(): string[] { + return [ + `data: ${JSON.stringify({ + id: 'c1', + object: 'chat.completion.chunk', + model: 'fake-model', + choices: [{ index: 0, delta: { role: 'assistant', content: 'oi' }, finish_reason: null }], + })}\n\n`, + `data: ${JSON.stringify({ + id: 'c1', + object: 'chat.completion.chunk', + model: 'fake-model', + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + })}\n\n`, + 'data: [DONE]\n\n', + ] +} + +async function drain(model = 'fake-model'): Promise { + const client = createOpenAIShimClient({}) as { + beta: { + messages: { + create: (p: Record) => { + withResponse: () => Promise<{ data: AsyncIterable }> + } + } + } + } + const result = await client.beta.messages + .create({ + model, + system: 'test', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 32, + stream: true, + }) + .withResponse() + for await (const _ of result.data) { + // consome o stream até o fim + } +} + +test('mostra warming-up por TEMPO quando o backend fica lento sem sinal do servidor', async () => { + // limiar curto pra teste; primeiro chunk atrasa mais que o limiar + process.env.VERBOO_SLOW_HINT_MS = '40' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler((s) => statusCalls.push(s)) + + globalThis.fetch = (async () => + makeDelayedSseResponse(contentChunks(), 150)) as unknown as FetchType + + await drain() + + // disparou o aviso 'warming-up' e depois limpou (null) ao chegar conteúdo + expect(statusCalls).toContain('warming-up') + expect(statusCalls[statusCalls.length - 1]).toBe(null) + const firstWarm = statusCalls.indexOf('warming-up') + const firstNull = statusCalls.indexOf(null) + expect(firstWarm).toBeGreaterThanOrEqual(0) + expect(firstNull).toBeGreaterThan(firstWarm) +}) + +test('mostra warming-up quando a REQUISIÇÃO fica pendurada antes de qualquer resposta (caso B)', async () => { + // Aqui o atraso está no próprio fetch (a resposta HTTP nem começa) — o + // cronômetro precisa contar desde o ENVIO, não só da leitura do corpo. + process.env.VERBOO_SLOW_HINT_MS = '40' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler((s) => statusCalls.push(s)) + + globalThis.fetch = (async () => { + await new Promise((r) => setTimeout(r, 150)) // requisição pendurada + return makeDelayedSseResponse(contentChunks(), 0) + }) as unknown as FetchType + + await drain() + + expect(statusCalls).toContain('warming-up') + expect(statusCalls[statusCalls.length - 1]).toBe(null) +}) + +test('NÃO mostra warming-up quando a resposta chega rápido (sem flicker)', async () => { + // limiar alto; chunk chega imediatamente + process.env.VERBOO_SLOW_HINT_MS = '5000' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler((s) => statusCalls.push(s)) + + globalThis.fetch = (async () => + makeDelayedSseResponse(contentChunks(), 0)) as unknown as FetchType + + await drain() + + expect(statusCalls).not.toContain('warming-up') +}) + +test('chunk inicial vazio não conta como conteúdo visível', async () => { + process.env.VERBOO_SLOW_HINT_MS = '40' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler(status => statusCalls.push(status)) + + const metadataChunk = `data: ${JSON.stringify({ + id: 'c1', + object: 'chat.completion.chunk', + model: 'fake-model', + choices: [ + { + index: 0, + delta: { role: 'assistant', content: '' }, + finish_reason: null, + }, + ], + })}\n\n` + globalThis.fetch = (async () => + makeSequencedSseResponse([ + { line: metadataChunk }, + { line: contentChunks()[0]!, delayMs: 150 }, + { line: contentChunks()[1]! }, + { line: contentChunks()[2]! }, + ])) as unknown as FetchType + + await drain() + + expect(statusCalls).toEqual(['warming-up', null]) +}) + +test('requisição rápida paralela não cancela o aviso da requisição lenta', async () => { + process.env.VERBOO_SLOW_HINT_MS = '40' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler(status => statusCalls.push(status)) + let fetchCount = 0 + + globalThis.fetch = (async () => { + fetchCount++ + return makeDelayedSseResponse( + contentChunks(), + fetchCount === 1 ? 150 : 0, + ) + }) as unknown as FetchType + + const slowRequest = drain('slow-model') + await new Promise(resolve => setTimeout(resolve, 10)) + const fastRequest = drain('fast-model') + await Promise.all([slowRequest, fastRequest]) + + expect(fetchCount).toBe(2) + expect(statusCalls).toEqual(['warming-up', null]) +}) + +test('stream Responses rápido finaliza o timer sem aviso tardio', async () => { + process.env.OPENAI_API_FORMAT = 'responses' + process.env.VERBOO_SLOW_HINT_MS = '40' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler(status => statusCalls.push(status)) + + globalThis.fetch = (async () => + makeDelayedSseResponse( + [ + `event: response.output_text.delta\ndata: ${JSON.stringify({ delta: 'oi' })}\n\n`, + `event: response.completed\ndata: ${JSON.stringify({ + response: { + id: 'resp_1', + model: 'fake-model', + status: 'completed', + output: [], + usage: { input_tokens: 1, output_tokens: 1 }, + }, + })}\n\n`, + ], + 0, + )) as unknown as FetchType + + await drain() + await new Promise(resolve => setTimeout(resolve, 80)) + + expect(statusCalls).not.toContain('warming-up') +}) + +test('falha antes da resposta cancela o timer', async () => { + process.env.VERBOO_SLOW_HINT_MS = '20' + const statusCalls: (string | null)[] = [] + setOpenAIShimRouterStatusHandler(status => statusCalls.push(status)) + globalThis.fetch = (async () => { + throw new Error('network unavailable') + }) as unknown as FetchType + + await expect(drain()).rejects.toThrow() + await new Promise(resolve => setTimeout(resolve, 40)) + + expect(statusCalls).not.toContain('warming-up') +})