From ea9021ed86e2da98a11c8cfd5fc67e6f203720e5 Mon Sep 17 00:00:00 2001 From: Robin Braumann <50590409+bjorger@users.noreply.github.com> Date: Fri, 2 Oct 2026 13:10:50 +0000 Subject: [PATCH] feat(core): Add durable pause checkpoints for background sub-agents (no-changelog) (#40118) --- docs/generated/postgres-schema/README.md | 3 +- .../public.agent_background_job.md | 6 +- .../postgres-schema/public.agents.md | 1 + docs/generated/sqlite-schema/README.md | 3 +- .../sqlite-schema/agent_background_job.md | 8 +- docs/generated/sqlite-schema/agents.md | 1 + .../runtime/__tests__/agent-runtime.test.ts | 320 +++++++++++++++++- .../agents/src/runtime/loop/agent-runtime.ts | 82 ++++- .../agents/src/runtime/loop/generate-sink.ts | 2 +- .../agents/src/runtime/loop/stream-sink.ts | 2 +- packages/@n8n/agents/src/sdk/agent.ts | 14 + .../agents/src/types/runtime/agent-loop.ts | 1 + packages/@n8n/agents/src/types/sdk/agent.ts | 10 + .../api-types/src/agents/background-job.ts | 4 +- ...790937677019-AddAgentBackgroundJobPause.ts | 60 ++++ .../discovery/__tests__/stream-status.test.ts | 9 +- .../evaluations/discovery/stream-status.ts | 2 + .../agent-background-job.service.test.ts | 28 +- .../agent-background-job.service.ts | 17 +- .../agents/background/agent-wake.service.ts | 4 +- .../entities/agent-background-job.entity.ts | 3 + .../integrations/n8n-checkpoint-storage.ts | 19 ++ .../agent-background-job.repository.ts | 177 ++++++++-- .../agent-checkpoint.repository.ts | 14 + .../agent-background-job.repository.test.ts | 298 +++++++++++++++- .../agent-execution.repository.test.ts | 64 ++++ ...019-add-agent-background-job-pause.test.ts | 157 +++++++++ .../agents/components/AgentChatPanel.vue | 31 +- 28 files changed, 1244 insertions(+), 96 deletions(-) create mode 100644 packages/@n8n/db/src/migrations/common/1790937677019-AddAgentBackgroundJobPause.ts create mode 100644 packages/cli/test/migration/1790937677019-add-agent-background-job-pause.test.ts diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 7b3747ea6f1..26392970d80 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -9,7 +9,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | Name | Columns | Comment | Type | | ---- | ------- | ------- | ---- | | [public.activity_event](public.activity_event.md) | 11 | | BASE TABLE | -| [public.agent_background_job](public.agent_background_job.md) | 19 | | BASE TABLE | +| [public.agent_background_job](public.agent_background_job.md) | 20 | | BASE TABLE | | [public.agent_channel_status](public.agent_channel_status.md) | 11 | | BASE TABLE | | [public.agent_chat_attachments](public.agent_chat_attachments.md) | 12 | | BASE TABLE | | [public.agent_chat_subscriptions](public.agent_chat_subscriptions.md) | 6 | | BASE TABLE | @@ -429,6 +429,7 @@ erDiagram varchar_64_ parentPrincipalHash varchar_255_ parentResourceId varchar_128_ parentThreadId + uuid pauseRequestId text result timestamp_3__with_time_zone settledAt varchar_16_ status diff --git a/docs/generated/postgres-schema/public.agent_background_job.md b/docs/generated/postgres-schema/public.agent_background_job.md index 31762fb20a8..ef567001ba6 100644 --- a/docs/generated/postgres-schema/public.agent_background_job.md +++ b/docs/generated/postgres-schema/public.agent_background_job.md @@ -15,9 +15,10 @@ | parentPrincipalHash | varchar(64) | | false | | | Sandbox principal hash of the parent agent run | | parentResourceId | varchar(255) | | false | | | Memory resource of the parent agent run | | parentThreadId | varchar(128) | | false | | | | +| pauseRequestId | uuid | | true | | | Groups jobs stopped by one user request | | result | text | | true | | | Final answer of a settled sub-agent job | | settledAt | timestamp(3) with time zone | | true | | | | -| status | varchar(16) | | false | | | running: child works; suspended: child waits for an approval; completed, failed, cancelled: terminal | +| status | varchar(16) | | false | | | running: child works; suspended: child awaits approval; paused: user stopped the child; completed, failed, cancelled: terminal | | subAgentId | varchar(36) | | true | | | Sub-agent jobs only | | timeoutAt | timestamp(3) with time zone | | true | | | When reconciliation fails the job as timed out; NULL means no timeout | | title | varchar(255) | | false | | | Task name or workflow name, echoed in status-check listings | @@ -29,7 +30,7 @@ | Name | Type | Definition | | ---- | ---- | ---------- | | CHK_agent_background_job_kind | CHECK | CHECK (((kind)::text = ANY ((ARRAY['subagent'::character varying, 'workflow'::character varying])::text[]))) | -| CHK_agent_background_job_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['running'::character varying, 'suspended'::character varying, 'completed'::character varying, 'failed'::character varying, 'cancelled'::character varying])::text[]))) | +| CHK_agent_background_job_status | CHECK | CHECK (((status)::text = ANY ((ARRAY['running'::character varying, 'suspended'::character varying, 'paused'::character varying, 'completed'::character varying, 'failed'::character varying, 'cancelled'::character varying])::text[]))) | | FK_d46c6f00730c2ef8bcb6ee24b67 | FOREIGN KEY | FOREIGN KEY ("parentAgentId") REFERENCES agents(id) ON DELETE CASCADE | | PK_6e0db58281aa2b4c956dc0d58e9 | PRIMARY KEY | PRIMARY KEY (id) | | agent_background_job_createdAt_not_null | n | NOT NULL "createdAt" | @@ -74,6 +75,7 @@ erDiagram varchar_64_ parentPrincipalHash varchar_255_ parentResourceId varchar_128_ parentThreadId + uuid pauseRequestId text result timestamp_3__with_time_zone settledAt varchar_16_ status diff --git a/docs/generated/postgres-schema/public.agents.md b/docs/generated/postgres-schema/public.agents.md index 36eab48dc94..3d52647188f 100644 --- a/docs/generated/postgres-schema/public.agents.md +++ b/docs/generated/postgres-schema/public.agents.md @@ -112,6 +112,7 @@ erDiagram varchar_64_ parentPrincipalHash varchar_255_ parentResourceId varchar_128_ parentThreadId + uuid pauseRequestId text result timestamp_3__with_time_zone settledAt varchar_16_ status diff --git a/docs/generated/sqlite-schema/README.md b/docs/generated/sqlite-schema/README.md index 2de14e1e9ba..6e1b8d32305 100644 --- a/docs/generated/sqlite-schema/README.md +++ b/docs/generated/sqlite-schema/README.md @@ -9,7 +9,7 @@ Auto-generated from the SQLite migrations in @n8n/db. Do not edit by hand. | Name | Columns | Comment | Type | | ---- | ------- | ------- | ---- | | [activity_event](activity_event.md) | 11 | | table | -| [agent_background_job](agent_background_job.md) | 19 | | table | +| [agent_background_job](agent_background_job.md) | 20 | | table | | [agent_channel_status](agent_channel_status.md) | 11 | | table | | [agent_chat_attachments](agent_chat_attachments.md) | 12 | | table | | [agent_chat_subscriptions](agent_chat_subscriptions.md) | 6 | | table | @@ -415,6 +415,7 @@ erDiagram varchar_64_ parentPrincipalHash varchar_255_ parentResourceId varchar_128_ parentThreadId + varchar pauseRequestId TEXT result datetime_3_ settledAt varchar_16_ status diff --git a/docs/generated/sqlite-schema/agent_background_job.md b/docs/generated/sqlite-schema/agent_background_job.md index 5b034bef97e..f7e22cd9e97 100644 --- a/docs/generated/sqlite-schema/agent_background_job.md +++ b/docs/generated/sqlite-schema/agent_background_job.md @@ -6,7 +6,7 @@ Table Definition ```sql -CREATE TABLE "agent_background_job" ("id" varchar(36) PRIMARY KEY NOT NULL, "kind" varchar(16) NOT NULL, "status" varchar(16) NOT NULL, "parentAgentId" varchar(36) NOT NULL, "parentThreadId" varchar(128) NOT NULL, "title" varchar(255) NOT NULL, "subAgentId" varchar(36), "childThreadId" varchar(128), "childExecutionId" varchar(36), "workflowId" varchar(36), "timeoutAt" datetime(3), "result" text, "error" text, "settledAt" datetime(3), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "notifiedAt" datetime(3), "parentResourceId" varchar(255) NOT NULL, "parentPrincipalHash" varchar(64) NOT NULL, CONSTRAINT "CHK_agent_background_job_kind" CHECK ((("kind" IN ('subagent', 'workflow')))), CONSTRAINT "CHK_agent_background_job_status" CHECK ("status" IN ('running', 'suspended', 'completed', 'failed', 'cancelled')), CONSTRAINT "FK_d46c6f00730c2ef8bcb6ee24b67" FOREIGN KEY ("parentAgentId") REFERENCES "agents" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) +CREATE TABLE "agent_background_job" ("id" varchar(36) PRIMARY KEY NOT NULL, "kind" varchar(16) NOT NULL, "status" varchar(16) NOT NULL, "parentAgentId" varchar(36) NOT NULL, "parentThreadId" varchar(128) NOT NULL, "title" varchar(255) NOT NULL, "subAgentId" varchar(36), "childThreadId" varchar(128), "childExecutionId" varchar(36), "workflowId" varchar(36), "timeoutAt" datetime(3), "result" text, "error" text, "settledAt" datetime(3), "createdAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "updatedAt" datetime(3) NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')), "notifiedAt" datetime(3), "parentResourceId" varchar(255) NOT NULL, "parentPrincipalHash" varchar(64) NOT NULL, "pauseRequestId" varchar, CONSTRAINT "CHK_agent_background_job_kind" CHECK (((("kind" IN ('subagent', 'workflow'))))), CONSTRAINT "CHK_agent_background_job_status" CHECK ("status" IN ('running', 'suspended', 'paused', 'completed', 'failed', 'cancelled')), CONSTRAINT "FK_d46c6f00730c2ef8bcb6ee24b67" FOREIGN KEY ("parentAgentId") REFERENCES "agents" ("id") ON DELETE CASCADE ON UPDATE NO ACTION) ``` @@ -26,6 +26,7 @@ CREATE TABLE "agent_background_job" ("id" varchar(36) PRIMARY KEY NOT NULL, "kin | parentPrincipalHash | varchar(64) | | false | | | | | parentResourceId | varchar(255) | | false | | | | | parentThreadId | varchar(128) | | false | | | | +| pauseRequestId | varchar | | true | | | | | result | TEXT | | true | | | | | settledAt | datetime(3) | | true | | | | | status | varchar(16) | | false | | | | @@ -39,8 +40,8 @@ CREATE TABLE "agent_background_job" ("id" varchar(36) PRIMARY KEY NOT NULL, "kin | Name | Type | Definition | | ---- | ---- | ---------- | -| - | CHECK | CHECK ((("kind" IN ('subagent', 'workflow')))) | -| - | CHECK | CHECK ("status" IN ('running', 'suspended', 'completed', 'failed', 'cancelled')) | +| - | CHECK | CHECK (((("kind" IN ('subagent', 'workflow'))))) | +| - | CHECK | CHECK ("status" IN ('running', 'suspended', 'paused', 'completed', 'failed', 'cancelled')) | | - (Foreign key ID: 0) | FOREIGN KEY | FOREIGN KEY (parentAgentId) REFERENCES agents (id) ON UPDATE NO ACTION ON DELETE CASCADE MATCH NONE | | id | PRIMARY KEY | PRIMARY KEY (id) | | sqlite_autoindex_agent_background_job_1 | PRIMARY KEY | PRIMARY KEY (id) | @@ -76,6 +77,7 @@ erDiagram varchar_64_ parentPrincipalHash varchar_255_ parentResourceId varchar_128_ parentThreadId + varchar pauseRequestId TEXT result datetime_3_ settledAt varchar_16_ status diff --git a/docs/generated/sqlite-schema/agents.md b/docs/generated/sqlite-schema/agents.md index ad02fc08076..1b0d460634d 100644 --- a/docs/generated/sqlite-schema/agents.md +++ b/docs/generated/sqlite-schema/agents.md @@ -114,6 +114,7 @@ erDiagram varchar_64_ parentPrincipalHash varchar_255_ parentResourceId varchar_128_ parentThreadId + varchar pauseRequestId TEXT result datetime_3_ settledAt varchar_16_ status diff --git a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts index d348e68d3ad..95e79945175 100644 --- a/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts +++ b/packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts @@ -1070,6 +1070,39 @@ describe('AgentRuntime — execution counters', () => { ); }); + it.each(['generate', 'stream'] as const)( + 'omits local and provider tools for one %s execution when disabled', + async (mode) => { + generateText.mockResolvedValue(makeGenerateSuccess()); + streamText.mockImplementation(() => makeStreamSuccess()); + const runtime = new AgentRuntime({ + name: 'test', + model: 'openai/gpt-4o-mini', + instructions: 'Report the saved progress.', + tools: [makeMockTool('work', vi.fn())], + providerTools: [{ name: 'openai.web_search', args: {} }], + }); + + for (const toolsEnabled of [false, undefined]) { + if (mode === 'generate') await runtime.generate('Report', { toolsEnabled }); + else { + const result = await runtime.stream('Report', { toolsEnabled }); + await collectChunks(result.stream); + } + const model = mode === 'generate' ? generateText : streamText; + const request = model.mock.lastCall?.[0]; + if (toolsEnabled === false) expect(request.tools).toBeUndefined(); + else { + expect(request.tools.work).toBeDefined(); + expect(request.tools['openai.web_search']).toMatchObject({ + type: 'provider', + isProviderExecuted: true, + }); + } + } + }, + ); + it('counts provider-executed tool calls when surfaced by the model', async () => { generateText .mockResolvedValueOnce({ @@ -3417,6 +3450,236 @@ function makeStreamWithToolCalls( }; } +describe('AgentRuntime — user pause', () => { + beforeEach(() => { + generateText.mockReset(); + streamText.mockReset(); + }); + + it.each(['generate', 'stream'] as const)( + 'finishes the current %s tool batch and resumes once without replay', + async (mode) => { + const store = makeClaimingCheckpointStore(); + const entered = createDeferredPromise(); + const release = createDeferredPromise(); + const handler = vi.fn(async () => { + entered.resolve(); + await release.promise; + return 'Saved partial result'; + }); + const tool = makeMockTool('work', handler); + const calls = ['first', 'second'].map((toolCallId) => ({ + toolCallId, + toolName: 'work', + args: {}, + })); + generateText.mockResolvedValueOnce(makeGenerateWithToolCalls(calls)); + streamText.mockReturnValueOnce(makeStreamWithToolCalls(calls)); + const runtime = createRuntimeWithCheckpointStore([tool], store); + let requested = false; + let finished = false; + let runId = ''; + const options = { + shouldPause: async () => requested, + persistence: { threadId: 'child', resourceId: 'user' }, + }; + const completion = (async () => { + if (mode === 'generate') { + const result = await runtime.generate('Do two steps', options); + runId = result.runId; + return result; + } + const result = await runtime.stream('Do two steps', options); + runId = result.runId; + return await collectChunks(result.stream); + })().finally(() => { + finished = true; + }); + await entered.promise; + requested = true; + expect(finished).toBe(false); + expect(store.save).not.toHaveBeenCalled(); + release.resolve(); + const result = await completion; + expect(Array.isArray(result) ? result.at(-1) : result).toMatchObject({ + finishReason: 'paused', + }); + expect(handler).toHaveBeenCalledTimes(2); + const state = runtime.getState(); + expect(state.status).toBe('suspended'); + const checkpoint = await store.load(runId); + expect(checkpoint).toMatchObject({ + finishReason: 'paused', + pendingToolCalls: {}, + iterationCount: 1, + persistence: options.persistence, + }); + + streamText.mockReset().mockReturnValue(makeStreamSuccess('Finished from the saved results')); + const resumes = await Promise.allSettled( + [runtime, createRuntimeWithCheckpointStore([tool], store)].map( + async (resumeRuntime) => await resumeRuntime.resumePaused({ runId }), + ), + ); + expect(resumes.filter((resume) => resume.status === 'fulfilled')).toHaveLength(1); + for (const resume of resumes) { + if (resume.status === 'rejected') expect(resume.reason).toBeInstanceOf(StaleResumeError); + else + expect((await collectChunks(resume.value.stream)).at(-1)).toMatchObject({ + finishReason: 'stop', + }); + } + expect(runtime.getState()).toMatchObject({ status: 'success', finishReason: undefined }); + expect(handler).toHaveBeenCalledTimes(2); + expect(JSON.stringify(streamText.mock.calls[0][0].messages)).toContain( + 'Saved partial result', + ); + expect(await store.load(runId)).toBeUndefined(); + }, + ); + + it.each([ + ['admission', 'Admission failed'], + ['options', 'Cannot decrease maxIterations'], + ['metadata', 'Cannot update host metadata without persistence'], + ] as const)( + 'pauses before the first model call and permits a retry after %s fails', + async (failure, message) => { + const store = makeClaimingCheckpointStore(); + const runtime = createRuntimeWithCheckpointStore([], store); + const result = await runtime.generate('Start later', { + shouldPause: async () => true, + maxIterations: 10, + persistence: + failure === 'metadata' + ? undefined + : { + threadId: 'child', + resourceId: 'user', + hostMetadata: { actor: 'saved', source: 'checkpoint' }, + }, + }); + expect(result.finishReason).toBe('paused'); + expect(generateText).not.toHaveBeenCalled(); + const original = await store.load(result.runId); + const resumeRuntime = createRuntimeWithCheckpointStore([], store); + await expect( + resumeRuntime.resumePaused({ + runId: result.runId, + maxIterations: failure === 'options' ? 1 : 10, + hostMetadata: { actor: 'selected' }, + onResumeClaimed: async () => { + throw new Error('Admission failed'); + }, + }), + ).rejects.toThrow(message); + const expectedCheckpoint = + failure === 'admission' + ? { + ...original, + persistence: { + threadId: 'child', + resourceId: 'user', + hostMetadata: { actor: 'selected', source: 'checkpoint' }, + }, + } + : original; + expect(await store.load(result.runId)).toEqual(expectedCheckpoint); + if (failure === 'admission') expect(resumeRuntime.getState()).toEqual(expectedCheckpoint); + expect(streamText).not.toHaveBeenCalled(); + + streamText.mockReturnValue(makeStreamSuccess()); + const restarted = createRuntimeWithCheckpointStore([], store); + const resumed = await restarted.resumePaused({ runId: result.runId, maxIterations: 10 }); + expect((await collectChunks(resumed.stream)).at(-1)).toMatchObject({ + finishReason: 'stop', + }); + expect(await store.load(result.runId)).toBeUndefined(); + }, + ); + + it.each(['pause polling', 'checkpoint persistence'] as const)( + 'cancels and removes the checkpoint when abort arrives during %s', + async (phase) => { + const bus = new AgentEventBus(); + const store = makeClaimingCheckpointStore(); + const entered = createDeferredPromise(); + const release = createDeferredPromise(); + const save = store.save; + store.save = async (...args) => { + await save(...args); + if (phase !== 'checkpoint persistence') return; + entered.resolve(); + await release.promise; + }; + const runtime = createRuntimeWithCheckpointStore([], store, bus); + const completion = runtime.generate('Start later', { + shouldPause: async () => { + if (phase === 'pause polling') { + entered.resolve(); + await release.promise; + } + return true; + }, + }); + await entered.promise; + bus.abort(); + release.resolve(); + const result = await completion; + + expect(result.finishReason).toBe('error'); + expect(runtime.getState().status).toBe('cancelled'); + expect(await store.load(result.runId)).toBeUndefined(); + }, + ); + + it.each([ + ['stop', 'success'], + ['error', 'failed'], + ] as const)( + 'keeps the terminal %s result when pause is requested during the model call', + async (finishReason, status) => { + const store = makeClaimingCheckpointStore(); + const runtime = createRuntimeWithCheckpointStore([], store); + let requested = false; + generateText.mockImplementationOnce(async () => { + requested = true; + if (finishReason === 'error') throw new Error('Model failed'); + return makeGenerateSuccess(); + }); + const result = await runtime.generate('Start', { + shouldPause: async () => requested, + onInputBoundary: async () => [], + }); + + expect(result.finishReason).toBe(finishReason); + expect(runtime.getState().status).toBe(status); + expect(await store.load(result.runId)).toBeUndefined(); + }, + ); + + it('preserves an approval reached by the current step', async () => { + const handler = vi.fn(); + const tool = wrapToolForApproval(makeMockTool('approve_work', handler), { + requireApproval: true, + }); + const store = makeClaimingCheckpointStore(); + const runtime = createRuntimeWithCheckpointStore([tool], store); + let requested = false; + generateText.mockImplementationOnce(async () => { + requested = true; + return makeGenerateWithToolCall('approval', tool.name, {}); + }); + const result = await runtime.generate('Start', { shouldPause: async () => requested }); + expect(handler).not.toHaveBeenCalled(); + expect(result.pendingSuspend).toHaveLength(1); + await expect(runtime.resumePaused({ runId: result.runId })).rejects.toBeInstanceOf( + StaleResumeError, + ); + expect((await store.load(result.runId))?.pendingToolCalls.approval.suspended).toBe(true); + }); +}); + describe('AgentRuntime — deferred tool loading', () => { beforeEach(() => { generateText.mockReset(); @@ -6946,31 +7209,58 @@ describe('AgentRuntime.resume() — checkpoint lifecycle', () => { expect((await checkpointStore.load(runId))?.status).toBe('suspended'); }); - it('claims the checkpoint after resume validation passes', async () => { + it('restores a rejected resume with current metadata and permits a retry', async () => { const checkpointStore = makeClaimingCheckpointStore(); - const runtime = createRuntimeWithCheckpointStore([makeApprovalTool()], checkpointStore); - const onResumeClaimed = vi.fn(); + const tool = makeApprovalTool(); + const handler = vi.spyOn(tool, 'handler'); + const runtime = createRuntimeWithCheckpointStore([tool], checkpointStore); + const persistence = { + threadId: 'thread-1', + resourceId: 'resource-1', + hostMetadata: { owner: 'previous', scope: 'retained' }, + }; generateText.mockResolvedValueOnce( makeGenerateWithToolCalls([{ toolCallId: 'tc-1', toolName: 'suspend_tool', args: {} }]), ); - const first = await runtime.generate('run tool'); + const first = await runtime.generate('run tool', { persistence }); const { runId, toolCallId } = first.pendingSuspend![0]; + const checkpoint = await checkpointStore.load(runId); + handler.mockClear(); + const rejectedRuntime = createRuntimeWithCheckpointStore([tool], checkpointStore); + const error = new Error('Resume admission rejected'); + await expect( + rejectedRuntime.resume( + 'stream', + { approved: true }, + { + runId, + toolCallId, + hostMetadata: { owner: 'current' }, + onResumeClaimed: async () => { + expect((await checkpointStore.load(runId))?.status).toBe('running'); + throw error; + }, + }, + ), + ).rejects.toBe(error); + expect(await checkpointStore.load(runId)).toEqual({ + ...checkpoint, + persistence: { + ...persistence, + hostMetadata: { owner: 'current', scope: 'retained' }, + }, + }); + expect(rejectedRuntime.getState().status).toBe('suspended'); + expect(handler).not.toHaveBeenCalled(); + expect(streamText).not.toHaveBeenCalled(); generateText.mockResolvedValueOnce(makeGenerateSuccess('done')); - const resumed = await runtime.resume( - 'generate', - { approved: true }, - { runId, toolCallId, onResumeClaimed }, - ); + const resumed = await runtime.resume('generate', { approved: true }, { runId, toolCallId }); expect(resumed.finishReason).toBe('stop'); - expect(checkpointStore.claimForResume).toHaveBeenCalledTimes(1); - expect(checkpointStore.claimForResume).toHaveBeenCalledWith( - runId, - expect.objectContaining({ status: 'suspended' }), - ); - expect(onResumeClaimed).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledOnce(); + expect(await checkpointStore.load(runId)).toBeUndefined(); }); it('does not emit a runtime error when another resume wins the checkpoint claim', async () => { diff --git a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts index 4dc4a1c1f05..f7de929b014 100644 --- a/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts +++ b/packages/@n8n/agents/src/runtime/loop/agent-runtime.ts @@ -341,6 +341,7 @@ export class AgentRuntime { const { state, list, resumeData } = await this.prepareToolResume(data, options); let abortScope: AgentAbortScope | undefined; let resumeClaimed = false; + let resumeAdmitted = false; try { // Merge persisted execution options with fresh caller options @@ -364,6 +365,7 @@ export class AgentRuntime { }; this.updateState({ persistence: resumeOptions.persistence }); await options.onResumeClaimed?.(); + resumeAdmitted = true; abortScope = this.eventBus.createAbortScope(resumeOptions.abortSignal); const activeAbortScope = abortScope; @@ -385,10 +387,57 @@ export class AgentRuntime { if (method === 'generate') return await this.generateResumedRun(ctx); return this.createResumedStream(ctx); } catch (error) { + if (resumeClaimed && !resumeAdmitted) { + const suspendedState = { ...state, persistence: this.currentState.persistence }; + await this.runState.suspend(this.runId, suspendedState); + this.updateState(suspendedState); + throw error; + } return await this.handleResumeFailure(method, error, abortScope, resumeClaimed); } } + /** Resume a user pause without settling or replaying a tool call. */ + async resumePaused( + options: Omit & ExecutionOptions, + ): Promise { + this.runId = options.runId; + const state = await this.runState.resume(this.runId); + if (!state || state.finishReason !== 'paused' || Object.keys(state.pendingToolCalls).length) { + throw new StaleResumeError('This run is not paused'); + } + if (options.hostMetadata !== undefined && !state.persistence) { + throw new Error('Cannot update host metadata without persistence'); + } + const { runId: _runId, hostMetadata, onResumeClaimed, ...executionOptions } = options; + const resumeOptions: RuntimeExecutionOptions = { + ...mergeResumeExecutionOptions(state, executionOptions), + persistence: mergeResumePersistence(state.persistence, hostMetadata), + }; + const list = await this.restoreCheckpointMessages(state); + if (!(await this.runState.claimResume(this.runId, state))) { + throw new StaleResumeError('This run has already resumed'); + } + this.updateState({ + status: 'running', + finishReason: undefined, + persistence: resumeOptions.persistence, + }); + const abortScope = this.eventBus.createAbortScope(resumeOptions.abortSignal); + try { + await onResumeClaimed?.(); + await this.prepareResumeMemory(list, state.persistence); + return this.createResumedStream({ list, options: resumeOptions, abortScope }); + } catch (error) { + abortScope.dispose(); + // No task action has started. Keep the checkpoint available for a retry. + const suspendedState = { ...state, persistence: resumeOptions.persistence }; + await this.runState.suspend(this.runId, suspendedState); + this.updateState(suspendedState); + throw error; + } + } + /** * Durable-log RFC (resilience phase): re-drive a run from a `running`-status * step checkpoint after a process crash. Unlike resume(), there is no @@ -619,7 +668,7 @@ export class AgentRuntime { result.runId = this.runId; result.usage = this.applyCost(result.usage); result.model = this.modelIdString; - if (!result.pendingSuspend?.length) { + if (!result.pendingSuspend?.length && result.finishReason !== 'paused') { this.updateState({ status: 'success', messageList: list.serialize() }); this.eventBus.emit({ type: AgentEvent.AgentEnd, messages: result.messages }); } @@ -734,6 +783,26 @@ export class AgentRuntime { // consuming it. End the run even if a consumer still returned messages. if (state.guardrailStop) break; if (state.reachedStopCondition && !hasInput) break; + this.assertNotAborted(prepared.abortScope); + if (await prepared.options?.shouldPause?.()) { + await this.persistSuspension( + {}, + prepared.options, + prepared.list, + state.totalUsage, + state.maxIterations, + state.iterationCount, + 'paused', + ); + this.assertNotAborted(prepared.abortScope); + return await sink.finishSuspended({ + suspendRunId: this.runId, + list: prepared.list, + usage: state.totalUsage, + suspensions: [], + finishReason: 'paused', + }); + } state.reachedStopCondition = false; const settlement = await this.runLoopIteration(prepared, sink, state); if (settlement.suspended) return settlement.result; @@ -966,7 +1035,7 @@ export class AgentRuntime { system: prompt.system, messages: prompt.messages, abortSignal: abortScope.signal, - hasTools: tools.hasTools, + hasTools: options?.toolsEnabled !== false && tools.hasTools, aiTools: prompt.aiTools, reasoning: staticContext.reasoning, providerOptions: staticContext.providerOptions, @@ -1217,6 +1286,7 @@ export class AgentRuntime { totalUsage: TokenUsage | undefined, maxIterations?: number, iterationCount?: number, + finishReason?: 'paused', ): Promise { const checkpointOptions = buildCheckpointOptions(options, maxIterations, iterationCount); @@ -1228,10 +1298,16 @@ export class AgentRuntime { messageList: list.serialize(), pendingToolCalls, usage: totalUsage, + finishReason, ...checkpointOptions, }; await this.runState.suspend(this.runId, state); - this.updateState({ status: 'suspended', pendingToolCalls, messageList: list.serialize() }); + this.updateState({ + status: 'suspended', + pendingToolCalls, + messageList: list.serialize(), + finishReason, + }); await this.memory.persistTurnDelta(list, options); } diff --git a/packages/@n8n/agents/src/runtime/loop/generate-sink.ts b/packages/@n8n/agents/src/runtime/loop/generate-sink.ts index d793acc4d62..c0a69612251 100644 --- a/packages/@n8n/agents/src/runtime/loop/generate-sink.ts +++ b/packages/@n8n/agents/src/runtime/loop/generate-sink.ts @@ -78,7 +78,7 @@ export class GenerateSink implements RunOutputSink { return { runId: suspendRunId, messages: list.responseDelta(), - finishReason: 'tool-calls', + finishReason: emission.finishReason ?? 'tool-calls', usage, pendingSuspend: suspensions.map((s) => ({ runId: suspendRunId, diff --git a/packages/@n8n/agents/src/runtime/loop/stream-sink.ts b/packages/@n8n/agents/src/runtime/loop/stream-sink.ts index 4edb93c1e3c..3382f36d0e7 100644 --- a/packages/@n8n/agents/src/runtime/loop/stream-sink.ts +++ b/packages/@n8n/agents/src/runtime/loop/stream-sink.ts @@ -454,7 +454,7 @@ export class StreamSink implements RunOutputSink { const costUsage = this.services.applyCost(emission.usage); await this.guard.write({ type: 'finish', - finishReason: 'tool-calls', + finishReason: emission.finishReason ?? 'tool-calls', ...(costUsage && { usage: costUsage }), model: this.services.modelId, }); diff --git a/packages/@n8n/agents/src/sdk/agent.ts b/packages/@n8n/agents/src/sdk/agent.ts index ebc68573865..5c774beb417 100644 --- a/packages/@n8n/agents/src/sdk/agent.ts +++ b/packages/@n8n/agents/src/sdk/agent.ts @@ -837,6 +837,20 @@ export class Agent implements BuiltAgent, AgentBuilder { } } + async resumePaused( + options: Omit & ExecutionOptions, + ): Promise { + const config = await this.ensureBuilt(); + const active = this.createRuntime(config, options.runId); + try { + const result = await active.runtime.resumePaused(options); + return { ...result, stream: this.trackStreamRuntime(result.stream, active) }; + } catch (error) { + await this.cleanupRuntime(active); + throw error; + } + } + /** * Durable-log RFC (resilience phase): re-drive a run from a `running`-status * step checkpoint after a process crash. There is no pending tool call to diff --git a/packages/@n8n/agents/src/types/runtime/agent-loop.ts b/packages/@n8n/agents/src/types/runtime/agent-loop.ts index bd03dd5ca20..e4dd070895d 100644 --- a/packages/@n8n/agents/src/types/runtime/agent-loop.ts +++ b/packages/@n8n/agents/src/types/runtime/agent-loop.ts @@ -75,6 +75,7 @@ export interface ModelCallContext { /** Data needed to emit a terminal suspension result. */ export interface SuspendEmission { + finishReason?: 'paused'; suspendRunId: string; list: AgentMessageList; usage: TokenUsage | undefined; diff --git a/packages/@n8n/agents/src/types/sdk/agent.ts b/packages/@n8n/agents/src/types/sdk/agent.ts index 04047018a01..cbef9f8fa0e 100644 --- a/packages/@n8n/agents/src/types/sdk/agent.ts +++ b/packages/@n8n/agents/src/types/sdk/agent.ts @@ -32,6 +32,7 @@ export const FINISH_REASONS = [ 'tool-calls', 'error', 'other', + 'paused', ] as const; export type FinishReason = (typeof FINISH_REASONS)[number]; @@ -222,6 +223,10 @@ export interface AgentInputBoundary { } export interface ExecutionOptions { + /** Expose local and provider tools to the model. Defaults to true. */ + toolsEnabled?: boolean; + /** Request a cooperative pause before the next model step. */ + shouldPause?: () => Promise; /** Commit additional input between model calls. Stream consumers must acknowledge input-boundary chunks. */ onInputBoundary?: (boundary: AgentInputBoundary) => Promise; maxIterations?: number; @@ -434,6 +439,11 @@ export interface BuiltAgent { options: ResumeOptions & ExecutionOptions, ): Promise; + /** Resume a user pause without repeating completed tools. */ + resumePaused( + options: Omit & ExecutionOptions, + ): Promise; + /** Approve a tool that uses requireApproval or needsApprovalFn */ approve(method: 'generate', options: ResumeOptions & ExecutionOptions): Promise; approve(method: 'stream', options: ResumeOptions & ExecutionOptions): Promise; diff --git a/packages/@n8n/api-types/src/agents/background-job.ts b/packages/@n8n/api-types/src/agents/background-job.ts index 4fac2e1f4e9..4d5778073fc 100644 --- a/packages/@n8n/api-types/src/agents/background-job.ts +++ b/packages/@n8n/api-types/src/agents/background-job.ts @@ -5,7 +5,7 @@ export interface AgentBackgroundJobDto { title: string; /** Source of the background job. */ kind: 'subagent' | 'workflow'; - status: 'running' | 'suspended' | 'completed' | 'failed' | 'cancelled'; + status: 'running' | 'suspended' | 'paused' | 'completed' | 'failed' | 'cancelled'; /** The next approval for this child. The parent conversation remains available. */ approval?: AgentBuilderOpenSuspension; /** ISO timestamp when the background job started. */ @@ -17,7 +17,7 @@ export interface AgentBackgroundJobDto { export interface AgentBackgroundJobSignal { tasks: Array< Pick & { - status: Exclude; + status: Exclude; } >; } diff --git a/packages/@n8n/db/src/migrations/common/1790937677019-AddAgentBackgroundJobPause.ts b/packages/@n8n/db/src/migrations/common/1790937677019-AddAgentBackgroundJobPause.ts new file mode 100644 index 00000000000..259d7ac4257 --- /dev/null +++ b/packages/@n8n/db/src/migrations/common/1790937677019-AddAgentBackgroundJobPause.ts @@ -0,0 +1,60 @@ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +export class AddAgentBackgroundJobPause1790937677019 implements ReversibleMigration { + async up({ + schemaBuilder: { addColumns, column, dropEnumCheck, addEnumCheck }, + escape, + runQuery, + isPostgres, + }: MigrationContext) { + await addColumns( + 'agent_background_job', + [column('pauseRequestId').uuid.comment('Groups jobs stopped by one user request')], + { recreatesOnSqlite: true }, + ); + await dropEnumCheck('agent_background_job', 'status', { recreatesOnSqlite: true }); + await addEnumCheck( + 'agent_background_job', + 'status', + ['running', 'suspended', 'paused', 'completed', 'failed', 'cancelled'], + { recreatesOnSqlite: true }, + ); + if (isPostgres) { + await runQuery( + `COMMENT ON COLUMN ${escape.tableName('agent_background_job')}.${escape.columnName('status')} IS 'running: child works; suspended: child awaits approval; paused: user stopped the child; completed, failed, cancelled: terminal'`, + ); + } + } + + async down({ + schemaBuilder: { dropColumns, dropEnumCheck, addEnumCheck }, + escape, + runQuery, + isPostgres, + }: MigrationContext) { + const table = escape.tableName('agent_background_job'); + const status = escape.columnName('status'); + const pausedJobs = await runQuery>( + `SELECT ${escape.columnName('id')} FROM ${table} WHERE ${status} = 'paused' LIMIT 1`, + ); + if (pausedJobs.length > 0) { + throw new Error( + 'Cannot revert background job pause support while paused jobs exist. Resume or cancel the jobs before you retry.', + ); + } + + await dropEnumCheck('agent_background_job', 'status', { recreatesOnSqlite: true }); + await addEnumCheck( + 'agent_background_job', + 'status', + ['running', 'suspended', 'completed', 'failed', 'cancelled'], + { recreatesOnSqlite: true }, + ); + await dropColumns('agent_background_job', ['pauseRequestId'], { recreatesOnSqlite: true }); + if (isPostgres) { + await runQuery( + `COMMENT ON COLUMN ${table}.${status} IS 'running: child works; suspended: child waits for an approval; completed, failed, cancelled: terminal'`, + ); + } + } +} diff --git a/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts b/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts index 08e9b9a5fe3..7fcf994fac0 100644 --- a/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts +++ b/packages/@n8n/instance-ai/evaluations/discovery/__tests__/stream-status.test.ts @@ -58,9 +58,12 @@ describe('resolveStreamStatus', () => { expect(resolveStreamStatus('timed-out', false)).toBe('timed-out'); }); - it('reports a suspended run as suspended', () => { - expect(resolveStreamStatus(streamResult({ status: 'suspended' }), false)).toBe('suspended'); - }); + it.each([streamResult({ status: 'suspended' }), streamResult({ finishReason: 'paused' })])( + 'reports a suspended or paused run as suspended', + (result) => { + expect(resolveStreamStatus(result, false)).toBe('suspended'); + }, + ); it('never reads a cancelled run as a clean finish', () => { expect(resolveStreamStatus(streamResult({ status: 'cancelled' }), false)).toBe('timed-out'); diff --git a/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts b/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts index f7a13f18786..04a975d509f 100644 --- a/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts +++ b/packages/@n8n/instance-ai/evaluations/discovery/stream-status.ts @@ -21,6 +21,8 @@ function statusForFinishReason(finishReason: FinishReason): DiscoveryStreamStatu return 'completed'; case 'max-iterations': return 'step-exhausted'; + case 'paused': + return 'suspended'; case 'tool-calls': case 'length': case 'content-filter': diff --git a/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts b/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts index e5b777dd9a1..8dedc72f80d 100644 --- a/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts +++ b/packages/cli/src/modules/agents/background/__tests__/agent-background-job.service.test.ts @@ -76,8 +76,7 @@ function setup(options: { backgroundTasksEnabled?: boolean } = {}) { }); (logger.scoped as Mock).mockReturnValue(logger); - jobRepository.countActiveSubAgentsByParentThread.mockResolvedValue(0); - jobRepository.insertJob.mockResolvedValue(undefined); + jobRepository.insertSubAgentJobIfCapacity.mockResolvedValue(true); jobRepository.insertWorkflowJobOrGetExisting.mockResolvedValue({ inserted: true }); jobRepository.settleIfActive.mockResolvedValue(true); jobRepository.findByParentThread.mockResolvedValue([]); @@ -150,14 +149,12 @@ describe('background task notifications', () => { 'agent-1', 'thread-1', ); - expect(jobRepository.insertJob.mock.invocationCallOrder[0]).toBeLessThan( + expect(jobRepository.insertSubAgentJobIfCapacity.mock.invocationCallOrder[0]).toBeLessThan( updateBroadcaster.notifyBackgroundJobsUpdated.mock.invocationCallOrder[0], ); - jobRepository.countActiveSubAgentsByParentThread.mockResolvedValueOnce( - MAX_RUNNING_JOBS_PER_THREAD, - ); + jobRepository.insertSubAgentJobIfCapacity.mockResolvedValueOnce(false); await service.registerSubAgentJob(registerParams); - jobRepository.insertJob.mockRejectedValueOnce(new Error('insert failed')); + jobRepository.insertSubAgentJobIfCapacity.mockRejectedValueOnce(new Error('insert failed')); await expect(service.registerSubAgentJob(registerParams)).rejects.toThrow('insert failed'); expect(updateBroadcaster.notifyBackgroundJobsUpdated).toHaveBeenCalledOnce(); }); @@ -226,30 +223,23 @@ describe('registerSubAgentJob', () => { const receipt = await service.registerSubAgentJob(registerParams); expect(receipt).toEqual({ status: 'started', jobId: 'job-1' }); - const inserted = jobRepository.insertJob.mock.calls[0][0]; + const inserted = jobRepository.insertSubAgentJobIfCapacity.mock.calls[0][0]; if (inserted.kind !== 'subagent') throw new Error('expected a subagent job insert'); const expectedTimeout = Date.now() + SUB_AGENT_BACKGROUND_TIMEOUT_MS; expect(inserted.timeoutAt.getTime()).toBeGreaterThan(expectedTimeout - 5000); expect(inserted.timeoutAt.getTime()).toBeLessThanOrEqual(expectedTimeout); + expect(jobRepository.insertSubAgentJobIfCapacity.mock.calls[0][1]).toBe( + MAX_RUNNING_JOBS_PER_THREAD, + ); }); it('returns limit-reached when the thread is at the running-job cap', async () => { const { service, jobRepository } = setup(); - jobRepository.countActiveSubAgentsByParentThread.mockResolvedValue(MAX_RUNNING_JOBS_PER_THREAD); + jobRepository.insertSubAgentJobIfCapacity.mockResolvedValue(false); const receipt = await service.registerSubAgentJob(registerParams); expect(receipt).toEqual({ status: 'limit-reached' }); - expect(jobRepository.insertJob).not.toHaveBeenCalled(); - }); - - it('counts only running sub-agent jobs toward the cap', async () => { - const { service, jobRepository } = setup(); - - const receipt = await service.registerSubAgentJob(registerParams); - - expect(receipt).toEqual({ status: 'started', jobId: 'job-1' }); - expect(jobRepository.countActiveSubAgentsByParentThread).toHaveBeenCalledWith('thread-1'); }); }); diff --git a/packages/cli/src/modules/agents/background/agent-background-job.service.ts b/packages/cli/src/modules/agents/background/agent-background-job.service.ts index d1e813dfba1..f4b3bd6c10b 100644 --- a/packages/cli/src/modules/agents/background/agent-background-job.service.ts +++ b/packages/cli/src/modules/agents/background/agent-background-job.service.ts @@ -153,16 +153,15 @@ export class AgentBackgroundJobService { async registerSubAgentJob( params: Omit, ): Promise { - const running = await this.jobRepository.countActiveSubAgentsByParentThread( - params.parentThreadId, + const inserted = await this.jobRepository.insertSubAgentJobIfCapacity( + { + ...params, + kind: 'subagent', + timeoutAt: new Date(Date.now() + SUB_AGENT_BACKGROUND_TIMEOUT_MS), + }, + MAX_RUNNING_JOBS_PER_THREAD, ); - if (running >= MAX_RUNNING_JOBS_PER_THREAD) return { status: 'limit-reached' }; - - await this.jobRepository.insertJob({ - ...params, - kind: 'subagent', - timeoutAt: new Date(Date.now() + SUB_AGENT_BACKGROUND_TIMEOUT_MS), - }); + if (!inserted) return { status: 'limit-reached' }; this.updateBroadcaster.notifyBackgroundJobsUpdated(params.parentAgentId, params.parentThreadId); return { status: 'started', jobId: params.id }; diff --git a/packages/cli/src/modules/agents/background/agent-wake.service.ts b/packages/cli/src/modules/agents/background/agent-wake.service.ts index 1e7a5d1285e..091ea1a8b44 100644 --- a/packages/cli/src/modules/agents/background/agent-wake.service.ts +++ b/packages/cli/src/modules/agents/background/agent-wake.service.ts @@ -306,7 +306,9 @@ export class AgentWakeService { message: formatWakeMessage(jobs), backgroundJobSignal: { tasks: jobs.flatMap(({ id, title, kind, status }) => - status === 'running' || status === 'suspended' ? [] : [{ id, title, kind, status }], + status === 'running' || status === 'suspended' || status === 'paused' + ? [] + : [{ id, title, kind, status }], ), }, memory: { threadId, resourceId }, diff --git a/packages/cli/src/modules/agents/entities/agent-background-job.entity.ts b/packages/cli/src/modules/agents/entities/agent-background-job.entity.ts index 99d5442fba8..d32e7f1409d 100644 --- a/packages/cli/src/modules/agents/entities/agent-background-job.entity.ts +++ b/packages/cli/src/modules/agents/entities/agent-background-job.entity.ts @@ -26,6 +26,9 @@ export class AgentBackgroundJob extends WithTimestampsAndStringId { @Column({ type: 'varchar', length: 16 }) status: AgentBackgroundJobStatus; + @Column({ type: 'uuid', nullable: true, comment: 'Groups jobs stopped by one user request' }) + pauseRequestId: string | null; + @Column({ type: 'varchar', length: 36 }) parentAgentId: string; diff --git a/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts b/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts index 1ef3ac36d0f..dcd655cdf9b 100644 --- a/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts +++ b/packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts @@ -302,6 +302,25 @@ export class N8NCheckpointStorage { } } + async markUserPaused( + agentId: string, + suspension: { + runId: string; + checkpoint: SerializableAgentState; + serializedState: string; + updatedAt: Date; + }, + ): Promise { + // Preserve the approval and its expiry. A concurrent resume must win or lose atomically. + return await this.agentCheckpointRepository.markUserPaused( + suspension.runId, + agentId, + suspension.serializedState, + JSON.stringify({ ...suspension.checkpoint, finishReason: 'paused' }), + suspension.updatedAt, + ); + } + private async deleteDelegation( runId: string, agentId: string, 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 6a453a9d677..67be3cdd62e 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 @@ -1,6 +1,7 @@ +import { BaseRepository, TransactionRunner, type OperationContext } from '@n8n/db'; import { Service } from '@n8n/di'; -import { DataSource, In, IsNull, LessThan, Not, Repository } from '@n8n/typeorm'; -import { OperationalError } from 'n8n-workflow'; +import { DataSource, In, IsNull, LessThan, Not } from '@n8n/typeorm'; +import { OperationalError, UserError } from 'n8n-workflow'; import { AgentBackgroundJob, @@ -8,6 +9,8 @@ import { type AgentBackgroundJobStatus, } from '../entities/agent-background-job.entity'; import { AgentCheckpoint } from '../entities/agent-checkpoint.entity'; +import { AgentExecution } from '../entities/agent-execution.entity'; +import { AgentExecutionThreadRepository } from './agent-execution-thread.repository'; type NewAgentBackgroundJobBase = { id: string; @@ -31,17 +34,16 @@ export type NewWorkflowJob = NewAgentBackgroundJobBase & { childExecutionId: string; }; -export type NewAgentBackgroundJob = NewSubAgentJob | NewWorkflowJob; - export type AgentBackgroundJobSettlement = { - status: Exclude; + status: Exclude; result?: string | null; error?: string | null; }; export type ExpectedBackgroundJobState = { - status: 'running' | 'suspended'; + status: 'running' | 'suspended' | 'paused'; timeoutAt?: Date | null; + pauseRequestId?: string | null; }; export type BackgroundJobGroupItem = Pick< @@ -50,13 +52,25 @@ export type BackgroundJobGroupItem = Pick< >; @Service() -export class AgentBackgroundJobRepository extends Repository { - constructor(dataSource: DataSource) { - super(AgentBackgroundJob, dataSource.manager); +export class AgentBackgroundJobRepository extends BaseRepository { + constructor( + dataSource: DataSource, + transactionRunner: TransactionRunner, + private readonly threadRepository: AgentExecutionThreadRepository, + ) { + super(AgentBackgroundJob, dataSource.manager, transactionRunner); } - async insertJob(job: NewAgentBackgroundJob): Promise { - await this.insert({ ...job, status: 'running' }); + async insertSubAgentJobIfCapacity(job: NewSubAgentJob, limit: number): Promise { + return await this.runInTransaction({}, async (manager, ctx) => { + if (!(await this.threadRepository.lockById(job.parentThreadId, ctx))) { + throw new UserError('Session not found'); + } + if ((await this.countActiveSubAgentsByParentThread(job.parentThreadId, ctx)) >= limit) + return false; + await manager.insert(AgentBackgroundJob, { ...job, status: 'running' }); + return true; + }); } /** @@ -83,8 +97,11 @@ export class AgentBackgroundJobRepository extends Repository } /** Suspended children still occupy a job slot. */ - async countActiveSubAgentsByParentThread(parentThreadId: string): Promise { - return await this.count({ + async countActiveSubAgentsByParentThread( + parentThreadId: string, + ctx: OperationContext = {}, + ): Promise { + return await this.managerFor(ctx).count(AgentBackgroundJob, { where: { parentThreadId, kind: 'subagent', status: In(['running', 'suspended']) }, }); } @@ -113,25 +130,44 @@ export class AgentBackgroundJobRepository extends Repository /** Results and approval requests that still need delivery. */ async findWakeableUnconsumed(parentThreadId: string): Promise { + const activePauseGroup = this.createQueryBuilder() + .subQuery() + .select('1') + .from(AgentBackgroundJob, 'member') + .leftJoin( + AgentExecution, + 'execution', + "execution.threadId = member.childThreadId AND execution.status = 'running'", + ) + .where('member.pauseRequestId = job.pauseRequestId') + .andWhere('member.parentThreadId = job.parentThreadId') + .andWhere("(member.status IN ('running', 'suspended') OR execution.id IS NOT NULL)") + .getQuery(); return await this.createQueryBuilder('job') .where('job.parentThreadId = :parentThreadId', { parentThreadId }) .andWhere("job.status <> 'running'") .andWhere('job.notifiedAt IS NULL') + .andWhere(`(job.pauseRequestId IS NULL OR NOT EXISTS ${activePauseGroup})`) .orderBy('COALESCE(job.settledAt, job.updatedAt)', 'ASC') .addOrderBy('job.createdAt', 'ASC') .getMany(); } - async markMailConsumed(parentThreadId: string, ids: string[]): Promise { + async markMailConsumed( + parentThreadId: string, + ids: string[], + includePauseReports = false, + ): Promise { if (ids.length === 0) return 0; - const result = await this.createQueryBuilder() + const query = this.createQueryBuilder() .update() .set({ notifiedAt: new Date() }) .where({ parentThreadId, id: In(ids) }) .andWhere('settledAt IS NOT NULL') - .andWhere('notifiedAt IS NULL') - .execute(); + .andWhere('notifiedAt IS NULL'); + if (!includePauseReports) query.andWhere('pauseRequestId IS NULL'); + const result = await query.execute(); return result.affected ?? 0; } @@ -179,6 +215,17 @@ export class AgentBackgroundJobRepository extends Repository return await this.find({ where: kind ? { status: 'running', kind } : { status: 'running' } }); } + async findRequestedPauses(parentThreadId?: string): Promise { + return await this.find({ + where: { + kind: 'subagent', + pauseRequestId: Not(IsNull()), + status: In(['running', 'suspended', 'paused']), + ...(parentThreadId ? { parentThreadId } : {}), + }, + }); + } + async findSettledSubAgentsWithCheckpoints(): Promise { return await this.createQueryBuilder('job') .innerJoin( @@ -187,12 +234,33 @@ export class AgentBackgroundJobRepository extends Repository 'checkpoint.agentId = job.subAgentId AND checkpoint.threadId = job.childThreadId', ) .where("job.kind = 'subagent'") - .andWhere("job.status NOT IN ('running', 'suspended')") + .andWhere("job.status NOT IN ('running', 'suspended', 'paused')") .andWhere('checkpoint.state IS NOT NULL') .distinct(true) .getMany(); } + async findPausedWithoutCheckpoint( + updatedAfter: Date, + parentThreadId?: string, + ): Promise { + const checkpoint = this.createQueryBuilder() + .subQuery() + .select('1') + .from(AgentCheckpoint, 'checkpoint') + .where('checkpoint.agentId = job.subAgentId') + .andWhere('checkpoint.threadId = job.childThreadId') + .andWhere('checkpoint.expired = :expired') + .andWhere('checkpoint.state IS NOT NULL') + .andWhere('checkpoint.updatedAt > :updatedAfter') + .getQuery(); + const query = this.createQueryBuilder('job') + .where({ kind: 'subagent', status: 'paused' }) + .andWhere(`NOT EXISTS ${checkpoint}`, { expired: false, updatedAfter }); + if (parentThreadId) query.andWhere('job.parentThreadId = :parentThreadId', { parentThreadId }); + return await query.getMany(); + } + /** Active jobs whose execution or approval deadline has passed. */ async findActivePastTimeout(now: Date): Promise { return await this.find({ @@ -217,12 +285,73 @@ export class AgentBackgroundJobRepository extends Repository async resumeIfSuspended(id: string, timeoutAt: Date): Promise { const result = await this.update( - { id, status: 'suspended' }, + { id, status: 'suspended', pauseRequestId: IsNull() }, { status: 'running', timeoutAt, notifiedAt: null }, ); return result.affected === 1; } + async requestPause( + parentAgentId: string, + parentThreadId: string, + parentResourceId: string, + pauseRequestId: string, + ): Promise { + await this.update( + { + parentAgentId, + parentThreadId, + parentResourceId, + kind: 'subagent', + status: In(['running', 'suspended']), + pauseRequestId: IsNull(), + }, + { pauseRequestId, notifiedAt: null }, + ); + } + + async pauseIfRequested( + job: AgentBackgroundJob, + resultText: string, + runId: string, + state: string, + ): Promise { + const running = this.createQueryBuilder() + .subQuery() + .select('1') + .from(AgentExecution, 'execution') + .where('execution.threadId = :childThreadId') + .andWhere("execution.status = 'running'") + .getQuery(); + const result = await this.createQueryBuilder() + .update() + .set({ + status: 'paused', + result: resultText, + timeoutAt: null, + settledAt: new Date(), + notifiedAt: null, + }) + .where({ id: job.id, status: In(['running', 'suspended']), pauseRequestId: Not(IsNull()) }) + .andWhere(this.matchCheckpoint(), { runId, state, expired: false }) + .andWhere(`NOT EXISTS ${running}`, { childThreadId: job.childThreadId }) + .execute(); + return result.affected === 1; + } + + async resumeIfPaused( + id: string, + pauseRequestId: string, + status: 'running' | 'suspended', + timeoutAt: Date, + ): Promise { + const result = await this.update( + { id, status: 'paused', pauseRequestId, notifiedAt: Not(IsNull()) }, + { status, pauseRequestId: null, timeoutAt, settledAt: null, notifiedAt: null, result: null }, + ); + return result.affected === 1; + } + /** * Settle the job if it is still active. Every writer that ends a job — * child settle, cancel, timeout, sweeper reconciliation — funnels through @@ -236,15 +365,21 @@ export class AgentBackgroundJobRepository extends Repository const result = await this.update( { id, - status: expected?.status ?? In(['running', 'suspended']), + status: expected?.status ?? In(['running', 'suspended', 'paused']), ...(expected?.timeoutAt !== undefined ? { timeoutAt: expected.timeoutAt ?? IsNull() } : {}), + ...(expected?.pauseRequestId !== undefined + ? { pauseRequestId: expected.pauseRequestId ?? IsNull() } + : {}), }, { status: settlement.status, result: settlement.result ?? null, error: settlement.error ?? null, settledAt: new Date(), - notifiedAt: null, + notifiedAt: + settlement.status === 'cancelled' || expected?.status === 'paused' + ? () => 'CASE WHEN "status" = \'paused\' THEN "notifiedAt" ELSE NULL END' + : null, }, ); return result.affected === 1; diff --git a/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts b/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts index efa039676c7..6452495b41d 100644 --- a/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-checkpoint.repository.ts @@ -80,6 +80,20 @@ export class AgentCheckpointRepository extends BaseRepository { return (result.affected ?? 0) > 0; } + async markUserPaused( + runId: string, + agentId: string, + state: string, + pausedState: string, + updatedAt: Date, + ): Promise { + const result = await this.update( + { runId, agentId, state, expired: false, updatedAt }, + { state: pausedState, updatedAt }, + ); + return result.affected === 1; + } + async expireByRunIdAndAgentId( runId: string, agentId: string, 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 3055a33a5d9..0e691ef8c27 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 @@ -1,3 +1,4 @@ +import type { SerializableAgentState } from '@n8n/agents'; import { LockService, type Logger } from '@n8n/backend-common'; import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils'; import type { AgentsConfig } from '@n8n/config'; @@ -17,11 +18,14 @@ import { AgentBackgroundJobService } from '@/modules/agents/background/agent-bac import { AgentWakeService, WAKE_DEBOUNCE_MS } from '@/modules/agents/background/agent-wake.service'; import type { AgentBackgroundJob } from '@/modules/agents/entities/agent-background-job.entity'; import type { Agent } from '@/modules/agents/entities/agent.entity'; -import type { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage'; +import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage'; import type { ChatIntegrationRegistry } from '@/modules/agents/integrations/agent-chat-integration'; import { AgentBackgroundJobRepository } from '@/modules/agents/repositories/agent-background-job.repository'; import type { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository'; import { AgentRepository } from '@/modules/agents/repositories/agent.repository'; +import { AgentCheckpointRepository } from '@/modules/agents/repositories/agent-checkpoint.repository'; +import { AgentExecutionThreadRepository } from '@/modules/agents/repositories/agent-execution-thread.repository'; +import { AgentExecution } from '@/modules/agents/entities/agent-execution.entity'; import { createOwner } from '../../shared/db/users'; @@ -77,6 +81,94 @@ describe('AgentBackgroundJobRepository', () => { }); } + it('admits five active tasks beside a paused group and fences concurrent spawns', async () => { + const parentThreadId = uuid(); + const agent = await agentRepository.findOneByOrFail({ id: agentId }); + await Container.get(AgentExecutionThreadRepository).insert({ + id: parentThreadId, + projectId: agent.projectId, + agentId, + agentName: agent.name, + }); + for (let index = 0; index < 5; index++) { + await insertJob({ id: uuid(), parentThreadId, status: 'paused', pauseRequestId: agentId }); + } + await insertJob({ id: uuid(), parentThreadId, kind: 'workflow', status: 'running' }); + const jobs = Array.from({ length: 6 }, () => ({ + id: uuid(), + kind: 'subagent' as const, + parentAgentId: agentId, + parentThreadId, + parentResourceId: 'draft-chat:user-1', + parentPrincipalHash: 'principal-hash', + title: 'New task', + subAgentId: agentId, + childThreadId: uuid(), + timeoutAt: new Date(Date.now() + 60_000), + })); + const admitted = await Promise.all( + jobs.map(async (job) => await repository.insertSubAgentJobIfCapacity(job, 5)), + ); + expect(admitted.filter(Boolean)).toHaveLength(5); + const active = (await repository.findByParentThread(parentThreadId)).filter( + (job) => job.kind === 'subagent' && job.status === 'running', + ); + await repository.update(active[0].id, { status: 'suspended' }); + expect(await repository.countActiveSubAgentsByParentThread(parentThreadId)).toBe(5); + expect(await repository.insertSubAgentJobIfCapacity({ ...jobs[0], id: uuid() }, 5)).toBe(false); + }); + + it('finds expired paused tasks and preserves their report state during guarded cleanup', async () => { + const checkpoints = Container.get(AgentCheckpointRepository); + const pauseRequestId = uuid(); + const notifiedAt = new Date('2026-10-01T10:00:00Z'); + const cutoff = new Date('2026-10-01T00:00:00Z'); + const ids: string[] = []; + for (const checkpoint of [ + undefined, + { expired: true, state: 'saved', updatedAt: new Date('2026-10-02T00:00:00Z') }, + { expired: false, state: null, updatedAt: new Date('2026-10-02T00:00:00Z') }, + { expired: false, state: 'saved', updatedAt: new Date('2026-09-30T00:00:00Z') }, + { expired: false, state: 'saved', updatedAt: new Date('2026-10-02T00:00:00Z') }, + ]) { + const id = uuid(); + const childThreadId = uuid(); + ids.push(id); + await insertJob({ + id, + parentThreadId: 'parent', + subAgentId: agentId, + childThreadId, + status: 'paused', + pauseRequestId, + notifiedAt, + }); + if (checkpoint) + await checkpoints.insert({ + runId: uuid(), + agentId, + threadId: childThreadId, + ...checkpoint, + }); + } + await insertJob({ id: uuid(), parentThreadId: 'other', status: 'paused' }); + expect( + (await repository.findPausedWithoutCheckpoint(cutoff, 'parent')).map((job) => job.id).sort(), + ).toEqual(ids.slice(0, 4).sort()); + await repository.update(ids[0], { status: 'running' }); + await repository.update(ids[1], { pauseRequestId: uuid() }); + for (let index = 0; index < 3; index++) { + expect( + await repository.settleIfActive( + ids[index], + { status: 'failed', error: 'Checkpoint expired' }, + { status: 'paused', pauseRequestId }, + ), + ).toBe(index === 2); + } + expect(await repository.findById(ids[2])).toMatchObject({ status: 'failed', notifiedAt }); + }); + it('returns only group fields for the selected agent and thread', async () => { const id = uuid(); const createdAt = new Date('2026-09-01T10:00:00Z'); @@ -144,6 +236,105 @@ describe('AgentBackgroundJobRepository', () => { expect(remaining.map((job) => job.id)).toEqual([suspendedId]); }); + it('keeps one stop group until every selected job settles and the combined report is delivered', async () => { + const first = uuid(); + const second = uuid(); + const workflow = uuid(); + const otherThread = uuid(); + const pauseRequestId = uuid(); + await insertJob({ id: first, parentThreadId: 'parent', status: 'running', settledAt: null }); + await insertJob({ id: second, parentThreadId: 'parent', status: 'suspended', settledAt: null }); + await insertJob({ + id: workflow, + parentThreadId: 'parent', + kind: 'workflow', + status: 'running', + settledAt: null, + }); + await insertJob({ + id: otherThread, + parentThreadId: 'other', + status: 'running', + settledAt: null, + }); + await repository.requestPause(agentId, 'parent', 'draft-chat:user-1', pauseRequestId); + await repository.requestPause(agentId, 'parent', 'draft-chat:user-1', uuid()); + expect( + (await repository.findByParentThread('parent')) + .filter((job) => job.pauseRequestId === pauseRequestId) + .map((job) => job.id) + .sort(), + ).toEqual([first, second].sort()); + expect((await repository.findById(workflow))?.pauseRequestId).toBeNull(); + expect((await repository.findById(otherThread))?.pauseRequestId).toBeNull(); + expect(await repository.resumeIfSuspended(second, new Date())).toBe(false); + expect(await repository.findById(second)).toMatchObject({ + status: 'suspended', + pauseRequestId, + timeoutAt: null, + }); + + const later = uuid(); + await insertJob({ id: later, parentThreadId: 'parent', status: 'suspended', settledAt: null }); + expect(await repository.resumeIfSuspended(later, new Date())).toBe(true); + await repository.update(first, { status: 'paused', settledAt: new Date(), timeoutAt: null }); + expect(await repository.findWakeableUnconsumed('parent')).toEqual([]); + await repository.settleIfActive(second, { status: 'completed', result: 'Won the race' }); + expect((await repository.findWakeableUnconsumed('parent')).map((job) => job.id).sort()).toEqual( + [first, second].sort(), + ); + expect(await repository.markMailConsumed('parent', [first, second])).toBe(0); + expect(await repository.resumeIfPaused(first, pauseRequestId, 'running', new Date())).toBe( + false, + ); + expect(await repository.markMailConsumed('parent', [first, second], true)).toBe(2); + expect(await repository.markMailConsumed('parent', [first, second], true)).toBe(0); + expect(await repository.findWakeableUnconsumed('parent')).toEqual([]); + expect(await repository.countActiveSubAgentsByParentThread('parent')).toBe(1); + await repository.deleteSettledBefore(new Date(Date.now() + 60_000)); + expect((await repository.findById(first))?.status).toBe('paused'); + const deadline = new Date(Date.now() + 30 * 60_000); + expect(await repository.resumeIfPaused(first, pauseRequestId, 'running', deadline)).toBe(true); + expect(await repository.resumeIfPaused(first, pauseRequestId, 'running', deadline)).toBe(false); + expect(await repository.findById(first)).toMatchObject({ + status: 'running', + pauseRequestId: null, + timeoutAt: deadline, + settledAt: null, + notifiedAt: null, + }); + }); + + it.each([false, true])( + 'keeps the pause report delivery state when cancelling a paused job (delivered: %s)', + async (delivered) => { + const cancelledId = uuid(); + const remainingId = uuid(); + const pauseRequestId = uuid(); + for (const id of [cancelledId, remainingId]) { + await insertJob({ id, parentThreadId: 'parent', status: 'paused', pauseRequestId }); + } + if (delivered) { + await repository.markMailConsumed('parent', [cancelledId, remainingId], true); + } + const notifiedAt = (await repository.findById(cancelledId))?.notifiedAt; + + expect(await repository.settleIfActive(cancelledId, { status: 'cancelled' })).toBe(true); + expect(await repository.markMailConsumed('parent', [cancelledId])).toBe(0); + expect(await repository.findById(cancelledId)).toMatchObject({ + status: 'cancelled', + pauseRequestId, + notifiedAt, + }); + expect( + (await repository.findWakeableUnconsumed('parent')).map((job) => job.id).sort(), + ).toEqual(delivered ? [] : [cancelledId, remainingId].sort()); + expect( + await repository.resumeIfPaused(remainingId, pauseRequestId, 'running', new Date()), + ).toBe(delivered); + }, + ); + it('consumes only selected settled rows from the requested thread', async () => { const selectedId = uuid(); const otherId = uuid(); @@ -175,6 +366,111 @@ describe('AgentBackgroundJobRepository', () => { expect(foreign?.notifiedAt).toBeNull(); }); + it('pauses only after checkpoint persistence and execution finalization', async () => { + const checkpoints = Container.get(AgentCheckpointRepository); + const checkpointStorage = Container.get(N8NCheckpointStorage); + const threads = Container.get(AgentExecutionThreadRepository); + const childThreadId = uuid(); + const runId = uuid(); + const id = uuid(); + const executionId = uuid(); + const agent = await agentRepository.findOneByOrFail({ id: agentId }); + await threads.insert({ + id: childThreadId, + projectId: agent.projectId, + agentId, + agentName: agent.name, + }); + await repository.manager.insert(AgentExecution, { + id: executionId, + threadId: childThreadId, + status: 'running', + }); + await insertJob({ + id, + parentThreadId: 'parent', + subAgentId: agentId, + childThreadId, + status: 'running', + pauseRequestId: uuid(), + settledAt: null, + }); + const job = await repository.findOneByOrFail({ id }); + const checkpoint: SerializableAgentState = { + status: 'suspended', + messageList: { messages: [], historyIds: [], inputIds: [], responseIds: [] }, + persistence: { + threadId: childThreadId, + resourceId: 'draft-chat:user-1', + delegated: true, + }, + pendingToolCalls: { + approval: { + toolCallId: 'approval', + toolName: 'approve_action', + input: { id: 'item-1' }, + suspended: true, + suspendPayload: { + type: 'approval', + toolName: 'approve_action', + args: { id: 'item-1' }, + }, + resumeSchema: { type: 'object', properties: { approved: { type: 'boolean' } } }, + runId, + }, + }, + }; + const original = JSON.stringify(checkpoint); + const state = JSON.stringify({ ...checkpoint, finishReason: 'paused' }); + try { + expect(await repository.pauseIfRequested(job, 'Progress', runId, state)).toBe(false); + const updatedAt = new Date(Date.now() - 60_000); + await checkpoints.insert({ + runId, + agentId, + threadId: childThreadId, + state: original, + updatedAt, + }); + const suspension = { runId, checkpoint, serializedState: original, updatedAt }; + expect(await checkpointStorage.markUserPaused(uuid(), suspension)).toBe(false); + expect( + await checkpointStorage.markUserPaused(agentId, { + ...suspension, + updatedAt: new Date(updatedAt.getTime() - 1), + }), + ).toBe(false); + expect(await checkpoints.findByRunId(runId)).toMatchObject({ + agentId, + state: original, + updatedAt, + }); + expect(await checkpointStorage.markUserPaused(agentId, suspension)).toBe(true); + expect(await checkpointStorage.markUserPaused(agentId, suspension)).toBe(false); + expect(await checkpoints.findByRunId(runId)).toMatchObject({ + agentId, + state, + updatedAt, + }); + expect(await repository.pauseIfRequested(job, 'Progress', runId, state)).toBe(false); + await repository.manager.update(AgentExecution, executionId, { status: 'success' }); + expect(await repository.pauseIfRequested(job, 'Progress', runId, original)).toBe(false); + expect(await repository.pauseIfRequested(job, 'Progress', runId, state)).toBe(true); + expect(await repository.settleIfActive(id, { status: 'failed' }, { status: 'running' })).toBe( + false, + ); + expect(await repository.findById(id)).toMatchObject({ + status: 'paused', + timeoutAt: null, + result: 'Progress', + }); + expect(await repository.findSettledSubAgentsWithCheckpoints()).toEqual([]); + } finally { + await checkpoints.delete(runId); + await threads.delete(childThreadId); + } + }); + it('deletes old settled jobs only if their results are marked as delivered', async () => { const old = new Date(Date.now() - 40 * 24 * 60 * 60 * 1000); const consumedId = uuid(); 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 8ce293d63b0..5be018ebf7a 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 @@ -2768,6 +2768,70 @@ describe('AgentExecutionRepository', () => { }, ); + it('restores a paused checkpoint for the current execution when resume admission fails', async () => { + const services = recordingServices(); + const threadId = uuid(); + await enqueue(services, input(threadId, 'Start', 'new')); + const active = await claim(services, threadId); + const { makeAgent } = createApprovalAgentFactory(threadId, owner.id, 0); + const store = services.checkpointStorage.getStorage(agentId); + const agent = makeAgent(store); + try { + const paused = await agent.stream('Start', { + shouldPause: async () => true, + persistence: { + threadId, + resourceId: `draft-chat:${owner.id}`, + hostMetadata: { [EXECUTION_METADATA_KEY]: active.admission.executionId }, + }, + }); + expect((await collect(paused.stream)).at(-1)).toMatchObject({ finishReason: 'paused' }); + await finish(services, active); + + const recording = { ...active.recording, userMessage: null, resumeRunId: paused.runId }; + const failed = await services.executionService.startExecutionRecording( + recording, + new Date(), + ); + await expect( + agent.resumePaused({ + runId: paused.runId, + hostMetadata: { [EXECUTION_METADATA_KEY]: failed.executionId }, + onResumeClaimed: async () => { + throw new Error('Resume admission failed'); + }, + }), + ).rejects.toThrow('Resume admission failed'); + expect(await store.load(paused.runId)).toMatchObject({ + status: 'suspended', + finishReason: 'paused', + persistence: { + hostMetadata: { [EXECUTION_METADATA_KEY]: failed.executionId }, + }, + }); + await finish(services, { ...active, admission: failed }, 'error'); + + const retry = await services.executionService.startExecutionRecording( + recording, + new Date(), + ); + const resumed = await agent.resumePaused({ + runId: paused.runId, + hostMetadata: { [EXECUTION_METADATA_KEY]: retry.executionId }, + }); + expect((await collect(resumed.stream)).at(-1)).toMatchObject({ finishReason: 'stop' }); + expect( + await Container.get(AgentCheckpointRepository).findByRunId(paused.runId), + ).toMatchObject({ + expired: true, + state: null, + }); + await finish(services, { ...active, admission: retry }); + } finally { + await agent.close(); + } + }); + it('advances after approval expiry without waiting for pruning', async () => { const services = recordingServices(); const threadId = uuid(); diff --git a/packages/cli/test/migration/1790937677019-add-agent-background-job-pause.test.ts b/packages/cli/test/migration/1790937677019-add-agent-background-job-pause.test.ts new file mode 100644 index 00000000000..5c8ffb9153b --- /dev/null +++ b/packages/cli/test/migration/1790937677019-add-agent-background-job-pause.test.ts @@ -0,0 +1,157 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + undoLastSingleMigration, + type TestMigrationContext, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; +import { randomUUID } from 'node:crypto'; + +const MIGRATION_NAME = 'AddAgentBackgroundJobPause1790937677019'; + +describe('AddAgentBackgroundJobPause migration', () => { + async function withContext(fn: (context: TestMigrationContext) => Promise): Promise { + const context = createTestMigrationContext(Container.get(DataSource)); + try { + return await fn(context); + } finally { + await context.queryRunner.release(); + } + } + + beforeAll(async () => { + await Container.get(DbConnection).init(); + await withContext(async ({ queryRunner }) => await queryRunner.clearDatabase()); + await initDbUpToMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + await Container.get(DbConnection).close(); + }); + + it('preserves jobs and blocks rollback until paused jobs are settled', async () => { + const agentId = randomUUID(); + const projectId = randomUUID(); + const runningId = randomUUID(); + const pausedId = randomUUID(); + const now = new Date(); + const pauseRequestId = randomUUID(); + await withContext(async ({ escape, runQuery }) => { + await runQuery( + `INSERT INTO ${escape.tableName('project')} ("id", "name", "type", "createdAt", "updatedAt") + VALUES (:projectId, 'Project', 'personal', :now, :now)`, + { projectId, now }, + ); + await runQuery( + `INSERT INTO ${escape.tableName('agents')} + ("id", "name", "projectId", "integrations", "tools", "skills", "createdAt", "updatedAt") + VALUES (:agentId, 'Agent', :projectId, '[]', '{}', '{}', :now, :now)`, + { agentId, projectId, now }, + ); + for (const id of [runningId, pausedId]) { + await runQuery( + `INSERT INTO ${escape.tableName('agent_background_job')} + ("id", "kind", "status", "parentAgentId", "parentThreadId", "parentResourceId", "parentPrincipalHash", "title", "subAgentId", "childThreadId", "createdAt", "updatedAt") + VALUES (:id, 'subagent', 'running', :agentId, 'parent', :resourceId, 'principal', 'Research', :agentId, :id, :now, :now)`, + { id, agentId, now, resourceId: 'draft-chat:user-1' }, + ); + } + }); + + await runSingleMigration(MIGRATION_NAME); + await withContext(async ({ escape, runQuery, queryRunner, tablePrefix }) => { + await runQuery( + `UPDATE ${escape.tableName('agent_background_job')} SET "status" = 'paused', "pauseRequestId" = :pauseRequestId, "notifiedAt" = :now WHERE "id" = :pausedId`, + { pausedId, now, pauseRequestId }, + ); + const rows = await runQuery< + Array<{ id: string; status: string; pauseRequestId: string | null }> + >(`SELECT "id", "status", "pauseRequestId" FROM ${escape.tableName('agent_background_job')}`); + expect(rows).toEqual( + expect.arrayContaining([ + { id: runningId, status: 'running', pauseRequestId: null }, + { id: pausedId, status: 'paused', pauseRequestId }, + ]), + ); + const table = await queryRunner.getTable(`${tablePrefix}agent_background_job`); + expect(table?.indices).toContainEqual( + expect.objectContaining({ + columnNames: ['childExecutionId'], + isUnique: true, + }), + ); + await expect( + runQuery( + `UPDATE ${escape.tableName('agent_background_job')} SET "status" = 'unknown' WHERE "id" = :pausedId`, + { pausedId }, + ), + ).rejects.toThrow(); + }); + + await expect(undoLastSingleMigration()).rejects.toThrow( + 'Cannot revert background job pause support while paused jobs exist.', + ); + await withContext(async ({ escape, runQuery }) => { + const rows = await runQuery< + Array<{ + id: string; + status: string; + pauseRequestId: string | null; + error: string | null; + notifiedAt: Date | null; + settledAt: Date | null; + }> + >( + `SELECT ${escape.columnName('id')}, ${escape.columnName('status')}, ${escape.columnName('pauseRequestId')}, + ${escape.columnName('error')}, ${escape.columnName('notifiedAt')}, ${escape.columnName('settledAt')} + FROM ${escape.tableName('agent_background_job')}`, + ); + expect(rows.find(({ id }) => id === runningId)).toMatchObject({ + status: 'running', + pauseRequestId: null, + error: null, + notifiedAt: null, + settledAt: null, + }); + expect(rows.find(({ id }) => id === pausedId)).toMatchObject({ + status: 'paused', + pauseRequestId, + error: null, + notifiedAt: expect.anything(), + settledAt: null, + }); + await runQuery( + `UPDATE ${escape.tableName('agent_background_job')} + SET ${escape.columnName('status')} = 'completed', ${escape.columnName('settledAt')} = :now + WHERE ${escape.columnName('id')} = :pausedId`, + { pausedId, now }, + ); + }); + + await undoLastSingleMigration(); + await withContext(async ({ escape, runQuery, queryRunner, tablePrefix }) => { + expect( + await queryRunner.hasColumn(`${tablePrefix}agent_background_job`, 'pauseRequestId'), + ).toBe(false); + const rows = await runQuery>( + `SELECT ${escape.columnName('id')}, ${escape.columnName('status')} + FROM ${escape.tableName('agent_background_job')}`, + ); + expect(rows).toEqual( + expect.arrayContaining([ + { id: runningId, status: 'running' }, + { id: pausedId, status: 'completed' }, + ]), + ); + await expect( + runQuery( + `UPDATE ${escape.tableName('agent_background_job')} SET "status" = 'paused' WHERE "id" = :pausedId`, + { pausedId }, + ), + ).rejects.toThrow(); + }); + }); +}); diff --git a/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue b/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue index 568f84ed5aa..d07061009b5 100644 --- a/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue +++ b/packages/frontend/editor-ui/src/features/agents/components/AgentChatPanel.vue @@ -397,19 +397,24 @@ async function respondToBackgroundApproval( } } const backgroundJobRows = computed(() => - backgroundJobs.value.map((job) => ({ - ...job, - label: locale.baseText( - job.kind === 'workflow' - ? 'agents.chat.backgroundTasks.workflow' - : 'agents.chat.backgroundTasks.subagent', - { interpolate: { title: job.title } }, - ), - indicator: - backgroundJobStatuses.value[ - job.kind === 'workflow' && job.status === 'running' ? 'waiting' : job.status - ], - })), + backgroundJobs.value.flatMap((job) => { + if (job.status === 'paused') return []; + return [ + { + ...job, + label: locale.baseText( + job.kind === 'workflow' + ? 'agents.chat.backgroundTasks.workflow' + : 'agents.chat.backgroundTasks.subagent', + { interpolate: { title: job.title } }, + ), + indicator: + backgroundJobStatuses.value[ + job.kind === 'workflow' && job.status === 'running' ? 'waiting' : job.status + ], + }, + ]; + }), ); const now = ref(Date.now()); const documentVisibility = useDocumentVisibility();