Fixed opanai early termination from tool calls

package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@ztimson/ai-utils",
-	"version": "1.6.8",
+	"version": "1.6.10",
 	"description": "AI Utility library",
 	"author": "Zak Timson",
 	"license": "MIT",

src/open-ai.ts

@@ -41,7 +41,7 @@ export class OpenAi extends LLMProvider {
 				wire.push({
 					role: 'assistant',
 					content: null,
-					tool_calls: [{id: h.id, type: 'function', function: {name: h.name, arguments: JSON.stringify(h.args)}],
+					tool_calls: [{id: h.id, type: 'function', function: {name: h.name, arguments: JSON.stringify(h.args)}}],
 				}, {
 					role: 'tool',
 					tool_call_id: h.id,
@@ -98,144 +98,85 @@ export class OpenAi extends LLMProvider {
 						throw err;
 					});
 
-					let usage: any, msg: any = {content: '', tool_calls: []};
+					let usage: any, finishReason: string | undefined, msg: any = {content: '', tool_calls: []};
 					if(options.stream) {
-						const toolCallState: Record<number, any> = {};
-						let pendingToolCalls = 0;
-
 						for await (const chunk of resp) {
 							if(controller.signal.aborted) break;
-
 							if(chunk.usage) usage = chunk.usage;
-
+							if(chunk.choices[0]?.finish_reason) finishReason = chunk.choices[0].finish_reason;
-							// Handle content deltas
 							if(chunk.choices[0]?.delta?.content) {
 								msg.content += chunk.choices[0].delta.content;
-								options.stream?.({text: chunk.choices[0].delta.content});
+								options.stream({text: chunk.choices[0].delta.content});
 							}
-
-							// Handle tool_call deltas
 							if(chunk.choices[0]?.delta?.tool_calls) {
 								for(const deltaTC of chunk.choices[0].delta.tool_calls) {
-									const tc = toolCallState[deltaTC.index];
+									const existing = deltaTC.index != null
-
+										? msg.tool_calls.find((tc: any) => tc.index === deltaTC.index)
-									if(!tc && deltaTC.id) {
+										: (deltaTC.id ? msg.tool_calls.find((tc: any) => tc.id === deltaTC.id) : undefined);
-										// New tool call delta
+									if(existing) {
-										toolCallState[deltaTC.index] = {
+										if(deltaTC.id) existing.id = deltaTC.id;
-											id: deltaTC.id,
+										if(deltaTC.function?.name) existing.function.name = deltaTC.function.name;
-											name: deltaTC.function?.name || '',
+										if(deltaTC.function?.arguments) existing.function.arguments += deltaTC.function.arguments;
-											arguments: deltaTC.function?.arguments || '',
+									} else {
-											complete: false
-										};
 										msg.tool_calls.push({
 											index: deltaTC.index,
 											id: deltaTC.id || '',
 											function: {name: deltaTC.function?.name || '', arguments: deltaTC.function?.arguments || ''}
 										});
-										pendingToolCalls++;
-									} else if(tc && deltaTC.function?.name) {
-										// Update existing tool call
-										if(deltaTC.id) tc.id = deltaTC.id;
-										if(deltaTC.function.name) tc.name = deltaTC.function.name;
-										if(deltaTC.function.arguments) tc.arguments += deltaTC.function.arguments;
-									}
-
-									if(tc && deltaTC.function?.arguments && !tc.complete) {
-										tc.complete = true;
-										pendingToolCalls--;
 									}
 								}
 							}
-
-							// Execute completed tools immediately when all deltas are received
-							if(!pendingToolCalls && msg.tool_calls.length > 0) {
-								const completedTools: any[] = [];
-
-								for(const tc of msg.tool_calls) {
-									const tcState = toolCallState[tc.index];
-									if(!tcState) continue;
-
-									const entry: any = {role: 'tool', id: tc.id, name: tcState.name, args: JSONAttemptParse(tcState.arguments, {}), content: undefined, timestamp: Date.now()};
-									history.push(entry);
-									completedTools.push({tc, entry});
-
-									const tool = tools.find(findByProp('name', tcState.name));
-									if(options.stream) options.stream?.({tool: tcState.name});
-
-									if(!tool) {
-										entry.error = 'Tool not found';
-										continue;
-									}
-
-									try {
-										const toolStream = options.stream && ((chunk: any) => {
-											if(chunk.done) { terminal = true; return; }
-											options.stream?.(chunk);
-										});
-										const result = await tool.fn(entry.args, toolStream, this.ai, tc.id);
-										entry.content = typeof result === 'object' ? JSONSanitize(result) : result;
-									} catch(err: any) {
-										entry.error = err?.message || err?.toString() || 'Unknown';
-									}
-								}
-
-								if(options.stream) options.stream?.({done: true});
-
-								for(const {entry} of completedTools) {
-									if(entry.error) {
-										options.stream?.({error: entry.error});
-									} else if(entry.content) {
-										options.stream?.({toolResult: true});
-									}
-								}
-
-								msg.tool_calls = []; // Clear after execution
-							}
-						}
-
-						// Handle finish_reason for proper termination detection
-						if(chunk.choices[0]?.finish_reason) {
-							const finishReason = chunk.choices[0].finish_reason;
-							if(finishReason === 'stop' || finishReason === 'tool_calls' || finishReason === 'length') {
-								terminal = true;
-							}
 						}
 					} else {
 						usage = resp.usage;
+						finishReason = resp.choices[0].finish_reason;
 						msg = resp.choices[0].message;
-						if(msg.tool_calls) {
-							// Non-streaming: execute tools immediately
-							for(const tc of msg.tool_calls) {
-								const entry: any = {role: 'tool', id: tc.id, name: tc.function.name, args: JSONAttemptParse(tc.function.arguments, {}), content: undefined, timestamp: Date.now()};
-								history.push(entry);
-
-								const tool = tools.find(findByProp('name', tc.function.name));
-								if(!tool) {
-									entry.error = 'Tool not found';
-									continue;
-								}
-
-								try {
-									const result = await tool.fn(entry.args, undefined, this.ai, tc.id);
-									entry.content = typeof result === 'object' ? JSONSanitize(result) : result;
-								} catch(err: any) {
-									entry.error = err?.message || err?.toString() || 'Unknown';
-								}
-							}
-						}
 					}
-
 					const duration = Date.now() - callStart;
 					const tps = usage?.completion_tokens && duration > 0 ? usage.completion_tokens / (duration / 1000) : 0;
 
-					// Capture assistant messages (before or after tools)
+					if(finishReason === 'length' && !controller.signal.aborted) {
-					if(msg.content?.trim()) {
+						if(msg.content?.trim()) history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps});
-						history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps});
+						throw new Error(`[OpenAI] Response hit token limit before completing`);
 					}
 
+					if(!finishReason && !controller.signal.aborted) {
+						throw new Error('[OpenAI] Stream ended prematurely - connection likely dropped');
+					}
+
+					const toolCalls = msg.tool_calls || [];
+					if(toolCalls.length && !controller.signal.aborted) {
+						if(msg.content?.trim()) history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps});
+
+						const entries = toolCalls.map((tc: any) => {
+							const entry: any = {role: 'tool', id: tc.id, name: tc.function.name, args: JSONAttemptParse(tc.function.arguments, {}), content: undefined, timestamp: Date.now()};
+							history.push(entry);
+							return {tc, entry};
+						});
+
+						await Promise.all(entries.map(async ({tc, entry}: any) => {
+							const tool = tools.find(findByProp('name', tc.function.name));
+							if(options.stream) options.stream({tool: tc.function.name});
+							if(!tool) { entry.error = 'Tool not found'; return; }
+							try {
+								const toolStream = options.stream && ((chunk: any) => {
+									if(chunk.done) return;
+									options.stream!(chunk);
+								});
+								const result = await tool.fn(entry.args, toolStream, this.ai, tc.id);
+								entry.content = typeof result === 'object' ? JSONSanitize(result) : result;
+							} catch(err: any) {
+								entry.error = err?.message || err?.toString() || 'Unknown';
+							}
+						}));
+					} else {
+						terminal = true;
+						const text = (msg.content || '').trim();
+						if(text) history.push({role: 'assistant', content: text, timestamp: Date.now(), duration, tps});
+					}
 				} while(!terminal && !controller.signal.aborted);
 
-				if(options.stream) options.stream?.({done: true});
+				if(options.stream) options.stream({done: true});
 
 				const turnStart = history.map(h => h.role).lastIndexOf('user');
 				const finalContent = history.slice(turnStart + 1).reduce((str, h) => h.role === 'assistant' ? str + (h.content || '') : str, '').trim();
@@ -245,4 +186,4 @@ export class OpenAi extends LLMProvider {
 			}
 		}), {abort: () => controller.abort()});
 	}
-}
+}