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..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 @@ -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,136 @@ 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(); + }); + + /** 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); + + expectRefused(item); + }); + + 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); + + expectRefused(item); + expect(provider.authorize).not.toHaveBeenCalled(); + }); + + it('records a failure for a system message without a registered provider', async () => { + const item = systemClaim('session'); + + await consumeOne(item); + + expectRefused(item); + }); + + 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..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 @@ -20,6 +20,10 @@ 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 { chatSurfaceForKind, type AgentChatSurface } from './utils/agent-thread-access'; @Service() export class AgentMessageQueueConsumer { @@ -40,6 +44,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 { @@ -133,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); + if (surface && sender) { + const user = await this.getThreadOwner(thread, surface); this.chatExecutionService.register( { projectId: thread.projectId, @@ -145,7 +151,7 @@ export class AgentMessageQueueConsumer { threadId: thread.id, userId: user.id, executionId: admission.executionId, - surface: payload.kind === 'n8n_chat' ? 'n8n-chat' : 'preview', + surface, }, controller, ); @@ -156,11 +162,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 +188,35 @@ 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, + 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 userHasScopes(user, ['agent:execute'], false, { projectId: thread.projectId })) - ) { + if (!user || user.disabled || !(await this.canExecute(user, thread, surface))) { throw new UserError('You can no longer execute this agent'); } return user; } + private async canExecute( + user: User, + thread: AgentExecutionThread, + surface: AgentChatSurface, + ): Promise { + if (surface !== 'system') { + 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, 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..76b73eb417a 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,8 +34,8 @@ import { type QueuedUserChatMessage, } from './types/agent-queued-message'; import { - canContinueThreadInN8nChat, - canContinueThreadInPreview, + canContinueThreadOnSurface, + chatSurfaceForKind, type AgentSessionMode, } from './utils/agent-thread-access'; import { buildInboundUserMessage, readInboundUserMessage } from './utils/inbound-attachments'; @@ -76,6 +77,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 +92,20 @@ 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. + // System agents run even when the admin turns off custom agents. + 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) => { - await this.settingsService.assertEnabled(ctx); + if (!isSystem) await this.settingsService.assertEnabled(ctx); await this.executionService.prepareThread( { ...input, @@ -185,6 +192,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 +343,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)) + !canContinueThreadOnSurface(chatSurfaceForKind(input.kind), thread, input.userId, source) ) { throw new NotFoundError('Session not found'); } @@ -353,9 +363,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); 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 new file mode 100644 index 00000000000..c591236a510 --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -0,0 +1,1093 @@ +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 { AgentChatAttachmentService } from '../../agent-chat-attachment.service'; +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' }); +// 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); + +/** 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 attachmentService = mock(); + + const onChunk = vi.fn(); + const onSettled = vi.fn(async () => {}); + const handle: SystemAgentTurnHandle = { + 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, + 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.mockImplementation(async () => ({ ...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, + attachmentService, + ); + + return { + service, + provider, + handle, + onChunk, + onSettled, + prepared, + threadRepository, + executionRepository, + turnExecutionService, + messageQueue, + checkpointStorage, + txRunner, + executionService, + chatExecutionService, + agentRepository, + registry, + attachmentService, + }; +} + +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, + ]); + const send = 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 () => { + 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; + }); + + const send = vi.fn(); + + await expect( + 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 () => { + 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(); + 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('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, + ]); + 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 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); + + 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('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); + + 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')).toEqual(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 } = setup(); + + 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 } = setup(); + + 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('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); + + 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('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); + + 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('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' }, + 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' }, + 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 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); + + 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..e0082e008b6 --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -0,0 +1,665 @@ +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'; + +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; + 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 ?? {}; + return runOptions; +} + +/** + * 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 || + // 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'); + } + 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'); + } + + /** Whether the thread is not idle. It uses the same rule as `getStatus`. */ + async isBusy(thread: AgentExecutionThread): Promise { + return (await this.resolveStatus(thread)).status !== 'idle'; + } + + async getStatus(thread: AgentExecutionThread): Promise<{ + status: SystemAgentThreadStatus; + latestExecutionId: string | null; + checkpoint: SerializableAgentState | null; + }> { + const { status, checkpoint } = await this.resolveStatus(thread); + const latest = await this.executionRepository.findLatestByThreadId(thread.id); + 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 ───────────────────────────────────────────────────────────── + + /** + * 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: SystemMessageInput & { agentId: string; user: User; threadId: string }, + ) { + const thread = await this.getUsableThread(params.agentId, params.user, params.threadId); + return await this.messageQueue.enqueue(this.queueInput(thread, params.user, params)); + } + + /** + * 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 controller = new AbortController(); + const handle = await provider.prepareTurn({ + type: 'start', + user, + thread, + resourceId, + abortSignal: controller.signal, + executionId: admission.executionId, + message: payload.message, + attachments, + options: payload.options ?? {}, + }); + const text = handle.input ?? payload.message; + this.attachFileStore(handle, thread); + const input = + attachments.length > 0 && typeof text === 'string' + ? buildInboundUserMessage(text, attachments) + : text; + 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 } : {}), + }, + }), + }); + } + + /** + * 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( + thread.agentId, + thread.id, + ); + const pending = Object.values(checkpoint?.pendingToolCalls ?? {}).find( + (toolCall): toolCall is SuspendedToolCall => + toolCall.suspended && + (params.toolCallId === undefined || toolCall.toolCallId === params.toolCallId), + ); + // 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'); + } + 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, + }); + this.attachFileStore(handle, thread); + 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 done = this.runTurn({ + thread, + handle, + send: params.send ?? (() => {}), + controller, + prepare: async () => ({ + type: 'resume', + resumeData: params.resumeData, + options: { + ...runOptionsOf(handle), + runId, + toolCallId: pending.toolCallId, + hostMetadata: handle.hostMetadata, + }, + recording, + }), + }); + 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 { + 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 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: thread.agentId, + projectId: thread.projectId, + threadId: thread.id, + sessionMode: 'existing', + source: SYSTEM_CHAT_SOURCE, + payload: { + kind: 'system', + 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 } : {}), + }, + }; + } + + /** 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'); + // A failure rejects, so the route can send an error event. + const { done } = await this.startResume({ + 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. 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); + if (!latest) return false; + return await this.chatExecutionService.requestCancel({ + projectId: thread.projectId, + agentId, + threadId: thread.id, + executionId: latest.id, + userId: user.id, + surface: 'system', + }); + } + + /** + * 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 + * options of the handle stops the turn at any time. + */ + 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; + 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, + agentInstance: handle.agent, + toolRegistry: handle.toolRegistry ?? new Map(), + mcpServerAttributions: new Map(), + context: { projectId: thread.projectId, agentId: thread.agentId, threadId: thread.id }, + chatSurface: 'system', + chatController: controller, + 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 = controller.signal.aborted ? 'cancelled' : 'errored'; + throw caught; + } finally { + providerSignal?.removeEventListener('abort', stopTurn); + 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, + }); + } + // 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-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..75933dc5043 --- /dev/null +++ b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts @@ -0,0 +1,98 @@ +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; + /** Aborts when the turn stops: a user cancel, the queue, or `runOptions.abortSignal`. */ + 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. `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; + 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/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 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'); + }); +}); 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..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 @@ -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,98 @@ 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 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(); + + 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 +2804,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 +2842,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 +2856,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 +2878,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,