mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
feat(core): Add durable pause checkpoints for background sub-agents (no-changelog) (#40118)
This commit is contained in:
Robin Braumann
GitHub
parent
a3ae96e87b
commit
ea9021ed86
3
docs/generated/postgres-schema/README.md
generated
3
docs/generated/postgres-schema/README.md
generated
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
1
docs/generated/postgres-schema/public.agents.md
generated
1
docs/generated/postgres-schema/public.agents.md
generated
|
|
@ -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
|
||||
|
|
|
|||
3
docs/generated/sqlite-schema/README.md
generated
3
docs/generated/sqlite-schema/README.md
generated
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
<summary><strong>Table Definition</strong></summary>
|
||||
|
||||
```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)
|
||||
```
|
||||
|
||||
</details>
|
||||
|
|
@ -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
|
||||
|
|
|
|||
1
docs/generated/sqlite-schema/agents.md
generated
1
docs/generated/sqlite-schema/agents.md
generated
|
|
@ -114,6 +114,7 @@ erDiagram
|
|||
varchar_64_ parentPrincipalHash
|
||||
varchar_255_ parentResourceId
|
||||
varchar_128_ parentThreadId
|
||||
varchar pauseRequestId
|
||||
TEXT result
|
||||
datetime_3_ settledAt
|
||||
varchar_16_ status
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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<ResumeOptions, 'toolCallId'> & ExecutionOptions,
|
||||
): Promise<StreamResult> {
|
||||
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<void> {
|
||||
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);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -78,7 +78,7 @@ export class GenerateSink implements RunOutputSink<GenerateResult> {
|
|||
return {
|
||||
runId: suspendRunId,
|
||||
messages: list.responseDelta(),
|
||||
finishReason: 'tool-calls',
|
||||
finishReason: emission.finishReason ?? 'tool-calls',
|
||||
usage,
|
||||
pendingSuspend: suspensions.map((s) => ({
|
||||
runId: suspendRunId,
|
||||
|
|
|
|||
|
|
@ -454,7 +454,7 @@ export class StreamSink implements RunOutputSink<void> {
|
|||
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,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -837,6 +837,20 @@ export class Agent implements BuiltAgent, AgentBuilder {
|
|||
}
|
||||
}
|
||||
|
||||
async resumePaused(
|
||||
options: Omit<ResumeOptions, 'toolCallId'> & ExecutionOptions,
|
||||
): Promise<StreamResult> {
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<boolean>;
|
||||
/** Commit additional input between model calls. Stream consumers must acknowledge input-boundary chunks. */
|
||||
onInputBoundary?: (boundary: AgentInputBoundary) => Promise<AgentDbMessage[]>;
|
||||
maxIterations?: number;
|
||||
|
|
@ -434,6 +439,11 @@ export interface BuiltAgent {
|
|||
options: ResumeOptions & ExecutionOptions,
|
||||
): Promise<StreamResult>;
|
||||
|
||||
/** Resume a user pause without repeating completed tools. */
|
||||
resumePaused(
|
||||
options: Omit<ResumeOptions, 'toolCallId'> & ExecutionOptions,
|
||||
): Promise<StreamResult>;
|
||||
|
||||
/** Approve a tool that uses requireApproval or needsApprovalFn */
|
||||
approve(method: 'generate', options: ResumeOptions & ExecutionOptions): Promise<GenerateResult>;
|
||||
approve(method: 'stream', options: ResumeOptions & ExecutionOptions): Promise<StreamResult>;
|
||||
|
|
|
|||
|
|
@ -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<AgentBackgroundJobDto, 'id' | 'title' | 'kind'> & {
|
||||
status: Exclude<AgentBackgroundJobDto['status'], 'running' | 'suspended'>;
|
||||
status: Exclude<AgentBackgroundJobDto['status'], 'running' | 'suspended' | 'paused'>;
|
||||
}
|
||||
>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Array<{ id: string }>>(
|
||||
`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'`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -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':
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -153,16 +153,15 @@ export class AgentBackgroundJobService {
|
|||
async registerSubAgentJob(
|
||||
params: Omit<NewSubAgentJob, 'kind' | 'timeoutAt'>,
|
||||
): Promise<BackgroundJobReceipt> {
|
||||
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 };
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -302,6 +302,25 @@ export class N8NCheckpointStorage {
|
|||
}
|
||||
}
|
||||
|
||||
async markUserPaused(
|
||||
agentId: string,
|
||||
suspension: {
|
||||
runId: string;
|
||||
checkpoint: SerializableAgentState;
|
||||
serializedState: string;
|
||||
updatedAt: Date;
|
||||
},
|
||||
): Promise<boolean> {
|
||||
// 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,
|
||||
|
|
|
|||
|
|
@ -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<AgentBackgroundJobStatus, 'running' | 'suspended'>;
|
||||
status: Exclude<AgentBackgroundJobStatus, 'running' | 'suspended' | 'paused'>;
|
||||
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<AgentBackgroundJob> {
|
||||
constructor(dataSource: DataSource) {
|
||||
super(AgentBackgroundJob, dataSource.manager);
|
||||
export class AgentBackgroundJobRepository extends BaseRepository<AgentBackgroundJob> {
|
||||
constructor(
|
||||
dataSource: DataSource,
|
||||
transactionRunner: TransactionRunner,
|
||||
private readonly threadRepository: AgentExecutionThreadRepository,
|
||||
) {
|
||||
super(AgentBackgroundJob, dataSource.manager, transactionRunner);
|
||||
}
|
||||
|
||||
async insertJob(job: NewAgentBackgroundJob): Promise<void> {
|
||||
await this.insert({ ...job, status: 'running' });
|
||||
async insertSubAgentJobIfCapacity(job: NewSubAgentJob, limit: number): Promise<boolean> {
|
||||
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<AgentBackgroundJob>
|
|||
}
|
||||
|
||||
/** Suspended children still occupy a job slot. */
|
||||
async countActiveSubAgentsByParentThread(parentThreadId: string): Promise<number> {
|
||||
return await this.count({
|
||||
async countActiveSubAgentsByParentThread(
|
||||
parentThreadId: string,
|
||||
ctx: OperationContext = {},
|
||||
): Promise<number> {
|
||||
return await this.managerFor(ctx).count(AgentBackgroundJob, {
|
||||
where: { parentThreadId, kind: 'subagent', status: In(['running', 'suspended']) },
|
||||
});
|
||||
}
|
||||
|
|
@ -113,25 +130,44 @@ export class AgentBackgroundJobRepository extends Repository<AgentBackgroundJob>
|
|||
|
||||
/** Results and approval requests that still need delivery. */
|
||||
async findWakeableUnconsumed(parentThreadId: string): Promise<AgentBackgroundJob[]> {
|
||||
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<number> {
|
||||
async markMailConsumed(
|
||||
parentThreadId: string,
|
||||
ids: string[],
|
||||
includePauseReports = false,
|
||||
): Promise<number> {
|
||||
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<AgentBackgroundJob>
|
|||
return await this.find({ where: kind ? { status: 'running', kind } : { status: 'running' } });
|
||||
}
|
||||
|
||||
async findRequestedPauses(parentThreadId?: string): Promise<AgentBackgroundJob[]> {
|
||||
return await this.find({
|
||||
where: {
|
||||
kind: 'subagent',
|
||||
pauseRequestId: Not(IsNull()),
|
||||
status: In(['running', 'suspended', 'paused']),
|
||||
...(parentThreadId ? { parentThreadId } : {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async findSettledSubAgentsWithCheckpoints(): Promise<AgentBackgroundJob[]> {
|
||||
return await this.createQueryBuilder('job')
|
||||
.innerJoin(
|
||||
|
|
@ -187,12 +234,33 @@ export class AgentBackgroundJobRepository extends Repository<AgentBackgroundJob>
|
|||
'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<AgentBackgroundJob[]> {
|
||||
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<AgentBackgroundJob[]> {
|
||||
return await this.find({
|
||||
|
|
@ -217,12 +285,73 @@ export class AgentBackgroundJobRepository extends Repository<AgentBackgroundJob>
|
|||
|
||||
async resumeIfSuspended(id: string, timeoutAt: Date): Promise<boolean> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
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<boolean> {
|
||||
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<AgentBackgroundJob>
|
|||
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;
|
||||
|
|
|
|||
|
|
@ -80,6 +80,20 @@ export class AgentCheckpointRepository extends BaseRepository<AgentCheckpoint> {
|
|||
return (result.affected ?? 0) > 0;
|
||||
}
|
||||
|
||||
async markUserPaused(
|
||||
runId: string,
|
||||
agentId: string,
|
||||
state: string,
|
||||
pausedState: string,
|
||||
updatedAt: Date,
|
||||
): Promise<boolean> {
|
||||
const result = await this.update(
|
||||
{ runId, agentId, state, expired: false, updatedAt },
|
||||
{ state: pausedState, updatedAt },
|
||||
);
|
||||
return result.affected === 1;
|
||||
}
|
||||
|
||||
async expireByRunIdAndAgentId(
|
||||
runId: string,
|
||||
agentId: string,
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<T>(fn: (context: TestMigrationContext) => Promise<T>): Promise<T> {
|
||||
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<Array<{ id: string; status: string }>>(
|
||||
`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();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user