diff --git a/src/server/main.ts b/src/server/main.ts index 5b4df4e..ff68507 100644 --- a/src/server/main.ts +++ b/src/server/main.ts @@ -22,9 +22,13 @@ export async function bootstrapNestServer(config: ProxyConfig): Promise setServerConfig(config); try { - app = await NestFactory.create(AppModule, new FastifyAdapter(), { - logger: ['error', 'warn', 'log'], - }); + app = await NestFactory.create( + AppModule, + new FastifyAdapter({ bodyLimit: 10 * 1024 * 1024 }), + { + logger: ['error', 'warn', 'log'], + }, + ); // Enable CORS app.enableCors(); diff --git a/src/server/modules/proxy/proxy.service.ts b/src/server/modules/proxy/proxy.service.ts index c184edc..8cd5af5 100644 --- a/src/server/modules/proxy/proxy.service.ts +++ b/src/server/modules/proxy/proxy.service.ts @@ -844,6 +844,7 @@ export class ProxyService { request: OpenAIChatRequest, context?: ProxyRequestContext, configOverride?: ProxyConfigOverride, + _requestId?: string, ): Promise> { const targetModel = this.resolveTargetModel(request.model); const extraHeaders = this.createModelSpecificHeaders(request.model); @@ -1292,7 +1293,11 @@ export class ProxyService { if (part.functionCall) { pushRoleChunk(); const currentToolIndex = toolCallIndex++; - const toolCallChunk = { + const toolCallId = part.functionCall.id || `${part.functionCall.name}-${uuidv4()}`; + const argsJson = JSON.stringify(part.functionCall.args || {}); + + // First delta: id, type, name, empty arguments + pushChunk({ id: streamId, object: 'chat.completion.chunk', created: created, @@ -1304,11 +1309,11 @@ export class ProxyService { tool_calls: [ { index: currentToolIndex, - id: part.functionCall.id || `${part.functionCall.name}-${uuidv4()}`, + id: toolCallId, type: 'function', function: { name: part.functionCall.name, - arguments: JSON.stringify(part.functionCall.args || {}), + arguments: '', }, }, ], @@ -1316,8 +1321,35 @@ export class ProxyService { finish_reason: null, }, ], - }; - pushChunk(toolCallChunk); + }); + + // Subsequent deltas: only arguments (split if large to respect SSE framing) + const ARG_CHUNK_SIZE = 1000; + for (let i = 0; i < argsJson.length; i += ARG_CHUNK_SIZE) { + const slice = argsJson.slice(i, i + ARG_CHUNK_SIZE); + pushChunk({ + id: streamId, + object: 'chat.completion.chunk', + created: created, + model: model, + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: currentToolIndex, + function: { + arguments: slice, + }, + }, + ], + }, + finish_reason: null, + }, + ], + }); + } continue; } diff --git a/src/tests/unit/proxy-tool-streaming.test.ts b/src/tests/unit/proxy-tool-streaming.test.ts new file mode 100644 index 0000000..e60cdbc --- /dev/null +++ b/src/tests/unit/proxy-tool-streaming.test.ts @@ -0,0 +1,173 @@ +import { describe, expect, it } from 'vitest'; +import { EventEmitter } from 'events'; +import { ProxyService } from '../../server/modules/proxy/proxy.service'; +import { Observable } from 'rxjs'; + +// Subclass to access private method +class TestableProxyService extends ProxyService { + constructor() { + super({} as any, {} as any); + } + + public testProcessStream( + stream: any, + model: string = 'model', + streamOptions?: { include_usage?: boolean }, + ): Observable { + return (this as any).processStreamResponse(stream, model, streamOptions); + } +} + +function parseSseChunk(raw: string): Record { + const lines = raw.split('\n').filter((l) => l.trim().startsWith('data: ')); + if (lines.length === 0) return {}; + const payload = lines[0].slice(6).trim(); + try { + return JSON.parse(payload) as Record; + } catch { + return {}; + } +} + +describe('ProxyService Tool Streaming (OpenAI format)', () => { + it('should emit tool call deltas with id+name first, then arguments separately', async () => { + const service = new TestableProxyService(); + const stream = new EventEmitter(); + const observable = service.testProcessStream(stream, 'gpt-4o'); + + const chunks: string[] = []; + const done = new Promise((resolve, reject) => { + observable.subscribe({ + next: (chunk) => chunks.push(chunk), + error: reject, + complete: resolve, + }); + }); + + const payload = JSON.stringify({ + candidates: [ + { + content: { + parts: [ + { functionCall: { id: 'call_123', name: 'get_weather', args: { location: 'NYC' } } }, + ], + }, + finishReason: 'STOP', + }, + ], + }); + + stream.emit('data', Buffer.from(`data: ${payload}\n`)); + stream.emit('end'); + + await done; + + const toolCallChunks = chunks.filter((c) => c.includes('"tool_calls"')); + expect(toolCallChunks.length).toBeGreaterThanOrEqual(2); + + const first = parseSseChunk(toolCallChunks[0]); + const firstDeltas = (first.choices as any)?.[0]?.delta?.tool_calls as any[]; + expect(firstDeltas).toBeDefined(); + expect(firstDeltas[0].index).toBe(0); + expect(firstDeltas[0].id).toBe('call_123'); + expect(firstDeltas[0].type).toBe('function'); + expect(firstDeltas[0].function.name).toBe('get_weather'); + expect(firstDeltas[0].function.arguments).toBe(''); + + const second = parseSseChunk(toolCallChunks[1]); + const secondDeltas = (second.choices as any)?.[0]?.delta?.tool_calls as any[]; + expect(secondDeltas).toBeDefined(); + expect(secondDeltas[0].index).toBe(0); + expect(secondDeltas[0].function.arguments).toBe('{"location":"NYC"}'); + // Subsequent deltas must NOT repeat id/type/name to keep strict clients happy + expect(secondDeltas[0].id).toBeUndefined(); + expect(secondDeltas[0].type).toBeUndefined(); + expect(secondDeltas[0].function.name).toBeUndefined(); + }); + + it('should split large tool arguments into multiple argument deltas', async () => { + const service = new TestableProxyService(); + const stream = new EventEmitter(); + const observable = service.testProcessStream(stream, 'gpt-4o'); + + const chunks: string[] = []; + const done = new Promise((resolve, reject) => { + observable.subscribe({ + next: (chunk) => chunks.push(chunk), + error: reject, + complete: resolve, + }); + }); + + const bigArgs = { code: 'a'.repeat(5000) }; + const payload = JSON.stringify({ + candidates: [ + { + content: { + parts: [{ functionCall: { id: 'call_456', name: 'write_file', args: bigArgs } }], + }, + finishReason: 'STOP', + }, + ], + }); + + stream.emit('data', Buffer.from(`data: ${payload}\n`)); + stream.emit('end'); + + await done; + + const toolCallChunks = chunks.filter((c) => c.includes('"tool_calls"')); + // At least: 1st (id+name) + one or more argument deltas + expect(toolCallChunks.length).toBeGreaterThanOrEqual(2); + + const first = parseSseChunk(toolCallChunks[0]); + expect((first.choices as any)[0].delta.tool_calls[0].function.arguments).toBe(''); + + // Concatenate all argument deltas + let argsConcat = ''; + for (let i = 1; i < toolCallChunks.length; i++) { + const parsed = parseSseChunk(toolCallChunks[i]); + const argDelta = (parsed.choices as any)?.[0]?.delta?.tool_calls?.[0]?.function?.arguments; + if (typeof argDelta === 'string') { + argsConcat += argDelta; + } + } + expect(argsConcat).toBe(JSON.stringify(bigArgs)); + }); + + it('should not break normal text streaming when no tools are present', async () => { + const service = new TestableProxyService(); + const stream = new EventEmitter(); + const observable = service.testProcessStream(stream, 'gpt-4o'); + + const chunks: string[] = []; + const done = new Promise((resolve, reject) => { + observable.subscribe({ + next: (chunk) => chunks.push(chunk), + error: reject, + complete: resolve, + }); + }); + + const payload = JSON.stringify({ + candidates: [ + { + content: { + parts: [{ text: 'hello world' }], + }, + finishReason: 'STOP', + }, + ], + }); + + stream.emit('data', Buffer.from(`data: ${payload}\n`)); + stream.emit('end'); + + await done; + + const output = chunks.join(''); + expect(output).toContain('"content":"hello world"'); + expect(output).toContain('data: [DONE]'); + expect(output).not.toContain('"tool_calls"'); + }); +});