From b36c2b506b92f391f859ec10067e57d96c8b09c8 Mon Sep 17 00:00:00 2001 From: Riqwan Thamir Date: Tue, 6 Oct 2026 16:17:43 +0000 Subject: [PATCH] perf(core): Prepare Instance AI skill sandbox in the background (#40363) --- packages/@n8n/agents/src/sdk/agent.ts | 1 - .../skills/__tests__/runtime-skills.test.ts | 20 ++----- packages/@n8n/agents/src/skills/types.ts | 6 ++ .../agent/__tests__/instance-agent.test.ts | 57 ++++++++++++++++++ .../instance-ai/src/agent/instance-agent.ts | 6 +- .../skills/__tests__/runtime-skills.test.ts | 58 ++++++++++++++++++ .../instance-ai/src/skills/runtime-skills.ts | 27 +++++++++ .../__tests__/langsmith-tracing.test.ts | 60 +++++++++++++++++++ .../src/tracing/langsmith-tracing.ts | 50 +++++++++++++++- packages/@n8n/instance-ai/src/types.ts | 5 ++ 10 files changed, 272 insertions(+), 18 deletions(-) diff --git a/packages/@n8n/agents/src/sdk/agent.ts b/packages/@n8n/agents/src/sdk/agent.ts index 5c774beb417..53d548a4de5 100644 --- a/packages/@n8n/agents/src/sdk/agent.ts +++ b/packages/@n8n/agents/src/sdk/agent.ts @@ -1099,7 +1099,6 @@ export class Agent implements BuiltAgent, AgentBuilder { let instructions = this.instructionsText; if (this.skillSource) { - await this.skillSource.prepare?.(); instructions = appendSkillCatalogToInstructions(instructions, this.skillSource.registry); } if (this.workspaceInstance) { diff --git a/packages/@n8n/agents/src/skills/__tests__/runtime-skills.test.ts b/packages/@n8n/agents/src/skills/__tests__/runtime-skills.test.ts index 65ce6a3a6ed..15c83b1dd74 100644 --- a/packages/@n8n/agents/src/skills/__tests__/runtime-skills.test.ts +++ b/packages/@n8n/agents/src/skills/__tests__/runtime-skills.test.ts @@ -490,25 +490,16 @@ Use the workflow SDK.`, expect(prepare).toHaveBeenCalledTimes(1); }); - it('prepares the runtime skill source before injecting the agent skill catalog', async () => { + it('injects the agent skill catalog without preparing the runtime skill source', async () => { const source = createRuntimeSkillSource([ { id: 'summarize_notes', name: 'Summarize notes', description: 'Use for meeting notes.', - instructions: 'Extract decisions.', + instructions: 'Full private skill body: Extract decisions.', }, ]); - const prepare = vi.fn(async () => { - await Promise.resolve(); - source.registry = { - ...source.registry, - skills: source.registry.skills.map((skill) => ({ - ...skill, - description: 'Use for materialized meeting notes.', - })), - }; - }); + const prepare = vi.fn(async () => {}); source.prepare = prepare; const agent = new Agent('assistant') @@ -521,11 +512,10 @@ Use the workflow SDK.`, const { instructions } = runtimeConfig; expect(runtimeConfig.skillSource).toBe(source); - expect(prepare).toHaveBeenCalledTimes(1); + expect(prepare).not.toHaveBeenCalled(); expect(instructions).toContain('name: "Summarize notes"'); expect(instructions).toContain('id: "summarize_notes"'); - expect(instructions).toContain('description: "Use for materialized meeting notes."'); - expect(instructions).not.toContain('description: "Use for meeting notes."'); + expect(instructions).toContain('description: "Use for meeting notes."'); expect(instructions).not.toContain('Full private skill body'); }); diff --git a/packages/@n8n/agents/src/skills/types.ts b/packages/@n8n/agents/src/skills/types.ts index a91581b03c0..7d2be46a705 100644 --- a/packages/@n8n/agents/src/skills/types.ts +++ b/packages/@n8n/agents/src/skills/types.ts @@ -162,6 +162,12 @@ export type RuntimeSkillFileLoader = ( export interface RuntimeSkillSource { registry: RuntimeSkillRegistry; + /** + * Lazy setup that runs before a skill is loaded, not when the agent is built. + * It must not change catalog fields (id, name, description, category, + * recommendedTools), because the catalog is rendered before it runs. + * Only load_skill calls it, so loadSkill and loadFile must not depend on it. + */ prepare?: () => Promise; loadSkill: RuntimeSkillLoader; loadFile?: RuntimeSkillFileLoader; diff --git a/packages/@n8n/instance-ai/src/agent/__tests__/instance-agent.test.ts b/packages/@n8n/instance-ai/src/agent/__tests__/instance-agent.test.ts index 8d50921d15b..5849f2424d2 100644 --- a/packages/@n8n/instance-ai/src/agent/__tests__/instance-agent.test.ts +++ b/packages/@n8n/instance-ai/src/agent/__tests__/instance-agent.test.ts @@ -455,6 +455,63 @@ describe('createInstanceAgent', () => { ); }); + it('starts preparing runtime skills in the background without waiting for them', async () => { + let finishPrepare: () => void = () => {}; + const prepare = vi.fn( + async () => + await new Promise((resolve) => { + finishPrepare = resolve; + }), + ); + const runtimeSkills = { + registry: { + schemaVersion: 1, + skillsHash: 'skills-hash', + skills: [{ id: 'data-table-manager', name: 'data-table-manager', description: 'x' }], + }, + prepare, + loadSkill: vi.fn(), + }; + + await createInstanceAgent({ + modelId: 'test-model', + context: {}, + orchestrationContext: { runId: 'skills-test', runtimeSkills }, + memoryConfig: {}, + mcpManager: createMcpManagerStub(), + } as never); + + expect(prepare).toHaveBeenCalledTimes(1); + finishPrepare(); + }); + + it('logs a background skill preparation failure instead of rejecting', async () => { + const logger = { warn: vi.fn(), info: vi.fn(), error: vi.fn(), debug: vi.fn() }; + const runtimeSkills = { + registry: { + schemaVersion: 1, + skillsHash: 'skills-hash', + skills: [{ id: 'data-table-manager', name: 'data-table-manager', description: 'x' }], + }, + prepare: vi.fn(async () => await Promise.reject(new Error('sandbox unavailable'))), + loadSkill: vi.fn(), + }; + + await createInstanceAgent({ + modelId: 'test-model', + context: {}, + orchestrationContext: { runId: 'skills-test', runtimeSkills, logger }, + memoryConfig: {}, + mcpManager: createMcpManagerStub(), + } as never); + + await vi.waitFor(() => + expect(logger.warn).toHaveBeenCalledWith('Failed to warm runtime skills in the background', { + error: 'sandbox unavailable', + }), + ); + }); + it('passes the selected catalog to domain tools before workspace materialization', async () => { const runtimeSkillCatalog = { registry: { schemaVersion: 1, skillsHash: 'selected-skills', skills: [] }, diff --git a/packages/@n8n/instance-ai/src/agent/instance-agent.ts b/packages/@n8n/instance-ai/src/agent/instance-agent.ts index c51dd78cf03..be69761a82f 100644 --- a/packages/@n8n/instance-ai/src/agent/instance-agent.ts +++ b/packages/@n8n/instance-ai/src/agent/instance-agent.ts @@ -14,7 +14,7 @@ import { import { attachRuntimeWorkspaceCapabilities } from './runtime-workspace'; import { listConnectedMcpServices } from '../mcp/connected-mcp-services'; import { getVersionedSystemPrompt, resolvePromptProfile } from '../prompts/prompt-profiles'; -import { hasRuntimeSkills } from '../skills/runtime-skills'; +import { hasRuntimeSkills, warmRuntimeSkills } from '../skills/runtime-skills'; import { createToolRegistry, mergeToolRegistries, toolRegistryValues } from '../tool-registry'; import { createOrchestratorDomainTools, @@ -254,6 +254,10 @@ export async function createInstanceAgent( const runtimeSkills = orchestrationContext?.runtimeSkills; if (hasRuntimeSkills(runtimeSkills)) { agent.skills(runtimeSkills); + warmRuntimeSkills(runtimeSkills, { + logger: orchestrationContext?.logger, + tracing: orchestrationContext?.tracing, + }); } if (telemetry) { agent.telemetry(telemetry); diff --git a/packages/@n8n/instance-ai/src/skills/__tests__/runtime-skills.test.ts b/packages/@n8n/instance-ai/src/skills/__tests__/runtime-skills.test.ts index f125610386c..2c186b7e56a 100644 --- a/packages/@n8n/instance-ai/src/skills/__tests__/runtime-skills.test.ts +++ b/packages/@n8n/instance-ai/src/skills/__tests__/runtime-skills.test.ts @@ -7,6 +7,7 @@ import { INSTANCE_AI_SKILLS_DIR, loadInstanceAiRuntimeSkillSource, loadInstanceAiRuntimeSkillSourceForBuildMode, + warmRuntimeSkills, } from '../runtime-skills'; import { CONFIG_EVALS_SKILL_ID, disabledInstanceAiSkillIds } from '../skill-gates'; @@ -614,3 +615,60 @@ async function loadRuntimeSkillSourceWithEnabledModules(enabledModules: string | const { loadInstanceAiRuntimeSkillSource } = await import('../runtime-skills.js'); return loadInstanceAiRuntimeSkillSource(); } + +describe('warmRuntimeSkills', () => { + const registry = { + schemaVersion: 1 as const, + skillsHash: 'skills-hash', + skills: [{ id: 'data-table-manager', name: 'data-table-manager', description: 'x' }], + }; + + it('runs the background preparation in the actor span of the chat trace', async () => { + let insideSpan = false; + let preparedInsideSpan = false; + const actorRun = { id: 'actor-run' }; + const withActiveSpan = vi.fn(async (_run: unknown, fn: () => Promise) => { + insideSpan = true; + try { + return await fn(); + } finally { + insideSpan = false; + } + }); + const prepare = vi.fn(async () => { + preparedInsideSpan = insideSpan; + }); + + warmRuntimeSkills({ registry, prepare, loadSkill: vi.fn() } as never, { + tracing: { actorRun, withActiveSpan } as never, + }); + + await vi.waitFor(() => expect(prepare).toHaveBeenCalledTimes(1)); + expect(withActiveSpan).toHaveBeenCalledWith(actorRun, expect.any(Function)); + expect(preparedInsideSpan).toBe(true); + }); + + it('keeps the chat trace open until the background preparation settles', () => { + const preparation = Promise.resolve(); + const keepOpenUntilSettled = vi.fn(); + const tracing = { + actorRun: { id: 'actor-run' }, + withActiveSpan: vi.fn(async () => await preparation), + keepOpenUntilSettled, + }; + + warmRuntimeSkills({ registry, prepare: vi.fn(), loadSkill: vi.fn() } as never, { + tracing: tracing as never, + }); + + expect(keepOpenUntilSettled).toHaveBeenCalledWith(tracing.withActiveSpan.mock.results[0].value); + }); + + it('prepares the source without a span when tracing is off', async () => { + const prepare = vi.fn(async () => {}); + + warmRuntimeSkills({ registry, prepare, loadSkill: vi.fn() } as never); + + await vi.waitFor(() => expect(prepare).toHaveBeenCalledTimes(1)); + }); +}); diff --git a/packages/@n8n/instance-ai/src/skills/runtime-skills.ts b/packages/@n8n/instance-ai/src/skills/runtime-skills.ts index 16c7aa7f1f4..a6e7dc52d46 100644 --- a/packages/@n8n/instance-ai/src/skills/runtime-skills.ts +++ b/packages/@n8n/instance-ai/src/skills/runtime-skills.ts @@ -1,11 +1,14 @@ import { loadRuntimeSkillSourceFromDirectory, type RuntimeSkillSource } from '@n8n/agents'; import type { InstanceAiBuildMode } from '@n8n/api-types'; +import { getErrorMessage } from '@n8n/utils/errors/get-error-message'; import { GROUPING_GUIDANCE } from '@n8n/workflow-sdk/prompts/sdk-reference'; import { TOP_LEVEL_ITEM_CEILING } from 'n8n-workflow'; import { resolve } from 'node:path'; import { isAgentFeatureEnabled } from '@/utils/agent-feature-enabled'; +import type { Logger } from '../logger'; +import type { InstanceAiTraceContext } from '../types'; import { PROMPT_FRAGMENT_SKILLS, resolvePromptProfile, @@ -63,3 +66,27 @@ export function hasRuntimeSkills( ): source is RuntimeSkillSource { return (source?.registry.skills.length ?? 0) > 0; } + +/** + * Prepare the skill source in the background so it overlaps with the first model + * call. load_skill awaits the same promise and retries if this one fails. It runs + * in the actor span, and keeps the trace open until it settles, so the sandbox + * spans stay in the chat trace even when the reply finishes first. + */ +export function warmRuntimeSkills( + source: RuntimeSkillSource, + options: { logger?: Logger; tracing?: InstanceAiTraceContext } = {}, +): void { + const { prepare } = source; + if (!prepare) return; + const { logger, tracing } = options; + const preparation = tracing + ? tracing.withActiveSpan(tracing.actorRun, async () => await prepare.call(source)) + : prepare.call(source); + tracing?.keepOpenUntilSettled?.(preparation); + preparation.catch((error: unknown) => { + logger?.warn('Failed to warm runtime skills in the background', { + error: getErrorMessage(error), + }); + }); +} diff --git a/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts b/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts index 131113f963b..92a8360ecd9 100644 --- a/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts +++ b/packages/@n8n/instance-ai/src/tracing/__tests__/langsmith-tracing.test.ts @@ -1185,6 +1185,66 @@ describe('createInstanceAiTraceContext', () => { expect(provider.shutdown).toHaveBeenCalledTimes(1); }); + it('keeps the trace open for a background operation that outlives the root run', async () => { + const tracing = await createInstanceAiTraceContext({ + threadId: 'thread-background-op', + messageId: 'message-background-op', + runId: 'run-background-op', + userId: 'user-background-op', + input: { message: 'hi' }, + }); + expect(tracing).toBeDefined(); + await startForegroundActor(tracing!); + const provider = agentsMock.getProvider(); + + const acquired = createDeferredPromise(); + const operation = tracing!.withActiveSpan(tracing!.actorRun, async () => { + await withCurrentTraceSpan({ name: 'sandbox: acquire' }, async () => await acquired.promise); + await withCurrentTraceSpan({ name: 'sandbox: sync-skills' }, async () => {}); + }); + tracing!.keepOpenUntilSettled?.(operation); + + await tracing!.finishRun(tracing!.rootRun, { outputs: { status: 'done' } }); + expect(provider.shutdown).not.toHaveBeenCalled(); + expect(tracing?.isLive?.()).toBe(true); + + acquired.resolve(undefined); + await operation; + await vi.waitFor(() => expect(provider.shutdown).toHaveBeenCalledTimes(1)); + + const spans = agentsMock.getSpans(); + const acquireSpan = spans.find((span) => span.name === 'sandbox: acquire'); + expect(acquireSpan?.ended).toBe(true); + expect(acquireSpan?.status?.message).toBeUndefined(); + expect(spans.find((span) => span.name === 'sandbox: sync-skills')?.ended).toBe(true); + expect(tracing?.isLive?.()).toBe(false); + }); + + it('releases the trace after a time limit when a background operation does not settle', async () => { + const tracing = await createInstanceAiTraceContext({ + threadId: 'thread-stuck-op', + messageId: 'message-stuck-op', + runId: 'run-stuck-op', + userId: 'user-stuck-op', + input: { message: 'hi' }, + }); + expect(tracing).toBeDefined(); + const provider = agentsMock.getProvider(); + + vi.useFakeTimers(); + try { + tracing!.keepOpenUntilSettled?.(new Promise(() => {})); + await tracing!.finishRun(tracing!.rootRun, { outputs: { status: 'done' } }); + expect(provider.shutdown).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(2 * 60 * 1000); + expect(provider.shutdown).toHaveBeenCalledTimes(1); + expect(tracing?.isLive?.()).toBe(false); + } finally { + vi.useRealTimers(); + } + }); + it('keeps the provider alive for a queued memory task until it releases, then exports its span and shuts down', async () => { const tracing = await createInstanceAiTraceContext({ threadId: 'thread-late-memory-task', diff --git a/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts b/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts index 2d30eac2cac..4ae0c599270 100644 --- a/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts +++ b/packages/@n8n/instance-ai/src/tracing/langsmith-tracing.ts @@ -185,10 +185,16 @@ interface ProductOtelTraceRuntime { spans: Map; contexts: Map; pendingOperations: Map; + /** Work started in the trace that can outlive the root run. The release waits for it. */ + backgroundOperations: Set>; + releaseDeferred: boolean; shutdown: boolean; lifetime: ProductTelemetryLifetime; } +/** Upper limit for how long a background operation can keep a closed trace open. */ +const BACKGROUND_OPERATION_RELEASE_TIMEOUT_MS = 2 * 60 * 1000; + interface OTelTracer { startSpan( name: string, @@ -488,7 +494,19 @@ async function releaseProductOtelRuntime( runtime: ProductOtelTraceRuntime, traceId: string, ): Promise { - if (runtime.shutdown) return; + if (runtime.shutdown || runtime.releaseDeferred) return; + + if (runtime.backgroundOperations.size > 0) { + // Do not block the caller. Release after the background work settles, so its + // spans finish normally instead of being cancelled. + runtime.releaseDeferred = true; + void settleBackgroundOperations(runtime).then(async () => { + runtime.releaseDeferred = false; + runtime.backgroundOperations.clear(); + await releaseProductOtelRuntime(runtime, traceId); + }); + return; + } runtime.shutdown = true; for (const run of runtime.pendingOperations.values()) { @@ -505,6 +523,27 @@ async function releaseProductOtelRuntime( await runtime.lifetime.release(); } +async function settleBackgroundOperations(runtime: ProductOtelTraceRuntime): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(resolve, BACKGROUND_OPERATION_RELEASE_TIMEOUT_MS); + timer.unref?.(); + }); + const allSettled = (async () => { + // Operations can register while earlier ones are in flight. + while (runtime.backgroundOperations.size > 0) { + const operations = [...runtime.backgroundOperations]; + await Promise.allSettled(operations); + for (const operation of operations) runtime.backgroundOperations.delete(operation); + } + })(); + try { + await Promise.race([allSettled, timeout]); + } finally { + clearTimeout(timer); + } +} + async function withProxyHeaders( proxyConfig: ServiceProxyConfig | undefined, fn: () => T | Promise, @@ -1318,6 +1357,13 @@ function createTraceContext( finishRun, failRun, onMemoryTaskEvent, + keepOpenUntilSettled: (operation) => { + if (otelRuntime.shutdown) return; + otelRuntime.backgroundOperations.add(operation); + void operation + .catch(() => undefined) + .finally(() => otelRuntime.backgroundOperations.delete(operation)); + }, ...(telemetryFactory ? { getTelemetry: telemetryFactory } : {}), wrapTools: (tools, traceOptions) => { if (ctx.replayMode === 'replay' && ctx.traceIndex && ctx.idRemapper) { @@ -1767,6 +1813,8 @@ async function createProductOtelRuntime( spans: new Map(), contexts: new Map(), pendingOperations: new Map(), + backgroundOperations: new Set(), + releaseDeferred: false, shutdown: false, lifetime: new ProductTelemetryLifetime(telemetry), }; diff --git a/packages/@n8n/instance-ai/src/types.ts b/packages/@n8n/instance-ai/src/types.ts index 54c0d070c23..7f83edcfcc3 100644 --- a/packages/@n8n/instance-ai/src/types.ts +++ b/packages/@n8n/instance-ai/src/types.ts @@ -2156,6 +2156,11 @@ export interface InstanceAiTraceContext { * for any agent (main or sub-agent) whose spans should join this trace. */ onMemoryTaskEvent?: (event: ScopedMemoryTaskEvent) => void; + /** + * Keep the trace open until a background operation settles, so its spans + * finish normally when it outlives the root run. + */ + keepOpenUntilSettled?: (operation: Promise) => void; /** Trace replay mode: 'record' captures tool I/O, 'replay' remaps IDs, 'off' disables. */ replayMode: TraceReplayMode; /** Shared ID remapper instance — available in 'replay' mode. */