Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
28 kB · 1036 lines
TypeScript
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037/** * Agent loop that works with AgentMessage throughout. * Transforms to Message[] only at the LLM call boundary. */
import { type AssistantMessage, type AssistantMessageEvent, type Context, EventStream, type Model, streamSimple, type ToolResultMessage, validateToolArguments,} from "@earendil-works/pi-ai";import type { AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, AgentTool, AgentToolCall, AgentToolResult, StreamFn,} from "./types.js";
export type AgentEventSink = (event: AgentEvent) => Promise<void> | void;
const ABORT_ERROR_MESSAGE = "Request was aborted";const TOOL_UPDATE_MIN_INTERVAL_MS = 50;const EMPTY_USAGE: AssistantMessage["usage"] = { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },};
class ToolUpdateRelay { private accepting = true; private latest: AgentToolResult<unknown> | undefined; private inFlight: Promise<void> | undefined; private failure: { error: unknown } | undefined; private lastEmitAtMs = Number.NEGATIVE_INFINITY;
constructor(private readonly emitUpdate: (partialResult: AgentToolResult<unknown>) => Promise<void> | void) {}
publish(partialResult: AgentToolResult<unknown>): void { if (!this.accepting) { return; } this.latest = partialResult; this.emitLatest(false); }
close(): void { this.accepting = false; }
async flush(): Promise<void> { await this.inFlight; this.throwIfFailed(); this.emitLatest(true); await this.inFlight; this.throwIfFailed(); }
private emitLatest(force: boolean): void { if (this.inFlight || this.failure || !this.latest) { return; } const nowMs = performance.now(); if (!force && nowMs - this.lastEmitAtMs < TOOL_UPDATE_MIN_INTERVAL_MS) { return; }
const partialResult = this.latest; this.latest = undefined; this.lastEmitAtMs = nowMs; let emission: Promise<void> | void; try { emission = this.emitUpdate(partialResult); } catch (error) { this.failure = { error }; throw error; } this.inFlight = Promise.resolve(emission).then( () => { this.inFlight = undefined; }, (error: unknown) => { this.failure = { error }; this.inFlight = undefined; }, ); }
private throwIfFailed(): void { if (this.failure) { throw this.failure.error; } }}
function createAbortError(): Error { return new Error(ABORT_ERROR_MESSAGE);}
function throwIfAborted(signal: AbortSignal | undefined): void { if (signal?.aborted) { throw createAbortError(); }}
function raceWithAbort<T>(operation: Promise<T>, signal: AbortSignal | undefined, onAbort?: () => void): Promise<T> { if (!signal) { return operation; } if (signal.aborted) { onAbort?.(); void operation.catch(() => undefined); return Promise.reject(createAbortError()); }
return new Promise<T>((resolve, reject) => { let settled = false; const cleanup = () => { signal.removeEventListener("abort", abort); }; const abort = () => { if (settled) { return; } settled = true; cleanup(); onAbort?.(); reject(createAbortError()); }; signal.addEventListener("abort", abort, { once: true }); operation.then( (value) => { if (settled) { return; } settled = true; cleanup(); resolve(value); }, (error: unknown) => { if (settled) { return; } settled = true; cleanup(); reject(error); }, ); });}
function maybePromiseWithAbort<T>( operation: T | Promise<T>, signal: AbortSignal | undefined, onAbort?: () => void,): Promise<T> { return raceWithAbort(Promise.resolve(operation), signal, onAbort);}
function isAbortError(error: unknown): boolean { return error instanceof Error && (error.message === ABORT_ERROR_MESSAGE || error.name === "AbortError");}
type PostTurnResult<T> = { status: "completed"; value: T } | { status: "aborted" };
async function settlePostTurn<T>(operation: Promise<T>, signal: AbortSignal | undefined): Promise<PostTurnResult<T>> { try { return { status: "completed", value: await operation }; } catch (error) { if (signal?.aborted && isAbortError(error)) { return { status: "aborted" }; } throw error; }}
function cloneAssistantContent(content: AssistantMessage["content"]): AssistantMessage["content"] { return content.map((part) => { if (part.type === "toolCall") { return { ...part, arguments: { ...part.arguments } }; } return { ...part }; });}
function cloneUsage(usage: AssistantMessage["usage"]): AssistantMessage["usage"] { return { ...usage, cost: { ...usage.cost } };}
/** * The model for this request. A loop that resolves its model per call can be * retargeted between turns; the snapshot stays the fallback. */function resolveRequestModel(config: AgentLoopConfig): Model<any> { return config.getModel?.() ?? config.model;}
function createAbortedAssistantMessage(model: Model<any>, partialMessage: AssistantMessage | null): AssistantMessage { return { role: "assistant", content: partialMessage ? cloneAssistantContent(partialMessage.content) : [{ type: "text", text: "" }], api: partialMessage?.api ?? model.api, provider: partialMessage?.provider ?? model.provider, model: partialMessage?.model ?? model.id, usage: cloneUsage(partialMessage?.usage ?? EMPTY_USAGE), stopReason: "aborted", errorMessage: ABORT_ERROR_MESSAGE, timestamp: Date.now(), };}
function getTerminalMessage(event: Extract<AssistantMessageEvent, { type: "done" | "error" }>): AssistantMessage { return event.type === "done" ? event.message : event.error;}
function endAgentStreamOnError( stream: EventStream<AgentEvent, AgentMessage[]>, promise: Promise<AgentMessage[]>,): void { void promise.then( (messages) => { stream.end(messages); }, () => { stream.end([]); }, );}
async function pollMessagesUnlessAborted( poll: (() => AgentMessage[] | Promise<AgentMessage[]>) | undefined, signal: AbortSignal | undefined,): Promise<AgentMessage[]> { if (!poll || signal?.aborted) { return []; } return (await maybePromiseWithAbort(poll(), signal)) || [];}
/** * Start an agent loop with a new prompt message. * The prompt is added to the context and events are emitted for it. */export function agentLoop( prompts: AgentMessage[], context: AgentContext, config: AgentLoopConfig, signal?: AbortSignal, streamFn?: StreamFn,): EventStream<AgentEvent, AgentMessage[]> { const stream = createAgentStream();
endAgentStreamOnError( stream, runAgentLoop( prompts, context, config, async (event) => { stream.push(event); }, signal, streamFn, ), );
return stream;}
/** * Continue an agent loop from the current context without adding a new message. * Used for retries - context already has user message or tool results. * * **Important:** The last message in context must convert to a `user` or `toolResult` message * via `convertToLlm`. If it doesn't, the LLM provider will reject the request. * This cannot be validated here since `convertToLlm` is only called once per turn. */export function agentLoopContinue( context: AgentContext, config: AgentLoopConfig, signal?: AbortSignal, streamFn?: StreamFn,): EventStream<AgentEvent, AgentMessage[]> { if (context.messages.length === 0) { throw new Error("Cannot continue: no messages in context"); }
if (context.messages[context.messages.length - 1].role === "assistant") { throw new Error("Cannot continue from message role: assistant"); }
const stream = createAgentStream();
endAgentStreamOnError( stream, runAgentLoopContinue( context, config, async (event) => { stream.push(event); }, signal, streamFn, ), );
return stream;}
export async function runAgentLoop( prompts: AgentMessage[], context: AgentContext, config: AgentLoopConfig, emit: AgentEventSink, signal?: AbortSignal, streamFn?: StreamFn,): Promise<AgentMessage[]> { const newMessages: AgentMessage[] = [...prompts]; const currentContext: AgentContext = { ...context, messages: [...context.messages, ...prompts], };
await emit({ type: "agent_start" }); await emit({ type: "turn_start" }); for (const prompt of prompts) { await emit({ type: "message_start", message: prompt }); await emit({ type: "message_end", message: prompt }); }
await runLoop(currentContext, newMessages, config, signal, emit, streamFn); return newMessages;}
export async function runAgentLoopContinue( context: AgentContext, config: AgentLoopConfig, emit: AgentEventSink, signal?: AbortSignal, streamFn?: StreamFn,): Promise<AgentMessage[]> { if (context.messages.length === 0) { throw new Error("Cannot continue: no messages in context"); }
if (context.messages[context.messages.length - 1].role === "assistant") { throw new Error("Cannot continue from message role: assistant"); }
const newMessages: AgentMessage[] = []; const currentContext: AgentContext = { ...context };
await emit({ type: "agent_start" }); await emit({ type: "turn_start" });
await runLoop(currentContext, newMessages, config, signal, emit, streamFn); return newMessages;}
function createAgentStream(): EventStream<AgentEvent, AgentMessage[]> { return new EventStream<AgentEvent, AgentMessage[]>( (event: AgentEvent) => event.type === "agent_end", (event: AgentEvent) => (event.type === "agent_end" ? event.messages : []), );}
async function runLoop( currentContext: AgentContext, newMessages: AgentMessage[], config: AgentLoopConfig, signal: AbortSignal | undefined, emit: AgentEventSink, streamFn?: StreamFn,): Promise<void> { let firstTurn = true; let lastTurn: Parameters<NonNullable<AgentLoopConfig["getContinuationMessages"]>>[0] | undefined; let pendingMessages: AgentMessage[] = await pollMessagesUnlessAborted(config.getSteeringMessages, signal);
const shouldStopBeforeTurn = (): boolean => !firstTurn && (config.shouldStopBeforeTurn?.() ?? false);
while (true) { throwIfAborted(signal); let hasMoreToolCalls = true;
while (hasMoreToolCalls || pendingMessages.length > 0) { throwIfAborted(signal); if (!firstTurn) { await emit({ type: "turn_start" }); } else { firstTurn = false; }
if (pendingMessages.length > 0) { for (const message of pendingMessages) { await emit({ type: "message_start", message }); await emit({ type: "message_end", message }); currentContext.messages.push(message); newMessages.push(message); } pendingMessages = []; }
const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn); newMessages.push(message);
if (message.stopReason === "error" || message.stopReason === "aborted") { await emit({ type: "turn_end", message, toolResults: [] }); await emit({ type: "agent_end", messages: newMessages }); return; }
const toolCalls = message.content.filter((c) => c.type === "toolCall");
const toolResults: ToolResultMessage[] = []; hasMoreToolCalls = false; if (toolCalls.length > 0) { const executedToolBatch = await executeToolCalls(currentContext, message, config, signal, emit); toolResults.push(...executedToolBatch.messages); hasMoreToolCalls = !executedToolBatch.terminate;
for (const result of toolResults) { currentContext.messages.push(result); newMessages.push(result); } }
await emit({ type: "turn_end", message, toolResults }); if (signal?.aborted) { await emit({ type: "agent_end", messages: newMessages }); return; } lastTurn = { message, toolResults, context: currentContext, newMessages, };
const shouldStopResult = await settlePostTurn( maybePromiseWithAbort( config.shouldStopAfterTurn?.({ message, toolResults, context: currentContext, newMessages, }) ?? false, signal, ), signal, ); if (shouldStopResult.status === "aborted") { await emit({ type: "agent_end", messages: newMessages }); return; } if (shouldStopResult.value || shouldStopBeforeTurn()) { await emit({ type: "agent_end", messages: newMessages }); return; }
const steeringMessagesResult = await settlePostTurn( pollMessagesUnlessAborted(config.getSteeringMessages, signal), signal, ); if (steeringMessagesResult.status === "aborted") { await emit({ type: "agent_end", messages: newMessages }); return; } pendingMessages = steeringMessagesResult.value; // Steering drained by this poll owns the turn boundary; stop only when it was empty. if (pendingMessages.length === 0 && shouldStopBeforeTurn()) { await emit({ type: "agent_end", messages: newMessages }); return; } }
if (shouldStopBeforeTurn()) break; const followUpMessagesResult = await settlePostTurn( pollMessagesUnlessAborted(config.getFollowUpMessages, signal), signal, ); if (followUpMessagesResult.status === "aborted") { await emit({ type: "agent_end", messages: newMessages }); return; } const followUpMessages = followUpMessagesResult.value; if (followUpMessages.length > 0) { pendingMessages = followUpMessages; continue; }
if (shouldStopBeforeTurn()) break; const continuationMessagesResult = lastTurn ? await settlePostTurn( maybePromiseWithAbort(config.getContinuationMessages?.(lastTurn, signal) ?? [], signal), signal, ) : ({ status: "completed", value: [] } satisfies PostTurnResult<AgentMessage[]>); if (continuationMessagesResult.status === "aborted") { await emit({ type: "agent_end", messages: newMessages }); return; } const continuationMessages = continuationMessagesResult.value || []; if (continuationMessages.length > 0) { pendingMessages = continuationMessages; continue; }
break; }
await emit({ type: "agent_end", messages: newMessages });}
async function streamAssistantResponse( context: AgentContext, config: AgentLoopConfig, signal: AbortSignal | undefined, emit: AgentEventSink, streamFn?: StreamFn,): Promise<AssistantMessage> { const model = resolveRequestModel(config); let partialMessage: AssistantMessage | null = null; let addedPartial = false; const finishAbortedMessage = async () => { const finalMessage = createAbortedAssistantMessage(model, partialMessage); if (addedPartial) { context.messages[context.messages.length - 1] = finalMessage; } else { context.messages.push(finalMessage); await emit({ type: "message_start", message: { ...finalMessage } }); } await emit({ type: "message_end", message: finalMessage }); return finalMessage; };
try { throwIfAborted(signal); let messages = context.messages; if (config.transformContext) { messages = await maybePromiseWithAbort(config.transformContext(messages, signal), signal); }
const llmMessages = await maybePromiseWithAbort(config.convertToLlm(messages), signal);
const streamFunction = streamFn || streamSimple;
const resolvedApiKey = (config.getApiKey ? await maybePromiseWithAbort(config.getApiKey(model.provider), signal) : undefined) || config.apiKey;
const llmContext: Context = { systemPrompt: config.getSystemPrompt?.() ?? context.systemPrompt, messages: llmMessages, tools: context.tools, };
const response = await maybePromiseWithAbort( streamFunction(model, llmContext, { ...config, apiKey: resolvedApiKey, signal, }), signal, ); const iterator = response[Symbol.asyncIterator](); const closeIterator = () => { void Promise.resolve(iterator.return?.()).catch(() => undefined); }; while (true) { const next = await raceWithAbort<IteratorResult<AssistantMessageEvent>>( iterator.next(), signal, closeIterator, ); if (next.done) { break; } const event = next.value; switch (event.type) { case "start": partialMessage = event.partial; context.messages.push(partialMessage); addedPartial = true; await emit({ type: "message_start", message: { ...partialMessage } }); break;
case "text_start": case "text_delta": case "text_end": case "thinking_start": case "thinking_delta": case "thinking_end": case "toolcall_start": case "toolcall_delta": case "toolcall_end": if (partialMessage) { partialMessage = event.partial; context.messages[context.messages.length - 1] = partialMessage; await emit({ type: "message_update", assistantMessageEvent: event, message: { ...partialMessage }, }); } break;
case "done": case "error": { let finalMessage = getTerminalMessage(event); try { finalMessage = await maybePromiseWithAbort(response.result(), signal); } catch (error) { if (!signal?.aborted || !isAbortError(error)) { throw error; } } if (addedPartial) { context.messages[context.messages.length - 1] = finalMessage; } else { context.messages.push(finalMessage); } if (!addedPartial) { await emit({ type: "message_start", message: { ...finalMessage } }); } await emit({ type: "message_end", message: finalMessage }); return finalMessage; } } }
const finalMessage = await maybePromiseWithAbort(response.result(), signal); if (addedPartial) { context.messages[context.messages.length - 1] = finalMessage; } else { context.messages.push(finalMessage); await emit({ type: "message_start", message: { ...finalMessage } }); } await emit({ type: "message_end", message: finalMessage }); return finalMessage; } catch (error) { if (signal?.aborted && isAbortError(error)) { return finishAbortedMessage(); } throw error; }}
async function executeToolCalls( currentContext: AgentContext, assistantMessage: AssistantMessage, config: AgentLoopConfig, signal: AbortSignal | undefined, emit: AgentEventSink,): Promise<ExecutedToolCallBatch> { const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall"); const hasSequentialToolCall = toolCalls.some( (tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential", ); if (config.toolExecution === "sequential" || hasSequentialToolCall) { return executeToolCallsSequential(currentContext, assistantMessage, toolCalls, config, signal, emit); } return executeToolCallsParallel(currentContext, assistantMessage, toolCalls, config, signal, emit);}
type ExecutedToolCallBatch = { messages: ToolResultMessage[]; terminate: boolean;};
async function executeToolCallsSequential( currentContext: AgentContext, assistantMessage: AssistantMessage, toolCalls: AgentToolCall[], config: AgentLoopConfig, signal: AbortSignal | undefined, emit: AgentEventSink,): Promise<ExecutedToolCallBatch> { const finalizedCalls: FinalizedToolCallOutcome[] = []; const messages: ToolResultMessage[] = [];
for (const toolCall of toolCalls) { if (signal?.aborted) { break; }
const startedAtMs = performance.now(); await emit({ type: "tool_execution_start", toolCallId: toolCall.id, toolName: toolCall.name, args: toolCall.arguments, });
const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal); let finalized: FinalizedToolCallOutcome; if (preparation.kind === "immediate") { finalized = { toolCall, result: preparation.result, isError: preparation.isError, }; } else { const executed = await executePreparedToolCall(preparation, signal, emit); finalized = await finalizeExecutedToolCall( currentContext, assistantMessage, preparation, executed, config, signal, ); }
finalized.durationMs = Math.round(performance.now() - startedAtMs); await emitToolExecutionEnd(finalized, emit); const toolResultMessage = createToolResultMessage(finalized); await emitToolResultMessage(toolResultMessage, emit); finalizedCalls.push(finalized); messages.push(toolResultMessage);
if (signal?.aborted) { break; } }
return { messages, terminate: shouldTerminateToolBatch(finalizedCalls), };}
async function executeToolCallsParallel( currentContext: AgentContext, assistantMessage: AssistantMessage, toolCalls: AgentToolCall[], config: AgentLoopConfig, signal: AbortSignal | undefined, emit: AgentEventSink,): Promise<ExecutedToolCallBatch> { const finalizedCalls: FinalizedToolCallEntry[] = [];
for (const toolCall of toolCalls) { const startedAtMs = performance.now(); await emit({ type: "tool_execution_start", toolCallId: toolCall.id, toolName: toolCall.name, args: toolCall.arguments, });
const preparation = await prepareToolCall(currentContext, assistantMessage, toolCall, config, signal); if (preparation.kind === "immediate") { const finalized = { toolCall, result: preparation.result, isError: preparation.isError, durationMs: Math.round(performance.now() - startedAtMs), } satisfies FinalizedToolCallOutcome; await emitToolExecutionEnd(finalized, emit); finalizedCalls.push(finalized); continue; }
finalizedCalls.push(async () => { const executed = await executePreparedToolCall(preparation, signal, emit); const finalized = await finalizeExecutedToolCall( currentContext, assistantMessage, preparation, executed, config, signal, ); finalized.durationMs = Math.round(performance.now() - startedAtMs); await emitToolExecutionEnd(finalized, emit); return finalized; }); }
const orderedFinalizedCalls = await Promise.all( finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))), ); const messages: ToolResultMessage[] = []; for (const finalized of orderedFinalizedCalls) { const toolResultMessage = createToolResultMessage(finalized); await emitToolResultMessage(toolResultMessage, emit); messages.push(toolResultMessage); }
return { messages, terminate: shouldTerminateToolBatch(orderedFinalizedCalls), };}
type PreparedToolCall = { kind: "prepared"; toolCall: AgentToolCall; tool: AgentTool<any>; args: unknown;};
type ImmediateToolCallOutcome = { kind: "immediate"; result: AgentToolResult<any>; isError: boolean;};
type ExecutedToolCallOutcome = { result: AgentToolResult<any>; isError: boolean;};
type FinalizedToolCallOutcome = { toolCall: AgentToolCall; result: AgentToolResult<any>; isError: boolean; durationMs?: number;};
type FinalizedToolCallEntry = FinalizedToolCallOutcome | (() => Promise<FinalizedToolCallOutcome>);
function shouldTerminateToolBatch(finalizedCalls: FinalizedToolCallOutcome[]): boolean { return finalizedCalls.length > 0 && finalizedCalls.every((finalized) => finalized.result.terminate === true);}
function prepareToolCallArguments(tool: AgentTool<any>, toolCall: AgentToolCall): AgentToolCall { if (!tool.prepareArguments) { return toolCall; } const preparedArguments = tool.prepareArguments(toolCall.arguments); if (preparedArguments === toolCall.arguments) { return toolCall; } return { ...toolCall, arguments: preparedArguments as Record<string, any>, };}
async function prepareToolCall( currentContext: AgentContext, assistantMessage: AssistantMessage, toolCall: AgentToolCall, config: AgentLoopConfig, signal: AbortSignal | undefined,): Promise<PreparedToolCall | ImmediateToolCallOutcome> { const tool = currentContext.tools?.find((t) => t.name === toolCall.name); if (!tool) { return { kind: "immediate", result: createErrorToolResult(`Tool ${toolCall.name} not found`), isError: true, }; }
try { const preparedToolCall = prepareToolCallArguments(tool, toolCall); const validatedArgs = validateToolArguments(tool, preparedToolCall); if (config.beforeToolCall) { const beforeResult = await maybePromiseWithAbort( config.beforeToolCall( { assistantMessage, toolCall, args: validatedArgs, context: currentContext, }, signal, ), signal, ); if (beforeResult?.block) { return { kind: "immediate", result: createErrorToolResult(beforeResult.reason || "Tool execution was blocked"), isError: true, }; } } return { kind: "prepared", toolCall, tool, args: validatedArgs, }; } catch (error) { return { kind: "immediate", result: createErrorToolResult(error instanceof Error ? error.message : String(error)), isError: true, }; }}
async function executePreparedToolCall( prepared: PreparedToolCall, signal: AbortSignal | undefined, emit: AgentEventSink,): Promise<ExecutedToolCallOutcome> { const updates = new ToolUpdateRelay((partialResult) => emit({ type: "tool_execution_update", toolCallId: prepared.toolCall.id, toolName: prepared.toolCall.name, args: prepared.toolCall.arguments, partialResult, }), );
try { throwIfAborted(signal); const result = await raceWithAbort( prepared.tool.execute(prepared.toolCall.id, prepared.args as never, signal, (partialResult) => { if (!signal?.aborted) { updates.publish(partialResult); } }), signal, ); updates.close(); try { await raceWithAbort(updates.flush(), signal); } catch (error) { if (!signal?.aborted || !isAbortError(error)) { throw error; } } // A tool that returns instead of throwing still gets to declare failure; without // this, a tool's own isError never reaches the tool result or the event. return { result, isError: result.isError === true }; } catch (error) { updates.close(); await raceWithAbort(updates.flush(), signal).catch(() => undefined); return { result: createErrorToolResult( signal?.aborted ? "Tool execution aborted" : error instanceof Error ? error.message : String(error), ), isError: true, }; }}
async function finalizeExecutedToolCall( currentContext: AgentContext, assistantMessage: AssistantMessage, prepared: PreparedToolCall, executed: ExecutedToolCallOutcome, config: AgentLoopConfig, signal: AbortSignal | undefined,): Promise<FinalizedToolCallOutcome> { let result = executed.result; let isError = executed.isError;
if (config.afterToolCall) { try { const afterResult = await maybePromiseWithAbort( config.afterToolCall( { assistantMessage, toolCall: prepared.toolCall, args: prepared.args, result, isError, context: currentContext, }, signal, ), signal, ); if (afterResult) { result = { content: afterResult.content ?? result.content, details: afterResult.details ?? result.details, terminate: afterResult.terminate ?? result.terminate, }; isError = afterResult.isError ?? isError; } } catch (error) { result = createErrorToolResult(error instanceof Error ? error.message : String(error)); isError = true; } }
return { toolCall: prepared.toolCall, result, isError, };}
function createErrorToolResult(message: string): AgentToolResult<any> { return { content: [{ type: "text", text: message }], details: {}, };}
async function emitToolExecutionEnd(finalized: FinalizedToolCallOutcome, emit: AgentEventSink): Promise<void> { await emit({ type: "tool_execution_end", toolCallId: finalized.toolCall.id, toolName: finalized.toolCall.name, result: finalized.result, isError: finalized.isError, durationMs: finalized.durationMs, });}
function createToolResultMessage(finalized: FinalizedToolCallOutcome): ToolResultMessage { return { role: "toolResult", toolCallId: finalized.toolCall.id, toolName: finalized.toolCall.name, content: finalized.result.content, details: finalized.result.details, isError: finalized.isError, timestamp: Date.now(), durationMs: finalized.durationMs, };}
async function emitToolResultMessage(toolResultMessage: ToolResultMessage, emit: AgentEventSink): Promise<void> { await emit({ type: "message_start", message: toolResultMessage }); await emit({ type: "message_end", message: toolResultMessage });}