fix(core): Keep Agent builder sessions out of system agent session routes (no-changelog)

- findOwnedById returns top-level sessions only, so the /agents/system
  routes cannot open, change or post into a builder session.
- createThread refuses the id of a child session.
- Builder recording is best-effort: a failed admission runs the turn
  without a record, and a failed finalize does not change the turn result.
- Document that future retention must delete through deleteThread.
- Add tests for user deletion: builder sessions share the fate of their
  parent.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jaakko Husso 2026-10-08 11:50:33 +03:00
parent c30eb0ac20
commit 2a7be25420
No known key found for this signature in database
8 changed files with 516 additions and 32 deletions

View File

@ -10,6 +10,7 @@ import type { NodeCatalogService } from '@/node-catalog';
import type { InstanceAiCreditService } from '../../../instance-ai/instance-ai-credit.service';
import type { StartExecutionParams } from '../../agent-execution.service';
import type { AgentTurnExecutionService } from '../../agent-turn-execution.service';
import { AgentTurnAlreadyRunningError } from '../../agent-turn-already-running.error';
import type { AgentsService } from '../../agents.service';
import { ExecutionRecorder } from '../../execution-recorder';
import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository';
@ -1134,6 +1135,160 @@ describe('AgentsBuilderService execution records', () => {
expect.objectContaining({ agentId: 'agent-1' }),
);
});
describe('best-effort recording', () => {
it('runs the turn without a record when the admission fails', async () => {
const { service, logger, user, credentialProvider, credentialService, turnExecutionService } =
setup();
turnExecutionService.startExecution.mockRejectedValue(new AgentTurnAlreadyRunningError());
agentsSdkMocks.nextChunks.push(finishChunk);
const chunks = await drain(
service.buildAgent(
'agent-1',
'project-1',
'hi',
credentialProvider,
credentialService,
user,
{ ...baseSession, parentExecution },
),
);
expect(chunks).toEqual([finishChunk]);
expect(turnExecutionService.finalizeExecution).not.toHaveBeenCalled();
expect(agentsSdkMocks.streamCalls[0]?.message).toBe('hi');
expect(agentsSdkMocks.streamCalls[0]?.options).toEqual({
persistence: { threadId: 'ia-builder:t:agent-1', resourceId: 'user-1' },
abortSignal: baseSession.abortSignal,
recoverUsageOnAbort: true,
executionCounter,
});
expect(logger.warn).toHaveBeenCalledWith(
'Failed to start the builder execution record, running without it',
expect.objectContaining({ agentId: 'agent-1', threadId: 'ia-builder:t:agent-1' }),
);
});
it('runs the turn without a record when the memory thread cannot be saved', async () => {
const {
service,
user,
credentialProvider,
credentialService,
turnExecutionService,
memoryImplementation,
} = setup();
memoryImplementation.saveThread.mockRejectedValue(new Error('db down'));
agentsSdkMocks.nextChunks.push(finishChunk);
const chunks = await drain(
service.buildAgent(
'agent-1',
'project-1',
'hi',
credentialProvider,
credentialService,
user,
{ ...baseSession, parentExecution },
),
);
expect(chunks).toEqual([finishChunk]);
expect(turnExecutionService.startExecution).not.toHaveBeenCalled();
expect(turnExecutionService.finalizeExecution).not.toHaveBeenCalled();
});
it('resumes without a record when the admission fails', async () => {
const {
service,
user,
credentialProvider,
credentialService,
turnExecutionService,
n8nCheckpointStorage,
} = setup();
n8nCheckpointStorage.getStatus.mockResolvedValue({
status: 'active',
checkpoint: {} as never,
});
turnExecutionService.startExecution.mockRejectedValue(new Error('db down'));
agentsSdkMocks.nextChunks.push(finishChunk);
const chunks = await drain(
service.resumeBuild(
'agent-1',
'project-1',
'builder-run-1',
'tool-call-1',
{ approved: true },
credentialProvider,
credentialService,
user,
{ ...baseSession, parentExecution },
),
);
expect(chunks).toEqual([finishChunk]);
expect(turnExecutionService.finalizeExecution).not.toHaveBeenCalled();
expect(agentsSdkMocks.resumeCalls[0]?.options).toEqual({
runId: 'builder-run-1',
toolCallId: 'tool-call-1',
abortSignal: baseSession.abortSignal,
recoverUsageOnAbort: true,
executionCounter,
});
});
it('keeps the turn result when the record cannot be finalized', async () => {
const { service, logger, user, credentialProvider, credentialService, turnExecutionService } =
setup();
turnExecutionService.finalizeExecution.mockRejectedValue(new Error('db down'));
agentsSdkMocks.nextChunks.push(finishChunk);
const chunks = await drain(
service.buildAgent(
'agent-1',
'project-1',
'hi',
credentialProvider,
credentialService,
user,
{ ...baseSession, parentExecution },
),
);
expect(chunks).toEqual([finishChunk]);
expect(logger.warn).toHaveBeenCalledWith(
'Failed to finalize the builder execution record',
expect.objectContaining({ executionId: 'builder-execution-1' }),
);
});
it('rethrows the runtime error when the record cannot be finalized', async () => {
const { service, user, credentialProvider, credentialService, turnExecutionService } =
setup();
turnExecutionService.finalizeExecution.mockRejectedValue(new Error('db down'));
const streamSpy = vi
.spyOn(agentsSdkMocks.MockAgent.prototype, 'stream')
.mockRejectedValueOnce(new Error('model unavailable'));
await expect(
drain(
service.buildAgent(
'agent-1',
'project-1',
'hi',
credentialProvider,
credentialService,
user,
{ ...baseSession, parentExecution },
),
),
).rejects.toThrow('model unavailable');
streamSpy.mockRestore();
});
});
});
describe('AgentsBuilderService state owner', () => {

View File

@ -40,7 +40,10 @@ import {
AgentExecutionRepository,
type AgentExecutionLinks,
} from '../repositories/agent-execution.repository';
import { EXECUTION_METADATA_KEY } from '../types/agent-queued-message';
import {
EXECUTION_METADATA_KEY,
type AgentExecutionAdmission,
} from '../types/agent-queued-message';
import { bindExecutionInput } from '../utils/execution-input';
import { modelStreamStallOptions } from '../model-stream-stall-options';
import { buildAgentPreviewPath } from './agent-builder-preview-path';
@ -225,7 +228,15 @@ export class AgentsBuilderService {
yield* this.streamRecordedTurn(
{ agentId, user, session, parentExecution, userMessage: message },
async ({ executionId, inputMessageIds, markStarted }) => {
async (turn) => {
if (!turn) {
return await builder.stream(message, {
persistence: { threadId: session.threadId, resourceId },
...options,
...executionCounterOption(parentExecution),
});
}
const { executionId, inputMessageIds, markStarted } = turn;
markStarted();
return await builder.stream(bindExecutionInput(message, inputMessageIds), {
persistence: {
@ -310,8 +321,15 @@ export class AgentsBuilderService {
yield* this.streamRecordedTurn(
{ agentId, user, session, parentExecution, userMessage: null, resumeRunId: runId },
async ({ executionId, recorder, markStarted }) =>
await builder.resume('stream', resumeData, {
async (turn) => {
if (!turn) {
return await builder.resume('stream', resumeData, {
...options,
...executionCounterOption(parentExecution),
});
}
const { executionId, recorder, markStarted } = turn;
return await builder.resume('stream', resumeData, {
...options,
...executionCounterOption(parentExecution),
hostMetadata: { [EXECUTION_METADATA_KEY]: executionId },
@ -319,7 +337,8 @@ export class AgentsBuilderService {
markStarted();
recorder.recordHitlResponse(toolCallId, resumeData);
},
}),
});
},
);
}
@ -436,10 +455,15 @@ export class AgentsBuilderService {
* and memory belong to the same agent (see `getBuilderStateOwnerAgentId`),
* so the admission finds them without an override. The memory thread keeps
* the built agent id as information.
*
* Recording is best-effort. When the record cannot start, the turn runs
* without one (`openStream` gets `undefined`). When the record cannot
* finalize, the turn result does not change. Failures of the turn itself
* still go to the caller.
*/
private async *streamRecordedTurn(
scope: BuilderTurnScope,
openStream: (turn: RecordedBuilderTurn) => Promise<StreamResult>,
openStream: (turn: RecordedBuilderTurn | undefined) => Promise<StreamResult>,
): AsyncGenerator<StreamChunk> {
const { session, parentExecution } = scope;
const executionLinks = await this.resolveExecutionLinks(scope);
@ -469,12 +493,12 @@ export class AgentsBuilderService {
recording,
);
session.abortSignal.throwIfAborted();
await this.n8nMemory.getImplementation(parentExecution.agentId).saveThread({
id: session.threadId,
resourceId: scope.user.id,
metadata: { [BUILT_AGENT_ID_METADATA_KEY]: scope.agentId },
});
const admission = await this.turnExecutionService.startExecution(recording, recorder.startedAt);
const admission = await this.startBuilderRecord(scope, recording, recorder.startedAt);
if (!admission) {
session.abortSignal.throwIfAborted();
yield* this.streamFromAgent(await openStream(undefined));
return;
}
const { executionId } = admission;
let executionStarted = false;
let executionError: unknown;
@ -503,18 +527,56 @@ export class AgentsBuilderService {
let hitlStatus: 'suspended' | 'resumed' | undefined;
if (recorder.suspended) hitlStatus = 'suspended';
else if (scope.resumeRunId !== undefined && executionStarted) hitlStatus = 'resumed';
await this.turnExecutionService.finalizeExecution({
executionId,
executionStarted,
executionError,
params: {
...recording,
record: session.abortSignal.aborted
? { ...record, finishReason: 'cancelled', error: null }
: record,
hitlStatus,
},
try {
await this.turnExecutionService.finalizeExecution({
executionId,
executionStarted,
executionError,
params: {
...recording,
record: session.abortSignal.aborted
? { ...record, finishReason: 'cancelled', error: null }
: record,
hitlStatus,
},
});
} catch (error) {
// A throw here would replace the turn result or the turn error.
this.logger.warn('Failed to finalize the builder execution record', {
agentId: scope.agentId,
threadId: session.threadId,
executionId,
error: error instanceof Error ? error.message : String(error),
});
}
}
}
/**
* Store the built agent id on the memory thread and admit the execution.
* Returns `undefined` and logs a warning when this fails, for example when
* another turn already runs on the thread.
*/
private async startBuilderRecord(
scope: BuilderTurnScope,
recording: StartExecutionParams,
startedAt: Date,
): Promise<AgentExecutionAdmission | undefined> {
const { threadId } = scope.session;
try {
await this.n8nMemory.getImplementation(scope.parentExecution.agentId).saveThread({
id: threadId,
resourceId: scope.user.id,
metadata: { [BUILT_AGENT_ID_METADATA_KEY]: scope.agentId },
});
return await this.turnExecutionService.startExecution(recording, startedAt);
} catch (error) {
this.logger.warn('Failed to start the builder execution record, running without it', {
agentId: scope.agentId,
threadId,
error: error instanceof Error ? error.message : String(error),
});
return undefined;
}
}

View File

@ -432,13 +432,23 @@ export class AgentExecutionThreadRepository extends BaseRepository<AgentExecutio
});
}
/** One private session of a user with an agent. */
/**
* One top-level private session of a user with an agent. A child session
* (for example an Agent builder session) has the same agent and owner as
* its parent, so the parent filter keeps it out of the session routes.
*/
async findOwnedById(
agentId: string,
ownerId: string,
threadId: string,
): Promise<AgentExecutionThread | null> {
return await this.findOneBy({ id: threadId, agentId, ownerId, accessScope: 'user' });
return await this.findOneBy({
id: threadId,
agentId,
ownerId,
accessScope: 'user',
parentThreadId: IsNull(),
});
}
/**
@ -504,7 +514,13 @@ export class AgentExecutionThreadRepository extends BaseRepository<AgentExecutio
return { threads, hasMore };
}
/** Sessions of an agent last updated before `cutoff`, oldest first, for pruning. */
/**
* Sessions of an agent last updated before `cutoff`, oldest first, for
* pruning. It returns only threads of the given agent, and it does not
* filter out child sessions. A future retention job must delete through
* `deleteThread` of the execution service. Then child sessions (for
* example Agent builder sessions) go with their parent.
*/
async findByAgentUpdatedBefore(
agentId: string,
cutoff: Date,

View File

@ -44,6 +44,7 @@ const thread = {
projectId: 'project-1',
accessScope: 'user',
ownerId: 'user-1',
parentThreadId: null,
} as AgentExecutionThread;
/** The checkpoint of a turn that waits for the answer to one tool call. */
@ -1145,5 +1146,39 @@ describe('SystemAgentExecutionService', () => {
).rejects.toThrow(NotFoundError);
expect(txRunner.run).not.toHaveBeenCalled();
});
it('refuses to create a session with the id of a child session of the user', async () => {
const { service, threadRepository, txRunner } = setup();
threadRepository.findOneBy.mockResolvedValue({
...thread,
id: 'ia-builder:thread-1:agent-a',
parentThreadId: 'thread-1',
} as never);
await expect(
service.createThread({
agentId: AGENT_ID,
user,
projectId: 'project-1',
threadId: 'ia-builder:thread-1:agent-a',
}),
).rejects.toThrow(NotFoundError);
expect(txRunner.run).not.toHaveBeenCalled();
});
it('returns an existing top-level session of the user', async () => {
const { service, threadRepository, txRunner } = setup();
threadRepository.findOneBy.mockResolvedValue(thread);
await expect(
service.createThread({
agentId: AGENT_ID,
user,
projectId: 'project-1',
threadId: 'thread-1',
}),
).resolves.toBe(thread);
expect(txRunner.run).not.toHaveBeenCalled();
});
});
});

View File

@ -148,10 +148,12 @@ export class SystemAgentExecutionService {
const threadId = params.threadId ?? randomUUID();
const existing = await this.threadRepository.findOneBy({ id: threadId });
if (existing) {
// A child session (for example an Agent builder session) is not a top-level session.
if (
existing.agentId !== params.agentId ||
existing.accessScope !== 'user' ||
existing.ownerId !== params.user.id
existing.ownerId !== params.user.id ||
existing.parentThreadId !== null
) {
throw new NotFoundError('Session not found');
}

View File

@ -1,7 +1,8 @@
import type { SerializableAgentState } from '@n8n/agents';
import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
import type { User } from '@n8n/db';
import { ProjectRepository, type User } from '@n8n/db';
import { Container } from '@n8n/di';
import { NotFoundError } from '@n8n/errors';
import { v4 as uuid } from 'uuid';
import {
@ -24,8 +25,9 @@ import { AgentObservationRepository } from '@/modules/agents/repositories/agent-
import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
import { SystemAgentExecutionService } from '@/modules/agents/system-agents/system-agent-execution.service';
import { EXECUTION_METADATA_KEY } from '@/modules/agents/types/agent-queued-message';
import { UserService } from '@/services/user.service';
import { createMember } from '../shared/db/users';
import { createMember, createOwner } from '../shared/db/users';
const INSTANCE_AGENT_ID = 'test-instance-assistant';
@ -50,6 +52,7 @@ describe('Agent builder execution records', () => {
let checkpointRepo: AgentCheckpointRepository;
let observationRepo: AgentObservationRepository;
let owner: User;
let admin: User;
let workingProjectId: string;
let targetProjectId: string;
let targetAgentId: string;
@ -67,6 +70,7 @@ describe('Agent builder execution records', () => {
checkpointRepo = Container.get(AgentCheckpointRepository);
observationRepo = Container.get(AgentObservationRepository);
owner = await createMember();
admin = await createOwner();
await systemAgentService.register({
agentId: INSTANCE_AGENT_ID,
name: 'Assistant',
@ -415,10 +419,135 @@ describe('Agent builder execution records', () => {
const page = await threadRepo.findOwnedHistoryPage(INSTANCE_AGENT_ID, owner.id, 10);
expect(page.threads.map(({ id }) => id)).toEqual([parent.threadId]);
// Known gap, closed in the next stacked PR: the single-session lookup
// has no parent filter, so it returns the builder session.
expect(
await threadRepo.findOwnedById(INSTANCE_AGENT_ID, owner.id, builder.threadId),
).toMatchObject({ id: builder.threadId, parentThreadId: parent.threadId });
).toBeNull();
expect((await threadRepo.findOwnedById(INSTANCE_AGENT_ID, owner.id, parent.threadId))?.id).toBe(
parent.threadId,
);
});
it('does not open a builder session through the system agent session entry points', async () => {
const parent = await createParentTurn();
const builder = await recordBuilderTurn(parent);
await executionRepo.update({ threadId: parent.threadId }, { status: 'success' });
await expect(
systemAgentService.getThread(INSTANCE_AGENT_ID, owner, builder.threadId),
).rejects.toThrow(NotFoundError);
await expect(
systemAgentService.getUsableThread(INSTANCE_AGENT_ID, owner, builder.threadId),
).rejects.toThrow(NotFoundError);
await expect(
systemAgentService.updateThread(INSTANCE_AGENT_ID, owner, builder.threadId, {
title: 'Renamed',
}),
).rejects.toThrow(NotFoundError);
await expect(
systemAgentService.sendMessage({
agentId: INSTANCE_AGENT_ID,
user: owner,
threadId: builder.threadId,
message: 'hi',
}),
).rejects.toThrow(NotFoundError);
await expect(
systemAgentService.cancel(INSTANCE_AGENT_ID, owner, builder.threadId),
).rejects.toThrow(NotFoundError);
await expect(
systemAgentService.deleteThread(INSTANCE_AGENT_ID, owner, builder.threadId),
).rejects.toThrow(NotFoundError);
// A client cannot start a new session with the id of a builder session.
await expect(
systemAgentService.createThread({
agentId: INSTANCE_AGENT_ID,
user: owner,
projectId: workingProjectId,
threadId: builder.threadId,
}),
).rejects.toThrow(NotFoundError);
await expect(
systemAgentService.prepareChatMessage({
agentId: INSTANCE_AGENT_ID,
user: owner,
sessionId: builder.threadId,
message: 'hi',
}),
).rejects.toThrow();
await expect(
systemAgentService.prepareChatMessage({
agentId: INSTANCE_AGENT_ID,
user: owner,
projectId: workingProjectId,
sessionId: builder.threadId,
message: 'hi',
}),
).rejects.toThrow(NotFoundError);
expect(await threadRepo.findOneBy({ id: builder.threadId })).toMatchObject({
title: null,
parentThreadId: parent.threadId,
});
expect(await executionRepo.findByThreadIdOrdered(builder.threadId)).toHaveLength(1);
});
/**
* No service deletes system agent sessions when their owner is deleted.
* The database decides: the owner column is set to null, and deleting the
* personal project cascades to the sessions in it. A builder session has
* the working project of its parent, so it always shares the fate of its
* parent.
*/
describe('when the session owner is deleted', () => {
async function deleteUserWithSessions(projectOf: (user: User) => Promise<string>) {
const user = await createMember();
workingProjectId = await projectOf(user);
const parentThread = await threadRepo.save(
threadRepo.create({
id: uuid(),
agentId: INSTANCE_AGENT_ID,
agentName: 'Assistant',
projectId: workingProjectId,
accessScope: 'user',
ownerId: user.id,
}),
);
const parentExecution = await executionRepo.save(
executionRepo.create({ id: uuid(), threadId: parentThread.id, status: 'success' }),
);
const previousOwner = owner;
owner = user;
try {
const builder = await recordBuilderTurn({
threadId: parentThread.id,
executionId: parentExecution.id,
});
await Container.get(UserService).deleteUser(admin, user.id);
return { parentThreadId: parentThread.id, builderThreadId: builder.threadId };
} finally {
owner = previousOwner;
}
}
it('keeps the parent and the builder session of a team project, without an owner', async () => {
const { parentThreadId, builderThreadId } = await deleteUserWithSessions(
async (user) => (await createTeamProject(undefined, user)).id,
);
for (const id of [parentThreadId, builderThreadId]) {
expect(await threadRepo.findOneBy({ id })).toMatchObject({ ownerId: null });
}
expect(await executionRepo.findByThreadIdOrdered(builderThreadId)).toHaveLength(1);
});
it('deletes the parent and the builder session of the personal project together', async () => {
const { parentThreadId, builderThreadId } = await deleteUserWithSessions(
async (user) =>
(await Container.get(ProjectRepository).getPersonalProjectForUserOrFail(user.id)).id,
);
expect(await threadRepo.findOneBy({ id: parentThreadId })).toBeNull();
expect(await threadRepo.findOneBy({ id: builderThreadId })).toBeNull();
});
});
});

View File

@ -125,6 +125,13 @@ describe('Agent thread repository queries for system agents', () => {
expect(await threadRepo.findOwnedById(agentId, ownerId, shared.id)).toBeNull();
expect(await threadRepo.findOwnedById(agentId, ownerId, 'missing')).toBeNull();
});
it('does not return a child session with the same agent and owner', async () => {
const parent = await createThread();
const child = await createThread({ parentThreadId: parent.id, parentAgentId: agentId });
expect(await threadRepo.findOwnedById(agentId, ownerId, child.id)).toBeNull();
});
});
describe('updateOwned', () => {

View File

@ -0,0 +1,78 @@
import { randomUUID } from 'node:crypto';
import { createTeamProject, testModules } from '@n8n/backend-test-utils';
import type { User } from '@n8n/db';
import { Container } from '@n8n/di';
import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository';
import { SystemAgentExecutionService } from '@/modules/agents/system-agents/system-agent-execution.service';
import { createMember } from '../shared/db/users';
import { setupTestServer } from '../shared/utils';
const INSTANCE_AGENT_ID = 'test-instance-assistant';
beforeAll(async () => {
await testModules.loadModules(['agents']);
});
const server = setupTestServer({ endpointGroups: ['ai'] });
// An Agent builder session has the agent and owner of its parent Assistant
// session. The system agent routes must still treat it as unknown.
describe('System agent routes with an Agent builder session id', () => {
let user: User;
let parentThreadId: string;
let builderThreadId: string;
beforeAll(async () => {
await Container.get(SystemAgentExecutionService).register({
agentId: INSTANCE_AGENT_ID,
name: 'Assistant',
authorize: async () => true,
prepareTurn: vi.fn(),
});
user = await createMember();
const projectId = (await createTeamProject(undefined, user)).id;
const threadRepository = Container.get(AgentExecutionThreadRepository);
const session = {
agentId: INSTANCE_AGENT_ID,
projectId,
accessScope: 'user' as const,
ownerId: user.id,
};
parentThreadId = randomUUID();
builderThreadId = `ia-builder:${parentThreadId}:${randomUUID()}`;
await threadRepository.save(
threadRepository.create({ ...session, id: parentThreadId, agentName: 'Assistant' }),
);
await threadRepository.save(
threadRepository.create({
...session,
id: builderThreadId,
agentName: 'agent-builder',
parentThreadId,
parentAgentId: INSTANCE_AGENT_ID,
}),
);
});
it('returns the usage of the parent session', async () => {
await server
.authAgentFor(user)
.get(`/agents/system/${INSTANCE_AGENT_ID}/chat/${parentThreadId}/usage`)
.expect(200);
});
it.each(['usage', 'background-tasks'])(
'returns 404 for the %s of the builder session',
async (route) => {
await server
.authAgentFor(user)
.get(
`/agents/system/${INSTANCE_AGENT_ID}/chat/${encodeURIComponent(builderThreadId)}/${route}`,
)
.expect(404);
},
);
});