From da6dbc7ed3da0cfd336000981cb5a5f1a8030871 Mon Sep 17 00:00:00 2001 From: Eugene Molodkin Date: Fri, 16 Oct 2026 14:11:53 +0200 Subject: [PATCH] feat(core): Enforce saved Agent plan stop boundaries (no-changelog) --- ...ent-execution-orchestrator.service.test.ts | 151 ++++- .../__tests__/agent-execution.service.test.ts | 5 + .../agent-message-queue.service.test.ts | 2 + ...e-reconstruction-integration-tools.test.ts | 2 + .../agent-turn-execution.service.test.ts | 2 + .../agent-workflow-execution.service.test.ts | 4 +- .../agent-execution-orchestrator.service.ts | 10 +- .../modules/agents/agent-execution.service.ts | 20 + .../agents/agent-message-queue.service.ts | 3 + .../agents/agent-turn-execution.service.ts | 17 + .../plans/__tests__/agent-plan-tools.test.ts | 11 + .../modules/agents/plans/agent-plan-tools.ts | 11 +- .../agent-background-job.repository.ts | 42 +- .../agent-message-queue.repository.ts | 23 + .../repositories/agent-plan.repository.ts | 43 +- .../agent-task-cancellation.repository.ts | 346 ++++++++++++ .../__tests__/sub-agent-runner.test.ts | 53 +- .../agents/sub-agents/sub-agent-runner.ts | 23 +- .../agents/types/agent-queued-message.ts | 6 +- .../integration/agent-plan.service.test.ts | 9 +- .../agent-background-job.repository.test.ts | 2 + .../agent-execution.repository.test.ts | 90 +++ .../agent-plan.repository.test.ts | 7 +- ...agent-task-cancellation.repository.test.ts | 515 ++++++++++++++++++ 24 files changed, 1356 insertions(+), 41 deletions(-) create mode 100644 packages/cli/src/modules/agents/repositories/agent-task-cancellation.repository.ts create mode 100644 packages/cli/test/integration/database/repositories/agent-task-cancellation.repository.test.ts diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index ba19535f79a..f9c3c850031 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -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(); 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(); + 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 () => { diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts index 2f27f4f85e5..e99ac23f71e 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts @@ -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(), mockLogger(), agentExecutionRepository, agentExecutionThreadRepository, @@ -769,6 +771,7 @@ describe('AgentExecutionService', () => { it('writes the timeline to blob storage in non-db mode', async () => { storageConfig = mock({ modeTag: 'fs' }); service = new AgentExecutionService( + mock(), mockLogger(), agentExecutionRepository, agentExecutionThreadRepository, @@ -866,6 +869,7 @@ describe('AgentExecutionService', () => { async (failure, currentLocation, shouldDeleteBlob) => { storageConfig = mock({ modeTag: 'fs' }); service = new AgentExecutionService( + mock(), mockLogger(), agentExecutionRepository, agentExecutionThreadRepository, @@ -1480,6 +1484,7 @@ describe('AgentExecutionService', () => { it('preserves an interrupted execution inline without overwriting blob storage', async () => { storageConfig = mock({ modeTag: 'fs' }); service = new AgentExecutionService( + mock(), mockLogger(), agentExecutionRepository, agentExecutionThreadRepository, diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts index 727d5a713c2..fd55d2db491 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue.service.test.ts @@ -19,6 +19,7 @@ import type { AgentExecutionThreadRepository } from '../repositories/agent-execu import type { AgentMessageQueueRepository } from '../repositories/agent-message-queue.repository'; import type { AgentMessageRepository } from '../repositories/agent-message.repository'; import type { AgentRepository } from '../repositories/agent.repository'; +import type { AgentTaskCancellationRepository } from '../repositories/agent-task-cancellation.repository'; function makeThread(overrides: Partial = {}): AgentExecutionThread { return mock({ @@ -56,6 +57,7 @@ describe('AgentMessageQueueService', () => { new Map([['thread-1', N8N_CHAT_PRODUCTION_SOURCE]]), ); service = new AgentMessageQueueService( + mock(), txRunner, repository, threadRepository, diff --git a/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts b/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts index 3b32b5e9248..8ca74e6e9cd 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-runtime-reconstruction-integration-tools.test.ts @@ -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(), logger, agentExecutionService, mock(), diff --git a/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts index a900549b6d5..120ce00cc40 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-turn-execution.service.test.ts @@ -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(); const service = new AgentTurnExecutionService( + mock(), mock(), executions, mock(), diff --git a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts index e469ea114bb..4c1e5b051e6 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts @@ -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(), mockLogger(), executionService, mock(), diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts index 13021e94f3b..810fd7eb9b9 100644 --- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts +++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts @@ -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; 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, }); } } diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts index b9719a35193..79240faa453 100644 --- a/packages/cli/src/modules/agents/agent-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-execution.service.ts @@ -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 { + wake?: { jobIds: string[]; planStopId?: string }; resourceId: string; messageOrigin?: Omit; hideUserMessageFromTranscript?: boolean; @@ -182,6 +184,7 @@ export class AgentExecutionService { private readonly sideCallUsageInFlightByExecution = new Map>>(); 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 }; diff --git a/packages/cli/src/modules/agents/agent-message-queue.service.ts b/packages/cli/src/modules/agents/agent-message-queue.service.ts index 19225e50712..6a97fa9b842 100644 --- a/packages/cli/src/modules/agents/agent-message-queue.service.ts +++ b/packages/cli/src/modules/agents/agent-message-queue.service.ts @@ -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); diff --git a/packages/cli/src/modules/agents/agent-turn-execution.service.ts b/packages/cli/src/modules/agents/agent-turn-execution.service.ts index 53e9cc8c801..6266d9a0805 100644 --- a/packages/cli/src/modules/agents/agent-turn-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-turn-execution.service.ts @@ -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; 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> { const { executionId, inputMessageIds } = admission; + const checkCancellation = async (): Promise => + (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 diff --git a/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts b/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts index aa538541ffc..33e117166df 100644 --- a/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts +++ b/packages/cli/src/modules/agents/plans/__tests__/agent-plan-tools.test.ts @@ -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], diff --git a/packages/cli/src/modules/agents/plans/agent-plan-tools.ts b/packages/cli/src/modules/agents/plans/agent-plan-tools.ts index e4c03490bf1..49031415289 100644 --- a/packages/cli/src/modules/agents/plans/agent-plan-tools.ts +++ b/packages/cli/src/modules/agents/plans/agent-plan-tools.ts @@ -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( handler: ( input: z.output, threadId: string, - ctx: OperationContext, + ctx: AgentPlanOperationContext, ) => Promise, ) { return new Tool(name) @@ -267,7 +268,11 @@ function scopedTool( } 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 { diff --git a/packages/cli/src/modules/agents/repositories/agent-background-job.repository.ts b/packages/cli/src/modules/agents/repositories/agent-background-job.repository.ts index 0a21df59314..9b3f98345ee 100644 --- a/packages/cli/src/modules/agents/repositories/agent-background-job.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-background-job.repository.ts @@ -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 { 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 { - 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 '"result"' : null), error: settlement.error ?? null, settledAt: new Date(), notifiedAt: @@ -520,6 +541,11 @@ export class AgentBackgroundJobRepository extends BaseRepository { + // 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 { await this.delete({ diff --git a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts index 8f164083802..eb56cd96f7c 100644 --- a/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-message-queue.repository.ts @@ -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 + 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 }, diff --git a/packages/cli/src/modules/agents/repositories/agent-plan.repository.ts b/packages/cli/src/modules/agents/repositories/agent-plan.repository.ts index 1013317ce36..bae963a55b2 100644 --- a/packages/cli/src/modules/agents/repositories/agent-plan.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-plan.repository.ts @@ -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; +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 { - 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 { 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 { }); } - async findActivePlan(threadId: string, ctx: OperationContext): Promise { + 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 { return await this.managerFor(ctx).findOneBy(AgentPlan, { threadId, closedAt: IsNull() }); } async findPlan( threadId: string, planId: string, - ctx: OperationContext, + ctx: AgentPlanOperationContext, ): Promise { return await this.managerFor(ctx).findOneBy(AgentPlan, { id: planId, threadId }); } @@ -98,7 +117,7 @@ export class AgentPlanRepository extends BaseRepository { threadId: string, planId: string, revision: number, - ctx: OperationContext, + ctx: AgentPlanOperationContext, ): Promise { return await this.managerFor(ctx).findOne(AgentPlanHistory, { where: { planId, revision, plan: { threadId } }, @@ -109,7 +128,7 @@ export class AgentPlanRepository extends BaseRepository { 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 { async replacePlan( input: PlanWrite & PlanDocument, - ctx: OperationContext, + ctx: AgentPlanOperationContext, ): Promise { this.validateVersion(input.formatVersion); return await this.writeRevision(input, { ...input, data: this.serializeData(input.data) }, ctx); } - async closePlan(input: PlanWrite, ctx: OperationContext): Promise { + async closePlan(input: PlanWrite, ctx: AgentPlanOperationContext): Promise { return await this.writeRevision(input, null, ctx); } private async writeRevision( input: PlanWrite, document: { formatVersion: number; data: string } | null, - ctx: OperationContext, + ctx: AgentPlanOperationContext, ): Promise { 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() diff --git a/packages/cli/src/modules/agents/repositories/agent-task-cancellation.repository.ts b/packages/cli/src/modules/agents/repositories/agent-task-cancellation.repository.ts new file mode 100644 index 00000000000..7924ace78bb --- /dev/null +++ b/packages/cli/src/modules/agents/repositories/agent-task-cancellation.repository.ts @@ -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 { + 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 { + 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(); + 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(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(); + 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; + } +} diff --git a/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts b/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts index 5d3fc1c063b..86cdf6bb096 100644 --- a/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts +++ b/packages/cli/src/modules/agents/sub-agents/__tests__/sub-agent-runner.test.ts @@ -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; let checkpointStorage: Mocked; let credentialProvider: Mocked; + let cancellations: Mocked; beforeEach(() => { vi.clearAllMocks(); @@ -150,9 +152,12 @@ describe('SubAgentRunner', () => { agentExecutionService.finalizeExecution.mockResolvedValue('agent-execution-1'); checkpointStorage = mock(); logger = mock(); + cancellations = mock(); runner = new SubAgentRunner( + cancellations, sourceResolver, new AgentTurnExecutionService( + cancellations, logger, agentExecutionService, mock(), @@ -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({ diff --git a/packages/cli/src/modules/agents/sub-agents/sub-agent-runner.ts b/packages/cli/src/modules/agents/sub-agents/sub-agent-runner.ts index 258a1289d1c..0706ff1d7ee 100644 --- a/packages/cli/src/modules/agents/sub-agents/sub-agent-runner.ts +++ b/packages/cli/src/modules/agents/sub-agents/sub-agent-runner.ts @@ -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 => + (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, diff --git a/packages/cli/src/modules/agents/types/agent-queued-message.ts b/packages/cli/src/modules/agents/types/agent-queued-message.ts index 2c0fa264976..1b3ffc2a412 100644 --- a/packages/cli/src/modules/agents/types/agent-queued-message.ts +++ b/packages/cli/src/modules/agents/types/agent-queued-message.ts @@ -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 | undefined { + const id = metadata?.[EXECUTION_METADATA_KEY]; return typeof id === 'string' ? id : undefined; } diff --git a/packages/cli/test/integration/agent-plan.service.test.ts b/packages/cli/test/integration/agent-plan.service.test.ts index 8baa3679b0a..2c3c5871e81 100644 --- a/packages/cli/test/integration/agent-plan.service.test.ts +++ b/packages/cli/test/integration/agent-plan.service.test.ts @@ -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); diff --git a/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts index cc9d8524c48..b63e8ada534 100644 --- a/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts @@ -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', diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts index 9b2ab244d22..7d217d4445c 100644 --- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts @@ -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(); const executionLogStore = mock(); + 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(), ); 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(), 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) => { diff --git a/packages/cli/test/integration/database/repositories/agent-plan.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-plan.repository.test.ts index d2cc900f23d..f6a1b994682 100644 --- a/packages/cli/test/integration/database/repositories/agent-plan.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-plan.repository.test.ts @@ -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); diff --git a/packages/cli/test/integration/database/repositories/agent-task-cancellation.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-task-cancellation.repository.test.ts new file mode 100644 index 00000000000..80f2e66ad6e --- /dev/null +++ b/packages/cli/test/integration/database/repositories/agent-task-cancellation.repository.test.ts @@ -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) }); + }); +});