mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-19 16:55:02 +02:00
refactor(core): Split agent runtime lifecycle functions
This commit is contained in:
Robin Braumann
parent
8a4fd13423
commit
f2563da22b
File diff suppressed because it is too large
Load Diff
|
|
@ -39,6 +39,13 @@ function wrapBuiltInRules(fragments: string[]): string | undefined {
|
|||
return `<built_in_rules>\n${fragments.map((f) => `- ${f}`).join('\n')}\n</built_in_rules>`;
|
||||
}
|
||||
|
||||
function prependBuiltInRules(fragments: string[], instructions: string): string {
|
||||
const rules = wrapBuiltInRules(fragments);
|
||||
if (!rules) return instructions;
|
||||
if (!instructions) return rules;
|
||||
return `${rules}\n\n${instructions}`;
|
||||
}
|
||||
|
||||
export interface StaticLoopContext {
|
||||
model: LanguageModel;
|
||||
aiProviderTools: ReturnType<typeof toAiSdkProviderTools>;
|
||||
|
|
@ -68,7 +75,7 @@ export class RuntimeContextBuilder {
|
|||
buildStaticLoopContext(
|
||||
execOptions?: ExecutionOptions & { persistence?: AgentPersistenceOptions },
|
||||
): StaticLoopContext {
|
||||
const { Output, jsonSchema } = loadAi();
|
||||
const ai = loadAi();
|
||||
const aiProviderTools = toAiSdkProviderTools(this.config.providerTools);
|
||||
const model = createModel(this.config.model, this.config.modelFetch);
|
||||
const outputSchema = this.config.structuredOutput;
|
||||
|
|
@ -78,16 +85,7 @@ export class RuntimeContextBuilder {
|
|||
isRawJsonSchemaOutput,
|
||||
);
|
||||
|
||||
const outputSpec = outputSchema
|
||||
? Output.object({
|
||||
// Zod schemas pass through directly; a raw JSON Schema gets
|
||||
// `additionalProperties: false` locked onto every object (required by Anthropic)
|
||||
// and wrapped with the AI SDK's `jsonSchema()` helper.
|
||||
schema: isZodSchema(outputSchema)
|
||||
? outputSchema
|
||||
: jsonSchema(lockAdditionalProperties(outputSchema)),
|
||||
})
|
||||
: undefined;
|
||||
const outputSpec = this.buildOutputSpec(outputSchema, ai);
|
||||
|
||||
return {
|
||||
model,
|
||||
|
|
@ -99,6 +97,18 @@ export class RuntimeContextBuilder {
|
|||
};
|
||||
}
|
||||
|
||||
private buildOutputSpec(
|
||||
outputSchema: AgentRuntimeConfig['structuredOutput'],
|
||||
{ Output, jsonSchema }: ReturnType<typeof loadAi>,
|
||||
): StaticLoopContext['outputSpec'] {
|
||||
if (!outputSchema) return undefined;
|
||||
// Anthropic requires additionalProperties: false on each raw schema object.
|
||||
const schema = isZodSchema(outputSchema)
|
||||
? outputSchema
|
||||
: jsonSchema(lockAdditionalProperties(outputSchema));
|
||||
return Output.object({ schema });
|
||||
}
|
||||
|
||||
/** Build the current local tool view; deferred loads can change this between iterations. */
|
||||
buildToolLoopContext(
|
||||
aiProviderTools: ReturnType<typeof toAiSdkProviderTools>,
|
||||
|
|
@ -260,6 +270,14 @@ export class RuntimeContextBuilder {
|
|||
instructions: string;
|
||||
volatileInstructions: string | undefined;
|
||||
} {
|
||||
const { stableFragments, volatileFragments } = this.collectInstructionFragments(tools);
|
||||
return {
|
||||
instructions: prependBuiltInRules(stableFragments, this.config.instructions),
|
||||
volatileInstructions: wrapBuiltInRules(volatileFragments),
|
||||
};
|
||||
}
|
||||
|
||||
private collectInstructionFragments(tools: BuiltTool[]) {
|
||||
const loadedToolNames = new Set(
|
||||
this.deferredToolManager?.getLoadedTools().map((t) => t.name) ?? [],
|
||||
);
|
||||
|
|
@ -272,9 +290,8 @@ export class RuntimeContextBuilder {
|
|||
) {
|
||||
continue;
|
||||
}
|
||||
(loadedToolNames.has(tool.name) ? volatileFragments : stableFragments).push(
|
||||
tool.systemInstruction,
|
||||
);
|
||||
const fragments = loadedToolNames.has(tool.name) ? volatileFragments : stableFragments;
|
||||
fragments.push(tool.systemInstruction);
|
||||
}
|
||||
|
||||
// Define the untrusted-data boundary ahead of the first wrapped result.
|
||||
|
|
@ -288,18 +305,7 @@ export class RuntimeContextBuilder {
|
|||
target.unshift(UNTRUSTED_OUTPUT_DOCTRINE);
|
||||
}
|
||||
|
||||
const userInstructions = this.config.instructions;
|
||||
const stableBlock = wrapBuiltInRules(stableFragments);
|
||||
const instructions = stableBlock
|
||||
? userInstructions
|
||||
? `${stableBlock}\n\n${userInstructions}`
|
||||
: stableBlock
|
||||
: userInstructions;
|
||||
|
||||
return {
|
||||
instructions,
|
||||
volatileInstructions: wrapBuiltInRules(volatileFragments),
|
||||
};
|
||||
return { stableFragments, volatileFragments };
|
||||
}
|
||||
|
||||
/** Build the providerOptions object for provider-specific thinking config. */
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import type { StreamTextTransform, ToolSet } from 'ai';
|
||||
import type { StreamTextTransform, TextStreamPart, ToolSet } from 'ai';
|
||||
|
||||
import type {
|
||||
CompleteEmission,
|
||||
|
|
@ -160,7 +160,7 @@ export class StreamSink implements RunOutputSink<void> {
|
|||
return transforms.length > 0 ? { experimental_transform: transforms } : {};
|
||||
}
|
||||
|
||||
async callModel(ctx: ModelCallContext): Promise<ModelTurnResult> {
|
||||
private getStreamDeadlines(): { idleMs: number; firstOutputMs: number } {
|
||||
// Clamped to MAX_MODEL_STREAM_TIMEOUT_MS: above it setTimeout fires after
|
||||
// 1ms, turning an "effectively disable" override into instant stalls.
|
||||
const idleMs = Math.min(
|
||||
|
|
@ -177,48 +177,55 @@ export class StreamSink implements RunOutputSink<void> {
|
|||
),
|
||||
MAX_MODEL_STREAM_TIMEOUT_MS,
|
||||
);
|
||||
return { idleMs, firstOutputMs };
|
||||
}
|
||||
|
||||
async callModel(ctx: ModelCallContext): Promise<ModelTurnResult> {
|
||||
const deadlines = this.getStreamDeadlines();
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
// Opt-in: only build the reader when the host bills stopped runs, and only
|
||||
// for providers with a reader. Re-created per attempt so a retried turn
|
||||
// starts from a clean raw capture.
|
||||
this.rawUsageReader = this.options?.recoverUsageOnAbort
|
||||
? createRawUsageReader(this.services.modelId)
|
||||
: undefined;
|
||||
// Some providers report failures (e.g. prompt safety blocks) only on the
|
||||
// raw stream — no error, no content — so raw chunks are required to
|
||||
// explain an otherwise silent empty response.
|
||||
this.rawErrorReader = createRawErrorReader(this.services.modelId);
|
||||
this.resetRawReaders();
|
||||
// Per-attempt controller: a stall must be able to cancel this attempt's
|
||||
// fetch (releasing the socket the 1h network timeout would otherwise
|
||||
// hold) without touching the run-level signal.
|
||||
const turnAbort = new AbortController();
|
||||
const attemptState = { streamedContent: false };
|
||||
try {
|
||||
return await this.streamModelTurn(ctx, turnAbort, { idleMs, firstOutputMs }, attemptState);
|
||||
return await this.streamModelTurn(ctx, turnAbort, deadlines, attemptState);
|
||||
} catch (error) {
|
||||
if (isAttachmentValidationError(error)) {
|
||||
if (!attemptState.streamedContent) await ctx.onInputRejected?.(error);
|
||||
throw error;
|
||||
}
|
||||
// A stalled or empty stream before any content is invisible to the user
|
||||
// (and to the host's persistence) — re-issue the request instead of
|
||||
// failing the run for what is usually a dead connection at request time.
|
||||
// Once content has streamed, a retry would duplicate segments and orphan
|
||||
// already-persisted tool-call facts, so the error surfaces instead.
|
||||
// The abandoned attempt's prompt processing goes unbilled — accepted:
|
||||
// it only has usage at all when the stream died between message_start
|
||||
// and the first content chunk.
|
||||
const retryable =
|
||||
(error instanceof ModelStreamStallError ||
|
||||
loadAi().NoOutputGeneratedError.isInstance(error)) &&
|
||||
attempt < MAX_MODEL_STREAM_STALL_RETRIES &&
|
||||
!attemptState.streamedContent &&
|
||||
!ctx.abortSignal.aborted;
|
||||
if (!retryable) throw error;
|
||||
if (!this.canRetryStream(error, attempt, attemptState, ctx.abortSignal)) throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private resetRawReaders(): void {
|
||||
// Each attempt needs fresh readers so a retry does not reuse captured usage.
|
||||
this.rawUsageReader = this.options?.recoverUsageOnAbort
|
||||
? createRawUsageReader(this.services.modelId)
|
||||
: undefined;
|
||||
// Some providers report a blocked prompt only in raw chunks.
|
||||
this.rawErrorReader = createRawErrorReader(this.services.modelId);
|
||||
}
|
||||
|
||||
private canRetryStream(
|
||||
error: unknown,
|
||||
attempt: number,
|
||||
attemptState: { streamedContent: boolean },
|
||||
abortSignal: AbortSignal,
|
||||
): boolean {
|
||||
// Retry only before content is visible or persisted. The abandoned attempt stays unbilled.
|
||||
return (
|
||||
(error instanceof ModelStreamStallError ||
|
||||
loadAi().NoOutputGeneratedError.isInstance(error)) &&
|
||||
attempt < MAX_MODEL_STREAM_STALL_RETRIES &&
|
||||
!attemptState.streamedContent &&
|
||||
!abortSignal.aborted
|
||||
);
|
||||
}
|
||||
|
||||
private async streamModelTurn(
|
||||
ctx: ModelCallContext,
|
||||
turnAbort: AbortController,
|
||||
|
|
@ -227,6 +234,36 @@ export class StreamSink implements RunOutputSink<void> {
|
|||
): Promise<ModelTurnResult> {
|
||||
const { idleMs, firstOutputMs } = deadlines;
|
||||
const { NoOutputGeneratedError, streamText } = loadAi();
|
||||
const { result, activity } = this.startModelStream(ctx, turnAbort, idleMs, streamText);
|
||||
|
||||
// Use the longer deadline before the first output to allow prompt processing.
|
||||
const chunkStream =
|
||||
idleMs > 0
|
||||
? withChunkIdleTimeout(
|
||||
result.stream,
|
||||
() => (attemptState.streamedContent ? idleMs : firstOutputMs),
|
||||
() => turnAbort.abort(),
|
||||
() => activity.lastAt,
|
||||
)
|
||||
: result.stream;
|
||||
|
||||
for await (const chunk of chunkStream) {
|
||||
if (chunk.type === 'error' && isAttachmentValidationError(chunk.error)) throw chunk.error;
|
||||
// Result promises report this error. Do not forward it or count it as content before a retry.
|
||||
if (chunk.type === 'error' && NoOutputGeneratedError.isInstance(chunk.error)) continue;
|
||||
if (!this.captureStreamChunk(chunk, attemptState)) continue;
|
||||
await this.writeStreamChunk(chunk);
|
||||
}
|
||||
|
||||
return await this.resolveModelTurn(result, ctx, idleMs, turnAbort);
|
||||
}
|
||||
|
||||
private startModelStream(
|
||||
ctx: ModelCallContext,
|
||||
turnAbort: AbortController,
|
||||
idleMs: number,
|
||||
streamText: ReturnType<typeof loadAi>['streamText'],
|
||||
) {
|
||||
// Raw chunks serve two consumers: the usage/error readers parse them, and
|
||||
// the stall watchdog counts them as liveness — keepalive events the SDK
|
||||
// otherwise drops (e.g. Anthropic `ping`, a gateway's empty-delta frames)
|
||||
|
|
@ -264,83 +301,58 @@ export class StreamSink implements RunOutputSink<void> {
|
|||
: undefined,
|
||||
),
|
||||
});
|
||||
return { result, activity };
|
||||
}
|
||||
|
||||
// A healthy streaming response emits chunks continuously (raw provider
|
||||
// events included), so prolonged silence means the connection or provider
|
||||
// is wedged — fail the turn instead of hanging until the network timeout.
|
||||
// Pre-first-output silence gets the longer deadline (prompt processing on
|
||||
// large cache-miss prompts sends nothing for minutes); once content has
|
||||
// streamed, the tighter idle limit applies.
|
||||
const chunkStream =
|
||||
idleMs > 0
|
||||
? withChunkIdleTimeout(
|
||||
result.stream,
|
||||
() => (attemptState.streamedContent ? idleMs : firstOutputMs),
|
||||
() => turnAbort.abort(),
|
||||
() => activity.lastAt,
|
||||
)
|
||||
: result.stream;
|
||||
private captureStreamChunk(
|
||||
chunk: TextStreamPart<ToolSet>,
|
||||
attemptState: { streamedContent: boolean },
|
||||
): boolean {
|
||||
// Capture raw chunks here when a stream does not use the tap, such as a test stream.
|
||||
if (chunk.type === 'raw') {
|
||||
this.rawUsageReader?.capture(chunk.rawValue);
|
||||
this.rawErrorReader?.capture(chunk.rawValue);
|
||||
return false;
|
||||
}
|
||||
if (!STALL_RETRY_SAFE_CHUNK_TYPES.has(chunk.type)) attemptState.streamedContent = true;
|
||||
// The runtime writes its own finish chunk after the loop completes.
|
||||
if (chunk.type === 'finish') return false;
|
||||
// Retain partial text so cancellation can save an unfinished response.
|
||||
if (chunk.type === 'text-delta') this.partialText += chunk.text ?? '';
|
||||
return true;
|
||||
}
|
||||
|
||||
// Consume the stream. When the AbortSignal fires mid-stream the AI SDK
|
||||
// cancels the underlying fetch and the async iterator throws; the error
|
||||
// propagates to the StreamSession which closes the consumer stream.
|
||||
for await (const chunk of chunkStream) {
|
||||
if (chunk.type === 'error' && isAttachmentValidationError(chunk.error)) throw chunk.error;
|
||||
// The rejected result promises surface this error after the stream
|
||||
// closes. Forwarding it here would mark a successful retry as failed.
|
||||
// Runs before the `streamedContent` check below so a skipped chunk
|
||||
// cannot make a stalled attempt look unretryable.
|
||||
if (chunk.type === 'error' && NoOutputGeneratedError.isInstance(chunk.error)) continue;
|
||||
// Raw chunks are normally consumed by the raw-chunk tap inside the SDK
|
||||
// pipeline and never reach this loop; this branch is the fallback for
|
||||
// streams the tap was not applied to (e.g. mocked streams in tests) and
|
||||
// is why 'raw' stays in the retry-safe set. Capturing usage here also
|
||||
// lets an aborted turn still be billed via getTerminalFinish.
|
||||
if (chunk.type === 'raw') {
|
||||
this.rawUsageReader?.capture(chunk.rawValue);
|
||||
this.rawErrorReader?.capture(chunk.rawValue);
|
||||
continue;
|
||||
}
|
||||
// Anything beyond transport bookkeeping counts as content: once seen,
|
||||
// a stalled attempt is no longer silently retryable (see callModel).
|
||||
if (!STALL_RETRY_SAFE_CHUNK_TYPES.has(chunk.type)) attemptState.streamedContent = true;
|
||||
// Filter only the SDK's terminal `finish` chunk — the runtime emits its
|
||||
// own consolidated `finish` after the loop completes. `start-step` /
|
||||
// `finish-step` are passed through as LLM-iteration boundaries.
|
||||
if (chunk.type === 'finish') continue;
|
||||
|
||||
// Accumulate streamed text so an abort mid-response can persist it (the
|
||||
// turn's `newMessages` are only built below, which the abort never reaches).
|
||||
if (chunk.type === 'text-delta') this.partialText += chunk.text ?? '';
|
||||
|
||||
// Provider-executed tools (e.g. native web search) skip the local
|
||||
// execution loop that emits tool-execution lifecycle events via the
|
||||
// event bus. Stamp them here at chunk-arrival time so live chat and the
|
||||
// persisted timeline both show a duration. A failed call arrives as a
|
||||
// `tool-error` part (never a `tool-result`), so close its timing there.
|
||||
if ((chunk.type === 'tool-result' || chunk.type === 'tool-error') && chunk.providerExecuted) {
|
||||
await this.guard.write({
|
||||
type: 'tool-execution-end',
|
||||
toolCallId: chunk.toolCallId,
|
||||
toolName: chunk.toolName ?? '',
|
||||
isError: chunk.type === 'tool-error',
|
||||
endTime: Date.now(),
|
||||
});
|
||||
}
|
||||
|
||||
const converted = convertChunk(chunk);
|
||||
if (converted) await this.guard.write(converted);
|
||||
|
||||
if (chunk.type === 'tool-call' && chunk.providerExecuted) {
|
||||
await this.guard.write({
|
||||
type: 'tool-execution-start',
|
||||
toolCallId: chunk.toolCallId,
|
||||
toolName: chunk.toolName ?? '',
|
||||
startTime: Date.now(),
|
||||
});
|
||||
}
|
||||
private async writeStreamChunk(chunk: TextStreamPart<ToolSet>): Promise<void> {
|
||||
// Provider tools skip the local executor. Record their timing when chunks arrive.
|
||||
if ((chunk.type === 'tool-result' || chunk.type === 'tool-error') && chunk.providerExecuted) {
|
||||
await this.guard.write({
|
||||
type: 'tool-execution-end',
|
||||
toolCallId: chunk.toolCallId,
|
||||
toolName: chunk.toolName ?? '',
|
||||
isError: chunk.type === 'tool-error',
|
||||
endTime: Date.now(),
|
||||
});
|
||||
}
|
||||
|
||||
const converted = convertChunk(chunk);
|
||||
if (converted) await this.guard.write(converted);
|
||||
|
||||
if (chunk.type === 'tool-call' && chunk.providerExecuted) {
|
||||
await this.guard.write({
|
||||
type: 'tool-execution-start',
|
||||
toolCallId: chunk.toolCallId,
|
||||
toolName: chunk.toolName ?? '',
|
||||
startTime: Date.now(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async resolveModelTurn(
|
||||
result: ReturnType<StreamSink['startModelStream']>['result'],
|
||||
ctx: ModelCallContext,
|
||||
idleMs: number,
|
||||
turnAbort: AbortController,
|
||||
): Promise<ModelTurnResult> {
|
||||
// The result promises settle as part of stream close on a healthy turn;
|
||||
// guard them with the same stall deadline so an SDK-internal promise that
|
||||
// never settles cannot hang the turn after the chunk loop ended.
|
||||
|
|
|
|||
|
|
@ -447,9 +447,25 @@ function resolveDelegateSubAgentSystemInstruction(
|
|||
* }));
|
||||
*/
|
||||
export function createDelegateSubAgentTool(options: CreateDelegateSubAgentToolOptions = {}) {
|
||||
// Per-parent child path index for stable task paths (/root/name_0, /root/name_1, ...).
|
||||
// Each parent has its own index so child task paths stay stable.
|
||||
const childPathIndexes = new Map<string, number>();
|
||||
const toolName = resolveDelegateSubAgentToolName(options.name);
|
||||
const resolvedOptions = resolveDelegateSubAgentOptions(options, toolName);
|
||||
const tool = buildDelegateSubAgentTool(toolName, resolvedOptions, childPathIndexes);
|
||||
return withSdkOwnedBuiltInMetadata({
|
||||
...tool,
|
||||
...createDelegateCancellationHandler(resolvedOptions, childPathIndexes),
|
||||
metadata: {
|
||||
...tool.metadata,
|
||||
[INLINE_DELEGATE_SUB_AGENT_TOOL_METADATA_KEY]: buildDelegateToolMetadata(resolvedOptions),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function resolveDelegateSubAgentOptions(
|
||||
options: CreateDelegateSubAgentToolOptions,
|
||||
toolName: string,
|
||||
): CreateDelegateSubAgentToolOptions {
|
||||
const resolvedOptions: CreateDelegateSubAgentToolOptions = {
|
||||
...options,
|
||||
policy: resolveDelegateSubAgentPolicy(options.policy, toolName),
|
||||
|
|
@ -462,6 +478,14 @@ export function createDelegateSubAgentTool(options: CreateDelegateSubAgentToolOp
|
|||
`${toolName} requires resumeSubAgent and cancelSubAgent to be configured together`,
|
||||
);
|
||||
}
|
||||
return resolvedOptions;
|
||||
}
|
||||
|
||||
function buildDelegateSubAgentTool(
|
||||
toolName: string,
|
||||
resolvedOptions: CreateDelegateSubAgentToolOptions,
|
||||
childPathIndexes: Map<string, number>,
|
||||
) {
|
||||
const inlineProviderToolInstruction = resolvedOptions.resolveInlineSubAgentProviderTools
|
||||
? "Provider-defined tools are loaded for the inline child's selected model provider."
|
||||
: 'Inline children do not inherit provider-defined tools.';
|
||||
|
|
@ -483,73 +507,77 @@ export function createDelegateSubAgentTool(options: CreateDelegateSubAgentToolOp
|
|||
) => await handleDelegateSubAgent(input, ctx, resolvedOptions, childPathIndexes);
|
||||
const toModelOutput = (output: z.infer<typeof delegateSubAgentOutputSchema>) =>
|
||||
resolvedOptions.toModelOutput ? resolvedOptions.toModelOutput(output) : output;
|
||||
const tool = resolvedOptions.resumeSubAgent
|
||||
? toolBuilder
|
||||
.suspend(delegateSubAgentSuspendSchema)
|
||||
.resume(delegateSubAgentResumeSchema)
|
||||
.handler(handler)
|
||||
.toModelOutput(toModelOutput)
|
||||
.build()
|
||||
: toolBuilder.handler(handler).toModelOutput(toModelOutput).build();
|
||||
return withSdkOwnedBuiltInMetadata({
|
||||
...tool,
|
||||
...(resolvedOptions.cancelSubAgent !== undefined
|
||||
if (resolvedOptions.resumeSubAgent) {
|
||||
return toolBuilder
|
||||
.suspend(delegateSubAgentSuspendSchema)
|
||||
.resume(delegateSubAgentResumeSchema)
|
||||
.handler(handler)
|
||||
.toModelOutput(toModelOutput)
|
||||
.build();
|
||||
}
|
||||
return toolBuilder.handler(handler).toModelOutput(toModelOutput).build();
|
||||
}
|
||||
|
||||
function createDelegateCancellationHandler(
|
||||
options: CreateDelegateSubAgentToolOptions,
|
||||
childPathIndexes: Map<string, number>,
|
||||
): Pick<BuiltTool, 'onCancellation'> {
|
||||
if (options.cancelSubAgent === undefined) return {};
|
||||
return {
|
||||
onCancellation: async (rawInput: unknown, ctx: ToolCancellationContext) => {
|
||||
const parsedInput = delegateSubAgentInputSchema.safeParse(rawInput);
|
||||
if (!parsedInput.success) {
|
||||
throw new Error('Delegated child input is missing or invalid');
|
||||
}
|
||||
await cancelDelegatedSubAgent(parsedInput.data, ctx, options, childPathIndexes);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function buildDelegateToolMetadata(
|
||||
resolvedOptions: CreateDelegateSubAgentToolOptions,
|
||||
): DelegateSubAgentToolMetadata {
|
||||
return {
|
||||
...(resolvedOptions.name !== undefined ? { name: resolvedOptions.name } : {}),
|
||||
...(resolvedOptions.availableSubAgents !== undefined
|
||||
? { availableSubAgents: resolvedOptions.availableSubAgents }
|
||||
: {}),
|
||||
policy: resolvedOptions.policy,
|
||||
...(resolvedOptions.inlineSubAgentBlockedTools !== undefined
|
||||
? { inlineSubAgentBlockedTools: resolvedOptions.inlineSubAgentBlockedTools }
|
||||
: {}),
|
||||
...(resolvedOptions.inlineSubAgentModelsByDifficulty !== undefined
|
||||
? {
|
||||
onCancellation: async (rawInput: unknown, ctx: ToolCancellationContext) => {
|
||||
const parsedInput = delegateSubAgentInputSchema.safeParse(rawInput);
|
||||
if (!parsedInput.success) {
|
||||
throw new Error('Delegated child input is missing or invalid');
|
||||
}
|
||||
await cancelDelegatedSubAgent(parsedInput.data, ctx, resolvedOptions, childPathIndexes);
|
||||
},
|
||||
inlineSubAgentModelsByDifficulty: resolvedOptions.inlineSubAgentModelsByDifficulty,
|
||||
}
|
||||
: {}),
|
||||
metadata: {
|
||||
...tool.metadata,
|
||||
[INLINE_DELEGATE_SUB_AGENT_TOOL_METADATA_KEY]: {
|
||||
...(resolvedOptions.name !== undefined ? { name: resolvedOptions.name } : {}),
|
||||
...(resolvedOptions.availableSubAgents !== undefined
|
||||
? { availableSubAgents: resolvedOptions.availableSubAgents }
|
||||
: {}),
|
||||
policy: resolvedOptions.policy,
|
||||
...(resolvedOptions.inlineSubAgentBlockedTools !== undefined
|
||||
? { inlineSubAgentBlockedTools: resolvedOptions.inlineSubAgentBlockedTools }
|
||||
: {}),
|
||||
...(resolvedOptions.inlineSubAgentModelsByDifficulty !== undefined
|
||||
? {
|
||||
inlineSubAgentModelsByDifficulty: resolvedOptions.inlineSubAgentModelsByDifficulty,
|
||||
}
|
||||
: {}),
|
||||
...(resolvedOptions.resolveInlineSubAgentProviderTools !== undefined
|
||||
? {
|
||||
resolveInlineSubAgentProviderTools:
|
||||
resolvedOptions.resolveInlineSubAgentProviderTools,
|
||||
}
|
||||
: {}),
|
||||
...(resolvedOptions.runSubAgent !== undefined
|
||||
? { runSubAgent: resolvedOptions.runSubAgent }
|
||||
: {}),
|
||||
...(resolvedOptions.resumeSubAgent !== undefined
|
||||
? { resumeSubAgent: resolvedOptions.resumeSubAgent }
|
||||
: {}),
|
||||
...(resolvedOptions.shouldRetrySubAgentResumeError !== undefined
|
||||
? { shouldRetrySubAgentResumeError: resolvedOptions.shouldRetrySubAgentResumeError }
|
||||
: {}),
|
||||
...(resolvedOptions.cancelSubAgent !== undefined
|
||||
? { cancelSubAgent: resolvedOptions.cancelSubAgent }
|
||||
: {}),
|
||||
...(resolvedOptions.systemInstruction !== undefined
|
||||
? { systemInstruction: resolvedOptions.systemInstruction }
|
||||
: {}),
|
||||
...(resolvedOptions.description !== undefined
|
||||
? { description: resolvedOptions.description }
|
||||
: {}),
|
||||
...(resolvedOptions.toModelOutput !== undefined
|
||||
? { toModelOutput: resolvedOptions.toModelOutput }
|
||||
: {}),
|
||||
} satisfies DelegateSubAgentToolMetadata,
|
||||
},
|
||||
});
|
||||
...(resolvedOptions.resolveInlineSubAgentProviderTools !== undefined
|
||||
? {
|
||||
resolveInlineSubAgentProviderTools: resolvedOptions.resolveInlineSubAgentProviderTools,
|
||||
}
|
||||
: {}),
|
||||
...(resolvedOptions.runSubAgent !== undefined
|
||||
? { runSubAgent: resolvedOptions.runSubAgent }
|
||||
: {}),
|
||||
...(resolvedOptions.resumeSubAgent !== undefined
|
||||
? { resumeSubAgent: resolvedOptions.resumeSubAgent }
|
||||
: {}),
|
||||
...(resolvedOptions.shouldRetrySubAgentResumeError !== undefined
|
||||
? { shouldRetrySubAgentResumeError: resolvedOptions.shouldRetrySubAgentResumeError }
|
||||
: {}),
|
||||
...(resolvedOptions.cancelSubAgent !== undefined
|
||||
? { cancelSubAgent: resolvedOptions.cancelSubAgent }
|
||||
: {}),
|
||||
...(resolvedOptions.systemInstruction !== undefined
|
||||
? { systemInstruction: resolvedOptions.systemInstruction }
|
||||
: {}),
|
||||
...(resolvedOptions.description !== undefined
|
||||
? { description: resolvedOptions.description }
|
||||
: {}),
|
||||
...(resolvedOptions.toModelOutput !== undefined
|
||||
? { toModelOutput: resolvedOptions.toModelOutput }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
export function getInlineDelegateSubAgentToolOptions(
|
||||
|
|
@ -632,26 +660,7 @@ async function handleDelegateSubAgent(
|
|||
request = restored.request;
|
||||
startedAt = Date.now();
|
||||
emitSubAgentStarted(ctx, request, startedAt);
|
||||
let output: DelegateSubAgentToolOutput;
|
||||
try {
|
||||
output = await options.resumeSubAgent(
|
||||
{
|
||||
...request,
|
||||
...getChildCheckpointTarget(checkpoint),
|
||||
resumeData: ctx.resumeData,
|
||||
},
|
||||
createRunnerHelpers(ctx, request, options.name),
|
||||
);
|
||||
} catch (error) {
|
||||
if (ctx.abortSignal?.aborted || isAbortError(error)) throw error;
|
||||
if (options.shouldRetrySubAgentResumeError?.(error) === false) throw error;
|
||||
return await ctx.suspend(ctx.suspendPayload);
|
||||
}
|
||||
emitSubAgentCompleted(ctx, request, output, startedAt);
|
||||
if (output.status === 'suspended') {
|
||||
return await cascadeChildSuspension(ctx, request, output);
|
||||
}
|
||||
return output;
|
||||
return await resumeDelegatedChild(ctx, restored, startedAt, options, options.resumeSubAgent);
|
||||
}
|
||||
|
||||
const childPathIndexKey = getChildPathIndexKey(ctx);
|
||||
|
|
@ -664,42 +673,90 @@ async function handleDelegateSubAgent(
|
|||
|
||||
startedAt = Date.now();
|
||||
emitSubAgentStarted(ctx, request, startedAt);
|
||||
const toolName = options.name ?? DELEGATE_SUB_AGENT_TOOL_NAME;
|
||||
if (!options.runSubAgent) {
|
||||
throw new Error(
|
||||
`${toolName} was registered without a runSubAgent callback, and no host runner was provided. Register it on an Agent (for inline delegation) or pass runSubAgent.`,
|
||||
);
|
||||
}
|
||||
const output = await options.runSubAgent(
|
||||
request,
|
||||
return await runDelegatedChild(ctx, request, startedAt, options);
|
||||
} catch (error) {
|
||||
return handleDelegateFailure(error, ctx, taskPath, request, startedAt);
|
||||
}
|
||||
}
|
||||
|
||||
async function resumeDelegatedChild(
|
||||
ctx: InterruptibleToolContext,
|
||||
{ request, checkpoint }: ReturnType<typeof restoreDelegateRequest>,
|
||||
startedAt: number,
|
||||
options: CreateDelegateSubAgentToolOptions,
|
||||
resumeSubAgent: DelegateSubAgentResumeRunner,
|
||||
): Promise<DelegateSubAgentToolOutput> {
|
||||
let output: DelegateSubAgentToolOutput;
|
||||
try {
|
||||
output = await resumeSubAgent(
|
||||
{
|
||||
...request,
|
||||
...getChildCheckpointTarget(checkpoint),
|
||||
resumeData: ctx.resumeData,
|
||||
},
|
||||
createRunnerHelpers(ctx, request, options.name),
|
||||
);
|
||||
emitSubAgentCompleted(ctx, request, output, startedAt);
|
||||
if (
|
||||
output.status === 'suspended' &&
|
||||
options.resumeSubAgent !== undefined &&
|
||||
isInterruptibleToolContext(ctx)
|
||||
) {
|
||||
return await cascadeChildSuspension(ctx, request, output);
|
||||
}
|
||||
return output;
|
||||
} catch (error) {
|
||||
// When the parent has a signal it is the authority: `isAbortError` also
|
||||
// matches by message text, and an unrelated child error must not be
|
||||
// mistaken for a cancellation and kill the parent run.
|
||||
const aborted = ctx.abortSignal ? ctx.abortSignal.aborted : isAbortError(error);
|
||||
const output: DelegateSubAgentToolOutput = {
|
||||
status: aborted ? 'cancelled' : 'failed',
|
||||
...(taskPath !== undefined ? { taskPath } : {}),
|
||||
answer: '',
|
||||
...(aborted ? {} : { error: stringifyUnknown(error) }),
|
||||
};
|
||||
if (request !== undefined && startedAt !== undefined) {
|
||||
emitSubAgentCompleted(ctx, request, output, startedAt);
|
||||
}
|
||||
if (aborted) throw error;
|
||||
return output;
|
||||
if (ctx.abortSignal?.aborted || isAbortError(error)) throw error;
|
||||
if (options.shouldRetrySubAgentResumeError?.(error) === false) throw error;
|
||||
return await ctx.suspend(ctx.suspendPayload);
|
||||
}
|
||||
emitSubAgentCompleted(ctx, request, output, startedAt);
|
||||
if (output.status === 'suspended') {
|
||||
return await cascadeChildSuspension(ctx, request, output);
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
async function runDelegatedChild(
|
||||
ctx: ToolContext | InterruptibleToolContext,
|
||||
request: DelegateSubAgentRequest,
|
||||
startedAt: number,
|
||||
options: CreateDelegateSubAgentToolOptions,
|
||||
): Promise<DelegateSubAgentToolOutput> {
|
||||
const toolName = options.name ?? DELEGATE_SUB_AGENT_TOOL_NAME;
|
||||
if (!options.runSubAgent) {
|
||||
throw new Error(
|
||||
`${toolName} was registered without a runSubAgent callback, and no host runner was provided. Register it on an Agent (for inline delegation) or pass runSubAgent.`,
|
||||
);
|
||||
}
|
||||
const output = await options.runSubAgent(
|
||||
request,
|
||||
createRunnerHelpers(ctx, request, options.name),
|
||||
);
|
||||
emitSubAgentCompleted(ctx, request, output, startedAt);
|
||||
if (
|
||||
output.status === 'suspended' &&
|
||||
options.resumeSubAgent !== undefined &&
|
||||
isInterruptibleToolContext(ctx)
|
||||
) {
|
||||
return await cascadeChildSuspension(ctx, request, output);
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
function handleDelegateFailure(
|
||||
error: unknown,
|
||||
ctx: ToolContext,
|
||||
taskPath: SubAgentTaskPath | undefined,
|
||||
request: DelegateSubAgentRequest | undefined,
|
||||
startedAt: number | undefined,
|
||||
): DelegateSubAgentToolOutput {
|
||||
// When the parent has a signal it is the authority: `isAbortError` also
|
||||
// matches by message text, and an unrelated child error must not be
|
||||
// mistaken for a cancellation and kill the parent run.
|
||||
const aborted = ctx.abortSignal ? ctx.abortSignal.aborted : isAbortError(error);
|
||||
const output: DelegateSubAgentToolOutput = {
|
||||
status: aborted ? 'cancelled' : 'failed',
|
||||
...(taskPath !== undefined ? { taskPath } : {}),
|
||||
answer: '',
|
||||
...(aborted ? {} : { error: stringifyUnknown(error) }),
|
||||
};
|
||||
if (request !== undefined && startedAt !== undefined) {
|
||||
emitSubAgentCompleted(ctx, request, output, startedAt);
|
||||
}
|
||||
if (aborted) throw error;
|
||||
return output;
|
||||
}
|
||||
|
||||
function createDelegateSubAgentRequest(
|
||||
|
|
@ -1001,12 +1058,7 @@ export function generateResultToDelegateSubAgentOutput(
|
|||
...(result.structuredOutput !== undefined ? { structuredOutput: result.structuredOutput } : {}),
|
||||
...(result.usage !== undefined
|
||||
? {
|
||||
usage: {
|
||||
promptTokens: result.usage.promptTokens,
|
||||
completionTokens: result.usage.completionTokens,
|
||||
totalTokens: result.usage.totalTokens,
|
||||
...(result.usage.cost !== undefined ? { cost: result.usage.cost } : {}),
|
||||
},
|
||||
usage: toDelegateUsage(result.usage),
|
||||
}
|
||||
: {}),
|
||||
...(result.finishReason !== undefined ? { finishReason: result.finishReason } : {}),
|
||||
|
|
@ -1017,6 +1069,15 @@ export function generateResultToDelegateSubAgentOutput(
|
|||
};
|
||||
}
|
||||
|
||||
function toDelegateUsage(usage: TokenUsage): DelegateSubAgentToolOutput['usage'] {
|
||||
return {
|
||||
promptTokens: usage.promptTokens,
|
||||
completionTokens: usage.completionTokens,
|
||||
totalTokens: usage.totalTokens,
|
||||
...(usage.cost !== undefined ? { cost: usage.cost } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
/** Last non-empty assistant text across the run's messages. */
|
||||
function lastText(messages: AgentMessage[]): string {
|
||||
const llmMessages = filterLlmMessages(messages);
|
||||
|
|
|
|||
|
|
@ -117,42 +117,14 @@ export async function executeTool(
|
|||
throw new Error(`No handler found for tool "${builtTool.name}"`);
|
||||
}
|
||||
|
||||
// Anchor programmatic skill activations to the calling tool's result by
|
||||
// default, so the skill body rides on that result (kept out of `system`)
|
||||
// instead of forcing a full prompt-prefix rewrite. An explicit anchor wins.
|
||||
const baseLoadSkill = executionContext.loadSkill;
|
||||
const loadSkill: ToolExecutionContext['loadSkill'] = baseLoadSkill
|
||||
? async (skillId, anchor) =>
|
||||
await baseLoadSkill(skillId, anchor ?? (toolCallId ? { toolCallId } : undefined))
|
||||
: undefined;
|
||||
|
||||
const loadSkill = anchorSkillLoader(executionContext.loadSkill, toolCallId);
|
||||
if (builtTool.suspendSchema) {
|
||||
const isCancelled = isCancellation(resumeData);
|
||||
const ctx: InterruptibleToolContext = {
|
||||
suspend: async (payload: unknown, options?: ToolSuspendOptions): Promise<never> => {
|
||||
const resolvedOptions: ToolSuspendOptions = {
|
||||
continuation: executionContext.continuation,
|
||||
...options,
|
||||
resumeSchema: options?.resumeSchema ?? executionContext.resumeSchema,
|
||||
};
|
||||
await executionContext.onSuspend?.(payload, resolvedOptions);
|
||||
return await Promise.resolve({
|
||||
[SUSPEND_BRAND]: true,
|
||||
payload,
|
||||
...resolvedOptions,
|
||||
} as never);
|
||||
},
|
||||
suspend: createSuspendHandler(executionContext),
|
||||
resumeData: isCancelled ? undefined : resumeData,
|
||||
cancellation: isCancelled ? { message: resumeData.message } : undefined,
|
||||
parentTelemetry,
|
||||
toolCallId,
|
||||
toolName: builtTool.name,
|
||||
runId: executionContext.runId,
|
||||
...(loadSkill ? { loadSkill } : {}),
|
||||
persistence: executionContext.persistence,
|
||||
emitEvent: executionContext.emitEvent,
|
||||
abortSignal: executionContext.abortSignal,
|
||||
executionCounter: executionContext.executionCounter,
|
||||
...createToolContext(builtTool, executionContext, parentTelemetry, toolCallId, loadSkill),
|
||||
suspendPayload: executionContext.suspendPayload,
|
||||
continuation: executionContext.continuation,
|
||||
resumeSchema: executionContext.resumeSchema,
|
||||
|
|
@ -160,7 +132,34 @@ export async function executeTool(
|
|||
return await builtTool.handler(args, ctx);
|
||||
}
|
||||
|
||||
const ctx: ToolContext = {
|
||||
const ctx = createToolContext(
|
||||
builtTool,
|
||||
executionContext,
|
||||
parentTelemetry,
|
||||
toolCallId,
|
||||
loadSkill,
|
||||
);
|
||||
return await builtTool.handler(args, ctx);
|
||||
}
|
||||
|
||||
function anchorSkillLoader(
|
||||
loadSkill: ToolExecutionContext['loadSkill'],
|
||||
toolCallId: string | undefined,
|
||||
): ToolExecutionContext['loadSkill'] {
|
||||
if (!loadSkill) return undefined;
|
||||
// Keep skill content on the tool result. An explicit anchor takes precedence.
|
||||
return async (skillId, anchor) =>
|
||||
await loadSkill(skillId, anchor ?? (toolCallId ? { toolCallId } : undefined));
|
||||
}
|
||||
|
||||
function createToolContext(
|
||||
builtTool: BuiltTool,
|
||||
executionContext: ToolExecutionContext,
|
||||
parentTelemetry: BuiltTelemetry | undefined,
|
||||
toolCallId: string | undefined,
|
||||
loadSkill: ToolExecutionContext['loadSkill'],
|
||||
): ToolContext {
|
||||
return {
|
||||
parentTelemetry,
|
||||
toolCallId,
|
||||
toolName: builtTool.name,
|
||||
|
|
@ -171,7 +170,24 @@ export async function executeTool(
|
|||
abortSignal: executionContext.abortSignal,
|
||||
executionCounter: executionContext.executionCounter,
|
||||
};
|
||||
return await builtTool.handler(args, ctx);
|
||||
}
|
||||
|
||||
function createSuspendHandler(
|
||||
executionContext: ToolExecutionContext,
|
||||
): InterruptibleToolContext['suspend'] {
|
||||
return async (payload: unknown, options?: ToolSuspendOptions): Promise<never> => {
|
||||
const resolvedOptions: ToolSuspendOptions = {
|
||||
continuation: executionContext.continuation,
|
||||
...options,
|
||||
resumeSchema: options?.resumeSchema ?? executionContext.resumeSchema,
|
||||
};
|
||||
await executionContext.onSuspend?.(payload, resolvedOptions);
|
||||
return await Promise.resolve({
|
||||
[SUSPEND_BRAND]: true,
|
||||
payload,
|
||||
...resolvedOptions,
|
||||
} as never);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -122,13 +122,28 @@ export interface ToolCallBatchResult {
|
|||
pending: Record<string, PendingToolCall>;
|
||||
}
|
||||
|
||||
interface RuntimeToolCall {
|
||||
interface ToolCallInput {
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
input: JSONObject;
|
||||
input: unknown;
|
||||
providerExecuted?: boolean;
|
||||
}
|
||||
|
||||
interface RuntimeToolCall extends ToolCallInput {
|
||||
input: JSONObject;
|
||||
}
|
||||
|
||||
type ToolCallIdentity = Pick<PendingToolCall, 'toolCallId' | 'toolName' | 'input'>;
|
||||
type SuspendedToolOutcome = Extract<ToolCallOutcome, { outcome: 'suspended' }>;
|
||||
|
||||
interface InterruptedToolSuspension {
|
||||
didSuspend: boolean;
|
||||
abortObserved: boolean;
|
||||
payload: unknown;
|
||||
options?: ToolSuspendOptions;
|
||||
cleanup?: Promise<void>;
|
||||
}
|
||||
|
||||
/** Shared input for the tool-call batch iterators. */
|
||||
export interface ToolBatchContext {
|
||||
toolMap: Map<string, BuiltTool>;
|
||||
|
|
@ -288,33 +303,54 @@ export class ToolCallExecutor {
|
|||
* even if one throws, then re-throws the first error.
|
||||
*/
|
||||
async iterateToolCallsConcurrent(
|
||||
ctx: ToolBatchContext & {
|
||||
toolCalls: Array<{
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
input: unknown;
|
||||
providerExecuted?: boolean;
|
||||
}>;
|
||||
},
|
||||
ctx: ToolBatchContext & { toolCalls: ToolCallInput[] },
|
||||
): Promise<ToolCallBatchResult> {
|
||||
const {
|
||||
toolCalls,
|
||||
toolMap,
|
||||
list,
|
||||
runId,
|
||||
telemetry: resolvedTelemetry,
|
||||
executionCounter,
|
||||
abortSignal,
|
||||
} = ctx;
|
||||
const errors: ToolCallError[] = [];
|
||||
const runtimeToolCalls: RuntimeToolCall[] = [];
|
||||
const runtimeToolCalls = this.normalizeToolCalls(ctx.toolCalls, ctx, errors);
|
||||
const executableCalls = runtimeToolCalls.filter((tc) => !tc.providerExecuted);
|
||||
const providerExecutedCount = runtimeToolCalls.length - executableCalls.length;
|
||||
for (let i = 0; i < providerExecutedCount; i++) {
|
||||
incrementToolCallCount(ctx.executionCounter);
|
||||
}
|
||||
const executableCallsById = new Map(executableCalls.map((tc) => [tc.toolCallId, tc]));
|
||||
const unexecutedIds = new Set(executableCalls.map((tc) => tc.toolCallId));
|
||||
const result: ToolCallBatchResult = { results: [], suspensions: [], errors, pending: {} };
|
||||
|
||||
for (let batchStart = 0; batchStart < executableCalls.length; ) {
|
||||
if (ctx.isAborted()) {
|
||||
this.skipAbortedToolCalls(unexecutedIds, executableCallsById, ctx.list, result.results);
|
||||
return await this.finalizeBatch(result, ctx);
|
||||
}
|
||||
const batch = this.takeNextToolCallBatch(executableCalls, batchStart, ctx.toolMap);
|
||||
batchStart += batch.length;
|
||||
const settled = await this.executeToolCallBatch(batch, ctx);
|
||||
for (const tc of batch) unexecutedIds.delete(tc.toolCallId);
|
||||
const hasSuspension = this.collectBatchResults(batch, settled, result, ctx);
|
||||
|
||||
if (ctx.isAborted()) {
|
||||
this.skipAbortedToolCalls(unexecutedIds, executableCallsById, ctx.list, result.results);
|
||||
return await this.finalizeBatch(result, ctx);
|
||||
}
|
||||
if (hasSuspension) {
|
||||
this.retainUnexecutedToolCalls(unexecutedIds, executableCallsById, result.pending);
|
||||
break;
|
||||
}
|
||||
}
|
||||
return await this.finalizeBatch(result, ctx);
|
||||
}
|
||||
|
||||
private normalizeToolCalls(
|
||||
toolCalls: ToolCallInput[],
|
||||
ctx: ToolBatchContext,
|
||||
errors: ToolCallError[],
|
||||
): RuntimeToolCall[] {
|
||||
const calls: RuntimeToolCall[] = [];
|
||||
for (const toolCall of toolCalls) {
|
||||
const normalizedInput = normalizeToolInputForModel(toolCall.input);
|
||||
if (!normalizedInput.ok) {
|
||||
const error = new Error(normalizedInput.error);
|
||||
incrementToolCallCount(executionCounter);
|
||||
list.setToolCallError(toolCall.toolCallId, error);
|
||||
incrementToolCallCount(ctx.executionCounter);
|
||||
ctx.list.setToolCallError(toolCall.toolCallId, error);
|
||||
errors.push({
|
||||
toolCallId: toolCall.toolCallId,
|
||||
toolName: toolCall.toolName,
|
||||
|
|
@ -323,177 +359,180 @@ export class ToolCallExecutor {
|
|||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
runtimeToolCalls.push({ ...toolCall, input: normalizedInput.input });
|
||||
calls.push({ ...toolCall, input: normalizedInput.input });
|
||||
}
|
||||
return calls;
|
||||
}
|
||||
|
||||
const executableCalls = runtimeToolCalls.filter((tc) => !tc.providerExecuted);
|
||||
const providerExecutedCount = runtimeToolCalls.length - executableCalls.length;
|
||||
for (let i = 0; i < providerExecutedCount; i++) {
|
||||
incrementToolCallCount(executionCounter);
|
||||
private async executeToolCallBatch(
|
||||
batch: RuntimeToolCall[],
|
||||
ctx: ToolBatchContext,
|
||||
): Promise<Array<PromiseSettledResult<ToolCallOutcome>>> {
|
||||
return await Promise.allSettled(
|
||||
batch.map(
|
||||
async (tc) =>
|
||||
await this.processToolCall({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: tc.input,
|
||||
toolMap: ctx.toolMap,
|
||||
list: ctx.list,
|
||||
runId: ctx.runId,
|
||||
persistence: ctx.persistence,
|
||||
resolvedTelemetry: ctx.telemetry,
|
||||
executionCounter: ctx.executionCounter,
|
||||
abortSignal: ctx.abortSignal,
|
||||
countToolCall: true,
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
private collectBatchResults(
|
||||
calls: RuntimeToolCall[],
|
||||
settled: Array<PromiseSettledResult<ToolCallOutcome>>,
|
||||
batch: ToolCallBatchResult,
|
||||
ctx: ToolBatchContext,
|
||||
): boolean {
|
||||
let hasSuspension = false;
|
||||
for (let i = 0; i < settled.length; i++) {
|
||||
if (this.collectSettledToolCall(calls[i], settled[i], batch, ctx)) hasSuspension = true;
|
||||
}
|
||||
const executableCallsById = new Map(executableCalls.map((tc) => [tc.toolCallId, tc]));
|
||||
const unexecutedIds = new Set(executableCalls.map((tc) => tc.toolCallId));
|
||||
const results: ToolCallSuccess[] = [];
|
||||
const suspensions: ToolCallSuspension[] = [];
|
||||
const pending: Record<string, PendingToolCall> = {};
|
||||
return hasSuspension;
|
||||
}
|
||||
|
||||
for (let batchStart = 0; batchStart < executableCalls.length; ) {
|
||||
if (ctx.isAborted()) {
|
||||
this.deps.onCancelled();
|
||||
for (const id of unexecutedIds) {
|
||||
const tc = executableCallsById.get(id)!;
|
||||
const modelOutput = '[Skipped: run was aborted]';
|
||||
list.setToolCallResult(tc.toolCallId, modelOutput, { canceled: true });
|
||||
results.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: tc.input,
|
||||
toolEntry: {
|
||||
tool: tc.toolName,
|
||||
input: tc.input,
|
||||
output: modelOutput,
|
||||
transformed: false,
|
||||
canceled: true,
|
||||
},
|
||||
modelOutput,
|
||||
});
|
||||
}
|
||||
return await this.finalizeBatch({ results, suspensions, errors, pending }, ctx);
|
||||
}
|
||||
|
||||
const batch = this.takeNextToolCallBatch(executableCalls, batchStart, toolMap);
|
||||
batchStart += batch.length;
|
||||
|
||||
const settledResults = await Promise.allSettled(
|
||||
batch.map(
|
||||
async (tc) =>
|
||||
await this.processToolCall({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: tc.input,
|
||||
toolMap,
|
||||
list,
|
||||
runId,
|
||||
persistence: ctx.persistence,
|
||||
resolvedTelemetry,
|
||||
executionCounter,
|
||||
abortSignal,
|
||||
countToolCall: true,
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
for (const tc of batch) {
|
||||
unexecutedIds.delete(tc.toolCallId);
|
||||
}
|
||||
|
||||
let hasSuspension = false;
|
||||
|
||||
for (let i = 0; i < settledResults.length; i++) {
|
||||
const result = settledResults[i];
|
||||
const tc = batch[i];
|
||||
const toolInput = tc.input;
|
||||
|
||||
if (result.status === 'rejected') {
|
||||
list.setToolCallError(tc.toolCallId, result.reason);
|
||||
errors.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: toolInput,
|
||||
error: result.reason,
|
||||
});
|
||||
} else if (result.value.outcome === 'suspended') {
|
||||
hasSuspension = true;
|
||||
suspensions.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: toolInput,
|
||||
payload: result.value.payload,
|
||||
resumeSchema: result.value.resumeSchema,
|
||||
});
|
||||
pending[tc.toolCallId] = {
|
||||
suspended: true,
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: toolInput,
|
||||
suspendPayload: result.value.payload,
|
||||
resumeSchema: result.value.resumeSchema,
|
||||
...(result.value.continuation !== undefined
|
||||
? { continuation: result.value.continuation }
|
||||
: {}),
|
||||
runId,
|
||||
};
|
||||
} else if (result.value.outcome === 'success') {
|
||||
results.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: toolInput,
|
||||
toolEntry: result.value.toolEntry,
|
||||
modelOutput: result.value.modelOutput,
|
||||
customMessage: result.value.customMessage,
|
||||
...(result.value.mcpServerName !== undefined
|
||||
? { mcpServerName: result.value.mcpServerName }
|
||||
: {}),
|
||||
});
|
||||
} else if (result.value.outcome === 'cancelled') {
|
||||
results.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: toolInput,
|
||||
toolEntry: result.value.toolEntry,
|
||||
modelOutput: result.value.modelOutput,
|
||||
});
|
||||
} else if (result.value.outcome === 'error') {
|
||||
errors.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: toolInput,
|
||||
error: result.value.error,
|
||||
});
|
||||
} else if (result.value.outcome === 'noop') {
|
||||
// noop
|
||||
}
|
||||
}
|
||||
|
||||
if (ctx.isAborted()) {
|
||||
this.deps.onCancelled();
|
||||
for (const id of unexecutedIds) {
|
||||
const tc = executableCallsById.get(id)!;
|
||||
const modelOutput = '[Skipped: run was aborted]';
|
||||
list.setToolCallResult(tc.toolCallId, modelOutput, { canceled: true });
|
||||
results.push({
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: tc.input,
|
||||
toolEntry: {
|
||||
tool: tc.toolName,
|
||||
input: tc.input,
|
||||
output: modelOutput,
|
||||
transformed: false,
|
||||
canceled: true,
|
||||
},
|
||||
modelOutput,
|
||||
});
|
||||
}
|
||||
return await this.finalizeBatch({ results, suspensions, errors, pending }, ctx);
|
||||
}
|
||||
|
||||
if (hasSuspension) {
|
||||
for (const id of unexecutedIds) {
|
||||
const tc = executableCallsById.get(id)!;
|
||||
pending[tc.toolCallId] = {
|
||||
suspended: false,
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
input: tc.input,
|
||||
};
|
||||
}
|
||||
private collectSettledToolCall(
|
||||
call: RuntimeToolCall,
|
||||
result: PromiseSettledResult<ToolCallOutcome>,
|
||||
batch: ToolCallBatchResult,
|
||||
ctx: ToolBatchContext,
|
||||
): boolean {
|
||||
if (result.status === 'rejected') {
|
||||
ctx.list.setToolCallError(call.toolCallId, result.reason);
|
||||
batch.errors.push({
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
error: result.reason,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
const outcome = result.value;
|
||||
switch (outcome.outcome) {
|
||||
case 'suspended':
|
||||
batch.suspensions.push(this.buildToolCallSuspension(call, outcome));
|
||||
batch.pending[call.toolCallId] = this.buildPendingToolCall(call, outcome, ctx.runId);
|
||||
return true;
|
||||
case 'success':
|
||||
case 'cancelled':
|
||||
this.appendCompletedToolCall(call, outcome, batch.results);
|
||||
break;
|
||||
case 'error':
|
||||
batch.errors.push({
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
error: outcome.error,
|
||||
});
|
||||
break;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
return await this.finalizeBatch({ results, suspensions, errors, pending }, ctx);
|
||||
private buildToolCallSuspension(
|
||||
call: ToolCallIdentity,
|
||||
outcome: SuspendedToolOutcome,
|
||||
): ToolCallSuspension {
|
||||
return {
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
payload: outcome.payload,
|
||||
resumeSchema: outcome.resumeSchema,
|
||||
};
|
||||
}
|
||||
|
||||
private buildPendingToolCall(
|
||||
call: ToolCallIdentity,
|
||||
outcome: SuspendedToolOutcome,
|
||||
runId: string,
|
||||
): PendingToolCall {
|
||||
return {
|
||||
suspended: true,
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
suspendPayload: outcome.payload,
|
||||
resumeSchema: outcome.resumeSchema,
|
||||
...(outcome.continuation !== undefined ? { continuation: outcome.continuation } : {}),
|
||||
runId,
|
||||
};
|
||||
}
|
||||
|
||||
private appendCompletedToolCall(
|
||||
call: ToolCallIdentity,
|
||||
outcome: Extract<ToolCallOutcome, { outcome: 'success' | 'cancelled' }>,
|
||||
results: ToolCallSuccess[],
|
||||
): void {
|
||||
const result: ToolCallSuccess = {
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
toolEntry: outcome.toolEntry,
|
||||
modelOutput: outcome.modelOutput,
|
||||
};
|
||||
if (outcome.outcome === 'success') {
|
||||
result.customMessage = outcome.customMessage;
|
||||
if (outcome.mcpServerName !== undefined) result.mcpServerName = outcome.mcpServerName;
|
||||
}
|
||||
results.push(result);
|
||||
}
|
||||
|
||||
private skipAbortedToolCalls(
|
||||
ids: Set<string>,
|
||||
calls: Map<string, RuntimeToolCall>,
|
||||
list: AgentMessageList,
|
||||
results: ToolCallSuccess[],
|
||||
): void {
|
||||
this.deps.onCancelled();
|
||||
for (const id of ids) {
|
||||
const call = calls.get(id)!;
|
||||
const modelOutput = '[Skipped: run was aborted]';
|
||||
list.setToolCallResult(call.toolCallId, modelOutput, { canceled: true });
|
||||
results.push(this.buildSkippedToolResult(call, modelOutput));
|
||||
}
|
||||
}
|
||||
|
||||
private buildSkippedToolResult(call: ToolCallIdentity, modelOutput: string): ToolCallSuccess {
|
||||
return {
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
toolEntry: {
|
||||
tool: call.toolName,
|
||||
input: call.input,
|
||||
output: modelOutput,
|
||||
transformed: false,
|
||||
canceled: true,
|
||||
},
|
||||
modelOutput,
|
||||
};
|
||||
}
|
||||
|
||||
private retainUnexecutedToolCalls(
|
||||
ids: Set<string>,
|
||||
calls: Map<string, RuntimeToolCall>,
|
||||
pending: Record<string, PendingToolCall>,
|
||||
): void {
|
||||
for (const id of ids) {
|
||||
const call = calls.get(id)!;
|
||||
pending[call.toolCallId] = {
|
||||
suspended: false,
|
||||
toolCallId: call.toolCallId,
|
||||
toolName: call.toolName,
|
||||
input: call.input,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -509,172 +548,112 @@ export class ToolCallExecutor {
|
|||
async iteratePendingToolCallsConcurrent(
|
||||
ctx: ToolBatchContext & { pendingResume: PendingResume },
|
||||
): Promise<ToolCallBatchResult> {
|
||||
const {
|
||||
pendingResume,
|
||||
toolMap,
|
||||
list,
|
||||
runId,
|
||||
persistence,
|
||||
telemetry: resolvedTelemetry,
|
||||
executionCounter,
|
||||
abortSignal,
|
||||
} = ctx;
|
||||
const { pendingResume } = ctx;
|
||||
const resumedId = pendingResume.resumeToolCallId;
|
||||
const resumedEntry = pendingResume.pendingToolCalls[resumedId];
|
||||
if (!resumedEntry) {
|
||||
throw new Error(`No pending tool call found for toolCallId: ${resumedId}`);
|
||||
}
|
||||
|
||||
const resumedToolName = resumedEntry.toolName;
|
||||
const results: ToolCallSuccess[] = [];
|
||||
const suspensions: ToolCallSuspension[] = [];
|
||||
const errors: ToolCallError[] = [];
|
||||
const pending: Record<string, PendingToolCall> = {};
|
||||
|
||||
// 1. Execute the resumed tool
|
||||
const processResult = await this.processToolCall(
|
||||
if (!resumedEntry) throw new Error(`No pending tool call found for toolCallId: ${resumedId}`);
|
||||
const batch: ToolCallBatchResult = { results: [], suspensions: [], errors: [], pending: {} };
|
||||
const outcome = await this.processToolCall(
|
||||
this.pendingToolCallParams(resumedEntry, ctx, pendingResume.resumeData),
|
||||
);
|
||||
|
||||
if (processResult.outcome === 'suspended') {
|
||||
pending[resumedId] = {
|
||||
suspended: true,
|
||||
toolCallId: resumedEntry.toolCallId,
|
||||
toolName: resumedToolName,
|
||||
input: resumedEntry.input,
|
||||
suspendPayload: processResult.payload,
|
||||
resumeSchema: processResult.resumeSchema,
|
||||
...(processResult.continuation !== undefined
|
||||
? { continuation: processResult.continuation }
|
||||
: {}),
|
||||
runId,
|
||||
};
|
||||
suspensions.push({
|
||||
toolCallId: resumedId,
|
||||
toolName: resumedToolName,
|
||||
input: resumedEntry.input,
|
||||
payload: processResult.payload,
|
||||
resumeSchema: processResult.resumeSchema,
|
||||
});
|
||||
} else if (processResult.outcome === 'success') {
|
||||
results.push({
|
||||
toolCallId: resumedEntry.toolCallId,
|
||||
toolName: resumedToolName,
|
||||
input: resumedEntry.input,
|
||||
toolEntry: processResult.toolEntry,
|
||||
modelOutput: processResult.modelOutput,
|
||||
customMessage: processResult.customMessage,
|
||||
...(processResult.mcpServerName !== undefined
|
||||
? { mcpServerName: processResult.mcpServerName }
|
||||
: {}),
|
||||
});
|
||||
} else if (processResult.outcome === 'cancelled') {
|
||||
results.push({
|
||||
toolCallId: resumedEntry.toolCallId,
|
||||
toolName: resumedToolName,
|
||||
input: resumedEntry.input,
|
||||
toolEntry: processResult.toolEntry,
|
||||
modelOutput: processResult.modelOutput,
|
||||
});
|
||||
list.addInput([
|
||||
{ role: 'user', content: [{ type: 'text', text: processResult.userMessage }] },
|
||||
]);
|
||||
|
||||
for (const id of Object.keys(pendingResume.pendingToolCalls)) {
|
||||
if (id !== resumedId) {
|
||||
const siblingEntry = pendingResume.pendingToolCalls[id];
|
||||
const siblingTool = toolMap.get(siblingEntry.toolName);
|
||||
const siblingParams = this.pendingToolCallParams(
|
||||
siblingEntry,
|
||||
ctx,
|
||||
pendingResume.resumeData,
|
||||
);
|
||||
if (siblingEntry.suspended && siblingTool?.onCancellation) {
|
||||
try {
|
||||
await this.runCancellationCleanup(
|
||||
siblingParams,
|
||||
siblingTool,
|
||||
processResult.userMessage,
|
||||
);
|
||||
} catch {
|
||||
this.retainPendingToolCall(id, siblingEntry, pending, suspensions);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
const modelOutput = '[Skipped: a sibling tool call was cancelled]';
|
||||
list.setToolCallResult(id, modelOutput, {
|
||||
canceled: true,
|
||||
});
|
||||
results.push({
|
||||
toolCallId: siblingEntry.toolCallId,
|
||||
toolName: siblingEntry.toolName,
|
||||
input: siblingEntry.input,
|
||||
toolEntry: {
|
||||
tool: siblingEntry.toolName,
|
||||
input: siblingEntry.input,
|
||||
output: modelOutput,
|
||||
transformed: false,
|
||||
canceled: true,
|
||||
},
|
||||
modelOutput,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return await this.finalizeBatch({ results, suspensions, errors, pending }, ctx);
|
||||
} else if (processResult.outcome === 'retryable-error') {
|
||||
if (outcome.outcome === 'cancelled') {
|
||||
this.appendCompletedToolCall(resumedEntry, outcome, batch.results);
|
||||
ctx.list.addInput([{ role: 'user', content: [{ type: 'text', text: outcome.userMessage }] }]);
|
||||
await this.cancelSiblingToolCalls(ctx, batch, outcome.userMessage);
|
||||
return await this.finalizeBatch(batch, ctx);
|
||||
}
|
||||
if (outcome.outcome === 'retryable-error') {
|
||||
for (const [id, entry] of Object.entries(pendingResume.pendingToolCalls)) {
|
||||
this.retainPendingToolCall(id, entry, pending, suspensions);
|
||||
this.retainPendingToolCall(id, entry, batch.pending, batch.suspensions);
|
||||
}
|
||||
return await this.finalizeBatch({ results, suspensions, errors, pending }, ctx);
|
||||
} else if (processResult.outcome === 'error') {
|
||||
errors.push({
|
||||
toolCallId: resumedEntry.toolCallId,
|
||||
toolName: resumedToolName,
|
||||
input: resumedEntry.input,
|
||||
error: processResult.error,
|
||||
});
|
||||
} else if (processResult.outcome === 'noop') {
|
||||
// noop
|
||||
return await this.finalizeBatch(batch, ctx);
|
||||
}
|
||||
|
||||
// 2. Process remaining items
|
||||
const unexecuted: Array<{ toolCallId: string; toolName: string; input: unknown }> = [];
|
||||
this.collectResumedToolCall(resumedId, resumedEntry, outcome, batch, ctx.runId);
|
||||
await this.executeRemainingToolCalls(ctx, batch);
|
||||
return await this.finalizeBatch(batch, ctx);
|
||||
}
|
||||
|
||||
for (const [id, entry] of Object.entries(pendingResume.pendingToolCalls)) {
|
||||
if (id === resumedId) continue;
|
||||
|
||||
if (entry.suspended) {
|
||||
this.retainPendingToolCall(id, entry, pending, suspensions);
|
||||
} else {
|
||||
// Unexecuted — collect for batch execution
|
||||
unexecuted.push({
|
||||
toolCallId: id,
|
||||
private collectResumedToolCall(
|
||||
id: string,
|
||||
entry: PendingToolCall,
|
||||
outcome: ToolCallOutcome,
|
||||
batch: ToolCallBatchResult,
|
||||
runId: string,
|
||||
): void {
|
||||
switch (outcome.outcome) {
|
||||
case 'suspended':
|
||||
batch.pending[id] = this.buildPendingToolCall(entry, outcome, runId);
|
||||
batch.suspensions.push(this.buildToolCallSuspension({ ...entry, toolCallId: id }, outcome));
|
||||
break;
|
||||
case 'success':
|
||||
this.appendCompletedToolCall(entry, outcome, batch.results);
|
||||
break;
|
||||
case 'error':
|
||||
batch.errors.push({
|
||||
toolCallId: entry.toolCallId,
|
||||
toolName: entry.toolName,
|
||||
input: entry.input,
|
||||
error: outcome.error,
|
||||
});
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private async cancelSiblingToolCalls(
|
||||
ctx: ToolBatchContext & { pendingResume: PendingResume },
|
||||
batch: ToolCallBatchResult,
|
||||
userMessage: string,
|
||||
): Promise<void> {
|
||||
for (const id of Object.keys(ctx.pendingResume.pendingToolCalls)) {
|
||||
if (id === ctx.pendingResume.resumeToolCallId) continue;
|
||||
const entry = ctx.pendingResume.pendingToolCalls[id];
|
||||
const tool = ctx.toolMap.get(entry.toolName);
|
||||
const params = this.pendingToolCallParams(entry, ctx, ctx.pendingResume.resumeData);
|
||||
if (entry.suspended && tool?.onCancellation) {
|
||||
try {
|
||||
await this.runCancellationCleanup(params, tool, userMessage);
|
||||
} catch {
|
||||
this.retainPendingToolCall(id, entry, batch.pending, batch.suspensions);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
const modelOutput = '[Skipped: a sibling tool call was cancelled]';
|
||||
ctx.list.setToolCallResult(id, modelOutput, { canceled: true });
|
||||
batch.results.push(this.buildSkippedToolResult(entry, modelOutput));
|
||||
}
|
||||
}
|
||||
|
||||
private async executeRemainingToolCalls(
|
||||
ctx: ToolBatchContext & { pendingResume: PendingResume },
|
||||
result: ToolCallBatchResult,
|
||||
): Promise<void> {
|
||||
const unexecuted: ToolCallInput[] = [];
|
||||
for (const [id, entry] of Object.entries(ctx.pendingResume.pendingToolCalls)) {
|
||||
if (id === ctx.pendingResume.resumeToolCallId) continue;
|
||||
if (entry.suspended) {
|
||||
this.retainPendingToolCall(id, entry, result.pending, result.suspensions);
|
||||
} else {
|
||||
unexecuted.push({ toolCallId: id, toolName: entry.toolName, input: entry.input });
|
||||
}
|
||||
}
|
||||
|
||||
// Execute unexecuted tools via iterateToolCallsConcurrent
|
||||
if (unexecuted.length > 0) {
|
||||
const batch = await this.iterateToolCallsConcurrent({
|
||||
toolCalls: unexecuted,
|
||||
toolMap,
|
||||
list,
|
||||
runId,
|
||||
persistence,
|
||||
telemetry: resolvedTelemetry,
|
||||
executionCounter,
|
||||
abortSignal,
|
||||
isAborted: ctx.isAborted,
|
||||
});
|
||||
results.push(...batch.results);
|
||||
suspensions.push(...batch.suspensions);
|
||||
errors.push(...batch.errors);
|
||||
Object.assign(pending, batch.pending);
|
||||
}
|
||||
return await this.finalizeBatch({ results, suspensions, errors, pending }, ctx);
|
||||
if (unexecuted.length === 0) return;
|
||||
const batch = await this.iterateToolCallsConcurrent({
|
||||
toolCalls: unexecuted,
|
||||
toolMap: ctx.toolMap,
|
||||
list: ctx.list,
|
||||
runId: ctx.runId,
|
||||
persistence: ctx.persistence,
|
||||
telemetry: ctx.telemetry,
|
||||
executionCounter: ctx.executionCounter,
|
||||
abortSignal: ctx.abortSignal,
|
||||
isAborted: ctx.isAborted,
|
||||
});
|
||||
result.results.push(...batch.results);
|
||||
result.suspensions.push(...batch.suspensions);
|
||||
result.errors.push(...batch.errors);
|
||||
Object.assign(result.pending, batch.pending);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -728,61 +707,83 @@ export class ToolCallExecutor {
|
|||
});
|
||||
}
|
||||
|
||||
return await this.executeValidatedToolCall(params, builtTool, input);
|
||||
}
|
||||
|
||||
private async executeValidatedToolCall(
|
||||
params: ProcessToolCallParams,
|
||||
builtTool: BuiltTool,
|
||||
input: JSONValue,
|
||||
): Promise<ToolCallOutcome> {
|
||||
let toolResult: unknown;
|
||||
let interruptedSuspendPayload: unknown;
|
||||
let interruptedSuspendOptions: ToolSuspendOptions | undefined;
|
||||
let didSuspend = false;
|
||||
let abortObserved = false;
|
||||
let suspensionCleanup: Promise<void> | undefined;
|
||||
const cleanupInterruptedSuspension = async () => {
|
||||
suspensionCleanup ??= this.runCancellationCleanup(
|
||||
{
|
||||
...params,
|
||||
input,
|
||||
suspendPayload: interruptedSuspendPayload,
|
||||
continuation: interruptedSuspendOptions?.continuation,
|
||||
resumeSchema: getToolResumeJsonSchema(builtTool, interruptedSuspendOptions?.resumeSchema),
|
||||
},
|
||||
builtTool,
|
||||
'Run aborted',
|
||||
).catch(() => undefined);
|
||||
await suspensionCleanup;
|
||||
const suspension: InterruptedToolSuspension = {
|
||||
didSuspend: false,
|
||||
abortObserved: false,
|
||||
payload: undefined,
|
||||
};
|
||||
try {
|
||||
toolResult = await this.runToolHandler(params, builtTool, input, async (payload, options) => {
|
||||
didSuspend = true;
|
||||
interruptedSuspendPayload = payload;
|
||||
interruptedSuspendOptions = options;
|
||||
if (abortObserved || params.abortSignal?.aborted) {
|
||||
await cleanupInterruptedSuspension();
|
||||
suspension.didSuspend = true;
|
||||
suspension.payload = payload;
|
||||
suspension.options = options;
|
||||
if (suspension.abortObserved || params.abortSignal?.aborted) {
|
||||
await this.cleanupInterruptedSuspension(params, builtTool, input, suspension);
|
||||
}
|
||||
});
|
||||
} catch (error) {
|
||||
if (isAbortError(error) || params.abortSignal?.aborted) {
|
||||
abortObserved = true;
|
||||
if (didSuspend) {
|
||||
await cleanupInterruptedSuspension();
|
||||
} else if (params.suspendPayload !== undefined || params.continuation !== undefined) {
|
||||
try {
|
||||
await this.runCancellationCleanup({ ...params, input }, builtTool, 'Run aborted');
|
||||
} catch {
|
||||
// Parent shutdown must continue; persistent stores will prune stale checkpoints.
|
||||
}
|
||||
}
|
||||
suspension.abortObserved = true;
|
||||
await this.cleanupAbortedTool(params, builtTool, input, suspension);
|
||||
this.deps.onCancelled();
|
||||
return this.buildCancelledOutcome(params, 'Run aborted');
|
||||
}
|
||||
|
||||
return await this.toolError(params, error, builtTool);
|
||||
}
|
||||
|
||||
if (isSuspendedToolResult(toolResult)) {
|
||||
return await this.buildSuspendedOutcome(params, builtTool, toolResult);
|
||||
}
|
||||
|
||||
return await this.buildSuccessOutcome(params, builtTool, input, toolResult);
|
||||
}
|
||||
|
||||
private async cleanupInterruptedSuspension(
|
||||
params: ProcessToolCallParams,
|
||||
builtTool: BuiltTool,
|
||||
input: JSONValue,
|
||||
suspension: InterruptedToolSuspension,
|
||||
): Promise<void> {
|
||||
// Both abort and late suspension can reach this path. Run cleanup only once.
|
||||
suspension.cleanup ??= this.runCancellationCleanup(
|
||||
{
|
||||
...params,
|
||||
input,
|
||||
suspendPayload: suspension.payload,
|
||||
continuation: suspension.options?.continuation,
|
||||
resumeSchema: getToolResumeJsonSchema(builtTool, suspension.options?.resumeSchema),
|
||||
},
|
||||
builtTool,
|
||||
'Run aborted',
|
||||
).catch(() => undefined);
|
||||
await suspension.cleanup;
|
||||
}
|
||||
|
||||
private async cleanupAbortedTool(
|
||||
params: ProcessToolCallParams,
|
||||
builtTool: BuiltTool,
|
||||
input: JSONValue,
|
||||
suspension: InterruptedToolSuspension,
|
||||
): Promise<void> {
|
||||
if (suspension.didSuspend) {
|
||||
await this.cleanupInterruptedSuspension(params, builtTool, input, suspension);
|
||||
} else if (params.suspendPayload !== undefined || params.continuation !== undefined) {
|
||||
try {
|
||||
await this.runCancellationCleanup({ ...params, input }, builtTool, 'Run aborted');
|
||||
} catch {
|
||||
// Parent shutdown must continue; persistent stores will prune stale checkpoints.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async runCancellationCleanup(
|
||||
params: ProcessToolCallParams,
|
||||
builtTool: BuiltTool,
|
||||
|
|
@ -1095,24 +1096,12 @@ export class ToolCallExecutor {
|
|||
|
||||
list.setToolCallResult(toolCallId, guardedResult.historyOutput);
|
||||
|
||||
let customMessage = await builtTool.toMessage?.(toolResult);
|
||||
if (customMessage && builtTool.outputTrust === 'untrusted') {
|
||||
customMessage = protectUntrustedToolMessage(customMessage, builtTool);
|
||||
}
|
||||
let guardedCustomMessage = customMessage
|
||||
? await guardToolMessageForModel(customMessage, this.deps.tokenCounter, storage)
|
||||
: undefined;
|
||||
// Stamp tool provenance so derived transcripts (e.g. the observation
|
||||
// log observer) can keep this content inside untrusted-data boundaries.
|
||||
if (guardedCustomMessage && 'role' in guardedCustomMessage) {
|
||||
guardedCustomMessage = {
|
||||
...guardedCustomMessage,
|
||||
origin: { kind: 'tool', toolName },
|
||||
};
|
||||
}
|
||||
if (guardedCustomMessage) {
|
||||
list.addResponse([guardedCustomMessage]);
|
||||
}
|
||||
const guardedCustomMessage = await this.appendCustomToolMessage(
|
||||
params,
|
||||
builtTool,
|
||||
toolResult,
|
||||
storage,
|
||||
);
|
||||
|
||||
return {
|
||||
outcome: 'success',
|
||||
|
|
@ -1128,6 +1117,33 @@ export class ToolCallExecutor {
|
|||
};
|
||||
}
|
||||
|
||||
private async appendCustomToolMessage(
|
||||
params: ProcessToolCallParams,
|
||||
builtTool: BuiltTool,
|
||||
toolResult: unknown,
|
||||
storage: ToolResultGuardStorage | undefined,
|
||||
): Promise<AgentMessage | undefined> {
|
||||
let customMessage = await builtTool.toMessage?.(toolResult);
|
||||
if (customMessage && builtTool.outputTrust === 'untrusted') {
|
||||
customMessage = protectUntrustedToolMessage(customMessage, builtTool);
|
||||
}
|
||||
let guardedCustomMessage = customMessage
|
||||
? await guardToolMessageForModel(customMessage, this.deps.tokenCounter, storage)
|
||||
: undefined;
|
||||
// Stamp tool provenance so derived transcripts (e.g. the observation
|
||||
// log observer) can keep this content inside untrusted-data boundaries.
|
||||
if (guardedCustomMessage && 'role' in guardedCustomMessage) {
|
||||
guardedCustomMessage = {
|
||||
...guardedCustomMessage,
|
||||
origin: { kind: 'tool', toolName: params.toolName },
|
||||
};
|
||||
}
|
||||
if (guardedCustomMessage) {
|
||||
params.list.addResponse([guardedCustomMessage]);
|
||||
}
|
||||
return guardedCustomMessage;
|
||||
}
|
||||
|
||||
private getResultStorage(params: ProcessToolCallParams): ToolResultGuardStorage | undefined {
|
||||
const filesystem = this.deps.workspaceFilesystem;
|
||||
if (!filesystem) return undefined;
|
||||
|
|
|
|||
|
|
@ -197,39 +197,41 @@ export function sanitizeOffloadedToolResultsForMemory(
|
|||
return messages.map((message) => {
|
||||
if (!('content' in message)) return { ...message };
|
||||
|
||||
const content = message.content.map((block): MessageContent => {
|
||||
if (block.type === 'tool-call') {
|
||||
if (block.state === 'resolved' && isOffloadedToolResult(block.output)) {
|
||||
return { ...block, output: { ...EXPIRED_OFFLOADED_TOOL_RESULT } };
|
||||
}
|
||||
if (block.state === 'resolved' && isContentToolResultOutput(block.output)) {
|
||||
return {
|
||||
...block,
|
||||
output: toJsonValue({
|
||||
type: 'content',
|
||||
value: block.output.value.map((part) =>
|
||||
part.type === 'text' && isSerializedOffloadedToolResult(part.text)
|
||||
? { ...part, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON }
|
||||
: part,
|
||||
),
|
||||
}),
|
||||
};
|
||||
}
|
||||
if (block.state === 'rejected' && isSerializedOffloadedToolResult(block.error)) {
|
||||
return { ...block, error: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
|
||||
}
|
||||
}
|
||||
|
||||
if (block.type === 'text' && isSerializedOffloadedToolResult(block.text)) {
|
||||
return { ...block, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
|
||||
}
|
||||
|
||||
return { ...block };
|
||||
});
|
||||
const content = message.content.map(sanitizeOffloadedContentBlock);
|
||||
return { ...message, content };
|
||||
});
|
||||
}
|
||||
|
||||
function sanitizeOffloadedContentBlock(block: MessageContent): MessageContent {
|
||||
if (block.type === 'tool-call') {
|
||||
if (block.state === 'resolved' && isOffloadedToolResult(block.output)) {
|
||||
return { ...block, output: { ...EXPIRED_OFFLOADED_TOOL_RESULT } };
|
||||
}
|
||||
if (block.state === 'resolved' && isContentToolResultOutput(block.output)) {
|
||||
return { ...block, output: sanitizeOffloadedContentResult(block.output) };
|
||||
}
|
||||
if (block.state === 'rejected' && isSerializedOffloadedToolResult(block.error)) {
|
||||
return { ...block, error: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
|
||||
}
|
||||
}
|
||||
|
||||
if (block.type === 'text' && isSerializedOffloadedToolResult(block.text)) {
|
||||
return { ...block, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
|
||||
}
|
||||
|
||||
return { ...block };
|
||||
}
|
||||
|
||||
function sanitizeOffloadedContentResult(output: ContentToolResultOutput): JSONValue {
|
||||
return toJsonValue({
|
||||
type: 'content',
|
||||
value: output.value.map((part) => {
|
||||
if (part.type !== 'text' || !isSerializedOffloadedToolResult(part.text)) return part;
|
||||
return { ...part, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
async function tryOffloadResult(
|
||||
serialized: string,
|
||||
tokenCount: number,
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user