mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-18 14:56:50 +02:00
feat(ai-builder): Emit an agent-snapshot trace event for eval seeding (no-changelog) (#35584)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
9ac12d212f
commit
fa694eb5da
|
|
@ -27,6 +27,7 @@ import type * as UsageAccumulatorMod from './stream/usage-accumulator';
|
|||
import type * as ToolsMod from './tools';
|
||||
import type * as AgentPersistenceMod from './tools/orchestration/agent-persistence';
|
||||
import type * as SanitizeWebContentMod from './tools/web-research/sanitize-web-content';
|
||||
import type * as AgentSnapshotEventMod from './tracing/agent-snapshot-event';
|
||||
import type * as LangsmithTracingMod from './tracing/langsmith-tracing';
|
||||
import type * as TraceReplayMod from './tracing/trace-replay';
|
||||
import type * as AgentTreeMod from './utils/agent-tree';
|
||||
|
|
@ -88,6 +89,9 @@ const loadLangsmithTracing = lazyModule(
|
|||
const loadTraceReplay = lazyModule(
|
||||
() => require('./tracing/trace-replay') as typeof TraceReplayMod,
|
||||
);
|
||||
const loadAgentSnapshotEvent = lazyModule(
|
||||
() => require('./tracing/agent-snapshot-event') as typeof AgentSnapshotEventMod,
|
||||
);
|
||||
const loadInstanceAgent = lazyModule(
|
||||
() => require('./agent/instance-agent') as typeof InstanceAgentMod,
|
||||
);
|
||||
|
|
@ -216,6 +220,13 @@ export const createDomainAccessTracker: typeof DomainAccessMod.createDomainAcces
|
|||
export type { DomainAccessTracker } from './domain-access';
|
||||
export type { SubmitLangsmithUserFeedbackOptions } from './tracing/langsmith-tracing';
|
||||
|
||||
export const emitAgentSnapshotTraceEvent: typeof AgentSnapshotEventMod.emitAgentSnapshotTraceEvent =
|
||||
lazyFunction(() => loadAgentSnapshotEvent().emitAgentSnapshotTraceEvent);
|
||||
export type {
|
||||
AgentSnapshotArtifact,
|
||||
AgentSnapshotReason,
|
||||
} from './tracing/agent-snapshot-event';
|
||||
|
||||
export const createInstanceAiTraceContext: typeof LangsmithTracingMod.createInstanceAiTraceContext =
|
||||
lazyFunction(() => loadLangsmithTracing().createInstanceAiTraceContext);
|
||||
|
||||
|
|
|
|||
|
|
@ -577,6 +577,128 @@ describe('build-agent tool', () => {
|
|||
});
|
||||
});
|
||||
|
||||
// An eval case needs the agent BEFORE the turn and again after a pass that
|
||||
// changed it. Nothing else in the trace records either.
|
||||
describe('agent-snapshot', () => {
|
||||
const ARTIFACT = {
|
||||
config: { name: 'Support Triage' },
|
||||
skills: {},
|
||||
configHash: 'hash-1',
|
||||
} as unknown as Awaited<
|
||||
ReturnType<NonNullable<InstanceAiBuilderDelegate['readAgentArtifact']>>
|
||||
>;
|
||||
|
||||
/** The read only runs under a live trace, so every case here needs one. */
|
||||
function makeTracedContext() {
|
||||
const made = makeContext();
|
||||
made.context.tracing = makeTracingStub().tracing;
|
||||
return made;
|
||||
}
|
||||
|
||||
it('snapshots an adopted agent before the builder can change it', async () => {
|
||||
const { context, delegate } = makeTracedContext();
|
||||
vi.mocked(delegate.readAgentArtifact!).mockResolvedValue(ARTIFACT);
|
||||
vi.mocked(delegate.streamBuild).mockImplementation(async () => {
|
||||
// Ordering is the point: read before the builder runs, or the
|
||||
// "baseline" is the state the turn produced.
|
||||
expect(delegate.readAgentArtifact).toHaveBeenCalledWith('agent-existing');
|
||||
return await Promise.resolve(fakeStream([], 'Editing it.'));
|
||||
});
|
||||
|
||||
await runTool(context, { message: 'Add a tool', agentId: 'agent-existing' });
|
||||
|
||||
expect(delegate.readAgentArtifact).toHaveBeenCalledWith('agent-existing');
|
||||
});
|
||||
|
||||
it('takes no baseline for an agent it just created — there is no prior state', async () => {
|
||||
const { context, delegate } = makeTracedContext();
|
||||
vi.mocked(delegate.createAgent).mockResolvedValue({
|
||||
agentId: 'agent-1',
|
||||
projectId: 'proj-1',
|
||||
});
|
||||
vi.mocked(delegate.readAgentArtifact!).mockResolvedValue(ARTIFACT);
|
||||
vi.mocked(delegate.streamBuild).mockResolvedValue(fakeStream([], 'Built it.'));
|
||||
|
||||
await runTool(context, { message: 'Build it', name: 'New Agent' });
|
||||
|
||||
expect(delegate.readAgentArtifact).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('snapshots again after a pass that changed the config', async () => {
|
||||
const { context, delegate } = makeTracedContext();
|
||||
vi.mocked(delegate.readAgentArtifact!).mockResolvedValue(ARTIFACT);
|
||||
vi.mocked(delegate.streamBuild).mockResolvedValue(
|
||||
fakeStream(
|
||||
[toolCallChunk('call-1', 'patch_config'), toolResultChunk('call-1')],
|
||||
'Updated the config.',
|
||||
),
|
||||
);
|
||||
|
||||
await runTool(context, { message: 'Add a tool', agentId: 'agent-existing' });
|
||||
|
||||
// Baseline + outcome.
|
||||
expect(delegate.readAgentArtifact).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('does not re-read after a pass that changed nothing', async () => {
|
||||
const { context, delegate } = makeTracedContext();
|
||||
vi.mocked(delegate.readAgentArtifact!).mockResolvedValue(ARTIFACT);
|
||||
vi.mocked(delegate.streamBuild).mockResolvedValue(fakeStream([], 'Nothing to change.'));
|
||||
|
||||
await runTool(context, { message: 'What does it do?', agentId: 'agent-existing' });
|
||||
|
||||
expect(delegate.readAgentArtifact).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('reads nothing when tracing is off — there is nowhere to emit', async () => {
|
||||
// Tracing is disabled on most instances, and the read costs a scope
|
||||
// check plus two queries on every non-create build-agent turn.
|
||||
const { context, delegate } = makeContext();
|
||||
vi.mocked(delegate.streamBuild).mockResolvedValue(
|
||||
fakeStream(
|
||||
[toolCallChunk('call-1', 'patch_config'), toolResultChunk('call-1')],
|
||||
'Updated the config.',
|
||||
),
|
||||
);
|
||||
|
||||
await runTool(context, { message: 'Add a tool', agentId: 'agent-existing' });
|
||||
|
||||
expect(delegate.readAgentArtifact).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('emits the event with the config, the agent id and the reason', async () => {
|
||||
const { context, delegate } = makeContext();
|
||||
const { tracing } = makeTracingStub();
|
||||
context.tracing = tracing;
|
||||
vi.mocked(delegate.readAgentArtifact!).mockResolvedValue(ARTIFACT);
|
||||
vi.mocked(delegate.streamBuild).mockResolvedValue(fakeStream([], 'Editing it.'));
|
||||
|
||||
await runTool(context, { message: 'Add a tool', agentId: 'agent-existing' });
|
||||
|
||||
const snapshotRuns = vi
|
||||
.mocked(tracing.startChildRun)
|
||||
.mock.calls.filter(([, init]) => (init as { name?: string }).name === 'agent-snapshot');
|
||||
expect(snapshotRuns).toHaveLength(1);
|
||||
expect(snapshotRuns[0][1]).toMatchObject({
|
||||
runType: 'chain',
|
||||
metadata: { agent_id: 'agent-existing', snapshot_reason: 'target-resolved' },
|
||||
});
|
||||
});
|
||||
|
||||
it('builds normally on a host whose delegate cannot read agents', async () => {
|
||||
// An older host doesn't implement the read; that must not break building.
|
||||
const delegate = mock<InstanceAiBuilderDelegate>();
|
||||
delegate.readAgentArtifact = undefined;
|
||||
const { context } = makeContext({ delegate });
|
||||
context.tracing = makeTracingStub().tracing;
|
||||
vi.mocked(delegate.streamBuild).mockResolvedValue(fakeStream([], 'Editing it.'));
|
||||
|
||||
const result = await runTool(context, { message: 'Add a tool', agentId: 'agent-existing' });
|
||||
|
||||
expect(result).toMatchObject({ ok: true, builderReply: 'Editing it.' });
|
||||
});
|
||||
});
|
||||
|
||||
describe('configUpdated', () => {
|
||||
it.each(['write_config', 'patch_config', 'publish_agent', 'unpublish_agent'])(
|
||||
'is true when the work summary has a succeeded %s call',
|
||||
|
|
@ -1771,6 +1893,44 @@ describe('build-agent tool', () => {
|
|||
},
|
||||
);
|
||||
|
||||
it('snapshots the config a friendly-mapped resume failure left behind', async () => {
|
||||
// A pass mutated the config, suspended for confirmation, and the resume
|
||||
// failed. The tool still reports configUpdated, so the post-state a
|
||||
// repair-shaped case grades has to exist in the trace.
|
||||
const { context, delegate } = makeContext();
|
||||
const { tracing } = makeTracingStub();
|
||||
context.tracing = tracing;
|
||||
context.domainContext!.agentBuilderTarget = { agentId: 'agent-1', projectId: 'proj-1' };
|
||||
vi.mocked(delegate.findOpenSuspensions).mockResolvedValue([
|
||||
{ runId: 'builder-run-1', toolCallId: 'builder-call-1' },
|
||||
]);
|
||||
vi.mocked(delegate.readAgentArtifact!).mockResolvedValue({
|
||||
config: { name: 'Support Triage' },
|
||||
configHash: 'hash-1',
|
||||
} as unknown as Awaited<
|
||||
ReturnType<NonNullable<InstanceAiBuilderDelegate['readAgentArtifact']>>
|
||||
>);
|
||||
vi.mocked(delegate.resumeBuild).mockResolvedValue(
|
||||
throwingStream(new FakeBuilderCheckpointUnavailableError('Checkpoint expired.')),
|
||||
);
|
||||
|
||||
const result = await runToolWithCtx(
|
||||
context,
|
||||
{ message: 'Build it', name: 'New Agent' },
|
||||
{
|
||||
resumeData: { approved: true },
|
||||
suspendPayload: suspendPayloadWithCheckpoint({ configUpdated: true }),
|
||||
},
|
||||
);
|
||||
|
||||
expect(result).toMatchObject({ ok: false, configUpdated: true });
|
||||
const snapshots = vi
|
||||
.mocked(tracing.startChildRun)
|
||||
.mock.calls.filter(([, init]) => (init as { name?: string }).name === 'agent-snapshot');
|
||||
expect(snapshots).toHaveLength(1);
|
||||
expect(snapshots[0][1]).toMatchObject({ metadata: { snapshot_reason: 'config-updated' } });
|
||||
});
|
||||
|
||||
it('still rethrows an unrelated error thrown mid-stream during resume', async () => {
|
||||
const { context, delegate } = makeContext();
|
||||
context.domainContext!.agentBuilderTarget = { agentId: 'agent-1', projectId: 'proj-1' };
|
||||
|
|
|
|||
|
|
@ -47,11 +47,17 @@ import {
|
|||
type AgentBuilderTarget,
|
||||
} from './agent-target-binding';
|
||||
import { instanceAiBuilderThreadPrefix } from './builder-thread-id';
|
||||
import { failTraceRun, finishTraceRun, startSubAgentTrace, withTraceRun } from './tracing-utils';
|
||||
import {
|
||||
consumeStreamCascading,
|
||||
type ConsumeStreamCascadingResult,
|
||||
} from '../../stream/consume-with-hitl';
|
||||
import type { WorkSummary } from '../../stream/work-summary-accumulator';
|
||||
import {
|
||||
emitAgentSnapshotTraceEvent,
|
||||
type AgentSnapshotArtifact,
|
||||
type AgentSnapshotReason,
|
||||
} from '../../tracing/agent-snapshot-event';
|
||||
import type {
|
||||
BuilderTurnStream,
|
||||
InstanceAiBuilderDelegate,
|
||||
|
|
@ -60,7 +66,6 @@ import type {
|
|||
SessionWorkflowRef,
|
||||
} from '../../types';
|
||||
import { ORCHESTRATION_TOOL_IDS } from '../tool-ids';
|
||||
import { failTraceRun, finishTraceRun, startSubAgentTrace, withTraceRun } from './tracing-utils';
|
||||
|
||||
const BUILDER_SUB_AGENT_ROLE = 'agent-builder';
|
||||
const BUILDER_SUB_AGENT_KIND = 'agent-builder';
|
||||
|
|
@ -333,6 +338,36 @@ function publishAgentBuilderCancelled(context: OrchestrationContext, builderAgen
|
|||
});
|
||||
}
|
||||
|
||||
/** Emit an `agent-snapshot` for the builder's target. Best-effort at both ends. */
|
||||
async function snapshotAgent(
|
||||
context: OrchestrationContext,
|
||||
delegate: InstanceAiBuilderDelegate,
|
||||
target: AgentBuilderTarget,
|
||||
reason: AgentSnapshotReason,
|
||||
): Promise<void> {
|
||||
// No trace, no read — the delegate read costs a scope check and two queries.
|
||||
// Matches the service-side call site, which early-returns on `!tracing`.
|
||||
if (!context.tracing) return;
|
||||
let artifact: AgentSnapshotArtifact | null = null;
|
||||
try {
|
||||
// An optional method may be absent, or return a non-promise on a mocked host.
|
||||
artifact = (await delegate.readAgentArtifact?.(target.agentId)) ?? null;
|
||||
} catch (error) {
|
||||
context.logger.debug(
|
||||
`[agent-snapshot] ${reason} read for ${target.agentId} failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (!artifact) return;
|
||||
await emitAgentSnapshotTraceEvent(context.tracing, {
|
||||
agentId: target.agentId,
|
||||
projectId: target.projectId,
|
||||
reason,
|
||||
artifact,
|
||||
logger: context.logger,
|
||||
});
|
||||
}
|
||||
|
||||
/** Publish the terminal `agent-completed` event and map the result to the tool output.
|
||||
* A cancelled turn is intercepted by the caller, so that status never arrives here. */
|
||||
async function finishTurn(
|
||||
|
|
@ -412,6 +447,15 @@ async function runBuilderConsumeLoop(params: {
|
|||
dedupeBase,
|
||||
} = params;
|
||||
|
||||
// Every settled return goes through here, so the state a pass left behind is
|
||||
// snapshotted on the error returns too — a pass that mutated the config, then
|
||||
// suspended and failed on resume, is exactly the post-state a repair case
|
||||
// grades. A suspend resumes and settles through here; a cancel throws past it.
|
||||
const settle = async (output: BuildAgentOutput): Promise<BuildAgentOutput> => {
|
||||
if (output.configUpdated) await snapshotAgent(context, delegate, target, 'config-updated');
|
||||
return output;
|
||||
};
|
||||
|
||||
const traceRun = await startSubAgentTrace(context, {
|
||||
agentId: builderAgentId,
|
||||
role: BUILDER_SUB_AGENT_ROLE,
|
||||
|
|
@ -445,12 +489,12 @@ async function runBuilderConsumeLoop(params: {
|
|||
// not from the `delegate.streamBuild`/`resumeBuild` call sites.
|
||||
const message = publishAgentBuilderFailure(context, builderAgentId, error);
|
||||
if (isFriendlyMappableBuilderError(error)) {
|
||||
return {
|
||||
return await settle({
|
||||
ok: false,
|
||||
error: message,
|
||||
configUpdated: carriedConfigUpdated,
|
||||
...targetIdentity(target),
|
||||
};
|
||||
});
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
|
@ -499,7 +543,7 @@ async function runBuilderConsumeLoop(params: {
|
|||
await failTraceRun(context, traceRun, new Error(output.error ?? 'builder run failed'));
|
||||
}
|
||||
await context.claimSubAgentUsage?.(dedupeBase, result.usage?.usage ?? [], result.status);
|
||||
return { ...output, ...targetIdentity(target) };
|
||||
return await settle({ ...output, ...targetIdentity(target) });
|
||||
}
|
||||
|
||||
const configUpdatedSoFar = carriedConfigUpdated || didUpdateConfig(result.workSummary);
|
||||
|
|
@ -528,12 +572,12 @@ async function runBuilderConsumeLoop(params: {
|
|||
result.usage?.usage ?? [],
|
||||
'errored',
|
||||
);
|
||||
return {
|
||||
return await settle({
|
||||
ok: false,
|
||||
error: message,
|
||||
configUpdated: configUpdatedSoFar,
|
||||
...targetIdentity(target),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
// The builder-level requestId must not leak up: the FE confirms against the
|
||||
|
|
@ -954,6 +998,12 @@ export function createBuildAgentTool(context: OrchestrationContext) {
|
|||
|
||||
publishAgentSpawned(context, builderAgentId, boundTarget);
|
||||
|
||||
// Before the builder touches it: a repair-shaped eval case seeds from the
|
||||
// state the turn opened on. A new agent has no prior state.
|
||||
if (resolution.mode !== 'create') {
|
||||
await snapshotAgent(context, delegate, boundTarget, 'target-resolved');
|
||||
}
|
||||
|
||||
let turn: BuilderTurnStream;
|
||||
try {
|
||||
turn = await delegate.streamBuild(boundTarget.agentId, outboundMessage, session);
|
||||
|
|
|
|||
|
|
@ -24,6 +24,10 @@ export const DOMAIN_TOOL_IDS = {
|
|||
* SDK re-parse; excluded by name from rebuilt transcripts. */
|
||||
export const COMPILED_WORKFLOW_TRACE_RUN_NAME = 'compiled-workflow';
|
||||
|
||||
/** Trace-only chain run carrying an agent's config + skills, for seed
|
||||
* reconstruction — the agent counterpart of the event above. */
|
||||
export const AGENT_SNAPSHOT_TRACE_RUN_NAME = 'agent-snapshot';
|
||||
|
||||
export const ORCHESTRATION_TOOL_IDS = {
|
||||
CREATE_TASKS: 'create-tasks',
|
||||
TASK_CONTROL: 'task-control',
|
||||
|
|
|
|||
|
|
@ -0,0 +1,207 @@
|
|||
// The `agent-snapshot` trace event. Mirrors the `compiled-workflow` tests in
|
||||
// `tools/workflows/__tests__/build-workflow.tool.test.ts`.
|
||||
import type { AgentJsonConfig, AgentSkill } from '@n8n/api-types';
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import type { InstanceAiTraceContext } from '../../types';
|
||||
import { emitAgentSnapshotTraceEvent } from '../agent-snapshot-event';
|
||||
|
||||
const CONFIG = {
|
||||
name: 'Support Triage',
|
||||
instructions: 'You triage inbound support tickets.',
|
||||
} as unknown as AgentJsonConfig;
|
||||
|
||||
const SKILLS: Record<string, AgentSkill> = {
|
||||
skill_triage_rules: {
|
||||
name: 'Triage rules',
|
||||
instructions: 'Assign every ticket exactly one severity.',
|
||||
} as unknown as AgentSkill,
|
||||
};
|
||||
|
||||
function makeTracing() {
|
||||
const actorRun = { id: 'actor-run-1' };
|
||||
const startChildRun = vi.fn<
|
||||
(parent: unknown, init: Record<string, unknown>) => Promise<{ id: string }>
|
||||
>(async () => await Promise.resolve({ id: 'snapshot-run-1' }));
|
||||
const finishRun = vi.fn<(run: { id: string }, opts: Record<string, unknown>) => Promise<void>>(
|
||||
async () => await Promise.resolve(),
|
||||
);
|
||||
const tracing = { actorRun, startChildRun, finishRun } as unknown as InstanceAiTraceContext;
|
||||
return { tracing, actorRun, startChildRun, finishRun };
|
||||
}
|
||||
|
||||
const BASE = {
|
||||
agentId: 'TrIaGe1234567890',
|
||||
projectId: 'proj-1',
|
||||
reason: 'attached' as const,
|
||||
};
|
||||
|
||||
describe('emitAgentSnapshotTraceEvent', () => {
|
||||
it('emits a trace-only child run carrying the config and skills', async () => {
|
||||
const { tracing, actorRun, startChildRun, finishRun } = makeTracing();
|
||||
|
||||
const outcome = await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: CONFIG, skills: SKILLS, configHash: 'hash-1' },
|
||||
});
|
||||
|
||||
expect(outcome).toBe('emitted');
|
||||
expect(startChildRun).toHaveBeenCalledTimes(1);
|
||||
const [parentRun, init] = startChildRun.mock.calls[0];
|
||||
expect(parentRun).toBe(actorRun);
|
||||
expect(init).toMatchObject({
|
||||
name: 'agent-snapshot',
|
||||
// Bookkeeping span: a tool-typed run reads as a real agent tool call.
|
||||
runType: 'chain',
|
||||
metadata: {
|
||||
agent_id: BASE.agentId,
|
||||
config_hash: 'hash-1',
|
||||
snapshot_reason: 'attached',
|
||||
// Lifts the export scrubber's structural depth cap for this payload.
|
||||
raw_trace_payload: true,
|
||||
},
|
||||
});
|
||||
const [, finishOpts] = finishRun.mock.calls[0];
|
||||
expect(finishOpts).toMatchObject({
|
||||
rawOutputs: true,
|
||||
outputs: {
|
||||
agentId: BASE.agentId,
|
||||
projectId: 'proj-1',
|
||||
configHash: 'hash-1',
|
||||
reason: 'attached',
|
||||
config: CONFIG,
|
||||
skills: SKILLS,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('emits a truncated marker instead of an oversized payload', async () => {
|
||||
const { tracing, startChildRun, finishRun } = makeTracing();
|
||||
const huge = {
|
||||
name: 'Fat Agent',
|
||||
instructions: 'x'.repeat(1_000_001),
|
||||
} as unknown as AgentJsonConfig;
|
||||
|
||||
const outcome = await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: huge, configHash: 'hash-big' },
|
||||
});
|
||||
|
||||
expect(outcome).toBe('emitted');
|
||||
expect(startChildRun).toHaveBeenCalledTimes(1);
|
||||
const [, finishOpts] = finishRun.mock.calls[0];
|
||||
// The marker records that a state existed without shipping a rejected run.
|
||||
expect(finishOpts.outputs).toEqual({
|
||||
agentId: BASE.agentId,
|
||||
projectId: 'proj-1',
|
||||
configHash: 'hash-big',
|
||||
reason: 'attached',
|
||||
truncated: true,
|
||||
});
|
||||
expect(finishOpts).not.toHaveProperty('rawOutputs');
|
||||
});
|
||||
|
||||
it('emits one event per agent+hash on a trace, so two triggers agree', async () => {
|
||||
// Attach + target baselines both fire when a turn attaches then edits.
|
||||
const { tracing, startChildRun } = makeTracing();
|
||||
const artifact = { config: CONFIG, configHash: 'hash-1' };
|
||||
|
||||
expect(await emitAgentSnapshotTraceEvent(tracing, { ...BASE, artifact })).toBe('emitted');
|
||||
expect(
|
||||
await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
reason: 'target-resolved',
|
||||
artifact,
|
||||
}),
|
||||
).toBe('duplicate');
|
||||
expect(startChildRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('emits again once the config actually changes', async () => {
|
||||
const { tracing, startChildRun } = makeTracing();
|
||||
|
||||
await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: CONFIG, configHash: 'hash-1' },
|
||||
});
|
||||
const outcome = await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
reason: 'config-updated',
|
||||
artifact: { config: CONFIG, configHash: 'hash-2' },
|
||||
});
|
||||
|
||||
expect(outcome).toBe('emitted');
|
||||
expect(startChildRun).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('emits without a hash rather than treating one snapshot as a duplicate', async () => {
|
||||
const { tracing, startChildRun } = makeTracing();
|
||||
const artifact = { config: CONFIG, configHash: null };
|
||||
|
||||
expect(await emitAgentSnapshotTraceEvent(tracing, { ...BASE, artifact })).toBe('emitted');
|
||||
expect(await emitAgentSnapshotTraceEvent(tracing, { ...BASE, artifact })).toBe('emitted');
|
||||
// No hash means no dedupe key — better a duplicate row than a lost state.
|
||||
expect(startChildRun).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('reports a broken exporter as skipped instead of failing the caller', async () => {
|
||||
// Tracing is bookkeeping: a broken exporter must not fail an agent build.
|
||||
// The tracing layer swallows its own error, so this reads as "exported
|
||||
// nothing" rather than as a bug here.
|
||||
const tracing = {
|
||||
actorRun: { id: 'actor-run-1' },
|
||||
startChildRun: vi.fn(async () => await Promise.reject(new Error('exporter down'))),
|
||||
finishRun: vi.fn(),
|
||||
} as unknown as InstanceAiTraceContext;
|
||||
|
||||
await expect(
|
||||
emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: CONFIG, configHash: 'hash-1' },
|
||||
}),
|
||||
).resolves.toBe('skipped');
|
||||
});
|
||||
|
||||
it('retries after a skip instead of burning the hash on a trace that exported nothing', async () => {
|
||||
const { tracing, startChildRun } = makeTracing();
|
||||
const artifact = { config: CONFIG, configHash: 'hash-1' };
|
||||
startChildRun.mockRejectedValueOnce(new Error('exporter down'));
|
||||
|
||||
expect(await emitAgentSnapshotTraceEvent(tracing, { ...BASE, artifact })).toBe('skipped');
|
||||
expect(await emitAgentSnapshotTraceEvent(tracing, { ...BASE, artifact })).toBe('emitted');
|
||||
expect(startChildRun).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('retries after a failure instead of burning the hash', async () => {
|
||||
const { tracing, startChildRun } = makeTracing();
|
||||
const circular: Record<string, unknown> = { name: 'Looper' };
|
||||
circular.self = circular;
|
||||
|
||||
expect(
|
||||
await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: circular as unknown as AgentJsonConfig, configHash: 'hash-1' },
|
||||
}),
|
||||
).toBe('failed');
|
||||
expect(
|
||||
await emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: CONFIG, configHash: 'hash-1' },
|
||||
}),
|
||||
).toBe('emitted');
|
||||
expect(startChildRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('reports a config it cannot serialize as failed, without throwing', async () => {
|
||||
const { tracing } = makeTracing();
|
||||
const circular: Record<string, unknown> = { name: 'Looper' };
|
||||
circular.self = circular;
|
||||
|
||||
await expect(
|
||||
emitAgentSnapshotTraceEvent(tracing, {
|
||||
...BASE,
|
||||
artifact: { config: circular as unknown as AgentJsonConfig, configHash: 'hash-1' },
|
||||
}),
|
||||
).resolves.toBe('failed');
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,134 @@
|
|||
import { jsonParse } from 'n8n-workflow';
|
||||
import { describe, it, expect } from 'vitest';
|
||||
|
||||
import { AGENT_SNAPSHOT_TRACE_RUN_NAME } from '../../tools/tool-ids';
|
||||
import { GEN_AI_COMPLETION, redactLangSmithTelemetrySpan } from '../trace-payloads';
|
||||
|
||||
/** A config nested past the sanitizer's depth cap: a node tool's parameters sit
|
||||
* at depth 5+ under the payload root, and its filter conditions at depth 9+. */
|
||||
function deepAgentConfig() {
|
||||
return {
|
||||
name: 'Support Triage',
|
||||
model: 'anthropic/claude-sonnet-4-5',
|
||||
credential: 'cred-anthropic-1',
|
||||
instructions:
|
||||
'Escalate anything a customer marks as production-down to the ops rota before replying. Older key sk-ant-api03-aaaaaaaaaaaaaaaa was rotated.',
|
||||
tools: [
|
||||
{
|
||||
type: 'node',
|
||||
name: 'send_reply',
|
||||
node: {
|
||||
nodeType: 'n8n-nodes-base.gmail',
|
||||
nodeTypeVersion: 2.1,
|
||||
nodeParameters: {
|
||||
resource: 'message',
|
||||
options: {
|
||||
filters: {
|
||||
conditions: [
|
||||
{
|
||||
leftValue: '={{ $json.priority }}',
|
||||
rightValue: 'urgent',
|
||||
operator: { type: 'string', operation: 'equals' },
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
credentials: { gmailOAuth2: { id: 'cred-2', name: 'Support Inbox' } },
|
||||
},
|
||||
},
|
||||
],
|
||||
} as const;
|
||||
}
|
||||
|
||||
function deepSkills() {
|
||||
return {
|
||||
skill_triage_rules: {
|
||||
name: 'Triage rules',
|
||||
description: 'How to grade an inbound ticket.',
|
||||
instructions:
|
||||
'Assign every ticket exactly one severity, and never downgrade one a customer raised.',
|
||||
references: [
|
||||
// `agentSkillReferenceSchema` requires a markdown path under `references/`,
|
||||
// so a fixture using anything else isn't the shape a real snapshot carries.
|
||||
{
|
||||
path: 'references/severity.md',
|
||||
content: 'Sev1 means the product is unusable for everyone.',
|
||||
},
|
||||
],
|
||||
},
|
||||
} as const;
|
||||
}
|
||||
|
||||
function exportSpan(rawPayload: boolean) {
|
||||
const span = redactLangSmithTelemetrySpan({
|
||||
name: AGENT_SNAPSHOT_TRACE_RUN_NAME,
|
||||
attributes: {
|
||||
// Producer-set flag (emitTraceOnlyChildRun with rawOutputs) — the deep
|
||||
// tier keys on this, not on the (claimable) span name.
|
||||
...(rawPayload ? { 'langsmith.metadata.raw_trace_payload': true } : {}),
|
||||
[GEN_AI_COMPLETION]: JSON.stringify({
|
||||
agentId: 'TrIaGe1234567890',
|
||||
projectId: 'proj-1',
|
||||
configHash: 'h1',
|
||||
reason: 'config-updated',
|
||||
config: deepAgentConfig(),
|
||||
skills: deepSkills(),
|
||||
}),
|
||||
},
|
||||
}) as { attributes: Record<string, string> };
|
||||
return span.attributes[GEN_AI_COMPLETION];
|
||||
}
|
||||
|
||||
type ExportedSnapshot = {
|
||||
agentId: string;
|
||||
configHash: string;
|
||||
config: Omit<ReturnType<typeof deepAgentConfig>, 'credential' | 'tools'> & {
|
||||
credential: string;
|
||||
tools: [{ node: Omit<ReturnType<typeof deepAgentConfig>['tools'][0]['node'], 'credentials'> }];
|
||||
};
|
||||
skills: ReturnType<typeof deepSkills>;
|
||||
};
|
||||
|
||||
describe('agent-snapshot trace payload fidelity through export scrubbing', () => {
|
||||
it('keeps the config and skill bodies while still scrubbing secrets and credentials', () => {
|
||||
const completion = exportSpan(true);
|
||||
const parsed = jsonParse<ExportedSnapshot>(completion);
|
||||
|
||||
// Consumer keying survives.
|
||||
expect(parsed.agentId).toBe('TrIaGe1234567890');
|
||||
expect(parsed.configHash).toBe('h1');
|
||||
|
||||
// The node tool's parameters survive intact, conditions (depth 9+) included.
|
||||
expect(
|
||||
parsed.config.tools[0].node.nodeParameters.options.filters.conditions[0]?.operator,
|
||||
).toEqual({ type: 'string', operation: 'equals' });
|
||||
|
||||
// Authored prose is NOT scrubbed — a seed carries it, so a case built from
|
||||
// one has to go through the scrub recipe.
|
||||
expect(parsed.config.instructions).toContain('production-down to the ops rota');
|
||||
expect(parsed.skills.skill_triage_rules.instructions).toContain('never downgrade one');
|
||||
expect(parsed.skills.skill_triage_rules.references[0].content).toContain('unusable for');
|
||||
|
||||
// Sensitive-named keys go WHOLESALE, not value-by-value: a consumer gets no
|
||||
// credential id or name back, only the marker.
|
||||
expect(parsed.config.credential).toBe('[redacted]');
|
||||
expect(parsed.config.tools[0].node).not.toHaveProperty('credentials.gmailOAuth2');
|
||||
expect((parsed.config.tools[0].node as { credentials?: unknown }).credentials).toBe(
|
||||
'[redacted]',
|
||||
);
|
||||
|
||||
// Secret patterns still fire inside ordinary prose.
|
||||
expect(parsed.config.instructions).not.toContain('sk-ant-');
|
||||
expect(parsed.config.instructions).toContain('[REDACTED]');
|
||||
|
||||
// No structural placeholders anywhere in the payload.
|
||||
expect(completion).not.toMatch(/\[array\(\d+\)\]|\[object \d+ keys\]|\[redacted-depth-limit\]/);
|
||||
});
|
||||
|
||||
it('keeps the default depth cap without the producer flag — even for a same-named span', () => {
|
||||
// Without the flag the config is destroyed at depth 5, which is the node
|
||||
// tool's parameters: the flag is load-bearing, not decorative.
|
||||
expect(exportSpan(false)).toContain('[redacted-depth-limit]');
|
||||
});
|
||||
});
|
||||
105
packages/@n8n/instance-ai/src/tracing/agent-snapshot-event.ts
Normal file
105
packages/@n8n/instance-ai/src/tracing/agent-snapshot-event.ts
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
/**
|
||||
* Trace-only `agent-snapshot` event — an agent's config + skills at a point in a
|
||||
* conversation, so eval seeding can reconstruct it. Sibling of
|
||||
* `build-workflow`'s `compiled-workflow` event: the agents-module builder is a
|
||||
* separate service, so no tool I/O in the trace carries the config.
|
||||
*
|
||||
* Payload is shaped for `instanceAiEvalSeedAgentSchema` — the consumer maps
|
||||
* `agentId` → `id` and drops the trace-only keys. It carries authored prose that
|
||||
* export scrubbing does NOT remove — scrub anything built from it before it
|
||||
* lands in a case.
|
||||
*/
|
||||
|
||||
import type { AgentJsonConfig, AgentSkill } from '@n8n/api-types';
|
||||
|
||||
import { emitTraceOnlyChildRun } from './langsmith-tracing';
|
||||
import { AGENT_SNAPSHOT_TRACE_RUN_NAME } from '../tools/tool-ids';
|
||||
import type { InstanceAiTraceContext } from '../types';
|
||||
|
||||
/** Matches the compiled-workflow gate: LangSmith rejects an oversized run whole. */
|
||||
const MAX_AGENT_SNAPSHOT_TRACE_CHARS = 1_000_000;
|
||||
|
||||
/** `attached`/`target-resolved` are BASELINES (state the turn opened on). */
|
||||
export type AgentSnapshotReason = 'attached' | 'target-resolved' | 'config-updated';
|
||||
|
||||
export interface AgentSnapshotArtifact {
|
||||
config: AgentJsonConfig;
|
||||
skills?: Record<string, AgentSkill>;
|
||||
/** The builder's own hash (`getAgentConfigHash`) — consumers dedupe on it. */
|
||||
configHash?: string | null;
|
||||
}
|
||||
|
||||
/** Emitted hashes per trace, so two triggers on one turn don't double-emit. */
|
||||
const emittedByTrace = new WeakMap<InstanceAiTraceContext, Set<string>>();
|
||||
|
||||
/** Emit one snapshot. Best-effort: tracing must never fail a turn. */
|
||||
export async function emitAgentSnapshotTraceEvent(
|
||||
tracing: InstanceAiTraceContext | undefined,
|
||||
args: {
|
||||
agentId: string;
|
||||
projectId: string;
|
||||
reason: AgentSnapshotReason;
|
||||
artifact: AgentSnapshotArtifact;
|
||||
logger?: { debug: (message: string) => void };
|
||||
},
|
||||
// `skipped` = the tracing layer exported nothing; `failed` = we threw. Only
|
||||
// the second is our bug.
|
||||
): Promise<'emitted' | 'duplicate' | 'skipped' | 'failed'> {
|
||||
const { agentId, projectId, reason, artifact } = args;
|
||||
// Claimed before the emit so two concurrent triggers can't double-write, and
|
||||
// released below unless it landed, so a skip or a throw stays retryable.
|
||||
let claim: { seen: Set<string>; key: string } | undefined;
|
||||
try {
|
||||
const hash = artifact.configHash ?? null;
|
||||
if (tracing && hash) {
|
||||
let seen = emittedByTrace.get(tracing);
|
||||
if (!seen) {
|
||||
seen = new Set();
|
||||
emittedByTrace.set(tracing, seen);
|
||||
}
|
||||
const key = `${agentId}:${hash}`;
|
||||
if (seen.has(key)) return 'duplicate';
|
||||
seen.add(key);
|
||||
claim = { seen, key };
|
||||
}
|
||||
|
||||
const payload = {
|
||||
agentId,
|
||||
projectId,
|
||||
configHash: hash,
|
||||
reason,
|
||||
config: artifact.config,
|
||||
...(artifact.skills ? { skills: artifact.skills } : {}),
|
||||
};
|
||||
const withinSizeGate = JSON.stringify(payload).length <= MAX_AGENT_SNAPSHOT_TRACE_CHARS;
|
||||
const emittedVia = await emitTraceOnlyChildRun(
|
||||
tracing,
|
||||
{
|
||||
name: AGENT_SNAPSHOT_TRACE_RUN_NAME,
|
||||
// 'chain' like the compiled-workflow event: a tool-typed run reads as
|
||||
// a real agent tool call in trace UIs.
|
||||
runType: 'chain',
|
||||
canonicalName: `instance-ai.${AGENT_SNAPSHOT_TRACE_RUN_NAME}`,
|
||||
tags: [AGENT_SNAPSHOT_TRACE_RUN_NAME],
|
||||
metadata: { agent_id: agentId, config_hash: hash, snapshot_reason: reason },
|
||||
},
|
||||
withinSizeGate
|
||||
? { outputs: payload, rawOutputs: true }
|
||||
: { outputs: { agentId, projectId, configHash: hash, reason, truncated: true } },
|
||||
);
|
||||
args.logger?.debug(
|
||||
`[agent-snapshot] ${reason} for ${agentId}: ${emittedVia}${withinSizeGate ? '' : ' (over size gate, emitted truncated marker)'}`,
|
||||
);
|
||||
if (emittedVia === 'skipped') {
|
||||
claim?.seen.delete(claim.key);
|
||||
return 'skipped';
|
||||
}
|
||||
return 'emitted';
|
||||
} catch (error) {
|
||||
claim?.seen.delete(claim.key);
|
||||
args.logger?.debug(
|
||||
`[agent-snapshot] ${reason} for ${agentId} failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
return 'failed';
|
||||
}
|
||||
}
|
||||
|
|
@ -13,6 +13,8 @@ import type {
|
|||
Workspace,
|
||||
} from '@n8n/agents';
|
||||
import type {
|
||||
AgentJsonConfig,
|
||||
AgentSkill,
|
||||
EvaluationMetric,
|
||||
TaskList,
|
||||
InstanceAiFileAttachment,
|
||||
|
|
@ -994,6 +996,15 @@ export interface InstanceAiBuilderDelegate {
|
|||
>;
|
||||
/** Current display name of the agent, or undefined when not found. */
|
||||
resolveAgentName(agentId: string): Promise<string | undefined>;
|
||||
/** Config + skills for the `agent-snapshot` trace event; `null` when the agent
|
||||
* has no config yet. Optional: the host supplies this delegate across a
|
||||
* package boundary, so an unwired host emits no snapshots instead of
|
||||
* breaking agent building. */
|
||||
readAgentArtifact?(agentId: string): Promise<{
|
||||
config: AgentJsonConfig;
|
||||
skills: Record<string, AgentSkill>;
|
||||
configHash: string | null;
|
||||
} | null>;
|
||||
}
|
||||
|
||||
// ── Local gateway status ─────────────────────────────────────────────────────
|
||||
|
|
|
|||
|
|
@ -4,11 +4,14 @@ import type {
|
|||
SerializableAgentState,
|
||||
StreamChunk,
|
||||
} from '@n8n/agents';
|
||||
import type { AgentJsonConfig, AgentSkill } from '@n8n/api-types';
|
||||
import type { User } from '@n8n/db';
|
||||
import { Like } from '@n8n/typeorm';
|
||||
import { UserError } from 'n8n-workflow';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { ForbiddenError } from '@/errors/response-errors/forbidden.error';
|
||||
import { NotFoundError } from '@/errors/response-errors/not-found.error';
|
||||
import * as checkAccess from '@/permissions.ee/check-access';
|
||||
|
||||
import type { AgentsService } from '../agents.service';
|
||||
|
|
@ -19,6 +22,9 @@ import {
|
|||
INSTANCE_AI_BUILDER_ADDENDUM,
|
||||
InstanceAiBuilderDelegateAdapterService,
|
||||
} from '../instance-ai-builder-delegate.adapter';
|
||||
import type { AgentConfigService } from '../agent-config.service';
|
||||
import { getAgentConfigHash } from '../utils/agent-config-hash';
|
||||
import type { AgentSkillsService } from '../agent-skills.service';
|
||||
import type { N8nMemory, N8nMemoryImpl } from '../integrations/n8n-memory';
|
||||
import type { AgentThreadRepository } from '../repositories/agent-thread.repository';
|
||||
|
||||
|
|
@ -27,12 +33,16 @@ function setup() {
|
|||
const agentsBuilderService = mock<AgentsBuilderService>();
|
||||
const n8nMemory = mock<N8nMemory>();
|
||||
const agentThreadRepository = mock<AgentThreadRepository>();
|
||||
const agentConfig = mock<AgentConfigService>();
|
||||
const agentSkills = mock<AgentSkillsService>();
|
||||
|
||||
const service = new InstanceAiBuilderDelegateAdapterService(
|
||||
agentsService,
|
||||
agentsBuilderService,
|
||||
n8nMemory,
|
||||
agentThreadRepository,
|
||||
agentConfig,
|
||||
agentSkills,
|
||||
);
|
||||
|
||||
const user = mock<User>({ id: 'user-1' });
|
||||
|
|
@ -47,6 +57,8 @@ function setup() {
|
|||
agentsBuilderService,
|
||||
n8nMemory,
|
||||
agentThreadRepository,
|
||||
agentConfig,
|
||||
agentSkills,
|
||||
credentialProvider,
|
||||
};
|
||||
}
|
||||
|
|
@ -353,6 +365,54 @@ describe('InstanceAiBuilderDelegateAdapterService', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('readAgentArtifact', () => {
|
||||
const CONFIG = { name: 'Support Triage' } as unknown as AgentJsonConfig;
|
||||
|
||||
it('returns the config, the skill bodies, and the builder-s own config hash', async () => {
|
||||
const { delegate, agentConfig, agentSkills } = setup();
|
||||
vi.spyOn(checkAccess, 'userHasScopes').mockResolvedValue(true);
|
||||
agentConfig.getConfig.mockResolvedValue(CONFIG);
|
||||
const skills = { skill_triage_rules: mock<AgentSkill>({ name: 'Triage rules' }) };
|
||||
agentSkills.listSkills.mockResolvedValue(skills);
|
||||
|
||||
const result = await delegate.readAgentArtifact!('agent-1');
|
||||
|
||||
expect(agentConfig.getConfig).toHaveBeenCalledWith('agent-1', 'project-1');
|
||||
expect(agentSkills.listSkills).toHaveBeenCalledWith('agent-1', 'project-1');
|
||||
// The same value read_config hands the model, so a consumer can dedupe on it.
|
||||
expect(result).toEqual({ config: CONFIG, skills, configHash: getAgentConfigHash(CONFIG) });
|
||||
});
|
||||
|
||||
it('returns null for an agent with no config yet, rather than throwing', async () => {
|
||||
// A freshly created agent the builder has not written to: nothing to
|
||||
// snapshot, and not a failure worth surfacing on a build.
|
||||
const { delegate, agentConfig, agentSkills } = setup();
|
||||
vi.spyOn(checkAccess, 'userHasScopes').mockResolvedValue(true);
|
||||
agentConfig.getConfig.mockRejectedValue(new UserError('Agent has no JSON config yet.'));
|
||||
|
||||
await expect(delegate.readAgentArtifact!('agent-1')).resolves.toBeNull();
|
||||
expect(agentSkills.listSkills).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('propagates a read failure instead of reporting it as no config', async () => {
|
||||
// A missing agent or a dead DB is not "nothing to snapshot" — the callers
|
||||
// treat a throw as no snapshot and log it, so it stays diagnosable.
|
||||
const { delegate, agentConfig } = setup();
|
||||
vi.spyOn(checkAccess, 'userHasScopes').mockResolvedValue(true);
|
||||
agentConfig.getConfig.mockRejectedValue(new NotFoundError('Agent not found'));
|
||||
|
||||
await expect(delegate.readAgentArtifact!('agent-1')).rejects.toThrow('Agent not found');
|
||||
});
|
||||
|
||||
it('rejects when the user lacks agent:read scope', async () => {
|
||||
const { delegate, agentConfig } = setup();
|
||||
vi.spyOn(checkAccess, 'userHasScopes').mockResolvedValue(false);
|
||||
|
||||
await expect(delegate.readAgentArtifact!('agent-1')).rejects.toThrow(ForbiddenError);
|
||||
expect(agentConfig.getConfig).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('createAgent', () => {
|
||||
it('enforces agent:create scope and delegates to AgentsService', async () => {
|
||||
const { delegate, agentsService } = setup();
|
||||
|
|
|
|||
|
|
@ -9,15 +9,19 @@ import type {
|
|||
} from '@n8n/instance-ai';
|
||||
import { type Scope } from '@n8n/permissions';
|
||||
import { Like } from '@n8n/typeorm';
|
||||
import { UserError } from 'n8n-workflow';
|
||||
|
||||
import { ForbiddenError } from '@/errors/response-errors/forbidden.error';
|
||||
import { userHasScopes } from '@/permissions.ee/check-access';
|
||||
|
||||
import { AgentConfigService } from './agent-config.service';
|
||||
import { AgentSkillsService } from './agent-skills.service';
|
||||
import { AgentsService } from './agents.service';
|
||||
import { AgentsBuilderService } from './builder/agents-builder.service';
|
||||
import type { InstanceAiBuilderSessionOptions } from './builder/agents-builder.service';
|
||||
import { N8nMemory } from './integrations/n8n-memory';
|
||||
import { AgentThreadRepository } from './repositories/agent-thread.repository';
|
||||
import { getAgentConfigHash } from './utils/agent-config-hash';
|
||||
|
||||
/** Prompt addendum for sub-agent runs; exported for tests. */
|
||||
export const INSTANCE_AI_BUILDER_ADDENDUM = `## Instance AI session rules
|
||||
|
|
@ -78,6 +82,8 @@ export class InstanceAiBuilderDelegateAdapterService {
|
|||
private readonly agentsBuilderService: AgentsBuilderService,
|
||||
private readonly n8nMemory: N8nMemory,
|
||||
private readonly agentThreadRepository: AgentThreadRepository,
|
||||
private readonly agentConfig: AgentConfigService,
|
||||
private readonly agentSkills: AgentSkillsService,
|
||||
) {}
|
||||
|
||||
/** Builder session options for the sub-agent surface: appends the sub-agent prompt rules. */
|
||||
|
|
@ -181,6 +187,23 @@ export class InstanceAiBuilderDelegateAdapterService {
|
|||
await assertProjectScope('agent:read');
|
||||
return (await this.agentsService.findById(agentId, projectId))?.name;
|
||||
},
|
||||
readAgentArtifact: async (agentId) => {
|
||||
await assertProjectScope('agent:read');
|
||||
// No JSON config yet (freshly created) is an empty snapshot, not an error.
|
||||
// Anything else propagates: the callers already treat a throw as "no
|
||||
// snapshot", and it gets logged there instead of vanishing here.
|
||||
const config = await this.agentConfig.getConfig(agentId, projectId).catch((error) => {
|
||||
if (error instanceof UserError) return null;
|
||||
throw error;
|
||||
});
|
||||
if (!config) return null;
|
||||
return {
|
||||
config,
|
||||
skills: await this.agentSkills.listSkills(agentId, projectId),
|
||||
// The same hash `read_config` hands the model, so consumers can dedupe.
|
||||
configHash: getAgentConfigHash(config),
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ vi.mock('@n8n/instance-ai', async () => {
|
|||
disconnect = vi.fn();
|
||||
},
|
||||
createDomainAccessTracker: vi.fn(),
|
||||
emitAgentSnapshotTraceEvent: vi.fn(async () => await Promise.resolve('emitted')),
|
||||
createSandbox: vi.fn(),
|
||||
createWorkspace: vi.fn(),
|
||||
createLazyRuntimeWorkspace: vi.fn(
|
||||
|
|
@ -196,6 +197,7 @@ import {
|
|||
resumeAgentRun,
|
||||
setupSandboxWorkspace,
|
||||
shutdownProductTelemetryProviders,
|
||||
emitAgentSnapshotTraceEvent,
|
||||
type BuilderUsageItem,
|
||||
type ManagedBackgroundTask,
|
||||
type InstanceAiTraceContext,
|
||||
|
|
@ -3926,6 +3928,117 @@ describe('InstanceAiService — editor handoff context resources', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('InstanceAiService — agent snapshots for attached agents', () => {
|
||||
type SnapshotService = {
|
||||
logger: { debug: Mock };
|
||||
snapshotAttachedAgents: (
|
||||
attachments: Array<Record<string, unknown>>,
|
||||
orchestrationContext: Record<string, unknown>,
|
||||
tracing: unknown,
|
||||
) => Promise<void>;
|
||||
};
|
||||
|
||||
const emitSnapshot = emitAgentSnapshotTraceEvent as unknown as Mock;
|
||||
const TRACING = { actorRun: { id: 'actor-1' } };
|
||||
const ARTIFACT = { config: { name: 'Support Triage' }, skills: {}, configHash: 'hash-1' };
|
||||
|
||||
function createService(): SnapshotService {
|
||||
const service = Object.create(InstanceAiService.prototype) as unknown as SnapshotService;
|
||||
service.logger = { debug: vi.fn() };
|
||||
return service;
|
||||
}
|
||||
|
||||
function makeContext(readAgentArtifact?: Mock) {
|
||||
return {
|
||||
domainContext: readAgentArtifact ? { builderDelegate: { readAgentArtifact } } : {},
|
||||
};
|
||||
}
|
||||
|
||||
beforeEach(() => emitSnapshot.mockClear());
|
||||
|
||||
it('snapshots an attached agent as it stood when the turn opened', async () => {
|
||||
const service = createService();
|
||||
const readAgentArtifact = vi.fn(async () => await Promise.resolve(ARTIFACT));
|
||||
|
||||
await service.snapshotAttachedAgents(
|
||||
[{ type: 'agent', id: 'agent-1', projectId: 'proj-1' }],
|
||||
makeContext(readAgentArtifact),
|
||||
TRACING,
|
||||
);
|
||||
|
||||
expect(readAgentArtifact).toHaveBeenCalledWith('agent-1');
|
||||
expect(emitSnapshot).toHaveBeenCalledTimes(1);
|
||||
expect(emitSnapshot.mock.calls[0][1]).toMatchObject({
|
||||
agentId: 'agent-1',
|
||||
projectId: 'proj-1',
|
||||
reason: 'attached',
|
||||
artifact: ARTIFACT,
|
||||
});
|
||||
});
|
||||
|
||||
it('ignores a workflow attachment — the workflow side has its own event', async () => {
|
||||
const service = createService();
|
||||
const readAgentArtifact = vi.fn();
|
||||
|
||||
await service.snapshotAttachedAgents(
|
||||
[{ type: 'workflow', id: 'wf-1' }],
|
||||
makeContext(readAgentArtifact),
|
||||
TRACING,
|
||||
);
|
||||
|
||||
expect(readAgentArtifact).not.toHaveBeenCalled();
|
||||
expect(emitSnapshot).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does nothing when the agents module is off, rather than failing the turn', async () => {
|
||||
const service = createService();
|
||||
|
||||
await expect(
|
||||
service.snapshotAttachedAgents(
|
||||
[{ type: 'agent', id: 'agent-1', projectId: 'proj-1' }],
|
||||
makeContext(undefined),
|
||||
TRACING,
|
||||
),
|
||||
).resolves.toBeUndefined();
|
||||
expect(emitSnapshot).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does nothing without a trace to attach the event to', async () => {
|
||||
const service = createService();
|
||||
const readAgentArtifact = vi.fn();
|
||||
|
||||
await service.snapshotAttachedAgents(
|
||||
[{ type: 'agent', id: 'agent-1', projectId: 'proj-1' }],
|
||||
makeContext(readAgentArtifact),
|
||||
undefined,
|
||||
);
|
||||
|
||||
expect(readAgentArtifact).not.toHaveBeenCalled();
|
||||
expect(emitSnapshot).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('skips an agent it cannot read and keeps going', async () => {
|
||||
const service = createService();
|
||||
const readAgentArtifact = vi.fn(async (agentId: string) =>
|
||||
agentId === 'agent-broken'
|
||||
? await Promise.reject(new Error('gone'))
|
||||
: await Promise.resolve(ARTIFACT),
|
||||
);
|
||||
|
||||
await service.snapshotAttachedAgents(
|
||||
[
|
||||
{ type: 'agent', id: 'agent-broken', projectId: 'proj-1' },
|
||||
{ type: 'agent', id: 'agent-ok', projectId: 'proj-1' },
|
||||
],
|
||||
makeContext(readAgentArtifact),
|
||||
TRACING,
|
||||
);
|
||||
|
||||
expect(emitSnapshot).toHaveBeenCalledTimes(1);
|
||||
expect(emitSnapshot.mock.calls[0][1]).toMatchObject({ agentId: 'agent-ok' });
|
||||
});
|
||||
});
|
||||
|
||||
describe('InstanceAiService — workflow verification follow-up gate', () => {
|
||||
type VerificationGateService = {
|
||||
workflowObligations: { getObligation: Mock };
|
||||
|
|
|
|||
|
|
@ -39,6 +39,7 @@ import {
|
|||
disabledInstanceAiSkillIds,
|
||||
createInstanceAiTraceContext,
|
||||
createInternalOperationTraceContext,
|
||||
emitAgentSnapshotTraceEvent,
|
||||
createInstanceAiLivenessPolicyConfig,
|
||||
InstanceAiLivenessPolicy,
|
||||
McpClientManager,
|
||||
|
|
@ -77,6 +78,7 @@ import {
|
|||
type ManagedBackgroundTask,
|
||||
type McpServerConfig,
|
||||
type ModelConfig,
|
||||
type AgentSnapshotArtifact,
|
||||
type OrchestrationContext,
|
||||
type InstanceAiTraceContext,
|
||||
type PlannedTaskGraph,
|
||||
|
|
@ -3499,6 +3501,10 @@ export class InstanceAiService {
|
|||
}
|
||||
}
|
||||
|
||||
// The builder never sees an attach-only turn, so nothing else records what
|
||||
// the attached agent looked like when the conversation opened.
|
||||
await this.snapshotAttachedAgents(contextAttachments, orchestrationContext, tracing);
|
||||
|
||||
const enrichedMessage = await this.buildMessageWithRunningTasks(threadId, message);
|
||||
const contextResourcesBlock = buildContextResourcesBlock(contextAttachments);
|
||||
|
||||
|
|
@ -5726,6 +5732,37 @@ export class InstanceAiService {
|
|||
return { status: 'limit-reached' };
|
||||
}
|
||||
|
||||
/** Snapshot every Agent the editor attached, as it stands before this turn
|
||||
* acts on it. Best-effort: it only feeds eval-seed authoring. */
|
||||
private async snapshotAttachedAgents(
|
||||
attachments: InstanceAiResourceAttachment[],
|
||||
orchestrationContext: OrchestrationContext,
|
||||
tracing: InstanceAiTraceContext | undefined,
|
||||
): Promise<void> {
|
||||
const delegate = orchestrationContext.domainContext?.builderDelegate;
|
||||
if (!delegate || !tracing) return;
|
||||
for (const attachment of attachments) {
|
||||
if (attachment.type !== 'agent') continue;
|
||||
let artifact: AgentSnapshotArtifact | null = null;
|
||||
try {
|
||||
artifact = (await delegate.readAgentArtifact?.(attachment.id)) ?? null;
|
||||
} catch (error) {
|
||||
this.logger.debug(
|
||||
`[agent-snapshot] attached read for ${attachment.id} failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
if (!artifact) continue;
|
||||
await emitAgentSnapshotTraceEvent(tracing, {
|
||||
agentId: attachment.id,
|
||||
projectId: attachment.projectId,
|
||||
reason: 'attached',
|
||||
artifact,
|
||||
logger: this.logger,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private async buildMessageWithRunningTasks(threadId: string, message: string): Promise<string> {
|
||||
return await enrichMessageWithBackgroundTasks(
|
||||
message,
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user