From 60b56086e12c883f4d678aa48d65a88eea5cd103 Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Thu, 15 Oct 2026 23:55:45 +0300 Subject: [PATCH] feat(core): Add a system chat surface to the Agents runtime (no-changelog) - Add the system-chat: memory resource, the system_chat transcript source and the system value of AgentChatSurface. canContinueSystemThread is the access rule of the surface. - The cross-main cancel payload carries surface. It keeps productionN8nChat for mains on the previous version. Co-Authored-By: Claude Opus 5.5 --- .../agent-chat-execution.service.test.ts | 37 +++++++++++++- .../agents/agent-chat-execution.service.ts | 49 ++++++++++++------- .../agents/utils/agent-memory-scope.ts | 12 +++++ .../agents/utils/agent-thread-access.ts | 38 +++++++++++--- .../src/scaling/pubsub/pubsub.event-map.ts | 2 + 5 files changed, 113 insertions(+), 25 deletions(-) diff --git a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts index d355f97a5ff..65db519d33c 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts @@ -165,7 +165,7 @@ it('relays Stop to the owning main and leaves other and later executions running const { surface: _surface, ...wireContext } = context; expect(mainB.publisher.publishCommand).toHaveBeenCalledExactlyOnceWith({ command: 'cancel-agent-chat-execution', - payload: { ...wireContext, productionN8nChat: false }, + payload: { ...wireContext, surface: 'preview', productionN8nChat: false }, }); await mainA.service.handleCancel(context); expect(mainA.backgroundJobs.cancelForParent).toHaveBeenCalledWith( @@ -197,7 +197,7 @@ it('keeps pubsub wire compatibility with a main still on the old productionN8nCh const { surface: _surface, ...wireContext } = production; expect(mainB.publisher.publishCommand).toHaveBeenCalledExactlyOnceWith({ command: 'cancel-agent-chat-execution', - payload: { ...wireContext, productionN8nChat: true }, + payload: { ...wireContext, surface: 'n8n-chat', productionN8nChat: true }, }); // An old main on the previous version relays this shape, with no `surface` field. @@ -210,6 +210,39 @@ it('keeps pubsub wire compatibility with a main still on the old productionN8nCh ); }); +it('relays Stop of a system-agent turn with its own surface and memory scope', async () => { + const mainA = makeService(); + const mainB = makeService(); + const controller = new AbortController(); + const system = { ...context, surface: 'system' as const }; + mainA.service.register(system, controller); + const systemExecution = { ...running, source: 'system_chat' }; + mainA.repository.findOneBy.mockResolvedValue(systemExecution); + mainB.repository.findOneBy.mockResolvedValue(systemExecution); + + expect(await mainB.service.requestCancel(system)).toBe(true); + const { surface: _surface, ...wireContext } = system; + expect(mainB.publisher.publishCommand).toHaveBeenCalledExactlyOnceWith({ + command: 'cancel-agent-chat-execution', + payload: { ...wireContext, surface: 'system', productionN8nChat: false }, + }); + + await mainA.service.handleCancel({ ...wireContext, surface: 'system', productionN8nChat: false }); + expect(controller.signal.aborted).toBe(true); + expect(mainA.backgroundJobs.cancelForParent).toHaveBeenCalledWith( + 'agent-1', + 'thread-1', + 'system-chat:user-1', + ); +}); + +it('does not let the Preview surface stop a system-agent turn', async () => { + const { service, repository } = makeService(false); + repository.findOneBy.mockResolvedValue({ ...running, source: 'system_chat' }); + + await expect(service.requestCancel(context)).rejects.toThrow('Execution not found'); +}); + it.each(['local', 'remote', 'during validation'] as const)( 'applies an early %s Stop when the execution registers', async (arrival) => { diff --git a/packages/cli/src/modules/agents/agent-chat-execution.service.ts b/packages/cli/src/modules/agents/agent-chat-execution.service.ts index eb970155e3a..1bf3d43dbf9 100644 --- a/packages/cli/src/modules/agents/agent-chat-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-chat-execution.service.ts @@ -12,13 +12,16 @@ import { AgentMessageSteeringService } from './agent-message-steering.service'; import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster'; import { AgentBackgroundJobService } from './background/agent-background-job.service'; import { PARENT_TASK_CANCELLED_REASON } from './background/sub-agent-background-state'; +import type { AgentExecutionThread } from './entities/agent-execution-thread.entity'; import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage'; import { AgentExecutionRepository } from './repositories/agent-execution.repository'; import { userIdFromDraftChatMemoryResourceId, userIdFromProductionChatMemoryResourceId, + userIdFromSystemChatMemoryResourceId, } from './utils/agent-memory-scope'; import { + canContinueSystemThread, canContinueThreadInPreview, chatSurfaceMemoryResourceId, N8N_CHAT_PRODUCTION_SOURCE, @@ -29,9 +32,9 @@ import { getDelegatedChildCheckpoints } from './utils/delegated-child-checkpoint type CancelExecutionPayload = PubSubCommandMap['cancel-agent-chat-execution']; // In-process callers already know which chat surface a run is on. The wire -// payload keeps the older `productionN8nChat` flag for rolling-deploy +// payload also keeps the older `productionN8nChat` flag for rolling-deploy // compatibility with mains still running the previous version. -type ExecutionContext = Omit & { +type ExecutionContext = Omit & { surface: AgentChatSurface; }; @@ -137,7 +140,7 @@ export class AgentChatExecutionService { const { surface, ...rest } = context; await this.publisher.publishCommand({ command: 'cancel-agent-chat-execution', - payload: { ...rest, productionN8nChat: surface === 'n8n-chat' }, + payload: { ...rest, surface, productionN8nChat: surface === 'n8n-chat' }, }); return true; } finally { @@ -153,10 +156,10 @@ export class AgentChatExecutionService { @OnPubSubEvent('cancel-agent-chat-execution', { instanceType: 'main' }) async handleCancel(payload: CancelExecutionPayload): Promise { - const { productionN8nChat, ...rest } = payload; + const { productionN8nChat, surface, ...rest } = payload; const context: ExecutionContext = { ...rest, - surface: productionN8nChat ? 'n8n-chat' : 'preview', + surface: surface ?? (productionN8nChat ? 'n8n-chat' : 'preview'), }; if (await this.cancelLocalWithChildren(context)) return; await this.lockService.withLease( @@ -212,20 +215,29 @@ export class AgentChatExecutionService { const thread = await this.executionService.findThreadById(threadId); if (!thread || !threadBelongsTo(thread, projectId, agentId, userId)) return null; const execution = await this.executionRepository.findOneBy({ id: executionId, threadId }); - if ( - !execution || - (context.surface === 'n8n-chat' - ? !( - thread.accessScope === 'user' && - thread.ownerId === userId && - execution.source === N8N_CHAT_PRODUCTION_SOURCE - ) - : !canContinueThreadInPreview(thread, userId, execution.source)) - ) - return null; + if (!execution || !this.canAccessExecution(context, thread, execution.source)) return null; return execution; } + private canAccessExecution( + context: ExecutionContext, + thread: AgentExecutionThread, + source: string | null, + ): boolean { + switch (context.surface) { + case 'n8n-chat': + return ( + thread.accessScope === 'user' && + thread.ownerId === context.userId && + source === N8N_CHAT_PRODUCTION_SOURCE + ); + case 'system': + return canContinueSystemThread(thread, context.userId, source); + case 'preview': + return canContinueThreadInPreview(thread, context.userId, source); + } + } + private async cancelRecordedSuspension(context: ExecutionContext): Promise { const latest = await this.executionRepository.findLatestByThreadId(context.threadId); if (latest?.id !== context.executionId || latest.hitlStatus !== 'suspended') return false; @@ -257,7 +269,10 @@ export class AgentChatExecutionService { return false; const thread = await this.executionService.findThreadById(checkpoint.persistence.threadId); const productionUserId = userIdFromProductionChatMemoryResourceId(params.resourceId); - const userId = userIdFromDraftChatMemoryResourceId(params.resourceId) ?? productionUserId; + const userId = + userIdFromDraftChatMemoryResourceId(params.resourceId) ?? + userIdFromSystemChatMemoryResourceId(params.resourceId) ?? + productionUserId; if ( productionUserId && (!thread || diff --git a/packages/cli/src/modules/agents/utils/agent-memory-scope.ts b/packages/cli/src/modules/agents/utils/agent-memory-scope.ts index 61491638586..0d42d1af76d 100644 --- a/packages/cli/src/modules/agents/utils/agent-memory-scope.ts +++ b/packages/cli/src/modules/agents/utils/agent-memory-scope.ts @@ -1,6 +1,7 @@ const INTEGRATION_MEMORY_RESOURCE_PREFIX = 'integration:'; const DRAFT_CHAT_MEMORY_RESOURCE_PREFIX = 'draft-chat:'; const PRODUCTION_CHAT_MEMORY_RESOURCE_PREFIX = 'n8n-chat-production:'; +const SYSTEM_CHAT_MEMORY_RESOURCE_PREFIX = 'system-chat:'; const THREAD_MEMORY_RESOURCE_PREFIX = 'thread:'; export function threadMemoryResourceId(threadId: string): string { @@ -28,6 +29,17 @@ export function productionChatMemoryResourceId(userId: string): string { return `${PRODUCTION_CHAT_MEMORY_RESOURCE_PREFIX}${userId}`; } +/** A system-agent chat. It is not a Preview chat, so it has its own resource scope. */ +export function systemChatMemoryResourceId(userId: string): string { + return `${SYSTEM_CHAT_MEMORY_RESOURCE_PREFIX}${userId}`; +} + +export function userIdFromSystemChatMemoryResourceId(resourceId: string): string | undefined { + if (!resourceId.startsWith(SYSTEM_CHAT_MEMORY_RESOURCE_PREFIX)) return undefined; + const userId = resourceId.slice(SYSTEM_CHAT_MEMORY_RESOURCE_PREFIX.length); + return userId.length > 0 ? userId : undefined; +} + export function userIdFromProductionChatMemoryResourceId(resourceId: string): string | undefined { if (!resourceId.startsWith(PRODUCTION_CHAT_MEMORY_RESOURCE_PREFIX)) return undefined; const userId = resourceId.slice(PRODUCTION_CHAT_MEMORY_RESOURCE_PREFIX.length); 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 4021179ad6d..3aae271c525 100644 --- a/packages/cli/src/modules/agents/utils/agent-thread-access.ts +++ b/packages/cli/src/modules/agents/utils/agent-thread-access.ts @@ -1,19 +1,33 @@ import type { AgentExecutionThread } from '../entities/agent-execution-thread.entity'; -import { draftChatMemoryResourceId, productionChatMemoryResourceId } from './agent-memory-scope'; +import { + draftChatMemoryResourceId, + productionChatMemoryResourceId, + systemChatMemoryResourceId, +} from './agent-memory-scope'; export type AgentSessionMode = 'new' | 'existing'; export const PREVIEW_THREAD_SOURCES = ['', 'chat', 'n8n_chat'] as const; export const N8N_CHAT_PRODUCTION_SOURCE = 'n8n_chat_production' as const; +/** Transcript source of a system-agent turn. It is not a Preview source. */ +export const SYSTEM_CHAT_SOURCE = 'system_chat' as const; -/** The n8n UI chat that a person watches the turn in. Integrations and tasks have none. */ -export type AgentChatSurface = 'preview' | 'n8n-chat'; +/** + * The n8n UI chat that a person watches the turn in. Integrations and tasks have none. + * `system` is the chat with a code-defined system agent. + */ +export type AgentChatSurface = 'preview' | 'n8n-chat' | 'system'; /** 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 { - return surface === 'n8n-chat' - ? productionChatMemoryResourceId(userId) - : draftChatMemoryResourceId(userId); + switch (surface) { + case 'n8n-chat': + return productionChatMemoryResourceId(userId); + case 'system': + return systemChatMemoryResourceId(userId); + case 'preview': + return draftChatMemoryResourceId(userId); + } } export function threadBelongsTo( @@ -56,6 +70,18 @@ export function canContinueThreadInN8nChat( ); } +export function canContinueSystemThread( + thread: AgentExecutionThread, + userId: string, + source?: string | null, +): boolean { + return ( + canUseTopLevelDraftThread(thread, userId) && + thread.taskId === null && + source === SYSTEM_CHAT_SOURCE + ); +} + export function canUseTopLevelDraftThread(thread: AgentExecutionThread, userId: string): boolean { return ( thread.accessScope === 'user' && thread.ownerId === userId && thread.parentThreadId === null diff --git a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts index 586eb262c27..1c111ab37d7 100644 --- a/packages/cli/src/scaling/pubsub/pubsub.event-map.ts +++ b/packages/cli/src/scaling/pubsub/pubsub.event-map.ts @@ -150,6 +150,8 @@ export type PubSubCommandMap = { threadId: string; executionId: string; userId: string; + /** Older mains read only `productionN8nChat`, so senders set both. */ + surface?: 'preview' | 'n8n-chat' | 'system'; productionN8nChat?: boolean; };