diff --git a/src/glmt/delta-accumulator.ts b/src/glmt/delta-accumulator.ts index 4988c9e8..7923a0ae 100644 --- a/src/glmt/delta-accumulator.ts +++ b/src/glmt/delta-accumulator.ts @@ -40,6 +40,7 @@ interface ToolCall { name: string; arguments: string; }; + blockIndex: number; } interface ToolCallDelta { @@ -240,7 +241,6 @@ export class DeltaAccumulator { addToolCallDelta(toolCallDelta: ToolCallDelta): void { const index = toolCallDelta.index; - // Initialize tool call if not exists if (!this.toolCallsIndex[index]) { const toolCall: ToolCall = { index: index, @@ -250,6 +250,7 @@ export class DeltaAccumulator { name: '', arguments: '', }, + blockIndex: -1, }; this.toolCalls.push(toolCall); this.toolCallsIndex[index] = toolCall; @@ -257,27 +258,39 @@ export class DeltaAccumulator { const toolCall = this.toolCallsIndex[index]; - // Update id if present if (toolCallDelta.id) { toolCall.id = toolCallDelta.id; } - // Update type if present if (toolCallDelta.type) { toolCall.type = toolCallDelta.type; } - // Update function name if present if (toolCallDelta.function?.name) { toolCall.function.name += toolCallDelta.function.name; } - // Update function arguments if present if (toolCallDelta.function?.arguments) { toolCall.function.arguments += toolCallDelta.function.arguments; } } + setToolCallBlockIndex(toolCallIndex: number, blockIndex: number): void { + const toolCall = this.toolCallsIndex[toolCallIndex]; + if (toolCall) { + toolCall.blockIndex = blockIndex; + } + } + + getToolCallBlockIndex(toolCallIndex: number): number { + const toolCall = this.toolCallsIndex[toolCallIndex]; + return toolCall?.blockIndex ?? 0; + } + + getUnstoppedBlocks(): ContentBlock[] { + return this.contentBlocks.filter((b) => !b.stopped); + } + /** * Get all tool calls * @returns Tool calls array diff --git a/src/glmt/pipeline/stream-parser.ts b/src/glmt/pipeline/stream-parser.ts index c0b6fc71..b72218a2 100644 --- a/src/glmt/pipeline/stream-parser.ts +++ b/src/glmt/pipeline/stream-parser.ts @@ -209,9 +209,8 @@ export class StreamParser { if (!toolCallDeltas) return events; - // Close current content block ONCE before processing any tool calls const currentBlock = accumulator.getCurrentBlock(); - if (currentBlock && !currentBlock.stopped) { + if (currentBlock && !currentBlock.stopped && currentBlock.type !== 'tool_use') { if (currentBlock.type === 'thinking') { events.push(...this.closeThinkingBlock(currentBlock, accumulator)); } else { @@ -220,7 +219,6 @@ export class StreamParser { } } - // Process tool call deltas events.push(...this.toolCallHandler.processToolCallDeltas(toolCallDeltas, accumulator)); return events; @@ -251,44 +249,41 @@ export class StreamParser { private forceFinalization(accumulator: DeltaAccumulator): AnthropicSSEEvent[] { const events: AnthropicSSEEvent[] = []; - // Close current block if any - const currentBlock = accumulator.getCurrentBlock(); - if (currentBlock && !currentBlock.stopped) { - if (currentBlock.type === 'thinking') { - events.push(...this.closeThinkingBlock(currentBlock, accumulator)); - } else { - events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock)); - accumulator.stopCurrentBlock(); + const unstoppedBlocks = accumulator.getUnstoppedBlocks(); + for (const block of unstoppedBlocks) { + if (block.type === 'thinking') { + const signatureEvent = this.responseBuilder.createSignatureDeltaEvent(block); + if (signatureEvent) { + events.push(signatureEvent); + } } + events.push(this.responseBuilder.createContentBlockStopEvent(block)); + block.stopped = true; } - // Force finalization events.push(...this.finalizeDelta(accumulator)); return events; } - /** - * Finalize streaming and generate closing events - */ finalizeDelta(accumulator: DeltaAccumulator): AnthropicSSEEvent[] { if (accumulator.isFinalized()) { - return []; // Already finalized + return []; } const events: AnthropicSSEEvent[] = []; - // Close current content block if any - const currentBlock = accumulator.getCurrentBlock(); - if (currentBlock && !currentBlock.stopped) { - if (currentBlock.type === 'thinking') { - events.push(...this.closeThinkingBlock(currentBlock, accumulator)); - } else { - events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock)); - accumulator.stopCurrentBlock(); + const unstoppedBlocks = accumulator.getUnstoppedBlocks(); + for (const block of unstoppedBlocks) { + if (block.type === 'thinking') { + const signatureEvent = this.responseBuilder.createSignatureDeltaEvent(block); + if (signatureEvent) { + events.push(signatureEvent); + } } + events.push(this.responseBuilder.createContentBlockStopEvent(block)); + block.stopped = true; } - // Message delta (stop reason + usage) and message stop const stopReason = this.responseBuilder.mapStopReason(accumulator.getFinishReason() || 'stop'); events.push(...this.responseBuilder.createFinalizationEvents(accumulator, stopReason)); diff --git a/src/glmt/pipeline/tool-call-handler.ts b/src/glmt/pipeline/tool-call-handler.ts index ffe0d670..5bc7f46d 100644 --- a/src/glmt/pipeline/tool-call-handler.ts +++ b/src/glmt/pipeline/tool-call-handler.ts @@ -5,15 +5,20 @@ * - Process tool call deltas from OpenAI * - Generate tool_use content blocks for Anthropic format * - Handle input_json_delta events + * - Close previous tool_use blocks before starting new ones (Anthropic sequential block requirement) */ import type { DeltaAccumulator } from '../delta-accumulator'; import type { OpenAIToolCallDelta, OpenAIToolCall, ContentBlock, AnthropicSSEEvent } from './types'; +import { ResponseBuilder } from './response-builder'; export class ToolCallHandler { - /** - * Process tool calls from non-streaming response - */ + private responseBuilder: ResponseBuilder; + + constructor() { + this.responseBuilder = new ResponseBuilder(false); + } + processToolCalls(toolCalls: OpenAIToolCall[]): ContentBlock[] { const content: ContentBlock[] = []; @@ -38,10 +43,6 @@ export class ToolCallHandler { return content; } - /** - * Process tool call deltas during streaming - * Returns events for tool use blocks and input_json deltas - */ processToolCallDeltas( toolCallDeltas: OpenAIToolCallDelta[], accumulator: DeltaAccumulator @@ -49,15 +50,19 @@ export class ToolCallHandler { const events: AnthropicSSEEvent[] = []; for (const toolCallDelta of toolCallDeltas) { - // Track tool call state const isNewToolCall = !accumulator.hasToolCall(toolCallDelta.index); accumulator.addToolCallDelta(toolCallDelta); - // Emit tool use events (start + input_json deltas) if (isNewToolCall) { - // Start new tool_use block in accumulator + const previousBlock = accumulator.getCurrentBlock(); + if (previousBlock && previousBlock.type === 'tool_use' && !previousBlock.stopped) { + events.push(this.responseBuilder.createContentBlockStopEvent(previousBlock)); + previousBlock.stopped = true; + } + const block = accumulator.startBlock('tool_use'); const toolCall = accumulator.getToolCall(toolCallDelta.index); + accumulator.setToolCallBlockIndex(toolCallDelta.index, block.index); events.push({ event: 'content_block_start', @@ -68,27 +73,25 @@ export class ToolCallHandler { type: 'tool_use', id: toolCall?.id || `tool_${toolCallDelta.index}`, name: toolCall?.function?.name || '', + input: {}, }, }, }); } - // Emit input_json delta if arguments present if (toolCallDelta.function?.arguments) { - const currentToolBlock = accumulator.getCurrentBlock(); - if (currentToolBlock && currentToolBlock.type === 'tool_use') { - events.push({ - event: 'content_block_delta', - data: { - type: 'content_block_delta', - index: currentToolBlock.index, - delta: { - type: 'input_json_delta', - partial_json: toolCallDelta.function.arguments, - }, + const toolCallBlockIndex = accumulator.getToolCallBlockIndex(toolCallDelta.index); + events.push({ + event: 'content_block_delta', + data: { + type: 'content_block_delta', + index: toolCallBlockIndex, + delta: { + type: 'input_json_delta', + partial_json: toolCallDelta.function.arguments, }, - }); - } + }, + }); } } diff --git a/src/proxy/transformers/request-transformer.ts b/src/proxy/transformers/request-transformer.ts index c1ebd5d7..a9c6ab69 100644 --- a/src/proxy/transformers/request-transformer.ts +++ b/src/proxy/transformers/request-transformer.ts @@ -457,8 +457,7 @@ function transformMessages(messagesValue: unknown): OpenAIMessage[] { } if (role === 'user') { - const userParts: OpenAIContentPart[] = []; - let sawToolResult = false; + const preToolResultParts: OpenAIContentPart[] = []; const resolvedToolUseIds = new Set(); content.forEach((block, blockIndex) => { @@ -472,23 +471,26 @@ function transformMessages(messagesValue: unknown): OpenAIMessage[] { } if (parsed.type === 'text') { - if (sawToolResult) { - throw new Error( - `messages[${messageIndex}].content[${blockIndex}] text is not allowed after tool_result blocks` - ); - } const text = typeof parsed.text === 'string' ? parsed.text : ''; - userParts.push({ type: 'text', text }); + if (pendingToolUseIds && pendingToolUseIds.size > 0 && !resolvedToolUseIds.size) { + preToolResultParts.push({ type: 'text', text }); + } else { + translatedMessages.push({ role: 'user', content: text }); + } return; } if (isImageBlock(parsed)) { - if (sawToolResult) { - throw new Error( - `messages[${messageIndex}].content[${blockIndex}] image is not allowed after tool_result blocks` + if (pendingToolUseIds && pendingToolUseIds.size > 0 && !resolvedToolUseIds.size) { + preToolResultParts.push( + toImagePart(parsed, `messages[${messageIndex}].content[${blockIndex}]`) ); + } else { + translatedMessages.push({ + role: 'user', + content: [toImagePart(parsed, `messages[${messageIndex}].content[${blockIndex}]`)], + }); } - userParts.push(toImagePart(parsed, `messages[${messageIndex}].content[${blockIndex}]`)); return; } @@ -498,11 +500,6 @@ function transformMessages(messagesValue: unknown): OpenAIMessage[] { `messages[${messageIndex}].content[${blockIndex}] tool_result requires a preceding assistant tool_use` ); } - if (userParts.length > 0) { - throw new Error( - `messages[${messageIndex}].content[${blockIndex}] tool_result blocks must come before other user content` - ); - } if (typeof parsed.tool_use_id !== 'string' || parsed.tool_use_id.trim().length === 0) { throw new Error( `messages[${messageIndex}].content[${blockIndex}].tool_use_id must be a non-empty string` @@ -518,7 +515,6 @@ function transformMessages(messagesValue: unknown): OpenAIMessage[] { `messages[${messageIndex}].content[${blockIndex}].tool_use_id "${parsed.tool_use_id}" is duplicated` ); } - sawToolResult = true; resolvedToolUseIds.add(parsed.tool_use_id); translatedMessages.push({ role: 'tool', @@ -543,7 +539,7 @@ function transformMessages(messagesValue: unknown): OpenAIMessage[] { ); }); - if (sawToolResult) { + if (resolvedToolUseIds.size > 0) { if (resolvedToolUseIds.size !== pendingToolUseIds?.size) { throw new Error( `messages[${messageIndex}].content must provide tool_result blocks for all pending tool_use ids` @@ -551,17 +547,16 @@ function transformMessages(messagesValue: unknown): OpenAIMessage[] { } pendingToolUseIds = null; hasPendingToolUseIds = false; - return; } if (pendingToolUseIds && pendingToolUseIds.size > 0) { throw new Error( - `messages[${messageIndex}].content must start with tool_result blocks for pending tool_use ids` + `messages[${messageIndex}].content must include tool_result blocks for pending tool_use ids` ); } - if (userParts.length > 0) { - flushUserContent(translatedMessages, userParts); + if (preToolResultParts.length > 0) { + flushUserContent(translatedMessages, preToolResultParts); } return; } diff --git a/tests/integration/proxy/messages-endpoint.test.ts b/tests/integration/proxy/messages-endpoint.test.ts index 263a1594..cf1d408b 100644 --- a/tests/integration/proxy/messages-endpoint.test.ts +++ b/tests/integration/proxy/messages-endpoint.test.ts @@ -319,4 +319,94 @@ describe('openai proxy messages endpoint', () => { expect(body.ok).toBe(true); expect(body.endpoints).toContain('/v1/messages'); }); + + it('translates user messages with tool_result followed by text', async () => { + const response = await requestProxy({ + model: 'hf-model', + stream: false, + messages: [ + { role: 'user', content: 'search docs' }, + { + role: 'assistant', + content: [ + { type: 'text', text: 'Let me search.' }, + { type: 'tool_use', id: 'toolu_01', name: 'search', input: { q: 'docs' } }, + ], + }, + { + role: 'user', + content: [ + { type: 'tool_result', tool_use_id: 'toolu_01', content: 'found 3 docs' }, + { type: 'text', text: 'What should I do next?' }, + ], + }, + ], + }); + + expect(response.status).toBe(200); + const parsedUpstream = upstreamBody as { + messages?: Array<{ role: string; content: string; tool_call_id?: string }>; + }; + const roles = parsedUpstream.messages?.map((m) => m.role); + expect(roles).toEqual(['user', 'assistant', 'tool', 'user']); + const toolMsg = parsedUpstream.messages?.find((m) => m.role === 'tool'); + expect(toolMsg?.tool_call_id).toBe('toolu_01'); + expect(toolMsg?.content).toBe('found 3 docs'); + const userAfterTool = parsedUpstream.messages?.filter((m) => m.role === 'user'); + expect(userAfterTool?.[1]?.content).toBe('What should I do next?'); + }); + + it('translates parallel tool calls with streaming', async () => { + const response = await requestProxy({ + model: 'hf-model', + stream: true, + messages: [ + { role: 'user', content: 'read both files' }, + { + role: 'assistant', + content: [ + { type: 'tool_use', id: 'toolu_01', name: 'Read', input: { file_path: 'a.ts' } }, + { type: 'tool_use', id: 'toolu_02', name: 'Read', input: { file_path: 'b.ts' } }, + ], + }, + { + role: 'user', + content: [ + { type: 'tool_result', tool_use_id: 'toolu_01', content: 'content of a' }, + { type: 'tool_result', tool_use_id: 'toolu_02', content: 'content of b' }, + { type: 'text', text: 'Now compare them' }, + ], + }, + ], + tools: [ + { + name: 'Read', + description: 'Read a file', + input_schema: { + type: 'object', + properties: { file_path: { type: 'string' } }, + required: ['file_path'], + }, + }, + ], + }); + + expect(response.status).toBe(200); + const parsedUpstream = upstreamBody as { + messages?: Array<{ + role: string; + content: string; + tool_call_id?: string; + tool_calls?: Array<{ id: string; function: { name: string; arguments: string } }>; + }>; + }; + const assistantMsg = parsedUpstream.messages?.find((m) => m.role === 'assistant'); + expect(assistantMsg?.tool_calls?.length).toBe(2); + expect(assistantMsg?.tool_calls?.[0]?.function.name).toBe('Read'); + expect(assistantMsg?.tool_calls?.[1]?.function.name).toBe('Read'); + const toolMsgs = parsedUpstream.messages?.filter((m) => m.role === 'tool'); + expect(toolMsgs?.length).toBe(2); + expect(toolMsgs?.[0]?.tool_call_id).toBe('toolu_01'); + expect(toolMsgs?.[1]?.tool_call_id).toBe('toolu_02'); + }); });