From ab2ff827a5e0081c0c3363e5a80383be41507cee Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Thu, 15 Oct 2026 23:55:47 +0300 Subject: [PATCH 01/10] feat(core): Add a system-agent layer to the Agents runtime (no-changelog) - A module can register a code-defined instance agent through a SystemAgentProvider. SystemAgentRegistry keeps the providers. - SystemAgentExecutionService manages the threads and turns of a system agent. It checks the access floor (owned thread, project:read on the working project) and then the provider. - The queue accepts the system kind only for a registered provider. The consumer sends system turns to the execution service. Co-Authored-By: Claude Opus 5.5 --- ...ent-message-queue-consumer.service.test.ts | 135 +++ .../agent-message-queue.service.test.ts | 2 + .../agent-message-queue-consumer.service.ts | 55 +- .../agents/agent-message-queue.service.ts | 47 +- .../system-agent-execution.service.test.ts | 791 ++++++++++++++++++ .../system-agents/system-agent-access.ts | 22 + .../system-agent-execution.service.ts | 599 +++++++++++++ .../system-agents/system-agent-registry.ts | 29 + .../system-agents/system-agent.types.ts | 94 +++ .../agent-execution.repository.test.ts | 123 ++- 10 files changed, 1866 insertions(+), 31 deletions(-) create mode 100644 packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts create mode 100644 packages/cli/src/modules/agents/system-agents/system-agent-access.ts create mode 100644 packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts create mode 100644 packages/cli/src/modules/agents/system-agents/system-agent-registry.ts create mode 100644 packages/cli/src/modules/agents/system-agents/system-agent.types.ts diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts index 2bf38c8a226..7adccc5d7b3 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts @@ -21,6 +21,9 @@ import type { AgentMessageEntity } from '../entities/agent-message.entity'; import type { AgentChatBridge } from '../integrations/agent-chat-bridge'; import type { ChatIntegrationService } from '../integrations/chat-integration.service'; import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository'; +import type { SystemAgentExecutionService } from '../system-agents/system-agent-execution.service'; +import { SystemAgentRegistry } from '../system-agents/system-agent-registry'; +import type { SystemAgentProvider } from '../system-agents/system-agent.types'; import type { AgentQueueDispatch, QueuedIntegrationMessage } from '../types/agent-queued-message'; vi.mock('@/permissions.ee/check-access', () => ({ userHasScopes: vi.fn() })); @@ -36,6 +39,8 @@ describe('AgentMessageQueueConsumer', () => { const integrations = mock(); const orchestrator = mock(); const sender = { send: vi.fn(), close: vi.fn(async () => {}) }; + const systemAgentExecution = mock(); + let systemAgents: SystemAgentRegistry; let consumer: AgentMessageQueueConsumer; function claim(threadId: string, integration = false): ClaimedAgentMessage { @@ -90,6 +95,7 @@ describe('AgentMessageQueueConsumer', () => { beforeEach(() => { vi.resetAllMocks(); vi.mocked(userHasScopes).mockResolvedValue(true); + systemAgents = new SystemAgentRegistry(); users.findByIdWithRole.mockResolvedValue(mock({ id: 'user', disabled: false })); repository.findThreadIds.mockResolvedValue([]); executions.getAbortSignal.mockImplementation(() => new AbortController().signal); @@ -117,6 +123,8 @@ describe('AgentMessageQueueConsumer', () => { integrations, orchestrator, mockLogger(), + systemAgents, + systemAgentExecution, ); }); @@ -440,4 +448,131 @@ describe('AgentMessageQueueConsumer', () => { expect.any(AbortSignal), ); }); + describe('system agents', () => { + function registerSystemAgent(authorized = true) { + const provider = { + agentId: 'agent', + name: 'System Agent', + authorize: vi.fn(async () => authorized), + prepareTurn: vi.fn(), + } satisfies SystemAgentProvider; + systemAgents.register(provider); + return provider; + } + + function systemClaim(threadId: string): ClaimedAgentMessage { + const item = claim(threadId); + return { + ...item, + payload: { kind: 'system', message: 'input', resourceId: 'system-chat:user' }, + item: mock({ ...item.item, payload: { kind: 'system' } }), + }; + } + + async function consumeOne(item: ClaimedAgentMessage) { + repository.findThreadIds.mockResolvedValue([item.thread.id]); + queue.claimNext.mockResolvedValueOnce(item).mockResolvedValue(null); + consumer.start(); + await vi.waitFor(() => + expect(queue.settle).toHaveBeenCalledWith(item.thread.id, item.admission.executionId), + ); + } + + it('routes a system message to the system-agent runtime after the floor check', async () => { + const provider = registerSystemAgent(); + const item = systemClaim('session'); + systemAgentExecution.consume.mockImplementation(async (_claim, _user, _signal, send) => { + send({ type: 'done', sessionId: 'session', executionId: item.admission.executionId }); + }); + + await consumeOne(item); + + expect(systemAgentExecution.consume).toHaveBeenCalledWith( + item, + expect.objectContaining({ id: 'user' }), + expect.any(AbortSignal), + sender.send, + ); + expect(chatExecutions.register).toHaveBeenCalledWith( + expect.objectContaining({ userId: 'user', surface: 'system' }), + expect.any(AbortController), + ); + // The runtime floor (`project:read`) and the provider decide access, + // not the project `agent:execute` scope. + expect(userHasScopes).toHaveBeenCalledWith( + expect.objectContaining({ id: 'user' }), + ['project:read'], + false, + { projectId: 'project' }, + ); + expect(userHasScopes).not.toHaveBeenCalledWith( + expect.anything(), + ['agent:execute'], + expect.anything(), + expect.anything(), + ); + expect(provider.authorize).toHaveBeenCalledWith( + expect.objectContaining({ id: 'user' }), + 'project', + ); + expect(testRuns.prepareDraftRun).not.toHaveBeenCalled(); + expect(chatExecutions.register).toHaveBeenCalledWith( + expect.objectContaining({ userId: 'user', threadId: 'session' }), + expect.any(AbortController), + ); + expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + expect(queue.recordFailure).not.toHaveBeenCalled(); + }); + + it('records a failure when the provider no longer authorizes the owner', async () => { + registerSystemAgent(false); + const item = systemClaim('session'); + + await consumeOne(item); + + expect(queue.recordFailure).toHaveBeenCalledWith( + item, + expect.objectContaining({ message: 'You can no longer execute this agent' }), + expect.any(AbortSignal), + ); + expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + }); + + it('records a failure when the owner can no longer read the working project', async () => { + const provider = registerSystemAgent(); + vi.mocked(userHasScopes).mockResolvedValue(false); + const item = systemClaim('session'); + + await consumeOne(item); + + expect(queue.recordFailure).toHaveBeenCalled(); + expect(provider.authorize).not.toHaveBeenCalled(); + expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + }); + + it('records a failure for a system message without a registered provider', async () => { + const item = systemClaim('session'); + + await consumeOne(item); + + expect(queue.recordFailure).toHaveBeenCalled(); + expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + }); + + it('keeps the agent:execute check for Preview messages of a system agent id', async () => { + registerSystemAgent(); + const item = claim('session'); + + await consumeOne(item); + + expect(userHasScopes).toHaveBeenCalledWith( + expect.objectContaining({ id: 'user' }), + ['agent:execute'], + false, + { projectId: 'project' }, + ); + expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expect(testRuns.prepareDraftRun).toHaveBeenCalled(); + }); + }); }); diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts index 727d5a713c2..78391fd066b 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts @@ -19,6 +19,7 @@ import type { AgentExecutionThreadRepository } from '../repositories/agent-execu import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository'; import type { AgentMessageRepository } from '../repositories/agent-message.repository'; import type { AgentRepository } from '../repositories/agent.repository'; +import { SystemAgentRegistry } from '../system-agents/system-agent-registry'; function makeThread(overrides: Partial = {}): AgentExecutionThread { return mock({ @@ -68,6 +69,7 @@ describe('AgentMessageQueueService', () => { messages, steering, settingsService, + new SystemAgentRegistry(), ); }); diff --git a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts index a2f4f51f853..8f93a0a2641 100644 --- a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts @@ -20,6 +20,11 @@ import type { AgentExecutionThread } from './entities/agent-execution-thread.ent import type { AgentChatBridge } from './integrations/agent-chat-bridge'; import { ChatIntegrationService } from './integrations/chat-integration.service'; import { AgentMessageQueueRepository } from './repositories/agent-message-queue.repository'; +import { canUseSystemAgent } from './system-agents/system-agent-access'; +import { SystemAgentExecutionService } from './system-agents/system-agent-execution.service'; +import { SystemAgentRegistry } from './system-agents/system-agent-registry'; +import type { AgentQueuedMessage } from './types/agent-queued-message'; +import type { AgentChatSurface } from './utils/agent-thread-access'; @Service() export class AgentMessageQueueConsumer { @@ -40,6 +45,8 @@ export class AgentMessageQueueConsumer { private readonly integrations: ChatIntegrationService, private readonly orchestrator: AgentExecutionOrchestratorService, private readonly logger: Logger, + private readonly systemAgents: SystemAgentRegistry, + private readonly systemAgentExecution: SystemAgentExecutionService, ) {} start(): void { @@ -137,7 +144,7 @@ export class AgentMessageQueueConsumer { payload.kind === 'integration' ? undefined : this.previewStreams.createSender(item.id); try { if (sender) { - const user = await this.getThreadOwner(thread); + const user = await this.getThreadOwner(thread, payload.kind === 'system'); this.chatExecutionService.register( { projectId: thread.projectId, @@ -145,7 +152,7 @@ export class AgentMessageQueueConsumer { threadId: thread.id, userId: user.id, executionId: admission.executionId, - surface: payload.kind === 'n8n_chat' ? 'n8n-chat' : 'preview', + surface: chatSurfaceFor(payload.kind), }, controller, ); @@ -156,11 +163,15 @@ export class AgentMessageQueueConsumer { inputMessageIds: admission.inputMessageIds, message: payload.message, }); - await this.chatExecutionService.settle(admission.executionId, async () => - payload.kind === 'n8n_chat' - ? await this.consumeN8nChat(claim, user, signal, sender.send) - : await this.consumePreview(claim, user, signal, sender.send), - ); + await this.chatExecutionService.settle(admission.executionId, async () => { + if (payload.kind === 'system') { + await this.systemAgentExecution.consume(claim, user, signal, sender.send); + } else if (payload.kind === 'n8n_chat') { + await this.consumeN8nChat(claim, user, signal, sender.send); + } else { + await this.consumePreview(claim, user, signal, sender.send); + } + }); } else { await this.consumeIntegration(claim, signal, bridge); } @@ -178,19 +189,32 @@ export class AgentMessageQueueConsumer { } } - private async getThreadOwner(thread: AgentExecutionThread): Promise { + /** + * Load the thread owner and check that they can still run the agent. A + * system agent uses the runtime floor and its provider checks instead of + * the project `agent:execute` scope. + */ + private async getThreadOwner(thread: AgentExecutionThread, systemKind: boolean): Promise { if (!thread.ownerId) throw new UserError('You can no longer execute this agent'); const user = await this.userRepository.findByIdWithRole(thread.ownerId); - if ( - !user || - user.disabled || - !(await userHasScopes(user, ['agent:execute'], false, { projectId: thread.projectId })) - ) { + if (!user || user.disabled || !(await this.canExecute(user, thread, systemKind))) { throw new UserError('You can no longer execute this agent'); } return user; } + private async canExecute( + user: User, + thread: AgentExecutionThread, + systemKind: boolean, + ): Promise { + if (!systemKind) { + return await userHasScopes(user, ['agent:execute'], false, { projectId: thread.projectId }); + } + const provider = this.systemAgents.get(thread.agentId); + return provider !== undefined && (await canUseSystemAgent(provider, user, thread.projectId)); + } + private async consumePreview( claim: ClaimedAgentMessage, user: User, @@ -293,3 +317,8 @@ export class AgentMessageQueueConsumer { await bridge.consumeQueuedMessage(payload, thread.id, claim.admission, signal, connection); } } + +function chatSurfaceFor(kind: AgentQueuedMessage['kind']): AgentChatSurface { + if (kind === 'n8n_chat') return 'n8n-chat'; + return kind === 'system' ? 'system' : 'preview'; +} diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 19225e50712..20daba4463f 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -24,6 +24,7 @@ import { AgentMessageRepository, } from './repositories/agent-message.repository'; import { AgentRepository } from './repositories/agent.repository'; +import { SystemAgentRegistry } from './system-agents/system-agent-registry'; import { acceptsSteering, type AgentExecutionAdmission, @@ -33,6 +34,7 @@ import { type QueuedUserChatMessage, } from './types/agent-queued-message'; import { + canContinueSystemThread, canContinueThreadInN8nChat, canContinueThreadInPreview, type AgentSessionMode, @@ -76,6 +78,7 @@ export class AgentMessageQueueService { private readonly messages: AgentMessageRepository, private readonly steering: AgentMessageSteeringService, private readonly settingsService: AgentsSettingsService, + private readonly systemAgents: SystemAgentRegistry, ) {} /** Save a pending message. It is durably accepted when the transaction commits. */ @@ -90,15 +93,22 @@ export class AgentMessageQueueService { }, onInserted?: (queueId: string) => void, ): Promise<{ status: 'accepted'; item: AgentMessageQueue } | { status: 'duplicate' }> { - await this.settingsService.assertEnabled(); - const agent = await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId); - if (!agent) throw new UserError('Agent not found'); const { payload } = input; + // Only the system kind reaches a system agent. Project lookups never return instance agents. + const systemAgent = + payload.kind === 'system' ? this.systemAgents.get(input.agentId) : undefined; + // System agents run even when the admin turns off custom agents. + if (!systemAgent) await this.settingsService.assertEnabled(); + const agent = + payload.kind === 'system' + ? systemAgent + : await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId); + if (!agent) throw new UserError('Agent not found'); const messageId = queuedMessageId(input.agentId, input.threadId, payload); let item: AgentMessageQueue; try { item = await this.txRunner.run({}, async (ctx) => { - await this.settingsService.assertEnabled(ctx); + if (!systemAgent) await this.settingsService.assertEnabled(ctx); await this.executionService.prepareThread( { ...input, @@ -185,6 +195,11 @@ export class AgentMessageQueueService { return await this.repository.enqueue(threadId, input.id, queueDispatch, ctx); } + /** Whether a message waits in the thread queue, hidden machine turns included. */ + async hasQueuedMessages(threadId: string): Promise { + return await this.repository.hasItems(threadId); + } + async listPending(input: PendingMessageScope): Promise { const { kind } = input; const thread = await this.threadRepository.findOneBy({ id: input.threadId }); @@ -331,9 +346,7 @@ export class AgentMessageQueueService { if ( thread.projectId !== input.projectId || thread.agentId !== input.agentId || - !(input.kind === 'n8n_chat' - ? canContinueThreadInN8nChat(thread, input.userId, source) - : canContinueThreadInPreview(thread, input.userId, source)) + !canContinueThread(input.kind, thread, input.userId, source) ) { throw new NotFoundError('Session not found'); } @@ -353,9 +366,11 @@ export class AgentMessageQueueService { ): Promise { let steeringChanged = false; const claimed = await this.txRunner.run({}, async (ctx) => { - if (!(await this.settingsService.getEnabled(ctx))) return null; const thread = await this.threadRepository.lockById(threadId, ctx); if (!thread) return null; + // The admin setting for custom agents does not apply to system agents. + if (!this.systemAgents.has(thread.agentId) && !(await this.settingsService.getEnabled(ctx))) + return null; steeringChanged = await this.steering.releaseInactive(thread, ctx); if (await this.isBlocked(thread, ctx)) return null; const active = await this.repository.findActive(threadId, ctx); @@ -493,3 +508,19 @@ export class AgentMessageQueueService { function isHidden(item: AgentMessageQueue): boolean { return item.message.origin?.hidden === true; } + +function canContinueThread( + kind: QueuedUserChatMessage['kind'], + thread: AgentExecutionThread, + userId: string, + source: string | null | undefined, +): boolean { + switch (kind) { + case 'n8n_chat': + return canContinueThreadInN8nChat(thread, userId, source); + case 'system': + return canContinueSystemThread(thread, userId, source); + case 'preview': + return canContinueThreadInPreview(thread, userId, source); + } +} diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts new file mode 100644 index 00000000000..28aa3009e89 --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -0,0 +1,791 @@ +import type { Logger } from '@n8n/backend-common'; +import type { TransactionRunner, User } from '@n8n/db'; +import { NotFoundError } from '@n8n/errors'; +import { mock } from 'vitest-mock-extended'; + +import { userHasScopes } from '@/permissions.ee/check-access'; + +import type { AgentChatExecutionService } from '../../agent-chat-execution.service'; +import type { AgentExecutionService } from '../../agent-execution.service'; +import type { + AgentMessageQueueService, + ClaimedAgentMessage, +} from '../../agent-message-queue.service'; +import type { + AgentTurnExecutionService, + AgentTurnRequest, +} from '../../agent-turn-execution.service'; +import type { AgentExecutionThread } from '../../entities/agent-execution-thread.entity'; +import type { N8NCheckpointStorage } from '../../integrations/n8n-checkpoint-storage'; +import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository'; +import type { AgentExecutionThreadRepository } from '../../repositories/agent-execution-thread.repository'; +import type { AgentRepository } from '../../repositories/agent.repository'; +import type { AgentExecutionStreamChunk } from '../../types/agent-steering'; +import { SystemAgentExecutionService } from '../system-agent-execution.service'; +import { SystemAgentRegistry } from '../system-agent-registry'; +import type { + SystemAgentProvider, + SystemAgentTurn, + SystemAgentTurnHandle, +} from '../system-agent.types'; + +vi.mock('@/permissions.ee/check-access', () => ({ userHasScopes: vi.fn() })); + +const AGENT_ID = 'test-assistant'; +const user = mock({ id: 'user-1' }); +const thread = { + id: 'thread-1', + agentId: AGENT_ID, + agentName: 'Test Assistant', + projectId: 'project-1', + accessScope: 'user', + ownerId: 'user-1', +} as AgentExecutionThread; + +/** The checkpoint of a turn that waits for the answer to one tool call. */ +function suspendedCheckpoint(resourceId = 'system-chat:user-1') { + return { + runId: 'run-1', + pendingToolCalls: { + 'tc-1': { + toolCallId: 'tc-1', + toolName: 'ask-user', + input: {}, + suspended: true, + suspendPayload: { requestId: 'req-1' }, + resumeSchema: {}, + runId: 'run-1', + }, + }, + persistence: { + threadId: 'thread-1', + resourceId, + hostMetadata: { turnOptions: { runId: 'host-run-1' } }, + }, + }; +} + +function setup(chunks: AgentExecutionStreamChunk[] = []) { + const registry = new SystemAgentRegistry(); + const threadRepository = mock(); + const executionRepository = mock(); + const turnExecutionService = mock(); + const messageQueue = mock(); + const checkpointStorage = mock(); + const executionService = mock(); + const chatExecutionService = mock(); + const agentRepository = mock(); + + const onChunk = vi.fn(); + const onSettled = vi.fn(async () => {}); + const handle: SystemAgentTurnHandle = { + agent: { name: 'runtime-agent' } as unknown as SystemAgentTurnHandle['agent'], + input: 'model input', + hostMetadata: { turnOptions: { runId: 'host-run-1' } }, + onChunk, + onSettled, + }; + const provider = { + agentId: AGENT_ID, + name: 'Test Assistant', + authorize: vi.fn(async (_user: User, _projectId: string) => true), + prepareTurn: vi.fn(async (_turn: SystemAgentTurn) => handle), + normalizeResumeData: vi.fn((data: unknown) => ({ normalized: data })), + } satisfies SystemAgentProvider; + registry.register(provider); + + threadRepository.findOwnedById.mockResolvedValue(thread); + executionRepository.findExecution.mockResolvedValue(null); + // Collect the prepared request, then stream the scripted chunks. + const prepared: AgentTurnRequest[] = []; + turnExecutionService.execute.mockImplementation(async function* (config) { + prepared.push(await config.prepare()); + yield* chunks; + }); + + const txRunner = mock(); + const service = new SystemAgentExecutionService( + mock(), + registry, + agentRepository, + threadRepository, + executionRepository, + executionService, + turnExecutionService, + messageQueue, + chatExecutionService, + checkpointStorage, + txRunner, + mock(), + ); + + return { + service, + provider, + handle, + onChunk, + onSettled, + prepared, + threadRepository, + executionRepository, + turnExecutionService, + messageQueue, + checkpointStorage, + txRunner, + executionService, + chatExecutionService, + agentRepository, + registry, + }; +} + +const suspendedChunk = { + type: 'tool-call-suspended', + runId: 'run-1', + toolCallId: 'tc-1', + toolName: 'ask-user', + input: {}, + suspendPayload: { requestId: 'req-1' }, +} as unknown as AgentExecutionStreamChunk; + +function claimFor(options: Record = {}): ClaimedAgentMessage { + return { + item: { id: 'queue-1' }, + thread, + payload: { + kind: 'system', + message: 'Build me a workflow', + resourceId: 'system-chat:user-1', + options, + }, + admission: { executionId: 'exec-1', startedAt: new Date(), inputMessageIds: ['m-1'] }, + recording: { threadId: 'thread-1' }, + } as unknown as ClaimedAgentMessage; +} + +const textChunk = { type: 'text-delta', id: 't-1', delta: 'Hello' } as AgentExecutionStreamChunk; + +describe('SystemAgentExecutionService', () => { + beforeEach(() => { + vi.mocked(userHasScopes).mockReset().mockResolvedValue(true); + }); + + describe('sendMessage', () => { + it('enqueues a system message with the provider turn options', async () => { + const { service, messageQueue, provider } = setup(); + + await service.sendMessage({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + message: 'hello', + options: { runId: 'run-1', timeZone: 'Europe/Helsinki' }, + hidden: true, + }); + + expect(provider.authorize).toHaveBeenCalledWith(user, 'project-1'); + expect(messageQueue.enqueue).toHaveBeenCalledWith({ + agentId: AGENT_ID, + projectId: 'project-1', + threadId: 'thread-1', + sessionMode: 'existing', + source: 'system_chat', + payload: { + kind: 'system', + userId: 'user-1', + message: 'hello', + resourceId: 'system-chat:user-1', + options: { runId: 'run-1', timeZone: 'Europe/Helsinki' }, + hidden: true, + }, + }); + }); + + it('refuses a user the provider does not authorize', async () => { + const { service, messageQueue, provider } = setup(); + provider.authorize.mockResolvedValue(false); + + await expect( + service.sendMessage({ agentId: AGENT_ID, user, threadId: 'thread-1', message: 'hello' }), + ).rejects.toThrow(NotFoundError); + expect(messageQueue.enqueue).not.toHaveBeenCalled(); + }); + + it('refuses a thread the user does not own', async () => { + const { service, threadRepository, messageQueue } = setup(); + threadRepository.findOwnedById.mockResolvedValue(null); + + await expect( + service.sendMessage({ agentId: AGENT_ID, user, threadId: 'thread-1', message: 'hello' }), + ).rejects.toThrow(NotFoundError); + expect(messageQueue.enqueue).not.toHaveBeenCalled(); + }); + }); + + describe('consume', () => { + it('prepares the turn and runs the provider handle on the Agents runtime', async () => { + const { service, provider, handle, turnExecutionService, prepared } = setup([textChunk]); + const signal = new AbortController().signal; + + await service.consume(claimFor({ runId: 'run-1' }), user, signal, vi.fn()); + + expect(provider.prepareTurn).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'start', + user, + thread, + resourceId: 'system-chat:user-1', + executionId: 'exec-1', + message: 'Build me a workflow', + attachments: [], + options: { runId: 'run-1' }, + }), + ); + expect(turnExecutionService.execute).toHaveBeenCalledWith( + expect.objectContaining({ + agentInstance: handle.agent, + admittedExecution: expect.objectContaining({ executionId: 'exec-1' }), + context: { projectId: 'project-1', agentId: AGENT_ID, threadId: 'thread-1' }, + }), + ); + expect(prepared[0]).toMatchObject({ + type: 'start', + input: 'model input', + options: { + persistence: { + threadId: 'thread-1', + resourceId: 'system-chat:user-1', + hostMetadata: { turnOptions: { runId: 'host-run-1' } }, + }, + }, + }); + }); + + it('hides the user message from the transcript when the handle asks for it', async () => { + const { service, handle, prepared } = setup(); + handle.hideUserMessage = true; + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(prepared[0].recording).toMatchObject({ hideUserMessageFromTranscript: true }); + }); + + it('forwards chunks to the handle and the client, and settles as completed', async () => { + const { service, onChunk, onSettled } = setup([textChunk]); + const send = vi.fn(); + + await service.consume(claimFor(), user, new AbortController().signal, send); + + expect(onChunk).toHaveBeenCalledWith(textChunk); + expect(send).toHaveBeenCalledWith(expect.objectContaining({ type: 'text-delta' })); + expect(onSettled).toHaveBeenCalledWith({ + status: 'completed', + executionId: 'exec-1', + error: undefined, + }); + expect(send).toHaveBeenLastCalledWith({ + type: 'done', + sessionId: 'thread-1', + executionId: 'exec-1', + }); + }); + + it('settles as suspended after a tool suspends, without a done event', async () => { + const { service, onSettled } = setup([suspendedChunk]); + const send = vi.fn(); + + await service.consume(claimFor(), user, new AbortController().signal, send); + + expect(onSettled).toHaveBeenCalledWith(expect.objectContaining({ status: 'suspended' })); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + }); + + it('settles as errored with the error of an error chunk', async () => { + const failure = new Error('model failed'); + const { service, onSettled } = setup([ + { type: 'error', error: failure } as AgentExecutionStreamChunk, + ]); + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(onSettled).toHaveBeenCalledWith({ + status: 'errored', + executionId: 'exec-1', + error: failure, + }); + }); + + it('settles as cancelled when the recorded execution was cancelled', async () => { + const { service, onSettled, executionRepository } = setup([textChunk]); + executionRepository.findExecution.mockResolvedValue({ status: 'cancelled' } as never); + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(onSettled).toHaveBeenCalledWith(expect.objectContaining({ status: 'cancelled' })); + }); + + it('settles as errored and rethrows when the turn throws', async () => { + const { service, onSettled, turnExecutionService } = setup(); + const failure = new Error('stream broke'); + // eslint-disable-next-line require-yield + turnExecutionService.execute.mockImplementation(async function* () { + throw failure; + }); + + await expect( + service.consume(claimFor(), user, new AbortController().signal, vi.fn()), + ).rejects.toThrow('stream broke'); + expect(onSettled).toHaveBeenCalledWith( + expect.objectContaining({ status: 'errored', error: failure }), + ); + }); + + it('does not prepare a turn when the claim is already aborted', async () => { + const { service, provider } = setup(); + const controller = new AbortController(); + controller.abort(); + + await expect(service.consume(claimFor(), user, controller.signal, vi.fn())).rejects.toThrow(); + expect(provider.prepareTurn).not.toHaveBeenCalled(); + }); + }); + + describe('resume', () => { + it('finds the suspended tool call in the checkpoint and prepares a resume turn', async () => { + const { service, provider, checkpointStorage, prepared } = setup([textChunk]); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + const result = await service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + resumeData: { approved: true }, + }); + await result.done; + + expect(result).toMatchObject({ runId: 'run-1', toolCallId: 'tc-1' }); + expect(checkpointStorage.findSuspendedForThread).toHaveBeenCalledWith(AGENT_ID, 'thread-1'); + expect(provider.prepareTurn).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'resume', + thread, + resourceId: 'system-chat:user-1', + runId: 'run-1', + toolCallId: 'tc-1', + checkpointHostMetadata: { turnOptions: { runId: 'host-run-1' } }, + resumeData: { approved: true }, + }), + ); + expect(prepared[0]).toMatchObject({ + type: 'resume', + resumeData: { approved: true }, + options: { runId: 'run-1', toolCallId: 'tc-1' }, + }); + }); + + it('refuses when no tool call waits for input', async () => { + const { service, provider, checkpointStorage } = setup(); + checkpointStorage.findSuspendedForThread.mockResolvedValue(null); + + await expect( + service.resume({ agentId: AGENT_ID, user, threadId: 'thread-1', resumeData: {} }), + ).rejects.toThrow('This action is no longer waiting for input'); + expect(provider.prepareTurn).not.toHaveBeenCalled(); + }); + + it('refuses a checkpoint stored for another memory resource', async () => { + const { service, provider, checkpointStorage } = setup(); + checkpointStorage.findSuspendedForThread.mockResolvedValue( + suspendedCheckpoint('system-chat:user-2') as never, + ); + + await expect( + service.resume({ agentId: AGENT_ID, user, threadId: 'thread-1', resumeData: {} }), + ).rejects.toThrow(NotFoundError); + expect(provider.prepareTurn).not.toHaveBeenCalled(); + }); + + it('throws NotFoundError when the provider does not authorize the user', async () => { + const { service, provider, checkpointStorage } = setup(); + provider.authorize.mockResolvedValue(false); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + await expect( + service.resume({ agentId: AGENT_ID, user, threadId: 'thread-1', resumeData: {} }), + ).rejects.toThrow(NotFoundError); + expect(provider.prepareTurn).not.toHaveBeenCalled(); + }); + }); + + describe('resumeRun', () => { + it('normalizes the client resume data with the provider before it resumes', async () => { + const { service, provider, checkpointStorage } = setup([textChunk]); + checkpointStorage.load.mockResolvedValue({ + persistence: { threadId: 'thread-1' }, + } as never); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + await service.resumeRun({ + agentId: AGENT_ID, + user, + runId: 'run-1', + toolCallId: 'tc-1', + resumeData: { kind: 'approval', approved: true }, + send: vi.fn(), + }); + + expect(checkpointStorage.load).toHaveBeenCalledWith('run-1', AGENT_ID); + expect(provider.normalizeResumeData).toHaveBeenCalledWith({ + kind: 'approval', + approved: true, + }); + expect(provider.prepareTurn).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'resume', + resumeData: { normalized: { kind: 'approval', approved: true } }, + }), + ); + }); + + it('refuses a run without a stored checkpoint', async () => { + const { service, checkpointStorage } = setup(); + checkpointStorage.load.mockResolvedValue(undefined); + + await expect( + service.resumeRun({ + agentId: AGENT_ID, + user, + runId: 'run-1', + toolCallId: 'tc-1', + resumeData: {}, + send: vi.fn(), + }), + ).rejects.toThrow('This action is no longer waiting for input'); + }); + }); + + describe('getStatus', () => { + it('reports a thread with a queued message as running', async () => { + const { service, executionRepository, checkpointStorage, messageQueue } = setup(); + executionRepository.existsRunningByThread.mockResolvedValue(false); + checkpointStorage.findSuspendedForThread.mockResolvedValue(null); + messageQueue.hasQueuedMessages.mockResolvedValue(true); + + expect((await service.getStatus(thread)).status).toBe('running'); + }); + + it('keeps a suspended thread suspended while a message waits', async () => { + const { service, executionRepository, checkpointStorage, messageQueue } = setup(); + executionRepository.existsRunningByThread.mockResolvedValue(false); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + messageQueue.hasQueuedMessages.mockResolvedValue(true); + + expect((await service.getStatus(thread)).status).toBe('suspended'); + }); + + it('reports an idle thread with an empty queue', async () => { + const { service, executionRepository, checkpointStorage, messageQueue } = setup(); + executionRepository.existsRunningByThread.mockResolvedValue(false); + checkpointStorage.findSuspendedForThread.mockResolvedValue(null); + messageQueue.hasQueuedMessages.mockResolvedValue(false); + + expect((await service.getStatus(thread)).status).toBe('idle'); + }); + }); + + describe('access floor', () => { + it('refuses a user who cannot read the working project, before the provider checks', async () => { + const { service, provider } = setup(); + vi.mocked(userHasScopes).mockResolvedValue(false); + + await expect(service.assertCanUse(AGENT_ID, user, 'project-1')).rejects.toThrow( + NotFoundError, + ); + expect(userHasScopes).toHaveBeenCalledWith(user, ['project:read'], false, { + projectId: 'project-1', + }); + expect(provider.authorize).not.toHaveBeenCalled(); + }); + + it('loads a usable thread with the floor on its working project', async () => { + const { service, provider } = setup(); + + expect(await service.getUsableThread(AGENT_ID, user, 'thread-1')).toBe(thread); + expect(userHasScopes).toHaveBeenCalledWith(user, ['project:read'], false, { + projectId: 'project-1', + }); + expect(provider.authorize).toHaveBeenCalledWith(user, 'project-1'); + }); + + it('refuses a thread the user does not own', async () => { + const { service, threadRepository } = setup(); + threadRepository.findOwnedById.mockResolvedValue(null); + + await expect(service.getUsableThread(AGENT_ID, user, 'thread-1')).rejects.toThrow( + NotFoundError, + ); + }); + }); + + describe('prepareChatMessage', () => { + it('keeps the working project of an existing session', async () => { + const { service, threadRepository } = setup(); + threadRepository.findOwnedById.mockResolvedValue(thread); + + const input = await service.prepareChatMessage({ + agentId: AGENT_ID, + user, + sessionId: 'thread-1', + message: 'hi', + }); + + expect(input).toMatchObject({ + projectId: 'project-1', + threadId: 'thread-1', + payload: { kind: 'system' }, + }); + }); + + it('refuses a new session without a working project', async () => { + const { service, threadRepository } = setup(); + threadRepository.findOwnedById.mockResolvedValue(null); + + await expect( + service.prepareChatMessage({ agentId: AGENT_ID, user, message: 'hi' }), + ).rejects.toThrow('A new session needs a working project'); + }); + + it('refuses a project that does not match the existing session', async () => { + const { service, threadRepository } = setup(); + threadRepository.findOwnedById.mockResolvedValue(thread); + + await expect( + service.prepareChatMessage({ + agentId: AGENT_ID, + user, + projectId: 'other-project', + sessionId: 'thread-1', + message: 'hi', + }), + ).rejects.toThrow(NotFoundError); + }); + }); + + describe('assertCanUse', () => { + it('throws NotFoundError for an agent id without a registered provider', async () => { + const { service } = setup(); + + await expect(service.assertCanUse('unknown-agent', user, 'project-1')).rejects.toThrow( + NotFoundError, + ); + }); + }); + describe('register', () => { + it('seeds the instance agent row before it registers the provider', async () => { + const { service, agentRepository, registry } = setup(); + const provider = { + agentId: 'other-agent', + name: 'Other Agent', + authorize: vi.fn(async () => true), + prepareTurn: vi.fn(), + } satisfies SystemAgentProvider; + + await service.register(provider); + + expect(agentRepository.ensureInstanceAgent).toHaveBeenCalledWith( + 'other-agent', + 'Other Agent', + ); + expect(registry.get('other-agent')).toBe(provider); + }); + + it('does not register a provider whose agent row cannot be seeded', async () => { + const { service, agentRepository, registry } = setup(); + agentRepository.ensureInstanceAgent.mockRejectedValue(new Error('project agent')); + + await expect( + service.register({ + agentId: 'other-agent', + name: 'Other Agent', + authorize: vi.fn(async () => true), + prepareTurn: vi.fn(), + }), + ).rejects.toThrow('project agent'); + expect(registry.has('other-agent')).toBe(false); + }); + }); + + describe('hidden turns', () => { + it('runs a hidden queued message like any other system turn', async () => { + const { service, provider, prepared } = setup([textChunk]); + const claim = claimFor({ followUp: true }); + claim.payload = { + kind: 'system', + message: 'Build me a workflow', + resourceId: 'system-chat:user-1', + options: { followUp: true }, + hidden: true, + }; + + await service.consume(claim, user, new AbortController().signal, vi.fn()); + + expect(provider.prepareTurn).toHaveBeenCalledWith( + expect.objectContaining({ type: 'start', options: { followUp: true } }), + ); + expect(prepared[0]).toMatchObject({ type: 'start', input: 'model input' }); + }); + }); + + describe('steerIntoRunningTurn', () => { + it('steers a queued system message into the running turn', async () => { + const { service, executionRepository, messageQueue } = setup(); + executionRepository.findSteerable.mockResolvedValue({ id: 'exec-1' } as never); + + expect(await service.steerIntoRunningTurn(thread, user, '7')).toBe(true); + expect(messageQueue.steer).toHaveBeenCalledWith({ + projectId: 'project-1', + agentId: AGENT_ID, + threadId: 'thread-1', + userId: 'user-1', + queueId: '7', + executionId: 'exec-1', + kind: 'system', + }); + }); + + it('leaves the message queued when no turn accepts steering', async () => { + const { service, executionRepository, messageQueue } = setup(); + executionRepository.findSteerable.mockResolvedValue(null); + + expect(await service.steerIntoRunningTurn(thread, user, '7')).toBe(false); + expect(messageQueue.steer).not.toHaveBeenCalled(); + }); + + it('leaves the message queued when the queue refuses the steer', async () => { + const { service, executionRepository, messageQueue } = setup(); + executionRepository.findSteerable.mockResolvedValue({ id: 'exec-1' } as never); + messageQueue.steer.mockRejectedValue(new Error('This message is no longer available')); + + expect(await service.steerIntoRunningTurn(thread, user, '7')).toBe(false); + }); + }); + + describe('cancel', () => { + it('requests a cancel of the latest execution of an owned thread', async () => { + const { service, executionRepository, chatExecutionService } = setup(); + executionRepository.findLatestByThreadId.mockResolvedValue({ id: 'exec-2' } as never); + chatExecutionService.requestCancel.mockResolvedValue(true); + + expect(await service.cancel(AGENT_ID, user, 'thread-1')).toBe(true); + expect(chatExecutionService.requestCancel).toHaveBeenCalledWith({ + projectId: 'project-1', + agentId: AGENT_ID, + threadId: 'thread-1', + executionId: 'exec-2', + userId: 'user-1', + surface: 'system', + }); + }); + + it('returns false for a thread without executions', async () => { + const { service, executionRepository, chatExecutionService } = setup(); + executionRepository.findLatestByThreadId.mockResolvedValue(null); + + expect(await service.cancel(AGENT_ID, user, 'thread-1')).toBe(false); + expect(chatExecutionService.requestCancel).not.toHaveBeenCalled(); + }); + + it('refuses a thread the user does not own', async () => { + const { service, threadRepository, chatExecutionService } = setup(); + threadRepository.findOwnedById.mockResolvedValue(null); + + await expect(service.cancel(AGENT_ID, user, 'thread-1')).rejects.toThrow(NotFoundError); + expect(chatExecutionService.requestCancel).not.toHaveBeenCalled(); + }); + }); + + describe('threads', () => { + it('deletes an owned thread through the Agents execution service', async () => { + const { service, executionService } = setup(); + executionService.deleteThread.mockResolvedValue(true); + + await service.deleteThread(AGENT_ID, user, 'thread-1'); + + expect(executionService.deleteThread).toHaveBeenCalledWith( + 'project-1', + AGENT_ID, + 'thread-1', + 'user-1', + ); + }); + + it('refuses to move a thread into a project the user cannot use', async () => { + const { service, provider, threadRepository } = setup(); + provider.authorize.mockImplementation( + async (_user: User, projectId: string) => projectId === 'project-1', + ); + + await expect( + service.updateThread(AGENT_ID, user, 'thread-1', { projectId: 'project-2' }), + ).rejects.toThrow(NotFoundError); + expect(threadRepository.moveOwned).not.toHaveBeenCalled(); + }); + + it('renames and moves a thread in one transaction', async () => { + const { service, threadRepository, txRunner, executionRepository, checkpointStorage } = + setup(); + const ctx = {}; + executionRepository.existsRunningByThread.mockResolvedValue(false); + checkpointStorage.findSuspendedForThread.mockResolvedValue(null); + txRunner.run.mockImplementation(async (_ctx, fn) => await fn(ctx)); + threadRepository.renameOwned.mockResolvedValue(true); + threadRepository.moveOwned.mockResolvedValue(true); + + const updated = await service.updateThread(AGENT_ID, user, 'thread-1', { + title: 'Renamed', + projectId: 'project-2', + }); + + expect(threadRepository.renameOwned).toHaveBeenCalledWith( + AGENT_ID, + user.id, + 'thread-1', + 'Renamed', + ctx, + ); + expect(threadRepository.moveOwned).toHaveBeenCalledWith( + AGENT_ID, + user.id, + 'thread-1', + 'project-2', + ctx, + ); + expect(updated).toMatchObject({ title: 'Renamed', projectId: 'project-2' }); + }); + + it('reports a thread that was deleted during the update as not found', async () => { + const { service, threadRepository, txRunner } = setup(); + txRunner.run.mockImplementation(async (ctx, fn) => await fn(ctx)); + threadRepository.renameOwned.mockResolvedValue(false); + + await expect( + service.updateThread(AGENT_ID, user, 'thread-1', { title: 'Renamed' }), + ).rejects.toThrow(NotFoundError); + }); + + it('refuses to create a session with the id of a thread of another user', async () => { + const { service, threadRepository, txRunner } = setup(); + threadRepository.findOneBy.mockResolvedValue({ ...thread, ownerId: 'user-2' } as never); + + await expect( + service.createThread({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + threadId: 'thread-1', + }), + ).rejects.toThrow(NotFoundError); + expect(txRunner.run).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-access.ts b/packages/cli/src/modules/agents/system-agents/system-agent-access.ts new file mode 100644 index 00000000000..d3b58bdecfe --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/system-agent-access.ts @@ -0,0 +1,22 @@ +import type { User } from '@n8n/db'; + +import { userHasScopes } from '@/permissions.ee/check-access'; + +import type { SystemAgentProvider } from './system-agent.types'; + +/** + * Whether the user can use a system agent in a working project. The runtime + * floor applies to every system agent: the user must be able to read the + * working project, because the agent acts in it with the user's permissions. + * The floor uses `project:read`, not `agent:execute`: `agent:execute` gates + * the custom agents of a project. The provider then adds its own checks. + * Thread ownership is checked where the thread is loaded. + */ +export async function canUseSystemAgent( + provider: SystemAgentProvider, + user: User, + projectId: string, +): Promise { + if (!(await userHasScopes(user, ['project:read'], false, { projectId }))) return false; + return await provider.authorize(user, projectId); +} diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts new file mode 100644 index 00000000000..77b8daab765 --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -0,0 +1,599 @@ +import type { SerializableAgentState } from '@n8n/agents'; +import type { AgentSseEvent } from '@n8n/api-types'; +import { Logger } from '@n8n/backend-common'; +import { TransactionRunner, type User } from '@n8n/db'; +import { Service } from '@n8n/di'; +import { NotFoundError } from '@n8n/errors'; +import { UserError } from 'n8n-workflow'; +import { randomUUID } from 'node:crypto'; + +import { AgentChatAttachmentService } from '../agent-chat-attachment.service'; +import { AgentChatExecutionService } from '../agent-chat-execution.service'; +import { AgentExecutionService, type StartExecutionParams } from '../agent-execution.service'; +import type { ClaimedAgentMessage } from '../agent-message-queue.service'; +import { AgentMessageQueueService } from '../agent-message-queue.service'; +import { emitChunkEvents } from '../agent-sse-stream'; +import { AgentTurnExecutionService, type AgentTurnRequest } from '../agent-turn-execution.service'; +import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity'; +import { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; +import { AgentExecutionRepository } from '../repositories/agent-execution.repository'; +import { AgentExecutionThreadRepository } from '../repositories/agent-execution-thread.repository'; +import { AgentRepository } from '../repositories/agent.repository'; +import type { StoredAttachmentRef } from '../types/agent-chat-attachment'; +import type { AgentExecutionAdmission } from '../types/agent-queued-message'; +import { systemChatMemoryResourceId } from '../utils/agent-memory-scope'; +import { buildInboundUserMessage } from '../utils/inbound-attachments'; +import { SYSTEM_CHAT_SOURCE } from '../utils/agent-thread-access'; +import { canUseSystemAgent } from './system-agent-access'; +import { SystemAgentRegistry } from './system-agent-registry'; +import type { + SystemAgentProvider, + SystemAgentTurnHandle, + SystemAgentTurnOptions, + SystemAgentTurnStatus, +} from './system-agent.types'; + +type SuspendedToolCall = Extract< + SerializableAgentState['pendingToolCalls'][string], + { suspended: true } +>; + +export type SystemAgentThreadStatus = 'running' | 'suspended' | 'idle'; + +function combineSignals(signal: AbortSignal, extra: AbortSignal | undefined): AbortSignal { + return extra ? AbortSignal.any([signal, extra]) : signal; +} + +/** + * Runs code-defined instance agents on the Agents runtime. A conversation is + * a private session (`accessScope = 'user'`). Its `projectId` is the working + * project. The queue, steering, checkpoints, cancel and recording are shared + * with project-agent chats, but the conversation has its own `system` chat + * surface: its own memory resource scope, transcript source and access rule. + * + * Every entry point that starts or continues a turn checks the runtime floor: + * the user owns the thread, can read its working project, and passes the + * provider's own checks. + */ +@Service() +export class SystemAgentExecutionService { + constructor( + private readonly logger: Logger, + private readonly registry: SystemAgentRegistry, + private readonly agentRepository: AgentRepository, + private readonly threadRepository: AgentExecutionThreadRepository, + private readonly executionRepository: AgentExecutionRepository, + private readonly executionService: AgentExecutionService, + private readonly turnExecutionService: AgentTurnExecutionService, + private readonly messageQueue: AgentMessageQueueService, + private readonly chatExecutionService: AgentChatExecutionService, + private readonly checkpointStorage: N8NCheckpointStorage, + private readonly txRunner: TransactionRunner, + private readonly attachmentService: AgentChatAttachmentService, + ) {} + + /** Register a provider and make sure its instance agent row exists. */ + async register(provider: SystemAgentProvider): Promise { + await this.agentRepository.ensureInstanceAgent(provider.agentId, provider.name); + this.registry.register(provider); + } + + resourceIdFor(user: Pick): string { + return systemChatMemoryResourceId(user.id); + } + + getProvider(agentId: string): SystemAgentProvider { + const provider = this.registry.get(agentId); + if (!provider) throw new NotFoundError(`Agent "${agentId}" not found`); + return provider; + } + + /** Throws NotFoundError when no provider is registered for the id. */ + assertRegistered(agentId: string): void { + this.getProvider(agentId); + } + + /** The runtime floor plus the provider checks, for a working project. */ + async assertCanUse(agentId: string, user: User, projectId: string): Promise { + const provider = this.getProvider(agentId); + // Refuse with "not found", so a user without access learns nothing about the agent. + if (!(await canUseSystemAgent(provider, user, projectId))) { + throw new NotFoundError(`Agent "${agentId}" not found`); + } + return provider; + } + + /** Load a thread the user owns and can still use: thread ownership, the floor and the provider. */ + async getUsableThread( + agentId: string, + user: User, + threadId: string, + ): Promise { + const thread = await this.getThread(agentId, user, threadId); + await this.assertCanUse(agentId, user, thread.projectId); + return thread; + } + + // ── Threads ────────────────────────────────────────────────────────────── + + async createThread(params: { + agentId: string; + user: User; + projectId: string; + threadId?: string; + title?: string; + }): Promise { + const provider = await this.assertCanUse(params.agentId, params.user, params.projectId); + const threadId = params.threadId ?? randomUUID(); + const existing = await this.threadRepository.findOneBy({ id: threadId }); + if (existing) { + if ( + existing.agentId !== params.agentId || + existing.accessScope !== 'user' || + existing.ownerId !== params.user.id + ) { + throw new NotFoundError('Session not found'); + } + return existing; + } + return await this.txRunner.run({}, async (ctx) => { + const { thread } = await this.threadRepository.findOrCreate( + threadId, + provider.agentId, + provider.name, + params.projectId, + { accessScope: 'user', ownerId: params.user.id }, + ctx, + ); + if (params.title) { + await this.threadRepository.renameOwned( + provider.agentId, + params.user.id, + thread.id, + params.title, + ctx, + ); + thread.title = params.title; + } + return thread; + }); + } + + async listThreads(agentId: string, user: User, limit?: number) { + this.assertRegistered(agentId); + return await this.threadRepository.findOwnedByAgent(agentId, user.id, { limit }); + } + + /** Load a thread the user owns. It checks ownership only, not the floor. */ + async getThread(agentId: string, user: User, threadId: string): Promise { + this.assertRegistered(agentId); + const thread = await this.threadRepository.findOwnedById(agentId, user.id, threadId); + if (!thread) throw new NotFoundError('Session not found'); + return thread; + } + + async updateThread( + agentId: string, + user: User, + threadId: string, + changes: { title?: string; projectId?: string }, + ): Promise { + const thread = await this.getThread(agentId, user, threadId); + if (changes.projectId && changes.projectId !== thread.projectId) { + await this.assertCanUse(agentId, user, changes.projectId); + if (await this.isBusy(thread)) { + throw new UserError('Wait for the current turn to finish before you change the project'); + } + } + await this.txRunner.run({}, async (ctx) => { + const { title, projectId } = changes; + const renamed = + title === undefined || + (await this.threadRepository.renameOwned(agentId, user.id, threadId, title, ctx)); + const moved = + projectId === undefined || + (await this.threadRepository.moveOwned(agentId, user.id, threadId, projectId, ctx)); + if (!renamed || !moved) throw new NotFoundError('Session not found'); + }); + if (changes.title !== undefined) thread.title = changes.title; + if (changes.projectId !== undefined) thread.projectId = changes.projectId; + return thread; + } + + /** Delete a thread with its executions, memory and stored files. */ + async deleteThread(agentId: string, user: User, threadId: string): Promise { + const thread = await this.getThread(agentId, user, threadId); + const deleted = await this.executionService.deleteThread( + thread.projectId, + agentId, + thread.id, + user.id, + ); + if (!deleted) throw new NotFoundError('Session not found'); + } + + async isBusy(thread: AgentExecutionThread): Promise { + if (await this.executionRepository.existsRunningByThread(thread.id)) return true; + return ( + (await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id)) !== null + ); + } + + /** + * Status of the thread. A queued message (for example a hidden follow-up + * turn) counts as running, so a client does not see the thread as idle + * between a turn and its follow-up. An open suspension still wins: the + * queue waits for the user to answer it. + */ + async getStatus(thread: AgentExecutionThread): Promise<{ + status: SystemAgentThreadStatus; + latestExecutionId: string | null; + checkpoint: SerializableAgentState | null; + }> { + const running = await this.executionRepository.existsRunningByThread(thread.id); + const checkpoint = running + ? null + : await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id); + const latest = await this.executionRepository.findLatestByThreadId(thread.id); + const queued = + !running && !checkpoint && (await this.messageQueue.hasQueuedMessages(thread.id)); + let status: SystemAgentThreadStatus = 'idle'; + if (running || queued) status = 'running'; + else if (checkpoint) status = 'suspended'; + return { status, latestExecutionId: latest?.id ?? null, checkpoint }; + } + + // ── Messages ───────────────────────────────────────────────────────────── + + /** + * Queue a message. The queue starts it when the conversation is idle. + * While a turn runs, the caller can steer the queued message into it. + * A `hidden` message is a machine turn: it stays out of the transcript. + */ + async sendMessage(params: { + agentId: string; + user: User; + threadId: string; + message: string; + attachments?: StoredAttachmentRef[]; + options?: SystemAgentTurnOptions; + messageId?: string; + hidden?: boolean; + }) { + const thread = await this.getUsableThread(params.agentId, params.user, params.threadId); + return await this.messageQueue.enqueue({ + agentId: params.agentId, + projectId: thread.projectId, + threadId: thread.id, + sessionMode: 'existing', + source: SYSTEM_CHAT_SOURCE, + payload: { + kind: 'system', + userId: params.user.id, + message: params.message, + resourceId: this.resourceIdFor(params.user), + ...(params.attachments?.length ? { attachments: params.attachments } : {}), + ...(params.messageId ? { messageId: params.messageId } : {}), + ...(params.options ? { options: params.options } : {}), + ...(params.hidden ? { hidden: true } : {}), + }, + }); + } + + /** + * Steer a queued message into the running turn. Returns false when no turn + * accepts steering; the message then waits in the queue. + */ + async steerIntoRunningTurn( + thread: AgentExecutionThread, + user: User, + queueId: string, + ): Promise { + const execution = await this.executionRepository.findSteerable(thread.id); + if (!execution) return false; + try { + await this.messageQueue.steer({ + projectId: thread.projectId, + agentId: thread.agentId, + threadId: thread.id, + userId: user.id, + queueId, + executionId: execution.id, + kind: 'system', + }); + return true; + } catch (error) { + this.logger.debug('Could not steer system agent message', { threadId: thread.id, error }); + return false; + } + } + + /** + * Run a claimed system-agent message. The queue consumer calls this after + * it checked the runtime floor for the thread owner. + */ + async consume( + claim: ClaimedAgentMessage, + user: User, + signal: AbortSignal, + send: (event: AgentSseEvent) => void, + ): Promise { + const { thread, admission, payload } = claim; + if (payload.kind !== 'system') return; + const provider = this.getProvider(thread.agentId); + signal.throwIfAborted(); + const { resourceId } = payload; + const attachments = payload.attachments ?? []; + const handle = await provider.prepareTurn({ + type: 'start', + user, + thread, + resourceId, + abortSignal: signal, + executionId: admission.executionId, + message: payload.message, + attachments, + options: payload.options ?? {}, + }); + const text = handle.input ?? payload.message; + if (attachments.length > 0) { + // Messages keep file references. The runtime loads bytes from binary data per call. + handle.agent.fileStore( + this.attachmentService.getFileStore( + { agentId: thread.agentId, projectId: thread.projectId }, + handle.agent.snapshot.model.provider ?? '', + ), + ); + } + const input = + attachments.length > 0 && typeof text === 'string' + ? buildInboundUserMessage(text, attachments) + : text; + const abortSignal = combineSignals(signal, handle.runOptions?.abortSignal); + await this.runTurn(provider, thread, handle, send, abortSignal, admission, async () => ({ + type: 'start', + input, + options: { + persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, + ...handle.runOptions, + abortSignal, + }, + recording: { + ...claim.recording, + ...(handle.hideUserMessage ? { hideUserMessageFromTranscript: true } : {}), + }, + })); + } + + /** Resume a suspended tool call. Runs the continuation in the background. */ + async resume(params: { + agentId: string; + user: User; + threadId: string; + resumeData: unknown; + runId?: string; + toolCallId?: string; + send?: (event: AgentSseEvent) => void; + }): Promise<{ runId: string; toolCallId: string; done: Promise }> { + const thread = await this.getUsableThread(params.agentId, params.user, params.threadId); + const provider = this.getProvider(params.agentId); + const checkpoint = await this.checkpointStorage.findSuspendedForThread( + thread.agentId, + thread.id, + ); + const pending = Object.values(checkpoint?.pendingToolCalls ?? {}).find( + (toolCall): toolCall is SuspendedToolCall => + toolCall.suspended && + (params.toolCallId === undefined || toolCall.toolCallId === params.toolCallId), + ); + if (!checkpoint || !pending) throw new UserError('This action is no longer waiting for input'); + const runId = params.runId ?? pending.runId; + const resourceId = this.resourceIdFor(params.user); + if (checkpoint.persistence?.resourceId !== resourceId) { + throw new NotFoundError('Session not found'); + } + const controller = new AbortController(); + const handle = await provider.prepareTurn({ + type: 'resume', + user: params.user, + thread, + resourceId, + abortSignal: controller.signal, + runId, + toolCallId: pending.toolCallId, + checkpointHostMetadata: checkpoint.persistence.hostMetadata ?? {}, + resumeData: params.resumeData, + }); + const recording: StartExecutionParams = { + threadId: thread.id, + agentId: thread.agentId, + agentName: thread.agentName, + projectId: thread.projectId, + resourceId, + access: { accessScope: 'user', ownerId: params.user.id }, + userMessage: null, + sessionMode: 'existing', + source: SYSTEM_CHAT_SOURCE, + }; + const abortSignal = combineSignals(controller.signal, handle.runOptions?.abortSignal); + const done = this.runTurn( + provider, + thread, + handle, + params.send ?? (() => {}), + abortSignal, + undefined, + async () => ({ + type: 'resume', + resumeData: params.resumeData, + options: { + ...handle.runOptions, + runId, + toolCallId: pending.toolCallId, + hostMetadata: handle.hostMetadata, + abortSignal, + }, + recording, + }), + ).catch((error: unknown) => { + this.logger.warn('System agent resume failed', { threadId: thread.id, error }); + }); + return { runId, toolCallId: pending.toolCallId, done }; + } + + /** + * Build the queue input for a message sent through the system-agent chat + * route. Creates the session when the client starts a new one. + */ + async prepareChatMessage(params: { + agentId: string; + user: User; + /** The working project of a new session. An existing session keeps its own. */ + projectId?: string; + sessionId?: string; + message: string; + messageId?: string; + clientContext?: Record; + storeAttachments?: ( + threadId: string, + projectId: string, + ) => Promise; + }): Promise[0]> { + this.assertRegistered(params.agentId); + const existing = params.sessionId + ? await this.threadRepository.findOwnedById(params.agentId, params.user.id, params.sessionId) + : null; + if (existing && params.projectId && existing.projectId !== params.projectId) { + throw new NotFoundError('Session not found'); + } + const projectId = existing?.projectId ?? params.projectId; + if (!projectId) throw new UserError('A new session needs a working project'); + const provider = await this.assertCanUse(params.agentId, params.user, projectId); + const thread = + existing ?? + (await this.createThread({ + agentId: params.agentId, + user: params.user, + projectId, + ...(params.sessionId ? { threadId: params.sessionId } : {}), + })); + const options = + (await provider.chatTurnOptions?.(params.user, thread, params.clientContext)) ?? {}; + const attachments = await params.storeAttachments?.(thread.id, thread.projectId); + return { + agentId: params.agentId, + projectId: thread.projectId, + threadId: thread.id, + sessionMode: 'existing', + source: SYSTEM_CHAT_SOURCE, + payload: { + kind: 'system', + userId: params.user.id, + message: params.message, + resourceId: this.resourceIdFor(params.user), + ...(params.messageId ? { messageId: params.messageId } : {}), + ...(attachments?.length ? { attachments } : {}), + options, + }, + }; + } + + /** Resume by run id, for the system-agent chat resume route. Streams to `send`. */ + async resumeRun(params: { + agentId: string; + user: User; + runId: string; + toolCallId: string; + resumeData: unknown; + send: (event: AgentSseEvent) => void; + }): Promise { + const provider = this.getProvider(params.agentId); + const state = await this.checkpointStorage.load(params.runId, params.agentId); + const threadId = state?.persistence?.threadId; + if (!threadId) throw new UserError('This action is no longer waiting for input'); + const { done } = await this.resume({ + agentId: params.agentId, + user: params.user, + threadId, + runId: params.runId, + toolCallId: params.toolCallId, + resumeData: provider.normalizeResumeData + ? provider.normalizeResumeData(params.resumeData) + : params.resumeData, + send: params.send, + }); + await done; + } + + /** Stop the running turn, or cancel the suspended one. */ + async cancel(agentId: string, user: User, threadId: string): Promise { + const thread = await this.getThread(agentId, user, threadId); + const latest = await this.executionRepository.findLatestByThreadId(thread.id); + if (!latest) return false; + return await this.chatExecutionService.requestCancel({ + projectId: thread.projectId, + agentId, + threadId: thread.id, + executionId: latest.id, + userId: user.id, + surface: 'system', + }); + } + + private async runTurn( + provider: SystemAgentProvider, + thread: AgentExecutionThread, + handle: SystemAgentTurnHandle, + send: (event: AgentSseEvent) => void, + abortSignal: AbortSignal, + admission: AgentExecutionAdmission | undefined, + prepare: () => Promise, + ): Promise { + let status: SystemAgentTurnStatus = 'completed'; + let executionId = admission?.executionId; + let error: unknown; + try { + const stream = this.turnExecutionService.execute({ + admittedExecution: admission, + agentInstance: handle.agent, + toolRegistry: handle.toolRegistry ?? new Map(), + mcpServerAttributions: new Map(), + context: { projectId: thread.projectId, agentId: provider.agentId, threadId: thread.id }, + chatSurface: 'system', + onExecutionStarted: (id) => { + executionId = id; + }, + prepare, + }); + for await (const chunk of stream) { + emitChunkEvents(chunk, send); + if (chunk.type === 'tool-call-suspended') status = 'suspended'; + if (chunk.type === 'error' && status !== 'suspended') { + status = 'errored'; + error = chunk.error; + } + handle.onChunk?.(chunk); + } + } catch (caught) { + error = caught; + status = abortSignal.aborted ? 'cancelled' : 'errored'; + throw caught; + } finally { + const execution = executionId + ? await this.executionRepository.findExecution(executionId) + : null; + if (execution?.status === 'cancelled') status = 'cancelled'; + try { + await handle.onSettled?.({ status, executionId, error }); + } catch (settleError) { + this.logger.warn('System agent turn settle hook failed', { + threadId: thread.id, + error: settleError, + }); + } + if (status !== 'suspended' && status !== 'cancelled') { + send({ type: 'done', sessionId: thread.id, executionId: executionId ?? '' }); + } + } + } +} diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-registry.ts b/packages/cli/src/modules/agents/system-agents/system-agent-registry.ts new file mode 100644 index 00000000000..0db5c3a2aee --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/system-agent-registry.ts @@ -0,0 +1,29 @@ +import { Service } from '@n8n/di'; +import { UnexpectedError } from 'n8n-workflow'; + +import type { SystemAgentProvider } from './system-agent.types'; + +/** + * Holds the code-defined instance agents. A module registers its provider at + * init through `SystemAgentExecutionService.register`. With no provider, every + * system-agent hook in the Agents runtime is inert. + */ +@Service() +export class SystemAgentRegistry { + private readonly providers = new Map(); + + register(provider: SystemAgentProvider): void { + if (this.providers.has(provider.agentId)) { + throw new UnexpectedError(`System agent "${provider.agentId}" is already registered`); + } + this.providers.set(provider.agentId, provider); + } + + get(agentId: string): SystemAgentProvider | undefined { + return this.providers.get(agentId); + } + + has(agentId: string): boolean { + return this.providers.has(agentId); + } +} diff --git a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts new file mode 100644 index 00000000000..24df7a3bab0 --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts @@ -0,0 +1,94 @@ +import type { + Agent as RuntimeAgent, + ExecutionOptions, + JSONObject, + Message, + ResumeOptions, + RunOptions, +} from '@n8n/agents'; +import type { User } from '@n8n/db'; + +import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity'; +import type { ToolRegistry } from '../tool-registry'; +import type { StoredAttachmentRef } from '../types/agent-chat-attachment'; +import type { AgentExecutionStreamChunk } from '../types/agent-steering'; + +/** Provider-defined, JSON-safe turn options. The queue stores them with the message. */ +export type SystemAgentTurnOptions = Record; + +interface SystemAgentTurnBase { + user: User; + thread: AgentExecutionThread; + resourceId: string; + abortSignal: AbortSignal; +} + +export interface SystemAgentStartTurn extends SystemAgentTurnBase { + type: 'start'; + /** Execution id when the queue admitted the turn. */ + executionId?: string; + /** The message as the user wrote it. It is the stored transcript text. */ + message: string; + attachments: StoredAttachmentRef[]; + options: SystemAgentTurnOptions; +} + +export interface SystemAgentResumeTurn extends SystemAgentTurnBase { + type: 'resume'; + runId: string; + toolCallId: string; + /** Host metadata the suspended run stored in its checkpoint. */ + checkpointHostMetadata: JSONObject; + resumeData: unknown; +} + +export type SystemAgentTurn = SystemAgentStartTurn | SystemAgentResumeTurn; + +export type SystemAgentTurnStatus = 'completed' | 'suspended' | 'errored' | 'cancelled'; + +export interface SystemAgentTurnOutcome { + status: SystemAgentTurnStatus; + executionId?: string; + error?: unknown; +} + +/** A built runtime for one turn, plus the hooks the host calls while the turn runs. */ +export interface SystemAgentTurnHandle { + agent: RuntimeAgent; + toolRegistry?: ToolRegistry; + /** Model input for a start turn. Defaults to the stored message. */ + input?: string | Message[]; + /** Stored on the checkpoint. A resume receives it back. */ + hostMetadata?: JSONObject; + /** Extra SDK run options, such as telemetry. */ + runOptions?: Partial & Partial; + /** Hide the user message from the transcript (machine turns). */ + hideUserMessage?: boolean; + onChunk?: (chunk: AgentExecutionStreamChunk) => void; + onSettled?: (outcome: SystemAgentTurnOutcome) => Promise; +} + +/** + * A code-defined, instance-level agent. The Agents runtime owns the queue, + * steering, checkpoints, HITL resume and recording. The provider builds a + * fresh runtime for each turn and adds its own access checks on top of the + * runtime floor (see `canUseSystemAgent`). + */ +export interface SystemAgentProvider { + readonly agentId: string; + readonly name: string; + /** + * Agent-specific access checks for the user in the working project. The + * runtime calls this only after its own floor check passed. + */ + authorize(user: User, projectId: string): Promise; + prepareTurn(turn: SystemAgentTurn): Promise; + /** Turn options for a message sent through the system-agent chat route. */ + chatTurnOptions?( + user: User, + thread: AgentExecutionThread, + clientContext?: Record, + ): Promise; + /** Convert a resume payload from a client into the tool's resume data. */ + normalizeResumeData?(resumeData: unknown): unknown; +} diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts index 9b2ab244d22..4f11d64a8c7 100644 --- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts @@ -95,6 +95,7 @@ import { AgentChatAttachmentRepository } from '@/modules/agents/repositories/age import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository'; import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository'; import { AgentRepository } from '@/modules/agents/repositories/agent.repository'; +import { SystemAgentRegistry } from '@/modules/agents/system-agents/system-agent-registry'; import { N8N_CHAT_PRODUCTION_SOURCE } from '@/modules/agents/utils/agent-thread-access'; import { createMember, createAdmin } from '../../shared/db/users'; @@ -163,6 +164,7 @@ describe('AgentExecutionRepository', () => { function recordingServices( memoryBackend: ReturnType = mock(), connection?: DataSource, + systemAgents = new SystemAgentRegistry(), ) { const txRunner = new TypeOrmTransactionRunner( connection ?? repository.manager.connection, @@ -243,6 +245,7 @@ describe('AgentExecutionRepository', () => { messageRepository, steering, settingsService, + systemAgents, ); const chatExecutionService = mock(); chatExecutionService.settle.mockImplementation(async (_executionId, finalize) => { @@ -2627,6 +2630,88 @@ describe('AgentExecutionRepository', () => { ); }); + describe('system agents', () => { + const systemAgentId = 'test-system-agent'; + + async function systemServices() { + const registry = new SystemAgentRegistry(); + registry.register({ + agentId: systemAgentId, + name: 'Test System Agent', + authorize: async () => true, + prepareTurn: vi.fn(), + }); + await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent'); + return recordingServices(undefined, undefined, registry); + } + + function systemInput( + threadId: string, + message: string, + sessionMode: 'new' | 'existing' = 'existing', + kind: 'system' | 'preview' = 'system', + ): Parameters[0] { + return { + agentId: systemAgentId, + projectId, + threadId, + sessionMode, + source: 'system_chat', + payload: { kind, message, userId: owner.id, resourceId: `system-chat:${owner.id}` }, + }; + } + + it('queues and claims system messages while custom agents are turned off', async () => { + const services = await systemServices(); + await services.settingsService.setEnabled(false); + const threadId = uuid(); + + const item = await enqueue(services, systemInput(threadId, 'Hello', 'new')); + expect(await services.queue.hasQueuedMessages(threadId)).toBe(true); + expect(await threadRepo.findOneByOrFail({ id: threadId })).toMatchObject({ + agentId: systemAgentId, + projectId, + accessScope: 'user', + ownerId: owner.id, + }); + + const claimed = await claim(services, threadId); + expect(claimed.item.id).toBe(item.id); + expect(claimed.payload).toMatchObject({ kind: 'system', message: 'Hello' }); + // A system thread has its own chat surface, so its recording is not a Preview chat. + expect(claimed.recording.previewChat).toBe(false); + await finish(services, claimed); + expect(await services.queue.hasQueuedMessages(threadId)).toBe(false); + }); + + it('refuses a system message for an agent without a registered provider', async () => { + const services = recordingServices(); + await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent'); + + await expect(services.queue.enqueue(systemInput(uuid(), 'Hello', 'new'))).rejects.toThrow( + 'Agent not found', + ); + }); + + it('refuses a Preview message for a system agent', async () => { + const services = await systemServices(); + + await expect( + services.queue.enqueue(systemInput(uuid(), 'Hello', 'new', 'preview')), + ).rejects.toThrow('Agent not found'); + }); + + it('refuses a system message for a project agent', async () => { + const services = await systemServices(); + await expect( + services.queue.enqueue({ + ...systemInput(uuid(), 'Hello', 'new'), + agentId, + }), + ).rejects.toThrow('Agent not found'); + }); + }); + it('consumes a hidden turn without listing it and passes turn options through unchanged', async () => { const services = recordingServices(); const threadId = uuid(); @@ -2709,17 +2794,35 @@ describe('AgentExecutionRepository', () => { expect((await claim(services, threadId)).item.id).toBe(reordered.id); }); - it('lets a user steer and reorder system agent input like Preview input', async () => { - const services = recordingServices(); + it('lets a user steer and reorder system agent input on its own surface', async () => { + // The system kind needs a registered provider and its instance agent row. + const systemAgentId = 'test-system-agent'; + const registry = new SystemAgentRegistry(); + registry.register({ + agentId: systemAgentId, + name: 'Test System Agent', + authorize: async () => true, + prepareTurn: vi.fn(), + }); + await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent'); + const services = recordingServices(undefined, undefined, registry); const threadId = uuid(); - const target = { projectId, agentId, threadId, userId: owner.id, kind: 'system' as const }; + const target = { + projectId, + agentId: systemAgentId, + threadId, + userId: owner.id, + kind: 'system' as const, + }; const systemInput = (message: string, sessionMode: 'new' | 'existing' = 'existing') => ({ ...input(threadId, message, sessionMode), + agentId: systemAgentId, + source: 'system_chat', payload: { kind: 'system' as const, message, userId: owner.id, - resourceId: `draft-chat:${owner.id}`, + resourceId: `system-chat:${owner.id}`, }, }); await enqueue(services, systemInput('Active', 'new')); @@ -2729,10 +2832,10 @@ describe('AgentExecutionRepository', () => { const c = await enqueue(services, systemInput('C')); const steered = await enqueue(services, systemInput('Steer')); - expect(await services.queue.listPending({ ...target, kind: 'preview' })).toEqual({ - items: [], - steerableExecutionId: active.admission.executionId, - }); + // A system thread has its own chat surface, so the Preview access rule refuses it. + await expect(services.queue.listPending({ ...target, kind: 'preview' })).rejects.toThrow( + 'Session not found', + ); const pending = await services.queue.listPending(target); expect(pending.steerableExecutionId).toBe(active.admission.executionId); expect(pending.items.map(({ id }) => id)).toEqual([b.id, c.id, steered.id]); @@ -2743,7 +2846,7 @@ describe('AgentExecutionRepository', () => { queueId: steered.id, executionId: active.admission.executionId, }), - ).rejects.toThrow('no longer available'); + ).rejects.toThrow('Session not found'); await services.queue.steer({ ...target, queueId: steered.id, @@ -2765,7 +2868,7 @@ describe('AgentExecutionRepository', () => { { ...target, executionId: active.admission.executionId, - resourceId: `draft-chat:${owner.id}`, + resourceId: `system-chat:${owner.id}`, }, { messages: [], lastCreatedAt: 0, completing: false, canContinue: true }, recorder, From d77f5b6bedf2947c6dd8f29987af2faa684034cc Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 20:50:10 +0300 Subject: [PATCH 02/10] fix(core): Link system agent turn signals to the turn controller (no-changelog) A system turn now runs under one controller. The provider gets its signal in prepareTurn, so a user cancel reaches the provider. The abort signal in the run options of the handle now stops the turn also after admission. The turn runtime takes an optional chat controller for this. Other chat surfaces keep their default controller. isBusy and getStatus use one status rule. A queued message now blocks a project change. Cancel checks thread ownership only, and a test covers it. Co-Authored-By: Claude Opus 5.5 --- .../agents/agent-turn-execution.service.ts | 14 +- .../system-agent-execution.service.test.ts | 98 +++++++++- .../system-agent-execution.service.ts | 86 ++++---- .../system-agents/system-agent.types.ts | 3 +- .../agents/system-agent-turn-abort.test.ts | 183 ++++++++++++++++++ 5 files changed, 346 insertions(+), 38 deletions(-) create mode 100644 packages/cli/test/integration/agents/system-agent-turn-abort.test.ts diff --git a/packages/cli/src/modules/agents/agent-turn-execution.service.ts b/packages/cli/src/modules/agents/agent-turn-execution.service.ts index 4a3b59966f7..3422a8d644c 100644 --- a/packages/cli/src/modules/agents/agent-turn-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-turn-execution.service.ts @@ -58,6 +58,12 @@ interface ExecuteTurnConfig { backgroundJobSignal?: AgentBackgroundJobSignal; /** The chat surface this turn runs under. Undefined for integrations and tasks. */ chatSurface?: AgentChatSurface; + /** + * The controller of the chat turn. Defaults to a new controller. A cancel + * aborts it, and the caller can abort it to stop the turn at any time. The + * request signal stops only a turn that is not admitted yet. + */ + chatController?: AbortController; automaticPreviewContinuation?: boolean; /** The abort signal is a wake lease, not a chat request. */ isWakeRun?: boolean; @@ -132,7 +138,11 @@ export class AgentTurnExecutionService { turn = await config.prepare(); const preparedTurn = turn; if (config.chatSurface !== undefined) { - chatControl = this.createChatExecutionControl(preparedTurn, config.chatSurface); + chatControl = this.createChatExecutionControl( + preparedTurn, + config.chatSurface, + config.chatController, + ); preparedTurn.options.abortSignal = chatControl.controller.signal; } const stream = await this.admitTurn(preparedTurn, config, recorder, state, chatControl); @@ -158,13 +168,13 @@ export class AgentTurnExecutionService { private createChatExecutionControl( turn: AgentTurnRequest, surface: AgentChatSurface, + controller = new AbortController(), ): ChatExecutionControl { const userId = turn.recording.access.ownerId; if (turn.recording.access.accessScope !== 'user' || !userId) { throw new UnexpectedError('A preview execution must have an owning user.'); } const requestSignal = turn.options.abortSignal; - const controller = new AbortController(); const abort = () => controller.abort(requestSignal?.reason); if (requestSignal?.aborted) abort(); else requestSignal?.addEventListener('abort', abort, { once: true }); diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts index 28aa3009e89..9ddb16cc751 100644 --- a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -340,6 +340,49 @@ describe('SystemAgentExecutionService', () => { ); }); + it('runs the turn under the controller of the provider turn signal', async () => { + const { service, provider, turnExecutionService, prepared } = setup([textChunk]); + const claimSignal = new AbortController().signal; + + await service.consume(claimFor(), user, claimSignal, vi.fn()); + + const [config] = turnExecutionService.execute.mock.calls[0]; + const [turn] = provider.prepareTurn.mock.calls[0]; + expect(config.chatController?.signal).toBe(turn.abortSignal); + // The claim signal is the request signal: it stops the turn only before admission. + expect(prepared[0].options.abortSignal).toBe(claimSignal); + }); + + it('stops following the provider run signal after the turn settles', async () => { + const { service, handle, turnExecutionService, prepared } = setup([textChunk]); + const providerAbort = new AbortController(); + handle.runOptions = { abortSignal: providerAbort.signal }; + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + providerAbort.abort(); + + const [config] = turnExecutionService.execute.mock.calls[0]; + expect(config.chatController?.signal.aborted).toBe(false); + expect(prepared[0].options.abortSignal).not.toBe(providerAbort.signal); + }); + + it('aborts the turn controller while the turn runs when the provider aborts', async () => { + const { service, handle, turnExecutionService } = setup(); + const providerAbort = new AbortController(); + handle.runOptions = { abortSignal: providerAbort.signal }; + let abortedInTurn: boolean | undefined; + turnExecutionService.execute.mockImplementation(async function* (config) { + await config.prepare(); + providerAbort.abort(); + abortedInTurn = config.chatController?.signal.aborted; + yield textChunk; + }); + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(abortedInTurn).toBe(true); + }); + it('does not prepare a turn when the claim is already aborted', async () => { const { service, provider } = setup(); const controller = new AbortController(); @@ -383,6 +426,26 @@ describe('SystemAgentExecutionService', () => { }); }); + it('runs the resumed turn under the controller of the provider turn signal', async () => { + const { service, provider, checkpointStorage, turnExecutionService, prepared } = setup([ + textChunk, + ]); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + const result = await service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + resumeData: { approved: true }, + }); + await result.done; + + const [config] = turnExecutionService.execute.mock.calls[0]; + const [turn] = provider.prepareTurn.mock.calls[0]; + expect(config.chatController?.signal).toBe(turn.abortSignal); + expect(prepared[0].options.abortSignal).toBeUndefined(); + }); + it('refuses when no tool call waits for input', async () => { const { service, provider, checkpointStorage } = setup(); checkpointStorage.findSuspendedForThread.mockResolvedValue(null); @@ -687,6 +750,17 @@ describe('SystemAgentExecutionService', () => { }); }); + it('cancels a turn of a user who can no longer use the working project', async () => { + const { service, provider, executionRepository, chatExecutionService } = setup(); + vi.mocked(userHasScopes).mockResolvedValue(false); + provider.authorize.mockResolvedValue(false); + executionRepository.findLatestByThreadId.mockResolvedValue({ id: 'exec-2' } as never); + chatExecutionService.requestCancel.mockResolvedValue(true); + + expect(await service.cancel(AGENT_ID, user, 'thread-1')).toBe(true); + expect(chatExecutionService.requestCancel).toHaveBeenCalled(); + }); + it('returns false for a thread without executions', async () => { const { service, executionRepository, chatExecutionService } = setup(); executionRepository.findLatestByThreadId.mockResolvedValue(null); @@ -731,12 +805,32 @@ describe('SystemAgentExecutionService', () => { expect(threadRepository.moveOwned).not.toHaveBeenCalled(); }); - it('renames and moves a thread in one transaction', async () => { - const { service, threadRepository, txRunner, executionRepository, checkpointStorage } = + it('refuses to move a thread while a message waits in its queue', async () => { + const { service, threadRepository, executionRepository, checkpointStorage, messageQueue } = setup(); + executionRepository.existsRunningByThread.mockResolvedValue(false); + checkpointStorage.findSuspendedForThread.mockResolvedValue(null); + messageQueue.hasQueuedMessages.mockResolvedValue(true); + + await expect( + service.updateThread(AGENT_ID, user, 'thread-1', { projectId: 'project-2' }), + ).rejects.toThrow('Wait for the current turn to finish before you change the project'); + expect(threadRepository.moveOwned).not.toHaveBeenCalled(); + }); + + it('renames and moves a thread in one transaction', async () => { + const { + service, + threadRepository, + txRunner, + executionRepository, + checkpointStorage, + messageQueue, + } = setup(); const ctx = {}; executionRepository.existsRunningByThread.mockResolvedValue(false); checkpointStorage.findSuspendedForThread.mockResolvedValue(null); + messageQueue.hasQueuedMessages.mockResolvedValue(false); txRunner.run.mockImplementation(async (_ctx, fn) => await fn(ctx)); threadRepository.renameOwned.mockResolvedValue(true); threadRepository.moveOwned.mockResolvedValue(true); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index 77b8daab765..a5f01638c37 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -40,8 +40,10 @@ type SuspendedToolCall = Extract< export type SystemAgentThreadStatus = 'running' | 'suspended' | 'idle'; -function combineSignals(signal: AbortSignal, extra: AbortSignal | undefined): AbortSignal { - return extra ? AbortSignal.any([signal, extra]) : signal; +/** The SDK run options of a handle, without its abort signal. `runTurn` links that signal. */ +function runOptionsOf(handle: SystemAgentTurnHandle) { + const { abortSignal: _abortSignal, ...runOptions } = handle.runOptions ?? {}; + return runOptions; } /** @@ -212,37 +214,42 @@ export class SystemAgentExecutionService { if (!deleted) throw new NotFoundError('Session not found'); } + /** Whether the thread is not idle. It uses the same rule as `getStatus`. */ async isBusy(thread: AgentExecutionThread): Promise { - if (await this.executionRepository.existsRunningByThread(thread.id)) return true; - return ( - (await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id)) !== null - ); + return (await this.resolveStatus(thread)).status !== 'idle'; } - /** - * Status of the thread. A queued message (for example a hidden follow-up - * turn) counts as running, so a client does not see the thread as idle - * between a turn and its follow-up. An open suspension still wins: the - * queue waits for the user to answer it. - */ async getStatus(thread: AgentExecutionThread): Promise<{ status: SystemAgentThreadStatus; latestExecutionId: string | null; checkpoint: SerializableAgentState | null; }> { - const running = await this.executionRepository.existsRunningByThread(thread.id); - const checkpoint = running - ? null - : await this.checkpointStorage.findSuspendedForThread(thread.agentId, thread.id); + const { status, checkpoint } = await this.resolveStatus(thread); const latest = await this.executionRepository.findLatestByThreadId(thread.id); - const queued = - !running && !checkpoint && (await this.messageQueue.hasQueuedMessages(thread.id)); - let status: SystemAgentThreadStatus = 'idle'; - if (running || queued) status = 'running'; - else if (checkpoint) status = 'suspended'; return { status, latestExecutionId: latest?.id ?? null, checkpoint }; } + /** + * A queued message (for example a hidden follow-up turn) counts as running, + * so the thread is not idle between a turn and its follow-up. An open + * suspension still wins: the queue waits for the user to answer it. + */ + private async resolveStatus(thread: AgentExecutionThread): Promise<{ + status: SystemAgentThreadStatus; + checkpoint: SerializableAgentState | null; + }> { + if (await this.executionRepository.existsRunningByThread(thread.id)) { + return { status: 'running', checkpoint: null }; + } + const checkpoint = await this.checkpointStorage.findSuspendedForThread( + thread.agentId, + thread.id, + ); + if (checkpoint) return { status: 'suspended', checkpoint }; + const queued = await this.messageQueue.hasQueuedMessages(thread.id); + return { status: queued ? 'running' : 'idle', checkpoint: null }; + } + // ── Messages ───────────────────────────────────────────────────────────── /** @@ -324,12 +331,13 @@ export class SystemAgentExecutionService { signal.throwIfAborted(); const { resourceId } = payload; const attachments = payload.attachments ?? []; + const controller = new AbortController(); const handle = await provider.prepareTurn({ type: 'start', user, thread, resourceId, - abortSignal: signal, + abortSignal: controller.signal, executionId: admission.executionId, message: payload.message, attachments, @@ -349,14 +357,14 @@ export class SystemAgentExecutionService { attachments.length > 0 && typeof text === 'string' ? buildInboundUserMessage(text, attachments) : text; - const abortSignal = combineSignals(signal, handle.runOptions?.abortSignal); - await this.runTurn(provider, thread, handle, send, abortSignal, admission, async () => ({ + await this.runTurn(provider, thread, handle, send, controller, admission, async () => ({ type: 'start', input, options: { persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, - ...handle.runOptions, - abortSignal, + ...runOptionsOf(handle), + // The claim signal stops the turn only until the runtime admits it. + abortSignal: signal, }, recording: { ...claim.recording, @@ -415,23 +423,21 @@ export class SystemAgentExecutionService { sessionMode: 'existing', source: SYSTEM_CHAT_SOURCE, }; - const abortSignal = combineSignals(controller.signal, handle.runOptions?.abortSignal); const done = this.runTurn( provider, thread, handle, params.send ?? (() => {}), - abortSignal, + controller, undefined, async () => ({ type: 'resume', resumeData: params.resumeData, options: { - ...handle.runOptions, + ...runOptionsOf(handle), runId, toolCallId: pending.toolCallId, hostMetadata: handle.hostMetadata, - abortSignal, }, recording, }), @@ -525,7 +531,10 @@ export class SystemAgentExecutionService { await done; } - /** Stop the running turn, or cancel the suspended one. */ + /** + * Stop the running turn, or cancel the suspended one. It checks thread + * ownership only, not the floor: a user who lost access can still stop a turn. + */ async cancel(agentId: string, user: User, threadId: string): Promise { const thread = await this.getThread(agentId, user, threadId); const latest = await this.executionRepository.findLatestByThreadId(thread.id); @@ -540,18 +549,27 @@ export class SystemAgentExecutionService { }); } + /** + * Run a prepared turn under `controller`, the controller the provider got + * the signal of. A cancel aborts it, and the abort signal in the run + * options of the handle stops the turn at any time. + */ private async runTurn( provider: SystemAgentProvider, thread: AgentExecutionThread, handle: SystemAgentTurnHandle, send: (event: AgentSseEvent) => void, - abortSignal: AbortSignal, + controller: AbortController, admission: AgentExecutionAdmission | undefined, prepare: () => Promise, ): Promise { let status: SystemAgentTurnStatus = 'completed'; let executionId = admission?.executionId; let error: unknown; + const providerSignal = handle.runOptions?.abortSignal; + const stopTurn = () => controller.abort(providerSignal?.reason); + if (providerSignal?.aborted) stopTurn(); + else providerSignal?.addEventListener('abort', stopTurn, { once: true }); try { const stream = this.turnExecutionService.execute({ admittedExecution: admission, @@ -560,6 +578,7 @@ export class SystemAgentExecutionService { mcpServerAttributions: new Map(), context: { projectId: thread.projectId, agentId: provider.agentId, threadId: thread.id }, chatSurface: 'system', + chatController: controller, onExecutionStarted: (id) => { executionId = id; }, @@ -576,9 +595,10 @@ export class SystemAgentExecutionService { } } catch (caught) { error = caught; - status = abortSignal.aborted ? 'cancelled' : 'errored'; + status = controller.signal.aborted ? 'cancelled' : 'errored'; throw caught; } finally { + providerSignal?.removeEventListener('abort', stopTurn); const execution = executionId ? await this.executionRepository.findExecution(executionId) : null; diff --git a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts index 24df7a3bab0..fb59aa40efb 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts @@ -20,6 +20,7 @@ interface SystemAgentTurnBase { user: User; thread: AgentExecutionThread; resourceId: string; + /** Aborts when the turn stops: a user cancel, the queue, or `runOptions.abortSignal`. */ abortSignal: AbortSignal; } @@ -60,7 +61,7 @@ export interface SystemAgentTurnHandle { input?: string | Message[]; /** Stored on the checkpoint. A resume receives it back. */ hostMetadata?: JSONObject; - /** Extra SDK run options, such as telemetry. */ + /** Extra SDK run options, such as telemetry. `abortSignal` stops the turn at any time. */ runOptions?: Partial & Partial; /** Hide the user message from the transcript (machine turns). */ hideUserMessage?: boolean; diff --git a/packages/cli/test/integration/agents/system-agent-turn-abort.test.ts b/packages/cli/test/integration/agents/system-agent-turn-abort.test.ts new file mode 100644 index 00000000000..98e564c6d9c --- /dev/null +++ b/packages/cli/test/integration/agents/system-agent-turn-abort.test.ts @@ -0,0 +1,183 @@ +import { Agent as RuntimeAgent, Tool } from '@n8n/agents'; +import { testDb, testModules } from '@n8n/backend-test-utils'; +import { ProjectRepository, type User } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { convertArrayToReadableStream, MockLanguageModelV3 } from 'ai/test'; +import { z } from 'zod'; + +import { AgentMessageQueueConsumer } from '@/modules/agents/agent-message-queue-consumer.service'; +import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage'; +import { N8nMemory } from '@/modules/agents/integrations/n8n-memory'; +import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository'; +import { SystemAgentExecutionService } from '@/modules/agents/system-agents/system-agent-execution.service'; +import type { + SystemAgentProvider, + SystemAgentTurn, + SystemAgentTurnOutcome, +} from '@/modules/agents/system-agents/system-agent.types'; + +import { createOwner } from '../shared/db/users'; + +type ModelStreamPart = Awaited< + ReturnType +>['stream'] extends ReadableStream + ? Part + : never; + +const AGENT_ID = 'test-system-agent'; +const usage = { + inputTokens: { total: 1, noCache: 1, cacheRead: 0, cacheWrite: 0 }, + outputTokens: { total: 1, text: 1, reasoning: 0 }, +}; + +/** The first model call asks for the `wait` tool. The next call answers. */ +function scriptedModel() { + let calls = 0; + return new MockLanguageModelV3({ + provider: 'mock', + modelId: 'system-turn', + doStream: async () => { + const first = calls++ === 0; + return { + stream: convertArrayToReadableStream([ + { type: 'stream-start', warnings: [] }, + ...(first + ? [{ type: 'tool-call' as const, toolCallId: 'wait-1', toolName: 'wait', input: '{}' }] + : [ + { type: 'text-start' as const, id: 'answer' }, + { type: 'text-delta' as const, id: 'answer', delta: 'Done' }, + { type: 'text-end' as const, id: 'answer' }, + ]), + { + type: 'finish', + finishReason: { unified: first ? 'tool-calls' : 'stop', raw: undefined }, + usage, + }, + ]), + }; + }, + }); +} + +async function until(check: () => boolean, timeoutMs = 10_000) { + const start = Date.now(); + while (!check()) { + if (Date.now() - start > timeoutMs) throw new Error('Timed out'); + await new Promise((resolve) => setTimeout(resolve, 20)); + } +} + +interface TurnProbe { + turnSignal?: AbortSignal; + providerAbort?: AbortController; + toolStarted?: boolean; + toolEnd?: 'aborted' | 'timeout'; + outcome?: SystemAgentTurnOutcome; +} + +describe('System agent turn abort', () => { + let probe: TurnProbe; + let service: SystemAgentExecutionService; + let owner: User; + let projectId: string; + + const provider: SystemAgentProvider = { + agentId: AGENT_ID, + name: 'Test System Agent', + authorize: async () => true, + async prepareTurn(turn: SystemAgentTurn) { + probe.turnSignal = turn.abortSignal; + probe.providerAbort = new AbortController(); + // The tool waits for the run signal, so the turn stays active until it aborts. + const wait = new Tool('wait') + .description('Wait') + .input(z.object({})) + .handler(async (_input, ctx) => { + probe.toolStarted = true; + probe.toolEnd = await new Promise((resolve) => { + const timer = setTimeout(() => resolve('timeout'), 5_000); + ctx.abortSignal?.addEventListener('abort', () => { + clearTimeout(timer); + resolve('aborted'); + }); + }); + return 'waited'; + }); + const agent = new RuntimeAgent('Test System Agent') + .instructions('Use the wait tool.') + .model(scriptedModel()) + .tool(wait) + .memory(Container.get(N8nMemory).getImplementation(AGENT_ID)) + .checkpoint(Container.get(N8NCheckpointStorage).getStorage(AGENT_ID)); + return { + agent, + runOptions: { abortSignal: probe.providerAbort.signal }, + onSettled: async (outcome) => { + probe.outcome = outcome; + }, + }; + }, + }; + + beforeAll(async () => { + await testModules.loadModules(['agents']); + await testDb.init(); + service = Container.get(SystemAgentExecutionService); + await service.register(provider); + Container.get(AgentMessageQueueConsumer).start(); + owner = await createOwner(); + projectId = (await Container.get(ProjectRepository).getPersonalProjectForUserOrFail(owner.id)) + .id; + }); + + afterAll(async () => { + Container.get(AgentMessageQueueConsumer).stop(); + await testDb.terminate(); + }); + + beforeEach(() => { + probe = {}; + }); + + /** Queue a message and wait until the turn runs the `wait` tool. */ + async function startTurn() { + const thread = await service.createThread({ agentId: AGENT_ID, user: owner, projectId }); + await service.sendMessage({ + agentId: AGENT_ID, + user: owner, + threadId: thread.id, + message: 'Go', + }); + await until(() => probe.toolStarted === true); + return thread; + } + + async function latestStatus(threadId: string) { + return (await Container.get(AgentExecutionRepository).findLatestByThreadId(threadId))?.status; + } + + it('aborts the turn signal of the provider when the user cancels the turn', async () => { + const thread = await startTurn(); + expect(probe.turnSignal?.aborted).toBe(false); + + expect(await service.cancel(AGENT_ID, owner, thread.id)).toBe(true); + await until(() => probe.outcome !== undefined); + + expect(probe.turnSignal?.aborted).toBe(true); + expect(probe.toolEnd).toBe('aborted'); + expect(probe.outcome?.status).toBe('cancelled'); + expect(await latestStatus(thread.id)).toBe('cancelled'); + }); + + it('stops an admitted turn when the provider aborts its run signal', async () => { + const thread = await startTurn(); + + probe.providerAbort?.abort(); + await until(() => probe.outcome !== undefined); + + expect(probe.toolEnd).toBe('aborted'); + expect(probe.turnSignal?.aborted).toBe(true); + expect(probe.outcome?.status).toBe('cancelled'); + expect(await latestStatus(thread.id)).toBe('cancelled'); + }); +}); From 79c25ea2184d5f20ee719b5e50a700bb60d3e49e Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:13:18 +0300 Subject: [PATCH 03/10] fix(core): Load earlier files and report failed resumes for system agents (no-changelog) Every system turn, start and resume, now attaches the attachment store. A later turn can then load the files of earlier turns. Only a completed system turn sends `done`, like Preview. `resumeRun` rejects when the resumed turn fails, so the route can send an error. The background `resume` still logs the failure. The host sets `persistence` after the provider run options, and the provider run options type omits it. Tests cover thread creation for a new session and the error event of a refused system turn. Co-Authored-By: Claude Opus 5.5 --- ...ent-message-queue-consumer.service.test.ts | 25 +-- .../system-agent-execution.service.test.ts | 170 +++++++++++++++++- .../system-agent-execution.service.ts | 76 +++++--- .../system-agents/system-agent.types.ts | 7 +- 4 files changed, 237 insertions(+), 41 deletions(-) diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts index 7adccc5d7b3..7954ce5f77b 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts @@ -524,18 +524,25 @@ describe('AgentMessageQueueConsumer', () => { expect(queue.recordFailure).not.toHaveBeenCalled(); }); + /** A refused system turn fails the item, streams an error and never runs. */ + function expectRefused(item: ReturnType) { + expect(queue.recordFailure).toHaveBeenCalledWith( + item, + expect.objectContaining({ message: 'You can no longer execute this agent' }), + expect.any(AbortSignal), + ); + expect(chatExecutions.register).not.toHaveBeenCalled(); + expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); + } + it('records a failure when the provider no longer authorizes the owner', async () => { registerSystemAgent(false); const item = systemClaim('session'); await consumeOne(item); - expect(queue.recordFailure).toHaveBeenCalledWith( - item, - expect.objectContaining({ message: 'You can no longer execute this agent' }), - expect.any(AbortSignal), - ); - expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expectRefused(item); }); it('records a failure when the owner can no longer read the working project', async () => { @@ -545,9 +552,8 @@ describe('AgentMessageQueueConsumer', () => { await consumeOne(item); - expect(queue.recordFailure).toHaveBeenCalled(); + expectRefused(item); expect(provider.authorize).not.toHaveBeenCalled(); - expect(systemAgentExecution.consume).not.toHaveBeenCalled(); }); it('records a failure for a system message without a registered provider', async () => { @@ -555,8 +561,7 @@ describe('AgentMessageQueueConsumer', () => { await consumeOne(item); - expect(queue.recordFailure).toHaveBeenCalled(); - expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expectRefused(item); }); it('keeps the agent:execute check for Preview messages of a system agent id', async () => { diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts index 9ddb16cc751..2dec618cf97 100644 --- a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -5,6 +5,7 @@ import { mock } from 'vitest-mock-extended'; import { userHasScopes } from '@/permissions.ee/check-access'; +import type { AgentChatAttachmentService } from '../../agent-chat-attachment.service'; import type { AgentChatExecutionService } from '../../agent-chat-execution.service'; import type { AgentExecutionService } from '../../agent-execution.service'; import type { @@ -75,11 +76,16 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { const executionService = mock(); const chatExecutionService = mock(); const agentRepository = mock(); + const attachmentService = mock(); const onChunk = vi.fn(); const onSettled = vi.fn(async () => {}); const handle: SystemAgentTurnHandle = { - agent: { name: 'runtime-agent' } as unknown as SystemAgentTurnHandle['agent'], + agent: { + name: 'runtime-agent', + fileStore: vi.fn(), + snapshot: { model: { provider: 'anthropic' } }, + } as unknown as SystemAgentTurnHandle['agent'], input: 'model input', hostMetadata: { turnOptions: { runId: 'host-run-1' } }, onChunk, @@ -116,7 +122,7 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { chatExecutionService, checkpointStorage, txRunner, - mock(), + attachmentService, ); return { @@ -136,6 +142,7 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { chatExecutionService, agentRepository, registry, + attachmentService, }; } @@ -305,14 +312,17 @@ describe('SystemAgentExecutionService', () => { const { service, onSettled } = setup([ { type: 'error', error: failure } as AgentExecutionStreamChunk, ]); + const send = vi.fn(); - await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + await service.consume(claimFor(), user, new AbortController().signal, send); expect(onSettled).toHaveBeenCalledWith({ status: 'errored', executionId: 'exec-1', error: failure, }); + expect(send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); }); it('settles as cancelled when the recorded execution was cancelled', async () => { @@ -332,12 +342,42 @@ describe('SystemAgentExecutionService', () => { throw failure; }); + const send = vi.fn(); + await expect( - service.consume(claimFor(), user, new AbortController().signal, vi.fn()), + service.consume(claimFor(), user, new AbortController().signal, send), ).rejects.toThrow('stream broke'); expect(onSettled).toHaveBeenCalledWith( expect.objectContaining({ status: 'errored', error: failure }), ); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + }); + + it('attaches the attachment store to a turn without attachments', async () => { + const { service, handle, attachmentService } = setup([textChunk]); + const store = mock>(); + attachmentService.getFileStore.mockReturnValue(store); + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(attachmentService.getFileStore).toHaveBeenCalledWith( + { agentId: AGENT_ID, projectId: 'project-1' }, + 'anthropic', + ); + expect(handle.agent.fileStore).toHaveBeenCalledWith(store); + }); + + it('keeps the host persistence when the provider run options set one', async () => { + const { service, handle, prepared } = setup([textChunk]); + handle.runOptions = { + persistence: { threadId: 'other-thread', resourceId: 'other-resource' }, + } as never; + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(prepared[0].options).toMatchObject({ + persistence: { threadId: 'thread-1', resourceId: 'system-chat:user-1' }, + }); }); it('runs the turn under the controller of the provider turn signal', async () => { @@ -426,6 +466,39 @@ describe('SystemAgentExecutionService', () => { }); }); + it('attaches the attachment store to a resumed turn', async () => { + const { service, handle, checkpointStorage } = setup([textChunk]); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + const result = await service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + resumeData: { approved: true }, + }); + await result.done; + + expect(handle.agent.fileStore).toHaveBeenCalled(); + }); + + it('logs a failed background resume instead of rejecting', async () => { + const { service, checkpointStorage, turnExecutionService } = setup(); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + // eslint-disable-next-line require-yield + turnExecutionService.execute.mockImplementation(async function* () { + throw new Error('resume broke'); + }); + + const result = await service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + resumeData: { approved: true }, + }); + + await expect(result.done).resolves.toBeUndefined(); + }); + it('runs the resumed turn under the controller of the provider turn signal', async () => { const { service, provider, checkpointStorage, turnExecutionService, prepared } = setup([ textChunk, @@ -510,6 +583,31 @@ describe('SystemAgentExecutionService', () => { ); }); + it('rejects when the resumed turn fails, so the route can report it', async () => { + const { service, checkpointStorage, turnExecutionService } = setup(); + checkpointStorage.load.mockResolvedValue({ + persistence: { threadId: 'thread-1' }, + } as never); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + // eslint-disable-next-line require-yield + turnExecutionService.execute.mockImplementation(async function* () { + throw new Error('resume broke'); + }); + const send = vi.fn(); + + await expect( + service.resumeRun({ + agentId: AGENT_ID, + user, + runId: 'run-1', + toolCallId: 'tc-1', + resumeData: {}, + send, + }), + ).rejects.toThrow('resume broke'); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + }); + it('refuses a run without a stored checkpoint', async () => { const { service, checkpointStorage } = setup(); checkpointStorage.load.mockResolvedValue(undefined); @@ -867,6 +965,70 @@ describe('SystemAgentExecutionService', () => { ).rejects.toThrow(NotFoundError); }); + it('creates a private session for a new thread id', async () => { + const { service, threadRepository, txRunner } = setup(); + const ctx = {}; + threadRepository.findOneBy.mockResolvedValue(null); + txRunner.run.mockImplementation(async (_ctx, fn) => await fn(ctx)); + threadRepository.findOrCreate.mockResolvedValue({ + thread: { ...thread, id: 'thread-new', projectId: 'project-1' }, + created: true, + } as never); + threadRepository.renameOwned.mockResolvedValue(true); + + const created = await service.createThread({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + threadId: 'thread-new', + title: 'First chat', + }); + + expect(threadRepository.findOrCreate).toHaveBeenCalledWith( + 'thread-new', + AGENT_ID, + 'Test Assistant', + 'project-1', + { accessScope: 'user', ownerId: 'user-1' }, + ctx, + ); + expect(threadRepository.renameOwned).toHaveBeenCalledWith( + AGENT_ID, + 'user-1', + 'thread-new', + 'First chat', + ctx, + ); + expect(created).toMatchObject({ id: 'thread-new', title: 'First chat' }); + }); + + it('queues the first message of a new session in the created thread', async () => { + const { service, threadRepository, txRunner } = setup(); + threadRepository.findOwnedById.mockResolvedValue(null); + threadRepository.findOneBy.mockResolvedValue(null); + txRunner.run.mockImplementation(async (_ctx, fn) => await fn({})); + threadRepository.findOrCreate.mockResolvedValue({ + thread: { ...thread, id: 'thread-new', projectId: 'project-1' }, + created: true, + } as never); + + const input = await service.prepareChatMessage({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + sessionId: 'thread-new', + message: 'Hello', + }); + + expect(input).toMatchObject({ + agentId: AGENT_ID, + projectId: 'project-1', + threadId: 'thread-new', + source: 'system_chat', + payload: { kind: 'system', userId: 'user-1', message: 'Hello' }, + }); + }); + it('refuses to create a session with the id of a thread of another user', async () => { const { service, threadRepository, txRunner } = setup(); threadRepository.findOneBy.mockResolvedValue({ ...thread, ownerId: 'user-2' } as never); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index a5f01638c37..59c2a06193e 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -40,6 +40,22 @@ type SuspendedToolCall = Extract< export type SystemAgentThreadStatus = 'running' | 'suspended' | 'idle'; +interface ResumeParams { + agentId: string; + user: User; + threadId: string; + resumeData: unknown; + runId?: string; + toolCallId?: string; + send?: (event: AgentSseEvent) => void; +} + +interface ResumeHandle { + runId: string; + toolCallId: string; + done: Promise; +} + /** The SDK run options of a handle, without its abort signal. `runTurn` links that signal. */ function runOptionsOf(handle: SystemAgentTurnHandle) { const { abortSignal: _abortSignal, ...runOptions } = handle.runOptions ?? {}; @@ -344,15 +360,7 @@ export class SystemAgentExecutionService { options: payload.options ?? {}, }); const text = handle.input ?? payload.message; - if (attachments.length > 0) { - // Messages keep file references. The runtime loads bytes from binary data per call. - handle.agent.fileStore( - this.attachmentService.getFileStore( - { agentId: thread.agentId, projectId: thread.projectId }, - handle.agent.snapshot.model.provider ?? '', - ), - ); - } + this.attachFileStore(handle, thread); const input = attachments.length > 0 && typeof text === 'string' ? buildInboundUserMessage(text, attachments) @@ -361,8 +369,8 @@ export class SystemAgentExecutionService { type: 'start', input, options: { - persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, ...runOptionsOf(handle), + persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, // The claim signal stops the turn only until the runtime admits it. abortSignal: signal, }, @@ -373,16 +381,20 @@ export class SystemAgentExecutionService { })); } - /** Resume a suspended tool call. Runs the continuation in the background. */ - async resume(params: { - agentId: string; - user: User; - threadId: string; - resumeData: unknown; - runId?: string; - toolCallId?: string; - send?: (event: AgentSseEvent) => void; - }): Promise<{ runId: string; toolCallId: string; done: Promise }> { + /** + * Resume a suspended tool call. Runs the continuation in the background: + * `done` settles when the turn ends and logs a failure instead of rejecting. + */ + async resume(params: ResumeParams): Promise { + const resumed = await this.startResume(params); + const done = resumed.done.catch((error: unknown) => { + this.logger.warn('System agent resume failed', { threadId: params.threadId, error }); + }); + return { ...resumed, done }; + } + + /** Start a resume. `done` rejects when the turn fails. */ + private async startResume(params: ResumeParams): Promise { const thread = await this.getUsableThread(params.agentId, params.user, params.threadId); const provider = this.getProvider(params.agentId); const checkpoint = await this.checkpointStorage.findSuspendedForThread( @@ -412,6 +424,7 @@ export class SystemAgentExecutionService { checkpointHostMetadata: checkpoint.persistence.hostMetadata ?? {}, resumeData: params.resumeData, }); + this.attachFileStore(handle, thread); const recording: StartExecutionParams = { threadId: thread.id, agentId: thread.agentId, @@ -441,9 +454,7 @@ export class SystemAgentExecutionService { }, recording, }), - ).catch((error: unknown) => { - this.logger.warn('System agent resume failed', { threadId: thread.id, error }); - }); + ); return { runId, toolCallId: pending.toolCallId, done }; } @@ -517,7 +528,8 @@ export class SystemAgentExecutionService { const state = await this.checkpointStorage.load(params.runId, params.agentId); const threadId = state?.persistence?.threadId; if (!threadId) throw new UserError('This action is no longer waiting for input'); - const { done } = await this.resume({ + // A failure rejects, so the route can send an error event. + const { done } = await this.startResume({ agentId: params.agentId, user: params.user, threadId, @@ -549,6 +561,19 @@ export class SystemAgentExecutionService { }); } + /** + * Attach the attachment store on every turn. Memory keeps file references, + * and without the store a later turn cannot load the files of earlier turns. + */ + private attachFileStore(handle: SystemAgentTurnHandle, thread: AgentExecutionThread): void { + handle.agent.fileStore( + this.attachmentService.getFileStore( + { agentId: thread.agentId, projectId: thread.projectId }, + handle.agent.snapshot.model.provider ?? '', + ), + ); + } + /** * Run a prepared turn under `controller`, the controller the provider got * the signal of. A cancel aborts it, and the abort signal in the run @@ -611,7 +636,8 @@ export class SystemAgentExecutionService { error: settleError, }); } - if (status !== 'suspended' && status !== 'cancelled') { + // Like Preview, only a completed turn sends `done`. An error has its own event. + if (status === 'completed') { send({ type: 'done', sessionId: thread.id, executionId: executionId ?? '' }); } } diff --git a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts index fb59aa40efb..75933dc5043 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts @@ -61,8 +61,11 @@ export interface SystemAgentTurnHandle { input?: string | Message[]; /** Stored on the checkpoint. A resume receives it back. */ hostMetadata?: JSONObject; - /** Extra SDK run options, such as telemetry. `abortSignal` stops the turn at any time. */ - runOptions?: Partial & Partial; + /** + * Extra SDK run options, such as telemetry. `abortSignal` stops the turn at + * any time. The host sets `persistence`, so a provider cannot change it. + */ + runOptions?: Omit & Partial, 'persistence'>; /** Hide the user message from the transcript (machine turns). */ hideUserMessage?: boolean; onChunk?: (chunk: AgentExecutionStreamChunk) => void; From 509b359a6d6aa910a2031261fa68aad2c286e8af Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:48:01 +0300 Subject: [PATCH 04/10] fix(core): Check the project of an existing system agent thread (no-changelog) createThread checks access on the requested working project. It now returns an existing thread only when that thread is in the same project. A thread in another project is not found. Co-Authored-By: Claude Opus 5.5 --- .../system-agent-execution.service.test.ts | 31 +++++++++++++++++++ .../system-agent-execution.service.ts | 4 ++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts index 2dec618cf97..c4f77521c94 100644 --- a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -1029,6 +1029,37 @@ describe('SystemAgentExecutionService', () => { }); }); + it('refuses to return an existing session of the user from another project', async () => { + const { service, threadRepository, txRunner } = setup(); + threadRepository.findOneBy.mockResolvedValue({ ...thread, projectId: 'project-2' } as never); + + await expect( + service.createThread({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + threadId: 'thread-1', + }), + ).rejects.toThrow(NotFoundError); + expect(txRunner.run).not.toHaveBeenCalled(); + }); + + it('returns an existing session of the user in the same project', async () => { + const { service, threadRepository, txRunner } = setup(); + const existing = { ...thread, projectId: 'project-1' }; + threadRepository.findOneBy.mockResolvedValue(existing as never); + + expect( + await service.createThread({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + threadId: 'thread-1', + }), + ).toBe(existing); + expect(txRunner.run).not.toHaveBeenCalled(); + }); + it('refuses to create a session with the id of a thread of another user', async () => { const { service, threadRepository, txRunner } = setup(); threadRepository.findOneBy.mockResolvedValue({ ...thread, ownerId: 'user-2' } as never); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index 59c2a06193e..b3847d6c24c 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -148,7 +148,9 @@ export class SystemAgentExecutionService { if ( existing.agentId !== params.agentId || existing.accessScope !== 'user' || - existing.ownerId !== params.user.id + existing.ownerId !== params.user.id || + // The access check above ran for `params.projectId`, not for the project of this thread. + existing.projectId !== params.projectId ) { throw new NotFoundError('Session not found'); } From 2f5e116dc6d8b3add028b1f8767b514008ba937e Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:54:23 +0300 Subject: [PATCH 05/10] refactor(core): Build system agent queue input in one place (no-changelog) sendMessage and prepareChatMessage now share one queueInput builder. The payload rules for options, attachments, message ids and hidden turns are in one place. Co-Authored-By: Claude Opus 5.5 --- .../system-agent-execution.service.ts | 73 ++++++++++--------- 1 file changed, 38 insertions(+), 35 deletions(-) diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index b3847d6c24c..3dd52977102 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -40,6 +40,17 @@ type SuspendedToolCall = Extract< export type SystemAgentThreadStatus = 'running' | 'suspended' | 'idle'; +type SystemQueueInput = Parameters[0]; + +interface SystemMessageInput { + message: string; + attachments?: StoredAttachmentRef[]; + options?: SystemAgentTurnOptions; + messageId?: string; + /** A machine turn: it stays out of the transcript. */ + hidden?: boolean; +} + interface ResumeParams { agentId: string; user: User; @@ -275,34 +286,11 @@ export class SystemAgentExecutionService { * While a turn runs, the caller can steer the queued message into it. * A `hidden` message is a machine turn: it stays out of the transcript. */ - async sendMessage(params: { - agentId: string; - user: User; - threadId: string; - message: string; - attachments?: StoredAttachmentRef[]; - options?: SystemAgentTurnOptions; - messageId?: string; - hidden?: boolean; - }) { + async sendMessage( + params: SystemMessageInput & { agentId: string; user: User; threadId: string }, + ) { const thread = await this.getUsableThread(params.agentId, params.user, params.threadId); - return await this.messageQueue.enqueue({ - agentId: params.agentId, - projectId: thread.projectId, - threadId: thread.id, - sessionMode: 'existing', - source: SYSTEM_CHAT_SOURCE, - payload: { - kind: 'system', - userId: params.user.id, - message: params.message, - resourceId: this.resourceIdFor(params.user), - ...(params.attachments?.length ? { attachments: params.attachments } : {}), - ...(params.messageId ? { messageId: params.messageId } : {}), - ...(params.options ? { options: params.options } : {}), - ...(params.hidden ? { hidden: true } : {}), - }, - }); + return await this.messageQueue.enqueue(this.queueInput(thread, params.user, params)); } /** @@ -477,7 +465,7 @@ export class SystemAgentExecutionService { threadId: string, projectId: string, ) => Promise; - }): Promise[0]> { + }): Promise { this.assertRegistered(params.agentId); const existing = params.sessionId ? await this.threadRepository.findOwnedById(params.agentId, params.user.id, params.sessionId) @@ -499,20 +487,35 @@ export class SystemAgentExecutionService { const options = (await provider.chatTurnOptions?.(params.user, thread, params.clientContext)) ?? {}; const attachments = await params.storeAttachments?.(thread.id, thread.projectId); + return this.queueInput(thread, params.user, { + message: params.message, + messageId: params.messageId, + attachments, + options, + }); + } + + /** The queue input of a system message in an existing thread. */ + private queueInput( + thread: AgentExecutionThread, + user: User, + input: SystemMessageInput, + ): SystemQueueInput { return { - agentId: params.agentId, + agentId: thread.agentId, projectId: thread.projectId, threadId: thread.id, sessionMode: 'existing', source: SYSTEM_CHAT_SOURCE, payload: { kind: 'system', - userId: params.user.id, - message: params.message, - resourceId: this.resourceIdFor(params.user), - ...(params.messageId ? { messageId: params.messageId } : {}), - ...(attachments?.length ? { attachments } : {}), - options, + userId: user.id, + message: input.message, + resourceId: this.resourceIdFor(user), + ...(input.attachments?.length ? { attachments: input.attachments } : {}), + ...(input.messageId ? { messageId: input.messageId } : {}), + ...(input.options ? { options: input.options } : {}), + ...(input.hidden ? { hidden: true } : {}), }, }; } From 55fbd500d0059db7b8698d95a27fe110a3bf9927 Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:55:14 +0300 Subject: [PATCH 06/10] refactor(core): Pass one object to the system agent runTurn (no-changelog) runTurn took seven positional arguments, and the resume path passed `undefined` for the admission. It now takes one object. It also reads the agent id from the thread, so it does not need the provider. Co-Authored-By: Claude Opus 5.5 --- .../system-agent-execution.service.ts | 72 +++++++++++-------- 1 file changed, 42 insertions(+), 30 deletions(-) diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index 3dd52977102..80933695c3f 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -355,20 +355,27 @@ export class SystemAgentExecutionService { attachments.length > 0 && typeof text === 'string' ? buildInboundUserMessage(text, attachments) : text; - await this.runTurn(provider, thread, handle, send, controller, admission, async () => ({ - type: 'start', - input, - options: { - ...runOptionsOf(handle), - persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, - // The claim signal stops the turn only until the runtime admits it. - abortSignal: signal, - }, - recording: { - ...claim.recording, - ...(handle.hideUserMessage ? { hideUserMessageFromTranscript: true } : {}), - }, - })); + await this.runTurn({ + thread, + handle, + send, + controller, + admission, + prepare: async () => ({ + type: 'start', + input, + options: { + ...runOptionsOf(handle), + persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, + // The claim signal stops the turn only until the runtime admits it. + abortSignal: signal, + }, + recording: { + ...claim.recording, + ...(handle.hideUserMessage ? { hideUserMessageFromTranscript: true } : {}), + }, + }), + }); } /** @@ -426,14 +433,12 @@ export class SystemAgentExecutionService { sessionMode: 'existing', source: SYSTEM_CHAT_SOURCE, }; - const done = this.runTurn( - provider, + const done = this.runTurn({ thread, handle, - params.send ?? (() => {}), + send: params.send ?? (() => {}), controller, - undefined, - async () => ({ + prepare: async () => ({ type: 'resume', resumeData: params.resumeData, options: { @@ -444,7 +449,7 @@ export class SystemAgentExecutionService { }, recording, }), - ); + }); return { runId, toolCallId: pending.toolCallId, done }; } @@ -584,15 +589,22 @@ export class SystemAgentExecutionService { * the signal of. A cancel aborts it, and the abort signal in the run * options of the handle stops the turn at any time. */ - private async runTurn( - provider: SystemAgentProvider, - thread: AgentExecutionThread, - handle: SystemAgentTurnHandle, - send: (event: AgentSseEvent) => void, - controller: AbortController, - admission: AgentExecutionAdmission | undefined, - prepare: () => Promise, - ): Promise { + private async runTurn({ + thread, + handle, + send, + controller, + admission, + prepare, + }: { + thread: AgentExecutionThread; + handle: SystemAgentTurnHandle; + send: (event: AgentSseEvent) => void; + controller: AbortController; + /** Set when the queue admitted the turn. A resume admits its own turn. */ + admission?: AgentExecutionAdmission; + prepare: () => Promise; + }): Promise { let status: SystemAgentTurnStatus = 'completed'; let executionId = admission?.executionId; let error: unknown; @@ -606,7 +618,7 @@ export class SystemAgentExecutionService { agentInstance: handle.agent, toolRegistry: handle.toolRegistry ?? new Map(), mcpServerAttributions: new Map(), - context: { projectId: thread.projectId, agentId: provider.agentId, threadId: thread.id }, + context: { projectId: thread.projectId, agentId: thread.agentId, threadId: thread.id }, chatSurface: 'system', chatController: controller, onExecutionStarted: (id) => { From ba9df52f74d42583dc95c11c0b37a9f61beeb683 Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:56:26 +0300 Subject: [PATCH 07/10] refactor(core): Resume a system agent turn with the run id of the suspension (no-changelog) resume used the client run id when one was sent and the suspension run id otherwise. It now always uses the run id of the open suspension. A client run id that does not match it is refused up front. Co-Authored-By: Claude Opus 5.5 --- .../system-agent-execution.service.test.ts | 16 ++++++++++++++++ .../system-agent-execution.service.ts | 7 +++++-- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts index c4f77521c94..bef9a2bff7b 100644 --- a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -519,6 +519,22 @@ describe('SystemAgentExecutionService', () => { expect(prepared[0].options.abortSignal).toBeUndefined(); }); + it('refuses a run id that does not match the open suspension', async () => { + const { service, provider, checkpointStorage } = setup(); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + await expect( + service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + runId: 'run-old', + resumeData: {}, + }), + ).rejects.toThrow('This action is no longer waiting for input'); + expect(provider.prepareTurn).not.toHaveBeenCalled(); + }); + it('refuses when no tool call waits for input', async () => { const { service, provider, checkpointStorage } = setup(); checkpointStorage.findSuspendedForThread.mockResolvedValue(null); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index 80933695c3f..e0082e008b6 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -403,8 +403,11 @@ export class SystemAgentExecutionService { toolCall.suspended && (params.toolCallId === undefined || toolCall.toolCallId === params.toolCallId), ); - if (!checkpoint || !pending) throw new UserError('This action is no longer waiting for input'); - const runId = params.runId ?? pending.runId; + // A client run id must name the open suspension, not an earlier run. + if (!checkpoint || !pending || (params.runId !== undefined && params.runId !== pending.runId)) { + throw new UserError('This action is no longer waiting for input'); + } + const { runId } = pending; const resourceId = this.resourceIdFor(params.user); if (checkpoint.persistence?.resourceId !== resourceId) { throw new NotFoundError('Session not found'); From 5b1964908e39b6c5a9836582187440d7cbcccbef Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:59:04 +0300 Subject: [PATCH 08/10] refactor(core): Map queued message kinds to chat surfaces in one place (no-changelog) The queue and the queue consumer each mapped a message kind to its chat surface. chatSurfaceForKind and canContinueThreadOnSurface in agent-thread-access now hold that mapping and the access rule of each surface. The consumer passes the surface to its access check instead of a boolean. Co-Authored-By: Claude Opus 5.5 --- .../agent-message-queue-consumer.service.ts | 29 ++++++------ .../agents/agent-message-queue.service.ts | 23 ++-------- .../__tests__/agent-thread-access.test.ts | 46 +++++++++++++++++++ .../agents/utils/agent-thread-access.ts | 30 ++++++++++++ 4 files changed, 92 insertions(+), 36 deletions(-) create mode 100644 packages/cli/src/modules/agents/utils/__tests__/agent-thread-access.test.ts diff --git a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts index 8f93a0a2641..f3808037e88 100644 --- a/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue-consumer.service.ts @@ -23,8 +23,7 @@ import { AgentMessageQueueRepository } from './repositories/agent-message-queue. import { canUseSystemAgent } from './system-agents/system-agent-access'; import { SystemAgentExecutionService } from './system-agents/system-agent-execution.service'; import { SystemAgentRegistry } from './system-agents/system-agent-registry'; -import type { AgentQueuedMessage } from './types/agent-queued-message'; -import type { AgentChatSurface } from './utils/agent-thread-access'; +import { chatSurfaceForKind, type AgentChatSurface } from './utils/agent-thread-access'; @Service() export class AgentMessageQueueConsumer { @@ -140,11 +139,11 @@ export class AgentMessageQueueConsumer { controller.signal, this.executionService.getAbortSignal(admission.executionId), ]); - const sender = - payload.kind === 'integration' ? undefined : this.previewStreams.createSender(item.id); + const surface = payload.kind === 'integration' ? undefined : chatSurfaceForKind(payload.kind); + const sender = surface && this.previewStreams.createSender(item.id); try { - if (sender) { - const user = await this.getThreadOwner(thread, payload.kind === 'system'); + if (surface && sender) { + const user = await this.getThreadOwner(thread, surface); this.chatExecutionService.register( { projectId: thread.projectId, @@ -152,7 +151,7 @@ export class AgentMessageQueueConsumer { threadId: thread.id, userId: user.id, executionId: admission.executionId, - surface: chatSurfaceFor(payload.kind), + surface, }, controller, ); @@ -194,10 +193,13 @@ export class AgentMessageQueueConsumer { * system agent uses the runtime floor and its provider checks instead of * the project `agent:execute` scope. */ - private async getThreadOwner(thread: AgentExecutionThread, systemKind: boolean): Promise { + private async getThreadOwner( + thread: AgentExecutionThread, + surface: AgentChatSurface, + ): Promise { if (!thread.ownerId) throw new UserError('You can no longer execute this agent'); const user = await this.userRepository.findByIdWithRole(thread.ownerId); - if (!user || user.disabled || !(await this.canExecute(user, thread, systemKind))) { + if (!user || user.disabled || !(await this.canExecute(user, thread, surface))) { throw new UserError('You can no longer execute this agent'); } return user; @@ -206,9 +208,9 @@ export class AgentMessageQueueConsumer { private async canExecute( user: User, thread: AgentExecutionThread, - systemKind: boolean, + surface: AgentChatSurface, ): Promise { - if (!systemKind) { + if (surface !== 'system') { return await userHasScopes(user, ['agent:execute'], false, { projectId: thread.projectId }); } const provider = this.systemAgents.get(thread.agentId); @@ -317,8 +319,3 @@ export class AgentMessageQueueConsumer { await bridge.consumeQueuedMessage(payload, thread.id, claim.admission, signal, connection); } } - -function chatSurfaceFor(kind: AgentQueuedMessage['kind']): AgentChatSurface { - if (kind === 'n8n_chat') return 'n8n-chat'; - return kind === 'system' ? 'system' : 'preview'; -} diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 20daba4463f..5aa5e90e8c8 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -34,9 +34,8 @@ import { type QueuedUserChatMessage, } from './types/agent-queued-message'; import { - canContinueSystemThread, - canContinueThreadInN8nChat, - canContinueThreadInPreview, + canContinueThreadOnSurface, + chatSurfaceForKind, type AgentSessionMode, } from './utils/agent-thread-access'; import { buildInboundUserMessage, readInboundUserMessage } from './utils/inbound-attachments'; @@ -346,7 +345,7 @@ export class AgentMessageQueueService { if ( thread.projectId !== input.projectId || thread.agentId !== input.agentId || - !canContinueThread(input.kind, thread, input.userId, source) + !canContinueThreadOnSurface(chatSurfaceForKind(input.kind), thread, input.userId, source) ) { throw new NotFoundError('Session not found'); } @@ -508,19 +507,3 @@ export class AgentMessageQueueService { function isHidden(item: AgentMessageQueue): boolean { return item.message.origin?.hidden === true; } - -function canContinueThread( - kind: QueuedUserChatMessage['kind'], - thread: AgentExecutionThread, - userId: string, - source: string | null | undefined, -): boolean { - switch (kind) { - case 'n8n_chat': - return canContinueThreadInN8nChat(thread, userId, source); - case 'system': - return canContinueSystemThread(thread, userId, source); - case 'preview': - return canContinueThreadInPreview(thread, userId, source); - } -} diff --git a/packages/cli/src/modules/agents/utils/__tests__/agent-thread-access.test.ts b/packages/cli/src/modules/agents/utils/__tests__/agent-thread-access.test.ts new file mode 100644 index 00000000000..0079bafe180 --- /dev/null +++ b/packages/cli/src/modules/agents/utils/__tests__/agent-thread-access.test.ts @@ -0,0 +1,46 @@ +import type { AgentExecutionThread } from '../../entities/agent-execution-thread.entity'; +import { + canContinueThreadOnSurface, + chatSurfaceForKind, + N8N_CHAT_PRODUCTION_SOURCE, + SYSTEM_CHAT_SOURCE, +} from '../agent-thread-access'; + +const thread = { + accessScope: 'user', + ownerId: 'user-1', + parentThreadId: null, + taskId: null, +} as AgentExecutionThread; + +describe('chatSurfaceForKind', () => { + it.each([ + ['preview', 'preview'], + ['n8n_chat', 'n8n-chat'], + ['system', 'system'], + ] as const)('maps the %s kind to the %s surface', (kind, surface) => { + expect(chatSurfaceForKind(kind)).toBe(surface); + }); +}); + +describe('canContinueThreadOnSurface', () => { + it.each([ + ['preview', 'chat'], + ['n8n-chat', N8N_CHAT_PRODUCTION_SOURCE], + ['system', SYSTEM_CHAT_SOURCE], + ] as const)('lets the owner continue a %s thread with its own source', (surface, source) => { + expect(canContinueThreadOnSurface(surface, thread, 'user-1', source)).toBe(true); + }); + + it.each([ + ['preview', SYSTEM_CHAT_SOURCE], + ['n8n-chat', SYSTEM_CHAT_SOURCE], + ['system', 'chat'], + ] as const)('refuses a %s thread with a %s source', (surface, source) => { + expect(canContinueThreadOnSurface(surface, thread, 'user-1', source)).toBe(false); + }); + + it('refuses a user who does not own the thread', () => { + expect(canContinueThreadOnSurface('system', thread, 'user-2', SYSTEM_CHAT_SOURCE)).toBe(false); + }); +}); diff --git a/packages/cli/src/modules/agents/utils/agent-thread-access.ts b/packages/cli/src/modules/agents/utils/agent-thread-access.ts index 3aae271c525..2c744f279f6 100644 --- a/packages/cli/src/modules/agents/utils/agent-thread-access.ts +++ b/packages/cli/src/modules/agents/utils/agent-thread-access.ts @@ -1,4 +1,5 @@ import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity'; +import type { QueuedUserChatMessage } from '../types/agent-queued-message'; import { draftChatMemoryResourceId, productionChatMemoryResourceId, @@ -18,6 +19,18 @@ export const SYSTEM_CHAT_SOURCE = 'system_chat' as const; */ export type AgentChatSurface = 'preview' | 'n8n-chat' | 'system'; +/** The chat surface that a queued user chat message runs on. */ +export function chatSurfaceForKind(kind: QueuedUserChatMessage['kind']): AgentChatSurface { + switch (kind) { + case 'n8n_chat': + return 'n8n-chat'; + case 'system': + return 'system'; + case 'preview': + return 'preview'; + } +} + /** The memory resource ID that scopes a thread to its owning user on the given chat surface. */ export function chatSurfaceMemoryResourceId(surface: AgentChatSurface, userId: string): string { switch (surface) { @@ -82,6 +95,23 @@ export function canContinueSystemThread( ); } +/** Whether the user can continue the thread with a message on the chat surface. */ +export function canContinueThreadOnSurface( + surface: AgentChatSurface, + thread: AgentExecutionThread, + userId: string, + source?: string | null, +): boolean { + switch (surface) { + case 'n8n-chat': + return canContinueThreadInN8nChat(thread, userId, source); + case 'system': + return canContinueSystemThread(thread, userId, source); + case 'preview': + return canContinueThreadInPreview(thread, userId, source); + } +} + export function canUseTopLevelDraftThread(thread: AgentExecutionThread, userId: string): boolean { return ( thread.accessScope === 'user' && thread.ownerId === userId && thread.parentThreadId === null From edde04df32d203b065b25fb6bd1be3eaaa43ff4a Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 22:00:34 +0300 Subject: [PATCH 09/10] refactor(core): Look up the queued agent once by message kind (no-changelog) enqueue now decides on the system kind once. The custom-agents setting applies to every other kind. A system message without a registered provider now fails with "Agent not found", also when the admin turned off custom agents. Co-Authored-By: Claude Opus 5.5 --- .../modules/agents/agent-message-queue.service.ts | 14 ++++++-------- .../agent-execution.repository.test.ts | 10 ++++++++++ 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 5aa5e90e8c8..76b73eb417a 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -94,20 +94,18 @@ export class AgentMessageQueueService { ): Promise<{ status: 'accepted'; item: AgentMessageQueue } | { status: 'duplicate' }> { const { payload } = input; // Only the system kind reaches a system agent. Project lookups never return instance agents. - const systemAgent = - payload.kind === 'system' ? this.systemAgents.get(input.agentId) : undefined; // System agents run even when the admin turns off custom agents. - if (!systemAgent) await this.settingsService.assertEnabled(); - const agent = - payload.kind === 'system' - ? systemAgent - : await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId); + const isSystem = payload.kind === 'system'; + if (!isSystem) await this.settingsService.assertEnabled(); + const agent = isSystem + ? this.systemAgents.get(input.agentId) + : await this.agentRepository.findByIdAndProjectId(input.agentId, input.projectId); if (!agent) throw new UserError('Agent not found'); const messageId = queuedMessageId(input.agentId, input.threadId, payload); let item: AgentMessageQueue; try { item = await this.txRunner.run({}, async (ctx) => { - if (!systemAgent) await this.settingsService.assertEnabled(ctx); + if (!isSystem) await this.settingsService.assertEnabled(ctx); await this.executionService.prepareThread( { ...input, diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts index 4f11d64a8c7..0eb003209e3 100644 --- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts @@ -2693,6 +2693,16 @@ describe('AgentExecutionRepository', () => { ); }); + it('refuses a system message without a provider while custom agents are turned off', async () => { + const services = recordingServices(); + await services.settingsService.setEnabled(false); + await agentRepo.ensureInstanceAgent(systemAgentId, 'Test System Agent'); + + await expect(services.queue.enqueue(systemInput(uuid(), 'Hello', 'new'))).rejects.toThrow( + 'Agent not found', + ); + }); + it('refuses a Preview message for a system agent', async () => { const services = await systemServices(); From e7440ab1f05ee501925e6ba23cb2c7f0bf82647f Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 22:01:36 +0300 Subject: [PATCH 10/10] test(core): Give each system agent service test its own thread (no-changelog) updateThread changes the thread that the repository returns. The tests shared one thread object, so a project move in one test leaked into the next. The fixture is now frozen and each lookup returns a copy. Co-Authored-By: Claude Opus 5.5 --- .../system-agent-execution.service.test.ts | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts index bef9a2bff7b..c591236a510 100644 --- a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -34,14 +34,15 @@ vi.mock('@/permissions.ee/check-access', () => ({ userHasScopes: vi.fn() })); const AGENT_ID = 'test-assistant'; const user = mock({ id: 'user-1' }); -const thread = { +// Frozen: the service changes the threads it returns, so each test gets a copy. +const thread = Object.freeze({ id: 'thread-1', agentId: AGENT_ID, agentName: 'Test Assistant', projectId: 'project-1', accessScope: 'user', ownerId: 'user-1', -} as AgentExecutionThread; +} as AgentExecutionThread); /** The checkpoint of a turn that waits for the answer to one tool call. */ function suspendedCheckpoint(resourceId = 'system-chat:user-1') { @@ -100,7 +101,7 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { } satisfies SystemAgentProvider; registry.register(provider); - threadRepository.findOwnedById.mockResolvedValue(thread); + threadRepository.findOwnedById.mockImplementation(async () => ({ ...thread })); executionRepository.findExecution.mockResolvedValue(null); // Collect the prepared request, then stream the scripted chunks. const prepared: AgentTurnRequest[] = []; @@ -687,7 +688,7 @@ describe('SystemAgentExecutionService', () => { it('loads a usable thread with the floor on its working project', async () => { const { service, provider } = setup(); - expect(await service.getUsableThread(AGENT_ID, user, 'thread-1')).toBe(thread); + expect(await service.getUsableThread(AGENT_ID, user, 'thread-1')).toEqual(thread); expect(userHasScopes).toHaveBeenCalledWith(user, ['project:read'], false, { projectId: 'project-1', }); @@ -706,8 +707,7 @@ describe('SystemAgentExecutionService', () => { describe('prepareChatMessage', () => { it('keeps the working project of an existing session', async () => { - const { service, threadRepository } = setup(); - threadRepository.findOwnedById.mockResolvedValue(thread); + const { service } = setup(); const input = await service.prepareChatMessage({ agentId: AGENT_ID, @@ -733,8 +733,7 @@ describe('SystemAgentExecutionService', () => { }); it('refuses a project that does not match the existing session', async () => { - const { service, threadRepository } = setup(); - threadRepository.findOwnedById.mockResolvedValue(thread); + const { service } = setup(); await expect( service.prepareChatMessage({ @@ -987,7 +986,7 @@ describe('SystemAgentExecutionService', () => { threadRepository.findOneBy.mockResolvedValue(null); txRunner.run.mockImplementation(async (_ctx, fn) => await fn(ctx)); threadRepository.findOrCreate.mockResolvedValue({ - thread: { ...thread, id: 'thread-new', projectId: 'project-1' }, + thread: { ...thread, id: 'thread-new' }, created: true, } as never); threadRepository.renameOwned.mockResolvedValue(true); @@ -1024,7 +1023,7 @@ describe('SystemAgentExecutionService', () => { threadRepository.findOneBy.mockResolvedValue(null); txRunner.run.mockImplementation(async (_ctx, fn) => await fn({})); threadRepository.findOrCreate.mockResolvedValue({ - thread: { ...thread, id: 'thread-new', projectId: 'project-1' }, + thread: { ...thread, id: 'thread-new' }, created: true, } as never);