From 5b130aca66c1576f34abd137afe376fd5dee37b8 Mon Sep 17 00:00:00 2001 From: Gemini Nexus Bot Date: Wed, 22 Apr 2026 03:26:51 -0300 Subject: [PATCH] fix(stream): restore processStreamResponse to original working behavior The initial role chunk (role:'assistant', content:'') added in a3eabc2 was causing the OpenCode TUI to receive an empty first delta and discard all subsequent content, showing only the completion time with no text output. Restored to the exact streaming behavior from e524a340 that worked: - Remove the pre-emptive initial chunk entirely - Move hasEmittedChunk=true back inside pushChunk (original position) - Use reasoning_content delta field for thought parts instead of wrapping in tags (which corrupts plain text responses) - Remove streamOptions/include_usage/lastUsageMetadata tracking (not present in the working version) - Remove isThinking state machine (was adding noise to output) Kept from the spec compliance work (a3eabc2): - toolCallIndex counter for correct finish_reason:'tool_calls' mapping - Comment explaining why we override Gemini's STOP finish reason --- src/server/modules/proxy/proxy.service.ts | 99 +++-------------------- 1 file changed, 13 insertions(+), 86 deletions(-) diff --git a/src/server/modules/proxy/proxy.service.ts b/src/server/modules/proxy/proxy.service.ts index 1db980d..fce7eca 100644 --- a/src/server/modules/proxy/proxy.service.ts +++ b/src/server/modules/proxy/proxy.service.ts @@ -751,7 +751,7 @@ export class ProxyService { token.token.upstream_proxy_url, extraHeaders, ); - return this.processStreamResponse(stream, request.model, request.stream_options); + return this.processStreamResponse(stream, request.model); } catch (streamError) { this.logger.warn( `Stream path failed for model=${request.model}; falling back to non-stream generation: ${ @@ -810,7 +810,7 @@ export class ProxyService { token.token.upstream_proxy_url, extraHeaders, ); - return this.processStreamResponse(stream, request.model, request.stream_options); + return this.processStreamResponse(stream, request.model); } const response = await this.generateInternalWithStreamFallback( @@ -957,38 +957,19 @@ export class ProxyService { private processStreamResponse( upstreamStream: NodeJS.ReadableStream, model: string, - streamOptions?: { include_usage?: boolean }, ): Observable { return new Observable((subscriber) => { const decoder = new TextDecoder(); let buffer = ''; let hasEmittedChunk = false; let hasSentDone = false; - let isThinking = false; let toolCallIndex = 0; - let lastUsageMetadata: GeminiResponse['usageMetadata']; const streamId = `chatcmpl-${uuidv4()}`; const created = Math.floor(Date.now() / 1000); - // Emit initial role chunk immediately to prevent strict clients (like Hermes) from timing out during long TTFT - const initialChunk = { - id: streamId, - object: 'chat.completion.chunk', - created: created, - model: model, - choices: [ - { - index: 0, - delta: { role: 'assistant', content: '' }, - finish_reason: null, - }, - ], - }; - subscriber.next(`data: ${JSON.stringify(initialChunk)}\n\n`); - hasEmittedChunk = true; - const pushChunk = (payload: Record): void => { + hasEmittedChunk = true; subscriber.next(`data: ${JSON.stringify(payload)}\n\n`); }; @@ -1009,17 +990,10 @@ export class ProxyService { const candidate = json.candidates?.[0]; const parts = candidate?.content?.parts || []; - if (json.usageMetadata) { - lastUsageMetadata = json.usageMetadata; - } - for (const part of parts) { if (part.thought && part.text) { - let textToEmit = part.text; - if (!isThinking) { - isThinking = true; - textToEmit = `\n${textToEmit}`; - } + // Reasoning/thinking parts: emit via reasoning_content delta field + // so clients that support it can display thinking separately. const reasoningChunk = { id: streamId, object: 'chat.completion.chunk', @@ -1028,7 +1002,7 @@ export class ProxyService { choices: [ { index: 0, - delta: { content: textToEmit }, + delta: { reasoning_content: part.text }, finish_reason: null, }, ], @@ -1090,11 +1064,6 @@ export class ProxyService { } if (part.text) { - let textToEmit = part.text; - if (isThinking && !part.thought) { - isThinking = false; - textToEmit = `\n\n\n${textToEmit}`; - } const contentChunk = { id: streamId, object: 'chat.completion.chunk', @@ -1103,7 +1072,7 @@ export class ProxyService { choices: [ { index: 0, - delta: { content: textToEmit }, + delta: { content: part.text }, finish_reason: null, }, ], @@ -1113,26 +1082,9 @@ export class ProxyService { } if (candidate?.finishReason) { - if (isThinking) { - isThinking = false; - const closingChunk = { - id: streamId, - object: 'chat.completion.chunk', - created: created, - model: model, - choices: [ - { - index: 0, - delta: { content: '\n\n\n' }, - finish_reason: null, - }, - ], - }; - pushChunk(closingChunk); - } - // Per the OpenAI spec, finish_reason must be "tool_calls" when the - // response contains function calls, even though Gemini always sends "STOP". + // response contains function calls. Gemini always sends "STOP" even + // for tool call responses, so we infer from toolCallIndex. const mappedFinishReason = toolCallIndex > 0 ? 'tool_calls' @@ -1157,7 +1109,7 @@ export class ProxyService { subscriber.complete(); } } catch { - // ignore parse errors + // ignore parse errors on individual SSE lines } } }); @@ -1178,26 +1130,6 @@ export class ProxyService { ], }); } - if (streamOptions?.include_usage && lastUsageMetadata) { - const usageChunk = { - id: streamId, - object: 'chat.completion.chunk', - created, - model, - choices: [], - usage: { - prompt_tokens: lastUsageMetadata.promptTokenCount || 0, - completion_tokens: - (lastUsageMetadata.candidatesTokenCount || 0) + - (lastUsageMetadata.thoughtsTokenCount || 0), - total_tokens: - (lastUsageMetadata.promptTokenCount || 0) + - (lastUsageMetadata.candidatesTokenCount || 0) + - (lastUsageMetadata.thoughtsTokenCount || 0), - }, - }; - pushChunk(usageChunk); - } if (!hasSentDone) { subscriber.next('data: [DONE]\n\n'); hasSentDone = true; @@ -1205,17 +1137,12 @@ export class ProxyService { subscriber.complete(); }); - upstreamStream.on('error', (err) => { - this.logger.error('OpenAI-compatible stream error:', err); + upstreamStream.on('error', (err: unknown) => { + // Convert to clean Error to avoid circular reference issues (socket objects) const cleanError = err instanceof Error ? new Error(err.message) : new Error(String(err)); + this.logger.error(`OpenAI-compatible stream error: ${cleanError.message}`); subscriber.error(cleanError); }); - - return () => { - if (typeof (upstreamStream as any).destroy === 'function') { - (upstreamStream as any).destroy(); - } - }; }); }