Fix open-ai early termination

src/open-ai.ts

@@ -36,21 +36,37 @@ export class OpenAi extends LLMProvider {
 	private toWire(history: LLMMessage[], system?: string): any[] {
 		const wire: any[] = [];
 		if(system) wire.push({role: 'system', content: system});
-		for(const h of history) {
-			if(h.role === 'tool') {
-				wire.push({
-					role: 'assistant',
-					content: null,
-					tool_calls: [{id: h.id, type: 'function', function: {name: h.name, arguments: JSON.stringify(h.args)}}],
-				}, {
-					role: 'tool',
-					tool_call_id: h.id,
-					content: h.error || h.content || '',
-				});
-			} else {
+		for(let i = 0; i < history.length; i++) {
+			const h = history[i];
+			if(h.role !== 'tool') {
 				wire.push({role: h.role, content: this.toWireContent(h.content)});
+				continue;
 			}
+
+			const calls: any[] = [];
+			const results: any[] = [];
+			while(i < history.length && history[i].role === 'tool') {
+				const tool = <any>history[i];
+				calls.push({
+					id: tool.id,
+					type: 'function',
+					function: {
+						name: tool.name,
+						arguments: JSON.stringify(tool.args)
+					}
+				});
+				results.push({
+					role: 'tool',
+					tool_call_id: tool.id,
+					content: tool.error || tool.content || ''
+				});
+				i++;
+			}
+			wire.push({role: 'assistant', content: null, tool_calls: calls});
+			wire.push(...results);
+			i--;
 		}
+
 		return wire;
 	}
 
@@ -74,8 +90,12 @@ export class OpenAi extends LLMProvider {
 						description: t.description,
 						parameters: {
 							type: 'object',
-							properties: t.args ? objectMap(t.args, (key, value) => ({...value, required: undefined})) : {},
-							required: t.args ? Object.entries(t.args).filter(t => t[1].required).map(t => t[0]) : []
+							properties: t.args
+								? objectMap(t.args, (key, value) => ({...value, required: undefined}))
+								: {},
+							required: t.args
+								? Object.entries(t.args).filter(t => t[1].required).map(t => t[0])
+								: []
 						}
 					}
 				}))
@@ -83,7 +103,10 @@ export class OpenAi extends LLMProvider {
 
 			if(options.schema) {
 				const schema = convertSchema(options.schema);
-				requestParams.response_format = {type: 'json_schema', json_schema: {name: 'response', strict: true, schema}};
+				requestParams.response_format = {
+					type: 'json_schema',
+					json_schema: {name: 'response', strict: true, schema}
+				};
 			}
 			if(options.stream) requestParams.stream_options = {include_usage: true};
 
@@ -93,47 +116,57 @@ export class OpenAi extends LLMProvider {
 					requestParams.messages = this.toWire(history.filter(h => h.role !== 'system'), options.system);
 
 					const callStart = Date.now();
-					const resp: any = await this.tokenPool.run(token => this.getClient(token).chat.completions.create(requestParams)).catch(err => {
+					const resp: any = await this.tokenPool.run(token =>
+						this.getClient(token).chat.completions.create(requestParams)
+					).catch(err => {
 						err.message += `\n\nMessages:\n${JSON.stringify(requestParams.messages, null, 2)}`;
 						throw err;
 					});
 
-					let usage: any, finishReason: string | undefined, msg: any = {content: '', tool_calls: []};
+					let usage: any;
+					let finishReason: string | undefined;
+					let msg: any = {content: '', tool_calls: []};
 					if(options.stream) {
-						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;
-							if(chunk.choices[0]?.delta?.content) {
-								msg.content += chunk.choices[0].delta.content;
-								options.stream({text: chunk.choices[0].delta.content});
-							}
-							if(chunk.choices[0]?.delta?.tool_calls) {
-								for(const deltaTC of chunk.choices[0].delta.tool_calls) {
-									const existing = deltaTC.index != null
-										? msg.tool_calls.find((tc: any) => tc.index === deltaTC.index)
-										: (deltaTC.id ? msg.tool_calls.find((tc: any) => tc.id === deltaTC.id) : undefined);
-									if(existing) {
+						let streamCompleted = false;
+						try {
+							for await (const chunk of resp) {
+								if(controller.signal.aborted) break;
+								if(chunk.usage) usage = chunk.usage;
+								const choice = chunk.choices?.[0];
+								if(choice?.finish_reason) finishReason = choice.finish_reason;
+								if(choice?.delta?.content) {
+									msg.content += choice.delta.content;
+									options.stream({text: choice.delta.content});
+								}
+								if(choice?.delta?.tool_calls) {
+									for(const deltaTC of choice.delta.tool_calls) {
+										const index = deltaTC.index ?? msg.tool_calls.length;
+										let existing = msg.tool_calls.find((tc: any) => tc.index === index);
+										if(!existing) {
+											existing = {index, id: '', function: {name: '', arguments: ''}};
+											msg.tool_calls.push(existing);
+										}
 										if(deltaTC.id) existing.id = deltaTC.id;
 										if(deltaTC.function?.name) existing.function.name = deltaTC.function.name;
 										if(deltaTC.function?.arguments) existing.function.arguments += deltaTC.function.arguments;
-									} else {
-										msg.tool_calls.push({
-											index: deltaTC.index,
-											id: deltaTC.id || '',
-											function: {name: deltaTC.function?.name || '', arguments: deltaTC.function?.arguments || ''}
-										});
 									}
 								}
 							}
+							streamCompleted = true;
+						} catch(err) {
+							if(!controller.signal.aborted) throw err;
 						}
+						if(streamCompleted && !finishReason) finishReason = msg.tool_calls.length ? 'tool_calls' : 'stop';
 					} else {
 						usage = resp.usage;
 						finishReason = resp.choices[0].finish_reason;
 						msg = resp.choices[0].message;
 					}
+
 					const duration = Date.now() - callStart;
-					const tps = usage?.completion_tokens && duration > 0 ? usage.completion_tokens / (duration / 1000) : 0;
+					const tps = usage?.completion_tokens && duration > 0
+						? usage.completion_tokens / (duration / 1000)
+						: 0;
 
 					if(finishReason === 'length' && !controller.signal.aborted) {
 						if(msg.content?.trim()) history.push({role: 'assistant', content: msg.content.trim(), timestamp: Date.now(), duration, tps});
@@ -141,13 +174,12 @@ export class OpenAi extends LLMProvider {
 					}
 
 					if(!finishReason && !controller.signal.aborted) {
-						throw new Error('[OpenAI] Stream ended prematurely - connection likely dropped');
+						throw new Error('[OpenAI] Completion ended without a usable response');
 					}
 
 					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);
@@ -157,7 +189,7 @@ export class OpenAi extends LLMProvider {
 						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; }
+							if(!tool) return entry.error = 'Tool not found';
 							try {
 								const toolStream = options.stream && ((chunk: any) => {
 									if(chunk.done) return;
@@ -177,7 +209,6 @@ export class OpenAi extends LLMProvider {
 				} while(!terminal && !controller.signal.aborted);
 
 				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();
 				res(options.schema ? JSONAttemptParse(finalContent, finalContent) : finalContent);