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,