mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
perf(core): Prepare Instance AI skill sandbox in the background (#40363)
This commit is contained in:
parent
ee1db0bc12
commit
b36c2b506b
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
loadSkill: RuntimeSkillLoader;
|
||||
loadFile?: RuntimeSkillFileLoader;
|
||||
|
|
|
|||
|
|
@ -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<void>((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: [] },
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<unknown>) => {
|
||||
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));
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<undefined>();
|
||||
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<void>(() => {}));
|
||||
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',
|
||||
|
|
|
|||
|
|
@ -185,10 +185,16 @@ interface ProductOtelTraceRuntime {
|
|||
spans: Map<string, OtelApiSpan>;
|
||||
contexts: Map<string, OtelContext>;
|
||||
pendingOperations: Map<string, InstanceAiTraceRun>;
|
||||
/** Work started in the trace that can outlive the root run. The release waits for it. */
|
||||
backgroundOperations: Set<Promise<unknown>>;
|
||||
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<void> {
|
||||
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<void> {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const timeout = new Promise<void>((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<T>(
|
||||
proxyConfig: ServiceProxyConfig | undefined,
|
||||
fn: () => T | Promise<T>,
|
||||
|
|
@ -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),
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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<unknown>) => void;
|
||||
/** Trace replay mode: 'record' captures tool I/O, 'replay' remaps IDs, 'off' disables. */
|
||||
replayMode: TraceReplayMode;
|
||||
/** Shared ID remapper instance — available in 'replay' mode. */
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user