fix(streaming): resolve Gemini v1internal SSE wrapped format and OpenCode compatibility

- Fix processStreamResponse to unwrap json.response ?? json for Gemini v1internal wrapped SSE chunks
- Add initial role:assistant delta chunk for Vercel AI SDK / OpenCode compatibility
- Add trafficLogger console transport for real-time debugging
- Add upstream request/response logging throughout proxy paths
- Update proxy-controller.integration and proxy-retry-mock tests for new signatures
- Add comprehensive proxy-stream-debug unit tests (8 cases including wrapped format regression)
- Add OpenCode integration documentation

src/server/modules/proxy/proxy.service.ts

@@ -33,6 +33,7 @@ import {
 import { getMaxOutputTokens, getThinkingBudget } from '../../../lib/geminiNexus/ModelSpecs';
 import { resolveRequestUserAgent } from './request-user-agent';
 import { UpstreamRequestError } from './clients/upstream-error';
+import { trafficLogger } from '../../../utils/traffic-logger';
 
 @Injectable()
 export class ProxyService {
@@ -691,7 +692,7 @@ export class ProxyService {
 
   async handleChatCompletions(
     request: OpenAIChatRequest,
-    _requestId?: string,
+    requestId?: string,
     headerSessionKey?: string,
   ): Promise<OpenAIChatResponse | Observable<string>> {
     // Prefer the x-session-affinity header value (sent by OpenCode) over body-level session ids.
@@ -745,13 +746,14 @@ export class ProxyService {
         // Use v1internal API (same as Anthropic handler)
         if (request.stream) {
           try {
+            trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', geminiBody, extraHeaders, { model: effectiveTargetModel });
             const stream = await this.geminiClient.streamGenerateInternal(
               geminiBody,
               token.token.access_token,
               token.token.upstream_proxy_url,
               extraHeaders,
             );
-            return this.processStreamResponse(stream, request.model);
+            return this.processStreamResponse(stream, request.model, requestId);
           } catch (streamError) {
             this.logger.warn(
               `Stream path failed for model=${request.model}; falling back to non-stream generation: ${
@@ -759,12 +761,14 @@ export class ProxyService {
               }`,
             );
 
+            trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', geminiBody, extraHeaders, { model: effectiveTargetModel });
             const response = await this.generateInternalWithStreamFallback(
               geminiBody,
               token.token.access_token,
               token.token.upstream_proxy_url,
               extraHeaders,
             );
+            trafficLogger.logUpstreamResponse(requestId || 'unknown', '/v1/chat/completions', 200, response, 0);
             this.logger.log(
               `Upstream response snippet after stream fallback: ${JSON.stringify(response).substring(0, 500)}`,
             );
@@ -804,21 +808,24 @@ export class ProxyService {
             fallbackBody.model = effectiveTargetModel;
             this.applyInternalGenerationConstraints(fallbackBody, effectiveTargetModel, token.id);
             if (request.stream) {
+              trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', fallbackBody, extraHeaders, { model: effectiveTargetModel });
               const stream = await this.geminiClient.streamGenerateInternal(
                 fallbackBody,
                 token.token.access_token,
                 token.token.upstream_proxy_url,
                 extraHeaders,
               );
-              return this.processStreamResponse(stream, request.model);
+              return this.processStreamResponse(stream, request.model, requestId);
             }
 
+            trafficLogger.logUpstream(requestId || 'unknown', '/v1/chat/completions', fallbackBody, extraHeaders, { model: effectiveTargetModel });
             const response = await this.generateInternalWithStreamFallback(
               fallbackBody,
               token.token.access_token,
               token.token.upstream_proxy_url,
               extraHeaders,
             );
+            trafficLogger.logUpstreamResponse(requestId || 'unknown', '/v1/chat/completions', 200, response, 0);
             const claudeResponse = transformResponse(response);
             return this.convertClaudeToOpenAIResponse(claudeResponse, request.model);
           } catch (fallbackErr) {
@@ -957,17 +964,38 @@ export class ProxyService {
   private processStreamResponse(
     upstreamStream: NodeJS.ReadableStream,
     model: string,
+    requestId?: string,
   ): Observable<string> {
     return new Observable<string>((subscriber) => {
       const decoder = new TextDecoder();
       let buffer = '';
       let hasEmittedChunk = false;
+      let hasSentRoleChunk = false;
       let hasSentDone = false;
       let toolCallIndex = 0;
 
       const streamId = `chatcmpl-${uuidv4()}`;
       const created = Math.floor(Date.now() / 1000);
 
+      const pushRoleChunk = (): void => {
+        if (hasSentRoleChunk) return;
+        hasSentRoleChunk = true;
+        const roleChunk = {
+          id: streamId,
+          object: 'chat.completion.chunk',
+          created,
+          model,
+          choices: [
+            {
+              index: 0,
+              delta: { role: 'assistant', content: '' },
+              finish_reason: null,
+            },
+          ],
+        };
+        subscriber.next(`data: ${JSON.stringify(roleChunk)}\n\n`);
+      };
+
       const pushChunk = (payload: Record<string, unknown>): void => {
         hasEmittedChunk = true;
         subscriber.next(`data: ${JSON.stringify(payload)}\n\n`);
@@ -987,7 +1015,10 @@ export class ProxyService {
 
           try {
             const json = JSON.parse(dataStr);
-            const candidate = json.candidates?.[0];
+            trafficLogger.logUpstreamResponse(requestId || 'unknown', '/v1/chat/completions', 200, json, 0, { rawLine: trimmed });
+            // v1internal API wraps response in {"response": {"candidates": [...]}}
+            const responsePayload = json.response ?? json;
+            const candidate = responsePayload.candidates?.[0];
             const parts = candidate?.content?.parts || [];
 
             for (const part of parts) {
@@ -1007,11 +1038,13 @@ export class ProxyService {
                     },
                   ],
                 };
+                pushRoleChunk();
                 pushChunk(reasoningChunk);
                 continue;
               }
 
               if (part.functionCall) {
+                pushRoleChunk();
                 const currentToolIndex = toolCallIndex++;
                 const toolCallChunk = {
                   id: streamId,
@@ -1059,11 +1092,13 @@ export class ProxyService {
                     },
                   ],
                 };
+                pushRoleChunk();
                 pushChunk(imageChunk);
                 continue;
               }
 
               if (part.text) {
+                pushRoleChunk();
                 const contentChunk = {
                   id: streamId,
                   object: 'chat.completion.chunk',

src/tests/unit/proxy-controller.integration.test.ts

@@ -11,6 +11,10 @@ function createReplyMock() {
   return reply;
 }
 
+function createReqMock() {
+  return { headers: {} } as any;
+}
+
 describe('ProxyController Integration', () => {
   it('routes Claude OpenAI requests to protocol parity path', async () => {
     const proxyService = {
@@ -19,6 +23,7 @@ describe('ProxyController Integration', () => {
     };
     const controller = new ProxyController(proxyService as any);
     const reply = createReplyMock();
+    const req = createReqMock();
 
     await controller.chatCompletions(
       {
@@ -26,6 +31,7 @@ describe('ProxyController Integration', () => {
         stream: false,
         messages: [{ role: 'user', content: 'hello' }],
       } as any,
+      req,
       reply as any,
     );
 
@@ -42,6 +48,7 @@ describe('ProxyController Integration', () => {
     };
     const controller = new ProxyController(proxyService as any);
     const reply = createReplyMock();
+    const req = createReqMock();
 
     await controller.chatCompletions(
       {
@@ -49,6 +56,7 @@ describe('ProxyController Integration', () => {
         stream: true,
         messages: [{ role: 'user', content: 'hello' }],
       } as any,
+      req,
       reply as any,
     );
 

src/tests/unit/proxy-retry-mock.test.ts

@@ -648,6 +648,7 @@ describe('ProxyService Protocol Parity Fixtures', () => {
     });
 
     const output = chunks.join('');
+    expect(output).toContain('"role":"assistant"');
     expect(output).toContain('"reasoning_content":"reasoning text"');
     expect(output).toContain('"tool_calls"');
     expect(output).toContain('"content":"final answer"');
@@ -689,6 +690,7 @@ describe('ProxyService Protocol Parity Fixtures', () => {
 
     expect(streamResult.completed).not.toBe(true);
     expect(streamResult.error?.message).toContain('socket hang up');
+    expect(chunks.join('')).toContain('"role":"assistant"');
     expect(chunks.join('')).toContain('"content":"partial output"');
     expect(chunks.join('')).not.toContain('data: [DONE]');
   });

src/tests/unit/proxy-stream-debug.test.ts

@@ -0,0 +1,329 @@
+import { describe, it, expect, vi } from 'vitest';
+import { EventEmitter } from 'events';
+import { ProxyService } from '../../server/modules/proxy/proxy.service';
+
+// Minimal mocks so ProxyService can instantiate
+const mockTokenManager = {
+  getNextToken: vi.fn(),
+  markAsRateLimited: vi.fn(),
+  markAsForbidden: vi.fn(),
+  forceRefreshToken: vi.fn(),
+  markFromUpstreamError: vi.fn(),
+  recordParityError: vi.fn(),
+  getModelOutputLimitForAccount: vi.fn(),
+  getModelThinkingBudgetForAccount: vi.fn(),
+  resolveDynamicModelForAccount: vi.fn((_token: unknown, model: string) => model),
+};
+const mockGeminiClient = { streamGenerateInternal: vi.fn(), generateInternal: vi.fn() };
+
+class TestableProxyService extends ProxyService {
+  constructor() {
+    super(mockTokenManager as any, mockGeminiClient as any);
+  }
+
+  public testProcessStreamResponse(
+    stream: any,
+    model: string = 'gemini-3.1-pro-low',
+    requestId?: string,
+  ) {
+    return (this as any).processStreamResponse(stream, model, requestId);
+  }
+}
+
+function collectChunks(observable: any): Promise<string[]> {
+  return new Promise((resolve, reject) => {
+    const chunks: string[] = [];
+    observable.subscribe({
+      next: (chunk: string) => chunks.push(chunk),
+      error: reject,
+      complete: () => resolve(chunks),
+    });
+  });
+}
+
+function emitSseLine(stream: EventEmitter, payload: object): void {
+  stream.emit('data', Buffer.from(`data: ${JSON.stringify(payload)}\n`));
+}
+
+describe('ProxyService Stream Debug', () => {
+  it('emits role chunk followed by content for a simple text response', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    emitSseLine(stream, {
+      candidates: [
+        {
+          content: {
+            parts: [{ text: 'Hello world' }],
+          },
+        },
+      ],
+    });
+
+    emitSseLine(stream, {
+      candidates: [
+        {
+          content: {
+            parts: [{ text: '!' }],
+          },
+          finishReason: 'STOP',
+        },
+      ],
+    });
+
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Simple text response chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    // Should have at least role + content + finish + DONE
+    expect(chunks.length).toBeGreaterThanOrEqual(4);
+    expect(chunks[0]).toContain('"role":"assistant"');
+    expect(chunks.some((c) => c.includes('"content":"Hello world"'))).toBe(true);
+    expect(chunks.some((c) => c.includes('"finish_reason":"stop"'))).toBe(true);
+    expect(chunks[chunks.length - 1]).toContain('[DONE]');
+  });
+
+  it('handles empty parts array gracefully', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    emitSseLine(stream, {
+      candidates: [
+        {
+          content: {
+            parts: [],
+          },
+          finishReason: 'STOP',
+        },
+      ],
+    });
+
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Empty parts response chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    // With empty parts, only finish + DONE should be emitted
+    expect(chunks.length).toBeGreaterThanOrEqual(2);
+    expect(chunks.some((c) => c.includes('"finish_reason":"stop"'))).toBe(true);
+    expect(chunks[chunks.length - 1]).toContain('[DONE]');
+  });
+
+  it('handles reasoning content (thought parts)', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    emitSseLine(stream, {
+      candidates: [
+        {
+          content: {
+            parts: [{ thought: true, text: 'Let me think...' }],
+          },
+        },
+      ],
+    });
+
+    emitSseLine(stream, {
+      candidates: [
+        {
+          content: {
+            parts: [{ text: 'Final answer' }],
+          },
+          finishReason: 'STOP',
+        },
+      ],
+    });
+
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Reasoning response chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    expect(chunks.some((c) => c.includes('"reasoning_content":"Let me think..."'))).toBe(true);
+    expect(chunks.some((c) => c.includes('"content":"Final answer"'))).toBe(true);
+  });
+
+  it('handles function call parts', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    emitSseLine(stream, {
+      candidates: [
+        {
+          content: {
+            parts: [
+              {
+                functionCall: {
+                  id: 'fc-1',
+                  name: 'get_weather',
+                  args: { city: 'London' },
+                },
+              },
+            ],
+          },
+          finishReason: 'STOP',
+        },
+      ],
+    });
+
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Function call response chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    expect(chunks.some((c) => c.includes('"tool_calls"'))).toBe(true);
+    expect(chunks.some((c) => c.includes('"finish_reason":"tool_calls"'))).toBe(true);
+  });
+
+  it('handles multiple SSE lines in a single buffer', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    // Simulate two complete SSE lines in one data chunk
+    const payload1 = JSON.stringify({
+      candidates: [{ content: { parts: [{ text: 'First' }] } }],
+    });
+    const payload2 = JSON.stringify({
+      candidates: [{ content: { parts: [{ text: 'Second' }] }, finishReason: 'STOP' }],
+    });
+    stream.emit('data', Buffer.from(`data: ${payload1}\ndata: ${payload2}\n`));
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Multi-line buffer chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    expect(chunks.some((c) => c.includes('"content":"First"'))).toBe(true);
+    expect(chunks.some((c) => c.includes('"content":"Second"'))).toBe(true);
+    expect(chunks[chunks.length - 1]).toContain('[DONE]');
+  });
+
+  it('handles partial SSE lines across multiple data events', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    const payload = JSON.stringify({
+      candidates: [{ content: { parts: [{ text: 'Split' }] }, finishReason: 'STOP' }],
+    });
+    const line = `data: ${payload}\n`;
+
+    // Split the line across two data events
+    const mid = Math.floor(line.length / 2);
+    stream.emit('data', Buffer.from(line.slice(0, mid)));
+    stream.emit('data', Buffer.from(line.slice(mid)));
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Split-line buffer chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    expect(chunks.some((c) => c.includes('"content":"Split"'))).toBe(true);
+    expect(chunks[chunks.length - 1]).toContain('[DONE]');
+  });
+
+  it('handles Gemini v1internal wrapped response format (response.candidates)', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    // v1internal API wraps the payload in {"response": {"candidates": [...]}}
+    stream.emit(
+      'data',
+      Buffer.from(
+        `data: ${JSON.stringify({
+          response: {
+            candidates: [
+              {
+                content: {
+                  parts: [{ text: 'Hello from wrapped' }],
+                },
+              },
+            ],
+          },
+          traceId: 'abc123',
+          metadata: {},
+        })}\n`,
+      ),
+    );
+
+    stream.emit(
+      'data',
+      Buffer.from(
+        `data: ${JSON.stringify({
+          response: {
+            candidates: [
+              {
+                content: {
+                  parts: [{ text: ' response' }],
+                },
+                finishReason: 'STOP',
+              },
+            ],
+          },
+          traceId: 'abc123',
+          metadata: {},
+        })}\n`,
+      ),
+    );
+
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== Wrapped response format chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    expect(chunks.some((c) => c.includes('"content":"Hello from wrapped"'))).toBe(true);
+    expect(chunks.some((c) => c.includes('"content":" response"'))).toBe(true);
+    expect(chunks[chunks.length - 1]).toContain('[DONE]');
+  });
+
+  it('handles Gemini streaming chunks with no content field', async () => {
+    const service = new TestableProxyService();
+    const stream = new EventEmitter();
+    const observable = service.testProcessStreamResponse(stream, 'gemini-3.1-pro-low');
+
+    const promise = collectChunks(observable);
+
+    // Sometimes Gemini sends candidates without content at all
+    emitSseLine(stream, {
+      candidates: [{ index: 0 }],
+    });
+
+    emitSseLine(stream, {
+      candidates: [{ content: { parts: [{ text: 'Late content' }] }, finishReason: 'STOP' }],
+    });
+
+    stream.emit('end');
+
+    const chunks = await promise;
+    console.log('=== No-content-then-late-content chunks ===');
+    chunks.forEach((c, i) => console.log(`Chunk ${i}:`, c.trim()));
+
+    expect(chunks.some((c) => c.includes('"content":"Late content"'))).toBe(true);
+  });
+});

src/utils/traffic-logger.ts

@@ -87,9 +87,22 @@ class TrafficLogger {
       console.error('Traffic DailyRotateFile transport error', error);
     });
 
+    const consoleFormat = winston.format.combine(
+      winston.format.colorize(),
+      winston.format.printf(({ level, message }) => {
+        return `[TRAFFIC] ${level}: ${message}`;
+      }),
+    );
+
     this.winstonLogger = winston.createLogger({
       level: 'debug',
-      transports: [rotateTransport],
+      transports: [
+        rotateTransport,
+        new winston.transports.Console({
+          level: 'debug',
+          format: consoleFormat,
+        }),
+      ],
       exitOnError: false,
     });
   }