feat(core): Add production n8n Chat channel infrastructure (no-changelog) (#39476)

Co-authored-by: Benjamin Schroth <benjamin@n8n.io>
This commit is contained in:
Michael Drury 2026-09-28 11:47:35 +00:00 committed by GitHub
parent 256e581aa7
commit 68f856fb9d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
59 changed files with 2359 additions and 108 deletions

View File

@ -34,6 +34,8 @@
"**/*.vue",
// Generated DB schema docs
"docs/generated/**",
// Recorded model responses
"**/cassettes/n8n-chat-production/**",
// Generated / vendored test-impact data
".github/test-metrics/**",
// Vendored SPDX license data and overrides

View File

@ -1,6 +1,25 @@
import { AgentIntegrationSchema } from '../agent-integration.schema';
import { AgentIntegrationConfigSchema, AgentIntegrationSchema } from '../agent-integration.schema';
import { AgentJsonConfigSchema } from '../agent-json-config.schema';
describe('AgentIntegrationSchema', () => {
it('accepts n8n Chat without a credential in agent configuration', () => {
expect(AgentIntegrationConfigSchema.parse({ type: 'n8n_chat' })).toEqual({
type: 'n8n_chat',
credentialId: '',
});
expect(AgentIntegrationSchema.safeParse({ type: 'n8n_chat' }).success).toBe(false);
});
it('rejects duplicate n8n Chat channel entries', () => {
expect(
AgentJsonConfigSchema.safeParse({
name: 'Agent',
model: 'openai/gpt-4o-mini',
instructions: 'Help',
integrations: [{ type: 'n8n_chat' }, { type: 'n8n_chat' }],
}).success,
).toBe(false);
});
it('accepts a telegram integration with credential id', () => {
const result = AgentIntegrationSchema.safeParse({
type: 'telegram',

View File

@ -1,7 +1,7 @@
/**
* Draft-lifecycle predicates. A capability is a "draft" while its setup is
* pending: the agent config until a model is chosen (`model: ""`), an
* integration entry until a credential is connected (`credentialId: ""`).
* credential-backed integration entry until a credential is connected.
*/
/** True while no model has been chosen yet (setup pending). */
@ -10,6 +10,6 @@ export function isDraftAgentConfig(config: { model?: string } | null | undefined
}
/** True while no credential is connected yet (setup pending). */
export function isDraftIntegration(integration: { credentialId: string }): boolean {
return integration.credentialId.trim() === '';
export function isDraftIntegration(integration: { type: string; credentialId: string }): boolean {
return integration.type !== 'n8n_chat' && integration.credentialId.trim() === '';
}

View File

@ -214,12 +214,30 @@ const draftCredentialIntegrations = [
}),
] as const;
const n8nChatIntegration = z
.object({
type: z.literal('n8n_chat'),
credentialId: z.literal('').default(''),
})
.strict();
export const AgentIntegrationSchema = z.discriminatedUnion('type', credentialIntegrations);
/** Draft config variant that allows cleared stale credential IDs. */
export const AgentIntegrationConfigSchema = z.discriminatedUnion(
'type',
draftCredentialIntegrations,
);
export const AgentIntegrationConfigSchema = z.discriminatedUnion('type', [
...draftCredentialIntegrations,
n8nChatIntegration,
]);
export type AgentIntegrationConfig = z.infer<typeof AgentIntegrationConfigSchema>;
export type CredentialAgentIntegrationConfig = Exclude<
AgentIntegrationConfig,
{ type: 'n8n_chat' }
>;
export function isCredentialAgentIntegration(
integration: AgentIntegrationConfig,
): integration is CredentialAgentIntegrationConfig {
return integration.type !== 'n8n_chat';
}

View File

@ -482,7 +482,15 @@ export const AgentJsonConfigBaseSchema = z.object({
.optional(),
tasks: z.array(AgentJsonTaskConfigSchema).optional(),
providerTools: z.record(z.record(z.unknown())).optional(),
integrations: z.array(AgentIntegrationConfigSchema).optional(),
integrations: z
.array(AgentIntegrationConfigSchema)
.refine(
(integrations) => integrations.filter((entry) => entry.type === 'n8n_chat').length <= 1,
{
message: 'Only one n8n Chat channel is allowed',
},
)
.optional(),
mcpServers: z
.array(McpServerConfigSchema)
.max(20)

View File

@ -40,6 +40,7 @@ export const AGENT_SESSION_ORIGINS = [
'sub-agent',
'schedule',
'workflow',
'n8n_chat_production',
'slack',
'telegram',
'linear',

View File

@ -164,6 +164,7 @@ export interface AgentIntegrationStatusResponse {
*/
status: 'configured' | 'connected' | 'disconnected' | 'partial' | 'error';
integrations: AgentIntegrationStatusEntry[];
n8nChat: { draftEnabled: boolean; publishedEnabled: boolean };
}
export interface AgentDisconnectIntegrationResponse {
@ -408,8 +409,8 @@ export interface AgentSessionLangSmithExportResponse {
}
/**
* Internal integration type for the in-app chat channel. Injected per-run for
* `/chat` executions — never persisted in an agent's `integrations` array.
* Integration type for n8n Chat. Preview injects its tools per run. A configured
* channel entry enables production chat after publish.
*/
export const N8N_CHAT_INTEGRATION_TYPE = 'n8n_chat' as const;
/** Fixed tool names for the implicit in-app chat integration (no credential suffixes). */

View File

@ -228,3 +228,51 @@ it('leaves a checkpoint claimed by a later execution intact during cleanup', asy
expect(checkpointStorage.cancelSuspended).not.toHaveBeenCalled();
expect(checkpointStorage.delete).not.toHaveBeenCalled();
});
it('requires the production source before cancelling a suspended run', async () => {
const { service, executionService, checkpointStorage } = makeService();
checkpointStorage.getStatus.mockResolvedValue({
status: 'active',
checkpoint: {
...checkpoint,
persistence: {
threadId: context.threadId,
resourceId: 'n8n-chat-production:user-1',
},
},
});
executionService.canUseProductionChatThread.mockResolvedValue(false);
expect(
await service.cancelSuspended({
agentId: context.agentId,
runId: 'run-1',
resourceId: 'n8n-chat-production:user-1',
}),
).toBe(false);
expect(checkpointStorage.cancelSuspended).not.toHaveBeenCalled();
executionService.canUseProductionChatThread.mockResolvedValue(true);
expect(
await service.cancelSuspended({
agentId: context.agentId,
runId: 'run-1',
resourceId: 'n8n-chat-production:user-1',
}),
).toBe(true);
expect(executionService.canUseProductionChatThread).toHaveBeenCalledWith(
context.threadId,
context.projectId,
context.agentId,
context.userId,
'existing',
);
});
it('rejects preview and foreign executions on the production cancel route', async () => {
const { service, repository } = makeService();
const production = { ...context, productionN8nChat: true };
await expect(service.requestCancel(production)).rejects.toBeInstanceOf(NotFoundError);
repository.findOneBy.mockResolvedValue({ ...running, source: 'n8n_chat_production' });
await expect(
service.requestCancel({ ...production, userId: 'other-user' }),
).rejects.toBeInstanceOf(NotFoundError);
});

View File

@ -40,7 +40,12 @@ import {
function makeController() {
const agentsService =
mock<Pick<AgentsService, 'findById' | 'findByProjectId' | 'findByProjectIdPaginated'>>();
mock<
Pick<
AgentsService,
'findById' | 'findByProjectId' | 'findByProjectIdPaginated' | 'isN8nChatPublished'
>
>();
const agentExecutionOrchestratorService = mock<AgentExecutionOrchestratorService>();
const agentsBuilderService = mock<AgentsBuilderService>();
const agentChatAttachmentService = mock<AgentChatAttachmentService>();
@ -59,6 +64,7 @@ function makeController() {
const chatExecutionService = mock<AgentChatExecutionService>();
agentExecutionService.findThreadById.mockResolvedValue(null);
agentExecutionService.canUseDraftThread.mockResolvedValue(true);
agentExecutionService.canUseProductionChatThread.mockResolvedValue(true);
agentValidationService.validateAgentIsRunnable.mockResolvedValue({ missing: [] });
const agentTestRunService = new AgentTestRunService(
agentExecutionService,
@ -108,9 +114,10 @@ function makeController() {
agentChatAttachmentService,
agentsService: {
findById: agentsService.findById,
isN8nChatPublished: agentsService.isN8nChatPublished,
getConversationHistory: agentExecutionOrchestratorService.getConversationHistory,
} as Mocked<
Pick<AgentsService, 'findById'> &
Pick<AgentsService, 'findById' | 'isN8nChatPublished'> &
Pick<AgentExecutionOrchestratorService, 'getConversationHistory'>
>,
};
@ -153,6 +160,12 @@ describe('AgentChatController route access scopes', () => {
const routes = getRoutesByHandlerName(AgentChatController);
it.each([
['productionChat', 'agent:execute'],
['productionChatResume', 'agent:execute'],
['cancelProductionChatExecution', 'agent:execute'],
['cancelProductionChatRun', 'agent:execute'],
['getProductionChatMessages', 'agent:read'],
['getProductionChatAttachment', 'agent:read'],
['chat', 'agent:execute'],
['chatResume', 'agent:execute'],
['cancelChatRun', 'agent:execute'],
@ -942,3 +955,118 @@ describe('AgentChatController attachment download', () => {
expect((res as { setHeader: ReturnType<typeof vi.fn> }).setHeader).not.toHaveBeenCalled();
});
});
describe('AgentChatController production n8n Chat', () => {
const request = { params: { projectId: 'project-1' }, user: { id: 'user-1' } };
it('rejects an unpublished channel before starting a turn', async () => {
const { controller, agentsService, agentExecutionOrchestratorService } = makeController();
agentsService.isN8nChatPublished.mockResolvedValue(false);
const writes: string[] = [];
await controller.productionChat(request as never, makeSseResponse(writes), 'agent-1', {
message: 'hello',
} as never);
expect(writes).toContain(
'data: {"type":"error","message":"This agent is not available in n8n Chat.","errorCode":"agent_unavailable"}\n\n',
);
expect(agentExecutionOrchestratorService.executeForN8nChatPublished).not.toHaveBeenCalled();
});
it('records a published turn with a user-owned memory resource', async () => {
const { controller, agentsService, agentExecutionOrchestratorService } = makeController();
agentsService.isN8nChatPublished.mockResolvedValue(true);
agentExecutionOrchestratorService.executeForN8nChatPublished.mockImplementation(
async function* () {
yield { type: 'text-delta', id: 'text-1', delta: 'Hi' };
},
);
const writes: string[] = [];
await controller.productionChat(request as never, makeSseResponse(writes), 'agent-1', {
message: 'hello',
} as never);
expect(agentExecutionOrchestratorService.executeForN8nChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
agentId: 'agent-1',
message: 'hello',
sessionMode: 'new',
memory: expect.objectContaining({ resourceId: 'n8n-chat-production:user-1' }),
}),
);
expect(writes.some((line) => line.includes('"delta":"Hi"'))).toBe(true);
});
it('rejects a foreign session before saving an attachment', async () => {
const { controller, agentsService, agentExecutionService, agentChatAttachmentService } =
makeController();
agentsService.isN8nChatPublished.mockResolvedValue(true);
agentExecutionService.canUseProductionChatThread.mockResolvedValue(false);
const writes: string[] = [];
await controller.productionChat(request as never, makeSseResponse(writes), 'agent-1', {
message: 'hello',
sessionId: 'foreign-thread',
attachments: [
{
fileName: 'note.txt',
mimeType: 'text/plain',
data: Buffer.from('hi').toString('base64'),
},
],
} as never);
expect(agentChatAttachmentService.storeInbound).not.toHaveBeenCalled();
expect(writes.some((line) => line.includes('Session not found'))).toBe(true);
});
it('checks publication and the production checkpoint scope before resuming', async () => {
const { controller, agentsService, agentExecutionOrchestratorService } = makeController();
agentsService.isN8nChatPublished.mockResolvedValue(true);
agentExecutionOrchestratorService.resumeForChat.mockImplementation(async function* () {});
await controller.productionChatResume(request as never, makeSseResponse([]), 'agent-1', {
runId: 'run-1',
toolCallId: 'call-1',
resumeData: { approved: true },
} as never);
expect(agentExecutionOrchestratorService.resumeForChat).toHaveBeenCalledWith(
expect.objectContaining({
usePublishedVersion: true,
source: 'n8n_chat_production',
expectedMemory: { resourceId: 'n8n-chat-production:user-1' },
}),
);
});
it('rejects production resume when the channel is not published', async () => {
const { controller, agentsService, agentExecutionOrchestratorService } = makeController();
agentsService.isN8nChatPublished.mockResolvedValue(false);
const writes: string[] = [];
await controller.productionChatResume(request as never, makeSseResponse(writes), 'agent-1', {
runId: 'run-1',
toolCallId: 'call-1',
resumeData: { approved: true },
} as never);
expect(agentExecutionOrchestratorService.resumeForChat).not.toHaveBeenCalled();
expect(writes.some((line) => line.includes('agent_unavailable'))).toBe(true);
});
it('does not finish a turn that suspended for HITL', async () => {
const { controller, agentsService, agentExecutionOrchestratorService } = makeController();
agentsService.isN8nChatPublished.mockResolvedValue(true);
agentExecutionOrchestratorService.executeForN8nChatPublished.mockImplementation(
async function* () {
yield {
type: 'tool-call-suspended',
toolCallId: 'call-1',
toolName: 'ask_questions',
runId: 'run-1',
suspendPayload: { type: 'questions', questions: [] },
};
},
);
const writes: string[] = [];
await controller.productionChat(request as never, makeSseResponse(writes), 'agent-1', {
message: 'Ask me first',
} as never);
expect(writes.some((line) => line.includes('"type":"tool-call-suspended"'))).toBe(true);
expect(writes.some((line) => line.includes('"type":"done"'))).toBe(false);
});
});

View File

@ -424,6 +424,29 @@ describe('AgentConfigService', () => {
expect(runtimeCacheService.clearRuntimes).toHaveBeenCalledWith(agentId);
});
it('persists n8n Chat as a draft channel', async () => {
const { service, agentRepository } = makeService();
const agent = makeAgent({ integrations: [{ type: 'slack', credentialId: 'slack-cred' }] });
agentRepository.findByIdAndProjectId.mockResolvedValue(agent);
await service.updateConfig(
agentId,
projectId,
{
...baseConfig,
integrations: [{ type: 'slack', credentialId: 'slack-cred' }, { type: 'n8n_chat' }],
},
user,
fencedOn(agent),
);
expect(agent.integrations).toEqual([
{ type: 'slack', credentialId: '' },
{ type: 'n8n_chat', credentialId: '' },
]);
const saved = agentRepository.saveDraftFenced.mock.calls.at(-1)?.[0];
expect(saved?.integrations).toEqual(agent.integrations);
expect(composeJsonConfig(agent)?.integrations).toEqual(agent.integrations);
});
it('persists modelDeploymentName, retains it when omitted, and drops it on clearOmittedOptionalFields', async () => {
const { service, agentRepository } = makeService();
const agent = makeAgent();

View File

@ -1440,6 +1440,124 @@ describe('AgentExecutionOrchestratorService', () => {
);
});
it('runs production n8n Chat with a published user-owned session', async () => {
const {
service,
agentRepository,
runtimeCacheService,
executionService,
integrationMessageContextService,
} = makeService();
agentRepository.isN8nChatPublished.mockResolvedValue(true);
executionService.canUseProductionChatThread.mockResolvedValue(true);
runtimeCacheService.getRuntime.mockResolvedValue(makeRuntime());
await collect(
service.executeForN8nChatPublished({
agentId,
projectId,
user,
message: 'Hello',
memory: { threadId: 'thread-1', resourceId: 'n8n-chat-production:user-1' },
}),
);
expect(runtimeCacheService.getRuntime).toHaveBeenCalledWith(
expect.objectContaining({
integrationType: N8N_CHAT_INTEGRATION_TYPE,
usePublishedVersion: true,
allowBackgroundTasks: false,
attributionUserId: user.id,
}),
);
expect(integrationMessageContextService.setLatest).toHaveBeenCalledWith(
'thread-1',
'n8n-chat-production:user-1',
expect.objectContaining({
platform: N8N_CHAT_INTEGRATION_TYPE,
interactingUserId: user.id,
target: { type: 'dm', userId: user.id, threadId: 'thread-1' },
}),
);
expect(executionService.startExecutionRecording).toHaveBeenCalledWith(
expect.objectContaining({
access: { accessScope: 'user', ownerId: user.id },
source: 'n8n_chat_production',
telemetry: expect.objectContaining({ runType: 'production', userId: user.id }),
}),
expect.any(Date),
);
});
it('rejects a production turn with a foreign thread or memory scope', async () => {
const { service, agentRepository, runtimeCacheService, executionService } = makeService();
agentRepository.isN8nChatPublished.mockResolvedValue(true);
for (const resourceId of ['draft-chat:user-1', 'n8n-chat-production:other-user']) {
await expect(
collect(
service.executeForN8nChatPublished({
agentId,
projectId,
user,
message: 'Hello',
memory: { threadId: 'thread-1', resourceId },
}),
),
).rejects.toThrow('Session not found');
}
executionService.canUseProductionChatThread.mockResolvedValue(false);
await expect(
collect(
service.executeForN8nChatPublished({
agentId,
projectId,
user,
message: 'Hello',
memory: { threadId: 'thread-1', resourceId: 'n8n-chat-production:user-1' },
}),
),
).rejects.toThrow('Session not found');
expect(runtimeCacheService.getRuntime).not.toHaveBeenCalled();
});
it('rejects a production resume with a different sandbox principal', async () => {
const { service, agentRepository, checkpointStorage, executionService, runtimeCacheService } =
makeService();
agentRepository.isN8nChatPublished.mockResolvedValue(true);
executionService.canUseProductionChatThread.mockResolvedValue(true);
checkpointStorage.getStatus.mockResolvedValue({
status: 'active',
checkpoint: makeCheckpoint(
{},
{
threadId: 'thread-1',
resourceId: 'n8n-chat-production:user-1',
hostMetadata: encodeAgentSandboxHostMetadata({
projectId,
principalHash: integrationPrincipalHash,
}),
},
),
});
await expect(
collect(
service.resumeForChat({
agentId,
projectId,
user,
runId: 'run-1',
toolCallId: 'call-1',
resumeData: { approved: true },
usePublishedVersion: true,
integrationType: N8N_CHAT_INTEGRATION_TYPE,
source: 'n8n_chat_production',
}),
),
).rejects.toThrow('unavailable');
expect(runtimeCacheService.getRuntime).not.toHaveBeenCalled();
});
it.each(['none', 'startExecutionRecording', 'finalizeExecution'] as const)(
'reports preview initialization failures with the recording outcome (%s)',
async (failure) => {
@ -2044,6 +2162,50 @@ describe('AgentExecutionOrchestratorService', () => {
).toEqual(messageContext);
});
it('records a production n8n Chat wake in the owning user session', async () => {
const {
service,
agentRepository,
executionService,
runtimeCacheService,
chatIntegrationService,
} = makeService();
const runtime = makeRuntime();
agentRepository.isN8nChatPublished.mockResolvedValue(true);
executionService.canUseProductionChatThread.mockResolvedValue(true);
runtimeCacheService.getRuntime.mockResolvedValue(runtime);
await service.executeForWake({
backgroundJobSignal,
agentId,
projectId,
message: 'The job is done.',
memory: { threadId: 'thread-1', resourceId: 'n8n-chat-production:user-1' },
identity: {
type: 'published',
integrationType: N8N_CHAT_INTEGRATION_TYPE,
principalHash: userPrincipalHash,
},
abortSignal: new AbortController().signal,
});
expect(runtimeCacheService.getRuntime).toHaveBeenCalledWith(
expect.objectContaining({
integrationType: N8N_CHAT_INTEGRATION_TYPE,
usePublishedVersion: true,
attributionUserId: userId,
}),
);
expect(executionService.startExecutionRecording).toHaveBeenCalledWith(
expect.objectContaining({
source: 'n8n_chat_production',
access: { accessScope: 'user', ownerId: userId },
}),
expect.any(Date),
);
expect(chatIntegrationService.getBridge).not.toHaveBeenCalled();
});
it('rejects a wake when chat delivery fails and releases the runtime', async () => {
const { service, executionService, runtimeCacheService, bridge } = makeService();
const runtime = makeRuntime();

View File

@ -1240,6 +1240,53 @@ describe('AgentExecutionService', () => {
});
});
describe('canUseProductionChatThread', () => {
it.each([
{ source: 'n8n_chat_production', allowed: true },
{ source: 'n8n_chat', allowed: false },
{ source: 'chat', allowed: false },
])('checks the first source: $source', async ({ source, allowed }) => {
agentExecutionThreadRepository.findOneBy.mockResolvedValue(makeThread());
agentExecutionRepository.findFirstSourceByThreadIds.mockResolvedValue(
new Map([['thread-1', source]]),
);
expect(
await service.canUseProductionChatThread(
'thread-1',
'project-1',
'agent-1',
'user-1',
'existing',
),
).toBe(allowed);
});
it('rejects another owner and an unknown existing session', async () => {
agentExecutionThreadRepository.findOneBy.mockResolvedValue(
makeThread({ ownerId: 'other-user' }),
);
expect(
await service.canUseProductionChatThread(
'thread-1',
'project-1',
'agent-1',
'user-1',
'existing',
),
).toBe(false);
agentExecutionThreadRepository.findOneBy.mockResolvedValue(null);
expect(
await service.canUseProductionChatThread(
'thread-1',
'project-1',
'agent-1',
'user-1',
'existing',
),
).toBe(false);
});
});
describe('getThreadDetail', () => {
it.each(['user', 'project'] as const)(
'returns readable %s thread executions',

View File

@ -469,6 +469,39 @@ describe('AgentIntegrationsController channel status', () => {
]);
});
it('reports draft and active n8n Chat availability separately', async () => {
const agent = {
...publishedAgent,
activeVersion: {
schema: {
name: 'Agent',
model: 'openai:gpt-4o-mini',
instructions: 'Help',
integrations: [{ type: 'n8n_chat', credentialId: '' }],
},
},
} as Agent;
const { response } = await statusOf(agent, []);
expect(response.n8nChat).toEqual({ draftEnabled: false, publishedEnabled: true });
expect(response.integrations).toEqual([
{ type: slack.type, credentialId: slack.credentialId, status: 'starting' },
{ type: telegram.type, credentialId: telegram.credentialId, status: 'starting' },
{ type: 'n8n_chat', status: 'connected' },
]);
const unpublished = {
...agent,
activeVersionId: null,
activeVersion: null,
integrations: [...agent.integrations, { type: 'n8n_chat', credentialId: '' }],
} as Agent;
const { response: unpublishedResponse } = await statusOf(unpublished, []);
expect(unpublishedResponse.n8nChat).toEqual({ draftEnabled: true, publishedEnabled: false });
expect(unpublishedResponse.integrations).toContainEqual({
type: 'n8n_chat',
status: 'configured',
});
});
it('reports the reason a channel could not start', async () => {
const failed = {
...liveRow(telegram),

View File

@ -404,7 +404,7 @@ describe('AgentPublishService', () => {
tasks: [{ type: 'task', id: 'task-1', enabled: true }],
},
skills: configuredSkills,
integrations,
integrations: [...integrations, { type: 'n8n_chat', credentialId: '' }],
});
const draftValidation = { status: 'valid' as const, issues: [] };
const task = {
@ -435,7 +435,7 @@ describe('AgentPublishService', () => {
{
versionId,
agentId,
schema: agent.schema,
schema: { ...agent.schema, integrations: [{ type: 'n8n_chat', credentialId: '' }] },
tools: configuredTools,
skills: configuredSkills,
publishedBy: user,
@ -605,8 +605,12 @@ describe('AgentPublishService', () => {
it('switches to an existing history row when publishing a specific version', async () => {
const { service, agentRepository, agentHistoryRepository, trx } = makeService();
const agent = makeAgent({ versionId: 'draft-v2', activeVersionId: 'v0' });
const target = makeHistory({ versionId: 'v1' });
const agent = makeAgent({
versionId: 'draft-v2',
activeVersionId: 'v0',
integrations: [{ type: 'n8n_chat', credentialId: '' }],
});
const target = makeHistory({ versionId: 'v1', schema: { ...schema, integrations: [] } });
agentRepository.findByIdAndProjectId.mockResolvedValue(agent);
agentHistoryRepository.findByVersionAndAgentId.mockResolvedValue(target);
@ -619,6 +623,9 @@ describe('AgentPublishService', () => {
expect(agentHistoryRepository.findByVersionAndAgentId).toHaveBeenCalledWith('v1', agentId);
expect(agent.activeVersionId).toBe('v1');
expect(agent.activeVersion).toBe(target);
expect(agent.integrations).toEqual([{ type: 'n8n_chat', credentialId: '' }]);
expect(target.schema?.integrations).toEqual([]);
expect(agentHistoryRepository.saveVersion).not.toHaveBeenCalled();
expect(agent.versionId).not.toBe('draft-v2');
expect(agentRepository.setActiveVersionFenced).toHaveBeenCalledWith(
agent.id,
@ -706,10 +713,11 @@ describe('AgentPublishService', () => {
const agent = makeAgent({
activeVersionId: 'current-active',
activeVersion: makeHistory({ versionId: 'current-active' }),
integrations: [{ type: 'n8n_chat', credentialId: '' }],
});
const target = makeHistory({
versionId: 'older-version',
schema: { ...schema, name: 'Older Agent' },
schema: { ...schema, name: 'Older Agent', integrations: [] },
});
agentRepository.findByIdAndProjectId.mockResolvedValue(agent);
@ -726,9 +734,10 @@ describe('AgentPublishService', () => {
await service.revertToVersion(agentId, projectId, 'older-version', user, 'user');
expect(agent.schema).toEqual(target.schema);
expect(agent.schema).toEqual({ ...schema, name: 'Older Agent' });
expect(agent.name).toBe('Older Agent');
expect(agent.activeVersionId).toBe('current-active');
expect(agent.integrations).toEqual([{ type: 'n8n_chat', credentialId: '' }]);
expect(agent.versionId).not.toBe('older-version');
expect(taskRepo.delete).toHaveBeenCalledWith(['draft-only']);
expect(taskRepo.update).toHaveBeenCalledWith(

View File

@ -624,6 +624,51 @@ describe('AgentRuntimeCacheService', () => {
);
});
it('keeps production n8n Chat on published instructions with project-scoped tools', async () => {
const { service, agentRepository, reconstructionService } = makeService();
const agent = makeAgent({
schema: {
name: 'Draft',
model: 'openai:gpt-4o',
instructions: 'Draft instructions',
},
activeVersion: {
schema: {
name: 'Published',
model: 'openai:gpt-4o',
instructions: 'Published instructions',
},
tools: {},
skills: {},
} as Agent['activeVersion'],
});
agentRepository.findByIdAndProjectId.mockResolvedValue(agent);
reconstructionService.reconstructFromAgentEntity.mockResolvedValue(makeRuntime());
await service.getRuntime({
agentId,
projectId,
usePublishedVersion: true,
integrationType: 'n8n_chat',
attributionUserId: 'user-1',
allowBackgroundTasks: false,
});
expect(reconstructionService.reconstructFromAgentEntity).toHaveBeenCalledWith(
expect.objectContaining({
schema: expect.objectContaining({ instructions: 'Published instructions' }),
}),
expect.anything(),
'production',
'n8n_chat',
undefined,
undefined,
'integrated',
undefined,
{ previewChat: undefined, allowBackgroundTasks: false, attributionUserId: 'user-1' },
);
});
it('rejects missing agents and unpublished runtime requests', async () => {
const { service, agentRepository } = makeService();

View File

@ -1,9 +1,19 @@
import type * as agents from '@n8n/agents';
import type { CredentialProvider } from '@n8n/agents';
import type { AgentJsonConfig, AgentJsonToolConfig } from '@n8n/api-types';
import {
N8N_CHAT_INTEGRATION_TYPE,
type AgentJsonConfig,
type AgentJsonToolConfig,
} from '@n8n/api-types';
import type { Logger } from '@n8n/backend-common';
import type { CustomFetch, HttpTransport, OutboundHttp } from '@n8n/backend-network';
import type { CredentialsEntity, User, WorkflowEntity, WorkflowRepository } from '@n8n/db';
import type {
CredentialsEntity,
User,
UserRepository,
WorkflowEntity,
WorkflowRepository,
} from '@n8n/db';
import { Container } from '@n8n/di';
import { mock } from 'vitest-mock-extended';
@ -23,8 +33,13 @@ import { AgentRuntimeReconstructionService } from '../agent-runtime-reconstructi
import type { AgentSandboxRuntimeService } from '../agent-sandbox-runtime.service';
import type { AgentWorkspaceService } from '../agent-workspace.service';
import type { Agent } from '../entities/agent.entity';
import { ChatIntegrationRegistry } from '../integrations/agent-chat-integration';
import { ChatIntegrationActionExecutor } from '../integrations/integration-action-executor';
import { ChatIntegrationContextQueryExecutor } from '../integrations/integration-context-query-executor';
import { IntegrationMessageContextService } from '../integrations/integration-message-context.service';
import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage';
import type { N8nMemory } from '../integrations/n8n-memory';
import { N8nChatIntegration } from '../integrations/platforms/n8n-chat-integration';
import type * as FromJsonConfig from '../json-config/from-json-config';
import type { BuildFromJsonOptions, ToolExecutor } from '../json-config/from-json-config';
import type { AgentFileRepository } from '../repositories/agent-file.repository';
@ -175,6 +190,15 @@ function buildFromJsonResolvingTools(resolved: Array<agents.BuiltTool | null | u
);
}
function setupN8nChatToolDependencies() {
const registry = new ChatIntegrationRegistry();
registry.register(new N8nChatIntegration(mock<UserRepository>()));
Container.set(ChatIntegrationRegistry, registry);
Container.set(IntegrationMessageContextService, mock<IntegrationMessageContextService>());
Container.set(ChatIntegrationActionExecutor, mock<ChatIntegrationActionExecutor>());
Container.set(ChatIntegrationContextQueryExecutor, mock<ChatIntegrationContextQueryExecutor>());
}
function toolNamesPassedToBuildFromJson(): string[] {
const [config] = buildFromJsonMock.mock.calls.at(-1) as [AgentJsonConfig];
return (config.tools ?? []).map((ref) =>
@ -206,6 +230,54 @@ describe('AgentRuntimeReconstructionService — per-user tool filtering', () =>
);
});
it('forwards the production n8n Chat marker to workflow tools', async () => {
const { service } = makeService({});
setupN8nChatToolDependencies();
buildFromJsonResolvingTools([]);
await service.reconstructFromAgentEntity(
makeAgentEntity([workflowTool]),
mock<CredentialProvider>(),
'production',
N8N_CHAT_INTEGRATION_TYPE,
undefined,
undefined,
'manual',
undefined,
{ attributionUserId: userId },
);
expect(resolveWorkflowToolMock).toHaveBeenCalledWith(
workflowTool,
expect.objectContaining({
integrationType: N8N_CHAT_INTEGRATION_TYPE,
userId,
publishedN8nChat: true,
}),
);
});
it('forwards the preview marker to workflow tools', async () => {
const { service } = makeService({});
setupN8nChatToolDependencies();
buildFromJsonResolvingTools([]);
await service.reconstructFromAgentEntity(
makeAgentEntity([workflowTool]),
mock<CredentialProvider>(),
'test',
N8N_CHAT_INTEGRATION_TYPE,
undefined,
undefined,
'manual',
undefined,
{ previewChat: true },
);
expect(resolveWorkflowToolMock).toHaveBeenCalledWith(
workflowTool,
expect.objectContaining({ previewChat: true, publishedN8nChat: false }),
);
});
it('drops node and workflow tools for a user without workflow:execute, keeps custom tools', async () => {
vi.mocked(userHasScopes).mockResolvedValue(false);
const { service } = makeService({});

View File

@ -897,6 +897,41 @@ describe('AgentValidationService — structured issues', () => {
]);
});
it('accepts credential-free n8n Chat without relaxing other channel checks', async () => {
const { service, agentRepository } = makeService();
agentRepository.findByIdAndProjectId.mockResolvedValue(
makeAgent(runnableConfig, {}, { integrations: [{ type: 'n8n_chat', credentialId: '' }] }),
);
const credentials = makeCredentialProvider([{ id: 'openai-main', type: 'openAiApi' }]);
await expect(
service.validateAgentConfiguration(agentId, projectId, credentials),
).resolves.toEqual({ status: 'valid', issues: [] });
agentRepository.findByIdAndProjectId.mockResolvedValue(
makeAgent(
runnableConfig,
{},
{
integrations: [
{ type: 'n8n_chat', credentialId: '' },
{ type: 'slack', credentialId: '' },
],
},
),
);
const result = await service.validateAgentConfiguration(agentId, projectId, credentials);
expect(result.status).toBe('invalid');
expect(result.issues).toEqual([
expect.objectContaining({
code: 'missing_credential',
path: 'integrations.1.credentialId',
capability: { kind: 'channel', id: 'slack', index: 1 },
}),
]);
});
it('flags a custom tool without a saved body', async () => {
const { service, agentRepository } = makeService();
agentRepository.findByIdAndProjectId.mockResolvedValue(

View File

@ -12,6 +12,8 @@ import { mock } from 'vitest-mock-extended';
import type { Publisher } from '@/scaling/pubsub/publisher.service';
import type { AgentExecutionUpdateBroadcaster } from '../agent-execution-update-broadcaster';
import type { AgentExecutionOrchestratorService } from '../agent-execution-orchestrator.service';
import type { AgentRepository } from '../repositories/agent.repository';
import type { AgentTestRunService } from '../agent-test-run.service';
import { AgentWorkflowToolResumeService } from '../agent-workflow-tool-resume.service';
import type { AgentBackgroundJobService } from '../background/agent-background-job.service';
@ -58,6 +60,8 @@ function setup() {
},
} as never);
const backgroundJobService = mock<AgentBackgroundJobService>();
const orchestratorService = mock<AgentExecutionOrchestratorService>();
const agentRepository = mock<AgentRepository>();
const service = new AgentWorkflowToolResumeService(
logger,
userRepository,
@ -69,6 +73,8 @@ function setup() {
instanceSettings,
publisher,
backgroundJobService,
orchestratorService,
agentRepository,
);
return {
service,
@ -83,9 +89,42 @@ function setup() {
instanceSettings,
messageContextService,
backgroundJobService,
orchestratorService,
agentRepository,
};
}
describe('AgentWorkflowToolResumeService production n8n Chat', () => {
it('resumes the published runtime for the owning user', async () => {
const { service, userRepository, agentRepository, orchestratorService, agentTestRunService } =
setup();
userRepository.findOneBy.mockResolvedValue({ id: 'user-1' } as User);
agentRepository.isN8nChatPublished.mockResolvedValue(true);
orchestratorService.resumeForChat.mockImplementation(async function* () {});
await service.resume({ ...previewRun, publishedN8nChat: true }, 'success');
expect(agentTestRunService.resumeDraftRun).not.toHaveBeenCalled();
expect(orchestratorService.resumeForChat).toHaveBeenCalledWith(
expect.objectContaining({
usePublishedVersion: true,
source: 'n8n_chat_production',
user: expect.objectContaining({ id: 'user-1' }),
expectedMemory: {
threadId: previewRun.threadId,
resourceId: 'n8n-chat-production:user-1',
},
}),
);
});
it('does not resume an unpublished agent', async () => {
const { service, userRepository, agentRepository, orchestratorService } = setup();
userRepository.findOneBy.mockResolvedValue({ id: 'user-1' } as User);
agentRepository.isN8nChatPublished.mockResolvedValue(false);
await service.resume({ ...previewRun, publishedN8nChat: true }, 'success');
expect(orchestratorService.resumeForChat).not.toHaveBeenCalled();
});
});
/** A `workflowExecuteAfter` context for a sub-execution carrying an agent marker. */
function afterContext(
status: IRun['status'],

View File

@ -13,9 +13,15 @@ import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage';
import { AgentExecutionRepository } from './repositories/agent-execution.repository';
import {
draftChatMemoryResourceId,
productionChatMemoryResourceId,
userIdFromDraftChatMemoryResourceId,
userIdFromProductionChatMemoryResourceId,
} from './utils/agent-memory-scope';
import { canContinueThreadInPreview, threadBelongsTo } from './utils/agent-thread-access';
import {
canContinueThreadInPreview,
N8N_CHAT_PRODUCTION_SOURCE,
threadBelongsTo,
} from './utils/agent-thread-access';
import { getDelegatedChildCheckpoints } from './utils/delegated-child-checkpoints';
type ExecutionContext = PubSubCommandMap['cancel-agent-chat-execution'];
@ -87,7 +93,9 @@ export class AgentChatExecutionService {
await this.cancelSuspended({
agentId: context.agentId,
runId: suspendedRunId,
resourceId: draftChatMemoryResourceId(context.userId),
resourceId: context.productionN8nChat
? productionChatMemoryResourceId(context.userId)
: draftChatMemoryResourceId(context.userId),
});
},
);
@ -162,7 +170,17 @@ 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 || !canContinueThreadInPreview(thread, userId, execution.source)) return null;
if (
!execution ||
(context.productionN8nChat
? !(
thread.accessScope === 'user' &&
thread.ownerId === userId &&
execution.source === N8N_CHAT_PRODUCTION_SOURCE
)
: !canContinueThreadInPreview(thread, userId, execution.source))
)
return null;
return execution;
}
@ -180,7 +198,9 @@ export class AgentChatExecutionService {
return await this.cancelSuspended({
agentId: context.agentId,
runId: pending.runId,
resourceId: draftChatMemoryResourceId(context.userId),
resourceId: context.productionN8nChat
? productionChatMemoryResourceId(context.userId)
: draftChatMemoryResourceId(context.userId),
});
}
@ -196,7 +216,20 @@ export class AgentChatExecutionService {
)
return false;
const thread = await this.executionService.findThreadById(checkpoint.persistence.threadId);
const userId = userIdFromDraftChatMemoryResourceId(params.resourceId);
const productionUserId = userIdFromProductionChatMemoryResourceId(params.resourceId);
const userId = userIdFromDraftChatMemoryResourceId(params.resourceId) ?? productionUserId;
if (
productionUserId &&
(!thread ||
!(await this.executionService.canUseProductionChatThread(
thread.id,
thread.projectId,
params.agentId,
productionUserId,
'existing',
)))
)
return false;
if (thread && (!userId || !threadBelongsTo(thread, thread.projectId, params.agentId, userId)))
return false;
const childCheckpoints = getDelegatedChildCheckpoints(checkpoint, params.agentId);

View File

@ -26,6 +26,7 @@ import { sanitizeFilename } from '@n8n/utils/files/sanitize-filename';
import type { Response } from 'express';
import { FileNotFoundError, getHtmlSandboxCSP } from 'n8n-core';
import { pipeline } from 'node:stream/promises';
import { randomUUID } from 'node:crypto';
import { CredentialsService } from '@/credentials/credentials.service';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
@ -33,8 +34,10 @@ import { NotFoundError } from '@/errors/response-errors/not-found.error';
import { AgentsCredentialProvider } from './adapters/agents-credential-provider';
import { AgentChatAttachmentService } from './agent-chat-attachment.service';
import type { AgentChatAttachment } from './entities/agent-chat-attachment.entity';
import type { StoredAttachmentRef } from './types/agent-chat-attachment';
import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service';
import { AgentExecutionRecordingError } from './agent-execution-recording.error';
import { AgentMessageQueueService } from './agent-message-queue.service';
import { AgentQueuedPreviewStreamService } from './agent-queued-preview-stream.service';
import {
@ -42,7 +45,7 @@ import {
AgentTurnAlreadyRunningError,
} from './agent-chat-execution.service';
import { AgentExecutionService } from './agent-execution.service';
import { threadBelongsTo } from './utils/agent-thread-access';
import { N8N_CHAT_PRODUCTION_SOURCE, threadBelongsTo } from './utils/agent-thread-access';
import { messagesToDto } from './agent-message-mapper';
import { type FlushableResponse, initSseStream } from './agent-sse-stream';
import { AgentTestChatService, chatThreadId } from './agent-test-chat.service';
@ -52,6 +55,7 @@ import { AgentsBuilderService } from './builder/agents-builder.service';
import { AgentBackgroundJobService } from './background/agent-background-job.service';
import {
draftChatMemoryResourceId,
productionChatMemoryResourceId,
userIdFromDraftChatMemoryResourceId,
} from './utils/agent-memory-scope';
import { resolveInboundMimeType } from './utils/inbound-attachments';
@ -102,8 +106,9 @@ export class AgentChatController {
projectId: string;
threadId: string;
resourceId: string;
source?: string;
}): Promise<StoredAttachmentRef[] | undefined> {
const { attachments, agentId, projectId, threadId, resourceId } = params;
const { attachments, agentId, projectId, threadId, resourceId, source = 'chat' } = params;
if (!attachments?.length) return undefined;
const stored: StoredAttachmentRef[] = [];
@ -125,7 +130,7 @@ export class AgentChatController {
projectId,
threadId,
resourceId,
source: 'chat',
source,
fileName: attachment.fileName,
mimeType,
data,
@ -145,6 +150,184 @@ export class AgentChatController {
return stored;
}
private async requireProductionChat(agentId: string, projectId: string): Promise<void> {
if (!(await this.agentsService.isN8nChatPublished(agentId, projectId))) {
throw new NotFoundError('Agent is not available in n8n Chat');
}
}
private async requireProductionThread(
threadId: string,
projectId: string,
agentId: string,
userId: string,
): Promise<void> {
if (
!(await this.agentExecutionService.canUseProductionChatThread(
threadId,
projectId,
agentId,
userId,
'existing',
))
)
throw new NotFoundError('Session not found');
}
@Post('/:agentId/n8n-chat', { usesTemplates: true })
@ProjectScope('agent:execute')
async productionChat(
req: AuthenticatedRequest<{ projectId: string }>,
res: FlushableResponse,
@Param('agentId') agentId: string,
@Body payload: AgentChatMessageDto,
) {
const { projectId } = req.params;
const execution = this.createChatExecution(res);
const { send, onChunk, abortSignal } = execution;
let executionId: string | undefined;
// `onExecutionRecorded` only fires once the turn finalizes. A disconnect or an
// error before that leaves the started id as the only way to name the execution.
let startedExecutionId: string | undefined;
const onExecutionStarted = (id: string, sessionId: string) => {
startedExecutionId = id;
execution.onExecutionStarted(id, sessionId);
};
let storedAttachments: StoredAttachmentRef[] | undefined;
try {
if (!(await this.agentsService.isN8nChatPublished(agentId, projectId))) {
send({
type: 'error',
message: 'This agent is not available in n8n Chat.',
errorCode: 'agent_unavailable',
});
return;
}
const sessionMode = payload.sessionId && !payload.newSession ? 'existing' : 'new';
const threadId = sessionMode === 'existing' ? payload.sessionId : randomUUID();
if (!threadId) throw new NotFoundError('Session not found');
if (sessionMode === 'existing') {
await this.requireProductionThread(threadId, projectId, agentId, req.user.id);
}
storedAttachments = await this.storeChatAttachments({
attachments: payload.attachments,
agentId,
projectId,
threadId,
resourceId: productionChatMemoryResourceId(req.user.id),
source: N8N_CHAT_PRODUCTION_SOURCE,
});
abortSignal.throwIfAborted();
const stream = this.agentExecutionOrchestratorService.executeForN8nChatPublished({
agentId,
projectId,
user: req.user,
message: payload.message,
memory: { threadId, resourceId: productionChatMemoryResourceId(req.user.id) },
attachments: storedAttachments,
sessionMode,
onExecutionStarted,
onExecutionRecorded: (id) => {
executionId = id;
},
abortSignal,
});
let suspended = false;
let failed = false;
for await (const chunk of stream) {
onChunk(chunk);
if (chunk.type === 'tool-call-suspended') suspended = true;
if (chunk.type === 'error') failed = true;
}
executionId ??= startedExecutionId;
if (!suspended && !failed) {
send({ type: 'done', sessionId: threadId, ...(executionId ? { executionId } : {}) });
}
} catch (error) {
executionId ??= startedExecutionId;
if (error instanceof AgentExecutionRecordingError) executionId ??= error.executionId;
if (!executionId && storedAttachments?.length) {
await this.agentChatAttachmentService
.deleteByIds(storedAttachments.map((ref) => ref.id))
.catch(() => {});
}
send({
type: 'error',
message: error instanceof Error ? error.message : 'Chat failed',
...(error instanceof AgentTurnAlreadyRunningError
? { errorCode: 'turn_already_running' }
: {}),
});
} finally {
execution.close();
}
}
@Post('/:agentId/n8n-chat/resume', { usesTemplates: true })
@ProjectScope('agent:execute')
async productionChatResume(
req: AuthenticatedRequest<{ projectId: string }>,
res: FlushableResponse,
@Param('agentId') agentId: string,
@Body payload: AgentChatResumeDto,
) {
const execution = this.createChatExecution(res);
const { send, onChunk, abortSignal } = execution;
let startedExecutionId: string | undefined;
const onExecutionStarted = (id: string, sessionId: string) => {
startedExecutionId = id;
execution.onExecutionStarted(id, sessionId);
};
try {
if (!(await this.agentsService.isN8nChatPublished(agentId, req.params.projectId))) {
send({
type: 'error',
message: 'This agent is not available in n8n Chat.',
errorCode: 'agent_unavailable',
});
return;
}
abortSignal.throwIfAborted();
const stream = this.agentExecutionOrchestratorService.resumeForChat({
agentId,
projectId: req.params.projectId,
runId: payload.runId,
toolCallId: payload.toolCallId,
resumeData: payload.resumeData,
user: req.user,
usePublishedVersion: true,
integrationType: 'n8n_chat',
source: N8N_CHAT_PRODUCTION_SOURCE,
expectedMemory: { resourceId: productionChatMemoryResourceId(req.user.id) },
onExecutionStarted,
abortSignal,
});
let suspended = false;
let failed = false;
for await (const chunk of stream) {
onChunk(chunk);
if (chunk.type === 'tool-call-suspended') suspended = true;
if (chunk.type === 'error') failed = true;
}
if (!suspended && !failed) {
send({
type: 'done',
...(startedExecutionId ? { executionId: startedExecutionId } : {}),
});
}
} catch (error) {
send({
type: 'error',
message: error instanceof Error ? error.message : 'Resume failed',
...(error instanceof AgentTurnAlreadyRunningError
? { errorCode: 'turn_already_running' }
: {}),
});
} finally {
execution.close();
}
}
@Post('/:agentId/chat', { usesTemplates: true })
@ProjectScope('agent:execute')
async chat(
@ -334,6 +517,102 @@ export class AgentChatController {
return { cancelled };
}
@Delete('/:agentId/n8n-chat/:threadId/executions/:executionId')
@ProjectScope('agent:execute')
async cancelProductionChatExecution(
req: AuthenticatedRequest<{ projectId: string }>,
_res: Response,
@Param('agentId') agentId: string,
@Param('threadId') threadId: string,
@Param('executionId') executionId: string,
) {
await this.requireProductionChat(agentId, req.params.projectId);
await this.requireProductionThread(threadId, req.params.projectId, agentId, req.user.id);
return {
cancelRequested: await this.chatExecutionService.requestCancel({
projectId: req.params.projectId,
agentId,
threadId,
executionId,
userId: req.user.id,
productionN8nChat: true,
}),
};
}
@Delete('/:agentId/n8n-chat/runs/:runId')
@ProjectScope('agent:execute')
async cancelProductionChatRun(
req: AuthenticatedRequest<{ projectId: string }>,
_res: Response,
@Param('agentId') agentId: string,
@Param('runId') runId: string,
) {
await this.requireProductionChat(agentId, req.params.projectId);
return {
cancelled: await this.agentExecutionOrchestratorService.cancelChatRun({
agentId,
runId,
resourceId: productionChatMemoryResourceId(req.user.id),
}),
};
}
@Get('/:agentId/n8n-chat/:threadId/messages')
@ProjectScope('agent:read')
async getProductionChatMessages(
req: AuthenticatedRequest<{ projectId: string; agentId: string; threadId: string }>,
): Promise<AgentChatMessagesResponse> {
const { projectId, agentId, threadId } = req.params;
await this.requireProductionChat(agentId, projectId);
await this.requireProductionThread(threadId, projectId, agentId, req.user.id);
const history = await this.agentExecutionOrchestratorService.getConversationHistory({
threadId,
projectId,
agentId,
userId: req.user.id,
});
if (!history) throw new NotFoundError('Session not found');
const checkpoint = await this.agentsBuilderService.findOpenCheckpointForThread(
agentId,
threadId,
);
return {
...withOpenSuspensions(
history.messages,
checkpoint?.persistence?.resourceId === productionChatMemoryResourceId(req.user.id)
? checkpoint
: null,
{ appendInactiveCheckpointMessages: false },
),
activeExecutionId: history.activeExecutionId,
};
}
@Get('/:agentId/n8n-chat/attachments/:attachmentId')
@ProjectScope('agent:read')
async getProductionChatAttachment(
req: AuthenticatedRequest<{ projectId: string; agentId: string; attachmentId: string }>,
res: Response,
) {
const { projectId, agentId, attachmentId } = req.params;
await this.requireProductionChat(agentId, projectId);
const attachment = await this.agentChatAttachmentService.getForAgent(attachmentId, {
agentId,
projectId,
userId: req.user.id,
});
if (
!attachment ||
attachment.source !== N8N_CHAT_PRODUCTION_SOURCE ||
attachment.resourceId !== productionChatMemoryResourceId(req.user.id)
) {
throw new NotFoundError(`Attachment "${attachmentId}" not found`);
}
await this.requireProductionThread(attachment.threadId, projectId, agentId, req.user.id);
await this.streamAttachment(attachment, res);
}
@Get('/:agentId/chat/:threadId/queue')
@ProjectScope('agent:read')
async getQueuedMessages(
@ -399,6 +678,17 @@ export class AgentChatController {
if (!threadBelongsTo(thread, projectId, agentId, req.user.id)) {
throw new NotFoundError(`Thread "${threadId}" not found`);
}
if (
thread.accessScope === 'user' &&
!(await this.agentExecutionService.canUseDraftThread(
threadId,
projectId,
agentId,
req.user.id,
{ previewChat: true, sessionMode: 'existing' },
))
)
throw new NotFoundError(`Thread "${threadId}" not found`);
const jobs = await this.backgroundJobService.listCurrentGroupForThread(agentId, threadId);
return {
@ -428,6 +718,17 @@ export class AgentChatController {
if (thread && !threadBelongsTo(thread, projectId, agentId, req.user.id)) {
throw new NotFoundError(`Thread "${threadId}" not found`);
}
if (
thread?.accessScope === 'user' &&
!(await this.agentExecutionService.canUseDraftThread(
threadId,
projectId,
agentId,
req.user.id,
{ previewChat: true, sessionMode: 'existing' },
))
)
throw new NotFoundError(`Thread "${threadId}" not found`);
const history = await this.agentExecutionOrchestratorService.getConversationHistory({
threadId,
projectId,
@ -513,7 +814,14 @@ export class AgentChatController {
userId: req.user.id,
});
if (!attachment) throw new NotFoundError(`Attachment "${attachmentId}" not found`);
if (attachment.source === N8N_CHAT_PRODUCTION_SOURCE) {
throw new NotFoundError(`Attachment "${attachmentId}" not found`);
}
await this.streamAttachment(attachment, res);
}
private async streamAttachment(attachment: AgentChatAttachment, res: Response) {
const attachmentId = attachment.id;
// Open the stream before writing headers: bytes can be gone while the row
// remains (out-of-band storage cleanup), and that must surface as a clean
// 404 rather than a half-written response.

View File

@ -323,7 +323,13 @@ export class AgentConfigService {
previousIntegrations,
nextIntegrations,
);
return { nextSchema, nextIntegrations, previousSchema, previousIntegrations, changedParts };
return {
nextSchema,
nextIntegrations,
previousSchema,
previousIntegrations,
changedParts,
};
}
private mergeConfigSchema(

View File

@ -29,8 +29,11 @@ import type { AgentSessionMode } from './utils/agent-thread-access';
import {
draftChatMemoryResourceId,
isTaskRunMemoryResourceId,
productionChatMemoryResourceId,
userIdFromDraftChatMemoryResourceId,
userIdFromProductionChatMemoryResourceId,
} from './utils/agent-memory-scope';
import { N8N_CHAT_PRODUCTION_SOURCE } from './utils/agent-thread-access';
import { AgentRunTracingService, modelIdFromSnapshot } from './agent-run-tracing.service';
import {
AgentRuntimeCacheService,
@ -133,6 +136,13 @@ export interface ExecuteForChatPublishedConfig extends ChatExecutionInput {
// applies regardless.
}
export interface ExecuteForN8nChatPublishedConfig
extends ChatExecutionInput,
ChatExecutionCallbacks {
user: User;
abortSignal?: AbortSignal;
}
export interface ResumeForChatConfig extends ChatExecutionCallbacks {
messageContext?: IntegrationMessageContext | null;
/** Platform conversation metadata scope. Execution memory can belong to a task. */
@ -207,6 +217,7 @@ export interface ExecuteForWakeConfig extends AgentExecutionInput {
export interface StreamChatResponseConfig extends ChatExecutionInput, ChatExecutionCallbacks {
onAdmitted?: () => Promise<void>;
access: AgentThreadAccess;
productionN8nChat?: boolean;
messageContext?: IntegrationMessageContext | null;
agentInstance: RuntimeAgent;
toolRegistry: ToolRegistry;
@ -342,22 +353,34 @@ export class AgentExecutionOrchestratorService {
}
private async resolveResumeAccess(
{ agentId, projectId, runId, user, usePublishedVersion, previewChat }: ResumeChatConfig,
{ agentId, projectId, runId, user, usePublishedVersion, previewChat, source }: ResumeChatConfig,
memoryScope: ResumeCheckpoint['memoryScope'],
): Promise<AgentThreadAccess> {
const isPreview = !usePublishedVersion && !isTaskRunMemoryResourceId(memoryScope.resourceId);
const isProductionChat = source === N8N_CHAT_PRODUCTION_SOURCE;
let access: AgentThreadAccess = { accessScope: 'project', ownerId: null };
if (isPreview) {
if (isPreview || isProductionChat) {
if (
!user ||
memoryScope.resourceId !== draftChatMemoryResourceId(user.id) ||
!(await this.agentExecutionService.canUseDraftThread(
memoryScope.threadId,
projectId,
agentId,
user.id,
{ previewChat, sessionMode: 'existing' },
))
memoryScope.resourceId !==
(isPreview
? draftChatMemoryResourceId(user.id)
: productionChatMemoryResourceId(user.id)) ||
(isPreview
? !(await this.agentExecutionService.canUseDraftThread(
memoryScope.threadId,
projectId,
agentId,
user.id,
{ previewChat, sessionMode: 'existing' },
))
: !(await this.agentExecutionService.canUseProductionChatThread(
memoryScope.threadId,
projectId,
agentId,
user.id,
'existing',
)))
) {
throw new UserError(`Checkpoint ${runId} does not belong to this chat`);
}
@ -366,6 +389,7 @@ export class AgentExecutionOrchestratorService {
const thread = await this.agentExecutionService.findThreadById(memoryScope.threadId);
if (
userIdFromDraftChatMemoryResourceId(memoryScope.resourceId) ||
userIdFromProductionChatMemoryResourceId(memoryScope.resourceId) ||
!thread ||
thread.projectId !== projectId ||
thread.agentId !== agentId ||
@ -378,17 +402,17 @@ export class AgentExecutionOrchestratorService {
}
private validateResumeSandbox(
{ projectId, runId, user, usePublishedVersion }: ResumeChatConfig,
{ projectId, runId, user, usePublishedVersion, source }: ResumeChatConfig,
memoryScope: ResumeCheckpoint['memoryScope'],
): AgentSandboxPrincipalHash | undefined {
const sandboxScope = decodeAgentSandboxHostMetadata(memoryScope.hostMetadata);
const sandboxPrincipalHash = sandboxScope?.principalHash;
if (
this.agentSandboxRuntimeService.isEnabled() &&
(this.agentSandboxRuntimeService.isEnabled() || source === N8N_CHAT_PRODUCTION_SOURCE) &&
(!sandboxScope ||
sandboxScope.projectId !== projectId ||
!sandboxPrincipalHash ||
(!usePublishedVersion &&
((!usePublishedVersion || source === N8N_CHAT_PRODUCTION_SOURCE) &&
(!user ||
sandboxPrincipalHash !==
hashAgentSandboxPrincipal({ type: 'n8n-user', userId: user.id }))))
@ -405,6 +429,12 @@ export class AgentExecutionOrchestratorService {
*/
async *resumeForChat(config: ResumeForChatConfig): AsyncGenerator<StreamChunk> {
const resume = { ...config, usePublishedVersion: config.usePublishedVersion ?? true };
if (
resume.source === N8N_CHAT_PRODUCTION_SOURCE &&
!(await this.agentRepository.isN8nChatPublished(resume.agentId, resume.projectId))
) {
throw new UserError('This agent is not available in n8n Chat');
}
const checkpoint = await this.loadResumeCheckpoint(resume);
yield* this.withRuntimeLease(
async () => await this.getResumeRuntime(resume, checkpoint),
@ -521,6 +551,92 @@ export class AgentExecutionOrchestratorService {
);
}
async *executeForN8nChatPublished(
config: ExecuteForN8nChatPublishedConfig,
): AsyncGenerator<StreamChunk> {
const {
agentId,
projectId,
message,
memory,
attachments,
user,
abortSignal,
sessionMode = 'new',
} = config;
if (!(await this.agentRepository.isN8nChatPublished(agentId, projectId))) {
throw new UserError('This agent is not available in n8n Chat');
}
if (
memory.resourceId !== productionChatMemoryResourceId(user.id) ||
!(await this.agentExecutionService.canUseProductionChatThread(
memory.threadId,
projectId,
agentId,
user.id,
sessionMode,
))
) {
throw new UserError('Session not found');
}
const sandboxPrincipalHash = hashAgentSandboxPrincipal({ type: 'n8n-user', userId: user.id });
await this.externalHooks.run('agent.preExecute', [agentId]);
yield* this.withRuntimeLease(
async () =>
await this.getRuntimeOrRecordFailure(
{
agentId,
projectId,
integrationType: N8N_CHAT_INTEGRATION_TYPE,
usePublishedVersion: true,
sandboxPrincipalHash,
allowBackgroundTasks: false,
attributionUserId: user.id,
},
{
threadId: memory.threadId,
userMessage: message,
attachments,
source: N8N_CHAT_PRODUCTION_SOURCE,
access: { accessScope: 'user', ownerId: user.id },
sessionMode,
abortSignal,
},
),
async (runtime) => {
const messageContext = this.createN8nChatMessageContext(memory, user.id);
return this.streamChatResponse({
access: { accessScope: 'user', ownerId: user.id },
messageContext,
onAdmitted: async () =>
await this.integrationMessageContextService.setLatest(
memory.threadId,
memory.resourceId,
messageContext,
),
agentInstance: runtime.agent,
toolRegistry: runtime.toolRegistry,
mcpServerAttributions: runtime.mcpServerAttributions,
agentId,
projectId,
message,
memory,
attachments,
userId: user.id,
productionN8nChat: true,
source: N8N_CHAT_PRODUCTION_SOURCE,
telemetry: { runType: 'production', configuration: runtime.telemetryConfiguration },
includeHitlToolDetails: true,
sandboxPrincipalHash,
sessionMode,
abortSignal,
onExecutionStarted: config.onExecutionStarted,
onExecutionRecorded: config.onExecutionRecorded,
});
},
);
}
/**
* Execute a published agent for a scheduled task, stamping `source='task'`
* and the originating `taskId` on the recorded session for traceability.
@ -632,9 +748,30 @@ export class AgentExecutionOrchestratorService {
async executeForWake(config: ExecuteForWakeConfig): Promise<void> {
const { agentId, memory, identity, abortSignal } = config;
const isDraft = identity.type === 'draft';
const productionUserId = isDraft
? undefined
: userIdFromProductionChatMemoryResourceId(memory.resourceId);
if (
productionUserId &&
(isDraft ||
identity.integrationType !== N8N_CHAT_INTEGRATION_TYPE ||
identity.principalHash !==
hashAgentSandboxPrincipal({ type: 'n8n-user', userId: productionUserId }) ||
!(await this.agentRepository.isN8nChatPublished(agentId, config.projectId)) ||
!(await this.agentExecutionService.canUseProductionChatThread(
memory.threadId,
config.projectId,
agentId,
productionUserId,
'existing',
)))
)
return;
const access: AgentThreadAccess = isDraft
? { accessScope: 'user', ownerId: identity.user.id }
: { accessScope: 'project', ownerId: null };
: productionUserId
? { accessScope: 'user', ownerId: productionUserId }
: { accessScope: 'project', ownerId: null };
if (isDraft) {
await this.assertDraftChatAccess({ ...config, user: identity.user, sessionMode: 'existing' });
} else {
@ -642,14 +779,22 @@ export class AgentExecutionOrchestratorService {
}
const integrationType = isDraft ? N8N_CHAT_INTEGRATION_TYPE : identity.integrationType;
const messageContext = await this.integrationMessageContextService.getLatest(memory.threadId);
const delivery = isDraft
? undefined
: await this.getWakeDelivery(agentId, integrationType, messageContext);
const delivery =
isDraft || productionUserId
? undefined
: await this.getWakeDelivery(agentId, integrationType, messageContext);
const stream = this.withRuntimeLease(
async () => await this.getWakeRuntime(config, access, integrationType),
async () => await this.getWakeRuntime(config, access, integrationType, productionUserId),
(runtime) =>
this.streamWakeResponse(
this.streamWakeTurn(config, runtime, access, integrationType, messageContext),
this.streamWakeTurn(
config,
runtime,
access,
integrationType,
messageContext,
productionUserId,
),
abortSignal,
delivery,
messageContext?.interactingUserId,
@ -731,6 +876,7 @@ export class AgentExecutionOrchestratorService {
includeHitlToolDetails: config.includeHitlToolDetails,
onExecutionRecorded: config.onExecutionRecorded,
previewChat: config.previewChat,
productionN8nChat: config.source === N8N_CHAT_PRODUCTION_SOURCE,
onExecutionStarted: config.onExecutionStarted,
onSettled: config.isWakeRun
? undefined
@ -817,7 +963,7 @@ export class AgentExecutionOrchestratorService {
agentName: selected.schema?.name ?? agent.name,
projectId,
telemetry: {
userId: params.user?.id,
userId: params.attributionUserId ?? params.user?.id,
runType: params.usePublishedVersion ? 'production' : 'test',
configuration: buildAgentConfigurationTelemetry(selected),
},
@ -852,6 +998,8 @@ export class AgentExecutionOrchestratorService {
usePublishedVersion,
integrationType,
user: usePublishedVersion ? undefined : user,
attributionUserId: source === N8N_CHAT_PRODUCTION_SOURCE ? user?.id : undefined,
allowBackgroundTasks: source === N8N_CHAT_PRODUCTION_SOURCE ? false : undefined,
...(sandboxPrincipalHash ? { sandboxPrincipalHash } : {}),
previewChat,
},
@ -881,8 +1029,10 @@ export class AgentExecutionOrchestratorService {
toolRegistry: runtime.toolRegistry,
mcpServerAttributions: runtime.mcpServerAttributions,
context: { projectId: config.projectId, agentId: config.agentId, threadId },
includeHitlToolDetails: !config.usePublishedVersion,
includeHitlToolDetails:
!config.usePublishedVersion || config.source === N8N_CHAT_PRODUCTION_SOURCE,
previewChat: config.previewChat,
productionN8nChat: config.source === N8N_CHAT_PRODUCTION_SOURCE,
automaticPreviewContinuation: config.automaticPreviewContinuation,
onExecutionStarted: config.onExecutionStarted,
onExecutionRecorded: config.onExecutionRecorded,
@ -1096,7 +1246,7 @@ export class AgentExecutionOrchestratorService {
previewChat,
sessionMode,
} = config;
const messageContext = this.createDraftMessageContext(memory, user.id);
const messageContext = this.createN8nChatMessageContext(memory, user.id);
return this.streamChatResponse({
admittedExecution: config.admittedExecution,
onAdmitted: async () =>
@ -1128,7 +1278,7 @@ export class AgentExecutionOrchestratorService {
});
}
private createDraftMessageContext(
private createN8nChatMessageContext(
memory: AgentMemoryScope,
userId: string,
): IntegrationMessageContext {
@ -1237,6 +1387,7 @@ export class AgentExecutionOrchestratorService {
config: ExecuteForWakeConfig,
access: AgentThreadAccess,
integrationType: string,
productionUserId?: string,
) {
const { agentId, projectId, memory, identity, abortSignal } = config;
const isDraft = identity.type === 'draft';
@ -1246,13 +1397,16 @@ export class AgentExecutionOrchestratorService {
projectId,
integrationType,
usePublishedVersion: !isDraft,
...(productionUserId
? { attributionUserId: productionUserId, allowBackgroundTasks: false }
: {}),
...(isDraft ? { user: identity.user } : {}),
sandboxPrincipalHash: identity.principalHash,
},
{
threadId: memory.threadId,
userMessage: null,
source: integrationType,
source: productionUserId ? N8N_CHAT_PRODUCTION_SOURCE : integrationType,
abortSignal,
access,
sessionMode: 'existing',
@ -1266,6 +1420,7 @@ export class AgentExecutionOrchestratorService {
access: AgentThreadAccess,
integrationType: string,
messageContext: IntegrationMessageContext | null,
productionUserId?: string,
) {
const { agentId, message, memory, identity, abortSignal } = config;
const isDraft = identity.type === 'draft';
@ -1276,17 +1431,21 @@ export class AgentExecutionOrchestratorService {
toolRegistry: runtime.toolRegistry,
mcpServerAttributions: runtime.mcpServerAttributions,
agentId,
...(isDraft ? { userId: identity.user.id } : {}),
...(isDraft
? { userId: identity.user.id }
: productionUserId
? { userId: productionUserId }
: {}),
message,
memory,
projectId: runtime.projectId,
source: integrationType,
source: productionUserId ? N8N_CHAT_PRODUCTION_SOURCE : integrationType,
telemetry: {
runType: isDraft ? 'test' : 'production',
configuration: runtime.telemetryConfiguration,
},
abortSignal,
includeHitlToolDetails: isDraft,
includeHitlToolDetails: isDraft || Boolean(productionUserId),
sandboxPrincipalHash: identity.principalHash,
hideUserMessageFromTranscript: true,
isWakeRun: true,

View File

@ -40,6 +40,7 @@ import { draftChatMemoryResourceId } from './utils/agent-memory-scope';
import {
canContinueThreadInPreview,
canUseTopLevelDraftThread,
N8N_CHAT_PRODUCTION_SOURCE,
threadBelongsTo,
type AgentSessionMode,
} from './utils/agent-thread-access';
@ -732,6 +733,26 @@ export class AgentExecutionService {
return await this.canUseUnrecordedDraftThread(threadId, agentId, userId);
}
async canUseProductionChatThread(
threadId: string,
projectId: string,
agentId: string,
userId: string,
sessionMode: AgentSessionMode,
): Promise<boolean> {
const thread = await this.findThreadById(threadId);
if (!thread) return sessionMode === 'new';
if (
thread.projectId !== projectId ||
thread.agentId !== agentId ||
!canUseTopLevelDraftThread(thread, userId) ||
thread.taskId !== null
)
return false;
const sources = await this.agentExecutionRepository.findFirstSourceByThreadIds([threadId]);
return sources.get(threadId) === N8N_CHAT_PRODUCTION_SOURCE;
}
private async canUseUnrecordedDraftThread(
threadId: string,
agentId: string,

View File

@ -107,9 +107,25 @@ export class AgentIntegrationsController {
const statuses = await this.channelStatusRepository.findByAgentId(agentId);
const now = new Date();
return buildChannelStatusReport(agent.integrations, agent.activeVersionId, statuses, (row) =>
this.statusReporter.isLive(row, now),
);
return {
...buildChannelStatusReport(
agent.integrations,
agent.activeVersionId,
statuses,
(row) => this.statusReporter.isLive(row, now),
agent.activeVersion?.schema?.integrations ?? [],
),
n8nChat: {
draftEnabled:
agent.integrations?.some((integration) => integration.type === 'n8n_chat') ?? false,
publishedEnabled:
agent.activeVersionId !== null &&
(agent.activeVersion?.schema?.integrations?.some(
(integration) => integration.type === 'n8n_chat',
) ??
false),
},
};
}
// Third-party webhook callback: do not add @ProjectScope. Auth happens

View File

@ -1,4 +1,6 @@
import {
N8N_CHAT_INTEGRATION_TYPE,
isCredentialAgentIntegration,
isDraftIntegration,
type AgentConfigValidationResponse,
type AgentJsonConfig,
@ -79,6 +81,13 @@ function requireValidValidation(
}
}
function draftSchemaFromVersion(schema: AgentJsonConfig | null): AgentJsonConfig | null {
if (!schema) return null;
const draft = deepCopy(schema);
delete draft.integrations;
return draft;
}
export interface PublishAgentResult {
agent: Agent;
/**
@ -173,8 +182,8 @@ export class AgentPublishService {
* about to become live, independent of any frontend check. Validating the
* current draft is not enough when a specific historical `versionId` is
* being republished — that snapshot's schema/tool/skill bodies must be
* checked instead. Integrations are never versioned, so the agent's
* *current* integrations are always part of the check.
* checked instead. Credential-backed integrations use the current draft.
* The n8n Chat entry is saved in the published schema.
*/
private async assertPublishable(
agent: Agent,
@ -227,6 +236,7 @@ export class AgentPublishService {
private async assertChannelsStartable(agent: Agent, projectId: string): Promise<void> {
const chatIntegrationService = Container.get(ChatIntegrationService);
for (const integration of agent.integrations ?? []) {
if (!isCredentialAgentIntegration(integration)) continue;
if (isDraftIntegration(integration)) continue;
await chatIntegrationService.assertStartupPreconditions(agent.id, integration, projectId);
}
@ -366,7 +376,7 @@ export class AgentPublishService {
let tasksChanged = false;
await this.agentRepository.manager.transaction(async (trx) => {
agent.schema = activeVersion.schema ? deepCopy(activeVersion.schema) : null;
agent.schema = draftSchemaFromVersion(activeVersion.schema);
agent.tools = deepCopy(activeVersion.tools ?? {});
agent.skills = deepCopy(activeVersion.skills ?? {});
agent.versionId = activeVersion.versionId;
@ -417,7 +427,7 @@ export class AgentPublishService {
throw new NotFoundError(`Version "${versionId}" not found`);
}
agent.schema = target.schema ? deepCopy(target.schema) : null;
agent.schema = draftSchemaFromVersion(target.schema);
agent.tools = deepCopy(target.tools ?? {});
agent.skills = deepCopy(target.skills ?? {});
agent.versionId = uuid();
@ -448,9 +458,8 @@ export class AgentPublishService {
}
/**
* A revert restores a stored schema wholesale, so it is a modification like
* any other config write. Integrations live outside the schema and survive
* the revert untouched, hence the same list on both sides of the diff.
* A revert restores the schema but leaves draft integrations unchanged.
* The published n8n Chat entry is removed from the restored draft schema.
* Sidecar body flags cover tool/skill/task bodies restored outside the schema.
*/
private async recordRevert(
@ -633,7 +642,7 @@ export class AgentPublishService {
private async startPublishedServices(agent: Agent): Promise<void> {
const agentId = agent.id;
const credentialIntegrations = agent.integrations ?? [];
const credentialIntegrations = (agent.integrations ?? []).filter(isCredentialAgentIntegration);
if (credentialIntegrations.length > 0) {
await Container.get(ChatIntegrationService)
.syncToConfig(agent, [], credentialIntegrations)
@ -738,7 +747,14 @@ export class AgentPublishService {
{
versionId,
agentId: agent.id,
schema: agent.schema,
schema: agent.schema
? {
...agent.schema,
integrations: (agent.integrations ?? []).filter(
(integration) => integration.type === N8N_CHAT_INTEGRATION_TYPE,
),
}
: null,
tools: this.customToolsService.snapshotConfiguredTools(agent.schema, agent.tools ?? {}),
skills: this.pickConfiguredSkillBodies(agent.schema, agent.skills ?? {}),
publishedBy: user,

View File

@ -45,6 +45,7 @@ export interface GetRuntimeParams {
sandboxPrincipalHash?: AgentSandboxPrincipalHash;
/** Disable background-job tools and wake hints for task-triggered runtimes. */
allowBackgroundTasks?: boolean;
attributionUserId?: string;
/**
* Build for the in-app preview chat, which gets an extra instruction saying
* the agent cannot change its own setup. It makes the runtime unusable for
@ -132,6 +133,7 @@ export class AgentRuntimeCacheService {
if (params.previewChat) parts.push('preview');
if (params.integrationType) parts.push(params.integrationType);
if (params.allowBackgroundTasks === false) parts.push('no-background-tasks');
if (params.attributionUserId) parts.push(`attribution:${params.attributionUserId}`);
// Per-user runtimes have node/workflow tools filtered by that user's
// access — keying by user id keeps them from colliding with each other
// or with the unscoped (no-user) runtime.
@ -309,6 +311,7 @@ export class AgentRuntimeCacheService {
sandboxPrincipalHash,
allowBackgroundTasks,
previewChat,
attributionUserId,
} = params;
const agentEntity = await getAgentOrThrow(
@ -343,7 +346,7 @@ export class AgentRuntimeCacheService {
undefined,
usePublishedVersion ? 'integrated' : 'manual',
sandboxPrincipalHash,
{ previewChat, allowBackgroundTasks },
{ previewChat, allowBackgroundTasks, attributionUserId },
);
const {
agent: agentInstance,

View File

@ -14,6 +14,7 @@ import {
SUB_AGENT_MAX_CHILDREN_DEFAULT,
SUB_AGENT_TASK_DIFFICULTIES,
buildProxyHeaders,
isCredentialAgentIntegration,
type AgentIntegrationConfig,
type AgentJsonConfig,
type AgentJsonMcpServerConfig,
@ -141,6 +142,7 @@ export interface ReconstructAgentRuntimeParams extends AgentRuntimeAssets {
* integration parents, which keep the project-scoped trust boundary.
*/
user?: User;
attributionUserId?: string;
/** Runtime seams inherited from the delegating parent run (see {@link AgentRuntimeInstrumentation}). */
instrumentation?: AgentRuntimeInstrumentation;
sandboxPrincipalHash?: AgentSandboxPrincipalHash;
@ -192,6 +194,7 @@ interface ToolRunIdentity {
integrationType?: string;
userId?: string;
previewChat?: boolean;
publishedN8nChat?: boolean;
supportsHitl: boolean;
backgroundTasksEnabled: boolean;
}
@ -287,6 +290,7 @@ export class AgentRuntimeReconstructionService {
supportsHitl,
previewChat,
allowBackgroundTasks = true,
attributionUserId,
}: {
/** Pass false when the caller cannot resume a suspended run (workflow executions). */
supportsHitl?: boolean;
@ -294,6 +298,7 @@ export class AgentRuntimeReconstructionService {
previewChat?: boolean;
/** Disable background jobs for task-triggered runtimes. */
allowBackgroundTasks?: boolean;
attributionUserId?: string;
} = {},
): Promise<ReconstructedAgentRuntime & { userToolAccessSnapshot?: UserToolAccessSnapshot }> {
let config = agentEntity.schema;
@ -338,6 +343,7 @@ export class AgentRuntimeReconstructionService {
credentialIntegrations: agentEntity.integrations ?? [],
subAgentDelegation,
user,
attributionUserId,
instrumentation,
sandboxPrincipalHash,
unavailableTools,
@ -612,6 +618,7 @@ export class AgentRuntimeReconstructionService {
memoryOwnerAgentId,
integrationType,
user,
attributionUserId,
previewChat,
supportsHitl,
runtimeProfile,
@ -626,8 +633,12 @@ export class AgentRuntimeReconstructionService {
usePublishedWorkflowVersion: runType === 'production',
agentId: memoryOwnerAgentId,
integrationType,
userId: user?.id,
userId: attributionUserId ?? user?.id,
previewChat,
publishedN8nChat:
runType === 'production' &&
integrationType === N8N_CHAT_INTEGRATION_TYPE &&
attributionUserId !== undefined,
// Sub-agent checkpoints are rejected on resume and inline agents have no
// checkpoint storage, so neither can be woken again.
supportsHitl: canResume,
@ -794,6 +805,8 @@ export class AgentRuntimeReconstructionService {
agentId,
integrationType,
userId,
previewChat,
publishedN8nChat,
supportsHitl,
backgroundTasksEnabled,
} = runIdentity;
@ -812,6 +825,8 @@ export class AgentRuntimeReconstructionService {
agentId,
integrationType,
userId,
previewChat,
publishedN8nChat,
supportsHitl,
backgroundTasksEnabled,
};
@ -922,7 +937,7 @@ export class AgentRuntimeReconstructionService {
const descriptors = this.createIntegrationDescriptors(
agentId,
credentialIntegrations,
credentialIntegrations.filter(isCredentialAgentIntegration),
integrationRegistry,
);

View File

@ -53,6 +53,7 @@ interface ExecuteTurnConfig {
includeHitlToolDetails?: boolean;
backgroundJobSignal?: AgentBackgroundJobSignal;
previewChat?: boolean;
productionN8nChat?: boolean;
automaticPreviewContinuation?: boolean;
onExecutionStarted?: (executionId: string, sessionId: string) => void;
onExecutionRecorded?: (executionId: string) => void;
@ -134,7 +135,7 @@ export class AgentTurnExecutionService {
try {
turn = await config.prepare();
const preparedTurn = turn;
if (config.previewChat) {
if (config.previewChat || config.productionN8nChat) {
previewControl = this.createPreviewExecutionControl(preparedTurn);
preparedTurn.options.abortSignal = previewControl.controller.signal;
}
@ -238,7 +239,7 @@ export class AgentTurnExecutionService {
): Promise<void> {
const finalize = async () =>
await this.finalizeTurn(turn, config, recorder, executionId, state);
if (config.previewChat) {
if (config.previewChat || config.productionN8nChat) {
await this.chatExecutionService.settle(executionId, finalize, state.suspendedRunId);
} else {
await finalize();
@ -433,7 +434,12 @@ export class AgentTurnExecutionService {
}
if (previewControl) {
this.chatExecutionService.register(
{ ...config.context, userId: previewControl.userId, executionId },
{
...config.context,
userId: previewControl.userId,
executionId,
...(config.productionN8nChat ? { productionN8nChat: true } : {}),
},
previewControl.controller,
);
previewControl.detachRequest();

View File

@ -210,9 +210,8 @@ export class AgentValidationService {
/**
* Same as {@link validateAgentConfiguration}, but against a specific
* published history snapshot instead of the live draft. Used before
* re-publishing a previously published version. Integrations are not
* versioned, so the agent's *current* integrations are always validated,
* regardless of which historical schema is checked.
* re-publishing a previously published version. Credential-backed
* integrations use the current draft; n8n Chat needs no credential check.
*/
async validateAgentHistoryConfiguration(
agentId: string,
@ -486,6 +485,7 @@ export class AgentValidationService {
) {
for (let index = 0; index < integrations.length; index++) {
const integration = integrations[index];
if (integration.type === 'n8n_chat') continue;
const path = `integrations.${index}.credentialId`;
const capability: AgentConfigValidationIssue['capability'] = {
kind: 'channel',

View File

@ -11,6 +11,10 @@ import { isTerminalExecutionStatus } from 'n8n-workflow';
import { Publisher } from '@/scaling/pubsub/publisher.service';
import { AgentExecutionUpdateBroadcaster } from './agent-execution-update-broadcaster';
import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service';
import { AgentRepository } from './repositories/agent.repository';
import { productionChatMemoryResourceId } from './utils/agent-memory-scope';
import { N8N_CHAT_PRODUCTION_SOURCE } from './utils/agent-thread-access';
import { AgentTestRunService } from './agent-test-run.service';
import {
AgentBackgroundJobService,
@ -42,6 +46,8 @@ export class AgentWorkflowToolResumeService {
private readonly instanceSettings: InstanceSettings,
private readonly publisher: Publisher,
private readonly backgroundJobService: AgentBackgroundJobService,
private readonly agentExecutionOrchestratorService: AgentExecutionOrchestratorService,
private readonly agentRepository: AgentRepository,
) {
this.logger = this.logger.scoped('agents');
}
@ -140,6 +146,10 @@ export class AgentWorkflowToolResumeService {
/** The tool handler re-reads the execution, so this payload only says why it woke. */
async resume(agentRun: RelatedAgentRun, status: string): Promise<void> {
const resumeData = { type: 'workflow_finished', value: status };
if (agentRun.publishedN8nChat === true) {
await this.resumeInProductionChat(agentRun, resumeData);
return;
}
if (agentRun.integrationType === N8N_CHAT_INTEGRATION_TYPE) {
await this.resumeInPreviewChat(agentRun, resumeData);
@ -191,6 +201,49 @@ export class AgentWorkflowToolResumeService {
);
}
private async resumeInProductionChat(
agentRun: RelatedAgentRun,
resumeData: unknown,
): Promise<void> {
const user = agentRun.userId
? await this.userRepository.findOneBy({ id: agentRun.userId })
: null;
if (
!user ||
!(await this.agentRepository.isN8nChatPublished(agentRun.agentId, agentRun.projectId))
)
return;
let executionId: string | undefined;
const stream = this.agentExecutionOrchestratorService.resumeForChat({
agentId: agentRun.agentId,
projectId: agentRun.projectId,
runId: agentRun.runId,
toolCallId: agentRun.toolCallId,
resumeData,
user,
usePublishedVersion: true,
integrationType: N8N_CHAT_INTEGRATION_TYPE,
source: N8N_CHAT_PRODUCTION_SOURCE,
expectedMemory: {
threadId: agentRun.threadId,
resourceId: productionChatMemoryResourceId(user.id),
},
onExecutionRecorded: (id) => {
executionId = id;
},
});
for await (const _chunk of stream) {
// Persist the streamed turn before notifying the user's session.
}
if (executionId)
this.executionUpdateBroadcaster.notify({
projectId: agentRun.projectId,
agentId: agentRun.agentId,
threadId: agentRun.threadId,
executionId,
});
}
/**
* The preview's SSE stream closed when the run suspended, so there is nothing to
* stream into: draining headlessly is what records the turn, and the push then

View File

@ -180,7 +180,10 @@ export class AgentsService {
schema: AgentJsonConfig,
projectId: string,
user: User,
): Promise<{ schemaConfig: AgentJsonConfig; integrations: AgentIntegrationConfig[] }> {
): Promise<{
schemaConfig: AgentJsonConfig;
integrations: AgentIntegrationConfig[];
}> {
const accessibleCredentialIds = new Set(
(await createAgentCredentialProvider(this.credentialsService, projectId, user).list()).map(
(credential) => credential.id,
@ -195,10 +198,9 @@ export class AgentsService {
// The credential-claim check ignores publish state, so a copy holding the
// source's channel credentialId would block the original from republishing
// or reconnecting (and the reconciler records that 409 on the source's row).
const draftIntegrations = integrations.map((integration) => ({
...integration,
credentialId: '',
}));
const draftIntegrations = integrations.map((integration) =>
integration.type === 'n8n_chat' ? integration : { ...integration, credentialId: '' },
);
return { schemaConfig, integrations: draftIntegrations };
}
@ -217,6 +219,10 @@ export class AgentsService {
return await this.agentRepository.findByIdAndProjectId(agentId, projectId);
}
async isN8nChatPublished(agentId: string, projectId: string): Promise<boolean> {
return await this.agentRepository.isN8nChatPublished(agentId, projectId);
}
/**
* Lightweight capability metadata for the AI Agent node card: the agent's
* model plus per-item labels for channels / tools / skills / tasks. Reads the

View File

@ -28,6 +28,7 @@ import {
AgentJsonConfigSchema,
isDraftAgentConfig,
isDraftIntegration,
isCredentialAgentIntegration,
sanitizeAgentJsonConfig,
tryParseConfigJson,
type AgentJsonConfig,
@ -456,7 +457,10 @@ export class AgentsBuilderToolsService {
listIntegrationCredentialIds: async () => {
const agent = await this.agentsService.findById(agentId, projectId);
return (agent?.integrations ?? [])
.filter((integration) => !isDraftIntegration(integration))
.filter(
(integration) =>
isCredentialAgentIntegration(integration) && !isDraftIntegration(integration),
)
.map((integration) => integration.credentialId);
},
listChatIntegrationTypes: () =>
@ -513,7 +517,10 @@ export class AgentsBuilderToolsService {
listIntegrationCredentialIds: async () => {
const agent = await this.agentsService.findById(agentId, projectId);
return (agent?.integrations ?? [])
.filter((integration) => !isDraftIntegration(integration))
.filter(
(integration) =>
isCredentialAgentIntegration(integration) && !isDraftIntegration(integration),
)
.map((integration) => integration.credentialId);
},
track,

View File

@ -11,6 +11,7 @@ const telegram: AgentIntegrationConfig = {
credentialId: 'cred-telegram',
settings: { accessMode: 'public', allowedUsers: [] },
};
const n8nChat: AgentIntegrationConfig = { type: 'n8n_chat', credentialId: '' };
/** Live unless a test says otherwise — expiry is exercised on its own below. */
const isLive = (row: AgentChannelStatus) =>
@ -52,6 +53,20 @@ function erroredRow(
}
describe('buildChannelStatusReport', () => {
it('uses the published n8n Chat entry while the draft changes', () => {
expect(buildChannelStatusReport([], PUBLISHED, [], isLive, [n8nChat])).toEqual({
status: 'connected',
integrations: [{ type: 'n8n_chat', status: 'connected' }],
});
expect(buildChannelStatusReport([n8nChat], PUBLISHED, [], isLive)).toEqual({
status: 'configured',
integrations: [{ type: 'n8n_chat', status: 'configured' }],
});
expect(buildChannelStatusReport([n8nChat], null, [], isLive, [n8nChat])).toEqual({
status: 'configured',
integrations: [{ type: 'n8n_chat', status: 'configured' }],
});
});
it('reports no channels as disconnected', () => {
expect(buildChannelStatusReport([], PUBLISHED, [], isLive)).toEqual({
status: 'disconnected',

View File

@ -201,6 +201,18 @@ describe('ChatIntegrationService.syncToConfig — publish gate', () => {
broadcastSpy = vi.spyOn(service, 'broadcastIntegrationChange').mockResolvedValue();
});
it('does not start an external adapter for n8n Chat', async () => {
const agent = makeAgent({ activeVersionId: 'published-version-1' });
const n8nChatIntegration = { type: 'n8n_chat', credentialId: '' } as const;
await service.syncToConfig(agent, [], [n8nChatIntegration]);
await service.syncToConfig(agent, [n8nChatIntegration], []);
expect(connectSpy).not.toHaveBeenCalled();
expect(disconnectSpy).not.toHaveBeenCalled();
expect(broadcastSpy).not.toHaveBeenCalled();
});
it('skips connect when the agent is not published', async () => {
const agent = makeAgent({ activeVersionId: null });

View File

@ -1,4 +1,8 @@
import { isDraftIntegration, type AgentIntegrationConfig } from '@n8n/api-types';
import {
isCredentialAgentIntegration,
isDraftIntegration,
type AgentIntegrationConfig,
} from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
import { AgentsConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
@ -349,7 +353,10 @@ export class AgentChannelReconciler {
const wanted: WantedChannels = new Map();
for (const agent of agents) {
const integrations = (agent.integrations ?? []).filter(
(integration) => !isDraftIntegration(integration) && this.runsHere(integration),
(integration) =>
isCredentialAgentIntegration(integration) &&
!isDraftIntegration(integration) &&
this.runsHere(integration),
);
for (const integration of integrations) {
wanted.set(agentChannelKey(agentChannelRef(agent.id, integration)), { agent, integration });

View File

@ -539,7 +539,10 @@ export class AgentChatBridge {
*/
private async resolveActiveThreadId(thread: Thread): Promise<InternalThread> {
const baseId = this.baseThreadId(thread);
const idleTimeoutMinutes = this.integration.settings?.sessionIdleTimeoutMinutes ?? null;
const idleTimeoutMinutes =
'settings' in this.integration
? (this.integration.settings?.sessionIdleTimeoutMinutes ?? null)
: null;
const id = await this.withSessionLock(
baseId,
async () => await this.computeGeneration(baseId, false, idleTimeoutMinutes),

View File

@ -1,4 +1,5 @@
import {
isCredentialAgentIntegration,
isDraftIntegration,
type AgentChannelRuntimeStatus,
type AgentIntegrationConfig,
@ -36,7 +37,8 @@ export function buildChannelStatusReport(
activeVersionId: string | null,
statuses: AgentChannelStatus[],
isLive: IsLiveRow,
): AgentIntegrationStatusResponse {
publishedIntegrations: AgentIntegrationConfig[] = [],
): Omit<AgentIntegrationStatusResponse, 'n8nChat'> {
const liveByChannel = new Map<string, AgentChannelStatus[]>();
for (const row of statuses) {
if (!isLive(row)) continue;
@ -49,6 +51,7 @@ export function buildChannelStatusReport(
// keeps channel-setup UIs from rendering a configured state and hiding their
// own setup form.
const entries: AgentIntegrationStatusEntry[] = (integrations ?? [])
.filter(isCredentialAgentIntegration)
.filter((integration) => !isDraftIntegration(integration))
.map((integration) => {
const rows = liveByChannel.get(channelKey(integration.type, integration.credentialId)) ?? [];
@ -64,6 +67,13 @@ export function buildChannelStatusReport(
...(failure?.errorMessage ? { errorMessage: failure.errorMessage } : {}),
};
});
const draftChat = integrations?.some((integration) => integration.type === 'n8n_chat') ?? false;
const publishedChat =
activeVersionId !== null &&
publishedIntegrations.some((integration) => integration.type === 'n8n_chat');
if (draftChat || publishedChat) {
entries.push({ type: 'n8n_chat', status: publishedChat ? 'connected' : 'configured' });
}
return { status: rollUp(entries), integrations: entries };
}

View File

@ -1,4 +1,8 @@
import { AgentIntegrationConfig, type AgentIntegrationSettings } from '@n8n/api-types';
import {
AgentIntegrationConfig,
isCredentialAgentIntegration,
type AgentIntegrationSettings,
} from '@n8n/api-types';
import { Logger } from '@n8n/backend-common';
import { GlobalConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
@ -206,6 +210,7 @@ export class ChatIntegrationService {
integration: AgentIntegrationConfig,
projectId: string,
): Promise<void> {
if (!isCredentialAgentIntegration(integration)) return;
const implementation = this.integrationRegistry.require(integration.type);
// Deliberately not `validateConfig`: publishing already validates the whole
// configuration, and running it again here would newly reject an agent whose
@ -223,6 +228,7 @@ export class ChatIntegrationService {
integration: AgentIntegrationConfig,
projectId: string,
): Promise<void> {
if (!isCredentialAgentIntegration(integration)) return;
const implementation = this.integrationRegistry.require(integration.type);
implementation.validateConfig?.(integration);
if (!implementation.onBeforeConnect) return;
@ -258,6 +264,7 @@ export class ChatIntegrationService {
projectId: string,
options: ConnectOptions = {},
): Promise<void> {
if (!isCredentialAgentIntegration(integration)) return;
const ingress = options.ingressEnabled ?? true;
if (!this.shouldRouteToLeader(integration.type, ingress)) {
return await this.connectLocal(agentId, integration, projectId, options);
@ -455,6 +462,7 @@ export class ChatIntegrationService {
integration: AgentIntegrationConfig,
options: DisconnectChannelOptions = {},
): Promise<void> {
if (!isCredentialAgentIntegration(integration)) return;
const { deleteSubscriptions = true } = options;
try {
@ -568,16 +576,20 @@ export class ChatIntegrationService {
previous: AgentIntegrationConfig[],
next: AgentIntegrationConfig[],
): Promise<void> {
const previousKeys = new Set(previous.map(buildIntegrationConnectionId));
const nextKeys = new Set(next.map(buildIntegrationConnectionId));
const previousExternal = previous.filter(isCredentialAgentIntegration);
const nextExternal = next.filter(isCredentialAgentIntegration);
const previousKeys = new Set(previousExternal.map(buildIntegrationConnectionId));
const nextKeys = new Set(nextExternal.map(buildIntegrationConnectionId));
for (const integration of previous) {
for (const integration of previousExternal) {
if (!nextKeys.has(buildIntegrationConnectionId(integration))) {
await this.disconnectChannel(agent.id, integration);
}
}
const additions = next.filter((i) => !previousKeys.has(buildIntegrationConnectionId(i)));
const additions = nextExternal.filter(
(integration) => !previousKeys.has(buildIntegrationConnectionId(integration)),
);
if (additions.length > 0 && !agent.activeVersionId) {
this.logger.debug(

View File

@ -105,7 +105,9 @@ export function getIntegrationToolConnectionDescriptors(
actionToolDefinitions,
contextToolGuidance: capabilities?.contextToolGuidance,
actionToolGuidance: capabilities?.actionToolGuidance,
...(integration.approval ? { approval: integration.approval } : {}),
...('approval' in integration && integration.approval
? { approval: integration.approval }
: {}),
};
});
}

View File

@ -82,9 +82,8 @@ export interface IntegrationSubjectPerson {
}
/**
* Source of a tool connection: a persisted credential integration, or the
* implicit credential-less in-app chat channel (injected per-run, never
* stored on the agent).
* Source of a tool connection: a configured integration, or the implicit
* preview chat channel injected per run.
*/
export type IntegrationToolConnectionSource =
| AgentIntegrationConfig

View File

@ -21,10 +21,10 @@ import type { IntegrationActionResult } from '../integration-tools';
/**
* The in-app chat channel ("n8n Chat").
*
* Implicit and credential-less: never persisted in an agent's `integrations`
* array and never shown in the integrations catalog. A synthetic tool
* descriptor is injected per-run when the execution comes through the in-app
* chat endpoints (see agent-runtime-reconstruction.service.ts). `respond`
* Credential-less. The integration implementation is always registered.
* A draft channel entry enables production chat after publish. Preview stays
* available without an entry. A synthetic tool descriptor is injected per run
* when execution comes through the chat endpoints. `respond`
* does not post anywhere — the card renders from the tool-call input in the
* chat UI, and interactive cards suspend through the generic machinery.
*/

View File

@ -25,5 +25,8 @@ export function decomposeJsonConfig(config: AgentJsonConfig): {
integrations: AgentIntegrationConfig[];
} {
const { integrations, ...schemaConfig } = config;
return { schemaConfig, integrations: integrations ?? [] };
return {
schemaConfig,
integrations: integrations ?? [],
};
}

View File

@ -1,3 +1,4 @@
import { N8N_CHAT_INTEGRATION_TYPE } from '@n8n/api-types';
import type { AgentIntegrationConfig, ListAgentsQueryDto } from '@n8n/api-types';
import { Service } from '@n8n/di';
import {
@ -163,6 +164,15 @@ export class AgentRepository extends Repository<Agent> {
});
}
async isN8nChatPublished(id: string, projectId: string): Promise<boolean> {
const agent = await this.findByIdAndProjectId(id, projectId);
return (
agent?.activeVersion?.schema?.integrations?.some(
(integration) => integration.type === N8N_CHAT_INTEGRATION_TYPE,
) ?? false
);
}
/**
* Finds an agent by ID alone. Agent IDs are globally unique, so this is safe
* for callers whose access check does not hinge on a specific project (e.g.

View File

@ -776,6 +776,22 @@ describe('workflow tool → parentAgentRun stamping', () => {
expect(executionData?.parentAgentRun).toEqual(expect.objectContaining({ previewChat: true }));
});
it('stamps the production n8n Chat marker so the wake-up uses the published runtime', async () => {
const executionData = await runToolWith(
{
agentId: 'agent-1',
integrationType: N8N_CHAT_INTEGRATION_TYPE,
userId: 'user-1',
publishedN8nChat: true,
},
agentCtx,
);
expect(executionData?.parentAgentRun).toEqual(
expect.objectContaining({ publishedN8nChat: true, userId: 'user-1' }),
);
});
it('omits the preview marker for every other draft surface', async () => {
const executionData = await runToolWith(
{ agentId: 'agent-1', integrationType: N8N_CHAT_INTEGRATION_TYPE },

View File

@ -167,6 +167,7 @@ export interface WorkflowToolContext {
integrationType?: string;
/** The in-app preview chat started this run — see `RelatedAgentRun.previewChat`. */
previewChat?: boolean;
publishedN8nChat?: boolean;
userId?: string;
/** Whether a suspension can be resumed at all. Defaults to true. */
supportsHitl?: boolean;
@ -643,6 +644,7 @@ function agentRunOf(
...(context.integrationType ? { integrationType: context.integrationType } : {}),
...(context.userId ? { userId: context.userId } : {}),
...(context.previewChat ? { previewChat: true } : {}),
...(context.publishedN8nChat ? { publishedN8nChat: true } : {}),
};
}

View File

@ -1,5 +1,6 @@
const INTEGRATION_MEMORY_RESOURCE_PREFIX = 'integration:';
const DRAFT_CHAT_MEMORY_RESOURCE_PREFIX = 'draft-chat:';
const PRODUCTION_CHAT_MEMORY_RESOURCE_PREFIX = 'n8n-chat-production:';
const THREAD_MEMORY_RESOURCE_PREFIX = 'thread:';
export function threadMemoryResourceId(threadId: string): string {
@ -23,6 +24,16 @@ export function draftChatMemoryResourceId(userId: string): string {
return `${DRAFT_CHAT_MEMORY_RESOURCE_PREFIX}${userId}`;
}
export function productionChatMemoryResourceId(userId: string): string {
return `${PRODUCTION_CHAT_MEMORY_RESOURCE_PREFIX}${userId}`;
}
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);
return userId.length > 0 ? userId : undefined;
}
export function userIdFromDraftChatMemoryResourceId(resourceId: string): string | undefined {
if (!resourceId.startsWith(DRAFT_CHAT_MEMORY_RESOURCE_PREFIX)) return undefined;
const userId = resourceId.slice(DRAFT_CHAT_MEMORY_RESOURCE_PREFIX.length);

View File

@ -3,6 +3,7 @@ import type { AgentExecutionThread } from '../entities/agent-execution-thread.en
export type AgentSessionMode = 'new' | 'existing';
export const PREVIEW_THREAD_SOURCES = ['', 'chat', 'n8n_chat'] as const;
export const N8N_CHAT_PRODUCTION_SOURCE = 'n8n_chat_production';
export function threadBelongsTo(
thread: AgentExecutionThread,

View File

@ -145,6 +145,7 @@ export type PubSubCommandMap = {
threadId: string;
executionId: string;
userId: string;
productionN8nChat?: boolean;
};
'relay-agent-message-queue-update': {

View File

@ -0,0 +1,214 @@
import { randomUUID } from 'node:crypto';
import { linkUserToProject, testModules } from '@n8n/backend-test-utils';
import { ProjectRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
import { AgentHistoryRepository } from '@/modules/agents/repositories/agent-history.repository';
import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository';
import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
import { createMember, createOwner } from '../shared/db/users';
import { setupTestServer } from '../shared/utils';
beforeAll(async () => {
await testModules.loadModules(['agents']);
});
const server = setupTestServer({ endpointGroups: ['ai'] });
const chatIntegration = { type: 'n8n_chat', credentialId: '' } as const;
describe('production n8n Chat HTTP route', () => {
async function createAgent(ownerId: string) {
const project = await Container.get(ProjectRepository).getPersonalProjectForUserOrFail(ownerId);
const repository = Container.get(AgentRepository);
const agent = await repository.save(
repository.create({
id: randomUUID(),
name: 'Agent',
projectId: project.id,
schema: { name: 'Agent', model: 'openai:gpt-4o-mini', instructions: 'Help' },
integrations: [chatIntegration],
tools: {},
skills: {},
versionId: randomUUID(),
}),
);
return { agent, project };
}
async function activateChat(agentId: string, enabled = true) {
const versionId = randomUUID();
await Container.get(AgentHistoryRepository).saveVersion({
versionId,
agentId,
schema: {
name: 'Agent',
model: 'openai:gpt-4o-mini',
instructions: 'Help',
integrations: enabled ? [chatIntegration] : [],
},
tools: {},
skills: {},
publishedBy: 'test',
});
await Container.get(AgentRepository).update({ id: agentId }, { activeVersionId: versionId });
}
async function createThread(agentId: string, projectId: string, ownerId: string, source: string) {
const threadRepository = Container.get(AgentExecutionThreadRepository);
const thread = await threadRepository.save(
threadRepository.create({
id: randomUUID(),
agentId,
agentName: 'Agent',
projectId,
accessScope: 'user',
ownerId,
sessionNumber: 1,
}),
);
const executionRepository = Container.get(AgentExecutionRepository);
const execution = await executionRepository.save(
executionRepository.create({
id: randomUUID(),
threadId: thread.id,
status: 'success',
userMessage: 'Hello',
source,
}),
);
return { thread, execution };
}
it('rejects an unpublished agent before a model request', async () => {
const owner = await createOwner();
const { project, agent } = await createAgent(owner.id);
const response = await server
.authAgentFor(owner)
.post(`/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`)
.send({ message: 'Hello' })
.expect(200);
expect(response.text).toContain('"errorCode":"agent_unavailable"');
});
it('rejects a published version with n8n Chat disabled', async () => {
const owner = await createOwner();
const { project, agent } = await createAgent(owner.id);
await activateChat(agent.id, false);
const response = await server
.authAgentFor(owner)
.post(`/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`)
.send({ message: 'Hello' })
.expect(200);
expect(response.text).toContain('"errorCode":"agent_unavailable"');
});
it('rejects a preview session in the production namespace', async () => {
const owner = await createOwner();
const { project, agent } = await createAgent(owner.id);
await activateChat(agent.id);
const { thread, execution } = await createThread(agent.id, project.id, owner.id, 'chat');
const url = `/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`;
await server.authAgentFor(owner).get(`${url}/${thread.id}/messages`).expect(404);
const response = await server
.authAgentFor(owner)
.post(url)
.send({ message: 'Continue', sessionId: thread.id })
.expect(200);
expect(response.text).toContain('Session not found');
await server
.authAgentFor(owner)
.delete(`${url}/${thread.id}/executions/${execution.id}`)
.expect(404);
});
it('rejects a production session in the preview namespace', async () => {
const owner = await createOwner();
const { project, agent } = await createAgent(owner.id);
await activateChat(agent.id);
const { thread, execution } = await createThread(
agent.id,
project.id,
owner.id,
'n8n_chat_production',
);
const url = `/projects/${project.id}/agents/v2/${agent.id}/chat`;
await server.authAgentFor(owner).get(`${url}/${thread.id}/messages`).expect(404);
const response = await server
.authAgentFor(owner)
.post(url)
.send({ message: 'Continue', sessionId: thread.id })
.expect(200);
expect(response.text).toContain('Session not found');
await server
.authAgentFor(owner)
.delete(`${url}/${thread.id}/executions/${execution.id}`)
.expect(404);
const sessionsUrl = `/projects/${project.id}/agents/v2/${agent.id}/threads`;
const previewSessions = await server
.authAgentFor(owner)
.get(`${sessionsUrl}?origin=preview`)
.expect(200);
expect(previewSessions.body.data.threads).toEqual([]);
const productionSessions = await server
.authAgentFor(owner)
.get(`${sessionsUrl}?origin=n8n_chat_production`)
.expect(200);
expect(productionSessions.body.data.threads).toEqual([
expect.objectContaining({ id: thread.id }),
]);
const detail = await server.authAgentFor(owner).get(`${sessionsUrl}/${thread.id}`).expect(200);
expect(detail.body.data.thread.canContinueInPreview).toBe(false);
});
it('keeps another project member out of a private production session', async () => {
const owner = await createOwner();
const other = await createMember();
const { project, agent } = await createAgent(owner.id);
await linkUserToProject(other, project, 'project:editor');
await activateChat(agent.id);
const { thread, execution } = await createThread(
agent.id,
project.id,
owner.id,
'n8n_chat_production',
);
const url = `/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`;
await server.authAgentFor(other).get(`${url}/${thread.id}/messages`).expect(404);
const response = await server
.authAgentFor(other)
.post(url)
.send({ message: 'Continue', sessionId: thread.id })
.expect(200);
expect(response.text).toContain('Session not found');
await server
.authAgentFor(other)
.delete(`${url}/${thread.id}/executions/${execution.id}`)
.expect(404);
});
it('requires project access before starting production chat', async () => {
const owner = await createOwner();
const other = await createMember();
const { project, agent } = await createAgent(owner.id);
await activateChat(agent.id);
await server
.authAgentFor(other)
.post(`/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`)
.send({ message: 'Hello' })
.expect(403);
});
it('removes production reads when the agent is unpublished', async () => {
const owner = await createOwner();
const { project, agent } = await createAgent(owner.id);
await activateChat(agent.id);
const { thread } = await createThread(agent.id, project.id, owner.id, 'n8n_chat_production');
const url = `/projects/${project.id}/agents/v2/${agent.id}/n8n-chat/${thread.id}/messages`;
await server.authAgentFor(owner).get(url).expect(200);
await Container.get(AgentRepository).update({ id: agent.id }, { activeVersionId: null });
await server.authAgentFor(owner).get(url).expect(404);
});
});

View File

@ -0,0 +1,439 @@
import { randomUUID } from 'node:crypto';
import { mkdirSync } from 'node:fs';
import path from 'node:path';
import { brotliDecompressSync, gunzipSync } from 'node:zlib';
import type { ToolDescriptor } from '@n8n/agents';
import type { AgentJsonConfig } from '@n8n/api-types';
import { linkUserToProject, testModules } from '@n8n/backend-test-utils';
import { ProjectRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import nock, { back as nockBack } from 'nock';
import { BinaryDataConfig, BinaryDataService } from 'n8n-core';
import type { Agent } from '@/modules/agents/entities/agent.entity';
import { AgentHistoryRepository } from '@/modules/agents/repositories/agent-history.repository';
import { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository';
import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
import { saveCredential } from '../shared/db/credentials';
import { createOwner, createMember } from '../shared/db/users';
import { initCredentialsTypes, setupTestServer } from '../shared/utils';
vi.mock('@/utils/ai-proxy-fetch', async (importOriginal) => {
const original = await importOriginal<typeof import('@/utils/ai-proxy-fetch')>();
return {
...original,
createAiProxyFetch: () => async (input: RequestInfo | URL, init?: RequestInit) =>
await globalThis.fetch(input, init),
};
});
const enabled = process.env.AGENT_N8N_CHAT_MODEL_TESTS === '1' || Boolean(process.env.CI);
const record = process.env.VCR_MODE === 'record';
const replay = !record && (process.env.VCR_MODE === 'replay' || Boolean(process.env.CI));
const cassettes = path.join(__dirname, 'cassettes', 'n8n-chat-production');
const chatIntegration = { type: 'n8n_chat', credentialId: '' } as const;
const redPixel =
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR4nGP4z8AAAAMBAQDJ/pLvAAAAAElFTkSuQmCC';
nock.restore();
const directFetch = globalThis.fetch;
function bypassLocalFetchInterceptors() {
const interceptedFetch = globalThis.fetch;
Object.defineProperty(globalThis, 'fetch', {
value: async (input: RequestInfo | URL, init?: RequestInit) =>
await (/localhost|127\.0\.0\.1/.test(input.toString())
? directFetch(input, init)
: interceptedFetch(input, init)),
configurable: true,
});
}
const approvalToolCode = `
import { Tool } from '@n8n/agents';
import { z } from 'zod';
export default new Tool('approve_marker')
.description('Approve a marker before recording it')
.input(z.object({ marker: z.string() }))
.handler(async ({ marker }) => ({ recorded: marker }));
`;
const approvalToolDescriptor: ToolDescriptor = {
name: 'approve_marker',
description: 'Approve a marker before recording it',
systemInstruction: null,
inputSchema: {
type: 'object',
properties: { marker: { type: 'string' } },
required: ['marker'],
},
outputSchema: null,
hasSuspend: false,
hasResume: false,
hasToMessage: false,
requireApproval: false,
providerOptions: null,
};
function events(body: string): Array<Record<string, unknown>> {
return body
.split('\n\n')
.filter((block) => block.startsWith('data: '))
.map((block) => JSON.parse(block.slice(6)) as Record<string, unknown>);
}
function eventOfType(items: Array<Record<string, unknown>>, type: string) {
return items.find((event) => event.type === type);
}
function streamedText(items: Array<Record<string, unknown>>): string {
return items
.filter((event) => event.type === 'text-delta')
.map((event) => (typeof event.delta === 'string' ? event.delta : ''))
.join('');
}
function scrubCassette(definitions: nock.Definition[]): nock.Definition[] {
return definitions
.filter((definition) => !/localhost|127\.0\.0\.1|models\.dev/.test(String(definition.scope)))
.map((definition) => {
const headers = definition.rawHeaders;
if (
headers?.['content-type']?.includes('application/json') &&
(headers['content-encoding'] === 'br' || headers['content-encoding'] === 'gzip') &&
Array.isArray(definition.response) &&
typeof definition.response[0] === 'string'
) {
const bytes = Buffer.from(definition.response[0], 'hex');
definition.response =
headers['content-encoding'] === 'br'
? brotliDecompressSync(bytes).toString('utf8')
: gunzipSync(bytes).toString('utf8');
for (const header of ['content-encoding', 'content-length', 'transfer-encoding']) {
delete headers[header];
}
}
for (const header of ['authorization', 'api-key', 'x-api-key']) {
delete definition.reqheaders?.[header];
}
for (const header of ['set-cookie', 'openai-organization', 'openai-project']) {
delete definition.rawHeaders?.[header];
}
return definition;
});
}
function formatCassette(definitions: nock.Definition[]): string {
return `[
${definitions
.map((definition) => {
const fields = Object.entries(definition)
.map(([key, value]) => `\t\t${JSON.stringify(key)}: ${JSON.stringify(value)}`)
.join(',\n');
return `\t{\n${fields}\n\t}`;
})
.join(',\n')}
]\n`;
}
beforeAll(async () => {
await testModules.loadModules(['agents']);
});
const server = setupTestServer({ endpointGroups: ['ai'] });
describe.skipIf(!enabled)('production n8n Chat with a real model', () => {
let finishCassette: (() => void) | undefined;
let fetchBeforeCassette: typeof globalThis.fetch | undefined;
beforeAll(async () => {
await initCredentialsTypes();
const testFolder = process.env.N8N_USER_FOLDER;
if (!testFolder) throw new Error('N8N_USER_FOLDER is required');
const binaryDataConfig = Container.get(BinaryDataConfig);
binaryDataConfig.mode = 'filesystem';
binaryDataConfig.localStoragePath = path.join(testFolder, '.n8n', 'storage');
await Container.get(BinaryDataService).init();
if (!replay && !process.env.OPENAI_API_KEY) {
throw new Error('Set OPENAI_API_KEY for live or record mode');
}
nockBack.fixtures = cassettes;
if (record) mkdirSync(cassettes, { recursive: true });
nockBack.setMode(record ? 'update' : replay ? 'lockdown' : 'wild');
if (replay) process.env.OPENAI_API_KEY = 'sk-test-replay';
});
beforeEach(async (context) => {
if (!record && !replay) return;
if (!nock.isActive()) nock.activate();
const name = `${context.task.name.replace(/[^a-z0-9]+/gi, '-').toLowerCase()}.json`;
const cassette = await nockBack(name, {
afterRecord: (definitions) => formatCassette(scrubCassette(definitions)),
});
finishCassette = cassette.nockDone;
fetchBeforeCassette = globalThis.fetch;
bypassLocalFetchInterceptors();
nock.enableNetConnect(record ? /127\.0\.0\.1|api\.openai\.com/ : /127\.0\.0\.1/);
nock('https://models.dev')
.get('/api.json')
.reply(200, {
openai: {
id: 'openai',
name: 'OpenAI',
models: {
'gpt-4o-mini': {
id: 'gpt-4o-mini',
name: 'GPT-4o mini',
tool_call: true,
limit: { context: 128000, output: 16384 },
},
},
},
})
.persist();
});
afterEach(() => {
finishCassette?.();
finishCassette = undefined;
if (fetchBeforeCassette) {
Object.defineProperty(globalThis, 'fetch', {
value: fetchBeforeCassette,
configurable: true,
});
fetchBeforeCassette = undefined;
}
nock.cleanAll();
nock.restore();
});
async function createPublishedAgent(options: { approvalTool?: boolean; instructions: string }) {
const owner = await createOwner();
const project = await Container.get(ProjectRepository).getPersonalProjectForUserOrFail(
owner.id,
);
const credential = await saveCredential(
{
name: 'OpenAI',
type: 'openAiApi',
data: { apiKey: process.env.OPENAI_API_KEY ?? 'sk-test-replay' },
},
{ project, role: 'credential:owner' },
);
const schema: AgentJsonConfig = {
name: 'Chat test agent',
model: 'openai/gpt-4o-mini',
credential: credential.id,
instructions: options.instructions,
memory: { enabled: true, storage: 'n8n' },
...(options.approvalTool
? { tools: [{ type: 'custom' as const, id: 'approve_marker', requireApproval: true }] }
: {}),
};
const tools: Agent['tools'] = options.approvalTool
? {
approve_marker: {
code: approvalToolCode,
descriptor: approvalToolDescriptor,
},
}
: {};
const repository = Container.get(AgentRepository);
const agent = await repository.save(
repository.create({
id: randomUUID(),
name: schema.name,
projectId: project.id,
schema,
tools,
skills: {},
integrations: [chatIntegration],
versionId: randomUUID(),
}),
);
const versionId = randomUUID();
await Container.get(AgentHistoryRepository).saveVersion({
versionId,
agentId: agent.id,
schema: { ...schema, integrations: [chatIntegration] },
tools,
skills: {},
publishedBy: owner,
});
await repository.update({ id: agent.id }, { activeVersionId: versionId });
return { owner, project, agent, versionId };
}
it('streams the published instructions and records a private production session', async () => {
const { owner, project, agent } = await createPublishedAgent({
instructions: 'Reply with exactly PUBLISHED_MARKER to greeting requests.',
});
const repository = Container.get(AgentRepository);
if (!agent.schema) throw new Error('Agent schema is missing');
agent.schema = {
...agent.schema,
instructions: 'Reply with exactly DRAFT_MARKER to greeting requests.',
};
expect(await repository.saveDraftFenced(agent)).toBe(true);
const response = await server
.authAgentFor(owner)
.post(`/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`)
.send({ message: 'Give me the greeting marker.' })
.expect(200);
const streamed = events(response.text);
const done = eventOfType(streamed, 'done');
expect(done).toBeDefined();
expect(streamedText(streamed)).toContain('PUBLISHED_MARKER');
expect(streamedText(streamed)).not.toContain('DRAFT_MARKER');
const threadId = done?.sessionId;
if (typeof threadId !== 'string') throw new Error('Session ID is missing');
const thread = await Container.get(AgentExecutionThreadRepository).findOneBy({ id: threadId });
expect(thread).toMatchObject({ ownerId: owner.id, accessScope: 'user' });
const execution = await Container.get(AgentExecutionRepository).findLatestByThreadId(threadId);
expect(execution).toMatchObject({ source: 'n8n_chat_production' });
const preview = await server
.authAgentFor(owner)
.post(`/projects/${project.id}/agents/v2/${agent.id}/chat`)
.send({ message: 'Give me the greeting marker.' })
.expect(200);
const previewEvents = events(preview.text);
const previewDone = eventOfType(previewEvents, 'done');
expect(previewDone).toBeDefined();
expect(streamedText(previewEvents)).toContain('DRAFT_MARKER');
expect(streamedText(previewEvents)).not.toContain('PUBLISHED_MARKER');
if (typeof previewDone?.sessionId !== 'string') {
throw new Error('Preview session ID is missing');
}
const previewExecution = await Container.get(AgentExecutionRepository).findLatestByThreadId(
previewDone.sessionId,
);
// Queued preview turns carry the queue item source, not the production marker.
expect(previewExecution?.source).toBe('chat');
}, 120_000);
it('suspends a tool call and resumes it only for the session owner', async () => {
const { owner, project, agent } = await createPublishedAgent({
approvalTool: true,
instructions:
'When asked to record a marker, call approve_marker with the exact marker. Then confirm the tool result.',
});
const base = `/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`;
const response = await server
.authAgentFor(owner)
.post(base)
.send({ message: 'Record marker A42 using approve_marker.' })
.expect(200);
const streamed = events(response.text);
const suspended = eventOfType(streamed, 'tool-call-suspended');
expect(suspended).toBeDefined();
expect(eventOfType(streamed, 'done')).toBeUndefined();
const payload = suspended?.payload;
if (
!payload ||
typeof payload !== 'object' ||
!('runId' in payload) ||
!('toolCallId' in payload) ||
typeof payload.runId !== 'string' ||
typeof payload.toolCallId !== 'string'
)
throw new Error('Suspended tool call is missing its identifiers');
const started = eventOfType(streamed, 'execution-started');
const threadId = started?.sessionId as string;
const other = await createMember();
await linkUserToProject(other, project, 'project:editor');
await server.authAgentFor(other).get(`${base}/${threadId}/messages`).expect(404);
const denied = await server
.authAgentFor(other)
.post(`${base}/resume`)
.send({
runId: payload.runId,
toolCallId: payload.toolCallId,
resumeData: { approved: true },
})
.expect(200);
expect(denied.text).toMatch(/does not belong|Session not found/);
const history = await server
.authAgentFor(owner)
.get(`${base}/${threadId}/messages`)
.expect(200);
expect(history.body.data.openSuspensions).toHaveLength(1);
const resumed = await server
.authAgentFor(owner)
.post(`${base}/resume`)
.send({
runId: payload.runId,
toolCallId: payload.toolCallId,
resumeData: { approved: true },
})
.expect(200);
expect(eventOfType(events(resumed.text), 'done')).toBeDefined();
}, 120_000);
it('sends an image to the model and reloads the stored attachment', async () => {
const { owner, project, agent } = await createPublishedAgent({
instructions: 'Describe the color of an attached image in one word.',
});
const base = `/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`;
const response = await server
.authAgentFor(owner)
.post(base)
.send({
message: 'What color is this image?',
attachments: [{ fileName: 'pixel.png', mimeType: 'image/png', data: redPixel }],
})
.expect(200);
const streamed = events(response.text);
const done = eventOfType(streamed, 'done');
expect(done, JSON.stringify(streamed)).toBeDefined();
expect(streamedText(streamed)).toMatch(/red/i);
const threadId = done?.sessionId as string;
const history = await server
.authAgentFor(owner)
.get(`${base}/${threadId}/messages`)
.expect(200);
expect(history.body.data.messages).toEqual(expect.any(Array));
const attachment = await Container.get(AgentExecutionRepository).findLatestByThreadId(threadId);
const attachmentId = attachment?.attachments?.[0]?.id;
expect(attachmentId).toBeDefined();
const downloaded = await server
.authAgentFor(owner)
.get(`${base}/attachments/${attachmentId}`)
.expect(200);
expect(downloaded.body).toEqual(Buffer.from(redPixel, 'base64'));
}, 120_000);
it('allows only the session owner to cancel a suspended run', async () => {
const { owner, project, agent } = await createPublishedAgent({
approvalTool: true,
instructions: 'When asked to record a marker, call approve_marker with that marker.',
});
const base = `/projects/${project.id}/agents/v2/${agent.id}/n8n-chat`;
const response = await server
.authAgentFor(owner)
.post(base)
.send({ message: 'Record marker B17 using approve_marker.' })
.expect(200);
const suspension = eventOfType(events(response.text), 'tool-call-suspended');
const payload = suspension?.payload;
if (!payload || typeof payload !== 'object' || !('runId' in payload)) {
throw new Error('Suspended tool call is missing its run ID');
}
if (typeof payload.runId !== 'string') throw new Error('Run ID is invalid');
const other = await createMember();
await linkUserToProject(other, project, 'project:editor');
const foreignCancel = await server
.authAgentFor(other)
.delete(`${base}/runs/${payload.runId}`)
.expect(200);
expect(foreignCancel.body.data.cancelled).toBe(false);
const ownerCancel = await server
.authAgentFor(owner)
.delete(`${base}/runs/${payload.runId}`)
.expect(200);
expect(ownerCancel.body.data.cancelled).toBe(true);
}, 120_000);
});

View File

@ -59,6 +59,48 @@ describe('AgentRepository', () => {
await testDb.terminate();
});
describe('published n8n Chat availability', () => {
it('uses the active version instead of the mutable draft switch', async () => {
const agent = await createAgent();
expect(agent.integrations).toEqual([]);
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(false);
const firstVersion = uuid();
await agentHistoryRepo.saveVersion({
versionId: firstVersion,
agentId: agent.id,
schema: agent.schema
? { ...agent.schema, integrations: [{ type: 'n8n_chat', credentialId: '' }] }
: null,
tools: {},
skills: {},
publishedBy: 'test',
});
await agentRepo.update({ id: agent.id }, { activeVersionId: firstVersion });
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(true);
await agentRepo.update({ id: agent.id }, { integrations: [] });
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(true);
const secondVersion = uuid();
await agentHistoryRepo.saveVersion({
versionId: secondVersion,
agentId: agent.id,
schema: agent.schema,
tools: {},
skills: {},
publishedBy: 'test',
});
await agentRepo.update({ id: agent.id }, { activeVersionId: secondVersion });
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(false);
await agentRepo.update({ id: agent.id }, { activeVersionId: firstVersion });
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(true);
await agentRepo.update({ id: agent.id }, { activeVersionId: null });
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(false);
expect(await agentRepo.isN8nChatPublished(agent.id, uuid())).toBe(false);
await agentRepo.delete({ id: agent.id });
expect(await agentRepo.isN8nChatPublished(agent.id, projectId)).toBe(false);
});
});
describe('saveDraftFenced', () => {
it('persists draft columns and bumps revision when the fence is won', async () => {
const agent = await createAgent();

View File

@ -215,6 +215,7 @@ describe('useAgentApi', () => {
integrations: [
{ type: 'telegram', credentialId: 'cred-telegram-1' },
{ type: 'slack', credentialId: 'cred-slack-1', settings: { channel: 'C1' } },
{ type: 'n8n_chat', credentialId: '' },
],
} as unknown as AgentJsonConfig;
const cloned = { id: 'agent-2', name: 'Support Agent (copy)' } as unknown as AgentResource;
@ -236,6 +237,7 @@ describe('useAgentApi', () => {
expect(postBody.schema.integrations).toEqual([
{ type: 'telegram', credentialId: '' },
{ type: 'slack', credentialId: '', settings: { channel: 'C1' } },
{ type: 'n8n_chat', credentialId: '' },
]);
});
});

View File

@ -146,10 +146,9 @@ export const duplicateAgent = async (
// time (and break the source's channel). Blank to drafts so the builder
// opens the copy with a "connect a channel" chip instead.
const { tasks: _tasks, integrations: sourceIntegrations, ...rest } = configResponse.config;
const draftIntegrations = (sourceIntegrations ?? []).map((integration) => ({
...integration,
credentialId: '',
}));
const draftIntegrations = (sourceIntegrations ?? []).map((integration) =>
integration.type === 'n8n_chat' ? integration : { ...integration, credentialId: '' },
);
return await createAgent(context, projectId, name, {
schema: { ...rest, name, integrations: draftIntegrations },
tools: agent.tools,

View File

@ -52,7 +52,9 @@ function integrationStatusEntriesFromConfig(
type: integration.type,
credentialId: integration.credentialId,
...('settings' in integration ? { settings: integration.settings } : {}),
...(integration.approval ? { approval: integration.approval } : {}),
...('approval' in integration && integration.approval
? { approval: integration.approval }
: {}),
status: isPublished ? 'starting' : 'configured',
});
}

View File

@ -3476,6 +3476,8 @@ export interface RelatedAgentRun {
* resume on the runtime they started on.
*/
previewChat?: boolean;
/** The published n8n Chat channel owns this run. */
publishedN8nChat?: boolean;
/**
* The interactive n8n user, when there is one. The preview chat resumes the draft
* agent version, which gates node and workflow tools by this user's access.