github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Enforce saved Agent plan stop boundaries (no-changelog)

Browse Source
This commit is contained in:
Eugene Molodkin 2026-10-16 14:11:53 +02:00
parent cf4de22c92
commit da6dbc7ed3
No known key found for this signature in database
24 changed files with 1356 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,4 +1,3 @@
import type { AgentMessageSteeringService } from '../agent-message-steering.service';
import type {
Agent as RuntimeAgent,
CredentialProvider,
@ -26,6 +25,8 @@ import { OperationalError, UserError } from 'n8n-workflow';
import type { InstanceSettings } from 'n8n-core';
import type { Mock } from 'vitest';
import { mock } from 'vitest-mock-extended';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import type { AgentMessageSteeringService } from '../agent-message-steering.service';
import type { AgentsSettingsService } from '../agents-settings.service';
import type { ExternalHooks } from '@/external-hooks';
@ -270,12 +271,14 @@ function makeService(sandboxEnabled = false) {
}));
executionService.finalizeExecution.mockResolvedValue('execution-1');
agentRunTracingService.build.mockResolvedValue(undefined);
const cancellations = mock<AgentTaskCancellationRepository>();
const service = new AgentExecutionOrchestratorService(
mockLogger(),
checkpointStorage,
executionService,
new AgentTurnExecutionService(
cancellations,
mockLogger(),
executionService,
chatExecutionService,
@ -300,6 +303,7 @@ function makeService(sandboxEnabled = false) {
return {
service,
cancellations,
settingsService,
backgroundJobRepository,
backgroundJobs,
@ -990,6 +994,26 @@ describe('AgentExecutionOrchestratorService', () => {
expect(sdkStart.mock.calls[0].at(-1)).toMatchObject({ approvalContext });
});
it('marks task cancellation in the main model and tool checks', async () => {
const { stream, sdkStart, cancellations } = makeTurn({ previewChat: true });
await collect(stream);
cancellations.isCancelled.mockResolvedValue(true);
const options = sdkStart.mock.calls[0]?.[operation === 'start' ? 1 : 2] as
| (RunOptions & ExecutionOptions)
| undefined;
const hook = options?.guardrails?.hooks[0];
expect(await hook?.before?.(mock())).toEqual({
action: 'stop',
code: 'tasks-cancelled',
canceled: true,
});
expect(await hook?.beforeTool?.(mock())).toEqual({
action: 'stop',
code: 'tasks-cancelled',
canceled: true,
});
});
it.each([true, false])(
'wires onBudgetNotice into the budget guardrail only when previewChat is %s',
async (previewChat) => {
@ -1004,7 +1028,7 @@ describe('AgentExecutionOrchestratorService', () => {
const options = sdkStart.mock.calls[0]?.[operation === 'start' ? 1 : 2] as
| (RunOptions & ExecutionOptions)
| undefined;
const hook = options?.guardrails?.hooks[0];
const hook = options?.guardrails?.hooks.find((candidate) => candidate.after);
if (!hook?.before || !hook.after) throw new Error('Expected a budget guardrail hook');
const ctx = {
@ -1099,7 +1123,7 @@ describe('AgentExecutionOrchestratorService', () => {
await expect(resumeAgain()).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
await chatExecutionService.handleCancel({
await chatExecutionService.handleForegroundCancel({
projectId,
agentId,
threadId: 'thread-1',
@ -1129,7 +1153,7 @@ describe('AgentExecutionOrchestratorService', () => {
const { stream, onExecutionStarted, sdkStart, executionService, chatExecutionService } =
makeTurn({ previewChat: true });
onExecutionStarted.mockImplementation(() => {
void chatExecutionService.handleCancel({
void chatExecutionService.handleForegroundCancel({
projectId,
agentId,
threadId: 'thread-1',
@ -1147,6 +1171,77 @@ describe('AgentExecutionOrchestratorService', () => {
);
});
it.each(['chunk', 'rejection'] as const)(
'ends a stopped chat without an error when the runtime emits an abort %s',
async (failure) => {
const { stream, sdkStart, chatExecutionService, executionService } = makeTurn({
previewChat: true,
});
const runtimeStream = new TransformStream<StreamChunk, StreamChunk>();
const writer = runtimeStream.writable.getWriter();
sdkStart.mockResolvedValue({ runId: 'runtime-run-1', stream: runtimeStream.readable });
const result = collect(stream);
const settled = expect(result).resolves.not.toContainEqual(
expect.objectContaining({ type: 'error' }),
);
await writer.write({ type: 'text-delta', id: 'text-1', delta: 'Saved partial response.' });
await chatExecutionService.handleForegroundCancel({
projectId,
agentId,
threadId: 'thread-1',
executionId: 'execution-1',
userId,
});
const error = new Error('Agent run was aborted');
if (failure === 'chunk') {
await writer.write({ type: 'error', error });
await writer.write({
type: 'finish',
finishReason: 'error',
usage: { promptTokens: 10, completionTokens: 5, totalTokens: 15 },
});
await writer.close();
} else {
await writer.abort(error);
}
await settled;
expect(executionService.finalizeExecution).toHaveBeenCalledWith(
'execution-1',
expect.objectContaining({
record: expect.objectContaining({
finishReason: 'cancelled',
error: null,
assistantResponse: 'Saved partial response.',
...(failure === 'chunk'
? { usage: { promptTokens: 10, completionTokens: 5, totalTokens: 15 } }
: {}),
}),
}),
);
},
);
it.each([false, true])(
'keeps chat errors visible when execution ownership is lost: %s',
async (ownershipLost) => {
const { stream, sdkStart, executionService } = makeTurn({ previewChat: true });
const controller = new AbortController();
const error = new OperationalError('Execution failed');
executionService.getAbortSignal.mockReturnValue(controller.signal);
sdkStart.mockImplementation(async () => {
if (ownershipLost) controller.abort(error);
return {
runId: 'runtime-run-1',
stream: makeReadableStream([
{ type: 'error', error },
{ type: 'finish', finishReason: 'error' },
]),
};
});
await expect(collect(stream)).resolves.toContainEqual({ type: 'error', error });
},
);
it.each(['startExecutionRecording', 'finalizeExecution'] as const)(
'reports the recording phase when %s fails',
async (recordingOperation) => {
@ -2242,7 +2337,9 @@ describe('AgentExecutionOrchestratorService', () => {
}),
);
expect(runtime.agent.stream.mock.calls[0][1]?.guardrails?.hooks).toHaveLength(1);
const options = runtime.agent.stream.mock.calls[0][1] as ExecutionOptions;
const hooks = options.guardrails?.hooks ?? [];
expect(hooks.filter((hook) => hook.after)).toHaveLength(1);
});
it('rejects a production turn with a foreign thread or memory scope', async () => {
@ -2776,6 +2873,41 @@ describe('AgentExecutionOrchestratorService', () => {
).resolves.toEqual(expect.any(Array));
});
it.each([{ tasks: [] }, backgroundJobSignal])(
'records a stop acknowledgement without a results event for tasks %j',
async ({ tasks }) => {
const { service, runtimeCacheService, executionService } = makeService();
const runtime = makeRuntime([
{ type: 'text-delta', id: 'text-1', delta: 'The plan is stopped.' },
{ type: 'finish', finishReason: 'stop' },
]);
runtimeCacheService.getRuntime.mockResolvedValue(runtime);
await service.executeForWake({
backgroundJobSignal: { tasks },
pauseReport: true,
planStopId: 'plan-stop',
wakeJobIds: ['paused-job'],
agentId,
projectId,
message: 'Acknowledge the stopped plan.',
memory: { threadId: 'thread-1', resourceId: 'draft-chat:user-1' },
identity: { type: 'draft', user, principalHash: userPrincipalHash },
abortSignal: new AbortController().signal,
});
const recording = executionService.startExecutionRecording.mock.calls[0][0];
expect(recording.initialTimeline).toBeUndefined();
expect(recording.wake).toEqual({ jobIds: ['paused-job'], planStopId: 'plan-stop' });
expect(executionService.finalizeExecution).toHaveBeenCalledWith(
'execution-1',
expect.objectContaining({
record: expect.objectContaining({
timeline: [expect.objectContaining({ type: 'text', content: 'The plan is stopped.' })],
}),
}),
);
},
);
it('blocks tools and rejects a pause report interrupted by its guardrail', async () => {
const { service, runtimeCacheService } = makeService();
const runtime = makeRuntime([
@ -2804,11 +2936,10 @@ describe('AgentExecutionOrchestratorService', () => {
});
const options = runtime.agent.stream.mock.calls[0][1] as ExecutionOptions;
expect(options.toolsEnabled).toBe(false);
const hook = options.guardrails?.hooks?.find((candidate) => candidate.beforeTool);
expect(await hook?.beforeTool?.(mock())).toMatchObject({
action: 'stop',
code: 'background-pause-report',
});
const decisions = await Promise.all(
(options.guardrails?.hooks ?? []).map(async (hook) => await hook.beforeTool?.(mock())),
);
expect(decisions).toContainEqual({ action: 'stop', code: 'background-pause-report' });
});
it('records a background continuation while Agents is disabled', async () => {

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

@ -5,6 +5,7 @@ import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import { mock } from 'vitest-mock-extended';
import type { ErrorReporter, StorageConfig } from 'n8n-core';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import type { Telemetry } from '@/telemetry';
import type { AgentChatAttachmentService } from '../agent-chat-attachment.service';
@ -134,6 +135,7 @@ describe('AgentExecutionService', () => {
txRunner.run.mockImplementation(async (ctx, fn) => await fn(ctx));
service = new AgentExecutionService(
mock<AgentTaskCancellationRepository>(),
mockLogger(),
agentExecutionRepository,
agentExecutionThreadRepository,
@ -769,6 +771,7 @@ describe('AgentExecutionService', () => {
it('writes the timeline to blob storage in non-db mode', async () => {
storageConfig = mock<StorageConfig>({ modeTag: 'fs' });
service = new AgentExecutionService(
mock<AgentTaskCancellationRepository>(),
mockLogger(),
agentExecutionRepository,
agentExecutionThreadRepository,
@ -866,6 +869,7 @@ describe('AgentExecutionService', () => {
async (failure, currentLocation, shouldDeleteBlob) => {
storageConfig = mock<StorageConfig>({ modeTag: 'fs' });
service = new AgentExecutionService(
mock<AgentTaskCancellationRepository>(),
mockLogger(),
agentExecutionRepository,
agentExecutionThreadRepository,
@ -1480,6 +1484,7 @@ describe('AgentExecutionService', () => {
it('preserves an interrupted execution inline without overwriting blob storage', async () => {
storageConfig = mock<StorageConfig>({ modeTag: 'fs' });
service = new AgentExecutionService(
mock<AgentTaskCancellationRepository>(),
mockLogger(),
agentExecutionRepository,
agentExecutionThreadRepository,

2
packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts
View File

@ -19,6 +19,7 @@ import type { AgentExecutionThreadRepository } from '../repositories/agent-execu
import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository';
import type { AgentMessageRepository } from '../repositories/agent-message.repository';
import type { AgentRepository } from '../repositories/agent.repository';
import type { AgentTaskCancellationRepository } from '../repositories/agent-task-cancellation.repository';
function makeThread(overrides: Partial<AgentExecutionThread> = {}): AgentExecutionThread {
return mock<AgentExecutionThread>({
@ -56,6 +57,7 @@ describe('AgentMessageQueueService', () => {
new Map([['thread-1', N8N_CHAT_PRODUCTION_SOURCE]]),
);
service = new AgentMessageQueueService(
mock<AgentTaskCancellationRepository>(),
txRunner,
repository,
threadRepository,

2
packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts
View File

@ -3,6 +3,7 @@ import {
type EventService,
type ProjectScopeService,
} from '@n8n/backend-services';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import type { AgentMessageSteeringService } from '../agent-message-steering.service';
import type { Mocked } from 'vitest';
import { type AgentJsonConfig } from '@n8n/api-types';
@ -305,6 +306,7 @@ describe('AgentRuntimeReconstructionService integration tools', () => {
n8nCheckpointStorage,
agentExecutionService,
new AgentTurnExecutionService(
mock<AgentTaskCancellationRepository>(),
logger,
agentExecutionService,
mock<AgentChatExecutionService>(),

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

@ -7,10 +7,12 @@ import { AgentMessageQueueService } from '../agent-message-queue.service';
import { AgentMessageSteeringService } from '../agent-message-steering.service';
import { AgentToolApprovalService } from '../agent-tool-approval.service';
import { AgentTurnExecutionService, type AgentTurnRequest } from '../agent-turn-execution.service';
import { AgentTaskCancellationRepository } from '../repositories/agent-task-cancellation.repository';
it('ends a preview turn stopped during preparation without recording an error', async () => {
const executions = mock<AgentExecutionService>();
const service = new AgentTurnExecutionService(
mock<AgentTaskCancellationRepository>(),
mock<Logger>(),
executions,
mock<AgentChatExecutionService>(),

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

@ -1,4 +1,3 @@
import type { AgentMessageSteeringService } from '../agent-message-steering.service';
import type { Agent as RuntimeAgent, ExecutionOptions, RunOptions, StreamChunk } from '@n8n/agents';
import type { AgentJsonConfig } from '@n8n/api-types';
import { mockInstance, mockLogger } from '@n8n/backend-test-utils';
@ -10,6 +9,8 @@ import { OperationalError, UserError } from 'n8n-workflow';
import type { ExecuteAgentWorkflowContext, IRunExecutionData } from 'n8n-workflow';
import type { Mock } from 'vitest';
import { mock } from 'vitest-mock-extended';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import type { AgentMessageSteeringService } from '../agent-message-steering.service';
import type { AgentsSettingsService } from '../agents-settings.service';
import { SkillService } from '../skills/skill.service';
@ -165,6 +166,7 @@ function makeService() {
logger,
agentRepository,
new AgentTurnExecutionService(
mock<AgentTaskCancellationRepository>(),
mockLogger(),
executionService,
mock<AgentChatExecutionService>(),

10
packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts
View File

@ -214,6 +214,8 @@ export interface ExecuteForTaskNowConfig extends AgentExecutionInput {
}
export interface ExecuteForWakeConfig extends AgentExecutionInput {
wakeJobIds?: string[];
planStopId?: string;
backgroundJobSignal: AgentBackgroundJobSignal;
pauseReport?: boolean;
abortSignal: AbortSignal;
@ -227,6 +229,7 @@ export interface ExecuteForWakeConfig extends AgentExecutionInput {
}
export interface StreamChatResponseConfig extends ChatExecutionInput, ChatExecutionCallbacks {
wake?: { jobIds: string[]; planStopId?: string };
onAdmitted?: () => Promise<void>;
access: AgentThreadAccess;
messageContext?: IntegrationMessageContext | null;
@ -1118,6 +1121,7 @@ export class AgentExecutionOrchestratorService {
threadId: config.memory.threadId,
},
backgroundJobSignal: config.backgroundJobSignal,
wake: config.wake,
onExecutionRecorded: config.onExecutionRecorded,
chatSurface: config.chatSurface,
isWakeRun: config.isWakeRun,
@ -1833,9 +1837,13 @@ export class AgentExecutionOrchestratorService {
sandboxPrincipalHash: identity.principalHash,
hideUserMessageFromTranscript: true,
isWakeRun: true,
wake: {
jobIds: config.wakeJobIds ?? config.backgroundJobSignal.tasks.map((job) => job.id),
...(config.planStopId ? { planStopId: config.planStopId } : {}),
},
pauseReport: config.pauseReport,
sessionMode: 'existing',
backgroundJobSignal: config.backgroundJobSignal,
backgroundJobSignal: config.pauseReport ? undefined : config.backgroundJobSignal,
});
}
}

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

@ -16,6 +16,7 @@ import { ErrorReporter, StorageConfig } from 'n8n-core';
import { OperationalError, UnexpectedError } from 'n8n-workflow';
import { ConflictError } from '@n8n/errors';
import { AgentTaskCancellationRepository } from './repositories/agent-task-cancellation.repository';
import type { AgentRunTelemetryType, IAgentConfigurationTelemetryProperties } from '@/interfaces';
import { Telemetry } from '@/telemetry';
@ -100,6 +101,7 @@ export interface RecordMessageParams {
}
export interface StartExecutionParams extends Omit<RecordMessageParams, 'record' | 'hitlStatus'> {
wake?: { jobIds: string[]; planStopId?: string };
resourceId: string;
messageOrigin?: Omit<AgentMessageOrigin, 'source' | 'hidden'>;
hideUserMessageFromTranscript?: boolean;
@ -182,6 +184,7 @@ export class AgentExecutionService {
private readonly sideCallUsageInFlightByExecution = new Map<string, Set<Promise<void>>>();
constructor(
private readonly cancellations: AgentTaskCancellationRepository,
private readonly logger: Logger,
private readonly agentExecutionRepository: AgentExecutionRepository,
private readonly agentExecutionThreadRepository: AgentExecutionThreadRepository,
@ -224,6 +227,8 @@ export class AgentExecutionService {
const prepared = lockedQueueThread
? { thread: lockedQueueThread, created: false }
: await this.prepareThread(params, ctx);
if (params.wake)
await this.cancellations.assertWakeAdmission(params.threadId, params.wake, ctx);
const { queueItem, predecessorId } = await this.checkAdmission(params, ctx);
const execution = this.agentExecutionRepository.create({
threadId: params.threadId,
@ -249,6 +254,19 @@ export class AgentExecutionService {
attachments: null,
});
const inserted = await this.agentExecutionRepository.saveInContext(execution, ctx);
await this.cancellations.recordAdmission(
params.threadId,
inserted.id,
{
planStopId: params.wake?.planStopId,
userInitiated:
!params.wake &&
!params.hideUserMessageFromTranscript &&
params.userMessage !== null &&
!params.resumeRunId,
},
ctx,
);
const inputMessageIds = await this.reserveInput(
params,
inserted.id,
@ -348,6 +366,8 @@ export class AgentExecutionService {
) {
throw new AgentTurnAlreadyRunningError();
}
// A validated stop report must finish before later user input can run.
if (params.wake?.planStopId) return { queueItem: null, predecessorId: undefined };
const head = await this.queueRepository.findHead(threadId, ctx);
if (head?.id !== queueItemId && (head || queueItemId)) throw new AgentTurnAlreadyRunningError();
return { queueItem: head, predecessorId: undefined };

3
packages/cli/src/modules/agents/agent-message-queue.service.ts
View File

@ -6,6 +6,7 @@ import { OperationalError, UnexpectedError, UserError } from 'n8n-workflow';
import { ConflictError, BadRequestError, NotFoundError } from '@n8n/errors';
import { AgentTaskCancellationRepository } from './repositories/agent-task-cancellation.repository';
import { AgentChatAttachmentService } from './agent-chat-attachment.service';
import { AgentMessageSteeringService } from './agent-message-steering.service';
import { AgentsSettingsService } from './agents-settings.service';
@ -64,6 +65,7 @@ export class AgentMessageQueueService {
onAvailable?: (threadId: string) => void;
constructor(
private readonly cancellations: AgentTaskCancellationRepository,
private readonly txRunner: TransactionRunner,
private readonly repository: AgentMessageQueueRepository,
private readonly threadRepository: AgentExecutionThreadRepository,
@ -358,6 +360,7 @@ export class AgentMessageQueueService {
if (!thread) return null;
steeringChanged = await this.steering.releaseInactive(thread, ctx);
if (await this.isBlocked(thread, ctx)) return null;
if (await this.cancellations.blocksQueue(threadId, ctx)) return null;
const active = await this.repository.findActive(threadId, ctx);
if (active?.executionId)
await this.repository.removeActive(threadId, active.executionId, ctx);

17
packages/cli/src/modules/agents/agent-turn-execution.service.ts
View File

@ -1,6 +1,7 @@
import type {
Agent as RuntimeAgent,
ExecutionOptions,
GuardrailDecision,
ResumeOptions,
RunOptions,
SideCallUsageReport,
@ -11,6 +12,7 @@ import { Logger } from '@n8n/backend-common';
import { Service } from '@n8n/di';
import { UnexpectedError } from 'n8n-workflow';
import { AgentTaskCancellationRepository } from './repositories/agent-task-cancellation.repository';
import type { AgentChatSurface, AgentSessionMode } from './utils/agent-thread-access';
import { AgentExecutionRecordingError } from './agent-execution-recording.error';
import { AgentTurnAlreadyRunningError } from './agent-turn-already-running.error';
@ -48,6 +50,7 @@ export type AgentTurnRequest = { recording: StartExecutionParams } & (
);
interface ExecuteTurnConfig {
wake?: StartExecutionParams['wake'];
admittedExecution?: AgentExecutionAdmission;
onAdmitted?: () => Promise<void>;
agentInstance: RuntimeAgent;
@ -99,6 +102,7 @@ function getMaxIterationsChunks(): StreamChunk[] {
@Service()
export class AgentTurnExecutionService {
constructor(
private readonly cancellations: AgentTaskCancellationRepository,
private readonly logger: Logger,
private readonly agentExecutionService: AgentExecutionService,
private readonly chatExecutionService: AgentChatExecutionService,
@ -460,6 +464,7 @@ export class AgentTurnExecutionService {
const admission = await this.startExecution(
{
...turn.recording,
wake: config.wake,
previewChat: config.chatSurface === 'preview' ? true : undefined,
acceptsSteering: chatControl !== undefined,
resumeRunId: turn.type === 'resume' ? turn.options.runId : undefined,
@ -484,6 +489,18 @@ export class AgentTurnExecutionService {
chatControl?: ChatExecutionControl,
): Promise<ReadableStream<StreamChunk>> {
const { executionId, inputMessageIds } = admission;
const checkCancellation = async (): Promise<GuardrailDecision | undefined> =>
(await this.cancellations.isCancelled(config.context.threadId, executionId))
? { action: 'stop', code: 'tasks-cancelled', canceled: true }
: undefined;
turn.options.guardrails = {
...turn.options.guardrails,
hooks: [
{ before: checkCancellation, beforeTool: checkCancellation },
...(turn.options.guardrails?.hooks ?? []),
],
};
if (turn.type === 'start') turn.input = bindExecutionInput(turn.input, inputMessageIds);
const executionSignal = this.agentExecutionService.getAbortSignal(executionId);
turn.options.abortSignal = turn.options.abortSignal

11
packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts
View File

@ -572,6 +572,17 @@ describe('Agent plan tools', () => {
});
});
it.each(['execution-1', 42, undefined])(
'passes only string execution metadata to plan writes: %s',
async (id) => {
await call('update_plan', { ...write, document: { description: 'New detail' } }, {
persistence: { ...context.persistence!, hostMetadata: { n8nExecutionId: id } },
} as ToolContext);
expect(service.replacePlan).toHaveBeenCalledWith(expect.anything(), {
sourceExecutionId: typeof id === 'string' ? id : undefined,
});
},
);
it('uses the last task end time when all leaves finish before the plan closes', () => {
const first = {
...original.data.items[0],

11
packages/cli/src/modules/agents/plans/agent-plan-tools.ts
View File

@ -1,7 +1,8 @@
import type { BuiltTool } from '@n8n/agents';
import { Tool } from '@n8n/agents/tool';
import { planLeaves } from './agent-plan-work-state';
import type { OperationContext } from '@n8n/db';
import { executionIdFromMetadata } from '../types/agent-queued-message';
import type { AgentPlanOperationContext } from '../repositories/agent-plan.repository';
import { randomUUID } from 'node:crypto';
import { z } from 'zod';
@ -251,7 +252,7 @@ function scopedTool<Schema extends z.ZodType>(
handler: (
input: z.output<Schema>,
threadId: string,
ctx: OperationContext,
ctx: AgentPlanOperationContext,
) => Promise<AgentPlanSnapshot | null>,
) {
return new Tool(name)
@ -267,7 +268,11 @@ function scopedTool<Schema extends z.ZodType>(
}
try {
return presentPlan(await handler(schema.parse(input), threadId, {}));
return presentPlan(
await handler(schema.parse(input), threadId, {
sourceExecutionId: executionIdFromMetadata(context.persistence?.hostMetadata),
}),
);
} catch (error) {
if (error instanceof AgentPlanWriteConflictError) {
return {

42
packages/cli/src/modules/agents/repositories/agent-background-job.repository.ts
View File

@ -8,11 +8,13 @@ import {
type AgentBackgroundJobKind,
type AgentBackgroundJobStatus,
} from '../entities/agent-background-job.entity';
import { AgentTaskCancellationRepository } from './agent-task-cancellation.repository';
import { AgentCheckpoint } from '../entities/agent-checkpoint.entity';
import { AgentExecution } from '../entities/agent-execution.entity';
import { AgentExecutionThreadRepository } from './agent-execution-thread.repository';
type NewAgentBackgroundJobBase = {
sourceExecutionId?: string;
id: string;
parentAgentId: string;
parentThreadId: string;
@ -48,7 +50,16 @@ export type ExpectedBackgroundJobState = {
export type BackgroundJobGroupItem = Pick<
AgentBackgroundJob,
'id' | 'kind' | 'title' | 'status' | 'createdAt' | 'settledAt' | 'notifiedAt' | 'pauseRequestId'
| 'id'
| 'kind'
| 'title'
| 'status'
| 'createdAt'
| 'settledAt'
| 'notifiedAt'
| 'pauseRequestId'
| 'planId'
| 'planTaskId'
>;
@Service()
@ -57,12 +68,15 @@ export class AgentBackgroundJobRepository extends BaseRepository<AgentBackground
dataSource: DataSource,
transactionRunner: TransactionRunner,
private readonly threadRepository: AgentExecutionThreadRepository,
private readonly cancellations: AgentTaskCancellationRepository,
) {
super(AgentBackgroundJob, dataSource.manager, transactionRunner);
}
async insertSubAgentJobIfCapacity(job: NewSubAgentJob, limit: number): Promise<boolean> {
return await this.runInTransaction({}, async (manager, ctx) => {
await this.cancellations.lockScope(job.parentThreadId, ctx);
await this.cancellations.assertAdmission(job.parentThreadId, job.sourceExecutionId, ctx);
if (!(await this.threadRepository.lockById(job.parentThreadId, ctx))) {
throw new UserError('Session not found');
}
@ -80,12 +94,17 @@ export class AgentBackgroundJobRepository extends BaseRepository<AgentBackground
async insertWorkflowJobOrGetExisting(
job: NewWorkflowJob,
): Promise<{ inserted: true } | { inserted: false; existing: AgentBackgroundJob }> {
await this.createQueryBuilder()
.insert()
.into(AgentBackgroundJob)
.values({ ...job, status: 'running' })
.orIgnore()
.execute();
await this.runInTransaction({}, async (manager, ctx) => {
await this.cancellations.lockScope(job.parentThreadId, ctx);
await manager
.createQueryBuilder()
.insert()
.into(AgentBackgroundJob)
.values({ ...job, status: 'running' })
.orIgnore()
.execute();
// The service stops late workflow receipts before returning them to the caller.
});
const inserted = await this.existsBy({ id: job.id });
if (inserted) return { inserted: true };
@ -128,6 +147,8 @@ export class AgentBackgroundJobRepository extends BaseRepository<AgentBackground
'settledAt',
'notifiedAt',
'pauseRequestId',
'planId',
'planTaskId',
],
order: { createdAt: 'ASC' },
});
@ -507,7 +528,7 @@ export class AgentBackgroundJobRepository extends BaseRepository<AgentBackground
},
{
status: settlement.status,
result: settlement.result ?? null,
result: settlement.result ?? (settlement.status === 'cancelled' ? () => '"result"' : null),
error: settlement.error ?? null,
settledAt: new Date(),
notifiedAt:
@ -520,6 +541,11 @@ export class AgentBackgroundJobRepository extends BaseRepository<AgentBackground
return result.affected === 1;
}
async preserveLateCompletedResult(id: string, result: string): Promise<void> {
// A confirmed output remains useful even when cancellation won the terminal-state race.
await this.update({ id, status: 'cancelled', result: IsNull() }, { result });
}
/** Delete settled jobs older than the cutoff only if their results are marked as delivered. */
async deleteSettledBefore(cutoff: Date): Promise<void> {
await this.delete({

23
packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts
View File

@ -3,6 +3,8 @@ import { Service } from '@n8n/di';
import { DataSource, In, IsNull, Not } from '@n8n/typeorm';
import { isDraftIntegration } from '@n8n/api-types';
import { readInboundUserMessage } from '../utils/inbound-attachments';
import { AgentMessageEntity } from '../entities/agent-message.entity';
import { AgentMessageQueue } from '../entities/agent-message-queue.entity';
import { Agent } from '../entities/agent.entity';
import type { AgentQueueDispatch, QueuedUserChatMessage } from '../types/agent-queued-message';
@ -109,6 +111,27 @@ export class AgentMessageQueueRepository extends BaseRepository<AgentMessageQueu
return result.affected === 1;
}
async discardPending(threadId: string, ctx: OperationContext) {
const manager = this.managerFor(ctx);
const items = await this.listPending(threadId, ctx);
if (!items.length) return [];
const attachmentIds = items.flatMap((item) =>
readInboundUserMessage(item.message.content).attachments.map((attachment) => attachment.id),
);
await manager.delete(AgentMessageQueue, { threadId, executionId: IsNull() });
await manager.update(
AgentMessageEntity,
{ id: In(items.map((item) => item.messageId)) },
{
content: { role: 'user', content: [] },
author: null,
modelContent: null,
modelContextAt: null,
},
);
return attachmentIds;
}
async findHead(threadId: string, ctx: OperationContext) {
return await this.managerFor(ctx).findOne(AgentMessageQueue, {
where: { threadId },

43
packages/cli/src/modules/agents/repositories/agent-plan.repository.ts
View File

@ -12,6 +12,7 @@ import { validate as isUuid } from 'uuid';
import { AgentPlanHistory } from '../entities/agent-plan-history.entity';
import { AgentPlan } from '../entities/agent-plan.entity';
import { AgentTaskCancellationRepository } from './agent-task-cancellation.repository';
export type AgentPlanRecord = Pick<
AgentPlan,
@ -25,6 +26,8 @@ export type AgentPlanRevision = Pick<
export type AgentPlanRevisionMetadata = Omit<AgentPlanRevision, 'planId' | 'data'>;
export type AgentPlanOperationContext = OperationContext & { sourceExecutionId?: string };
type PlanWrite = { threadId: string; planId: string; expectedRevision: number };
type PlanDocument = { formatVersion: number; data: JsonObject };
@ -36,19 +39,25 @@ export class AgentPlanWriteConflictError extends UserError {
@Service()
export class AgentPlanRepository extends BaseRepository<AgentPlan> {
constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
constructor(
dataSource: DataSource,
transactionRunner: TransactionRunner,
private readonly cancellations: AgentTaskCancellationRepository,
) {
super(AgentPlan, dataSource.manager, transactionRunner);
}
async createActivePlan(
input: PlanDocument & { id: string; threadId: string },
ctx: OperationContext,
ctx: AgentPlanOperationContext,
): Promise<AgentPlanRecord> {
if (!isUuid(input.id)) throw new UserError('The plan ID must be a UUID');
this.validateVersion(input.formatVersion);
const data = this.serializeData(input.data);
return await this.runInTransaction(ctx, async (manager) => {
return await this.runInTransaction(ctx, async (manager, tx) => {
await this.cancellations.lockScope(input.threadId, tx);
await this.cancellations.assertAdmission(input.threadId, ctx.sourceExecutionId, tx);
const now = new Date();
try {
await manager
@ -82,14 +91,24 @@ export class AgentPlanRepository extends BaseRepository<AgentPlan> {
});
}
async findActivePlan(threadId: string, ctx: OperationContext): Promise<AgentPlanRecord | null> {
async findLatestPlan(threadId: string, ctx: AgentPlanOperationContext) {
return await this.managerFor(ctx).findOne(AgentPlan, {
where: { threadId },
order: { createdAt: 'DESC' },
});
}
async findActivePlan(
threadId: string,
ctx: AgentPlanOperationContext,
): Promise<AgentPlanRecord | null> {
return await this.managerFor(ctx).findOneBy(AgentPlan, { threadId, closedAt: IsNull() });
}
async findPlan(
threadId: string,
planId: string,
ctx: OperationContext,
ctx: AgentPlanOperationContext,
): Promise<AgentPlanRecord | null> {
return await this.managerFor(ctx).findOneBy(AgentPlan, { id: planId, threadId });
}
@ -98,7 +117,7 @@ export class AgentPlanRepository extends BaseRepository<AgentPlan> {
threadId: string,
planId: string,
revision: number,
ctx: OperationContext,
ctx: AgentPlanOperationContext,
): Promise<AgentPlanRevision | null> {
return await this.managerFor(ctx).findOne(AgentPlanHistory, {
where: { planId, revision, plan: { threadId } },
@ -109,7 +128,7 @@ export class AgentPlanRepository extends BaseRepository<AgentPlan> {
threadId: string,
planId: string,
options: { afterRevision?: number; limit?: number },
ctx: OperationContext,
ctx: AgentPlanOperationContext,
): Promise<{ items: AgentPlanRevisionMetadata[]; nextCursor: number | null }> {
const limit = options.limit ?? 50;
const afterRevision = options.afterRevision ?? 0;
@ -149,24 +168,26 @@ export class AgentPlanRepository extends BaseRepository<AgentPlan> {
async replacePlan(
input: PlanWrite & PlanDocument,
ctx: OperationContext,
ctx: AgentPlanOperationContext,
): Promise<AgentPlanRecord> {
this.validateVersion(input.formatVersion);
return await this.writeRevision(input, { ...input, data: this.serializeData(input.data) }, ctx);
}
async closePlan(input: PlanWrite, ctx: OperationContext): Promise<AgentPlanRecord> {
async closePlan(input: PlanWrite, ctx: AgentPlanOperationContext): Promise<AgentPlanRecord> {
return await this.writeRevision(input, null, ctx);
}
private async writeRevision(
input: PlanWrite,
document: { formatVersion: number; data: string } | null,
ctx: OperationContext,
ctx: AgentPlanOperationContext,
): Promise<AgentPlanRecord> {
this.validateVersion(input.expectedRevision);
return await this.runInTransaction(ctx, async (manager) => {
return await this.runInTransaction(ctx, async (manager, tx) => {
await this.cancellations.lockScope(input.threadId, tx);
await this.cancellations.assertAdmission(input.threadId, ctx.sourceExecutionId, tx);
const now = new Date();
const query = manager
.createQueryBuilder()

346
packages/cli/src/modules/agents/repositories/agent-task-cancellation.repository.ts Normal file
View File

@ -0,0 +1,346 @@
import type { SerializableAgentState } from '@n8n/agents';
import { AgentsConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db';
import { Service } from '@n8n/di';
import { DataSource, In, IsNull, MoreThanOrEqual, Not } from '@n8n/typeorm';
import { jsonParse, UserError } from 'n8n-workflow';
import { checkpointExecutionId } from '../types/agent-queued-message';
import type { AgentTaskStop } from '../types/agent-task-stop';
import { AgentExecution } from '../entities/agent-execution.entity';
import { AgentExecutionThread } from '../entities/agent-execution-thread.entity';
import { AgentBackgroundJob } from '../entities/agent-background-job.entity';
import { AgentPlan } from '../entities/agent-plan.entity';
import { AgentCheckpoint } from '../entities/agent-checkpoint.entity';
import { AgentExecutionThreadRepository } from './agent-execution-thread.repository';
type TaskStopScope = AgentTaskStop & { threadId: string };
@Service()
export class AgentTaskCancellationRepository extends BaseRepository<AgentExecutionThread> {
constructor(
dataSource: DataSource,
transactionRunner: TransactionRunner,
private readonly threads: AgentExecutionThreadRepository,
private readonly agentsConfig: AgentsConfig,
) {
super(AgentExecutionThread, dataSource.manager, transactionRunner);
}
async latest(threadId: string, ctx: OperationContext = {}): Promise<TaskStopScope | null> {
const thread = await this.managerFor(ctx).findOneBy(AgentExecutionThread, { id: threadId });
return this.readStop(thread);
}
async saveStop(stop: TaskStopScope, ctx: OperationContext) {
const { threadId, ...taskStop } = stop;
await this.managerFor(ctx).update(AgentExecutionThread, { id: threadId }, { taskStop });
return stop;
}
async captureGeneration(threadId: string, ctx: OperationContext) {
const manager = this.managerFor(ctx);
const [executions, jobs, children] = await Promise.all([
manager.find(AgentExecution, {
where: { threadId },
select: ['id'],
}),
manager.find(AgentBackgroundJob, {
where: { parentThreadId: threadId },
select: ['id', 'sourceExecutionId', 'childThreadId'],
}),
manager.find(AgentExecutionThread, { where: { parentThreadId: threadId }, select: ['id'] }),
]);
return {
executionIds: [
...new Set([
...executions.map((execution) => execution.id),
...jobs.flatMap((job) => (job.sourceExecutionId ? [job.sourceExecutionId] : [])),
]),
],
jobIds: jobs.map((job) => job.id),
threadIds: [
...new Set([
...children.map((child) => child.id),
...jobs.flatMap((job) => (job.childThreadId ? [job.childThreadId] : [])),
]),
],
};
}
async blocksQueue(threadId: string, ctx: OperationContext) {
const stop = await this.latest(threadId, ctx);
return Boolean(stop && !stop.pause.reportedAt);
}
/** Lock ancestors first so dispatch and cancellation use the same admission boundary. */
async lockScope(threadId: string, ctx: OperationContext) {
const lineage = await this.lineage(threadId, ctx);
for (const thread of lineage.toReversed()) await this.threads.lockById(thread.id, ctx);
}
async assertWakeAdmission(
threadId: string,
wake: { jobIds: string[]; planStopId?: string },
ctx: OperationContext,
) {
const pause = await this.latest(threadId, ctx);
if (wake.planStopId) {
if (
!pause ||
pause.pause.id !== wake.planStopId ||
pause.pause.reportExecutionId ||
pause.pause.reportedAt ||
pause.pause.resumedAt ||
(await this.unfinishedWork(pause, ctx)).length
) {
throw new UserError('This plan stop report is no longer available');
}
return;
}
if (await this.hasPausedAncestor(threadId, ctx)) {
throw new UserError('The plan is stopped. Wait for a new user request.');
}
}
async assertAdmission(threadId: string, executionId: string | undefined, ctx: OperationContext) {
if (
(await this.isCancelled(threadId, executionId, ctx)) ||
(await this.hasPausedAncestor(threadId, ctx))
) {
throw new UserError('These tasks were canceled. Wait for a new user request.');
}
}
async isCancelled(threadId: string, executionId?: string, ctx: OperationContext = {}) {
const request = await this.latest(threadId, ctx);
if (!request) return false;
const execution = executionId
? await this.managerFor(ctx).findOneBy(AgentExecution, { id: executionId, threadId })
: await this.managerFor(ctx).findOne(AgentExecution, {
where: { threadId, status: 'running' },
order: { createdAt: 'DESC' },
});
return Boolean(execution && request.generation.executionIds.includes(execution.id));
}
async targetedJobs(request: TaskStopScope, ctx: OperationContext = {}) {
const manager = this.managerFor(ctx);
const jobs: AgentBackgroundJob[] = [];
const visited = new Set<string>();
const pending = [request.threadId, ...request.generation.threadIds];
while (pending.length) {
const threadId = pending.shift()!;
if (visited.has(threadId)) continue;
visited.add(threadId);
const children = await manager.find(AgentBackgroundJob, {
where:
threadId === request.threadId
? [
{ parentThreadId: threadId, id: In(request.generation.jobIds) },
{
parentThreadId: threadId,
sourceExecutionId: In(request.generation.executionIds),
},
]
: { parentThreadId: threadId },
});
jobs.push(...children);
pending.push(...children.flatMap((job) => (job.childThreadId ? [job.childThreadId] : [])));
const delegated = await manager.find(AgentExecutionThread, {
where: {
parentThreadId: threadId,
...(threadId === request.threadId ? { id: In(request.generation.threadIds) } : {}),
},
});
pending.push(...delegated.map((thread) => thread.id));
}
return jobs;
}
async targetedDescendants(request: TaskStopScope, ctx: OperationContext = {}) {
const jobs = await this.targetedJobs(request, ctx);
const threads = [
...request.generation.threadIds,
...jobs.flatMap((job) => (job.childThreadId ? [job.childThreadId] : [])),
];
const seen = new Set(threads);
let parents = [...threads];
while (parents.length) {
const descendants = await this.managerFor(ctx).find(AgentExecutionThread, {
where: { parentThreadId: In(parents) },
});
parents = descendants.filter((thread) => !seen.has(thread.id)).map((thread) => thread.id);
for (const id of parents) seen.add(id);
}
return await this.managerFor(ctx).findBy(AgentExecutionThread, {
id: In([...seen].filter((id) => id !== request.threadId)),
});
}
async unfinishedWork(request: TaskStopScope, ctx: OperationContext = {}) {
const manager = this.managerFor(ctx);
const children = await this.targetedDescendants(request, ctx);
const childIds = children.map((thread) => thread.id);
const executions = await manager.find(AgentExecution, {
where: [
{ threadId: request.threadId, id: In(request.generation.executionIds), status: 'running' },
{ threadId: In(childIds), status: 'running' },
],
});
const checkpoints = await manager.find(AgentCheckpoint, {
where: {
threadId: request.threadId,
state: Not(IsNull()),
expired: false,
updatedAt: MoreThanOrEqual(
new Date(
Date.now() - this.agentsConfig.checkpointTtlSeconds * Time.seconds.toMilliseconds,
),
),
},
});
const jobs = await this.targetedJobs(request, ctx);
return [
...jobs
.filter((job) => ['running', 'suspended'].includes(job.status))
.map((job) => ({ jobId: job.id, title: job.title })),
...executions.map((execution) => ({
jobId: execution.id,
title:
children.find((child) => child.id === execution.threadId)?.agentName ??
'Current response',
})),
...checkpoints.flatMap((checkpoint) => {
const state = jsonParse<SerializableAgentState | null>(checkpoint.state ?? '', {
fallbackValue: null,
});
if (state?.status !== 'suspended') return [];
const executionId = checkpointExecutionId(state);
if (executionId && !request.generation.executionIds.includes(executionId)) return [];
return [{ jobId: executionId ?? request.threadId, title: 'Current response' }];
}),
];
}
async pausedScope(threadId: string, ctx: OperationContext = {}) {
return (
(await this.lineage(threadId, ctx))
.map((thread) => this.readStop(thread))
.find((stop) => stop && !stop.pause.resumedAt) ?? null
);
}
async hasPausedAncestor(threadId: string, ctx: OperationContext) {
return Boolean(await this.pausedScope(threadId, ctx));
}
async recordAdmission(
threadId: string,
executionId: string,
input: {
planStopId?: string;
userInitiated: boolean;
},
ctx: OperationContext,
) {
const stop = await this.latest(threadId, ctx);
if (!stop) return;
if (input.planStopId) {
stop.pause.reportExecutionId = executionId;
} else if (input.userInitiated && !stop.pause.resumedAt) {
if (!stop.pause.reportedAt) throw new UserError('The plan is still stopping');
stop.pause.resumedAt = new Date().toISOString();
} else return;
await this.saveStop(stop, ctx);
}
async pendingPauseReports() {
const isPostgres = this.manager.connection.options.type === 'postgres';
const pause = isPostgres
? 'thread."taskStop"->\'pause\''
: "json_extract(thread.taskStop, '$.pause')";
const reportedAt = isPostgres
? "thread.\"taskStop\"->'pause'->>'reportedAt'"
: "json_extract(thread.taskStop, '$.pause.reportedAt')";
const threads = await this.createQueryBuilder('thread')
.select('thread.id')
.where(`${pause} IS NOT NULL`)
.andWhere(`${reportedAt} IS NULL`)
.getMany();
return threads.map((thread) => thread.id);
}
async pauseReportPlan(stop: TaskStopScope) {
return stop.planId
? await this.managerFor({}).findOneBy(AgentPlan, {
threadId: stop.threadId,
id: stop.planId,
})
: null;
}
async pauseReportTarget(threadId: string) {
return await this.findOneBy({ id: threadId });
}
async finishPauseReport(threadId: string, stopId: string, failed = false) {
return await this.runInTransaction({}, async (manager, ctx) => {
await this.threads.lockById(threadId, ctx);
const stop = await this.latest(threadId, ctx);
if (!stop || stop.pause.id !== stopId || stop.pause.reportedAt) return;
const execution = stop.pause.reportExecutionId
? await manager.findOneBy(AgentExecution, { id: stop.pause.reportExecutionId })
: null;
if (execution?.status === 'running' || (!execution && !failed)) return;
stop.pause.reportedAt = new Date().toISOString();
stop.pause.reportFailed =
failed || !execution || ['error', 'interrupted', 'cancelled'].includes(execution.status);
// Composer Stop also consumes this report. It must not trigger another reply.
await this.saveStop(stop, ctx);
const jobs = await this.targetedJobs(stop, ctx);
if (jobs.length)
await manager.update(
AgentBackgroundJob,
{
id: In(jobs.map((job) => job.id)),
status: In(['paused', 'completed', 'failed', 'cancelled']),
},
{ notifiedAt: new Date(stop.pause.reportedAt) },
);
});
}
private readStop(thread: AgentExecutionThread | null): TaskStopScope | null {
if (!thread?.taskStop) return null;
const stop = thread.taskStop;
return {
...stop,
threadId: thread.id,
// Legacy stops have no pending report and do not pause later turns.
pause: stop.pause ?? {
id: `legacy:${thread.id}:${stop.requestedAt}`,
reportedAt: stop.requestedAt,
resumedAt: stop.requestedAt,
},
};
}
private async lineage(threadId: string, ctx: OperationContext) {
const threads: AgentExecutionThread[] = [];
const visited = new Set<string>();
let id: string | null = threadId;
while (id && !visited.has(id)) {
visited.add(id);
const thread: AgentExecutionThread | null = await this.managerFor(ctx).findOneBy(
AgentExecutionThread,
{ id },
);
if (!thread) break;
threads.push(thread);
id = thread.parentThreadId;
}
return threads;
}
}

53
packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts
View File

@ -1,4 +1,3 @@
import type { AgentMessageSteeringService } from '../../agent-message-steering.service';
import {
INLINE_SUB_AGENT_ID,
type BuiltAgent,
@ -21,6 +20,8 @@ import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import type { Mocked } from 'vitest';
import { mock } from 'vitest-mock-extended';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import type { AgentMessageSteeringService } from '../../agent-message-steering.service';
import type { AgentExecutionService } from '../../agent-execution.service';
import type { AgentMessageQueueService } from '../../agent-message-queue.service';
import type { AgentChatExecutionService } from '../../agent-chat-execution.service';
@ -130,6 +131,7 @@ describe('SubAgentRunner', () => {
let logger: Mocked<Logger>;
let checkpointStorage: Mocked<N8NCheckpointStorage>;
let credentialProvider: Mocked<CredentialProvider>;
let cancellations: Mocked<AgentTaskCancellationRepository>;
beforeEach(() => {
vi.clearAllMocks();
@ -150,9 +152,12 @@ describe('SubAgentRunner', () => {
agentExecutionService.finalizeExecution.mockResolvedValue('agent-execution-1');
checkpointStorage = mock<N8NCheckpointStorage>();
logger = mock<Logger>();
cancellations = mock<AgentTaskCancellationRepository>();
runner = new SubAgentRunner(
cancellations,
sourceResolver,
new AgentTurnExecutionService(
cancellations,
logger,
agentExecutionService,
mock<AgentChatExecutionService>(),
@ -193,6 +198,47 @@ describe('SubAgentRunner', () => {
expect(reconstructionService.reconstructFromResolvedSource).toHaveBeenCalledTimes(1);
});
it('marks task cancellation in the child model and tool checks', async () => {
await runner.run(spawnRequest, {
parentAgentId,
projectId,
credentialProvider,
runType: 'production',
});
cancellations.isCancelled.mockResolvedValue(true);
const hooks = childAgent.stream.mock.calls[0][1]?.guardrails?.hooks ?? [];
const decisions = await Promise.all([
hooks[0]?.before?.(mock()),
hooks[0]?.beforeTool?.(mock()),
]);
expect(decisions).toEqual([
{ action: 'stop', code: 'tasks-cancelled', canceled: true },
{ action: 'stop', code: 'tasks-cancelled', canceled: true },
]);
});
it('blocks a tool returned after an ancestor stops and retains the pause checkpoint path', async () => {
const shouldPause = vi.fn().mockResolvedValue(false);
await runner.run(spawnRequest, {
parentAgentId,
projectId,
credentialProvider,
runType: 'production',
shouldPause,
});
cancellations.isCancelled.mockResolvedValue(false);
cancellations.hasPausedAncestor.mockResolvedValue(true);
const options = childAgent.stream.mock.calls[0][1];
expect(await options?.guardrails?.hooks?.[0]?.beforeTool?.(mock())).toEqual({
action: 'stop',
code: 'tasks-stopped',
canceled: true,
});
expect(await options?.shouldPause?.()).toBe(true);
expect(shouldPause).toHaveBeenCalled();
expect(cancellations.hasPausedAncestor).toHaveBeenCalled();
});
it('rebuilds the child through the shared reconstruction service and runs it with a fresh prompt', async () => {
agentExecutionService.startExecutionRecording.mockImplementation(
async (_params, startedAt) => ({
@ -795,7 +841,8 @@ describe('SubAgentRunner', () => {
});
const options = childAgent.stream.mock.calls[0]?.[1];
expect(options?.guardrails?.hooks).toEqual([]);
expect(options?.guardrails?.hooks).toHaveLength(1);
expect(options?.guardrails?.hooks[0].after).toBeUndefined();
expect(options?.guardrails?.onNotice).toBeDefined();
});
@ -903,7 +950,7 @@ describe('SubAgentRunner', () => {
expect.objectContaining({
runId: 'child-run-1',
hostMetadata: { n8nExecutionId: 'agent-execution-1' },
shouldPause,
shouldPause: expect.any(Function),
}),
);
expect(result).toMatchObject({

23
packages/cli/src/modules/agents/sub-agents/sub-agent-runner.ts
View File

@ -11,6 +11,7 @@ import {
type DelegateSubAgentCancelRequest,
type DelegateSubAgentResumeRequest,
type GenerateResult,
type GuardrailDecision,
type JSONObject,
type JSONValue,
type SerializableAgentState,
@ -34,6 +35,7 @@ import { isRecord } from '@n8n/utils/is-record';
import { jsonParse, UserError } from 'n8n-workflow';
import { v4 as uuid } from 'uuid';
import { AgentTaskCancellationRepository } from '../repositories/agent-task-cancellation.repository';
import type { AgentRunTelemetryType } from '@/interfaces';
import type { StartExecutionParams } from '../agent-execution.service';
@ -166,6 +168,7 @@ type ForegroundOperation = {
@Service()
export class SubAgentRunner {
constructor(
private readonly cancellations: AgentTaskCancellationRepository,
private readonly sourceResolver: SubAgentSourceResolver,
private readonly turnExecutionService: AgentTurnExecutionService,
private readonly checkpointStorage: N8NCheckpointStorage,
@ -357,6 +360,10 @@ export class SubAgentRunner {
{ ...notice, agentId: notice.agentId ?? runtimeSource.source.sourceId },
)
: undefined;
const checkCancellation = async (): Promise<GuardrailDecision | undefined> =>
(await this.cancellations.isCancelled(threadId, executionId))
? { action: 'stop', code: 'tasks-cancelled', canceled: true }
: undefined;
const executionOptions = withBudgetGuardrail(
{
approvalContext: await this.toolApprovalService.createContext(
@ -367,7 +374,21 @@ export class SubAgentRunner {
...(telemetry !== undefined ? { telemetry } : {}),
...modelStreamStallOptions(this.aiConfig),
executionCounter: context.executionCounter,
shouldPause: context.shouldPause,
shouldPause: async () =>
(await context.shouldPause?.()) ||
(await this.cancellations.hasPausedAncestor(threadId, {})),
guardrails: {
hooks: [
{
before: checkCancellation,
beforeTool: async () =>
(await checkCancellation()) ??
((await this.cancellations.hasPausedAncestor(threadId, {}))
? { action: 'stop' as const, code: 'tasks-stopped', canceled: true as const }
: undefined),
},
],
},
},
{
useRootSessionCap: true,

6
packages/cli/src/modules/agents/types/agent-queued-message.ts
View File

@ -79,6 +79,10 @@ export const EXECUTION_METADATA_KEY = 'n8nExecutionId';
export function checkpointExecutionId(state: SerializableAgentState): string | undefined {
if (state.persistence?.delegated) return undefined;
const id = state.persistence?.hostMetadata?.[EXECUTION_METADATA_KEY];
return executionIdFromMetadata(state.persistence?.hostMetadata);
}
export function executionIdFromMetadata(metadata?: Record<string, unknown>): string | undefined {
const id = metadata?.[EXECUTION_METADATA_KEY];
return typeof id === 'string' ? id : undefined;
}

9
packages/cli/test/integration/agent-plan.service.test.ts
View File

@ -1,3 +1,4 @@
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
import { GlobalConfig } from '@n8n/config';
import { TransactionRunner } from '@n8n/db';
@ -114,7 +115,13 @@ describe('AgentPlanService', () => {
},
{},
);
const fresh = new AgentPlanService(new AgentPlanRepository(dataSource, transactionRunner));
const fresh = new AgentPlanService(
new AgentPlanRepository(
dataSource,
transactionRunner,
Container.get(AgentTaskCancellationRepository),
),
);
expect(await fresh.findActivePlan(threadId, {})).toEqual(updated);
expect(updated.readiness.ready).toEqual([second.id]);
expect(updated.data.items[0].startedAt).toEqual(updated.data.items[0].endedAt);

2
packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts
View File

@ -401,6 +401,8 @@ describe('AgentBackgroundJobRepository', () => {
const jobs = await repository.findGroupCandidates(agentId, 'thread-1');
expect(jobs).toHaveLength(1);
expect({ ...jobs[0] }).toEqual({
planId: null,
planTaskId: null,
id,
kind: 'subagent',
title: 'Research',

90
packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
View File

@ -26,6 +26,7 @@ import { v4 as uuid } from 'uuid';
import { mock } from 'vitest-mock-extended';
import { z } from 'zod';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import type { Telemetry } from '@/telemetry';
import type { Publisher } from '@/scaling/pubsub/publisher.service';
import type { ExternalHooks } from '@/external-hooks';
@ -193,7 +194,14 @@ describe('AgentExecutionRepository', () => {
memory.getImplementation.mockReturnValue(memoryBackend);
const attachmentService = mock<AgentChatAttachmentService>();
const executionLogStore = mock<AgentExecutionLogStore>();
const cancellations = new AgentTaskCancellationRepository(
connection ?? repository.manager.connection,
txRunner,
threads,
new AgentsConfig(),
);
const executionService = new AgentExecutionService(
cancellations,
mockLogger(),
executions,
threads,
@ -231,6 +239,7 @@ describe('AgentExecutionRepository', () => {
mock<AgentExecutionUpdateBroadcaster>(),
);
const queue = new AgentMessageQueueService(
cancellations,
txRunner,
queueRepository,
threads,
@ -249,6 +258,7 @@ describe('AgentExecutionRepository', () => {
await finalize();
});
return {
cancellations,
steering,
settingsService,
txRunner,
@ -262,6 +272,7 @@ describe('AgentExecutionRepository', () => {
attachmentService,
executionLogStore,
turns: new AgentTurnExecutionService(
cancellations,
mockLogger(),
executionService,
chatExecutionService,
@ -493,6 +504,7 @@ describe('AgentExecutionRepository', () => {
);
const runner = new SubAgentBackgroundRunner(
new SubAgentRunner(
mock<AgentTaskCancellationRepository>(),
sourceResolver,
turns,
storage,
@ -1678,6 +1690,84 @@ describe('AgentExecutionRepository', () => {
await services.queue.settle(item.thread.id, item.admission.executionId);
}
it('admits one stop report before later input without consuming the queued message', async () => {
const services = recordingServices();
const threadId = uuid();
await enqueue(services, input(threadId, 'Start', 'new'));
const first = await claim(services, threadId);
await finish(services, first);
const stopId = uuid();
await services.cancellations.saveStop(
{
threadId,
planId: null,
requestedAt: new Date().toISOString(),
generation: await services.cancellations.captureGeneration(threadId, {}),
failures: [],
pause: { id: stopId },
},
{},
);
const pending = await enqueue(services, input(threadId, 'Continue'));
expect(await services.queue.claimNext(threadId, async () => true)).toBeNull();
const recording = {
...first.recording,
queueItemId: undefined,
userMessage: 'Report the stop',
hideUserMessageFromTranscript: true,
wake: { jobIds: [], planStopId: stopId },
};
const start = async (params = recording) =>
await services.executionService.startExecutionRecording(params, new Date());
await expect(
start({ ...recording, wake: { jobIds: [], planStopId: uuid() } }),
).rejects.toThrow('This plan stop report is no longer available');
await expect(start({ ...recording, wake: { jobIds: [], planStopId: '' } })).rejects.toThrow(
'The plan is stopped',
);
const blocker = await repository.save({ id: uuid(), threadId, status: 'running' });
await expect(start()).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
await repository.update(blocker.id, { status: 'success' });
await services.queueRepository.update(pending.id, { executionId: blocker.id });
await expect(start()).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
await services.queueRepository.update(pending.id, { executionId: null });
const checkpoints = Container.get(AgentCheckpointRepository);
const checkpoint = await checkpoints.save({
runId: uuid(),
agentId,
threadId,
expired: false,
state: JSON.stringify({
status: 'suspended',
persistence: {
threadId,
hostMetadata: { [EXECUTION_METADATA_KEY]: blocker.id },
},
}),
});
await expect(start()).rejects.toBeInstanceOf(AgentTurnAlreadyRunningError);
await checkpoints.delete(checkpoint.runId);
const report = await start();
expect(report.inputMessageIds).not.toContain(pending.messageId);
expect(await services.queueRepository.findItem(threadId, pending.id, {})).toMatchObject({
executionId: null,
});
await expect(start()).rejects.toThrow('This plan stop report is no longer available');
expect(await services.queue.claimNext(threadId, async () => true)).toBeNull();
const recorder = new ExecutionRecorder();
recorder.record({ type: 'text-delta', id: 'report', delta: 'Tasks stopped.' });
recorder.record({ type: 'finish', finishReason: 'stop' });
await services.executionService.finalizeExecution(report.executionId, {
...recording,
record: recorder.getMessageRecord(),
});
await services.cancellations.finishPauseReport(threadId, stopId);
const next = await claim(services, threadId);
expect(next.admission.inputMessageIds).toEqual([pending.messageId]);
await finish(services, next);
expect(await services.queue.claimNext(threadId, async () => true)).toBeNull();
});
it.each([false, true])(
'finishes active work and cancels pending messages when Agents turns off (attachment cleanup fails: %s)',
async (cleanupFails) => {

7
packages/cli/test/integration/database/repositories/agent-plan.repository.test.ts
View File

@ -6,6 +6,7 @@ import { DataSource, InsertQueryBuilder } from '@n8n/typeorm';
import type { JsonObject } from 'n8n-workflow';
import { randomUUID } from 'node:crypto';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import { AgentExecutionThread } from '@/modules/agents/entities/agent-execution-thread.entity';
import { AgentPlanHistory } from '@/modules/agents/entities/agent-plan-history.entity';
import { AgentPlan } from '@/modules/agents/entities/agent-plan.entity';
@ -76,7 +77,11 @@ describe('AgentPlanRepository', () => {
unknownField: { list: [null, true, 3, 'text', { nested: [] }] },
};
const plan = await createPlan(data);
const freshRepository = new AgentPlanRepository(dataSource, transactionRunner);
const freshRepository = new AgentPlanRepository(
dataSource,
transactionRunner,
Container.get(AgentTaskCancellationRepository),
);
expect(await freshRepository.findActivePlan(threadId, {})).toEqual(plan);
expect(await freshRepository.findPlan(threadId, plan.id, {})).toEqual(plan);

515
packages/cli/test/integration/database/repositories/agent-task-cancellation.repository.test.ts Normal file
View File

@ -0,0 +1,515 @@
import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
import { Container } from '@n8n/di';
import { DataSource } from '@n8n/typeorm';
import { randomUUID } from 'node:crypto';
import { Agent } from '@/modules/agents/entities/agent.entity';
import { AgentCheckpoint } from '@/modules/agents/entities/agent-checkpoint.entity';
import { AgentExecutionThread } from '@/modules/agents/entities/agent-execution-thread.entity';
import { AgentExecution } from '@/modules/agents/entities/agent-execution.entity';
import { AgentBackgroundJob } from '@/modules/agents/entities/agent-background-job.entity';
import { AgentTaskCancellationRepository } from '@/modules/agents/repositories/agent-task-cancellation.repository';
import { AgentMessageQueueRepository } from '@/modules/agents/repositories/agent-message-queue.repository';
import { AgentMessageRepository } from '@/modules/agents/repositories/agent-message.repository';
import { AgentPlanRepository } from '@/modules/agents/repositories/agent-plan.repository';
import { AgentBackgroundJobRepository } from '@/modules/agents/repositories/agent-background-job.repository';
const beforeStop = new Date('2026-10-01T10:00:00.000Z');
const cutoffAt = new Date('2026-10-01T10:01:00.000Z');
const afterStop = new Date('2026-10-01T10:02:00.000Z');
describe('Task cancellation persistence', () => {
let db: DataSource;
let repository: AgentTaskCancellationRepository;
let queue: AgentMessageQueueRepository;
let plans: AgentPlanRepository;
let thread: AgentExecutionThread;
let agentId: string;
let projectId: string;
beforeAll(async () => {
await testModules.loadModules(['agents']);
await testDb.init();
db = Container.get(DataSource);
repository = Container.get(AgentTaskCancellationRepository);
queue = Container.get(AgentMessageQueueRepository);
plans = Container.get(AgentPlanRepository);
({ id: projectId } = await createTeamProject());
agentId = randomUUID();
await db.getRepository(Agent).save({
id: agentId,
name: 'Cancellation tests',
projectId,
integrations: [],
tools: {},
skills: {},
});
});
beforeEach(async () => {
thread = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'Cancellation tests',
projectId,
sessionNumber: 1,
createdAt: beforeStop,
});
});
afterEach(async () => {
await db.getRepository(AgentExecutionThread).delete({ agentId });
});
afterAll(async () => await testDb.terminate());
async function request(planId: string | null = null) {
return await repository.saveStop(
{
threadId: thread.id,
planId,
requestedAt: cutoffAt.toISOString(),
generation: await repository.captureGeneration(thread.id, {}),
failures: [],
pause: { id: randomUUID() },
},
{},
);
}
async function enqueue(message: string) {
const input = await Container.get(AgentMessageRepository).createInput(
{
threadId: thread.id,
resourceId: 'draft-chat:test',
content: { role: 'user', content: [{ type: 'text', text: message }] },
origin: { source: 'chat' },
},
{},
);
return await queue.enqueue(thread.id, input.id, { kind: 'preview' }, {});
}
it('excludes legacy stops from pending reports without skipping current reports', async () => {
const legacy = {
planId: null,
requestedAt: cutoffAt.toISOString(),
generation: { executionIds: [], jobIds: [], threadIds: [] },
failures: [],
};
await db.getRepository(AgentExecutionThread).update(thread.id, {
taskStop: legacy,
});
const current = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'Current stop',
projectId,
});
await repository.saveStop({ ...legacy, threadId: current.id, pause: { id: randomUUID() } }, {});
const pending = await repository.pendingPauseReports();
expect(pending).not.toContain(thread.id);
expect(pending).toContain(current.id);
});
it('reads a legacy stop without scheduling a report or blocking new user input', async () => {
const old = await db.getRepository(AgentExecution).save({
id: randomUUID(),
threadId: thread.id,
status: 'cancelled',
});
const legacy = {
planId: null,
requestedAt: cutoffAt.toISOString(),
generation: await repository.captureGeneration(thread.id, {}),
failures: [],
};
await db.getRepository(AgentExecutionThread).update(thread.id, { taskStop: legacy });
const stop = await repository.latest(thread.id);
expect(stop).toMatchObject(legacy);
expect(await repository.latest(thread.id)).toEqual(stop);
expect(await repository.blocksQueue(thread.id, {})).toBe(false);
expect(await repository.pausedScope(thread.id)).toBeNull();
expect(await repository.isCancelled(thread.id, old.id)).toBe(true);
await expect(repository.assertAdmission(thread.id, old.id, {})).rejects.toThrow();
const next = await db.getRepository(AgentExecution).save({
id: randomUUID(),
threadId: thread.id,
status: 'running',
});
await expect(repository.assertAdmission(thread.id, next.id, {})).resolves.toBeUndefined();
await repository.recordAdmission(thread.id, next.id, { userInitiated: true }, {});
await expect(
repository.assertWakeAdmission(thread.id, { jobIds: [], planStopId: stop!.pause.id }, {}),
).rejects.toThrow('no longer available');
await repository.finishPauseReport(thread.id, stop!.pause.id, true);
expect(
(await db.getRepository(AgentExecutionThread).findOneByOrFail({ id: thread.id })).taskStop,
).toEqual(legacy);
});
it('discards pending and steering messages while preserving later input', async () => {
const first = await enqueue('First');
await enqueue('Second');
const execution = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'running' });
await queue.reserveSteering(thread.id, first.id, execution.id, {});
await queue.discardPending(thread.id, {});
expect(await queue.listPending(thread.id)).toEqual([]);
const next = await enqueue('Continue');
expect((await queue.findHead(thread.id, {}))?.id).toBe(next.id);
});
it('admits one stop report and waits for it before admitting new plan work', async () => {
const old = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'cancelled' });
const stop = await request();
const report = { jobIds: [], planStopId: stop.pause.id };
expect(await repository.pendingPauseReports()).toContain(thread.id);
expect(await repository.blocksQueue(thread.id, {})).toBe(true);
await expect(repository.assertWakeAdmission(thread.id, { jobIds: [] }, {})).rejects.toThrow(
'The plan is stopped',
);
await expect(repository.assertWakeAdmission(thread.id, report, {})).resolves.toBeUndefined();
const execution = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'running' });
await repository.recordAdmission(
thread.id,
execution.id,
{ planStopId: stop.pause.id, userInitiated: false },
{},
);
await expect(repository.assertWakeAdmission(thread.id, report, {})).rejects.toThrow(
'no longer available',
);
await repository.finishPauseReport(thread.id, stop.pause.id);
expect(await repository.blocksQueue(thread.id, {})).toBe(true);
await db.getRepository(AgentExecution).update(execution.id, { status: 'success' });
await repository.finishPauseReport(thread.id, stop.pause.id);
expect(await repository.blocksQueue(thread.id, {})).toBe(false);
expect(await repository.pendingPauseReports()).not.toContain(thread.id);
await repository.recordAdmission(thread.id, randomUUID(), { userInitiated: true }, {});
expect((await repository.latest(thread.id))?.pause.resumedAt).toBeDefined();
expect(await repository.isCancelled(thread.id, old.id)).toBe(true);
await expect(
repository.assertWakeAdmission(thread.id, { jobIds: [] }, {}),
).resolves.toBeUndefined();
});
it('keeps a stopped child checkpoint and blocks dispatch until Continue', async () => {
const child = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'Child',
projectId,
parentThreadId: thread.id,
sessionNumber: 2,
});
const job = await db.getRepository(AgentBackgroundJob).save({
id: randomUUID(),
parentThreadId: thread.id,
parentAgentId: agentId,
parentResourceId: 'draft-chat:test',
parentPrincipalHash: 'principal',
title: 'Child',
kind: 'subagent',
status: 'paused',
childThreadId: child.id,
});
const stop = await request();
await db
.getRepository(AgentCheckpoint)
.save({ runId: randomUUID(), agentId, threadId: child.id, state: '{}', expired: false });
expect(await repository.unfinishedWork(stop)).toEqual([]);
expect(await repository.isCancelled(child.id)).toBe(false);
expect(await repository.hasPausedAncestor(child.id, {})).toBe(true);
await expect(repository.assertAdmission(child.id, undefined, {})).rejects.toThrow();
await repository.finishPauseReport(thread.id, stop.pause.id, true);
await expect(
repository.assertWakeAdmission(thread.id, { jobIds: [], planStopId: stop.pause.id }, {}),
).rejects.toThrow('no longer available');
expect((await repository.latest(thread.id))?.pause).toMatchObject({
reportFailed: true,
reportedAt: expect.any(String),
});
expect(
(await db.getRepository(AgentBackgroundJob).findOneByOrFail({ id: job.id })).notifiedAt,
).not.toBeNull();
await repository.recordAdmission(thread.id, randomUUID(), { userInitiated: true }, {});
expect(await repository.hasPausedAncestor(child.id, {})).toBe(false);
await expect(repository.assertAdmission(child.id, undefined, {})).resolves.toBeUndefined();
});
it('stores only the latest stop boundary on its chat', async () => {
const original = await request();
const other = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'Other chat',
projectId,
sessionNumber: 2,
});
expect(
(await db.getRepository(AgentExecutionThread).findOneByOrFail({ id: thread.id })).taskStop,
).toMatchObject({ requestedAt: original.requestedAt });
expect(await repository.latest(other.id)).toBeNull();
const execution = await db.getRepository(AgentExecution).save({
id: randomUUID(),
threadId: thread.id,
status: 'success',
});
await request();
expect((await repository.latest(thread.id))?.generation.executionIds).toContain(execution.id);
const runner = db.createQueryRunner();
try {
expect(await runner.hasTable('agent_task_cancellation')).toBe(false);
} finally {
await runner.release();
}
});
it('blocks old executions after cancellation and permits new work', async () => {
const old = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'success', createdAt: beforeStop });
await request();
const newer = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'running', createdAt: afterStop });
expect(await repository.isCancelled(thread.id, old.id)).toBe(true);
expect(await repository.isCancelled(thread.id, newer.id)).toBe(false);
});
it('uses captured identities when another server has a later clock', async () => {
const execution = await db.getRepository(AgentExecution).save({
id: randomUUID(),
threadId: thread.id,
status: 'success',
createdAt: afterStop,
});
await request();
expect(await repository.isCancelled(thread.id, execution.id)).toBe(true);
});
it('permits work admitted after cancellation even when its timestamp is earlier', async () => {
const cancellation = await request();
const execution = await db.getRepository(AgentExecution).save({
id: randomUUID(),
threadId: thread.id,
status: 'running',
createdAt: beforeStop,
});
expect(await repository.isCancelled(thread.id, execution.id)).toBe(false);
const job = await db.getRepository(AgentBackgroundJob).save({
id: randomUUID(),
parentThreadId: thread.id,
parentAgentId: agentId,
parentResourceId: 'draft-chat:test',
parentPrincipalHash: 'principal',
title: 'New work',
kind: 'subagent',
status: 'running',
sourceExecutionId: execution.id,
createdAt: beforeStop,
});
expect((await repository.targetedJobs(cancellation)).map((item) => item.id)).not.toContain(
job.id,
);
});
it('does not include descendants from a later user request', async () => {
const stop = await request();
const child = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'New child',
projectId,
parentThreadId: thread.id,
sessionNumber: 2,
});
await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: child.id, status: 'running' });
expect(await repository.targetedDescendants(stop)).toEqual([]);
expect(await repository.unfinishedWork(stop)).toEqual([]);
expect(await repository.isCancelled(child.id)).toBe(false);
});
it('waits for a live foreground checkpoint to stop', async () => {
const execution = await db.getRepository(AgentExecution).save({
id: randomUUID(),
threadId: thread.id,
status: 'success',
hitlStatus: 'suspended',
});
const stop = await request();
const runId = randomUUID();
await db.getRepository(AgentCheckpoint).save({
runId,
agentId,
threadId: thread.id,
expired: false,
state: JSON.stringify({
status: 'suspended',
persistence: {
threadId: thread.id,
hostMetadata: { n8nExecutionId: execution.id },
},
}),
});
expect(await repository.unfinishedWork(stop)).toEqual([
{ jobId: execution.id, title: 'Current response' },
]);
await db
.getRepository(AgentCheckpoint)
.update({ runId }, { updatedAt: new Date('2000-01-01') });
expect(await repository.unfinishedWork(stop)).toEqual([]);
await db.getRepository(AgentCheckpoint).update({ runId }, { updatedAt: new Date() });
await db.getRepository(AgentCheckpoint).update({ runId }, { state: null, expired: true });
expect(await repository.unfinishedWork(stop)).toEqual([]);
});
it('pauses a child admitted before Stop even if its thread starts later', async () => {
const child = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'Child',
projectId,
sessionNumber: 2,
parentThreadId: thread.id,
createdAt: afterStop,
});
await db.getRepository(AgentBackgroundJob).save({
id: randomUUID(),
parentThreadId: thread.id,
parentAgentId: agentId,
parentResourceId: 'draft-chat:test',
parentPrincipalHash: 'principal',
title: 'Child',
kind: 'subagent',
status: 'cancelled',
childThreadId: child.id,
createdAt: beforeStop,
});
await request();
expect(await repository.hasPausedAncestor(child.id, {})).toBe(true);
});
it('preserves a descendant checkpoint saved while stopping', async () => {
const child = await db.getRepository(AgentExecutionThread).save({
id: randomUUID(),
agentId,
agentName: 'Child',
projectId,
sessionNumber: 2,
parentThreadId: thread.id,
createdAt: beforeStop,
});
const cancellation = await request();
const runId = randomUUID();
await db
.getRepository(AgentCheckpoint)
.save({ runId, agentId, threadId: child.id, state: '{}', expired: false });
expect(await repository.unfinishedWork(cancellation, {})).toEqual([]);
expect(await db.getRepository(AgentCheckpoint).findOneBy({ runId })).not.toBeNull();
await db.getRepository(AgentCheckpoint).update({ runId }, { state: null });
expect(await repository.unfinishedWork(cancellation, {})).toEqual([]);
});
it('rejects dispatch from a canceled execution', async () => {
const source = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'running' });
await request();
await expect(
Container.get(AgentBackgroundJobRepository).insertSubAgentJobIfCapacity(
{
id: randomUUID(),
kind: 'subagent',
parentThreadId: thread.id,
parentAgentId: agentId,
parentResourceId: 'draft-chat:test',
parentPrincipalHash: 'principal',
title: 'Late task',
sourceExecutionId: source.id,
subAgentId: agentId,
childThreadId: randomUUID(),
timeoutAt: afterStop,
},
5,
),
).rejects.toThrow('These tasks were canceled');
});
it('tracks a workflow that reaches Wait after cancellation so stopping can retry', async () => {
const source = await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'success', createdAt: beforeStop });
const cancellation = await request();
const id = randomUUID();
await Container.get(AgentBackgroundJobRepository).insertWorkflowJobOrGetExisting({
id,
kind: 'workflow',
parentThreadId: thread.id,
parentAgentId: agentId,
parentResourceId: 'draft-chat:test',
parentPrincipalHash: 'principal',
title: 'Waiting workflow',
workflowId: 'workflow',
childExecutionId: '123',
sourceExecutionId: source.id,
});
expect(await repository.latest(thread.id)).toEqual(cancellation);
expect((await repository.targetedJobs(cancellation)).map((job) => job.id)).toContain(id);
});
it('keeps plan history and blocks late writes after Stop', async () => {
await db
.getRepository(AgentExecution)
.save({ id: randomUUID(), threadId: thread.id, status: 'running' });
const plan = await plans.createActivePlan(
{ id: randomUUID(), threadId: thread.id, formatVersion: 1, data: { result: 'Saved' } },
{},
);
await request(plan.id);
await expect(
plans.replacePlan(
{ threadId: thread.id, planId: plan.id, expectedRevision: 1, formatVersion: 1, data: {} },
{},
),
).rejects.toThrow('These tasks were canceled');
const current = await plans.findPlan(thread.id, plan.id, {});
expect(current).toMatchObject({ closedAt: null, revision: 1, data: { result: 'Saved' } });
expect((await plans.findRevision(thread.id, plan.id, 1, {}))?.data).toEqual({
result: 'Saved',
});
expect(await plans.findRevision(thread.id, plan.id, 2, {})).toBeNull();
});
it('blocks an old wake even if the job finished before Stop', async () => {
const job = await db.getRepository(AgentBackgroundJob).save({
id: randomUUID(),
parentThreadId: thread.id,
parentAgentId: agentId,
parentResourceId: 'draft-chat:test',
parentPrincipalHash: 'principal',
title: 'Completed task',
kind: 'subagent',
status: 'completed',
result: 'Saved output',
});
const stop = await request();
await repository.finishPauseReport(thread.id, stop.pause.id, true);
await expect(
repository.assertWakeAdmission(thread.id, { jobIds: [job.id] }, {}),
).rejects.toThrow('The plan is stopped');
await expect(repository.assertWakeAdmission(thread.id, { jobIds: [] }, {})).rejects.toThrow(
'The plan is stopped',
);
expect(
await db.getRepository(AgentBackgroundJob).findOneByOrFail({ id: job.id }),
).toMatchObject({ result: 'Saved output', notifiedAt: expect.any(Date) });
});
});