github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-16 21:19:43 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Add a system chat surface to the Agents runtime (no-changelog)

Browse Source 
- Add the system-chat:<userId> 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 <noreply@anthropic.com>
This commit is contained in:
Jaakko Husso 2026-10-15 23:55:45 +03:00 committed by GitHub
parent bc2de3b222
commit 60b56086e1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 113 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
packages/cli/src/modules/agents/__tests__/agent-chat-execution.service.test.ts
View File

@ -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) => {

49
packages/cli/src/modules/agents/agent-chat-execution.service.ts
View File

@ -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<CancelExecutionPayload, 'productionN8nChat'> & {
type ExecutionContext = Omit<CancelExecutionPayload, 'productionN8nChat' | 'surface'> & {
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<void> {
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<boolean> {
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 ||

12
packages/cli/src/modules/agents/utils/agent-memory-scope.ts
View File

@ -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);

38
packages/cli/src/modules/agents/utils/agent-thread-access.ts
View File

@ -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

2
packages/cli/src/scaling/pubsub/pubsub.event-map.ts
View File

@ -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;
};